caching: Support out-of-thread (async) HttpCache implementations

source/extensions/filters/http/cache/cache_filter.cc

@@ -7,6 +7,7 @@
 #include "extensions/filters/http/cache/cacheability_utils.h"
 #include "extensions/filters/http/cache/inline_headers_handles.h"
 
+#include "absl/memory/memory.h"
 #include "absl/strings/string_view.h"
 
 namespace Envoy {
@@ -32,8 +33,12 @@ CacheFilter::CacheFilter(const envoy::extensions::filters::http::cache::v3alpha:
     : time_source_(time_source), cache_(http_cache) {}
 
 void CacheFilter::onDestroy() {
-  lookup_ = nullptr;
-  insert_ = nullptr;
+  if (lookup_) {
+    lookup_->onDestroy();
+  }
+  if (insert_) {
+    insert_->onDestroy();
+  }
 }
 
 Http::FilterHeadersStatus CacheFilter::decodeHeaders(Http::RequestHeaderMap& headers,
@@ -58,30 +63,18 @@ Http::FilterHeadersStatus CacheFilter::decodeHeaders(Http::RequestHeaderMap& hea
   lookup_ = cache_.makeLookupContext(std::move(lookup_request));
 
   ASSERT(lookup_);
+  getHeaders(headers);
   ENVOY_STREAM_LOG(debug, "CacheFilter::decodeHeaders starting lookup", *decoder_callbacks_);
 
-  lookup_->getHeaders(
-      [this, &headers](LookupResult&& result) { onHeaders(std::move(result), headers); });
-
-  // If the cache called onHeaders synchronously it will have advanced the filter_state_.
-  switch (filter_state_) {
-  case FilterState::Initial:
-    // Headers are not fetched from cache yet -- wait until cache lookup is completed.
-    filter_state_ = FilterState::WaitingForCacheLookup;
-    return Http::FilterHeadersStatus::StopAllIterationAndWatermark;
-  case FilterState::DecodeServingFromCache:
-  case FilterState::ResponseServedFromCache:
-    // A fresh cached response was found -- no need to continue the decoding stream.
-    return Http::FilterHeadersStatus::StopAllIterationAndWatermark;
-  default:
-    return Http::FilterHeadersStatus::Continue;
-  }
+  // Stop the decoding stream until the cache lookup result is ready.
+  return Http::FilterHeadersStatus::StopAllIterationAndWatermark;
 }
 
 Http::FilterHeadersStatus CacheFilter::encodeHeaders(Http::ResponseHeaderMap& headers,
                                                      bool end_stream) {
   if (filter_state_ == FilterState::DecodeServingFromCache) {
-    // This call was invoked by decoder_callbacks_->encodeHeaders -- ignore it.
+    // This call was invoked during decoding by decoder_callbacks_->encodeHeaders because a fresh
+    // cached response was found and is being added to the encoding stream -- ignore it.
     return Http::FilterHeadersStatus::Continue;
   }
 
@@ -92,12 +85,8 @@ Http::FilterHeadersStatus CacheFilter::encodeHeaders(Http::ResponseHeaderMap& he
 
   if (filter_state_ == FilterState::ValidatingCachedResponse && isResponseNotModified(headers)) {
     processSuccessfulValidation(headers);
-    if (filter_state_ != FilterState::ResponseServedFromCache) {
-      // Response is still being fetched from cache -- wait until it is fetched & encoded.
-      filter_state_ = FilterState::WaitingForCacheBody;
-      return Http::FilterHeadersStatus::StopIteration;
-    }
-    return Http::FilterHeadersStatus::Continue;
+    // Stop the encoding stream until the cached response is fetched & added to the encoding stream.
+    return Http::FilterHeadersStatus::StopIteration;
   }
 
   // Either a cache miss or a cache entry that is no longer valid.
@@ -113,11 +102,12 @@ Http::FilterHeadersStatus CacheFilter::encodeHeaders(Http::ResponseHeaderMap& he
 
 Http::FilterDataStatus CacheFilter::encodeData(Buffer::Instance& data, bool end_stream) {
   if (filter_state_ == FilterState::DecodeServingFromCache) {
-    // This call was invoked by decoder_callbacks_->encodeData -- ignore it.
+    // This call was invoked during decoding by decoder_callbacks_->encodeData because a fresh
+    // cached response was found and is being added to the encoding stream -- ignore it.
     return Http::FilterDataStatus::Continue;
   }
-  if (filter_state_ == FilterState::WaitingForCacheBody) {
-    // Encoding stream stopped waiting for cached body (and trailers) to be encoded.
+  if (filter_state_ == FilterState::EncodeServingFromCache) {
+    // Stop the encoding stream until the cached response is fetched & added to the encoding stream.
     return Http::FilterDataStatus::StopIterationAndBuffer;
   }
   if (insert_) {
@@ -129,23 +119,68 @@ Http::FilterDataStatus CacheFilter::encodeData(Buffer::Instance& data, bool end_
   return Http::FilterDataStatus::Continue;
 }
 
+void CacheFilter::getHeaders(Http::RequestHeaderMap& request_headers) {
+  ASSERT(lookup_, "CacheFilter is trying to call getHeaders with no LookupContext");
+
+  // The dispatcher needs to be captured because there's no guarantee that
+  // decoder_callbacks_->dispatcher() is thread-safe.
+  lookup_->getHeaders([this, &request_headers,
+                       &dispatcher = decoder_callbacks_->dispatcher()](LookupResult&& result) {
+    // The callback is posted to the dispatcher to make sure it is called on the worker thread.
+    // The lambda passed to dispatcher.post() needs to be copyable as it will be used to
+    // initialize a std::function. Therefore, it cannot capture anything non-copyable.
+    // LookupResult is non-copyable as LookupResult::headers_ is a unique_ptr, which is
+    // non-copyable. Hence, "result" is decomposed when captured, and re-instantiated inside the
+    // lambda so that "result.headers_" can be captured as a raw pointer, then wrapped in a
+    // unique_ptr when the result is re-instantiated.
+    dispatcher.post(
+        [this, &request_headers, status = result.cache_entry_status_,
+         headers = result.headers_.release(), response_ranges = std::move(result.response_ranges_),
+         content_length = result.content_length_, has_trailers = result.has_trailers_]() mutable {
+          onHeaders(LookupResult{status, absl::WrapUnique(headers), content_length, response_ranges,
+                                 has_trailers},
+                    request_headers);
+        });
+  });
+}
+
 void CacheFilter::getBody() {
   ASSERT(lookup_, "CacheFilter is trying to call getBody with no LookupContext");
   ASSERT(!remaining_body_.empty(), "No reason to call getBody when there's no body to get.");
-  lookup_->getBody(remaining_body_[0],
-                   [this](Buffer::InstancePtr&& body) { onBody(std::move(body)); });
+
+  // The dispatcher needs to be captured because there's no guarantee that
+  // decoder_callbacks_->dispatcher() is thread-safe.
+  lookup_->getBody(remaining_body_[0], [this, &dispatcher = decoder_callbacks_->dispatcher()](
+                                           Buffer::InstancePtr&& body) {
+    // The callback is posted to the dispatcher to make sure it is called on the worker thread.
+    // The lambda passed to dispatcher.post() needs to be copyable as it will be used to
+    // initialize a std::function. Therefore, it cannot capture anything non-copyable.
+    // "body" is a unique_ptr, which is non-copyable. Hence, it is captured as a raw pointer then
+    // wrapped in a unique_ptr inside the lambda.
+    dispatcher.post([this, body = body.release()] { onBody(absl::WrapUnique(body)); });
+  });
 }
 
 void CacheFilter::getTrailers() {
   ASSERT(lookup_, "CacheFilter is trying to call getTrailers with no LookupContext");
   ASSERT(response_has_trailers_, "No reason to call getTrailers when there's no trailers to get.");
-  lookup_->getTrailers(
-      [this](Http::ResponseTrailerMapPtr&& trailers) { onTrailers(std::move(trailers)); });
+
+  // The dispatcher needs to be captured because there's no guarantee that
+  // decoder_callbacks_->dispatcher() is thread-safe.
+  lookup_->getTrailers([this, &dispatcher = decoder_callbacks_->dispatcher()](
+                           Http::ResponseTrailerMapPtr&& trailers) {
+    // The callback is posted to the dispatcher to make sure it is called on the worker thread.
+    // The lambda passed to dispatcher.post() needs to be copyable as it will be used to
+    // initialize a std::function. Therefore, it cannot capture anything non-copyable.
+    // "trailers" is a unique_ptr, which is non-copyable. Hence, it is captured as a raw
+    // pointer then wrapped in a unique_ptr inside the lambda.
+    dispatcher.post(
+        [this, trailers = trailers.release()] { onTrailers(absl::WrapUnique(trailers)); });
+  });
 }
 
 void CacheFilter::onHeaders(LookupResult&& result, Http::RequestHeaderMap& request_headers) {
-  // TODO(yosrym93): Handle request only-if-cached directive.
-  bool should_continue_decoding = false;
+  // TODO(yosrym93): Handle request only-if-cached directive
   switch (result.cache_entry_status_) {
   case CacheEntryStatus::FoundNotModified:
   case CacheEntryStatus::NotSatisfiableRange: // TODO(#10132): create 416 response.
@@ -156,25 +191,23 @@ void CacheFilter::onHeaders(LookupResult&& result, Http::RequestHeaderMap& reque
     // If the cache entry was valid, the response status should be 304 (unmodified) and the cache
     // entry will be injected in the response body.
     lookup_result_ = std::make_unique<LookupResult>(std::move(result));
-    should_continue_decoding = filter_state_ == FilterState::WaitingForCacheLookup;
     filter_state_ = FilterState::ValidatingCachedResponse;
     injectValidationHeaders(request_headers);
     break;
   case CacheEntryStatus::Unusable:
-    should_continue_decoding = filter_state_ == FilterState::WaitingForCacheLookup;
-    filter_state_ = FilterState::NoCachedResponseFound;
     break;
   case CacheEntryStatus::SatisfiableRange: // TODO(#10132): break response content to the ranges
                                            // requested.
   case CacheEntryStatus::Ok:
     lookup_result_ = std::make_unique<LookupResult>(std::move(result));
     filter_state_ = FilterState::DecodeServingFromCache;
     encodeCachedResponse();
+    // Return here so that continueDecoding is not called.
+    // No need to continue the decoding stream as a cached response is already being served.
+    return;
   }
-  if (should_continue_decoding) {
-    // decodeHeaders returned StopIteration waiting for this callback -- continue decoding.
-    decoder_callbacks_->continueDecoding();
-  }
+  // decodeHeaders returned StopIteration waiting for this callback -- continue decoding
+  decoder_callbacks_->continueDecoding();
 }
 
 // TODO(toddmgreer): Handle downstream backpressure.
@@ -236,7 +269,9 @@ void CacheFilter::processSuccessfulValidation(Http::ResponseHeaderMap& response_
   // Check whether the cached entry should be updated before modifying the 304 response.
   const bool should_update_cached_entry = shouldUpdateCachedEntry(response_headers);
 
-  // Update the 304 response status code and content-length.
+  filter_state_ = FilterState::EncodeServingFromCache;
+
+  // Update the 304 response status code and content-length
   response_headers.setStatus(lookup_result_->headers_->getStatusValue());
   response_headers.setContentLength(lookup_result_->headers_->getContentLengthValue());
 
@@ -344,7 +379,7 @@ void CacheFilter::encodeCachedResponse() {
 }
 
 void CacheFilter::finalizeEncodingCachedResponse() {
-  if (filter_state_ == FilterState::WaitingForCacheBody) {
+  if (filter_state_ == FilterState::EncodeServingFromCache) {
     // encodeHeaders returned StopIteration waiting for finishing encoding the cached response --
     // continue encoding.
     encoder_callbacks_->continueEncoding();

source/extensions/filters/http/cache/cache_filter.h

@@ -38,7 +38,8 @@ class CacheFilter : public Http::PassThroughFilter,
   Http::FilterDataStatus encodeData(Buffer::Instance& buffer, bool end_stream) override;
 
 private:
-  // Utility functions: make any necessary checks and call the corresponding lookup_ functions.
+  // Utility functions; make any necessary checks and call the corresponding lookup_ functions
+  void getHeaders(Http::RequestHeaderMap& request_headers);
   void getBody();
   void getTrailers();
 
@@ -95,28 +96,20 @@ class CacheFilter : public Http::PassThroughFilter,
   enum class FilterState {
     Initial,
 
-    // CacheFilter::decodeHeaders called lookup->getHeaders() but onHeaders was not called yet
-    // (lookup result not ready) -- the decoding stream should be stopped until the cache lookup
-    // result is ready.
-    WaitingForCacheLookup,
-
-    // CacheFilter::encodeHeaders called encodeCachedResponse() but encoding the cached response is
-    // not finished yet -- the encoding stream should be stopped until it is finished.
-    WaitingForCacheBody,
-
-    // Cache lookup did not find a cached response for this request.
-    NoCachedResponseFound,
-
-    // Cache lookup found a cached response that requires validation.
+    // Cache lookup found a cached response that requires validation
     ValidatingCachedResponse,
 
     // Cache lookup found a fresh cached response and it is being added to the encoding stream.
     DecodeServingFromCache,
 
+    // A cached response was successfully validated and it is being added to the encoding stream
+    EncodeServingFromCache,
+
     // The cached response was successfully added to the encoding stream (either during decoding or
     // encoding).
     ResponseServedFromCache
   };
+
   FilterState filter_state_ = FilterState::Initial;
 };
 

source/extensions/filters/http/cache/http_cache.h

@@ -241,6 +241,14 @@ class InsertContext {
   // Inserts trailers into the cache.
   virtual void insertTrailers(const Http::ResponseTrailerMap& trailers) PURE;
 
+  // This routine is called prior to an InsertContext being destroyed. InsertContext is responsible
+  // for making sure that any async events are cleaned up in the context of this routine. This
+  // includes timers, network calls, etc. The reason there is an onDestroy() method vs. doing this
+  // type of cleanup in the destructor is due to the deferred deletion model that Envoy uses to
+  // avoid stack unwind complications. InsertContext must not invoke any callbacks after having
+  // onDestroy() invoked.
+  virtual void onDestroy() PURE;
+
   virtual ~InsertContext() = default;
 };
 using InsertContextPtr = std::unique_ptr<InsertContext>;
@@ -277,6 +285,14 @@ class LookupContext {
   // Http::ResponseTrailerMapPtr passed to cb must not be null.
   virtual void getTrailers(LookupTrailersCallback&& cb) PURE;
 
+  // This routine is called prior to an LookupContext being destroyed. LookupContext is responsible
+  // for making sure that any async events are cleaned up in the context of this routine. This
+  // includes timers, network calls, etc. The reason there is an onDestroy() method vs. doing this
+  // type of cleanup in the destructor is due to the deferred deletion model that Envoy uses to
+  // avoid stack unwind complications. LookupContext must not invoke any callbacks after having
+  // onDestroy() invoked.
+  virtual void onDestroy() PURE;
+
   virtual ~LookupContext() = default;
 };
 using LookupContextPtr = std::unique_ptr<LookupContext>;

source/extensions/filters/http/cache/simple_http_cache/simple_http_cache.cc

@@ -37,6 +37,7 @@ class SimpleLookupContext : public LookupContext {
   }
 
   const LookupRequest& request() const { return request_; }
+  void onDestroy() override {}
 
 private:
   SimpleHttpCache& cache_;
@@ -74,6 +75,8 @@ class SimpleInsertContext : public InsertContext {
     NOT_IMPLEMENTED_GCOVR_EXCL_LINE; // TODO(toddmgreer): support trailers
   }
 
+  void onDestroy() override {}
+
 private:
   void commit() {
     committed_ = true;