core: dispatcher test suite and stream completion

.gitmodules

@@ -1,3 +1,3 @@
 [submodule "envoy"]
 	path = envoy
-	url = https://github.com/envoyproxy/envoy.git
+	url = https://github.com/goaway/envoy.git

library/common/http/dispatcher.cc

@@ -9,38 +9,41 @@ namespace Http {
 Dispatcher::DirectStreamCallbacks::DirectStreamCallbacks(envoy_stream_t stream,
                                                          envoy_observer observer,
                                                          Dispatcher& http_dispatcher)
-    : stream_(stream), observer_(observer), http_dispatcher_(http_dispatcher) {}
+    : stream_id_(stream), observer_(observer), http_dispatcher_(http_dispatcher) {}
 
 void Dispatcher::DirectStreamCallbacks::onHeaders(HeaderMapPtr&& headers, bool end_stream) {
-  ENVOY_LOG(debug, "response headers for stream (end_stream={}):\n{}", end_stream, *headers);
-  if (end_stream) {
-    http_dispatcher_.removeStream(stream_);
-  }
+  ENVOY_LOG(debug, "[S{}] response headers for stream (end_stream={}):\n{}", stream_id_, end_stream,
+            *headers);
   observer_.on_headers_f(Utility::transformHeaders(*headers), end_stream, observer_.context);
 }
 
 void Dispatcher::DirectStreamCallbacks::onData(Buffer::Instance& data, bool end_stream) {
-  ENVOY_LOG(debug, "response data for stream (length={} end_stream={})", data.length(), end_stream);
-  if (end_stream) {
-    http_dispatcher_.removeStream(stream_);
-  }
+  ENVOY_LOG(debug, "[S{}] response data for stream (length={} end_stream={})", stream_id_,
+            data.length(), end_stream);
   observer_.on_data_f(Envoy::Buffer::Utility::transformData(data), end_stream, observer_.context);
 }
 
 void Dispatcher::DirectStreamCallbacks::onTrailers(HeaderMapPtr&& trailers) {
-  ENVOY_LOG(debug, "response trailers for stream:\n{}", *trailers);
-  http_dispatcher_.removeStream(stream_);
+  ENVOY_LOG(debug, "[S{}] response trailers for stream:\n{}", stream_id_, *trailers);
   observer_.on_trailers_f(Utility::transformHeaders(*trailers), observer_.context);
 }
 
+void Dispatcher::DirectStreamCallbacks::onComplete() {
+  ENVOY_LOG(debug, "[S{}] complete stream", stream_id_);
+  observer_.on_complete_f(observer_.context);
+  http_dispatcher_.cleanup(stream_id_);
+}
+
 void Dispatcher::DirectStreamCallbacks::onReset() {
-  http_dispatcher_.removeStream(stream_);
+  ENVOY_LOG(debug, "[S{}] remote reset stream", stream_id_);
   observer_.on_error_f({ENVOY_STREAM_RESET, {0, nullptr}}, observer_.context);
 }
 
-Dispatcher::DirectStream::DirectStream(AsyncClient::Stream& underlying_stream,
+Dispatcher::DirectStream::DirectStream(envoy_stream_t stream_id,
+                                       AsyncClient::Stream& underlying_stream,
                                        DirectStreamCallbacksPtr&& callbacks)
-    : underlying_stream_(underlying_stream), callbacks_(std::move(callbacks)) {}
+    : stream_id_(stream_id), underlying_stream_(underlying_stream),
+      callbacks_(std::move(callbacks)) {}
 
 Dispatcher::Dispatcher(Event::Dispatcher& event_dispatcher,
                        Upstream::ClusterManager& cluster_manager)
@@ -62,30 +65,30 @@ envoy_stream_t Dispatcher::startStream(envoy_observer observer) {
       callbacks->onReset();
     } else {
       DirectStreamPtr direct_stream =
-          std::make_unique<DirectStream>(*underlying_stream, std::move(callbacks));
+          std::make_unique<DirectStream>(new_stream_id, *underlying_stream, std::move(callbacks));
       streams_.emplace(new_stream_id, std::move(direct_stream));
-      ENVOY_LOG(debug, "started stream [{}]", new_stream_id);
+      ENVOY_LOG(debug, "[S{}] start stream", new_stream_id);
     }
   });
 
   return new_stream_id;
 }
 
-envoy_status_t Dispatcher::sendHeaders(envoy_stream_t stream_id, envoy_headers headers,
+envoy_status_t Dispatcher::sendHeaders(envoy_stream_t stream, envoy_headers headers,
                                        bool end_stream) {
-  event_dispatcher_.post([this, stream_id, headers, end_stream]() -> void {
-    DirectStream* direct_stream = getStream(stream_id);
+  event_dispatcher_.post([this, stream, headers, end_stream]() -> void {
+    DirectStream* direct_stream = getStream(stream);
     // If direct_stream is not found, it means the stream has already closed or been reset
     // and the appropriate callback has been issued to the caller. There's nothing to do here
     // except silently swallow this.
     // TODO: handle potential race condition with cancellation or failure get a stream in the
-    // first place. Additionally it is possible to get a nullptr due to bogus stream_id
+    // first place. Additionally it is possible to get a nullptr due to bogus envoy_stream_t
     // from the caller.
     // https://github.com/lyft/envoy-mobile/issues/301
     if (direct_stream != nullptr) {
       direct_stream->headers_ = Utility::transformHeaders(headers);
-      ENVOY_LOG(debug, "request headers for stream [{}] (end_stream={}):\n{}", stream_id,
-                end_stream, *direct_stream->headers_);
+      ENVOY_LOG(debug, "[S{}] request headers for stream (end_stream={}):\n{}", stream, end_stream,
+                *direct_stream->headers_);
       direct_stream->underlying_stream_.sendHeaders(*direct_stream->headers_, end_stream);
     }
   });
@@ -99,19 +102,35 @@ envoy_status_t Dispatcher::sendMetadata(envoy_stream_t, envoy_headers, bool) {
   return ENVOY_FAILURE;
 }
 envoy_status_t Dispatcher::sendTrailers(envoy_stream_t, envoy_headers) { return ENVOY_FAILURE; }
-envoy_status_t Dispatcher::locallyCloseStream(envoy_stream_t) { return ENVOY_FAILURE; }
-envoy_status_t Dispatcher::resetStream(envoy_stream_t) { return ENVOY_FAILURE; }
 
-Dispatcher::DirectStream* Dispatcher::getStream(envoy_stream_t stream_id) {
+envoy_status_t Dispatcher::resetStream(envoy_stream_t stream) {
+  event_dispatcher_.post([this, stream]() -> void {
+    DirectStream* direct_stream = getStream(stream);
+    if (direct_stream) {
+      direct_stream->underlying_stream_.reset();
+    }
+  });
+  return ENVOY_SUCCESS;
+}
+
+Dispatcher::DirectStream* Dispatcher::getStream(envoy_stream_t stream) {
   ASSERT(event_dispatcher_.isThreadSafe(),
          "stream interaction must be performed on the event_dispatcher_'s thread.");
-  auto direct_stream_pair_it = streams_.find(stream_id);
+  auto direct_stream_pair_it = streams_.find(stream);
   return (direct_stream_pair_it != streams_.end()) ? direct_stream_pair_it->second.get() : nullptr;
 }
 
-// TODO: implement. Note: the stream might not be in the map if for example startStream called
-// onReset due to its inability to get an underlying stream.
-envoy_status_t Dispatcher::removeStream(envoy_stream_t) { return ENVOY_FAILURE; }
+void Dispatcher::cleanup(envoy_stream_t stream_id) {
+  DirectStream* direct_stream = getStream(stream_id);
+
+  RELEASE_ASSERT(direct_stream,
+                 "cleanup is a private method that is only called with stream ids that exist");
+
+  // TODO: think about thread safety of deleting the DirectStream immediately.
+  size_t erased = streams_.erase(stream_id);
+  ASSERT(erased == 1, "cleanup should always remove one entry from the streams map");
+  ENVOY_LOG(debug, "[S{}] remove stream", stream_id);
+}
 
 } // namespace Http
 } // namespace Envoy

library/common/http/dispatcher.h

@@ -36,13 +36,11 @@ class Dispatcher : public Logger::Loggable<Logger::Id::http> {
   envoy_status_t sendData(envoy_stream_t stream, envoy_headers headers, bool end_stream);
   envoy_status_t sendMetadata(envoy_stream_t stream, envoy_headers headers, bool end_stream);
   envoy_status_t sendTrailers(envoy_stream_t stream, envoy_headers headers);
-  envoy_status_t locallyCloseStream(envoy_stream_t stream);
   // TODO: when implementing this function we have to make sure to prevent races with already
   // scheduled and potentially scheduled callbacks. In order to do so the platform callbacks need to
   // check for atomic state (boolean most likely) that will be updated here to mark the stream as
   // closed.
   envoy_status_t resetStream(envoy_stream_t stream);
-  envoy_status_t removeStream(envoy_stream_t stream);
 
 private:
   /**
@@ -55,17 +53,18 @@ class Dispatcher : public Logger::Loggable<Logger::Id::http> {
   class DirectStreamCallbacks : public AsyncClient::StreamCallbacks,
                                 public Logger::Loggable<Logger::Id::http> {
   public:
-    DirectStreamCallbacks(envoy_stream_t stream, envoy_observer observer,
+    DirectStreamCallbacks(envoy_stream_t stream_id, envoy_observer observer,
                           Dispatcher& http_dispatcher);
 
     // AsyncClient::StreamCallbacks
     void onHeaders(HeaderMapPtr&& headers, bool end_stream);
     void onData(Buffer::Instance& data, bool end_stream);
     void onTrailers(HeaderMapPtr&& trailers);
+    void onComplete();
     void onReset();
 
   private:
-    const envoy_stream_t stream_;
+    const envoy_stream_t stream_id_;
     const envoy_observer observer_;
     Dispatcher& http_dispatcher_;
   };
@@ -77,13 +76,11 @@ class Dispatcher : public Logger::Loggable<Logger::Id::http> {
    * AsyncClient::Stream and in the incoming direction via DirectStreamCallbacks.
    */
   class DirectStream {
-    // TODO: Bookkeeping for this class is insufficient to fully cover all cases necessary to
-    // track the lifecycle of the underlying_stream_. One way or another, we must fix this
-    // to prevent bugs in the future. (Enhanced internal bookkeeping is probably good enough,
-    // but other options include upstream modifications to AsyncClient and friends.
   public:
-    DirectStream(AsyncClient::Stream& underlying_stream, DirectStreamCallbacksPtr&& callbacks);
+    DirectStream(envoy_stream_t stream_id, AsyncClient::Stream& underlying_stream,
+                 DirectStreamCallbacksPtr&& callbacks);
 
+    const envoy_stream_t stream_id_;
     // Used to issue outgoing HTTP stream operations.
     AsyncClient::Stream& underlying_stream_;
     // Used to receive incoming HTTP stream operations.
@@ -102,6 +99,7 @@ class Dispatcher : public Logger::Loggable<Logger::Id::http> {
   // Everything in the below interface must only be accessed from the event_dispatcher's thread.
   // This allows us to generally avoid synchronization.
   DirectStream* getStream(envoy_stream_t stream_id);
+  void cleanup(envoy_stream_t stream_id);
 
   std::unordered_map<envoy_stream_t, DirectStreamPtr> streams_;
   std::atomic<envoy_stream_t> current_stream_id_;

library/common/include/c_types.h

@@ -114,6 +114,7 @@ typedef void (*on_metadata)(envoy_headers metadata, void* context);
  * execution.
  */
 typedef void (*on_trailers)(envoy_headers trailers, void* context);
+typedef void (*on_complete)(void* context);
 /**
  * Called when the async HTTP stream has an error.
  * @param envoy_error, the error received/caused by the async HTTP stream.
@@ -134,6 +135,7 @@ typedef struct {
   on_data on_data_f;
   on_metadata on_metadata_f;
   on_trailers on_trailers_f;
+  on_complete on_complete_f;
   on_error on_error_f;
   void* context; // Will be passed through to callbacks to provide dispatch and execution state.
 } envoy_observer;

library/common/main_interface.cc

@@ -24,16 +24,16 @@ envoy_stream start_stream(envoy_observer observer) {
   return {ENVOY_SUCCESS, http_dispatcher_->startStream(observer)};
 }
 
-envoy_status_t send_headers(envoy_stream_t stream_id, envoy_headers headers, bool end_stream) {
-  return http_dispatcher_->sendHeaders(stream_id, headers, end_stream);
+envoy_status_t send_headers(envoy_stream_t stream, envoy_headers headers, bool end_stream) {
+  return http_dispatcher_->sendHeaders(stream, headers, end_stream);
 }
 
 // TODO: implement.
 envoy_status_t send_data(envoy_stream_t, envoy_data, bool) { return ENVOY_FAILURE; }
 envoy_status_t send_metadata(envoy_stream_t, envoy_headers, bool) { return ENVOY_FAILURE; }
 envoy_status_t send_trailers(envoy_stream_t, envoy_headers) { return ENVOY_FAILURE; }
-envoy_status_t locally_close_stream(envoy_stream_t) { return ENVOY_FAILURE; }
-envoy_status_t reset_stream(envoy_stream_t) { return ENVOY_FAILURE; }
+
+envoy_status_t reset_stream(envoy_stream_t stream) { return http_dispatcher_->resetStream(stream); }
 
 /*
  * Setup envoy for interaction via the main interface.

library/common/main_interface.h

@@ -55,14 +55,6 @@ envoy_status_t send_metadata(envoy_stream_t stream, envoy_headers metadata, bool
  */
 envoy_status_t send_trailers(envoy_stream_t stream, envoy_headers trailers);
 
-/**
- * Half-close an HTTP stream. The stream will be observable and may return further data
- * via the observer callbacks. However, nothing further may be sent.
- * @param stream, the stream to close.
- * @return envoy_status_t, the resulting status of the operation.
- */
-envoy_status_t locally_close_stream(envoy_stream_t stream);
-
 /**
  * Detach all observers from a stream and send an interrupt upstream if supported by transport.
  * @param stream, the stream to evict.

test/common/http/BUILD

@@ -4,6 +4,23 @@ load("@envoy//bazel:envoy_build_system.bzl", "envoy_cc_test", "envoy_package")
 
 envoy_package()
 
+envoy_cc_test(
+    name = "dispatcher_test",
+    srcs = ["dispatcher_test.cc"],
+    repository = "@envoy",
+    deps = [
+        "//library/common/http:dispatcher_lib",
+        "//library/common/http:header_utility_lib",
+        "//library/common/include:c_types_interface",
+        "@envoy//source/common/http:async_client_lib",
+        "@envoy//source/common/http:context_lib",
+        "@envoy//test/common/http:common_lib",
+        "@envoy//test/mocks/event:event_mocks",
+        "@envoy//test/mocks/local_info:local_info_mocks",
+        "@envoy//test/mocks/upstream:upstream_mocks",
+    ],
+)
+
 envoy_cc_test(
     name = "header_utility_test",
     srcs = ["header_utility_test.cc"],