src/cadet/test_cadet_flow.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_flow.c
  22  * @author Bart Polot
  23  * @author Christian Grothoff
  24  * @brief Test for flow control of CADET service
  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_DEFAULT 500
  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 ( \
  64     GNUNET_TIME_UNIT_MILLISECONDS, 10)
  65
  66
  67 /**
  68  * How many packets to send.
  69  */
  70 static unsigned int total_packets = TOTAL_PACKETS_DEFAULT;
  71
  72 /**
  73  * Time to wait for fast operations.
  74  */
  75 static struct GNUNET_TIME_Relative short_time;
  76
  77 /**
  78  * Size of each test packet's payload
  79  */
  80 static size_t size_payload = sizeof(uint32_t);
  81
  82 /**
  83  * Operation to get peer ids.
  84  */
  85 static struct GNUNET_TESTBED_Operation *t_op[2];
  86
  87 /**
  88  * Peer ids.
  89  */
  90 static struct GNUNET_PeerIdentity *p_id[2];
  91
  92 /**
  93  * Port ID
  94  */
  95 static struct GNUNET_HashCode port;
  96
  97 /**
  98  * Peer ids counter.
  99  */
 100 static unsigned int p_ids;
 101
 102 /**
 103  * Is the setup initialized?
 104  */
 105 static int initialized;
 106
 107 /**
 108  * Number of payload packes sent.
 109  */
 110 static int data_sent;
 111
 112 /**
 113  * Number of payload packets received.
 114  */
 115 static int data_received;
 116
 117 /**
 118  * Number of payload packed acknowledgements sent.
 119  */
 120 static int ack_sent;
 121
 122 /**
 123  * Number of payload packed explicitly (app level) acknowledged.
 124  */
 125 static int ack_received;
 126
 127 /**
 128  * Total number of peers asked to run.
 129  */
 130 static unsigned int peers_requested = 2;
 131
 132 /**
 133  * Number of currently running peers (should be same as @c peers_requested).
 134  */
 135 static unsigned int peers_running;
 136
 137 /**
 138  * Test context (to shut down).
 139  */
 140 struct GNUNET_CADET_TEST_Context *test_ctx;
 141
 142 /**
 143  * Task called to disconnect peers.
 144  */
 145 static struct GNUNET_SCHEDULER_Task *disconnect_task;
 146
 147 /**
 148  * Task To perform tests
 149  */
 150 static struct GNUNET_SCHEDULER_Task *test_task;
 151
 152 /**
 153  * Task runnining #send_next_msg().
 154  */
 155 static struct GNUNET_SCHEDULER_Task *send_next_msg_task;
 156
 157 /**
 158  * Cadet handle for the root peer
 159  */
 160 static struct GNUNET_CADET_Handle *h1;
 161
 162 /**
 163  * Cadet handle for the first leaf peer
 164  */
 165 static struct GNUNET_CADET_Handle *h2;
 166
 167 /**
 168  * Channel handle for the root peer
 169  */
 170 static struct GNUNET_CADET_Channel *outgoing_ch;
 171
 172 /**
 173  * Channel handle for the dest peer
 174  */
 175 static struct GNUNET_CADET_Channel *incoming_ch;
 176
 177 /**
 178  * Time we started the data transmission (after channel has been established
 179  * and initilized).
 180  */
 181 static struct GNUNET_TIME_Absolute start_time;
 182
 183 /**
 184  * Peers handle.
 185  */
 186 static struct GNUNET_TESTBED_Peer **testbed_peers;
 187
 188 /**
 189  * Statistics operation handle.
 190  */
 191 static struct GNUNET_TESTBED_Operation *stats_op;
 192
 193 /**
 194  * Keepalives sent.
 195  */
 196 static unsigned int ka_sent;
 197
 198 /**
 199  * Keepalives received.
 200  */
 201 static unsigned int ka_received;
 202
 203 /**
 204  * How many messages were dropped by CADET because of full buffers?
 205  */
 206 static unsigned int msg_dropped;
 207
 208
 209 /**
 210  * Show the results of the test (banwidth acheived) and log them to GAUGER
 211  */
 212 static void
 213 show_end_data (void)
 214 {
 215   static struct GNUNET_TIME_Absolute end_time;
 216   static struct GNUNET_TIME_Relative total_time;
 217
 218   end_time = GNUNET_TIME_absolute_get ();
 219   total_time = GNUNET_TIME_absolute_get_difference (start_time, end_time);
 220   fprintf (stderr,
 221            "\nResults of test \"%s\"\n",
 222            test_name);
 223   fprintf (stderr,
 224            "Test time %s\n",
 225            GNUNET_STRINGS_relative_time_to_string (total_time, GNUNET_YES));
 226   fprintf (stderr,
 227            "Test bandwidth: %f kb/s\n",
 228            4 * total_packets * 1.0 / (total_time.rel_value_us / 1000));    // 4bytes * ms
 229   fprintf (stderr,
 230            "Test throughput: %f packets/s\n\n",
 231            total_packets * 1000.0 / (total_time.rel_value_us / 1000));     // packets * ms
 232   GAUGER ("CADET",
 233           test_name,
 234           total_packets * 1000.0 / (total_time.rel_value_us / 1000),
 235           "packets/s");
 236 }
 237
 238
 239 /**
 240  * Shut down peergroup, clean up.
 241  *
 242  * @param cls Closure (unused).
 243  * @param tc Task Context.
 244  */
 245 static void
 246 shutdown_task (void *cls)
 247 {
 248   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 249               "Ending test.\n");
 250   if (NULL != send_next_msg_task)
 251   {
 252     GNUNET_SCHEDULER_cancel (send_next_msg_task);
 253     send_next_msg_task = NULL;
 254   }
 255   if (NULL != test_task)
 256   {
 257     GNUNET_SCHEDULER_cancel (test_task);
 258     test_task = NULL;
 259   }
 260   for (unsigned int i = 0; i < 2; i++)
 261     GNUNET_TESTBED_operation_done (t_op[i]);
 262   if (NULL != outgoing_ch)
 263   {
 264     GNUNET_CADET_channel_destroy (outgoing_ch);
 265     outgoing_ch = NULL;
 266   }
 267   if (NULL != incoming_ch)
 268   {
 269     GNUNET_CADET_channel_destroy (incoming_ch);
 270     incoming_ch = NULL;
 271   }
 272   GNUNET_CADET_TEST_cleanup (test_ctx);
 273 }
 274
 275
 276 /**
 277  * Stats callback. Finish the stats testbed operation and when all stats have
 278  * been iterated, shutdown the test.
 279  *
 280  * @param cls Closure (line number from which termination was requested).
 281  * @param op the operation that has been finished
 282  * @param emsg error message in case the operation has failed; will be NULL if
 283  *          operation has executed successfully.
 284  */
 285 static void
 286 stats_cont (void *cls,
 287             struct GNUNET_TESTBED_Operation *op,
 288             const char *emsg)
 289 {
 290   GNUNET_log (GNUNET_ERROR_TYPE_INFO,
 291               "KA sent: %u, KA received: %u\n",
 292               ka_sent,
 293               ka_received);
 294   if ((KEEPALIVE == test) && ((ka_sent < 2) || (ka_sent > ka_received + 1)))
 295   {
 296     GNUNET_break (0);
 297     ok--;
 298   }
 299   GNUNET_TESTBED_operation_done (stats_op);
 300
 301   if (NULL != disconnect_task)
 302     GNUNET_SCHEDULER_cancel (disconnect_task);
 303   disconnect_task = GNUNET_SCHEDULER_add_now (&disconnect_cadet_peers,
 304                                               cls);
 305 }
 306
 307
 308 /**
 309  * Process statistic values.
 310  *
 311  * @param cls closure (line number, unused)
 312  * @param peer the peer the statistic belong to
 313  * @param subsystem name of subsystem that created the statistic
 314  * @param name the name of the datum
 315  * @param value the current value
 316  * @param is_persistent #GNUNET_YES if the value is persistent, #GNUNET_NO if not
 317  * @return #GNUNET_OK to continue, #GNUNET_SYSERR to abort iteration
 318  */
 319 static int
 320 stats_iterator (void *cls,
 321                 const struct GNUNET_TESTBED_Peer *peer,
 322                 const char *subsystem,
 323                 const char *name,
 324                 uint64_t value,
 325                 int is_persistent)
 326 {
 327   static const char *s_sent = "# keepalives sent";
 328   static const char *s_recv = "# keepalives received";
 329   static const char *rdrops = "# messages dropped due to full buffer";
 330   static const char *cdrops = "# messages dropped due to slow client";
 331   uint32_t i;
 332
 333   i = GNUNET_TESTBED_get_index (peer);
 334   GNUNET_log (GNUNET_ERROR_TYPE_INFO, "STATS PEER %u - %s [%s]: %llu\n", i,
 335               subsystem, name, (unsigned long long) value);
 336   if ((0 == strncmp (s_sent, name, strlen (s_sent))) && (0 == i))
 337     ka_sent = value;
 338   if ((0 == strncmp (s_recv, name, strlen (s_recv))) && (peers_requested - 1 ==
 339                                                          i) )
 340     ka_received = value;
 341   if (0 == strncmp (rdrops, name, strlen (rdrops)))
 342     msg_dropped += value;
 343   if (0 == strncmp (cdrops, name, strlen (cdrops)))
 344     msg_dropped += value;
 345
 346   return GNUNET_OK;
 347 }
 348
 349
 350 /**
 351  * Task to gather all statistics.
 352  *
 353  * @param cls Closure (line from which the task was scheduled).
 354  */
 355 static void
 356 gather_stats_and_exit (void *cls)
 357 {
 358   long l = (long) cls;
 359
 360   disconnect_task = NULL;
 361   GNUNET_log (GNUNET_ERROR_TYPE_INFO,
 362               "gathering statistics from line %ld\n",
 363               l);
 364   if (NULL != outgoing_ch)
 365   {
 366     GNUNET_CADET_channel_destroy (outgoing_ch);
 367     outgoing_ch = NULL;
 368   }
 369   stats_op = GNUNET_TESTBED_get_statistics (peers_running,
 370                                             testbed_peers,
 371                                             "cadet",
 372                                             NULL,
 373                                             &stats_iterator,
 374                                             stats_cont,
 375                                             cls);
 376 }
 377
 378
 379 /**
 380  * Abort test: schedule disconnect and shutdown immediately
 381  *
 382  * @param line Line in the code the abort is requested from (__LINE__).
 383  */
 384 static void
 385 abort_test (long line)
 386 {
 387   if (NULL != disconnect_task)
 388   {
 389     GNUNET_SCHEDULER_cancel (disconnect_task);
 390     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 391                 "Aborting test from %ld\n",
 392                 line);
 393     disconnect_task =
 394       GNUNET_SCHEDULER_add_now (&disconnect_cadet_peers,
 395                                 (void *) line);
 396   }
 397 }
 398
 399
 400 /**
 401  * Send a message on the channel with the appropriate size and payload.
 402  *
 403  * Update the appropriate *_sent counter.
 404  *
 405  * @param channel Channel to send the message on.
 406  */
 407 static void
 408 send_test_message (struct GNUNET_CADET_Channel *channel)
 409 {
 410   struct GNUNET_MQ_Envelope *env;
 411   struct GNUNET_MessageHeader *msg;
 412   uint32_t *data;
 413   int payload;
 414   int size;
 415
 416   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 417               "Sending test message on channel %p\n",
 418               channel);
 419   size = size_payload;
 420   if (GNUNET_NO == initialized)
 421   {
 422     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Sending INITIALIZER\n");
 423     size += 1000;
 424     payload = data_sent;
 425     if (SPEED_ACK == test)   // FIXME unify SPEED_ACK with an initializer
 426       data_sent++;
 427   }
 428   else if ((SPEED == test) || (SPEED_ACK == test))
 429   {
 430     if (get_target_channel () == channel)
 431     {
 432       payload = ack_sent;
 433       size += ack_sent;
 434       ack_sent++;
 435       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 436                   "Sending ACK %u [%d bytes]\n",
 437                   payload, size);
 438     }
 439     else
 440     {
 441       payload = data_sent;
 442       size += data_sent;
 443       data_sent++;
 444       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 445                   "Sending DATA %u [%d bytes]\n",
 446                   data_sent, size);
 447     }
 448   }
 449   else if (FORWARD == test)
 450   {
 451     payload = ack_sent;
 452   }
 453   else if (P2P_SIGNAL == test)
 454   {
 455     payload = data_sent;
 456   }
 457   else
 458   {
 459     GNUNET_assert (0);
 460   }
 461   env = GNUNET_MQ_msg_extra (msg, size, GNUNET_MESSAGE_TYPE_DUMMY);
 462
 463   data = (uint32_t *) &msg[1];
 464   *data = htonl (payload);
 465   GNUNET_MQ_send (GNUNET_CADET_get_mq (channel), env);
 466 }
 467
 468
 469 /**
 470  * Task to request a new data transmission in a SPEED test, without waiting
 471  * for previous messages to be sent/arrrive.
 472  *
 473  * @param cls Closure (unused).
 474  */
 475 static void
 476 send_next_msg (void *cls)
 477 {
 478   struct GNUNET_CADET_Channel *channel;
 479
 480   send_next_msg_task = NULL;
 481   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 482               "Sending next message: %d\n",
 483               data_sent);
 484
 485   channel = GNUNET_YES == test_backwards ? incoming_ch : outgoing_ch;
 486   GNUNET_assert (NULL != channel);
 487   GNUNET_assert (SPEED == test);
 488   send_test_message (channel);
 489   if (data_sent < total_packets)
 490   {
 491     /* SPEED test: Send all messages as soon as possible */
 492     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 493                 "Scheduling message %d\n",
 494                 data_sent + 1);
 495     send_next_msg_task =
 496       GNUNET_SCHEDULER_add_delayed (SEND_INTERVAL,
 497                                     &send_next_msg,
 498                                     NULL);
 499   }
 500 }
 501
 502
 503 /**
 504  * Check if payload is sane (size contains payload).
 505  *
 506  * @param cls should match #ch
 507  * @param message The actual message.
 508  * @return #GNUNET_OK to keep the channel open,
 509  *         #GNUNET_SYSERR to close it (signal serious error).
 510  */
 511 static int
 512 check_data (void *cls,
 513             const struct GNUNET_MessageHeader *message)
 514 {
 515   return GNUNET_OK;             /* all is well-formed */
 516 }
 517
 518
 519 /**
 520  * Function is called whenever a message is received.
 521  *
 522  * @param cls closure (set from GNUNET_CADET_connect(), peer number)
 523  * @param message the actual message
 524  */
 525 static void
 526 handle_data (void *cls,
 527              const struct GNUNET_MessageHeader *message)
 528 {
 529   struct CadetTestChannelWrapper *ch = cls;
 530   struct GNUNET_CADET_Channel *channel = ch->ch;
 531   uint32_t *data;
 532   uint32_t payload;
 533   int *counter;
 534
 535   GNUNET_CADET_receive_done (channel);
 536   counter = get_target_channel () == channel ? &data_received : &ack_received;
 537   if (channel == outgoing_ch)
 538   {
 539     GNUNET_log (GNUNET_ERROR_TYPE_INFO,
 540                 "Root client got a message.\n");
 541   }
 542   else if (channel == incoming_ch)
 543   {
 544     GNUNET_log (GNUNET_ERROR_TYPE_INFO,
 545                 "Leaf client got a message.\n");
 546   }
 547   else
 548   {
 549     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 550                 "Unknown channel %p.\n",
 551                 channel);
 552     GNUNET_assert (0);
 553   }
 554
 555   data = (uint32_t *) &message[1];
 556   payload = ntohl (*data);
 557   if (payload == *counter)
 558   {
 559     GNUNET_log (GNUNET_ERROR_TYPE_INFO,
 560                 "Payload as expected: %u\n",
 561                 payload);
 562   }
 563   else
 564   {
 565     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 566                 "Received payload %u, expected: %u\n",
 567                 payload, *counter);
 568   }
 569   (*counter)++;
 570   if (get_target_channel () == channel)  /* Got "data" */
 571   {
 572     GNUNET_log (GNUNET_ERROR_TYPE_INFO,
 573                 " received data %u\n",
 574                 data_received);
 575     if (data_received < total_packets)
 576       return;
 577   }
 578   else /* Got "ack" */
 579   {
 580     if ((SPEED_ACK == test) || (SPEED == test) )
 581     {
 582       GNUNET_log (GNUNET_ERROR_TYPE_INFO, " received ack %u\n", ack_received);
 583       /* Send more data */
 584       send_test_message (channel);
 585       if ((ack_received < total_packets) && (SPEED != test) )
 586         return;
 587       if ((ok == 2) && (SPEED == test) )
 588         return;
 589       show_end_data ();
 590     }
 591     if (test == P2P_SIGNAL)
 592     {
 593       GNUNET_CADET_channel_destroy (incoming_ch);
 594       incoming_ch = NULL;
 595     }
 596     else
 597     {
 598       GNUNET_CADET_channel_destroy (outgoing_ch);
 599       outgoing_ch = NULL;
 600     }
 601   }
 602 }
 603
 604
 605 /**
 606  * Method called whenever a peer connects to a port in MQ-based CADET.
 607  *
 608  * @param cls Closure from #GNUNET_CADET_open_port (peer # as long).
 609  * @param channel New handle to the channel.
 610  * @param source Peer that started this channel.
 611  * @return Closure for the incoming @a channel. It's given to:
 612  *         - The #GNUNET_CADET_DisconnectEventHandler (given to
 613  *           #GNUNET_CADET_open_port) when the channel dies.
 614  *         - Each the #GNUNET_MQ_MessageCallback handlers for each message
 615  *           received on the @a channel.
 616  */
 617 static void *
 618 connect_handler (void *cls,
 619                  struct GNUNET_CADET_Channel *channel,
 620                  const struct GNUNET_PeerIdentity *source)
 621 {
 622   struct CadetTestChannelWrapper *ch;
 623   long peer = (long) cls;
 624
 625   GNUNET_log (GNUNET_ERROR_TYPE_INFO,
 626               "Incoming channel from %s to %ld: %p\n",
 627               GNUNET_i2s (source),
 628               peer,
 629               channel);
 630   if (peer == peers_requested - 1)
 631   {
 632     if (NULL != incoming_ch)
 633     {
 634       GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 635                   "Duplicate incoming channel for client %lu\n",
 636                   (long) cls);
 637       GNUNET_assert (0);
 638     }
 639     incoming_ch = channel;
 640   }
 641   else
 642   {
 643     GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 644                 "Incoming channel for unexpected peer #%lu\n",
 645                 (long) cls);
 646     GNUNET_assert (0);
 647   }
 648   ch = GNUNET_new (struct CadetTestChannelWrapper);
 649   ch->ch = channel;
 650
 651   return ch;
 652 }
 653
 654
 655 /**
 656  * Function called whenever an MQ-channel is destroyed, even if the destruction
 657  * was requested by #GNUNET_CADET_channel_destroy.
 658  * It must NOT call #GNUNET_CADET_channel_destroy on the channel.
 659  *
 660  * It should clean up any associated state, including cancelling any pending
 661  * transmission on this channel.
 662  *
 663  * @param cls Channel closure (channel wrapper).
 664  * @param channel Connection to the other end (henceforth invalid).
 665  */
 666 static void
 667 disconnect_handler (void *cls,
 668                     const struct GNUNET_CADET_Channel *channel)
 669 {
 670   struct CadetTestChannelWrapper *ch_w = cls;
 671
 672   GNUNET_log (GNUNET_ERROR_TYPE_INFO,
 673               "Channel disconnected at %d\n",
 674               ok);
 675   GNUNET_assert (ch_w->ch == channel);
 676   if (channel == incoming_ch)
 677     incoming_ch = NULL;
 678   else if (outgoing_ch == channel)
 679     outgoing_ch = NULL;
 680   else
 681     GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 682                 "Disconnect on unknown channel %p\n",
 683                 channel);
 684   if (NULL != disconnect_task)
 685     GNUNET_SCHEDULER_cancel (disconnect_task);
 686   disconnect_task = GNUNET_SCHEDULER_add_now (&gather_stats_and_exit,
 687                                               (void *) __LINE__);
 688   GNUNET_free (ch_w);
 689 }
 690
 691
 692 /**
 693  * Start the testcase, we know the peers and have handles to CADET.
 694  *
 695  * Testcase continues when the root receives confirmation of connected peers,
 696  * on callback function ch.
 697  *
 698  * @param cls Closure (unused).
 699  */
 700 static void
 701 start_test (void *cls)
 702 {
 703   struct GNUNET_MQ_MessageHandler handlers[] = {
 704     GNUNET_MQ_hd_var_size (data,
 705                            GNUNET_MESSAGE_TYPE_DUMMY,
 706                            struct GNUNET_MessageHeader,
 707                            NULL),
 708     GNUNET_MQ_handler_end ()
 709   };
 710   struct CadetTestChannelWrapper *ch;
 711   enum GNUNET_CADET_ChannelOption flags;
 712
 713   test_task = NULL;
 714   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 715               "In start_test\n");
 716   start_time = GNUNET_TIME_absolute_get ();
 717   ch = GNUNET_new (struct CadetTestChannelWrapper);
 718   outgoing_ch = GNUNET_CADET_channel_create (h1,
 719                                              ch,
 720                                              p_id[1],
 721                                              &port,
 722                                              flags,
 723                                              NULL,
 724                                              &disconnect_handler,
 725                                              handlers);
 726   ch->ch = outgoing_ch;
 727   GNUNET_assert (NULL == disconnect_task);
 728   disconnect_task
 729     = GNUNET_SCHEDULER_add_delayed (short_time,
 730                                     &gather_stats_and_exit,
 731                                     (void *) __LINE__);
 732   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 733               "Sending data initializer on channel %p...\n",
 734               outgoing_ch);
 735   send_test_message (outgoing_ch);
 736 }
 737
 738
 739 /**
 740  * Callback to be called when the requested peer information is available
 741  *
 742  * @param cls the closure from GNUNET_TESTBED_peer_get_information()
 743  * @param op the operation this callback corresponds to
 744  * @param pinfo the result; will be NULL if the operation has failed
 745  * @param emsg error message if the operation has failed;
 746  *             NULL if the operation is successfull
 747  */
 748 static void
 749 pi_cb (void *cls,
 750        struct GNUNET_TESTBED_Operation *op,
 751        const struct GNUNET_TESTBED_PeerInformation *pinfo,
 752        const char *emsg)
 753 {
 754   long i = (long) cls;
 755
 756   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 757               "ID callback for %ld\n",
 758               i);
 759   if ((NULL == pinfo) ||
 760       (NULL != emsg))
 761   {
 762     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 763                 "pi_cb: %s\n",
 764                 emsg);
 765     abort_test (__LINE__);
 766     return;
 767   }
 768   p_id[i] = pinfo->result.id;
 769   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 770               "id: %s\n",
 771               GNUNET_i2s (p_id[i]));
 772   p_ids++;
 773   if (p_ids < 2)
 774     return;
 775   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 776               "Got all IDs, starting test\n");
 777   test_task = GNUNET_SCHEDULER_add_now (&start_test,
 778                                         NULL);
 779 }
 780
 781
 782 /**
 783  * test main: start test when all peers are connected
 784  *
 785  * @param cls Closure.
 786  * @param ctx Argument to give to GNUNET_CADET_TEST_cleanup on test end.
 787  * @param num_peers Number of peers that are running.
 788  * @param peers Array of peers.
 789  * @param cadets Handle to each of the CADETs of the peers.
 790  */
 791 static void
 792 tmain (void *cls,
 793        struct GNUNET_CADET_TEST_Context *ctx,
 794        unsigned int num_peers,
 795        struct GNUNET_TESTBED_Peer **peers,
 796        struct GNUNET_CADET_Handle **cadets)
 797 {
 798   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 799               "test main\n");
 800   test_ctx = ctx;
 801   peers_running = num_peers;
 802   GNUNET_assert (peers_running == peers_requested);
 803   testbed_peers = peers;
 804   h1 = cadets[0];
 805   h2 = cadets[num_peers - 1];
 806   GNUNET_SCHEDULER_add_shutdown (&shutdown_task,
 807                                  NULL);
 808   p_ids = 0;
 809   t_op[0] = GNUNET_TESTBED_peer_get_information (peers[0],
 810                                                  GNUNET_TESTBED_PIT_IDENTITY,
 811                                                  &pi_cb,
 812                                                  (void *) 0L);
 813   t_op[1] = GNUNET_TESTBED_peer_get_information (peers[num_peers - 1],
 814                                                  GNUNET_TESTBED_PIT_IDENTITY,
 815                                                  &pi_cb,
 816                                                  (void *) 1L);
 817   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 818               "requested peer ids\n");
 819 }
 820
 821
 822 /**
 823  * Main: start test
 824  */
 825 int
 826 main (int argc,
 827       char *argv[])
 828 {
 829   static const struct GNUNET_HashCode *ports[2];
 830   struct GNUNET_MQ_MessageHandler handlers[] = {
 831     GNUNET_MQ_hd_var_size (data,
 832                            GNUNET_MESSAGE_TYPE_DUMMY,
 833                            struct GNUNET_MessageHeader,
 834                            NULL),
 835     GNUNET_MQ_handler_end ()
 836   };
 837   const char *config_file = "test_cadet.conf";
 838   char port_id[] = "test port";
 839   struct GNUNET_GETOPT_CommandLineOption options[] = {
 840     GNUNET_GETOPT_option_relative_time ('t',
 841                                         "time",
 842                                         "short_time",
 843                                         gettext_noop ("set short timeout"),
 844                                         &short_time),
 845     GNUNET_GETOPT_option_uint ('m',
 846                                "messages",
 847                                "NUM_MESSAGES",
 848                                gettext_noop ("set number of messages to send"),
 849                                &total_packets),
 850     GNUNET_GETOPT_option_uint ('p',
 851                                "peers",
 852                                "NUM_PEERS",
 853                                gettext_noop ("number of peers to launch"),
 854                                &peers_requested),
 855     GNUNET_GETOPT_OPTION_END
 856   };
 857
 858   GNUNET_log_setup ("test-cadet-flow",
 859                     "DEBUG",
 860                     NULL);
 861   total_packets = TOTAL_PACKETS;
 862   short_time = SHORT_TIME;
 863   if (-1 == GNUNET_GETOPT_run (argv[0],
 864                                options,
 865                                argc,
 866                                argv))
 867   {
 868     fprintf (stderr,
 869              "test failed: problem with CLI parameters\n");
 870     return 1;
 871   }
 872   GNUNET_CRYPTO_hash (port_id,
 873                       sizeof(port_id),
 874                       &port);
 875   ports[0] = &port;
 876   ports[1] = NULL;
 877   GNUNET_CADET_TEST_ruN ("test_cadet_flow",
 878                          config_file,
 879                          peers_requested,
 880                          &tmain,
 881                          NULL,        /* tmain cls */
 882                          &connect_handler,
 883                          NULL,
 884                          &disconnect_handler,
 885                          handlers,
 886                          ports);
 887   return 0;
 888 }
 889
 890
 891 /* end of test_cadet_flow.c */