Merge pull request #35 from JoyStream/development

v0.3.1

Author: mnaamani

conan_package/base.py

@@ -3,14 +3,14 @@
 
 class ProtocolSessionBase(ConanFile):
     name = "ProtocolSession"
-    version = "0.3.0"
+    version = "0.3.1"
     license = "(c) JoyStream Inc. 2016-2017"
     url = "https://github.com/JoyStream/protocol_session-cpp.git"
     repo_ssh_url = "git@github.com:JoyStream/protocol_session-cpp.git"
     repo_https_url = "https://github.com/JoyStream/protocol_session-cpp.git"
     settings = "os", "compiler", "build_type", "arch"
     generators = "cmake"
-    requires = "ProtocolStateMachine/0.3.0@joystream/stable"
+    requires = "ProtocolStateMachine/0.3.1@joystream/stable"
     build_policy = "missing"
 
     def source(self):

sources/CMakeLists.txt

@@ -17,6 +17,7 @@ set(
     src/PieceInformation.cpp
     src/common.cpp
     src/PieceDeliveryPipeline.cpp
+    src/SpeedTestPolicy.cpp
 )
 
 # === build library ===

sources/include/protocol_session/Callbacks.hpp

@@ -52,6 +52,8 @@ enum class DisconnectCause {
 
     seller_message_overflow,
 
+    seller_failed_speed_test,
+
     //// selling
 
     //buyer_invited_with_bad_terms,
@@ -62,7 +64,11 @@ enum class DisconnectCause {
 
     buyer_sent_invalid_payment,
 
-    buyer_message_overflow
+    buyer_message_overflow,
+
+    buyer_requested_too_many_speed_tests,
+
+    buyer_speed_test_payload_requested_too_large
 };
 
 // Removal of a connection from the session: c++11 alias declaration

sources/include/protocol_session/Session.cpp

@@ -47,7 +47,8 @@ namespace protocol_session {
         , _observing(nullptr)
         , _selling(nullptr)
         , _buying(nullptr)
-        , _network(network) {
+        , _network(network)
+        , _getTime(std::chrono::high_resolution_clock::now) {
 
         time(&_started);
     }
@@ -911,6 +912,26 @@ namespace protocol_session {
         assert(false);
     }
 
+    template<class ConnectionIdType>
+    void Session<ConnectionIdType>::sellerCompletedSpeedTest(const ConnectionIdType & id, bool successful) {
+
+        assert(hasConnection(id));
+        assert(_mode == SessionMode::buying);
+        assert(_observing == nullptr && _buying != nullptr && _selling == nullptr);
+
+        _buying->sellerCompletedSpeedTest(id, successful);
+    }
+
+    template<class ConnectionIdType>
+    void Session<ConnectionIdType>::buyerRequestedSpeedTest(const ConnectionIdType & id, uint32_t payloadSize) {
+
+        assert(hasConnection(id));
+        assert(_mode == SessionMode::selling);
+        assert(_observing == nullptr && _buying == nullptr && _selling != nullptr);
+
+        _selling->buyerRequestedSpeedTest(id, payloadSize);
+    }
+
     template<class ConnectionIdType>
     detail::Connection<ConnectionIdType> * Session<ConnectionIdType>::createConnection(const ConnectionIdType & id, const SendMessageOnConnectionCallbacks & sendMessageCallbacks) {
 
@@ -931,7 +952,10 @@ namespace protocol_session {
         [this, id](const protocol_wire::PieceData & p) { this->receivedFullPiece(id, p); },
         [this, id]() { this->remoteMessageOverflow(id); },
         [this, id]() { this->localMessageOverflow(id); },
-        _network);
+        [this, id](bool successful) { this->sellerCompletedSpeedTest(id, successful); },
+        [this, id](uint32_t payloadSize) { this->buyerRequestedSpeedTest(id, payloadSize); },
+        _network,
+        _getTime);
     }
 
     template <class ConnectionIdType>
@@ -1014,5 +1038,21 @@ namespace protocol_session {
 
         return connection;
     }
+
+    template <class ConnectionIdType>
+    SpeedTestPolicy Session<ConnectionIdType>::speedTestPolicy() const {
+      return _speedTestPolicy;
+    }
+
+    template <class ConnectionIdType>
+    void Session<ConnectionIdType>::setSpeedTestPolicy(const SpeedTestPolicy policy) {
+      _speedTestPolicy = policy;
+    }
+
+    template <class ConnectionIdType>
+    void Session<ConnectionIdType>::setTimeGetter(const std::function<std::chrono::high_resolution_clock::time_point()> & timeGetter) {
+      _getTime = timeGetter;
+    }
+
 }
 }

sources/include/protocol_session/Session.hpp

@@ -13,6 +13,7 @@
 #include <protocol_session/Callbacks.hpp>
 #include <protocol_session/SessionMode.hpp>
 #include <protocol_session/SessionState.hpp>
+#include <protocol_session/SpeedTestPolicy.hpp>
 
 #include <unordered_map>
 #include <chrono>
@@ -205,6 +206,12 @@ namespace detail {
 
         Coin::Network network() const;
 
+        SpeedTestPolicy speedTestPolicy() const;
+
+        void setSpeedTestPolicy(const SpeedTestPolicy);
+
+        void setTimeGetter(const std::function<std::chrono::high_resolution_clock::time_point()> &);
+
     private:
 
         // Session mode
@@ -221,6 +228,11 @@ namespace detail {
 
         Coin::Network _network;
 
+        std::function<std::chrono::high_resolution_clock::time_point()> _getTime;
+
+        SpeedTestPolicy _speedTestPolicy;
+
+
         //// Substates
 
         // Each pointer is != nullptr only when _mode corresponds
@@ -254,6 +266,8 @@ namespace detail {
         void receivedFullPiece(const ConnectionIdType &, const protocol_wire::PieceData &);
         void remoteMessageOverflow(const ConnectionIdType &);
         void localMessageOverflow(const ConnectionIdType &);
+        void buyerRequestedSpeedTest(const ConnectionIdType&, uint32_t);
+        void sellerCompletedSpeedTest(const ConnectionIdType&, bool);
 
         //// Utility routines
 

sources/include/protocol_session/SpeedTestPolicy.hpp

@@ -0,0 +1,43 @@
+/**
+ * Copyright (C) JoyStream - All Rights Reserved
+ */
+
+#ifndef JOYSTREAM_PROTOCOLSESSION_SPEEDTESTPOLICY_HPP
+#define JOYSTREAM_PROTOCOLSESSION_SPEEDTESTPOLICY_HPP
+
+#include <chrono>
+
+namespace joystream {
+namespace protocol_session {
+
+  class SpeedTestPolicy {
+    public:
+      SpeedTestPolicy();
+
+      uint32_t payloadSize() const;
+      uint32_t maxPayloadSize() const;
+      std::chrono::seconds maxTimeToRespond() const;
+      bool isEnabled() const;
+      bool disconnectIfSlow() const;
+
+      void setPayloadSize(uint32_t);
+      void setMaxPayloadSize(uint32_t);
+      void setMaxTimeToRespond(std::chrono::seconds);
+      void enable();
+      void disable();
+      void setDisconnectIfSlow(bool);
+
+    private:
+
+      uint32_t _payloadSize;
+      uint32_t _maxPayloadSize;
+      std::chrono::seconds _maxTimeToRespond;
+      bool _enabled;
+      bool _disconnectIfSlow;
+
+  };
+
+}
+}
+
+#endif // JOYSTREAM_PROTOCOLSESSION_SPEEDTESTPOLICY_HPP

sources/include/protocol_session/Status.hpp

@@ -30,11 +30,13 @@ namespace status {
         CBStateMachine(const std::type_index & innerStateTypeIndex,
                        const protocol_statemachine::AnnouncedModeAndTerms & announcedModeAndTermsFromPeer,
                        const paymentchannel::Payor & payor,
-                       const paymentchannel::Payee & payee)
+                       const paymentchannel::Payee & payee,
+                       const int32_t latency)
             : innerStateTypeIndex(innerStateTypeIndex)
             , announcedModeAndTermsFromPeer(announcedModeAndTermsFromPeer)
             , payor(payor)
-            , payee(payee) {
+            , payee(payee)
+            , latency(latency) {
         }
 
         // Type index of innermost currently active state
@@ -54,6 +56,9 @@ namespace status {
         // Payee side of payment channel interaction
         paymentchannel::Payee payee;
 
+        // Number of ticks from underlying clock used to measure how long it took to send test payload
+        int32_t latency;
+
     };
 
     template <class ConnectionIdType>

sources/include/protocol_session/detail/Buying.cpp

@@ -137,14 +137,10 @@ namespace detail {
 
             // Check that this peer is seller,
             protocol_statemachine::ModeAnnounced m = a.modeAnnounced();
+
             assert(m != protocol_statemachine::ModeAnnounced::none);
 
-            // and has good enough terms to warrant an invitation,
-            // then send invitation
-            if(m == protocol_statemachine::ModeAnnounced::sell && _terms.satisfiedBy(a.sellModeTerms())) {
-                c->processEvent(protocol_statemachine::event::InviteSeller());
-                std::cout << "Invited: " << IdToString(id) << std::endl;
-            }
+            maybeInviteSeller(c);
         }
     }
 
@@ -210,14 +206,68 @@ namespace detail {
     template<class ConnectionIdType>
     void Buying<ConnectionIdType>::remoteMessageOverflow(const ConnectionIdType & id) {
       std::clog << "Error: remoteMessageOverflow from seller connection " << id << std::endl;
+
       removeConnection(id, DisconnectCause::seller_message_overflow);
+
+      // Notify state machine about deletion
+      throw protocol_statemachine::exception::StateMachineDeletedException();
+    }
+
+    template<class ConnectionIdType>
+    void Buying<ConnectionIdType>::sellerCompletedSpeedTest(const ConnectionIdType & id, bool successful) {
+
+      detail::Connection<ConnectionIdType> * c = _session->get(id);
+
+      if (!successful) {
+        // Remove connection
+        removeConnection(id, DisconnectCause::seller_failed_speed_test);
+
+        // Notify state machine about deletion
+        throw protocol_statemachine::exception::StateMachineDeletedException();
+
+      } else {
+        // record completion time
+        c->endingSpeedTest();
+
+        if (!_session->speedTestPolicy().disconnectIfSlow()) {
+          maybeInviteSeller(c);
+        } else {
+
+          if (c->speedTestCompletedInLessThan(_session->speedTestPolicy().maxTimeToRespond())) {
+            // Invite seller if they met the speedTestPolicy requirement
+            maybeInviteSeller(c);
+
+          } else {
+            // otherwise disconnect them
+            removeConnection(id, DisconnectCause::seller_failed_speed_test);
+
+            // Notify state machine about deletion
+            throw protocol_statemachine::exception::StateMachineDeletedException();
+          }
+        }
+
+      }
+
     }
 
     template <class ConnectionIdType>
     void Buying<ConnectionIdType>::leavingState() {
 
         // Prepare sellers before we interrupt with new mode
         politeSellerCompensation();
+
+        // Reset speed testing state for all connections that have not completed a speed test
+        // This is to ensure when the session comes back to buying mode it will make sure to send a speed test request
+        // to sellers that have not completed yet.
+        for(auto itr = _session->_connections.cbegin();itr != _session->_connections.cend();) {
+            auto connection = itr->second;
+
+            if (!connection->hasCompletedSpeedTest()) {
+              connection->abandonSpeedTest();
+            }
+
+            itr++;
+        }
     }
 
     template <class ConnectionIdType>
@@ -506,18 +556,36 @@ namespace detail {
 
           detail::Connection<ConnectionIdType> * c = mapping.second;
 
-          // Check that this peer is seller,
-          protocol_statemachine::AnnouncedModeAndTerms a = c->announcedModeAndTermsFromPeer();
+          maybeInviteSeller(c);
+      }
 
-          // and has good enough terms to warrant an invitation,
-          // then send invitation
-          if(a.modeAnnounced() == protocol_statemachine::ModeAnnounced::sell && _terms.satisfiedBy(a.sellModeTerms())) {
+    }
 
-              c->processEvent(protocol_statemachine::event::InviteSeller());
+    template <class ConnectionIdType>
+    void Buying<ConnectionIdType>::maybeInviteSeller(detail::Connection<ConnectionIdType> * c) const {
 
-              std::cout << "Invited: " << IdToString(mapping.first) << std::endl;
-          }
-      }
+        assert(_session->_state == SessionState::started);
+        assert(_state == BuyingState::sending_invitations);
+
+        // Check that this peer is seller,
+        protocol_statemachine::AnnouncedModeAndTerms a = c->announcedModeAndTermsFromPeer();
+
+        // Do not send invitations if peer is not announcing sell mode or has incompatible terms
+        if(a.modeAnnounced() != protocol_statemachine::ModeAnnounced::sell || !_terms.satisfiedBy(a.sellModeTerms())) {
+          return;
+        }
+
+        // Does seller need to complete a speed test to be invited?
+        if (_session->_speedTestPolicy.isEnabled() && !c->hasCompletedSpeedTest()) {
+            if (c->hasStartedSpeedTest()) return;
+            c->startingSpeedTest(); // record starting time
+            c->processEvent(protocol_statemachine::event::TestSellerSpeed(_session->_speedTestPolicy.payloadSize()));
+            return;
+        }
+
+        // Seller has previously completed a speed test/or no speed test was required.. invite them
+        c->processEvent(protocol_statemachine::event::InviteSeller());
+        std::cout << "Invited: " << IdToString(c->connectionId()) << std::endl;
     }
 
     template <class ConnectionIdType>
@@ -590,8 +658,8 @@ namespace detail {
     void Buying<ConnectionIdType>::removeSeller(detail::Seller<ConnectionIdType> & s) {
         if (s.isGone()) return;
 
-        // we must be downloading
-        assert(_state == BuyingState::downloading);
+        // we must be downloading or just finish downloading
+        assert(_state == BuyingState::downloading || _state == BuyingState::download_completed);
 
         // If this seller has assigned piecees, then we must unassign them
         for(uint i = 0;i < _pieces.size();i++) {