refactor to use variant

Summary:
Replaces the unique pointer with a std::variant and wires up the AlignedParser.

Makes some changes to run the AlignedParser in the stress test tool.

Reviewed By: praihan

Differential Revision: D70032449

fbshipit-source-id: 426ddc69118e3ff32c2a55cdb912ead5db558109

Author: Robert Roeser

thrift/conformance/stresstest/example/ExampleScenarios.cpp

@@ -55,11 +55,21 @@ THRIFT_STRESS_TEST(Echo512k) {
   co_await client->co_echo(s);
 }
 
+THRIFT_STRESS_TEST(Echo512k_eb) {
+  static std::string const s(1 << 19, '?');
+  co_await client->co_echoEb(s);
+}
+
 THRIFT_STRESS_TEST(Echo4M) {
   static std::string const s(4096000, '?');
   co_await client->co_echo(s);
 }
 
+THRIFT_STRESS_TEST(Echo4M_eb) {
+  static std::string const s(4096000, '?');
+  co_await client->co_echoEb(s);
+}
+
 /**
  * Send a request with a small payload, have the server "process" it for 50ms on
  * the CPU threadpool via sleeping instead of the default busy wait, and return

thrift/conformance/stresstest/server/StressTestServer.cpp

@@ -24,8 +24,10 @@
 
 #include <scripts/rroeser/src/executor/WorkStealingExecutor.h>
 #include <folly/executors/CPUThreadPoolExecutor.h>
+#include <thrift/lib/cpp2/Flags.h>
 #include <thrift/lib/cpp2/server/ParallelConcurrencyController.h>
 #include <thrift/lib/cpp2/server/ThriftServer.h>
+#include <thrift/lib/cpp2/transport/rocket/framing/Parser.h>
 
 DEFINE_int32(port, 5000, "Server port");
 DEFINE_int32(io_threads, 0, "Number of IO threads (0 == number of cores)");
@@ -65,6 +67,7 @@ DEFINE_bool(enable_resource_pools, false, "Enable resource pools");
 DEFINE_bool(
     disable_active_request_tracking, false, "Disabled Active Request Tracking");
 DEFINE_bool(enable_checksum, false, "Enable Server Side Checksum support");
+DEFINE_bool(aligned_parser, false, "Enable AlignedParser");
 
 namespace apache {
 namespace thrift {
@@ -165,6 +168,11 @@ std::shared_ptr<ThriftServer> createStressTestServer(
   if (!handler) {
     handler = createStressTestHandler();
   }
+
+  if (FLAGS_aligned_parser) {
+    THRIFT_FLAG_SET_MOCK(rocket_frame_parser, "aligned");
+  }
+
   auto server = std::make_shared<ThriftServer>();
   server->setInterface(std::move(handler));
   server->setPort(FLAGS_port);

thrift/lib/cpp2/transport/rocket/framing/Parser.cpp

@@ -38,6 +38,8 @@ ParserMode stringToMode(const std::string& modeStr) noexcept {
     return ParserMode::STRATEGY;
   } else if (modeStr == "allocating") {
     return ParserMode::ALLOCATING;
+  } else if (modeStr == "aligned") {
+    return ParserMode::ALIGNED;
   }
 
   LOG(WARNING) << "Invalid parser mode: '" << modeStr

thrift/lib/cpp2/transport/rocket/framing/Parser.h

@@ -26,6 +26,7 @@
 
 #include <thrift/lib/cpp2/Flags.h>
 #include <thrift/lib/cpp2/async/RpcOptions.h>
+#include <thrift/lib/cpp2/transport/rocket/framing/parser/AlignedParserStrategy.h>
 #include <thrift/lib/cpp2/transport/rocket/framing/parser/AllocatingParserStrategy.h>
 #include <thrift/lib/cpp2/transport/rocket/framing/parser/FrameLengthParserStrategy.h>
 #include <thrift/lib/cpp2/transport/rocket/framing/parser/ParserStrategy.h>
@@ -35,33 +36,42 @@ THRIFT_FLAG_DECLARE_string(rocket_frame_parser);
 namespace apache::thrift::rocket {
 
 namespace detail {
-enum class ParserMode { STRATEGY, ALLOCATING };
+enum class ParserMode { STRATEGY, ALLOCATING, ALIGNED };
 ParserMode stringToMode(const std::string& modeStr) noexcept;
 ParserAllocatorType& getDefaultAllocator();
 } // namespace detail
 
-// TODO (T160861572): deprecate most of logic in this class and replace with
-// either AllocatingParserStrategy or FrameLengthParserStrategy
 template <class T>
 class Parser final : public folly::AsyncTransport::ReadCallback {
+  using FrameLengthParser = ParserStrategy<T, FrameLengthParserStrategy>;
+  using AllocatingParser =
+      ParserStrategy<T, AllocatingParserStrategy, ParserAllocatorType>;
+  using AlignedParser = ParserStrategy<T, AlignedParserStrategy>;
+  using ParserVariant = std::variant<
+      std::monostate,
+      FrameLengthParser,
+      AllocatingParser,
+      AlignedParser>;
+
  public:
   explicit Parser(
       T& owner, std::shared_ptr<ParserAllocatorType> alloc = nullptr)
       : owner_(owner),
         mode_(detail::stringToMode(THRIFT_FLAG(rocket_frame_parser))) {
-    if (mode_ == detail::ParserMode::STRATEGY) {
-      frameLengthParser_ =
-          std::make_unique<ParserStrategy<T, FrameLengthParserStrategy>>(
-              owner_);
-    }
-    if (mode_ == detail::ParserMode::ALLOCATING) {
-      allocatingParser_ = std::make_unique<
-          ParserStrategy<T, AllocatingParserStrategy, ParserAllocatorType>>(
-          owner_, alloc ? *alloc : detail::getDefaultAllocator());
+    switch (mode_) {
+      case detail::ParserMode::STRATEGY: {
+        parser_.template emplace<FrameLengthParser>(owner_);
+      } break;
+      case detail::ParserMode::ALLOCATING: {
+        parser_.template emplace<AllocatingParser>(
+            owner_, alloc ? *alloc : detail::getDefaultAllocator());
+      } break;
+      case detail::ParserMode::ALIGNED:
+        parser_.template emplace<AlignedParser>(owner_);
+        break;
     }
   }
 
-  // AsyncTransport::ReadCallback implementation
   FOLLY_NOINLINE void getReadBuffer(void** bufout, size_t* lenout) override;
   FOLLY_NOINLINE void readDataAvailable(size_t nbytes) noexcept override;
   FOLLY_NOINLINE void readEOF() noexcept override;
@@ -80,37 +90,46 @@ class Parser final : public folly::AsyncTransport::ReadCallback {
   T& owner_;
 
   detail::ParserMode mode_;
-  std::unique_ptr<ParserStrategy<T, FrameLengthParserStrategy>>
-      frameLengthParser_;
-  std::unique_ptr<
-      ParserStrategy<T, AllocatingParserStrategy, ParserAllocatorType>>
-      allocatingParser_;
+  ParserVariant parser_;
+
+  template <typename DelegateFunc>
+  FOLLY_ALWAYS_INLINE decltype(auto) visit(DelegateFunc&& delegate);
 };
 
 template <class T>
-void Parser<T>::getReadBuffer(void** bufout, size_t* lenout) {
+template <typename DelegateFunc>
+FOLLY_ALWAYS_INLINE decltype(auto) Parser<T>::visit(DelegateFunc&& delegate) {
+  FOLLY_SAFE_DCHECK(
+      std::holds_alternative<std::monostate>(parser_) == false,
+      "parser variant must be set");
   switch (mode_) {
-    case (detail::ParserMode::STRATEGY):
-      frameLengthParser_->getReadBuffer(bufout, lenout);
-      break;
-    case (detail::ParserMode::ALLOCATING):
-      allocatingParser_->getReadBuffer(bufout, lenout);
-      break;
+    case detail::ParserMode::STRATEGY:
+      return std::invoke(
+          std::forward<DelegateFunc>(delegate),
+          std::get<FrameLengthParser>(parser_));
+    case detail::ParserMode::ALLOCATING:
+      return std::invoke(
+          std::forward<DelegateFunc>(delegate),
+          std::get<AllocatingParser>(parser_));
+    case detail::ParserMode::ALIGNED:
+      return std::invoke(
+          std::forward<DelegateFunc>(delegate),
+          std::get<AlignedParser>(parser_));
+    default:
+      LOG(FATAL) << "Unknown parser type";
   }
 }
 
+template <class T>
+void Parser<T>::getReadBuffer(void** bufout, size_t* lenout) {
+  visit([&](auto& parser) { parser.getReadBuffer(bufout, lenout); });
+}
+
 template <class T>
 void Parser<T>::readDataAvailable(size_t nbytes) noexcept {
   folly::DelayedDestruction::DestructorGuard dg(&this->owner_);
   try {
-    switch (mode_) {
-      case (detail::ParserMode::STRATEGY):
-        frameLengthParser_->readDataAvailable(nbytes);
-        break;
-      case (detail::ParserMode::ALLOCATING):
-        allocatingParser_->readDataAvailable(nbytes);
-        break;
-    }
+    visit([&](auto& parser) { parser.readDataAvailable(nbytes); });
   } catch (...) {
     auto exceptionStr =
         folly::exceptionStr(folly::current_exception()).toStdString();
@@ -140,15 +159,7 @@ void Parser<T>::readBufferAvailable(
     std::unique_ptr<folly::IOBuf> buf) noexcept {
   folly::DelayedDestruction::DestructorGuard dg(&this->owner_);
   try {
-    switch (mode_) {
-      case (detail::ParserMode::STRATEGY):
-        frameLengthParser_->readBufferAvailable(std::move(buf));
-        break;
-      case (detail::ParserMode::ALLOCATING):
-        // Will throw not implemented runtime exception
-        allocatingParser_->readBufferAvailable(std::move(buf));
-        break;
-    }
+    visit([&](auto& parser) { parser.readBufferAvailable(std::move(buf)); });
   } catch (...) {
     auto exceptionStr =
         folly::exceptionStr(folly::current_exception()).toStdString();

thrift/lib/cpp2/transport/rocket/framing/parser/test/AlignedParserStrategyTest.cpp

@@ -17,6 +17,7 @@
 #include <gtest/gtest.h>
 #include <folly/String.h>
 #include <folly/io/IOBuf.h>
+#include <thrift/lib/cpp2/server/ThriftServer.h>
 #include <thrift/lib/cpp2/transport/rocket/framing/Frames.h>
 #include <thrift/lib/cpp2/transport/rocket/framing/parser/AlignedParserStrategy.h>
 #include <thrift/lib/cpp2/transport/rocket/framing/parser/test/TestUtil.h>