Don't reuse closed tcp connections (#70)

Signed-off-by: turuslan <[email protected]>

Author: turuslan

src/network/impl/connection_manager_impl.cpp

@@ -22,13 +22,13 @@ namespace libp2p::network {
   ConnectionManager::ConnectionSPtr
   ConnectionManagerImpl::getBestConnectionForPeer(const peer::PeerId &p) const {
     // TODO(warchant): maybe make pluggable strategies
-    auto c = getConnectionsToPeer(p);
-    if (c.empty()) {
-      return nullptr;
+    for (auto &conn : getConnectionsToPeer(p)) {
+      if (!conn->isClosed()) {
+        // for now, return first connection
+        return conn;
+      }
     }
-
-    // for now, return first connection
-    return c[0];
+    return nullptr;
   }
 
   ConnectionManager::Connectedness ConnectionManagerImpl::connectedness(

src/transport/tcp/tcp_connection.cpp

@@ -76,48 +76,42 @@ namespace libp2p::transport {
                                });
   }
 
+  template <typename Callback>
+  auto closeOnError(TcpConnection &conn, Callback &&cb) {
+    return [cb{std::move(cb)}, conn{conn.shared_from_this()}](auto &&ec,
+                                                              auto &&result) {
+      if (ec == boost::asio::error::broken_pipe) {
+        std::ignore = conn->close();
+      }
+      if (ec) {
+        return cb(std::forward<decltype(ec)>(ec));
+      }
+      cb(result);
+    };
+  }
+
   void TcpConnection::read(gsl::span<uint8_t> out, size_t bytes,
                            TcpConnection::ReadCallbackFunc cb) {
     boost::asio::async_read(socket_, detail::makeBuffer(out, bytes),
-                            [cb = std::move(cb)](auto &&ec, auto &&read) {
-                              if (ec) {
-                                return cb(std::forward<decltype(ec)>(ec));
-                              }
-                              return cb(read);
-                            });
+                            closeOnError(*this, cb));
   }
 
   void TcpConnection::readSome(gsl::span<uint8_t> out, size_t bytes,
                                TcpConnection::ReadCallbackFunc cb) {
     socket_.async_read_some(detail::makeBuffer(out, bytes),
-                            [cb = std::move(cb)](auto &&ec, auto &&read) {
-                              if (ec) {
-                                return cb(std::forward<decltype(ec)>(ec));
-                              }
-                              return cb(read);
-                            });
+                            closeOnError(*this, cb));
   }
 
   void TcpConnection::write(gsl::span<const uint8_t> in, size_t bytes,
                             TcpConnection::WriteCallbackFunc cb) {
     boost::asio::async_write(socket_, detail::makeBuffer(in, bytes),
-                             [cb = std::move(cb)](auto &&ec, auto &&written) {
-                               if (ec) {
-                                 return cb(std::forward<decltype(ec)>(ec));
-                               }
-                               return cb(written);
-                             });
+                             closeOnError(*this, cb));
   }
 
   void TcpConnection::writeSome(gsl::span<const uint8_t> in, size_t bytes,
                                 TcpConnection::WriteCallbackFunc cb) {
     socket_.async_write_some(detail::makeBuffer(in, bytes),
-                             [cb = std::move(cb)](auto &&ec, auto &&written) {
-                               if (ec) {
-                                 return cb(std::forward<decltype(ec)>(ec));
-                               }
-                               return cb(written);
-                             });
+                             closeOnError(*this, cb));
   }
 
 }  // namespace libp2p::transport