Fix pending callback not called when ServerConnection destructs (#3572)

src/ray/common/client_connection.cc

@@ -33,6 +33,14 @@ ServerConnection<T>::ServerConnection(boost::asio::basic_stream_socket<T> &&sock
       async_write_queue_(),
       async_write_in_flight_(false) {}
 
+template <class T>
+ServerConnection<T>::~ServerConnection() {
+  // If there are any pending messages, invoke their callbacks with an IOError status.
+  for (const auto &write_buffer : async_write_queue_) {
+    write_buffer->handler(Status::IOError("Connection closed."));
+  }
+}
+
 template <class T>
 Status ServerConnection<T>::WriteBuffer(
     const std::vector<boost::asio::const_buffer> &buffer) {

src/ray/common/client_connection.h

@@ -29,6 +29,9 @@ ray::Status TcpConnect(boost::asio::ip::tcp::socket &socket,
 template <typename T>
 class ServerConnection : public std::enable_shared_from_this<ServerConnection<T>> {
  public:
+  /// ServerConnection destructor.
+  virtual ~ServerConnection();
+
   /// Allocate a new server connection.
   ///
   /// \param socket A reference to the server socket.