src/stream/perf_stream_api.c

   1  /*
   2      This file is part of GNUnet.
   3      (C) 2008--2013 Christian Grothoff (and other contributing authors)
   4
   5      GNUnet is free software; you can redistribute it and/or modify
   6      it under the terms of the GNU General Public License as published
   7      by the Free Software Foundation; either version 3, or (at your
   8      option) any later version.
   9
  10      GNUnet is distributed in the hope that it will be useful, but
  11      WITHOUT ANY WARRANTY; without even the implied warranty of
  12      MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13      General Public License for more details.
  14
  15      You should have received a copy of the GNU General Public License
  16      along with GNUnet; see the file COPYING.  If not, write to the
  17      Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  18      Boston, MA 02111-1307, USA.
  19 */
  20
  21 /**
  22  * @file stream/perf_stream_api.c
  23  * @brief performance benchmarks for stream api
  24  * @author Sree Harsha Totakura
  25  */
  26
  27 #define LOG(kind, ...)                         \
  28   GNUNET_log (kind, __VA_ARGS__);
  29
  30 /****************************************************************************************/
  31 /* Test is setup into the following major steps:                                        */
  32 /*    1. Measurements over loopback (1 hop). i.e. we use only one peer and open         */
  33 /*       stream connections over loopback. Messages will go through                     */
  34 /*       STREAM_API->MESH_API->MESH_SERVICE->MESH_API->STREAM_API.                      */
  35 /*    2. Measurements over 2 peers (2 hops). We use testbed to create 2 peers,          */
  36 /*       connect them and then create stream connections. Messages will go through      */
  37 /*       STREAM_API->MESH_API->MESH_SERVICE->CORE1.....CORE2->MESH_API->STREAM_API      */
  38 /*    3. Measurements over 3 peers (3 hops). We use testbed to create 3 peers,          */
  39 /*       connect them in a line topology: peer1->peer2->peer3. Messages will go         */
  40 /*       through                                                                        */
  41 /*       STREAM_API->MESH_API->MESH_SERVICE->CORE1..CORE2..CORE3->MESH_API->STREAM_API. */
  42 /****************************************************************************************/
  43
  44 #include "platform.h"
  45 #include "gnunet_common.h"
  46 #include "gnunet_util_lib.h"
  47 #include "gnunet_testbed_service.h"
  48 #include "gnunet_stream_lib.h"
  49
  50 /**
  51  * Simple struct to keep track of progress, and print a
  52  * nice little percentage meter for long running tasks.
  53  */
  54 struct ProgressMeter
  55 {
  56   unsigned int total;
  57
  58   unsigned int modnum;
  59
  60   unsigned int dotnum;
  61
  62   unsigned int completed;
  63
  64   int print;
  65
  66   char *startup_string;
  67 };
  68
  69
  70 /**
  71  * Steps in testing
  72  */
  73 enum TestStep
  74 {
  75   /**
  76    * Single hop loopback testing
  77    */
  78   TEST_STEP_1_HOP,
  79
  80   /**
  81    * Testing with 2 peers
  82    */
  83   TEST_STEP_2_HOP,
  84
  85   /**
  86    * Testing with 3 peers
  87    */
  88   TEST_STEP_3_HOP
  89 };
  90
  91
  92 /**
  93  * Structure for holding peer's sockets and IO Handles
  94  */
  95 struct PeerData
  96 {
  97   /**
  98    * Peer's stream socket
  99    */
 100   struct GNUNET_STREAM_Socket *socket;
 101
 102   /**
 103    * Peer's io write handle
 104    */
 105   struct GNUNET_STREAM_WriteHandle *io_write_handle;
 106
 107   /**
 108    * Peer's io read handle
 109    */
 110   struct GNUNET_STREAM_ReadHandle *io_read_handle;
 111
 112   /**
 113    * The peer handle when we use the testbed servie
 114    */
 115   struct GNUNET_TESTBED_Peer *peer;
 116
 117   /**
 118    * Handle to peer specific opearations while using testbed service
 119    */
 120   struct GNUNET_TESTBED_Operation *op;
 121
 122   /**
 123    * The identity of this peer
 124    */
 125   struct GNUNET_PeerIdentity id;
 126
 127   /**
 128    * Peer's shutdown handle
 129    */
 130   struct GNUNET_STREAM_ShutdownHandle *shutdown_handle;
 131
 132   /**
 133    * Bytes the peer has written
 134    */
 135   size_t bytes_wrote;
 136
 137   /**
 138    * Byte the peer has read
 139    */
 140   size_t bytes_read;
 141
 142   /**
 143    * number of packets sent
 144    */
 145   unsigned int packets_wrote;
 146
 147   /**
 148    * number of packets read
 149    */
 150   unsigned int packets_read;
 151 };
 152
 153
 154 /**
 155  * Enumeration of stages in this testing
 156  */
 157 enum TestStage
 158 {
 159   /**
 160    * The initial stage
 161    */
 162   INIT,
 163
 164   /**
 165    * Uplink testing stage
 166    */
 167   UPLINK_OK,
 168
 169   /**
 170    * Downlink testing stage
 171    */
 172   DOWNLINK_OK
 173 };
 174
 175
 176 /**
 177  * Maximum size of the data which we will transfer during tests
 178  */
 179 #define DATA_SIZE 5000000      /* 5mB */
 180
 181 /**
 182  * Fixed number of packets we send in each direction during each subtest
 183  */
 184 #define MAX_PACKETS 1000
 185
 186 /**
 187  * Listen socket of peer2
 188  */
 189 struct GNUNET_STREAM_ListenSocket *peer2_listen_socket;
 190
 191 /**
 192  * Handle to configuration during TEST_STEP_1_HOP
 193  */
 194 const struct GNUNET_CONFIGURATION_Handle *config;
 195
 196 /**
 197  * Handle for the progress meter
 198  */
 199 static struct ProgressMeter *meter;
 200
 201 /**
 202  * Placeholder for peer data
 203  */
 204 static struct PeerData peer_data[3];
 205
 206 /**
 207  * Handle to common operations while using testbed
 208  */
 209 static struct GNUNET_TESTBED_Operation *common_op;
 210
 211 /**
 212  * Task ID for abort task
 213  */
 214 static GNUNET_SCHEDULER_TaskIdentifier abort_task;
 215
 216 /**
 217  * Task ID for write task
 218  */
 219 static GNUNET_SCHEDULER_TaskIdentifier write_task;
 220
 221 /**
 222  * Task ID for read task
 223  */
 224 static GNUNET_SCHEDULER_TaskIdentifier read_task;
 225
 226 /**
 227  * Absolute time when profiling starts
 228  */
 229 static struct GNUNET_TIME_Absolute prof_start_time;
 230
 231 /**
 232  * Test time taken for sending the data
 233  */
 234 static struct GNUNET_TIME_Relative prof_time;
 235
 236 /**
 237  * Random data block. Should generate data first
 238  */
 239 static uint32_t data[DATA_SIZE / 4];
 240
 241 /**
 242  * Payload sizes to test each major test with
 243  */
 244 static uint16_t payload_size[] =
 245 { 20, 500, 2000, 7000, 13000, 25000, 30000};//, 50000, 60000, 63000, 64000 };
 246
 247 /**
 248  * Current step of testing
 249  */
 250 static enum TestStep test_step;
 251
 252 /**
 253  * Index for choosing payload size
 254  */
 255 static unsigned int payload_size_index;
 256
 257 /**
 258  * Number of peers we want to create while using the testbed service
 259  */
 260 static int num_peers;
 261
 262 /**
 263  * Flag to indicate that the other peer should reset its data read source index
 264  */
 265 static int reset_read;
 266
 267 /**
 268  * Testing result of a major test
 269  */
 270 static enum TestStage result;
 271
 272 /**
 273  * Create a meter to keep track of the progress of some task.
 274  *
 275  * @param total the total number of items to complete
 276  * @param start_string a string to prefix the meter with (if printing)
 277  * @param print GNUNET_YES to print the meter, GNUNET_NO to count
 278  *              internally only
 279  *
 280  * @return the progress meter
 281  */
 282 static struct ProgressMeter *
 283 create_meter (unsigned int total, char *start_string, int print)
 284 {
 285   struct ProgressMeter *ret;
 286
 287   ret = GNUNET_malloc (sizeof (struct ProgressMeter));
 288   ret->print = print;
 289   ret->total = total;
 290   ret->modnum = total / 4;
 291   if (ret->modnum == 0)         /* Divide by zero check */
 292     ret->modnum = 1;
 293   ret->dotnum = (total / 50) + 1;
 294   if (start_string != NULL)
 295     ret->startup_string = GNUNET_strdup (start_string);
 296   else
 297     ret->startup_string = GNUNET_strdup ("");
 298
 299   return ret;
 300 }
 301
 302
 303 /**
 304  * Update progress meter (increment by one).
 305  *
 306  * @param meter the meter to update and print info for
 307  *
 308  * @return GNUNET_YES if called the total requested,
 309  *         GNUNET_NO if more items expected
 310  */
 311 static int
 312 update_meter (struct ProgressMeter *meter)
 313 {
 314   if (meter->print == GNUNET_YES)
 315   {
 316     if (meter->completed % meter->modnum == 0)
 317     {
 318       if (meter->completed == 0)
 319       {
 320         FPRINTF (stdout, "%sProgress: [0%%", meter->startup_string);
 321       }
 322       else
 323         FPRINTF (stdout, "%d%%",
 324                  (int) (((float) meter->completed / meter->total) * 100));
 325     }
 326     else if (meter->completed % meter->dotnum == 0)
 327       FPRINTF (stdout, "%s",  ".");
 328
 329     if (meter->completed + 1 == meter->total)
 330       FPRINTF (stdout, "%d%%]\n", 100);
 331     fflush (stdout);
 332   }
 333   meter->completed++;
 334
 335   if (meter->completed == meter->total)
 336     return GNUNET_YES;
 337   if (meter->completed > meter->total)
 338     GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "Progress meter overflow!!\n");
 339   return GNUNET_NO;
 340 }
 341
 342
 343 /**
 344  * Reset progress meter.
 345  *
 346  * @param meter the meter to reset
 347  *
 348  * @return GNUNET_YES if meter reset,
 349  *         GNUNET_SYSERR on error
 350  */
 351 static int
 352 reset_meter (struct ProgressMeter *meter)
 353 {
 354   if (meter == NULL)
 355     return GNUNET_SYSERR;
 356
 357   meter->completed = 0;
 358   return GNUNET_YES;
 359 }
 360
 361
 362 /**
 363  * Release resources for meter
 364  *
 365  * @param meter the meter to free
 366  */
 367 static void
 368 free_meter (struct ProgressMeter *meter)
 369 {
 370   GNUNET_free_non_null (meter->startup_string);
 371   GNUNET_free (meter);
 372 }
 373
 374
 375 /**
 376  * Shutdown nicely
 377  */
 378 static void
 379 do_close (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 380 {
 381   switch (test_step)
 382   {
 383   case TEST_STEP_1_HOP:
 384     if (NULL != peer_data[0].socket)
 385       GNUNET_STREAM_close (peer_data[0].socket);
 386     if (NULL != peer_data[1].socket)
 387       GNUNET_STREAM_close (peer_data[1].socket);
 388     if (NULL != peer2_listen_socket)
 389       GNUNET_STREAM_listen_close (peer2_listen_socket); /* Close listen socket */
 390     break;
 391   case TEST_STEP_2_HOP:
 392     if (NULL != peer_data[1].socket)
 393       GNUNET_STREAM_close (peer_data[1].socket);
 394     if (NULL != peer_data[0].op)
 395       GNUNET_TESTBED_operation_done (peer_data[0].op);
 396     if (NULL != peer_data[1].op)
 397       GNUNET_TESTBED_operation_done (peer_data[1].op);
 398     break;
 399   case TEST_STEP_3_HOP:
 400     GNUNET_break (0);
 401   }
 402   if (GNUNET_SCHEDULER_NO_TASK != abort_task)
 403     GNUNET_SCHEDULER_cancel (abort_task);
 404   if (GNUNET_SCHEDULER_NO_TASK != write_task)
 405     GNUNET_SCHEDULER_cancel (write_task);
 406   GNUNET_SCHEDULER_shutdown (); /* Shutdown this testcase */
 407   if (NULL != meter)
 408   {
 409     free_meter (meter);
 410     meter = NULL;
 411   }
 412 }
 413
 414
 415 /**
 416  * Something went wrong and timed out. Kill everything and set error flag
 417  */
 418 static void
 419 do_abort (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 420 {
 421   abort_task = GNUNET_SCHEDULER_NO_TASK;
 422   GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "test: ABORT\n");
 423   if (GNUNET_SCHEDULER_NO_TASK != read_task)
 424     GNUNET_SCHEDULER_cancel (read_task);
 425   result = GNUNET_SYSERR;
 426   do_close (cls, tc);
 427 }
 428
 429
 430 /**
 431  * Completion callback for shutdown
 432  *
 433  * @param cls the closure from GNUNET_STREAM_shutdown call
 434  * @param operation the operation that was shutdown (SHUT_RD, SHUT_WR,
 435  *          SHUT_RDWR)
 436  */
 437 static void
 438 shutdown_completion (void *cls,
 439                      int operation)
 440 {
 441   static int shutdowns;
 442
 443   if (++shutdowns == 1)
 444   {
 445     peer_data[0].shutdown_handle = NULL;
 446     peer_data[1].shutdown_handle = GNUNET_STREAM_shutdown (peer_data[1].socket, SHUT_RDWR,
 447                                                            &shutdown_completion, cls);
 448     return;
 449   }
 450   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "STREAM shutdown successful\n");
 451   GNUNET_SCHEDULER_add_now (&do_close, cls);
 452 }
 453
 454
 455 /**
 456  * Shutdown sockets gracefully
 457  */
 458 static void
 459 do_shutdown (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 460 {
 461   peer_data[0].shutdown_handle = GNUNET_STREAM_shutdown (peer_data[0].socket, SHUT_RDWR,
 462                                                          &shutdown_completion, cls);
 463 }
 464
 465
 466 /**
 467  * Scheduler call back; to be executed when a new stream is connected
 468  * Called from listen connect for peer2
 469  */
 470 static void
 471 stream_read_task (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc);
 472
 473
 474 /**
 475  * Task for calling STREAM_write with a chunk of random data
 476  *
 477  * @param cls the peer data entity
 478  * @param tc the task context
 479  */
 480 static void
 481 stream_write_task (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc);
 482
 483
 484 /**
 485  * The write completion function; called upon writing some data to stream or
 486  * upon error
 487  *
 488  * @param cls the closure from GNUNET_STREAM_write/read
 489  * @param status the status of the stream at the time this function is called
 490  * @param size the number of bytes written
 491  */
 492 static void
 493 write_completion (void *cls, enum GNUNET_STREAM_Status status, size_t size)
 494 {
 495   struct PeerData *pdata = cls;
 496   double throughput;
 497   double prof_time_sec;
 498   unsigned int packets_wrote;
 499
 500   if (GNUNET_STREAM_OK != status)
 501   {
 502     GNUNET_SCHEDULER_cancel (abort_task);
 503     abort_task = GNUNET_SCHEDULER_add_now (&do_abort, NULL);
 504     return;
 505   }
 506   GNUNET_assert (size <= DATA_SIZE);
 507   packets_wrote = (size + payload_size[payload_size_index] - 1)
 508       / payload_size[payload_size_index];
 509   pdata->bytes_wrote += size;
 510   for (;packets_wrote > 0; packets_wrote--)
 511   {
 512     update_meter (meter);
 513     pdata->packets_wrote++;
 514   }
 515   if (pdata->packets_wrote < MAX_PACKETS) /* Have more data to send */
 516   {
 517     size_t write_amount;
 518
 519     if (GNUNET_SCHEDULER_NO_TASK != abort_task)
 520     {
 521       GNUNET_SCHEDULER_cancel (abort_task);
 522       abort_task =
 523           GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply
 524                                         (GNUNET_TIME_UNIT_SECONDS, 300), &do_abort,
 525                                     NULL);
 526     }
 527     write_amount = (MAX_PACKETS - pdata->packets_wrote) *
 528         payload_size[payload_size_index];
 529     if (write_amount > DATA_SIZE)
 530       write_amount = DATA_SIZE;
 531     reset_read = GNUNET_YES;
 532     pdata->io_write_handle = GNUNET_STREAM_write (pdata->socket, data,
 533                                                   write_amount,
 534                                                   GNUNET_TIME_UNIT_FOREVER_REL,
 535                                                   &write_completion, pdata);
 536     GNUNET_assert (NULL != pdata->io_write_handle);
 537   }
 538   else
 539   {
 540     free_meter (meter);
 541     meter = NULL;
 542     prof_time = GNUNET_TIME_absolute_get_duration (prof_start_time);
 543     prof_time_sec = (((double) prof_time.rel_value)/ ((double) 1000));
 544     throughput = ((float) pdata->bytes_wrote) / prof_time_sec;
 545     PRINTF ("Throughput %.2f kB/sec\n", throughput / 1000.00);
 546     switch (result)
 547     {
 548     case INIT:
 549       result = UPLINK_OK;
 550       GNUNET_assert (GNUNET_SCHEDULER_NO_TASK == read_task);
 551       GNUNET_assert (GNUNET_SCHEDULER_NO_TASK == write_task);
 552       pdata->bytes_read = 0;
 553       pdata->packets_read = 0;
 554       meter = create_meter (MAX_PACKETS, "Testing Downlink\n", GNUNET_YES);
 555       read_task = GNUNET_SCHEDULER_add_now (&stream_read_task, &peer_data[0]);
 556       write_task = GNUNET_SCHEDULER_add_now (&stream_write_task, &peer_data[1]);
 557       break;
 558     case UPLINK_OK:
 559       result = DOWNLINK_OK;
 560       GNUNET_SCHEDULER_add_now (&do_shutdown, NULL);
 561       break;
 562     case DOWNLINK_OK:
 563       GNUNET_assert (0);
 564     }
 565   }
 566 }
 567
 568
 569 /**
 570  * Task for calling STREAM_write with a chunk of random data
 571  *
 572  * @param cls the peer data entity
 573  * @param tc the task context
 574  */
 575 static void
 576 stream_write_task (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 577 {
 578   struct PeerData *pdata = cls;
 579   size_t write_amount;
 580
 581   if (GNUNET_SCHEDULER_NO_TASK != abort_task)
 582   {
 583     GNUNET_SCHEDULER_cancel (abort_task);
 584     abort_task =
 585       GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply
 586                                     (GNUNET_TIME_UNIT_SECONDS, 300), &do_abort,
 587                                     NULL);
 588   }
 589   write_task = GNUNET_SCHEDULER_NO_TASK;
 590   prof_start_time = GNUNET_TIME_absolute_get ();
 591   pdata->bytes_wrote = 0;
 592   pdata->packets_wrote = 0;
 593   write_amount = MAX_PACKETS * payload_size[payload_size_index];
 594   if (write_amount > DATA_SIZE)
 595     write_amount = DATA_SIZE;
 596   reset_read = GNUNET_YES;
 597   pdata->io_write_handle = GNUNET_STREAM_write (pdata->socket, data,
 598                                                 write_amount,
 599                                                 GNUNET_TIME_UNIT_FOREVER_REL,
 600                                                 &write_completion, pdata);
 601   GNUNET_assert (NULL != pdata->io_write_handle);
 602 }
 603
 604
 605 /**
 606  * Scheduler call back; to be executed when a new stream is connected
 607  * Called from listen connect for peer2
 608  */
 609 static void
 610 stream_read_task (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc);
 611
 612
 613 /**
 614  * Input processor
 615  *
 616  * @param cls peer2
 617  * @param status the status of the stream at the time this function is called
 618  * @param data traffic from the other side
 619  * @param size the number of bytes available in data read
 620  * @return number of bytes of processed from 'data' (any data remaining should be
 621  *         given to the next time the read processor is called).
 622  */
 623 static size_t
 624 input_processor (void *cls, enum GNUNET_STREAM_Status status,
 625                  const void *input_data, size_t size)
 626 {
 627   struct PeerData *pdata = cls;
 628
 629   if (GNUNET_STREAM_OK != status)
 630   {
 631     GNUNET_SCHEDULER_cancel (abort_task);
 632     abort_task = GNUNET_SCHEDULER_add_now (&do_abort, NULL);
 633     return 0;
 634   }
 635   GNUNET_assert (size <= DATA_SIZE);
 636   if (GNUNET_YES == reset_read)
 637   {
 638     pdata->bytes_read = 0;
 639     reset_read = GNUNET_NO;
 640   }
 641   GNUNET_assert ((pdata->bytes_read + size) <= DATA_SIZE);
 642   GNUNET_assert (0 == memcmp (((void *)data ) + pdata->bytes_read,
 643                               input_data, size));
 644   pdata->bytes_read += size;
 645   pdata->packets_read += (size + payload_size[payload_size_index] - 1)
 646       / payload_size[payload_size_index];
 647   if (pdata->packets_read < MAX_PACKETS)
 648   {
 649     GNUNET_assert (GNUNET_SCHEDULER_NO_TASK == read_task);
 650     read_task = GNUNET_SCHEDULER_add_now (&stream_read_task, pdata);
 651   }
 652   else
 653   {
 654     LOG (GNUNET_ERROR_TYPE_DEBUG, "Reading finished successfully\n");
 655   }
 656   return size;
 657 }
 658
 659
 660 /**
 661  * Scheduler call back; to be executed when a new stream is connected
 662  * Called from listen connect for peer2
 663  */
 664 static void
 665 stream_read_task (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 666 {
 667   struct PeerData *pdata = cls;
 668
 669   read_task = GNUNET_SCHEDULER_NO_TASK;
 670   pdata->io_read_handle =
 671       GNUNET_STREAM_read (pdata->socket, GNUNET_TIME_UNIT_FOREVER_REL,
 672                           &input_processor, pdata);
 673   GNUNET_assert (NULL != pdata->io_read_handle);
 674 }
 675
 676
 677 /**
 678  * Functions of this type are called upon new stream connection from other peers
 679  *
 680  * @param cls the closure from GNUNET_STREAM_listen
 681  * @param socket the socket representing the stream
 682  * @param initiator the identity of the peer who wants to establish a stream
 683  *            with us
 684  * @return GNUNET_OK to keep the socket open, GNUNET_SYSERR to close the
 685  *             stream (the socket will be invalid after the call)
 686  */
 687 static int
 688 stream_listen_cb (void *cls, struct GNUNET_STREAM_Socket *socket,
 689                   const struct GNUNET_PeerIdentity *initiator)
 690 {
 691   struct PeerData *pdata = cls;
 692
 693
 694   if ((NULL == socket) || (NULL == initiator))
 695   {
 696     GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "Binding error\n");
 697     if (GNUNET_SCHEDULER_NO_TASK != abort_task)
 698       GNUNET_SCHEDULER_cancel (abort_task);
 699     abort_task = GNUNET_SCHEDULER_add_now (&do_abort, NULL);
 700     return GNUNET_OK;
 701   }
 702   GNUNET_assert (NULL != socket);
 703   GNUNET_assert (pdata == &peer_data[1]);
 704   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Peer connected: %s\n",
 705               GNUNET_i2s(initiator));
 706   pdata->socket = socket;
 707   pdata->bytes_read = 0;
 708   read_task = GNUNET_SCHEDULER_add_now (&stream_read_task, pdata);
 709   return GNUNET_OK;
 710 }
 711
 712
 713 /**
 714  * Function executed after stream has been established
 715  *
 716  * @param cls the closure from GNUNET_STREAM_open
 717  * @param socket socket to use to communicate with the other side (read/write)
 718  */
 719 static void
 720 stream_open_cb (void *cls,
 721                 struct GNUNET_STREAM_Socket *socket)
 722 {
 723   struct PeerData *pdata = cls;
 724
 725   GNUNET_assert (socket == pdata->socket);
 726   meter = create_meter (MAX_PACKETS, "Testing Uplink\n", GNUNET_YES);
 727   write_task = GNUNET_SCHEDULER_add_now (&stream_write_task, pdata);
 728 }
 729
 730
 731 /**
 732  * Listen success callback; connects a peer to stream as client
 733  */
 734 static void
 735 stream_connect (void)
 736 {
 737   peer_data[0].socket =
 738       GNUNET_STREAM_open (config, &peer_data[1].id, 10, &stream_open_cb,
 739                           &peer_data[0],
 740                           GNUNET_STREAM_OPTION_MAX_PAYLOAD_SIZE,
 741                           payload_size[payload_size_index],
 742                           GNUNET_STREAM_OPTION_END);
 743   GNUNET_assert (NULL != peer_data[0].socket);
 744 }
 745
 746
 747 /**
 748  * Initialize framework and start test
 749  *
 750  * @param cls closure
 751  * @param cfg configuration of the peer that was started
 752  * @param peer identity of the peer that was created
 753  */
 754 static void
 755 run (void *cls,
 756      const struct GNUNET_CONFIGURATION_Handle *cfg,
 757      struct GNUNET_TESTING_Peer *peer)
 758 {
 759   struct GNUNET_PeerIdentity id;
 760
 761   GNUNET_TESTING_peer_get_identity (peer, &id);
 762   config = cfg;
 763   peer2_listen_socket =
 764       GNUNET_STREAM_listen (config, 10, &stream_listen_cb, &peer_data[1],
 765                             GNUNET_STREAM_OPTION_SIGNAL_LISTEN_SUCCESS,
 766                             &stream_connect,
 767                             GNUNET_STREAM_OPTION_MAX_PAYLOAD_SIZE,
 768                             payload_size[payload_size_index],
 769                             GNUNET_STREAM_OPTION_END);
 770   GNUNET_assert (NULL != peer2_listen_socket);
 771   peer_data[0].id = id;
 772   peer_data[1].id = id;
 773   abort_task =
 774       GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply
 775                                     (GNUNET_TIME_UNIT_SECONDS, 300), &do_abort,
 776                                     NULL);
 777 }
 778
 779
 780 /**
 781  * Adapter function called to establish a connection to
 782  * a service.
 783  *
 784  * @param cls closure
 785  * @param cfg configuration of the peer to connect to; will be available until
 786  *          GNUNET_TESTBED_operation_done() is called on the operation returned
 787  *          from GNUNET_TESTBED_service_connect()
 788  * @return service handle to return in 'op_result', NULL on error
 789  */
 790 static void *
 791 stream_ca (void *cls, const struct GNUNET_CONFIGURATION_Handle *cfg);
 792
 793
 794 /**
 795  * Adapter function called to destroy a connection to
 796  * a service.
 797  *
 798  * @param cls closure
 799  * @param op_result service handle returned from the connect adapter
 800  */
 801 static void
 802 stream_da (void *cls, void *op_result)
 803 {
 804   if (&peer_data[1] == cls)
 805   {
 806     GNUNET_STREAM_listen_close (op_result);
 807     return;
 808   }
 809   else if (&peer_data[0] == cls)
 810   {
 811     GNUNET_STREAM_close (op_result);
 812     return;
 813   }
 814   GNUNET_assert (0);
 815 }
 816
 817
 818 /**
 819  * Listen success callback; connects a peer to stream as client. Called from
 820  * testbed stream_ca
 821  */
 822 static void
 823 stream_connect2 (void)
 824 {
 825   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Stream listen open successful\n");
 826   peer_data[0].op =
 827       GNUNET_TESTBED_service_connect (&peer_data[0], peer_data[0].peer,
 828                                       "stream", NULL, NULL, stream_ca,
 829                                       stream_da, &peer_data[0]);
 830 }
 831
 832
 833 /**
 834  * Adapter function called to establish a connection to
 835  * a service.
 836  *
 837  * @param cls closure
 838  * @param cfg configuration of the peer to connect to; will be available until
 839  *          GNUNET_TESTBED_operation_done() is called on the operation returned
 840  *          from GNUNET_TESTBED_service_connect()
 841  * @return service handle to return in 'op_result', NULL on error
 842  */
 843 static void *
 844 stream_ca (void *cls, const struct GNUNET_CONFIGURATION_Handle *cfg)
 845 {
 846   struct PeerData *pdata = cls;
 847
 848   if (&peer_data[1] == pdata)
 849   {
 850     peer2_listen_socket = NULL;
 851     peer2_listen_socket =
 852         GNUNET_STREAM_listen (cfg, 10, &stream_listen_cb, &peer_data[1],
 853                               GNUNET_STREAM_OPTION_SIGNAL_LISTEN_SUCCESS,
 854                               &stream_connect2,
 855                               GNUNET_STREAM_OPTION_MAX_PAYLOAD_SIZE,
 856                               payload_size[payload_size_index],
 857                               GNUNET_STREAM_OPTION_END);
 858     GNUNET_assert (NULL != peer2_listen_socket);
 859     return peer2_listen_socket;
 860   }
 861   if (&peer_data[0] == pdata)
 862   {
 863     pdata->socket =
 864         GNUNET_STREAM_open (cfg, &peer_data[1].id, 10, &stream_open_cb,
 865                             &peer_data[0],
 866                             GNUNET_STREAM_OPTION_MAX_PAYLOAD_SIZE,
 867                             payload_size[payload_size_index],
 868                             GNUNET_STREAM_OPTION_END);
 869     GNUNET_assert (NULL != pdata->socket);
 870     return pdata->socket;
 871   }
 872   GNUNET_assert (0);
 873   return NULL;
 874 }
 875
 876
 877 /**
 878  * Callback to be called when the requested peer information is available
 879  *
 880  * @param cb_cls the closure from GNUNET_TETSBED_peer_get_information()
 881  * @param op the operation this callback corresponds to
 882  * @param pinfo the result; will be NULL if the operation has failed
 883  * @param emsg error message if the operation has failed; will be NULL if the
 884  *          operation is successfull
 885  */
 886 static void
 887 peerinfo_cb (void *cb_cls, struct GNUNET_TESTBED_Operation *op,
 888              const struct GNUNET_TESTBED_PeerInformation *pinfo,
 889              const char *emsg)
 890 {
 891   struct PeerData *pdata = cb_cls;
 892
 893   GNUNET_assert (NULL == emsg);
 894   GNUNET_assert (common_op == op);
 895   GNUNET_assert (NULL != pdata);
 896   memcpy (&pdata->id, pinfo->result.id, sizeof (struct GNUNET_PeerIdentity));
 897   GNUNET_TESTBED_operation_done (op);
 898   if (pdata == &peer_data[0])
 899   {
 900     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Peer 1 id: %s\n",
 901                 GNUNET_i2s (&pdata->id));
 902     common_op = GNUNET_TESTBED_peer_get_information (peer_data[1].peer,
 903                                                      GNUNET_TESTBED_PIT_IDENTITY,
 904                                                      &peerinfo_cb, &peer_data[1]);
 905   }
 906   else if (pdata == &peer_data[1])
 907   {
 908     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Peer 2 id: %s\n",
 909                 GNUNET_i2s (&pdata->id));
 910     if (TEST_STEP_2_HOP == test_step)
 911       peer_data[1].op =
 912           GNUNET_TESTBED_service_connect (&peer_data[1], peer_data[1].peer,
 913                                           "stream", NULL, NULL, stream_ca,
 914                                           stream_da, &peer_data[1]);
 915     else
 916       GNUNET_break (0);         /* FIXME: 3 hop test case here... */
 917   }
 918 }
 919
 920
 921 /**
 922  * Controller event callback
 923  *
 924  * @param cls NULL
 925  * @param event the controller event
 926  */
 927 static void
 928 controller_event_cb (void *cls,
 929                      const struct GNUNET_TESTBED_EventInformation *event)
 930 {
 931   switch (event->type)
 932   {
 933   case GNUNET_TESTBED_ET_OPERATION_FINISHED:
 934     if (NULL != event->details.operation_finished.emsg)
 935     {
 936       FPRINTF (stderr, "Error while expecting an operation to succeed:%s \n",
 937                event->details.operation_finished.emsg);
 938       GNUNET_assert (0);
 939     }
 940     break;
 941   default:
 942     GNUNET_assert (0);
 943   }
 944 }
 945
 946
 947 /**
 948  * Signature of a main function for a testcase.
 949  *
 950  * @param cls closure
 951  * @param num_peers number of peers in 'peers'
 952  * @param peers handle to peers run in the testbed
 953  * @param links_succeeded the number of overlay link connection attempts that
 954  *          succeeded
 955  * @param links_failed the number of overlay link connection attempts that
 956  *          failed
 957  */
 958 static void
 959 test_master (void *cls, unsigned int num_peers_,
 960              struct GNUNET_TESTBED_Peer **peers,
 961              unsigned int links_succeeded,
 962              unsigned int links_failed)
 963 {
 964   GNUNET_assert (NULL != peers);
 965   GNUNET_assert (NULL != peers[0]);
 966   GNUNET_assert (NULL != peers[1]);
 967   GNUNET_assert (num_peers_ == num_peers);
 968   peer_data[0].peer = peers[0];
 969   peer_data[1].peer = peers[1];
 970   if (2 == num_peers)
 971         /* Get the peer identity and configuration of peers */
 972     common_op =
 973         GNUNET_TESTBED_peer_get_information (peer_data[0].peer,
 974                                              GNUNET_TESTBED_PIT_IDENTITY,
 975                                              &peerinfo_cb, &peer_data[0]);
 976   else
 977     GNUNET_break (0);
 978   abort_task =
 979       GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply
 980                                     (GNUNET_TIME_UNIT_SECONDS, 120), &do_abort,
 981                                     NULL);
 982 }
 983
 984
 985 /**
 986  * Main function
 987  */
 988 int main (int argc, char **argv)
 989 {
 990   char *test_name = "perf_stream_api";
 991   char *cfg_file = "test_stream_local.conf";
 992   uint64_t event_mask;
 993   unsigned int count;
 994   int ret;
 995
 996   meter = create_meter ((sizeof (data) / 4), "Generating random data\n", GNUNET_YES);
 997   for (count=0; count < (sizeof (data) / 4); count++)
 998   {
 999     data[count] = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
1000                                             UINT32_MAX);
1001     update_meter (meter);
1002   }
1003   reset_meter (meter);
1004   free_meter (meter);
1005   meter = NULL;
1006   test_step = TEST_STEP_1_HOP;
1007   for (payload_size_index = 0;
1008        payload_size_index < (sizeof (payload_size) / sizeof (uint16_t));
1009        payload_size_index++)
1010   {
1011     PRINTF ("\nTesting over loopback with payload size %hu\n",
1012             payload_size[payload_size_index]);
1013     (void) memset (peer_data, 0, sizeof (peer_data));
1014     result = INIT;
1015     reset_read = GNUNET_NO;
1016     ret = GNUNET_TESTING_peer_run (test_name, cfg_file, &run, NULL);
1017     if ((0 != ret) || (DOWNLINK_OK != result))
1018       goto return_fail;
1019   }
1020   test_step = TEST_STEP_2_HOP;
1021   num_peers = 2;
1022   event_mask = 0;
1023   event_mask |= (1LL << GNUNET_TESTBED_ET_OPERATION_FINISHED);
1024   for (payload_size_index = 0;
1025        payload_size_index < (sizeof (payload_size) / sizeof (uint16_t));
1026        payload_size_index++)
1027   {
1028     PRINTF ("\nTesting over 1 hop with payload size %hu\n",
1029             payload_size[payload_size_index]);
1030     (void) memset (peer_data, 0, sizeof (peer_data));
1031     result = INIT;
1032     reset_read = GNUNET_NO;
1033     (void) GNUNET_TESTBED_test_run (test_name, cfg_file, num_peers, event_mask,
1034                                     &controller_event_cb, NULL, &test_master,
1035                                     NULL);
1036     if (DOWNLINK_OK != result)
1037       goto return_fail;
1038   }
1039   test_step = TEST_STEP_3_HOP;
1040   for (payload_size_index = 0;
1041        payload_size_index < (sizeof (payload_size) / sizeof (uint16_t));
1042        payload_size_index++)
1043   {
1044     /* Initialize testbed here */
1045   }
1046   return 0;
1047
1048  return_fail:
1049   GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "Test failed\n");
1050   return 1;
1051 }
1052
1053 /* end of perf_stream_api.c */