SERVER-34636 enable opportunistic writes with ssl

Add a side channel to the asio ssl streams which allows us to capture
the work remaining for the tcp send in the event of a full send buffer.

This makes opportunistic writes safe for ssl sockets

Author: hanumantmk

src/mongo/transport/session_asio.h

@@ -379,13 +379,19 @@ class TransportLayerASIO::ASIOSession final : public Session {
 #ifdef MONGO_CONFIG_SSL
         _ranHandshake = true;
         if (_sslSocket) {
+#ifdef __linux__
+            // We do some trickery in asio (see moreToSend), which appears to work well on linux,
+            // but fails on other platforms.
+            return opportunisticWrite(*_sslSocket, buffers);
+#else
             if (_blockingMode == Async) {
                 // Opportunistic writes are broken for async egress SSL (switching between blocking
                 // and non-blocking mode corrupts the TLS exchange).
                 return asio::async_write(*_sslSocket, buffers, UseFuture{});
             } else {
                 return opportunisticWrite(*_sslSocket, buffers);
             }
+#endif
         }
 #endif
         return opportunisticWrite(_socket, buffers);
@@ -414,19 +420,60 @@ class TransportLayerASIO::ASIOSession final : public Session {
         }
     }
 
+    /**
+     * moreToSend checks the ssl socket after an opportunisticWrite.  If there are still bytes to
+     * send, we manually send them off the underlying socket.  Then we hook that up with a future
+     * that gets us back to sending from the ssl side.
+     *
+     * There are two variants because we call opportunisticWrite on generic sockets and ssl sockets.
+     * The generic socket impl never has more to send (because it doesn't have an inner socket it
+     * needs to keep sending).
+     */
+    template <typename ConstBufferSequence>
+    boost::optional<Future<size_t>> moreToSend(GenericSocket& socket,
+                                               const ConstBufferSequence& buffers,
+                                               size_t size) {
+        return boost::none;
+    }
+
+#ifdef MONGO_CONFIG_SSL
+    template <typename ConstBufferSequence>
+    boost::optional<Future<size_t>> moreToSend(asio::ssl::stream<GenericSocket>& socket,
+                                               const ConstBufferSequence& buffers,
+                                               size_t sizeFromBefore) {
+        if (_sslSocket->getCoreOutputBuffer().size()) {
+            return opportunisticWrite(getSocket(), _sslSocket->getCoreOutputBuffer())
+                .then([this, &socket, buffers, sizeFromBefore](size_t) {
+                    return opportunisticWrite(socket, buffers)
+                        .then([sizeFromBefore](size_t justWritten) {
+                            return justWritten + sizeFromBefore;
+                        });
+                });
+        }
+
+        return boost::none;
+    }
+#endif
+
     template <typename Stream, typename ConstBufferSequence>
     Future<size_t> opportunisticWrite(Stream& stream, const ConstBufferSequence& buffers) {
         std::error_code ec;
         auto size = asio::write(stream, buffers, ec);
         if (((ec == asio::error::would_block) || (ec == asio::error::try_again)) &&
             (_blockingMode == Async)) {
+
             // asio::write is a loop internally, so some of buffers may have been read into already.
             // So we need to adjust the buffers passed into async_write to be offset by size, if
             // size is > 0.
             ConstBufferSequence asyncBuffers(buffers);
             if (size > 0) {
                 asyncBuffers += size;
             }
+
+            if (auto more = moreToSend(stream, asyncBuffers, size)) {
+                return std::move(*more);
+            }
+
             return asio::async_write(stream, asyncBuffers, UseFuture{})
                 .then([size](size_t asyncSize) {
                     // Add back in the size written opportunistically.

src/mongo/util/net/ssl/detail/io.hpp

@@ -48,18 +48,20 @@ std::size_t io(Stream& next_layer, stream_core& core, const Operation& op, asio:
 
             case engine::want_output_and_retry:
 
+                core.output_ = core.engine_.get_output(core.output_buffer_);
                 // Get output data from the engine and write it to the underlying
                 // transport.
-                asio::write(next_layer, core.engine_.get_output(core.output_buffer_), ec);
+                core.output_ += asio::write(next_layer, core.output_, ec);
 
                 // Try the operation again.
                 continue;
 
             case engine::want_output:
 
+                core.output_ = core.engine_.get_output(core.output_buffer_);
                 // Get output data from the engine and write it to the underlying
                 // transport.
-                asio::write(next_layer, core.engine_.get_output(core.output_buffer_), ec);
+                core.output_ += asio::write(next_layer, core.output_, ec);
 
                 // Operation is complete. Return result to caller.
                 core.engine_.map_error_code(ec);
@@ -75,7 +77,7 @@ std::size_t io(Stream& next_layer, stream_core& core, const Operation& op, asio:
 
     // Operation failed. Return result to caller.
     core.engine_.map_error_code(ec);
-    return 0;
+    return bytes_transferred;
 }
 
 template <typename Stream, typename Operation, typename Handler>

src/mongo/util/net/ssl/detail/stream_core.hpp

@@ -125,6 +125,9 @@ struct stream_core {
 
     // The buffer pointing to the engine's unconsumed input.
     asio::const_buffer input_;
+
+    // The buffer pointing to the engine's unconsumed output.
+    asio::mutable_buffer output_;
 };
 
 }  // namespace detail

src/mongo/util/net/ssl/stream.hpp

@@ -577,6 +577,10 @@ class stream : public stream_base, private noncopyable {
         return init.result.get();
     }
 
+    asio::mutable_buffer& getCoreOutputBuffer() {
+        return core_.output_;
+    }
+
 private:
     Stream next_layer_;
     detail::stream_core core_;