blob: d46e3de85f0ee0f405d0fc62710e3c96d5a4b076 [file] [log] [blame]
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -08001/*
2 *
3 * Copyright 2014, Google Inc.
4 * 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/server.h"
35
36#include <stdlib.h>
37#include <string.h>
38
39#include "src/core/channel/census_filter.h"
40#include "src/core/channel/channel_args.h"
41#include "src/core/channel/connected_channel.h"
ctiller18b49ab2014-12-09 14:39:16 -080042#include "src/core/iomgr/iomgr.h"
Craig Tiller485d7762015-01-23 12:54:05 -080043#include "src/core/support/string.h"
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -080044#include "src/core/surface/call.h"
45#include "src/core/surface/channel.h"
46#include "src/core/surface/completion_queue.h"
Craig Tillercce17ac2015-01-20 09:29:28 -080047#include "src/core/transport/metadata.h"
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -080048#include <grpc/support/alloc.h>
49#include <grpc/support/log.h>
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -080050#include <grpc/support/useful.h>
51
52typedef enum { PENDING_START, ALL_CALLS, CALL_LIST_COUNT } call_list;
53
54typedef struct listener {
55 void *arg;
ctiller58393c22015-01-07 14:03:30 -080056 void (*start)(grpc_server *server, void *arg, grpc_pollset *pollset);
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -080057 void (*destroy)(grpc_server *server, void *arg);
58 struct listener *next;
59} listener;
60
61typedef struct call_data call_data;
62typedef struct channel_data channel_data;
63
64struct channel_data {
65 grpc_server *server;
66 grpc_channel *channel;
Craig Tillercce17ac2015-01-20 09:29:28 -080067 grpc_mdstr *path_key;
68 grpc_mdstr *authority_key;
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -080069 /* linked list of all channels on a server */
70 channel_data *next;
71 channel_data *prev;
72};
73
Craig Tillercce17ac2015-01-20 09:29:28 -080074typedef void (*new_call_cb)(grpc_server *server, grpc_completion_queue *cq, grpc_metadata_array *initial_metadata, call_data *calld, void *user_data);
75
76typedef struct {
77 void *user_data;
78 grpc_completion_queue *cq;
79 grpc_metadata_array *initial_metadata;
80 new_call_cb cb;
81} requested_call;
82
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -080083struct grpc_server {
84 size_t channel_filter_count;
85 const grpc_channel_filter **channel_filters;
86 grpc_channel_args *channel_args;
87 grpc_completion_queue *cq;
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -080088
89 gpr_mu mu;
90
Craig Tillercce17ac2015-01-20 09:29:28 -080091 requested_call *requested_calls;
92 size_t requested_call_count;
93 size_t requested_call_capacity;
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -080094
95 gpr_uint8 shutdown;
Craig Tiller4ffdcd52015-01-16 11:34:55 -080096 gpr_uint8 have_shutdown_tag;
97 void *shutdown_tag;
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -080098
99 call_data *lists[CALL_LIST_COUNT];
100 channel_data root_channel_data;
101
102 listener *listeners;
103 gpr_refcount internal_refcount;
104};
105
106typedef struct {
107 call_data *next;
108 call_data *prev;
109} call_link;
110
111typedef enum {
112 /* waiting for metadata */
113 NOT_STARTED,
114 /* inital metadata read, not flow controlled in yet */
115 PENDING,
116 /* flow controlled in, on completion queue */
117 ACTIVATED,
118 /* cancelled before being queued */
119 ZOMBIED
120} call_state;
121
Craig Tillercce17ac2015-01-20 09:29:28 -0800122typedef struct legacy_data {
123 grpc_metadata_array client_metadata;
124} legacy_data;
125
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800126struct call_data {
127 grpc_call *call;
128
129 call_state state;
130 gpr_timespec deadline;
Craig Tillercce17ac2015-01-20 09:29:28 -0800131 grpc_mdstr *path;
132 grpc_mdstr *host;
133
134 legacy_data *legacy;
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800135
136 gpr_uint8 included[CALL_LIST_COUNT];
137 call_link links[CALL_LIST_COUNT];
138};
139
140#define SERVER_FROM_CALL_ELEM(elem) \
141 (((channel_data *)(elem)->channel_data)->server)
142
143static void do_nothing(void *unused, grpc_op_error ignored) {}
144
145static int call_list_join(grpc_server *server, call_data *call,
146 call_list list) {
147 if (call->included[list]) return 0;
148 call->included[list] = 1;
149 if (!server->lists[list]) {
150 server->lists[list] = call;
151 call->links[list].next = call->links[list].prev = call;
152 } else {
153 call->links[list].next = server->lists[list];
154 call->links[list].prev = server->lists[list]->links[list].prev;
155 call->links[list].next->links[list].prev =
156 call->links[list].prev->links[list].next = call;
157 }
158 return 1;
159}
160
161static call_data *call_list_remove_head(grpc_server *server, call_list list) {
162 call_data *out = server->lists[list];
163 if (out) {
164 out->included[list] = 0;
165 if (out->links[list].next == out) {
166 server->lists[list] = NULL;
167 } else {
168 server->lists[list] = out->links[list].next;
169 out->links[list].next->links[list].prev = out->links[list].prev;
170 out->links[list].prev->links[list].next = out->links[list].next;
171 }
172 }
173 return out;
174}
175
176static int call_list_remove(grpc_server *server, call_data *call,
177 call_list list) {
178 if (!call->included[list]) return 0;
179 call->included[list] = 0;
180 if (server->lists[list] == call) {
181 server->lists[list] = call->links[list].next;
182 if (server->lists[list] == call) {
183 server->lists[list] = NULL;
184 return 1;
185 }
186 }
187 GPR_ASSERT(server->lists[list] != call);
188 call->links[list].next->links[list].prev = call->links[list].prev;
189 call->links[list].prev->links[list].next = call->links[list].next;
190 return 1;
191}
192
193static void server_ref(grpc_server *server) {
194 gpr_ref(&server->internal_refcount);
195}
196
197static void server_unref(grpc_server *server) {
198 if (gpr_unref(&server->internal_refcount)) {
199 grpc_channel_args_destroy(server->channel_args);
200 gpr_mu_destroy(&server->mu);
201 gpr_free(server->channel_filters);
Craig Tillercce17ac2015-01-20 09:29:28 -0800202 gpr_free(server->requested_calls);
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800203 gpr_free(server);
204 }
205}
206
207static int is_channel_orphaned(channel_data *chand) {
208 return chand->next == chand;
209}
210
211static void orphan_channel(channel_data *chand) {
212 chand->next->prev = chand->prev;
213 chand->prev->next = chand->next;
214 chand->next = chand->prev = chand;
215}
216
ctiller58393c22015-01-07 14:03:30 -0800217static void finish_destroy_channel(void *cd, int success) {
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800218 channel_data *chand = cd;
219 grpc_server *server = chand->server;
220 /*gpr_log(GPR_INFO, "destroy channel %p", chand->channel);*/
221 grpc_channel_destroy(chand->channel);
222 server_unref(server);
223}
224
225static void destroy_channel(channel_data *chand) {
226 if (is_channel_orphaned(chand)) return;
227 GPR_ASSERT(chand->server != NULL);
228 orphan_channel(chand);
229 server_ref(chand->server);
ctiller18b49ab2014-12-09 14:39:16 -0800230 grpc_iomgr_add_callback(finish_destroy_channel, chand);
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800231}
232
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800233static void start_new_rpc(grpc_call_element *elem) {
234 channel_data *chand = elem->channel_data;
235 call_data *calld = elem->call_data;
236 grpc_server *server = chand->server;
237
238 gpr_mu_lock(&server->mu);
Craig Tillercce17ac2015-01-20 09:29:28 -0800239 if (server->requested_call_count > 0) {
240 requested_call rc = server->requested_calls[--server->requested_call_count];
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800241 calld->state = ACTIVATED;
Craig Tillercce17ac2015-01-20 09:29:28 -0800242 gpr_mu_unlock(&server->mu);
243 rc.cb(server, rc.cq, rc.initial_metadata, calld, rc.user_data);
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800244 } else {
245 calld->state = PENDING;
246 call_list_join(server, calld, PENDING_START);
Craig Tillercce17ac2015-01-20 09:29:28 -0800247 gpr_mu_unlock(&server->mu);
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800248 }
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800249}
250
ctiller58393c22015-01-07 14:03:30 -0800251static void kill_zombie(void *elem, int success) {
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800252 grpc_call_destroy(grpc_call_from_top_element(elem));
253}
254
Craig Tillercce17ac2015-01-20 09:29:28 -0800255static void stream_closed(grpc_call_element *elem) {
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800256 call_data *calld = elem->call_data;
257 channel_data *chand = elem->channel_data;
258 gpr_mu_lock(&chand->server->mu);
259 switch (calld->state) {
260 case ACTIVATED:
Craig Tillercce17ac2015-01-20 09:29:28 -0800261 grpc_call_stream_closed(elem);
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800262 break;
263 case PENDING:
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800264 call_list_remove(chand->server, calld, PENDING_START);
265 /* fallthrough intended */
266 case NOT_STARTED:
267 calld->state = ZOMBIED;
ctiller18b49ab2014-12-09 14:39:16 -0800268 grpc_iomgr_add_callback(kill_zombie, elem);
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800269 break;
270 case ZOMBIED:
271 break;
272 }
273 gpr_mu_unlock(&chand->server->mu);
274}
275
Craig Tillercce17ac2015-01-20 09:29:28 -0800276static void read_closed(grpc_call_element *elem) {
277 call_data *calld = elem->call_data;
278 channel_data *chand = elem->channel_data;
279 gpr_mu_lock(&chand->server->mu);
280 switch (calld->state) {
281 case ACTIVATED:
282 case PENDING:
283 grpc_call_read_closed(elem);
284 break;
285 case NOT_STARTED:
286 calld->state = ZOMBIED;
287 grpc_iomgr_add_callback(kill_zombie, elem);
288 break;
289 case ZOMBIED:
290 break;
291 }
292 gpr_mu_unlock(&chand->server->mu);
293}
294
ctillerf962f522014-12-10 15:28:27 -0800295static void call_op(grpc_call_element *elem, grpc_call_element *from_elemn,
296 grpc_call_op *op) {
Craig Tillercce17ac2015-01-20 09:29:28 -0800297 channel_data *chand = elem->channel_data;
298 call_data *calld = elem->call_data;
299 grpc_mdelem *md;
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800300 GRPC_CALL_LOG_OP(GPR_INFO, elem, op);
301 switch (op->type) {
302 case GRPC_RECV_METADATA:
Craig Tillercce17ac2015-01-20 09:29:28 -0800303 md = op->data.metadata;
304 if (md->key == chand->path_key) {
305 calld->path = grpc_mdstr_ref(md->value);
306 grpc_mdelem_unref(md);
307 } else if (md->key == chand->authority_key) {
308 calld->host = grpc_mdstr_ref(md->value);
309 grpc_mdelem_unref(md);
310 } else {
311 grpc_call_recv_metadata(elem, md);
312 }
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800313 break;
314 case GRPC_RECV_END_OF_INITIAL_METADATA:
315 start_new_rpc(elem);
316 break;
317 case GRPC_RECV_MESSAGE:
Craig Tillercce17ac2015-01-20 09:29:28 -0800318 grpc_call_recv_message(elem, op->data.message);
319 op->done_cb(op->user_data, GRPC_OP_OK);
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800320 break;
321 case GRPC_RECV_HALF_CLOSE:
Craig Tillercce17ac2015-01-20 09:29:28 -0800322 read_closed(elem);
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800323 break;
324 case GRPC_RECV_FINISH:
Craig Tillercce17ac2015-01-20 09:29:28 -0800325 stream_closed(elem);
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800326 break;
327 case GRPC_RECV_DEADLINE:
328 grpc_call_set_deadline(elem, op->data.deadline);
329 ((call_data *)elem->call_data)->deadline = op->data.deadline;
330 break;
331 default:
332 GPR_ASSERT(op->dir == GRPC_CALL_DOWN);
333 grpc_call_next_op(elem, op);
334 break;
335 }
336}
337
ctillerf962f522014-12-10 15:28:27 -0800338static void channel_op(grpc_channel_element *elem,
339 grpc_channel_element *from_elem, grpc_channel_op *op) {
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800340 channel_data *chand = elem->channel_data;
341
342 switch (op->type) {
343 case GRPC_ACCEPT_CALL:
344 /* create a call */
345 grpc_call_create(chand->channel,
346 op->data.accept_call.transport_server_data);
347 break;
348 case GRPC_TRANSPORT_CLOSED:
349 /* if the transport is closed for a server channel, we destroy the
350 channel */
351 gpr_mu_lock(&chand->server->mu);
352 server_ref(chand->server);
353 destroy_channel(chand);
354 gpr_mu_unlock(&chand->server->mu);
355 server_unref(chand->server);
356 break;
nnoble0c475f02014-12-05 15:37:39 -0800357 case GRPC_TRANSPORT_GOAWAY:
358 gpr_slice_unref(op->data.goaway.message);
359 break;
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800360 default:
361 GPR_ASSERT(op->dir == GRPC_CALL_DOWN);
362 grpc_channel_next_op(elem, op);
363 break;
364 }
365}
366
ctiller58393c22015-01-07 14:03:30 -0800367static void finish_shutdown_channel(void *cd, int success) {
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800368 channel_data *chand = cd;
369 grpc_channel_op op;
nnoble0c475f02014-12-05 15:37:39 -0800370 op.type = GRPC_CHANNEL_DISCONNECT;
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800371 op.dir = GRPC_CALL_DOWN;
372 channel_op(grpc_channel_stack_element(
373 grpc_channel_get_channel_stack(chand->channel), 0),
ctillerf962f522014-12-10 15:28:27 -0800374 NULL, &op);
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800375 grpc_channel_internal_unref(chand->channel);
376}
377
378static void shutdown_channel(channel_data *chand) {
379 grpc_channel_internal_ref(chand->channel);
ctiller18b49ab2014-12-09 14:39:16 -0800380 grpc_iomgr_add_callback(finish_shutdown_channel, chand);
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800381}
382
383static void init_call_elem(grpc_call_element *elem,
384 const void *server_transport_data) {
385 call_data *calld = elem->call_data;
386 channel_data *chand = elem->channel_data;
387 memset(calld, 0, sizeof(call_data));
388 calld->deadline = gpr_inf_future;
389 calld->call = grpc_call_from_top_element(elem);
390
391 gpr_mu_lock(&chand->server->mu);
392 call_list_join(chand->server, calld, ALL_CALLS);
393 gpr_mu_unlock(&chand->server->mu);
394
395 server_ref(chand->server);
396}
397
398static void destroy_call_elem(grpc_call_element *elem) {
399 channel_data *chand = elem->channel_data;
400 int i;
401
402 gpr_mu_lock(&chand->server->mu);
403 for (i = 0; i < CALL_LIST_COUNT; i++) {
404 call_list_remove(chand->server, elem->call_data, i);
405 }
Craig Tiller4ffdcd52015-01-16 11:34:55 -0800406 if (chand->server->shutdown && chand->server->have_shutdown_tag &&
407 chand->server->lists[ALL_CALLS] == NULL) {
408 grpc_cq_end_server_shutdown(chand->server->cq, chand->server->shutdown_tag);
409 }
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800410 gpr_mu_unlock(&chand->server->mu);
411
412 server_unref(chand->server);
413}
414
415static void init_channel_elem(grpc_channel_element *elem,
416 const grpc_channel_args *args,
417 grpc_mdctx *metadata_context, int is_first,
418 int is_last) {
419 channel_data *chand = elem->channel_data;
420 GPR_ASSERT(is_first);
421 GPR_ASSERT(!is_last);
422 chand->server = NULL;
423 chand->channel = NULL;
Craig Tillercce17ac2015-01-20 09:29:28 -0800424 chand->path_key = grpc_mdstr_from_string(metadata_context, ":path");
425 chand->authority_key = grpc_mdstr_from_string(metadata_context, ":authority");
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800426 chand->next = chand->prev = chand;
427}
428
429static void destroy_channel_elem(grpc_channel_element *elem) {
430 channel_data *chand = elem->channel_data;
431 if (chand->server) {
432 gpr_mu_lock(&chand->server->mu);
433 chand->next->prev = chand->prev;
434 chand->prev->next = chand->next;
435 chand->next = chand->prev = chand;
436 gpr_mu_unlock(&chand->server->mu);
Craig Tillercce17ac2015-01-20 09:29:28 -0800437 grpc_mdstr_unref(chand->path_key);
438 grpc_mdstr_unref(chand->authority_key);
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800439 server_unref(chand->server);
440 }
441}
442
443static const grpc_channel_filter server_surface_filter = {
Yang Gao5fd0d292015-01-26 00:19:48 -0800444 call_op, channel_op, sizeof(call_data),
445 init_call_elem, destroy_call_elem, sizeof(channel_data),
446 init_channel_elem, destroy_channel_elem, "server", };
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800447
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800448grpc_server *grpc_server_create_from_filters(grpc_completion_queue *cq,
449 grpc_channel_filter **filters,
450 size_t filter_count,
451 const grpc_channel_args *args) {
452 size_t i;
453 int census_enabled = grpc_channel_args_is_census_enabled(args);
454
455 grpc_server *server = gpr_malloc(sizeof(grpc_server));
456 memset(server, 0, sizeof(grpc_server));
457
458 gpr_mu_init(&server->mu);
459
460 server->cq = cq;
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800461 /* decremented by grpc_server_destroy */
462 gpr_ref_init(&server->internal_refcount, 1);
463 server->root_channel_data.next = server->root_channel_data.prev =
464 &server->root_channel_data;
465
466 /* Server filter stack is:
467
468 server_surface_filter - for making surface API calls
469 grpc_server_census_filter (optional) - for stats collection and tracing
470 {passed in filter stack}
471 grpc_connected_channel_filter - for interfacing with transports */
472 server->channel_filter_count = filter_count + 1 + census_enabled;
473 server->channel_filters =
474 gpr_malloc(server->channel_filter_count * sizeof(grpc_channel_filter *));
475 server->channel_filters[0] = &server_surface_filter;
476 if (census_enabled) {
477 server->channel_filters[1] = &grpc_server_census_filter;
478 }
479 for (i = 0; i < filter_count; i++) {
480 server->channel_filters[i + 1 + census_enabled] = filters[i];
481 }
482
483 server->channel_args = grpc_channel_args_copy(args);
484
485 return server;
486}
487
488void grpc_server_start(grpc_server *server) {
489 listener *l;
490
491 for (l = server->listeners; l; l = l->next) {
ctiller58393c22015-01-07 14:03:30 -0800492 l->start(server, l->arg, grpc_cq_pollset(server->cq));
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800493 }
494}
495
496grpc_transport_setup_result grpc_server_setup_transport(
497 grpc_server *s, grpc_transport *transport,
498 grpc_channel_filter const **extra_filters, size_t num_extra_filters,
499 grpc_mdctx *mdctx) {
500 size_t num_filters = s->channel_filter_count + num_extra_filters + 1;
501 grpc_channel_filter const **filters =
502 gpr_malloc(sizeof(grpc_channel_filter *) * num_filters);
503 size_t i;
504 grpc_channel *channel;
505 channel_data *chand;
506
507 for (i = 0; i < s->channel_filter_count; i++) {
508 filters[i] = s->channel_filters[i];
509 }
510 for (; i < s->channel_filter_count + num_extra_filters; i++) {
511 filters[i] = extra_filters[i - s->channel_filter_count];
512 }
513 filters[i] = &grpc_connected_channel_filter;
514
ctillerd79b4862014-12-17 16:36:59 -0800515 grpc_transport_add_to_pollset(transport, grpc_cq_pollset(s->cq));
516
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800517 channel = grpc_channel_create_from_filters(filters, num_filters,
518 s->channel_args, mdctx, 0);
519 chand = (channel_data *)grpc_channel_stack_element(
520 grpc_channel_get_channel_stack(channel), 0)->channel_data;
521 chand->server = s;
522 server_ref(s);
523 chand->channel = channel;
524
525 gpr_mu_lock(&s->mu);
526 chand->next = &s->root_channel_data;
527 chand->prev = chand->next->prev;
528 chand->next->prev = chand->prev->next = chand;
529 gpr_mu_unlock(&s->mu);
530
531 gpr_free(filters);
532
533 return grpc_connected_channel_bind_transport(
534 grpc_channel_get_channel_stack(channel), transport);
535}
536
Craig Tiller4ffdcd52015-01-16 11:34:55 -0800537void shutdown_internal(grpc_server *server, gpr_uint8 have_shutdown_tag,
538 void *shutdown_tag) {
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800539 listener *l;
Craig Tillercce17ac2015-01-20 09:29:28 -0800540 requested_call *requested_calls;
541 size_t requested_call_count;
nnoble0c475f02014-12-05 15:37:39 -0800542 channel_data **channels;
543 channel_data *c;
544 size_t nchannels;
545 size_t i;
546 grpc_channel_op op;
547 grpc_channel_element *elem;
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800548
549 /* lock, and gather up some stuff to do */
550 gpr_mu_lock(&server->mu);
551 if (server->shutdown) {
552 gpr_mu_unlock(&server->mu);
553 return;
554 }
555
nnoble0c475f02014-12-05 15:37:39 -0800556 nchannels = 0;
557 for (c = server->root_channel_data.next; c != &server->root_channel_data;
558 c = c->next) {
559 nchannels++;
560 }
561 channels = gpr_malloc(sizeof(channel_data *) * nchannels);
562 i = 0;
563 for (c = server->root_channel_data.next; c != &server->root_channel_data;
564 c = c->next) {
565 grpc_channel_internal_ref(c->channel);
566 channels[i] = c;
567 i++;
568 }
569
Craig Tillercce17ac2015-01-20 09:29:28 -0800570 requested_calls = server->requested_calls;
571 requested_call_count = server->requested_call_count;
572 server->requested_calls = NULL;
573 server->requested_call_count = 0;
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800574
575 server->shutdown = 1;
Craig Tiller4ffdcd52015-01-16 11:34:55 -0800576 server->have_shutdown_tag = have_shutdown_tag;
577 server->shutdown_tag = shutdown_tag;
578 if (have_shutdown_tag) {
579 grpc_cq_begin_op(server->cq, NULL, GRPC_SERVER_SHUTDOWN);
580 if (server->lists[ALL_CALLS] == NULL) {
581 grpc_cq_end_server_shutdown(server->cq, shutdown_tag);
582 }
583 }
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800584 gpr_mu_unlock(&server->mu);
585
nnoble0c475f02014-12-05 15:37:39 -0800586 for (i = 0; i < nchannels; i++) {
587 c = channels[i];
588 elem = grpc_channel_stack_element(
589 grpc_channel_get_channel_stack(c->channel), 0);
590
591 op.type = GRPC_CHANNEL_GOAWAY;
592 op.dir = GRPC_CALL_DOWN;
593 op.data.goaway.status = GRPC_STATUS_OK;
594 op.data.goaway.message = gpr_slice_from_copied_string("Server shutdown");
ctillerf962f522014-12-10 15:28:27 -0800595 elem->filter->channel_op(elem, NULL, &op);
nnoble0c475f02014-12-05 15:37:39 -0800596
597 grpc_channel_internal_unref(c->channel);
598 }
599 gpr_free(channels);
600
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800601 /* terminate all the requested calls */
Craig Tillercce17ac2015-01-20 09:29:28 -0800602 for (i = 0; i < requested_call_count; i++) {
603 requested_calls[i].cb(server, requested_calls[i].cq, requested_calls[i].initial_metadata, NULL, requested_calls[i].user_data);
604 }
605 gpr_free(requested_calls);
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800606
607 /* Shutdown listeners */
608 for (l = server->listeners; l; l = l->next) {
609 l->destroy(server, l->arg);
610 }
611 while (server->listeners) {
612 l = server->listeners;
613 server->listeners = l->next;
614 gpr_free(l);
615 }
616}
617
Craig Tiller4ffdcd52015-01-16 11:34:55 -0800618void grpc_server_shutdown(grpc_server *server) {
619 shutdown_internal(server, 0, NULL);
620}
621
622void grpc_server_shutdown_and_notify(grpc_server *server, void *tag) {
623 shutdown_internal(server, 1, tag);
624}
625
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800626void grpc_server_destroy(grpc_server *server) {
627 channel_data *c;
628 gpr_mu_lock(&server->mu);
629 for (c = server->root_channel_data.next; c != &server->root_channel_data;
630 c = c->next) {
631 shutdown_channel(c);
632 }
633 gpr_mu_unlock(&server->mu);
634
635 server_unref(server);
636}
637
638void grpc_server_add_listener(grpc_server *server, void *arg,
ctiller58393c22015-01-07 14:03:30 -0800639 void (*start)(grpc_server *server, void *arg,
640 grpc_pollset *pollset),
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800641 void (*destroy)(grpc_server *server, void *arg)) {
642 listener *l = gpr_malloc(sizeof(listener));
643 l->arg = arg;
644 l->start = start;
645 l->destroy = destroy;
646 l->next = server->listeners;
647 server->listeners = l;
648}
649
Craig Tillercce17ac2015-01-20 09:29:28 -0800650static grpc_call_error queue_call_request(grpc_server *server, grpc_completion_queue *cq, grpc_metadata_array *initial_metadata, new_call_cb cb, void *user_data) {
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800651 call_data *calld;
Craig Tillercce17ac2015-01-20 09:29:28 -0800652 requested_call *rc;
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800653 gpr_mu_lock(&server->mu);
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800654 if (server->shutdown) {
655 gpr_mu_unlock(&server->mu);
Craig Tillercce17ac2015-01-20 09:29:28 -0800656 cb(server, cq, initial_metadata, NULL, user_data);
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800657 return GRPC_CALL_OK;
658 }
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800659 calld = call_list_remove_head(server, PENDING_START);
660 if (calld) {
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800661 calld->state = ACTIVATED;
Craig Tillercce17ac2015-01-20 09:29:28 -0800662 GPR_ASSERT(calld->state == PENDING);
663 gpr_mu_unlock(&server->mu);
664 cb(server, cq, initial_metadata, calld, user_data);
665 return GRPC_CALL_OK;
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800666 } else {
Craig Tillercce17ac2015-01-20 09:29:28 -0800667 if (server->requested_call_count == server->requested_call_capacity) {
668 server->requested_call_capacity = GPR_MAX(server->requested_call_capacity + 8, server->requested_call_capacity * 2);
669 server->requested_calls = gpr_realloc(server->requested_calls,
670 sizeof(requested_call) * server->requested_call_capacity);
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800671 }
Craig Tillercce17ac2015-01-20 09:29:28 -0800672 rc = &server->requested_calls[server->requested_call_count++];
673 rc->cb = cb;
674 rc->cq = cq;
675 rc->user_data = user_data;
676 rc->initial_metadata = initial_metadata;
677 gpr_mu_unlock(&server->mu);
678 return GRPC_CALL_OK;
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800679 }
Craig Tillercce17ac2015-01-20 09:29:28 -0800680}
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800681
Craig Tillercce17ac2015-01-20 09:29:28 -0800682static void begin_request(grpc_server *server, grpc_completion_queue *cq, grpc_metadata_array *initial_metadata, call_data *call_data, void *tag) {
683 abort();
684}
685
686grpc_call_error grpc_server_request_call(
687 grpc_server *server, grpc_completion_queue *cq, grpc_call_details *details,
688 grpc_metadata_array *initial_metadata, void *tag) {
689 grpc_cq_begin_op(cq, NULL, GRPC_IOREQ);
690 return queue_call_request(server, cq, initial_metadata, begin_request, tag);
691}
692
693static void publish_legacy_request(grpc_call *call, grpc_op_error status, void *tag) {
694 grpc_call_element *elem = grpc_call_stack_element(grpc_call_get_call_stack(call), 0);
695 call_data *calld = elem->call_data;
696 channel_data *chand = elem->channel_data;
697 grpc_server *server = chand->server;
698
699 if (status == GRPC_OP_OK) {
700 grpc_cq_end_new_rpc(server->cq, tag, call,
701 do_nothing, NULL, grpc_mdstr_as_c_string(calld->path),
702 grpc_mdstr_as_c_string(calld->host), calld->deadline,
703 calld->legacy->client_metadata.count, calld->legacy->client_metadata.metadata);
704 }
705}
706
707static void begin_legacy_request(grpc_server *server, grpc_completion_queue *cq, grpc_metadata_array *initial_metadata, call_data *calld, void *tag) {
708 grpc_ioreq req;
709 if (!calld) {
710 gpr_free(initial_metadata);
711 grpc_cq_end_new_rpc(cq, tag, NULL, do_nothing, NULL, NULL, NULL,
712 gpr_inf_past, 0, NULL);
713 return;
714 }
715 req.op = GRPC_IOREQ_RECV_INITIAL_METADATA;
716 req.data.recv_metadata = initial_metadata;
717 grpc_call_start_ioreq_and_call_back(calld->call, &req, 1, publish_legacy_request, tag);
718}
719
720grpc_call_error grpc_server_request_call_old(grpc_server *server, void *tag_new) {
721 grpc_metadata_array *client_metadata = gpr_malloc(sizeof(grpc_metadata_array));
722 memset(client_metadata, 0, sizeof(*client_metadata));
723 grpc_cq_begin_op(server->cq, NULL, GRPC_SERVER_RPC_NEW);
724 return queue_call_request(server, server->cq, client_metadata, begin_legacy_request, tag_new);
Nicolas Nobleb7ebd3b2014-11-26 16:33:03 -0800725}
726
727const grpc_channel_args *grpc_server_get_channel_args(grpc_server *server) {
728 return server->channel_args;
729}