diff options
Diffstat (limited to 'src/core/iomgr')
47 files changed, 661 insertions, 145 deletions
diff --git a/src/core/iomgr/alarm.c b/src/core/iomgr/alarm.c index 68d33b9cf6..ddb30dc4bb 100644 --- a/src/core/iomgr/alarm.c +++ b/src/core/iomgr/alarm.c @@ -105,8 +105,7 @@ void grpc_alarm_list_init(gpr_timespec now) { void grpc_alarm_list_shutdown(void) { int i; - while (run_some_expired_alarms(NULL, gpr_inf_future(g_clock_type), NULL, - 0)) + while (run_some_expired_alarms(NULL, gpr_inf_future(g_clock_type), NULL, 0)) ; for (i = 0; i < NUM_SHARDS; i++) { shard_type *shard = &g_shards[i]; @@ -362,7 +361,7 @@ static int run_some_expired_alarms(gpr_mu *drop_mu, gpr_timespec now, int grpc_alarm_check(gpr_mu *drop_mu, gpr_timespec now, gpr_timespec *next) { GPR_ASSERT(now.clock_type == g_clock_type); return run_some_expired_alarms( - drop_mu, now, next, + drop_mu, now, next, gpr_time_cmp(now, gpr_inf_future(now.clock_type)) != 0); } diff --git a/src/core/iomgr/alarm.h b/src/core/iomgr/alarm.h index c067a0b8a3..4a13527e64 100644 --- a/src/core/iomgr/alarm.h +++ b/src/core/iomgr/alarm.h @@ -86,4 +86,4 @@ void grpc_alarm_init(grpc_alarm *alarm, gpr_timespec deadline, Requires: cancel() must happen after add() on a given alarm */ void grpc_alarm_cancel(grpc_alarm *alarm); -#endif /* GRPC_INTERNAL_CORE_IOMGR_ALARM_H */ +#endif /* GRPC_INTERNAL_CORE_IOMGR_ALARM_H */ diff --git a/src/core/iomgr/alarm_heap.c b/src/core/iomgr/alarm_heap.c index d912178fda..daed251982 100644 --- a/src/core/iomgr/alarm_heap.c +++ b/src/core/iomgr/alarm_heap.c @@ -66,11 +66,11 @@ static void adjust_downwards(grpc_alarm **first, int i, int length, int next_i; if (left_child >= length) break; right_child = left_child + 1; - next_i = - right_child < length && gpr_time_cmp(first[left_child]->deadline, - first[right_child]->deadline) < 0 - ? right_child - : left_child; + next_i = right_child < length && + gpr_time_cmp(first[left_child]->deadline, + first[right_child]->deadline) < 0 + ? right_child + : left_child; if (gpr_time_cmp(t->deadline, first[next_i]->deadline) >= 0) break; first[i] = first[next_i]; first[i]->heap_index = i; diff --git a/src/core/iomgr/alarm_heap.h b/src/core/iomgr/alarm_heap.h index c5adfc6d31..60db6c991b 100644 --- a/src/core/iomgr/alarm_heap.h +++ b/src/core/iomgr/alarm_heap.h @@ -54,4 +54,4 @@ void grpc_alarm_heap_pop(grpc_alarm_heap *heap); int grpc_alarm_heap_is_empty(grpc_alarm_heap *heap); -#endif /* GRPC_INTERNAL_CORE_IOMGR_ALARM_HEAP_H */ +#endif /* GRPC_INTERNAL_CORE_IOMGR_ALARM_HEAP_H */ diff --git a/src/core/iomgr/alarm_internal.h b/src/core/iomgr/alarm_internal.h index 0268a01bad..e9f98a3444 100644 --- a/src/core/iomgr/alarm_internal.h +++ b/src/core/iomgr/alarm_internal.h @@ -59,4 +59,4 @@ gpr_timespec grpc_alarm_list_next_timeout(void); void grpc_kick_poller(void); -#endif /* GRPC_INTERNAL_CORE_IOMGR_ALARM_INTERNAL_H */ +#endif /* GRPC_INTERNAL_CORE_IOMGR_ALARM_INTERNAL_H */ diff --git a/src/core/iomgr/endpoint.c b/src/core/iomgr/endpoint.c index 744fe7656c..8ee14bce9b 100644 --- a/src/core/iomgr/endpoint.c +++ b/src/core/iomgr/endpoint.c @@ -50,7 +50,8 @@ void grpc_endpoint_add_to_pollset(grpc_endpoint *ep, grpc_pollset *pollset) { ep->vtable->add_to_pollset(ep, pollset); } -void grpc_endpoint_add_to_pollset_set(grpc_endpoint *ep, grpc_pollset_set *pollset_set) { +void grpc_endpoint_add_to_pollset_set(grpc_endpoint *ep, + grpc_pollset_set *pollset_set) { ep->vtable->add_to_pollset_set(ep, pollset_set); } diff --git a/src/core/iomgr/endpoint.h b/src/core/iomgr/endpoint.h index a2216925f9..ea92a500e8 100644 --- a/src/core/iomgr/endpoint.h +++ b/src/core/iomgr/endpoint.h @@ -103,10 +103,11 @@ void grpc_endpoint_destroy(grpc_endpoint *ep); /* Add an endpoint to a pollset, so that when the pollset is polled, events from this endpoint are considered */ void grpc_endpoint_add_to_pollset(grpc_endpoint *ep, grpc_pollset *pollset); -void grpc_endpoint_add_to_pollset_set(grpc_endpoint *ep, grpc_pollset_set *pollset_set); +void grpc_endpoint_add_to_pollset_set(grpc_endpoint *ep, + grpc_pollset_set *pollset_set); struct grpc_endpoint { const grpc_endpoint_vtable *vtable; }; -#endif /* GRPC_INTERNAL_CORE_IOMGR_ENDPOINT_H */ +#endif /* GRPC_INTERNAL_CORE_IOMGR_ENDPOINT_H */ diff --git a/src/core/iomgr/endpoint_pair.h b/src/core/iomgr/endpoint_pair.h index 25087be0c7..095ec5fcc9 100644 --- a/src/core/iomgr/endpoint_pair.h +++ b/src/core/iomgr/endpoint_pair.h @@ -44,4 +44,4 @@ typedef struct { grpc_endpoint_pair grpc_iomgr_create_endpoint_pair(const char *name, size_t read_slice_size); -#endif /* GRPC_INTERNAL_CORE_IOMGR_ENDPOINT_PAIR_H */ +#endif /* GRPC_INTERNAL_CORE_IOMGR_ENDPOINT_PAIR_H */ diff --git a/src/core/iomgr/endpoint_pair_windows.c b/src/core/iomgr/endpoint_pair_windows.c index e8295df8b3..db9d092dca 100644 --- a/src/core/iomgr/endpoint_pair_windows.c +++ b/src/core/iomgr/endpoint_pair_windows.c @@ -52,21 +52,26 @@ static void create_sockets(SOCKET sv[2]) { SOCKADDR_IN addr; int addr_len = sizeof(addr); - lst_sock = WSASocket(AF_INET, SOCK_STREAM, IPPROTO_TCP, NULL, 0, WSA_FLAG_OVERLAPPED); + lst_sock = WSASocket(AF_INET, SOCK_STREAM, IPPROTO_TCP, NULL, 0, + WSA_FLAG_OVERLAPPED); GPR_ASSERT(lst_sock != INVALID_SOCKET); memset(&addr, 0, sizeof(addr)); addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK); addr.sin_family = AF_INET; - GPR_ASSERT(bind(lst_sock, (struct sockaddr*)&addr, sizeof(addr)) != SOCKET_ERROR); + GPR_ASSERT(bind(lst_sock, (struct sockaddr *)&addr, sizeof(addr)) != + SOCKET_ERROR); GPR_ASSERT(listen(lst_sock, SOMAXCONN) != SOCKET_ERROR); - GPR_ASSERT(getsockname(lst_sock, (struct sockaddr*)&addr, &addr_len) != SOCKET_ERROR); + GPR_ASSERT(getsockname(lst_sock, (struct sockaddr *)&addr, &addr_len) != + SOCKET_ERROR); - cli_sock = WSASocket(AF_INET, SOCK_STREAM, IPPROTO_TCP, NULL, 0, WSA_FLAG_OVERLAPPED); + cli_sock = WSASocket(AF_INET, SOCK_STREAM, IPPROTO_TCP, NULL, 0, + WSA_FLAG_OVERLAPPED); GPR_ASSERT(cli_sock != INVALID_SOCKET); - GPR_ASSERT(WSAConnect(cli_sock, (struct sockaddr*)&addr, addr_len, NULL, NULL, NULL, NULL) == 0); - svr_sock = accept(lst_sock, (struct sockaddr*)&addr, &addr_len); + GPR_ASSERT(WSAConnect(cli_sock, (struct sockaddr *)&addr, addr_len, NULL, + NULL, NULL, NULL) == 0); + svr_sock = accept(lst_sock, (struct sockaddr *)&addr, &addr_len); GPR_ASSERT(svr_sock != INVALID_SOCKET); closesocket(lst_sock); @@ -77,7 +82,8 @@ static void create_sockets(SOCKET sv[2]) { sv[0] = svr_sock; } -grpc_endpoint_pair grpc_iomgr_create_endpoint_pair(const char *name, size_t read_slice_size) { +grpc_endpoint_pair grpc_iomgr_create_endpoint_pair(const char *name, + size_t read_slice_size) { SOCKET sv[2]; grpc_endpoint_pair p; create_sockets(sv); diff --git a/src/core/iomgr/iocp_windows.c b/src/core/iomgr/iocp_windows.c index 8741241fb8..09a457dd9a 100644 --- a/src/core/iomgr/iocp_windows.c +++ b/src/core/iomgr/iocp_windows.c @@ -65,18 +65,17 @@ static void do_iocp_work() { LPOVERLAPPED overlapped; grpc_winsocket *socket; grpc_winsocket_callback_info *info; - void(*f)(void *, int) = NULL; + void (*f)(void *, int) = NULL; void *opaque = NULL; - success = GetQueuedCompletionStatus(g_iocp, &bytes, - &completion_key, &overlapped, - INFINITE); + success = GetQueuedCompletionStatus(g_iocp, &bytes, &completion_key, + &overlapped, INFINITE); /* success = 0 and overlapped = NULL means the deadline got attained. Which is impossible. since our wait time is +inf */ GPR_ASSERT(success || overlapped); GPR_ASSERT(completion_key && overlapped); if (overlapped == &g_iocp_custom_overlap) { gpr_atm_full_fetch_add(&g_custom_events, -1); - if (completion_key == (ULONG_PTR) &g_iocp_kick_token) { + if (completion_key == (ULONG_PTR)&g_iocp_kick_token) { /* We were awoken from a kick. */ return; } @@ -84,7 +83,7 @@ static void do_iocp_work() { abort(); } - socket = (grpc_winsocket*) completion_key; + socket = (grpc_winsocket *)completion_key; if (overlapped == &socket->write_info.overlapped) { info = &socket->write_info; } else if (overlapped == &socket->read_info.overlapped) { @@ -121,8 +120,7 @@ static void do_iocp_work() { } static void iocp_loop(void *p) { - while (gpr_atm_acq_load(&g_orphans) || - gpr_atm_acq_load(&g_custom_events) || + while (gpr_atm_acq_load(&g_orphans) || gpr_atm_acq_load(&g_custom_events) || !gpr_event_get(&g_shutdown_iocp)) { grpc_maybe_call_delayed_callbacks(NULL, 1); do_iocp_work(); @@ -134,8 +132,8 @@ static void iocp_loop(void *p) { void grpc_iocp_init(void) { gpr_thd_id id; - g_iocp = CreateIoCompletionPort(INVALID_HANDLE_VALUE, - NULL, (ULONG_PTR)NULL, 0); + g_iocp = + CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, (ULONG_PTR)NULL, 0); GPR_ASSERT(g_iocp); gpr_event_init(&g_iocp_done); @@ -147,8 +145,7 @@ void grpc_iocp_kick(void) { BOOL success; gpr_atm_full_fetch_add(&g_custom_events, 1); - success = PostQueuedCompletionStatus(g_iocp, 0, - (ULONG_PTR) &g_iocp_kick_token, + success = PostQueuedCompletionStatus(g_iocp, 0, (ULONG_PTR)&g_iocp_kick_token, &g_iocp_custom_overlap); GPR_ASSERT(success); } @@ -165,8 +162,8 @@ void grpc_iocp_shutdown(void) { void grpc_iocp_add_socket(grpc_winsocket *socket) { HANDLE ret; if (socket->added_to_iocp) return; - ret = CreateIoCompletionPort((HANDLE)socket->socket, - g_iocp, (gpr_uintptr) socket, 0); + ret = CreateIoCompletionPort((HANDLE)socket->socket, g_iocp, + (gpr_uintptr)socket, 0); if (!ret) { char *utf8_message = gpr_format_message(WSAGetLastError()); gpr_log(GPR_ERROR, "Unable to add socket to iocp: %s", utf8_message); @@ -189,7 +186,7 @@ void grpc_iocp_socket_orphan(grpc_winsocket *socket) { the callback now. -) The IOCP hasn't completed yet, and we're queuing it for later. */ static void socket_notify_on_iocp(grpc_winsocket *socket, - void(*cb)(void *, int), void *opaque, + void (*cb)(void *, int), void *opaque, grpc_winsocket_callback_info *info) { int run_now = 0; GPR_ASSERT(!info->cb); @@ -206,13 +203,13 @@ static void socket_notify_on_iocp(grpc_winsocket *socket, } void grpc_socket_notify_on_write(grpc_winsocket *socket, - void(*cb)(void *, int), void *opaque) { + void (*cb)(void *, int), void *opaque) { socket_notify_on_iocp(socket, cb, opaque, &socket->write_info); } -void grpc_socket_notify_on_read(grpc_winsocket *socket, - void(*cb)(void *, int), void *opaque) { +void grpc_socket_notify_on_read(grpc_winsocket *socket, void (*cb)(void *, int), + void *opaque) { socket_notify_on_iocp(socket, cb, opaque, &socket->read_info); } -#endif /* GPR_WINSOCK_SOCKET */ +#endif /* GPR_WINSOCK_SOCKET */ diff --git a/src/core/iomgr/iocp_windows.h b/src/core/iomgr/iocp_windows.h index 9df6476917..ee3847a229 100644 --- a/src/core/iomgr/iocp_windows.h +++ b/src/core/iomgr/iocp_windows.h @@ -44,10 +44,10 @@ void grpc_iocp_shutdown(void); void grpc_iocp_add_socket(grpc_winsocket *); void grpc_iocp_socket_orphan(grpc_winsocket *); -void grpc_socket_notify_on_write(grpc_winsocket *, void(*cb)(void *, int success), - void *opaque); +void grpc_socket_notify_on_write(grpc_winsocket *, + void (*cb)(void *, int success), void *opaque); -void grpc_socket_notify_on_read(grpc_winsocket *, void(*cb)(void *, int success), - void *opaque); +void grpc_socket_notify_on_read(grpc_winsocket *, + void (*cb)(void *, int success), void *opaque); -#endif /* GRPC_INTERNAL_CORE_IOMGR_IOCP_WINDOWS_H */ +#endif /* GRPC_INTERNAL_CORE_IOMGR_IOCP_WINDOWS_H */ diff --git a/src/core/iomgr/iomgr.h b/src/core/iomgr/iomgr.h index 6d4a82917b..261c17366a 100644 --- a/src/core/iomgr/iomgr.h +++ b/src/core/iomgr/iomgr.h @@ -77,4 +77,4 @@ void grpc_iomgr_add_callback(grpc_iomgr_closure *closure); argument. */ void grpc_iomgr_add_delayed_callback(grpc_iomgr_closure *iocb, int success); -#endif /* GRPC_INTERNAL_CORE_IOMGR_IOMGR_H */ +#endif /* GRPC_INTERNAL_CORE_IOMGR_IOMGR_H */ diff --git a/src/core/iomgr/iomgr_internal.h b/src/core/iomgr/iomgr_internal.h index 6c1e0e1799..4cec973ba0 100644 --- a/src/core/iomgr/iomgr_internal.h +++ b/src/core/iomgr/iomgr_internal.h @@ -52,4 +52,4 @@ void grpc_iomgr_unregister_object(grpc_iomgr_object *obj); void grpc_iomgr_platform_init(void); void grpc_iomgr_platform_shutdown(void); -#endif /* GRPC_INTERNAL_CORE_IOMGR_IOMGR_INTERNAL_H */ +#endif /* GRPC_INTERNAL_CORE_IOMGR_IOMGR_INTERNAL_H */ diff --git a/src/core/iomgr/iomgr_posix.c b/src/core/iomgr/iomgr_posix.c index 758ae77b86..2425e59941 100644 --- a/src/core/iomgr/iomgr_posix.c +++ b/src/core/iomgr/iomgr_posix.c @@ -51,4 +51,4 @@ void grpc_iomgr_platform_shutdown(void) { grpc_fd_global_shutdown(); } -#endif /* GRPC_POSIX_SOCKET */ +#endif /* GRPC_POSIX_SOCKET */ diff --git a/src/core/iomgr/iomgr_posix.h b/src/core/iomgr/iomgr_posix.h index a404f6433e..716fedb636 100644 --- a/src/core/iomgr/iomgr_posix.h +++ b/src/core/iomgr/iomgr_posix.h @@ -39,4 +39,4 @@ void grpc_pollset_global_init(void); void grpc_pollset_global_shutdown(void); -#endif /* GRPC_INTERNAL_CORE_IOMGR_IOMGR_POSIX_H */ +#endif /* GRPC_INTERNAL_CORE_IOMGR_IOMGR_POSIX_H */ diff --git a/src/core/iomgr/iomgr_windows.c b/src/core/iomgr/iomgr_windows.c index 74cd5a829b..b49cb87e97 100644 --- a/src/core/iomgr/iomgr_windows.c +++ b/src/core/iomgr/iomgr_windows.c @@ -68,4 +68,4 @@ void grpc_iomgr_platform_shutdown(void) { winsock_shutdown(); } -#endif /* GRPC_WINSOCK_SOCKET */ +#endif /* GRPC_WINSOCK_SOCKET */ diff --git a/src/core/iomgr/pollset_multipoller_with_epoll.c b/src/core/iomgr/pollset_multipoller_with_epoll.c index 1320c64579..5ea9dd2101 100644 --- a/src/core/iomgr/pollset_multipoller_with_epoll.c +++ b/src/core/iomgr/pollset_multipoller_with_epoll.c @@ -234,8 +234,7 @@ static void multipoll_with_epoll_pollset_destroy(grpc_pollset *pollset) { } static const grpc_pollset_vtable multipoll_with_epoll_pollset = { - multipoll_with_epoll_pollset_add_fd, - multipoll_with_epoll_pollset_del_fd, + multipoll_with_epoll_pollset_add_fd, multipoll_with_epoll_pollset_del_fd, multipoll_with_epoll_pollset_maybe_work, multipoll_with_epoll_pollset_finish_shutdown, multipoll_with_epoll_pollset_destroy}; diff --git a/src/core/iomgr/pollset_multipoller_with_poll_posix.c b/src/core/iomgr/pollset_multipoller_with_poll_posix.c index b5b2d7534d..001fcecf76 100644 --- a/src/core/iomgr/pollset_multipoller_with_poll_posix.c +++ b/src/core/iomgr/pollset_multipoller_with_poll_posix.c @@ -74,7 +74,7 @@ static void multipoll_with_poll_pollset_add_fd(grpc_pollset *pollset, } h->fds[h->fd_count++] = fd; GRPC_FD_REF(fd, "multipoller"); -exit: +exit: if (and_unlock_pollset) { gpr_mu_unlock(&pollset->mu); } @@ -202,8 +202,7 @@ static void multipoll_with_poll_pollset_destroy(grpc_pollset *pollset) { } static const grpc_pollset_vtable multipoll_with_poll_pollset = { - multipoll_with_poll_pollset_add_fd, - multipoll_with_poll_pollset_del_fd, + multipoll_with_poll_pollset_add_fd, multipoll_with_poll_pollset_del_fd, multipoll_with_poll_pollset_maybe_work, multipoll_with_poll_pollset_finish_shutdown, multipoll_with_poll_pollset_destroy}; diff --git a/src/core/iomgr/pollset_posix.c b/src/core/iomgr/pollset_posix.c index d3a9193af1..a01f9ff727 100644 --- a/src/core/iomgr/pollset_posix.c +++ b/src/core/iomgr/pollset_posix.c @@ -140,10 +140,10 @@ void grpc_pollset_init(grpc_pollset *pollset) { void grpc_pollset_add_fd(grpc_pollset *pollset, grpc_fd *fd) { gpr_mu_lock(&pollset->mu); pollset->vtable->add_fd(pollset, fd, 1); - /* the following (enabled only in debug) will reacquire and then release - our lock - meaning that if the unlocking flag passed to del_fd above is - not respected, the code will deadlock (in a way that we have a chance of - debugging) */ +/* the following (enabled only in debug) will reacquire and then release + our lock - meaning that if the unlocking flag passed to del_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); @@ -153,10 +153,10 @@ void grpc_pollset_add_fd(grpc_pollset *pollset, grpc_fd *fd) { void grpc_pollset_del_fd(grpc_pollset *pollset, grpc_fd *fd) { gpr_mu_lock(&pollset->mu); pollset->vtable->del_fd(pollset, fd, 1); - /* the following (enabled only in debug) will reacquire and then release - our lock - meaning that if the unlocking flag passed to del_fd above is - not respected, the code will deadlock (in a way that we have a chance of - debugging) */ +/* the following (enabled only in debug) will reacquire and then release + our lock - meaning that if the unlocking flag passed to del_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); diff --git a/src/core/iomgr/pollset_posix.h b/src/core/iomgr/pollset_posix.h index 1c1b736193..a3ea353de6 100644 --- a/src/core/iomgr/pollset_posix.h +++ b/src/core/iomgr/pollset_posix.h @@ -102,7 +102,8 @@ void grpc_kick_drain(grpc_pollset *p); - longer than a millisecond polls are rounded up to the next nearest millisecond to avoid spinning - infinite timeouts are converted to -1 */ -int grpc_poll_deadline_to_millis_timeout(gpr_timespec deadline, gpr_timespec now); +int grpc_poll_deadline_to_millis_timeout(gpr_timespec deadline, + gpr_timespec now); /* turn a pollset into a multipoller: platform specific */ typedef void (*grpc_platform_become_multipoller_type)(grpc_pollset *pollset, diff --git a/src/core/iomgr/pollset_windows.c b/src/core/iomgr/pollset_windows.c index 22dc5891c3..8710395ab3 100644 --- a/src/core/iomgr/pollset_windows.c +++ b/src/core/iomgr/pollset_windows.c @@ -56,8 +56,7 @@ static grpc_pollset_worker *pop_front_worker(grpc_pollset *p) { grpc_pollset_worker *w = p->root_worker.next; remove_worker(p, w); return w; - } - else { + } else { return NULL; } } @@ -100,7 +99,8 @@ void grpc_pollset_destroy(grpc_pollset *pollset) { gpr_mu_destroy(&pollset->mu); } -int grpc_pollset_work(grpc_pollset *pollset, grpc_pollset_worker *worker, gpr_timespec deadline) { +int grpc_pollset_work(grpc_pollset *pollset, grpc_pollset_worker *worker, + gpr_timespec deadline) { gpr_timespec now; int added_worker = 0; now = gpr_now(GPR_CLOCK_MONOTONIC); @@ -134,8 +134,8 @@ 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.next; - specific_worker != &p->root_worker; - specific_worker = specific_worker->next) { + specific_worker != &p->root_worker; + specific_worker = specific_worker->next) { gpr_cv_signal(&specific_worker->cv); } p->kicked_without_pollers = 1; diff --git a/src/core/iomgr/resolve_address.h b/src/core/iomgr/resolve_address.h index 8f1d7a22bb..cc1bd428b0 100644 --- a/src/core/iomgr/resolve_address.h +++ b/src/core/iomgr/resolve_address.h @@ -66,4 +66,4 @@ void grpc_resolved_addresses_destroy(grpc_resolved_addresses *addresses); grpc_resolved_addresses *grpc_blocking_resolve_address( const char *addr, const char *default_port); -#endif /* GRPC_INTERNAL_CORE_IOMGR_RESOLVE_ADDRESS_H */ +#endif /* GRPC_INTERNAL_CORE_IOMGR_RESOLVE_ADDRESS_H */ diff --git a/src/core/iomgr/resolve_address_posix.c b/src/core/iomgr/resolve_address_posix.c index dbf884c769..ce6972b797 100644 --- a/src/core/iomgr/resolve_address_posix.c +++ b/src/core/iomgr/resolve_address_posix.c @@ -105,10 +105,7 @@ grpc_resolved_addresses *grpc_blocking_resolve_address( s = getaddrinfo(host, port, &hints, &result); if (s != 0) { /* Retry if well-known service name is recognized */ - char *svc[][2] = { - {"http", "80"}, - {"https", "443"} - }; + char *svc[][2] = {{"http", "80"}, {"https", "443"}}; int i; for (i = 0; i < (int)(sizeof(svc) / sizeof(svc[0])); i++) { if (strcmp(port, svc[i][0]) == 0) { diff --git a/src/core/iomgr/sockaddr.h b/src/core/iomgr/sockaddr.h index 7528db73b8..e41e1ec6b4 100644 --- a/src/core/iomgr/sockaddr.h +++ b/src/core/iomgr/sockaddr.h @@ -44,4 +44,4 @@ #include "src/core/iomgr/sockaddr_posix.h" #endif -#endif /* GRPC_INTERNAL_CORE_IOMGR_SOCKADDR_H */ +#endif /* GRPC_INTERNAL_CORE_IOMGR_SOCKADDR_H */ diff --git a/src/core/iomgr/sockaddr_posix.h b/src/core/iomgr/sockaddr_posix.h index 2a3d932f70..388abb3306 100644 --- a/src/core/iomgr/sockaddr_posix.h +++ b/src/core/iomgr/sockaddr_posix.h @@ -41,4 +41,4 @@ #include <netdb.h> #include <unistd.h> -#endif /* GRPC_INTERNAL_CORE_IOMGR_SOCKADDR_POSIX_H */ +#endif /* GRPC_INTERNAL_CORE_IOMGR_SOCKADDR_POSIX_H */ diff --git a/src/core/iomgr/sockaddr_utils.c b/src/core/iomgr/sockaddr_utils.c index 65ec1f94ac..efdc480365 100644 --- a/src/core/iomgr/sockaddr_utils.c +++ b/src/core/iomgr/sockaddr_utils.c @@ -206,7 +206,8 @@ int grpc_sockaddr_get_port(const struct sockaddr *addr) { case AF_UNIX: return 1; default: - gpr_log(GPR_ERROR, "Unknown socket family %d in grpc_sockaddr_get_port", addr->sa_family); + gpr_log(GPR_ERROR, "Unknown socket family %d in grpc_sockaddr_get_port", + addr->sa_family); return 0; } } @@ -220,7 +221,8 @@ int grpc_sockaddr_set_port(const struct sockaddr *addr, int port) { ((struct sockaddr_in6 *)addr)->sin6_port = htons(port); return 1; default: - gpr_log(GPR_ERROR, "Unknown socket family %d in grpc_sockaddr_set_port", addr->sa_family); + gpr_log(GPR_ERROR, "Unknown socket family %d in grpc_sockaddr_set_port", + addr->sa_family); return 0; } } diff --git a/src/core/iomgr/sockaddr_utils.h b/src/core/iomgr/sockaddr_utils.h index 99f1ed54da..6f7a279900 100644 --- a/src/core/iomgr/sockaddr_utils.h +++ b/src/core/iomgr/sockaddr_utils.h @@ -86,4 +86,4 @@ int grpc_sockaddr_to_string(char **out, const struct sockaddr *addr, char *grpc_sockaddr_to_uri(const struct sockaddr *addr); -#endif /* GRPC_INTERNAL_CORE_IOMGR_SOCKADDR_UTILS_H */ +#endif /* GRPC_INTERNAL_CORE_IOMGR_SOCKADDR_UTILS_H */ diff --git a/src/core/iomgr/sockaddr_win32.h b/src/core/iomgr/sockaddr_win32.h index be55db805a..fe2be99145 100644 --- a/src/core/iomgr/sockaddr_win32.h +++ b/src/core/iomgr/sockaddr_win32.h @@ -43,4 +43,4 @@ const char *inet_ntop(int af, const void *src, char *dst, socklen_t size); #endif -#endif /* GRPC_INTERNAL_CORE_IOMGR_SOCKADDR_WIN32_H */ +#endif /* GRPC_INTERNAL_CORE_IOMGR_SOCKADDR_WIN32_H */ diff --git a/src/core/iomgr/socket_utils_posix.h b/src/core/iomgr/socket_utils_posix.h index d2a315b462..d330d1986e 100644 --- a/src/core/iomgr/socket_utils_posix.h +++ b/src/core/iomgr/socket_utils_posix.h @@ -110,4 +110,4 @@ extern int grpc_forbid_dualstack_sockets_for_testing; int grpc_create_dualstack_socket(const struct sockaddr *addr, int type, int protocol, grpc_dualstack_mode *dsmode); -#endif /* GRPC_INTERNAL_CORE_IOMGR_SOCKET_UTILS_POSIX_H */ +#endif /* GRPC_INTERNAL_CORE_IOMGR_SOCKET_UTILS_POSIX_H */ diff --git a/src/core/iomgr/socket_windows.c b/src/core/iomgr/socket_windows.c index f6ddfff0ad..7d8421376b 100644 --- a/src/core/iomgr/socket_windows.c +++ b/src/core/iomgr/socket_windows.c @@ -106,4 +106,4 @@ void grpc_winsocket_destroy(grpc_winsocket *winsocket) { gpr_free(winsocket); } -#endif /* GPR_WINSOCK_SOCKET */ +#endif /* GPR_WINSOCK_SOCKET */ diff --git a/src/core/iomgr/socket_windows.h b/src/core/iomgr/socket_windows.h index 346fde8edd..ecf2530173 100644 --- a/src/core/iomgr/socket_windows.h +++ b/src/core/iomgr/socket_windows.h @@ -54,7 +54,7 @@ typedef struct grpc_winsocket_callback_info { OVERLAPPED overlapped; /* The callback information for the pending operation. May be empty if the caller hasn't registered a callback yet. */ - void(*cb)(void *opaque, int success); + void (*cb)(void *opaque, int success); void *opaque; /* A boolean to describe if the IO Completion Port got a notification for that operation. This will happen if the operation completed before the @@ -118,4 +118,4 @@ void grpc_winsocket_orphan(grpc_winsocket *socket); or by grpc_winsocket_orphan if there's no pending operation. */ void grpc_winsocket_destroy(grpc_winsocket *socket); -#endif /* GRPC_INTERNAL_CORE_IOMGR_SOCKET_WINDOWS_H */ +#endif /* GRPC_INTERNAL_CORE_IOMGR_SOCKET_WINDOWS_H */ diff --git a/src/core/iomgr/tcp_client.h b/src/core/iomgr/tcp_client.h index 0fa08b52b0..8ad9b818e1 100644 --- a/src/core/iomgr/tcp_client.h +++ b/src/core/iomgr/tcp_client.h @@ -41,7 +41,7 @@ /* Asynchronously connect to an address (specified as (addr, len)), and call cb with arg and the completed connection when done (or call cb with arg and - NULL on failure). + NULL on failure). interested_parties points to a set of pollsets that would be interested in this connection being established (in order to continue their work) */ void grpc_tcp_client_connect(void (*cb)(void *arg, grpc_endpoint *tcp), diff --git a/src/core/iomgr/tcp_client_posix.c b/src/core/iomgr/tcp_client_posix.c index 9572ce5980..66027f87a0 100644 --- a/src/core/iomgr/tcp_client_posix.c +++ b/src/core/iomgr/tcp_client_posix.c @@ -264,7 +264,8 @@ void grpc_tcp_client_connect(void (*cb)(void *arg, grpc_endpoint *ep), ac->write_closure.cb_arg = ac; gpr_mu_lock(&ac->mu); - grpc_alarm_init(&ac->alarm, gpr_convert_clock_type(deadline, GPR_CLOCK_MONOTONIC), + grpc_alarm_init(&ac->alarm, + gpr_convert_clock_type(deadline, GPR_CLOCK_MONOTONIC), tc_on_alarm, ac, gpr_now(GPR_CLOCK_MONOTONIC)); grpc_fd_notify_on_write(ac->fd, &ac->write_closure); gpr_mu_unlock(&ac->mu); diff --git a/src/core/iomgr/tcp_posix.c b/src/core/iomgr/tcp_posix.c index 24fee0596f..360e6ebd8c 100644 --- a/src/core/iomgr/tcp_posix.c +++ b/src/core/iomgr/tcp_posix.c @@ -572,7 +572,8 @@ static void grpc_tcp_add_to_pollset(grpc_endpoint *ep, grpc_pollset *pollset) { grpc_pollset_add_fd(pollset, tcp->em_fd); } -static void grpc_tcp_add_to_pollset_set(grpc_endpoint *ep, grpc_pollset_set *pollset_set) { +static void grpc_tcp_add_to_pollset_set(grpc_endpoint *ep, + grpc_pollset_set *pollset_set) { grpc_tcp *tcp = (grpc_tcp *)ep; grpc_pollset_set_add_fd(pollset_set, tcp->em_fd); } diff --git a/src/core/iomgr/tcp_posix.h b/src/core/iomgr/tcp_posix.h index d752feaeea..40b3ae2679 100644 --- a/src/core/iomgr/tcp_posix.h +++ b/src/core/iomgr/tcp_posix.h @@ -56,4 +56,4 @@ extern int grpc_tcp_trace; grpc_endpoint *grpc_tcp_create(grpc_fd *fd, size_t read_slice_size, const char *peer_string); -#endif /* GRPC_INTERNAL_CORE_IOMGR_TCP_POSIX_H */ +#endif /* GRPC_INTERNAL_CORE_IOMGR_TCP_POSIX_H */ diff --git a/src/core/iomgr/tcp_server_windows.c b/src/core/iomgr/tcp_server_windows.c index 0adbe9507c..d0478d3604 100644 --- a/src/core/iomgr/tcp_server_windows.c +++ b/src/core/iomgr/tcp_server_windows.c @@ -79,7 +79,8 @@ struct grpc_tcp_server { /* active port count: how many ports are actually still listening */ int active_ports; - /* number of iomgr callbacks that have been explicitly scheduled during shutdown */ + /* number of iomgr callbacks that have been explicitly scheduled during + * shutdown */ int iomgr_callbacks_pending; /* all listening ports */ @@ -292,7 +293,7 @@ static void on_accept(void *arg, int from_iocp) { and act accordingly. */ transfered_bytes = 0; wsa_success = WSAGetOverlappedResult(sock, &info->overlapped, - &transfered_bytes, FALSE, &flags); + &transfered_bytes, FALSE, &flags); if (!wsa_success) { if (sp->shutting_down) { /* During the shutdown case, we ARE expecting an error. So that's well, @@ -309,16 +310,15 @@ static void on_accept(void *arg, int from_iocp) { if (!sp->shutting_down) { peer_name_string = NULL; err = setsockopt(sock, SOL_SOCKET, SO_UPDATE_ACCEPT_CONTEXT, - (char *)&sp->socket->socket, - sizeof(sp->socket->socket)); + (char *)&sp->socket->socket, sizeof(sp->socket->socket)); if (err) { char *utf8_message = gpr_format_message(WSAGetLastError()); gpr_log(GPR_ERROR, "setsockopt error: %s", utf8_message); gpr_free(utf8_message); } - err = getpeername(sock, (struct sockaddr*)&peer_name, &peer_name_len); + err = getpeername(sock, (struct sockaddr *)&peer_name, &peer_name_len); if (!err) { - peer_name_string = grpc_sockaddr_to_uri((struct sockaddr*)&peer_name); + peer_name_string = grpc_sockaddr_to_uri((struct sockaddr *)&peer_name); } else { char *utf8_message = gpr_format_message(WSAGetLastError()); gpr_log(GPR_ERROR, "getpeername error: %s", utf8_message); diff --git a/src/core/iomgr/tcp_windows.c b/src/core/iomgr/tcp_windows.c index 89aa741470..123f46d71d 100644 --- a/src/core/iomgr/tcp_windows.c +++ b/src/core/iomgr/tcp_windows.c @@ -55,24 +55,22 @@ static int set_non_block(SOCKET sock) { int status; unsigned long param = 1; DWORD ret; - status = WSAIoctl(sock, FIONBIO, ¶m, sizeof(param), NULL, 0, &ret, - NULL, NULL); + status = + WSAIoctl(sock, FIONBIO, ¶m, sizeof(param), NULL, 0, &ret, NULL, NULL); return status == 0; } static int set_dualstack(SOCKET sock) { int status; unsigned long param = 0; - status = setsockopt(sock, IPPROTO_IPV6, IPV6_V6ONLY, - (const char *) ¶m, sizeof(param)); + status = setsockopt(sock, IPPROTO_IPV6, IPV6_V6ONLY, (const char *)¶m, + sizeof(param)); return status == 0; } int grpc_tcp_prepare_socket(SOCKET sock) { - if (!set_non_block(sock)) - return 0; - if (!set_dualstack(sock)) - return 0; + if (!set_non_block(sock)) return 0; + if (!set_dualstack(sock)) return 0; return 1; } @@ -100,9 +98,7 @@ typedef struct grpc_tcp { char *peer_string; } grpc_tcp; -static void tcp_ref(grpc_tcp *tcp) { - gpr_ref(&tcp->refcount); -} +static void tcp_ref(grpc_tcp *tcp) { gpr_ref(&tcp->refcount); } static void tcp_unref(grpc_tcp *tcp) { if (gpr_unref(&tcp->refcount)) { @@ -116,7 +112,7 @@ static void tcp_unref(grpc_tcp *tcp) { /* Asynchronous callback from the IOCP, or the background thread. */ static void on_read(void *tcpp, int from_iocp) { - grpc_tcp *tcp = (grpc_tcp *) tcpp; + grpc_tcp *tcp = (grpc_tcp *)tcpp; grpc_winsocket *socket = tcp->socket; gpr_slice sub; gpr_slice *slice = NULL; @@ -175,9 +171,9 @@ static void on_read(void *tcpp, int from_iocp) { cb(opaque, slice, nslices, status); } -static void win_notify_on_read(grpc_endpoint *ep, - grpc_endpoint_read_cb cb, void *arg) { - grpc_tcp *tcp = (grpc_tcp *) ep; +static void win_notify_on_read(grpc_endpoint *ep, grpc_endpoint_read_cb cb, + void *arg) { + grpc_tcp *tcp = (grpc_tcp *)ep; grpc_winsocket *handle = tcp->socket; grpc_winsocket_callback_info *info = &handle->read_info; int status; @@ -201,8 +197,8 @@ static void win_notify_on_read(grpc_endpoint *ep, buffer.buf = (char *)GPR_SLICE_START_PTR(tcp->read_slice); /* First let's try a synchronous, non-blocking read. */ - status = WSARecv(tcp->socket->socket, &buffer, 1, &bytes_read, &flags, - NULL, NULL); + status = + WSARecv(tcp->socket->socket, &buffer, 1, &bytes_read, &flags, NULL, NULL); info->wsa_error = status == 0 ? 0 : WSAGetLastError(); /* Did we get data immediately ? Yay. */ @@ -232,7 +228,7 @@ static void win_notify_on_read(grpc_endpoint *ep, /* Asynchronous callback from the IOCP, or the background thread. */ static void on_write(void *tcpp, int from_iocp) { - grpc_tcp *tcp = (grpc_tcp *) tcpp; + grpc_tcp *tcp = (grpc_tcp *)tcpp; grpc_winsocket *handle = tcp->socket; grpc_winsocket_callback_info *info = &handle->write_info; grpc_endpoint_cb_status status = GRPC_ENDPOINT_CB_OK; @@ -286,7 +282,7 @@ static grpc_endpoint_write_status win_write(grpc_endpoint *ep, gpr_slice *slices, size_t nslices, grpc_endpoint_write_cb cb, void *arg) { - grpc_tcp *tcp = (grpc_tcp *) ep; + grpc_tcp *tcp = (grpc_tcp *)ep; grpc_winsocket *socket = tcp->socket; grpc_winsocket_callback_info *info = &socket->write_info; unsigned i; @@ -309,7 +305,7 @@ static grpc_endpoint_write_status win_write(grpc_endpoint *ep, gpr_slice_buffer_addn(&tcp->write_slices, slices, nslices); if (tcp->write_slices.count > GPR_ARRAY_SIZE(local_buffers)) { - buffers = (WSABUF *) gpr_malloc(sizeof(WSABUF) * tcp->write_slices.count); + buffers = (WSABUF *)gpr_malloc(sizeof(WSABUF) * tcp->write_slices.count); allocated = buffers; } @@ -370,15 +366,15 @@ static grpc_endpoint_write_status win_write(grpc_endpoint *ep, static void win_add_to_pollset(grpc_endpoint *ep, grpc_pollset *ps) { grpc_tcp *tcp; - (void) ps; - tcp = (grpc_tcp *) ep; + (void)ps; + tcp = (grpc_tcp *)ep; grpc_iocp_add_socket(tcp->socket); } static void win_add_to_pollset_set(grpc_endpoint *ep, grpc_pollset_set *pss) { grpc_tcp *tcp; - (void) pss; - tcp = (grpc_tcp *) ep; + (void)pss; + tcp = (grpc_tcp *)ep; grpc_iocp_add_socket(tcp->socket); } @@ -389,7 +385,7 @@ static void win_add_to_pollset_set(grpc_endpoint *ep, grpc_pollset_set *pss) { callback will happen from another thread, so we need to protect against concurrent access of the data structure in that regard. */ static void win_shutdown(grpc_endpoint *ep) { - grpc_tcp *tcp = (grpc_tcp *) ep; + grpc_tcp *tcp = (grpc_tcp *)ep; int extra_refs = 0; gpr_mu_lock(&tcp->mu); /* At that point, what may happen is that we're already inside the IOCP @@ -401,7 +397,7 @@ static void win_shutdown(grpc_endpoint *ep) { } static void win_destroy(grpc_endpoint *ep) { - grpc_tcp *tcp = (grpc_tcp *) ep; + grpc_tcp *tcp = (grpc_tcp *)ep; tcp_unref(tcp); } @@ -410,13 +406,12 @@ static char *win_get_peer(grpc_endpoint *ep) { return gpr_strdup(tcp->peer_string); } -static grpc_endpoint_vtable vtable = {win_notify_on_read, win_write, - win_add_to_pollset, win_add_to_pollset_set, - win_shutdown, win_destroy, - win_get_peer}; +static grpc_endpoint_vtable vtable = { + win_notify_on_read, win_write, win_add_to_pollset, win_add_to_pollset_set, + win_shutdown, win_destroy, win_get_peer}; grpc_endpoint *grpc_tcp_create(grpc_winsocket *socket, char *peer_string) { - grpc_tcp *tcp = (grpc_tcp *) gpr_malloc(sizeof(grpc_tcp)); + grpc_tcp *tcp = (grpc_tcp *)gpr_malloc(sizeof(grpc_tcp)); memset(tcp, 0, sizeof(grpc_tcp)); tcp->base.vtable = &vtable; tcp->socket = socket; @@ -427,4 +422,4 @@ grpc_endpoint *grpc_tcp_create(grpc_winsocket *socket, char *peer_string) { return &tcp->base; } -#endif /* GPR_WINSOCK_SOCKET */ +#endif /* GPR_WINSOCK_SOCKET */ diff --git a/src/core/iomgr/tcp_windows.h b/src/core/iomgr/tcp_windows.h index 7e301db250..deb3e48293 100644 --- a/src/core/iomgr/tcp_windows.h +++ b/src/core/iomgr/tcp_windows.h @@ -54,4 +54,4 @@ grpc_endpoint *grpc_tcp_create(grpc_winsocket *socket, char *peer_string); int grpc_tcp_prepare_socket(SOCKET sock); -#endif /* GRPC_INTERNAL_CORE_IOMGR_TCP_WINDOWS_H */ +#endif /* GRPC_INTERNAL_CORE_IOMGR_TCP_WINDOWS_H */ diff --git a/src/core/iomgr/time_averaged_stats.h b/src/core/iomgr/time_averaged_stats.h index 13894b2640..e6dec1b4cd 100644 --- a/src/core/iomgr/time_averaged_stats.h +++ b/src/core/iomgr/time_averaged_stats.h @@ -85,4 +85,4 @@ void grpc_time_averaged_stats_add_sample(grpc_time_averaged_stats *stats, value. */ double grpc_time_averaged_stats_update_average(grpc_time_averaged_stats *stats); -#endif /* GRPC_INTERNAL_CORE_IOMGR_TIME_AVERAGED_STATS_H */ +#endif /* GRPC_INTERNAL_CORE_IOMGR_TIME_AVERAGED_STATS_H */ diff --git a/src/core/iomgr/udp_server.c b/src/core/iomgr/udp_server.c new file mode 100644 index 0000000000..16482c08f7 --- /dev/null +++ b/src/core/iomgr/udp_server.c @@ -0,0 +1,438 @@ +/* + * + * Copyright 2015, 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. + * + */ + +/* FIXME: "posix" files shouldn't be depending on _GNU_SOURCE */ +#ifndef _GNU_SOURCE +#define _GNU_SOURCE +#endif + +#include <grpc/support/port_platform.h> + +#ifdef GPR_POSIX_SOCKET + +#include "src/core/iomgr/udp_server.h" + +#include <errno.h> +#include <fcntl.h> +#include <limits.h> +#include <netinet/in.h> +#include <netinet/tcp.h> +#include <string.h> +#include <sys/socket.h> +#include <sys/stat.h> +#include <sys/types.h> +#include <sys/un.h> +#include <unistd.h> + +#include "src/core/iomgr/fd_posix.h" +#include "src/core/iomgr/pollset_posix.h" +#include "src/core/iomgr/resolve_address.h" +#include "src/core/iomgr/sockaddr_utils.h" +#include "src/core/iomgr/socket_utils_posix.h" +#include "src/core/support/string.h" +#include <grpc/support/alloc.h> +#include <grpc/support/log.h> +#include <grpc/support/sync.h> +#include <grpc/support/string_util.h> +#include <grpc/support/time.h> + +#define INIT_PORT_CAP 2 + +/* one listening port */ +typedef struct { + int fd; + grpc_fd *emfd; + grpc_udp_server *server; + union { + gpr_uint8 untyped[GRPC_MAX_SOCKADDR_SIZE]; + struct sockaddr sockaddr; + struct sockaddr_un un; + } addr; + int addr_len; + grpc_iomgr_closure read_closure; + grpc_iomgr_closure destroyed_closure; + grpc_udp_server_read_cb read_cb; +} server_port; + +static void unlink_if_unix_domain_socket(const struct sockaddr_un *un) { + struct stat st; + + if (stat(un->sun_path, &st) == 0 && (st.st_mode & S_IFMT) == S_IFSOCK) { + unlink(un->sun_path); + } +} + +/* the overall server */ +struct grpc_udp_server { + grpc_udp_server_cb cb; + void *cb_arg; + + gpr_mu mu; + gpr_cv cv; + + /* active port count: how many ports are actually still listening */ + size_t active_ports; + /* destroyed port count: how many ports are completely destroyed */ + size_t destroyed_ports; + + /* is this server shutting down? (boolean) */ + int shutdown; + + /* all listening ports */ + server_port *ports; + size_t nports; + size_t port_capacity; + + /* shutdown callback */ + void (*shutdown_complete)(void *); + void *shutdown_complete_arg; + + /* all pollsets interested in new connections */ + grpc_pollset **pollsets; + /* number of pollsets in the pollsets array */ + size_t pollset_count; +}; + +grpc_udp_server *grpc_udp_server_create(void) { + grpc_udp_server *s = gpr_malloc(sizeof(grpc_udp_server)); + gpr_mu_init(&s->mu); + gpr_cv_init(&s->cv); + s->active_ports = 0; + s->destroyed_ports = 0; + s->shutdown = 0; + s->cb = NULL; + s->cb_arg = NULL; + s->ports = gpr_malloc(sizeof(server_port) * INIT_PORT_CAP); + s->nports = 0; + s->port_capacity = INIT_PORT_CAP; + + return s; +} + +static void finish_shutdown(grpc_udp_server *s) { + s->shutdown_complete(s->shutdown_complete_arg); + + gpr_mu_destroy(&s->mu); + gpr_cv_destroy(&s->cv); + + gpr_free(s->ports); + gpr_free(s); +} + +static void destroyed_port(void *server, int success) { + grpc_udp_server *s = server; + gpr_mu_lock(&s->mu); + s->destroyed_ports++; + if (s->destroyed_ports == s->nports) { + gpr_mu_unlock(&s->mu); + finish_shutdown(s); + } else { + gpr_mu_unlock(&s->mu); + } +} + +static void dont_care_about_shutdown_completion(void *ignored) {} + +/* called when all listening endpoints have been shutdown, so no further + events will be received on them - at this point it's safe to destroy + things */ +static void deactivated_all_ports(grpc_udp_server *s) { + size_t i; + + /* delete ALL the things */ + gpr_mu_lock(&s->mu); + + if (!s->shutdown) { + gpr_mu_unlock(&s->mu); + return; + } + + if (s->nports) { + for (i = 0; i < s->nports; i++) { + server_port *sp = &s->ports[i]; + if (sp->addr.sockaddr.sa_family == AF_UNIX) { + unlink_if_unix_domain_socket(&sp->addr.un); + } + sp->destroyed_closure.cb = destroyed_port; + sp->destroyed_closure.cb_arg = s; + grpc_fd_orphan(sp->emfd, &sp->destroyed_closure, "udp_listener_shutdown"); + } + gpr_mu_unlock(&s->mu); + } else { + gpr_mu_unlock(&s->mu); + finish_shutdown(s); + } +} + +void grpc_udp_server_destroy( + grpc_udp_server *s, void (*shutdown_complete)(void *shutdown_complete_arg), + void *shutdown_complete_arg) { + size_t i; + gpr_mu_lock(&s->mu); + + GPR_ASSERT(!s->shutdown); + s->shutdown = 1; + + s->shutdown_complete = shutdown_complete + ? shutdown_complete + : dont_care_about_shutdown_completion; + s->shutdown_complete_arg = shutdown_complete_arg; + + /* shutdown all fd's */ + if (s->active_ports) { + for (i = 0; i < s->nports; i++) { + grpc_fd_shutdown(s->ports[i].emfd); + } + gpr_mu_unlock(&s->mu); + } else { + gpr_mu_unlock(&s->mu); + deactivated_all_ports(s); + } +} + +/* Prepare a recently-created socket for listening. */ +static int prepare_socket(int fd, const struct sockaddr *addr, int addr_len) { + struct sockaddr_storage sockname_temp; + socklen_t sockname_len; + int get_local_ip; + int rc; + + if (fd < 0) { + goto error; + } + + get_local_ip = 1; + rc = setsockopt(fd, IPPROTO_IP, IP_PKTINFO, &get_local_ip, + sizeof(get_local_ip)); + if (rc == 0 && addr->sa_family == AF_INET6) { + rc = setsockopt(fd, IPPROTO_IPV6, IPV6_RECVPKTINFO, &get_local_ip, + sizeof(get_local_ip)); + } + + if (bind(fd, addr, addr_len) < 0) { + char *addr_str; + grpc_sockaddr_to_string(&addr_str, addr, 0); + gpr_log(GPR_ERROR, "bind addr=%s: %s", addr_str, strerror(errno)); + gpr_free(addr_str); + goto error; + } + + sockname_len = sizeof(sockname_temp); + if (getsockname(fd, (struct sockaddr *)&sockname_temp, &sockname_len) < 0) { + goto error; + } + + return grpc_sockaddr_get_port((struct sockaddr *)&sockname_temp); + +error: + if (fd >= 0) { + close(fd); + } + return -1; +} + +/* event manager callback when reads are ready */ +static void on_read(void *arg, int success) { + server_port *sp = arg; + + if (success == 0) { + gpr_mu_lock(&sp->server->mu); + if (0 == --sp->server->active_ports) { + gpr_mu_unlock(&sp->server->mu); + deactivated_all_ports(sp->server); + } else { + gpr_mu_unlock(&sp->server->mu); + } + return; + } + + /* Tell the registered callback that data is available to read. */ + GPR_ASSERT(sp->read_cb); + sp->read_cb(sp->fd, sp->server->cb, sp->server->cb_arg); + + /* Re-arm the notification event so we get another chance to read. */ + grpc_fd_notify_on_read(sp->emfd, &sp->read_closure); +} + +static int add_socket_to_server(grpc_udp_server *s, int fd, + const struct sockaddr *addr, int addr_len, + grpc_udp_server_read_cb read_cb) { + server_port *sp; + int port; + char *addr_str; + char *name; + + port = prepare_socket(fd, addr, addr_len); + if (port >= 0) { + grpc_sockaddr_to_string(&addr_str, (struct sockaddr *)&addr, 1); + gpr_asprintf(&name, "udp-server-listener:%s", addr_str); + gpr_mu_lock(&s->mu); + GPR_ASSERT(!s->cb && "must add ports before starting server"); + /* append it to the list under a lock */ + if (s->nports == s->port_capacity) { + s->port_capacity *= 2; + s->ports = gpr_realloc(s->ports, sizeof(server_port) * s->port_capacity); + } + sp = &s->ports[s->nports++]; + sp->server = s; + sp->fd = fd; + sp->emfd = grpc_fd_create(fd, name); + memcpy(sp->addr.untyped, addr, addr_len); + sp->addr_len = addr_len; + sp->read_cb = read_cb; + GPR_ASSERT(sp->emfd); + gpr_mu_unlock(&s->mu); + } + + return port; +} + +int grpc_udp_server_add_port(grpc_udp_server *s, const void *addr, int addr_len, + grpc_udp_server_read_cb read_cb) { + int allocated_port1 = -1; + int allocated_port2 = -1; + unsigned i; + int fd; + grpc_dualstack_mode dsmode; + struct sockaddr_in6 addr6_v4mapped; + struct sockaddr_in wild4; + struct sockaddr_in6 wild6; + struct sockaddr_in addr4_copy; + struct sockaddr *allocated_addr = NULL; + struct sockaddr_storage sockname_temp; + socklen_t sockname_len; + int port; + + if (((struct sockaddr *)addr)->sa_family == AF_UNIX) { + unlink_if_unix_domain_socket(addr); + } + + /* Check if this is a wildcard port, and if so, try to keep the port the same + as some previously created listener. */ + if (grpc_sockaddr_get_port(addr) == 0) { + for (i = 0; i < s->nports; i++) { + sockname_len = sizeof(sockname_temp); + if (0 == getsockname(s->ports[i].fd, (struct sockaddr *)&sockname_temp, + &sockname_len)) { + port = grpc_sockaddr_get_port((struct sockaddr *)&sockname_temp); + if (port > 0) { + allocated_addr = malloc(addr_len); + memcpy(allocated_addr, addr, addr_len); + grpc_sockaddr_set_port(allocated_addr, port); + addr = allocated_addr; + break; + } + } + } + } + + if (grpc_sockaddr_to_v4mapped(addr, &addr6_v4mapped)) { + addr = (const struct sockaddr *)&addr6_v4mapped; + addr_len = sizeof(addr6_v4mapped); + } + + /* Treat :: or 0.0.0.0 as a family-agnostic wildcard. */ + if (grpc_sockaddr_is_wildcard(addr, &port)) { + grpc_sockaddr_make_wildcards(port, &wild4, &wild6); + + /* Try listening on IPv6 first. */ + addr = (struct sockaddr *)&wild6; + addr_len = sizeof(wild6); + fd = grpc_create_dualstack_socket(addr, SOCK_DGRAM, IPPROTO_UDP, &dsmode); + allocated_port1 = add_socket_to_server(s, fd, addr, addr_len, read_cb); + if (fd >= 0 && dsmode == GRPC_DSMODE_DUALSTACK) { + goto done; + } + + /* If we didn't get a dualstack socket, also listen on 0.0.0.0. */ + if (port == 0 && allocated_port1 > 0) { + grpc_sockaddr_set_port((struct sockaddr *)&wild4, allocated_port1); + } + addr = (struct sockaddr *)&wild4; + addr_len = sizeof(wild4); + } + + fd = grpc_create_dualstack_socket(addr, SOCK_DGRAM, IPPROTO_UDP, &dsmode); + if (fd < 0) { + gpr_log(GPR_ERROR, "Unable to create socket: %s", strerror(errno)); + } + if (dsmode == GRPC_DSMODE_IPV4 && + grpc_sockaddr_is_v4mapped(addr, &addr4_copy)) { + addr = (struct sockaddr *)&addr4_copy; + addr_len = sizeof(addr4_copy); + } + allocated_port2 = add_socket_to_server(s, fd, addr, addr_len, read_cb); + +done: + gpr_free(allocated_addr); + return allocated_port1 >= 0 ? allocated_port1 : allocated_port2; +} + +int grpc_udp_server_get_fd(grpc_udp_server *s, unsigned index) { + return (index < s->nports) ? s->ports[index].fd : -1; +} + +void grpc_udp_server_start(grpc_udp_server *s, grpc_pollset **pollsets, + size_t pollset_count, + grpc_udp_server_cb new_transport_cb, void *cb_arg) { + size_t i, j; + GPR_ASSERT(new_transport_cb); + gpr_mu_lock(&s->mu); + GPR_ASSERT(!s->cb); + GPR_ASSERT(s->active_ports == 0); + s->cb = new_transport_cb; + s->cb_arg = cb_arg; + s->pollsets = pollsets; + for (i = 0; i < s->nports; i++) { + for (j = 0; j < pollset_count; j++) { + grpc_pollset_add_fd(pollsets[j], s->ports[i].emfd); + } + s->ports[i].read_closure.cb = on_read; + s->ports[i].read_closure.cb_arg = &s->ports[i]; + grpc_fd_notify_on_read(s->ports[i].emfd, &s->ports[i].read_closure); + s->active_ports++; + } + gpr_mu_unlock(&s->mu); +} + +/* TODO(rjshade): Add a test for this method. */ +void grpc_udp_server_write(server_port *sp, const char *buffer, size_t buf_len, + const struct sockaddr *peer_address) { + int rc; + rc = sendto(sp->fd, buffer, buf_len, 0, peer_address, sizeof(peer_address)); + if (rc < 0) { + gpr_log(GPR_ERROR, "Unable to send data: %s", strerror(errno)); + } +} + +#endif diff --git a/src/core/iomgr/udp_server.h b/src/core/iomgr/udp_server.h new file mode 100644 index 0000000000..fcc4ba6e97 --- /dev/null +++ b/src/core/iomgr/udp_server.h @@ -0,0 +1,85 @@ +/* + * + * Copyright 2015, 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. + * + */ + +#ifndef GRPC_INTERNAL_CORE_IOMGR_UDP_SERVER_H +#define GRPC_INTERNAL_CORE_IOMGR_UDP_SERVER_H + +#include "src/core/iomgr/endpoint.h" + +/* Forward decl of grpc_udp_server */ +typedef struct grpc_udp_server grpc_udp_server; + +/* New server callback: ep is the newly connected connection */ +typedef void (*grpc_udp_server_cb)(void *arg, grpc_endpoint *ep); + +/* Called when data is available to read from the socket. */ +typedef void (*grpc_udp_server_read_cb)(int fd, + grpc_udp_server_cb new_transport_cb, + void *cb_arg); + +/* Create a server, initially not bound to any ports */ +grpc_udp_server *grpc_udp_server_create(void); + +/* Start listening to bound ports */ +void grpc_udp_server_start(grpc_udp_server *server, grpc_pollset **pollsets, + size_t pollset_count, grpc_udp_server_cb cb, + void *cb_arg); + +int grpc_udp_server_get_fd(grpc_udp_server *s, unsigned index); + +/* Add a port to the server, returning port number on success, or negative + on failure. + + The :: and 0.0.0.0 wildcard addresses are treated identically, accepting + both IPv4 and IPv6 connections, but :: is the preferred style. This usually + creates one socket, but possibly two on systems which support IPv6, + but not dualstack sockets. */ + +/* TODO(ctiller): deprecate this, and make grpc_udp_server_add_ports to handle + all of the multiple socket port matching logic in one place */ +int grpc_udp_server_add_port(grpc_udp_server *s, const void *addr, int addr_len, + grpc_udp_server_read_cb read_cb); + +void grpc_udp_server_destroy(grpc_udp_server *server, + void (*shutdown_done)(void *shutdown_done_arg), + void *shutdown_done_arg); + +/* Write the contents of buffer to the underlying UDP socket. */ +/* +void grpc_udp_server_write(grpc_udp_server *s, + const char *buffer, + int buf_len, + const struct sockaddr* to); + */ + +#endif /* GRPC_INTERNAL_CORE_IOMGR_UDP_SERVER_H */ diff --git a/src/core/iomgr/wakeup_fd_eventfd.c b/src/core/iomgr/wakeup_fd_eventfd.c index 52912235f8..08fdc74f17 100644 --- a/src/core/iomgr/wakeup_fd_eventfd.c +++ b/src/core/iomgr/wakeup_fd_eventfd.c @@ -75,8 +75,7 @@ static int eventfd_check_availability(void) { } const grpc_wakeup_fd_vtable grpc_specialized_wakeup_fd_vtable = { - eventfd_create, eventfd_consume, eventfd_wakeup, eventfd_destroy, - eventfd_check_availability -}; + eventfd_create, eventfd_consume, eventfd_wakeup, eventfd_destroy, + eventfd_check_availability}; #endif /* GPR_LINUX_EVENTFD */ diff --git a/src/core/iomgr/wakeup_fd_nospecial.c b/src/core/iomgr/wakeup_fd_nospecial.c index c1038bf379..78d763c103 100644 --- a/src/core/iomgr/wakeup_fd_nospecial.c +++ b/src/core/iomgr/wakeup_fd_nospecial.c @@ -43,12 +43,9 @@ #include "src/core/iomgr/wakeup_fd_posix.h" #include <stddef.h> -static int check_availability_invalid(void) { - return 0; -} +static int check_availability_invalid(void) { return 0; } const grpc_wakeup_fd_vtable grpc_specialized_wakeup_fd_vtable = { - NULL, NULL, NULL, NULL, check_availability_invalid -}; + NULL, NULL, NULL, NULL, check_availability_invalid}; -#endif /* GPR_POSIX_NO_SPECIAL_WAKEUP_FD */ +#endif /* GPR_POSIX_NO_SPECIAL_WAKEUP_FD */ diff --git a/src/core/iomgr/wakeup_fd_pipe.c b/src/core/iomgr/wakeup_fd_pipe.c index 9fc4ee2388..bd643e8061 100644 --- a/src/core/iomgr/wakeup_fd_pipe.c +++ b/src/core/iomgr/wakeup_fd_pipe.c @@ -94,4 +94,4 @@ const grpc_wakeup_fd_vtable grpc_pipe_wakeup_fd_vtable = { pipe_init, pipe_consume, pipe_wakeup, pipe_destroy, pipe_check_availability}; -#endif /* GPR_POSIX_WAKUP_FD */ +#endif /* GPR_POSIX_WAKUP_FD */ diff --git a/src/core/iomgr/wakeup_fd_pipe.h b/src/core/iomgr/wakeup_fd_pipe.h index aa8f977ddb..01a13a97c0 100644 --- a/src/core/iomgr/wakeup_fd_pipe.h +++ b/src/core/iomgr/wakeup_fd_pipe.h @@ -38,4 +38,4 @@ extern grpc_wakeup_fd_vtable grpc_pipe_wakeup_fd_vtable; -#endif /* GRPC_INTERNAL_CORE_IOMGR_WAKEUP_FD_PIPE_H */ +#endif /* GRPC_INTERNAL_CORE_IOMGR_WAKEUP_FD_PIPE_H */ diff --git a/src/core/iomgr/wakeup_fd_posix.c b/src/core/iomgr/wakeup_fd_posix.c index e48f5223fa..d09fb78d12 100644 --- a/src/core/iomgr/wakeup_fd_posix.c +++ b/src/core/iomgr/wakeup_fd_posix.c @@ -53,9 +53,7 @@ void grpc_wakeup_fd_global_init_force_fallback(void) { wakeup_fd_vtable = &grpc_pipe_wakeup_fd_vtable; } -void grpc_wakeup_fd_global_destroy(void) { - wakeup_fd_vtable = NULL; -} +void grpc_wakeup_fd_global_destroy(void) { wakeup_fd_vtable = NULL; } void grpc_wakeup_fd_init(grpc_wakeup_fd *fd_info) { wakeup_fd_vtable->init(fd_info); @@ -73,4 +71,4 @@ void grpc_wakeup_fd_destroy(grpc_wakeup_fd *fd_info) { wakeup_fd_vtable->destroy(fd_info); } -#endif /* GPR_POSIX_WAKEUP_FD */ +#endif /* GPR_POSIX_WAKEUP_FD */ diff --git a/src/core/iomgr/wakeup_fd_posix.h b/src/core/iomgr/wakeup_fd_posix.h index a4da4df51f..b6c086900d 100644 --- a/src/core/iomgr/wakeup_fd_posix.h +++ b/src/core/iomgr/wakeup_fd_posix.h @@ -96,4 +96,4 @@ void grpc_wakeup_fd_destroy(grpc_wakeup_fd *fd_info); * wakeup_fd_nospecial.c if no such implementation exists. */ extern const grpc_wakeup_fd_vtable grpc_specialized_wakeup_fd_vtable; -#endif /* GRPC_INTERNAL_CORE_IOMGR_WAKEUP_FD_POSIX_H */ +#endif /* GRPC_INTERNAL_CORE_IOMGR_WAKEUP_FD_POSIX_H */ |