diff --git a/include/anyhttp/asio/runtime.hpp b/include/anyhttp/asio/runtime.hpp index 3bf8aea..a82cb80 100644 --- a/include/anyhttp/asio/runtime.hpp +++ b/include/anyhttp/asio/runtime.hpp @@ -521,8 +521,13 @@ class Event /** * A timer that calls back when it expires, which is what an engine with timeouts of its own * (ngtcp2) needs: it tells when, and wants to be called then. Arming the timer again replaces - * both the expiry and the callback, and a callback that has been replaced or cancelled is never - * called. Callbacks run on the timer's executor. + * both the expiry and the callback. Callbacks run on the timer's executor. + * + * Arming it again, cancel() and the destructor stop a wait that is still pending, but no + * asynchronous operation can be taken back once it has completed: when the delay has expired + * already, the callback is on its way, and it still runs, once, soon after. So a callback holds + * what it uses rather than refer to it -- a `weak_ptr` to its session, not `this` -- and a call + * that is late does no harm. */ class Timer { diff --git a/include/anyhttp/h2/session_details.hpp b/include/anyhttp/h2/session_details.hpp index 10a9a17..b0e7f81 100644 --- a/include/anyhttp/h2/session_details.hpp +++ b/include/anyhttp/h2/session_details.hpp @@ -235,7 +235,8 @@ Task NGHttp2SessionImpl::end_connection() // if (auto ec = io::shutdown(stream_, io::Shutdown::send); !ec) { - if (auto drained = co_await io::drain(stream_, this->get_executor(), std::chrono::seconds(2))) + if (auto drained = co_await io::drain(stream_, this->weak_from_this(), this->get_executor(), + std::chrono::seconds(2))) mlogd("dropped {} bytes the peer sent after the session ended", drained); } else if (ec != errc::not_connected) // the peer may be gone already @@ -456,7 +457,11 @@ IoTask<> ClientSession::shutdown() this->start_write(); Timer timer(this->get_executor()); - timer.arm(std::chrono::seconds(2), [this] { io::cancel(stream_); }); + timer.arm(std::chrono::seconds(2), [weak = this->weak_from_this()] { + // may come late, see Timer + if (auto self = std::static_pointer_cast(weak.lock())) + io::cancel(self->stream_); + }); co_await this->session_done_.wait(); timer.cancel(); diff --git a/include/anyhttp/h2v2/session_details.hpp b/include/anyhttp/h2v2/session_details.hpp index 905ddd3..f10d5db 100644 --- a/include/anyhttp/h2v2/session_details.hpp +++ b/include/anyhttp/h2v2/session_details.hpp @@ -234,7 +234,8 @@ Task NGHttp2SessionImpl::end_connection() // if (auto ec = io::shutdown(stream_, io::Shutdown::send); !ec) { - if (auto drained = co_await io::drain(stream_, this->get_executor(), std::chrono::seconds(2))) + if (auto drained = co_await io::drain(stream_, this->weak_from_this(), this->get_executor(), + std::chrono::seconds(2))) mlogd("dropped {} bytes the peer sent after the session ended", drained); } else if (ec != errc::not_connected) // the peer may be gone already @@ -388,7 +389,11 @@ IoTask<> ClientSession::shutdown() this->start_write(); Timer timer(this->get_executor()); - timer.arm(std::chrono::seconds(2), [this] { io::cancel(stream_); }); + timer.arm(std::chrono::seconds(2), [weak = this->weak_from_this()] { + // may come late, see Timer + if (auto self = std::static_pointer_cast(weak.lock())) + io::cancel(self->stream_); + }); co_await this->session_done_.wait(); timer.cancel(); diff --git a/include/anyhttp/net.hpp b/include/anyhttp/net.hpp index b6ed347..9204975 100644 --- a/include/anyhttp/net.hpp +++ b/include/anyhttp/net.hpp @@ -43,6 +43,7 @@ #include #include #include +#include #include #include @@ -70,13 +71,20 @@ namespace anyhttp::inline asio::io * * The data is of no interest, not even decrypted: this reads the socket itself, by hand, as that * is what every stream type has in common. + * + * \p owner is what keeps \p stream alive: a timeout that expires as the peer ends its side may + * still cancel the stream after this has returned (see Timer), and does nothing once \p owner is + * gone. */ template -Task drain(Stream& stream, const Executor& executor, +Task drain(Stream& stream, std::weak_ptr owner, const Executor& executor, std::chrono::steady_clock::duration timeout) { Timer timer(executor); - timer.arm(timeout, [&stream] { io::cancel(stream); }); + timer.arm(timeout, [&stream, weak = std::move(owner)] { + if (auto self = weak.lock()) + io::cancel(stream); + }); auto& socket = get_socket(stream); std::array buffer; diff --git a/src/h1/session.cpp b/src/h1/session.cpp index c56b673..fae8854 100644 --- a/src/h1/session.cpp +++ b/src/h1/session.cpp @@ -866,7 +866,7 @@ Task BeastSession::end_connection(bool notify) // if (auto ec = io::shutdown(stream_, io::Shutdown::send); !ec) { - if (auto drained = co_await io::drain(stream_, get_executor(), 2s)) + if (auto drained = co_await io::drain(stream_, this->weak_from_this(), get_executor(), 2s)) mlogd("dropped {} bytes the peer sent after the last message", drained); } else if (ec != errc::not_connected) // the peer may be gone already diff --git a/test/test_corosio.cpp b/test/test_corosio.cpp index 98662d2..370bc40 100644 --- a/test/test_corosio.cpp +++ b/test/test_corosio.cpp @@ -422,21 +422,6 @@ TEST(CorosioRuntime, WHEN_event_is_set_before_and_while_waiting_THEN_waits_end) EXPECT_EQ(woken, 2); } -TEST(CorosioRuntime, WHEN_timer_is_rearmed_or_cancelled_THEN_only_the_last_callback_runs) -{ - boost::corosio::io_context context; - Timer timer(context.get_executor()); - std::vector fired; - - timer.arm(std::chrono::milliseconds(1), [&] { fired.push_back(1); }); - timer.arm(std::chrono::milliseconds(2), [&] { fired.push_back(2); }); - Timer other(context.get_executor()); - other.arm(std::chrono::milliseconds(1), [&] { fired.push_back(3); }); - other.cancel(); - context.run(); - EXPECT_THAT(fired, ElementsAre(2)); -} - TEST(CorosioRuntime, WHEN_both_tasks_run_THEN_when_both_waits_for_both) { boost::corosio::io_context context; diff --git a/test/test_runtime.cpp b/test/test_runtime.cpp index 3be038f..b6b6584 100644 --- a/test/test_runtime.cpp +++ b/test/test_runtime.cpp @@ -22,6 +22,7 @@ #include #include #include +#include using namespace anyhttp; using namespace std::chrono_literals; @@ -165,3 +166,20 @@ TEST(Runtime, WHEN_partial_or_total_cancellation_alone_is_requested_THEN_delay_r EXPECT_FALSE(*result) << result->message(); } #endif + +// ================================================================================================= + +TEST(Runtime, WHEN_timer_is_rearmed_or_cancelled_THEN_only_the_last_callback_runs) +{ + IoContext context; + Timer timer(context.get_executor()); + std::vector fired; + + timer.arm(1ms, [&] { fired.push_back(1); }); + timer.arm(2ms, [&] { fired.push_back(2); }); + Timer other(context.get_executor()); + other.arm(1ms, [&] { fired.push_back(3); }); + other.cancel(); + context.run(); + EXPECT_THAT(fired, ElementsAre(2)); +}