diff --git a/src/core/lib/iomgr/ev_epoll_linux.c b/src/core/lib/iomgr/ev_epoll_linux.c index d6664aead2e..715d057c517 100644 --- a/src/core/lib/iomgr/ev_epoll_linux.c +++ b/src/core/lib/iomgr/ev_epoll_linux.c @@ -796,7 +796,7 @@ static polling_island *polling_island_merge(polling_island *p, gpr_atm_rel_store(&p->merged_to, (gpr_atm)q); PI_ADD_REF(q, "pi_merge"); /* To account for the new incoming ref from p */ - workqueue_move_items_to_parent(q); + workqueue_move_items_to_parent(p); } /* else if p == q, nothing needs to be done */ diff --git a/test/core/iomgr/ev_epoll_linux_test.c b/test/core/iomgr/ev_epoll_linux_test.c index 40ae91bc6de..d95d5527731 100644 --- a/test/core/iomgr/ev_epoll_linux_test.c +++ b/test/core/iomgr/ev_epoll_linux_test.c @@ -45,6 +45,7 @@ #include #include "src/core/lib/iomgr/iomgr.h" +#include "src/core/lib/iomgr/workqueue.h" #include "test/core/util/test_config.h" typedef struct test_pollset { @@ -60,6 +61,22 @@ typedef struct test_fd { /* num_fds should be an even number */ static void test_fd_init(test_fd *tfds, int *fds, int num_fds) { int i; + int r; + + /* Create some dummy file descriptors. Currently using pipe file descriptors + * for this test but we could use any other type of file descriptors. Also, + * since pipe() used in this test creates two fds in each call, num_fds should + * be an even number */ + GPR_ASSERT((num_fds % 2) == 0); + for (i = 0; i < num_fds; i = i + 2) { + r = pipe(fds + i); + if (r != 0) { + gpr_log(GPR_ERROR, "Error in creating pipe. %d (%s)", errno, + strerror(errno)); + return; + } + } + for (i = 0; i < num_fds; i++) { tfds[i].inner_fd = fds[i]; tfds[i].fd = grpc_fd_create(fds[i], "test_fd"); @@ -111,8 +128,80 @@ static void test_pollset_cleanup(grpc_exec_ctx *exec_ctx, } } -#define NUM_FDS 8 -#define NUM_POLLSETS 4 +static void increment(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { + ++*(int *)arg; +} + +/* + * Validate that merging two workqueues preserves the closures in each queue. + * This is a regression test for a bug in + * polling_island_merge()[ev_epoll_linux.c], where the parent relationship was + * inverted. + */ +static void test_pollset_queue_merge_items() { + grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; + const int num_fds = 2; + const int num_pollsets = 2; + const int num_closures = 4; + test_fd tfds[num_fds]; + int fds[num_fds]; + test_pollset pollsets[num_pollsets]; + grpc_closure closures[num_closures]; + int i; + int result = 0; + + test_fd_init(tfds, fds, num_fds); + test_pollset_init(pollsets, num_pollsets); + + /* Two distinct polling islands, each with their own FD and pollset. */ + for (i = 0; i < num_fds; i++) { + grpc_pollset_add_fd(&exec_ctx, pollsets[i].pollset, tfds[i].fd); + grpc_exec_ctx_flush(&exec_ctx); + } + + /* Enqeue the closures, 3 to polling island 0 and 1 to polling island 1. */ + grpc_closure_init( + closures, increment, &result, + grpc_workqueue_scheduler(grpc_fd_get_polling_island(tfds[0].fd))); + grpc_closure_init( + closures + 1, increment, &result, + grpc_workqueue_scheduler(grpc_fd_get_polling_island(tfds[0].fd))); + grpc_closure_init( + closures + 2, increment, &result, + grpc_workqueue_scheduler(grpc_fd_get_polling_island(tfds[0].fd))); + grpc_closure_init( + closures + 3, increment, &result, + grpc_workqueue_scheduler(grpc_fd_get_polling_island(tfds[1].fd))); + for (i = 0; i < num_closures; ++i) { + grpc_closure_sched(&exec_ctx, closures + i, GRPC_ERROR_NONE); + } + + /* Merge the two polling islands. */ + grpc_pollset_add_fd(&exec_ctx, pollsets[0].pollset, tfds[1].fd); + grpc_exec_ctx_flush(&exec_ctx); + + /* + * Execute the closures, verify we see each one execute when executing work on + * the merged polling island. + */ + grpc_pollset_worker *worker = NULL; + for (i = 0; i < num_closures; ++i) { + const gpr_timespec deadline = gpr_time_add( + gpr_now(GPR_CLOCK_MONOTONIC), gpr_time_from_seconds(2, GPR_TIMESPAN)); + gpr_mu_lock(pollsets[1].mu); + GRPC_LOG_IF_ERROR( + "grpc_pollset_work", + grpc_pollset_work(&exec_ctx, pollsets[1].pollset, &worker, + gpr_now(GPR_CLOCK_MONOTONIC), deadline)); + gpr_mu_unlock(pollsets[1].mu); + } + GPR_ASSERT(result == num_closures); + + test_fd_cleanup(&exec_ctx, tfds, num_fds); + test_pollset_cleanup(&exec_ctx, pollsets, num_pollsets); + grpc_exec_ctx_finish(&exec_ctx); +} + /* * Cases to test: * case 1) Polling islands of both fd and pollset are NULL @@ -125,28 +214,16 @@ static void test_pollset_cleanup(grpc_exec_ctx *exec_ctx, * */ static void test_add_fd_to_pollset() { grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; - test_fd tfds[NUM_FDS]; - int fds[NUM_FDS]; - test_pollset pollsets[NUM_POLLSETS]; + const int num_fds = 8; + const int num_pollsets = 4; + test_fd tfds[num_fds]; + int fds[num_fds]; + test_pollset pollsets[num_pollsets]; void *expected_pi = NULL; int i; - int r; - /* Create some dummy file descriptors. Currently using pipe file descriptors - * for this test but we could use any other type of file descriptors. Also, - * since pipe() used in this test creates two fds in each call, NUM_FDS should - * be an even number */ - for (i = 0; i < NUM_FDS; i = i + 2) { - r = pipe(fds + i); - if (r != 0) { - gpr_log(GPR_ERROR, "Error in creating pipe. %d (%s)", errno, - strerror(errno)); - return; - } - } - - test_fd_init(tfds, fds, NUM_FDS); - test_pollset_init(pollsets, NUM_POLLSETS); + test_fd_init(tfds, fds, num_fds); + test_pollset_init(pollsets, num_pollsets); /*Step 1. * Create three polling islands (This will exercise test case 1 and 2) with @@ -207,19 +284,19 @@ static void test_add_fd_to_pollset() { /* Compare Fd:0's polling island with that of all other Fds */ expected_pi = grpc_fd_get_polling_island(tfds[0].fd); - for (i = 1; i < NUM_FDS; i++) { + for (i = 1; i < num_fds; i++) { GPR_ASSERT(grpc_are_polling_islands_equal( expected_pi, grpc_fd_get_polling_island(tfds[i].fd))); } /* Compare Fd:0's polling island with that of all other pollsets */ - for (i = 0; i < NUM_POLLSETS; i++) { + for (i = 0; i < num_pollsets; i++) { GPR_ASSERT(grpc_are_polling_islands_equal( expected_pi, grpc_pollset_get_polling_island(pollsets[i].pollset))); } - test_fd_cleanup(&exec_ctx, tfds, NUM_FDS); - test_pollset_cleanup(&exec_ctx, pollsets, NUM_POLLSETS); + test_fd_cleanup(&exec_ctx, tfds, num_fds); + test_pollset_cleanup(&exec_ctx, pollsets, num_pollsets); grpc_exec_ctx_finish(&exec_ctx); } @@ -231,6 +308,7 @@ int main(int argc, char **argv) { poll_strategy = grpc_get_poll_strategy_name(); if (poll_strategy != NULL && strcmp(poll_strategy, "epoll") == 0) { test_add_fd_to_pollset(); + test_pollset_queue_merge_items(); } else { gpr_log(GPR_INFO, "Skipping the test. The test is only relevant for 'epoll' "