blob: ee8bd1fd3940d385ccdd87cbb0694e439d2830f1 [file] [log] [blame]
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -08001/*
2 *
Craig Tiller06059952015-02-18 08:34:56 -08003 * Copyright 2015, Google Inc.
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -08004 * All rights reserved.
5 *
6 * Redistribution and use in source and binary forms, with or without
7 * modification, are permitted provided that the following conditions are
8 * met:
9 *
10 * * Redistributions of source code must retain the above copyright
11 * notice, this list of conditions and the following disclaimer.
12 * * Redistributions in binary form must reproduce the above
13 * copyright notice, this list of conditions and the following disclaimer
14 * in the documentation and/or other materials provided with the
15 * distribution.
16 * * Neither the name of Google Inc. nor the names of its
17 * contributors may be used to endorse or promote products derived from
18 * this software without specific prior written permission.
19 *
20 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
21 * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
22 * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
23 * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
24 * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
25 * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
26 * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
27 * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
28 * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
29 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
30 * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
31 *
32 */
33
34#include "src/core/surface/completion_queue.h"
35
36#include <stdio.h>
37#include <string.h>
38
ctiller58393c22015-01-07 14:03:30 -080039#include "src/core/iomgr/pollset.h"
Craig Tiller485d7762015-01-23 12:54:05 -080040#include "src/core/support/string.h"
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -080041#include "src/core/surface/call.h"
42#include "src/core/surface/event_string.h"
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -080043#include "src/core/surface/surface_trace.h"
44#include <grpc/support/alloc.h>
45#include <grpc/support/atm.h>
46#include <grpc/support/log.h>
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -080047
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -080048/* Completion queue structure */
49struct grpc_completion_queue {
Craig Tiller97fc6a32015-07-08 15:31:35 -070050 /** completed events */
51 grpc_cq_completion completed_head;
52 grpc_cq_completion *completed_tail;
53 /** Number of pending events (+1 if we're not shutdown) */
54 gpr_refcount pending_events;
55 /** Once owning_refs drops to zero, we will destroy the cq */
Craig Tiller5717a982015-04-27 12:01:49 -070056 gpr_refcount owning_refs;
Craig Tiller97fc6a32015-07-08 15:31:35 -070057 /** the set of low level i/o things that concern this cq */
ctillerd79b4862014-12-17 16:36:59 -080058 grpc_pollset pollset;
Craig Tiller97fc6a32015-07-08 15:31:35 -070059 /** 0 initially, 1 once we've begun shutting down */
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -080060 int shutdown;
Craig Tillerf5fd4ba2015-03-02 18:01:21 +000061 int shutdown_called;
Craig Tillerb56975c2015-06-15 10:11:16 -070062 int is_server_cq;
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -080063};
64
Craig Tiller32946d32015-01-15 11:37:30 -080065grpc_completion_queue *grpc_completion_queue_create(void) {
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -080066 grpc_completion_queue *cc = gpr_malloc(sizeof(grpc_completion_queue));
67 memset(cc, 0, sizeof(*cc));
68 /* Initial ref is dropped by grpc_completion_queue_shutdown */
Craig Tiller97fc6a32015-07-08 15:31:35 -070069 gpr_ref_init(&cc->pending_events, 1);
Craig Tiller70730b42015-05-22 14:42:38 -070070 /* One for destroy(), one for pollset_shutdown */
71 gpr_ref_init(&cc->owning_refs, 2);
ctillerd79b4862014-12-17 16:36:59 -080072 grpc_pollset_init(&cc->pollset);
Craig Tiller97fc6a32015-07-08 15:31:35 -070073 cc->completed_tail = &cc->completed_head;
Craig Tiller12cf5372015-07-09 13:48:11 -070074 cc->completed_head.next = (gpr_uintptr)cc->completed_tail;
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -080075 return cc;
76}
77
Craig Tiller463f2372015-05-28 16:16:15 -070078#ifdef GRPC_CQ_REF_COUNT_DEBUG
Nicolas "Pixel" Nobleaaf39ab2015-06-24 23:22:53 +020079void grpc_cq_internal_ref(grpc_completion_queue *cc, const char *reason,
80 const char *file, int line) {
Craig Tiller12cf5372015-07-09 13:48:11 -070081 gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, "CQ:%p ref %d -> %d %s", cc,
82 (int)cc->owning_refs.count, (int)cc->owning_refs.count + 1, reason);
Craig Tiller463f2372015-05-28 16:16:15 -070083#else
Craig Tiller5717a982015-04-27 12:01:49 -070084void grpc_cq_internal_ref(grpc_completion_queue *cc) {
Craig Tiller463f2372015-05-28 16:16:15 -070085#endif
Craig Tiller5717a982015-04-27 12:01:49 -070086 gpr_ref(&cc->owning_refs);
87}
88
89static void on_pollset_destroy_done(void *arg) {
90 grpc_completion_queue *cc = arg;
Craig Tiller463f2372015-05-28 16:16:15 -070091 GRPC_CQ_INTERNAL_UNREF(cc, "pollset_destroy");
Craig Tiller5717a982015-04-27 12:01:49 -070092}
93
Craig Tiller463f2372015-05-28 16:16:15 -070094#ifdef GRPC_CQ_REF_COUNT_DEBUG
Nicolas "Pixel" Nobleaaf39ab2015-06-24 23:22:53 +020095void grpc_cq_internal_unref(grpc_completion_queue *cc, const char *reason,
96 const char *file, int line) {
Craig Tiller12cf5372015-07-09 13:48:11 -070097 gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, "CQ:%p unref %d -> %d %s", cc,
98 (int)cc->owning_refs.count, (int)cc->owning_refs.count - 1, reason);
Craig Tiller463f2372015-05-28 16:16:15 -070099#else
Craig Tiller5717a982015-04-27 12:01:49 -0700100void grpc_cq_internal_unref(grpc_completion_queue *cc) {
Craig Tiller463f2372015-05-28 16:16:15 -0700101#endif
Craig Tiller5717a982015-04-27 12:01:49 -0700102 if (gpr_unref(&cc->owning_refs)) {
Craig Tiller12cf5372015-07-09 13:48:11 -0700103 GPR_ASSERT(cc->completed_head.next == (gpr_uintptr)&cc->completed_head);
Craig Tiller70730b42015-05-22 14:42:38 -0700104 grpc_pollset_destroy(&cc->pollset);
105 gpr_free(cc);
Craig Tiller5717a982015-04-27 12:01:49 -0700106 }
107}
108
Craig Tiller97fc6a32015-07-08 15:31:35 -0700109void grpc_cq_begin_op(grpc_completion_queue *cc) {
Craig Tiller402acf62015-08-05 10:43:10 -0700110#ifndef NDEBUG
111 gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset));
112 GPR_ASSERT(!cc->shutdown_called);
113 gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
114#endif
Craig Tiller97fc6a32015-07-08 15:31:35 -0700115 gpr_ref(&cc->pending_events);
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800116}
117
118/* Signal the end of an operation - if this is the last waiting-to-be-queued
119 event, then enter shutdown mode */
Craig Tiller97fc6a32015-07-08 15:31:35 -0700120/* Queue a GRPC_OP_COMPLETED operation */
Craig Tiller12cf5372015-07-09 13:48:11 -0700121void grpc_cq_end_op(grpc_completion_queue *cc, void *tag, int success,
122 void (*done)(void *done_arg, grpc_cq_completion *storage),
123 void *done_arg, grpc_cq_completion *storage) {
Craig Tiller304048c2015-07-17 11:19:58 -0700124 int shutdown;
Craig Tiller97fc6a32015-07-08 15:31:35 -0700125
126 storage->tag = tag;
127 storage->done = done;
128 storage->done_arg = done_arg;
Craig Tiller12cf5372015-07-09 13:48:11 -0700129 storage->next =
130 ((gpr_uintptr)&cc->completed_head) | ((gpr_uintptr)(success != 0));
Craig Tiller97fc6a32015-07-08 15:31:35 -0700131
Craig Tiller304048c2015-07-17 11:19:58 -0700132 gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset));
133 shutdown = gpr_unref(&cc->pending_events);
Craig Tiller97fc6a32015-07-08 15:31:35 -0700134 if (!shutdown) {
Craig Tiller12cf5372015-07-09 13:48:11 -0700135 cc->completed_tail->next =
136 ((gpr_uintptr)storage) | (1u & (gpr_uintptr)cc->completed_tail->next);
Craig Tiller97fc6a32015-07-08 15:31:35 -0700137 cc->completed_tail = storage;
Craig Tiller86062bb2015-07-09 16:56:21 -0700138 grpc_pollset_kick(&cc->pollset);
Craig Tiller97fc6a32015-07-08 15:31:35 -0700139 gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
140 } else {
Craig Tiller12cf5372015-07-09 13:48:11 -0700141 cc->completed_tail->next =
142 ((gpr_uintptr)storage) | (1u & (gpr_uintptr)cc->completed_tail->next);
Craig Tiller97fc6a32015-07-08 15:31:35 -0700143 cc->completed_tail = storage;
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800144 GPR_ASSERT(!cc->shutdown);
Craig Tillerf5fd4ba2015-03-02 18:01:21 +0000145 GPR_ASSERT(cc->shutdown_called);
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800146 cc->shutdown = 1;
Craig Tiller97fc6a32015-07-08 15:31:35 -0700147 gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
Craig Tiller70730b42015-05-22 14:42:38 -0700148 grpc_pollset_shutdown(&cc->pollset, on_pollset_destroy_done, cc);
Craig Tiller6f2d6472015-05-22 21:34:15 -0700149 }
Craig Tillercce17ac2015-01-20 09:29:28 -0800150}
151
Craig Tiller64be9f72015-05-04 14:53:51 -0700152grpc_event grpc_completion_queue_next(grpc_completion_queue *cc,
153 gpr_timespec deadline) {
Craig Tiller64be9f72015-05-04 14:53:51 -0700154 grpc_event ret;
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800155
Craig Tiller6a7626c2015-07-19 22:21:41 -0700156 deadline = gpr_convert_clock_type(deadline, GPR_CLOCK_MONOTONIC);
157
Craig Tiller463f2372015-05-28 16:16:15 -0700158 GRPC_CQ_INTERNAL_REF(cc, "next");
ctiller58393c22015-01-07 14:03:30 -0800159 gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset));
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800160 for (;;) {
Craig Tiller97fc6a32015-07-08 15:31:35 -0700161 if (cc->completed_tail != &cc->completed_head) {
Craig Tiller12cf5372015-07-09 13:48:11 -0700162 grpc_cq_completion *c = (grpc_cq_completion *)cc->completed_head.next;
Craig Tiller97fc6a32015-07-08 15:31:35 -0700163 cc->completed_head.next = c->next & ~(gpr_uintptr)1;
164 if (c == cc->completed_tail) {
165 cc->completed_tail = &cc->completed_head;
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800166 }
Craig Tiller97fc6a32015-07-08 15:31:35 -0700167 gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
168 ret.type = GRPC_OP_COMPLETE;
169 ret.success = c->next & 1u;
170 ret.tag = c->tag;
171 c->done(c->done_arg, c);
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800172 break;
173 }
174 if (cc->shutdown) {
Craig Tiller97fc6a32015-07-08 15:31:35 -0700175 gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
176 memset(&ret, 0, sizeof(ret));
177 ret.type = GRPC_QUEUE_SHUTDOWN;
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800178 break;
179 }
Craig Tiller44883302015-05-28 12:44:27 -0700180 if (!grpc_pollset_work(&cc->pollset, deadline)) {
ctiller58393c22015-01-07 14:03:30 -0800181 gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
Craig Tiller64be9f72015-05-04 14:53:51 -0700182 memset(&ret, 0, sizeof(ret));
183 ret.type = GRPC_QUEUE_TIMEOUT;
Craig Tiller97fc6a32015-07-08 15:31:35 -0700184 break;
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800185 }
186 }
Craig Tiller64be9f72015-05-04 14:53:51 -0700187 GRPC_SURFACE_TRACE_RETURNED_EVENT(cc, &ret);
Craig Tiller463f2372015-05-28 16:16:15 -0700188 GRPC_CQ_INTERNAL_UNREF(cc, "next");
Craig Tiller64be9f72015-05-04 14:53:51 -0700189 return ret;
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800190}
191
Craig Tiller64be9f72015-05-04 14:53:51 -0700192grpc_event grpc_completion_queue_pluck(grpc_completion_queue *cc, void *tag,
193 gpr_timespec deadline) {
Craig Tiller64be9f72015-05-04 14:53:51 -0700194 grpc_event ret;
Craig Tiller97fc6a32015-07-08 15:31:35 -0700195 grpc_cq_completion *c;
196 grpc_cq_completion *prev;
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800197
Craig Tiller6a7626c2015-07-19 22:21:41 -0700198 deadline = gpr_convert_clock_type(deadline, GPR_CLOCK_MONOTONIC);
199
Craig Tiller463f2372015-05-28 16:16:15 -0700200 GRPC_CQ_INTERNAL_REF(cc, "pluck");
ctiller58393c22015-01-07 14:03:30 -0800201 gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset));
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800202 for (;;) {
Craig Tiller97fc6a32015-07-08 15:31:35 -0700203 prev = &cc->completed_head;
Craig Tiller12cf5372015-07-09 13:48:11 -0700204 while ((c = (grpc_cq_completion *)(prev->next & ~(gpr_uintptr)1)) !=
205 &cc->completed_head) {
Craig Tiller97fc6a32015-07-08 15:31:35 -0700206 if (c->tag == tag) {
Craig Tiller12cf5372015-07-09 13:48:11 -0700207 prev->next =
208 (prev->next & (gpr_uintptr)1) | (c->next & ~(gpr_uintptr)1);
Craig Tiller97fc6a32015-07-08 15:31:35 -0700209 if (c == cc->completed_tail) {
210 cc->completed_tail = prev;
211 }
212 gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
213 ret.type = GRPC_OP_COMPLETE;
214 ret.success = c->next & 1u;
215 ret.tag = c->tag;
216 c->done(c->done_arg, c);
217 goto done;
218 }
219 prev = c;
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800220 }
221 if (cc->shutdown) {
Craig Tiller97fc6a32015-07-08 15:31:35 -0700222 gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
223 memset(&ret, 0, sizeof(ret));
224 ret.type = GRPC_QUEUE_SHUTDOWN;
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800225 break;
226 }
Craig Tiller44883302015-05-28 12:44:27 -0700227 if (!grpc_pollset_work(&cc->pollset, deadline)) {
ctiller58393c22015-01-07 14:03:30 -0800228 gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
Craig Tiller64be9f72015-05-04 14:53:51 -0700229 memset(&ret, 0, sizeof(ret));
230 ret.type = GRPC_QUEUE_TIMEOUT;
Craig Tiller97fc6a32015-07-08 15:31:35 -0700231 break;
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800232 }
233 }
Craig Tiller97fc6a32015-07-08 15:31:35 -0700234done:
Craig Tiller4751fb12015-05-15 14:42:54 -0700235 GRPC_SURFACE_TRACE_RETURNED_EVENT(cc, &ret);
Craig Tiller463f2372015-05-28 16:16:15 -0700236 GRPC_CQ_INTERNAL_UNREF(cc, "pluck");
Craig Tiller64be9f72015-05-04 14:53:51 -0700237 return ret;
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800238}
239
240/* Shutdown simply drops a ref that we reserved at creation time; if we drop
241 to zero here, then enter shutdown mode and wake up any waiters */
242void grpc_completion_queue_shutdown(grpc_completion_queue *cc) {
Craig Tillerf5fd4ba2015-03-02 18:01:21 +0000243 gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset));
Craig Tiller91c069c2015-05-29 11:32:10 -0700244 if (cc->shutdown_called) {
245 gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
246 return;
247 }
Craig Tillerf5fd4ba2015-03-02 18:01:21 +0000248 cc->shutdown_called = 1;
249 gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
250
Craig Tiller97fc6a32015-07-08 15:31:35 -0700251 if (gpr_unref(&cc->pending_events)) {
ctiller58393c22015-01-07 14:03:30 -0800252 gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset));
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800253 GPR_ASSERT(!cc->shutdown);
254 cc->shutdown = 1;
ctiller58393c22015-01-07 14:03:30 -0800255 gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
Craig Tiller70730b42015-05-22 14:42:38 -0700256 grpc_pollset_shutdown(&cc->pollset, on_pollset_destroy_done, cc);
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800257 }
258}
259
David Klempnerb5056612015-02-24 14:22:50 -0800260void grpc_completion_queue_destroy(grpc_completion_queue *cc) {
Craig Tiller91c069c2015-05-29 11:32:10 -0700261 grpc_completion_queue_shutdown(cc);
Craig Tiller463f2372015-05-28 16:16:15 -0700262 GRPC_CQ_INTERNAL_UNREF(cc, "destroy");
David Klempnerb5056612015-02-24 14:22:50 -0800263}
264
ctillerd79b4862014-12-17 16:36:59 -0800265grpc_pollset *grpc_cq_pollset(grpc_completion_queue *cc) {
266 return &cc->pollset;
Craig Tiller190d3602015-02-18 09:23:38 -0800267}
Craig Tilleraec96aa2015-04-07 14:32:15 -0700268
269void grpc_cq_hack_spin_pollset(grpc_completion_queue *cc) {
270 gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset));
271 grpc_pollset_kick(&cc->pollset);
Craig Tiller58bbc862015-07-13 09:51:17 -0700272 grpc_pollset_work(&cc->pollset,
273 gpr_time_add(gpr_now(GPR_CLOCK_REALTIME),
274 gpr_time_from_millis(100, GPR_TIMESPAN)));
Craig Tilleraec96aa2015-04-07 14:32:15 -0700275 gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
276}
Craig Tillerb56975c2015-06-15 10:11:16 -0700277
278void grpc_cq_mark_server_cq(grpc_completion_queue *cc) { cc->is_server_cq = 1; }
279
280int grpc_cq_is_server_cq(grpc_completion_queue *cc) { return cc->is_server_cq; }