1 /*
2 *
3 * Copyright 2019 gRPC authors.
4 *
5 * Licensed under the Apache License, Version 2.0 (the "License");
6 * you may not use this file except in compliance with the License.
7 * You may obtain a copy of the License at
8 *
9 * http://www.apache.org/licenses/LICENSE-2.0
10 *
11 * Unless required by applicable law or agreed to in writing, software
12 * distributed under the License is distributed on an "AS IS" BASIS,
13 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 * See the License for the specific language governing permissions and
15 * limitations under the License.
16 *
17 */
18 #include <vector>
19
20 #include <grpc/grpc.h>
21 #include <grpc/support/time.h>
22
23 #include "src/core/lib/gprpp/thd.h"
24 #include "src/core/lib/iomgr/exec_ctx.h"
25 #include "src/core/lib/iomgr/iocp_windows.h"
26 #include "src/core/lib/iomgr/iomgr_internal.h"
27 #include "src/core/lib/iomgr/pollset.h"
28 #include "src/core/lib/iomgr/pollset_windows.h"
29 #include "src/core/lib/surface/init.h"
30 #include "test/core/util/test_config.h"
31
32 #if defined(GRPC_WINSOCK_SOCKET)
33
34 // At least three threads are required to reproduce #18848
35 const size_t THREADS = 3;
36
37 struct ThreadParams {
38 gpr_cv cv;
39 gpr_mu mu;
40 int complete;
41 int queuing;
42 gpr_mu* pollset_mu[THREADS];
43 };
44
main(int argc,char ** argv)45 int main(int argc, char** argv) {
46 grpc_init();
47
48 // Create the threads that all start queueing for work.
49 //
50 // The first one becomes the active poller for work and the two other
51 // threads go into the poller queue.
52 //
53 // When work arrives, the first one notifies the next queued poller,
54 // this wakes the second thread - however all this does is return from
55 // the grpc_pollset_work function. It's up to that thread to figure
56 // out if it still wants to queue for more work or if it should kick
57 // other pollers.
58 //
59 // Previously that kick only affected pollers in the same pollset, thus
60 // leaving the other threads stuck in the poller queue. Now the pollset-
61 // specific grpc_pollset_kick will also kick pollers from other pollsets
62 // if there are no pollers in the current pollset. This frees up the
63 // last threads and completes the test.
64 ThreadParams params = {};
65 gpr_cv_init(¶ms.cv);
66 gpr_mu_init(¶ms.mu);
67 std::vector<grpc_core::Thread> threads;
68 for (int i = 0; i < THREADS; i++) {
69 grpc_core::Thread thd(
70 "Poller",
71 [](void* params) {
72 ThreadParams* tparams = static_cast<ThreadParams*>(params);
73 grpc_core::ExecCtx exec_ctx;
74
75 gpr_mu* mu;
76 grpc_pollset pollset = {};
77 grpc_pollset_init(&pollset, &mu);
78
79 // Lock the pollset mutex before notifying the test runner thread that
80 // one more thread is queuing. This allows the test runner thread to
81 // wait for all threads to be queued before sending the first kick by
82 // waiting for the mutexes to be released, which happens in
83 // gpr_pollset_work when the poller is queued.
84 gpr_mu_lock(mu);
85
86 gpr_mu_lock(&tparams->mu);
87 tparams->pollset_mu[tparams->queuing] = mu;
88 tparams->queuing++;
89 gpr_cv_signal(&tparams->cv);
90 gpr_mu_unlock(&tparams->mu);
91
92 // Queue for work and once we're done, make sure to kick the remaining
93 // threads.
94 grpc_error* error;
95 error = grpc_pollset_work(&pollset, NULL, GRPC_MILLIS_INF_FUTURE);
96 error = grpc_pollset_kick(&pollset, NULL);
97
98 gpr_mu_unlock(mu);
99
100 gpr_mu_lock(&tparams->mu);
101 tparams->complete++;
102 gpr_cv_signal(&tparams->cv);
103 gpr_mu_unlock(&tparams->mu);
104 },
105 ¶ms);
106 thd.Start();
107 threads.push_back(std::move(thd));
108 }
109
110 // Wait for all three threads to be queuing.
111 gpr_mu_lock(¶ms.mu);
112 while (
113 params.queuing != THREADS &&
114 !gpr_cv_wait(¶ms.cv, ¶ms.mu, gpr_inf_future(GPR_CLOCK_REALTIME)))
115 ;
116 gpr_mu_unlock(¶ms.mu);
117
118 // Wait for the mutexes to be released. This indicates that the threads have
119 // entered the work wait.
120 //
121 // At least currently these are essentially all references to the same global
122 // pollset mutex, but we are still waiting on them once for each thread in
123 // the case this ever changes.
124 for (int i = 0; i < THREADS; i++) {
125 gpr_mu_lock(params.pollset_mu[i]);
126 gpr_mu_unlock(params.pollset_mu[i]);
127 }
128
129 grpc_iocp_kick();
130
131 // Wait for the threads to complete.
132 gpr_mu_lock(¶ms.mu);
133 while (
134 params.complete != THREADS &&
135 !gpr_cv_wait(¶ms.cv, ¶ms.mu, gpr_inf_future(GPR_CLOCK_REALTIME)))
136 ;
137 gpr_mu_unlock(¶ms.mu);
138
139 for (auto& t : threads) t.Join();
140 return EXIT_SUCCESS;
141 }
142 #else /* defined(GRPC_WINSOCK_SOCKET) */
main(int,char **)143 int main(int /*argc*/, char** /*argv*/) { return 0; }
144 #endif
145