Line data Source code
1 : /*
2 : *
3 : * Copyright 2015, 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 <limits.h>
37 : #include <stdlib.h>
38 : #include <string.h>
39 :
40 : #include <grpc/support/alloc.h>
41 : #include <grpc/support/log.h>
42 : #include <grpc/support/string_util.h>
43 : #include <grpc/support/useful.h>
44 :
45 : #include "src/core/census/grpc_filter.h"
46 : #include "src/core/channel/channel_args.h"
47 : #include "src/core/channel/connected_channel.h"
48 : #include "src/core/iomgr/iomgr.h"
49 : #include "src/core/support/stack_lockfree.h"
50 : #include "src/core/support/string.h"
51 : #include "src/core/surface/api_trace.h"
52 : #include "src/core/surface/call.h"
53 : #include "src/core/surface/channel.h"
54 : #include "src/core/surface/completion_queue.h"
55 : #include "src/core/surface/init.h"
56 : #include "src/core/transport/metadata.h"
57 :
58 : typedef struct listener {
59 : void *arg;
60 : void (*start)(grpc_exec_ctx *exec_ctx, grpc_server *server, void *arg,
61 : grpc_pollset **pollsets, size_t pollset_count);
62 : void (*destroy)(grpc_exec_ctx *exec_ctx, grpc_server *server, void *arg,
63 : grpc_closure *closure);
64 : struct listener *next;
65 : grpc_closure destroy_done;
66 : } listener;
67 :
68 : typedef struct call_data call_data;
69 : typedef struct channel_data channel_data;
70 : typedef struct registered_method registered_method;
71 :
72 : typedef struct {
73 : call_data *next;
74 : call_data *prev;
75 : } call_link;
76 :
77 : typedef enum { BATCH_CALL, REGISTERED_CALL } requested_call_type;
78 :
79 : typedef struct requested_call {
80 : requested_call_type type;
81 : void *tag;
82 : grpc_server *server;
83 : grpc_completion_queue *cq_bound_to_call;
84 : grpc_completion_queue *cq_for_notification;
85 : grpc_call **call;
86 : grpc_cq_completion completion;
87 : union {
88 : struct {
89 : grpc_call_details *details;
90 : grpc_metadata_array *initial_metadata;
91 : } batch;
92 : struct {
93 : registered_method *registered_method;
94 : gpr_timespec *deadline;
95 : grpc_metadata_array *initial_metadata;
96 : grpc_byte_buffer **optional_payload;
97 : } registered;
98 : } data;
99 : } requested_call;
100 :
101 : typedef struct channel_registered_method {
102 : registered_method *server_registered_method;
103 : grpc_mdstr *method;
104 : grpc_mdstr *host;
105 : } channel_registered_method;
106 :
107 : struct channel_data {
108 : grpc_server *server;
109 : grpc_connectivity_state connectivity_state;
110 : grpc_channel *channel;
111 : grpc_mdstr *path_key;
112 : grpc_mdstr *authority_key;
113 : /* linked list of all channels on a server */
114 : channel_data *next;
115 : channel_data *prev;
116 : channel_registered_method *registered_methods;
117 : gpr_uint32 registered_method_slots;
118 : gpr_uint32 registered_method_max_probes;
119 : grpc_closure finish_destroy_channel_closure;
120 : grpc_closure channel_connectivity_changed;
121 : };
122 :
123 : typedef struct shutdown_tag {
124 : void *tag;
125 : grpc_completion_queue *cq;
126 : grpc_cq_completion completion;
127 : } shutdown_tag;
128 :
129 : typedef enum {
130 : /* waiting for metadata */
131 : NOT_STARTED,
132 : /* inital metadata read, not flow controlled in yet */
133 : PENDING,
134 : /* flow controlled in, on completion queue */
135 : ACTIVATED,
136 : /* cancelled before being queued */
137 : ZOMBIED
138 : } call_state;
139 :
140 : typedef struct request_matcher request_matcher;
141 :
142 : struct call_data {
143 : grpc_call *call;
144 :
145 : /** protects state */
146 : gpr_mu mu_state;
147 : /** the current state of a call - see call_state */
148 : call_state state;
149 :
150 : grpc_mdstr *path;
151 : grpc_mdstr *host;
152 : gpr_timespec deadline;
153 : int got_initial_metadata;
154 :
155 : grpc_completion_queue *cq_new;
156 :
157 : grpc_stream_op_buffer *recv_ops;
158 : grpc_stream_state *recv_state;
159 : grpc_closure *on_done_recv;
160 :
161 : grpc_closure server_on_recv;
162 : grpc_closure kill_zombie_closure;
163 :
164 : call_data *pending_next;
165 : };
166 :
167 : struct request_matcher {
168 : call_data *pending_head;
169 : call_data *pending_tail;
170 : gpr_stack_lockfree *requests;
171 : };
172 :
173 : struct registered_method {
174 : char *method;
175 : char *host;
176 : request_matcher request_matcher;
177 : registered_method *next;
178 : };
179 :
180 : typedef struct {
181 : grpc_channel **channels;
182 : size_t num_channels;
183 : } channel_broadcaster;
184 :
185 : struct grpc_server {
186 : size_t channel_filter_count;
187 : grpc_channel_filter const **channel_filters;
188 : grpc_channel_args *channel_args;
189 :
190 : grpc_completion_queue **cqs;
191 : grpc_pollset **pollsets;
192 : size_t cq_count;
193 :
194 : /* The two following mutexes control access to server-state
195 : mu_global controls access to non-call-related state (e.g., channel state)
196 : mu_call controls access to call-related state (e.g., the call lists)
197 :
198 : If they are ever required to be nested, you must lock mu_global
199 : before mu_call. This is currently used in shutdown processing
200 : (grpc_server_shutdown_and_notify and maybe_finish_shutdown) */
201 : gpr_mu mu_global; /* mutex for server and channel state */
202 : gpr_mu mu_call; /* mutex for call-specific state */
203 :
204 : registered_method *registered_methods;
205 : request_matcher unregistered_request_matcher;
206 : /** free list of available requested_calls indices */
207 : gpr_stack_lockfree *request_freelist;
208 : /** requested call backing data */
209 : requested_call *requested_calls;
210 : size_t max_requested_calls;
211 :
212 : gpr_atm shutdown_flag;
213 : gpr_uint8 shutdown_published;
214 : size_t num_shutdown_tags;
215 : shutdown_tag *shutdown_tags;
216 :
217 : channel_data root_channel_data;
218 :
219 : listener *listeners;
220 : int listeners_destroyed;
221 : gpr_refcount internal_refcount;
222 :
223 : /** when did we print the last shutdown progress message */
224 : gpr_timespec last_shutdown_message_time;
225 : };
226 :
227 : #define SERVER_FROM_CALL_ELEM(elem) \
228 : (((channel_data *)(elem)->channel_data)->server)
229 :
230 : static void begin_call(grpc_exec_ctx *exec_ctx, grpc_server *server,
231 : call_data *calld, requested_call *rc);
232 : static void fail_call(grpc_exec_ctx *exec_ctx, grpc_server *server,
233 : requested_call *rc);
234 : /* Before calling maybe_finish_shutdown, we must hold mu_global and not
235 : hold mu_call */
236 : static void maybe_finish_shutdown(grpc_exec_ctx *exec_ctx, grpc_server *server);
237 :
238 : /*
239 : * channel broadcaster
240 : */
241 :
242 : /* assumes server locked */
243 2285 : static void channel_broadcaster_init(grpc_server *s, channel_broadcaster *cb) {
244 : channel_data *c;
245 2285 : size_t count = 0;
246 4081 : for (c = s->root_channel_data.next; c != &s->root_channel_data; c = c->next) {
247 1796 : count++;
248 : }
249 2285 : cb->num_channels = count;
250 2285 : cb->channels = gpr_malloc(sizeof(*cb->channels) * cb->num_channels);
251 2285 : count = 0;
252 4081 : for (c = s->root_channel_data.next; c != &s->root_channel_data; c = c->next) {
253 1796 : cb->channels[count++] = c->channel;
254 1796 : GRPC_CHANNEL_INTERNAL_REF(c->channel, "broadcast");
255 : }
256 2285 : }
257 :
258 : struct shutdown_cleanup_args {
259 : grpc_closure closure;
260 : gpr_slice slice;
261 : };
262 :
263 1796 : static void shutdown_cleanup(grpc_exec_ctx *exec_ctx, void *arg,
264 : int iomgr_status_ignored) {
265 1796 : struct shutdown_cleanup_args *a = arg;
266 1796 : gpr_slice_unref(a->slice);
267 1796 : gpr_free(a);
268 1796 : }
269 :
270 1796 : static void send_shutdown(grpc_exec_ctx *exec_ctx, grpc_channel *channel,
271 : int send_goaway, int send_disconnect) {
272 : grpc_transport_op op;
273 : struct shutdown_cleanup_args *sc;
274 : grpc_channel_element *elem;
275 :
276 1796 : memset(&op, 0, sizeof(op));
277 1796 : op.send_goaway = send_goaway;
278 1796 : sc = gpr_malloc(sizeof(*sc));
279 1796 : sc->slice = gpr_slice_from_copied_string("Server shutdown");
280 1796 : op.goaway_message = &sc->slice;
281 1796 : op.goaway_status = GRPC_STATUS_OK;
282 1796 : op.disconnect = send_disconnect;
283 1796 : grpc_closure_init(&sc->closure, shutdown_cleanup, sc);
284 1796 : op.on_consumed = &sc->closure;
285 :
286 1796 : elem = grpc_channel_stack_element(grpc_channel_get_channel_stack(channel), 0);
287 1796 : elem->filter->start_transport_op(exec_ctx, elem, &op);
288 1796 : }
289 :
290 2285 : static void channel_broadcaster_shutdown(grpc_exec_ctx *exec_ctx,
291 : channel_broadcaster *cb,
292 : int send_goaway,
293 : int force_disconnect) {
294 : size_t i;
295 :
296 4081 : for (i = 0; i < cb->num_channels; i++) {
297 1796 : send_shutdown(exec_ctx, cb->channels[i], send_goaway, force_disconnect);
298 1796 : GRPC_CHANNEL_INTERNAL_UNREF(exec_ctx, cb->channels[i], "broadcast");
299 : }
300 2285 : gpr_free(cb->channels);
301 2285 : }
302 :
303 : /*
304 : * request_matcher
305 : */
306 :
307 3475 : static void request_matcher_init(request_matcher *rm, size_t entries) {
308 3475 : memset(rm, 0, sizeof(*rm));
309 3475 : rm->requests = gpr_stack_lockfree_create(entries);
310 3475 : }
311 :
312 3475 : static void request_matcher_destroy(request_matcher *rm) {
313 3475 : GPR_ASSERT(gpr_stack_lockfree_pop(rm->requests) == -1);
314 3475 : gpr_stack_lockfree_destroy(rm->requests);
315 3475 : }
316 :
317 142 : static void kill_zombie(grpc_exec_ctx *exec_ctx, void *elem, int success) {
318 142 : grpc_call_destroy(grpc_call_from_top_element(elem));
319 142 : }
320 :
321 14967 : static void request_matcher_zombify_all_pending_calls(grpc_exec_ctx *exec_ctx,
322 : request_matcher *rm) {
323 30069 : while (rm->pending_head) {
324 135 : call_data *calld = rm->pending_head;
325 135 : rm->pending_head = calld->pending_next;
326 135 : gpr_mu_lock(&calld->mu_state);
327 135 : calld->state = ZOMBIED;
328 135 : gpr_mu_unlock(&calld->mu_state);
329 135 : grpc_closure_init(
330 : &calld->kill_zombie_closure, kill_zombie,
331 135 : grpc_call_stack_element(grpc_call_get_call_stack(calld->call), 0));
332 135 : grpc_exec_ctx_enqueue(exec_ctx, &calld->kill_zombie_closure, 1);
333 : }
334 14967 : }
335 :
336 14967 : static void request_matcher_kill_requests(grpc_exec_ctx *exec_ctx,
337 : grpc_server *server,
338 : request_matcher *rm) {
339 : int request_id;
340 111472 : while ((request_id = gpr_stack_lockfree_pop(rm->requests)) != -1) {
341 81538 : fail_call(exec_ctx, server, &server->requested_calls[request_id]);
342 : }
343 14967 : }
344 :
345 : /*
346 : * server proper
347 : */
348 :
349 2716476 : static void server_ref(grpc_server *server) {
350 2716476 : gpr_ref(&server->internal_refcount);
351 2717289 : }
352 :
353 2261 : static void server_delete(grpc_exec_ctx *exec_ctx, grpc_server *server) {
354 : registered_method *rm;
355 : size_t i;
356 2261 : grpc_channel_args_destroy(server->channel_args);
357 2261 : gpr_mu_destroy(&server->mu_global);
358 2261 : gpr_mu_destroy(&server->mu_call);
359 2261 : gpr_free((void *)server->channel_filters);
360 5736 : while ((rm = server->registered_methods) != NULL) {
361 1214 : server->registered_methods = rm->next;
362 1214 : request_matcher_destroy(&rm->request_matcher);
363 1214 : gpr_free(rm->method);
364 1214 : gpr_free(rm->host);
365 1214 : gpr_free(rm);
366 : }
367 4566 : for (i = 0; i < server->cq_count; i++) {
368 2305 : GRPC_CQ_INTERNAL_UNREF(server->cqs[i], "server");
369 : }
370 2261 : request_matcher_destroy(&server->unregistered_request_matcher);
371 2261 : gpr_stack_lockfree_destroy(server->request_freelist);
372 2261 : gpr_free(server->cqs);
373 2261 : gpr_free(server->pollsets);
374 2261 : gpr_free(server->shutdown_tags);
375 2261 : gpr_free(server->requested_calls);
376 2261 : gpr_free(server);
377 2261 : }
378 :
379 2719469 : static void server_unref(grpc_exec_ctx *exec_ctx, grpc_server *server) {
380 2719469 : if (gpr_unref(&server->internal_refcount)) {
381 2261 : server_delete(exec_ctx, server);
382 : }
383 2720043 : }
384 :
385 2022 : static int is_channel_orphaned(channel_data *chand) {
386 2022 : return chand->next == chand;
387 : }
388 :
389 2022 : static void orphan_channel(channel_data *chand) {
390 2022 : chand->next->prev = chand->prev;
391 2022 : chand->prev->next = chand->next;
392 2022 : chand->next = chand->prev = chand;
393 2022 : }
394 :
395 2022 : static void finish_destroy_channel(grpc_exec_ctx *exec_ctx, void *cd,
396 : int success) {
397 2022 : channel_data *chand = cd;
398 2022 : grpc_server *server = chand->server;
399 2022 : gpr_log(GPR_DEBUG, "finish_destroy_channel: %p", chand->channel);
400 2022 : GRPC_CHANNEL_INTERNAL_UNREF(exec_ctx, chand->channel, "server");
401 2022 : server_unref(exec_ctx, server);
402 2022 : }
403 :
404 2022 : static void destroy_channel(grpc_exec_ctx *exec_ctx, channel_data *chand) {
405 4044 : if (is_channel_orphaned(chand)) return;
406 2022 : GPR_ASSERT(chand->server != NULL);
407 2022 : orphan_channel(chand);
408 2022 : server_ref(chand->server);
409 2022 : maybe_finish_shutdown(exec_ctx, chand->server);
410 2022 : chand->finish_destroy_channel_closure.cb = finish_destroy_channel;
411 2022 : chand->finish_destroy_channel_closure.cb_arg = chand;
412 2022 : grpc_exec_ctx_enqueue(exec_ctx, &chand->finish_destroy_channel_closure, 1);
413 : }
414 :
415 1300602 : static void finish_start_new_rpc(grpc_exec_ctx *exec_ctx, grpc_server *server,
416 : grpc_call_element *elem, request_matcher *rm) {
417 1300602 : call_data *calld = elem->call_data;
418 : int request_id;
419 :
420 1300602 : if (gpr_atm_acq_load(&server->shutdown_flag)) {
421 0 : gpr_mu_lock(&calld->mu_state);
422 0 : calld->state = ZOMBIED;
423 0 : gpr_mu_unlock(&calld->mu_state);
424 0 : grpc_closure_init(&calld->kill_zombie_closure, kill_zombie, elem);
425 0 : grpc_exec_ctx_enqueue(exec_ctx, &calld->kill_zombie_closure, 1);
426 1301870 : return;
427 : }
428 :
429 1300602 : request_id = gpr_stack_lockfree_pop(rm->requests);
430 1301873 : if (request_id == -1) {
431 274601 : gpr_mu_lock(&server->mu_call);
432 274612 : gpr_mu_lock(&calld->mu_state);
433 274612 : calld->state = PENDING;
434 274612 : gpr_mu_unlock(&calld->mu_state);
435 274612 : if (rm->pending_head == NULL) {
436 1636 : rm->pending_tail = rm->pending_head = calld;
437 : } else {
438 272976 : rm->pending_tail->pending_next = calld;
439 272976 : rm->pending_tail = calld;
440 : }
441 274612 : calld->pending_next = NULL;
442 274612 : gpr_mu_unlock(&server->mu_call);
443 : } else {
444 1027272 : gpr_mu_lock(&calld->mu_state);
445 1027271 : calld->state = ACTIVATED;
446 1027271 : gpr_mu_unlock(&calld->mu_state);
447 1027265 : begin_call(exec_ctx, server, calld, &server->requested_calls[request_id]);
448 : }
449 : }
450 :
451 1301166 : static void start_new_rpc(grpc_exec_ctx *exec_ctx, grpc_call_element *elem) {
452 1301166 : channel_data *chand = elem->channel_data;
453 1301166 : call_data *calld = elem->call_data;
454 1301166 : grpc_server *server = chand->server;
455 : gpr_uint32 i;
456 : gpr_uint32 hash;
457 : channel_registered_method *rm;
458 :
459 1301166 : if (chand->registered_methods && calld->path && calld->host) {
460 : /* TODO(ctiller): unify these two searches */
461 : /* check for an exact match with host */
462 1298904 : hash = GRPC_MDSTR_KV_HASH(calld->host->hash, calld->path->hash);
463 2904827 : for (i = 0; i <= chand->registered_method_max_probes; i++) {
464 3211838 : rm = &chand->registered_methods[(hash + i) %
465 1605919 : chand->registered_method_slots];
466 1605919 : if (!rm) break;
467 1605924 : if (rm->host != calld->host) continue;
468 1 : if (rm->method != calld->path) continue;
469 1 : finish_start_new_rpc(exec_ctx, server, elem,
470 1 : &rm->server_registered_method->request_matcher);
471 1 : return;
472 : }
473 : /* check for a wildcard method definition (no host set) */
474 1298903 : hash = GRPC_MDSTR_KV_HASH(0, calld->path->hash);
475 1504760 : for (i = 0; i <= chand->registered_method_max_probes; i++) {
476 3010014 : rm = &chand->registered_methods[(hash + i) %
477 1505007 : chand->registered_method_slots];
478 1505007 : if (!rm) break;
479 1505008 : if (rm->host != NULL) continue;
480 1504888 : if (rm->method != calld->path) continue;
481 1299151 : finish_start_new_rpc(exec_ctx, server, elem,
482 1299151 : &rm->server_registered_method->request_matcher);
483 1299148 : return;
484 : }
485 : }
486 2014 : finish_start_new_rpc(exec_ctx, server, elem,
487 : &server->unregistered_request_matcher);
488 : }
489 :
490 7283 : static int num_listeners(grpc_server *server) {
491 : listener *l;
492 7283 : int n = 0;
493 13590 : for (l = server->listeners; l; l = l->next) {
494 6307 : n++;
495 : }
496 7283 : return n;
497 : }
498 :
499 2261 : static void done_shutdown_event(grpc_exec_ctx *exec_ctx, void *server,
500 : grpc_cq_completion *completion) {
501 2261 : server_unref(exec_ctx, server);
502 2261 : }
503 :
504 0 : static int num_channels(grpc_server *server) {
505 : channel_data *chand;
506 0 : int n = 0;
507 0 : for (chand = server->root_channel_data.next;
508 0 : chand != &server->root_channel_data; chand = chand->next) {
509 0 : n++;
510 : }
511 0 : return n;
512 : }
513 :
514 9221 : static void kill_pending_work_locked(grpc_exec_ctx *exec_ctx,
515 : grpc_server *server) {
516 : registered_method *rm;
517 9221 : request_matcher_kill_requests(exec_ctx, server,
518 : &server->unregistered_request_matcher);
519 9221 : request_matcher_zombify_all_pending_calls(
520 : exec_ctx, &server->unregistered_request_matcher);
521 14967 : for (rm = server->registered_methods; rm; rm = rm->next) {
522 5746 : request_matcher_kill_requests(exec_ctx, server, &rm->request_matcher);
523 5746 : request_matcher_zombify_all_pending_calls(exec_ctx, &rm->request_matcher);
524 : }
525 9221 : }
526 :
527 8078 : static void maybe_finish_shutdown(grpc_exec_ctx *exec_ctx,
528 : grpc_server *server) {
529 : size_t i;
530 8078 : if (!gpr_atm_acq_load(&server->shutdown_flag) || server->shutdown_published) {
531 1118 : return;
532 : }
533 :
534 6960 : kill_pending_work_locked(exec_ctx, server);
535 :
536 11982 : if (server->root_channel_data.next != &server->root_channel_data ||
537 5022 : server->listeners_destroyed < num_listeners(server)) {
538 4699 : if (gpr_time_cmp(gpr_time_sub(gpr_now(GPR_CLOCK_REALTIME),
539 : server->last_shutdown_message_time),
540 : gpr_time_from_seconds(1, GPR_TIMESPAN)) >= 0) {
541 0 : server->last_shutdown_message_time = gpr_now(GPR_CLOCK_REALTIME);
542 0 : gpr_log(GPR_DEBUG,
543 : "Waiting for %d channels and %d/%d listeners to be destroyed"
544 : " before shutting down server",
545 : num_channels(server),
546 0 : num_listeners(server) - server->listeners_destroyed,
547 : num_listeners(server));
548 : }
549 4699 : return;
550 : }
551 2261 : server->shutdown_published = 1;
552 4522 : for (i = 0; i < server->num_shutdown_tags; i++) {
553 2261 : server_ref(server);
554 4522 : grpc_cq_end_op(exec_ctx, server->shutdown_tags[i].cq,
555 2261 : server->shutdown_tags[i].tag, 1, done_shutdown_event, server,
556 2261 : &server->shutdown_tags[i].completion);
557 : }
558 : }
559 :
560 6507919 : static grpc_mdelem *server_filter(void *user_data, grpc_mdelem *md) {
561 6507919 : grpc_call_element *elem = user_data;
562 6507919 : channel_data *chand = elem->channel_data;
563 6507919 : call_data *calld = elem->call_data;
564 6507919 : if (md->key == chand->path_key) {
565 1301796 : calld->path = GRPC_MDSTR_REF(md->value);
566 1301879 : return NULL;
567 5206123 : } else if (md->key == chand->authority_key) {
568 1301740 : calld->host = GRPC_MDSTR_REF(md->value);
569 1301875 : return NULL;
570 : }
571 3904383 : return md;
572 : }
573 :
574 3219986 : static void server_on_recv(grpc_exec_ctx *exec_ctx, void *ptr, int success) {
575 3219986 : grpc_call_element *elem = ptr;
576 3219986 : call_data *calld = elem->call_data;
577 : gpr_timespec op_deadline;
578 :
579 3219986 : if (success && !calld->got_initial_metadata) {
580 : size_t i;
581 1301820 : size_t nops = calld->recv_ops->nops;
582 1301820 : grpc_stream_op *ops = calld->recv_ops->ops;
583 2603640 : for (i = 0; i < nops; i++) {
584 1301823 : grpc_stream_op *op = &ops[i];
585 1301823 : if (op->type != GRPC_OP_METADATA) continue;
586 1301823 : grpc_metadata_batch_filter(&op->data.metadata, server_filter, elem);
587 1301736 : op_deadline = op->data.metadata.deadline;
588 1301889 : if (0 !=
589 1301736 : gpr_time_cmp(op_deadline, gpr_inf_future(op_deadline.clock_type))) {
590 2698 : calld->deadline = op->data.metadata.deadline;
591 : }
592 1301889 : if (calld->host && calld->path) {
593 1301882 : calld->got_initial_metadata = 1;
594 1301882 : start_new_rpc(exec_ctx, elem);
595 : }
596 1301878 : break;
597 : }
598 : }
599 :
600 3220041 : switch (*calld->recv_state) {
601 : case GRPC_STREAM_OPEN:
602 609491 : break;
603 : case GRPC_STREAM_SEND_CLOSED:
604 0 : break;
605 : case GRPC_STREAM_RECV_CLOSED:
606 1310315 : gpr_mu_lock(&calld->mu_state);
607 1310303 : if (calld->state == NOT_STARTED) {
608 0 : calld->state = ZOMBIED;
609 0 : gpr_mu_unlock(&calld->mu_state);
610 0 : grpc_closure_init(&calld->kill_zombie_closure, kill_zombie, elem);
611 0 : grpc_exec_ctx_enqueue(exec_ctx, &calld->kill_zombie_closure, 1);
612 : } else {
613 1310303 : gpr_mu_unlock(&calld->mu_state);
614 : }
615 1310296 : break;
616 : case GRPC_STREAM_CLOSED:
617 1301855 : gpr_mu_lock(&calld->mu_state);
618 1301882 : if (calld->state == NOT_STARTED) {
619 7 : calld->state = ZOMBIED;
620 7 : gpr_mu_unlock(&calld->mu_state);
621 7 : grpc_closure_init(&calld->kill_zombie_closure, kill_zombie, elem);
622 7 : grpc_exec_ctx_enqueue(exec_ctx, &calld->kill_zombie_closure, 1);
623 1301875 : } else if (calld->state == PENDING) {
624 101 : calld->state = ZOMBIED;
625 101 : gpr_mu_unlock(&calld->mu_state);
626 : /* zombied call will be destroyed when it's removed from the pending
627 : queue... later */
628 : } else {
629 1301774 : gpr_mu_unlock(&calld->mu_state);
630 : }
631 1301878 : break;
632 : }
633 :
634 3220045 : calld->on_done_recv->cb(exec_ctx, calld->on_done_recv->cb_arg, success);
635 3221616 : }
636 :
637 6208287 : static void server_mutate_op(grpc_call_element *elem,
638 : grpc_transport_stream_op *op) {
639 6208287 : call_data *calld = elem->call_data;
640 :
641 6208287 : if (op->recv_ops) {
642 : /* substitute our callback for the higher callback */
643 3221011 : calld->recv_ops = op->recv_ops;
644 3221011 : calld->recv_state = op->recv_state;
645 3221011 : calld->on_done_recv = op->on_done_recv;
646 3221011 : op->on_done_recv = &calld->server_on_recv;
647 : }
648 6208287 : }
649 :
650 4906764 : static void server_start_transport_stream_op(grpc_exec_ctx *exec_ctx,
651 : grpc_call_element *elem,
652 : grpc_transport_stream_op *op) {
653 4906764 : GRPC_CALL_LOG_OP(GPR_INFO, elem, op);
654 4906764 : server_mutate_op(elem, op);
655 4908922 : grpc_call_next_op(exec_ctx, elem, op);
656 4910900 : }
657 :
658 1301771 : static void accept_stream(void *cd, grpc_transport *transport,
659 : const void *transport_server_data) {
660 1301771 : channel_data *chand = cd;
661 : /* create a call */
662 1301771 : grpc_call_create(chand->channel, NULL, 0, NULL, transport_server_data, NULL,
663 : 0, gpr_inf_future(GPR_CLOCK_MONOTONIC));
664 1301875 : }
665 :
666 4044 : static void channel_connectivity_changed(grpc_exec_ctx *exec_ctx, void *cd,
667 : int iomgr_status_ignored) {
668 4044 : channel_data *chand = cd;
669 4044 : grpc_server *server = chand->server;
670 4044 : if (chand->connectivity_state != GRPC_CHANNEL_FATAL_FAILURE) {
671 : grpc_transport_op op;
672 2022 : memset(&op, 0, sizeof(op));
673 2022 : op.on_connectivity_state_change = &chand->channel_connectivity_changed,
674 2022 : op.connectivity_state = &chand->connectivity_state;
675 2022 : grpc_channel_next_op(exec_ctx,
676 : grpc_channel_stack_element(
677 : grpc_channel_get_channel_stack(chand->channel), 0),
678 : &op);
679 : } else {
680 2022 : gpr_mu_lock(&server->mu_global);
681 2022 : destroy_channel(exec_ctx, chand);
682 2022 : gpr_mu_unlock(&server->mu_global);
683 2022 : GRPC_CHANNEL_INTERNAL_UNREF(exec_ctx, chand->channel, "connectivity");
684 : }
685 4044 : }
686 :
687 1301829 : static void init_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
688 : const void *server_transport_data,
689 : grpc_transport_stream_op *initial_op) {
690 1301829 : call_data *calld = elem->call_data;
691 1301829 : channel_data *chand = elem->channel_data;
692 1301829 : memset(calld, 0, sizeof(call_data));
693 1301829 : calld->deadline = gpr_inf_future(GPR_CLOCK_REALTIME);
694 1301879 : calld->call = grpc_call_from_top_element(elem);
695 1301869 : gpr_mu_init(&calld->mu_state);
696 :
697 1301862 : grpc_closure_init(&calld->server_on_recv, server_on_recv, elem);
698 :
699 1301852 : server_ref(chand->server);
700 :
701 1301875 : if (initial_op) server_mutate_op(elem, initial_op);
702 1301871 : }
703 :
704 1301726 : static void destroy_call_elem(grpc_exec_ctx *exec_ctx,
705 : grpc_call_element *elem) {
706 1301726 : channel_data *chand = elem->channel_data;
707 1301726 : call_data *calld = elem->call_data;
708 :
709 1301726 : GPR_ASSERT(calld->state != PENDING);
710 :
711 1301726 : if (calld->host) {
712 1301743 : GRPC_MDSTR_UNREF(calld->host);
713 : }
714 1301882 : if (calld->path) {
715 1301875 : GRPC_MDSTR_UNREF(calld->path);
716 : }
717 :
718 1301844 : gpr_mu_destroy(&calld->mu_state);
719 :
720 1301826 : server_unref(exec_ctx, chand->server);
721 1301885 : }
722 :
723 2022 : static void init_channel_elem(grpc_exec_ctx *exec_ctx,
724 : grpc_channel_element *elem, grpc_channel *master,
725 : const grpc_channel_args *args,
726 : grpc_mdctx *metadata_context, int is_first,
727 : int is_last) {
728 2022 : channel_data *chand = elem->channel_data;
729 2022 : GPR_ASSERT(is_first);
730 2022 : GPR_ASSERT(!is_last);
731 2022 : chand->server = NULL;
732 2022 : chand->channel = NULL;
733 2022 : chand->path_key = grpc_mdstr_from_string(metadata_context, ":path");
734 2022 : chand->authority_key = grpc_mdstr_from_string(metadata_context, ":authority");
735 2022 : chand->next = chand->prev = chand;
736 2022 : chand->registered_methods = NULL;
737 2022 : chand->connectivity_state = GRPC_CHANNEL_IDLE;
738 2022 : grpc_closure_init(&chand->channel_connectivity_changed,
739 : channel_connectivity_changed, chand);
740 2022 : }
741 :
742 2022 : static void destroy_channel_elem(grpc_exec_ctx *exec_ctx,
743 : grpc_channel_element *elem) {
744 : size_t i;
745 2022 : channel_data *chand = elem->channel_data;
746 2022 : if (chand->registered_methods) {
747 2487 : for (i = 0; i < chand->registered_method_slots; i++) {
748 2324 : if (chand->registered_methods[i].method) {
749 1162 : GRPC_MDSTR_UNREF(chand->registered_methods[i].method);
750 : }
751 2324 : if (chand->registered_methods[i].host) {
752 365 : GRPC_MDSTR_UNREF(chand->registered_methods[i].host);
753 : }
754 : }
755 163 : gpr_free(chand->registered_methods);
756 : }
757 2022 : if (chand->server) {
758 2022 : gpr_mu_lock(&chand->server->mu_global);
759 2022 : chand->next->prev = chand->prev;
760 2022 : chand->prev->next = chand->next;
761 2022 : chand->next = chand->prev = chand;
762 2022 : maybe_finish_shutdown(exec_ctx, chand->server);
763 2022 : gpr_mu_unlock(&chand->server->mu_global);
764 2022 : GRPC_MDSTR_UNREF(chand->path_key);
765 2022 : GRPC_MDSTR_UNREF(chand->authority_key);
766 2022 : server_unref(exec_ctx, chand->server);
767 : }
768 2022 : }
769 :
770 : static const grpc_channel_filter server_surface_filter = {
771 : server_start_transport_stream_op, grpc_channel_next_op, sizeof(call_data),
772 : init_call_elem, destroy_call_elem, sizeof(channel_data), init_channel_elem,
773 : destroy_channel_elem, grpc_call_next_get_peer, "server",
774 : };
775 :
776 2305 : void grpc_server_register_completion_queue(grpc_server *server,
777 : grpc_completion_queue *cq,
778 : void *reserved) {
779 : size_t i, n;
780 2305 : GRPC_API_TRACE(
781 : "grpc_server_register_completion_queue(server=%p, cq=%p, reserved=%p)", 3,
782 : (server, cq, reserved));
783 2305 : GPR_ASSERT(!reserved);
784 2383 : for (i = 0; i < server->cq_count; i++) {
785 2383 : if (server->cqs[i] == cq) return;
786 : }
787 2305 : GRPC_CQ_INTERNAL_REF(cq, "server");
788 2305 : grpc_cq_mark_server_cq(cq);
789 2305 : n = server->cq_count++;
790 2305 : server->cqs = gpr_realloc(server->cqs,
791 2305 : server->cq_count * sizeof(grpc_completion_queue *));
792 2305 : server->cqs[n] = cq;
793 : }
794 :
795 2261 : grpc_server *grpc_server_create_from_filters(
796 : const grpc_channel_filter **filters, size_t filter_count,
797 : const grpc_channel_args *args) {
798 : size_t i;
799 : /* TODO(census): restore this once we finalize census filter etc.
800 : int census_enabled = grpc_channel_args_is_census_enabled(args); */
801 2261 : int census_enabled = 0;
802 :
803 2261 : grpc_server *server = gpr_malloc(sizeof(grpc_server));
804 :
805 2261 : GPR_ASSERT(grpc_is_initialized() && "call grpc_init()");
806 :
807 2261 : memset(server, 0, sizeof(grpc_server));
808 :
809 2261 : gpr_mu_init(&server->mu_global);
810 2261 : gpr_mu_init(&server->mu_call);
811 :
812 : /* decremented by grpc_server_destroy */
813 2261 : gpr_ref_init(&server->internal_refcount, 1);
814 2261 : server->root_channel_data.next = server->root_channel_data.prev =
815 2261 : &server->root_channel_data;
816 :
817 : /* TODO(ctiller): expose a channel_arg for this */
818 2261 : server->max_requested_calls = 32768;
819 2261 : server->request_freelist =
820 2261 : gpr_stack_lockfree_create(server->max_requested_calls);
821 74090709 : for (i = 0; i < (size_t)server->max_requested_calls; i++) {
822 74088448 : gpr_stack_lockfree_push(server->request_freelist, (int)i);
823 : }
824 2261 : request_matcher_init(&server->unregistered_request_matcher,
825 : server->max_requested_calls);
826 2261 : server->requested_calls = gpr_malloc(server->max_requested_calls *
827 : sizeof(*server->requested_calls));
828 :
829 : /* Server filter stack is:
830 :
831 : server_surface_filter - for making surface API calls
832 : grpc_server_census_filter (optional) - for stats collection and tracing
833 : {passed in filter stack}
834 : grpc_connected_channel_filter - for interfacing with transports */
835 2261 : server->channel_filter_count = filter_count + 1u + (census_enabled ? 1u : 0u);
836 2261 : server->channel_filters =
837 2261 : gpr_malloc(server->channel_filter_count * sizeof(grpc_channel_filter *));
838 2261 : server->channel_filters[0] = &server_surface_filter;
839 2261 : if (census_enabled) {
840 0 : server->channel_filters[1] = &grpc_server_census_filter;
841 : }
842 4034 : for (i = 0; i < filter_count; i++) {
843 1773 : server->channel_filters[i + 1u + (census_enabled ? 1u : 0u)] = filters[i];
844 : }
845 :
846 2261 : server->channel_args = grpc_channel_args_copy(args);
847 :
848 2261 : return server;
849 : }
850 :
851 5458 : static int streq(const char *a, const char *b) {
852 5458 : if (a == NULL && b == NULL) return 1;
853 5458 : if (a == NULL) return 0;
854 5053 : if (b == NULL) return 0;
855 5053 : return 0 == strcmp(a, b);
856 : }
857 :
858 1214 : void *grpc_server_register_method(grpc_server *server, const char *method,
859 : const char *host) {
860 : registered_method *m;
861 1214 : GRPC_API_TRACE("grpc_server_register_method(server=%p, method=%s, host=%s)",
862 : 3, (server, method, host));
863 1214 : if (!method) {
864 0 : gpr_log(GPR_ERROR,
865 : "grpc_server_register_method method string cannot be NULL");
866 0 : return NULL;
867 : }
868 6267 : for (m = server->registered_methods; m; m = m->next) {
869 5053 : if (streq(m->method, method) && streq(m->host, host)) {
870 0 : gpr_log(GPR_ERROR, "duplicate registration for %s@%s", method,
871 : host ? host : "*");
872 0 : return NULL;
873 : }
874 : }
875 1214 : m = gpr_malloc(sizeof(registered_method));
876 1214 : memset(m, 0, sizeof(*m));
877 1214 : request_matcher_init(&m->request_matcher, server->max_requested_calls);
878 1214 : m->method = gpr_strdup(method);
879 1214 : m->host = gpr_strdup(host);
880 1214 : m->next = server->registered_methods;
881 1214 : server->registered_methods = m;
882 1214 : return m;
883 : }
884 :
885 2261 : void grpc_server_start(grpc_server *server) {
886 : listener *l;
887 : size_t i;
888 2261 : grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
889 :
890 2261 : GRPC_API_TRACE("grpc_server_start(server=%p)", 1, (server));
891 :
892 2261 : server->pollsets = gpr_malloc(sizeof(grpc_pollset *) * server->cq_count);
893 4566 : for (i = 0; i < server->cq_count; i++) {
894 2305 : server->pollsets[i] = grpc_cq_pollset(server->cqs[i]);
895 : }
896 :
897 4034 : for (l = server->listeners; l; l = l->next) {
898 1773 : l->start(&exec_ctx, server, l->arg, server->pollsets, server->cq_count);
899 : }
900 :
901 2261 : grpc_exec_ctx_finish(&exec_ctx);
902 2261 : }
903 :
904 2022 : void grpc_server_setup_transport(grpc_exec_ctx *exec_ctx, grpc_server *s,
905 : grpc_transport *transport,
906 : grpc_channel_filter const **extra_filters,
907 : size_t num_extra_filters, grpc_mdctx *mdctx,
908 : const grpc_channel_args *args) {
909 2022 : size_t num_filters = s->channel_filter_count + num_extra_filters + 1;
910 2022 : grpc_channel_filter const **filters =
911 2022 : gpr_malloc(sizeof(grpc_channel_filter *) * num_filters);
912 : size_t i;
913 : size_t num_registered_methods;
914 : size_t alloc;
915 : registered_method *rm;
916 : channel_registered_method *crm;
917 : grpc_channel *channel;
918 : channel_data *chand;
919 : grpc_mdstr *host;
920 : grpc_mdstr *method;
921 : gpr_uint32 hash;
922 : size_t slots;
923 : gpr_uint32 probes;
924 2022 : gpr_uint32 max_probes = 0;
925 : grpc_transport_op op;
926 :
927 5578 : for (i = 0; i < s->channel_filter_count; i++) {
928 3556 : filters[i] = s->channel_filters[i];
929 : }
930 4484 : for (; i < s->channel_filter_count + num_extra_filters; i++) {
931 2462 : filters[i] = extra_filters[i - s->channel_filter_count];
932 : }
933 2022 : filters[i] = &grpc_connected_channel_filter;
934 :
935 4171 : for (i = 0; i < s->cq_count; i++) {
936 2149 : memset(&op, 0, sizeof(op));
937 2149 : op.bind_pollset = grpc_cq_pollset(s->cqs[i]);
938 2149 : grpc_transport_perform_op(exec_ctx, transport, &op);
939 : }
940 :
941 2022 : channel = grpc_channel_create_from_filters(exec_ctx, NULL, filters,
942 : num_filters, args, mdctx, 0);
943 2022 : chand = (channel_data *)grpc_channel_stack_element(
944 : grpc_channel_get_channel_stack(channel), 0)->channel_data;
945 2022 : chand->server = s;
946 2022 : server_ref(s);
947 2022 : chand->channel = channel;
948 :
949 2022 : num_registered_methods = 0;
950 3184 : for (rm = s->registered_methods; rm; rm = rm->next) {
951 1162 : num_registered_methods++;
952 : }
953 : /* build a lookup table phrased in terms of mdstr's in this channels context
954 : to quickly find registered methods */
955 2022 : if (num_registered_methods > 0) {
956 163 : slots = 2 * num_registered_methods;
957 163 : alloc = sizeof(channel_registered_method) * slots;
958 163 : chand->registered_methods = gpr_malloc(alloc);
959 163 : memset(chand->registered_methods, 0, alloc);
960 1325 : for (rm = s->registered_methods; rm; rm = rm->next) {
961 1162 : host = rm->host ? grpc_mdstr_from_string(mdctx, rm->host) : NULL;
962 1162 : method = grpc_mdstr_from_string(mdctx, rm->method);
963 1162 : hash = GRPC_MDSTR_KV_HASH(host ? host->hash : 0, method->hash);
964 4136 : for (probes = 0; chand->registered_methods[(hash + probes) % slots]
965 1487 : .server_registered_method != NULL;
966 325 : probes++)
967 : ;
968 1162 : if (probes > max_probes) max_probes = probes;
969 1162 : crm = &chand->registered_methods[(hash + probes) % slots];
970 1162 : crm->server_registered_method = rm;
971 1162 : crm->host = host;
972 1162 : crm->method = method;
973 : }
974 163 : GPR_ASSERT(slots <= GPR_UINT32_MAX);
975 163 : chand->registered_method_slots = (gpr_uint32)slots;
976 163 : chand->registered_method_max_probes = max_probes;
977 : }
978 :
979 2022 : grpc_connected_channel_bind_transport(grpc_channel_get_channel_stack(channel),
980 : transport);
981 :
982 2022 : gpr_mu_lock(&s->mu_global);
983 2022 : chand->next = &s->root_channel_data;
984 2022 : chand->prev = chand->next->prev;
985 2022 : chand->next->prev = chand->prev->next = chand;
986 2022 : gpr_mu_unlock(&s->mu_global);
987 :
988 2022 : gpr_free((void *)filters);
989 :
990 2022 : GRPC_CHANNEL_INTERNAL_REF(channel, "connectivity");
991 2022 : memset(&op, 0, sizeof(op));
992 2022 : op.set_accept_stream = accept_stream;
993 2022 : op.set_accept_stream_user_data = chand;
994 2022 : op.on_connectivity_state_change = &chand->channel_connectivity_changed;
995 2022 : op.connectivity_state = &chand->connectivity_state;
996 2022 : op.disconnect = gpr_atm_acq_load(&s->shutdown_flag) != 0;
997 2022 : grpc_transport_perform_op(exec_ctx, transport, &op);
998 2022 : }
999 :
1000 0 : void done_published_shutdown(grpc_exec_ctx *exec_ctx, void *done_arg,
1001 : grpc_cq_completion *storage) {
1002 : (void)done_arg;
1003 0 : gpr_free(storage);
1004 0 : }
1005 :
1006 1773 : static void listener_destroy_done(grpc_exec_ctx *exec_ctx, void *s,
1007 : int success) {
1008 1773 : grpc_server *server = s;
1009 1773 : gpr_mu_lock(&server->mu_global);
1010 1773 : server->listeners_destroyed++;
1011 1773 : maybe_finish_shutdown(exec_ctx, server);
1012 1773 : gpr_mu_unlock(&server->mu_global);
1013 1773 : }
1014 :
1015 2261 : void grpc_server_shutdown_and_notify(grpc_server *server,
1016 : grpc_completion_queue *cq, void *tag) {
1017 : listener *l;
1018 : shutdown_tag *sdt;
1019 : channel_broadcaster broadcaster;
1020 2261 : grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
1021 :
1022 2261 : GRPC_API_TRACE("grpc_server_shutdown_and_notify(server=%p, cq=%p, tag=%p)", 3,
1023 : (server, cq, tag));
1024 :
1025 2261 : GRPC_SERVER_LOG_SHUTDOWN(GPR_INFO, server, cq, tag);
1026 :
1027 : /* lock, and gather up some stuff to do */
1028 2261 : gpr_mu_lock(&server->mu_global);
1029 2261 : grpc_cq_begin_op(cq);
1030 2261 : if (server->shutdown_published) {
1031 0 : grpc_cq_end_op(&exec_ctx, cq, tag, 1, done_published_shutdown, NULL,
1032 0 : gpr_malloc(sizeof(grpc_cq_completion)));
1033 0 : gpr_mu_unlock(&server->mu_global);
1034 0 : goto done;
1035 : }
1036 2261 : server->shutdown_tags =
1037 2261 : gpr_realloc(server->shutdown_tags,
1038 2261 : sizeof(shutdown_tag) * (server->num_shutdown_tags + 1));
1039 2261 : sdt = &server->shutdown_tags[server->num_shutdown_tags++];
1040 2261 : sdt->tag = tag;
1041 2261 : sdt->cq = cq;
1042 2261 : if (gpr_atm_acq_load(&server->shutdown_flag)) {
1043 0 : gpr_mu_unlock(&server->mu_global);
1044 0 : goto done;
1045 : }
1046 :
1047 2261 : server->last_shutdown_message_time = gpr_now(GPR_CLOCK_REALTIME);
1048 :
1049 2261 : channel_broadcaster_init(server, &broadcaster);
1050 :
1051 2261 : gpr_atm_rel_store(&server->shutdown_flag, 1);
1052 :
1053 : /* collect all unregistered then registered calls */
1054 2261 : gpr_mu_lock(&server->mu_call);
1055 2261 : kill_pending_work_locked(&exec_ctx, server);
1056 2261 : gpr_mu_unlock(&server->mu_call);
1057 :
1058 2261 : maybe_finish_shutdown(&exec_ctx, server);
1059 2261 : gpr_mu_unlock(&server->mu_global);
1060 :
1061 : /* Shutdown listeners */
1062 4034 : for (l = server->listeners; l; l = l->next) {
1063 1773 : grpc_closure_init(&l->destroy_done, listener_destroy_done, server);
1064 1773 : l->destroy(&exec_ctx, server, l->arg, &l->destroy_done);
1065 : }
1066 :
1067 2261 : channel_broadcaster_shutdown(&exec_ctx, &broadcaster, 1, 0);
1068 :
1069 : done:
1070 2261 : grpc_exec_ctx_finish(&exec_ctx);
1071 2261 : }
1072 :
1073 24 : void grpc_server_cancel_all_calls(grpc_server *server) {
1074 : channel_broadcaster broadcaster;
1075 24 : grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
1076 :
1077 24 : GRPC_API_TRACE("grpc_server_cancel_all_calls(server=%p)", 1, (server));
1078 :
1079 24 : gpr_mu_lock(&server->mu_global);
1080 24 : channel_broadcaster_init(server, &broadcaster);
1081 24 : gpr_mu_unlock(&server->mu_global);
1082 :
1083 24 : channel_broadcaster_shutdown(&exec_ctx, &broadcaster, 0, 1);
1084 24 : grpc_exec_ctx_finish(&exec_ctx);
1085 24 : }
1086 :
1087 2261 : void grpc_server_destroy(grpc_server *server) {
1088 : listener *l;
1089 2261 : grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
1090 :
1091 2261 : GRPC_API_TRACE("grpc_server_destroy(server=%p)", 1, (server));
1092 :
1093 2261 : gpr_mu_lock(&server->mu_global);
1094 2261 : GPR_ASSERT(gpr_atm_acq_load(&server->shutdown_flag) || !server->listeners);
1095 2261 : GPR_ASSERT(server->listeners_destroyed == num_listeners(server));
1096 :
1097 6295 : while (server->listeners) {
1098 1773 : l = server->listeners;
1099 1773 : server->listeners = l->next;
1100 1773 : gpr_free(l);
1101 : }
1102 :
1103 2261 : gpr_mu_unlock(&server->mu_global);
1104 :
1105 2261 : server_unref(&exec_ctx, server);
1106 2261 : grpc_exec_ctx_finish(&exec_ctx);
1107 2261 : }
1108 :
1109 1773 : void grpc_server_add_listener(
1110 : grpc_exec_ctx *exec_ctx, grpc_server *server, void *arg,
1111 : void (*start)(grpc_exec_ctx *exec_ctx, grpc_server *server, void *arg,
1112 : grpc_pollset **pollsets, size_t pollset_count),
1113 : void (*destroy)(grpc_exec_ctx *exec_ctx, grpc_server *server, void *arg,
1114 : grpc_closure *on_done)) {
1115 1773 : listener *l = gpr_malloc(sizeof(listener));
1116 1773 : l->arg = arg;
1117 1773 : l->start = start;
1118 1773 : l->destroy = destroy;
1119 1773 : l->next = server->listeners;
1120 1773 : server->listeners = l;
1121 1773 : }
1122 :
1123 1409264 : static grpc_call_error queue_call_request(grpc_exec_ctx *exec_ctx,
1124 : grpc_server *server,
1125 : requested_call *rc) {
1126 1409264 : call_data *calld = NULL;
1127 1409264 : request_matcher *rm = NULL;
1128 : int request_id;
1129 1409264 : if (gpr_atm_acq_load(&server->shutdown_flag)) {
1130 26308 : fail_call(exec_ctx, server, rc);
1131 26339 : return GRPC_CALL_OK;
1132 : }
1133 1382956 : request_id = gpr_stack_lockfree_pop(server->request_freelist);
1134 1383288 : if (request_id == -1) {
1135 : /* out of request ids: just fail this one */
1136 0 : fail_call(exec_ctx, server, rc);
1137 0 : return GRPC_CALL_OK;
1138 : }
1139 1383288 : switch (rc->type) {
1140 : case BATCH_CALL:
1141 3046 : rm = &server->unregistered_request_matcher;
1142 3046 : break;
1143 : case REGISTERED_CALL:
1144 1380242 : rm = &rc->data.registered.registered_method->request_matcher;
1145 1380242 : break;
1146 : }
1147 1383288 : server->requested_calls[request_id] = *rc;
1148 1383288 : gpr_free(rc);
1149 1383269 : if (gpr_stack_lockfree_push(rm->requests, request_id)) {
1150 : /* this was the first queued request: we need to lock and start
1151 : matching calls */
1152 342770 : gpr_mu_lock(&server->mu_call);
1153 960017 : while ((calld = rm->pending_head) != NULL) {
1154 547071 : request_id = gpr_stack_lockfree_pop(rm->requests);
1155 547071 : if (request_id == -1) break;
1156 274477 : rm->pending_head = calld->pending_next;
1157 274477 : gpr_mu_unlock(&server->mu_call);
1158 274477 : gpr_mu_lock(&calld->mu_state);
1159 274477 : if (calld->state == ZOMBIED) {
1160 0 : gpr_mu_unlock(&calld->mu_state);
1161 0 : grpc_closure_init(
1162 : &calld->kill_zombie_closure, kill_zombie,
1163 0 : grpc_call_stack_element(grpc_call_get_call_stack(calld->call), 0));
1164 0 : grpc_exec_ctx_enqueue(exec_ctx, &calld->kill_zombie_closure, 1);
1165 : } else {
1166 274477 : GPR_ASSERT(calld->state == PENDING);
1167 274477 : calld->state = ACTIVATED;
1168 274477 : gpr_mu_unlock(&calld->mu_state);
1169 548954 : begin_call(exec_ctx, server, calld,
1170 548954 : &server->requested_calls[request_id]);
1171 : }
1172 274477 : gpr_mu_lock(&server->mu_call);
1173 : }
1174 342770 : gpr_mu_unlock(&server->mu_call);
1175 : }
1176 1383285 : return GRPC_CALL_OK;
1177 : }
1178 :
1179 3046 : grpc_call_error grpc_server_request_call(
1180 : grpc_server *server, grpc_call **call, grpc_call_details *details,
1181 : grpc_metadata_array *initial_metadata,
1182 : grpc_completion_queue *cq_bound_to_call,
1183 : grpc_completion_queue *cq_for_notification, void *tag) {
1184 : grpc_call_error error;
1185 3046 : grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
1186 3046 : requested_call *rc = gpr_malloc(sizeof(*rc));
1187 3046 : GRPC_API_TRACE(
1188 : "grpc_server_request_call("
1189 : "server=%p, call=%p, details=%p, initial_metadata=%p, "
1190 : "cq_bound_to_call=%p, cq_for_notification=%p, tag%p)",
1191 : 7, (server, call, details, initial_metadata, cq_bound_to_call,
1192 : cq_for_notification, tag));
1193 3046 : GRPC_SERVER_LOG_REQUEST_CALL(GPR_INFO, server, call, details,
1194 : initial_metadata, cq_bound_to_call,
1195 : cq_for_notification, tag);
1196 3046 : if (!grpc_cq_is_server_cq(cq_for_notification)) {
1197 0 : gpr_free(rc);
1198 0 : error = GRPC_CALL_ERROR_NOT_SERVER_COMPLETION_QUEUE;
1199 0 : goto done;
1200 : }
1201 3046 : grpc_cq_begin_op(cq_for_notification);
1202 3046 : details->reserved = NULL;
1203 3046 : rc->type = BATCH_CALL;
1204 3046 : rc->server = server;
1205 3046 : rc->tag = tag;
1206 3046 : rc->cq_bound_to_call = cq_bound_to_call;
1207 3046 : rc->cq_for_notification = cq_for_notification;
1208 3046 : rc->call = call;
1209 3046 : rc->data.batch.details = details;
1210 3046 : rc->data.batch.initial_metadata = initial_metadata;
1211 3046 : error = queue_call_request(&exec_ctx, server, rc);
1212 : done:
1213 3046 : grpc_exec_ctx_finish(&exec_ctx);
1214 3046 : return error;
1215 : }
1216 :
1217 1405688 : grpc_call_error grpc_server_request_registered_call(
1218 : grpc_server *server, void *rmp, grpc_call **call, gpr_timespec *deadline,
1219 : grpc_metadata_array *initial_metadata, grpc_byte_buffer **optional_payload,
1220 : grpc_completion_queue *cq_bound_to_call,
1221 : grpc_completion_queue *cq_for_notification, void *tag) {
1222 : grpc_call_error error;
1223 1405688 : grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
1224 1405688 : requested_call *rc = gpr_malloc(sizeof(*rc));
1225 1406490 : registered_method *rm = rmp;
1226 1406490 : GRPC_API_TRACE(
1227 : "grpc_server_request_registered_call("
1228 : "server=%p, rmp=%p, call=%p, deadline=%p, initial_metadata=%p, "
1229 : "optional_payload=%p, cq_bound_to_call=%p, cq_for_notification=%p, "
1230 : "tag=%p)",
1231 : 9, (server, rmp, call, deadline, initial_metadata, optional_payload,
1232 : cq_bound_to_call, cq_for_notification, tag));
1233 1406490 : if (!grpc_cq_is_server_cq(cq_for_notification)) {
1234 0 : gpr_free(rc);
1235 0 : error = GRPC_CALL_ERROR_NOT_SERVER_COMPLETION_QUEUE;
1236 0 : goto done;
1237 : }
1238 1406431 : grpc_cq_begin_op(cq_for_notification);
1239 1406525 : rc->type = REGISTERED_CALL;
1240 1406525 : rc->server = server;
1241 1406525 : rc->tag = tag;
1242 1406525 : rc->cq_bound_to_call = cq_bound_to_call;
1243 1406525 : rc->cq_for_notification = cq_for_notification;
1244 1406525 : rc->call = call;
1245 1406525 : rc->data.registered.registered_method = rm;
1246 1406525 : rc->data.registered.deadline = deadline;
1247 1406525 : rc->data.registered.initial_metadata = initial_metadata;
1248 1406525 : rc->data.registered.optional_payload = optional_payload;
1249 1406525 : error = queue_call_request(&exec_ctx, server, rc);
1250 : done:
1251 1406578 : grpc_exec_ctx_finish(&exec_ctx);
1252 1406568 : return error;
1253 : }
1254 :
1255 : static void publish_registered_or_batch(grpc_exec_ctx *exec_ctx,
1256 : grpc_call *call, int success,
1257 : void *tag);
1258 0 : static void publish_was_not_set(grpc_exec_ctx *exec_ctx, grpc_call *call,
1259 : int success, void *tag) {
1260 0 : abort();
1261 : }
1262 :
1263 5162 : static void cpstr(char **dest, size_t *capacity, grpc_mdstr *value) {
1264 5162 : gpr_slice slice = value->slice;
1265 5162 : size_t len = GPR_SLICE_LENGTH(slice);
1266 :
1267 5162 : if (len + 1 > *capacity) {
1268 4847 : *capacity = GPR_MAX(len + 1, *capacity * 2);
1269 4847 : *dest = gpr_realloc(*dest, *capacity);
1270 : }
1271 5162 : memcpy(*dest, grpc_mdstr_as_c_string(value), len + 1);
1272 5162 : }
1273 :
1274 1301621 : static void begin_call(grpc_exec_ctx *exec_ctx, grpc_server *server,
1275 : call_data *calld, requested_call *rc) {
1276 1301621 : grpc_ioreq_completion_func publish = publish_was_not_set;
1277 : grpc_ioreq req[2];
1278 1301621 : grpc_ioreq *r = req;
1279 :
1280 : /* called once initial metadata has been read by the call, but BEFORE
1281 : the ioreq to fetch it out of the call has been executed.
1282 : This means metadata related fields can be relied on in calld, but to
1283 : fill in the metadata array passed by the client, we need to perform
1284 : an ioreq op, that should complete immediately. */
1285 :
1286 1301621 : grpc_call_set_completion_queue(exec_ctx, calld->call, rc->cq_bound_to_call);
1287 1301719 : *rc->call = calld->call;
1288 1301719 : calld->cq_new = rc->cq_for_notification;
1289 1301719 : switch (rc->type) {
1290 : case BATCH_CALL:
1291 2581 : GPR_ASSERT(calld->host != NULL);
1292 2581 : GPR_ASSERT(calld->path != NULL);
1293 5162 : cpstr(&rc->data.batch.details->host,
1294 2581 : &rc->data.batch.details->host_capacity, calld->host);
1295 5162 : cpstr(&rc->data.batch.details->method,
1296 2581 : &rc->data.batch.details->method_capacity, calld->path);
1297 2581 : rc->data.batch.details->deadline = calld->deadline;
1298 2581 : r->op = GRPC_IOREQ_RECV_INITIAL_METADATA;
1299 2581 : r->data.recv_metadata = rc->data.batch.initial_metadata;
1300 2581 : r->flags = 0;
1301 2581 : r++;
1302 2581 : publish = publish_registered_or_batch;
1303 2581 : break;
1304 : case REGISTERED_CALL:
1305 1299126 : *rc->data.registered.deadline = calld->deadline;
1306 1299126 : r->op = GRPC_IOREQ_RECV_INITIAL_METADATA;
1307 1299126 : r->data.recv_metadata = rc->data.registered.initial_metadata;
1308 1299126 : r->flags = 0;
1309 1299126 : r++;
1310 1299126 : if (rc->data.registered.optional_payload) {
1311 1299090 : r->op = GRPC_IOREQ_RECV_MESSAGE;
1312 1299090 : r->data.recv_message = rc->data.registered.optional_payload;
1313 1299090 : r->flags = 0;
1314 1299090 : r++;
1315 : }
1316 1299126 : publish = publish_registered_or_batch;
1317 1299126 : break;
1318 : }
1319 :
1320 1301719 : GRPC_CALL_INTERNAL_REF(calld->call, "server");
1321 1301740 : grpc_call_start_ioreq_and_call_back(exec_ctx, calld->call, req,
1322 1301740 : (size_t)(r - req), publish, rc);
1323 1301746 : }
1324 :
1325 1409572 : static void done_request_event(grpc_exec_ctx *exec_ctx, void *req,
1326 : grpc_cq_completion *c) {
1327 1409572 : requested_call *rc = req;
1328 1409572 : grpc_server *server = rc->server;
1329 :
1330 2792772 : if (rc >= server->requested_calls &&
1331 1383200 : rc < server->requested_calls + server->max_requested_calls) {
1332 1383202 : GPR_ASSERT(rc - server->requested_calls <= INT_MAX);
1333 1383202 : gpr_stack_lockfree_push(server->request_freelist,
1334 1383202 : (int)(rc - server->requested_calls));
1335 : } else {
1336 26370 : gpr_free(req);
1337 : }
1338 :
1339 1409642 : server_unref(exec_ctx, server);
1340 1409423 : }
1341 :
1342 107463 : static void fail_call(grpc_exec_ctx *exec_ctx, grpc_server *server,
1343 : requested_call *rc) {
1344 107463 : *rc->call = NULL;
1345 107463 : switch (rc->type) {
1346 : case BATCH_CALL:
1347 465 : rc->data.batch.initial_metadata->count = 0;
1348 465 : break;
1349 : case REGISTERED_CALL:
1350 106948 : rc->data.registered.initial_metadata->count = 0;
1351 106948 : break;
1352 : }
1353 107463 : server_ref(server);
1354 107619 : grpc_cq_end_op(exec_ctx, rc->cq_for_notification, rc->tag, 0,
1355 : done_request_event, rc, &rc->completion);
1356 107834 : }
1357 :
1358 1301707 : static void publish_registered_or_batch(grpc_exec_ctx *exec_ctx,
1359 : grpc_call *call, int success,
1360 : void *prc) {
1361 1301726 : grpc_call_element *elem =
1362 1301707 : grpc_call_stack_element(grpc_call_get_call_stack(call), 0);
1363 1301725 : requested_call *rc = prc;
1364 1301725 : call_data *calld = elem->call_data;
1365 1301725 : channel_data *chand = elem->channel_data;
1366 1301725 : server_ref(chand->server);
1367 1301745 : grpc_cq_end_op(exec_ctx, calld->cq_new, rc->tag, success, done_request_event,
1368 : rc, &rc->completion);
1369 1301727 : GRPC_CALL_INTERNAL_UNREF(exec_ctx, call, "server");
1370 1301750 : }
1371 :
1372 3556 : const grpc_channel_args *grpc_server_get_channel_args(grpc_server *server) {
1373 3556 : return server->channel_args;
1374 : }
1375 :
1376 132 : int grpc_server_has_open_connections(grpc_server *server) {
1377 : int r;
1378 132 : gpr_mu_lock(&server->mu_global);
1379 132 : r = server->root_channel_data.next != &server->root_channel_data;
1380 132 : gpr_mu_unlock(&server->mu_global);
1381 132 : return r;
1382 : }
|
