Enable custom WINDOW_UPDATE notifying criteria in WindowManager. This CL introduces ShouldNotifyListener(limit, window, delta) to WindowManager and adds a WindowManager optional constructor parameter ShouldNotifyListener to override WindowManager's default WINDOW_UPDATE strategy. The oghttp2 library will make use of this capability by providing a ShouldNotifyListener aligned with the nghttp2 WINDOW_UPDATE strategy. PiperOrigin-RevId: 432234434
diff --git a/http2/adapter/oghttp2_session.cc b/http2/adapter/oghttp2_session.cc index 8b87090..e1be0d8 100644 --- a/http2/adapter/oghttp2_session.cc +++ b/http2/adapter/oghttp2_session.cc
@@ -334,6 +334,7 @@ [this](size_t window_update_delta) { SendWindowUpdate(kConnectionStreamId, window_update_delta); }, + /*should_notify_listener=*/{}, /*update_window_on_notify=*/false), options_(options) { decoder_.set_visitor(&receive_logger_);
diff --git a/http2/adapter/oghttp2_session.h b/http2/adapter/oghttp2_session.h index a714825..63d54bc 100644 --- a/http2/adapter/oghttp2_session.h +++ b/http2/adapter/oghttp2_session.h
@@ -215,6 +215,7 @@ StreamState(int32_t stream_receive_window, int32_t stream_send_window, WindowManager::WindowUpdateListener listener) : window_manager(stream_receive_window, std::move(listener), + /*should_notify_listener=*/{}, /*update_window_on_notify=*/false), send_window(stream_send_window) {}
diff --git a/http2/adapter/window_manager.cc b/http2/adapter/window_manager.cc index 3dd9baf..c93356a 100644 --- a/http2/adapter/window_manager.cc +++ b/http2/adapter/window_manager.cc
@@ -8,14 +8,37 @@ namespace http2 { namespace adapter { +bool DefaultShouldNotifyListener(int64_t limit, int64_t window, int64_t delta) { + // For the sake of efficiency, we want to send window updates if less than + // half of the max quota is available to the peer at any point in time. + const int64_t kDesiredMinWindow = limit / 2; + const int64_t kDesiredMinDelta = limit / 3; + if (delta >= kDesiredMinDelta) { + // This particular window update was sent because the available delta + // exceeded the desired minimum. + return true; + } else if (window < kDesiredMinWindow) { + // This particular window update was sent because the quota available to the + // peer at this moment is less than the desired minimum. + return true; + } + return false; +} + WindowManager::WindowManager(int64_t window_size_limit, WindowUpdateListener listener, + ShouldNotifyListener should_notify_listener, bool update_window_on_notify) : limit_(window_size_limit), window_(window_size_limit), buffered_(0), listener_(std::move(listener)), - update_window_on_notify_(update_window_on_notify) {} + should_notify_listener_(std::move(should_notify_listener)), + update_window_on_notify_(update_window_on_notify) { + if (!should_notify_listener_) { + should_notify_listener_ = DefaultShouldNotifyListener; + } +} void WindowManager::OnWindowSizeLimitChange(const int64_t new_limit) { QUICHE_VLOG(2) << "WindowManager@" << this @@ -65,22 +88,8 @@ } void WindowManager::MaybeNotifyListener() { - // For the sake of efficiency, we want to send window updates if less than - // half of the max quota is available to the peer at any point in time. - const int64_t kDesiredMinWindow = limit_ / 2; - const int64_t kDesiredMinDelta = limit_ / 3; const int64_t delta = limit_ - (buffered_ + window_); - bool send_update = false; - if (delta >= kDesiredMinDelta) { - // This particular window update was sent because the available delta - // exceeded the desired minimum. - send_update = true; - } else if (window_ < kDesiredMinWindow) { - // This particular window update was sent because the quota available to the - // peer at this moment is less than the desired minimum. - send_update = true; - } - if (send_update && delta > 0) { + if (should_notify_listener_(limit_, window_, delta) && delta > 0) { QUICHE_VLOG(2) << "WindowManager@" << this << " Informing listener of delta: " << delta; listener_(delta);
diff --git a/http2/adapter/window_manager.h b/http2/adapter/window_manager.h index 45a6ddd..373e602 100644 --- a/http2/adapter/window_manager.h +++ b/http2/adapter/window_manager.h
@@ -19,9 +19,15 @@ class QUICHE_EXPORT_PRIVATE WindowManager { public: // A WindowUpdateListener is invoked when it is time to send a window update. - typedef std::function<void(int64_t)> WindowUpdateListener; + using WindowUpdateListener = std::function<void(int64_t)>; + + // Invoked to determine whether to call the listener based on the window + // limit, window size, and delta that would be sent. + using ShouldNotifyListener = + std::function<bool(int64_t limit, int64_t size, int64_t delta)>; WindowManager(int64_t window_size_limit, WindowUpdateListener listener, + ShouldNotifyListener should_notify_listener = {}, bool update_window_on_notify = true); int64_t CurrentWindowSize() const { return window_; } @@ -75,6 +81,8 @@ WindowUpdateListener listener_; + ShouldNotifyListener should_notify_listener_; + bool update_window_on_notify_; };
diff --git a/http2/adapter/window_manager_test.cc b/http2/adapter/window_manager_test.cc index c5a9ffb..27b5e21 100644 --- a/http2/adapter/window_manager_test.cc +++ b/http2/adapter/window_manager_test.cc
@@ -218,11 +218,13 @@ WindowManager wm1( kDefaultLimit, [&call_sequence1](int64_t delta) { call_sequence1.push_back(delta); }, + /*should_notify_listener=*/{}, /*update_window_on_notify=*/true); // default std::list<int64_t> call_sequence2; WindowManager wm2( kDefaultLimit, [&call_sequence2](int64_t delta) { call_sequence2.push_back(delta); }, + /*should_notify_listener=*/{}, /*update_window_on_notify=*/false); const int64_t consumed = kDefaultLimit / 3 - 1; @@ -263,6 +265,71 @@ EXPECT_EQ(wm2.CurrentWindowSize(), kDefaultLimit); } +// This test verifies that when the constructor option is specified, +// WindowManager uses the provided ShouldNotifyListener to determine when to +// notify the listener. +TEST(WindowManagerShouldUpdateTest, CustomShouldNotifyListener) { + const int64_t kDefaultLimit = 65535; + + // This window manager should always notify. + std::list<int64_t> call_sequence1; + WindowManager wm1( + kDefaultLimit, + [&call_sequence1](int64_t delta) { call_sequence1.push_back(delta); }, + [](int64_t /*limit*/, int64_t /*window*/, int64_t /*delta*/) { + return true; + }); + // This window manager should never notify. + std::list<int64_t> call_sequence2; + WindowManager wm2( + kDefaultLimit, + [&call_sequence2](int64_t delta) { call_sequence2.push_back(delta); }, + [](int64_t /*limit*/, int64_t /*window*/, int64_t /*delta*/) { + return false; + }); + // This window manager should notify as long as no data is buffered. + std::list<int64_t> call_sequence3; + WindowManager wm3( + kDefaultLimit, + [&call_sequence3](int64_t delta) { call_sequence3.push_back(delta); }, + [](int64_t limit, int64_t window, int64_t delta) { + return delta == limit - window; + }); + + const int64_t consumed = kDefaultLimit / 4; + + wm1.MarkWindowConsumed(consumed); + EXPECT_THAT(call_sequence1, testing::ElementsAre(consumed)); + wm2.MarkWindowConsumed(consumed); + EXPECT_TRUE(call_sequence2.empty()); + wm3.MarkWindowConsumed(consumed); + EXPECT_THAT(call_sequence3, testing::ElementsAre(consumed)); + + const int64_t buffered = 42; + + wm1.MarkDataBuffered(buffered); + EXPECT_THAT(call_sequence1, testing::ElementsAre(consumed)); + wm2.MarkDataBuffered(buffered); + EXPECT_TRUE(call_sequence2.empty()); + wm3.MarkDataBuffered(buffered); + EXPECT_THAT(call_sequence3, testing::ElementsAre(consumed)); + + wm1.MarkDataFlushed(buffered / 3); + EXPECT_THAT(call_sequence1, testing::ElementsAre(consumed, buffered / 3)); + wm2.MarkDataFlushed(buffered / 3); + EXPECT_TRUE(call_sequence2.empty()); + wm3.MarkDataFlushed(buffered / 3); + EXPECT_THAT(call_sequence3, testing::ElementsAre(consumed)); + + wm1.MarkDataFlushed(2 * buffered / 3); + EXPECT_THAT(call_sequence1, + testing::ElementsAre(consumed, buffered / 3, 2 * buffered / 3)); + wm2.MarkDataFlushed(2 * buffered / 3); + EXPECT_TRUE(call_sequence2.empty()); + wm3.MarkDataFlushed(2 * buffered / 3); + EXPECT_THAT(call_sequence3, testing::ElementsAre(consumed, buffered)); +} + } // namespace } // namespace test } // namespace adapter