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   GNUNET_TESTBED_operation_done (peer1.op);
 245 }
 246
 247
 248 /**
 249  * Completion callback for shutdown
 250  *
 251  * @param cls the closure from GNUNET_STREAM_shutdown call
 252  * @param operation the operation that was shutdown (SHUT_RD, SHUT_WR,
 253  *          SHUT_RDWR)
 254  */
 255 static void
 256 shutdown_completion (void *cls,
 257                      int operation)
 258 {
 259   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "STREAM shutdown successful\n");
 260   GNUNET_SCHEDULER_add_now (&do_close, cls);
 261 }
 262
 263
 264
 265 /**
 266  * Shutdown sockets gracefully
 267  */
 268 static void
 269 do_shutdown (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 270 {
 271   result = GNUNET_OK;
 272   peer1.shutdown_handle = GNUNET_STREAM_shutdown (peer1.socket, SHUT_RDWR,
 273                                                   &shutdown_completion, cls);
 274 }
 275
 276
 277 /**
 278  * Something went wrong and timed out. Kill everything and set error flag
 279  */
 280 static void
 281 do_abort (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 282 {
 283   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "test: ABORT\n");
 284   result = GNUNET_SYSERR;
 285   abort_task = 0;
 286   do_close (cls, tc);
 287 }
 288
 289
 290 /**
 291  * The write completion function; called upon writing some data to stream or
 292  * upon error
 293  *
 294  * @param cls the closure from GNUNET_STREAM_write/read
 295  * @param status the status of the stream at the time this function is called
 296  * @param size the number of bytes read or written
 297  */
 298 static void
 299 write_completion (void *cls,
 300                   enum GNUNET_STREAM_Status status,
 301                   size_t size)
 302 {
 303   struct PeerData *peer=cls;
 304
 305   GNUNET_assert (GNUNET_STREAM_OK == status);
 306   GNUNET_assert (size <= strlen (data));
 307   peer->bytes_wrote += size;
 308
 309   if (peer->bytes_wrote < strlen(data)) /* Have more data to send */
 310     {
 311       GNUNET_SCHEDULER_add_now (&stream_write_task, peer);
 312     }
 313   else
 314     {
 315       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 316                   "Writing completed\n");
 317
 318       if (&peer2 == peer)   /* Peer1 has finished writing; should read now */
 319         {
 320           peer->bytes_read = 0;
 321           GNUNET_SCHEDULER_add_now (&stream_read_task, peer);
 322         }
 323       else
 324         {
 325           writing_success = GNUNET_YES;
 326           if (GNUNET_YES == reading_success)
 327             GNUNET_SCHEDULER_add_now (&do_shutdown, NULL);
 328         }
 329     }
 330 }
 331
 332
 333 /**
 334  * Function executed after stream has been established
 335  *
 336  * @param cls the closure from GNUNET_STREAM_open
 337  * @param socket socket to use to communicate with the other side (read/write)
 338  */
 339 static void
 340 stream_open_cb (void *cls,
 341                 struct GNUNET_STREAM_Socket *socket)
 342 {
 343   struct PeerData *peer=cls;
 344
 345   GNUNET_assert (&peer2 == peer);
 346   GNUNET_assert (socket == peer2.socket);
 347   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "%s: Stream established from peer2\n",
 348               GNUNET_i2s (&peer1.our_id));
 349   peer->bytes_wrote = 0;
 350   GNUNET_SCHEDULER_add_now (&stream_write_task, peer);
 351 }
 352
 353
 354 /**
 355  * Input processor
 356  *
 357  * @param cls the closure from GNUNET_STREAM_write/read
 358  * @param status the status of the stream at the time this function is called
 359  * @param data traffic from the other side
 360  * @param size the number of bytes available in data read
 361  * @return number of bytes of processed from 'data' (any data remaining should be
 362  *         given to the next time the read processor is called).
 363  */
 364 static size_t
 365 input_processor (void *cls,
 366                  enum GNUNET_STREAM_Status status,
 367                  const void *input_data,
 368                  size_t size)
 369 {
 370   struct PeerData *peer;
 371
 372   peer = (struct PeerData *) cls;
 373
 374   if (GNUNET_STREAM_TIMEOUT == status)
 375     {
 376       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 377                   "Read operation timedout - reading again!\n");
 378       GNUNET_assert (0 == size);
 379       GNUNET_SCHEDULER_add_now (&stream_read_task, peer);
 380       return 0;
 381     }
 382
 383   GNUNET_assert (GNUNET_STREAM_OK == status);
 384   GNUNET_assert (size <= strlen (data));
 385   GNUNET_assert (0 == strncmp ((const char *) data + peer->bytes_read,
 386                                (const char *) input_data,
 387                                size));
 388   peer->bytes_read += size;
 389
 390   if (peer->bytes_read < strlen (data))
 391     {
 392       GNUNET_SCHEDULER_add_now (&stream_read_task, peer);
 393     }
 394   else
 395     {
 396       if (&peer1 == peer)    /* Peer2 has completed reading; should write */
 397         {
 398           peer->bytes_wrote = 0;
 399           GNUNET_SCHEDULER_add_now (&stream_write_task, peer);
 400         }
 401       else                      /* Peer1 has completed reading. End of tests */
 402         {
 403           reading_success = GNUNET_YES;
 404           if (GNUNET_YES == writing_success)
 405             GNUNET_SCHEDULER_add_now (&do_shutdown, NULL);
 406         }
 407     }
 408   return size;
 409 }
 410
 411
 412 /**
 413  * Functions of this type are called upon new stream connection from other peers
 414  *
 415  * @param cls the closure from GNUNET_STREAM_listen
 416  * @param socket the socket representing the stream
 417  * @param initiator the identity of the peer who wants to establish a stream
 418  *            with us
 419  * @return GNUNET_OK to keep the socket open, GNUNET_SYSERR to close the
 420  *             stream (the socket will be invalid after the call)
 421  */
 422 static int
 423 stream_listen_cb (void *cls,
 424                   struct GNUNET_STREAM_Socket *socket,
 425                   const struct GNUNET_PeerIdentity *initiator)
 426 {
 427   GNUNET_assert (NULL != socket);
 428   GNUNET_assert (NULL != initiator);
 429   GNUNET_assert (socket != peer2.socket);
 430   GNUNET_assert (0 == memcmp (initiator, &peer2.our_id,
 431                               sizeof (struct GNUNET_PeerIdentity)));
 432   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "%s: Peer connected: %s\n",
 433               GNUNET_i2s (&peer1.our_id), GNUNET_i2s (initiator));
 434   peer1.socket = socket;
 435   peer1.bytes_read = 0;
 436   GNUNET_SCHEDULER_add_now (&stream_read_task, &peer1);
 437   return GNUNET_OK;
 438 }
 439
 440
 441 /**
 442  * Listen success callback; connects a peer to stream as client
 443  */
 444 static void stream_connect (void);
 445
 446
 447 /**
 448  * Adapter function called to destroy a connection to
 449  * a service.
 450  *
 451  * @param cls closure
 452  * @param op_result service handle returned from the connect adapter
 453  */
 454 static void
 455 stream_da (void *cls, void *op_result)
 456 {
 457   struct GNUNET_STREAM_ListenSocket *lsocket;
 458   struct GNUNET_STREAM_Socket *socket;
 459
 460   if (&peer1 == cls)
 461   {
 462     lsocket = op_result;
 463     GNUNET_STREAM_listen_close (lsocket);
 464     GNUNET_TESTBED_operation_done (peer2.op);
 465     return;
 466   }
 467   if (&peer2 == cls)
 468   {
 469     socket = op_result;
 470     GNUNET_STREAM_close (socket);
 471     GNUNET_SCHEDULER_shutdown (); /* Exit point of the test */
 472     return;
 473   }
 474   GNUNET_assert (0);
 475 }
 476
 477
 478 /**
 479  * Adapter function called to establish a connection to
 480  * a service.
 481  *
 482  * @param cls closure
 483  * @param cfg configuration of the peer to connect to; will be available until
 484  *          GNUNET_TESTBED_operation_done() is called on the operation returned
 485  *          from GNUNET_TESTBED_service_connect()
 486  * @return service handle to return in 'op_result', NULL on error
 487  */
 488 static void *
 489 stream_ca (void *cls, const struct GNUNET_CONFIGURATION_Handle *cfg)
 490 {
 491   struct GNUNET_STREAM_ListenSocket *lsocket;
 492
 493   switch (setup_state)
 494   {
 495   case PEER1_STREAM_CONNECT:
 496     lsocket = GNUNET_STREAM_listen (cfg, 10, &stream_listen_cb, NULL,
 497                                     GNUNET_STREAM_OPTION_SIGNAL_LISTEN_SUCCESS,
 498                                     &stream_connect, GNUNET_STREAM_OPTION_END);
 499     return lsocket;
 500   case PEER2_STREAM_CONNECT:
 501     peer2.socket = GNUNET_STREAM_open (cfg, &peer1.our_id, 10, &stream_open_cb,
 502                                        &peer2, GNUNET_STREAM_OPTION_END);
 503     return peer2.socket;
 504   default:
 505     GNUNET_assert (0);
 506   }
 507 }
 508
 509
 510 /**
 511  * Listen success callback; connects a peer to stream as client
 512  */
 513 static void
 514 stream_connect (void)
 515 {
 516   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Stream listen open successful\n");
 517   peer2.op = GNUNET_TESTBED_service_connect (&peer2, peer2.peer, "stream",
 518                                              NULL, NULL,
 519                                              stream_ca, stream_da, &peer2);
 520   setup_state = PEER2_STREAM_CONNECT;
 521 }
 522
 523
 524 /**
 525  * Callback to be called when the requested peer information is available
 526  *
 527  * @param cb_cls the closure from GNUNET_TETSBED_peer_get_information()
 528  * @param op the operation this callback corresponds to
 529  * @param pinfo the result; will be NULL if the operation has failed
 530  * @param emsg error message if the operation has failed; will be NULL if the
 531  *          operation is successfull
 532  */
 533 static void
 534 peerinfo_cb (void *cb_cls, struct GNUNET_TESTBED_Operation *op_,
 535              const struct GNUNET_TESTBED_PeerInformation *pinfo,
 536              const char *emsg)
 537 {
 538   GNUNET_assert (NULL == emsg);
 539   GNUNET_assert (op == op_);
 540   switch (setup_state)
 541     {
 542     case PEER1_GET_IDENTITY:
 543       memcpy (&peer1.our_id, pinfo->result.id,
 544               sizeof (struct GNUNET_PeerIdentity));
 545       GNUNET_TESTBED_operation_done (op);
 546       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Peer 1 id: %s\n", GNUNET_i2s
 547                   (&peer1.our_id));
 548       op = GNUNET_TESTBED_peer_get_information (peer2.peer,
 549                                                 GNUNET_TESTBED_PIT_IDENTITY,
 550                                                 &peerinfo_cb, NULL);
 551       setup_state = PEER2_GET_IDENTITY;
 552       break;
 553     case PEER2_GET_IDENTITY:
 554       memcpy (&peer2.our_id, pinfo->result.id,
 555               sizeof (struct GNUNET_PeerIdentity));
 556       GNUNET_TESTBED_operation_done (op);
 557       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Peer 2 id: %s\n", GNUNET_i2s
 558                   (&peer2.our_id));
 559       peer1.op = GNUNET_TESTBED_service_connect (&peer1, peer1.peer, "stream",
 560                                                  NULL, NULL, stream_ca,
 561                                                  stream_da, &peer1);
 562       setup_state = PEER1_STREAM_CONNECT;
 563       break;
 564     default:
 565       GNUNET_assert (0);
 566     }
 567 }
 568
 569
 570 /**
 571  * Controller event callback
 572  *
 573  * @param cls NULL
 574  * @param event the controller event
 575  */
 576 static void
 577 controller_event_cb (void *cls,
 578                      const struct GNUNET_TESTBED_EventInformation *event)
 579 {
 580   switch (event->type)
 581   {
 582   case GNUNET_TESTBED_ET_CONNECT:
 583     GNUNET_assert (INIT == setup_state);
 584     GNUNET_TESTBED_operation_done (op);
 585     /* Get the peer identity and configuration of peers */
 586     op = GNUNET_TESTBED_peer_get_information (peer1.peer,
 587                                               GNUNET_TESTBED_PIT_IDENTITY,
 588                                               &peerinfo_cb, NULL);
 589     setup_state = PEER1_GET_IDENTITY;
 590     break;
 591   case GNUNET_TESTBED_ET_OPERATION_FINISHED:
 592     switch (setup_state)
 593     {
 594     case PEER1_STREAM_CONNECT:
 595     case PEER2_STREAM_CONNECT:
 596       GNUNET_assert (NULL == event->details.operation_finished.emsg);
 597       break;
 598     default:
 599       GNUNET_assert (0);
 600     }
 601     break;
 602   default:
 603     GNUNET_assert (0);
 604   }
 605 }
 606
 607
 608 /**
 609  * Signature of a main function for a testcase.
 610  *
 611  * @param cls closure
 612  * @param num_peers number of peers in 'peers'
 613  * @param peers handle to peers run in the testbed
 614  */
 615 static void
 616 test_master (void *cls, unsigned int num_peers,
 617              struct GNUNET_TESTBED_Peer **peers)
 618 {
 619   GNUNET_assert (NULL != peers);
 620   GNUNET_assert (NULL != peers[0]);
 621   GNUNET_assert (NULL != peers[1]);
 622   peer1.peer = peers[0];
 623   peer2.peer = peers[1];
 624   op = GNUNET_TESTBED_overlay_connect (NULL, NULL, NULL, peer2.peer, peer1.peer);
 625   setup_state = INIT;
 626   abort_task =
 627     GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply
 628                                   (GNUNET_TIME_UNIT_SECONDS, 40), &do_abort,
 629                                   NULL);
 630 }
 631
 632
 633 /**
 634  * Main function
 635  */
 636 int main (int argc, char **argv)
 637 {
 638   uint64_t event_mask;
 639
 640   result = GNUNET_NO;
 641   event_mask = 0;
 642   event_mask |= (1LL << GNUNET_TESTBED_ET_CONNECT);
 643   event_mask |= (1LL << GNUNET_TESTBED_ET_OPERATION_FINISHED);
 644   GNUNET_TESTBED_test_run ("test_stream_2peers", "test_stream_local.conf",
 645                            NUM_PEERS, event_mask, &controller_event_cb, NULL,
 646                            &test_master, NULL);
 647   if (GNUNET_SYSERR == result)
 648     return 1;
 649   return 0;
 650 }