src/cadet/test_cadet.c

   1 /*
   2      This file is part of GNUnet.
   3      Copyright (C) 2011, 2017 GNUnet e.V.
   4
   5      GNUnet is free software: you can redistribute it and/or modify it
   6      under the terms of the GNU Affero General Public License as published
   7      by the Free Software Foundation, either version 3 of the License,
   8      or (at your 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      Affero General Public License for more details.
  14
  15      You should have received a copy of the GNU Affero General Public License
  16      along with this program.  If not, see <http://www.gnu.org/licenses/>.
  17
  18      SPDX-License-Identifier: AGPL3.0-or-later
  19 */
  20 /**
  21  * @file cadet/test_cadet.c
  22  * @author Bart Polot
  23  * @author Christian Grothoff
  24  * @brief Test for the cadet service using mq API.
  25  */
  26 #include <stdio.h>
  27 #include "platform.h"
  28 #include "cadet_test_lib.h"
  29 #include "gnunet_cadet_service.h"
  30 #include "gnunet_statistics_service.h"
  31 #include <gauger.h>
  32
  33
  34 /**
  35  * Ugly workaround to unify data handlers on incoming and outgoing channels.
  36  */
  37 struct CadetTestChannelWrapper
  38 {
  39   /**
  40    * Channel pointer.
  41    */
  42   struct GNUNET_CADET_Channel *ch;
  43 };
  44
  45 /**
  46  * How many messages to send by default.
  47  */
  48 #define TOTAL_PACKETS 500       /* Cannot exceed 64k! */
  49
  50 /**
  51  * How long until we give up on connecting the peers?
  52  */
  53 #define TIMEOUT GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 120)
  54
  55 /**
  56  * Time to wait by default  for stuff that should be rather fast.
  57  */
  58 #define SHORT_TIME GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 20)
  59
  60 /**
  61  * How fast do we send messages?
  62  */
  63 #define SEND_INTERVAL GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MILLISECONDS, 10)
  64
  65 /**
  66  * DIFFERENT TESTS TO RUN
  67  */
  68 #define SETUP 0
  69 #define FORWARD 1
  70 #define KEEPALIVE 2
  71 #define SPEED 3
  72 #define SPEED_ACK 4
  73 #define SPEED_REL 8
  74 #define P2P_SIGNAL 10
  75 #define REOPEN 11
  76
  77 /**
  78  * Which test are we running?
  79  */
  80 static int test;
  81
  82 /**
  83  * String with test name
  84  */
  85 static char *test_name;
  86
  87 /**
  88  * Flag to send traffic leaf->root in speed tests to test BCK_ACK logic.
  89  */
  90 static int test_backwards = GNUNET_NO;
  91
  92 /**
  93  * How many packets to send.
  94  */
  95 static unsigned int total_packets;
  96
  97 /**
  98  * Time to wait for fast operations.
  99  */
 100 static struct GNUNET_TIME_Relative short_time;
 101
 102 /**
 103  * How many events have happened
 104  */
 105 static int ok;
 106
 107 /**
 108  * Number of events expected to conclude the test successfully.
 109  */
 110 static int ok_goal;
 111
 112 /**
 113  * Size of each test packet's payload
 114  */
 115 static size_t size_payload = sizeof (uint32_t);
 116
 117 /**
 118  * Operation to get peer ids.
 119  */
 120 static struct GNUNET_TESTBED_Operation *t_op[2];
 121
 122 /**
 123  * Peer ids.
 124  */
 125 static struct GNUNET_PeerIdentity *p_id[2];
 126
 127 /**
 128  * Port ID
 129  */
 130 static struct GNUNET_HashCode port;
 131
 132 /**
 133  * Peer ids counter.
 134  */
 135 static unsigned int p_ids;
 136
 137 /**
 138  * Is the setup initialized?
 139  */
 140 static int initialized;
 141
 142 /**
 143  * Number of payload packes sent.
 144  */
 145 static int data_sent;
 146
 147 /**
 148  * Number of payload packets received.
 149  */
 150 static int data_received;
 151
 152 /**
 153  * Number of payload packed acknowledgements sent.
 154  */
 155 static int ack_sent;
 156
 157 /**
 158  * Number of payload packed explicitly (app level) acknowledged.
 159  */
 160 static int ack_received;
 161
 162 /**
 163  * Total number of peers asked to run.
 164  */
 165 static unsigned long long peers_requested;
 166
 167 /**
 168  * Number of currently running peers (should be same as @c peers_requested).
 169  */
 170 static unsigned long long peers_running;
 171
 172 /**
 173  * Test context (to shut down).
 174  */
 175 struct GNUNET_CADET_TEST_Context *test_ctx;
 176
 177 /**
 178  * Task called to disconnect peers.
 179  */
 180 static struct GNUNET_SCHEDULER_Task *disconnect_task;
 181
 182 /**
 183  * Task called to reconnect peers.
 184  */
 185 static struct GNUNET_SCHEDULER_Task *reconnect_task;
 186
 187 /**
 188  * Task To perform tests
 189  */
 190 static struct GNUNET_SCHEDULER_Task *test_task;
 191
 192 /**
 193  * Task runnining #send_next_msg().
 194  */
 195 static struct GNUNET_SCHEDULER_Task *send_next_msg_task;
 196
 197 /**
 198  * Cadet handle for the root peer
 199  */
 200 static struct GNUNET_CADET_Handle *h1;
 201
 202 /**
 203  * Cadet handle for the first leaf peer
 204  */
 205 static struct GNUNET_CADET_Handle *h2;
 206
 207 /**
 208  * Channel handle for the root peer
 209  */
 210 static struct GNUNET_CADET_Channel *outgoing_ch;
 211
 212 /**
 213  * Channel handle for the dest peer
 214  */
 215 static struct GNUNET_CADET_Channel *incoming_ch;
 216
 217 /**
 218  * Time we started the data transmission (after channel has been established
 219  * and initilized).
 220  */
 221 static struct GNUNET_TIME_Absolute start_time;
 222
 223 /**
 224  * Peers handle.
 225  */
 226 static struct GNUNET_TESTBED_Peer **testbed_peers;
 227
 228 /**
 229  * Statistics operation handle.
 230  */
 231 static struct GNUNET_TESTBED_Operation *stats_op;
 232
 233 /**
 234  * Keepalives sent.
 235  */
 236 static unsigned int ka_sent;
 237
 238 /**
 239  * Keepalives received.
 240  */
 241 static unsigned int ka_received;
 242
 243 /**
 244  * How many messages were dropped by CADET because of full buffers?
 245  */
 246 static unsigned int msg_dropped;
 247
 248
 249 /******************************************************************************/
 250
 251
 252 /******************************************************************************/
 253
 254
 255 /**
 256  * Get the channel considered as the "target" or "receiver", depending on
 257  * the test type and size.
 258  *
 259  * @return Channel handle of the target client, either 0 (for backward tests)
 260  *         or the last peer in the line (for other tests).
 261  */
 262 static struct GNUNET_CADET_Channel *
 263 get_target_channel ()
 264 {
 265   if (SPEED == test && GNUNET_YES == test_backwards)
 266     return outgoing_ch;
 267   else
 268     return incoming_ch;
 269 }
 270
 271
 272 /**
 273  * Show the results of the test (banwidth acheived) and log them to GAUGER
 274  */
 275 static void
 276 show_end_data (void)
 277 {
 278   static struct GNUNET_TIME_Absolute end_time;
 279   static struct GNUNET_TIME_Relative total_time;
 280
 281   end_time = GNUNET_TIME_absolute_get ();
 282   total_time = GNUNET_TIME_absolute_get_difference (start_time, end_time);
 283   FPRINTF (stderr,
 284            "\nResults of test \"%s\"\n",
 285            test_name);
 286   FPRINTF (stderr,
 287            "Test time %s\n",
 288            GNUNET_STRINGS_relative_time_to_string (total_time, GNUNET_YES));
 289   FPRINTF (stderr,
 290            "Test bandwidth: %f kb/s\n",
 291            4 * total_packets * 1.0 / (total_time.rel_value_us / 1000));    // 4bytes * ms
 292   FPRINTF (stderr,
 293            "Test throughput: %f packets/s\n\n",
 294            total_packets * 1000.0 / (total_time.rel_value_us / 1000));     // packets * ms
 295   GAUGER ("CADET",
 296           test_name,
 297           total_packets * 1000.0 / (total_time.rel_value_us / 1000),
 298           "packets/s");
 299 }
 300
 301
 302 /**
 303  * Disconnect from cadet services af all peers, call shutdown.
 304  *
 305  * @param cls Closure (line number from which termination was requested).
 306  * @param tc Task Context.
 307  */
 308 static void
 309 disconnect_cadet_peers (void *cls)
 310 {
 311   long line = (long) cls;
 312
 313   disconnect_task = NULL;
 314   GNUNET_log (GNUNET_ERROR_TYPE_INFO,
 315               "disconnecting cadet service of peers, called from line %ld\n",
 316               line);
 317   for (unsigned int i = 0; i < 2; i++)
 318   {
 319     GNUNET_TESTBED_operation_done (t_op[i]);
 320   }
 321   if (NULL != outgoing_ch)
 322   {
 323     GNUNET_CADET_channel_destroy (outgoing_ch);
 324     outgoing_ch = NULL;
 325   }
 326   if (NULL != incoming_ch)
 327   {
 328     GNUNET_CADET_channel_destroy (incoming_ch);
 329     incoming_ch = NULL;
 330   }
 331   GNUNET_CADET_TEST_cleanup (test_ctx);
 332   GNUNET_SCHEDULER_shutdown ();
 333 }
 334
 335
 336 /**
 337  * Shut down peergroup, clean up.
 338  *
 339  * @param cls Closure (unused).
 340  * @param tc Task Context.
 341  */
 342 static void
 343 shutdown_task (void *cls)
 344 {
 345   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 346               "Ending test.\n");
 347   if (NULL != send_next_msg_task)
 348   {
 349     GNUNET_SCHEDULER_cancel (send_next_msg_task);
 350     send_next_msg_task = NULL;
 351   }
 352   if (NULL != test_task)
 353   {
 354     GNUNET_SCHEDULER_cancel (test_task);
 355     test_task = NULL;
 356   }
 357   if (NULL != disconnect_task)
 358   {
 359     GNUNET_SCHEDULER_cancel (disconnect_task);
 360     disconnect_task =
 361         GNUNET_SCHEDULER_add_now (&disconnect_cadet_peers,
 362                                   (void *) __LINE__);
 363   }
 364 }
 365
 366
 367 /**
 368  * Stats callback. Finish the stats testbed operation and when all stats have
 369  * been iterated, shutdown the test.
 370  *
 371  * @param cls Closure (line number from which termination was requested).
 372  * @param op the operation that has been finished
 373  * @param emsg error message in case the operation has failed; will be NULL if
 374  *          operation has executed successfully.
 375  */
 376 static void
 377 stats_cont (void *cls,
 378             struct GNUNET_TESTBED_Operation *op,
 379             const char *emsg)
 380 {
 381   GNUNET_log (GNUNET_ERROR_TYPE_INFO,
 382               "KA sent: %u, KA received: %u\n",
 383               ka_sent,
 384               ka_received);
 385   if ((KEEPALIVE == test || REOPEN == test) &&
 386       ((ka_sent < 2) || (ka_sent > ka_received + 1)))
 387   {
 388     GNUNET_break (0);
 389     ok--;
 390   }
 391   GNUNET_TESTBED_operation_done (stats_op);
 392
 393   if (NULL != disconnect_task)
 394     GNUNET_SCHEDULER_cancel (disconnect_task);
 395   disconnect_task = GNUNET_SCHEDULER_add_now (&disconnect_cadet_peers,
 396                                               cls);
 397 }
 398
 399
 400 /**
 401  * Process statistic values.
 402  *
 403  * @param cls closure (line number, unused)
 404  * @param peer the peer the statistic belong to
 405  * @param subsystem name of subsystem that created the statistic
 406  * @param name the name of the datum
 407  * @param value the current value
 408  * @param is_persistent #GNUNET_YES if the value is persistent, #GNUNET_NO if not
 409  * @return #GNUNET_OK to continue, #GNUNET_SYSERR to abort iteration
 410  */
 411 static int
 412 stats_iterator (void *cls,
 413                 const struct GNUNET_TESTBED_Peer *peer,
 414                 const char *subsystem,
 415                 const char *name,
 416                 uint64_t value,
 417                 int is_persistent)
 418 {
 419   static const char *s_sent = "# keepalives sent";
 420   static const char *s_recv = "# keepalives received";
 421   static const char *rdrops = "# messages dropped due to full buffer";
 422   static const char *cdrops = "# messages dropped due to slow client";
 423   uint32_t i;
 424
 425   i = GNUNET_TESTBED_get_index (peer);
 426   GNUNET_log (GNUNET_ERROR_TYPE_INFO, "STATS PEER %u - %s [%s]: %llu\n", i,
 427               subsystem, name, (unsigned long long) value);
 428   if (0 == strncmp (s_sent, name, strlen (s_sent)) && 0 == i)
 429     ka_sent = value;
 430   if (0 == strncmp (s_recv, name, strlen (s_recv)) && peers_requested - 1 == i)
 431     ka_received = value;
 432   if (0 == strncmp (rdrops, name, strlen (rdrops)))
 433     msg_dropped += value;
 434   if (0 == strncmp (cdrops, name, strlen (cdrops)))
 435     msg_dropped += value;
 436
 437   return GNUNET_OK;
 438 }
 439
 440
 441 /**
 442  * Task to gather all statistics.
 443  *
 444  * @param cls Closure (line from which the task was scheduled).
 445  */
 446 static void
 447 gather_stats_and_exit (void *cls)
 448 {
 449   long l = (long) cls;
 450
 451   disconnect_task = NULL;
 452   GNUNET_log (GNUNET_ERROR_TYPE_INFO,
 453               "gathering statistics from line %ld\n",
 454               l);
 455   if (NULL != outgoing_ch)
 456   {
 457     GNUNET_CADET_channel_destroy (outgoing_ch);
 458     outgoing_ch = NULL;
 459   }
 460   stats_op = GNUNET_TESTBED_get_statistics (peers_running,
 461                                             testbed_peers,
 462                                             "cadet",
 463                                             NULL,
 464                                             &stats_iterator,
 465                                             stats_cont,
 466                                             cls);
 467 }
 468
 469
 470 /**
 471  * Send a message on the channel with the appropriate size and payload.
 472  *
 473  * Update the appropriate *_sent counter.
 474  *
 475  * @param channel Channel to send the message on.
 476  */
 477 static void
 478 send_test_message (struct GNUNET_CADET_Channel *channel);
 479
 480 /**
 481  * Check if payload is sane (size contains payload).
 482  *
 483  * @param cls should match #ch
 484  * @param message The actual message.
 485  * @return #GNUNET_OK to keep the channel open,
 486  *         #GNUNET_SYSERR to close it (signal serious error).
 487  */
 488 static int
 489 check_data (void *cls,
 490             const struct GNUNET_MessageHeader *message);
 491
 492 /**
 493  * Function is called whenever a message is received.
 494  *
 495  * @param cls closure (set from GNUNET_CADET_connect(), peer number)
 496  * @param message the actual message
 497  */
 498 static void
 499 handle_data (void *cls,
 500              const struct GNUNET_MessageHeader *message);
 501
 502 /**
 503  * Function called whenever an MQ-channel is destroyed, even if the destruction
 504  * was requested by #GNUNET_CADET_channel_destroy.
 505  * It must NOT call #GNUNET_CADET_channel_destroy on the channel.
 506  *
 507  * It should clean up any associated state, including cancelling any pending
 508  * transmission on this channel.
 509  *
 510  * @param cls Channel closure (channel wrapper).
 511  * @param channel Connection to the other end (henceforth invalid).
 512  */
 513 static void
 514 disconnect_handler (void *cls,
 515                      const struct GNUNET_CADET_Channel *channel);
 516
 517
 518 /**
 519  * Task to reconnect to other peer.
 520  *
 521  * @param cls Closure (line from which the task was scheduled).
 522  */
 523 static void
 524 reconnect_op (void *cls)
 525 {
 526   struct GNUNET_MQ_MessageHandler handlers[] = {
 527     GNUNET_MQ_hd_var_size (data,
 528                            GNUNET_MESSAGE_TYPE_DUMMY,
 529                            struct GNUNET_MessageHeader,
 530                            NULL),
 531     GNUNET_MQ_handler_end ()
 532   };
 533   long l = (long) cls;
 534   struct CadetTestChannelWrapper *ch;
 535   enum GNUNET_CADET_ChannelOption flags;
 536
 537   reconnect_task = NULL;
 538   GNUNET_log (GNUNET_ERROR_TYPE_INFO,
 539               "reconnecting from line %ld\n",
 540               l);
 541   if (NULL != outgoing_ch)
 542   {
 543     GNUNET_CADET_channel_destroy (outgoing_ch);
 544     outgoing_ch = NULL;
 545   }
 546   flags = GNUNET_CADET_OPTION_DEFAULT;
 547   ch = GNUNET_new (struct CadetTestChannelWrapper);
 548   outgoing_ch = GNUNET_CADET_channel_create (h1,
 549                                              ch,
 550                                              p_id[1],
 551                                              &port,
 552                                              flags,
 553                                              NULL,
 554                                              &disconnect_handler,
 555                                              handlers);
 556   ch->ch = outgoing_ch;
 557   send_test_message (outgoing_ch);
 558 }
 559
 560 /**
 561  * Function called whenever an MQ-channel is destroyed, even if the destruction
 562  * was requested by #GNUNET_CADET_channel_destroy.
 563  * It must NOT call #GNUNET_CADET_channel_destroy on the channel.
 564  *
 565  * It should clean up any associated state, including cancelling any pending
 566  * transmission on this channel.
 567  *
 568  * @param cls Channel closure (channel wrapper).
 569  * @param channel Connection to the other end (henceforth invalid).
 570  */
 571 static void
 572 disconnect_handler (void *cls,
 573                      const struct GNUNET_CADET_Channel *channel)
 574 {
 575   struct CadetTestChannelWrapper *ch_w = cls;
 576
 577   GNUNET_log (GNUNET_ERROR_TYPE_INFO,
 578               "Channel disconnected at %d\n",
 579               ok);
 580   GNUNET_assert (ch_w->ch == channel);
 581   if (channel == incoming_ch)
 582   {
 583     ok++;
 584     incoming_ch = NULL;
 585   }
 586   else if (outgoing_ch == channel)
 587   {
 588     if (P2P_SIGNAL == test)
 589     {
 590       ok++;
 591     }
 592     outgoing_ch = NULL;
 593   }
 594   else
 595     GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 596                 "Unknown channel! %p\n",
 597                 channel);
 598   if (NULL != disconnect_task && REOPEN != test)
 599   {
 600     GNUNET_SCHEDULER_cancel (disconnect_task);
 601     disconnect_task =
 602         GNUNET_SCHEDULER_add_now (&gather_stats_and_exit,
 603                                   (void *) __LINE__);
 604   }
 605   else if (NULL != reconnect_task && REOPEN == test)
 606   {
 607     GNUNET_SCHEDULER_cancel (reconnect_task);
 608     reconnect_task =
 609         GNUNET_SCHEDULER_add_now (&reconnect_op,
 610                                   (void *) __LINE__);
 611   }
 612   GNUNET_free (ch_w);
 613 }
 614
 615
 616 /**
 617  * Abort test: schedule disconnect and shutdown immediately
 618  *
 619  * @param line Line in the code the abort is requested from (__LINE__).
 620  */
 621 static void
 622 abort_test (long line)
 623 {
 624   if (NULL != disconnect_task)
 625   {
 626     GNUNET_SCHEDULER_cancel (disconnect_task);
 627     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 628                 "Aborting test from %ld\n",
 629                 line);
 630     disconnect_task =
 631         GNUNET_SCHEDULER_add_now (&disconnect_cadet_peers,
 632                                   (void *) line);
 633   }
 634 }
 635
 636
 637 /**
 638  * Send a message on the channel with the appropriate size and payload.
 639  *
 640  * Update the appropriate *_sent counter.
 641  *
 642  * @param channel Channel to send the message on.
 643  */
 644 static void
 645 send_test_message (struct GNUNET_CADET_Channel *channel)
 646 {
 647   struct GNUNET_MQ_Envelope *env;
 648   struct GNUNET_MessageHeader *msg;
 649   uint32_t *data;
 650   int payload;
 651   int size;
 652
 653   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 654               "Sending test message on channel %p\n",
 655               channel);
 656   size = size_payload;
 657   if (GNUNET_NO == initialized)
 658   {
 659     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Sending INITIALIZER\n");
 660     size += 1000;
 661     payload = data_sent;
 662     if (SPEED_ACK == test) // FIXME unify SPEED_ACK with an initializer
 663         data_sent++;
 664   }
 665   else if (SPEED == test || SPEED_ACK == test)
 666   {
 667     if (get_target_channel() == channel)
 668     {
 669       payload = ack_sent;
 670       size += ack_sent;
 671       ack_sent++;
 672       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 673                   "Sending ACK %u [%d bytes]\n",
 674                   payload, size);
 675     }
 676     else
 677     {
 678       payload = data_sent;
 679       size += data_sent;
 680       data_sent++;
 681       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 682                   "Sending DATA %u [%d bytes]\n",
 683                   data_sent, size);
 684     }
 685   }
 686   else if (FORWARD == test)
 687   {
 688     payload = ack_sent;
 689   }
 690   else if (P2P_SIGNAL == test)
 691   {
 692     payload = data_sent;
 693   }
 694   else if (REOPEN == test)
 695   {
 696     payload = data_sent;
 697     data_sent++;
 698     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 699                 "Sending DATA %u [%d bytes]\n",
 700                 data_sent, size);
 701   }
 702   else
 703   {
 704     GNUNET_assert (0);
 705   }
 706   env = GNUNET_MQ_msg_extra (msg, size, GNUNET_MESSAGE_TYPE_DUMMY);
 707
 708   data = (uint32_t *) &msg[1];
 709   *data = htonl (payload);
 710   GNUNET_MQ_send (GNUNET_CADET_get_mq (channel), env);
 711 }
 712
 713
 714 /**
 715  * Task to request a new data transmission in a SPEED test, without waiting
 716  * for previous messages to be sent/arrrive.
 717  *
 718  * @param cls Closure (unused).
 719  */
 720 static void
 721 send_next_msg (void *cls)
 722 {
 723   struct GNUNET_CADET_Channel *channel;
 724
 725   send_next_msg_task = NULL;
 726   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 727               "Sending next message: %d\n",
 728               data_sent);
 729
 730   channel = GNUNET_YES == test_backwards ? incoming_ch : outgoing_ch;
 731   GNUNET_assert (NULL != channel);
 732   GNUNET_assert (SPEED == test);
 733   send_test_message (channel);
 734   if (data_sent < total_packets)
 735   {
 736     /* SPEED test: Send all messages as soon as possible */
 737     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 738                 "Scheduling message %d\n",
 739                 data_sent + 1);
 740     send_next_msg_task =
 741       GNUNET_SCHEDULER_add_delayed (SEND_INTERVAL,
 742                                       &send_next_msg,
 743                                       NULL);
 744   }
 745 }
 746
 747
 748 /**
 749  * Every few messages cancel the timeout task and re-schedule it again, to
 750  * avoid timing out when traffic keeps coming.
 751  *
 752  * @param line Code line number to log if a timeout occurs.
 753  */
 754 static void
 755 reschedule_timeout_task (long line)
 756 {
 757   if ((ok % 10) == 0)
 758   {
 759     if (NULL != disconnect_task)
 760     {
 761       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 762                   "reschedule timeout every 10 messages\n");
 763       GNUNET_SCHEDULER_cancel (disconnect_task);
 764       disconnect_task = GNUNET_SCHEDULER_add_delayed (short_time,
 765                                                       &gather_stats_and_exit,
 766                                                       (void *) line);
 767     }
 768   }
 769 }
 770
 771
 772 /**
 773  * Check if payload is sane (size contains payload).
 774  *
 775  * @param cls should match #ch
 776  * @param message The actual message.
 777  * @return #GNUNET_OK to keep the channel open,
 778  *         #GNUNET_SYSERR to close it (signal serious error).
 779  */
 780 static int
 781 check_data (void *cls,
 782             const struct GNUNET_MessageHeader *message)
 783 {
 784   return GNUNET_OK;             /* all is well-formed */
 785 }
 786
 787
 788 /**
 789  * Function is called whenever a message is received.
 790  *
 791  * @param cls closure (set from GNUNET_CADET_connect(), peer number)
 792  * @param message the actual message
 793  */
 794 static void
 795 handle_data (void *cls,
 796              const struct GNUNET_MessageHeader *message)
 797 {
 798   struct CadetTestChannelWrapper *ch = cls;
 799   struct GNUNET_CADET_Channel *channel = ch->ch;
 800   uint32_t *data;
 801   uint32_t payload;
 802   int *counter;
 803
 804   ok++;
 805   GNUNET_CADET_receive_done (channel);
 806   counter = get_target_channel () == channel ? &data_received : &ack_received;
 807
 808   reschedule_timeout_task ((long) __LINE__);
 809
 810   if (channel == outgoing_ch)
 811   {
 812     GNUNET_log (GNUNET_ERROR_TYPE_INFO,
 813                 "Root client got a message.\n");
 814   }
 815   else if (channel == incoming_ch)
 816   {
 817     GNUNET_log (GNUNET_ERROR_TYPE_INFO,
 818                 "Leaf client got a message.\n");
 819   }
 820   else
 821   {
 822     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 823                 "Unknown channel %p.\n",
 824                 channel);
 825     GNUNET_assert (0);
 826   }
 827
 828   GNUNET_log (GNUNET_ERROR_TYPE_INFO,
 829               " ok: (%d/%d)\n",
 830               ok,
 831               ok_goal);
 832   data = (uint32_t *) &message[1];
 833   payload = ntohl (*data);
 834   if (payload == *counter)
 835   {
 836     GNUNET_log (GNUNET_ERROR_TYPE_INFO,
 837                 " payload as expected: %u\n",
 838                 payload);
 839   }
 840   else
 841   {
 842     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 843                 " payload %u, expected: %u\n",
 844                 payload, *counter);
 845   }
 846
 847   if (GNUNET_NO == initialized)
 848   {
 849     initialized = GNUNET_YES;
 850     start_time = GNUNET_TIME_absolute_get ();
 851     if (SPEED == test)
 852     {
 853       GNUNET_assert (incoming_ch == channel);
 854       send_next_msg_task = GNUNET_SCHEDULER_add_now (&send_next_msg,
 855                                                      NULL);
 856       return;
 857     }
 858   }
 859
 860   (*counter)++;
 861   if (get_target_channel () == channel) /* Got "data" */
 862   {
 863     GNUNET_log (GNUNET_ERROR_TYPE_INFO, " received data %u\n", data_received);
 864     if (SPEED != test || (ok_goal - 2) == ok)
 865     {
 866       /* Send ACK */
 867       send_test_message (channel);
 868       return;
 869     }
 870     else
 871     {
 872       if (data_received < total_packets)
 873         return;
 874     }
 875   }
 876   else /* Got "ack" */
 877   {
 878     if (SPEED_ACK == test || SPEED == test)
 879     {
 880       GNUNET_log (GNUNET_ERROR_TYPE_INFO, " received ack %u\n", ack_received);
 881       /* Send more data */
 882       send_test_message (channel);
 883       if (ack_received < total_packets && SPEED != test)
 884         return;
 885       if (ok == 2 && SPEED == test)
 886         return;
 887       show_end_data ();
 888     }
 889     if (test == P2P_SIGNAL)
 890     {
 891       GNUNET_CADET_channel_destroy (incoming_ch);
 892       incoming_ch = NULL;
 893     }
 894     else
 895     {
 896       GNUNET_CADET_channel_destroy (outgoing_ch);
 897       outgoing_ch = NULL;
 898     }
 899   }
 900 }
 901
 902
 903 /**
 904  * Method called whenever a peer connects to a port in MQ-based CADET.
 905  *
 906  * @param cls Closure from #GNUNET_CADET_open_port (peer # as long).
 907  * @param channel New handle to the channel.
 908  * @param source Peer that started this channel.
 909  * @return Closure for the incoming @a channel. It's given to:
 910  *         - The #GNUNET_CADET_DisconnectEventHandler (given to
 911  *           #GNUNET_CADET_open_port) when the channel dies.
 912  *         - Each the #GNUNET_MQ_MessageCallback handlers for each message
 913  *           received on the @a channel.
 914  */
 915 static void *
 916 connect_handler (void *cls,
 917                  struct GNUNET_CADET_Channel *channel,
 918                  const struct GNUNET_PeerIdentity *source)
 919 {
 920   struct CadetTestChannelWrapper *ch;
 921   long peer = (long) cls;
 922
 923   GNUNET_log (GNUNET_ERROR_TYPE_INFO,
 924               "Incoming channel from %s to %ld: %p\n",
 925               GNUNET_i2s (source),
 926               peer,
 927               channel);
 928   ok++;
 929   GNUNET_log (GNUNET_ERROR_TYPE_INFO,
 930               " ok: %d\n",
 931               ok);
 932   if (peer == peers_requested - 1)
 933   {
 934     if (NULL != incoming_ch)
 935     {
 936       GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 937                   "Duplicate incoming channel for client %lu\n",
 938                   (long) cls);
 939       GNUNET_assert (0);
 940     }
 941     incoming_ch = channel;
 942   }
 943   else
 944   {
 945     GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 946                 "Incoming channel for unexpected peer #%lu\n",
 947                 (long) cls);
 948     GNUNET_assert (0);
 949   }
 950   if (NULL != disconnect_task && REOPEN != test)
 951   {
 952     GNUNET_SCHEDULER_cancel (disconnect_task);
 953     disconnect_task = GNUNET_SCHEDULER_add_delayed (short_time,
 954                                                     &gather_stats_and_exit,
 955                                                     (void *) __LINE__);
 956   }
 957   else if ((NULL != disconnect_task) && (REOPEN == test))
 958   {
 959     GNUNET_SCHEDULER_cancel (disconnect_task);
 960     disconnect_task = GNUNET_SCHEDULER_add_delayed (
 961         GNUNET_TIME_relative_multiply (short_time, 2),
 962         &gather_stats_and_exit,
 963         (void *) __LINE__);
 964   }
 965
 966   if ((NULL != reconnect_task) && (REOPEN == test))
 967   {
 968     GNUNET_SCHEDULER_cancel (reconnect_task);
 969     reconnect_task = GNUNET_SCHEDULER_add_delayed (short_time,
 970                                                    &reconnect_op,
 971                                                    (void *) __LINE__);
 972   }
 973
 974   /* TODO: cannot return channel as-is, in order to unify the data handlers */
 975   ch = GNUNET_new (struct CadetTestChannelWrapper);
 976   ch->ch = channel;
 977
 978   return ch;
 979 }
 980
 981
 982 /**
 983  * START THE TESTCASE ITSELF, AS WE ARE CONNECTED TO THE CADET SERVICES.
 984  *
 985  * Testcase continues when the root receives confirmation of connected peers,
 986  * on callback function ch.
 987  *
 988  * @param cls Closure (unused).
 989  */
 990 static void
 991 start_test (void *cls)
 992 {
 993   struct GNUNET_MQ_MessageHandler handlers[] = {
 994     GNUNET_MQ_hd_var_size (data,
 995                            GNUNET_MESSAGE_TYPE_DUMMY,
 996                            struct GNUNET_MessageHeader,
 997                            NULL),
 998     GNUNET_MQ_handler_end ()
 999   };
1000   struct CadetTestChannelWrapper *ch;
1001   enum GNUNET_CADET_ChannelOption flags;
1002
1003   test_task = NULL;
1004   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "start_test: %s\n", test_name);
1005   if (NULL != disconnect_task)
1006   {
1007     GNUNET_SCHEDULER_cancel (disconnect_task);
1008     disconnect_task = NULL;
1009   }
1010
1011   flags = GNUNET_CADET_OPTION_DEFAULT;
1012   if (SPEED_REL == test)
1013   {
1014     test = SPEED;
1015     flags |= GNUNET_CADET_OPTION_RELIABLE;
1016   }
1017
1018   ch = GNUNET_new (struct CadetTestChannelWrapper);
1019   outgoing_ch = GNUNET_CADET_channel_create (h1,
1020                                              ch,
1021                                              p_id[1],
1022                                              &port,
1023                                              flags,
1024                                              NULL,
1025                                              &disconnect_handler,
1026                                              handlers);
1027
1028   ch->ch = outgoing_ch;
1029
1030   disconnect_task = GNUNET_SCHEDULER_add_delayed (short_time,
1031                                                   &gather_stats_and_exit,
1032                                                   (void *) __LINE__);
1033   if (KEEPALIVE == test)
1034     return;                     /* Don't send any data. */
1035
1036   data_received = 0;
1037   data_sent = 0;
1038   ack_received = 0;
1039   ack_sent = 0;
1040   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1041               "Sending data initializer on channel %p...\n",
1042               outgoing_ch);
1043   send_test_message (outgoing_ch);
1044   if (REOPEN == test)
1045   {
1046     reconnect_task = GNUNET_SCHEDULER_add_delayed (short_time,
1047                                                    &reconnect_op,
1048                                                    (void *) __LINE__);
1049     GNUNET_SCHEDULER_cancel (disconnect_task);
1050     disconnect_task = GNUNET_SCHEDULER_add_delayed (
1051         GNUNET_TIME_relative_multiply (short_time, 2),
1052         &gather_stats_and_exit,
1053         (void *) __LINE__);
1054   }
1055
1056 }
1057
1058
1059 /**
1060  * Callback to be called when the requested peer information is available
1061  *
1062  * @param cls the closure from GNUNET_TESTBED_peer_get_information()
1063  * @param op the operation this callback corresponds to
1064  * @param pinfo the result; will be NULL if the operation has failed
1065  * @param emsg error message if the operation has failed;
1066  *             NULL if the operation is successfull
1067  */
1068 static void
1069 pi_cb (void *cls,
1070        struct GNUNET_TESTBED_Operation *op,
1071        const struct GNUNET_TESTBED_PeerInformation *pinfo,
1072        const char *emsg)
1073 {
1074   long i = (long) cls;
1075
1076   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1077               "ID callback for %ld\n",
1078               i);
1079   if ( (NULL == pinfo) ||
1080        (NULL != emsg) )
1081   {
1082     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
1083                 "pi_cb: %s\n",
1084                 emsg);
1085     abort_test (__LINE__);
1086     return;
1087   }
1088   p_id[i] = pinfo->result.id;
1089   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1090               "id: %s\n",
1091               GNUNET_i2s (p_id[i]));
1092   p_ids++;
1093   if (p_ids < 2)
1094     return;
1095   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1096               "Got all IDs, starting test\n");
1097   test_task = GNUNET_SCHEDULER_add_now (&start_test, NULL);
1098 }
1099
1100
1101 /**
1102  * test main: start test when all peers are connected
1103  *
1104  * @param cls Closure.
1105  * @param ctx Argument to give to GNUNET_CADET_TEST_cleanup on test end.
1106  * @param num_peers Number of peers that are running.
1107  * @param peers Array of peers.
1108  * @param cadets Handle to each of the CADETs of the peers.
1109  */
1110 static void
1111 tmain (void *cls,
1112        struct GNUNET_CADET_TEST_Context *ctx,
1113        unsigned int num_peers,
1114        struct GNUNET_TESTBED_Peer **peers,
1115        struct GNUNET_CADET_Handle **cadets)
1116 {
1117   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "test main\n");
1118   ok = 0;
1119   test_ctx = ctx;
1120   peers_running = num_peers;
1121   GNUNET_assert (peers_running == peers_requested);
1122   testbed_peers = peers;
1123   h1 = cadets[0];
1124   h2 = cadets[num_peers - 1];
1125   disconnect_task = GNUNET_SCHEDULER_add_delayed (short_time,
1126                                                   &disconnect_cadet_peers,
1127                                                   (void *) __LINE__);
1128   GNUNET_SCHEDULER_add_shutdown (&shutdown_task,
1129                                  NULL);
1130   t_op[0] = GNUNET_TESTBED_peer_get_information (peers[0],
1131                                                  GNUNET_TESTBED_PIT_IDENTITY,
1132                                                  &pi_cb,
1133                                                  (void *) 0L);
1134   t_op[1] = GNUNET_TESTBED_peer_get_information (peers[num_peers - 1],
1135                                                  GNUNET_TESTBED_PIT_IDENTITY,
1136                                                  &pi_cb,
1137                                                  (void *) 1L);
1138   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "requested peer ids\n");
1139 }
1140
1141
1142 /**
1143  * Main: start test
1144  */
1145 int
1146 main (int argc, char *argv[])
1147 {
1148   static const struct GNUNET_HashCode *ports[2];
1149   struct GNUNET_MQ_MessageHandler handlers[] = {
1150     GNUNET_MQ_hd_var_size (data,
1151                            GNUNET_MESSAGE_TYPE_DUMMY,
1152                            struct GNUNET_MessageHeader,
1153                            NULL),
1154     GNUNET_MQ_handler_end ()
1155   };
1156   const char *config_file;
1157   char port_id[] = "test port";
1158   struct GNUNET_GETOPT_CommandLineOption options[] = {
1159     GNUNET_GETOPT_option_relative_time ('t',
1160                                         "time",
1161                                         "short_time",
1162                                         gettext_noop ("set short timeout"),
1163                                         &short_time),
1164     GNUNET_GETOPT_option_uint ('m',
1165                                "messages",
1166                                "NUM_MESSAGES",
1167                                gettext_noop ("set number of messages to send"),
1168                                &total_packets),
1169
1170     GNUNET_GETOPT_OPTION_END
1171   };
1172
1173
1174   initialized = GNUNET_NO;
1175   GNUNET_log_setup ("test", "DEBUG", NULL);
1176
1177   total_packets = TOTAL_PACKETS;
1178   short_time = SHORT_TIME;
1179   if (-1 == GNUNET_GETOPT_run (argv[0], options, argc, argv))
1180   {
1181     FPRINTF (stderr, "test failed: problem with CLI parameters\n");
1182     exit (1);
1183   }
1184
1185   config_file = "test_cadet.conf";
1186   GNUNET_CRYPTO_hash (port_id, sizeof (port_id), &port);
1187
1188   /* Find out requested size */
1189   if (strstr (argv[0], "_2_") != NULL)
1190   {
1191     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "DIRECT CONNECTIONs\n");
1192     peers_requested = 2;
1193   }
1194   else if (strstr (argv[0], "_5_") != NULL)
1195   {
1196     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "5 PEER LINE\n");
1197     peers_requested = 5;
1198   }
1199   else if (strstr (argv[0], "_6_") != NULL)
1200   {
1201     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "6 PEER LINE\n");
1202     peers_requested = 6;
1203   }
1204   else
1205   {
1206     GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "SIZE UNKNOWN, USING 2\n");
1207     peers_requested = 2;
1208   }
1209
1210   /* Find out requested test */
1211   if (strstr (argv[0], "_forward") != NULL)
1212   {
1213     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "FORWARD\n");
1214     test = FORWARD;
1215     test_name = "unicast";
1216     ok_goal = 4;
1217   }
1218   else if (strstr (argv[0], "_signal") != NULL)
1219   {
1220     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "SIGNAL\n");
1221     test = P2P_SIGNAL;
1222     test_name = "signal";
1223     ok_goal = 4;
1224   }
1225   else if (strstr (argv[0], "_speed_ack") != NULL)
1226   {
1227     /* Test is supposed to generate the following callbacks:
1228      * 1 incoming channel (@dest)
1229      * total_packets received data packet (@dest)
1230      * total_packets received data packet (@orig)
1231      * 1 received channel destroy (@dest)
1232      */
1233     ok_goal = total_packets * 2 + 2;
1234     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "SPEED_ACK\n");
1235     test = SPEED_ACK;
1236     test_name = "speed ack";
1237   }
1238   else if (strstr (argv[0], "_speed") != NULL)
1239   {
1240     /* Test is supposed to generate the following callbacks:
1241      * 1 incoming channel (@dest)
1242      * 1 initial packet (@dest)
1243      * total_packets received data packet (@dest)
1244      * 1 received data packet (@orig)
1245      * 1 received channel destroy (@dest)
1246      */
1247     ok_goal = total_packets + 4;
1248     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "SPEED\n");
1249     if (strstr (argv[0], "_reliable") != NULL)
1250     {
1251       test = SPEED_REL;
1252       test_name = "speed reliable";
1253       config_file = "test_cadet_drop.conf";
1254     }
1255     else
1256     {
1257       test = SPEED;
1258       test_name = "speed";
1259     }
1260   }
1261   else if (strstr (argv[0], "_keepalive") != NULL)
1262   {
1263     test = KEEPALIVE;
1264     /* Test is supposed to generate the following callbacks:
1265      * 1 incoming channel (@dest)
1266      * [wait]
1267      * 1 received channel destroy (@dest)
1268      */
1269     ok_goal = 2;
1270   }
1271   else if (strstr (argv[0], "_reopen") != NULL)
1272   {
1273     test = REOPEN;
1274     test_name = "reopen";
1275     ///* Test is supposed to generate the following callbacks:
1276     // * 1 incoming channel (@dest)
1277     // * [wait]
1278     // * 1 received channel destroy (@dest)
1279     // */
1280     ok_goal = 7;
1281   }
1282   else
1283   {
1284     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "UNKNOWN\n");
1285     test = SETUP;
1286     ok_goal = 0;
1287   }
1288
1289   if (strstr (argv[0], "backwards") != NULL)
1290   {
1291     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "BACKWARDS (LEAF TO ROOT)\n");
1292     test_backwards = GNUNET_YES;
1293     GNUNET_asprintf (&test_name, "backwards %s", test_name);
1294   }
1295
1296   p_ids = 0;
1297   ports[0] = &port;
1298   ports[1] = NULL;
1299   GNUNET_CADET_TEST_ruN ("test_cadet_small",
1300                          config_file,
1301                          peers_requested,
1302                          &tmain,
1303                          NULL,        /* tmain cls */
1304                          &connect_handler,
1305                          NULL,
1306                          &disconnect_handler,
1307                          handlers,
1308                          ports);
1309   if (NULL != strstr (argv[0], "_reliable"))
1310     msg_dropped = 0;            /* dropped should be retransmitted */
1311
1312   if (ok_goal > ok - msg_dropped)
1313   {
1314     GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "FAILED! (%d/%d)\n", ok, ok_goal);
1315     return 1;
1316   }
1317   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "success\n");
1318   return 0;
1319 }
1320
1321 /* end of test_cadet.c */