s4:messaging: add local.messaging.multi_ctx.multi_ctx test
[kai/samba-autobuild/.git] / source4 / lib / messaging / tests / messaging.c
1 /* 
2    Unix SMB/CIFS implementation.
3
4    local test for messaging code
5
6    Copyright (C) Andrew Tridgell 2004
7    
8    This program is free software; you can redistribute it and/or modify
9    it under the terms of the GNU General Public License as published by
10    the Free Software Foundation; either version 3 of the License, or
11    (at your option) any later version.
12    
13    This program is distributed in the hope that it will be useful,
14    but WITHOUT ANY WARRANTY; without even the implied warranty of
15    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
16    GNU General Public License for more details.
17    
18    You should have received a copy of the GNU General Public License
19    along with this program.  If not, see <http://www.gnu.org/licenses/>.
20 */
21
22 #include "includes.h"
23 #include "lib/events/events.h"
24 #include "lib/messaging/irpc.h"
25 #include "torture/torture.h"
26 #include "cluster/cluster.h"
27 #include "param/param.h"
28 #include "torture/local/proto.h"
29 #include "system/select.h"
30 #include "system/filesys.h"
31 #include "lib/crypto/md5.h"
32
33 static uint32_t msg_pong;
34
35 static void ping_message(struct imessaging_context *msg, void *private_data,
36                          uint32_t msg_type, struct server_id src, DATA_BLOB *data)
37 {
38         NTSTATUS status;
39         status = imessaging_send(msg, src, msg_pong, data);
40         if (!NT_STATUS_IS_OK(status)) {
41                 printf("pong failed - %s\n", nt_errstr(status));
42         }
43 }
44
45 static void pong_message(struct imessaging_context *msg, void *private_data,
46                          uint32_t msg_type, struct server_id src, DATA_BLOB *data)
47 {
48         int *count = (int *)private_data;
49         (*count)++;
50 }
51
52 static void exit_message(struct imessaging_context *msg, void *private_data,
53                          uint32_t msg_type, struct server_id src, DATA_BLOB *data)
54 {
55         talloc_free(private_data);
56         exit(0);
57 }
58
59 /*
60   test ping speed
61 */
62 static bool test_ping_speed(struct torture_context *tctx)
63 {
64         struct tevent_context *ev;
65         struct imessaging_context *msg_client_ctx;
66         struct imessaging_context *msg_server_ctx;
67         int ping_count = 0;
68         int pong_count = 0;
69         struct timeval tv;
70         int timelimit = torture_setting_int(tctx, "timelimit", 10);
71         uint32_t msg_ping, msg_exit;
72
73         lpcfg_set_cmdline(tctx->lp_ctx, "pid directory", "piddir.tmp");
74
75         ev = tctx->ev;
76
77         msg_server_ctx = imessaging_init(tctx,
78                                          tctx->lp_ctx, cluster_id(0, 1),
79                                          ev);
80         
81         torture_assert(tctx, msg_server_ctx != NULL, "Failed to init ping messaging context");
82                 
83         imessaging_register_tmp(msg_server_ctx, NULL, ping_message, &msg_ping);
84         imessaging_register_tmp(msg_server_ctx, tctx, exit_message, &msg_exit);
85
86         msg_client_ctx = imessaging_init(tctx,
87                                          tctx->lp_ctx,
88                                          cluster_id(0, 2),
89                                          ev);
90
91         torture_assert(tctx, msg_client_ctx != NULL, 
92                        "msg_client_ctx imessaging_init() failed");
93
94         imessaging_register_tmp(msg_client_ctx, &pong_count, pong_message, &msg_pong);
95
96         tv = timeval_current();
97
98         torture_comment(tctx, "Sending pings for %d seconds\n", timelimit);
99         while (timeval_elapsed(&tv) < timelimit) {
100                 DATA_BLOB data;
101                 NTSTATUS status1, status2;
102
103                 data.data = discard_const_p(uint8_t, "testing");
104                 data.length = strlen((const char *)data.data);
105
106                 status1 = imessaging_send(msg_client_ctx, cluster_id(0, 1), msg_ping, &data);
107                 status2 = imessaging_send(msg_client_ctx, cluster_id(0, 1), msg_ping, NULL);
108
109                 torture_assert_ntstatus_ok(tctx, status1, "msg1 failed");
110                 ping_count++;
111
112                 torture_assert_ntstatus_ok(tctx, status2, "msg2 failed");
113                 ping_count++;
114
115                 while (ping_count > pong_count + 20) {
116                         tevent_loop_once(ev);
117                 }
118         }
119
120         torture_comment(tctx, "waiting for %d remaining replies (done %d)\n", 
121                ping_count - pong_count, pong_count);
122         while (timeval_elapsed(&tv) < 30 && pong_count < ping_count) {
123                 tevent_loop_once(ev);
124         }
125
126         torture_comment(tctx, "sending exit\n");
127         imessaging_send(msg_client_ctx, cluster_id(0, 1), msg_exit, NULL);
128
129         torture_assert_int_equal(tctx, ping_count, pong_count, "ping test failed");
130
131         torture_comment(tctx, "ping rate of %.0f messages/sec\n", 
132                (ping_count+pong_count)/timeval_elapsed(&tv));
133
134         talloc_free(msg_client_ctx);
135         talloc_free(msg_server_ctx);
136
137         return true;
138 }
139
140 static bool test_messaging_overflow(struct torture_context *tctx)
141 {
142         struct imessaging_context *msg_ctx;
143         ssize_t nwritten, nread;
144         pid_t child;
145         char c = 0;
146         int up_pipe[2], down_pipe[2];
147         int i, ret, child_status;
148
149         ret = pipe(up_pipe);
150         torture_assert(tctx, ret == 0, "pipe failed");
151         ret = pipe(down_pipe);
152         torture_assert(tctx, ret == 0, "pipe failed");
153
154         child = fork();
155         if (child < 0) {
156                 torture_fail(tctx, "fork failed");
157         }
158
159         if (child == 0) {
160                 ret = tevent_re_initialise(tctx->ev);
161                 torture_assert(tctx, ret == 0, "tevent_re_initialise failed");
162
163                 msg_ctx = imessaging_init(tctx, tctx->lp_ctx,
164                                           cluster_id(getpid(), 0),
165                                           tctx->ev);
166                 torture_assert(tctx, msg_ctx != NULL,
167                                "imessaging_init failed");
168
169                 do {
170                         nwritten = write(up_pipe[1], &c, 1);
171                 } while ((nwritten == -1) && (errno == EINTR));
172
173                 ret = close(down_pipe[1]);
174                 torture_assert(tctx, ret == 0, "close failed");
175
176                 do {
177                         nread = read(down_pipe[0], &c, 1);
178                 } while ((nread == -1) && (errno == EINTR));
179
180                 exit(0);
181         }
182
183         do {
184                 nread = read(up_pipe[0], &c, 1);
185         } while ((nread == -1) && (errno == EINTR));
186
187         msg_ctx = imessaging_init(tctx, tctx->lp_ctx, cluster_id(getpid(), 0),
188                                   tctx->ev);
189         torture_assert(tctx, msg_ctx != NULL, "imessaging_init failed");
190
191         for (i=0; i<1000; i++) {
192                 NTSTATUS status;
193                 status = imessaging_send(msg_ctx, cluster_id(child, 0),
194                                          MSG_PING, NULL);
195                 torture_assert_ntstatus_ok(tctx, status,
196                                            "imessaging_send failed");
197         }
198
199         tevent_loop_once(tctx->ev);
200
201         talloc_free(msg_ctx);
202
203         ret = close(down_pipe[1]);
204         torture_assert(tctx, ret == 0, "close failed");
205
206         ret = waitpid(child, &child_status, 0);
207         torture_assert(tctx, ret == child, "wrong child exited");
208         torture_assert(tctx, child_status == 0, "child failed");
209
210         poll(NULL, 0, 500);
211
212         return true;
213 }
214
215 struct overflow_parent_child {
216         MD5_CTX md5ctx;
217         bool done;
218 };
219
220 static void overflow_md5_child_handler(struct imessaging_context *msg,
221                                        void *private_data,
222                                        uint32_t msg_type,
223                                        struct server_id server_id,
224                                        DATA_BLOB *data)
225 {
226         struct overflow_parent_child *state = private_data;
227
228         if (data->length == 0) {
229                 state->done = true;
230                 return;
231         }
232
233         MD5Update(&state->md5ctx, data->data, data->length);
234 }
235
236 struct overflow_child_parent {
237         uint8_t final[16];
238         bool done;
239 };
240
241 static void overflow_md5_parent_handler(struct imessaging_context *msg_ctx,
242                                         void *private_data,
243                                         uint32_t msg_type,
244                                         struct server_id server_id,
245                                         DATA_BLOB *data)
246 {
247         struct overflow_child_parent *state = private_data;
248
249         if (data->length != sizeof(state->final)) {
250                 memset(state->final, 0, sizeof(state->final));
251                 state->done = true;
252                 return;
253         }
254         memcpy(state->final, data->data, 16);
255         state->done = true;
256 }
257
258 static bool test_messaging_overflow_check(struct torture_context *tctx)
259 {
260         struct imessaging_context *msg_ctx;
261         ssize_t nwritten, nread;
262         pid_t child;
263         char c = 0;
264         int up_pipe[2], down_pipe[2];
265         int i, ret, child_status;
266         MD5_CTX md5ctx;
267         uint8_t final[16];
268         struct overflow_child_parent child_msg = { .done = false };
269         NTSTATUS status;
270
271         ret = pipe(up_pipe);
272         torture_assert(tctx, ret == 0, "pipe failed");
273         ret = pipe(down_pipe);
274         torture_assert(tctx, ret == 0, "pipe failed");
275
276         child = fork();
277         if (child < 0) {
278                 torture_fail(tctx, "fork failed");
279         }
280
281         if (child == 0) {
282                 struct overflow_parent_child child_state = { .done = false };
283                 DATA_BLOB retblob = { .data = final, .length = sizeof(final) };
284
285                 ret = tevent_re_initialise(tctx->ev);
286                 torture_assert(tctx, ret == 0, "tevent_re_initialise failed");
287
288                 MD5Init(&child_state.md5ctx);
289
290                 msg_ctx = imessaging_init(tctx, tctx->lp_ctx,
291                                           cluster_id(getpid(), 0),
292                                           tctx->ev);
293                 torture_assert(tctx, msg_ctx != NULL,
294                                "imessaging_init failed");
295
296                 status = imessaging_register(msg_ctx, &child_state,
297                                              MSG_TMP_BASE-1,
298                                              overflow_md5_child_handler);
299                 torture_assert(tctx, NT_STATUS_IS_OK(status),
300                                "imessaging_register failed");
301
302                 do {
303                         nwritten = write(up_pipe[1], &c, 1);
304                 } while ((nwritten == -1) && (errno == EINTR));
305
306                 ret = close(down_pipe[1]);
307                 torture_assert(tctx, ret == 0, "close failed");
308
309                 do {
310                         nread = read(down_pipe[0], &c, 1);
311                 } while ((nread == -1) && (errno == EINTR));
312
313                 while (!child_state.done) {
314                         tevent_loop_once(tctx->ev);
315                 }
316
317                 MD5Final(final, &child_state.md5ctx);
318
319                 status = imessaging_send(msg_ctx,
320                                          cluster_id(getppid(), 0),
321                                          MSG_TMP_BASE-2,
322                                          &retblob);
323                 torture_assert(tctx, NT_STATUS_IS_OK(status),
324                                "imessaging_send failed");
325
326                 exit(0);
327         }
328
329         do {
330                 nread = read(up_pipe[0], &c, 1);
331         } while ((nread == -1) && (errno == EINTR));
332
333         msg_ctx = imessaging_init(tctx, tctx->lp_ctx, cluster_id(getpid(), 0),
334                                   tctx->ev);
335         torture_assert(tctx, msg_ctx != NULL, "imessaging_init failed");
336
337         status = imessaging_register(msg_ctx,
338                                      &child_msg,
339                                      MSG_TMP_BASE-2,
340                                      overflow_md5_parent_handler);
341         torture_assert(tctx,
342                        NT_STATUS_IS_OK(status),
343                        "imessaging_register failed");
344
345         MD5Init(&md5ctx);
346
347         for (i=0; i<1000; i++) {
348                 size_t len = ((random() % 100) + 1);
349                 uint8_t buf[len];
350                 DATA_BLOB blob = { .data = buf, .length = len };
351
352                 generate_random_buffer(buf, len);
353
354                 MD5Update(&md5ctx, buf, len);
355
356                 status = imessaging_send(msg_ctx, cluster_id(child, 0),
357                                          MSG_TMP_BASE-1, &blob);
358                 torture_assert_ntstatus_ok(tctx, status,
359                                            "imessaging_send failed");
360         }
361
362         status = imessaging_send(msg_ctx, cluster_id(child, 0),
363                                  MSG_TMP_BASE-1, NULL);
364         torture_assert_ntstatus_ok(tctx, status,
365                                    "imessaging_send failed");
366
367         MD5Final(final, &md5ctx);
368
369         do {
370                 nwritten = write(down_pipe[1], &c, 1);
371         } while ((nwritten == -1) && (errno == EINTR));
372
373         while (!child_msg.done) {
374                 tevent_loop_once(tctx->ev);
375         }
376
377         ret = close(down_pipe[1]);
378         torture_assert(tctx, ret == 0, "close failed");
379
380         talloc_free(msg_ctx);
381
382         ret = waitpid(child, &child_status, 0);
383         torture_assert(tctx, ret == child, "wrong child exited");
384         torture_assert(tctx, child_status == 0, "child failed");
385
386         if (memcmp(final, child_msg.final, 16) != 0) {
387                 dump_data_file(final, 16, false, stderr);
388                 dump_data_file(child_msg.final, 16, false, stderr);
389                 fflush(stderr);
390                 torture_fail(tctx, "checksum comparison failed");
391         }
392
393         return true;
394 }
395
396 struct test_multi_ctx {
397         struct torture_context *tctx;
398         struct imessaging_context *server_ctx;
399         struct imessaging_context *client_ctx[4];
400         size_t num_missing;
401         bool got_server;
402         bool got_client_0_1;
403         bool got_client_2_3;
404         bool ok;
405 };
406
407 static void multi_ctx_server_handler(struct imessaging_context *msg,
408                                      void *private_data,
409                                      uint32_t msg_type,
410                                      struct server_id server_id,
411                                      DATA_BLOB *data)
412 {
413         struct test_multi_ctx *state = private_data;
414         char *str = NULL;
415
416         torture_assert_goto(state->tctx, state->num_missing >= 1,
417                             state->ok, fail,
418                             "num_missing should be at least 1.");
419         state->num_missing -= 1;
420
421         torture_assert_goto(state->tctx, !state->got_server,
422                             state->ok, fail,
423                             "already got server.");
424         state->got_server = true;
425
426         /*
427          * We free the context itself and most likely reuse
428          * the memory immediately.
429          */
430         TALLOC_FREE(state->server_ctx);
431         str = generate_random_str(state->tctx, 128);
432         torture_assert_goto(state->tctx, str != NULL,
433                             state->ok, fail,
434                             "generate_random_str()");
435
436 fail:
437         return;
438 }
439
440 static void multi_ctx_client_0_1_handler(struct imessaging_context *msg,
441                                          void *private_data,
442                                          uint32_t msg_type,
443                                          struct server_id server_id,
444                                          DATA_BLOB *data)
445 {
446         struct test_multi_ctx *state = private_data;
447         char *str = NULL;
448
449         torture_assert_goto(state->tctx, state->num_missing >= 2,
450                             state->ok, fail,
451                             "num_missing should be at least 2.");
452         state->num_missing -= 2;
453
454         torture_assert_goto(state->tctx, !state->got_client_0_1,
455                             state->ok, fail,
456                             "already got client_0_1.");
457         state->got_client_0_1 = true;
458
459         /*
460          * We free two contexts and most likely reuse
461          * the memory immediately.
462          */
463         TALLOC_FREE(state->client_ctx[0]);
464         str = generate_random_str(state->tctx, 128);
465         torture_assert_goto(state->tctx, str != NULL,
466                             state->ok, fail,
467                             "generate_random_str()");
468         TALLOC_FREE(state->client_ctx[1]);
469         str = generate_random_str(state->tctx, 128);
470         torture_assert_goto(state->tctx, str != NULL,
471                             state->ok, fail,
472                             "generate_random_str()");
473
474 fail:
475         return;
476 }
477
478 static void multi_ctx_client_2_3_handler(struct imessaging_context *msg,
479                                          void *private_data,
480                                          uint32_t msg_type,
481                                          struct server_id server_id,
482                                          DATA_BLOB *data)
483 {
484         struct test_multi_ctx *state = private_data;
485         char *str = NULL;
486
487         torture_assert_goto(state->tctx, state->num_missing >= 2,
488                             state->ok, fail,
489                             "num_missing should be at least 2.");
490         state->num_missing -= 2;
491
492         torture_assert_goto(state->tctx, !state->got_client_2_3,
493                             state->ok, fail,
494                             "already got client_2_3.");
495         state->got_client_2_3 = true;
496
497         /*
498          * We free two contexts and most likely reuse
499          * the memory immediately.
500          */
501         TALLOC_FREE(state->client_ctx[2]);
502         str = generate_random_str(state->tctx, 128);
503         torture_assert_goto(state->tctx, str != NULL,
504                             state->ok, fail,
505                             "generate_random_str()");
506         TALLOC_FREE(state->client_ctx[3]);
507         str = generate_random_str(state->tctx, 128);
508         torture_assert_goto(state->tctx, str != NULL,
509                             state->ok, fail,
510                             "generate_random_str()");
511
512 fail:
513         return;
514 }
515
516 static bool test_multi_ctx(struct torture_context *tctx)
517 {
518         struct test_multi_ctx state = {
519                 .tctx = tctx,
520                 .ok = true,
521         };
522         struct timeval tv;
523         NTSTATUS status;
524
525         lpcfg_set_cmdline(tctx->lp_ctx, "pid directory", "piddir.tmp");
526
527         /*
528          * We use cluster_id(0, 0) as that gets for
529          * all task ids.
530          */
531         state.server_ctx = imessaging_init(tctx,
532                                            tctx->lp_ctx,
533                                            cluster_id(0, 0),
534                                            tctx->ev);
535         torture_assert(tctx, state.server_ctx != NULL,
536                        "Failed to init messaging context");
537
538         status = imessaging_register(state.server_ctx, &state,
539                                      MSG_TMP_BASE-1,
540                                      multi_ctx_server_handler);
541         torture_assert(tctx, NT_STATUS_IS_OK(status), "imessaging_register failed");
542
543         state.client_ctx[0] = imessaging_init(tctx,
544                                               tctx->lp_ctx,
545                                               cluster_id(0, 0),
546                                               tctx->ev);
547         torture_assert(tctx, state.client_ctx[0] != NULL,
548                        "msg_client_ctx imessaging_init() failed");
549         status = imessaging_register(state.client_ctx[0], &state,
550                                      MSG_TMP_BASE-1,
551                                      multi_ctx_client_0_1_handler);
552         torture_assert(tctx, NT_STATUS_IS_OK(status), "imessaging_register failed");
553         state.client_ctx[1] = imessaging_init(tctx,
554                                               tctx->lp_ctx,
555                                               cluster_id(0, 0),
556                                               tctx->ev);
557         torture_assert(tctx, state.client_ctx[1] != NULL,
558                        "msg_client_ctx imessaging_init() failed");
559         status = imessaging_register(state.client_ctx[1], &state,
560                                      MSG_TMP_BASE-1,
561                                      multi_ctx_client_0_1_handler);
562         torture_assert(tctx, NT_STATUS_IS_OK(status), "imessaging_register failed");
563         state.client_ctx[2] = imessaging_init(tctx,
564                                               tctx->lp_ctx,
565                                               cluster_id(0, 0),
566                                               tctx->ev);
567         torture_assert(tctx, state.client_ctx[2] != NULL,
568                        "msg_client_ctx imessaging_init() failed");
569         status = imessaging_register(state.client_ctx[2], &state,
570                                      MSG_TMP_BASE-1,
571                                      multi_ctx_client_2_3_handler);
572         torture_assert(tctx, NT_STATUS_IS_OK(status), "imessaging_register failed");
573         state.client_ctx[3] = imessaging_init(tctx,
574                                               tctx->lp_ctx,
575                                               cluster_id(0, 0),
576                                               tctx->ev);
577         torture_assert(tctx, state.client_ctx[3] != NULL,
578                        "msg_client_ctx imessaging_init() failed");
579         status = imessaging_register(state.client_ctx[3], &state,
580                                      MSG_TMP_BASE-1,
581                                      multi_ctx_client_2_3_handler);
582         torture_assert(tctx, NT_STATUS_IS_OK(status), "imessaging_register failed");
583
584         /*
585          * Send one message that need to arrive on 3 ( 5 - 2 ) handlers.
586          */
587         state.num_missing = 5;
588
589         status = imessaging_send(state.server_ctx,
590                                  cluster_id(0, 0),
591                                  MSG_TMP_BASE-1, NULL);
592         torture_assert_ntstatus_ok(tctx, status, "msg failed");
593
594         tv = timeval_current();
595         while (timeval_elapsed(&tv) < 30 && state.num_missing > 0 && state.ok) {
596                 int ret;
597
598                 ret = tevent_loop_once(tctx->ev);
599                 torture_assert_int_equal(tctx, ret, 0, "tevent_loop_once()");
600         }
601
602         if (!state.ok) {
603                 return false;
604         }
605
606         torture_assert_int_equal(tctx, state.num_missing, 0,
607                                  "wrong message count");
608
609         torture_assert(tctx, state.got_client_0_1, "got_client_0_1");
610         torture_assert(tctx, state.got_client_2_3, "got_client_2_3");
611         torture_assert(tctx, state.got_server, "got_server");
612
613         return true;
614 }
615
616 struct torture_suite *torture_local_messaging(TALLOC_CTX *mem_ctx)
617 {
618         struct torture_suite *s = torture_suite_create(mem_ctx, "messaging");
619         torture_suite_add_simple_test(s, "overflow", test_messaging_overflow);
620         torture_suite_add_simple_test(s, "overflow_check",
621                                       test_messaging_overflow_check);
622         torture_suite_add_simple_test(s, "ping_speed", test_ping_speed);
623         torture_suite_add_simple_test(s, "multi_ctx", test_multi_ctx);
624         return s;
625 }