diff options
Diffstat (limited to 'src/core/lib/support')
-rw-r--r-- | src/core/lib/support/cmdline.cc | 2 | ||||
-rw-r--r-- | src/core/lib/support/mpscq.cc | 37 | ||||
-rw-r--r-- | src/core/lib/support/mpscq.h | 30 |
3 files changed, 66 insertions, 3 deletions
diff --git a/src/core/lib/support/cmdline.cc b/src/core/lib/support/cmdline.cc index 49b34194c3..d2785d2f30 100644 --- a/src/core/lib/support/cmdline.cc +++ b/src/core/lib/support/cmdline.cc @@ -105,7 +105,7 @@ void gpr_cmdline_add_flag(gpr_cmdline* cl, const char* name, const char* help, } void gpr_cmdline_add_string(gpr_cmdline* cl, const char* name, const char* help, - char** value) { + const char** value) { add_arg(cl, name, help, ARGTYPE_STRING, value); } diff --git a/src/core/lib/support/mpscq.cc b/src/core/lib/support/mpscq.cc index db25f24264..b270777d5c 100644 --- a/src/core/lib/support/mpscq.cc +++ b/src/core/lib/support/mpscq.cc @@ -31,11 +31,12 @@ void gpr_mpscq_destroy(gpr_mpscq* q) { GPR_ASSERT(q->tail == &q->stub); } -void gpr_mpscq_push(gpr_mpscq* q, gpr_mpscq_node* n) { +bool gpr_mpscq_push(gpr_mpscq* q, gpr_mpscq_node* n) { gpr_atm_no_barrier_store(&n->next, (gpr_atm)NULL); gpr_mpscq_node* prev = (gpr_mpscq_node*)gpr_atm_full_xchg(&q->head, (gpr_atm)n); gpr_atm_rel_store(&prev->next, (gpr_atm)n); + return prev == &q->stub; } gpr_mpscq_node* gpr_mpscq_pop(gpr_mpscq* q) { @@ -77,3 +78,37 @@ gpr_mpscq_node* gpr_mpscq_pop_and_check_end(gpr_mpscq* q, bool* empty) { *empty = false; return NULL; } + +void gpr_locked_mpscq_init(gpr_locked_mpscq* q) { + gpr_mpscq_init(&q->queue); + gpr_mu_init(&q->mu); +} + +void gpr_locked_mpscq_destroy(gpr_locked_mpscq* q) { + gpr_mpscq_destroy(&q->queue); + gpr_mu_destroy(&q->mu); +} + +bool gpr_locked_mpscq_push(gpr_locked_mpscq* q, gpr_mpscq_node* n) { + return gpr_mpscq_push(&q->queue, n); +} + +gpr_mpscq_node* gpr_locked_mpscq_try_pop(gpr_locked_mpscq* q) { + if (gpr_mu_trylock(&q->mu)) { + gpr_mpscq_node* n = gpr_mpscq_pop(&q->queue); + gpr_mu_unlock(&q->mu); + return n; + } + return NULL; +} + +gpr_mpscq_node* gpr_locked_mpscq_pop(gpr_locked_mpscq* q) { + gpr_mu_lock(&q->mu); + bool empty = false; + gpr_mpscq_node* n; + do { + n = gpr_mpscq_pop_and_check_end(&q->queue, &empty); + } while (n == NULL && !empty); + gpr_mu_unlock(&q->mu); + return n; +} diff --git a/src/core/lib/support/mpscq.h b/src/core/lib/support/mpscq.h index 1cc9d89feb..fb22742050 100644 --- a/src/core/lib/support/mpscq.h +++ b/src/core/lib/support/mpscq.h @@ -20,6 +20,7 @@ #define GRPC_CORE_LIB_SUPPORT_MPSCQ_H #include <grpc/support/atm.h> +#include <grpc/support/sync.h> #include <stdbool.h> #include <stddef.h> @@ -49,13 +50,40 @@ typedef struct gpr_mpscq { void gpr_mpscq_init(gpr_mpscq* q); void gpr_mpscq_destroy(gpr_mpscq* q); // Push a node -void gpr_mpscq_push(gpr_mpscq* q, gpr_mpscq_node* n); +// Thread safe - can be called from multiple threads concurrently +// Returns true if this was possibly the first node (may return true +// sporadically, will not return false sporadically) +bool gpr_mpscq_push(gpr_mpscq* q, gpr_mpscq_node* n); // Pop a node (returns NULL if no node is ready - which doesn't indicate that // the queue is empty!!) +// Thread compatible - can only be called from one thread at a time gpr_mpscq_node* gpr_mpscq_pop(gpr_mpscq* q); // Pop a node; sets *empty to true if the queue is empty, or false if it is not gpr_mpscq_node* gpr_mpscq_pop_and_check_end(gpr_mpscq* q, bool* empty); +// An mpscq with a lock: it's safe to pop from multiple threads, but doing +// only one thread will succeed concurrently +typedef struct gpr_locked_mpscq { + gpr_mpscq queue; + gpr_mu mu; +} gpr_locked_mpscq; + +void gpr_locked_mpscq_init(gpr_locked_mpscq* q); +void gpr_locked_mpscq_destroy(gpr_locked_mpscq* q); +// Push a node +// Thread safe - can be called from multiple threads concurrently +// Returns true if this was possibly the first node (may return true +// sporadically, will not return false sporadically) +bool gpr_locked_mpscq_push(gpr_locked_mpscq* q, gpr_mpscq_node* n); + +// Pop a node (returns NULL if no node is ready - which doesn't indicate that +// the queue is empty!!) +// Thread safe - can be called from multiple threads concurrently +gpr_mpscq_node* gpr_locked_mpscq_try_pop(gpr_locked_mpscq* q); + +// Pop a node. Returns NULL only if the queue was empty at some point after +// calling this function +gpr_mpscq_node* gpr_locked_mpscq_pop(gpr_locked_mpscq* q); #ifdef __cplusplus } #endif |