Make AsyncProcessor::processInteraction pure virtual

Summary: `processInteraction()` is part of the critical path for interactions in Resource Pools and can lead to undefined behavior if it is not implemented. Let's make the contract more clear and force custom `AsyncProcessor` implementations to provide an implementation for it. Reviewed By: tlj77 Differential Revision: D62590107 fbshipit-source-id: 69fe9e9bee745766d891b5fc5f33d9953d6845eb
facebook · Oct 3, 2024 · c9482e5 · c9482e5
1 parent b125f36
commit c9482e5
Show file tree

Hide file tree

Showing 11 changed files with 72 additions and 1 deletion.
diff --git a/thrift/lib/cpp2/async/AsyncProcessor.h b/thrift/lib/cpp2/async/AsyncProcessor.h
@@ -411,7 +411,7 @@ class AsyncProcessor : public TProcessorBase {
   virtual void destroyAllInteractions(
       Cpp2ConnContext& conn, folly::EventBase&) noexcept;
 
-  virtual void processInteraction(ServerRequest&&) {}
+  virtual void processInteraction(ServerRequest&&) = 0;
 
   // This is the main interface we are migrating to. Eventually it should
   // replace all the processSerialized... methods.

diff --git a/thrift/lib/cpp2/async/PreprocessingAsyncProcessorWrapper.h b/thrift/lib/cpp2/async/PreprocessingAsyncProcessorWrapper.h
@@ -65,6 +65,12 @@ class PreprocessingAsyncProcessorWrapper : public AsyncProcessor {
 
   const char* getServiceName() override final;
 
+  void processInteraction(ServerRequest&&) override {
+    LOG(FATAL)
+        << "This AsyncProcessor doesn't support Thrift interactions. "
+        << "Please implement processInteraction to support interactions.";
+  }
+
  protected:
   using ProcessSerializedCompressedRequestReturnT = std::
       tuple<ResponseChannelRequest::UniquePtr, SerializedCompressedRequest>;

diff --git a/thrift/lib/cpp2/server/test/ParallelConcurrencyControllerTest.cpp b/thrift/lib/cpp2/server/test/ParallelConcurrencyControllerTest.cpp
@@ -61,6 +61,12 @@ class MockAsyncProcessor : public AsyncProcessor {
     }
   }
 
+  void processInteraction(apache::thrift::ServerRequest&&) override {
+    LOG(FATAL)
+        << "This AsyncProcessor doesn't support Thrift interactions. "
+        << "Please implement processInteraction to support interactions.";
+  }
+
   void setFunc(Func func) { executeRequestFunc_ = std::move(func); }
 
  private:
@@ -593,6 +599,12 @@ TEST(ParallelConcurrencyControllerTest, FinishCallbackExceptionSafe) {
       });
     }
 
+    void processInteraction(apache::thrift::ServerRequest&&) override {
+      LOG(FATAL)
+          << "This AsyncProcessor doesn't support Thrift interactions. "
+          << "Please implement processInteraction to support interactions.";
+    }
+
    private:
     folly::Baton<>& baton_;
   };

diff --git a/thrift/lib/cpp2/test/AsyncProcessorTest.cpp b/thrift/lib/cpp2/test/AsyncProcessorTest.cpp
@@ -512,6 +512,12 @@ TEST_P(
           processor_->executeRequest(std::move(request), methodMetadata);
         }
 
+        void processInteraction(apache::thrift::ServerRequest&&) override {
+          LOG(FATAL)
+              << "This AsyncProcessor doesn't support Thrift interactions. "
+              << "Please implement processInteraction to support interactions.";
+        }
+
         explicit Processor(std::unique_ptr<AsyncProcessor> processor)
             : processor_(std::move(processor)) {}
 

diff --git a/thrift/lib/cpp2/test/MultiplexAsyncProcessorTest.cpp b/thrift/lib/cpp2/test/MultiplexAsyncProcessorTest.cpp
@@ -404,6 +404,12 @@ class WildcardThrowsInternalError : public TProcessorFactory {
         inner_->destroyAllInteractions(ctx, eb);
       }
 
+      void processInteraction(ServerRequest&&) override {
+        LOG(FATAL)
+            << "This AsyncProcessor doesn't support Thrift interactions. "
+            << "Please implement processInteraction to support interactions.";
+      }
+
       explicit Processor(
           std::unique_ptr<AsyncProcessor>&& inner,
           const MessageVariant& message)

diff --git a/thrift/lib/cpp2/test/PreprocessingAsyncProcessorWrapperTest.cpp b/thrift/lib/cpp2/test/PreprocessingAsyncProcessorWrapperTest.cpp
@@ -43,6 +43,12 @@ class MockAsyncProcessor : public AsyncProcessor {
       void(
           ServerRequest&& request,
           const AsyncProcessorFactory::MethodMetadata& methodMetadata));
+
+  void processInteraction(apache::thrift::ServerRequest&&) override {
+    LOG(FATAL)
+        << "This AsyncProcessor doesn't support Thrift interactions. "
+        << "Please implement processInteraction to support interactions.";
+  }
 };
 
 class TestChannelRequest : public ResponseChannelRequest {

diff --git a/thrift/lib/cpp2/test/server/ThriftServerTest.cpp b/thrift/lib/cpp2/test/server/ThriftServerTest.cpp
@@ -4331,6 +4331,12 @@ class HeaderOrRocketCompression
           tm);
     }
 
+    void processInteraction(apache::thrift::ServerRequest&&) override {
+      LOG(FATAL)
+          << "This AsyncProcessor doesn't support Thrift interactions. "
+          << "Please implement processInteraction to support interactions.";
+    }
+
     std::unique_ptr<AsyncProcessor> underlyingProcessor_;
     CompressionAlgorithm compression_;
   };
@@ -4365,6 +4371,12 @@ class HeaderOrRocketCompression
           tm);
     }
 
+    void processInteraction(apache::thrift::ServerRequest&&) override {
+      LOG(FATAL)
+          << "This AsyncProcessor doesn't support Thrift interactions. "
+          << "Please implement processInteraction to support interactions.";
+    }
+
     std::unique_ptr<AsyncProcessor> underlyingProcessor_;
   };
 

diff --git a/thrift/lib/cpp2/transport/core/testutil/TransportCompatibilityTest.cpp b/thrift/lib/cpp2/transport/core/testutil/TransportCompatibilityTest.cpp
@@ -1310,6 +1310,12 @@ void TransportCompatibilityTest::TestCustomAsyncProcessor() {
           tm);
     }
 
+    void processInteraction(apache::thrift::ServerRequest&&) override {
+      LOG(FATAL)
+          << "This AsyncProcessor doesn't support Thrift interactions. "
+          << "Please implement processInteraction to support interactions.";
+    }
+
    private:
     std::unique_ptr<apache::thrift::AsyncProcessor> underlyingProcessor_;
   };

diff --git a/thrift/lib/cpp2/transport/rocket/test/fuzz/FuzzUtil.h b/thrift/lib/cpp2/transport/rocket/test/fuzz/FuzzUtil.h
@@ -98,6 +98,12 @@ class FakeProcessor final : public apache::thrift::AsyncProcessor {
             "place holder"),
         "1" /* doesnt matter */);
   }
+
+  void processInteraction(apache::thrift::ServerRequest&&) override {
+    LOG(FATAL)
+        << "This AsyncProcessor doesn't support Thrift interactions. "
+        << "Please implement processInteraction to support interactions.";
+  }
 };
 
 class FakeProcessorFactory final

diff --git a/thrift/lib/cpp2/util/EmptyAsyncProcessor.cpp b/thrift/lib/cpp2/util/EmptyAsyncProcessor.cpp
@@ -30,6 +30,11 @@ std::unique_ptr<AsyncProcessor> EmptyAsyncProcessorFactory::getProcessor() {
         concurrency::ThreadManager*) override {
       LOG(FATAL) << "This method should never be called on EmptyAsyncProcessor";
     }
+    void processInteraction(ServerRequest&&) override {
+      LOG(FATAL)
+          << "This AsyncProcessor doesn't support Thrift interactions. "
+          << "Please implement processInteraction to support interactions.";
+    }
   };
 
   return std::make_unique<EmptyAsyncProcessor>();

diff --git a/thrift/lib/py/server/CppServerWrapper.cpp b/thrift/lib/py/server/CppServerWrapper.cpp
@@ -307,6 +307,12 @@ class PythonAsyncProcessor : public AsyncProcessor {
         true /* fromExecuteRequest  */);
   }
 
+  void processInteraction(apache::thrift::ServerRequest&&) override {
+    LOG(FATAL)
+        << "This AsyncProcessor doesn't support Thrift interactions. "
+        << "Please implement processInteraction to support interactions.";
+  }
+
   /**
    * Get the priority of the request
    * Check the headers directly in C++ since noone seems to override that logic