b7f2a813917b3ccf376d0c6bc2cbf4f8253adc56
[oweals/gnunet.git] / src / core / test_core_api_reliability.c
1 /*
2      This file is part of GNUnet.
3      (C) 2009, 2010 Christian Grothoff (and other contributing authors)
4
5      GNUnet is free software; you can redistribute it and/or modify
6      it under the terms of the GNU General Public License as published
7      by the Free Software Foundation; either version 3, or (at your
8      option) any later version.
9
10      GNUnet is distributed in the hope that it will be useful, but
11      WITHOUT ANY WARRANTY; without even the implied warranty of
12      MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
13      General Public License for more details.
14
15      You should have received a copy of the GNU General Public License
16      along with GNUnet; see the file COPYING.  If not, write to the
17      Free Software Foundation, Inc., 59 Temple Place - Suite 330,
18      Boston, MA 02111-1307, USA.
19 */
20 /**
21  * @file core/test_core_api_reliability.c
22  * @brief testcase for core_api.c focusing on reliable transmission (with TCP)
23  *
24  * FIXME:
25  * - make sure connect callback is invoked properly as well!
26  */
27 #include "platform.h"
28 #include "gnunet_common.h"
29 #include "gnunet_constants.h"
30 #include "gnunet_arm_service.h"
31 #include "gnunet_core_service.h"
32 #include "gnunet_getopt_lib.h"
33 #include "gnunet_os_lib.h"
34 #include "gnunet_program_lib.h"
35 #include "gnunet_scheduler_lib.h"
36 #include "gnunet_transport_service.h"
37 #include <gauger.h>
38
39 #define VERBOSE GNUNET_NO
40
41 #define START_ARM GNUNET_YES
42
43 /**
44  * Note that this value must not significantly exceed
45  * 'MAX_PENDING' in 'gnunet-service-transport.c', otherwise
46  * messages may be dropped even for a reliable transport.
47  */
48 #define TOTAL_MSGS (600 * 10)
49
50 /**
51  * How long until we give up on transmitting the message?
52  */
53 #define TIMEOUT GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 6000)
54
55 /**
56  * What delay do we request from the core service for transmission?
57  * Any value smaller than the CORK delay will disable CORKing, which
58  * is what we want here.
59  */
60 #define FAST_TIMEOUT GNUNET_TIME_relative_divide (GNUNET_CONSTANTS_MAX_CORK_DELAY, 2)
61
62 #define MTYPE 12345
63
64
65 static unsigned long long total_bytes;
66
67 static struct GNUNET_TIME_Absolute start_time;
68
69 static GNUNET_SCHEDULER_TaskIdentifier err_task;
70
71
72 struct PeerContext
73 {
74   struct GNUNET_CONFIGURATION_Handle *cfg;
75   struct GNUNET_CORE_Handle *ch;
76   struct GNUNET_PeerIdentity id;   
77   struct GNUNET_TRANSPORT_Handle *th;
78   struct GNUNET_MessageHeader *hello;
79   int connect_status;
80 #if START_ARM
81   struct GNUNET_OS_Process *arm_proc;
82 #endif
83 };
84
85 static struct PeerContext p1;
86
87 static struct PeerContext p2;
88
89 static int ok;
90
91 static int32_t tr_n;
92
93
94 #if VERBOSE
95 #define OKPP do { ok++; fprintf (stderr, "Now at stage %u at %s:%u\n", ok, __FILE__, __LINE__); } while (0)
96 #else
97 #define OKPP do { ok++; } while (0)
98 #endif
99
100 struct TestMessage 
101 {
102   struct GNUNET_MessageHeader header;
103   uint32_t num;
104 };
105
106
107 static unsigned int
108 get_size (unsigned int iter)
109 {
110   unsigned int ret;
111   if (iter < 60000)
112     return iter + sizeof (struct TestMessage);
113   ret = (iter * iter * iter);
114   return sizeof (struct TestMessage) + (ret % 60000);
115 }
116
117
118 static void
119 terminate_task (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
120 {
121   unsigned long long delta;
122
123   GNUNET_CORE_disconnect (p1.ch);
124   p1.ch = NULL;
125   GNUNET_CORE_disconnect (p2.ch);
126   p2.ch = NULL;
127   GNUNET_TRANSPORT_disconnect (p1.th);
128   p1.th = NULL;
129   GNUNET_TRANSPORT_disconnect (p2.th);
130   p2.th = NULL;
131   delta = GNUNET_TIME_absolute_get_duration (start_time).rel_value;
132   fprintf (stderr,
133            "\nThroughput was %llu kb/s\n",
134            total_bytes * 1000 / 1024 / delta);
135   GAUGER ("CORE", "Core throughput/s", total_bytes * 1000 / 1024 / delta, "kb/s");
136   ok = 0;
137 }
138
139
140 static void
141 terminate_task_error (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
142 {
143   GNUNET_break (0);
144   GNUNET_CORE_disconnect (p1.ch);
145   p1.ch = NULL;
146   GNUNET_CORE_disconnect (p2.ch);
147   p2.ch = NULL;
148   GNUNET_TRANSPORT_disconnect (p1.th);
149   p1.th = NULL;
150   GNUNET_TRANSPORT_disconnect (p2.th);
151   p2.th = NULL;
152   ok = 42;
153 }
154
155
156 static size_t
157 transmit_ready (void *cls, size_t size, void *buf)
158 {
159   char *cbuf = buf;
160   struct TestMessage hdr;
161   unsigned int s;
162   unsigned int ret;
163
164   GNUNET_assert (size <= GNUNET_CONSTANTS_MAX_ENCRYPTED_MESSAGE_SIZE); 
165   if (buf == NULL)
166     {
167       if (p1.ch != NULL)
168         GNUNET_break (NULL != 
169                       GNUNET_CORE_notify_transmit_ready (p1.ch,
170                                                          0,
171                                                          FAST_TIMEOUT,
172                                                          &p2.id,
173                                                          get_size(tr_n),
174                                                          &transmit_ready, &p1));
175       return 0;
176     }
177   GNUNET_assert (tr_n < TOTAL_MSGS);
178   ret = 0;
179   s = get_size (tr_n);
180   GNUNET_assert (size >= s);
181   GNUNET_assert (buf != NULL);
182   cbuf = buf;
183   do
184     {
185 #if VERBOSE
186       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
187                   "Sending message %u of size %u at offset %u\n",
188                   tr_n,
189                   s,
190                   ret);
191 #endif
192       hdr.header.size = htons (s);
193       hdr.header.type = htons (MTYPE);
194       hdr.num = htonl (tr_n);
195       memcpy (&cbuf[ret], &hdr, sizeof (struct TestMessage));
196       ret += sizeof (struct TestMessage);
197       memset (&cbuf[ret], tr_n, s - sizeof (struct TestMessage));
198       ret += s - sizeof (struct TestMessage);
199       tr_n++;
200       s = get_size (tr_n);
201       if (0 == GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, 16))
202         break; /* sometimes pack buffer full, sometimes not */
203     }
204   while (size - ret >= s);
205   GNUNET_SCHEDULER_cancel (err_task);
206   err_task = 
207     GNUNET_SCHEDULER_add_delayed (TIMEOUT,
208                                   &terminate_task_error, 
209                                   NULL);
210   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
211               "Returning total message block of size %u\n",
212               ret);
213   total_bytes += ret;
214   return ret;
215 }
216
217
218
219 static void
220 connect_notify (void *cls,
221                 const struct GNUNET_PeerIdentity *peer,
222                 const struct GNUNET_TRANSPORT_ATS_Information *atsi)
223 {
224   struct PeerContext *pc = cls;
225
226   if (0 == memcmp (&pc->id,
227                    peer,
228                    sizeof (struct GNUNET_PeerIdentity)))
229     return;
230   GNUNET_assert (pc->connect_status == 0);
231   pc->connect_status = 1;
232   if (pc == &p1)
233     {
234       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
235                   "Encrypted connection established to peer `%4s'\n",
236                   GNUNET_i2s (peer));
237       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
238                   "Asking core (1) for transmission to peer `%4s'\n",
239                   GNUNET_i2s (&p2.id));
240       err_task = 
241         GNUNET_SCHEDULER_add_delayed (TIMEOUT,
242                                       &terminate_task_error, 
243                                       NULL);
244       start_time = GNUNET_TIME_absolute_get ();
245       GNUNET_break (NULL != 
246                     GNUNET_CORE_notify_transmit_ready (p1.ch,
247                                                        0,
248                                                        TIMEOUT,
249                                                        &p2.id,
250                                                        get_size (0),
251                                                        &transmit_ready, &p1));
252     }
253 }
254
255
256 static void
257 disconnect_notify (void *cls,
258                    const struct GNUNET_PeerIdentity *peer)
259 {
260   struct PeerContext *pc = cls;
261
262   if (0 == memcmp (&pc->id,
263                    peer,
264                    sizeof (struct GNUNET_PeerIdentity)))
265     return;
266   pc->connect_status = 0;
267   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
268               "Encrypted connection to `%4s' cut\n", GNUNET_i2s (peer));
269 }
270
271
272 static int
273 inbound_notify (void *cls,
274                 const struct GNUNET_PeerIdentity *other,
275                 const struct GNUNET_MessageHeader *message,
276                 const struct GNUNET_TRANSPORT_ATS_Information *atsi)
277 {
278 #if VERBOSE
279   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
280               "Core provides inbound data from `%4s'.\n", GNUNET_i2s (other));
281 #endif
282   return GNUNET_OK;
283 }
284
285
286 static int
287 outbound_notify (void *cls,
288                  const struct GNUNET_PeerIdentity *other,
289                  const struct GNUNET_MessageHeader *message,
290                  const struct GNUNET_TRANSPORT_ATS_Information *atsi)
291 {
292 #if VERBOSE
293   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
294               "Core notifies about outbound data for `%4s'.\n",
295               GNUNET_i2s (other));
296 #endif
297   return GNUNET_OK;
298 }
299
300
301 static size_t
302 transmit_ready (void *cls, size_t size, void *buf);
303
304 static int
305 process_mtype (void *cls,
306                const struct GNUNET_PeerIdentity *peer,
307                const struct GNUNET_MessageHeader *message,
308                const struct GNUNET_TRANSPORT_ATS_Information *atsi)
309 {
310   static int n;
311   unsigned int s;
312   const struct TestMessage *hdr;
313
314   hdr = (const struct TestMessage*) message;
315   s = get_size (n);
316   if (MTYPE != ntohs (message->type))
317     return GNUNET_SYSERR;
318   if (ntohs (message->size) != s)
319     {
320       GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
321                   "Expected message %u of size %u, got %u bytes of message %u\n",
322                   n, s,
323                   ntohs (message->size),
324                   ntohl (hdr->num));
325       GNUNET_SCHEDULER_cancel (err_task);
326       err_task = GNUNET_SCHEDULER_add_now (&terminate_task_error, NULL);
327       return GNUNET_SYSERR;
328     }
329   if (ntohl (hdr->num) != n)
330     {
331       GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
332                   "Expected message %u of size %u, got %u bytes of message %u\n",
333                   n, s,
334                   ntohs (message->size),
335                   ntohl (hdr->num));
336       GNUNET_SCHEDULER_cancel (err_task);
337       err_task = GNUNET_SCHEDULER_add_now (&terminate_task_error, NULL);
338       return GNUNET_SYSERR;
339     }
340 #if VERBOSE
341   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
342               "Got message %u of size %u\n",
343               ntohl (hdr->num),
344               ntohs (message->size));         
345 #endif
346   n++;
347   if (0 == (n % (TOTAL_MSGS/100)))
348     fprintf (stderr, ".");
349   if (n == TOTAL_MSGS)
350     {
351       GNUNET_SCHEDULER_cancel (err_task);
352       GNUNET_SCHEDULER_add_now (&terminate_task, NULL);
353     }
354   else
355     {
356       if (n == tr_n)
357         GNUNET_break (NULL != 
358                       GNUNET_CORE_notify_transmit_ready (p1.ch,
359                                                          0,
360                                                          FAST_TIMEOUT,
361                                                          &p2.id,
362                                                          get_size(tr_n),
363                                                          &transmit_ready, &p1));
364     }
365   return GNUNET_OK;
366 }
367
368
369 static struct GNUNET_CORE_MessageHandler handlers[] = {
370   {&process_mtype, MTYPE, 0},
371   {NULL, 0, 0}
372 };
373
374
375
376 static void
377 init_notify (void *cls,
378              struct GNUNET_CORE_Handle *server,
379              const struct GNUNET_PeerIdentity *my_identity,
380              const struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded *publicKey)
381 {
382   struct PeerContext *p = cls;
383
384   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
385               "Connection to CORE service of `%4s' established\n",
386               GNUNET_i2s (my_identity));
387   GNUNET_assert (server != NULL);
388   p->id = *my_identity;
389   p->ch = server;
390   if (cls == &p1)
391     {
392       GNUNET_assert (ok == 2);
393       OKPP;
394       /* connect p2 */
395       GNUNET_CORE_connect (p2.cfg, 1,
396                            &p2,
397                            &init_notify,                         
398                            &connect_notify,
399                            &disconnect_notify,
400                            NULL,
401                            &inbound_notify,
402                            GNUNET_YES,
403                            &outbound_notify, GNUNET_YES, handlers);
404     }
405   else
406     {
407       GNUNET_assert (ok == 3);
408       OKPP;
409       GNUNET_assert (cls == &p2);
410       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
411                   "Asking core (1) to connect to peer `%4s'\n",
412                   GNUNET_i2s (&p2.id));
413       GNUNET_CORE_peer_request_connect (p1.ch,
414                                         GNUNET_TIME_UNIT_SECONDS,
415                                         &p2.id,
416                                         NULL, NULL);
417     }
418 }
419
420
421 static void
422 process_hello (void *cls,
423                const struct GNUNET_MessageHeader *message)
424 {
425   struct PeerContext *p = cls;
426
427   GNUNET_TRANSPORT_get_hello_cancel (p->th, &process_hello, p);
428   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
429               "Received (my) `%s' from transport service\n",
430               "HELLO");
431   GNUNET_assert (message != NULL);
432   p->hello = GNUNET_malloc (ntohs (message->size));
433   memcpy (p->hello, message, ntohs (message->size));
434   if ((p == &p1) && (p2.th != NULL))
435     GNUNET_TRANSPORT_offer_hello (p2.th, message, NULL, NULL);
436   if ((p == &p2) && (p1.th != NULL))
437     GNUNET_TRANSPORT_offer_hello (p1.th, message, NULL, NULL);
438
439   if ((p == &p1) && (p2.hello != NULL))
440     GNUNET_TRANSPORT_offer_hello (p1.th, p2.hello, NULL, NULL);
441   if ((p == &p2) && (p1.hello != NULL))
442     GNUNET_TRANSPORT_offer_hello (p2.th, p1.hello, NULL, NULL);
443 }
444
445
446
447 static void
448 setup_peer (struct PeerContext *p, const char *cfgname)
449 {
450   p->cfg = GNUNET_CONFIGURATION_create ();
451 #if START_ARM
452   p->arm_proc = GNUNET_OS_start_process (NULL, NULL, "gnunet-service-arm",
453                                         "gnunet-service-arm",
454 #if VERBOSE
455                                         "-L", "DEBUG",
456 #endif
457                                         "-c", cfgname, NULL);
458 #endif
459   GNUNET_assert (GNUNET_OK == GNUNET_CONFIGURATION_load (p->cfg, cfgname));
460   p->th = GNUNET_TRANSPORT_connect (p->cfg, NULL, p, NULL, NULL, NULL);
461   GNUNET_assert (p->th != NULL);
462   GNUNET_TRANSPORT_get_hello (p->th, &process_hello, p);
463 }
464
465
466 static void
467 run (void *cls,
468      char *const *args,
469      const char *cfgfile,
470      const struct GNUNET_CONFIGURATION_Handle *cfg)
471 {
472   GNUNET_assert (ok == 1);
473   OKPP;
474   setup_peer (&p1, "test_core_api_peer1.conf");
475   setup_peer (&p2, "test_core_api_peer2.conf");
476   GNUNET_CORE_connect (p1.cfg, 1,
477                        &p1,
478                        &init_notify,
479                        &connect_notify,
480                        &disconnect_notify,
481                        NULL,
482                        &inbound_notify,
483                        GNUNET_YES, &outbound_notify, GNUNET_YES, handlers);
484 }
485
486
487 static void
488 stop_arm (struct PeerContext *p)
489 {
490 #if START_ARM
491   if (0 != GNUNET_OS_process_kill (p->arm_proc, SIGTERM))
492     GNUNET_log_strerror (GNUNET_ERROR_TYPE_WARNING, "kill");
493   if (GNUNET_OS_process_wait(p->arm_proc) != GNUNET_OK)
494     GNUNET_log_strerror (GNUNET_ERROR_TYPE_WARNING, "waitpid");
495   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
496               "ARM process %u stopped\n", GNUNET_OS_process_get_pid (p->arm_proc));
497   GNUNET_OS_process_close (p->arm_proc);
498   p->arm_proc = NULL;
499 #endif
500   GNUNET_CONFIGURATION_destroy (p->cfg);
501 }
502
503 static int
504 check ()
505 {
506   char *const argv[] = { "test-core-api-reliability",
507     "-c",
508     "test_core_api_data.conf",
509 #if VERBOSE
510     "-L", "DEBUG",
511 #endif
512     NULL
513   };
514   struct GNUNET_GETOPT_CommandLineOption options[] = {
515     GNUNET_GETOPT_OPTION_END
516   };
517   ok = 1;
518   GNUNET_PROGRAM_run ((sizeof (argv) / sizeof (char *)) - 1,
519                       argv, "test-core-api-reliability", "nohelp", options, &run, &ok);
520   stop_arm (&p1);
521   stop_arm (&p2);
522   return ok;
523 }
524
525 int
526 main (int argc, char *argv[])
527 {
528   int ret;
529
530   GNUNET_log_setup ("test-core-api",
531 #if VERBOSE
532                     "DEBUG",
533 #else
534                     "WARNING",
535 #endif
536                     NULL);
537   ret = check ();
538   GNUNET_DISK_directory_remove ("/tmp/test-gnunet-core-peer-1"); 
539   GNUNET_DISK_directory_remove ("/tmp/test-gnunet-core-peer-2");
540
541   return ret;
542 }
543
544 /* end of test_core_api_reliability.c */