• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1 /*
2  *
3  * Copyright 2017 gRPC authors.
4  *
5  * Licensed under the Apache License, Version 2.0 (the "License");
6  * you may not use this file except in compliance with the License.
7  * You may obtain a copy of the License at
8  *
9  *     http://www.apache.org/licenses/LICENSE-2.0
10  *
11  * Unless required by applicable law or agreed to in writing, software
12  * distributed under the License is distributed on an "AS IS" BASIS,
13  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14  * See the License for the specific language governing permissions and
15  * limitations under the License.
16  *
17  */
18 
19 #include <grpc/support/port_platform.h>
20 
21 #include "src/core/lib/iomgr/lockfree_event.h"
22 
23 #include <grpc/support/log.h>
24 
25 #include "src/core/lib/debug/trace.h"
26 #include "src/core/lib/iomgr/exec_ctx.h"
27 
28 extern grpc_core::DebugOnlyTraceFlag grpc_polling_trace;
29 
30 /* 'state' holds the to call when the fd is readable or writable respectively.
31    It can contain one of the following values:
32      kClosureReady     : The fd has an I/O event of interest but there is no
33                          closure yet to execute
34 
35      kClosureNotReady : The fd has no I/O event of interest
36 
37      closure ptr       : The closure to be executed when the fd has an I/O
38                          event of interest
39 
40      shutdown_error | kShutdownBit :
41                         'shutdown_error' field ORed with kShutdownBit.
42                          This indicates that the fd is shutdown. Since all
43                          memory allocations are word-aligned, the lower two
44                          bits of the shutdown_error pointer are always 0. So
45                          it is safe to OR these with kShutdownBit
46 
47    Valid state transitions:
48 
49      <closure ptr> <-----3------ kClosureNotReady -----1------->  kClosureReady
50        |  |                         ^   |    ^                         |  |
51        |  |                         |   |    |                         |  |
52        |  +--------------4----------+   6    +---------2---------------+  |
53        |                                |                                 |
54        |                                v                                 |
55        +-----5------->  [shutdown_error | kShutdownBit] <-------7---------+
56 
57     For 1, 4 : See SetReady() function
58     For 2, 3 : See NotifyOn() function
59     For 5,6,7: See SetShutdown() function */
60 
61 namespace grpc_core {
62 
LockfreeEvent()63 LockfreeEvent::LockfreeEvent() { InitEvent(); }
64 
InitEvent()65 void LockfreeEvent::InitEvent() {
66   /* Perform an atomic store to start the state machine.
67 
68      Note carefully that LockfreeEvent *MAY* be used whilst in a destroyed
69      state, while a file descriptor is on a freelist. In such a state it may
70      be SetReady'd, and so we need to perform an atomic operation here to
71      ensure no races */
72   gpr_atm_no_barrier_store(&state_, kClosureNotReady);
73 }
74 
DestroyEvent()75 void LockfreeEvent::DestroyEvent() {
76   gpr_atm curr;
77   do {
78     curr = gpr_atm_no_barrier_load(&state_);
79     if (curr & kShutdownBit) {
80       GRPC_ERROR_UNREF((grpc_error*)(curr & ~kShutdownBit));
81     } else {
82       GPR_ASSERT(curr == kClosureNotReady || curr == kClosureReady);
83     }
84     /* we CAS in a shutdown, no error value here. If this event is interacted
85        with post-deletion (see the note in the constructor) we want the bit
86        pattern to prevent error retention in a deleted object */
87   } while (!gpr_atm_no_barrier_cas(&state_, curr,
88                                    kShutdownBit /* shutdown, no error */));
89 }
90 
NotifyOn(grpc_closure * closure)91 void LockfreeEvent::NotifyOn(grpc_closure* closure) {
92   while (true) {
93     /* This load needs to be an acquire load because this can be a shutdown
94      * error that we might need to reference. Adding acquire semantics makes
95      * sure that the shutdown error has been initialized properly before us
96      * referencing it. */
97     gpr_atm curr = gpr_atm_acq_load(&state_);
98     if (GRPC_TRACE_FLAG_ENABLED(grpc_polling_trace)) {
99       gpr_log(GPR_DEBUG,
100               "LockfreeEvent::NotifyOn: %p curr=%" PRIxPTR " closure=%p", this,
101               curr, closure);
102     }
103     switch (curr) {
104       case kClosureNotReady: {
105         /* kClosureNotReady -> <closure>.
106 
107            We're guaranteed by API that there's an acquire barrier before here,
108            so there's no need to double-dip and this can be a release-only.
109 
110            The release itself pairs with the acquire half of a set_ready full
111            barrier. */
112         if (gpr_atm_rel_cas(&state_, kClosureNotReady,
113                             reinterpret_cast<gpr_atm>(closure))) {
114           return; /* Successful. Return */
115         }
116 
117         break; /* retry */
118       }
119 
120       case kClosureReady: {
121         /* Change the state to kClosureNotReady. Schedule the closure if
122            successful. If not, the state most likely transitioned to shutdown.
123            We should retry.
124 
125            This can be a no-barrier cas since the state is being transitioned to
126            kClosureNotReady; set_ready and set_shutdown do not schedule any
127            closure when transitioning out of CLOSURE_NO_READY state (i.e there
128            is no other code that needs to 'happen-after' this) */
129         if (gpr_atm_no_barrier_cas(&state_, kClosureReady, kClosureNotReady)) {
130           ExecCtx::Run(DEBUG_LOCATION, closure, GRPC_ERROR_NONE);
131           return; /* Successful. Return */
132         }
133 
134         break; /* retry */
135       }
136 
137       default: {
138         /* 'curr' is either a closure or the fd is shutdown(in which case 'curr'
139            contains a pointer to the shutdown-error). If the fd is shutdown,
140            schedule the closure with the shutdown error */
141         if ((curr & kShutdownBit) > 0) {
142           grpc_error* shutdown_err =
143               reinterpret_cast<grpc_error*>(curr & ~kShutdownBit);
144           ExecCtx::Run(DEBUG_LOCATION, closure,
145                        GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING(
146                            "FD Shutdown", &shutdown_err, 1));
147           return;
148         }
149 
150         /* There is already a closure!. This indicates a bug in the code */
151         gpr_log(GPR_ERROR,
152                 "LockfreeEvent::NotifyOn: notify_on called with a previous "
153                 "callback still pending");
154         abort();
155       }
156     }
157   }
158 
159   GPR_UNREACHABLE_CODE(return );
160 }
161 
SetShutdown(grpc_error * shutdown_error)162 bool LockfreeEvent::SetShutdown(grpc_error* shutdown_error) {
163   gpr_atm new_state = reinterpret_cast<gpr_atm>(shutdown_error) | kShutdownBit;
164 
165   while (true) {
166     gpr_atm curr = gpr_atm_no_barrier_load(&state_);
167     if (GRPC_TRACE_FLAG_ENABLED(grpc_polling_trace)) {
168       gpr_log(GPR_DEBUG,
169               "LockfreeEvent::SetShutdown: %p curr=%" PRIxPTR " err=%s",
170               &state_, curr, grpc_error_string(shutdown_error));
171     }
172     switch (curr) {
173       case kClosureReady:
174       case kClosureNotReady:
175         /* Need a full barrier here so that the initial load in notify_on
176            doesn't need a barrier */
177         if (gpr_atm_full_cas(&state_, curr, new_state)) {
178           return true; /* early out */
179         }
180         break; /* retry */
181 
182       default: {
183         /* 'curr' is either a closure or the fd is already shutdown */
184 
185         /* If fd is already shutdown, we are done */
186         if ((curr & kShutdownBit) > 0) {
187           GRPC_ERROR_UNREF(shutdown_error);
188           return false;
189         }
190 
191         /* Fd is not shutdown. Schedule the closure and move the state to
192            shutdown state.
193            Needs an acquire to pair with setting the closure (and get a
194            happens-after on that edge), and a release to pair with anything
195            loading the shutdown state. */
196         if (gpr_atm_full_cas(&state_, curr, new_state)) {
197           ExecCtx::Run(DEBUG_LOCATION, reinterpret_cast<grpc_closure*>(curr),
198                        GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING(
199                            "FD Shutdown", &shutdown_error, 1));
200           return true;
201         }
202 
203         /* 'curr' was a closure but now changed to a different state. We will
204           have to retry */
205         break;
206       }
207     }
208   }
209 
210   GPR_UNREACHABLE_CODE(return false);
211 }
212 
SetReady()213 void LockfreeEvent::SetReady() {
214   while (true) {
215     gpr_atm curr = gpr_atm_no_barrier_load(&state_);
216 
217     if (GRPC_TRACE_FLAG_ENABLED(grpc_polling_trace)) {
218       gpr_log(GPR_DEBUG, "LockfreeEvent::SetReady: %p curr=%" PRIxPTR, &state_,
219               curr);
220     }
221 
222     switch (curr) {
223       case kClosureReady: {
224         /* Already ready. We are done here */
225         return;
226       }
227 
228       case kClosureNotReady: {
229         /* No barrier required as we're transitioning to a state that does not
230            involve a closure */
231         if (gpr_atm_no_barrier_cas(&state_, kClosureNotReady, kClosureReady)) {
232           return; /* early out */
233         }
234         break; /* retry */
235       }
236 
237       default: {
238         /* 'curr' is either a closure or the fd is shutdown */
239         if ((curr & kShutdownBit) > 0) {
240           /* The fd is shutdown. Do nothing */
241           return;
242         }
243         /* Full cas: acquire pairs with this cas' release in the event of a
244            spurious set_ready; release pairs with this or the acquire in
245            notify_on (or set_shutdown) */
246         else if (gpr_atm_full_cas(&state_, curr, kClosureNotReady)) {
247           ExecCtx::Run(DEBUG_LOCATION, reinterpret_cast<grpc_closure*>(curr),
248                        GRPC_ERROR_NONE);
249           return;
250         }
251         /* else the state changed again (only possible by either a racing
252            set_ready or set_shutdown functions. In both these cases, the closure
253            would have been scheduled for execution. So we are done here */
254         return;
255       }
256     }
257   }
258 }
259 
260 }  // namespace grpc_core
261