Hotfix/streams (#55)

* hotfixes to yamux and dialer

* streams hotfix: yamux update

* unneeded comment removed

* unndeded std::forward changed

* one minor cosmetic fix

Author: art-gor

include/libp2p/common/trace.hpp

@@ -0,0 +1,33 @@
+/**
+ * Copyright Soramitsu Co., Ltd. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#include <libp2p/common/logger.hpp>
+
+#ifndef LIBP2P_COMMON_TRACE_HPP
+#define LIBP2P_COMMON_TRACE_HPP
+
+namespace libp2p::common {
+
+  /**
+   * Special debug utility function, allows for not having logger as member
+   * field
+   */
+  template <typename... Args>
+  inline void traceToDebugLogger(spdlog::string_view_t fmt,
+                                 const Args &... args) {
+    static const std::string tag("debug");
+    auto log = common::createLogger(tag);
+    log->trace(fmt, args...);
+  }
+
+}  // namespace libp2p::common
+
+#if TRACE_ENABLED
+#define TRACE(...) libp2p::common::traceToDebugLogger(__VA_ARGS__)
+#else
+#define TRACE(...)
+#endif
+
+#endif  // LIBP2P_COMMON_TRACE_HPP

include/libp2p/muxer/yamux/yamux_stream.hpp

@@ -90,7 +90,10 @@ namespace libp2p::connection {
     void write(gsl::span<const uint8_t> in, size_t bytes, WriteCallbackFunc cb,
                bool some);
 
+    /// this stream's connection
     std::weak_ptr<YamuxedConnection> yamuxed_connection_;
+
+    /// id of this stream
     YamuxedConnection::StreamId stream_id_;
 
     /// is the stream opened for reads?
@@ -99,8 +102,10 @@ namespace libp2p::connection {
     /// is the stream opened for writes?
     bool is_writable_ = true;
 
-    /// default sliding window size of the stream - how much unread bytes can be
-    /// on both sides
+    /**
+     * default sliding window size of the stream - how much unread bytes can be
+     * on both sides
+     */
     static constexpr uint32_t kDefaultWindowSize = 256 * 1024;  // in bytes
 
     /// how much unacked bytes can we have on our side
@@ -118,9 +123,48 @@ namespace libp2p::connection {
     /// is the stream reading right now?
     bool is_reading_ = false;
 
+    /// read callback, non-zero during async data receive
+    ReadCallbackFunc read_cb_;
+
+    /// client's read buffer
+    gsl::span<uint8_t> external_read_buffer_;
+
+    /// bytes count client is waiting for, non-zero during async data receive
+    size_t bytes_waiting_ = 0;
+
+    /// client makes readSome operation
+    bool reading_some_ = false;
+
+    /// starts async read operation
+    void beginRead(ReadCallbackFunc cb, gsl::span<uint8_t> out, size_t bytes,
+                   bool some);
+
+    /// ends async read operation
+    void endRead(outcome::result<size_t> result);
+
+    /// Tries to consume requested bytes from already received data
+    outcome::result<size_t> tryConsumeReadBuffer(gsl::span<uint8_t> out,
+                                                 size_t bytes, bool some);
+
+    /**
+     * Forwards read buffer and receive window and acknowledges bytes received
+     * in async manner
+     * @param bytes number of bytes to ack
+     */
+    void sendAck(size_t bytes);
+
     /// is the stream writing right now?
     bool is_writing_ = false;
 
+    /// write callback, non-zero during async sends
+    WriteCallbackFunc write_cb_;
+
+    /// starts async write operation
+    void beginWrite(WriteCallbackFunc cb);
+
+    /// ends async write operation
+    void endWrite(outcome::result<size_t> result);
+
     /// YamuxedConnection API starts here
     friend class YamuxedConnection;
 
@@ -137,6 +181,9 @@ namespace libp2p::connection {
      */
     outcome::result<void> commitData(gsl::span<const uint8_t> data,
                                      size_t data_size);
+
+    /// Called by connection on reset
+    void onConnectionReset(outcome::result<size_t> reason);
   };
 }  // namespace libp2p::connection
 

include/libp2p/muxer/yamux/yamuxed_connection.hpp

@@ -39,7 +39,8 @@ namespace libp2p::connection {
       TOO_MANY_STREAMS,
       FORBIDDEN_CALL,
       OTHER_SIDE_ERROR,
-      INTERNAL_ERROR
+      INTERNAL_ERROR,
+      CLOSED_BY_PEER,
     };
 
     /**
@@ -104,9 +105,6 @@ namespace libp2p::connection {
     // indicates whether start() has been executed or not
     bool started_ = false;
 
-    // XXX
-    bool new_stream_pending_ = false;
-
     /**
      * Write message to the connection; ensures no more than one wright
      * would be executed at one time
@@ -169,7 +167,7 @@ namespace libp2p::connection {
     /**
      * Reset all streams, which were created over this connection
      */
-    void resetAllStreams();
+    void resetAllStreams(outcome::result<void> reason);
 
     /**
      * Find stream with such id in local streams
@@ -236,13 +234,21 @@ namespace libp2p::connection {
     /**
      * Close this Yamux session
      */
-    void closeSession();
+    void closeSession(outcome::result<void> reason);
 
+    /// Underlying connection
     std::shared_ptr<SecureConnection> connection_;
+
+    /// Handler for new inbound streams
     NewStreamHandlerFunc new_stream_handler_;
+
+    /// Config constants
     muxer::MuxedConnectionConfig config_;
 
+    /// Last stream id to be incremented
     uint32_t last_created_stream_id_;
+
+    /// Streams
     std::unordered_map<StreamId, std::shared_ptr<YamuxStream>> streams_;
 
     libp2p::common::Logger log_ = libp2p::common::createLogger("yx-conn");
@@ -265,18 +271,6 @@ namespace libp2p::connection {
     void streamOnWindowUpdate(StreamId stream_id, NotifyeeCallback cb);
     std::map<StreamId, NotifyeeCallback> window_updates_subs_;
 
-    /**
-     * Add a handler function, which is called, when data for a particular
-     * stream is received
-     * @param stream_id of the stream which is to be notified
-     * @param handler to be called; if it returns true, it's removed from
-     * the list of handlers for that stream
-     * @note this is done through a function and not event emitters, as each
-     * stream is to receive that event independently based on id
-     */
-    void streamOnAddData(StreamId stream_id, NotifyeeCallback cb);
-    std::map<StreamId, NotifyeeCallback> data_subs_;
-
     /**
      * Write bytes to the connection; before calling this method, the stream
      * must ensure that no write operations are currently running

include/libp2p/network/impl/dialer_impl.hpp

@@ -8,6 +8,7 @@
 
 #include <libp2p/network/connection_manager.hpp>
 #include <libp2p/network/dialer.hpp>
+#include <libp2p/network/listener_manager.hpp>
 #include <libp2p/network/transport_manager.hpp>
 #include <libp2p/protocol_muxer/protocol_muxer.hpp>
 
@@ -19,7 +20,8 @@ namespace libp2p::network {
 
     DialerImpl(std::shared_ptr<protocol_muxer::ProtocolMuxer> multiselect,
                std::shared_ptr<TransportManager> tmgr,
-               std::shared_ptr<ConnectionManager> cmgr);
+               std::shared_ptr<ConnectionManager> cmgr,
+               std::shared_ptr<ListenerManager> listener);
 
     // Establishes a connection to a given peer
     void dial(const peer::PeerInfo &p, DialResultFunc cb) override;
@@ -33,8 +35,7 @@ namespace libp2p::network {
     std::shared_ptr<protocol_muxer::ProtocolMuxer> multiselect_;
     std::shared_ptr<TransportManager> tmgr_;
     std::shared_ptr<ConnectionManager> cmgr_;
-
-    common::Logger log_ = common::createLogger("debug"); // XXX
+    std::shared_ptr<ListenerManager> listener_;
   };
 
 }  // namespace libp2p::network

include/libp2p/network/impl/listener_manager_impl.hpp

@@ -50,6 +50,10 @@ namespace libp2p::network {
 
     Router &getRouter() override;
 
+    void onConnection(
+        outcome::result<std::shared_ptr<connection::CapableConnection>> rconn)
+        override;
+
    private:
     bool started = false;
 
@@ -61,9 +65,6 @@ namespace libp2p::network {
     std::shared_ptr<network::Router> router_;
     std::shared_ptr<TransportManager> tmgr_;
     std::shared_ptr<ConnectionManager> cmgr_;
-
-    void onConnection(
-        outcome::result<std::shared_ptr<connection::CapableConnection>> rconn);
   };
 
 }  // namespace libp2p::network

include/libp2p/network/impl/network_impl.hpp

@@ -15,7 +15,7 @@ namespace libp2p::network {
    public:
     ~NetworkImpl() override = default;
 
-    NetworkImpl(std::unique_ptr<ListenerManager> listener,
+    NetworkImpl(std::shared_ptr<ListenerManager> listener,
                 std::unique_ptr<Dialer> dialer,
                 std::shared_ptr<ConnectionManager> cmgr);
 
@@ -28,7 +28,7 @@ namespace libp2p::network {
     ConnectionManager &getConnectionManager() override;
 
    private:
-    std::unique_ptr<ListenerManager> listener_;
+    std::shared_ptr<ListenerManager> listener_;
     std::unique_ptr<Dialer> dialer_;
     std::shared_ptr<ConnectionManager> cmgr_;
   };

include/libp2p/network/listener_manager.hpp

@@ -11,8 +11,8 @@
 #include <libp2p/event/bus.hpp>
 #include <libp2p/multi/multiaddress.hpp>
 #include <libp2p/network/router.hpp>
-#include <libp2p/protocol/base_protocol.hpp>
 #include <libp2p/outcome/outcome.hpp>
+#include <libp2p/protocol/base_protocol.hpp>
 
 namespace libp2p::network {
 
@@ -117,6 +117,13 @@ namespace libp2p::network {
      * @brief Getter for Router.
      */
     virtual Router &getRouter() = 0;
+
+    /**
+     * @brief Allows new connections for accepting incoming streams
+     */
+    virtual void onConnection(
+        outcome::result<std::shared_ptr<connection::CapableConnection>>
+            rconn) = 0;
   };
 
 }  // namespace libp2p::network

src/muxer/yamux/yamux_frame.cpp

@@ -8,6 +8,9 @@
 #include <libp2p/common/byteutil.hpp>
 #include <libp2p/muxer/yamux/yamux_frame.hpp>
 
+#define TRACE_ENABLED 0
+#include <libp2p/common/trace.hpp>
+
 namespace libp2p::connection {
   YamuxFrame::ByteArray YamuxFrame::frameBytes(uint8_t version, FrameType type,
                                                Flag flag, uint32_t stream_id,
@@ -34,24 +37,28 @@ namespace libp2p::connection {
   }
 
   YamuxFrame::ByteArray newStreamMsg(YamuxFrame::StreamId stream_id) {
+    TRACE("yamux newStreamMsg, stream_id={}", stream_id);
     return YamuxFrame::frameBytes(YamuxFrame::kDefaultVersion,
                                   YamuxFrame::FrameType::DATA,
                                   YamuxFrame::Flag::SYN, stream_id, 0);
   }
 
   YamuxFrame::ByteArray ackStreamMsg(YamuxFrame::StreamId stream_id) {
+    TRACE("yamux ackStreamMsg, stream_id={}", stream_id);
     return YamuxFrame::frameBytes(YamuxFrame::kDefaultVersion,
                                   YamuxFrame::FrameType::DATA,
                                   YamuxFrame::Flag::ACK, stream_id, 0);
   }
 
   YamuxFrame::ByteArray closeStreamMsg(YamuxFrame::StreamId stream_id) {
+    TRACE("yamux closeStreamMsg, stream_id={}", stream_id);
     return YamuxFrame::frameBytes(YamuxFrame::kDefaultVersion,
                                   YamuxFrame::FrameType::DATA,
                                   YamuxFrame::Flag::FIN, stream_id, 0);
   }
 
   YamuxFrame::ByteArray resetStreamMsg(YamuxFrame::StreamId stream_id) {
+    TRACE("yamux resetStreamMsg, stream_id={}", stream_id);
     return YamuxFrame::frameBytes(YamuxFrame::kDefaultVersion,
                                   YamuxFrame::FrameType::DATA,
                                   YamuxFrame::Flag::RST, stream_id, 0);
@@ -71,13 +78,15 @@ namespace libp2p::connection {
 
   YamuxFrame::ByteArray dataMsg(YamuxFrame::StreamId stream_id,
                                 gsl::span<const uint8_t> data) {
+    TRACE("yamux dataMsg, stream_id={}, size={}", stream_id, data.size());
     return YamuxFrame::frameBytes(YamuxFrame::kDefaultVersion,
                                   YamuxFrame::FrameType::DATA,
                                   YamuxFrame::Flag::NONE, stream_id,
                                   static_cast<uint32_t>(data.size()), data);
   }
 
   YamuxFrame::ByteArray goAwayMsg(YamuxFrame::GoAwayError error) {
+    TRACE("yamux goAwayMsg");
     return YamuxFrame::frameBytes(
         YamuxFrame::kDefaultVersion, YamuxFrame::FrameType::GO_AWAY,
         YamuxFrame::Flag::NONE, 0, static_cast<uint32_t>(error));