Chiebot-Mirror
/
grpc
mirror of https://github.com/grpc/grpc.git
8
0
Fork 0
Code Issues Projects Releases Wiki Activity

[EventEngine] Revert "Revert "EventEngine::RunAt: C++ Alarm (#30024)" (#30147)" (#32461)

Browse Source 
This reverts commit 0fc0384b5a.

Major changes: this code calls `GetDefaultEventEngine` once on Alarm
init instead of 7 times throughout.

I will run benchmarks to ensure b/237283941 is not reproduced.


<!--

If you know who should review your pull request, please assign it to
that
person, otherwise the pull request would get assigned randomly.

If your pull request is for a specific language, please add the
appropriate
lang label.

-->

---------

Co-authored-by: drfloob <drfloob@users.noreply.github.com>
pull/32465/head
AJ Heller 2 years ago committed by GitHub
parent 8fdc82f33d
commit 070ce3630a
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 79 additions and 48 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 1
      BUILD
  2. 1
      include/grpcpp/alarm.h
  3. 125
      src/cpp/common/alarm.cc

1
BUILD
Unescape View File

@ -1919,6 +1919,7 @@ grpc_cc_library(
"//src/core:channel_args",
"//src/core:channel_init",
"//src/core:closure",
"//src/core:default_event_engine",
"//src/core:error",
"//src/core:gpr_atm",
"//src/core:gpr_manual_constructor",

1
include/grpcpp/alarm.h
Unescape View File

@ -23,6 +23,7 @@
#include <functional>
#include <grpc/event_engine/event_engine.h>
#include <grpc/grpc.h>
#include <grpcpp/completion_queue.h>
#include <grpcpp/impl/completion_queue_tag.h>

125
src/cpp/common/alarm.cc
Unescape View File

@ -18,8 +18,14 @@
#include <grpc/support/port_platform.h>
#include <functional>
#include <memory>
#include <utility>
#include "absl/base/thread_annotations.h"
#include "absl/status/status.h"
#include "absl/types/optional.h"
#include <grpc/event_engine/event_engine.h>
#include <grpc/grpc.h>
#include <grpc/support/log.h>
#include <grpc/support/sync.h>
@ -28,22 +34,28 @@
#include <grpcpp/completion_queue.h>
#include <grpcpp/impl/completion_queue_tag.h>
#include "src/core/lib/event_engine/default_event_engine.h"
#include "src/core/lib/gprpp/sync.h"
#include "src/core/lib/gprpp/time.h"
#include "src/core/lib/iomgr/closure.h"
#include "src/core/lib/iomgr/error.h"
#include "src/core/lib/iomgr/exec_ctx.h"
#include "src/core/lib/iomgr/executor.h"
#include "src/core/lib/iomgr/timer.h"
#include "src/core/lib/surface/completion_queue.h"
namespace grpc {
namespace internal {
namespace {
using ::grpc_event_engine::experimental::EventEngine;
} // namespace
class AlarmImpl : public grpc::internal::CompletionQueueTag {
public:
AlarmImpl() : cq_(nullptr), tag_(nullptr) {
AlarmImpl()
: event_engine_(grpc_event_engine::experimental::GetDefaultEventEngine()),
cq_(nullptr),
tag_(nullptr) {
gpr_ref_init(&refs_, 1);
grpc_timer_init_unset(&timer_);
}
~AlarmImpl() override {}
bool FinalizeResult(void** tag, bool* /*status*/) override {
@ -52,61 +64,46 @@ class AlarmImpl : public grpc::internal::CompletionQueueTag {
return true;
}
void Set(grpc::CompletionQueue* cq, gpr_timespec deadline, void* tag) {
grpc_core::ApplicationCallbackExecCtx callback_exec_ctx;
grpc_core::MutexLock lock(&mu_);
GPR_ASSERT(!cq_timer_handle_.has_value() &&
!callback_timer_handle_.has_value());
grpc_core::ExecCtx exec_ctx;
GRPC_CQ_INTERNAL_REF(cq->cq(), "alarm");
cq_ = cq->cq();
tag_ = tag;
GPR_ASSERT(grpc_cq_begin_op(cq_, this));
GRPC_CLOSURE_INIT(
&on_alarm_,
[](void* arg, grpc_error_handle error) {
// queue the op on the completion queue
AlarmImpl* alarm = static_cast<AlarmImpl*>(arg);
alarm->Ref();
// Preserve the cq and reset the cq_ so that the alarm
// can be reset when the alarm tag is delivered.
grpc_completion_queue* cq = alarm->cq_;
alarm->cq_ = nullptr;
grpc_cq_end_op(
cq, alarm, error,
[](void* /*arg*/, grpc_cq_completion* /*completion*/) {}, arg,
&alarm->completion_);
GRPC_CQ_INTERNAL_UNREF(cq, "alarm");
},
this, grpc_schedule_on_exec_ctx);
grpc_timer_init(&timer_,
grpc_core::Timestamp::FromTimespecRoundUp(deadline),
&on_alarm_);
Ref();
cq_timer_handle_ = event_engine_->RunAfter(
grpc_core::Timestamp::FromTimespecRoundUp(deadline) -
grpc_core::ExecCtx::Get()->Now(),
[this] { OnCQAlarm(absl::OkStatus()); });
}
void Set(gpr_timespec deadline, std::function<void(bool)> f) {
grpc_core::ApplicationCallbackExecCtx callback_exec_ctx;
grpc_core::MutexLock lock(&mu_);
GPR_ASSERT(!cq_timer_handle_.has_value() &&
!callback_timer_handle_.has_value());
grpc_core::ExecCtx exec_ctx;
// Don't use any CQ at all. Instead just use the timer to fire the function
callback_ = std::move(f);
Ref();
GRPC_CLOSURE_INIT(
&on_alarm_,
[](void* arg, grpc_error_handle error) {
grpc_core::Executor::Run(GRPC_CLOSURE_CREATE(
[](void* arg, grpc_error_handle error) {
AlarmImpl* alarm =
static_cast<AlarmImpl*>(arg);
alarm->callback_(error.ok());
alarm->Unref();
},
arg, nullptr),
error);
},
this, grpc_schedule_on_exec_ctx);
grpc_timer_init(&timer_,
grpc_core::Timestamp::FromTimespecRoundUp(deadline),
&on_alarm_);
callback_timer_handle_ = event_engine_->RunAfter(
grpc_core::Timestamp::FromTimespecRoundUp(deadline) -
grpc_core::ExecCtx::Get()->Now(),
[this] { OnCallbackAlarm(true); });
}
void Cancel() {
grpc_core::ApplicationCallbackExecCtx callback_exec_ctx;
grpc_core::ExecCtx exec_ctx;
grpc_timer_cancel(&timer_);
grpc_core::MutexLock lock(&mu_);
if (callback_timer_handle_.has_value() &&
event_engine_->Cancel(*callback_timer_handle_)) {
event_engine_->Run([this] { OnCallbackAlarm(/*is_ok=*/false); });
callback_timer_handle_.reset();
} else if (cq_timer_handle_.has_value() &&
event_engine_->Cancel(*cq_timer_handle_)) {
event_engine_->Run(
[this] { OnCQAlarm(absl::CancelledError("cancelled")); });
cq_timer_handle_.reset();
}
}
void Destroy() {
Cancel();
@ -114,6 +111,35 @@ class AlarmImpl : public grpc::internal::CompletionQueueTag {
}
private:
void OnCQAlarm(grpc_error_handle error) {
{
grpc_core::MutexLock lock(&mu_);
cq_timer_handle_.reset();
}
grpc_core::ApplicationCallbackExecCtx callback_exec_ctx;
grpc_core::ExecCtx exec_ctx;
// Preserve the cq and reset the cq_ so that the alarm
// can be reset when the alarm tag is delivered.
grpc_completion_queue* cq = cq_;
cq_ = nullptr;
grpc_cq_end_op(
cq, this, error,
[](void* /*arg*/, grpc_cq_completion* /*completion*/) {}, nullptr,
&completion_);
GRPC_CQ_INTERNAL_UNREF(cq, "alarm");
}
void OnCallbackAlarm(bool is_ok) {
{
grpc_core::MutexLock lock(&mu_);
callback_timer_handle_.reset();
}
grpc_core::ApplicationCallbackExecCtx callback_exec_ctx;
grpc_core::ExecCtx exec_ctx;
callback_(is_ok);
Unref();
}
void Ref() { gpr_ref(&refs_); }
void Unref() {
if (gpr_unref(&refs_)) {
@ -121,9 +147,12 @@ class AlarmImpl : public grpc::internal::CompletionQueueTag {
}
}
grpc_timer timer_;
grpc_core::Mutex mu_;
std::shared_ptr<grpc_event_engine::experimental::EventEngine> event_engine_;
absl::optional<EventEngine::TaskHandle> cq_timer_handle_ ABSL_GUARDED_BY(mu_);
absl::optional<EventEngine::TaskHandle> callback_timer_handle_
ABSL_GUARDED_BY(mu_);
gpr_refcount refs_;
grpc_closure on_alarm_;
grpc_cq_completion completion_;
// completion queue where events about this alarm will be posted
grpc_completion_queue* cq_;

Write Preview
Loading…
Cancel
Save