1 | /* |
2 | * Copyright 2018 gRPC authors. |
3 | * |
4 | * Licensed under the Apache License, Version 2.0 (the "License"); |
5 | * you may not use this file except in compliance with the License. |
6 | * You may obtain a copy of the License at |
7 | * |
8 | * http://www.apache.org/licenses/LICENSE-2.0 |
9 | * |
10 | * Unless required by applicable law or agreed to in writing, software |
11 | * distributed under the License is distributed on an "AS IS" BASIS, |
12 | * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
13 | * See the License for the specific language governing permissions and |
14 | * limitations under the License. |
15 | * |
16 | */ |
17 | |
18 | #include <grpcpp/alarm.h> |
19 | |
20 | #include <memory> |
21 | |
22 | #include <grpc/support/log.h> |
23 | #include <grpc/support/port_platform.h> |
24 | #include <grpcpp/completion_queue.h> |
25 | #include <grpcpp/impl/grpc_library.h> |
26 | #include <grpcpp/support/time.h> |
27 | #include "src/core/lib/iomgr/exec_ctx.h" |
28 | #include "src/core/lib/iomgr/timer.h" |
29 | #include "src/core/lib/surface/completion_queue.h" |
30 | |
31 | #include <grpc/support/log.h> |
32 | #include "src/core/lib/debug/trace.h" |
33 | |
34 | namespace grpc_impl { |
35 | |
36 | namespace internal { |
37 | class AlarmImpl : public ::grpc::internal::CompletionQueueTag { |
38 | public: |
39 | AlarmImpl() : cq_(nullptr), tag_(nullptr) { |
40 | gpr_ref_init(&refs_, 1); |
41 | grpc_timer_init_unset(&timer_); |
42 | } |
43 | ~AlarmImpl() {} |
44 | bool FinalizeResult(void** tag, bool* /*status*/) override { |
45 | *tag = tag_; |
46 | Unref(); |
47 | return true; |
48 | } |
49 | void Set(::grpc::CompletionQueue* cq, gpr_timespec deadline, void* tag) { |
50 | grpc_core::ApplicationCallbackExecCtx callback_exec_ctx; |
51 | grpc_core::ExecCtx exec_ctx; |
52 | GRPC_CQ_INTERNAL_REF(cq->cq(), "alarm" ); |
53 | cq_ = cq->cq(); |
54 | tag_ = tag; |
55 | GPR_ASSERT(grpc_cq_begin_op(cq_, this)); |
56 | GRPC_CLOSURE_INIT( |
57 | &on_alarm_, |
58 | [](void* arg, grpc_error* error) { |
59 | // queue the op on the completion queue |
60 | AlarmImpl* alarm = static_cast<AlarmImpl*>(arg); |
61 | alarm->Ref(); |
62 | // Preserve the cq and reset the cq_ so that the alarm |
63 | // can be reset when the alarm tag is delivered. |
64 | grpc_completion_queue* cq = alarm->cq_; |
65 | alarm->cq_ = nullptr; |
66 | grpc_cq_end_op( |
67 | cq, alarm, error, |
68 | [](void* /*arg*/, grpc_cq_completion* /*completion*/) {}, arg, |
69 | &alarm->completion_); |
70 | GRPC_CQ_INTERNAL_UNREF(cq, "alarm" ); |
71 | }, |
72 | this, grpc_schedule_on_exec_ctx); |
73 | grpc_timer_init(&timer_, grpc_timespec_to_millis_round_up(deadline), |
74 | &on_alarm_); |
75 | } |
76 | void Set(gpr_timespec deadline, std::function<void(bool)> f) { |
77 | grpc_core::ApplicationCallbackExecCtx callback_exec_ctx; |
78 | grpc_core::ExecCtx exec_ctx; |
79 | // Don't use any CQ at all. Instead just use the timer to fire the function |
80 | callback_ = std::move(f); |
81 | Ref(); |
82 | GRPC_CLOSURE_INIT(&on_alarm_, |
83 | [](void* arg, grpc_error* error) { |
84 | AlarmImpl* alarm = static_cast<AlarmImpl*>(arg); |
85 | alarm->callback_(error == GRPC_ERROR_NONE); |
86 | alarm->Unref(); |
87 | }, |
88 | this, grpc_schedule_on_exec_ctx); |
89 | grpc_timer_init(&timer_, grpc_timespec_to_millis_round_up(deadline), |
90 | &on_alarm_); |
91 | } |
92 | void Cancel() { |
93 | grpc_core::ApplicationCallbackExecCtx callback_exec_ctx; |
94 | grpc_core::ExecCtx exec_ctx; |
95 | grpc_timer_cancel(&timer_); |
96 | } |
97 | void Destroy() { |
98 | Cancel(); |
99 | Unref(); |
100 | } |
101 | |
102 | private: |
103 | void Ref() { gpr_ref(&refs_); } |
104 | void Unref() { |
105 | if (gpr_unref(&refs_)) { |
106 | delete this; |
107 | } |
108 | } |
109 | |
110 | grpc_timer timer_; |
111 | gpr_refcount refs_; |
112 | grpc_closure on_alarm_; |
113 | grpc_cq_completion completion_; |
114 | // completion queue where events about this alarm will be posted |
115 | grpc_completion_queue* cq_; |
116 | void* tag_; |
117 | std::function<void(bool)> callback_; |
118 | }; |
119 | } // namespace internal |
120 | |
121 | static ::grpc::internal::GrpcLibraryInitializer g_gli_initializer; |
122 | |
123 | Alarm::Alarm() : alarm_(new internal::AlarmImpl()) { |
124 | g_gli_initializer.summon(); |
125 | } |
126 | |
127 | void Alarm::SetInternal(::grpc::CompletionQueue* cq, gpr_timespec deadline, |
128 | void* tag) { |
129 | // Note that we know that alarm_ is actually an internal::AlarmImpl |
130 | // but we declared it as the base pointer to avoid a forward declaration |
131 | // or exposing core data structures in the C++ public headers. |
132 | // Thus it is safe to use a static_cast to the subclass here, and the |
133 | // C++ style guide allows us to do so in this case |
134 | static_cast<internal::AlarmImpl*>(alarm_)->Set(cq, deadline, tag); |
135 | } |
136 | |
137 | void Alarm::SetInternal(gpr_timespec deadline, std::function<void(bool)> f) { |
138 | // Note that we know that alarm_ is actually an internal::AlarmImpl |
139 | // but we declared it as the base pointer to avoid a forward declaration |
140 | // or exposing core data structures in the C++ public headers. |
141 | // Thus it is safe to use a static_cast to the subclass here, and the |
142 | // C++ style guide allows us to do so in this case |
143 | static_cast<internal::AlarmImpl*>(alarm_)->Set(deadline, std::move(f)); |
144 | } |
145 | |
146 | Alarm::~Alarm() { |
147 | if (alarm_ != nullptr) { |
148 | static_cast<internal::AlarmImpl*>(alarm_)->Destroy(); |
149 | } |
150 | } |
151 | |
152 | void Alarm::Cancel() { static_cast<internal::AlarmImpl*>(alarm_)->Cancel(); } |
153 | } // namespace grpc_impl |
154 | |