xref: /aosp_15_r20/external/grpc-grpc/test/core/iomgr/pollset_windows_starvation_test.cc (revision cc02d7e222339f7a4f6ba5f422e6413f4bd931f2) 
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(&params.cv);
66   gpr_mu_init(&params.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_handle error;
95           error = grpc_pollset_work(&pollset, NULL,
96                                     grpc_core::Timestamp::InfFuture());
97           error = grpc_pollset_kick(&pollset, NULL);
98 
99           gpr_mu_unlock(mu);
100 
101           gpr_mu_lock(&tparams->mu);
102           tparams->complete++;
103           gpr_cv_signal(&tparams->cv);
104           gpr_mu_unlock(&tparams->mu);
105         },
106         &params);
107     thd.Start();
108     threads.push_back(std::move(thd));
109   }
110 
111   // Wait for all three threads to be queuing.
112   gpr_mu_lock(&params.mu);
113   while (
114       params.queuing != THREADS &&
115       !gpr_cv_wait(&params.cv, &params.mu, gpr_inf_future(GPR_CLOCK_REALTIME)))
116     ;
117   gpr_mu_unlock(&params.mu);
118 
119   // Wait for the mutexes to be released. This indicates that the threads have
120   // entered the work wait.
121   //
122   // At least currently these are essentially all references to the same global
123   // pollset mutex, but we are still waiting on them once for each thread in
124   // the case this ever changes.
125   for (int i = 0; i < THREADS; i++) {
126     gpr_mu_lock(params.pollset_mu[i]);
127     gpr_mu_unlock(params.pollset_mu[i]);
128   }
129 
130   grpc_iocp_kick();
131 
132   // Wait for the threads to complete.
133   gpr_mu_lock(&params.mu);
134   while (
135       params.complete != THREADS &&
136       !gpr_cv_wait(&params.cv, &params.mu, gpr_inf_future(GPR_CLOCK_REALTIME)))
137     ;
138   gpr_mu_unlock(&params.mu);
139 
140   for (auto& t : threads) t.Join();
141   return EXIT_SUCCESS;
142 }
143 #else  // defined(GRPC_WINSOCK_SOCKET)
main(int,char **)144 int main(int /*argc*/, char** /*argv*/) { return 0; }
145 #endif
146