Allow sending/receiving frames without having a frame processor

Summary:
If we unset the frame processor we can no longer send terminal frames. However if we unset it we risk crashing if we receive any frames while it's unset (this happens outside the stack frame due to a thread hop).

Instead we explicitly support the case of no frame processor set by keeping a queue of incoming frames and flushing those when the processor is back.

Reviewed By: yschimke

Differential Revision: D4921303

fbshipit-source-id: bc3da38242540348318a1d754e64eba8c95639af

Author: lexs

src/FrameTransport.cpp

@@ -55,6 +55,17 @@ void FrameTransport::setFrameProcessor(
   }
 
   drainOutputFramesQueue();
+  if (frameProcessor_) {
+    while (!pendingReads_.empty()) {
+      auto frame = std::move(pendingReads_.front());
+      pendingReads_.pop_front();
+      frameProcessor_->processFrame(std::move(frame));
+    }
+    if (pendingTerminal_) {
+      terminateFrameProcessor(std::move(*pendingTerminal_));
+      pendingTerminal_ = folly::none;
+    }
+  }
 }
 
 void FrameTransport::close(folly::exception_wrapper ex) {
@@ -101,11 +112,10 @@ void FrameTransport::onSubscribe(
 void FrameTransport::onNext(std::unique_ptr<folly::IOBuf> frame) noexcept {
   std::lock_guard<std::recursive_mutex> lock(mutex_);
 
-  if (connection_) {
-    // if *this is not closed and is pulling frames, it should have
-    // frameProcessor
-    CHECK(frameProcessor_);
+  if (connection_ && frameProcessor_) {
     frameProcessor_->processFrame(std::move(frame));
+  } else {
+    pendingReads_.emplace_back(std::move(frame));
   }
 }
 
@@ -115,6 +125,10 @@ void FrameTransport::terminateFrameProcessor(folly::exception_wrapper ex) {
   std::shared_ptr<FrameProcessor> frameProcessor;
   {
     std::lock_guard<std::recursive_mutex> lock(mutex_);
+    if (!frameProcessor_) {
+      pendingTerminal_ = std::move(ex);
+      return;
+    }
     frameProcessor = std::move(frameProcessor_);
   }
 
@@ -159,9 +173,9 @@ void FrameTransport::cancel() noexcept {
 void FrameTransport::outputFrameOrEnqueue(std::unique_ptr<folly::IOBuf> frame) {
   std::lock_guard<std::recursive_mutex> lock(mutex_);
 
-  // we don't want to be sending frames when frameProcessor_ is not set because
-  // we wont have a way to process error/terminating signals
-  if (connection_ && frameProcessor_) {
+  // We allow sending frames even without a frame processor so it's possible
+  // to send terminal frames without expecting anything in return
+  if (connection_) {
     drainOutputFramesQueue();
     if (pendingWrites_.empty() && writeAllowance_.tryAcquire()) {
       // TODO: temporary disabling VLOG as we don't know the correct
@@ -187,7 +201,7 @@ void FrameTransport::outputFrameOrEnqueue(std::unique_ptr<folly::IOBuf> frame) {
 void FrameTransport::drainOutputFramesQueue() {
   std::lock_guard<std::recursive_mutex> lock(mutex_);
 
-  if (connection_ && frameProcessor_) {
+  if (connection_) {
     // Drain the queue or the allowance.
     while (!pendingWrites_.empty() && writeAllowance_.tryAcquire()) {
       auto frame = std::move(pendingWrites_.front());

src/FrameTransport.h

@@ -5,6 +5,10 @@
 #include <deque>
 #include <memory>
 #include <mutex>
+
+#include <folly/Optional.h>
+#include <folly/ExceptionWrapper.h>
+
 #include "src/AllowanceSemaphore.h"
 #include "src/Common.h"
 #include "src/FrameProcessor.h"
@@ -77,5 +81,7 @@ class FrameTransport :
   std::shared_ptr<Subscription> connectionInputSub_;
 
   std::deque<std::unique_ptr<folly::IOBuf>> pendingWrites_;
+  std::deque<std::unique_ptr<folly::IOBuf>> pendingReads_;
+  folly::Optional<folly::exception_wrapper> pendingTerminal_;
 };
 } // reactivesocket

src/ServerConnectionAcceptor.cpp

@@ -131,17 +131,16 @@ void ServerConnectionAcceptor::processFrame(
         break;
       }
 
+      removeConnection(transport);
       auto triedResume =
           connectionHandler->resumeSocket(transport, std::move(resumeParams));
-      if (triedResume) {
-        connections_.erase(transport);
-      } else {
+      if (!triedResume) {
         transport->outputFrameOrEnqueue(frameSerializer->serializeOut(
             Frame_ERROR::connectionError("can not resume")));
         transport->close(std::runtime_error("can not resume"));
-        connections_.erase(transport);
       }
-    } break;
+      break;
+    }
 
     case FrameType::CANCEL:
     case FrameType::ERROR:

test/ServerConnectionAcceptorTest.cpp

@@ -39,9 +39,9 @@ class MockConnectionHandler : public ConnectionHandler {
       void(std::shared_ptr<FrameTransport>, folly::exception_wrapper ex));
 };
 
-class MockFrameProcessor : public FrameProcessor {
-  void processFrame(std::unique_ptr<folly::IOBuf>) override {};
-  void onTerminal(folly::exception_wrapper) override {};
+struct MockFrameProcessor : public FrameProcessor {
+  MOCK_METHOD1(processFrame, void(std::unique_ptr<folly::IOBuf>));
+  MOCK_METHOD1(onTerminal, void(folly::exception_wrapper));
 };
 
 class ServerConnectionAcceptorTest : public Test {
@@ -221,3 +221,88 @@ TEST_F(ServerConnectionAcceptorTest, VerifyTransport) {
   clientOutput_->onComplete();
   transport_->close(folly::exception_wrapper());
 }
+
+TEST_F(ServerConnectionAcceptorTest, VerifyAsyncProcessorFrame) {
+  ResumeParameters resumeParams(
+      ResumeIdentificationToken::generateNew(),
+      1,
+      2,
+      FrameSerializer::getCurrentProtocolVersion());
+  std::shared_ptr<FrameTransport> transport_;
+  EXPECT_CALL(*handler_, resumeSocket(_, _))
+      .WillOnce(Invoke(
+          [&](std::shared_ptr<FrameTransport> transport,
+              ResumeParameters params) -> bool {
+            // Only capture the transport, wait with setting the processor.
+            transport_ = transport;
+            return true;
+          }));
+
+  auto frameSerializer = FrameSerializer::createCurrentVersion();
+  acceptor_.accept(std::move(serverConnection_), handler_);
+  clientOutput_->onNext(frameSerializer->serializeOut(Frame_RESUME(
+      resumeParams.token,
+      resumeParams.serverPosition,
+      resumeParams.clientPosition,
+      FrameSerializer::getCurrentProtocolVersion())));
+
+  // The transport won't have a processor now, try sending a frame
+  clientOutput_->onNext(frameSerializer->serializeOut(Frame_REQUEST_FNF(
+      1,
+      FrameFlags::EMPTY,
+      Payload())));
+
+  auto processor = std::make_shared<NiceMock<MockFrameProcessor>>();
+  EXPECT_CALL(*processor, onTerminal(_))
+    .Times(Exactly(0));
+  EXPECT_CALL(*processor, processFrame(_))
+    .WillOnce(Invoke([&](std::unique_ptr<folly::IOBuf> frame) {
+      Frame_REQUEST_FNF fnfFrame;
+      EXPECT_TRUE(frameSerializer->deserializeFrom(fnfFrame, std::move(frame)));
+    }));
+
+  transport_->setFrameProcessor(processor);
+
+  // Teardown will cause a terminal callback
+  Mock::VerifyAndClearExpectations(processor.get());
+  clientOutput_->onComplete();
+  transport_->close(folly::exception_wrapper());
+}
+
+TEST_F(ServerConnectionAcceptorTest, VerifyAsyncProcessorTerminal) {
+  ResumeParameters resumeParams(
+      ResumeIdentificationToken::generateNew(),
+      1,
+      2,
+      FrameSerializer::getCurrentProtocolVersion());
+  std::shared_ptr<FrameTransport> transport_;
+  EXPECT_CALL(*handler_, resumeSocket(_, _))
+      .WillOnce(Invoke(
+          [&](std::shared_ptr<FrameTransport> transport,
+              ResumeParameters params) -> bool {
+            // Only capture the transport, wait with setting the processor.
+            transport_ = transport;
+            return true;
+          }));
+
+  auto frameSerializer = FrameSerializer::createCurrentVersion();
+  acceptor_.accept(std::move(serverConnection_), handler_);
+  clientOutput_->onNext(frameSerializer->serializeOut(Frame_RESUME(
+      resumeParams.token,
+      resumeParams.serverPosition,
+      resumeParams.clientPosition,
+      FrameSerializer::getCurrentProtocolVersion())));
+
+  // The transport won't have a processor now, try sending terminal.
+  clientOutput_->onError(std::runtime_error("too bad"));
+
+  auto processor = std::make_shared<StrictMock<MockFrameProcessor>>();
+  EXPECT_CALL(*processor, onTerminal(_))
+    .WillOnce(Invoke([&](folly::exception_wrapper ex) {
+      EXPECT_THAT(ex.what().toStdString(), HasSubstr("too bad"));
+    }));
+
+  transport_->setFrameProcessor(processor);
+
+  transport_->close(folly::exception_wrapper());
+}