initial resume machinery (#113)

* initial resume machinery

* add placeholder calls to retransmit from position.

* use std::array for token storage.

* use unique_ptr to hold ResumeTracker and ResumeCache so they can be moved.

* add server side resume listener callback for handling RESUME association

* add client side ReactiveSocket tryClientResume logic

* initial framework for resume TCP tests

* fix bad function call from confusion over 'this'

* add token to SETUP

* send ERROR frames when can not resume. Minor fixes for position tracking. Avoid closing ReactiveSocket upon close to allow for resume.

* added token to fromClientConnection params. Reworked resume listener to pass position and added methods on ReactiveSocket to query cache position.

Author: tmontgomery

CMakeLists.txt

@@ -132,7 +132,11 @@ add_library(
   src/folly/FollyKeepaliveTimer.cpp
   src/folly/FollyKeepaliveTimer.h
   src/ConnectionSetupPayload.h
-  src/ConnectionSetupPayload.cpp)
+  src/ConnectionSetupPayload.cpp
+  src/ResumeTracker.h
+  src/ResumeCache.h
+  src/ResumeCache.cpp
+)
 
 target_link_libraries(
   ReactiveSocket
@@ -162,7 +166,8 @@ add_executable(
   test/ReactiveSocketTest.cpp
   test/ReactiveStreamsMocksCompat.h
   test/Test.cpp
-  test/MockStats.h)
+  test/MockStats.h
+)
 
 target_link_libraries(
   tests
@@ -240,4 +245,44 @@ target_link_libraries(
 
 add_dependencies(tckclient gmock)
 
+add_executable(
+        tcpresumeclient
+        test/resume/TcpResumeClient.cpp
+        test/simple/PrintSubscriber.cpp
+        test/simple/PrintSubscriber.h
+        src/ReactiveSocket.cpp
+        src/ReactiveSocket.h
+        test/simple/StatsPrinter.cpp
+        test/simple/StatsPrinter.h)
+
+target_link_libraries(
+        tcpresumeclient
+        ReactiveSocket
+        ${FOLLY_LIBRARIES}
+        gflags
+        ${GMOCK_LIBS}
+        glog
+        ${CMAKE_THREAD_LIBS_INIT})
+
+add_dependencies(tcpresumeclient gmock)
+
+add_executable(
+        tcpresumeserver
+        test/resume/TcpResumeServer.cpp
+        test/simple/PrintSubscriber.cpp
+        test/simple/PrintSubscriber.h
+        test/simple/StatsPrinter.cpp
+        test/simple/StatsPrinter.h)
+
+target_link_libraries(
+        tcpresumeserver
+        ReactiveSocket
+        ${FOLLY_LIBRARIES}
+        gflags
+        ${GMOCK_LIBS}
+        glog
+        ${CMAKE_THREAD_LIBS_INIT})
+
+add_dependencies(tcpresumeserver gmock)
+
 # EOF

src/ConnectionAutomaton.cpp

@@ -21,12 +21,17 @@ namespace reactivesocket {
 ConnectionAutomaton::ConnectionAutomaton(
     std::unique_ptr<DuplexConnection> connection,
     StreamAutomatonFactory factory,
+    ResumeListener resumeListener,
     Stats& stats,
     bool isServer)
     : connection_(std::move(connection)),
       factory_(std::move(factory)),
       stats_(stats),
-      isServer_(isServer) {
+      isServer_(isServer),
+      isResumable_(true),
+      resumeTracker_(new ResumeTracker()),
+      resumeCache_(new ResumeCache()),
+      resumeListener_(resumeListener) {
   // We deliberately do not "open" input or output to avoid having c'tor on the
   // stack when processing any signals from the connection. See ::connect and
   // ::onSubscribe.
@@ -66,6 +71,12 @@ void ConnectionAutomaton::disconnect() {
   }
 }
 
+void ConnectionAutomaton::reconnect(std::unique_ptr<DuplexConnection> newConnection) {
+    disconnect();
+    connection_ = std::move(newConnection);
+    connect();
+}
+
 ConnectionAutomaton::~ConnectionAutomaton() {
   VLOG(6) << "~ConnectionAutomaton";
   // We rely on SubscriptionPtr and SubscriberPtr to dispatch appropriate
@@ -127,6 +138,10 @@ void ConnectionAutomaton::onNext(std::unique_ptr<folly::IOBuf> frame) {
 
   stats_.frameRead(ss.str());
 
+  // TODO(tmont): If a frame is invalid, it will still be tracked. However, we actually want that. We want to keep
+  // each side in sync, even if a frame is invalid.
+  resumeTracker_->trackReceivedFrame(*frame);
+
   auto streamIdPtr = FrameHeader::peekStreamId(*frame);
   if (!streamIdPtr) {
     // Failed to deserialize the frame.
@@ -183,6 +198,9 @@ void ConnectionAutomaton::onConnectionFrame(
       return;
     }
     case FrameType::SETUP: {
+      // TODO(tmont): check for ENABLE_RESUME and make sure isResumable_ is true
+      // TODO(tmont): figure out how best to pass in ResumeIdentificationToken
+
       if (!factory_(0, std::move(payload))) {
         assert(false);
       }
@@ -194,6 +212,46 @@ void ConnectionAutomaton::onConnectionFrame(
       }
       return;
     }
+    case FrameType::RESUME: {
+      Frame_RESUME frame;
+      if (frame.deserializeFrom(std::move(payload))) {
+        bool canResume = false;
+
+        if (isServer_ && isResumable_) {
+            // find old ConnectionAutmaton via calling listener.
+            // Application will call resumeFromAutomaton to setup streams and resume information
+            canResume = resumeListener_(frame.token_, frame.position_);
+        }
+
+        if (canResume) {
+          outputFrameOrEnqueue(
+              Frame_RESUME_OK(resumeTracker_->impliedPosition()).serializeOut());
+          resumeCache_->retransmitFromPosition(frame.position_, *this);
+        } else {
+          outputFrameOrEnqueue(Frame_ERROR::canNotResume("can not resume").serializeOut());
+          disconnect();
+        }
+      } else {
+        outputFrameOrEnqueue(Frame_ERROR::unexpectedFrame().serializeOut());
+        disconnect();
+      }
+      return;
+    }
+    case FrameType::RESUME_OK: {
+      Frame_RESUME_OK frame;
+      if (frame.deserializeFrom(std::move(payload))) {
+        if (!isServer_ && isResumable_ && resumeCache_->isPositionAvailable(frame.position_)) {
+          resumeCache_->retransmitFromPosition(frame.position_, *this);
+        } else {
+            outputFrameOrEnqueue(Frame_ERROR::canNotResume("can not resume").serializeOut());
+            disconnect();
+        }
+      } else {
+        outputFrameOrEnqueue(Frame_ERROR::unexpectedFrame().serializeOut());
+        disconnect();
+      }
+      return;
+    }
     default:
       outputFrameOrEnqueue(Frame_ERROR::unexpectedFrame().serializeOut());
       disconnect();
@@ -263,6 +321,19 @@ void ConnectionAutomaton::sendKeepalive() {
   outputFrameOrEnqueue(pingFrame.serializeOut());
 }
 
+void ConnectionAutomaton::sendResume(const ResumeIdentificationToken &token) {
+  Frame_RESUME resumeFrame(token, resumeTracker_->impliedPosition());
+  outputFrameOrEnqueue(resumeFrame.serializeOut());
+}
+
+bool ConnectionAutomaton::isPositionAvailable(ResumePosition position) {
+    return resumeCache_->isPositionAvailable(position);
+}
+
+ResumePosition ConnectionAutomaton::positionDifference(ResumePosition position) {
+    return resumeCache_->position() - position;
+}
+
 void ConnectionAutomaton::onClose(ConnectionCloseListener listener) {
   closeListeners_.push_back(listener);
 }
@@ -299,6 +370,20 @@ void ConnectionAutomaton::outputFrame(
 
   stats_.frameWritten(ss.str());
 
+  resumeCache_->trackAndCacheSentFrame(*outputFrame);
   connectionOutput_.onNext(std::move(outputFrame));
 }
+
+void ConnectionAutomaton::resumeFromAutomaton(ConnectionAutomaton& oldAutomaton)
+{
+  if (isServer_ && isResumable_)
+  {
+    streams_ = std::move(oldAutomaton.streams_);
+    factory_ = oldAutomaton.factory_;
+
+    resumeTracker_ = std::move(oldAutomaton.resumeTracker_);
+    resumeCache_ = std::move(oldAutomaton.resumeCache_);
+  }
+}
+
 }

src/ConnectionAutomaton.h

@@ -12,6 +12,8 @@
 #include "src/Payload.h"
 #include "src/ReactiveStreamsCompat.h"
 #include "src/Stats.h"
+#include "src/ResumeTracker.h"
+#include "src/ResumeCache.h"
 
 namespace reactivesocket {
 
@@ -32,6 +34,9 @@ using StreamId = uint32_t;
 using StreamAutomatonFactory =
     std::function<bool(StreamId, std::unique_ptr<folly::IOBuf>)>;
 
+using ResumeListener =
+    std::function<bool(const ResumeIdentificationToken &token, ResumePosition position)>;
+
 using ConnectionCloseListener = std::function<void()>;
 
 /// Handles connection-level frames and (de)multiplexes streams.
@@ -51,6 +56,7 @@ class ConnectionAutomaton :
       std::unique_ptr<DuplexConnection> connection,
       // TODO(stupaq): for testing only, can devirtualise if necessary
       StreamAutomatonFactory factory,
+      ResumeListener resumeListener,
       Stats& stats,
       bool client);
 
@@ -66,6 +72,9 @@ class ConnectionAutomaton :
   /// AbstractStreamAutomaton attached to this ConnectionAutomaton.
   void disconnect();
 
+  /// Terminate underlying connection and connect new connection
+  void reconnect(std::unique_ptr<DuplexConnection> newConnection);
+
   ~ConnectionAutomaton();
 
   /// @{
@@ -107,9 +116,16 @@ class ConnectionAutomaton :
   ///   delivered multiple times as long as the caller holds shared_ptr to
   ///   ConnectionAutomaton.
   void endStream(StreamId streamId, StreamCompletionSignal signal);
+
+  /// Copy the streams and resumption information from a previous ConnectionAutomaton
+  void resumeFromAutomaton(ConnectionAutomaton& oldAutomaton);
   /// @}
 
   void sendKeepalive();
+  void sendResume(const ResumeIdentificationToken &token);
+
+  bool isPositionAvailable(ResumePosition position);
+  ResumePosition positionDifference(ResumePosition position);
 
   void onClose(ConnectionCloseListener listener);
 
@@ -164,6 +180,10 @@ class ConnectionAutomaton :
       pendingWrites_; // TODO(stupaq): two vectors?
   Stats& stats_;
   bool isServer_;
+  bool isResumable_;
   std::vector<ConnectionCloseListener> closeListeners_;
+  std::unique_ptr<ResumeTracker> resumeTracker_;
+  std::unique_ptr<ResumeCache> resumeCache_;
+  ResumeListener resumeListener_;
 };
 }

src/Frame.cpp

@@ -65,6 +65,10 @@ std::ostream& operator<<(std::ostream& os, FrameType type) {
       return os << "LEASE";
     case FrameType::METADATA_PUSH:
       return os << "METADATA_PUSH";
+    case FrameType::RESUME:
+      return os << "RESUME";
+    case FrameType::RESUME_OK:
+      return os << "RESUME_OK";
   }
   return os << "FrameType(" << static_cast<uint16_t>(type) << ")";
 }
@@ -321,6 +325,10 @@ Frame_ERROR Frame_ERROR::applicationError(
   return Frame_ERROR(streamId, ErrorCode::APPLICATION_ERROR, Payload(message));
 }
 
+Frame_ERROR Frame_ERROR::canNotResume(const std::string& message) {
+    return Frame_ERROR(0, ErrorCode::CONNECTION_ERROR, Payload(message));
+}
+
 std::unique_ptr<folly::IOBuf> Frame_ERROR::serializeOut() {
   auto queue = createBufferQueue(
       FrameHeader::kSize + sizeof(uint32_t) + payload_.framingSize());
@@ -383,7 +391,7 @@ std::ostream& operator<<(std::ostream& os, const Frame_KEEPALIVE& frame) {
 /// @{
 std::unique_ptr<folly::IOBuf> Frame_SETUP::serializeOut() {
   auto queue = createBufferQueue(
-      FrameHeader::kSize + 3 * sizeof(uint32_t) + 2 +
+      FrameHeader::kSize + 3 * sizeof(uint32_t) + token_.size() + 2 +
       metadataMimeType_.length() + dataMimeType_.length() +
       payload_.framingSize());
   folly::io::QueueAppender appender(&queue, /* do not grow */ 0);
@@ -392,8 +400,9 @@ std::unique_ptr<folly::IOBuf> Frame_SETUP::serializeOut() {
   appender.writeBE(static_cast<uint32_t>(version_));
   appender.writeBE(static_cast<uint32_t>(keepaliveTime_));
   appender.writeBE(static_cast<uint32_t>(maxLifetime_));
+  appender.push((const uint8_t*)token_.data(), token_.size());
 
-  CHECK(metadataMimeType_.length() <= std::numeric_limits<uint8_t>::max());
+    CHECK(metadataMimeType_.length() <= std::numeric_limits<uint8_t>::max());
   appender.writeBE(static_cast<uint8_t>(metadataMimeType_.length()));
   appender.push(
       (const uint8_t*)metadataMimeType_.data(), metadataMimeType_.length());
@@ -413,6 +422,7 @@ bool Frame_SETUP::deserializeFrom(std::unique_ptr<folly::IOBuf> in) {
     version_ = cur.readBE<uint32_t>();
     keepaliveTime_ = cur.readBE<uint32_t>();
     maxLifetime_ = cur.readBE<uint32_t>();
+    cur.pull(token_.data(), token_.size());
 
     int mdmtLen = cur.readBE<uint8_t>();
     metadataMimeType_ = cur.readFixedString(mdmtLen);
@@ -465,4 +475,64 @@ std::ostream& operator<<(std::ostream& os, const Frame_LEASE& frame) {
             << ")";
 }
 /// @}
+
+/// @{
+std::unique_ptr<folly::IOBuf> Frame_RESUME::serializeOut() {
+  auto queue = createBufferQueue(
+      FrameHeader::kSize + sizeof(ResumeIdentificationToken) + sizeof(ResumePosition));
+  folly::io::QueueAppender appender(&queue, /* do not grow */ 0);
+
+  header_.serializeInto(appender);
+  appender.push(token_.data(), sizeof(token_));
+  appender.writeBE(position_);
+
+  return queue.move();
+}
+
+bool Frame_RESUME::deserializeFrom(std::unique_ptr<folly::IOBuf> in) {
+  folly::io::Cursor cur(in.get());
+  try {
+    header_.deserializeFrom(cur);
+    cur.pull(token_.data(), sizeof(token_));
+    position_ = cur.readBE<ResumePosition>();
+  } catch (...) {
+    return false;
+  }
+  return true;
+}
+
+std::ostream& operator<<(std::ostream& os, const Frame_RESUME& frame) {
+  return os << frame.header_ << ", ("
+            << "token" << ", @" << frame.position_ << ")";
+}
+/// @}
+
+/// @{
+std::unique_ptr<folly::IOBuf> Frame_RESUME_OK::serializeOut() {
+  auto queue = createBufferQueue(
+      FrameHeader::kSize + sizeof(ResumePosition));
+  folly::io::QueueAppender appender(&queue, /* do not grow */ 0);
+
+  header_.serializeInto(appender);
+  appender.writeBE(position_);
+
+  return queue.move();
+}
+
+bool Frame_RESUME_OK::deserializeFrom(std::unique_ptr<folly::IOBuf> in) {
+  folly::io::Cursor cur(in.get());
+  try {
+    header_.deserializeFrom(cur);
+    position_ = cur.readBE<ResumePosition>();
+  } catch (...) {
+    return false;
+  }
+  return true;
+}
+
+std::ostream& operator<<(std::ostream& os, const Frame_RESUME_OK& frame) {
+  return os << frame.header_ << ", (@" << frame.position_ << ")";
+}
+/// @}
+
 }