diff --git a/test/core/iomgr/fd_posix_test.c b/test/core/iomgr/fd_posix_test.c index f97f33712eb..18cd825df00 100644 --- a/test/core/iomgr/fd_posix_test.c +++ b/test/core/iomgr/fd_posix_test.c @@ -518,6 +518,103 @@ static void destroy_pollset(grpc_exec_ctx *exec_ctx, void *p, bool success) { grpc_pollset_destroy(p); } +typedef struct read_notifier_test_fd_context { + grpc_fd *fd; + bool is_cb_called; +} read_notifier_test_fd_context; + +static void read_notifier_test_callback( + grpc_exec_ctx *exec_ctx, void *arg /* (read_notifier_test_fd_context *) */, + bool success) { + read_notifier_test_fd_context *fd_context = arg; + grpc_fd *fd = fd_context->fd; + + /* Verify that the read notifier pollset is set */ + GPR_ASSERT(grpc_fd_get_read_notifier_pollset(exec_ctx, fd) != NULL); + fd_context->is_cb_called = true; +} + +/* sv MUST to be an array of size 2 */ +static void get_socket_pair(int sv[]) { + int flags = 0; + 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); +} + +static grpc_pollset *create_grpc_pollset(gpr_mu **mu) { + grpc_pollset *pollset = gpr_malloc(grpc_pollset_size()); + grpc_pollset_init(pollset, mu); + return pollset; +} + +static void free_grpc_pollset(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset) { + grpc_closure destroyed; + grpc_closure_init(&destroyed, destroy_pollset, pollset); + grpc_pollset_shutdown(exec_ctx, pollset, &destroyed); + grpc_exec_ctx_finish(exec_ctx); + gpr_free(pollset); +} + +static void test_grpc_fd_read_notifier_pollset(void) { + grpc_fd *em_fd[2]; + read_notifier_test_fd_context fd_context; + int sv[2][2]; + char data; + ssize_t result; + int i; + grpc_closure on_read_closure; + gpr_mu *mu; + grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; + + grpc_pollset *pollset = create_grpc_pollset(&mu); + + for (i = 0; i < 2; i++) { + get_socket_pair(sv[i]); + + em_fd[i] = grpc_fd_create(sv[i][0], "test_grpc_fd_1_read_notifier_pollset"); + + grpc_pollset_add_fd(&exec_ctx, pollset, em_fd[i]); + + on_read_closure.cb = read_notifier_test_callback; + fd_context.fd = em_fd[i]; + fd_context.is_cb_called = false; + on_read_closure.cb_arg = &fd_context; + grpc_fd_notify_on_read(&exec_ctx, em_fd[i], &on_read_closure); + + data = 0; + result = write(sv[i][1], &data, sizeof(data)); + GPR_ASSERT(result == 1); + + gpr_mu_lock(mu); + while (!fd_context.is_cb_called) { + grpc_pollset_worker *worker = NULL; + grpc_pollset_work(&exec_ctx, pollset, &worker, + gpr_now(GPR_CLOCK_MONOTONIC), + gpr_inf_future(GPR_CLOCK_MONOTONIC)); + gpr_mu_unlock(mu); + grpc_exec_ctx_finish(&exec_ctx); + gpr_mu_lock(mu); + } + gpr_mu_unlock(mu); + + /* Drain the socket (Not really needed for the test) */ + result = read(sv[i][0], &data, 1); + GPR_ASSERT(result == 1); + } + + + for (i = 0; i < 2; i++) { + grpc_fd_orphan(&exec_ctx, em_fd[i], NULL, NULL, ""); + close(sv[i][1]); + } + + free_grpc_pollset(&exec_ctx, pollset); + grpc_exec_ctx_finish(&exec_ctx); +} + int main(int argc, char **argv) { grpc_closure destroyed; grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; @@ -527,6 +624,7 @@ int main(int argc, char **argv) { grpc_pollset_init(g_pollset, &g_mu); test_grpc_fd(); test_grpc_fd_change(); + test_grpc_fd_read_notifier_pollset(); grpc_closure_init(&destroyed, destroy_pollset, g_pollset); grpc_pollset_shutdown(&exec_ctx, g_pollset, &destroyed); grpc_exec_ctx_finish(&exec_ctx);