2 simple ctdb benchmark - send messages in a ring around cluster
4 Copyright (C) Amitay Isaacs 2015
6 This program is free software; you can redistribute it and/or modify
7 it under the terms of the GNU General Public License as published by
8 the Free Software Foundation; either version 3 of the License, or
9 (at your option) any later version.
11 This program is distributed in the hope that it will be useful,
12 but WITHOUT ANY WARRANTY; without even the implied warranty of
13 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14 GNU General Public License for more details.
16 You should have received a copy of the GNU General Public License
17 along with this program; if not, see <http://www.gnu.org/licenses/>.
21 #include "system/network.h"
23 #include "lib/util/debug.h"
24 #include "lib/util/time.h"
25 #include "lib/util/tevent_unix.h"
27 #include "client/client.h"
28 #include "tests/src/test_options.h"
29 #include "tests/src/cluster_wait.h"
31 #define MSG_ID_BENCH 0
33 struct message_ring_state
{
34 struct tevent_context
*ev
;
35 struct ctdb_client_context
*client
;
40 int msg_plus
, msg_minus
;
41 struct timeval start_time
;
44 static void message_ring_wait(struct tevent_req
*subreq
);
45 static void message_ring_start(struct tevent_req
*subreq
);
46 static void message_ring_each_second(struct tevent_req
*subreq
);
47 static void message_ring_msg_sent(struct tevent_req
*subreq
);
48 static void message_ring_msg_handler(uint64_t srvid
, TDB_DATA data
,
50 static void message_ring_finish(struct tevent_req
*subreq
);
52 static struct tevent_req
*message_ring_send(TALLOC_CTX
*mem_ctx
,
53 struct tevent_context
*ev
,
54 struct ctdb_client_context
*client
,
55 int num_nodes
, int timelimit
,
58 struct tevent_req
*req
, *subreq
;
59 struct message_ring_state
*state
;
61 req
= tevent_req_create(mem_ctx
, &state
, struct message_ring_state
);
67 state
->client
= client
;
68 state
->num_nodes
= num_nodes
;
69 state
->timelimit
= timelimit
;
70 state
->interactive
= interactive
;
72 subreq
= ctdb_client_set_message_handler_send(
73 state
, state
->ev
, state
->client
,
75 message_ring_msg_handler
, req
);
76 if (tevent_req_nomem(subreq
, req
)) {
77 return tevent_req_post(req
, ev
);
79 tevent_req_set_callback(subreq
, message_ring_wait
, req
);
84 static void message_ring_wait(struct tevent_req
*subreq
)
86 struct tevent_req
*req
= tevent_req_callback_data(
87 subreq
, struct tevent_req
);
88 struct message_ring_state
*state
= tevent_req_data(
89 req
, struct message_ring_state
);
93 status
= ctdb_client_set_message_handler_recv(subreq
, &ret
);
96 tevent_req_error(req
, ret
);
100 subreq
= cluster_wait_send(state
, state
->ev
, state
->client
,
102 if (tevent_req_nomem(subreq
, req
)) {
105 tevent_req_set_callback(subreq
, message_ring_start
, req
);
108 static void message_ring_start(struct tevent_req
*subreq
)
110 struct tevent_req
*req
= tevent_req_callback_data(
111 subreq
, struct tevent_req
);
112 struct message_ring_state
*state
= tevent_req_data(
113 req
, struct message_ring_state
);
117 status
= cluster_wait_recv(subreq
, &ret
);
120 tevent_req_error(req
, ret
);
124 state
->start_time
= tevent_timeval_current();
126 if (ctdb_client_pnn(state
->client
) == 0) {
127 subreq
= tevent_wakeup_send(state
, state
->ev
,
128 tevent_timeval_current_ofs(1, 0));
129 if (tevent_req_nomem(subreq
, req
)) {
132 tevent_req_set_callback(subreq
, message_ring_each_second
, req
);
135 subreq
= tevent_wakeup_send(state
, state
->ev
,
136 tevent_timeval_current_ofs(
137 state
->timelimit
, 0));
138 if (tevent_req_nomem(subreq
, req
)) {
141 tevent_req_set_callback(subreq
, message_ring_finish
, req
);
144 static uint32_t next_node(struct ctdb_client_context
*client
,
145 int num_nodes
, int incr
)
147 return (ctdb_client_pnn(client
) + num_nodes
+ incr
) % num_nodes
;
150 static void message_ring_each_second(struct tevent_req
*subreq
)
152 struct tevent_req
*req
= tevent_req_callback_data(
153 subreq
, struct tevent_req
);
154 struct message_ring_state
*state
= tevent_req_data(
155 req
, struct message_ring_state
);
156 struct ctdb_req_message msg
;
161 status
= tevent_wakeup_recv(subreq
);
164 tevent_req_error(req
, EIO
);
168 pnn
= ctdb_client_pnn(state
->client
);
169 if (pnn
== 0 && state
->interactive
== 1) {
172 t
= timeval_elapsed(&state
->start_time
);
173 printf("Ring[%u]: %.2f msgs/sec (+ve=%d -ve=%d)\n",
174 pnn
, state
->msg_count
/ t
,
175 state
->msg_plus
, state
->msg_minus
);
179 if (state
->msg_plus
== 0) {
183 msg
.data
.data
.dptr
= (uint8_t *)&incr
;
184 msg
.data
.data
.dsize
= sizeof(incr
);
186 pnn
= next_node(state
->client
, state
->num_nodes
, incr
);
188 subreq
= ctdb_client_message_send(state
, state
->ev
,
189 state
->client
, pnn
, &msg
);
190 if (tevent_req_nomem(subreq
, req
)) {
193 tevent_req_set_callback(subreq
, message_ring_msg_sent
, req
);
196 if (state
->msg_minus
== 0) {
200 msg
.data
.data
.dptr
= (uint8_t *)&incr
;
201 msg
.data
.data
.dsize
= sizeof(incr
);
203 pnn
= next_node(state
->client
, state
->num_nodes
, incr
);
205 subreq
= ctdb_client_message_send(state
, state
->ev
,
206 state
->client
, pnn
, &msg
);
207 if (tevent_req_nomem(subreq
, req
)) {
210 tevent_req_set_callback(subreq
, message_ring_msg_sent
, req
);
213 subreq
= tevent_wakeup_send(state
, state
->ev
,
214 tevent_timeval_current_ofs(1, 0));
215 if (tevent_req_nomem(subreq
, req
)) {
218 tevent_req_set_callback(subreq
, message_ring_each_second
, req
);
221 static void message_ring_msg_sent(struct tevent_req
*subreq
)
223 struct tevent_req
*req
= tevent_req_callback_data(
224 subreq
, struct tevent_req
);
228 status
= ctdb_client_message_recv(subreq
, &ret
);
231 tevent_req_error(req
, ret
);
235 static void message_ring_msg_handler(uint64_t srvid
, TDB_DATA data
,
238 struct tevent_req
*req
= talloc_get_type_abort(
239 private_data
, struct tevent_req
);
240 struct message_ring_state
*state
= tevent_req_data(
241 req
, struct message_ring_state
);
242 struct ctdb_req_message msg
;
243 struct tevent_req
*subreq
;
247 if (srvid
!= MSG_ID_BENCH
) {
251 if (data
.dsize
!= sizeof(int)) {
254 incr
= *(int *)data
.dptr
;
256 state
->msg_count
+= 1;
258 state
->msg_plus
+= 1;
260 state
->msg_minus
+= 1;
263 pnn
= next_node(state
->client
, state
->num_nodes
, incr
);
266 msg
.data
.data
= data
;
268 subreq
= ctdb_client_message_send(state
, state
->ev
, state
->client
,
270 if (tevent_req_nomem(subreq
, req
)) {
273 tevent_req_set_callback(subreq
, message_ring_msg_sent
, req
);
276 static void message_ring_finish(struct tevent_req
*subreq
)
278 struct tevent_req
*req
= tevent_req_callback_data(
279 subreq
, struct tevent_req
);
280 struct message_ring_state
*state
= tevent_req_data(
281 req
, struct message_ring_state
);
285 status
= tevent_wakeup_recv(subreq
);
288 tevent_req_error(req
, EIO
);
292 t
= timeval_elapsed(&state
->start_time
);
294 printf("Ring[%u]: %.2f msgs/sec (+ve=%d -ve=%d)\n",
295 ctdb_client_pnn(state
->client
), state
->msg_count
/ t
,
296 state
->msg_plus
, state
->msg_minus
);
298 tevent_req_done(req
);
301 static bool message_ring_recv(struct tevent_req
*req
)
305 if (tevent_req_is_unix_error(req
, &ret
)) {
311 int main(int argc
, const char *argv
[])
313 const struct test_options
*opts
;
315 struct tevent_context
*ev
;
316 struct ctdb_client_context
*client
;
317 struct tevent_req
*req
;
321 setup_logging("message_ring", DEBUG_STDERR
);
323 status
= process_options_basic(argc
, argv
, &opts
);
328 mem_ctx
= talloc_new(NULL
);
329 if (mem_ctx
== NULL
) {
330 fprintf(stderr
, "Memory allocation error\n");
334 ev
= tevent_context_init(mem_ctx
);
336 fprintf(stderr
, "Memory allocation error\n");
340 ret
= ctdb_client_init(mem_ctx
, ev
, opts
->socket
, &client
);
342 fprintf(stderr
, "Failed to initialize client, ret=%d\n", ret
);
346 if (! ctdb_recovery_wait(ev
, client
)) {
347 fprintf(stderr
, "Failed to wait for recovery\n");
351 req
= message_ring_send(mem_ctx
, ev
, client
,
352 opts
->num_nodes
, opts
->timelimit
,
355 fprintf(stderr
, "Memory allocation error\n");
359 tevent_req_poll(req
, ev
);
361 status
= message_ring_recv(req
);
363 fprintf(stderr
, "message ring test failed\n");
367 talloc_free(mem_ctx
);