diff options
| -rw-r--r-- | src/mongo/transport/session_asio.h | 47 | ||||
| -rw-r--r-- | src/mongo/util/net/ssl/detail/io.hpp | 8 | ||||
| -rw-r--r-- | src/mongo/util/net/ssl/detail/stream_core.hpp | 3 | ||||
| -rw-r--r-- | src/mongo/util/net/ssl/stream.hpp | 4 |
4 files changed, 59 insertions, 3 deletions
diff --git a/src/mongo/transport/session_asio.h b/src/mongo/transport/session_asio.h index 8baf0610f3e..994b0941b04 100644 --- a/src/mongo/transport/session_asio.h +++ b/src/mongo/transport/session_asio.h @@ -379,6 +379,11 @@ private: #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). @@ -386,6 +391,7 @@ private: } else { return opportunisticWrite(*_sslSocket, buffers); } +#endif } #endif return opportunisticWrite(_socket, buffers); @@ -414,12 +420,48 @@ private: } } + /** + * 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. @@ -427,6 +469,11 @@ private: 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. diff --git a/src/mongo/util/net/ssl/detail/io.hpp b/src/mongo/util/net/ssl/detail/io.hpp index a8f0cc42047..6c0bc5f69fc 100644 --- a/src/mongo/util/net/ssl/detail/io.hpp +++ b/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> diff --git a/src/mongo/util/net/ssl/detail/stream_core.hpp b/src/mongo/util/net/ssl/detail/stream_core.hpp index 6513043f348..98bb9f7ddab 100644 --- a/src/mongo/util/net/ssl/detail/stream_core.hpp +++ b/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 diff --git a/src/mongo/util/net/ssl/stream.hpp b/src/mongo/util/net/ssl/stream.hpp index f7f2a0f31a2..1bdf28e26b5 100644 --- a/src/mongo/util/net/ssl/stream.hpp +++ b/src/mongo/util/net/ssl/stream.hpp @@ -577,6 +577,10 @@ public: return init.result.get(); } + asio::mutable_buffer& getCoreOutputBuffer() { + return core_.output_; + } + private: Stream next_layer_; detail::stream_core core_; |