1 /*
2 *
3 * Copyright 2016 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
19 #include "src/core/lib/gpr/mpscq.h"
20
21 #include <inttypes.h>
22 #include <stdlib.h>
23
24 #include <grpc/support/alloc.h>
25 #include <grpc/support/log.h>
26 #include <grpc/support/sync.h>
27
28 #include "src/core/lib/gpr/useful.h"
29 #include "src/core/lib/gprpp/thd.h"
30 #include "test/core/util/test_config.h"
31
32 typedef struct test_node {
33 gpr_mpscq_node node;
34 size_t i;
35 size_t* ctr;
36 } test_node;
37
new_node(size_t i,size_t * ctr)38 static test_node* new_node(size_t i, size_t* ctr) {
39 test_node* n = static_cast<test_node*>(gpr_malloc(sizeof(test_node)));
40 n->i = i;
41 n->ctr = ctr;
42 return n;
43 }
44
test_serial(void)45 static void test_serial(void) {
46 gpr_log(GPR_DEBUG, "test_serial");
47 gpr_mpscq q;
48 gpr_mpscq_init(&q);
49 for (size_t i = 0; i < 10000000; i++) {
50 gpr_mpscq_push(&q, &new_node(i, nullptr)->node);
51 }
52 for (size_t i = 0; i < 10000000; i++) {
53 test_node* n = reinterpret_cast<test_node*>(gpr_mpscq_pop(&q));
54 GPR_ASSERT(n);
55 GPR_ASSERT(n->i == i);
56 gpr_free(n);
57 }
58 }
59
60 typedef struct {
61 size_t ctr;
62 gpr_mpscq* q;
63 gpr_event* start;
64 } thd_args;
65
66 #define THREAD_ITERATIONS 10000
67
test_thread(void * args)68 static void test_thread(void* args) {
69 thd_args* a = static_cast<thd_args*>(args);
70 gpr_event_wait(a->start, gpr_inf_future(GPR_CLOCK_REALTIME));
71 for (size_t i = 1; i <= THREAD_ITERATIONS; i++) {
72 gpr_mpscq_push(a->q, &new_node(i, &a->ctr)->node);
73 }
74 }
75
test_mt(void)76 static void test_mt(void) {
77 gpr_log(GPR_DEBUG, "test_mt");
78 gpr_event start;
79 gpr_event_init(&start);
80 grpc_core::Thread thds[100];
81 thd_args ta[GPR_ARRAY_SIZE(thds)];
82 gpr_mpscq q;
83 gpr_mpscq_init(&q);
84 for (size_t i = 0; i < GPR_ARRAY_SIZE(thds); i++) {
85 ta[i].ctr = 0;
86 ta[i].q = &q;
87 ta[i].start = &start;
88 thds[i] = grpc_core::Thread("grpc_mt_test", test_thread, &ta[i]);
89 thds[i].Start();
90 }
91 size_t num_done = 0;
92 size_t spins = 0;
93 gpr_event_set(&start, (void*)1);
94 while (num_done != GPR_ARRAY_SIZE(thds)) {
95 gpr_mpscq_node* n;
96 while ((n = gpr_mpscq_pop(&q)) == nullptr) {
97 spins++;
98 }
99 test_node* tn = reinterpret_cast<test_node*>(n);
100 GPR_ASSERT(*tn->ctr == tn->i - 1);
101 *tn->ctr = tn->i;
102 if (tn->i == THREAD_ITERATIONS) num_done++;
103 gpr_free(tn);
104 }
105 gpr_log(GPR_DEBUG, "spins: %" PRIdPTR, spins);
106 for (auto& th : thds) {
107 th.Join();
108 }
109 gpr_mpscq_destroy(&q);
110 }
111
112 typedef struct {
113 thd_args* ta;
114 size_t num_thds;
115 gpr_mu mu;
116 size_t num_done;
117 size_t spins;
118 gpr_mpscq* q;
119 gpr_event* start;
120 } pull_args;
121
pull_thread(void * arg)122 static void pull_thread(void* arg) {
123 pull_args* pa = static_cast<pull_args*>(arg);
124 gpr_event_wait(pa->start, gpr_inf_future(GPR_CLOCK_REALTIME));
125
126 for (;;) {
127 gpr_mu_lock(&pa->mu);
128 if (pa->num_done == pa->num_thds) {
129 gpr_mu_unlock(&pa->mu);
130 return;
131 }
132 gpr_mpscq_node* n;
133 while ((n = gpr_mpscq_pop(pa->q)) == nullptr) {
134 pa->spins++;
135 }
136 test_node* tn = reinterpret_cast<test_node*>(n);
137 GPR_ASSERT(*tn->ctr == tn->i - 1);
138 *tn->ctr = tn->i;
139 if (tn->i == THREAD_ITERATIONS) pa->num_done++;
140 gpr_free(tn);
141 gpr_mu_unlock(&pa->mu);
142 }
143 }
144
test_mt_multipop(void)145 static void test_mt_multipop(void) {
146 gpr_log(GPR_DEBUG, "test_mt_multipop");
147 gpr_event start;
148 gpr_event_init(&start);
149 grpc_core::Thread thds[50];
150 grpc_core::Thread pull_thds[50];
151 thd_args ta[GPR_ARRAY_SIZE(thds)];
152 gpr_mpscq q;
153 gpr_mpscq_init(&q);
154 for (size_t i = 0; i < GPR_ARRAY_SIZE(thds); i++) {
155 ta[i].ctr = 0;
156 ta[i].q = &q;
157 ta[i].start = &start;
158 thds[i] = grpc_core::Thread("grpc_multipop_test", test_thread, &ta[i]);
159 thds[i].Start();
160 }
161 pull_args pa;
162 pa.ta = ta;
163 pa.num_thds = GPR_ARRAY_SIZE(thds);
164 pa.spins = 0;
165 pa.num_done = 0;
166 pa.q = &q;
167 pa.start = &start;
168 gpr_mu_init(&pa.mu);
169 for (size_t i = 0; i < GPR_ARRAY_SIZE(pull_thds); i++) {
170 pull_thds[i] = grpc_core::Thread("grpc_multipop_pull", pull_thread, &pa);
171 pull_thds[i].Start();
172 }
173 gpr_event_set(&start, (void*)1);
174 for (auto& pth : pull_thds) {
175 pth.Join();
176 }
177 gpr_log(GPR_DEBUG, "spins: %" PRIdPTR, pa.spins);
178 for (auto& th : thds) {
179 th.Join();
180 }
181 gpr_mpscq_destroy(&q);
182 }
183
main(int argc,char ** argv)184 int main(int argc, char** argv) {
185 grpc_test_init(argc, argv);
186 test_serial();
187 test_mt();
188 test_mt_multipop();
189 return 0;
190 }
191