Chiebot-Mirror
/
grpc
mirror of https://github.com/grpc/grpc.git
8
0
Fork 0
Code Issues Projects Releases Wiki Activity

Rewrite completion queue internals to use pre-allocation of events

Browse Source
pull/2347/head
Craig Tiller 10 years ago
parent c6964b1d98
commit 97fc6a3f3f
6 changed files with 227 additions and 245 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 3
      src/core/iomgr/pollset_multipoller_with_epoll.c
  2. 42
      src/core/surface/call.c
  3. 212
      src/core/surface/completion_queue.c
  4. 22
      src/core/surface/completion_queue.h
  5. 167
      src/core/surface/server.c
  6. 24
      test/core/surface/completion_queue_test.c

3
src/core/iomgr/pollset_multipoller_with_epoll.c
Unescape View File

@ -105,10 +105,11 @@ static void multipoll_with_epoll_pollset_maybe_work(
* here. * here.
*/ */
timeout_ms = grpc_poll_deadline_to_millis_timeout(deadline, now);
pollset->counter += 1; pollset->counter += 1;
gpr_mu_unlock(&pollset->mu); gpr_mu_unlock(&pollset->mu);
timeout_ms = grpc_poll_deadline_to_millis_timeout(deadline, now);
do { do {
ep_rv = epoll_wait(h->epoll_fd, ep_ev, GRPC_EPOLL_MAX_EVENTS, timeout_ms); ep_rv = epoll_wait(h->epoll_fd, ep_ev, GRPC_EPOLL_MAX_EVENTS, timeout_ms);
if (ep_rv < 0) { if (ep_rv < 0) {

42
src/core/surface/call.c
Unescape View File

@ -162,6 +162,8 @@ struct grpc_call {
gpr_uint8 error_status_set; gpr_uint8 error_status_set;
/** should the alarm be cancelled */ /** should the alarm be cancelled */
gpr_uint8 cancel_alarm; gpr_uint8 cancel_alarm;
/** bitmask of allocated completion events in completions */
gpr_uint8 allocated_completions;
/* flags with bits corresponding to write states allowing us to determine /* flags with bits corresponding to write states allowing us to determine
what was sent */ what was sent */
@ -250,6 +252,9 @@ struct grpc_call {
grpc_iomgr_closure on_done_recv; grpc_iomgr_closure on_done_recv;
grpc_iomgr_closure on_done_send; grpc_iomgr_closure on_done_send;
grpc_iomgr_closure on_done_bind; grpc_iomgr_closure on_done_bind;
/** completion events - for completion queue use */
grpc_cq_completion completions[6];
}; };
#define CALL_STACK_FROM_CALL(call) ((grpc_call_stack *)((call) + 1)) #define CALL_STACK_FROM_CALL(call) ((grpc_call_stack *)((call) + 1))
@ -349,6 +354,27 @@ grpc_completion_queue *grpc_call_get_completion_queue(grpc_call *call) {
return call->cq; return call->cq;
} }
grpc_cq_completion *allocate_completion(grpc_call *call) {
gpr_uint8 i;
for (i = 0; i < GPR_ARRAY_SIZE(call->completions); i++) {
if (call->allocated_completions & (1u << i)) {
continue;
}
call->allocated_completions |= 1u << i;
return &call->completions[i];
}
gpr_log(GPR_ERROR, "should never reach here");
abort();
}
void done_completion(void *call, grpc_cq_completion *completion) {
grpc_call *c = call;
gpr_mu_lock(&c->mu);
c->allocated_completions &= ~(1u << (completion - c->completions));
gpr_mu_unlock(&c->mu);
GRPC_CALL_INTERNAL_UNREF(c, "completion", 1);
}
#ifdef GRPC_CALL_REF_COUNT_DEBUG #ifdef GRPC_CALL_REF_COUNT_DEBUG
void grpc_call_internal_ref(grpc_call *c, const char *reason) { void grpc_call_internal_ref(grpc_call *c, const char *reason) {
gpr_log(GPR_DEBUG, "CALL: ref %p %d -> %d [%s]", c, gpr_log(GPR_DEBUG, "CALL: ref %p %d -> %d [%s]", c,
@ -1316,11 +1342,15 @@ static void set_cancelled_value(grpc_status_code status, void *dest) {
} }
static void finish_batch(grpc_call *call, int success, void *tag) { static void finish_batch(grpc_call *call, int success, void *tag) {
grpc_cq_end_op(call->cq, tag, call, success); grpc_cq_end_op(call->cq,
tag, success, done_completion, call,
allocate_completion(call));
} }
static void finish_batch_with_close(grpc_call *call, int success, void *tag) { static void finish_batch_with_close(grpc_call *call, int success, void *tag) {
grpc_cq_end_op(call->cq, tag, call, 1); grpc_cq_end_op(call->cq,
tag, 1, done_completion, call,
allocate_completion(call));
} }
static int are_write_flags_valid(gpr_uint32 flags) { static int are_write_flags_valid(gpr_uint32 flags) {
@ -1343,8 +1373,9 @@ grpc_call_error grpc_call_start_batch(grpc_call *call, const grpc_op *ops,
GRPC_CALL_LOG_BATCH(GPR_INFO, call, ops, nops, tag); GRPC_CALL_LOG_BATCH(GPR_INFO, call, ops, nops, tag);
if (nops == 0) { if (nops == 0) {
grpc_cq_begin_op(call->cq, call); grpc_cq_begin_op(call->cq);
grpc_cq_end_op(call->cq, tag, call, 1); GRPC_CALL_INTERNAL_REF(call, "completion");
grpc_cq_end_op(call->cq, tag, 1, done_completion, call, allocate_completion(call));
return GRPC_CALL_OK; return GRPC_CALL_OK;
} }
@ -1466,7 +1497,8 @@ grpc_call_error grpc_call_start_batch(grpc_call *call, const grpc_op *ops,
} }
} }
grpc_cq_begin_op(call->cq, call); GRPC_CALL_INTERNAL_REF(call, "completion");
grpc_cq_begin_op(call->cq);
return grpc_call_start_ioreq_and_call_back(call, reqs, out, finish_func, tag); return grpc_call_start_ioreq_and_call_back(call, reqs, out, finish_func, tag);
} }

212
src/core/surface/completion_queue.c
Unescape View File

@ -45,34 +45,20 @@
#include <grpc/support/atm.h> #include <grpc/support/atm.h>
#include <grpc/support/log.h> #include <grpc/support/log.h>
#define NUM_TAG_BUCKETS 31
/* A single event: extends grpc_event to form a linked list with a destruction
function (on_finish) that is hidden from outside this module */
typedef struct event {
grpc_event base;
struct event *queue_next;
struct event *queue_prev;
struct event *bucket_next;
struct event *bucket_prev;
} event;
/* Completion queue structure */ /* Completion queue structure */
struct grpc_completion_queue { struct grpc_completion_queue {
/* When refs drops to zero, we are in shutdown mode, and will be destroyable /** completed events */
once all queued events are drained */ grpc_cq_completion completed_head;
gpr_refcount refs; grpc_cq_completion *completed_tail;
/* Once owning_refs drops to zero, we will destroy the cq */ /** Number of pending events (+1 if we're not shutdown) */
gpr_refcount pending_events;
/** Once owning_refs drops to zero, we will destroy the cq */
gpr_refcount owning_refs; gpr_refcount owning_refs;
/* the set of low level i/o things that concern this cq */ /** the set of low level i/o things that concern this cq */
grpc_pollset pollset; grpc_pollset pollset;
/* 0 initially, 1 once we've begun shutting down */ /** 0 initially, 1 once we've begun shutting down */
int shutdown; int shutdown;
int shutdown_called; int shutdown_called;
/* Head of a linked list of queued events (prev points to the last element) */
event *queue;
/* Fixed size chained hash table of events for pluck() */
event *buckets[NUM_TAG_BUCKETS];
int is_server_cq; int is_server_cq;
}; };
@ -80,10 +66,12 @@ grpc_completion_queue *grpc_completion_queue_create(void) {
grpc_completion_queue *cc = gpr_malloc(sizeof(grpc_completion_queue)); grpc_completion_queue *cc = gpr_malloc(sizeof(grpc_completion_queue));
memset(cc, 0, sizeof(*cc)); memset(cc, 0, sizeof(*cc));
/* Initial ref is dropped by grpc_completion_queue_shutdown */ /* Initial ref is dropped by grpc_completion_queue_shutdown */
gpr_ref_init(&cc->refs, 1); gpr_ref_init(&cc->pending_events, 1);
/* One for destroy(), one for pollset_shutdown */ /* One for destroy(), one for pollset_shutdown */
gpr_ref_init(&cc->owning_refs, 2); gpr_ref_init(&cc->owning_refs, 2);
grpc_pollset_init(&cc->pollset); grpc_pollset_init(&cc->pollset);
cc->completed_tail = &cc->completed_head;
cc->completed_head.next = (gpr_uintptr) cc->completed_tail;
return cc; return cc;
} }
@ -114,179 +102,127 @@ void grpc_cq_internal_unref(grpc_completion_queue *cc, const char *reason,
void grpc_cq_internal_unref(grpc_completion_queue *cc) { void grpc_cq_internal_unref(grpc_completion_queue *cc) {
#endif #endif
if (gpr_unref(&cc->owning_refs)) { if (gpr_unref(&cc->owning_refs)) {
GPR_ASSERT(cc->queue == NULL); GPR_ASSERT(cc->completed_head.next == (gpr_uintptr) &cc->completed_head);
grpc_pollset_destroy(&cc->pollset); grpc_pollset_destroy(&cc->pollset);
gpr_free(cc); gpr_free(cc);
} }
} }
/* Create and append an event to the queue. Returns the event so that its data void grpc_cq_begin_op(grpc_completion_queue *cc) {
members can be filled in. gpr_ref(&cc->pending_events);
Requires GRPC_POLLSET_MU(&cc->pollset) locked. */
static event *add_locked(grpc_completion_queue *cc, grpc_completion_type type,
void *tag, grpc_call *call) {
event *ev = gpr_malloc(sizeof(event));
gpr_uintptr bucket = ((gpr_uintptr)tag) % NUM_TAG_BUCKETS;
ev->base.type = type;
ev->base.tag = tag;
if (cc->queue == NULL) {
cc->queue = ev->queue_next = ev->queue_prev = ev;
} else {
ev->queue_next = cc->queue;
ev->queue_prev = cc->queue->queue_prev;
ev->queue_next->queue_prev = ev->queue_prev->queue_next = ev;
}
if (cc->buckets[bucket] == NULL) {
cc->buckets[bucket] = ev->bucket_next = ev->bucket_prev = ev;
} else {
ev->bucket_next = cc->buckets[bucket];
ev->bucket_prev = cc->buckets[bucket]->bucket_prev;
ev->bucket_next->bucket_prev = ev->bucket_prev->bucket_next = ev;
}
grpc_pollset_kick(&cc->pollset);
return ev;
}
void grpc_cq_begin_op(grpc_completion_queue *cc, grpc_call *call) {
gpr_ref(&cc->refs);
if (call) GRPC_CALL_INTERNAL_REF(call, "cq");
} }
/* Signal the end of an operation - if this is the last waiting-to-be-queued /* Signal the end of an operation - if this is the last waiting-to-be-queued
event, then enter shutdown mode */ event, then enter shutdown mode */
void grpc_cq_end_op(grpc_completion_queue *cc, void *tag, grpc_call *call, /* Queue a GRPC_OP_COMPLETED operation */
int success) { void grpc_cq_end_op(
event *ev; grpc_completion_queue *cc,
int shutdown = 0; void *tag,
int success,
void (*done)(void *done_arg, grpc_cq_completion *storage),
void *done_arg,
grpc_cq_completion *storage) {
int shutdown = gpr_unref(&cc->pending_events);
storage->tag = tag;
storage->done = done;
storage->done_arg = done_arg;
storage->next = ((gpr_uintptr)&cc->completed_head) | ((gpr_uintptr)(success != 0));
if (!shutdown) {
gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset)); gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset));
ev = add_locked(cc, GRPC_OP_COMPLETE, tag, call); cc->completed_tail->next = ((gpr_uintptr)storage) | (1u & (gpr_uintptr)cc->completed_tail->next);
ev->base.success = success; cc->completed_tail = storage;
if (gpr_unref(&cc->refs)) { gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
} else {
gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset));
cc->completed_tail->next = ((gpr_uintptr)storage) | (1u & (gpr_uintptr)cc->completed_tail->next);
cc->completed_tail = storage;
GPR_ASSERT(!cc->shutdown); GPR_ASSERT(!cc->shutdown);
GPR_ASSERT(cc->shutdown_called); GPR_ASSERT(cc->shutdown_called);
cc->shutdown = 1; cc->shutdown = 1;
shutdown = 1;
}
gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset)); gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
if (call) GRPC_CALL_INTERNAL_UNREF(call, "cq", 0);
if (shutdown) {
grpc_pollset_shutdown(&cc->pollset, on_pollset_destroy_done, cc); grpc_pollset_shutdown(&cc->pollset, on_pollset_destroy_done, cc);
} }
} }
/* Create a GRPC_QUEUE_SHUTDOWN event without queuing it anywhere */
static event *create_shutdown_event(void) {
event *ev = gpr_malloc(sizeof(event));
ev->base.type = GRPC_QUEUE_SHUTDOWN;
ev->base.tag = NULL;
return ev;
}
grpc_event grpc_completion_queue_next(grpc_completion_queue *cc, grpc_event grpc_completion_queue_next(grpc_completion_queue *cc,
gpr_timespec deadline) { gpr_timespec deadline) {
event *ev = NULL;
grpc_event ret; grpc_event ret;
GRPC_CQ_INTERNAL_REF(cc, "next"); GRPC_CQ_INTERNAL_REF(cc, "next");
gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset)); gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset));
for (;;) { for (;;) {
if (cc->queue != NULL) { if (cc->completed_tail != &cc->completed_head) {
gpr_uintptr bucket; grpc_cq_completion *c = (grpc_cq_completion *) cc->completed_head.next;
ev = cc->queue; cc->completed_head.next = c->next & ~(gpr_uintptr)1;
bucket = ((gpr_uintptr)ev->base.tag) % NUM_TAG_BUCKETS; if (c == cc->completed_tail) {
cc->queue = ev->queue_next; cc->completed_tail = &cc->completed_head;
ev->queue_next->queue_prev = ev->queue_prev;
ev->queue_prev->queue_next = ev->queue_next;
ev->bucket_next->bucket_prev = ev->bucket_prev;
ev->bucket_prev->bucket_next = ev->bucket_next;
if (ev == cc->buckets[bucket]) {
cc->buckets[bucket] = ev->bucket_next;
if (ev == cc->buckets[bucket]) {
cc->buckets[bucket] = NULL;
}
}
if (cc->queue == ev) {
cc->queue = NULL;
} }
gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
ret.type = GRPC_OP_COMPLETE;
ret.success = c->next & 1u;
ret.tag = c->tag;
c->done(c->done_arg, c);
break; break;
} }
if (cc->shutdown) { if (cc->shutdown) {
ev = create_shutdown_event(); gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
memset(&ret, 0, sizeof(ret));
ret.type = GRPC_QUEUE_SHUTDOWN;
break; break;
} }
if (!grpc_pollset_work(&cc->pollset, deadline)) { if (!grpc_pollset_work(&cc->pollset, deadline)) {
gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset)); gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
memset(&ret, 0, sizeof(ret)); memset(&ret, 0, sizeof(ret));
ret.type = GRPC_QUEUE_TIMEOUT; ret.type = GRPC_QUEUE_TIMEOUT;
GRPC_SURFACE_TRACE_RETURNED_EVENT(cc, &ret); break;
GRPC_CQ_INTERNAL_UNREF(cc, "next");
return ret;
} }
} }
gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
ret = ev->base;
gpr_free(ev);
GRPC_SURFACE_TRACE_RETURNED_EVENT(cc, &ret); GRPC_SURFACE_TRACE_RETURNED_EVENT(cc, &ret);
GRPC_CQ_INTERNAL_UNREF(cc, "next"); GRPC_CQ_INTERNAL_UNREF(cc, "next");
return ret; return ret;
} }
static event *pluck_event(grpc_completion_queue *cc, void *tag) {
gpr_uintptr bucket = ((gpr_uintptr)tag) % NUM_TAG_BUCKETS;
event *ev = cc->buckets[bucket];
if (ev == NULL) return NULL;
do {
if (ev->base.tag == tag) {
ev->queue_next->queue_prev = ev->queue_prev;
ev->queue_prev->queue_next = ev->queue_next;
ev->bucket_next->bucket_prev = ev->bucket_prev;
ev->bucket_prev->bucket_next = ev->bucket_next;
if (ev == cc->buckets[bucket]) {
cc->buckets[bucket] = ev->bucket_next;
if (ev == cc->buckets[bucket]) {
cc->buckets[bucket] = NULL;
}
}
if (cc->queue == ev) {
cc->queue = ev->queue_next;
if (cc->queue == ev) {
cc->queue = NULL;
}
}
return ev;
}
ev = ev->bucket_next;
} while (ev != cc->buckets[bucket]);
return NULL;
}
grpc_event grpc_completion_queue_pluck(grpc_completion_queue *cc, void *tag, grpc_event grpc_completion_queue_pluck(grpc_completion_queue *cc, void *tag,
gpr_timespec deadline) { gpr_timespec deadline) {
event *ev = NULL;
grpc_event ret; grpc_event ret;
grpc_cq_completion *c;
grpc_cq_completion *prev;
GRPC_CQ_INTERNAL_REF(cc, "pluck"); GRPC_CQ_INTERNAL_REF(cc, "pluck");
gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset)); gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset));
for (;;) { for (;;) {
if ((ev = pluck_event(cc, tag))) { prev = &cc->completed_head;
break; while ((c = (grpc_cq_completion*)(prev->next & ~(gpr_uintptr)1)) != &cc->completed_head) {
if (c->tag == tag) {
prev->next = (prev->next & (gpr_uintptr)1) | (c->next & ~(gpr_uintptr)1);
if (c == cc->completed_tail) {
cc->completed_tail = prev;
}
gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
ret.type = GRPC_OP_COMPLETE;
ret.success = c->next & 1u;
ret.tag = c->tag;
c->done(c->done_arg, c);
goto done;
}
prev = c;
} }
if (cc->shutdown) { if (cc->shutdown) {
ev = create_shutdown_event(); gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
memset(&ret, 0, sizeof(ret));
ret.type = GRPC_QUEUE_SHUTDOWN;
break; break;
} }
if (!grpc_pollset_work(&cc->pollset, deadline)) { if (!grpc_pollset_work(&cc->pollset, deadline)) {
gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset)); gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
memset(&ret, 0, sizeof(ret)); memset(&ret, 0, sizeof(ret));
ret.type = GRPC_QUEUE_TIMEOUT; ret.type = GRPC_QUEUE_TIMEOUT;
GRPC_SURFACE_TRACE_RETURNED_EVENT(cc, &ret); break;
GRPC_CQ_INTERNAL_UNREF(cc, "pluck");
return ret;
} }
} }
gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset)); done:
ret = ev->base;
gpr_free(ev);
GRPC_SURFACE_TRACE_RETURNED_EVENT(cc, &ret); GRPC_SURFACE_TRACE_RETURNED_EVENT(cc, &ret);
GRPC_CQ_INTERNAL_UNREF(cc, "pluck"); GRPC_CQ_INTERNAL_UNREF(cc, "pluck");
return ret; return ret;
@ -303,7 +239,7 @@ void grpc_completion_queue_shutdown(grpc_completion_queue *cc) {
cc->shutdown_called = 1; cc->shutdown_called = 1;
gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset)); gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
if (gpr_unref(&cc->refs)) { if (gpr_unref(&cc->pending_events)) {
gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset)); gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset));
GPR_ASSERT(!cc->shutdown); GPR_ASSERT(!cc->shutdown);
cc->shutdown = 1; cc->shutdown = 1;

22
src/core/surface/completion_queue.h
Unescape View File

@ -39,6 +39,17 @@
#include "src/core/iomgr/pollset.h" #include "src/core/iomgr/pollset.h"
#include <grpc/grpc.h> #include <grpc/grpc.h>
typedef struct grpc_cq_completion {
/** user supplied tag */
void *tag;
/** done callback - called when this queue element is no longer
needed by the completion queue */
void (*done)(void *done_arg, struct grpc_cq_completion *c);
void *done_arg;
/** next pointer; low bit is used to indicate success or not */
gpr_uintptr next;
} grpc_cq_completion;
#ifdef GRPC_CQ_REF_COUNT_DEBUG #ifdef GRPC_CQ_REF_COUNT_DEBUG
void grpc_cq_internal_ref(grpc_completion_queue *cc, const char *reason, void grpc_cq_internal_ref(grpc_completion_queue *cc, const char *reason,
const char *file, int line); const char *file, int line);
@ -57,11 +68,16 @@ void grpc_cq_internal_unref(grpc_completion_queue *cc);
/* Flag that an operation is beginning: the completion channel will not finish /* Flag that an operation is beginning: the completion channel will not finish
shutdown until a corrensponding grpc_cq_end_* call is made */ shutdown until a corrensponding grpc_cq_end_* call is made */
void grpc_cq_begin_op(grpc_completion_queue *cc, grpc_call *call); void grpc_cq_begin_op(grpc_completion_queue *cc);
/* Queue a GRPC_OP_COMPLETED operation */ /* Queue a GRPC_OP_COMPLETED operation */
void grpc_cq_end_op(grpc_completion_queue *cc, void *tag, grpc_call *call, void grpc_cq_end_op(
int success); grpc_completion_queue *cc,
void *tag,
int success,
void (*done)(void *done_arg, grpc_cq_completion *storage),
void *done_arg,
grpc_cq_completion *storage);
grpc_pollset *grpc_cq_pollset(grpc_completion_queue *cc); grpc_pollset *grpc_cq_pollset(grpc_completion_queue *cc);

167
src/core/surface/server.c
Unescape View File

@ -72,12 +72,14 @@ typedef struct {
typedef enum { BATCH_CALL, REGISTERED_CALL } requested_call_type; typedef enum { BATCH_CALL, REGISTERED_CALL } requested_call_type;
typedef struct { typedef struct requested_call {
requested_call_type type; requested_call_type type;
struct requested_call *next;
void *tag; void *tag;
grpc_completion_queue *cq_bound_to_call; grpc_completion_queue *cq_bound_to_call;
grpc_completion_queue *cq_for_notification; grpc_completion_queue *cq_for_notification;
grpc_call **call; grpc_call **call;
grpc_cq_completion completion;
union { union {
struct { struct {
grpc_call_details *details; grpc_call_details *details;
@ -92,17 +94,11 @@ typedef struct {
} data; } data;
} requested_call; } requested_call;
typedef struct {
requested_call *calls;
size_t count;
size_t capacity;
} requested_call_array;
struct registered_method { struct registered_method {
char *method; char *method;
char *host; char *host;
call_data *pending; call_data *pending;
requested_call_array requested; requested_call *requests;
registered_method *next; registered_method *next;
}; };
@ -131,6 +127,7 @@ struct channel_data {
typedef struct shutdown_tag { typedef struct shutdown_tag {
void *tag; void *tag;
grpc_completion_queue *cq; grpc_completion_queue *cq;
grpc_cq_completion completion;
} shutdown_tag; } shutdown_tag;
struct grpc_server { struct grpc_server {
@ -153,7 +150,7 @@ struct grpc_server {
gpr_mu mu_call; /* mutex for call-specific state */ gpr_mu mu_call; /* mutex for call-specific state */
registered_method *registered_methods; registered_method *registered_methods;
requested_call_array requested_calls; requested_call *requests;
gpr_uint8 shutdown; gpr_uint8 shutdown;
gpr_uint8 shutdown_published; gpr_uint8 shutdown_published;
@ -325,22 +322,6 @@ static int call_list_remove(call_data *call, call_list list) {
return 1; return 1;
} }
static void requested_call_array_destroy(requested_call_array *array) {
gpr_free(array->calls);
}
static requested_call *requested_call_array_add(requested_call_array *array) {
requested_call *rc;
if (array->count == array->capacity) {
array->capacity = GPR_MAX(array->capacity + 8, array->capacity * 2);
array->calls =
gpr_realloc(array->calls, sizeof(requested_call) * array->capacity);
}
rc = &array->calls[array->count++];
memset(rc, 0, sizeof(*rc));
return rc;
}
static void server_ref(grpc_server *server) { static void server_ref(grpc_server *server) {
gpr_ref(&server->internal_refcount); gpr_ref(&server->internal_refcount);
} }
@ -352,12 +333,10 @@ static void server_delete(grpc_server *server) {
gpr_mu_destroy(&server->mu_global); gpr_mu_destroy(&server->mu_global);
gpr_mu_destroy(&server->mu_call); gpr_mu_destroy(&server->mu_call);
gpr_free(server->channel_filters); gpr_free(server->channel_filters);
requested_call_array_destroy(&server->requested_calls);
while ((rm = server->registered_methods) != NULL) { while ((rm = server->registered_methods) != NULL) {
server->registered_methods = rm->next; server->registered_methods = rm->next;
gpr_free(rm->method); gpr_free(rm->method);
gpr_free(rm->host); gpr_free(rm->host);
requested_call_array_destroy(&rm->requested);
gpr_free(rm); gpr_free(rm);
} }
for (i = 0; i < server->cq_count; i++) { for (i = 0; i < server->cq_count; i++) {
@ -406,18 +385,18 @@ static void destroy_channel(channel_data *chand) {
static void finish_start_new_rpc_and_unlock(grpc_server *server, static void finish_start_new_rpc_and_unlock(grpc_server *server,
grpc_call_element *elem, grpc_call_element *elem,
call_data **pending_root, call_data **pending_root,
requested_call_array *array) { requested_call **requests) {
requested_call rc; requested_call *rc = *requests;
call_data *calld = elem->call_data; call_data *calld = elem->call_data;
if (array->count == 0) { if (rc == NULL) {
calld->state = PENDING; calld->state = PENDING;
call_list_join(pending_root, calld, PENDING_START); call_list_join(pending_root, calld, PENDING_START);
gpr_mu_unlock(&server->mu_call); gpr_mu_unlock(&server->mu_call);
} else { } else {
rc = array->calls[--array->count]; *requests = rc->next;
calld->state = ACTIVATED; calld->state = ACTIVATED;
gpr_mu_unlock(&server->mu_call); gpr_mu_unlock(&server->mu_call);
begin_call(server, calld, &rc); begin_call(server, calld, rc);
} }
} }
@ -442,7 +421,7 @@ static void start_new_rpc(grpc_call_element *elem) {
if (rm->method != calld->path) continue; if (rm->method != calld->path) continue;
finish_start_new_rpc_and_unlock(server, elem, finish_start_new_rpc_and_unlock(server, elem,
&rm->server_registered_method->pending, &rm->server_registered_method->pending,
&rm->server_registered_method->requested); &rm->server_registered_method->requests);
return; return;
} }
/* check for a wildcard method definition (no host set) */ /* check for a wildcard method definition (no host set) */
@ -455,12 +434,12 @@ static void start_new_rpc(grpc_call_element *elem) {
if (rm->method != calld->path) continue; if (rm->method != calld->path) continue;
finish_start_new_rpc_and_unlock(server, elem, finish_start_new_rpc_and_unlock(server, elem,
&rm->server_registered_method->pending, &rm->server_registered_method->pending,
&rm->server_registered_method->requested); &rm->server_registered_method->requests);
return; return;
} }
} }
finish_start_new_rpc_and_unlock(server, elem, &server->lists[PENDING_START], finish_start_new_rpc_and_unlock(server, elem, &server->lists[PENDING_START],
&server->requested_calls); &server->requests);
} }
static void kill_zombie(void *elem, int success) { static void kill_zombie(void *elem, int success) {
@ -476,6 +455,10 @@ static int num_listeners(grpc_server *server) {
return n; return n;
} }
static void done_shutdown_event(void *server, grpc_cq_completion *completion) {
server_unref(server);
}
static void maybe_finish_shutdown(grpc_server *server) { static void maybe_finish_shutdown(grpc_server *server) {
size_t i; size_t i;
if (!server->shutdown || server->shutdown_published) { if (!server->shutdown || server->shutdown_published) {
@ -494,8 +477,14 @@ static void maybe_finish_shutdown(grpc_server *server) {
} }
server->shutdown_published = 1; server->shutdown_published = 1;
for (i = 0; i < server->num_shutdown_tags; i++) { for (i = 0; i < server->num_shutdown_tags; i++) {
grpc_cq_end_op(server->shutdown_tags[i].cq, server->shutdown_tags[i].tag, server_ref(server);
NULL, 1); grpc_cq_end_op(server->shutdown_tags[i].cq,
server->shutdown_tags[i].tag,
1,
done_shutdown_event,
server,
&server->shutdown_tags[i].completion
);
} }
} }
@ -910,15 +899,14 @@ void grpc_server_setup_transport(grpc_server *s, grpc_transport *transport,
void grpc_server_shutdown_and_notify(grpc_server *server, void grpc_server_shutdown_and_notify(grpc_server *server,
grpc_completion_queue *cq, void *tag) { grpc_completion_queue *cq, void *tag) {
listener *l; listener *l;
requested_call_array requested_calls; requested_call *requests = NULL;
size_t i;
registered_method *rm; registered_method *rm;
shutdown_tag *sdt; shutdown_tag *sdt;
channel_broadcaster broadcaster; channel_broadcaster broadcaster;
/* lock, and gather up some stuff to do */ /* lock, and gather up some stuff to do */
gpr_mu_lock(&server->mu_global); gpr_mu_lock(&server->mu_global);
grpc_cq_begin_op(cq, NULL); grpc_cq_begin_op(cq);
server->shutdown_tags = server->shutdown_tags =
gpr_realloc(server->shutdown_tags, gpr_realloc(server->shutdown_tags,
sizeof(shutdown_tag) * (server->num_shutdown_tags + 1)); sizeof(shutdown_tag) * (server->num_shutdown_tags + 1));
@ -934,23 +922,15 @@ void grpc_server_shutdown_and_notify(grpc_server *server,
/* collect all unregistered then registered calls */ /* collect all unregistered then registered calls */
gpr_mu_lock(&server->mu_call); gpr_mu_lock(&server->mu_call);
requested_calls = server->requested_calls; requests = server->requests;
memset(&server->requested_calls, 0, sizeof(server->requested_calls)); server->requests = NULL;
for (rm = server->registered_methods; rm; rm = rm->next) { for (rm = server->registered_methods; rm; rm = rm->next) {
if (requested_calls.count + rm->requested.count > while (rm->requests != NULL) {
requested_calls.capacity) { requested_call *c = rm->requests;
requested_calls.capacity = rm->requests = c->next;
GPR_MAX(requested_calls.count + rm->requested.count, c->next = requests;
2 * requested_calls.capacity); requests = c;
requested_calls.calls = }
gpr_realloc(requested_calls.calls, sizeof(*requested_calls.calls) *
requested_calls.capacity);
}
memcpy(requested_calls.calls + requested_calls.count, rm->requested.calls,
sizeof(*requested_calls.calls) * rm->requested.count);
requested_calls.count += rm->requested.count;
gpr_free(rm->requested.calls);
memset(&rm->requested, 0, sizeof(rm->requested));
} }
gpr_mu_unlock(&server->mu_call); gpr_mu_unlock(&server->mu_call);
@ -959,10 +939,11 @@ void grpc_server_shutdown_and_notify(grpc_server *server,
gpr_mu_unlock(&server->mu_global); gpr_mu_unlock(&server->mu_global);
/* terminate all the requested calls */ /* terminate all the requested calls */
for (i = 0; i < requested_calls.count; i++) { while (requests != NULL) {
fail_call(server, &requested_calls.calls[i]); requested_call *next = requests->next;
fail_call(server, requests);
requests = next;
} }
gpr_free(requested_calls.calls);
/* Shutdown listeners */ /* Shutdown listeners */
for (l = server->listeners; l; l = l->next) { for (l = server->listeners; l; l = l->next) {
@ -1024,7 +1005,7 @@ void grpc_server_add_listener(grpc_server *server, void *arg,
static grpc_call_error queue_call_request(grpc_server *server, static grpc_call_error queue_call_request(grpc_server *server,
requested_call *rc) { requested_call *rc) {
call_data *calld = NULL; call_data *calld = NULL;
requested_call_array *requested_calls = NULL; requested_call **requests;
gpr_mu_lock(&server->mu_call); gpr_mu_lock(&server->mu_call);
if (server->shutdown) { if (server->shutdown) {
gpr_mu_unlock(&server->mu_call); gpr_mu_unlock(&server->mu_call);
@ -1035,12 +1016,12 @@ static grpc_call_error queue_call_request(grpc_server *server,
case BATCH_CALL: case BATCH_CALL:
calld = calld =
call_list_remove_head(&server->lists[PENDING_START], PENDING_START); call_list_remove_head(&server->lists[PENDING_START], PENDING_START);
requested_calls = &server->requested_calls; requests = &server->requests;
break; break;
case REGISTERED_CALL: case REGISTERED_CALL:
calld = call_list_remove_head( calld = call_list_remove_head(
&rc->data.registered.registered_method->pending, PENDING_START); &rc->data.registered.registered_method->pending, PENDING_START);
requested_calls = &rc->data.registered.registered_method->requested; requests = &rc->data.registered.registered_method->requests;
break; break;
} }
if (calld) { if (calld) {
@ -1050,7 +1031,8 @@ static grpc_call_error queue_call_request(grpc_server *server,
begin_call(server, calld, rc); begin_call(server, calld, rc);
return GRPC_CALL_OK; return GRPC_CALL_OK;
} else { } else {
*requested_call_array_add(requested_calls) = *rc; rc->next = *requests;
*requests = rc;
gpr_mu_unlock(&server->mu_call); gpr_mu_unlock(&server->mu_call);
return GRPC_CALL_OK; return GRPC_CALL_OK;
} }
@ -1061,22 +1043,23 @@ grpc_call_error grpc_server_request_call(
grpc_metadata_array *initial_metadata, grpc_metadata_array *initial_metadata,
grpc_completion_queue *cq_bound_to_call, grpc_completion_queue *cq_bound_to_call,
grpc_completion_queue *cq_for_notification, void *tag) { grpc_completion_queue *cq_for_notification, void *tag) {
requested_call rc; requested_call *rc = gpr_malloc(sizeof(*rc));
GRPC_SERVER_LOG_REQUEST_CALL(GPR_INFO, server, call, details, GRPC_SERVER_LOG_REQUEST_CALL(GPR_INFO, server, call, details,
initial_metadata, cq_bound_to_call, initial_metadata, cq_bound_to_call,
cq_for_notification, tag); cq_for_notification, tag);
if (!grpc_cq_is_server_cq(cq_for_notification)) { if (!grpc_cq_is_server_cq(cq_for_notification)) {
gpr_free(rc);
return GRPC_CALL_ERROR_NOT_SERVER_COMPLETION_QUEUE; return GRPC_CALL_ERROR_NOT_SERVER_COMPLETION_QUEUE;
} }
grpc_cq_begin_op(cq_for_notification, NULL); grpc_cq_begin_op(cq_for_notification);
rc.type = BATCH_CALL; rc->type = BATCH_CALL;
rc.tag = tag; rc->tag = tag;
rc.cq_bound_to_call = cq_bound_to_call; rc->cq_bound_to_call = cq_bound_to_call;
rc.cq_for_notification = cq_for_notification; rc->cq_for_notification = cq_for_notification;
rc.call = call; rc->call = call;
rc.data.batch.details = details; rc->data.batch.details = details;
rc.data.batch.initial_metadata = initial_metadata; rc->data.batch.initial_metadata = initial_metadata;
return queue_call_request(server, &rc); return queue_call_request(server, rc);
} }
grpc_call_error grpc_server_request_registered_call( grpc_call_error grpc_server_request_registered_call(
@ -1084,22 +1067,23 @@ grpc_call_error grpc_server_request_registered_call(
grpc_metadata_array *initial_metadata, grpc_byte_buffer **optional_payload, grpc_metadata_array *initial_metadata, grpc_byte_buffer **optional_payload,
grpc_completion_queue *cq_bound_to_call, grpc_completion_queue *cq_bound_to_call,
grpc_completion_queue *cq_for_notification, void *tag) { grpc_completion_queue *cq_for_notification, void *tag) {
requested_call rc; requested_call *rc = gpr_malloc(sizeof(*rc));
registered_method *registered_method = rm; registered_method *registered_method = rm;
if (!grpc_cq_is_server_cq(cq_for_notification)) { if (!grpc_cq_is_server_cq(cq_for_notification)) {
gpr_free(rc);
return GRPC_CALL_ERROR_NOT_SERVER_COMPLETION_QUEUE; return GRPC_CALL_ERROR_NOT_SERVER_COMPLETION_QUEUE;
} }
grpc_cq_begin_op(cq_for_notification, NULL); grpc_cq_begin_op(cq_for_notification);
rc.type = REGISTERED_CALL; rc->type = REGISTERED_CALL;
rc.tag = tag; rc->tag = tag;
rc.cq_bound_to_call = cq_bound_to_call; rc->cq_bound_to_call = cq_bound_to_call;
rc.cq_for_notification = cq_for_notification; rc->cq_for_notification = cq_for_notification;
rc.call = call; rc->call = call;
rc.data.registered.registered_method = registered_method; rc->data.registered.registered_method = registered_method;
rc.data.registered.deadline = deadline; rc->data.registered.deadline = deadline;
rc.data.registered.initial_metadata = initial_metadata; rc->data.registered.initial_metadata = initial_metadata;
rc.data.registered.optional_payload = optional_payload; rc->data.registered.optional_payload = optional_payload;
return queue_call_request(server, &rc); return queue_call_request(server, rc);
} }
static void publish_registered_or_batch(grpc_call *call, int success, static void publish_registered_or_batch(grpc_call *call, int success,
@ -1167,7 +1151,11 @@ static void begin_call(grpc_server *server, call_data *calld,
GRPC_CALL_INTERNAL_REF(calld->call, "server"); GRPC_CALL_INTERNAL_REF(calld->call, "server");
grpc_call_start_ioreq_and_call_back(calld->call, req, r - req, publish, grpc_call_start_ioreq_and_call_back(calld->call, req, r - req, publish,
rc->tag); rc);
}
static void done_request_event(void *req, grpc_cq_completion *c) {
gpr_free(req);
} }
static void fail_call(grpc_server *server, requested_call *rc) { static void fail_call(grpc_server *server, requested_call *rc) {
@ -1180,15 +1168,16 @@ static void fail_call(grpc_server *server, requested_call *rc) {
rc->data.registered.initial_metadata->count = 0; rc->data.registered.initial_metadata->count = 0;
break; break;
} }
grpc_cq_end_op(rc->cq_for_notification, rc->tag, NULL, 0); grpc_cq_end_op(rc->cq_for_notification, rc->tag, 0, done_request_event, rc, &rc->completion);
} }
static void publish_registered_or_batch(grpc_call *call, int success, static void publish_registered_or_batch(grpc_call *call, int success,
void *tag) { void *prc) {
grpc_call_element *elem = grpc_call_element *elem =
grpc_call_stack_element(grpc_call_get_call_stack(call), 0); grpc_call_stack_element(grpc_call_get_call_stack(call), 0);
requested_call *rc = prc;
call_data *calld = elem->call_data; call_data *calld = elem->call_data;
grpc_cq_end_op(calld->cq_new, tag, call, success); grpc_cq_end_op(calld->cq_new, rc->tag, success, done_request_event, rc, &rc->completion);
} }
const grpc_channel_args *grpc_server_get_channel_args(grpc_server *server) { const grpc_channel_args *grpc_server_get_channel_args(grpc_server *server) {

24
test/core/surface/completion_queue_test.c
Unescape View File

@ -74,17 +74,20 @@ static void test_wait_empty(void) {
shutdown_and_destroy(cc); shutdown_and_destroy(cc);
} }
static void do_nothing_end_completion(void *arg, grpc_cq_completion *c) {}
static void test_cq_end_op(void) { static void test_cq_end_op(void) {
grpc_event ev; grpc_event ev;
grpc_completion_queue *cc; grpc_completion_queue *cc;
grpc_cq_completion completion;
void *tag = create_test_tag(); void *tag = create_test_tag();
LOG_TEST("test_cq_end_op"); LOG_TEST("test_cq_end_op");
cc = grpc_completion_queue_create(); cc = grpc_completion_queue_create();
grpc_cq_begin_op(cc, NULL); grpc_cq_begin_op(cc);
grpc_cq_end_op(cc, tag, NULL, 1); grpc_cq_end_op(cc, tag, 1, do_nothing_end_completion, NULL, &completion);
ev = grpc_completion_queue_next(cc, gpr_inf_past); ev = grpc_completion_queue_next(cc, gpr_inf_past);
GPR_ASSERT(ev.type == GRPC_OP_COMPLETE); GPR_ASSERT(ev.type == GRPC_OP_COMPLETE);
@ -120,6 +123,7 @@ static void test_pluck(void) {
grpc_event ev; grpc_event ev;
grpc_completion_queue *cc; grpc_completion_queue *cc;
void *tags[128]; void *tags[128];
grpc_cq_completion completions[GPR_ARRAY_SIZE(tags)];
unsigned i, j; unsigned i, j;
LOG_TEST("test_pluck"); LOG_TEST("test_pluck");
@ -134,8 +138,8 @@ static void test_pluck(void) {
cc = grpc_completion_queue_create(); cc = grpc_completion_queue_create();
for (i = 0; i < GPR_ARRAY_SIZE(tags); i++) { for (i = 0; i < GPR_ARRAY_SIZE(tags); i++) {
grpc_cq_begin_op(cc, NULL); grpc_cq_begin_op(cc);
grpc_cq_end_op(cc, tags[i], NULL, 1); grpc_cq_end_op(cc, tags[i], 1, do_nothing_end_completion, NULL, &completions[i]);
} }
for (i = 0; i < GPR_ARRAY_SIZE(tags); i++) { for (i = 0; i < GPR_ARRAY_SIZE(tags); i++) {
@ -144,8 +148,8 @@ static void test_pluck(void) {
} }
for (i = 0; i < GPR_ARRAY_SIZE(tags); i++) { for (i = 0; i < GPR_ARRAY_SIZE(tags); i++) {
grpc_cq_begin_op(cc, NULL); grpc_cq_begin_op(cc);
grpc_cq_end_op(cc, tags[i], NULL, 1); grpc_cq_end_op(cc, tags[i], 1, do_nothing_end_completion, NULL, &completions[i]);
} }
for (i = 0; i < GPR_ARRAY_SIZE(tags); i++) { for (i = 0; i < GPR_ARRAY_SIZE(tags); i++) {
@ -174,6 +178,10 @@ gpr_timespec ten_seconds_time(void) {
return GRPC_TIMEOUT_SECONDS_TO_DEADLINE(1); return GRPC_TIMEOUT_SECONDS_TO_DEADLINE(1);
} }
static void free_completion(void *arg, grpc_cq_completion *completion) {
gpr_free(completion);
}
static void producer_thread(void *arg) { static void producer_thread(void *arg) {
test_thread_options *opt = arg; test_thread_options *opt = arg;
int i; int i;
@ -184,7 +192,7 @@ static void producer_thread(void *arg) {
gpr_log(GPR_INFO, "producer %d phase 1", opt->id); gpr_log(GPR_INFO, "producer %d phase 1", opt->id);
for (i = 0; i < TEST_THREAD_EVENTS; i++) { for (i = 0; i < TEST_THREAD_EVENTS; i++) {
grpc_cq_begin_op(opt->cc, NULL); grpc_cq_begin_op(opt->cc);
} }
gpr_log(GPR_INFO, "producer %d phase 1 done", opt->id); gpr_log(GPR_INFO, "producer %d phase 1 done", opt->id);
@ -193,7 +201,7 @@ static void producer_thread(void *arg) {
gpr_log(GPR_INFO, "producer %d phase 2", opt->id); gpr_log(GPR_INFO, "producer %d phase 2", opt->id);
for (i = 0; i < TEST_THREAD_EVENTS; i++) { for (i = 0; i < TEST_THREAD_EVENTS; i++) {
grpc_cq_end_op(opt->cc, (void *)(gpr_intptr)1, NULL, 1); grpc_cq_end_op(opt->cc, (void *)(gpr_intptr)1, 1, free_completion, NULL, gpr_malloc(sizeof(grpc_cq_completion)));
opt->events_triggered++; opt->events_triggered++;
} }

Write Preview
Loading…
Cancel
Save