diff options
author | 2014-12-09 14:39:16 -0800 | |
---|---|---|
committer | 2014-12-09 16:20:55 -0800 | |
commit | 18b49ab914ea5a57f22ed6d77520cd7d4372749b (patch) | |
tree | c2ec5971eebd10e3ef52c0c084c797b8d06bb267 /src/core/surface/completion_queue.c | |
parent | 98bffb779b8c47f4d76c72c7807d9f1b1074a795 (diff) |
Introducing iomgr.
Move eventmanager and platform dependent endpoint functionality into a single
library called 'iomgr'.
This is primarily to prepare for a Windows port - where posix socket semantics
lead to poor quality code.
Mostly this is a code movement CL, with some small changes to help prepare the
way for porting:
- em style fd objects can only be held internally in iomgr, and own their memory
- added grpc_iomgr_create_endpoint_pair() to accomodate the common pattern of
creating a tcp endpoint from the output of socketpair - this will help keep
our tests portable
- separated em alarm interface into a separate file, as this part of event
manager is needed higher up the stack
- made the eventmanager bits a true singleton, simplifying API's across the
stack as there's no longer a reason to carry a pointer there.
Initial design document is here:
https://docs.google.com/document/d/1VmafcHvvrP5kwtQkz84R5yXF7u7fW-9Pn0bkSUQHDt8/edit?disco=AAAAARNByxg
Change on 2014/12/09 by ctiller <ctiller@google.com>
-------------
Created by MOE: http://code.google.com/p/moe-java
MOE_MIGRATED_REVID=81716456
Diffstat (limited to 'src/core/surface/completion_queue.c')
-rw-r--r-- | src/core/surface/completion_queue.c | 65 |
1 files changed, 31 insertions, 34 deletions
diff --git a/src/core/surface/completion_queue.c b/src/core/surface/completion_queue.c index 2002476777..1f3074fd42 100644 --- a/src/core/surface/completion_queue.c +++ b/src/core/surface/completion_queue.c @@ -36,10 +36,9 @@ #include <stdio.h> #include <string.h> -#include "src/core/eventmanager/em.h" +#include "src/core/iomgr/iomgr_completion_queue_interface.h" #include "src/core/surface/call.h" #include "src/core/surface/event_string.h" -#include "src/core/surface/surface_em.h" #include "src/core/surface/surface_trace.h" #include <grpc/support/alloc.h> #include <grpc/support/atm.h> @@ -62,7 +61,6 @@ typedef struct event { /* Completion queue structure */ struct grpc_completion_queue { - grpc_em *em; int allow_polling; /* When refs drops to zero, we are in shutdown mode, and will be destroyable @@ -89,7 +87,6 @@ grpc_completion_queue *grpc_completion_queue_create() { memset(cc, 0, sizeof(*cc)); /* Initial ref is dropped by grpc_completion_queue_shutdown */ gpr_ref_init(&cc->refs, 1); - cc->em = grpc_surface_em(); cc->allow_polling = 1; return cc; } @@ -100,7 +97,7 @@ void grpc_completion_queue_dont_poll_test_only(grpc_completion_queue *cc) { /* Create and append an event to the queue. Returns the event so that its data members can be filled in. - Requires cc->em->mu locked. */ + Requires grpc_iomgr_mu locked. */ static event *add_locked(grpc_completion_queue *cc, grpc_completion_type type, void *tag, grpc_call *call, grpc_event_finish_func on_finish, void *user_data) { @@ -126,7 +123,7 @@ static event *add_locked(grpc_completion_queue *cc, grpc_completion_type type, ev->bucket_prev = cc->buckets[bucket]->bucket_prev; ev->bucket_next->bucket_prev = ev->bucket_prev->bucket_next = ev; } - gpr_cv_broadcast(&cc->em->cv); + gpr_cv_broadcast(&grpc_iomgr_cv); return ev; } @@ -149,7 +146,7 @@ static void end_op_locked(grpc_completion_queue *cc, if (gpr_unref(&cc->refs)) { GPR_ASSERT(!cc->shutdown); cc->shutdown = 1; - gpr_cv_broadcast(&cc->em->cv); + gpr_cv_broadcast(&grpc_iomgr_cv); } } @@ -157,11 +154,11 @@ void grpc_cq_end_read(grpc_completion_queue *cc, void *tag, grpc_call *call, grpc_event_finish_func on_finish, void *user_data, grpc_byte_buffer *read) { event *ev; - gpr_mu_lock(&cc->em->mu); + gpr_mu_lock(&grpc_iomgr_mu); ev = add_locked(cc, GRPC_READ, tag, call, on_finish, user_data); ev->base.data.read = read; end_op_locked(cc, GRPC_READ); - gpr_mu_unlock(&cc->em->mu); + gpr_mu_unlock(&grpc_iomgr_mu); } void grpc_cq_end_invoke_accepted(grpc_completion_queue *cc, void *tag, @@ -169,11 +166,11 @@ void grpc_cq_end_invoke_accepted(grpc_completion_queue *cc, void *tag, grpc_event_finish_func on_finish, void *user_data, grpc_op_error error) { event *ev; - gpr_mu_lock(&cc->em->mu); + gpr_mu_lock(&grpc_iomgr_mu); ev = add_locked(cc, GRPC_INVOKE_ACCEPTED, tag, call, on_finish, user_data); ev->base.data.invoke_accepted = error; end_op_locked(cc, GRPC_INVOKE_ACCEPTED); - gpr_mu_unlock(&cc->em->mu); + gpr_mu_unlock(&grpc_iomgr_mu); } void grpc_cq_end_write_accepted(grpc_completion_queue *cc, void *tag, @@ -181,11 +178,11 @@ void grpc_cq_end_write_accepted(grpc_completion_queue *cc, void *tag, grpc_event_finish_func on_finish, void *user_data, grpc_op_error error) { event *ev; - gpr_mu_lock(&cc->em->mu); + gpr_mu_lock(&grpc_iomgr_mu); ev = add_locked(cc, GRPC_WRITE_ACCEPTED, tag, call, on_finish, user_data); ev->base.data.write_accepted = error; end_op_locked(cc, GRPC_WRITE_ACCEPTED); - gpr_mu_unlock(&cc->em->mu); + gpr_mu_unlock(&grpc_iomgr_mu); } void grpc_cq_end_finish_accepted(grpc_completion_queue *cc, void *tag, @@ -193,11 +190,11 @@ void grpc_cq_end_finish_accepted(grpc_completion_queue *cc, void *tag, grpc_event_finish_func on_finish, void *user_data, grpc_op_error error) { event *ev; - gpr_mu_lock(&cc->em->mu); + gpr_mu_lock(&grpc_iomgr_mu); ev = add_locked(cc, GRPC_FINISH_ACCEPTED, tag, call, on_finish, user_data); ev->base.data.finish_accepted = error; end_op_locked(cc, GRPC_FINISH_ACCEPTED); - gpr_mu_unlock(&cc->em->mu); + gpr_mu_unlock(&grpc_iomgr_mu); } void grpc_cq_end_client_metadata_read(grpc_completion_queue *cc, void *tag, @@ -206,24 +203,24 @@ void grpc_cq_end_client_metadata_read(grpc_completion_queue *cc, void *tag, void *user_data, size_t count, grpc_metadata *elements) { event *ev; - gpr_mu_lock(&cc->em->mu); + gpr_mu_lock(&grpc_iomgr_mu); ev = add_locked(cc, GRPC_CLIENT_METADATA_READ, tag, call, on_finish, user_data); ev->base.data.client_metadata_read.count = count; ev->base.data.client_metadata_read.elements = elements; end_op_locked(cc, GRPC_CLIENT_METADATA_READ); - gpr_mu_unlock(&cc->em->mu); + gpr_mu_unlock(&grpc_iomgr_mu); } void grpc_cq_end_finished(grpc_completion_queue *cc, void *tag, grpc_call *call, grpc_event_finish_func on_finish, void *user_data, grpc_status status) { event *ev; - gpr_mu_lock(&cc->em->mu); + gpr_mu_lock(&grpc_iomgr_mu); ev = add_locked(cc, GRPC_FINISHED, tag, call, on_finish, user_data); ev->base.data.finished = status; end_op_locked(cc, GRPC_FINISHED); - gpr_mu_unlock(&cc->em->mu); + gpr_mu_unlock(&grpc_iomgr_mu); } void grpc_cq_end_new_rpc(grpc_completion_queue *cc, void *tag, grpc_call *call, @@ -232,7 +229,7 @@ void grpc_cq_end_new_rpc(grpc_completion_queue *cc, void *tag, grpc_call *call, gpr_timespec deadline, size_t metadata_count, grpc_metadata *metadata_elements) { event *ev; - gpr_mu_lock(&cc->em->mu); + gpr_mu_lock(&grpc_iomgr_mu); ev = add_locked(cc, GRPC_SERVER_RPC_NEW, tag, call, on_finish, user_data); ev->base.data.server_rpc_new.method = method; ev->base.data.server_rpc_new.host = host; @@ -240,7 +237,7 @@ void grpc_cq_end_new_rpc(grpc_completion_queue *cc, void *tag, grpc_call *call, ev->base.data.server_rpc_new.metadata_count = metadata_count; ev->base.data.server_rpc_new.metadata_elements = metadata_elements; end_op_locked(cc, GRPC_SERVER_RPC_NEW); - gpr_mu_unlock(&cc->em->mu); + gpr_mu_unlock(&grpc_iomgr_mu); } /* Create a GRPC_QUEUE_SHUTDOWN event without queuing it anywhere */ @@ -257,7 +254,7 @@ grpc_event *grpc_completion_queue_next(grpc_completion_queue *cc, gpr_timespec deadline) { event *ev = NULL; - gpr_mu_lock(&cc->em->mu); + gpr_mu_lock(&grpc_iomgr_mu); for (;;) { if (cc->queue != NULL) { gpr_uintptr bucket; @@ -283,15 +280,15 @@ grpc_event *grpc_completion_queue_next(grpc_completion_queue *cc, ev = create_shutdown_event(); break; } - if (cc->allow_polling && grpc_em_work(cc->em, deadline)) { + if (cc->allow_polling && grpc_iomgr_work(deadline)) { continue; } - if (gpr_cv_wait(&cc->em->cv, &cc->em->mu, deadline)) { - gpr_mu_unlock(&cc->em->mu); + if (gpr_cv_wait(&grpc_iomgr_cv, &grpc_iomgr_mu, deadline)) { + gpr_mu_unlock(&grpc_iomgr_mu); return NULL; } } - gpr_mu_unlock(&cc->em->mu); + gpr_mu_unlock(&grpc_iomgr_mu); GRPC_SURFACE_TRACE_RETURNED_EVENT(cc, &ev->base); return &ev->base; } @@ -329,7 +326,7 @@ grpc_event *grpc_completion_queue_pluck(grpc_completion_queue *cc, void *tag, gpr_timespec deadline) { event *ev = NULL; - gpr_mu_lock(&cc->em->mu); + gpr_mu_lock(&grpc_iomgr_mu); for (;;) { if ((ev = pluck_event(cc, tag))) { break; @@ -338,15 +335,15 @@ grpc_event *grpc_completion_queue_pluck(grpc_completion_queue *cc, void *tag, ev = create_shutdown_event(); break; } - if (cc->allow_polling && grpc_em_work(cc->em, deadline)) { + if (cc->allow_polling && grpc_iomgr_work(deadline)) { continue; } - if (gpr_cv_wait(&cc->em->cv, &cc->em->mu, deadline)) { - gpr_mu_unlock(&cc->em->mu); + if (gpr_cv_wait(&grpc_iomgr_cv, &grpc_iomgr_mu, deadline)) { + gpr_mu_unlock(&grpc_iomgr_mu); return NULL; } } - gpr_mu_unlock(&cc->em->mu); + gpr_mu_unlock(&grpc_iomgr_mu); GRPC_SURFACE_TRACE_RETURNED_EVENT(cc, &ev->base); return &ev->base; } @@ -355,11 +352,11 @@ grpc_event *grpc_completion_queue_pluck(grpc_completion_queue *cc, void *tag, to zero here, then enter shutdown mode and wake up any waiters */ void grpc_completion_queue_shutdown(grpc_completion_queue *cc) { if (gpr_unref(&cc->refs)) { - gpr_mu_lock(&cc->em->mu); + gpr_mu_lock(&grpc_iomgr_mu); GPR_ASSERT(!cc->shutdown); cc->shutdown = 1; - gpr_cv_broadcast(&cc->em->cv); - gpr_mu_unlock(&cc->em->mu); + gpr_cv_broadcast(&grpc_iomgr_cv); + gpr_mu_unlock(&grpc_iomgr_mu); } } |