mirror of https://github.com/grpc/grpc.git
[chttp2] Continue refactoring towards promises (#34437)
Isolate ping callback tracking to its own file. Also takes the opportunity to simplify keepalive code by applying the ping timeout to all pings. Adds an experiment to allow multiple pings outstanding too (this was originally an accidental behavior change of the work, but one that I think may be useful going forward). --------- Co-authored-by: ctiller <ctiller@users.noreply.github.com>pull/34564/head
Craig Tiller
1 year ago
committed by
GitHub
parent
e76ac4ab07
commit
a17f08b49d
34 changed files with 1103 additions and 240 deletions
@ -0,0 +1,99 @@ |
|||||||
|
// Copyright 2023 gRPC authors.
|
||||||
|
//
|
||||||
|
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||||
|
// you may not use this file except in compliance with the License.
|
||||||
|
// You may obtain a copy of the License at
|
||||||
|
//
|
||||||
|
// http://www.apache.org/licenses/LICENSE-2.0
|
||||||
|
//
|
||||||
|
// Unless required by applicable law or agreed to in writing, software
|
||||||
|
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||||
|
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||||
|
// See the License for the specific language governing permissions and
|
||||||
|
// limitations under the License.
|
||||||
|
|
||||||
|
#include <grpc/support/port_platform.h> |
||||||
|
|
||||||
|
#include "src/core/ext/transport/chttp2/transport/ping_callbacks.h" |
||||||
|
|
||||||
|
#include <utility> |
||||||
|
|
||||||
|
#include "absl/meta/type_traits.h" |
||||||
|
#include "absl/random/distributions.h" |
||||||
|
|
||||||
|
namespace grpc_core { |
||||||
|
|
||||||
|
void Chttp2PingCallbacks::OnPing(Callback on_start, Callback on_ack) { |
||||||
|
on_start_.emplace_back(std::move(on_start)); |
||||||
|
on_ack_.emplace_back(std::move(on_ack)); |
||||||
|
ping_requested_ = true; |
||||||
|
} |
||||||
|
|
||||||
|
void Chttp2PingCallbacks::OnPingAck(Callback on_ack) { |
||||||
|
auto it = inflight_.find(most_recent_inflight_); |
||||||
|
if (it != inflight_.end()) { |
||||||
|
it->second.on_ack.emplace_back(std::move(on_ack)); |
||||||
|
return; |
||||||
|
} |
||||||
|
ping_requested_ = true; |
||||||
|
on_ack_.emplace_back(std::move(on_ack)); |
||||||
|
} |
||||||
|
|
||||||
|
uint64_t Chttp2PingCallbacks::StartPing( |
||||||
|
absl::BitGenRef bitgen, Duration ping_timeout, Callback on_timeout, |
||||||
|
grpc_event_engine::experimental::EventEngine* event_engine) { |
||||||
|
uint64_t id; |
||||||
|
do { |
||||||
|
id = absl::Uniform<uint64_t>(bitgen); |
||||||
|
} while (inflight_.contains(id)); |
||||||
|
CallbackVec cbs = std::move(on_start_); |
||||||
|
CallbackVec().swap(on_start_); |
||||||
|
InflightPing inflight; |
||||||
|
inflight.on_ack.swap(on_ack_); |
||||||
|
if (ping_timeout != Duration::Infinity()) { |
||||||
|
inflight.on_timeout = |
||||||
|
event_engine->RunAfter(ping_timeout, std::move(on_timeout)); |
||||||
|
} else { |
||||||
|
inflight.on_timeout = |
||||||
|
grpc_event_engine::experimental::EventEngine::TaskHandle::kInvalid; |
||||||
|
} |
||||||
|
inflight_.emplace(id, std::move(inflight)); |
||||||
|
most_recent_inflight_ = id; |
||||||
|
ping_requested_ = false; |
||||||
|
for (auto& cb : cbs) { |
||||||
|
cb(); |
||||||
|
} |
||||||
|
return id; |
||||||
|
} |
||||||
|
|
||||||
|
bool Chttp2PingCallbacks::AckPing( |
||||||
|
uint64_t id, grpc_event_engine::experimental::EventEngine* event_engine) { |
||||||
|
auto ping = inflight_.extract(id); |
||||||
|
if (ping.empty()) return false; |
||||||
|
if (ping.mapped().on_timeout != |
||||||
|
grpc_event_engine::experimental::EventEngine::TaskHandle::kInvalid) { |
||||||
|
event_engine->Cancel(ping.mapped().on_timeout); |
||||||
|
} |
||||||
|
for (auto& cb : ping.mapped().on_ack) { |
||||||
|
cb(); |
||||||
|
} |
||||||
|
return true; |
||||||
|
} |
||||||
|
|
||||||
|
void Chttp2PingCallbacks::CancelAll( |
||||||
|
grpc_event_engine::experimental::EventEngine* event_engine) { |
||||||
|
CallbackVec().swap(on_start_); |
||||||
|
CallbackVec().swap(on_ack_); |
||||||
|
for (auto& cbs : inflight_) { |
||||||
|
CallbackVec().swap(cbs.second.on_ack); |
||||||
|
if (cbs.second.on_timeout != |
||||||
|
grpc_event_engine::experimental::EventEngine::TaskHandle::kInvalid) { |
||||||
|
event_engine->Cancel(std::exchange( |
||||||
|
cbs.second.on_timeout, |
||||||
|
grpc_event_engine::experimental::EventEngine::TaskHandle::kInvalid)); |
||||||
|
} |
||||||
|
} |
||||||
|
ping_requested_ = false; |
||||||
|
} |
||||||
|
|
||||||
|
} // namespace grpc_core
|
||||||
@ -0,0 +1,71 @@ |
|||||||
|
// Copyright 2023 gRPC authors.
|
||||||
|
//
|
||||||
|
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||||
|
// you may not use this file except in compliance with the License.
|
||||||
|
// You may obtain a copy of the License at
|
||||||
|
//
|
||||||
|
// http://www.apache.org/licenses/LICENSE-2.0
|
||||||
|
//
|
||||||
|
// Unless required by applicable law or agreed to in writing, software
|
||||||
|
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||||
|
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||||
|
// See the License for the specific language governing permissions and
|
||||||
|
// limitations under the License.
|
||||||
|
|
||||||
|
#ifndef GRPC_SRC_CORE_EXT_TRANSPORT_CHTTP2_TRANSPORT_PING_CALLBACKS_H |
||||||
|
#define GRPC_SRC_CORE_EXT_TRANSPORT_CHTTP2_TRANSPORT_PING_CALLBACKS_H |
||||||
|
|
||||||
|
#include <grpc/support/port_platform.h> |
||||||
|
|
||||||
|
#include <stddef.h> |
||||||
|
#include <stdint.h> |
||||||
|
|
||||||
|
#include <algorithm> |
||||||
|
#include <vector> |
||||||
|
|
||||||
|
#include "absl/container/flat_hash_map.h" |
||||||
|
#include "absl/functional/any_invocable.h" |
||||||
|
#include "absl/hash/hash.h" |
||||||
|
#include "absl/random/bit_gen_ref.h" |
||||||
|
|
||||||
|
#include <grpc/event_engine/event_engine.h> |
||||||
|
|
||||||
|
#include "src/core/lib/gprpp/time.h" |
||||||
|
|
||||||
|
namespace grpc_core { |
||||||
|
|
||||||
|
class Chttp2PingCallbacks { |
||||||
|
public: |
||||||
|
using Callback = absl::AnyInvocable<void()>; |
||||||
|
|
||||||
|
void RequestPing() { ping_requested_ = true; } |
||||||
|
void OnPing(Callback on_start, Callback on_ack); |
||||||
|
void OnPingAck(Callback on_ack); |
||||||
|
|
||||||
|
GRPC_MUST_USE_RESULT uint64_t |
||||||
|
StartPing(absl::BitGenRef bitgen, Duration ping_timeout, Callback on_timeout, |
||||||
|
grpc_event_engine::experimental::EventEngine* event_engine); |
||||||
|
bool AckPing(uint64_t id, |
||||||
|
grpc_event_engine::experimental::EventEngine* event_engine); |
||||||
|
|
||||||
|
void CancelAll(grpc_event_engine::experimental::EventEngine* event_engine); |
||||||
|
|
||||||
|
bool ping_requested() const { return ping_requested_; } |
||||||
|
size_t pings_inflight() const { return inflight_.size(); } |
||||||
|
|
||||||
|
private: |
||||||
|
using CallbackVec = std::vector<Callback>; |
||||||
|
struct InflightPing { |
||||||
|
grpc_event_engine::experimental::EventEngine::TaskHandle on_timeout; |
||||||
|
CallbackVec on_ack; |
||||||
|
}; |
||||||
|
absl::flat_hash_map<uint64_t, InflightPing> inflight_; |
||||||
|
uint64_t most_recent_inflight_ = 0; |
||||||
|
bool ping_requested_ = false; |
||||||
|
CallbackVec on_start_; |
||||||
|
CallbackVec on_ack_; |
||||||
|
}; |
||||||
|
|
||||||
|
} // namespace grpc_core
|
||||||
|
|
||||||
|
#endif // GRPC_SRC_CORE_EXT_TRANSPORT_CHTTP2_TRANSPORT_PING_CALLBACKS_H
|
||||||
@ -0,0 +1,568 @@ |
|||||||
|
// Copyright 2023 gRPC authors.
|
||||||
|
//
|
||||||
|
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||||
|
// you may not use this file except in compliance with the License.
|
||||||
|
// You may obtain a copy of the License at
|
||||||
|
//
|
||||||
|
// http://www.apache.org/licenses/LICENSE-2.0
|
||||||
|
//
|
||||||
|
// Unless required by applicable law or agreed to in writing, software
|
||||||
|
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||||
|
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||||
|
// See the License for the specific language governing permissions and
|
||||||
|
// limitations under the License.
|
||||||
|
|
||||||
|
#include "src/core/ext/transport/chttp2/transport/ping_callbacks.h" |
||||||
|
|
||||||
|
#include <chrono> |
||||||
|
|
||||||
|
#include "absl/random/random.h" |
||||||
|
#include "gmock/gmock.h" |
||||||
|
#include "gtest/gtest.h" |
||||||
|
|
||||||
|
#include "src/core/lib/gprpp/crash.h" |
||||||
|
#include "test/core/event_engine/mock_event_engine.h" |
||||||
|
|
||||||
|
using grpc_event_engine::experimental::EventEngine; |
||||||
|
using grpc_event_engine::experimental::MockEventEngine; |
||||||
|
using testing::_; |
||||||
|
using testing::Matcher; |
||||||
|
using testing::Return; |
||||||
|
using testing::StrictMock; |
||||||
|
|
||||||
|
namespace grpc_core { |
||||||
|
namespace { |
||||||
|
|
||||||
|
TEST(PingCallbacksTest, RequestPingRequestsPing) { |
||||||
|
Chttp2PingCallbacks callbacks; |
||||||
|
EXPECT_FALSE(callbacks.ping_requested()); |
||||||
|
callbacks.RequestPing(); |
||||||
|
EXPECT_TRUE(callbacks.ping_requested()); |
||||||
|
} |
||||||
|
|
||||||
|
TEST(PingCallbacksTest, OnPingRequestsPing) { |
||||||
|
Chttp2PingCallbacks callbacks; |
||||||
|
EXPECT_FALSE(callbacks.ping_requested()); |
||||||
|
callbacks.OnPing([] {}, [] {}); |
||||||
|
EXPECT_TRUE(callbacks.ping_requested()); |
||||||
|
} |
||||||
|
|
||||||
|
TEST(PingCallbacksTest, OnPingAckRequestsPing) { |
||||||
|
Chttp2PingCallbacks callbacks; |
||||||
|
EXPECT_FALSE(callbacks.ping_requested()); |
||||||
|
callbacks.OnPingAck([] {}); |
||||||
|
EXPECT_TRUE(callbacks.ping_requested()); |
||||||
|
} |
||||||
|
|
||||||
|
TEST(PingCallbacksTest, PingRoundtrips) { |
||||||
|
StrictMock<MockEventEngine> event_engine; |
||||||
|
absl::BitGen bitgen; |
||||||
|
Chttp2PingCallbacks callbacks; |
||||||
|
bool started = false; |
||||||
|
bool acked = false; |
||||||
|
EXPECT_FALSE(callbacks.ping_requested()); |
||||||
|
// Request ping
|
||||||
|
callbacks.OnPing( |
||||||
|
[&started] { |
||||||
|
EXPECT_FALSE(started); |
||||||
|
started = true; |
||||||
|
}, |
||||||
|
[&acked] { |
||||||
|
EXPECT_FALSE(acked); |
||||||
|
acked = true; |
||||||
|
}); |
||||||
|
EXPECT_TRUE(callbacks.ping_requested()); |
||||||
|
EXPECT_EQ(callbacks.pings_inflight(), 0); |
||||||
|
EXPECT_FALSE(started); |
||||||
|
EXPECT_FALSE(acked); |
||||||
|
// Start ping should call the start methods, set a timeout, and clear the
|
||||||
|
// request
|
||||||
|
EXPECT_CALL(event_engine, RunAfter(EventEngine::Duration(Duration::Hours(24)), |
||||||
|
Matcher<absl::AnyInvocable<void()>>(_))) |
||||||
|
.WillOnce([]() { |
||||||
|
return EventEngine::TaskHandle{123, 456}; |
||||||
|
}); |
||||||
|
auto id = callbacks.StartPing( |
||||||
|
bitgen, Duration::Hours(24), [] { Crash("should not reach here"); }, |
||||||
|
&event_engine); |
||||||
|
EXPECT_FALSE(callbacks.ping_requested()); |
||||||
|
EXPECT_EQ(callbacks.pings_inflight(), 1); |
||||||
|
EXPECT_TRUE(started); |
||||||
|
EXPECT_FALSE(acked); |
||||||
|
// Ack should cancel the timeout
|
||||||
|
EXPECT_CALL(event_engine, Cancel(EventEngine::TaskHandle{123, 456})) |
||||||
|
.WillOnce(Return(true)); |
||||||
|
EXPECT_TRUE(callbacks.AckPing(id, &event_engine)); |
||||||
|
EXPECT_EQ(callbacks.pings_inflight(), 0); |
||||||
|
EXPECT_FALSE(callbacks.ping_requested()); |
||||||
|
EXPECT_TRUE(started); |
||||||
|
EXPECT_TRUE(acked); |
||||||
|
} |
||||||
|
|
||||||
|
TEST(PingCallbacksTest, PingRoundtripsWithInfiniteTimeout) { |
||||||
|
StrictMock<MockEventEngine> event_engine; |
||||||
|
absl::BitGen bitgen; |
||||||
|
Chttp2PingCallbacks callbacks; |
||||||
|
bool started = false; |
||||||
|
bool acked = false; |
||||||
|
EXPECT_FALSE(callbacks.ping_requested()); |
||||||
|
// Request ping
|
||||||
|
callbacks.OnPing( |
||||||
|
[&started] { |
||||||
|
EXPECT_FALSE(started); |
||||||
|
started = true; |
||||||
|
}, |
||||||
|
[&acked] { |
||||||
|
EXPECT_FALSE(acked); |
||||||
|
acked = true; |
||||||
|
}); |
||||||
|
EXPECT_TRUE(callbacks.ping_requested()); |
||||||
|
EXPECT_EQ(callbacks.pings_inflight(), 0); |
||||||
|
EXPECT_FALSE(started); |
||||||
|
EXPECT_FALSE(acked); |
||||||
|
auto id = callbacks.StartPing( |
||||||
|
bitgen, Duration::Infinity(), [] { Crash("should not reach here"); }, |
||||||
|
&event_engine); |
||||||
|
EXPECT_FALSE(callbacks.ping_requested()); |
||||||
|
EXPECT_EQ(callbacks.pings_inflight(), 1); |
||||||
|
EXPECT_TRUE(started); |
||||||
|
EXPECT_FALSE(acked); |
||||||
|
EXPECT_TRUE(callbacks.AckPing(id, &event_engine)); |
||||||
|
EXPECT_EQ(callbacks.pings_inflight(), 0); |
||||||
|
EXPECT_FALSE(callbacks.ping_requested()); |
||||||
|
EXPECT_TRUE(started); |
||||||
|
EXPECT_TRUE(acked); |
||||||
|
} |
||||||
|
|
||||||
|
TEST(PingCallbacksTest, InvalidPingIdFlagsError) { |
||||||
|
StrictMock<MockEventEngine> event_engine; |
||||||
|
Chttp2PingCallbacks callbacks; |
||||||
|
EXPECT_FALSE(callbacks.AckPing(1234, &event_engine)); |
||||||
|
} |
||||||
|
|
||||||
|
TEST(PingCallbacksTest, DuplicatePingIdFlagsError) { |
||||||
|
StrictMock<MockEventEngine> event_engine; |
||||||
|
absl::BitGen bitgen; |
||||||
|
Chttp2PingCallbacks callbacks; |
||||||
|
bool started = false; |
||||||
|
bool acked = false; |
||||||
|
EXPECT_FALSE(callbacks.ping_requested()); |
||||||
|
callbacks.OnPing( |
||||||
|
[&started] { |
||||||
|
EXPECT_FALSE(started); |
||||||
|
started = true; |
||||||
|
}, |
||||||
|
[&acked] { |
||||||
|
EXPECT_FALSE(acked); |
||||||
|
acked = true; |
||||||
|
}); |
||||||
|
EXPECT_TRUE(callbacks.ping_requested()); |
||||||
|
EXPECT_FALSE(started); |
||||||
|
EXPECT_FALSE(acked); |
||||||
|
EXPECT_CALL(event_engine, RunAfter(EventEngine::Duration(Duration::Hours(24)), |
||||||
|
Matcher<absl::AnyInvocable<void()>>(_))) |
||||||
|
.WillOnce([]() { |
||||||
|
return EventEngine::TaskHandle{123, 456}; |
||||||
|
}); |
||||||
|
auto id = callbacks.StartPing( |
||||||
|
bitgen, Duration::Hours(24), [] { Crash("should not reach here"); }, |
||||||
|
&event_engine); |
||||||
|
EXPECT_FALSE(callbacks.ping_requested()); |
||||||
|
EXPECT_TRUE(started); |
||||||
|
EXPECT_FALSE(acked); |
||||||
|
EXPECT_CALL(event_engine, Cancel(EventEngine::TaskHandle{123, 456})) |
||||||
|
.WillOnce(Return(true)); |
||||||
|
EXPECT_TRUE(callbacks.AckPing(id, &event_engine)); |
||||||
|
EXPECT_FALSE(callbacks.ping_requested()); |
||||||
|
EXPECT_TRUE(started); |
||||||
|
EXPECT_TRUE(acked); |
||||||
|
// Second ping ack on the same id should fail
|
||||||
|
EXPECT_FALSE(callbacks.AckPing(id, &event_engine)); |
||||||
|
} |
||||||
|
|
||||||
|
TEST(PingCallbacksTest, OnPingAckCanPiggybackInflightPings) { |
||||||
|
StrictMock<MockEventEngine> event_engine; |
||||||
|
absl::BitGen bitgen; |
||||||
|
Chttp2PingCallbacks callbacks; |
||||||
|
bool started = false; |
||||||
|
bool acked_first = false; |
||||||
|
bool acked_second = false; |
||||||
|
EXPECT_FALSE(callbacks.ping_requested()); |
||||||
|
callbacks.OnPing( |
||||||
|
[&started] { |
||||||
|
EXPECT_FALSE(started); |
||||||
|
started = true; |
||||||
|
}, |
||||||
|
[&acked_first] { |
||||||
|
EXPECT_FALSE(acked_first); |
||||||
|
acked_first = true; |
||||||
|
}); |
||||||
|
EXPECT_TRUE(callbacks.ping_requested()); |
||||||
|
EXPECT_FALSE(started); |
||||||
|
EXPECT_FALSE(acked_first); |
||||||
|
EXPECT_FALSE(acked_second); |
||||||
|
EXPECT_CALL(event_engine, RunAfter(EventEngine::Duration(Duration::Hours(24)), |
||||||
|
Matcher<absl::AnyInvocable<void()>>(_))) |
||||||
|
.WillOnce([]() { |
||||||
|
return EventEngine::TaskHandle{123, 456}; |
||||||
|
}); |
||||||
|
auto id = callbacks.StartPing( |
||||||
|
bitgen, Duration::Hours(24), [] { Crash("should not reach here"); }, |
||||||
|
&event_engine); |
||||||
|
EXPECT_FALSE(callbacks.ping_requested()); |
||||||
|
EXPECT_TRUE(started); |
||||||
|
EXPECT_FALSE(acked_first); |
||||||
|
EXPECT_FALSE(acked_second); |
||||||
|
callbacks.OnPingAck([&acked_second] { |
||||||
|
EXPECT_FALSE(acked_second); |
||||||
|
acked_second = true; |
||||||
|
}); |
||||||
|
EXPECT_FALSE(callbacks.ping_requested()); |
||||||
|
EXPECT_TRUE(started); |
||||||
|
EXPECT_FALSE(acked_first); |
||||||
|
EXPECT_FALSE(acked_second); |
||||||
|
EXPECT_CALL(event_engine, Cancel(EventEngine::TaskHandle{123, 456})) |
||||||
|
.WillOnce(Return(true)); |
||||||
|
EXPECT_TRUE(callbacks.AckPing(id, &event_engine)); |
||||||
|
EXPECT_FALSE(callbacks.ping_requested()); |
||||||
|
EXPECT_TRUE(started); |
||||||
|
EXPECT_TRUE(acked_first); |
||||||
|
EXPECT_TRUE(acked_second); |
||||||
|
} |
||||||
|
|
||||||
|
TEST(PingCallbacksTest, PingAckRoundtrips) { |
||||||
|
StrictMock<MockEventEngine> event_engine; |
||||||
|
absl::BitGen bitgen; |
||||||
|
Chttp2PingCallbacks callbacks; |
||||||
|
bool acked = false; |
||||||
|
EXPECT_FALSE(callbacks.ping_requested()); |
||||||
|
callbacks.OnPingAck([&acked] { |
||||||
|
EXPECT_FALSE(acked); |
||||||
|
acked = true; |
||||||
|
}); |
||||||
|
EXPECT_TRUE(callbacks.ping_requested()); |
||||||
|
EXPECT_FALSE(acked); |
||||||
|
EXPECT_CALL(event_engine, RunAfter(EventEngine::Duration(Duration::Hours(24)), |
||||||
|
Matcher<absl::AnyInvocable<void()>>(_))) |
||||||
|
.WillOnce([]() { |
||||||
|
return EventEngine::TaskHandle{123, 456}; |
||||||
|
}); |
||||||
|
auto id = callbacks.StartPing( |
||||||
|
bitgen, Duration::Hours(24), [] { Crash("should not reach here"); }, |
||||||
|
&event_engine); |
||||||
|
EXPECT_FALSE(callbacks.ping_requested()); |
||||||
|
EXPECT_FALSE(acked); |
||||||
|
EXPECT_CALL(event_engine, Cancel(EventEngine::TaskHandle{123, 456})) |
||||||
|
.WillOnce(Return(true)); |
||||||
|
EXPECT_TRUE(callbacks.AckPing(id, &event_engine)); |
||||||
|
EXPECT_FALSE(callbacks.ping_requested()); |
||||||
|
EXPECT_TRUE(acked); |
||||||
|
} |
||||||
|
|
||||||
|
TEST(PingCallbacksTest, MultiPingRoundtrips) { |
||||||
|
StrictMock<MockEventEngine> event_engine; |
||||||
|
absl::BitGen bitgen; |
||||||
|
Chttp2PingCallbacks callbacks; |
||||||
|
bool started1 = false; |
||||||
|
bool acked1 = false; |
||||||
|
bool started2 = false; |
||||||
|
bool acked2 = false; |
||||||
|
EXPECT_FALSE(callbacks.ping_requested()); |
||||||
|
callbacks.OnPing( |
||||||
|
[&started1] { |
||||||
|
EXPECT_FALSE(started1); |
||||||
|
started1 = true; |
||||||
|
}, |
||||||
|
[&acked1] { |
||||||
|
EXPECT_FALSE(acked1); |
||||||
|
acked1 = true; |
||||||
|
}); |
||||||
|
EXPECT_TRUE(callbacks.ping_requested()); |
||||||
|
EXPECT_FALSE(started1); |
||||||
|
EXPECT_FALSE(acked1); |
||||||
|
EXPECT_FALSE(started2); |
||||||
|
EXPECT_FALSE(acked2); |
||||||
|
EXPECT_CALL(event_engine, RunAfter(EventEngine::Duration(Duration::Hours(24)), |
||||||
|
Matcher<absl::AnyInvocable<void()>>(_))) |
||||||
|
.WillOnce([]() { |
||||||
|
return EventEngine::TaskHandle{123, 456}; |
||||||
|
}); |
||||||
|
auto id1 = callbacks.StartPing( |
||||||
|
bitgen, Duration::Hours(24), [] { Crash("should not reach here"); }, |
||||||
|
&event_engine); |
||||||
|
EXPECT_FALSE(callbacks.ping_requested()); |
||||||
|
EXPECT_TRUE(started1); |
||||||
|
EXPECT_FALSE(acked1); |
||||||
|
EXPECT_FALSE(started2); |
||||||
|
EXPECT_FALSE(acked2); |
||||||
|
callbacks.OnPing( |
||||||
|
[&started2] { |
||||||
|
EXPECT_FALSE(started2); |
||||||
|
started2 = true; |
||||||
|
}, |
||||||
|
[&acked2] { |
||||||
|
EXPECT_FALSE(acked2); |
||||||
|
acked2 = true; |
||||||
|
}); |
||||||
|
EXPECT_TRUE(callbacks.ping_requested()); |
||||||
|
EXPECT_TRUE(started1); |
||||||
|
EXPECT_FALSE(acked1); |
||||||
|
EXPECT_FALSE(started2); |
||||||
|
EXPECT_FALSE(acked2); |
||||||
|
EXPECT_CALL(event_engine, RunAfter(EventEngine::Duration(Duration::Hours(24)), |
||||||
|
Matcher<absl::AnyInvocable<void()>>(_))) |
||||||
|
.WillOnce([]() { |
||||||
|
return EventEngine::TaskHandle{123, 789}; |
||||||
|
}); |
||||||
|
auto id2 = callbacks.StartPing( |
||||||
|
bitgen, Duration::Hours(24), [] { Crash("should not reach here"); }, |
||||||
|
&event_engine); |
||||||
|
EXPECT_NE(id1, id2); |
||||||
|
EXPECT_TRUE(started1); |
||||||
|
EXPECT_FALSE(acked1); |
||||||
|
EXPECT_TRUE(started2); |
||||||
|
EXPECT_FALSE(acked2); |
||||||
|
EXPECT_CALL(event_engine, Cancel(EventEngine::TaskHandle{123, 456})) |
||||||
|
.WillOnce(Return(true)); |
||||||
|
EXPECT_TRUE(callbacks.AckPing(id1, &event_engine)); |
||||||
|
EXPECT_FALSE(callbacks.ping_requested()); |
||||||
|
EXPECT_TRUE(started1); |
||||||
|
EXPECT_TRUE(acked1); |
||||||
|
EXPECT_TRUE(started2); |
||||||
|
EXPECT_FALSE(acked2); |
||||||
|
EXPECT_CALL(event_engine, Cancel(EventEngine::TaskHandle{123, 789})) |
||||||
|
.WillOnce(Return(true)); |
||||||
|
EXPECT_TRUE(callbacks.AckPing(id2, &event_engine)); |
||||||
|
EXPECT_FALSE(callbacks.ping_requested()); |
||||||
|
EXPECT_TRUE(started1); |
||||||
|
EXPECT_TRUE(acked1); |
||||||
|
EXPECT_TRUE(started2); |
||||||
|
EXPECT_TRUE(acked2); |
||||||
|
} |
||||||
|
|
||||||
|
TEST(PingCallbacksTest, MultiPingRoundtripsWithOutOfOrderAcks) { |
||||||
|
StrictMock<MockEventEngine> event_engine; |
||||||
|
absl::BitGen bitgen; |
||||||
|
Chttp2PingCallbacks callbacks; |
||||||
|
bool started1 = false; |
||||||
|
bool acked1 = false; |
||||||
|
bool started2 = false; |
||||||
|
bool acked2 = false; |
||||||
|
EXPECT_FALSE(callbacks.ping_requested()); |
||||||
|
callbacks.OnPing( |
||||||
|
[&started1] { |
||||||
|
EXPECT_FALSE(started1); |
||||||
|
started1 = true; |
||||||
|
}, |
||||||
|
[&acked1] { |
||||||
|
EXPECT_FALSE(acked1); |
||||||
|
acked1 = true; |
||||||
|
}); |
||||||
|
EXPECT_TRUE(callbacks.ping_requested()); |
||||||
|
EXPECT_FALSE(started1); |
||||||
|
EXPECT_FALSE(acked1); |
||||||
|
EXPECT_FALSE(started2); |
||||||
|
EXPECT_FALSE(acked2); |
||||||
|
EXPECT_CALL(event_engine, RunAfter(EventEngine::Duration(Duration::Hours(24)), |
||||||
|
Matcher<absl::AnyInvocable<void()>>(_))) |
||||||
|
.WillOnce([]() { |
||||||
|
return EventEngine::TaskHandle{123, 456}; |
||||||
|
}); |
||||||
|
auto id1 = callbacks.StartPing( |
||||||
|
bitgen, Duration::Hours(24), [] { Crash("should not reach here"); }, |
||||||
|
&event_engine); |
||||||
|
EXPECT_FALSE(callbacks.ping_requested()); |
||||||
|
EXPECT_TRUE(started1); |
||||||
|
EXPECT_FALSE(acked1); |
||||||
|
EXPECT_FALSE(started2); |
||||||
|
EXPECT_FALSE(acked2); |
||||||
|
callbacks.OnPing( |
||||||
|
[&started2] { |
||||||
|
EXPECT_FALSE(started2); |
||||||
|
started2 = true; |
||||||
|
}, |
||||||
|
[&acked2] { |
||||||
|
EXPECT_FALSE(acked2); |
||||||
|
acked2 = true; |
||||||
|
}); |
||||||
|
EXPECT_TRUE(callbacks.ping_requested()); |
||||||
|
EXPECT_TRUE(started1); |
||||||
|
EXPECT_FALSE(acked1); |
||||||
|
EXPECT_FALSE(started2); |
||||||
|
EXPECT_FALSE(acked2); |
||||||
|
EXPECT_CALL(event_engine, RunAfter(EventEngine::Duration(Duration::Hours(24)), |
||||||
|
Matcher<absl::AnyInvocable<void()>>(_))) |
||||||
|
.WillOnce([]() { |
||||||
|
return EventEngine::TaskHandle{123, 789}; |
||||||
|
}); |
||||||
|
auto id2 = callbacks.StartPing( |
||||||
|
bitgen, Duration::Hours(24), [] { Crash("should not reach here"); }, |
||||||
|
&event_engine); |
||||||
|
EXPECT_NE(id1, id2); |
||||||
|
EXPECT_TRUE(started1); |
||||||
|
EXPECT_FALSE(acked1); |
||||||
|
EXPECT_TRUE(started2); |
||||||
|
EXPECT_FALSE(acked2); |
||||||
|
EXPECT_CALL(event_engine, Cancel(EventEngine::TaskHandle{123, 789})) |
||||||
|
.WillOnce(Return(true)); |
||||||
|
EXPECT_TRUE(callbacks.AckPing(id2, &event_engine)); |
||||||
|
EXPECT_FALSE(callbacks.ping_requested()); |
||||||
|
EXPECT_TRUE(started1); |
||||||
|
EXPECT_FALSE(acked1); |
||||||
|
EXPECT_TRUE(started2); |
||||||
|
EXPECT_TRUE(acked2); |
||||||
|
EXPECT_CALL(event_engine, Cancel(EventEngine::TaskHandle{123, 456})) |
||||||
|
.WillOnce(Return(true)); |
||||||
|
EXPECT_TRUE(callbacks.AckPing(id1, &event_engine)); |
||||||
|
EXPECT_FALSE(callbacks.ping_requested()); |
||||||
|
EXPECT_TRUE(started1); |
||||||
|
EXPECT_TRUE(acked1); |
||||||
|
EXPECT_TRUE(started2); |
||||||
|
EXPECT_TRUE(acked2); |
||||||
|
} |
||||||
|
|
||||||
|
TEST(PingCallbacksTest, CoalescedPingsRoundtrip) { |
||||||
|
StrictMock<MockEventEngine> event_engine; |
||||||
|
absl::BitGen bitgen; |
||||||
|
Chttp2PingCallbacks callbacks; |
||||||
|
bool started1 = false; |
||||||
|
bool acked1 = false; |
||||||
|
bool started2 = false; |
||||||
|
bool acked2 = false; |
||||||
|
EXPECT_FALSE(callbacks.ping_requested()); |
||||||
|
callbacks.OnPing( |
||||||
|
[&started1] { |
||||||
|
EXPECT_FALSE(started1); |
||||||
|
started1 = true; |
||||||
|
}, |
||||||
|
[&acked1] { |
||||||
|
EXPECT_FALSE(acked1); |
||||||
|
acked1 = true; |
||||||
|
}); |
||||||
|
callbacks.OnPing( |
||||||
|
[&started2] { |
||||||
|
EXPECT_FALSE(started2); |
||||||
|
started2 = true; |
||||||
|
}, |
||||||
|
[&acked2] { |
||||||
|
EXPECT_FALSE(acked2); |
||||||
|
acked2 = true; |
||||||
|
}); |
||||||
|
EXPECT_TRUE(callbacks.ping_requested()); |
||||||
|
EXPECT_FALSE(started1); |
||||||
|
EXPECT_FALSE(acked1); |
||||||
|
EXPECT_FALSE(started2); |
||||||
|
EXPECT_FALSE(acked2); |
||||||
|
EXPECT_CALL(event_engine, RunAfter(EventEngine::Duration(Duration::Hours(24)), |
||||||
|
Matcher<absl::AnyInvocable<void()>>(_))) |
||||||
|
.WillOnce([]() { |
||||||
|
return EventEngine::TaskHandle{123, 456}; |
||||||
|
}); |
||||||
|
auto id = callbacks.StartPing( |
||||||
|
bitgen, Duration::Hours(24), [] { Crash("should not reach here"); }, |
||||||
|
&event_engine); |
||||||
|
EXPECT_FALSE(callbacks.ping_requested()); |
||||||
|
EXPECT_TRUE(started1); |
||||||
|
EXPECT_FALSE(acked1); |
||||||
|
EXPECT_TRUE(started2); |
||||||
|
EXPECT_FALSE(acked2); |
||||||
|
EXPECT_CALL(event_engine, Cancel(EventEngine::TaskHandle{123, 456})) |
||||||
|
.WillOnce(Return(true)); |
||||||
|
EXPECT_TRUE(callbacks.AckPing(id, &event_engine)); |
||||||
|
EXPECT_FALSE(callbacks.ping_requested()); |
||||||
|
EXPECT_TRUE(started1); |
||||||
|
EXPECT_TRUE(acked1); |
||||||
|
EXPECT_TRUE(started2); |
||||||
|
EXPECT_TRUE(acked2); |
||||||
|
} |
||||||
|
|
||||||
|
TEST(PingCallbacksTest, CancelAllCancelsCallbacks) { |
||||||
|
StrictMock<MockEventEngine> event_engine; |
||||||
|
absl::BitGen bitgen; |
||||||
|
Chttp2PingCallbacks callbacks; |
||||||
|
bool started = false; |
||||||
|
bool acked = false; |
||||||
|
EXPECT_FALSE(callbacks.ping_requested()); |
||||||
|
callbacks.OnPing( |
||||||
|
[&started] { |
||||||
|
EXPECT_FALSE(started); |
||||||
|
started = true; |
||||||
|
}, |
||||||
|
[&acked] { |
||||||
|
EXPECT_FALSE(acked); |
||||||
|
acked = true; |
||||||
|
}); |
||||||
|
EXPECT_TRUE(callbacks.ping_requested()); |
||||||
|
callbacks.CancelAll(&event_engine); |
||||||
|
EXPECT_FALSE(started); |
||||||
|
EXPECT_FALSE(acked); |
||||||
|
EXPECT_FALSE(callbacks.ping_requested()); |
||||||
|
// Can still send a ping, no callback should be invoked
|
||||||
|
EXPECT_CALL(event_engine, RunAfter(EventEngine::Duration(Duration::Hours(24)), |
||||||
|
Matcher<absl::AnyInvocable<void()>>(_))) |
||||||
|
.WillOnce([]() { |
||||||
|
return EventEngine::TaskHandle{123, 456}; |
||||||
|
}); |
||||||
|
auto id = callbacks.StartPing( |
||||||
|
bitgen, Duration::Hours(24), [] { Crash("should not reach here"); }, |
||||||
|
&event_engine); |
||||||
|
EXPECT_FALSE(started); |
||||||
|
EXPECT_FALSE(acked); |
||||||
|
EXPECT_CALL(event_engine, Cancel(EventEngine::TaskHandle{123, 456})) |
||||||
|
.WillOnce(Return(true)); |
||||||
|
EXPECT_TRUE(callbacks.AckPing(id, &event_engine)); |
||||||
|
EXPECT_FALSE(started); |
||||||
|
EXPECT_FALSE(acked); |
||||||
|
EXPECT_FALSE(callbacks.ping_requested()); |
||||||
|
} |
||||||
|
|
||||||
|
TEST(PingCallbacksTest, CancelAllCancelsInflightPings) { |
||||||
|
StrictMock<MockEventEngine> event_engine; |
||||||
|
absl::BitGen bitgen; |
||||||
|
Chttp2PingCallbacks callbacks; |
||||||
|
bool started = false; |
||||||
|
bool acked = false; |
||||||
|
EXPECT_FALSE(callbacks.ping_requested()); |
||||||
|
callbacks.OnPing( |
||||||
|
[&started] { |
||||||
|
EXPECT_FALSE(started); |
||||||
|
started = true; |
||||||
|
}, |
||||||
|
[&acked] { |
||||||
|
EXPECT_FALSE(acked); |
||||||
|
acked = true; |
||||||
|
}); |
||||||
|
EXPECT_TRUE(callbacks.ping_requested()); |
||||||
|
EXPECT_FALSE(started); |
||||||
|
EXPECT_FALSE(acked); |
||||||
|
EXPECT_CALL(event_engine, RunAfter(EventEngine::Duration(Duration::Hours(24)), |
||||||
|
Matcher<absl::AnyInvocable<void()>>(_))) |
||||||
|
.WillOnce([]() { |
||||||
|
return EventEngine::TaskHandle{123, 456}; |
||||||
|
}); |
||||||
|
auto id = callbacks.StartPing( |
||||||
|
bitgen, Duration::Hours(24), [] { Crash("should not reach here"); }, |
||||||
|
&event_engine); |
||||||
|
EXPECT_FALSE(callbacks.ping_requested()); |
||||||
|
EXPECT_TRUE(started); |
||||||
|
EXPECT_FALSE(acked); |
||||||
|
EXPECT_CALL(event_engine, Cancel(EventEngine::TaskHandle{123, 456})) |
||||||
|
.WillOnce(Return(true)); |
||||||
|
callbacks.CancelAll(&event_engine); |
||||||
|
// Ensure Cancel call comes from CancelAll
|
||||||
|
testing::Mock::VerifyAndClearExpectations(&event_engine); |
||||||
|
EXPECT_FALSE(acked); |
||||||
|
EXPECT_FALSE(callbacks.ping_requested()); |
||||||
|
// Ping should still be valid, but no callback should be invoked
|
||||||
|
EXPECT_TRUE(callbacks.AckPing(id, &event_engine)); |
||||||
|
EXPECT_FALSE(acked); |
||||||
|
EXPECT_FALSE(callbacks.ping_requested()); |
||||||
|
} |
||||||
|
|
||||||
|
} // namespace
|
||||||
|
} // namespace grpc_core
|
||||||
|
|
||||||
|
int main(int argc, char** argv) { |
||||||
|
::testing::InitGoogleTest(&argc, argv); |
||||||
|
return RUN_ALL_TESTS(); |
||||||
|
} |
||||||
Loading…
Reference in new issue