Extend AsyncClient to support streaming requests / responses #353

lizan · 2017-01-13T00:20:26Z

Refactor AsyncRequestImpl to AsyncStreamImpl and add streaming requests / responses
Implements AsyncRequestImpl based on AsyncStreamImpl

Existing AsyncClientImplTest covers most of AsyncStreamImpl. Working on more tests for streams.

@qiwzhang PTAL.

Closes #317.

mattklein123 · 2017-01-13T02:56:58Z

@lizan I'm OOO tomorrow, will take a look next week when all tests are in place. At a very high level approach is fine.

RomanDzhabarov · 2017-01-15T21:43:23Z

test/mocks/http/mocks.h

@@ -294,6 +294,14 @@ class MockAsyncClient : public AsyncClient {

  MOCK_METHOD3(send_, Request*(MessagePtr& request, Callbacks& callbacks,
                               const Optional<std::chrono::milliseconds>& timeout));
+
+  Stream* start(StreamCallbacks& callbacks,


no need to override method, you can simply
MOCK_METHOD2(start, Stream*(StreamCallbacks& callbacks,
const Optionalstd::chrono::milliseconds& timeout))
this should work just fine

RomanDzhabarov · 2017-01-15T21:48:36Z

source/common/http/async_client_impl.h

+public:
+  AsyncRequestImpl(MessagePtr&& request, AsyncClientImpl& parent, AsyncClient::Callbacks& callbacks,
+                   const Optional<std::chrono::milliseconds>& timeout);
+  virtual ~AsyncRequestImpl();


hm, AsyncRequestImpl is final there should not be a need for virtual destructor

RomanDzhabarov · 2017-01-15T21:52:23Z

source/common/http/async_client_impl.h

  Router::ProdFilter router_;
  std::function<void()> reset_callback_;
  AccessLog::RequestInfoImpl request_info_;
  RouteImpl route_;
-  bool complete_{};
+  Buffer::Instance* decoding_buffer_{nullptr};
+  bool internal_header_inserted_{false};


nit: bool internal_header_inserted_{};
same for * decoding_buffer_{};

RomanDzhabarov · 2017-01-15T22:53:11Z

include/envoy/http/async_client.h

@@ -50,6 +83,39 @@ class AsyncClient {
    virtual void cancel() PURE;
  };

+  /**
+   * An in-flight HTTP stream


super nit: full stop here and down below for comments on methods.

RomanDzhabarov · 2017-01-15T23:03:29Z

source/common/http/async_client_impl.cc

+AsyncClient::Stream* AsyncClientImpl::start(AsyncClient::StreamCallbacks& callbacks,
+                                            const Optional<std::chrono::milliseconds>& timeout) {
+  AsyncStreamImpl* async_request = new AsyncStreamImpl(*this, callbacks, timeout);
+  std::unique_ptr<AsyncStreamImpl> new_request{async_request};


nit: combine into one line
std::unique_ptr new_request(new AsyncStreamImpl(*this, callbacks, timeout));

RomanDzhabarov · 2017-01-15T23:04:16Z

source/common/http/async_client_impl.cc

+  headers.insertEnvoyInternalRequest().value(Headers::get().EnvoyInternalRequestValues.True);
+  headers.insertForwardedFor().value(parent_.config_.local_info_.address());
+  router_.decodeHeaders(headers, end_stream);
+}


nit: new line in between

qiwzhang · 2017-01-16T05:46:12Z

This is cool. Just wondering if this stream can be shared by multiple threads. 1) can multiple thread write data to this stream. 2) When response data is received, which thread will be called.

mattklein123 · 2017-01-16T17:41:07Z

@qiwzhang no, as written, AsyncClient is thread local and cannot be safely shared between threads. It would be possible to surround it with enough locking to make it safe, but my general recommendation would be to reframe your problem as a thread local problem. I have not yet found a problem that cannot be solved cleanly with this pattern. Happy to discuss this offline with you.

qiwzhang · 2017-01-17T22:22:35Z

@mattklein123 Thanks for your quick response. In this case, istio/proxy will need to create N gRPC streams to istio/mixer if it has N threads. I feel it will be better if they can share the same gRPC stream since some transport optimizations of mixer api are done based on the gRPC stream. We will go with one stream per thread for now to make thing easier, will visit this later if performance is an issue.

mattklein123 · 2017-01-18T03:06:51Z

@lizan can you tag me when the tests are finished and you address roman's comments. then I will take a look.

lizan · 2017-01-18T03:22:10Z

@mattklein123 sure, I'm making more changes since I discovered some problems in the tests. In the meanwhile it would be nice if you can take a look on the interfaces (i.e. async_client.h).

mattklein123 · 2017-01-18T03:26:36Z

@lizan sure will do.

mattklein123 · 2017-01-18T03:27:57Z

include/envoy/http/async_client.h

+    /***
+     * Close the stream.
+     */
+    virtual void close() PURE;


What does this do exactly? The above methods implicitly send end_stream. Is this a reset? If so can we call it reset() and provide a reset reason?

This is comparable to Request::cancel() and its implementation are almost resetting stream and cleanup. I can rename it to reset().

The underlying Filter (i.e. router_)'s interface doesn't have reset reason, so I didn't provide that here. Do you think it is needed?

Sorry you are right, I just looked at the code again. The fact that there is no reason is due to how we are wrapping the router filter to do all of this. I would just rename to reset() for clarity since IMO that makes more sense in the streaming case.

mattklein123 · 2017-01-18T03:30:56Z

include/envoy/http/async_client.h

+   *         the handle can be used to send more messages or close the stream.
+   */
+  virtual Stream* start(StreamCallbacks& callbacks,
+                        const Optional<std::chrono::milliseconds>& timeout) PURE;


It's questionable whether timeout makes sense on this interface, since the only reason I could see it really ever being used is for long lived streaming API requests like you are implementing it for. But I don't have a strong opinion and could go either way. I guess since it's optional, we can leave it. (Please update the comments though).

I don't have a strong opinion either. I will update comments soon.

Explicitly says if no timeout given, it's infinite.

What is behavior if timeout is hit, does it implicitly call Stream.reset() and by implication the Callback.onResetStream gets called & on what thread?

This is all single threaded. Yes, when timeout is hit the stream will be reset with callbacks called.

When timeout, it will call StreamCallbacks.onHeader with :status 504 and onData, the header and data is coming from underlying filters.

See this test:
https://github.com/lyft/envoy/pull/353/files#diff-918cf37446b66e452d1114b88b1e61c4R578

lizan · 2017-01-19T05:39:01Z

@mattklein123 Thanks for reviewing it, and sorry I was busy on other things today, will update the PR with tests soon.

lizan · 2017-01-22T20:53:16Z

@mattklein123 This is ready to review, PTAL. Thanks!

mattklein123 · 2017-01-22T21:10:02Z

@RomanDzhabarov please review. I will take a final pass once you are done.

RomanDzhabarov · 2017-01-23T21:31:51Z

I'll take another pass on this today.

fengli79 · 2017-01-24T18:05:50Z

include/envoy/http/async_client.h

+    virtual ~Stream() {}
+
+    /***
+     * Send headers to the stream.


Explicitly says the sendHeaders method cannot be invoked more than once?

RomanDzhabarov · 2017-01-24T04:41:21Z

source/common/http/async_client_impl.h

  Router::ProdFilter router_;
  std::function<void()> reset_callback_;
  AccessLog::RequestInfoImpl request_info_;
  RouteImpl route_;
-  bool complete_{};
+  Buffer::Instance* decoding_buffer_{nullptr};


nit: decoding_buffer_{}

RomanDzhabarov · 2017-01-24T04:43:13Z

test/mocks/http/mocks.h

+  void onResetStream() override { onResetStream_(); }
+
+  MOCK_METHOD2(onHeaders_, void(HeaderMap& headers, bool end_stream));
+  MOCK_METHOD2(onData_, void(Buffer::Instance& data, bool end_stream));


"void onData(Buffer::Instance& data, bool end_stream) override { onData_(data, end_stream); }" is not required, just define MOCK_METHOD2 properly for onData.

RomanDzhabarov · 2017-01-24T04:43:22Z

test/mocks/http/mocks.h

+  }
+  void onData(Buffer::Instance& data, bool end_stream) override { onData_(data, end_stream); }
+  void onTrailers(HeaderMapPtr&& trailers) override { onTrailers_(*trailers); }
+  void onResetStream() override { onResetStream_(); }


RomanDzhabarov · 2017-01-24T04:48:55Z

test/mocks/http/mocks.h

@@ -317,6 +338,19 @@ class MockAsyncClientRequest : public AsyncClient::Request {

  MockAsyncClient* client_;
 };
+
+class MockAsyncClientStream : public AsyncClient::Stream {


where is this used at all? and other added mocks.

This mock is not used, removed. What are others? MockAsyncClientStreamCallbacks is used in the AsyncClientImplTest.

louiscryan · 2017-01-24T18:41:02Z

include/envoy/http/async_client.h

@@ -38,7 +38,40 @@ class AsyncClient {
  };

  /**
-   * An in-flight HTTP request
+   * Notifies caller of async HTTP stream status.


It might be worth having some exposition about the full-duplex nature of streams in the documentation. For instance StreamCallbacks can continue to receive events even if the Stream has had sendXXX(..., end_stream=true) called.

louiscryan · 2017-01-24T18:51:29Z

source/common/http/async_client_impl.cc

+}
+
+void AsyncStreamImpl::closeLocal(bool end_stream) {
+  local_closed_ |= end_stream;


debug assertion if local already closed and end_stream = true

louiscryan · 2017-01-24T18:58:13Z

source/common/http/async_client_impl.cc


-void AsyncRequestImpl::encodeHeaders(HeaderMapPtr&& headers, bool end_stream) {
-  response_.reset(new ResponseMessageImpl(std::move(headers)));
+void AsyncStreamImpl::encodeHeaders(HeaderMapPtr&& headers, bool end_stream) {


@mattklein123 this is some odd nomenclature, why are headers received from the transport in a method called 'encode...' ?

I explained this above, this is an internal implementation detail, the response is being "encoded" into the client, which is then dispatching the appropriate callbacks.

mattklein123 · 2017-01-24T23:40:49Z

@lizan there are a huge amount of comments on this review. I'm fine with the overall approach. When everyone's concerns are satisfied or if there are other questions let me know and I can review in detail.

lizan · 2017-01-27T18:15:25Z

@fengli79 @louiscryan @RomanDzhabarov PTAL again, thanks!

mattklein123 · 2017-01-27T19:47:39Z

source/common/http/async_client_impl.cc

  }
 }

 AsyncClient::Request* AsyncClientImpl::send(MessagePtr&& request, AsyncClient::Callbacks& callbacks,
                                            const Optional<std::chrono::milliseconds>& timeout) {
-  std::unique_ptr<AsyncRequestImpl> new_request{
-      new AsyncRequestImpl(std::move(request), *this, callbacks, timeout)};
+  AsyncRequestImpl* async_request =


nit: just initialize this directly in unique_ptr and return front().get() like you do below.

Having a raw pointer here is to avoid downcasting from AsyncStreamImpl to AsyncRequestImpl when it returns. AsyncStreamImpl is not a subclass of AsyncClient::Request.

ok sorry sounds good

mattklein123 · 2017-01-27T19:49:26Z

source/common/http/async_client_impl.cc

+  ASSERT(!(local_closed_ && end_stream));
+
+  local_closed_ |= end_stream;
+  if (complete())


nit: braces around all if statements

mattklein123 · 2017-01-27T19:49:34Z

source/common/http/async_client_impl.cc

+
+void AsyncStreamImpl::closeRemote(bool end_stream) {
+  remote_closed_ |= end_stream;
+  if (complete())


nit: braces around all if statements

mattklein123 · 2017-01-27T19:51:58Z

source/common/http/async_client_impl.h

+  void onTrailers(HeaderMapPtr&& trailers) override;
+  void onReset() override;
+  void onComplete();
+  MessagePtr request_;


nit: newline before this line

mattklein123 · 2017-01-27T19:52:18Z

source/common/http/async_client_impl.h

+  void onData(Buffer::Instance& data, bool end_stream) override;
+  void onTrailers(HeaderMapPtr&& trailers) override;
+  void onReset() override;
+  void onComplete();


nit: move to top of section

mattklein123 · 2017-01-27T19:52:39Z

source/common/http/async_client_impl.h

+  virtual void cancel() override;
+
+private:
+  void onHeaders(HeaderMapPtr&& headers, bool end_stream) override;


prefix overrides with // AsyncClient::StreamCallbacks

mattklein123 · 2017-01-27T19:56:35Z

source/common/http/async_client_impl.cc

-  cleanup();
+void AsyncStreamImpl::sendData(Buffer::Instance& data, bool end_stream) {
+  router_.decodeData(data, end_stream);
+  decoding_buffer_ = &data;


This is not quite right and is kind of complicated. In short:

When you call router_.decodeData(...) it will return a response code which we currently ignore which tells you whether to buffer or not. Buffering is only done in certain cases, for example, if we might retry, or shadow. In this case, the "connection manager" (AsyncClient) is expected to buffer the entire request.

In the streaming case, it doesn't make sense to buffer. Therefore, I would probably check the return code of router_.decodeData(...) and if it asks for buffering, throw an exception of some kind to indicate programmer error.

so basically for the streaming case I think decodingBuffer() should return nullptr, in this normal case, it should return the request body.

Sorry just to further clarify, buffering should continue to work in the AsyncRequestImpl case (non-streaming).

mattklein123 · 2017-01-27T20:00:33Z

source/common/http/async_client_impl.cc

-  if (!new_request->complete_) {
-    new_request->moveIntoList(std::move(new_request), active_requests_);
-    return active_requests_.front().get();
+  if (!new_request->complete()) {


If I'm reading the code correctly, I think you actually want to check for remote_close_ here now, not complete_, or, arguably better, you need to set local_close_ in resetStream().

The reason for this is that the router might respond right away. If it responds before the request is complete, it will reset. In this case, you should return a null handle.

Sorry just looked at the code in detail (I haven't looked at this stuff in a while). The router will not actually call resetStream() on its own, it relies on the connection manager to reset if the request when the response is complete. (The main server makes the assumption that even in the streaming case, if the response is finished, the request should be reset if it is not finished).

So, in this case, I would probably just check for remote_close_ and if that is set, return a null handle and do not send any more data. You could also replicate the connection manager behavior, and if remote is closed before local is closed, close local and "reset."

P.S., this stuff is pretty complicated. If you want to chat in realtime find me in the public Gitter.

looked at the code again, check remote_close_ sounds reasonable, done.

lizan · 2017-02-01T01:33:08Z

@mattklein123 Sorry for the delay, PTAL when you have time, thanks!

mattklein123 · 2017-02-01T18:52:25Z

source/common/http/async_client_impl.cc

+  FilterDataStatus status = router_.decodeData(data, end_stream);
+
+  if (status == FilterDataStatus::StopIterationAndBuffer) {
+    decoding_buffer_ = &data;


This is still not right. To probably support buffering you actually need to buffer the data. For example, have Buffer::OwnedImpl, and add each data frame to it. I don't think this is worth supporting, as I don't think there is any reasonable case where for a streaming request we will need buffering. I would refactor the code slightly in such a way where in the base streaming case, you don't support buffering (throw an exception if something results in buffering being needed), but in the non-streaming case you do what we did previously, which is to just return the request's body data.

Done, refactored it to throw exception in AsyncStreamImpl, and return body in AsyncRequestImpl.

mattklein123 · 2017-02-01T21:06:15Z

source/common/http/async_client_impl.h

@@ -164,7 +164,7 @@ class AsyncStreamImpl : public AsyncClient::Stream,
  AccessLog::RequestInfo& requestInfo() override { return request_info_; }
  const std::string& downstreamAddress() override { return EMPTY_STRING; }
  void continueDecoding() override { NOT_IMPLEMENTED; }
-  const Buffer::Instance* decodingBuffer() override { return decoding_buffer_; }
+  const Buffer::Instance* decodingBuffer() override { NOT_IMPLEMENTED; }


this is going to just crash. Can you throw an actual EnvoyException (which will also end up crashing the process but indirectly) and add a test for this case please.

regression from #353

…09-rel14

Description: this PR fixes a couple bugs to allow e2e use of the Envoy Mobile library in the swift iOS demo. Risk Level: low - fixing bugs Testing: ./bazelw build --config=ios --xcode_version=10.3.0.10G8 //:ios_dist and then ./bazelw run --config=ios --xcode_version=10.3.0.10G8 //examples/swift/hello_world:app Signed-off-by: Jose Nino <jnino@lyft.com> Signed-off-by: JP Simard <jp@jpsim.com>

Extend AsyncClient to support streaming requests / responses

ebbb140

lizan added 2 commits January 12, 2017 20:39

Add comment for start

7266341

Add mocks

a1d2b14

RomanDzhabarov reviewed Jan 15, 2017

View reviewed changes

Address comments

373ba1f

mattklein123 reviewed Jan 18, 2017

View reviewed changes

lizan added 2 commits January 18, 2017 21:24

add BasicStream test

7c7ab57

Merge remote-tracking branch 'upstream/master' into async_http_stream

803f824

lizan added 8 commits January 19, 2017 01:43

Handle stream close in AsyncStreamImpl

f157b49

rename close() to reset()

0a491f7

rename close() to reset()

410e361

more tests

edbcb91

comment about timeout

6b5b227

more tests

46ac4ca

Merge remote-tracking branch 'upstream/master' into async_http_stream

663c6fc

full stops

afd4b0d

mattklein123 closed this Jan 22, 2017

mattklein123 reopened this Jan 22, 2017

fengli79 reviewed Jan 24, 2017

View reviewed changes

RomanDzhabarov reviewed Jan 24, 2017

View reviewed changes

louiscryan reviewed Jan 24, 2017

View reviewed changes

Address comments

05538d4

fengli79 mentioned this pull request Jan 27, 2017

Supports GRPC streaming based on Asyn HTTP2 client. #381

Closed

lizan added 2 commits January 26, 2017 19:38

Add comment for full-duplex stream

944bde5

call StreamCallbacks.onReset() when Stream.reset() is called

a1c75c7

mattklein123 reviewed Jan 27, 2017

View reviewed changes

lizan added 3 commits January 30, 2017 17:16

remote_closed_ instead of complete()

ab38439

buffer when needed

580edff

Merge remote-tracking branch 'upstream/master' into async_http_stream

5a1ccc6

mattklein123 reviewed Feb 1, 2017

View reviewed changes

only buffering for AsyncRequestImpl

6bc5ef5

mattklein123 reviewed Feb 1, 2017

View reviewed changes

throw EnvoyException instead

a321140

mattklein123 approved these changes Feb 2, 2017

View reviewed changes

mattklein123 merged commit 14d3e97 into envoyproxy:master Feb 2, 2017

mattklein123 added a commit that referenced this pull request Feb 2, 2017

async client: fix immediate response when request has body

d4c0900

regression from #353

mattklein123 mentioned this pull request Feb 2, 2017

async client: fix immediate response when request has body #411

Merged

mattklein123 added a commit that referenced this pull request Feb 2, 2017

async client: fix immediate response when request has body (#411)

1260794

regression from #353

lizan mentioned this pull request Jun 13, 2017

gRPC streaming client. #1054

Merged

GregHanson referenced this pull request in istio/envoy Jun 9, 2022

Merge pull request #353 from istio-private/envoy-cve-patches-2022-06-…

613c2df

…09-rel14

Extend AsyncClient to support streaming requests / responses #353

Extend AsyncClient to support streaming requests / responses #353

Conversation

lizan commented Jan 13, 2017

mattklein123 commented Jan 13, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

qiwzhang commented Jan 16, 2017

mattklein123 commented Jan 16, 2017

qiwzhang commented Jan 17, 2017

mattklein123 commented Jan 18, 2017

lizan commented Jan 18, 2017

mattklein123 commented Jan 18, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

louiscryan Jan 24, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lizan commented Jan 19, 2017

lizan commented Jan 22, 2017

mattklein123 commented Jan 22, 2017

RomanDzhabarov commented Jan 23, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mattklein123 commented Jan 24, 2017

lizan commented Jan 27, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lizan commented Feb 1, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

louiscryan Jan 24, 2017 •

edited

Loading