1 /*
2 *
3 * Copyright 2015 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/iomgr/combiner.h"
20
21 #include <grpc/grpc.h>
22 #include <grpc/support/alloc.h>
23 #include <grpc/support/log.h>
24
25 #include "src/core/lib/gpr/useful.h"
26 #include "src/core/lib/gprpp/thd.h"
27 #include "test/core/util/test_config.h"
28
test_no_op(void)29 static void test_no_op(void) {
30 gpr_log(GPR_DEBUG, "test_no_op");
31 grpc_core::ExecCtx exec_ctx;
32 GRPC_COMBINER_UNREF(grpc_combiner_create(), "test_no_op");
33 }
34
set_event_to_true(void * value,grpc_error *)35 static void set_event_to_true(void* value, grpc_error* /*error*/) {
36 gpr_event_set(static_cast<gpr_event*>(value), (void*)1);
37 }
38
test_execute_one(void)39 static void test_execute_one(void) {
40 gpr_log(GPR_DEBUG, "test_execute_one");
41
42 grpc_core::Combiner* lock = grpc_combiner_create();
43 gpr_event done;
44 gpr_event_init(&done);
45 grpc_core::ExecCtx exec_ctx;
46 lock->Run(GRPC_CLOSURE_CREATE(set_event_to_true, &done, nullptr),
47 GRPC_ERROR_NONE);
48 grpc_core::ExecCtx::Get()->Flush();
49 GPR_ASSERT(gpr_event_wait(&done, grpc_timeout_seconds_to_deadline(5)) !=
50 nullptr);
51 GRPC_COMBINER_UNREF(lock, "test_execute_one");
52 }
53
54 typedef struct {
55 size_t ctr;
56 grpc_core::Combiner* lock;
57 gpr_event done;
58 } thd_args;
59
60 typedef struct {
61 size_t* ctr;
62 size_t value;
63 } ex_args;
64
check_one(void * a,grpc_error *)65 static void check_one(void* a, grpc_error* /*error*/) {
66 ex_args* args = static_cast<ex_args*>(a);
67 GPR_ASSERT(*args->ctr == args->value - 1);
68 *args->ctr = args->value;
69 gpr_free(a);
70 }
71
execute_many_loop(void * a)72 static void execute_many_loop(void* a) {
73 thd_args* args = static_cast<thd_args*>(a);
74 grpc_core::ExecCtx exec_ctx;
75 size_t n = 1;
76 for (size_t i = 0; i < 10; i++) {
77 for (size_t j = 0; j < 10000; j++) {
78 ex_args* c = static_cast<ex_args*>(gpr_malloc(sizeof(*c)));
79 c->ctr = &args->ctr;
80 c->value = n++;
81 args->lock->Run(GRPC_CLOSURE_CREATE(check_one, c, nullptr),
82 GRPC_ERROR_NONE);
83 grpc_core::ExecCtx::Get()->Flush();
84 }
85 // sleep for a little bit, to test a combiner draining and another thread
86 // picking it up
87 gpr_sleep_until(grpc_timeout_milliseconds_to_deadline(100));
88 }
89 args->lock->Run(GRPC_CLOSURE_CREATE(set_event_to_true, &args->done, nullptr),
90 GRPC_ERROR_NONE);
91 }
92
test_execute_many(void)93 static void test_execute_many(void) {
94 gpr_log(GPR_DEBUG, "test_execute_many");
95
96 grpc_core::Combiner* lock = grpc_combiner_create();
97 grpc_core::Thread thds[100];
98 thd_args ta[GPR_ARRAY_SIZE(thds)];
99 for (size_t i = 0; i < GPR_ARRAY_SIZE(thds); i++) {
100 ta[i].ctr = 0;
101 ta[i].lock = lock;
102 gpr_event_init(&ta[i].done);
103 thds[i] = grpc_core::Thread("grpc_execute_many", execute_many_loop, &ta[i]);
104 thds[i].Start();
105 }
106 for (size_t i = 0; i < GPR_ARRAY_SIZE(thds); i++) {
107 GPR_ASSERT(gpr_event_wait(&ta[i].done,
108 gpr_inf_future(GPR_CLOCK_REALTIME)) != nullptr);
109 thds[i].Join();
110 }
111 grpc_core::ExecCtx exec_ctx;
112 GRPC_COMBINER_UNREF(lock, "test_execute_many");
113 }
114
115 static gpr_event got_in_finally;
116
in_finally(void *,grpc_error *)117 static void in_finally(void* /*arg*/, grpc_error* /*error*/) {
118 gpr_event_set(&got_in_finally, (void*)1);
119 }
120
add_finally(void * arg,grpc_error *)121 static void add_finally(void* arg, grpc_error* /*error*/) {
122 static_cast<grpc_core::Combiner*>(arg)->Run(
123 GRPC_CLOSURE_CREATE(in_finally, arg, nullptr), GRPC_ERROR_NONE);
124 }
125
test_execute_finally(void)126 static void test_execute_finally(void) {
127 gpr_log(GPR_DEBUG, "test_execute_finally");
128
129 grpc_core::Combiner* lock = grpc_combiner_create();
130 grpc_core::ExecCtx exec_ctx;
131 gpr_event_init(&got_in_finally);
132 lock->Run(GRPC_CLOSURE_CREATE(add_finally, lock, nullptr), GRPC_ERROR_NONE);
133 grpc_core::ExecCtx::Get()->Flush();
134 GPR_ASSERT(gpr_event_wait(&got_in_finally,
135 grpc_timeout_seconds_to_deadline(5)) != nullptr);
136 GRPC_COMBINER_UNREF(lock, "test_execute_finally");
137 }
138
main(int argc,char ** argv)139 int main(int argc, char** argv) {
140 grpc::testing::TestEnvironment env(argc, argv);
141 grpc_init();
142 test_no_op();
143 test_execute_one();
144 test_execute_finally();
145 test_execute_many();
146 grpc_shutdown();
147
148 return 0;
149 }
150