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