src/stream/test_stream_2peers.c

   1 /*
   2      This file is part of GNUnet.
   3      (C) 2011, 2012 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/test_stream_2peers.c
  23  * @brief Stream API testing between 2 peers using testing API
  24  * @author Sree Harsha Totakura
  25  */
  26
  27 #include <string.h>
  28
  29 #include "platform.h"
  30 #include "gnunet_util_lib.h"
  31 #include "gnunet_mesh_service.h"
  32 #include "gnunet_stream_lib.h"
  33 #include "gnunet_testbed_service.h"
  34
  35 /**
  36  * Number of peers; Do NOT change this
  37  */
  38 #define NUM_PEERS 2
  39
  40 /**
  41  * Shorthand for Relative time in seconds
  42  */
  43 #define TIME_REL_SECS(sec) \
  44   GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, sec)
  45
  46 /**
  47  * Structure for holding peer's sockets and IO Handles
  48  */
  49 struct PeerData
  50 {
  51   /**
  52    * Handle to testbed peer
  53    */
  54   struct GNUNET_TESTBED_Peer *peer;
  55
  56   /**
  57    * Peer's stream socket
  58    */
  59   struct GNUNET_STREAM_Socket *socket;
  60
  61   /**
  62    * Peer's io write handle
  63    */
  64   struct GNUNET_STREAM_IOWriteHandle *io_write_handle;
  65
  66   /**
  67    * Peer's io read handle
  68    */
  69   struct GNUNET_STREAM_IOReadHandle *io_read_handle;
  70
  71   /**
  72    * Peer's shutdown handle
  73    */
  74   struct GNUNET_STREAM_ShutdownHandle *shutdown_handle;
  75
  76   /**
  77    * The service connect operation to stream
  78    */
  79   struct GNUNET_TESTBED_Operation *op;
  80
  81   /**
  82    * Our Peer id
  83    */
  84   struct GNUNET_PeerIdentity our_id;
  85
  86   /**
  87    * Bytes the peer has written
  88    */
  89   unsigned int bytes_wrote;
  90
  91   /**
  92    * Byte the peer has read
  93    */
  94   unsigned int bytes_read;
  95 };
  96
  97
  98 /**
  99  * Different states in test setup
 100  */
 101 enum SetupState
 102 {
 103   /**
 104    * The initial state
 105    */
 106   INIT,
 107
 108   /**
 109    * Get the identity of peer 1
 110    */
 111   PEER1_GET_IDENTITY,
 112
 113   /**
 114    * Get the identity of peer 2
 115    */
 116   PEER2_GET_IDENTITY,
 117
 118   /**
 119    * Connect to stream service of peer 1
 120    */
 121   PEER1_STREAM_CONNECT,
 122
 123   /**
 124    * Connect to stream service of peer 2
 125    */
 126   PEER2_STREAM_CONNECT
 127
 128 };
 129
 130 /**
 131  * Various states during test setup
 132  */
 133 static enum SetupState setup_state;
 134
 135 /**
 136  * Data context for peer 1
 137  */
 138 static struct PeerData peer1;
 139
 140 /**
 141  * Data context for peer 2
 142  */
 143 static struct PeerData peer2;
 144
 145 /**
 146  * Testbed operation handle
 147  */
 148 static struct GNUNET_TESTBED_Operation *op;
 149
 150 static GNUNET_SCHEDULER_TaskIdentifier abort_task;
 151
 152 static char *data = "ABCD";
 153 static int result;
 154
 155 static int writing_success;
 156 static int reading_success;
 157
 158
 159 /**
 160  * Input processor
 161  *
 162  * @param cls the closure from GNUNET_STREAM_write/read
 163  * @param status the status of the stream at the time this function is called
 164  * @param data traffic from the other side
 165  * @param size the number of bytes available in data read
 166  * @return number of bytes of processed from 'data' (any data remaining should be
 167  *         given to the next time the read processor is called).
 168  */
 169 static size_t
 170 input_processor (void *cls,
 171                  enum GNUNET_STREAM_Status status,
 172                  const void *input_data,
 173                  size_t size);
 174
 175 /**
 176  * Task for calling STREAM_read
 177  *
 178  * @param cls the peer data entity
 179  * @param tc the task context
 180  */
 181 static void
 182 stream_read_task (void *cls,
 183                   const struct GNUNET_SCHEDULER_TaskContext *tc)
 184 {
 185   struct PeerData *peer = cls;
 186
 187   peer->io_read_handle = GNUNET_STREAM_read (peer->socket,
 188                                              GNUNET_TIME_relative_multiply
 189                                              (GNUNET_TIME_UNIT_SECONDS, 5),
 190                                              &input_processor,
 191                                              peer);
 192   GNUNET_assert (NULL != peer->io_read_handle);
 193 }
 194
 195 /**
 196  * The write completion function; called upon writing some data to stream or
 197  * upon error
 198  *
 199  * @param cls the closure from GNUNET_STREAM_write/read
 200  * @param status the status of the stream at the time this function is called
 201  * @param size the number of bytes read or written
 202  */
 203 static void
 204 write_completion (void *cls,
 205                   enum GNUNET_STREAM_Status status,
 206                   size_t size);
 207
 208
 209 /**
 210  * Task for calling STREAM_write
 211  *
 212  * @param cls the peer data entity
 213  * @param tc the task context
 214  */
 215 static void
 216 stream_write_task (void *cls,
 217                    const struct GNUNET_SCHEDULER_TaskContext *tc)
 218 {
 219   struct PeerData *peer = cls;
 220
 221   peer->io_write_handle =
 222     GNUNET_STREAM_write (peer->socket,
 223                          (void *) data,
 224                          strlen(data) - peer->bytes_wrote,
 225                          GNUNET_TIME_relative_multiply
 226                          (GNUNET_TIME_UNIT_SECONDS, 5),
 227                          &write_completion,
 228                          peer);
 229
 230   GNUNET_assert (NULL != peer->io_write_handle);
 231  }
 232
 233
 234 /**
 235  * Close sockets and stop testing deamons nicely
 236  */
 237 static void
 238 do_close (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 239 {
 240   if (GNUNET_SCHEDULER_NO_TASK != abort_task)
 241     GNUNET_SCHEDULER_cancel (abort_task);
 242   if (NULL != peer1.socket)
 243     GNUNET_STREAM_close (peer1.socket);
 244   if (NULL != peer1.op)
 245     GNUNET_TESTBED_operation_done (peer1.op);
 246   else
 247     GNUNET_SCHEDULER_shutdown (); /* For shutting down testbed */
 248 }
 249
 250
 251 /**
 252  * Completion callback for shutdown
 253  *
 254  * @param cls the closure from GNUNET_STREAM_shutdown call
 255  * @param operation the operation that was shutdown (SHUT_RD, SHUT_WR,
 256  *          SHUT_RDWR)
 257  */
 258 static void
 259 shutdown_completion (void *cls,
 260                      int operation)
 261 {
 262   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "STREAM shutdown successful\n");
 263   GNUNET_SCHEDULER_add_now (&do_close, cls);
 264 }
 265
 266
 267 /**
 268  * Shutdown sockets gracefully
 269  */
 270 static void
 271 do_shutdown (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 272 {
 273   result = GNUNET_OK;
 274   peer1.shutdown_handle = GNUNET_STREAM_shutdown (peer1.socket, SHUT_RDWR,
 275                                                   &shutdown_completion, cls);
 276 }
 277
 278
 279 /**
 280  * Something went wrong and timed out. Kill everything and set error flag
 281  */
 282 static void
 283 do_abort (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 284 {
 285   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "test: ABORT\n");
 286   result = GNUNET_SYSERR;
 287   abort_task = 0;
 288   do_close (cls, tc);
 289 }
 290
 291
 292 /**
 293  * The write completion function; called upon writing some data to stream or
 294  * upon error
 295  *
 296  * @param cls the closure from GNUNET_STREAM_write/read
 297  * @param status the status of the stream at the time this function is called
 298  * @param size the number of bytes read or written
 299  */
 300 static void
 301 write_completion (void *cls,
 302                   enum GNUNET_STREAM_Status status,
 303                   size_t size)
 304 {
 305   struct PeerData *peer=cls;
 306
 307   GNUNET_assert (GNUNET_STREAM_OK == status);
 308   GNUNET_assert (size <= strlen (data));
 309   peer->bytes_wrote += size;
 310
 311   if (peer->bytes_wrote < strlen(data)) /* Have more data to send */
 312     {
 313       GNUNET_SCHEDULER_add_now (&stream_write_task, peer);
 314     }
 315   else
 316     {
 317       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 318                   "Writing completed\n");
 319
 320       if (&peer2 == peer)   /* Peer1 has finished writing; should read now */
 321         {
 322           peer->bytes_read = 0;
 323           GNUNET_SCHEDULER_add_now (&stream_read_task, peer);
 324         }
 325       else
 326         {
 327           writing_success = GNUNET_YES;
 328           if (GNUNET_YES == reading_success)
 329             GNUNET_SCHEDULER_add_now (&do_shutdown, NULL);
 330         }
 331     }
 332 }
 333
 334
 335 /**
 336  * Function executed after stream has been established
 337  *
 338  * @param cls the closure from GNUNET_STREAM_open
 339  * @param socket socket to use to communicate with the other side (read/write)
 340  */
 341 static void
 342 stream_open_cb (void *cls,
 343                 struct GNUNET_STREAM_Socket *socket)
 344 {
 345   struct PeerData *peer=cls;
 346
 347   GNUNET_assert (&peer2 == peer);
 348   GNUNET_assert (socket == peer2.socket);
 349   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "%s: Stream established from peer2\n",
 350               GNUNET_i2s (&peer1.our_id));
 351   peer->bytes_wrote = 0;
 352   GNUNET_SCHEDULER_add_now (&stream_write_task, peer);
 353 }
 354
 355
 356 /**
 357  * Input processor
 358  *
 359  * @param cls the closure from GNUNET_STREAM_write/read
 360  * @param status the status of the stream at the time this function is called
 361  * @param data traffic from the other side
 362  * @param size the number of bytes available in data read
 363  * @return number of bytes of processed from 'data' (any data remaining should be
 364  *         given to the next time the read processor is called).
 365  */
 366 static size_t
 367 input_processor (void *cls,
 368                  enum GNUNET_STREAM_Status status,
 369                  const void *input_data,
 370                  size_t size)
 371 {
 372   struct PeerData *peer;
 373
 374   peer = (struct PeerData *) cls;
 375
 376   if (GNUNET_STREAM_TIMEOUT == status)
 377     {
 378       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 379                   "Read operation timedout - reading again!\n");
 380       GNUNET_assert (0 == size);
 381       GNUNET_SCHEDULER_add_now (&stream_read_task, peer);
 382       return 0;
 383     }
 384
 385   GNUNET_assert (GNUNET_STREAM_OK == status);
 386   GNUNET_assert (size <= strlen (data));
 387   GNUNET_assert (0 == strncmp ((const char *) data + peer->bytes_read,
 388                                (const char *) input_data,
 389                                size));
 390   peer->bytes_read += size;
 391
 392   if (peer->bytes_read < strlen (data))
 393     {
 394       GNUNET_SCHEDULER_add_now (&stream_read_task, peer);
 395     }
 396   else
 397     {
 398       if (&peer1 == peer)    /* Peer2 has completed reading; should write */
 399         {
 400           peer->bytes_wrote = 0;
 401           GNUNET_SCHEDULER_add_now (&stream_write_task, peer);
 402         }
 403       else                      /* Peer1 has completed reading. End of tests */
 404         {
 405           reading_success = GNUNET_YES;
 406           if (GNUNET_YES == writing_success)
 407             GNUNET_SCHEDULER_add_now (&do_shutdown, NULL);
 408         }
 409     }
 410   return size;
 411 }
 412
 413
 414 /**
 415  * Functions of this type are called upon new stream connection from other peers
 416  *
 417  * @param cls the closure from GNUNET_STREAM_listen
 418  * @param socket the socket representing the stream
 419  * @param initiator the identity of the peer who wants to establish a stream
 420  *            with us
 421  * @return GNUNET_OK to keep the socket open, GNUNET_SYSERR to close the
 422  *             stream (the socket will be invalid after the call)
 423  */
 424 static int
 425 stream_listen_cb (void *cls,
 426                   struct GNUNET_STREAM_Socket *socket,
 427                   const struct GNUNET_PeerIdentity *initiator)
 428 {
 429   if ((NULL == socket) || (NULL == initiator))
 430   {
 431     GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "Binding error\n");
 432     if (GNUNET_SCHEDULER_NO_TASK != abort_task)
 433       GNUNET_SCHEDULER_cancel (abort_task);
 434     abort_task = GNUNET_SCHEDULER_add_now (&do_abort, NULL);
 435     return GNUNET_OK;
 436   }
 437   GNUNET_assert (NULL != initiator);
 438   GNUNET_assert (socket != peer2.socket);
 439   GNUNET_assert (0 == memcmp (initiator, &peer2.our_id,
 440                               sizeof (struct GNUNET_PeerIdentity)));
 441   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "%s: Peer connected: %s\n",
 442               GNUNET_i2s (&peer1.our_id), GNUNET_i2s (initiator));
 443   peer1.socket = socket;
 444   peer1.bytes_read = 0;
 445   GNUNET_SCHEDULER_add_now (&stream_read_task, &peer1);
 446   return GNUNET_OK;
 447 }
 448
 449
 450 /**
 451  * Listen success callback; connects a peer to stream as client
 452  */
 453 static void stream_connect (void);
 454
 455
 456 /**
 457  * Adapter function called to destroy a connection to
 458  * a service.
 459  *
 460  * @param cls closure
 461  * @param op_result service handle returned from the connect adapter
 462  */
 463 static void
 464 stream_da (void *cls, void *op_result)
 465 {
 466   struct GNUNET_STREAM_ListenSocket *lsocket;
 467   struct GNUNET_STREAM_Socket *socket;
 468
 469   if (&peer1 == cls)
 470   {
 471     lsocket = op_result;
 472     GNUNET_STREAM_listen_close (lsocket);
 473     if (NULL != peer2.op)
 474       GNUNET_TESTBED_operation_done (peer2.op);
 475     else
 476       GNUNET_SCHEDULER_shutdown ();
 477     return;
 478   }
 479   if (&peer2 == cls)
 480   {
 481     socket = op_result;
 482     GNUNET_STREAM_close (socket);
 483     GNUNET_SCHEDULER_shutdown (); /* Exit point of the test */
 484     return;
 485   }
 486   GNUNET_assert (0);
 487 }
 488
 489
 490 /**
 491  * Adapter function called to establish a connection to
 492  * a service.
 493  *
 494  * @param cls closure
 495  * @param cfg configuration of the peer to connect to; will be available until
 496  *          GNUNET_TESTBED_operation_done() is called on the operation returned
 497  *          from GNUNET_TESTBED_service_connect()
 498  * @return service handle to return in 'op_result', NULL on error
 499  */
 500 static void *
 501 stream_ca (void *cls, const struct GNUNET_CONFIGURATION_Handle *cfg)
 502 {
 503   struct GNUNET_STREAM_ListenSocket *lsocket;
 504
 505   switch (setup_state)
 506   {
 507   case PEER1_STREAM_CONNECT:
 508     lsocket = GNUNET_STREAM_listen (cfg, 10, &stream_listen_cb, NULL,
 509                                     GNUNET_STREAM_OPTION_SIGNAL_LISTEN_SUCCESS,
 510                                     &stream_connect, GNUNET_STREAM_OPTION_END);
 511     return lsocket;
 512   case PEER2_STREAM_CONNECT:
 513     peer2.socket = GNUNET_STREAM_open (cfg, &peer1.our_id, 10, &stream_open_cb,
 514                                        &peer2, GNUNET_STREAM_OPTION_END);
 515     return peer2.socket;
 516   default:
 517     GNUNET_assert (0);
 518   }
 519 }
 520
 521
 522 /**
 523  * Listen success callback; connects a peer to stream as client
 524  */
 525 static void
 526 stream_connect (void)
 527 {
 528   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Stream listen open successful\n");
 529   peer2.op = GNUNET_TESTBED_service_connect (&peer2, peer2.peer, "stream",
 530                                              NULL, NULL,
 531                                              stream_ca, stream_da, &peer2);
 532   setup_state = PEER2_STREAM_CONNECT;
 533 }
 534
 535
 536 /**
 537  * Callback to be called when the requested peer information is available
 538  *
 539  * @param cb_cls the closure from GNUNET_TETSBED_peer_get_information()
 540  * @param op the operation this callback corresponds to
 541  * @param pinfo the result; will be NULL if the operation has failed
 542  * @param emsg error message if the operation has failed; will be NULL if the
 543  *          operation is successfull
 544  */
 545 static void
 546 peerinfo_cb (void *cb_cls, struct GNUNET_TESTBED_Operation *op_,
 547              const struct GNUNET_TESTBED_PeerInformation *pinfo,
 548              const char *emsg)
 549 {
 550   GNUNET_assert (NULL == emsg);
 551   GNUNET_assert (op == op_);
 552   switch (setup_state)
 553     {
 554     case PEER1_GET_IDENTITY:
 555       memcpy (&peer1.our_id, pinfo->result.id,
 556               sizeof (struct GNUNET_PeerIdentity));
 557       GNUNET_TESTBED_operation_done (op);
 558       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Peer 1 id: %s\n", GNUNET_i2s
 559                   (&peer1.our_id));
 560       op = GNUNET_TESTBED_peer_get_information (peer2.peer,
 561                                                 GNUNET_TESTBED_PIT_IDENTITY,
 562                                                 &peerinfo_cb, NULL);
 563       setup_state = PEER2_GET_IDENTITY;
 564       break;
 565     case PEER2_GET_IDENTITY:
 566       memcpy (&peer2.our_id, pinfo->result.id,
 567               sizeof (struct GNUNET_PeerIdentity));
 568       GNUNET_TESTBED_operation_done (op);
 569       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Peer 2 id: %s\n", GNUNET_i2s
 570                   (&peer2.our_id));
 571       peer1.op = GNUNET_TESTBED_service_connect (&peer1, peer1.peer, "stream",
 572                                                  NULL, NULL, stream_ca,
 573                                                  stream_da, &peer1);
 574       setup_state = PEER1_STREAM_CONNECT;
 575       break;
 576     default:
 577       GNUNET_assert (0);
 578     }
 579 }
 580
 581
 582 /**
 583  * Controller event callback
 584  *
 585  * @param cls NULL
 586  * @param event the controller event
 587  */
 588 static void
 589 controller_event_cb (void *cls,
 590                      const struct GNUNET_TESTBED_EventInformation *event)
 591 {
 592   switch (event->type)
 593   {
 594   case GNUNET_TESTBED_ET_CONNECT:
 595     GNUNET_assert (INIT == setup_state);
 596     GNUNET_TESTBED_operation_done (op);
 597     /* Get the peer identity and configuration of peers */
 598     op = GNUNET_TESTBED_peer_get_information (peer1.peer,
 599                                               GNUNET_TESTBED_PIT_IDENTITY,
 600                                               &peerinfo_cb, NULL);
 601     setup_state = PEER1_GET_IDENTITY;
 602     break;
 603   case GNUNET_TESTBED_ET_OPERATION_FINISHED:
 604     switch (setup_state)
 605     {
 606     case PEER1_STREAM_CONNECT:
 607     case PEER2_STREAM_CONNECT:
 608       GNUNET_assert (NULL == event->details.operation_finished.emsg);
 609       break;
 610     default:
 611       GNUNET_assert (0);
 612     }
 613     break;
 614   default:
 615     GNUNET_assert (0);
 616   }
 617 }
 618
 619
 620 /**
 621  * Signature of a main function for a testcase.
 622  *
 623  * @param cls closure
 624  * @param num_peers number of peers in 'peers'
 625  * @param peers handle to peers run in the testbed
 626  */
 627 static void
 628 test_master (void *cls, unsigned int num_peers,
 629              struct GNUNET_TESTBED_Peer **peers)
 630 {
 631   GNUNET_assert (NULL != peers);
 632   GNUNET_assert (NULL != peers[0]);
 633   GNUNET_assert (NULL != peers[1]);
 634   peer1.peer = peers[0];
 635   peer2.peer = peers[1];
 636   op = GNUNET_TESTBED_overlay_connect (NULL, NULL, NULL, peer2.peer, peer1.peer);
 637   setup_state = INIT;
 638   abort_task =
 639     GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply
 640                                   (GNUNET_TIME_UNIT_SECONDS, 40), &do_abort,
 641                                   NULL);
 642 }
 643
 644
 645 /**
 646  * Main function
 647  */
 648 int main (int argc, char **argv)
 649 {
 650   uint64_t event_mask;
 651
 652   result = GNUNET_NO;
 653   event_mask = 0;
 654   event_mask |= (1LL << GNUNET_TESTBED_ET_CONNECT);
 655   event_mask |= (1LL << GNUNET_TESTBED_ET_OPERATION_FINISHED);
 656   GNUNET_TESTBED_test_run ("test_stream_2peers", "test_stream_local.conf",
 657                            NUM_PEERS, event_mask, &controller_event_cb, NULL,
 658                            &test_master, NULL);
 659   if (GNUNET_SYSERR == result)
 660     return 1;
 661   return 0;
 662 }