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 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 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 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 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 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 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 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
