http: delaying attach pending requests #2871
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -20,6 +20,12 @@ namespace Envoy { | |
| namespace Http { | ||
| namespace Http1 { | ||
|
|
||
| ConnPoolImpl::ConnPoolImpl(Event::Dispatcher& dispatcher, Upstream::HostConstSharedPtr host, | ||
| Upstream::ResourcePriority priority, | ||
| const Network::ConnectionSocket::OptionsSharedPtr& options) | ||
| : dispatcher_(dispatcher), host_(host), priority_(priority), socket_options_(options), | ||
| upstream_ready_timer_(dispatcher_.createTimer([this]() { onUpstreamReady(); })) {} | ||
|
|
||
| ConnPoolImpl::~ConnPoolImpl() { | ||
| while (!ready_clients_.empty()) { | ||
| ready_clients_.front()->codec_client_->close(); | ||
|
|
@@ -180,7 +186,7 @@ void ConnPoolImpl::onConnectionEvent(ActiveClient& client, Network::ConnectionEv | |
| // whether the client is in the ready list (connected) or the busy list (failed to connect). | ||
| if (event == Network::ConnectionEvent::Connected) { | ||
| conn_connect_ms_->complete(); | ||
| processIdleClient(client, false); | ||
| } | ||
| } | ||
|
|
||
|
|
@@ -209,25 +215,48 @@ void ConnPoolImpl::onResponseComplete(ActiveClient& client) { | |
| host_->cluster().stats().upstream_cx_max_requests_.inc(); | ||
| onDownstreamReset(client); | ||
| } else { | ||
| // Upstream connection might be closed right after response is complete. Setting delay=true | ||
| // here to attach pending requests in next dispatcher loop to handle that case. | ||
| // https://github.com/envoyproxy/envoy/issues/2715 | ||
| processIdleClient(client, true); | ||
| } | ||
| } | ||
|
|
||
| void ConnPoolImpl::onUpstreamReady() { | ||
| upstream_ready_enabled_ = false; | ||
| while (!pending_requests_.empty() && !ready_clients_.empty()) { | ||
| ActiveClient& client = *ready_clients_.front(); | ||
| ENVOY_CONN_LOG(debug, "attaching to next request", *client.codec_client_); | ||
| // There is work to do so bind a request to the client and move it to the busy list. Pending | ||
| // requests are pushed onto the front, so pull from the back. | ||
| attachRequestToClient(client, pending_requests_.back()->decoder_, | ||
| pending_requests_.back()->callbacks_); | ||
| pending_requests_.pop_back(); | ||
| client.moveBetweenLists(ready_clients_, busy_clients_); | ||
| } | ||
| } | ||
|
|
||
| void ConnPoolImpl::processIdleClient(ActiveClient& client, bool delay) { | ||
| client.stream_wrapper_.reset(); | ||
| if (pending_requests_.empty() || delay) { | ||
| // There is nothing to service or delayed processing is requested, so just move the connection | ||
| // into the ready list. | ||
| ENVOY_CONN_LOG(debug, "moving to ready", *client.codec_client_); | ||
| client.moveBetweenLists(busy_clients_, ready_clients_); | ||
|
There was a problem hiding this comment. @lizan Too early to mark the client ready in this branch. Expecting anther read attempt to see if there is There was a problem hiding this comment. At the end of the upstream read event when we reach EOF the client will be removed so this is safe. https://github.com/envoyproxy/envoy/blob/v1.10.0/source/common/network/connection_impl.cc#L486 |
||
| } else { | ||
| // There is work to do immediately so bind a request to the client and move it to the busy list. | ||
| // Pending requests are pushed onto the front, so pull from the back. | ||
| ENVOY_CONN_LOG(debug, "attaching to next request", *client.codec_client_); | ||
| attachRequestToClient(client, pending_requests_.back()->decoder_, | ||
| pending_requests_.back()->callbacks_); | ||
| pending_requests_.pop_back(); | ||
| } | ||
|
|
||
| if (delay && !pending_requests_.empty() && !upstream_ready_enabled_) { | ||
| upstream_ready_enabled_ = true; | ||
| upstream_ready_timer_->enableTimer(std::chrono::milliseconds(0)); | ||
| } | ||
|
|
||
| checkForDrained(); | ||
| } | ||
|
|
||
|
|
||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -936,6 +936,57 @@ void HttpIntegrationTest::testIdleTimeoutWithTwoRequests() { | |
| test_server_->waitForCounterGe("cluster.cluster_0.upstream_cx_idle_timeout", 1); | ||
| } | ||
|
|
||
| void HttpIntegrationTest::testUpstreamDisconnectWithTwoRequests() { | ||
| initialize(); | ||
| fake_upstreams_[0]->set_allow_unexpected_disconnects(true); | ||
|
|
||
| codec_client_ = makeHttpConnection(lookupPort("http")); | ||
|
|
||
| // Request 1. | ||
| codec_client_->makeRequestWithBody(Http::TestHeaderMapImpl{{":method", "GET"}, | ||
| {":path", "/test/long/url"}, | ||
| {":scheme", "http"}, | ||
| {":authority", "host"}}, | ||
| 1024, *response_); | ||
| waitForNextUpstreamRequest(); | ||
|
|
||
| // Request 2. | ||
| IntegrationStreamDecoderPtr response2{new IntegrationStreamDecoder(*dispatcher_)}; | ||
| IntegrationCodecClientPtr codec_client2 = makeHttpConnection(lookupPort("http")); | ||
| codec_client2->makeRequestWithBody(Http::TestHeaderMapImpl{{":method", "GET"}, | ||
| {":path", "/test/long/url"}, | ||
| {":scheme", "http"}, | ||
| {":authority", "host"}}, | ||
| 512, *response2); | ||
|
|
||
| // Response 1. | ||
| upstream_request_->encodeHeaders(Http::TestHeaderMapImpl{{":status", "200"}}, false); | ||
| upstream_request_->encodeData(512, true); | ||
| fake_upstream_connection_->close(); | ||
|
There was a problem hiding this comment. Sorry, I may not have had sufficient caffeine, but how are we making sure that the test Envoy is receiving the FIN before assigning the next request? If there were a context switch between encodeData and the close() couldn't Envoy read the whole request and reassign the upstream before the close() occurs? We may need to unit test this rather than integration test. Alternately we could do a raw tcp connection, send the response and stray data in one write call. I think that'd guarantee no race and I'd hope any activity on the delayed-use connection would cause it to be removed. There was a problem hiding this comment. The dispatcher is not running until next wait* call, so in the right next line Also added a unit test in conn_pool_test too. There was a problem hiding this comment. Yeah, I'd like to think this will be fine, we've just had a spate of macos integration flakes due to slightly different network semantics. We can just keep an eye out for test flakes and remove this if it causes problems now that we have a unit test. @zuercher just so it's on his radar. |
||
| response_->waitForEndStream(); | ||
|
|
||
| EXPECT_TRUE(upstream_request_->complete()); | ||
| EXPECT_TRUE(response_->complete()); | ||
| EXPECT_STREQ("200", response_->headers().Status()->value().c_str()); | ||
| test_server_->waitForCounterGe("cluster.cluster_0.upstream_cx_total", 1); | ||
| test_server_->waitForCounterGe("cluster.cluster_0.upstream_rq_200", 1); | ||
|
|
||
| // Response 2. | ||
| fake_upstream_connection_->waitForDisconnect(); | ||
| fake_upstream_connection_.reset(); | ||
| waitForNextUpstreamRequest(); | ||
| upstream_request_->encodeHeaders(Http::TestHeaderMapImpl{{":status", "200"}}, false); | ||
| upstream_request_->encodeData(1024, true); | ||
| response2->waitForEndStream(); | ||
| codec_client2->close(); | ||
|
|
||
| EXPECT_TRUE(upstream_request_->complete()); | ||
| EXPECT_TRUE(response2->complete()); | ||
| EXPECT_STREQ("200", response2->headers().Status()->value().c_str()); | ||
| test_server_->waitForCounterGe("cluster.cluster_0.upstream_cx_total", 2); | ||
| test_server_->waitForCounterGe("cluster.cluster_0.upstream_rq_200", 2); | ||
| } | ||
|
|
||
| void HttpIntegrationTest::testTwoRequests() { | ||
| initialize(); | ||
|
|
||
|
|
||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
sanity check: this is simply deferring work we used to do in the last dispatcher loop to the next loop, so there's no corner case we'll end up batching together too much work and triggering the watchdog, right?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
The chance of batching together is when multiple upstream connections are completing responses at same time (same epoll callback), I don't think there will will be too much work to trigger watchdog.