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

Split cq_data into cq_next_data, cq_pluck_data

Browse Source
pull/11317/head
Craig Tiller 8 years ago
parent c019572f1a
commit be290851c0
3 changed files with 217 additions and 173 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. 382
      src/core/lib/surface/completion_queue.c
  2. 3
      src/core/lib/surface/completion_queue.h
  3. 5
      src/core/lib/surface/server.c

382
src/core/lib/surface/completion_queue.c
Unescape View File

@ -203,7 +203,10 @@ static const cq_poller_vtable g_poller_vtable_by_poller_type[] = {
typedef struct cq_vtable { typedef struct cq_vtable {
grpc_cq_completion_type cq_completion_type; grpc_cq_completion_type cq_completion_type;
size_t (*size)(); size_t data_size;
void (*init)(void *data);
void (*shutdown)(grpc_exec_ctx *exec_ctx, grpc_completion_queue *cq);
void (*destroy)(void *data);
void (*begin_op)(grpc_completion_queue *cq, void *tag); void (*begin_op)(grpc_completion_queue *cq, void *tag);
void (*end_op)(grpc_exec_ctx *exec_ctx, grpc_completion_queue *cq, void *tag, void (*end_op)(grpc_exec_ctx *exec_ctx, grpc_completion_queue *cq, void *tag,
grpc_error *error, grpc_error *error,
@ -232,25 +235,30 @@ typedef struct grpc_cq_event_queue {
gpr_atm num_queue_items; gpr_atm num_queue_items;
} grpc_cq_event_queue; } grpc_cq_event_queue;
/* TODO: sreek Refactor this based on the completion_type. Put completion-type typedef struct cq_next_data {
* specific data in a different structure (and co-allocate memory for it along /** Completed events for completion-queues of type GRPC_CQ_NEXT */
* with completion queue + pollset )*/ grpc_cq_event_queue queue;
typedef struct cq_data {
gpr_mu *mu; /** Number of pending events (+1 if we're not shutdown) */
gpr_refcount pending_events;
/** Counter of how many things have ever been queued on this completion queue
useful for avoiding locks to check the queue */
gpr_atm things_queued_ever;
/** 0 initially, 1 once we've begun shutting down */
gpr_atm shutdown;
int shutdown_called;
} cq_next_data;
typedef struct cq_pluck_data {
/** Completed events for completion-queues of type GRPC_CQ_PLUCK */ /** Completed events for completion-queues of type GRPC_CQ_PLUCK */
grpc_cq_completion completed_head; grpc_cq_completion completed_head;
grpc_cq_completion *completed_tail; grpc_cq_completion *completed_tail;
/** Completed events for completion-queues of type GRPC_CQ_NEXT */
grpc_cq_event_queue queue;
/** Number of pending events (+1 if we're not shutdown) */ /** Number of pending events (+1 if we're not shutdown) */
gpr_refcount pending_events; gpr_refcount pending_events;
/** Once owning_refs drops to zero, we will destroy the cq */
gpr_refcount owning_refs;
/** Counter of how many things have ever been queued on this completion queue /** Counter of how many things have ever been queued on this completion queue
useful for avoiding locks to check the queue */ useful for avoiding locks to check the queue */
gpr_atm things_queued_ever; gpr_atm things_queued_ever;
@ -259,34 +267,42 @@ typedef struct cq_data {
gpr_atm shutdown; gpr_atm shutdown;
int shutdown_called; int shutdown_called;
int is_server_cq;
int num_pluckers; int num_pluckers;
int num_polls;
plucker pluckers[GRPC_MAX_COMPLETION_QUEUE_PLUCKERS]; plucker pluckers[GRPC_MAX_COMPLETION_QUEUE_PLUCKERS];
grpc_closure pollset_shutdown_done; } cq_pluck_data;
/* Completion queue structure */
struct grpc_completion_queue {
/** Once owning_refs drops to zero, we will destroy the cq */
gpr_refcount owning_refs;
gpr_mu *mu;
const cq_vtable *vtable;
const cq_poller_vtable *poller_vtable;
#ifndef NDEBUG #ifndef NDEBUG
void **outstanding_tags; void **outstanding_tags;
size_t outstanding_tag_count; size_t outstanding_tag_count;
size_t outstanding_tag_capacity; size_t outstanding_tag_capacity;
#endif #endif
} cq_data;
/* Completion queue structure */ grpc_closure pollset_shutdown_done;
struct grpc_completion_queue { int num_polls;
cq_data data;
const cq_vtable *vtable;
const cq_poller_vtable *poller_vtable;
}; };
/* Forward declarations */ /* Forward declarations */
static void cq_finish_shutdown(grpc_exec_ctx *exec_ctx, static void cq_finish_shutdown_next(grpc_exec_ctx *exec_ctx,
grpc_completion_queue *cq); grpc_completion_queue *cq);
static void cq_finish_shutdown_pluck(grpc_exec_ctx *exec_ctx,
static size_t cq_size(grpc_completion_queue *cq); grpc_completion_queue *cq);
static void cq_shutdown_next(grpc_exec_ctx *exec_ctx,
static void cq_begin_op(grpc_completion_queue *cq, void *tag); grpc_completion_queue *cq);
static void cq_shutdown_pluck(grpc_exec_ctx *exec_ctx,
grpc_completion_queue *cq);
static void cq_begin_op_for_next(grpc_completion_queue *cq, void *tag);
static void cq_begin_op_for_pluck(grpc_completion_queue *cq, void *tag);
static void cq_end_op_for_next(grpc_exec_ctx *exec_ctx, static void cq_end_op_for_next(grpc_exec_ctx *exec_ctx,
grpc_completion_queue *cq, void *tag, grpc_completion_queue *cq, void *tag,
@ -310,26 +326,36 @@ static grpc_event cq_next(grpc_completion_queue *cq, gpr_timespec deadline,
static grpc_event cq_pluck(grpc_completion_queue *cq, void *tag, static grpc_event cq_pluck(grpc_completion_queue *cq, void *tag,
gpr_timespec deadline, void *reserved); gpr_timespec deadline, void *reserved);
static void cq_init_next(void *data);
static void cq_init_pluck(void *data);
static void cq_destroy_next(void *data);
static void cq_destroy_pluck(void *data);
/* Completion queue vtables based on the completion-type */ /* Completion queue vtables based on the completion-type */
static const cq_vtable g_cq_vtable[] = { static const cq_vtable g_cq_vtable[] = {
/* GRPC_CQ_NEXT */ /* GRPC_CQ_NEXT */
{.cq_completion_type = GRPC_CQ_NEXT, {.cq_completion_type = GRPC_CQ_NEXT,
.size = cq_size, .init = cq_init_next,
.begin_op = cq_begin_op, .shutdown = cq_shutdown_next,
.destroy = cq_destroy_next,
.begin_op = cq_begin_op_for_next,
.end_op = cq_end_op_for_next, .end_op = cq_end_op_for_next,
.next = cq_next, .next = cq_next,
.pluck = NULL}, .pluck = NULL},
/* GRPC_CQ_PLUCK */ /* GRPC_CQ_PLUCK */
{.cq_completion_type = GRPC_CQ_PLUCK, {.cq_completion_type = GRPC_CQ_PLUCK,
.size = cq_size, .init = cq_init_pluck,
.begin_op = cq_begin_op, .shutdown = cq_shutdown_pluck,
.destroy = cq_destroy_pluck,
.begin_op = cq_begin_op_for_pluck,
.end_op = cq_end_op_for_pluck, .end_op = cq_end_op_for_pluck,
.next = NULL, .next = NULL,
.pluck = cq_pluck}, .pluck = cq_pluck},
}; };
#define POLLSET_FROM_CQ(cq) ((grpc_pollset *)(cq + 1)) #define DATA_FROM_CQ(cq) ((void *)(cq + 1))
#define CQ_FROM_POLLSET(ps) (((grpc_completion_queue *)ps) - 1) #define POLLSET_FROM_CQ(cq) \
((grpc_pollset *)(cq->vtable->data_size + (char *)DATA_FROM_CQ(cq)))
grpc_tracer_flag grpc_cq_pluck_trace = GRPC_TRACER_INITIALIZER(true); grpc_tracer_flag grpc_cq_pluck_trace = GRPC_TRACER_INITIALIZER(true);
grpc_tracer_flag grpc_cq_event_timeout_trace = GRPC_TRACER_INITIALIZER(true); grpc_tracer_flag grpc_cq_event_timeout_trace = GRPC_TRACER_INITIALIZER(true);
@ -381,12 +407,6 @@ static long cq_event_queue_num_items(grpc_cq_event_queue *q) {
return (long)gpr_atm_no_barrier_load(&q->num_queue_items); return (long)gpr_atm_no_barrier_load(&q->num_queue_items);
} }
static size_t cq_size(grpc_completion_queue *cq) {
/* Size of the completion queue and the size of the pollset whose memory is
allocated right after that of completion queue */
return sizeof(grpc_completion_queue) + cq->poller_vtable->size();
}
grpc_completion_queue *grpc_completion_queue_create_internal( grpc_completion_queue *grpc_completion_queue_create_internal(
grpc_cq_completion_type completion_type, grpc_cq_completion_type completion_type,
grpc_cq_polling_type polling_type) { grpc_cq_polling_type polling_type) {
@ -403,41 +423,57 @@ grpc_completion_queue *grpc_completion_queue_create_internal(
const cq_poller_vtable *poller_vtable = const cq_poller_vtable *poller_vtable =
&g_poller_vtable_by_poller_type[polling_type]; &g_poller_vtable_by_poller_type[polling_type];
cq = gpr_zalloc(sizeof(grpc_completion_queue) + poller_vtable->size()); cq = gpr_zalloc(sizeof(grpc_completion_queue) + vtable->data_size +
cq_data *cqd = &cq->data; poller_vtable->size());
cq->vtable = vtable; cq->vtable = vtable;
cq->poller_vtable = poller_vtable; cq->poller_vtable = poller_vtable;
poller_vtable->init(POLLSET_FROM_CQ(cq), &cq->data.mu); /* One for destroy(), one for pollset_shutdown */
gpr_ref_init(&cq->owning_refs, 2);
#ifndef NDEBUG poller_vtable->init(POLLSET_FROM_CQ(cq), &cq->mu);
cqd->outstanding_tags = NULL; vtable->init(DATA_FROM_CQ(cq));
cqd->outstanding_tag_capacity = 0;
#endif grpc_closure_init(&cq->pollset_shutdown_done, on_pollset_shutdown_done, cq,
grpc_schedule_on_exec_ctx);
GPR_TIMER_END("grpc_completion_queue_create_internal", 0);
return cq;
}
static void cq_init_next(void *ptr) {
cq_next_data *cqd = ptr;
/* Initial ref is dropped by grpc_completion_queue_shutdown */
gpr_ref_init(&cqd->pending_events, 1);
gpr_atm_no_barrier_store(&cqd->shutdown, 0);
cqd->shutdown_called = 0;
gpr_atm_no_barrier_store(&cqd->things_queued_ever, 0);
cq_event_queue_init(&cqd->queue);
}
static void cq_destroy_next(void *ptr) {
cq_next_data *cqd = ptr;
GPR_ASSERT(cq_event_queue_num_items(&cqd->queue) == 0);
cq_event_queue_destroy(&cqd->queue);
}
static void cq_init_pluck(void *ptr) {
cq_pluck_data *cqd = ptr;
/* Initial ref is dropped by grpc_completion_queue_shutdown */ /* Initial ref is dropped by grpc_completion_queue_shutdown */
gpr_ref_init(&cqd->pending_events, 1); gpr_ref_init(&cqd->pending_events, 1);
/* One for destroy(), one for pollset_shutdown */
gpr_ref_init(&cqd->owning_refs, 2);
cqd->completed_tail = &cqd->completed_head; cqd->completed_tail = &cqd->completed_head;
cqd->completed_head.next = (uintptr_t)cqd->completed_tail; cqd->completed_head.next = (uintptr_t)cqd->completed_tail;
gpr_atm_no_barrier_store(&cqd->shutdown, 0); gpr_atm_no_barrier_store(&cqd->shutdown, 0);
cqd->shutdown_called = 0; cqd->shutdown_called = 0;
cqd->is_server_cq = 0;
cqd->num_pluckers = 0; cqd->num_pluckers = 0;
cqd->num_polls = 0;
gpr_atm_no_barrier_store(&cqd->things_queued_ever, 0); gpr_atm_no_barrier_store(&cqd->things_queued_ever, 0);
#ifndef NDEBUG }
cqd->outstanding_tag_count = 0;
#endif
cq_event_queue_init(&cqd->queue);
grpc_closure_init(&cqd->pollset_shutdown_done, on_pollset_shutdown_done, cq,
grpc_schedule_on_exec_ctx);
GPR_TIMER_END("grpc_completion_queue_create_internal", 0);
return cq; static void cq_destroy_pluck(void *ptr) {
cq_pluck_data *cqd = ptr;
GPR_ASSERT(cqd->completed_head.next == (uintptr_t)&cqd->completed_head);
} }
grpc_cq_completion_type grpc_get_cq_completion_type(grpc_completion_queue *cq) { grpc_cq_completion_type grpc_get_cq_completion_type(grpc_completion_queue *cq) {
@ -446,23 +482,21 @@ grpc_cq_completion_type grpc_get_cq_completion_type(grpc_completion_queue *cq) {
int grpc_get_cq_poll_num(grpc_completion_queue *cq) { int grpc_get_cq_poll_num(grpc_completion_queue *cq) {
int cur_num_polls; int cur_num_polls;
gpr_mu_lock(cq->data.mu); gpr_mu_lock(cq->mu);
cur_num_polls = cq->data.num_polls; cur_num_polls = cq->num_polls;
gpr_mu_unlock(cq->data.mu); gpr_mu_unlock(cq->mu);
return cur_num_polls; return cur_num_polls;
} }
#ifdef GRPC_CQ_REF_COUNT_DEBUG #ifdef GRPC_CQ_REF_COUNT_DEBUG
void grpc_cq_internal_ref(grpc_completion_queue *cq, const char *reason, void grpc_cq_internal_ref(grpc_completion_queue *cq, const char *reason,
const char *file, int line) { const char *file, int line) {
cq_data *cqd = &cq->data;
gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, "CQ:%p ref %d -> %d %s", cq, gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, "CQ:%p ref %d -> %d %s", cq,
(int)cqd->owning_refs.count, (int)cqd->owning_refs.count + 1, reason); (int)cq->owning_refs.count, (int)cq->owning_refs.count + 1, reason);
#else #else
void grpc_cq_internal_ref(grpc_completion_queue *cq) { void grpc_cq_internal_ref(grpc_completion_queue *cq) {
cq_data *cqd = &cq->data;
#endif #endif
gpr_ref(&cqd->owning_refs); gpr_ref(&cq->owning_refs);
} }
static void on_pollset_shutdown_done(grpc_exec_ctx *exec_ctx, void *arg, static void on_pollset_shutdown_done(grpc_exec_ctx *exec_ctx, void *arg,
@ -480,61 +514,63 @@ void grpc_cq_internal_unref(grpc_completion_queue *cq, const char *reason,
#else #else
void grpc_cq_internal_unref(grpc_exec_ctx *exec_ctx, void grpc_cq_internal_unref(grpc_exec_ctx *exec_ctx,
grpc_completion_queue *cq) { grpc_completion_queue *cq) {
cq_data *cqd = &cq->data;
#endif #endif
if (gpr_unref(&cqd->owning_refs)) { if (gpr_unref(&cq->owning_refs)) {
GPR_ASSERT(cqd->completed_head.next == (uintptr_t)&cqd->completed_head); cq->vtable->destroy(DATA_FROM_CQ(cq));
cq->poller_vtable->destroy(exec_ctx, POLLSET_FROM_CQ(cq)); cq->poller_vtable->destroy(exec_ctx, POLLSET_FROM_CQ(cq));
cq_event_queue_destroy(&cqd->queue);
#ifndef NDEBUG #ifndef NDEBUG
gpr_free(cqd->outstanding_tags); gpr_free(cq->outstanding_tags);
#endif #endif
gpr_free(cq); gpr_free(cq);
} }
} }
static void cq_begin_op(grpc_completion_queue *cq, void *tag) { static void cq_begin_op_for_next(grpc_completion_queue *cq, void *tag) {
cq_data *cqd = &cq->data; cq_next_data *cqd = DATA_FROM_CQ(cq);
#ifndef NDEBUG GPR_ASSERT(!cqd->shutdown_called);
gpr_mu_lock(cqd->mu); gpr_ref(&cqd->pending_events);
}
static void cq_begin_op_for_pluck(grpc_completion_queue *cq, void *tag) {
cq_pluck_data *cqd = DATA_FROM_CQ(cq);
GPR_ASSERT(!cqd->shutdown_called); GPR_ASSERT(!cqd->shutdown_called);
if (cqd->outstanding_tag_count == cqd->outstanding_tag_capacity) {
cqd->outstanding_tag_capacity =
GPR_MAX(4, 2 * cqd->outstanding_tag_capacity);
cqd->outstanding_tags =
gpr_realloc(cqd->outstanding_tags, sizeof(*cqd->outstanding_tags) *
cqd->outstanding_tag_capacity);
}
cqd->outstanding_tags[cqd->outstanding_tag_count++] = tag;
gpr_mu_unlock(cqd->mu);
#endif
gpr_ref(&cqd->pending_events); gpr_ref(&cqd->pending_events);
} }
void grpc_cq_begin_op(grpc_completion_queue *cq, void *tag) { void grpc_cq_begin_op(grpc_completion_queue *cq, void *tag) {
#ifndef NDEBUG
gpr_mu_lock(cq->mu);
if (cq->outstanding_tag_count == cq->outstanding_tag_capacity) {
cq->outstanding_tag_capacity = GPR_MAX(4, 2 * cq->outstanding_tag_capacity);
cq->outstanding_tags =
gpr_realloc(cq->outstanding_tags, sizeof(*cq->outstanding_tags) *
cq->outstanding_tag_capacity);
}
cq->outstanding_tags[cq->outstanding_tag_count++] = tag;
gpr_mu_unlock(cq->mu);
#endif
cq->vtable->begin_op(cq, tag); cq->vtable->begin_op(cq, tag);
} }
#ifndef NDEBUG #ifndef NDEBUG
static void cq_check_tag(grpc_completion_queue *cq, void *tag, bool lock_cq) { static void cq_check_tag(grpc_completion_queue *cq, void *tag, bool lock_cq) {
cq_data *cqd = &cq->data;
int found = 0; int found = 0;
if (lock_cq) { if (lock_cq) {
gpr_mu_lock(cqd->mu); gpr_mu_lock(cq->mu);
} }
for (int i = 0; i < (int)cqd->outstanding_tag_count; i++) { for (int i = 0; i < (int)cq->outstanding_tag_count; i++) {
if (cqd->outstanding_tags[i] == tag) { if (cq->outstanding_tags[i] == tag) {
cqd->outstanding_tag_count--; cq->outstanding_tag_count--;
GPR_SWAP(void *, cqd->outstanding_tags[i], GPR_SWAP(void *, cq->outstanding_tags[i],
cqd->outstanding_tags[cqd->outstanding_tag_count]); cq->outstanding_tags[cq->outstanding_tag_count]);
found = 1; found = 1;
break; break;
} }
} }
if (lock_cq) { if (lock_cq) {
gpr_mu_unlock(cqd->mu); gpr_mu_unlock(cq->mu);
} }
GPR_ASSERT(found); GPR_ASSERT(found);
@ -568,7 +604,7 @@ static void cq_end_op_for_next(grpc_exec_ctx *exec_ctx,
} }
} }
cq_data *cqd = &cq->data; cq_next_data *cqd = DATA_FROM_CQ(cq);
int is_success = (error == GRPC_ERROR_NONE); int is_success = (error == GRPC_ERROR_NONE);
storage->tag = tag; storage->tag = tag;
@ -586,10 +622,10 @@ static void cq_end_op_for_next(grpc_exec_ctx *exec_ctx,
if (!shutdown) { if (!shutdown) {
/* Only kick if this is the first item queued */ /* Only kick if this is the first item queued */
if (is_first) { if (is_first) {
gpr_mu_lock(cqd->mu); gpr_mu_lock(cq->mu);
grpc_error *kick_error = grpc_error *kick_error =
cq->poller_vtable->kick(POLLSET_FROM_CQ(cq), NULL); cq->poller_vtable->kick(POLLSET_FROM_CQ(cq), NULL);
gpr_mu_unlock(cqd->mu); gpr_mu_unlock(cq->mu);
if (kick_error != GRPC_ERROR_NONE) { if (kick_error != GRPC_ERROR_NONE) {
const char *msg = grpc_error_string(kick_error); const char *msg = grpc_error_string(kick_error);
@ -598,9 +634,9 @@ static void cq_end_op_for_next(grpc_exec_ctx *exec_ctx,
} }
} }
} else { } else {
gpr_mu_lock(cqd->mu); gpr_mu_lock(cq->mu);
cq_finish_shutdown(exec_ctx, cq); cq_finish_shutdown_next(exec_ctx, cq);
gpr_mu_unlock(cqd->mu); gpr_mu_unlock(cq->mu);
} }
GPR_TIMER_END("cq_end_op_for_next", 0); GPR_TIMER_END("cq_end_op_for_next", 0);
@ -617,7 +653,7 @@ static void cq_end_op_for_pluck(grpc_exec_ctx *exec_ctx,
void *done_arg, void *done_arg,
grpc_cq_completion *storage), grpc_cq_completion *storage),
void *done_arg, grpc_cq_completion *storage) { void *done_arg, grpc_cq_completion *storage) {
cq_data *cqd = &cq->data; cq_pluck_data *cqd = DATA_FROM_CQ(cq);
int is_success = (error == GRPC_ERROR_NONE); int is_success = (error == GRPC_ERROR_NONE);
GPR_TIMER_BEGIN("cq_end_op_for_pluck", 0); GPR_TIMER_BEGIN("cq_end_op_for_pluck", 0);
@ -641,7 +677,7 @@ static void cq_end_op_for_pluck(grpc_exec_ctx *exec_ctx,
storage->done_arg = done_arg; storage->done_arg = done_arg;
storage->next = ((uintptr_t)&cqd->completed_head) | ((uintptr_t)(is_success)); storage->next = ((uintptr_t)&cqd->completed_head) | ((uintptr_t)(is_success));
gpr_mu_lock(cqd->mu); gpr_mu_lock(cq->mu);
cq_check_tag(cq, tag, false); /* Used in debug builds only */ cq_check_tag(cq, tag, false); /* Used in debug builds only */
/* Add to the list of completions */ /* Add to the list of completions */
@ -663,7 +699,7 @@ static void cq_end_op_for_pluck(grpc_exec_ctx *exec_ctx,
grpc_error *kick_error = grpc_error *kick_error =
cq->poller_vtable->kick(POLLSET_FROM_CQ(cq), pluck_worker); cq->poller_vtable->kick(POLLSET_FROM_CQ(cq), pluck_worker);
gpr_mu_unlock(cqd->mu); gpr_mu_unlock(cq->mu);
if (kick_error != GRPC_ERROR_NONE) { if (kick_error != GRPC_ERROR_NONE) {
const char *msg = grpc_error_string(kick_error); const char *msg = grpc_error_string(kick_error);
@ -672,8 +708,8 @@ static void cq_end_op_for_pluck(grpc_exec_ctx *exec_ctx,
GRPC_ERROR_UNREF(kick_error); GRPC_ERROR_UNREF(kick_error);
} }
} else { } else {
cq_finish_shutdown(exec_ctx, cq); cq_finish_shutdown_pluck(exec_ctx, cq);
gpr_mu_unlock(cqd->mu); gpr_mu_unlock(cq->mu);
} }
GPR_TIMER_END("cq_end_op_for_pluck", 0); GPR_TIMER_END("cq_end_op_for_pluck", 0);
@ -701,7 +737,7 @@ typedef struct {
static bool cq_is_next_finished(grpc_exec_ctx *exec_ctx, void *arg) { static bool cq_is_next_finished(grpc_exec_ctx *exec_ctx, void *arg) {
cq_is_finished_arg *a = arg; cq_is_finished_arg *a = arg;
grpc_completion_queue *cq = a->cq; grpc_completion_queue *cq = a->cq;
cq_data *cqd = &cq->data; cq_next_data *cqd = DATA_FROM_CQ(cq);
GPR_ASSERT(a->stolen_completion == NULL); GPR_ASSERT(a->stolen_completion == NULL);
gpr_atm current_last_seen_things_queued_ever = gpr_atm current_last_seen_things_queued_ever =
@ -728,18 +764,16 @@ static bool cq_is_next_finished(grpc_exec_ctx *exec_ctx, void *arg) {
static void dump_pending_tags(grpc_completion_queue *cq) { static void dump_pending_tags(grpc_completion_queue *cq) {
if (!GRPC_TRACER_ON(grpc_trace_pending_tags)) return; if (!GRPC_TRACER_ON(grpc_trace_pending_tags)) return;
cq_data *cqd = &cq->data;
gpr_strvec v; gpr_strvec v;
gpr_strvec_init(&v); gpr_strvec_init(&v);
gpr_strvec_add(&v, gpr_strdup("PENDING TAGS:")); gpr_strvec_add(&v, gpr_strdup("PENDING TAGS:"));
gpr_mu_lock(cqd->mu); gpr_mu_lock(cq->mu);
for (size_t i = 0; i < cqd->outstanding_tag_count; i++) { for (size_t i = 0; i < cq->outstanding_tag_count; i++) {
char *s; char *s;
gpr_asprintf(&s, " %p", cqd->outstanding_tags[i]); gpr_asprintf(&s, " %p", cq->outstanding_tags[i]);
gpr_strvec_add(&v, s); gpr_strvec_add(&v, s);
} }
gpr_mu_unlock(cqd->mu); gpr_mu_unlock(cq->mu);
char *out = gpr_strvec_flatten(&v, NULL); char *out = gpr_strvec_flatten(&v, NULL);
gpr_strvec_destroy(&v); gpr_strvec_destroy(&v);
gpr_log(GPR_DEBUG, "%s", out); gpr_log(GPR_DEBUG, "%s", out);
@ -753,7 +787,7 @@ static grpc_event cq_next(grpc_completion_queue *cq, gpr_timespec deadline,
void *reserved) { void *reserved) {
grpc_event ret; grpc_event ret;
gpr_timespec now; gpr_timespec now;
cq_data *cqd = &cq->data; cq_next_data *cqd = DATA_FROM_CQ(cq);
GPR_TIMER_BEGIN("grpc_completion_queue_next", 0); GPR_TIMER_BEGIN("grpc_completion_queue_next", 0);
@ -842,11 +876,11 @@ static grpc_event cq_next(grpc_completion_queue *cq, gpr_timespec deadline,
} }
/* The main polling work happens in grpc_pollset_work */ /* The main polling work happens in grpc_pollset_work */
gpr_mu_lock(cqd->mu); gpr_mu_lock(cq->mu);
cqd->num_polls++; cq->num_polls++;
grpc_error *err = cq->poller_vtable->work(&exec_ctx, POLLSET_FROM_CQ(cq), grpc_error *err = cq->poller_vtable->work(&exec_ctx, POLLSET_FROM_CQ(cq),
NULL, now, iteration_deadline); NULL, now, iteration_deadline);
gpr_mu_unlock(cqd->mu); gpr_mu_unlock(cq->mu);
if (err != GRPC_ERROR_NONE) { if (err != GRPC_ERROR_NONE) {
const char *msg = grpc_error_string(err); const char *msg = grpc_error_string(err);
@ -868,9 +902,9 @@ static grpc_event cq_next(grpc_completion_queue *cq, gpr_timespec deadline,
if (cq_event_queue_num_items(&cqd->queue) > 0 && if (cq_event_queue_num_items(&cqd->queue) > 0 &&
gpr_atm_no_barrier_load(&cqd->shutdown) == 0) { gpr_atm_no_barrier_load(&cqd->shutdown) == 0) {
gpr_mu_lock(cqd->mu); gpr_mu_lock(cq->mu);
cq->poller_vtable->kick(POLLSET_FROM_CQ(cq), NULL); cq->poller_vtable->kick(POLLSET_FROM_CQ(cq), NULL);
gpr_mu_unlock(cqd->mu); gpr_mu_unlock(cq->mu);
} }
GPR_TIMER_END("grpc_completion_queue_next", 0); GPR_TIMER_END("grpc_completion_queue_next", 0);
@ -885,7 +919,7 @@ grpc_event grpc_completion_queue_next(grpc_completion_queue *cq,
static int add_plucker(grpc_completion_queue *cq, void *tag, static int add_plucker(grpc_completion_queue *cq, void *tag,
grpc_pollset_worker **worker) { grpc_pollset_worker **worker) {
cq_data *cqd = &cq->data; cq_pluck_data *cqd = DATA_FROM_CQ(cq);
if (cqd->num_pluckers == GRPC_MAX_COMPLETION_QUEUE_PLUCKERS) { if (cqd->num_pluckers == GRPC_MAX_COMPLETION_QUEUE_PLUCKERS) {
return 0; return 0;
} }
@ -897,7 +931,7 @@ static int add_plucker(grpc_completion_queue *cq, void *tag,
static void del_plucker(grpc_completion_queue *cq, void *tag, static void del_plucker(grpc_completion_queue *cq, void *tag,
grpc_pollset_worker **worker) { grpc_pollset_worker **worker) {
cq_data *cqd = &cq->data; cq_pluck_data *cqd = DATA_FROM_CQ(cq);
for (int i = 0; i < cqd->num_pluckers; i++) { for (int i = 0; i < cqd->num_pluckers; i++) {
if (cqd->pluckers[i].tag == tag && cqd->pluckers[i].worker == worker) { if (cqd->pluckers[i].tag == tag && cqd->pluckers[i].worker == worker) {
cqd->num_pluckers--; cqd->num_pluckers--;
@ -911,13 +945,13 @@ static void del_plucker(grpc_completion_queue *cq, void *tag,
static bool cq_is_pluck_finished(grpc_exec_ctx *exec_ctx, void *arg) { static bool cq_is_pluck_finished(grpc_exec_ctx *exec_ctx, void *arg) {
cq_is_finished_arg *a = arg; cq_is_finished_arg *a = arg;
grpc_completion_queue *cq = a->cq; grpc_completion_queue *cq = a->cq;
cq_data *cqd = &cq->data; cq_pluck_data *cqd = DATA_FROM_CQ(cq);
GPR_ASSERT(a->stolen_completion == NULL); GPR_ASSERT(a->stolen_completion == NULL);
gpr_atm current_last_seen_things_queued_ever = gpr_atm current_last_seen_things_queued_ever =
gpr_atm_no_barrier_load(&cqd->things_queued_ever); gpr_atm_no_barrier_load(&cqd->things_queued_ever);
if (current_last_seen_things_queued_ever != a->last_seen_things_queued_ever) { if (current_last_seen_things_queued_ever != a->last_seen_things_queued_ever) {
gpr_mu_lock(cqd->mu); gpr_mu_lock(cq->mu);
a->last_seen_things_queued_ever = a->last_seen_things_queued_ever =
gpr_atm_no_barrier_load(&cqd->things_queued_ever); gpr_atm_no_barrier_load(&cqd->things_queued_ever);
grpc_cq_completion *c; grpc_cq_completion *c;
@ -929,13 +963,13 @@ static bool cq_is_pluck_finished(grpc_exec_ctx *exec_ctx, void *arg) {
if (c == cqd->completed_tail) { if (c == cqd->completed_tail) {
cqd->completed_tail = prev; cqd->completed_tail = prev;
} }
gpr_mu_unlock(cqd->mu); gpr_mu_unlock(cq->mu);
a->stolen_completion = c; a->stolen_completion = c;
return true; return true;
} }
prev = c; prev = c;
} }
gpr_mu_unlock(cqd->mu); gpr_mu_unlock(cq->mu);
} }
return !a->first_loop && return !a->first_loop &&
gpr_time_cmp(a->deadline, gpr_now(a->deadline.clock_type)) < 0; gpr_time_cmp(a->deadline, gpr_now(a->deadline.clock_type)) < 0;
@ -948,7 +982,7 @@ static grpc_event cq_pluck(grpc_completion_queue *cq, void *tag,
grpc_cq_completion *prev; grpc_cq_completion *prev;
grpc_pollset_worker *worker = NULL; grpc_pollset_worker *worker = NULL;
gpr_timespec now; gpr_timespec now;
cq_data *cqd = &cq->data; cq_pluck_data *cqd = DATA_FROM_CQ(cq);
GPR_TIMER_BEGIN("grpc_completion_queue_pluck", 0); GPR_TIMER_BEGIN("grpc_completion_queue_pluck", 0);
@ -969,7 +1003,7 @@ static grpc_event cq_pluck(grpc_completion_queue *cq, void *tag,
deadline = gpr_convert_clock_type(deadline, GPR_CLOCK_MONOTONIC); deadline = gpr_convert_clock_type(deadline, GPR_CLOCK_MONOTONIC);
GRPC_CQ_INTERNAL_REF(cq, "pluck"); GRPC_CQ_INTERNAL_REF(cq, "pluck");
gpr_mu_lock(cqd->mu); gpr_mu_lock(cq->mu);
cq_is_finished_arg is_finished_arg = { cq_is_finished_arg is_finished_arg = {
.last_seen_things_queued_ever = .last_seen_things_queued_ever =
gpr_atm_no_barrier_load(&cqd->things_queued_ever), gpr_atm_no_barrier_load(&cqd->things_queued_ever),
@ -982,7 +1016,7 @@ static grpc_event cq_pluck(grpc_completion_queue *cq, void *tag,
GRPC_EXEC_CTX_INITIALIZER(0, cq_is_pluck_finished, &is_finished_arg); GRPC_EXEC_CTX_INITIALIZER(0, cq_is_pluck_finished, &is_finished_arg);
for (;;) { for (;;) {
if (is_finished_arg.stolen_completion != NULL) { if (is_finished_arg.stolen_completion != NULL) {
gpr_mu_unlock(cqd->mu); gpr_mu_unlock(cq->mu);
c = is_finished_arg.stolen_completion; c = is_finished_arg.stolen_completion;
is_finished_arg.stolen_completion = NULL; is_finished_arg.stolen_completion = NULL;
ret.type = GRPC_OP_COMPLETE; ret.type = GRPC_OP_COMPLETE;
@ -999,7 +1033,7 @@ static grpc_event cq_pluck(grpc_completion_queue *cq, void *tag,
if (c == cqd->completed_tail) { if (c == cqd->completed_tail) {
cqd->completed_tail = prev; cqd->completed_tail = prev;
} }
gpr_mu_unlock(cqd->mu); gpr_mu_unlock(cq->mu);
ret.type = GRPC_OP_COMPLETE; ret.type = GRPC_OP_COMPLETE;
ret.success = c->next & 1u; ret.success = c->next & 1u;
ret.tag = c->tag; ret.tag = c->tag;
@ -1009,7 +1043,7 @@ static grpc_event cq_pluck(grpc_completion_queue *cq, void *tag,
prev = c; prev = c;
} }
if (gpr_atm_no_barrier_load(&cqd->shutdown)) { if (gpr_atm_no_barrier_load(&cqd->shutdown)) {
gpr_mu_unlock(cqd->mu); gpr_mu_unlock(cq->mu);
memset(&ret, 0, sizeof(ret)); memset(&ret, 0, sizeof(ret));
ret.type = GRPC_QUEUE_SHUTDOWN; ret.type = GRPC_QUEUE_SHUTDOWN;
break; break;
@ -1019,7 +1053,7 @@ static grpc_event cq_pluck(grpc_completion_queue *cq, void *tag,
"Too many outstanding grpc_completion_queue_pluck calls: maximum " "Too many outstanding grpc_completion_queue_pluck calls: maximum "
"is %d", "is %d",
GRPC_MAX_COMPLETION_QUEUE_PLUCKERS); GRPC_MAX_COMPLETION_QUEUE_PLUCKERS);
gpr_mu_unlock(cqd->mu); gpr_mu_unlock(cq->mu);
memset(&ret, 0, sizeof(ret)); memset(&ret, 0, sizeof(ret));
/* TODO(ctiller): should we use a different result here */ /* TODO(ctiller): should we use a different result here */
ret.type = GRPC_QUEUE_TIMEOUT; ret.type = GRPC_QUEUE_TIMEOUT;
@ -1029,19 +1063,19 @@ static grpc_event cq_pluck(grpc_completion_queue *cq, void *tag,
now = gpr_now(GPR_CLOCK_MONOTONIC); now = gpr_now(GPR_CLOCK_MONOTONIC);
if (!is_finished_arg.first_loop && gpr_time_cmp(now, deadline) >= 0) { if (!is_finished_arg.first_loop && gpr_time_cmp(now, deadline) >= 0) {
del_plucker(cq, tag, &worker); del_plucker(cq, tag, &worker);
gpr_mu_unlock(cqd->mu); gpr_mu_unlock(cq->mu);
memset(&ret, 0, sizeof(ret)); memset(&ret, 0, sizeof(ret));
ret.type = GRPC_QUEUE_TIMEOUT; ret.type = GRPC_QUEUE_TIMEOUT;
dump_pending_tags(cq); dump_pending_tags(cq);
break; break;
} }
cqd->num_polls++; cq->num_polls++;
grpc_error *err = cq->poller_vtable->work(&exec_ctx, POLLSET_FROM_CQ(cq), grpc_error *err = cq->poller_vtable->work(&exec_ctx, POLLSET_FROM_CQ(cq),
&worker, now, deadline); &worker, now, deadline);
if (err != GRPC_ERROR_NONE) { if (err != GRPC_ERROR_NONE) {
del_plucker(cq, tag, &worker); del_plucker(cq, tag, &worker);
gpr_mu_unlock(cqd->mu); gpr_mu_unlock(cq->mu);
const char *msg = grpc_error_string(err); const char *msg = grpc_error_string(err);
gpr_log(GPR_ERROR, "Completion queue pluck failed: %s", msg); gpr_log(GPR_ERROR, "Completion queue pluck failed: %s", msg);
@ -1076,37 +1110,71 @@ grpc_event grpc_completion_queue_pluck(grpc_completion_queue *cq, void *tag,
- Must be called only once in completion queue's lifetime - Must be called only once in completion queue's lifetime
- grpc_completion_queue_shutdown() MUST have been called before calling - grpc_completion_queue_shutdown() MUST have been called before calling
this function */ this function */
static void cq_finish_shutdown(grpc_exec_ctx *exec_ctx, static void cq_finish_shutdown_next(grpc_exec_ctx *exec_ctx,
grpc_completion_queue *cq) { grpc_completion_queue *cq) {
cq_data *cqd = &cq->data; cq_next_data *cqd = DATA_FROM_CQ(cq);
GPR_ASSERT(cqd->shutdown_called); GPR_ASSERT(cqd->shutdown_called);
GPR_ASSERT(!gpr_atm_no_barrier_load(&cqd->shutdown)); GPR_ASSERT(!gpr_atm_no_barrier_load(&cqd->shutdown));
gpr_atm_no_barrier_store(&cqd->shutdown, 1); gpr_atm_no_barrier_store(&cqd->shutdown, 1);
cq->poller_vtable->shutdown(exec_ctx, POLLSET_FROM_CQ(cq), cq->poller_vtable->shutdown(exec_ctx, POLLSET_FROM_CQ(cq),
&cqd->pollset_shutdown_done); &cq->pollset_shutdown_done);
} }
/* Shutdown simply drops a ref that we reserved at creation time; if we drop static void cq_finish_shutdown_pluck(grpc_exec_ctx *exec_ctx,
to zero here, then enter shutdown mode and wake up any waiters */ grpc_completion_queue *cq) {
void grpc_completion_queue_shutdown(grpc_completion_queue *cq) { cq_pluck_data *cqd = DATA_FROM_CQ(cq);
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
GPR_TIMER_BEGIN("grpc_completion_queue_shutdown", 0); GPR_ASSERT(cqd->shutdown_called);
GRPC_API_TRACE("grpc_completion_queue_shutdown(cq=%p)", 1, (cq)); GPR_ASSERT(!gpr_atm_no_barrier_load(&cqd->shutdown));
cq_data *cqd = &cq->data; gpr_atm_no_barrier_store(&cqd->shutdown, 1);
cq->poller_vtable->shutdown(exec_ctx, POLLSET_FROM_CQ(cq),
&cq->pollset_shutdown_done);
}
static void cq_shutdown_next(grpc_exec_ctx *exec_ctx,
grpc_completion_queue *cq) {
cq_next_data *cqd = DATA_FROM_CQ(cq);
gpr_mu_lock(cqd->mu); gpr_mu_lock(cq->mu);
if (cqd->shutdown_called) { if (cqd->shutdown_called) {
gpr_mu_unlock(cqd->mu); gpr_mu_unlock(cq->mu);
GPR_TIMER_END("grpc_completion_queue_shutdown", 0); GPR_TIMER_END("grpc_completion_queue_shutdown", 0);
return; return;
} }
cqd->shutdown_called = 1; cqd->shutdown_called = 1;
if (gpr_unref(&cqd->pending_events)) { if (gpr_unref(&cqd->pending_events)) {
cq_finish_shutdown(&exec_ctx, cq); cq_finish_shutdown_next(exec_ctx, cq);
} }
gpr_mu_unlock(cqd->mu); gpr_mu_unlock(cq->mu);
}
static void cq_shutdown_pluck(grpc_exec_ctx *exec_ctx,
grpc_completion_queue *cq) {
cq_pluck_data *cqd = DATA_FROM_CQ(cq);
gpr_mu_lock(cq->mu);
if (cqd->shutdown_called) {
gpr_mu_unlock(cq->mu);
GPR_TIMER_END("grpc_completion_queue_shutdown", 0);
return;
}
cqd->shutdown_called = 1;
if (gpr_unref(&cqd->pending_events)) {
cq_finish_shutdown_pluck(exec_ctx, cq);
}
gpr_mu_unlock(cq->mu);
}
/* Shutdown simply drops a ref that we reserved at creation time; if we drop
to zero here, then enter shutdown mode and wake up any waiters */
void grpc_completion_queue_shutdown(grpc_completion_queue *cq) {
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
GPR_TIMER_BEGIN("grpc_completion_queue_shutdown", 0);
GRPC_API_TRACE("grpc_completion_queue_shutdown(cq=%p)", 1, (cq));
cq->vtable->shutdown(&exec_ctx, cq);
grpc_exec_ctx_finish(&exec_ctx); grpc_exec_ctx_finish(&exec_ctx);
GPR_TIMER_END("grpc_completion_queue_shutdown", 0); GPR_TIMER_END("grpc_completion_queue_shutdown", 0);
} }
@ -1116,12 +1184,6 @@ void grpc_completion_queue_destroy(grpc_completion_queue *cq) {
GPR_TIMER_BEGIN("grpc_completion_queue_destroy", 0); GPR_TIMER_BEGIN("grpc_completion_queue_destroy", 0);
grpc_completion_queue_shutdown(cq); grpc_completion_queue_shutdown(cq);
/* TODO (sreek): This should not ideally be here. Refactor it into the
* cq_vtable (perhaps have a create/destroy methods in the cq vtable) */
if (cq->vtable->cq_completion_type == GRPC_CQ_NEXT) {
GPR_ASSERT(cq_event_queue_num_items(&cq->data.queue) == 0);
}
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
GRPC_CQ_INTERNAL_UNREF(&exec_ctx, cq, "destroy"); GRPC_CQ_INTERNAL_UNREF(&exec_ctx, cq, "destroy");
grpc_exec_ctx_finish(&exec_ctx); grpc_exec_ctx_finish(&exec_ctx);
@ -1132,18 +1194,6 @@ grpc_pollset *grpc_cq_pollset(grpc_completion_queue *cq) {
return cq->poller_vtable->can_get_pollset ? POLLSET_FROM_CQ(cq) : NULL; return cq->poller_vtable->can_get_pollset ? POLLSET_FROM_CQ(cq) : NULL;
} }
grpc_completion_queue *grpc_cq_from_pollset(grpc_pollset *ps) {
return CQ_FROM_POLLSET(ps);
}
void grpc_cq_mark_server_cq(grpc_completion_queue *cq) {
cq->data.is_server_cq = 1;
}
bool grpc_cq_is_server_cq(grpc_completion_queue *cq) {
return cq->data.is_server_cq;
}
bool grpc_cq_can_listen(grpc_completion_queue *cq) { bool grpc_cq_can_listen(grpc_completion_queue *cq) {
return cq->poller_vtable->can_listen; return cq->poller_vtable->can_listen;
} }

3
src/core/lib/surface/completion_queue.h
Unescape View File

@ -99,10 +99,7 @@ void grpc_cq_end_op(grpc_exec_ctx *exec_ctx, grpc_completion_queue *cc,
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);
grpc_completion_queue *grpc_cq_from_pollset(grpc_pollset *ps);
void grpc_cq_mark_server_cq(grpc_completion_queue *cc);
bool grpc_cq_is_server_cq(grpc_completion_queue *cc);
bool grpc_cq_can_listen(grpc_completion_queue *cc); bool grpc_cq_can_listen(grpc_completion_queue *cc);
grpc_cq_completion_type grpc_get_cq_completion_type(grpc_completion_queue *cc); grpc_cq_completion_type grpc_get_cq_completion_type(grpc_completion_queue *cc);

5
src/core/lib/surface/server.c
Unescape View File

@ -964,8 +964,6 @@ static void register_completion_queue(grpc_server *server,
if (server->cqs[i] == cq) return; if (server->cqs[i] == cq) return;
} }
grpc_cq_mark_server_cq(cq);
GRPC_CQ_INTERNAL_REF(cq, "server"); GRPC_CQ_INTERNAL_REF(cq, "server");
n = server->cq_count++; n = server->cq_count++;
server->cqs = gpr_realloc(server->cqs, server->cqs = gpr_realloc(server->cqs,
@ -1129,9 +1127,8 @@ void grpc_server_setup_transport(grpc_exec_ctx *exec_ctx, grpc_server *s,
chand->channel = channel; chand->channel = channel;
size_t cq_idx; size_t cq_idx;
grpc_completion_queue *accepting_cq = grpc_cq_from_pollset(accepting_pollset);
for (cq_idx = 0; cq_idx < s->cq_count; cq_idx++) { for (cq_idx = 0; cq_idx < s->cq_count; cq_idx++) {
if (s->cqs[cq_idx] == accepting_cq) break; if (grpc_cq_pollset(s->cqs[cq_idx]) == accepting_pollset) break;
} }
if (cq_idx == s->cq_count) { if (cq_idx == s->cq_count) {
/* completion queue not found: pick a random one to publish new calls to */ /* completion queue not found: pick a random one to publish new calls to */

Write Preview
Loading…
Cancel
Save