diff --git a/test/core/iomgr/pollset_windows_starvation_test.cc b/test/core/iomgr/pollset_windows_starvation_test.cc index de26b9aa4b3..eed57b10075 100644 --- a/test/core/iomgr/pollset_windows_starvation_test.cc +++ b/test/core/iomgr/pollset_windows_starvation_test.cc @@ -15,8 +15,6 @@ * limitations under the License. * */ -#if defined(GRPC_WINSOCK_SOCKET) - #include #include @@ -31,36 +29,43 @@ #include "src/core/lib/surface/init.h" #include "test/core/util/test_config.h" +#if defined(GRPC_WINSOCK_SOCKET) + +// At least three threads are required to reproduce #18848 +const size_t THREADS = 3; + struct ThreadParams { gpr_cv cv; gpr_mu mu; int complete; + int queuing; + gpr_mu* pollset_mu[THREADS]; }; int main(int argc, char** argv) { grpc_init(); - // Create three threads that all start queueing for work. + // Create the threads that all start queueing for work. // // The first one becomes the active poller for work and the two other // threads go into the poller queue. // - // When work arrives, the first one notifies the next active poller, + // When work arrives, the first one notifies the next queued poller, // this wakes the second thread - however all this does is return from // the grpc_pollset_work function. It's up to that thread to figure // out if it still wants to queue for more work or if it should kick // other pollers. // // Previously that kick only affected pollers in the same pollset, thus - // leaving the third thread stuck in the poller queue. Now the pollset- + // leaving the other threads stuck in the poller queue. Now the pollset- // specific grpc_pollset_kick will also kick pollers from other pollsets // if there are no pollers in the current pollset. This frees up the - // last thread and completes the test. + // last threads and completes the test. ThreadParams params = {}; gpr_cv_init(¶ms.cv); gpr_mu_init(¶ms.mu); std::vector threads; - for (int i = 0; i < 3; i++) { + for (int i = 0; i < THREADS; i++) { grpc_core::Thread thd( "Poller", [](void* params) { @@ -71,46 +76,66 @@ int main(int argc, char** argv) { grpc_pollset pollset = {}; grpc_pollset_init(&pollset, &mu); + // Lock the pollset mutex before notifying the test runner thread that + // one more thread is queuing. This allows the test runner thread to + // wait for all threads to be queued before sending the first kick by + // waiting for the mutexes to be released, which happens in + // gpr_pollset_work when the poller is queued. gpr_mu_lock(mu); + gpr_mu_lock(&tparams->mu); + tparams->pollset_mu[tparams->queuing] = mu; + tparams->queuing++; + gpr_cv_signal(&tparams->cv); + gpr_mu_unlock(&tparams->mu); + // Queue for work and once we're done, make sure to kick the remaining // threads. - grpc_millis deadline = grpc_timespec_to_millis_round_up( - grpc_timeout_seconds_to_deadline(5)); grpc_error* error; - error = grpc_pollset_work(&pollset, NULL, deadline); + error = grpc_pollset_work(&pollset, NULL, GRPC_MILLIS_INF_FUTURE); error = grpc_pollset_kick(&pollset, NULL); gpr_mu_unlock(mu); - { - gpr_mu_lock(&tparams->mu); - tparams->complete++; - gpr_cv_signal(&tparams->cv); - gpr_mu_unlock(&tparams->mu); - } + gpr_mu_lock(&tparams->mu); + tparams->complete++; + gpr_cv_signal(&tparams->cv); + gpr_mu_unlock(&tparams->mu); }, ¶ms); thd.Start(); threads.push_back(std::move(thd)); } - // Wait for the threads to start working and then kick one of them. - gpr_sleep_until(grpc_timeout_milliseconds_to_deadline(10)); + // Wait for all three threads to be queuing. + gpr_mu_lock(¶ms.mu); + while ( + params.queuing != THREADS && + !gpr_cv_wait(¶ms.cv, ¶ms.mu, gpr_inf_future(GPR_CLOCK_REALTIME))) + ; + gpr_mu_unlock(¶ms.mu); + + // Wait for the mutexes to be released. This indicates that the threads have + // entered the work wait. + // + // At least currently these are essentially all references to the same global + // pollset mutex, but we are still waiting on them once for each thread in + // the case this ever changes. + for (int i = 0; i < THREADS; i++) { + gpr_mu_lock(params.pollset_mu[i]); + gpr_mu_unlock(params.pollset_mu[i]); + } + grpc_iocp_kick(); // Wait for the threads to complete. - gpr_timespec deadline = grpc_timeout_seconds_to_deadline(1); gpr_mu_lock(¶ms.mu); - while (params.complete != 3 && !gpr_cv_wait(¶ms.cv, ¶ms.mu, deadline)) + while ( + params.complete != THREADS && + !gpr_cv_wait(¶ms.cv, ¶ms.mu, gpr_inf_future(GPR_CLOCK_REALTIME))) ; - if (params.complete != 3) { - gpr_mu_unlock(¶ms.mu); - for (auto& t : threads) t.Join(); - return EXIT_FAILURE; - } - gpr_mu_unlock(¶ms.mu); + for (auto& t : threads) t.Join(); return EXIT_SUCCESS; }