/* * * Copyright 2015-2016, Google Inc. * All rights reserved. * * Redistribution and use in source and binary forms, with or without * modification, are permitted provided that the following conditions are * met: * * * Redistributions of source code must retain the above copyright * notice, this list of conditions and the following disclaimer. * * Redistributions in binary form must reproduce the above * copyright notice, this list of conditions and the following disclaimer * in the documentation and/or other materials provided with the * distribution. * * Neither the name of Google Inc. nor the names of its * contributors may be used to endorse or promote products derived from * this software without specific prior written permission. * * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. * */ #include #ifdef GPR_WINSOCK_SOCKET #include #include #include "src/core/iomgr/timer_internal.h" #include "src/core/iomgr/iomgr_internal.h" #include "src/core/iomgr/iocp_windows.h" #include "src/core/iomgr/pollset.h" #include "src/core/iomgr/pollset_windows.h" gpr_mu grpc_polling_mu; static grpc_pollset_worker *g_active_poller; static grpc_pollset_worker g_global_root_worker; void grpc_pollset_global_init() { gpr_mu_init(&grpc_polling_mu); g_active_poller = NULL; g_global_root_worker.links[GRPC_POLLSET_WORKER_LINK_GLOBAL].next = g_global_root_worker.links[GRPC_POLLSET_WORKER_LINK_GLOBAL].prev = &g_global_root_worker; } void grpc_pollset_global_shutdown() { gpr_mu_destroy(&grpc_polling_mu); } static void remove_worker(grpc_pollset_worker *worker, grpc_pollset_worker_link_type type) { worker->links[type].prev->links[type].next = worker->links[type].next; worker->links[type].next->links[type].prev = worker->links[type].prev; worker->links[type].next = worker->links[type].prev = worker; } static int has_workers(grpc_pollset_worker *root, grpc_pollset_worker_link_type type) { return root->links[type].next != root; } static grpc_pollset_worker *pop_front_worker( grpc_pollset_worker *root, grpc_pollset_worker_link_type type) { if (has_workers(root, type)) { grpc_pollset_worker *w = root->links[type].next; remove_worker(w, type); return w; } else { return NULL; } } static void push_front_worker(grpc_pollset_worker *root, grpc_pollset_worker_link_type type, grpc_pollset_worker *worker) { worker->links[type].prev = root; worker->links[type].next = worker->links[type].prev->links[type].next; worker->links[type].prev->links[type].next = worker->links[type].next->links[type].prev = worker; } /* There isn't really any such thing as a pollset under Windows, due to the nature of the IO completion ports. We're still going to provide a minimal set of features for the sake of the rest of grpc. But grpc_pollset_work won't actually do any polling, and return as quickly as possible. */ void grpc_pollset_init(grpc_pollset *pollset) { memset(pollset, 0, sizeof(*pollset)); pollset->root_worker.links[GRPC_POLLSET_WORKER_LINK_POLLSET].next = pollset->root_worker.links[GRPC_POLLSET_WORKER_LINK_POLLSET].prev = &pollset->root_worker; } void grpc_pollset_shutdown(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset, grpc_closure *closure) { pollset->shutting_down = 1; grpc_pollset_kick(pollset, GRPC_POLLSET_KICK_BROADCAST); if (!pollset->is_iocp_worker) { grpc_exec_ctx_enqueue(exec_ctx, closure, true, NULL); } else { pollset->on_shutdown = closure; } } void grpc_pollset_destroy(grpc_pollset *pollset) {} void grpc_pollset_reset(grpc_pollset *pollset) { GPR_ASSERT(pollset->shutting_down); GPR_ASSERT( !has_workers(&pollset->root_worker, GRPC_POLLSET_WORKER_LINK_POLLSET)); pollset->shutting_down = 0; pollset->is_iocp_worker = 0; pollset->kicked_without_pollers = 0; pollset->on_shutdown = NULL; } void grpc_pollset_work(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset, grpc_pollset_worker *worker, gpr_timespec now, gpr_timespec deadline) { int added_worker = 0; worker->links[GRPC_POLLSET_WORKER_LINK_POLLSET].next = worker->links[GRPC_POLLSET_WORKER_LINK_POLLSET].prev = worker->links[GRPC_POLLSET_WORKER_LINK_GLOBAL].next = worker->links[GRPC_POLLSET_WORKER_LINK_GLOBAL].prev = NULL; worker->kicked = 0; worker->pollset = pollset; gpr_cv_init(&worker->cv); if (grpc_timer_check(exec_ctx, now, &deadline)) { goto done; } if (!pollset->kicked_without_pollers && !pollset->shutting_down) { if (g_active_poller == NULL) { grpc_pollset_worker *next_worker; /* become poller */ pollset->is_iocp_worker = 1; g_active_poller = worker; gpr_mu_unlock(&grpc_polling_mu); grpc_iocp_work(exec_ctx, deadline); grpc_exec_ctx_flush(exec_ctx); gpr_mu_lock(&grpc_polling_mu); pollset->is_iocp_worker = 0; g_active_poller = NULL; /* try to get a worker from this pollsets worker list */ next_worker = pop_front_worker(&pollset->root_worker, GRPC_POLLSET_WORKER_LINK_POLLSET); if (next_worker == NULL) { /* try to get a worker from the global list */ next_worker = pop_front_worker(&g_global_root_worker, GRPC_POLLSET_WORKER_LINK_GLOBAL); } if (next_worker != NULL) { next_worker->kicked = 1; gpr_cv_signal(&next_worker->cv); } if (pollset->shutting_down && pollset->on_shutdown != NULL) { grpc_exec_ctx_enqueue(exec_ctx, pollset->on_shutdown, true, NULL); pollset->on_shutdown = NULL; } goto done; } push_front_worker(&g_global_root_worker, GRPC_POLLSET_WORKER_LINK_GLOBAL, worker); push_front_worker(&pollset->root_worker, GRPC_POLLSET_WORKER_LINK_POLLSET, worker); added_worker = 1; while (!worker->kicked) { if (gpr_cv_wait(&worker->cv, &grpc_polling_mu, deadline)) { break; } } } else { pollset->kicked_without_pollers = 0; } done: if (!grpc_closure_list_empty(exec_ctx->closure_list)) { gpr_mu_unlock(&grpc_polling_mu); grpc_exec_ctx_flush(exec_ctx); gpr_mu_lock(&grpc_polling_mu); } if (added_worker) { remove_worker(worker, GRPC_POLLSET_WORKER_LINK_GLOBAL); remove_worker(worker, GRPC_POLLSET_WORKER_LINK_POLLSET); } gpr_cv_destroy(&worker->cv); } void grpc_pollset_kick(grpc_pollset *p, grpc_pollset_worker *specific_worker) { if (specific_worker != NULL) { if (specific_worker == GRPC_POLLSET_KICK_BROADCAST) { for (specific_worker = p->root_worker.links[GRPC_POLLSET_WORKER_LINK_POLLSET].next; specific_worker != &p->root_worker; specific_worker = specific_worker->links[GRPC_POLLSET_WORKER_LINK_POLLSET].next) { specific_worker->kicked = 1; gpr_cv_signal(&specific_worker->cv); } p->kicked_without_pollers = 1; if (p->is_iocp_worker) { grpc_iocp_kick(); } } else { if (p->is_iocp_worker && g_active_poller == specific_worker) { grpc_iocp_kick(); } else { specific_worker->kicked = 1; gpr_cv_signal(&specific_worker->cv); } } } else { specific_worker = pop_front_worker(&p->root_worker, GRPC_POLLSET_WORKER_LINK_POLLSET); if (specific_worker != NULL) { grpc_pollset_kick(p, specific_worker); } else if (p->is_iocp_worker) { grpc_iocp_kick(); } else { p->kicked_without_pollers = 1; } } } void grpc_kick_poller(void) { grpc_iocp_kick(); } #endif /* GPR_WINSOCK_SOCKET */