rpc: Fix crash during connection teardown

_buf_write.close() races with _fd.shutdown_output(). If the former
happens first, it will close the socket. It is illegal to use a closed
socket, so shutdown_output() was causing a crash.

The code was protecting against this by checking if _send_loop_stopped
is ready, but there is a deferring point between finally() callback
containing _buf_write.close() is executed and the future associated
with it is resolved, so this check is not enough.

This patch fixes the problem by replacing the check to
_send_loop_stopped.available() with a fully reliable check of a flag,
which is set immediately before the stream is closed.

The complication could be avoided by changing implementation of posix
connected_socket to not close the whole socket but only shutdown their
read/write side. In which case shutdown_output() could be called
concurrently with _buf_write.close(). But this is a bigger change so
was postponed.

Fixes scylladb#210.

Message-Id: <[email protected]>

rpc/rpc.hh

@@ -112,6 +112,7 @@ class protocol {
         input_stream<char> _read_buf;
         output_stream<char> _write_buf;
         bool _error = false;
+        bool _write_side_closed = false;
         protocol& _proto;
         bool _connected = false;
         promise<> _stopped;
@@ -210,15 +211,15 @@ class protocol {
             }).handle_exception([this] (std::exception_ptr eptr) {
                 _error = true;
             }).finally([this] {
+                _write_side_closed = true;
                 return _write_buf.close();
             });
         }
 
         future<> stop_send_loop() {
             _error = true;
-            if (!_send_loop_stopped.available()) {
-                // if _send_loop_stopped is ready it means that _fd is closed already
-                // and nobody waits on _outgoing_queue_cond
+            // We must not call shutdown_output() concurrently with or after _write_buf.close()
+            if (_connected && !_write_side_closed) {
                 _outgoing_queue_cond.broken();
                 _fd.shutdown_output();
             }