/* * * 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. * */ #include "src/core/lib/iomgr/ev_posix.h" #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include "src/core/lib/iomgr/ev_posix.h" #include "src/core/lib/iomgr/iomgr.h" #include "src/core/lib/iomgr/socket_utils_posix.h" #include "test/core/util/test_config.h" static gpr_mu *g_mu; static grpc_pollset *g_pollset; /* buffer size used to send and receive data. 1024 is the minimal value to set TCP send and receive buffer. */ #define BUF_SIZE 1024 /* Create a test socket with the right properties for testing. port is the TCP port to listen or connect to. Return a socket FD and sockaddr_in. */ static void create_test_socket(int port, int *socket_fd, struct sockaddr_in *sin) { int fd; int one = 1; int buffer_size_bytes = BUF_SIZE; int flags; fd = socket(AF_INET, SOCK_STREAM, 0); setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one)); /* Reset the size of socket send buffer to the minimal value to facilitate buffer filling up and triggering notify_on_write */ GPR_ASSERT(grpc_set_socket_sndbuf(fd, buffer_size_bytes)); GPR_ASSERT(grpc_set_socket_rcvbuf(fd, buffer_size_bytes)); /* Make fd non-blocking */ flags = fcntl(fd, F_GETFL, 0); GPR_ASSERT(fcntl(fd, F_SETFL, flags | O_NONBLOCK) == 0); *socket_fd = fd; /* Use local address for test */ sin->sin_family = AF_INET; sin->sin_addr.s_addr = htonl(0x7f000001); GPR_ASSERT(port >= 0 && port < 65536); sin->sin_port = htons((uint16_t)port); } /* Dummy gRPC callback */ void no_op_cb(void *arg, int success) {} /* =======An upload server to test notify_on_read=========== The server simply reads and counts a stream of bytes. */ /* An upload server. */ typedef struct { grpc_fd *em_fd; /* listening fd */ ssize_t read_bytes_total; /* total number of received bytes */ int done; /* set to 1 when a server finishes serving */ grpc_closure listen_closure; } server; static void server_init(server *sv) { sv->read_bytes_total = 0; sv->done = 0; } /* An upload session. Created when a new upload request arrives in the server. */ typedef struct { server *sv; /* not owned by a single session */ grpc_fd *em_fd; /* fd to read upload bytes */ char read_buf[BUF_SIZE]; /* buffer to store upload bytes */ grpc_closure session_read_closure; } session; /* Called when an upload session can be safely shutdown. Close session FD and start to shutdown listen FD. */ static void session_shutdown_cb(grpc_exec_ctx *exec_ctx, void *arg, /*session */ bool success) { session *se = arg; server *sv = se->sv; grpc_fd_orphan(exec_ctx, se->em_fd, NULL, NULL, "a"); gpr_free(se); /* Start to shutdown listen fd. */ grpc_fd_shutdown(exec_ctx, sv->em_fd); } /* Called when data become readable in a session. */ static void session_read_cb(grpc_exec_ctx *exec_ctx, void *arg, /*session */ bool success) { session *se = arg; int fd = grpc_fd_wrapped_fd(se->em_fd); ssize_t read_once = 0; ssize_t read_total = 0; if (!success) { session_shutdown_cb(exec_ctx, arg, 1); return; } do { read_once = read(fd, se->read_buf, BUF_SIZE); if (read_once > 0) read_total += read_once; } while (read_once > 0); se->sv->read_bytes_total += read_total; /* read() returns 0 to indicate the TCP connection was closed by the client. read(fd, read_buf, 0) also returns 0 which should never be called as such. It is possible to read nothing due to spurious edge event or data has been drained, In such a case, read() returns -1 and set errno to EAGAIN. */ if (read_once == 0) { session_shutdown_cb(exec_ctx, arg, 1); } else if (read_once == -1) { if (errno == EAGAIN) { /* An edge triggered event is cached in the kernel until next poll. In the current single thread implementation, session_read_cb is called in the polling thread, such that polling only happens after this callback, and will catch read edge event if data is available again before notify_on_read. TODO(chenw): in multi-threaded version, callback and polling can be run in different threads. polling may catch a persist read edge event before notify_on_read is called. */ grpc_fd_notify_on_read(exec_ctx, se->em_fd, &se->session_read_closure); } else { gpr_log(GPR_ERROR, "Unhandled read error %s", strerror(errno)); abort(); } } } /* Called when the listen FD can be safely shutdown. Close listen FD and signal that server can be shutdown. */ static void listen_shutdown_cb(grpc_exec_ctx *exec_ctx, void *arg /*server */, int success) { server *sv = arg; grpc_fd_orphan(exec_ctx, sv->em_fd, NULL, NULL, "b"); gpr_mu_lock(g_mu); sv->done = 1; grpc_pollset_kick(g_pollset, NULL); gpr_mu_unlock(g_mu); } /* Called when a new TCP connection request arrives in the listening port. */ static void listen_cb(grpc_exec_ctx *exec_ctx, void *arg, /*=sv_arg*/ bool success) { server *sv = arg; int fd; int flags; session *se; struct sockaddr_storage ss; socklen_t slen = sizeof(ss); grpc_fd *listen_em_fd = sv->em_fd; if (!success) { listen_shutdown_cb(exec_ctx, arg, 1); return; } fd = accept(grpc_fd_wrapped_fd(listen_em_fd), (struct sockaddr *)&ss, &slen); GPR_ASSERT(fd >= 0); GPR_ASSERT(fd < FD_SETSIZE); flags = fcntl(fd, F_GETFL, 0); fcntl(fd, F_SETFL, flags | O_NONBLOCK); se = gpr_malloc(sizeof(*se)); se->sv = sv; se->em_fd = grpc_fd_create(fd, "listener"); grpc_pollset_add_fd(exec_ctx, g_pollset, se->em_fd); se->session_read_closure.cb = session_read_cb; se->session_read_closure.cb_arg = se; grpc_fd_notify_on_read(exec_ctx, se->em_fd, &se->session_read_closure); grpc_fd_notify_on_read(exec_ctx, listen_em_fd, &sv->listen_closure); } /* Max number of connections pending to be accepted by listen(). */ #define MAX_NUM_FD 1024 /* Start a test server, return the TCP listening port bound to listen_fd. listen_cb() is registered to be interested in reading from listen_fd. When connection request arrives, listen_cb() is called to accept the connection request. */ static int server_start(grpc_exec_ctx *exec_ctx, server *sv) { int port = 0; int fd; struct sockaddr_in sin; socklen_t addr_len; create_test_socket(port, &fd, &sin); addr_len = sizeof(sin); GPR_ASSERT(bind(fd, (struct sockaddr *)&sin, addr_len) == 0); GPR_ASSERT(getsockname(fd, (struct sockaddr *)&sin, &addr_len) == 0); port = ntohs(sin.sin_port); GPR_ASSERT(listen(fd, MAX_NUM_FD) == 0); sv->em_fd = grpc_fd_create(fd, "server"); grpc_pollset_add_fd(exec_ctx, g_pollset, sv->em_fd); /* Register to be interested in reading from listen_fd. */ sv->listen_closure.cb = listen_cb; sv->listen_closure.cb_arg = sv; grpc_fd_notify_on_read(exec_ctx, sv->em_fd, &sv->listen_closure); return port; } /* Wait and shutdown a sever. */ static void server_wait_and_shutdown(server *sv) { 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); grpc_exec_ctx_finish(&exec_ctx); gpr_mu_lock(g_mu); } gpr_mu_unlock(g_mu); } /* ===An upload client to test notify_on_write=== */ /* Client write buffer size */ #define CLIENT_WRITE_BUF_SIZE 10 /* Total number of times that the client fills up the write buffer */ #define CLIENT_TOTAL_WRITE_CNT 3 /* An upload client. */ typedef struct { grpc_fd *em_fd; char write_buf[CLIENT_WRITE_BUF_SIZE]; ssize_t write_bytes_total; /* Number of times that the client fills up the write buffer and calls notify_on_write to schedule another write. */ int client_write_cnt; int done; /* set to 1 when a client finishes sending */ grpc_closure write_closure; } client; static void client_init(client *cl) { memset(cl->write_buf, 0, sizeof(cl->write_buf)); cl->write_bytes_total = 0; cl->client_write_cnt = 0; cl->done = 0; } /* Called when a client upload session is ready to shutdown. */ static void client_session_shutdown_cb(grpc_exec_ctx *exec_ctx, void *arg /*client */, int success) { client *cl = arg; grpc_fd_orphan(exec_ctx, cl->em_fd, NULL, NULL, "c"); cl->done = 1; grpc_pollset_kick(g_pollset, NULL); } /* Write as much as possible, then register notify_on_write. */ static void client_session_write(grpc_exec_ctx *exec_ctx, void *arg, /*client */ bool success) { client *cl = arg; int fd = grpc_fd_wrapped_fd(cl->em_fd); ssize_t write_once = 0; if (!success) { gpr_mu_lock(g_mu); client_session_shutdown_cb(exec_ctx, arg, 1); gpr_mu_unlock(g_mu); return; } do { write_once = write(fd, cl->write_buf, CLIENT_WRITE_BUF_SIZE); if (write_once > 0) cl->write_bytes_total += write_once; } while (write_once > 0); if (errno == EAGAIN) { 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; grpc_fd_notify_on_write(exec_ctx, cl->em_fd, &cl->write_closure); cl->client_write_cnt++; } else { client_session_shutdown_cb(exec_ctx, arg, 1); } gpr_mu_unlock(g_mu); } else { gpr_log(GPR_ERROR, "unknown errno %s", strerror(errno)); abort(); } } /* Start a client to send a stream of bytes. */ static void client_start(grpc_exec_ctx *exec_ctx, client *cl, int port) { int fd; struct sockaddr_in sin; create_test_socket(port, &fd, &sin); if (connect(fd, (struct sockaddr *)&sin, sizeof(sin)) == -1) { if (errno == EINPROGRESS) { struct pollfd pfd; pfd.fd = fd; pfd.events = POLLOUT; pfd.revents = 0; if (poll(&pfd, 1, -1) == -1) { gpr_log(GPR_ERROR, "poll() failed during connect; errno=%d", errno); abort(); } } else { gpr_log(GPR_ERROR, "Failed to connect to the server (errno=%d)", errno); abort(); } } cl->em_fd = grpc_fd_create(fd, "client"); grpc_pollset_add_fd(exec_ctx, g_pollset, cl->em_fd); client_session_write(exec_ctx, cl, 1); } /* Wait for the signal to shutdown a client. */ static void client_wait_and_shutdown(client *cl) { 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); grpc_exec_ctx_finish(&exec_ctx); gpr_mu_lock(g_mu); } gpr_mu_unlock(g_mu); } /* Test grpc_fd. Start an upload server and client, upload a stream of bytes from the client to the server, and verify that the total number of sent bytes is equal to the total number of received bytes. */ static void test_grpc_fd(void) { server sv; client cl; int port; grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; server_init(&sv); port = server_start(&exec_ctx, &sv); client_init(&cl); client_start(&exec_ctx, &cl, port); grpc_exec_ctx_finish(&exec_ctx); client_wait_and_shutdown(&cl); server_wait_and_shutdown(&sv); GPR_ASSERT(sv.read_bytes_total == cl.write_bytes_total); gpr_log(GPR_INFO, "Total read bytes %" PRIdPTR, sv.read_bytes_total); } typedef struct fd_change_data { grpc_iomgr_cb_func cb_that_ran; } fd_change_data; void init_change_data(fd_change_data *fdc) { fdc->cb_that_ran = NULL; } void destroy_change_data(fd_change_data *fdc) {} 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); fdc->cb_that_ran = first_read_callback; grpc_pollset_kick(g_pollset, NULL); 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); fdc->cb_that_ran = second_read_callback; grpc_pollset_kick(g_pollset, NULL); gpr_mu_unlock(g_mu); } /* Test that changing the callback we use for notify_on_read actually works. Note that we have two different but almost identical callbacks above -- the point is to have two different function pointers and two different data pointers and make sure that changing both really works. */ static void test_grpc_fd_change(void) { grpc_fd *em_fd; fd_change_data a, b; int flags; int sv[2]; char data; ssize_t result; grpc_closure first_closure; grpc_closure second_closure; grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; first_closure.cb = first_read_callback; first_closure.cb_arg = &a; second_closure.cb = second_read_callback; second_closure.cb_arg = &b; init_change_data(&a); init_change_data(&b); GPR_ASSERT(socketpair(AF_UNIX, SOCK_STREAM, 0, sv) == 0); flags = fcntl(sv[0], F_GETFL, 0); GPR_ASSERT(fcntl(sv[0], F_SETFL, flags | O_NONBLOCK) == 0); flags = fcntl(sv[1], F_GETFL, 0); GPR_ASSERT(fcntl(sv[1], F_SETFL, flags | O_NONBLOCK) == 0); em_fd = grpc_fd_create(sv[0], "test_grpc_fd_change"); grpc_pollset_add_fd(&exec_ctx, g_pollset, em_fd); /* Register the first callback, then make its FD readable */ grpc_fd_notify_on_read(&exec_ctx, em_fd, &first_closure); data = 0; result = write(sv[1], &data, 1); GPR_ASSERT(result == 1); /* And now wait for it to run. */ 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); grpc_exec_ctx_finish(&exec_ctx); gpr_mu_lock(g_mu); } GPR_ASSERT(a.cb_that_ran == first_read_callback); gpr_mu_unlock(g_mu); /* And drain the socket so we can generate a new read edge */ result = read(sv[0], &data, 1); GPR_ASSERT(result == 1); /* Now register a second callback with distinct change data, and do the same thing again. */ grpc_fd_notify_on_read(&exec_ctx, em_fd, &second_closure); data = 0; result = write(sv[1], &data, 1); GPR_ASSERT(result == 1); 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); grpc_exec_ctx_finish(&exec_ctx); 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); grpc_fd_orphan(&exec_ctx, em_fd, NULL, NULL, "d"); grpc_exec_ctx_finish(&exec_ctx); destroy_change_data(&a); destroy_change_data(&b); close(sv[1]); } static void destroy_pollset(grpc_exec_ctx *exec_ctx, void *p, bool success) { grpc_pollset_destroy(p); } int main(int argc, char **argv) { grpc_closure destroyed; grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; grpc_test_init(argc, argv); grpc_iomgr_init(); g_pollset = gpr_malloc(grpc_pollset_size()); grpc_pollset_init(g_pollset, &g_mu); test_grpc_fd(); test_grpc_fd_change(); grpc_closure_init(&destroyed, destroy_pollset, g_pollset); grpc_pollset_shutdown(&exec_ctx, g_pollset, &destroyed); grpc_exec_ctx_finish(&exec_ctx); gpr_free(g_pollset); grpc_iomgr_shutdown(); return 0; }