Promise based sleeps (#28722)
* Promise based sleep * Embrace absl::Status * Automated change: Fix sanity tests * Add another test, fix bug * fix * fix * review feedback Co-authored-by: ctiller <ctiller@users.noreply.github.com>pull/28862/head
Craig Tiller
3 years ago
committed by
GitHub
parent
6565584c7b
commit
df943da2c4
8 changed files with 334 additions and 0 deletions
@ -0,0 +1,74 @@ |
|||||||
|
// Copyright 2022 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/lib/promise/sleep.h" |
||||||
|
|
||||||
|
namespace grpc_core { |
||||||
|
|
||||||
|
Sleep::Sleep(grpc_millis deadline) : state_(new State(deadline)) { |
||||||
|
GRPC_CLOSURE_INIT(&state_->on_timer, &OnTimer, state_, nullptr); |
||||||
|
} |
||||||
|
|
||||||
|
Sleep::~Sleep() { |
||||||
|
if (state_ == nullptr) return; |
||||||
|
{ |
||||||
|
MutexLock lock(&state_->mu); |
||||||
|
switch (state_->stage) { |
||||||
|
case Stage::kInitial: |
||||||
|
state_->Unref(); |
||||||
|
break; |
||||||
|
case Stage::kStarted: |
||||||
|
grpc_timer_cancel(&state_->timer); |
||||||
|
break; |
||||||
|
case Stage::kDone: |
||||||
|
break; |
||||||
|
} |
||||||
|
} |
||||||
|
state_->Unref(); |
||||||
|
} |
||||||
|
|
||||||
|
void Sleep::OnTimer(void* arg, grpc_error_handle) { |
||||||
|
auto* state = static_cast<State*>(arg); |
||||||
|
Waker waker; |
||||||
|
{ |
||||||
|
MutexLock lock(&state->mu); |
||||||
|
state->stage = Stage::kDone; |
||||||
|
waker = std::move(state->waker); |
||||||
|
} |
||||||
|
waker.Wakeup(); |
||||||
|
state->Unref(); |
||||||
|
} |
||||||
|
|
||||||
|
Poll<absl::Status> Sleep::operator()() { |
||||||
|
MutexLock lock(&state_->mu); |
||||||
|
switch (state_->stage) { |
||||||
|
case Stage::kInitial: |
||||||
|
if (state_->deadline <= ExecCtx::Get()->Now()) { |
||||||
|
return absl::OkStatus(); |
||||||
|
} |
||||||
|
state_->stage = Stage::kStarted; |
||||||
|
grpc_timer_init(&state_->timer, state_->deadline, &state_->on_timer); |
||||||
|
break; |
||||||
|
case Stage::kStarted: |
||||||
|
break; |
||||||
|
case Stage::kDone: |
||||||
|
return absl::OkStatus(); |
||||||
|
} |
||||||
|
state_->waker = Activity::current()->MakeNonOwningWaker(); |
||||||
|
return Pending{}; |
||||||
|
} |
||||||
|
|
||||||
|
} // namespace grpc_core
|
||||||
@ -0,0 +1,66 @@ |
|||||||
|
// Copyright 2022 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_CORE_LIB_PROMISE_SLEEP_H |
||||||
|
#define GRPC_CORE_LIB_PROMISE_SLEEP_H |
||||||
|
|
||||||
|
#include <grpc/support/port_platform.h> |
||||||
|
|
||||||
|
#include "src/core/lib/iomgr/timer.h" |
||||||
|
#include "src/core/lib/promise/activity.h" |
||||||
|
#include "src/core/lib/promise/poll.h" |
||||||
|
|
||||||
|
namespace grpc_core { |
||||||
|
|
||||||
|
// Promise that sleeps until a deadline and then finishes.
|
||||||
|
class Sleep { |
||||||
|
public: |
||||||
|
explicit Sleep(grpc_millis deadline); |
||||||
|
~Sleep(); |
||||||
|
|
||||||
|
Sleep(const Sleep&) = delete; |
||||||
|
Sleep& operator=(const Sleep&) = delete; |
||||||
|
Sleep(Sleep&& other) noexcept : state_(other.state_) { |
||||||
|
other.state_ = nullptr; |
||||||
|
} |
||||||
|
Sleep& operator=(Sleep&& other) noexcept { |
||||||
|
std::swap(state_, other.state_); |
||||||
|
return *this; |
||||||
|
} |
||||||
|
|
||||||
|
Poll<absl::Status> operator()(); |
||||||
|
|
||||||
|
private: |
||||||
|
static void OnTimer(void* arg, grpc_error_handle error); |
||||||
|
|
||||||
|
enum class Stage { kInitial, kStarted, kDone }; |
||||||
|
struct State { |
||||||
|
explicit State(grpc_millis deadline) : deadline(deadline) {} |
||||||
|
RefCount refs{2}; |
||||||
|
const grpc_millis deadline; |
||||||
|
grpc_timer timer; |
||||||
|
grpc_closure on_timer; |
||||||
|
Mutex mu; |
||||||
|
Stage stage ABSL_GUARDED_BY(mu) = Stage::kInitial; |
||||||
|
Waker waker ABSL_GUARDED_BY(mu); |
||||||
|
void Unref() { |
||||||
|
if (refs.Unref()) delete this; |
||||||
|
} |
||||||
|
}; |
||||||
|
State* state_; |
||||||
|
}; |
||||||
|
|
||||||
|
} // namespace grpc_core
|
||||||
|
|
||||||
|
#endif // GRPC_CORE_LIB_PROMISE_SLEEP_H
|
||||||
@ -0,0 +1,86 @@ |
|||||||
|
// Copyright 2022 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/lib/promise/sleep.h" |
||||||
|
|
||||||
|
#include <atomic> |
||||||
|
|
||||||
|
#include <gmock/gmock.h> |
||||||
|
#include <gtest/gtest.h> |
||||||
|
|
||||||
|
#include "absl/synchronization/notification.h" |
||||||
|
|
||||||
|
#include <grpc/grpc.h> |
||||||
|
|
||||||
|
#include "src/core/lib/promise/race.h" |
||||||
|
#include "src/core/lib/promise/seq.h" |
||||||
|
#include "test/core/promise/test_wakeup_schedulers.h" |
||||||
|
|
||||||
|
namespace grpc_core { |
||||||
|
namespace { |
||||||
|
|
||||||
|
TEST(Sleep, Zzzz) { |
||||||
|
ExecCtx exec_ctx; |
||||||
|
absl::Notification done; |
||||||
|
grpc_millis done_time = ExecCtx::Get()->Now() + 1000; |
||||||
|
// Sleep for one second then set done to true.
|
||||||
|
auto activity = MakeActivity(Sleep(done_time), InlineWakeupScheduler(), |
||||||
|
[&done](absl::Status r) { |
||||||
|
EXPECT_EQ(r, absl::OkStatus()); |
||||||
|
done.Notify(); |
||||||
|
}); |
||||||
|
done.WaitForNotification(); |
||||||
|
exec_ctx.InvalidateNow(); |
||||||
|
EXPECT_GE(ExecCtx::Get()->Now(), done_time); |
||||||
|
} |
||||||
|
|
||||||
|
TEST(Sleep, AlreadyDone) { |
||||||
|
ExecCtx exec_ctx; |
||||||
|
absl::Notification done; |
||||||
|
grpc_millis done_time = ExecCtx::Get()->Now() - 1000; |
||||||
|
// Sleep for no time at all then set done to true.
|
||||||
|
auto activity = MakeActivity(Sleep(done_time), InlineWakeupScheduler(), |
||||||
|
[&done](absl::Status r) { |
||||||
|
EXPECT_EQ(r, absl::OkStatus()); |
||||||
|
done.Notify(); |
||||||
|
}); |
||||||
|
done.WaitForNotification(); |
||||||
|
} |
||||||
|
|
||||||
|
TEST(Sleep, Cancel) { |
||||||
|
ExecCtx exec_ctx; |
||||||
|
absl::Notification done; |
||||||
|
grpc_millis done_time = ExecCtx::Get()->Now() + 1000; |
||||||
|
// Sleep for one second but race it to complete immediately
|
||||||
|
auto activity = MakeActivity( |
||||||
|
Race(Sleep(done_time), [] { return absl::CancelledError(); }), |
||||||
|
InlineWakeupScheduler(), [&done](absl::Status r) { |
||||||
|
EXPECT_EQ(r, absl::CancelledError()); |
||||||
|
done.Notify(); |
||||||
|
}); |
||||||
|
done.WaitForNotification(); |
||||||
|
exec_ctx.InvalidateNow(); |
||||||
|
EXPECT_LT(ExecCtx::Get()->Now(), done_time); |
||||||
|
} |
||||||
|
|
||||||
|
} // namespace
|
||||||
|
} // namespace grpc_core
|
||||||
|
|
||||||
|
int main(int argc, char** argv) { |
||||||
|
::testing::InitGoogleTest(&argc, argv); |
||||||
|
grpc_init(); |
||||||
|
int r = RUN_ALL_TESTS(); |
||||||
|
grpc_shutdown(); |
||||||
|
return r; |
||||||
|
} |
||||||
Loading…
Reference in new issue