diff --git a/src/common/events/events_io.cpp b/src/common/events/events_io.cpp index 6ba1e4911..23a3b37d3 100644 --- a/src/common/events/events_io.cpp +++ b/src/common/events/events_io.cpp @@ -104,19 +104,27 @@ ReaderFromAsyncReader::ReaderFromAsyncReader(EventLoop &event_loop, mio::AsyncRe mio::ExpectedSize ReaderFromAsyncReader::Read( vector::iterator start, vector::iterator end) { - mio::ExpectedSize read; - bool finished = false; - event_loop_.Post([start, end, this, &finished, &read]() { - auto err = - reader_->AsyncRead(start, end, [this, &finished, &read](mio::ExpectedSize num_read) { - read = num_read; - finished = true; - event_loop_.Stop(); + auto read = make_shared(); + auto loop_stopped_here = make_shared(false); + event_loop_.Post([start, end, this, read, loop_stopped_here]() { + auto err = reader_->AsyncRead( + start, end, [this, read, loop_stopped_here](mio::ExpectedSize num_read) { + *read = num_read; + // Make sure to only stop the event loop here *once* because + // it's only recursively run once below. If this handler gets + // called multiple time, always stopping the loop here could + // stop the loop for good. + if (!*loop_stopped_here) { + event_loop_.Stop(); + *loop_stopped_here = true; + } }); if (err != error::NoError) { - read = expected::unexpected(err); - finished = true; - event_loop_.Stop(); + *read = expected::unexpected(err); + if (!*loop_stopped_here) { + event_loop_.Stop(); + *loop_stopped_here = true; + } } }); @@ -124,7 +132,7 @@ mio::ExpectedSize ReaderFromAsyncReader::Read( // loop recursively to keep processing events. event_loop_.Run(); - if (!finished) { + if (!*loop_stopped_here) { // If this happens then it means that the event loop was stopped by somebody // else. We have no choice now but to return error, since we have to get out of this // stack frame. We also need to re-stop the event loop, since the first stop was @@ -134,7 +142,7 @@ mio::ExpectedSize ReaderFromAsyncReader::Read( error::Error(make_error_condition(errc::operation_canceled), "Event loop was stopped")); } - return read; + return *read; } TeeReader::ExpectedTeeReaderLeafPtr TeeReader::MakeAsyncReader() { diff --git a/src/common/http/http_resumer.cpp b/src/common/http/http_resumer.cpp index 172d6abd8..ae9bad905 100644 --- a/src/common/http/http_resumer.cpp +++ b/src/common/http/http_resumer.cpp @@ -176,7 +176,7 @@ void HeaderHandlerFunctor::HandleNextResponse( // If an error occurs during handling here, cancel the resuming and call the user handler. auto resp = exp_resp.value(); - auto resumer_reader = resumer_client->resumer_reader_.lock(); + auto resumer_reader = resumer_client->resumer_reader_; if (!resumer_reader) { // Errors should already have been handled as part of the Cancel() inside the // destructor of the reader. @@ -262,7 +262,7 @@ void BodyHandlerFunctor::operator()(http::ExpectedIncomingResponsePtr exp_resp) resumer_client->resumer_state_->offset < resumer_client->resumer_state_->content_length; if (!exp_resp || (is_range_response && is_data_missing)) { if (!exp_resp) { - auto resumer_reader = resumer_client->resumer_reader_.lock(); + auto resumer_reader = resumer_client->resumer_reader_; if (resumer_reader) { resumer_reader->inner_reader_.reset(); } @@ -301,6 +301,17 @@ DownloadResumerAsyncReader::~DownloadResumerAsyncReader() { Cancel(); } +void DownloadResumerAsyncReader::Fail(error::Error err) { + if (last_read_.handler) { + // Remove the handler first in case calling the handler causes this + // function to be called again (or another function relying on the + // handler). + auto handler = last_read_.handler; + last_read_.handler = nullptr; + handler(expected::unexpected(err)); + } +} + void DownloadResumerAsyncReader::Cancel() { auto resumer_client = resumer_client_.lock(); if (!*cancelled_ && resumer_client) { @@ -322,7 +333,7 @@ error::Error DownloadResumerAsyncReader::AsyncRead( "DownloadResumerAsyncReader::AsyncRead called after stream is destroyed"); } // Save user parameters for further resumes of the body read - resumer_client->last_read_ = {.start = start, .end = end, .handler = handler}; + last_read_ = {.start = start, .end = end, .handler = handler}; return AsyncReadResume(); } @@ -334,9 +345,7 @@ error::Error DownloadResumerAsyncReader::AsyncReadResume() { "DownloadResumerAsyncReader::AsyncReadResume called after client is destroyed"); } return inner_reader_->AsyncRead( - resumer_client->last_read_.start, - resumer_client->last_read_.end, - [this](io::ExpectedSize result) { + last_read_.start, last_read_.end, [this](io::ExpectedSize result) { if (!result) { logger_.Warning( "Reading error, a new request will be re-scheduled. " @@ -349,7 +358,7 @@ error::Error DownloadResumerAsyncReader::AsyncReadResume() { logger_.Debug("read " + to_string(result.value()) + " bytes"); auto resumer_client = resumer_client_.lock(); if (resumer_client) { - resumer_client->last_read_.handler(result); + last_read_.handler(result); } else { logger_.Error( "AsyncRead finish handler called after resumer client has been destroyed."); @@ -375,6 +384,7 @@ DownloadResumerClient::~DownloadResumerClient() { logger_.Warning("DownloadResumerClient destroyed while request is still active!"); } client_.Cancel(); + resumer_reader_.reset(); } error::Error DownloadResumerClient::AsyncCall( @@ -395,6 +405,7 @@ error::Error DownloadResumerClient::AsyncCall( *cancelled_ = false; retry_.backoff.Reset(); + resumer_reader_.reset(); resumer_state_->active_state = DownloadResumerActiveStatus::Inactive; resumer_state_->user_handlers_state = DownloadResumerUserHandlersStatus::None; return client_.AsyncCall(req, resumer_header_handler, resumer_body_handler); @@ -472,6 +483,17 @@ error::Error DownloadResumerClient::ScheduleNextResumeRequest() { void DownloadResumerClient::CallUserHandler(http::ExpectedIncomingResponsePtr exp_resp) { if (!exp_resp) { DoCancel(); + + // Fail the resumer_reader because that's the mechanism to deliver the + // information about a failure to the consumer of this + // (DownloadResumerClient) API handling incoming data through the body + // *reader* callback invoked for every chunk of data rather than by + // means of the body *handler* callback which is only called once full + // body is fetched (see the explanation of how this class works and is + // used near its declaration). + if (resumer_reader_) { + resumer_reader_->Fail(exp_resp.error()); + } } if (resumer_state_->user_handlers_state == DownloadResumerUserHandlersStatus::None) { resumer_state_->user_handlers_state = @@ -483,6 +505,8 @@ void DownloadResumerClient::CallUserHandler(http::ExpectedIncomingResponsePtr ex resumer_state_->user_handlers_state = DownloadResumerUserHandlersStatus::BodyHandlerCalled; DoCancel(); user_body_handler_(exp_resp); + // we are done, the body reader won't produce any more data + resumer_reader_.reset(); } else { string msg; if (!exp_resp) { @@ -498,6 +522,7 @@ void DownloadResumerClient::CallUserHandler(http::ExpectedIncomingResponsePtr ex void DownloadResumerClient::Cancel() { DoCancel(); client_.Cancel(); + resumer_reader_.reset(); }; void DownloadResumerClient::DoCancel() { diff --git a/src/common/http_resumer.hpp b/src/common/http_resumer.hpp index fb2099f2a..8e40cf6c4 100644 --- a/src/common/http_resumer.hpp +++ b/src/common/http_resumer.hpp @@ -74,6 +74,9 @@ class DownloadResumerAsyncReader : virtual public io::AsyncReader { vector::iterator end, io::AsyncIoHandler handler) override; + // Stop the reader with an error (unlike Cancel()) + void Fail(error::Error err); + void Cancel() override; private: @@ -90,6 +93,14 @@ class DownloadResumerAsyncReader : virtual public io::AsyncReader { weak_ptr resumer_client_; + // Parameters from the last time that user called AsyncRead. + // They are re-used when resuming the download + struct { + vector::iterator start; + vector::iterator end; + io::AsyncIoHandler handler; + } last_read_; + // The header handler needs to manipulate inner_reader_ in order to replace it in // subsequent requests. friend class HeaderHandlerFunctor; @@ -99,6 +110,26 @@ class DownloadResumerAsyncReader : virtual public io::AsyncReader { // Main class to download the Artifact, which will react to server // disconnections or other sorts of short read by scheduling new HTTP // requests with `Range` header. +// This is how it works: +// - it is an http::ClientInterface class so AsyncCall() is the entry point +// - the given header and body handlers are wrapped into handlers that +// transparently attempt multiple HTTP connections (and requests) in case of +// failures (incl. disconnections) +// - the caller calls MakeBodyAsyncReader() in the header handler and then the +// result's AsyncRead() to be able to process chunks of data from potentially +// multiple HTTP responses, or +// - the caller calls SetBodyWriter() in the header handler in order to +// capture all data from potentially multiple HTTP responses; however, +// SetBodyWriter() internally uses MakeBodyAsyncReader() as well, together +// with AsyncCopy() to move the data between the reader and the respective +// writer passed as argument +// - the body handler is only called when there are no more HTTP connections and +// requests to be done, i.e. when all data is fetched or when an error occurs, +// which may happen after processing any amount of response data (from none to +// all but the last byte) +// - MakeBodyAsyncReader() utilizes the above DownloadResumerAsyncReader class +// which then utilizes this class (DownloadResumerClient) to schedule multiple +// HTTP connections (and requests), if necessary, at the desired intervals // It needs to be used from a shared_ptr class DownloadResumerClient : virtual public http::ClientInterface, @@ -113,6 +144,14 @@ class DownloadResumerClient : http::ResponseHandler header_handler, http::ResponseHandler body_handler) override; + // NOTE: Unlike the basic http::Client which constructs the reader and + // returns it, leaving the API user code to manage the reader's + // lifetime, the DownloadResumerClient keeps a reference (shared + // pointer) to the reader too and thus manages its lifetime. The + // reason for this is that while the http::Client's reader (and it's + // lifetime) is bound to a socket that gets closed on an error or + // completion, only the DownloadResumerClient itself knows when the + // reader won't produce more data (when no more retries). io::ExpectedAsyncReaderPtr MakeBodyAsyncReader(http::IncomingResponsePtr resp) override; void Cancel() override; @@ -143,7 +182,7 @@ class DownloadResumerClient : void DoCancel(); shared_ptr resumer_state_; - weak_ptr resumer_reader_; + shared_ptr resumer_reader_; http::Client client_; log::Logger logger_; @@ -164,16 +203,6 @@ class DownloadResumerClient : events::Timer wait_timer; } retry_; - // Parameters from the last time that user called AsyncRead. - // They are re-used when resuming the download - struct { - vector::iterator start; - vector::iterator end; - io::AsyncIoHandler handler; - } last_read_; - - friend class DownloadResumerAsyncReader; - friend class HeaderHandlerFunctor; friend class BodyHandlerFunctor; }; diff --git a/tests/src/common/http_resumer/http_resumer_test.cpp b/tests/src/common/http_resumer/http_resumer_test.cpp index 1f7fb8099..f4fbbc5b4 100644 --- a/tests/src/common/http_resumer/http_resumer_test.cpp +++ b/tests/src/common/http_resumer/http_resumer_test.cpp @@ -664,9 +664,7 @@ TEST_F(DownloadResumerTest, ResponseBodyReaderSmallBuffer) { reader->RepeatedAsyncRead( buf.begin(), buf.end(), - // Note in particular the capture of `reader`, to keep it alive. - [&buf, reader, body_writer, &got_read_error, &got_read_success]( - io::ExpectedSize result) { + [&buf, body_writer, &got_read_error, &got_read_success](io::ExpectedSize result) { if (!result) { EXPECT_TRUE(false) << "Unexpected error: " << result.error().String(); got_read_error = true; @@ -692,6 +690,115 @@ TEST_F(DownloadResumerTest, ResponseBodyReaderSmallBuffer) { EXPECT_FALSE(got_read_error); } +// Regression test for the daemon hang reported in +// https://hub.mender.io/t/.../8299 : when the body is consumed through the async reader +// (as the update state machine does) and the server stays down until the resumer exhausts +// its retry backoff, the "giving up" error must be delivered through the reader's pending +// read handler. Before the fix it was only routed to the user body handler (which the update +// client intentionally ignores), so the pending read never completed and the daemon wedged +// forever without reporting a deployment failure. +TEST_F(DownloadResumerTest, ServerDownWhileReadingBodyReportsError) { + TestEventLoop loop(chrono::seconds(30)); + + http::ServerConfig server_config; + http::Server server(server_config, loop); + events::Timer kill_timer(loop); + bool kill_scheduled {false}; + + int server_num_requests = 0; + server.AsyncServeUrl( + "http://127.0.0.1:" TEST_PORT, + [](http::ExpectedIncomingRequestPtr exp_req) { + ASSERT_TRUE(exp_req) << exp_req.error().String(); + }, + [&server, &server_num_requests, &kill_timer, &kill_scheduled]( + http::ExpectedIncomingRequestPtr exp_req) { + ASSERT_TRUE(exp_req) << exp_req.error().String(); + + auto result = exp_req.value()->MakeResponse(); + ASSERT_TRUE(result); + auto resp = result.value(); + + server_num_requests++; + + auto size = RangeBodyOfXes::TARGET_BODY_SIZE; + resp->SetStatusCodeAndMessage(200, "Success"); + // Announce the full length, but only deliver a small part of it, so the client + // sees a short read and tries to resume. Keep it small so the first response + // finishes well before the server is taken down below. + resp->SetHeader("Content-Length", to_string(size)); + auto partial_body = make_shared(); + partial_body->SetRanges(0, 4999); + resp->SetBodyReader(partial_body); + + resp->AsyncReply([](error::Error err) { ASSERT_EQ(error::NoError, err); }); + + // After the first (partial) response, take the server down and keep it down, so + // every resume attempt fails until the resumer gives up. Cancelling from a timer + // (rather than the reply handler) avoids tearing down the stream mid-write. + if (!kill_scheduled) { + kill_scheduled = true; + kill_timer.AsyncWait( + chrono::milliseconds(100), [&server](error::Error) { server.Cancel(); }); + } + }); + + http::ClientConfig client_config; + shared_ptr client = + make_shared(client_config, loop); + // First resume waits longer than the server-kill delay above, so the server is already + // down by the time we retry, and the retries are then exhausted quickly. + client->SetSmallestWaitInterval(chrono::milliseconds(200)); + + auto req = make_shared(); + req->SetMethod(http::Method::GET); + req->SetAddress("http://127.0.0.1:" TEST_PORT); + + vector buf; + buf.resize(1235); + bool got_read_error {false}; + error::Error read_error; + + client->AsyncCall( + req, + [&buf, &got_read_error, &read_error, &loop](http::ExpectedIncomingResponsePtr exp_resp) { + ASSERT_TRUE(exp_resp) << exp_resp.error().String(); + auto resp = exp_resp.value(); + + auto exp_reader = resp->MakeBodyAsyncReader(); + ASSERT_TRUE(exp_reader); + auto reader = exp_reader.value(); + reader->RepeatedAsyncRead( + buf.begin(), + buf.end(), + [&got_read_error, &read_error, &loop](io::ExpectedSize result) { + if (!result) { + got_read_error = true; + read_error = result.error(); + loop.Stop(); + return io::Repeat::No; + } + if (result.value() == 0) { + // Unexpected clean EOF; stop so the test does not hang. + loop.Stop(); + return io::Repeat::No; + } + return io::Repeat::Yes; + }); + }, + [](http::ExpectedIncomingResponsePtr exp_resp) { + // Body handler intentionally ignores errors, exactly like the update state + // machine: the error must surface through the reader instead. + }); + + loop.Run(); + + EXPECT_TRUE(got_read_error) + << "resumer gave up but the reader never reported an error (the daemon would hang here)"; + EXPECT_EQ(read_error.code, http::MakeError(http::DownloadResumerError, "").code) + << "unexpected error: " << read_error.String(); +} + TEST_F(DownloadResumerTest, TwoRangesClientReuse) { TestEventLoop loop(chrono::seconds(10)); @@ -1152,7 +1259,7 @@ TEST_F(DownloadResumerTest, UserCancelInBodyHandler) { EXPECT_TRUE(body_handler_called); } -TEST_F(DownloadResumerTest, UserDestroysReader) { +TEST_F(DownloadResumerTest, UserCancelsReader) { TestEventLoop loop(chrono::seconds(10)); // Server @@ -1210,7 +1317,7 @@ TEST_F(DownloadResumerTest, UserDestroysReader) { ASSERT_EQ(err, error::NoError); events::Timer timer(loop); - timer.AsyncWait(chrono::milliseconds(10), [&reader](error::Error err) { reader.reset(); }); + timer.AsyncWait(chrono::milliseconds(10), [&reader](error::Error err) { reader->Cancel(); }); loop.Run(); @@ -1222,7 +1329,7 @@ TEST_F(DownloadResumerTest, UserDestroysReader) { EXPECT_TRUE(body_handler_called); } -TEST_F(DownloadResumerTest, CallerDestroysReaderInTheMiddleOfAWait) { +TEST_F(DownloadResumerTest, CallerCancelsReaderInTheMiddleOfAWait) { TestEventLoop loop(chrono::seconds(10)); // Server @@ -1291,10 +1398,9 @@ TEST_F(DownloadResumerTest, CallerDestroysReaderInTheMiddleOfAWait) { reader = exp_reader.value(); auto writer = make_shared(); - // Note the use of references instead of pointers. This is to make sure that - // the reader gets destroyed by the pointer reset below, otherwise there - // would still be a reference here. The writer we keep alive using the - // capture. + // Note the use of references instead of pointers. This is to avoid + // circular dependencies between callbacks and the writer and + // reader. io::AsyncCopy( *writer, *reader, [writer](error::Error err) { ASSERT_EQ(err, error::NoError); }); }; @@ -1304,12 +1410,10 @@ TEST_F(DownloadResumerTest, CallerDestroysReaderInTheMiddleOfAWait) { EXPECT_EQ(exp_resp.error().code, make_error_condition(errc::operation_canceled)); }; - events::Timer destroy_reader(loop); - destroy_reader.AsyncWait(chrono::milliseconds(500), [&loop, &reader](error::Error err) { + events::Timer cancel_reader(loop); + cancel_reader.AsyncWait(chrono::milliseconds(500), [&loop, &reader](error::Error err) { ASSERT_EQ(err, error::NoError); - // Make sure this destroys the reader. - EXPECT_EQ(reader.use_count(), 1); - reader.reset(); + reader->Cancel(); loop.Stop(); }); @@ -1410,4 +1514,4 @@ TEST_F(DownloadResumerTest, NoServerResponse) { EXPECT_EQ(server_num_requests, 2); EXPECT_EQ(user_num_callbacks, 1); -} \ No newline at end of file +}