Change pollset mutex ownership

pull/5333/head
Craig Tiller 9 years ago
parent 2167987565
commit 85371a2bb0
  1. 4
      src/core/iomgr/fd_posix.c
  2. 2
      src/core/iomgr/pollset.h
  3. 8
      src/core/iomgr/pollset_multipoller_with_epoll.c
  4. 4
      src/core/iomgr/pollset_multipoller_with_poll_posix.c
  5. 33
      src/core/iomgr/pollset_posix.c
  6. 2
      src/core/iomgr/pollset_posix.h
  7. 26
      src/core/security/google_default_credentials.c
  8. 47
      src/core/surface/completion_queue.c
  9. 12
      test/core/end2end/fixtures/h2_uchannel.c
  10. 24
      test/core/httpcli/httpcli_test.c
  11. 23
      test/core/httpcli/httpscli_test.c
  12. 6
      test/core/iomgr/endpoint_pair_test.c
  13. 56
      test/core/iomgr/fd_posix_test.c
  14. 44
      test/core/iomgr/tcp_client_posix_test.c
  15. 56
      test/core/iomgr/tcp_posix_test.c
  16. 16
      test/core/iomgr/tcp_server_posix_test.c
  17. 16
      test/core/iomgr/workqueue_test.c
  18. 10
      test/core/security/oauth2_utils.c
  19. 20
      test/core/security/print_google_default_creds_token.c
  20. 6
      test/core/security/secure_endpoint_test.c
  21. 16
      test/core/security/verify_jwt.c
  22. 25
      test/core/util/port_posix.c
  23. 6
      test/core/util/test_tcp_server.c
  24. 2
      test/core/util/test_tcp_server.h

@ -177,11 +177,11 @@ int grpc_fd_is_orphaned(grpc_fd *fd) {
}
static void pollset_kick_locked(grpc_fd_watcher *watcher) {
gpr_mu_lock(watcher->pollset->mu);
gpr_mu_lock(&watcher->pollset->mu);
GPR_ASSERT(watcher->worker);
grpc_pollset_kick_ext(watcher->pollset, watcher->worker,
GRPC_POLLSET_REEVALUATE_POLLING_ON_WAKEUP);
gpr_mu_unlock(watcher->pollset->mu);
gpr_mu_unlock(&watcher->pollset->mu);
}
static void maybe_wake_one_watcher_locked(grpc_fd *fd) {

@ -53,7 +53,7 @@ typedef struct grpc_pollset grpc_pollset;
typedef struct grpc_pollset_worker grpc_pollset_worker;
size_t grpc_pollset_size(void);
void grpc_pollset_init(grpc_pollset *pollset, gpr_mu *mu);
void grpc_pollset_init(grpc_pollset *pollset, gpr_mu **mu);
/* Begin shutting down the pollset, and call closure when done.
* GRPC_POLLSET_MU(pollset) must be held */
void grpc_pollset_shutdown(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,

@ -149,7 +149,7 @@ static void perform_delayed_add(grpc_exec_ctx *exec_ctx, void *arg,
finally_add_fd(exec_ctx, da->pollset, da->fd);
}
gpr_mu_lock(da->pollset->mu);
gpr_mu_lock(&da->pollset->mu);
da->pollset->in_flight_cbs--;
if (da->pollset->shutting_down) {
/* We don't care about this pollset anymore. */
@ -158,7 +158,7 @@ static void perform_delayed_add(grpc_exec_ctx *exec_ctx, void *arg,
grpc_exec_ctx_enqueue(exec_ctx, da->pollset->shutdown_done, true, NULL);
}
}
gpr_mu_unlock(da->pollset->mu);
gpr_mu_unlock(&da->pollset->mu);
GRPC_FD_UNREF(da->fd, "delayed_add");
@ -170,7 +170,7 @@ static void multipoll_with_epoll_pollset_add_fd(grpc_exec_ctx *exec_ctx,
grpc_fd *fd,
int and_unlock_pollset) {
if (and_unlock_pollset) {
gpr_mu_unlock(pollset->mu);
gpr_mu_unlock(&pollset->mu);
finally_add_fd(exec_ctx, pollset, fd);
} else {
delayed_add *da = gpr_malloc(sizeof(*da));
@ -202,7 +202,7 @@ static void multipoll_with_epoll_pollset_maybe_work_and_unlock(
* here.
*/
gpr_mu_unlock(pollset->mu);
gpr_mu_unlock(&pollset->mu);
timeout_ms = grpc_poll_deadline_to_millis_timeout(deadline, now);

@ -80,7 +80,7 @@ static void multipoll_with_poll_pollset_add_fd(grpc_exec_ctx *exec_ctx,
GRPC_FD_REF(fd, "multipoller");
exit:
if (and_unlock_pollset) {
gpr_mu_unlock(pollset->mu);
gpr_mu_unlock(&pollset->mu);
}
}
@ -132,7 +132,7 @@ static void multipoll_with_poll_pollset_maybe_work_and_unlock(
}
h->del_count = 0;
h->fd_count = fd_count;
gpr_mu_unlock(pollset->mu);
gpr_mu_unlock(&pollset->mu);
for (i = 2; i < pfd_count; i++) {
pfds[i].events = (short)grpc_fd_begin_poll(watchers[i].fd, pollset, worker,

@ -188,8 +188,9 @@ void grpc_kick_poller(void) { grpc_wakeup_fd_wakeup(&grpc_global_wakeup_fd); }
static void become_basic_pollset(grpc_pollset *pollset, grpc_fd *fd_or_null);
void grpc_pollset_init(grpc_pollset *pollset, gpr_mu *mu) {
pollset->mu = mu;
void grpc_pollset_init(grpc_pollset *pollset, gpr_mu **mu) {
gpr_mu_init(&pollset->mu);
*mu = &pollset->mu;
pollset->root_worker.next = pollset->root_worker.prev = &pollset->root_worker;
pollset->in_flight_cbs = 0;
pollset->shutting_down = 0;
@ -228,15 +229,15 @@ void grpc_pollset_reset(grpc_pollset *pollset) {
void grpc_pollset_add_fd(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
grpc_fd *fd) {
gpr_mu_lock(pollset->mu);
gpr_mu_lock(&pollset->mu);
pollset->vtable->add_fd(exec_ctx, pollset, fd, 1);
/* the following (enabled only in debug) will reacquire and then release
our lock - meaning that if the unlocking flag passed to add_fd above is
not respected, the code will deadlock (in a way that we have a chance of
debugging) */
#ifndef NDEBUG
gpr_mu_lock(pollset->mu);
gpr_mu_unlock(pollset->mu);
gpr_mu_lock(&pollset->mu);
gpr_mu_unlock(&pollset->mu);
#endif
}
@ -285,7 +286,7 @@ void grpc_pollset_work(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
/* Give do_promote priority so we don't starve it out */
if (pollset->in_flight_cbs) {
GPR_TIMER_MARK("grpc_pollset_work.in_flight_cbs", 0);
gpr_mu_unlock(pollset->mu);
gpr_mu_unlock(&pollset->mu);
locked = 0;
goto done;
}
@ -319,7 +320,7 @@ void grpc_pollset_work(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
done:
if (!locked) {
queued_work |= grpc_exec_ctx_flush(exec_ctx);
gpr_mu_lock(pollset->mu);
gpr_mu_lock(&pollset->mu);
locked = 1;
}
/* If we're forced to re-evaluate polling (via grpc_pollset_kick with
@ -349,19 +350,19 @@ void grpc_pollset_work(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
grpc_pollset_kick(pollset, NULL);
} else if (!pollset->called_shutdown && pollset->in_flight_cbs == 0) {
pollset->called_shutdown = 1;
gpr_mu_unlock(pollset->mu);
gpr_mu_unlock(&pollset->mu);
finish_shutdown(exec_ctx, pollset);
grpc_exec_ctx_flush(exec_ctx);
/* Continuing to access pollset here is safe -- it is the caller's
* responsibility to not destroy when it has outstanding calls to
* grpc_pollset_work.
* TODO(dklempner): Can we refactor the shutdown logic to avoid this? */
gpr_mu_lock(pollset->mu);
gpr_mu_lock(&pollset->mu);
} else if (!grpc_closure_list_empty(pollset->idle_jobs)) {
grpc_exec_ctx_enqueue_list(exec_ctx, &pollset->idle_jobs, NULL);
gpr_mu_unlock(pollset->mu);
gpr_mu_unlock(&pollset->mu);
grpc_exec_ctx_flush(exec_ctx);
gpr_mu_lock(pollset->mu);
gpr_mu_lock(&pollset->mu);
}
}
*worker_hdl = NULL;
@ -429,7 +430,7 @@ static void basic_do_promote(grpc_exec_ctx *exec_ctx, void *args,
* 4. The pollset may be shutting down.
*/
gpr_mu_lock(pollset->mu);
gpr_mu_lock(&pollset->mu);
/* First we need to ensure that nobody is polling concurrently */
GPR_ASSERT(!grpc_pollset_has_workers(pollset));
@ -470,7 +471,7 @@ static void basic_do_promote(grpc_exec_ctx *exec_ctx, void *args,
}
}
gpr_mu_unlock(pollset->mu);
gpr_mu_unlock(&pollset->mu);
/* Matching ref in basic_pollset_add_fd */
GRPC_FD_UNREF(fd, "basicpoll_add");
@ -523,7 +524,7 @@ static void basic_pollset_add_fd(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
exit:
if (and_unlock_pollset) {
gpr_mu_unlock(pollset->mu);
gpr_mu_unlock(&pollset->mu);
}
}
@ -559,14 +560,14 @@ static void basic_pollset_maybe_work_and_unlock(grpc_exec_ctx *exec_ctx,
pfd[2].fd = fd->fd;
pfd[2].revents = 0;
GRPC_FD_REF(fd, "basicpoll_begin");
gpr_mu_unlock(pollset->mu);
gpr_mu_unlock(&pollset->mu);
pfd[2].events = (short)grpc_fd_begin_poll(fd, pollset, worker, POLLIN,
POLLOUT, &fd_watcher);
if (pfd[2].events != 0) {
nfds++;
}
} else {
gpr_mu_unlock(pollset->mu);
gpr_mu_unlock(&pollset->mu);
}
/* TODO(vpai): Consider first doing a 0 timeout poll here to avoid

@ -69,7 +69,7 @@ struct grpc_pollset {
For example, we may choose a poll() based implementation on linux for
few fds, and an epoll() based implementation for many fds */
const grpc_pollset_vtable *vtable;
gpr_mu *mu;
gpr_mu mu;
grpc_pollset_worker root_worker;
int in_flight_cbs;
int shutting_down;

@ -52,10 +52,11 @@
static grpc_channel_credentials *default_credentials = NULL;
static int compute_engine_detection_done = 0;
static gpr_mu g_mu;
static gpr_mu g_state_mu;
static gpr_mu *g_polling_mu;
static gpr_once g_once = GPR_ONCE_INIT;
static void init_default_credentials(void) { gpr_mu_init(&g_mu); }
static void init_default_credentials(void) { gpr_mu_init(&g_state_mu); }
typedef struct {
grpc_pollset *pollset;
@ -80,10 +81,10 @@ static void on_compute_engine_detection_http_response(
}
}
}
gpr_mu_lock(&g_mu);
gpr_mu_lock(g_polling_mu);
detector->is_done = 1;
grpc_pollset_kick(detector->pollset, NULL);
gpr_mu_unlock(&g_mu);
gpr_mu_unlock(g_polling_mu);
}
static void destroy_pollset(grpc_exec_ctx *exec_ctx, void *p, bool s) {
@ -102,7 +103,7 @@ static int is_stack_running_on_compute_engine(void) {
gpr_timespec max_detection_delay = gpr_time_from_seconds(1, GPR_TIMESPAN);
detector.pollset = gpr_malloc(grpc_pollset_size());
grpc_pollset_init(detector.pollset, &g_mu);
grpc_pollset_init(detector.pollset, &g_polling_mu);
detector.is_done = 0;
detector.success = 0;
@ -121,19 +122,20 @@ static int is_stack_running_on_compute_engine(void) {
/* Block until we get the response. This is not ideal but this should only be
called once for the lifetime of the process by the default credentials. */
gpr_mu_lock(&g_mu);
gpr_mu_lock(g_polling_mu);
while (!detector.is_done) {
grpc_pollset_worker *worker = NULL;
grpc_pollset_work(&exec_ctx, detector.pollset, &worker,
gpr_now(GPR_CLOCK_MONOTONIC),
gpr_inf_future(GPR_CLOCK_MONOTONIC));
}
gpr_mu_unlock(&g_mu);
gpr_mu_unlock(g_polling_mu);
grpc_httpcli_context_destroy(&context);
grpc_closure_init(&destroy_closure, destroy_pollset, &detector.pollset);
grpc_closure_init(&destroy_closure, destroy_pollset, detector.pollset);
grpc_pollset_shutdown(&exec_ctx, detector.pollset, &destroy_closure);
grpc_exec_ctx_finish(&exec_ctx);
g_polling_mu = NULL;
gpr_free(detector.pollset);
@ -187,7 +189,7 @@ grpc_channel_credentials *grpc_google_default_credentials_create(void) {
gpr_once_init(&g_once, init_default_credentials);
gpr_mu_lock(&g_mu);
gpr_mu_lock(&g_state_mu);
if (default_credentials != NULL) {
result = grpc_channel_credentials_ref(default_credentials);
@ -233,19 +235,19 @@ end:
gpr_log(GPR_ERROR, "Could not create google default credentials.");
}
}
gpr_mu_unlock(&g_mu);
gpr_mu_unlock(&g_state_mu);
return result;
}
void grpc_flush_cached_google_default_credentials(void) {
gpr_once_init(&g_once, init_default_credentials);
gpr_mu_lock(&g_mu);
gpr_mu_lock(&g_state_mu);
if (default_credentials != NULL) {
grpc_channel_credentials_unref(default_credentials);
default_credentials = NULL;
}
compute_engine_detection_done = 0;
gpr_mu_unlock(&g_mu);
gpr_mu_unlock(&g_state_mu);
}
/* -- Well known credentials path. -- */

@ -57,7 +57,8 @@ typedef struct {
/* Completion queue structure */
struct grpc_completion_queue {
gpr_mu mu;
/** owned by pollset */
gpr_mu *mu;
/** completed events */
grpc_cq_completion completed_head;
grpc_cq_completion *completed_tail;
@ -97,7 +98,6 @@ void grpc_cq_global_shutdown(void) {
while (g_freelist) {
grpc_completion_queue *next = g_freelist->next_free;
grpc_pollset_destroy(POLLSET_FROM_CQ(g_freelist));
gpr_mu_destroy(&g_freelist->mu);
#ifndef NDEBUG
gpr_free(g_freelist->outstanding_tags);
#endif
@ -128,7 +128,6 @@ grpc_completion_queue *grpc_completion_queue_create(void *reserved) {
gpr_mu_unlock(&g_freelist_mu);
cc = gpr_malloc(sizeof(grpc_completion_queue) + grpc_pollset_size());
gpr_mu_init(&cc->mu);
grpc_pollset_init(POLLSET_FROM_CQ(cc), &cc->mu);
#ifndef NDEBUG
cc->outstanding_tags = NULL;
@ -198,7 +197,7 @@ void grpc_cq_internal_unref(grpc_completion_queue *cc) {
void grpc_cq_begin_op(grpc_completion_queue *cc, void *tag) {
#ifndef NDEBUG
gpr_mu_lock(&cc->mu);
gpr_mu_lock(cc->mu);
GPR_ASSERT(!cc->shutdown_called);
if (cc->outstanding_tag_count == cc->outstanding_tag_capacity) {
cc->outstanding_tag_capacity = GPR_MAX(4, 2 * cc->outstanding_tag_capacity);
@ -207,7 +206,7 @@ void grpc_cq_begin_op(grpc_completion_queue *cc, void *tag) {
cc->outstanding_tag_capacity);
}
cc->outstanding_tags[cc->outstanding_tag_count++] = tag;
gpr_mu_unlock(&cc->mu);
gpr_mu_unlock(cc->mu);
#endif
gpr_ref(&cc->pending_events);
}
@ -235,7 +234,7 @@ void grpc_cq_end_op(grpc_exec_ctx *exec_ctx, grpc_completion_queue *cc,
storage->next =
((uintptr_t)&cc->completed_head) | ((uintptr_t)(success != 0));
gpr_mu_lock(&cc->mu);
gpr_mu_lock(cc->mu);
#ifndef NDEBUG
for (i = 0; i < (int)cc->outstanding_tag_count; i++) {
if (cc->outstanding_tags[i] == tag) {
@ -261,7 +260,7 @@ void grpc_cq_end_op(grpc_exec_ctx *exec_ctx, grpc_completion_queue *cc,
}
}
grpc_pollset_kick(POLLSET_FROM_CQ(cc), pluck_worker);
gpr_mu_unlock(&cc->mu);
gpr_mu_unlock(cc->mu);
} else {
cc->completed_tail->next =
((uintptr_t)storage) | (1u & (uintptr_t)cc->completed_tail->next);
@ -271,7 +270,7 @@ void grpc_cq_end_op(grpc_exec_ctx *exec_ctx, grpc_completion_queue *cc,
cc->shutdown = 1;
grpc_pollset_shutdown(exec_ctx, POLLSET_FROM_CQ(cc),
&cc->pollset_shutdown_done);
gpr_mu_unlock(&cc->mu);
gpr_mu_unlock(cc->mu);
}
GPR_TIMER_END("grpc_cq_end_op", 0);
@ -299,7 +298,7 @@ grpc_event grpc_completion_queue_next(grpc_completion_queue *cc,
deadline = gpr_convert_clock_type(deadline, GPR_CLOCK_MONOTONIC);
GRPC_CQ_INTERNAL_REF(cc, "next");
gpr_mu_lock(&cc->mu);
gpr_mu_lock(cc->mu);
for (;;) {
if (cc->completed_tail != &cc->completed_head) {
grpc_cq_completion *c = (grpc_cq_completion *)cc->completed_head.next;
@ -307,7 +306,7 @@ grpc_event grpc_completion_queue_next(grpc_completion_queue *cc,
if (c == cc->completed_tail) {
cc->completed_tail = &cc->completed_head;
}
gpr_mu_unlock(&cc->mu);
gpr_mu_unlock(cc->mu);
ret.type = GRPC_OP_COMPLETE;
ret.success = c->next & 1u;
ret.tag = c->tag;
@ -315,14 +314,14 @@ grpc_event grpc_completion_queue_next(grpc_completion_queue *cc,
break;
}
if (cc->shutdown) {
gpr_mu_unlock(&cc->mu);
gpr_mu_unlock(cc->mu);
memset(&ret, 0, sizeof(ret));
ret.type = GRPC_QUEUE_SHUTDOWN;
break;
}
now = gpr_now(GPR_CLOCK_MONOTONIC);
if (!first_loop && gpr_time_cmp(now, deadline) >= 0) {
gpr_mu_unlock(&cc->mu);
gpr_mu_unlock(cc->mu);
memset(&ret, 0, sizeof(ret));
ret.type = GRPC_QUEUE_TIMEOUT;
break;
@ -335,9 +334,9 @@ grpc_event grpc_completion_queue_next(grpc_completion_queue *cc,
gpr_timespec iteration_deadline = deadline;
if (grpc_timer_check(&exec_ctx, now, &iteration_deadline)) {
GPR_TIMER_MARK("alarm_triggered", 0);
gpr_mu_unlock(&cc->mu);
gpr_mu_unlock(cc->mu);
grpc_exec_ctx_flush(&exec_ctx);
gpr_mu_lock(&cc->mu);
gpr_mu_lock(cc->mu);
continue;
} else {
grpc_pollset_work(&exec_ctx, POLLSET_FROM_CQ(cc), &worker, now,
@ -401,7 +400,7 @@ grpc_event grpc_completion_queue_pluck(grpc_completion_queue *cc, void *tag,
deadline = gpr_convert_clock_type(deadline, GPR_CLOCK_MONOTONIC);
GRPC_CQ_INTERNAL_REF(cc, "pluck");
gpr_mu_lock(&cc->mu);
gpr_mu_lock(cc->mu);
for (;;) {
prev = &cc->completed_head;
while ((c = (grpc_cq_completion *)(prev->next & ~(uintptr_t)1)) !=
@ -411,7 +410,7 @@ grpc_event grpc_completion_queue_pluck(grpc_completion_queue *cc, void *tag,
if (c == cc->completed_tail) {
cc->completed_tail = prev;
}
gpr_mu_unlock(&cc->mu);
gpr_mu_unlock(cc->mu);
ret.type = GRPC_OP_COMPLETE;
ret.success = c->next & 1u;
ret.tag = c->tag;
@ -421,7 +420,7 @@ grpc_event grpc_completion_queue_pluck(grpc_completion_queue *cc, void *tag,
prev = c;
}
if (cc->shutdown) {
gpr_mu_unlock(&cc->mu);
gpr_mu_unlock(cc->mu);
memset(&ret, 0, sizeof(ret));
ret.type = GRPC_QUEUE_SHUTDOWN;
break;
@ -431,7 +430,7 @@ grpc_event grpc_completion_queue_pluck(grpc_completion_queue *cc, void *tag,
"Too many outstanding grpc_completion_queue_pluck calls: maximum "
"is %d",
GRPC_MAX_COMPLETION_QUEUE_PLUCKERS);
gpr_mu_unlock(&cc->mu);
gpr_mu_unlock(cc->mu);
memset(&ret, 0, sizeof(ret));
/* TODO(ctiller): should we use a different result here */
ret.type = GRPC_QUEUE_TIMEOUT;
@ -440,7 +439,7 @@ grpc_event grpc_completion_queue_pluck(grpc_completion_queue *cc, void *tag,
now = gpr_now(GPR_CLOCK_MONOTONIC);
if (!first_loop && gpr_time_cmp(now, deadline) >= 0) {
del_plucker(cc, tag, &worker);
gpr_mu_unlock(&cc->mu);
gpr_mu_unlock(cc->mu);
memset(&ret, 0, sizeof(ret));
ret.type = GRPC_QUEUE_TIMEOUT;
break;
@ -453,9 +452,9 @@ grpc_event grpc_completion_queue_pluck(grpc_completion_queue *cc, void *tag,
gpr_timespec iteration_deadline = deadline;
if (grpc_timer_check(&exec_ctx, now, &iteration_deadline)) {
GPR_TIMER_MARK("alarm_triggered", 0);
gpr_mu_unlock(&cc->mu);
gpr_mu_unlock(cc->mu);
grpc_exec_ctx_flush(&exec_ctx);
gpr_mu_lock(&cc->mu);
gpr_mu_lock(cc->mu);
continue;
} else {
grpc_pollset_work(&exec_ctx, POLLSET_FROM_CQ(cc), &worker, now,
@ -479,9 +478,9 @@ void grpc_completion_queue_shutdown(grpc_completion_queue *cc) {
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
GPR_TIMER_BEGIN("grpc_completion_queue_shutdown", 0);
GRPC_API_TRACE("grpc_completion_queue_shutdown(cc=%p)", 1, (cc));
gpr_mu_lock(&cc->mu);
gpr_mu_lock(cc->mu);
if (cc->shutdown_called) {
gpr_mu_unlock(&cc->mu);
gpr_mu_unlock(cc->mu);
GPR_TIMER_END("grpc_completion_queue_shutdown", 0);
return;
}
@ -492,7 +491,7 @@ void grpc_completion_queue_shutdown(grpc_completion_queue *cc) {
grpc_pollset_shutdown(&exec_ctx, POLLSET_FROM_CQ(cc),
&cc->pollset_shutdown_done);
}
gpr_mu_unlock(&cc->mu);
gpr_mu_unlock(cc->mu);
grpc_exec_ctx_finish(&exec_ctx);
GPR_TIMER_END("grpc_completion_queue_shutdown", 0);
}

@ -253,8 +253,7 @@ static void destroy_pollset(grpc_exec_ctx *exec_ctx, void *arg, bool success) {
}
static grpc_connected_subchannel *connect_subchannel(grpc_subchannel *c) {
gpr_mu mu;
gpr_mu_init(&mu);
gpr_mu *mu;
grpc_pollset *pollset = gpr_malloc(grpc_pollset_size());
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
grpc_pollset_init(pollset, &mu);
@ -264,22 +263,21 @@ static grpc_connected_subchannel *connect_subchannel(grpc_subchannel *c) {
&g_state,
grpc_closure_create(state_changed, c));
grpc_exec_ctx_flush(&exec_ctx);
gpr_mu_lock(&mu);
gpr_mu_lock(mu);
while (g_state != GRPC_CHANNEL_READY) {
grpc_pollset_worker *worker = NULL;
grpc_pollset_work(&exec_ctx, pollset, &worker, gpr_now(GPR_CLOCK_MONOTONIC),
GRPC_TIMEOUT_SECONDS_TO_DEADLINE(1));
gpr_mu_unlock(&mu);
gpr_mu_unlock(mu);
grpc_exec_ctx_flush(&exec_ctx);
gpr_mu_lock(&mu);
gpr_mu_lock(mu);
}
grpc_pollset_shutdown(&exec_ctx, pollset,
grpc_closure_create(destroy_pollset, pollset));
grpc_pollset_set_destroy(&g_interested_parties);
gpr_mu_unlock(&mu);
gpr_mu_unlock(mu);
grpc_exec_ctx_finish(&exec_ctx);
gpr_free(pollset);
gpr_mu_destroy(&mu);
return grpc_subchannel_get_connected_subchannel(c);
}

@ -47,7 +47,7 @@
static int g_done = 0;
static grpc_httpcli_context g_context;
static gpr_mu g_mu;
static gpr_mu *g_mu;
static grpc_pollset *g_pollset;
static gpr_timespec n_seconds_time(int seconds) {
@ -64,10 +64,10 @@ static void on_finish(grpc_exec_ctx *exec_ctx, void *arg,
GPR_ASSERT(response->status == 200);
GPR_ASSERT(response->body_length == strlen(expect));
GPR_ASSERT(0 == memcmp(expect, response->body, response->body_length));
gpr_mu_lock(&g_mu);
gpr_mu_lock(g_mu);
g_done = 1;
grpc_pollset_kick(g_pollset, NULL);
gpr_mu_unlock(&g_mu);
gpr_mu_unlock(g_mu);
}
static void test_get(int port) {
@ -88,16 +88,16 @@ static void test_get(int port) {
grpc_httpcli_get(&exec_ctx, &g_context, g_pollset, &req, n_seconds_time(15),
on_finish, (void *)42);
gpr_mu_lock(&g_mu);
gpr_mu_lock(g_mu);
while (!g_done) {
grpc_pollset_worker *worker = NULL;
grpc_pollset_work(&exec_ctx, g_pollset, &worker,
gpr_now(GPR_CLOCK_MONOTONIC), n_seconds_time(20));
gpr_mu_unlock(&g_mu);
gpr_mu_unlock(g_mu);
grpc_exec_ctx_finish(&exec_ctx);
gpr_mu_lock(&g_mu);
gpr_mu_lock(g_mu);
}
gpr_mu_unlock(&g_mu);
gpr_mu_unlock(g_mu);
gpr_free(host);
}
@ -119,16 +119,16 @@ static void test_post(int port) {
grpc_httpcli_post(&exec_ctx, &g_context, g_pollset, &req, "hello", 5,
n_seconds_time(15), on_finish, (void *)42);
gpr_mu_lock(&g_mu);
gpr_mu_lock(g_mu);
while (!g_done) {
grpc_pollset_worker *worker = NULL;
grpc_pollset_work(&exec_ctx, g_pollset, &worker,
gpr_now(GPR_CLOCK_MONOTONIC), n_seconds_time(20));
gpr_mu_unlock(&g_mu);
gpr_mu_unlock(g_mu);
grpc_exec_ctx_finish(&exec_ctx);
gpr_mu_lock(&g_mu);
gpr_mu_lock(g_mu);
}
gpr_mu_unlock(&g_mu);
gpr_mu_unlock(g_mu);
gpr_free(host);
}
@ -177,7 +177,6 @@ int main(int argc, char **argv) {
grpc_init();
grpc_httpcli_context_init(&g_context);
g_pollset = gpr_malloc(grpc_pollset_size());
gpr_mu_init(&g_mu);
grpc_pollset_init(g_pollset, &g_mu);
test_get(port);
@ -189,7 +188,6 @@ int main(int argc, char **argv) {
grpc_exec_ctx_finish(&exec_ctx);
grpc_shutdown();
gpr_mu_destroy(&g_mu);
gpr_free(g_pollset);
gpr_subprocess_destroy(server);

@ -47,7 +47,7 @@
static int g_done = 0;
static grpc_httpcli_context g_context;
static gpr_mu g_mu;
static gpr_mu *g_mu;
static grpc_pollset *g_pollset;
static gpr_timespec n_seconds_time(int seconds) {
@ -64,10 +64,10 @@ static void on_finish(grpc_exec_ctx *exec_ctx, void *arg,
GPR_ASSERT(response->status == 200);
GPR_ASSERT(response->body_length == strlen(expect));
GPR_ASSERT(0 == memcmp(expect, response->body, response->body_length));
gpr_mu_lock(&g_mu);
gpr_mu_lock(g_mu);
g_done = 1;
grpc_pollset_kick(g_pollset, NULL);
gpr_mu_unlock(&g_mu);
gpr_mu_unlock(g_mu);
}
static void test_get(int port) {
@ -89,16 +89,16 @@ static void test_get(int port) {
grpc_httpcli_get(&exec_ctx, &g_context, g_pollset, &req, n_seconds_time(15),
on_finish, (void *)42);
gpr_mu_lock(&g_mu);
gpr_mu_lock(g_mu);
while (!g_done) {
grpc_pollset_worker *worker = NULL;
grpc_pollset_work(&exec_ctx, g_pollset, &worker,
gpr_now(GPR_CLOCK_MONOTONIC), n_seconds_time(20));
gpr_mu_unlock(&g_mu);
gpr_mu_unlock(g_mu);
grpc_exec_ctx_finish(&exec_ctx);
gpr_mu_lock(&g_mu);
gpr_mu_lock(g_mu);
}
gpr_mu_unlock(&g_mu);
gpr_mu_unlock(g_mu);
gpr_free(host);
}
@ -121,16 +121,16 @@ static void test_post(int port) {
grpc_httpcli_post(&exec_ctx, &g_context, g_pollset, &req, "hello", 5,
n_seconds_time(15), on_finish, (void *)42);
gpr_mu_lock(&g_mu);
gpr_mu_lock(g_mu);
while (!g_done) {
grpc_pollset_worker *worker = NULL;
grpc_pollset_work(&exec_ctx, g_pollset, &worker,
gpr_now(GPR_CLOCK_MONOTONIC), n_seconds_time(20));
gpr_mu_unlock(&g_mu);
gpr_mu_unlock(g_mu);
grpc_exec_ctx_finish(&exec_ctx);
gpr_mu_lock(&g_mu);
gpr_mu_lock(g_mu);
}
gpr_mu_unlock(&g_mu);
gpr_mu_unlock(g_mu);
gpr_free(host);
}
@ -192,7 +192,6 @@ int main(int argc, char **argv) {
grpc_shutdown();
gpr_free(g_pollset);
gpr_mu_destroy(&g_mu);
gpr_subprocess_destroy(server);

@ -42,7 +42,7 @@
#include "test/core/iomgr/endpoint_tests.h"
#include "test/core/util/test_config.h"
static gpr_mu g_mu;
static gpr_mu *g_mu;
static grpc_pollset *g_pollset;
static void clean_up(void) {}
@ -73,17 +73,15 @@ static void destroy_pollset(grpc_exec_ctx *exec_ctx, void *p, bool success) {
int main(int argc, char **argv) {
grpc_closure destroyed;
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
gpr_mu_init(&g_mu);
grpc_test_init(argc, argv);
grpc_init();
g_pollset = gpr_malloc(grpc_pollset_size());
grpc_pollset_init(g_pollset, &g_mu);
grpc_endpoint_tests(configs[0], g_pollset, &g_mu);
grpc_endpoint_tests(configs[0], g_pollset, g_mu);
grpc_closure_init(&destroyed, destroy_pollset, g_pollset);
grpc_pollset_shutdown(&exec_ctx, g_pollset, &destroyed);
grpc_exec_ctx_finish(&exec_ctx);
grpc_shutdown();
gpr_mu_destroy(&g_mu);
gpr_free(g_pollset);
return 0;

@ -53,7 +53,7 @@
#include "src/core/iomgr/pollset_posix.h"
#include "test/core/util/test_config.h"
static gpr_mu g_mu;
static gpr_mu *g_mu;
static grpc_pollset *g_pollset;
/* buffer size used to send and receive data.
@ -182,10 +182,10 @@ static void listen_shutdown_cb(grpc_exec_ctx *exec_ctx, void *arg /*server */,
grpc_fd_orphan(exec_ctx, sv->em_fd, NULL, NULL, "b");
gpr_mu_lock(&g_mu);
gpr_mu_lock(g_mu);
sv->done = 1;
grpc_pollset_kick(g_pollset, NULL);
gpr_mu_unlock(&g_mu);
gpr_mu_unlock(g_mu);
}
/* Called when a new TCP connection request arrives in the listening port. */
@ -252,18 +252,18 @@ static int server_start(grpc_exec_ctx *exec_ctx, server *sv) {
/* Wait and shutdown a sever. */
static void server_wait_and_shutdown(server *sv) {
gpr_mu_lock(&g_mu);
gpr_mu_lock(g_mu);
while (!sv->done) {
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
grpc_pollset_worker *worker = NULL;
grpc_pollset_work(&exec_ctx, g_pollset, &worker,
gpr_now(GPR_CLOCK_MONOTONIC),
gpr_inf_future(GPR_CLOCK_MONOTONIC));
gpr_mu_unlock(&g_mu);
gpr_mu_unlock(g_mu);
grpc_exec_ctx_finish(&exec_ctx);
gpr_mu_lock(&g_mu);
gpr_mu_lock(g_mu);
}
gpr_mu_unlock(&g_mu);
gpr_mu_unlock(g_mu);
}
/* ===An upload client to test notify_on_write=== */
@ -310,9 +310,9 @@ static void client_session_write(grpc_exec_ctx *exec_ctx, void *arg, /*client */
ssize_t write_once = 0;
if (!success) {
gpr_mu_lock(&g_mu);
gpr_mu_lock(g_mu);
client_session_shutdown_cb(exec_ctx, arg, 1);
gpr_mu_unlock(&g_mu);
gpr_mu_unlock(g_mu);
return;
}
@ -322,7 +322,7 @@ static void client_session_write(grpc_exec_ctx *exec_ctx, void *arg, /*client */
} while (write_once > 0);
if (errno == EAGAIN) {
gpr_mu_lock(&g_mu);
gpr_mu_lock(g_mu);
if (cl->client_write_cnt < CLIENT_TOTAL_WRITE_CNT) {
cl->write_closure.cb = client_session_write;
cl->write_closure.cb_arg = cl;
@ -331,7 +331,7 @@ static void client_session_write(grpc_exec_ctx *exec_ctx, void *arg, /*client */
} else {
client_session_shutdown_cb(exec_ctx, arg, 1);
}
gpr_mu_unlock(&g_mu);
gpr_mu_unlock(g_mu);
} else {
gpr_log(GPR_ERROR, "unknown errno %s", strerror(errno));
abort();
@ -367,18 +367,18 @@ static void client_start(grpc_exec_ctx *exec_ctx, client *cl, int port) {
/* Wait for the signal to shutdown a client. */
static void client_wait_and_shutdown(client *cl) {
gpr_mu_lock(&g_mu);
gpr_mu_lock(g_mu);
while (!cl->done) {
grpc_pollset_worker *worker = NULL;
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
grpc_pollset_work(&exec_ctx, g_pollset, &worker,
gpr_now(GPR_CLOCK_MONOTONIC),
gpr_inf_future(GPR_CLOCK_MONOTONIC));
gpr_mu_unlock(&g_mu);
gpr_mu_unlock(g_mu);
grpc_exec_ctx_finish(&exec_ctx);
gpr_mu_lock(&g_mu);
gpr_mu_lock(g_mu);
}
gpr_mu_unlock(&g_mu);
gpr_mu_unlock(g_mu);
}
/* Test grpc_fd. Start an upload server and client, upload a stream of
@ -413,20 +413,20 @@ static void first_read_callback(grpc_exec_ctx *exec_ctx,
void *arg /* fd_change_data */, bool success) {
fd_change_data *fdc = arg;
gpr_mu_lock(&g_mu);
gpr_mu_lock(g_mu);
fdc->cb_that_ran = first_read_callback;
grpc_pollset_kick(g_pollset, NULL);
gpr_mu_unlock(&g_mu);
gpr_mu_unlock(g_mu);
}
static void second_read_callback(grpc_exec_ctx *exec_ctx,
void *arg /* fd_change_data */, bool success) {
fd_change_data *fdc = arg;
gpr_mu_lock(&g_mu);
gpr_mu_lock(g_mu);
fdc->cb_that_ran = second_read_callback;
grpc_pollset_kick(g_pollset, NULL);
gpr_mu_unlock(&g_mu);
gpr_mu_unlock(g_mu);
}
/* Test that changing the callback we use for notify_on_read actually works.
@ -468,18 +468,18 @@ static void test_grpc_fd_change(void) {
GPR_ASSERT(result == 1);
/* And now wait for it to run. */
gpr_mu_lock(&g_mu);
gpr_mu_lock(g_mu);
while (a.cb_that_ran == NULL) {
grpc_pollset_worker *worker = NULL;
grpc_pollset_work(&exec_ctx, g_pollset, &worker,
gpr_now(GPR_CLOCK_MONOTONIC),
gpr_inf_future(GPR_CLOCK_MONOTONIC));
gpr_mu_unlock(&g_mu);
gpr_mu_unlock(g_mu);
grpc_exec_ctx_finish(&exec_ctx);
gpr_mu_lock(&g_mu);
gpr_mu_lock(g_mu);
}
GPR_ASSERT(a.cb_that_ran == first_read_callback);
gpr_mu_unlock(&g_mu);
gpr_mu_unlock(g_mu);
/* And drain the socket so we can generate a new read edge */
result = read(sv[0], &data, 1);
@ -492,19 +492,19 @@ static void test_grpc_fd_change(void) {
result = write(sv[1], &data, 1);
GPR_ASSERT(result == 1);
gpr_mu_lock(&g_mu);
gpr_mu_lock(g_mu);
while (b.cb_that_ran == NULL) {
grpc_pollset_worker *worker = NULL;
grpc_pollset_work(&exec_ctx, g_pollset, &worker,
gpr_now(GPR_CLOCK_MONOTONIC),
gpr_inf_future(GPR_CLOCK_MONOTONIC));
gpr_mu_unlock(&g_mu);
gpr_mu_unlock(g_mu);
grpc_exec_ctx_finish(&exec_ctx);
gpr_mu_lock(&g_mu);
gpr_mu_lock(g_mu);
}
/* Except now we verify that second_read_callback ran instead */
GPR_ASSERT(b.cb_that_ran == second_read_callback);
gpr_mu_unlock(&g_mu);
gpr_mu_unlock(g_mu);
grpc_fd_orphan(&exec_ctx, em_fd, NULL, NULL, "d");
grpc_exec_ctx_finish(&exec_ctx);
@ -522,7 +522,6 @@ int main(int argc, char **argv) {
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
grpc_test_init(argc, argv);
grpc_iomgr_init();
gpr_mu_init(&g_mu);
g_pollset = gpr_malloc(grpc_pollset_size());
grpc_pollset_init(g_pollset, &g_mu);
test_grpc_fd();
@ -531,7 +530,6 @@ int main(int argc, char **argv) {
grpc_pollset_shutdown(&exec_ctx, g_pollset, &destroyed);
grpc_exec_ctx_finish(&exec_ctx);
gpr_free(g_pollset);
gpr_mu_destroy(&g_mu);
grpc_iomgr_shutdown();
return 0;
}

@ -50,7 +50,7 @@
#include "test/core/util/test_config.h"
static grpc_pollset_set g_pollset_set;
static gpr_mu g_mu;
static gpr_mu *g_mu;
static grpc_pollset *g_pollset;
static int g_connections_complete = 0;
static grpc_endpoint *g_connecting = NULL;
@ -60,10 +60,10 @@ static gpr_timespec test_deadline(void) {
}
static void finish_connection() {
gpr_mu_lock(&g_mu);
gpr_mu_lock(g_mu);
g_connections_complete++;
grpc_pollset_kick(g_pollset, NULL);
gpr_mu_unlock(&g_mu);
gpr_mu_unlock(g_mu);
}
static void must_succeed(grpc_exec_ctx *exec_ctx, void *arg, bool success) {
@ -101,9 +101,9 @@ void test_succeeds(void) {
GPR_ASSERT(0 == bind(svr_fd, (struct sockaddr *)&addr, addr_len));
GPR_ASSERT(0 == listen(svr_fd, 1));
gpr_mu_lock(&g_mu);
gpr_mu_lock(g_mu);
connections_complete_before = g_connections_complete;
gpr_mu_unlock(&g_mu);
gpr_mu_unlock(g_mu);
/* connect to it */
GPR_ASSERT(getsockname(svr_fd, (struct sockaddr *)&addr, &addr_len) == 0);
@ -120,19 +120,19 @@ void test_succeeds(void) {
GPR_ASSERT(r >= 0);
close(r);
gpr_mu_lock(&g_mu);
gpr_mu_lock(g_mu);
while (g_connections_complete == connections_complete_before) {
grpc_pollset_worker *worker = NULL;
grpc_pollset_work(&exec_ctx, g_pollset, &worker,
gpr_now(GPR_CLOCK_MONOTONIC),
GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5));
gpr_mu_unlock(&g_mu);
gpr_mu_unlock(g_mu);
grpc_exec_ctx_flush(&exec_ctx);
gpr_mu_lock(&g_mu);
gpr_mu_lock(g_mu);
}
gpr_mu_unlock(&g_mu);
gpr_mu_unlock(g_mu);
grpc_exec_ctx_finish(&exec_ctx);
}
@ -149,9 +149,9 @@ void test_fails(void) {
memset(&addr, 0, sizeof(addr));
addr.sin_family = AF_INET;
gpr_mu_lock(&g_mu);
gpr_mu_lock(g_mu);
connections_complete_before = g_connections_complete;
gpr_mu_unlock(&g_mu);
gpr_mu_unlock(g_mu);
/* connect to a broken address */
grpc_closure_init(&done, must_fail, NULL);
@ -159,7 +159,7 @@ void test_fails(void) {
(struct sockaddr *)&addr, addr_len,
gpr_inf_future(GPR_CLOCK_REALTIME));
gpr_mu_lock(&g_mu);
gpr_mu_lock(g_mu);
/* wait for the connection callback to finish */
while (g_connections_complete == connections_complete_before) {
@ -169,12 +169,12 @@ void test_fails(void) {
if (!grpc_timer_check(&exec_ctx, now, &polling_deadline)) {
grpc_pollset_work(&exec_ctx, g_pollset, &worker, now, polling_deadline);
}
gpr_mu_unlock(&g_mu);
gpr_mu_unlock(g_mu);
grpc_exec_ctx_flush(&exec_ctx);
gpr_mu_lock(&g_mu);
gpr_mu_lock(g_mu);
}
gpr_mu_unlock(&g_mu);
gpr_mu_unlock(g_mu);
grpc_exec_ctx_finish(&exec_ctx);
}
@ -219,16 +219,16 @@ void test_times_out(void) {
connect_deadline = GRPC_TIMEOUT_SECONDS_TO_DEADLINE(1);
gpr_mu_lock(&g_mu);
gpr_mu_lock(g_mu);
connections_complete_before = g_connections_complete;
gpr_mu_unlock(&g_mu);
gpr_mu_unlock(g_mu);
grpc_closure_init(&done, must_fail, NULL);
grpc_tcp_client_connect(&exec_ctx, &done, &g_connecting, &g_pollset_set,
(struct sockaddr *)&addr, addr_len, connect_deadline);
/* Make sure the event doesn't trigger early */
gpr_mu_lock(&g_mu);
gpr_mu_lock(g_mu);
for (;;) {
grpc_pollset_worker *worker = NULL;
gpr_timespec now = gpr_now(connect_deadline.clock_type);
@ -256,11 +256,11 @@ void test_times_out(void) {
if (!grpc_timer_check(&exec_ctx, now, &polling_deadline)) {
grpc_pollset_work(&exec_ctx, g_pollset, &worker, now, polling_deadline);
}
gpr_mu_unlock(&g_mu);
gpr_mu_unlock(g_mu);
grpc_exec_ctx_flush(&exec_ctx);
gpr_mu_lock(&g_mu);
gpr_mu_lock(g_mu);
}
gpr_mu_unlock(&g_mu);
gpr_mu_unlock(g_mu);
grpc_exec_ctx_finish(&exec_ctx);
@ -281,7 +281,6 @@ int main(int argc, char **argv) {
grpc_init();
grpc_pollset_set_init(&g_pollset_set);
g_pollset = gpr_malloc(grpc_pollset_size());
gpr_mu_init(&g_mu);
grpc_pollset_init(g_pollset, &g_mu);
grpc_pollset_set_add_pollset(&exec_ctx, &g_pollset_set, g_pollset);
grpc_exec_ctx_finish(&exec_ctx);
@ -295,6 +294,5 @@ int main(int argc, char **argv) {
grpc_exec_ctx_finish(&exec_ctx);
grpc_shutdown();
gpr_free(g_pollset);
gpr_mu_destroy(&g_mu);
return 0;
}

@ -48,7 +48,7 @@
#include "test/core/iomgr/endpoint_tests.h"
#include "test/core/util/test_config.h"
static gpr_mu g_mu;
static gpr_mu *g_mu;
static grpc_pollset *g_pollset;
/*
@ -146,7 +146,7 @@ static void read_cb(grpc_exec_ctx *exec_ctx, void *user_data, bool success) {
GPR_ASSERT(success);
gpr_mu_lock(&g_mu);
gpr_mu_lock(g_mu);
current_data = state->read_bytes % 256;
read_bytes = count_slices(state->incoming.slices, state->incoming.count,
&current_data);
@ -154,10 +154,10 @@ static void read_cb(grpc_exec_ctx *exec_ctx, void *user_data, bool success) {
gpr_log(GPR_INFO, "Read %d bytes of %d", read_bytes,
state->target_read_bytes);
if (state->read_bytes >= state->target_read_bytes) {
gpr_mu_unlock(&g_mu);
gpr_mu_unlock(g_mu);
} else {
grpc_endpoint_read(exec_ctx, state->ep, &state->incoming, &state->read_cb);
gpr_mu_unlock(&g_mu);
gpr_mu_unlock(g_mu);
}
}
@ -189,17 +189,17 @@ static void read_test(size_t num_bytes, size_t slice_size) {
grpc_endpoint_read(&exec_ctx, ep, &state.incoming, &state.read_cb);
gpr_mu_lock(&g_mu);
gpr_mu_lock(g_mu);
while (state.read_bytes < state.target_read_bytes) {
grpc_pollset_worker *worker = NULL;
grpc_pollset_work(&exec_ctx, g_pollset, &worker,
gpr_now(GPR_CLOCK_MONOTONIC), deadline);
gpr_mu_unlock(&g_mu);
gpr_mu_unlock(g_mu);
grpc_exec_ctx_finish(&exec_ctx);
gpr_mu_lock(&g_mu);
gpr_mu_lock(g_mu);
}
GPR_ASSERT(state.read_bytes == state.target_read_bytes);
gpr_mu_unlock(&g_mu);
gpr_mu_unlock(g_mu);
gpr_slice_buffer_destroy(&state.incoming);
grpc_endpoint_destroy(&exec_ctx, ep);
@ -235,17 +235,17 @@ static void large_read_test(size_t slice_size) {
grpc_endpoint_read(&exec_ctx, ep, &state.incoming, &state.read_cb);
gpr_mu_lock(&g_mu);
gpr_mu_lock(g_mu);
while (state.read_bytes < state.target_read_bytes) {
grpc_pollset_worker *worker = NULL;
grpc_pollset_work(&exec_ctx, g_pollset, &worker,
gpr_now(GPR_CLOCK_MONOTONIC), deadline);
gpr_mu_unlock(&g_mu);
gpr_mu_unlock(g_mu);
grpc_exec_ctx_finish(&exec_ctx);
gpr_mu_lock(&g_mu);
gpr_mu_lock(g_mu);
}
GPR_ASSERT(state.read_bytes == state.target_read_bytes);
gpr_mu_unlock(&g_mu);
gpr_mu_unlock(g_mu);
gpr_slice_buffer_destroy(&state.incoming);
grpc_endpoint_destroy(&exec_ctx, ep);
@ -284,11 +284,11 @@ static void write_done(grpc_exec_ctx *exec_ctx,
void *user_data /* write_socket_state */, bool success) {
struct write_socket_state *state = (struct write_socket_state *)user_data;
gpr_log(GPR_INFO, "Write done callback called");
gpr_mu_lock(&g_mu);
gpr_mu_lock(g_mu);
gpr_log(GPR_INFO, "Signalling write done");
state->write_done = 1;
grpc_pollset_kick(g_pollset, NULL);
gpr_mu_unlock(&g_mu);
gpr_mu_unlock(g_mu);
}
void drain_socket_blocking(int fd, size_t num_bytes, size_t read_size) {
@ -305,11 +305,11 @@ void drain_socket_blocking(int fd, size_t num_bytes, size_t read_size) {
for (;;) {
grpc_pollset_worker *worker = NULL;
gpr_mu_lock(&g_mu);
gpr_mu_lock(g_mu);
grpc_pollset_work(&exec_ctx, g_pollset, &worker,
gpr_now(GPR_CLOCK_MONOTONIC),
GRPC_TIMEOUT_MILLIS_TO_DEADLINE(10));
gpr_mu_unlock(&g_mu);
gpr_mu_unlock(g_mu);
grpc_exec_ctx_finish(&exec_ctx);
do {
bytes_read =
@ -364,7 +364,7 @@ static void write_test(size_t num_bytes, size_t slice_size) {
grpc_endpoint_write(&exec_ctx, ep, &outgoing, &write_done_closure);
drain_socket_blocking(sv[0], num_bytes, num_bytes);
gpr_mu_lock(&g_mu);
gpr_mu_lock(g_mu);
for (;;) {
grpc_pollset_worker *worker = NULL;
if (state.write_done) {
@ -372,11 +372,11 @@ static void write_test(size_t num_bytes, size_t slice_size) {
}
grpc_pollset_work(&exec_ctx, g_pollset, &worker,
gpr_now(GPR_CLOCK_MONOTONIC), deadline);
gpr_mu_unlock(&g_mu);
gpr_mu_unlock(g_mu);
grpc_exec_ctx_finish(&exec_ctx);
gpr_mu_lock(&g_mu);
gpr_mu_lock(g_mu);
}
gpr_mu_unlock(&g_mu);
gpr_mu_unlock(g_mu);
gpr_slice_buffer_destroy(&outgoing);
grpc_endpoint_destroy(&exec_ctx, ep);
@ -424,27 +424,27 @@ static void release_fd_test(size_t num_bytes, size_t slice_size) {
grpc_endpoint_read(&exec_ctx, ep, &state.incoming, &state.read_cb);
gpr_mu_lock(&g_mu);
gpr_mu_lock(g_mu);
while (state.read_bytes < state.target_read_bytes) {
grpc_pollset_worker *worker = NULL;
grpc_pollset_work(&exec_ctx, g_pollset, &worker,
gpr_now(GPR_CLOCK_MONOTONIC), deadline);
gpr_mu_unlock(&g_mu);
gpr_mu_unlock(g_mu);
grpc_exec_ctx_finish(&exec_ctx);
gpr_mu_lock(&g_mu);
gpr_mu_lock(g_mu);
}
GPR_ASSERT(state.read_bytes == state.target_read_bytes);
gpr_mu_unlock(&g_mu);
gpr_mu_unlock(g_mu);
gpr_slice_buffer_destroy(&state.incoming);
grpc_tcp_destroy_and_release_fd(&exec_ctx, ep, &fd, &fd_released_cb);
gpr_mu_lock(&g_mu);
gpr_mu_lock(g_mu);
while (!fd_released_done) {
grpc_pollset_worker *worker = NULL;
grpc_pollset_work(&exec_ctx, g_pollset, &worker,
gpr_now(GPR_CLOCK_MONOTONIC), deadline);
}
gpr_mu_unlock(&g_mu);
gpr_mu_unlock(g_mu);
GPR_ASSERT(fd_released_done == 1);
GPR_ASSERT(fd == sv[1]);
grpc_exec_ctx_finish(&exec_ctx);
@ -514,16 +514,14 @@ int main(int argc, char **argv) {
grpc_test_init(argc, argv);
grpc_init();
g_pollset = gpr_malloc(grpc_pollset_size());
gpr_mu_init(&g_mu);
grpc_pollset_init(g_pollset, &g_mu);
run_tests();
grpc_endpoint_tests(configs[0], g_pollset, &g_mu);
grpc_endpoint_tests(configs[0], g_pollset, g_mu);
grpc_closure_init(&destroyed, destroy_pollset, g_pollset);
grpc_pollset_shutdown(&exec_ctx, g_pollset, &destroyed);
grpc_exec_ctx_finish(&exec_ctx);
grpc_shutdown();
gpr_free(g_pollset);
gpr_mu_destroy(&g_mu);
return 0;
}

@ -52,7 +52,7 @@
#define LOG_TEST(x) gpr_log(GPR_INFO, "%s", #x)
static gpr_mu g_mu;
static gpr_mu *g_mu;
static grpc_pollset *g_pollset;
static int g_nconnects = 0;
@ -117,11 +117,11 @@ static void on_connect(grpc_exec_ctx *exec_ctx, void *arg, grpc_endpoint *tcp,
grpc_endpoint_shutdown(exec_ctx, tcp);
grpc_endpoint_destroy(exec_ctx, tcp);
gpr_mu_lock(&g_mu);
gpr_mu_lock(g_mu);
on_connect_result_set(&g_result, acceptor);
g_nconnects++;
grpc_pollset_kick(g_pollset, NULL);
gpr_mu_unlock(&g_mu);
gpr_mu_unlock(g_mu);
}
static void test_no_op(void) {
@ -178,7 +178,7 @@ static void tcp_connect(grpc_exec_ctx *exec_ctx, const struct sockaddr *remote,
int clifd = socket(remote->sa_family, SOCK_STREAM, 0);
int nconnects_before;
gpr_mu_lock(&g_mu);
gpr_mu_lock(g_mu);
nconnects_before = g_nconnects;
on_connect_result_init(&g_result);
GPR_ASSERT(clifd >= 0);
@ -190,16 +190,16 @@ static void tcp_connect(grpc_exec_ctx *exec_ctx, const struct sockaddr *remote,
grpc_pollset_worker *worker = NULL;
grpc_pollset_work(exec_ctx, g_pollset, &worker,
gpr_now(GPR_CLOCK_MONOTONIC), deadline);
gpr_mu_unlock(&g_mu);
gpr_mu_unlock(g_mu);
grpc_exec_ctx_finish(exec_ctx);
gpr_mu_lock(&g_mu);
gpr_mu_lock(g_mu);
}
gpr_log(GPR_DEBUG, "wait done");
GPR_ASSERT(g_nconnects == nconnects_before + 1);
close(clifd);
*result = g_result;
gpr_mu_unlock(&g_mu);
gpr_mu_unlock(g_mu);
}
/* Tests a tcp server with multiple ports. TODO(daniel-j-born): Multiple fds for
@ -315,7 +315,6 @@ int main(int argc, char **argv) {
grpc_test_init(argc, argv);
grpc_init();
g_pollset = gpr_malloc(grpc_pollset_size());
gpr_mu_init(&g_mu);
grpc_pollset_init(g_pollset, &g_mu);
test_no_op();
@ -330,6 +329,5 @@ int main(int argc, char **argv) {
grpc_exec_ctx_finish(&exec_ctx);
grpc_shutdown();
gpr_free(g_pollset);
gpr_mu_destroy(&g_mu);
return 0;
}

@ -39,15 +39,15 @@
#include "test/core/util/test_config.h"
static gpr_mu g_mu;
static gpr_mu *g_mu;
static grpc_pollset *g_pollset;
static void must_succeed(grpc_exec_ctx *exec_ctx, void *p, bool success) {
GPR_ASSERT(success == 1);
gpr_mu_lock(&g_mu);
gpr_mu_lock(g_mu);
*(int *)p = 1;
grpc_pollset_kick(g_pollset, NULL);
gpr_mu_unlock(&g_mu);
gpr_mu_unlock(g_mu);
}
static void test_ref_unref(void) {
@ -71,11 +71,11 @@ static void test_add_closure(void) {
grpc_workqueue_push(wq, &c, 1);
grpc_workqueue_add_to_pollset(&exec_ctx, wq, g_pollset);
gpr_mu_lock(&g_mu);
gpr_mu_lock(g_mu);
GPR_ASSERT(!done);
grpc_pollset_work(&exec_ctx, g_pollset, &worker, gpr_now(deadline.clock_type),
deadline);
gpr_mu_unlock(&g_mu);
gpr_mu_unlock(g_mu);
grpc_exec_ctx_finish(&exec_ctx);
GPR_ASSERT(done);
@ -96,11 +96,11 @@ static void test_flush(void) {
grpc_workqueue_flush(&exec_ctx, wq);
grpc_workqueue_add_to_pollset(&exec_ctx, wq, g_pollset);
gpr_mu_lock(&g_mu);
gpr_mu_lock(g_mu);
GPR_ASSERT(!done);
grpc_pollset_work(&exec_ctx, g_pollset, &worker, gpr_now(deadline.clock_type),
deadline);
gpr_mu_unlock(&g_mu);
gpr_mu_unlock(g_mu);
grpc_exec_ctx_finish(&exec_ctx);
GPR_ASSERT(done);
@ -117,7 +117,6 @@ int main(int argc, char **argv) {
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
grpc_test_init(argc, argv);
grpc_init();
gpr_mu_init(&g_mu);
g_pollset = gpr_malloc(grpc_pollset_size());
grpc_pollset_init(g_pollset, &g_mu);
@ -131,6 +130,5 @@ int main(int argc, char **argv) {
grpc_shutdown();
gpr_free(g_pollset);
gpr_mu_destroy(&g_mu);
return 0;
}

@ -45,7 +45,7 @@
#include "src/core/security/credentials.h"
typedef struct {
gpr_mu mu;
gpr_mu *mu;
grpc_pollset *pollset;
int is_done;
char *token;
@ -67,11 +67,11 @@ static void on_oauth2_response(grpc_exec_ctx *exec_ctx, void *user_data,
GPR_SLICE_LENGTH(token_slice));
token[GPR_SLICE_LENGTH(token_slice)] = '\0';
}
gpr_mu_lock(&request->mu);
gpr_mu_lock(request->mu);
request->is_done = 1;
request->token = token;
grpc_pollset_kick(request->pollset, NULL);
gpr_mu_unlock(&request->mu);
gpr_mu_unlock(request->mu);
}
static void do_nothing(grpc_exec_ctx *exec_ctx, void *unused, bool success) {}
@ -95,14 +95,14 @@ char *grpc_test_fetch_oauth2_token_with_credentials(
grpc_exec_ctx_finish(&exec_ctx);
gpr_mu_lock(&request.mu);
gpr_mu_lock(request.mu);
while (!request.is_done) {
grpc_pollset_worker *worker = NULL;
grpc_pollset_work(&exec_ctx, request.pollset, &worker,
gpr_now(GPR_CLOCK_MONOTONIC),
gpr_inf_future(GPR_CLOCK_MONOTONIC));
}
gpr_mu_unlock(&request.mu);
gpr_mu_unlock(request.mu);
grpc_pollset_shutdown(&exec_ctx, request.pollset, &do_nothing_closure);
grpc_exec_ctx_finish(&exec_ctx);

@ -46,7 +46,7 @@
#include "src/core/support/string.h"
typedef struct {
gpr_mu mu;
gpr_mu *mu;
grpc_pollset *pollset;
int is_done;
} synchronizer;
@ -64,10 +64,10 @@ static void on_metadata_response(grpc_exec_ctx *exec_ctx, void *user_data,
printf("\nGot token: %s\n\n", token);
gpr_free(token);
}
gpr_mu_lock(&sync->mu);
gpr_mu_lock(sync->mu);
sync->is_done = 1;
grpc_pollset_kick(sync->pollset, NULL);
gpr_mu_unlock(&sync->mu);
gpr_mu_unlock(sync->mu);
}
int main(int argc, char **argv) {
@ -94,7 +94,6 @@ int main(int argc, char **argv) {
}
sync.pollset = gpr_malloc(grpc_pollset_size());
gpr_mu_init(&sync.mu);
grpc_pollset_init(sync.pollset, &sync.mu);
sync.is_done = 0;
@ -102,21 +101,22 @@ int main(int argc, char **argv) {
&exec_ctx, ((grpc_composite_channel_credentials *)creds)->call_creds,
sync.pollset, context, on_metadata_response, &sync);
gpr_mu_lock(&sync.mu);
gpr_mu_lock(sync.mu);
while (!sync.is_done) {
grpc_pollset_worker *worker = NULL;
grpc_pollset_work(&exec_ctx, sync.pollset, &worker,
gpr_now(GPR_CLOCK_MONOTONIC),
gpr_inf_future(GPR_CLOCK_MONOTONIC));
gpr_mu_unlock(&sync.mu);
grpc_exec_ctx_finish(&exec_ctx);
gpr_mu_lock(&sync.mu);
gpr_mu_unlock(sync.mu);
grpc_exec_ctx_flush(&exec_ctx);
gpr_mu_lock(sync.mu);
}
gpr_mu_unlock(&sync.mu);
gpr_mu_unlock(sync.mu);
grpc_exec_ctx_finish(&exec_ctx);
grpc_channel_credentials_release(creds);
gpr_free(sync.pollset);
gpr_mu_destroy(&sync.mu);
end:
gpr_cmdline_destroy(cl);

@ -45,7 +45,7 @@
#include "src/core/tsi/fake_transport_security.h"
#include "test/core/util/test_config.h"
static gpr_mu g_mu;
static gpr_mu *g_mu;
static grpc_pollset *g_pollset;
static grpc_endpoint_test_fixture secure_endpoint_create_fixture_tcp_socketpair(
@ -183,9 +183,8 @@ int main(int argc, char **argv) {
grpc_init();
g_pollset = gpr_malloc(grpc_pollset_size());
gpr_mu_init(&g_mu);
grpc_pollset_init(g_pollset, &g_mu);
grpc_endpoint_tests(configs[0], g_pollset, &g_mu);
grpc_endpoint_tests(configs[0], g_pollset, g_mu);
test_leftover(configs[1], 1);
grpc_closure_init(&destroyed, destroy_pollset, g_pollset);
grpc_pollset_shutdown(&exec_ctx, g_pollset, &destroyed);
@ -193,7 +192,6 @@ int main(int argc, char **argv) {
grpc_shutdown();
gpr_free(g_pollset);
gpr_mu_destroy(&g_mu);
return 0;
}

@ -46,7 +46,7 @@
typedef struct {
grpc_pollset *pollset;
gpr_mu mu;
gpr_mu *mu;
int is_done;
int success;
} synchronizer;
@ -79,10 +79,10 @@ static void on_jwt_verification_done(void *user_data,
grpc_jwt_verifier_status_to_string(status));
}
gpr_mu_lock(&sync->mu);
gpr_mu_lock(sync->mu);
sync->is_done = 1;
grpc_pollset_kick(sync->pollset, NULL);
gpr_mu_unlock(&sync->mu);
gpr_mu_unlock(sync->mu);
}
int main(int argc, char **argv) {
@ -106,26 +106,24 @@ int main(int argc, char **argv) {
grpc_init();
sync.pollset = gpr_malloc(grpc_pollset_size());
gpr_mu_init(&sync.mu);
grpc_pollset_init(sync.pollset, &sync.mu);
sync.is_done = 0;
grpc_jwt_verifier_verify(&exec_ctx, verifier, sync.pollset, jwt, aud,
on_jwt_verification_done, &sync);
gpr_mu_lock(&sync.mu);
gpr_mu_lock(sync.mu);
while (!sync.is_done) {
grpc_pollset_worker *worker = NULL;
grpc_pollset_work(&exec_ctx, sync.pollset, &worker,
gpr_now(GPR_CLOCK_MONOTONIC),
gpr_inf_future(GPR_CLOCK_MONOTONIC));
gpr_mu_unlock(&sync.mu);
gpr_mu_unlock(sync.mu);
grpc_exec_ctx_finish(&exec_ctx);
gpr_mu_lock(&sync.mu);
gpr_mu_lock(sync.mu);
}
gpr_mu_unlock(&sync.mu);
gpr_mu_unlock(sync.mu);
gpr_mu_destroy(&sync.mu);
gpr_free(sync.pollset);
grpc_jwt_verifier_destroy(verifier);

@ -69,7 +69,7 @@ static int has_port_been_chosen(int port) {
}
typedef struct freereq {
gpr_mu mu;
gpr_mu *mu;
grpc_pollset *pollset;
int done;
} freereq;
@ -83,10 +83,10 @@ static void destroy_pollset_and_shutdown(grpc_exec_ctx *exec_ctx, void *p,
static void freed_port_from_server(grpc_exec_ctx *exec_ctx, void *arg,
const grpc_httpcli_response *response) {
freereq *pr = arg;
gpr_mu_lock(&pr->mu);
gpr_mu_lock(pr->mu);
pr->done = 1;
grpc_pollset_kick(pr->pollset, NULL);
gpr_mu_unlock(&pr->mu);
gpr_mu_unlock(pr->mu);
}
static void free_port_using_server(char *server, int port) {
@ -103,7 +103,6 @@ static void free_port_using_server(char *server, int port) {
memset(&req, 0, sizeof(req));
pr.pollset = gpr_malloc(grpc_pollset_size());
gpr_mu_init(&pr.mu);
grpc_pollset_init(pr.pollset, &pr.mu);
grpc_closure_init(&shutdown_closure, destroy_pollset_and_shutdown,
pr.pollset);
@ -116,21 +115,20 @@ static void free_port_using_server(char *server, int port) {
grpc_httpcli_get(&exec_ctx, &context, pr.pollset, &req,
GRPC_TIMEOUT_SECONDS_TO_DEADLINE(10), freed_port_from_server,
&pr);
gpr_mu_lock(&pr.mu);
gpr_mu_lock(pr.mu);
while (!pr.done) {
grpc_pollset_worker *worker = NULL;
grpc_pollset_work(&exec_ctx, pr.pollset, &worker,
gpr_now(GPR_CLOCK_MONOTONIC),
GRPC_TIMEOUT_SECONDS_TO_DEADLINE(1));
}
gpr_mu_unlock(&pr.mu);
gpr_mu_unlock(pr.mu);
grpc_httpcli_context_destroy(&context);
grpc_exec_ctx_finish(&exec_ctx);
grpc_pollset_shutdown(&exec_ctx, pr.pollset, &shutdown_closure);
grpc_exec_ctx_finish(&exec_ctx);
gpr_free(pr.pollset);
gpr_mu_destroy(&pr.mu);
gpr_free(path);
}
@ -208,7 +206,7 @@ static int is_port_available(int *port, int is_tcp) {
}
typedef struct portreq {
gpr_mu mu;
gpr_mu *mu;
grpc_pollset *pollset;
int port;
int retries;
@ -253,10 +251,10 @@ static void got_port_from_server(grpc_exec_ctx *exec_ctx, void *arg,
port = port * 10 + response->body[i] - '0';
}
GPR_ASSERT(port > 1024);
gpr_mu_lock(&pr->mu);
gpr_mu_lock(pr->mu);
pr->port = port;
grpc_pollset_kick(pr->pollset, NULL);
gpr_mu_unlock(&pr->mu);
gpr_mu_unlock(pr->mu);
}
static int pick_port_using_server(char *server) {
@ -271,7 +269,6 @@ static int pick_port_using_server(char *server) {
memset(&pr, 0, sizeof(pr));
memset(&req, 0, sizeof(req));
pr.pollset = gpr_malloc(grpc_pollset_size());
gpr_mu_init(&pr.mu);
grpc_pollset_init(pr.pollset, &pr.mu);
grpc_closure_init(&shutdown_closure, destroy_pollset_and_shutdown,
pr.pollset);
@ -287,20 +284,20 @@ static int pick_port_using_server(char *server) {
GRPC_TIMEOUT_SECONDS_TO_DEADLINE(10), got_port_from_server,
&pr);
grpc_exec_ctx_finish(&exec_ctx);
gpr_mu_lock(&pr.mu);
gpr_mu_lock(pr.mu);
while (pr.port == -1) {
grpc_pollset_worker *worker = NULL;
grpc_pollset_work(&exec_ctx, pr.pollset, &worker,
gpr_now(GPR_CLOCK_MONOTONIC),
GRPC_TIMEOUT_SECONDS_TO_DEADLINE(1));
}
gpr_mu_unlock(&pr.mu);
gpr_mu_unlock(pr.mu);
grpc_httpcli_context_destroy(&context);
grpc_pollset_shutdown(&exec_ctx, pr.pollset, &shutdown_closure);
grpc_exec_ctx_finish(&exec_ctx);
gpr_free(pr.pollset);
gpr_mu_destroy(&pr.mu);
gpr_mu_destroy(pr.mu);
return pr.port;
}

@ -58,7 +58,6 @@ void test_tcp_server_init(test_tcp_server *server,
grpc_closure_init(&server->shutdown_complete, on_server_destroyed, server);
server->shutdown = 0;
server->pollset = gpr_malloc(grpc_pollset_size());
gpr_mu_init(&server->mu);
grpc_pollset_init(server->pollset, &server->mu);
server->on_connect = on_connect;
server->cb_data = user_data;
@ -91,10 +90,10 @@ void test_tcp_server_poll(test_tcp_server *server, int seconds) {
gpr_time_add(gpr_now(GPR_CLOCK_MONOTONIC),
gpr_time_from_seconds(seconds, GPR_TIMESPAN));
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
gpr_mu_lock(&server->mu);
gpr_mu_lock(server->mu);
grpc_pollset_work(&exec_ctx, server->pollset, &worker,
gpr_now(GPR_CLOCK_MONOTONIC), deadline);
gpr_mu_unlock(&server->mu);
gpr_mu_unlock(server->mu);
grpc_exec_ctx_finish(&exec_ctx);
}
@ -116,6 +115,5 @@ void test_tcp_server_destroy(test_tcp_server *server) {
grpc_exec_ctx_finish(&exec_ctx);
grpc_pollset_destroy(server->pollset);
gpr_free(server->pollset);
gpr_mu_destroy(&server->mu);
grpc_shutdown();
}

@ -41,7 +41,7 @@ typedef struct test_tcp_server {
grpc_tcp_server *tcp_server;
grpc_closure shutdown_complete;
int shutdown;
gpr_mu mu;
gpr_mu *mu;
grpc_pollset *pollset;
grpc_tcp_server_cb on_connect;
void *cb_data;

Loading…
Cancel
Save