Common: Introduce StopAllIteration filter status for decoding and encoding filters

include/envoy/http/filter.h

@@ -33,7 +33,30 @@ enum class FilterHeadersStatus {
   StopIteration,
   // Continue iteration to remaining filters, but ignore any subsequent data or trailers. This
   // results in creating a header only request/response.
-  ContinueAndEndStream
+  ContinueAndEndStream,
+  // Do not iterate for headers as well as data and trailers for the current filter and the filters
+  // following, and buffer body data for later dispatching. ContinueDecoding() MUST
+  // be called if continued filter iteration is desired.
+  //
+  // Used when a filter wants to stop iteration on data and trailers while waiting for headers'
+  // iteration to resume.
+  //
+  // If buffering the request causes buffered data to exceed the configured buffer limit, a 413 will
+  // be sent to the user. On the response path exceeding buffer limits will result in a 500.
+  //
+  // TODO(soya3129): stop metadata parsing when StopAllIterationAndBuffer is set.
+  StopAllIterationAndBuffer,
+  // Do not iterate for headers as well as data and trailers for the current filter and the filters
+  // following, and buffer body data for later dispatching. continueDecoding() MUST
+  // be called if continued filter iteration is desired.
+  //
+  // Used when a filter wants to stop iteration on data and trailers while waiting for headers'
+  // iteration to resume.
+  //
+  // This will cause the flow of incoming data to cease until continueDecoding() function is called.
+  //
+  // TODO(soya3129): stop metadata parsing when StopAllIterationAndWatermark is set.
+  StopAllIterationAndWatermark,
 };
 
 /**

source/common/http/conn_manager_impl.h

@@ -95,16 +95,26 @@ class ConnectionManagerImpl : Logger::Loggable<Logger::Id::http>,
    */
   struct ActiveStreamFilterBase : public virtual StreamFilterCallbacks {
     ActiveStreamFilterBase(ActiveStream& parent, bool dual_filter)
-        : parent_(parent), headers_continued_(false), continue_headers_continued_(false),
-          stopped_(false), end_stream_(false), dual_filter_(dual_filter) {}
+        : iteration_state_(IterationState::Continue), iterate_from_current_filter_(false),
+          parent_(parent), headers_continued_(false), continue_headers_continued_(false),
+          end_stream_(false), dual_filter_(dual_filter) {}
 
+    // Functions in the following block are called after the filter finishes processing
+    // corresponding data. Those functions handle state updates and data storage (if needed)
+    // according to the status returned by filter's callback functions.
     bool commonHandleAfter100ContinueHeadersCallback(FilterHeadersStatus status);
     bool commonHandleAfterHeadersCallback(FilterHeadersStatus status, bool& headers_only);
-    void commonHandleBufferData(Buffer::Instance& provided_data);
     bool commonHandleAfterDataCallback(FilterDataStatus status, Buffer::Instance& provided_data,
                                        bool& buffer_was_streaming);
     bool commonHandleAfterTrailersCallback(FilterTrailersStatus status);
 
+    // Buffers provided_data.
+    void commonHandleBufferData(Buffer::Instance& provided_data);
+
+    // If iteration has stopped for all frame types, calls this function to buffer the data before
+    // the filter processes data. The function also updates streaming state.
+    void commonBufferDataIfStopAll(Buffer::Instance& provided_data, bool& buffer_was_streaming);
+
     void commonContinue();
     virtual bool canContinue() PURE;
     virtual Buffer::WatermarkBufferPtr createBuffer() PURE;
@@ -128,10 +138,35 @@ class ConnectionManagerImpl : Logger::Loggable<Logger::Id::http>,
     Tracing::Span& activeSpan() override;
     Tracing::Config& tracingConfig() override;
 
+    // Functions to set or get iteration state.
+    bool canIterate() { return iteration_state_ == IterationState::Continue; }
+    bool stoppedAll() {
+      return iteration_state_ == IterationState::StopAllBuffer ||
+             iteration_state_ == IterationState::StopAllWatermark;
+    }
+    void allowIteration() {
+      ASSERT(iteration_state_ != IterationState::Continue);
+      iteration_state_ = IterationState::Continue;
+    }
+
+    // The state of iteration.
+    enum class IterationState {
+      Continue,            // Iteration has not stopped for any frame type.
+      StopSingleIteration, // Iteration has stopped for headers, 100-continue, or data.
+      StopAllBuffer,       // Iteration has stopped for all frame types, and following data should
+                           // be buffered.
+      StopAllWatermark,    // Iteration has stopped for all frame types, and following data should
+                           // be buffered until high watermark is reached.
+    };
+    IterationState iteration_state_;
+    // If the filter resumes iteration from a StopAllBuffer/Watermark state, the current filter
+    // hasn't parsed data and trailers. As a result, the filter iteration should start with the
+    // current filter instead of the next one. If true, filter iteration starts with the current
+    // filter. Otherwise, starts with the next filter in the chain.
+    bool iterate_from_current_filter_;
     ActiveStream& parent_;
     bool headers_continued_ : 1;
     bool continue_headers_continued_ : 1;
-    bool stopped_ : 1;
     // If true, end_stream is called for this filter.
     bool end_stream_ : 1;
     const bool dual_filter_ : 1;
@@ -164,7 +199,8 @@ class ConnectionManagerImpl : Logger::Loggable<Logger::Id::http>,
       parent_.decodeHeaders(this, *parent_.request_headers_, end_stream);
     }
     void doData(bool end_stream) override {
-      parent_.decodeData(this, *parent_.buffered_request_data_, end_stream);
+      parent_.decodeData(this, *parent_.buffered_request_data_, end_stream,
+                         ActiveStream::FilterIterationStartState::CanStartFromCurrent);
     }
     void doTrailers() override { parent_.decodeTrailers(this, *parent_.request_trailers_); }
     const HeaderMapPtr& trailers() override { return parent_.request_trailers_; }
@@ -247,7 +283,8 @@ class ConnectionManagerImpl : Logger::Loggable<Logger::Id::http>,
       parent_.encodeHeaders(this, *parent_.response_headers_, end_stream);
     }
     void doData(bool end_stream) override {
-      parent_.encodeData(this, *parent_.buffered_response_data_, end_stream);
+      parent_.encodeData(this, *parent_.buffered_response_data_, end_stream,
+                         ActiveStream::FilterIterationStartState::CanStartFromCurrent);
     }
     void doTrailers() override { parent_.encodeTrailers(this, *parent_.response_trailers_); }
     const HeaderMapPtr& trailers() override { return parent_.response_trailers_; }
@@ -290,16 +327,28 @@ class ConnectionManagerImpl : Logger::Loggable<Logger::Id::http>,
     ActiveStream(ConnectionManagerImpl& connection_manager);
     ~ActiveStream();
 
+    // Indicates which filter to start the iteration with.
+    enum class FilterIterationStartState { AlwaysStartFromNext, CanStartFromCurrent };
+
     void addStreamDecoderFilterWorker(StreamDecoderFilterSharedPtr filter, bool dual_filter);
     void addStreamEncoderFilterWorker(StreamEncoderFilterSharedPtr filter, bool dual_filter);
     void chargeStats(const HeaderMap& headers);
+    // Returns the encoder filter to start iteration with.
     std::list<ActiveStreamEncoderFilterPtr>::iterator
-    commonEncodePrefix(ActiveStreamEncoderFilter* filter, bool end_stream);
+    commonEncodePrefix(ActiveStreamEncoderFilter* filter, bool end_stream,
+                       FilterIterationStartState filter_iteration_start_state);
+    // Returns the decoder filter to start iteration with.
+    std::list<ActiveStreamDecoderFilterPtr>::iterator
+    commonDecodePrefix(ActiveStreamDecoderFilter* filter,
+                       FilterIterationStartState filter_iteration_start_state);
     const Network::Connection* connection();
     void addDecodedData(ActiveStreamDecoderFilter& filter, Buffer::Instance& data, bool streaming);
     HeaderMap& addDecodedTrailers();
     void decodeHeaders(ActiveStreamDecoderFilter* filter, HeaderMap& headers, bool end_stream);
-    void decodeData(ActiveStreamDecoderFilter* filter, Buffer::Instance& data, bool end_stream);
+    // Sends data through decoding filter chains. filter_iteration_start_state indicates which
+    // filter to start the iteration with.
+    void decodeData(ActiveStreamDecoderFilter* filter, Buffer::Instance& data, bool end_stream,
+                    FilterIterationStartState filter_iteration_start_state);
     void decodeTrailers(ActiveStreamDecoderFilter* filter, HeaderMap& trailers);
     void disarmRequestTimeout();
     void maybeEndDecode(bool end_stream);
@@ -311,11 +360,19 @@ class ConnectionManagerImpl : Logger::Loggable<Logger::Id::http>,
                         const absl::optional<Grpc::Status::GrpcStatus> grpc_status);
     void encode100ContinueHeaders(ActiveStreamEncoderFilter* filter, HeaderMap& headers);
     void encodeHeaders(ActiveStreamEncoderFilter* filter, HeaderMap& headers, bool end_stream);
-    void encodeData(ActiveStreamEncoderFilter* filter, Buffer::Instance& data, bool end_stream);
+    // Sends data through encoding filter chains. filter_iteration_start_state indicates which
+    // filter to start the iteration with.
+    void encodeData(ActiveStreamEncoderFilter* filter, Buffer::Instance& data, bool end_stream,
+                    FilterIterationStartState filter_iteration_start_state);
     void encodeTrailers(ActiveStreamEncoderFilter* filter, HeaderMap& trailers);
     void encodeMetadata(ActiveStreamEncoderFilter* filter, MetadataMapPtr&& metadata_map_ptr);
     void maybeEndEncode(bool end_stream);
     uint64_t streamId() { return stream_id_; }
+    // Returns true if filter has stopped iteration for all frame types. Otherwise, returns false.
+    // filter_streaming is the variable to indicate if stream is streaming, and its value may be
+    // changed by the function.
+    bool handleDataIfStopAll(ActiveStreamFilterBase& filter, Buffer::Instance& data,
+                             bool& filter_streaming);
 
     // Http::StreamCallbacks
     void onResetStream(StreamResetReason reason,