src/stream/test_stream_local.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_local.c
  23  * @brief Stream API testing between local peers
  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_testing_lib.h"
  34
  35 #define VERBOSE 1
  36
  37 /**
  38  * Structure for holding peer's sockets and IO Handles
  39  */
  40 struct PeerData
  41 {
  42   /**
  43    * Peer's stream socket
  44    */
  45   struct GNUNET_STREAM_Socket *socket;
  46
  47   /**
  48    * Peer's io write handle
  49    */
  50   struct GNUNET_STREAM_IOWriteHandle *io_write_handle;
  51
  52   /**
  53    * Peer's io read handle
  54    */
  55   struct GNUNET_STREAM_IOReadHandle *io_read_handle;
  56
  57   /**
  58    * Bytes the peer has written
  59    */
  60   unsigned int bytes_wrote;
  61
  62   /**
  63    * Byte the peer has read
  64    */
  65   unsigned int bytes_read;
  66 };
  67
  68 static struct GNUNET_OS_Process *arm_pid;
  69 static struct PeerData peer1;
  70 static struct PeerData peer2;
  71 static struct GNUNET_STREAM_ListenSocket *peer2_listen_socket;
  72 static struct GNUNET_CONFIGURATION_Handle *config_peer1;
  73 static struct GNUNET_CONFIGURATION_Handle *config_peer2;
  74
  75 static GNUNET_SCHEDULER_TaskIdentifier abort_task;
  76 static GNUNET_SCHEDULER_TaskIdentifier test_task;
  77
  78 static char *data = "ABCD";
  79 static int result;
  80
  81 static int writing_success;
  82 static int reading_success;
  83
  84 /**
  85  * Input processor
  86  *
  87  * @param cls the closure from GNUNET_STREAM_write/read
  88  * @param status the status of the stream at the time this function is called
  89  * @param data traffic from the other side
  90  * @param size the number of bytes available in data read
  91  * @return number of bytes of processed from 'data' (any data remaining should be
  92  *         given to the next time the read processor is called).
  93  */
  94 static size_t
  95 input_processor (void *cls,
  96                  enum GNUNET_STREAM_Status status,
  97                  const void *input_data,
  98                  size_t size);
  99
 100 /**
 101  * Task for calling STREAM_read
 102  *
 103  * @param cls the peer data entity
 104  * @param tc the task context
 105  */
 106 static void
 107 stream_read_task (void *cls,
 108                   const struct GNUNET_SCHEDULER_TaskContext *tc)
 109 {
 110   struct PeerData *peer = cls;
 111
 112   peer->io_read_handle = GNUNET_STREAM_read (peer->socket,
 113                                              GNUNET_TIME_relative_multiply
 114                                              (GNUNET_TIME_UNIT_SECONDS, 5),
 115                                              &input_processor,
 116                                              peer);
 117   GNUNET_assert (NULL != peer->io_read_handle);
 118 }
 119
 120 /**
 121  * The write completion function; called upon writing some data to stream or
 122  * upon error
 123  *
 124  * @param cls the closure from GNUNET_STREAM_write/read
 125  * @param status the status of the stream at the time this function is called
 126  * @param size the number of bytes read or written
 127  */
 128 static void
 129 write_completion (void *cls,
 130                   enum GNUNET_STREAM_Status status,
 131                   size_t size);
 132
 133
 134 /**
 135  * Task for calling STREAM_write
 136  *
 137  * @param cls the peer data entity
 138  * @param tc the task context
 139  */
 140 static void
 141 stream_write_task (void *cls,
 142                    const struct GNUNET_SCHEDULER_TaskContext *tc)
 143 {
 144   struct PeerData *peer = cls;
 145
 146   peer->io_write_handle =
 147     GNUNET_STREAM_write (peer->socket,
 148                          (void *) data,
 149                          strlen(data) - peer->bytes_wrote,
 150                          GNUNET_TIME_relative_multiply
 151                          (GNUNET_TIME_UNIT_SECONDS, 5),
 152                          &write_completion,
 153                          peer);
 154
 155   GNUNET_assert (NULL != peer->io_write_handle);
 156  }
 157
 158 /**
 159  * Shutdown nicely
 160  */
 161 static void
 162 do_shutdown (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 163 {
 164   GNUNET_STREAM_close (peer1.socket);
 165   if (NULL != peer2.socket)
 166     GNUNET_STREAM_close (peer2.socket);
 167   if (NULL != peer2_listen_socket)
 168     GNUNET_STREAM_listen_close (peer2_listen_socket);
 169   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "test: shutdown\n");
 170   if (0 != abort_task)
 171   {
 172     GNUNET_SCHEDULER_cancel (abort_task);
 173   }
 174   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "test: arm\n");
 175   if (0 != GNUNET_OS_process_kill (arm_pid, SIGTERM))
 176   {
 177     GNUNET_log_strerror (GNUNET_ERROR_TYPE_WARNING, "kill");
 178   }
 179   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "test: Wait\n");
 180   /* Free the duplicated configuration */
 181   GNUNET_CONFIGURATION_destroy (config_peer1);
 182   GNUNET_CONFIGURATION_destroy (config_peer2);
 183   GNUNET_assert (GNUNET_OK == GNUNET_OS_process_wait (arm_pid));
 184   GNUNET_OS_process_close (arm_pid);
 185 }
 186
 187
 188 /**
 189  * Something went wrong and timed out. Kill everything and set error flag
 190  */
 191 static void
 192 do_abort (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 193 {
 194   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "test: ABORT\n");
 195   if (0 != test_task)
 196   {
 197     GNUNET_SCHEDULER_cancel (test_task);
 198   }
 199
 200   result = GNUNET_SYSERR;
 201   abort_task = 0;
 202   do_shutdown (cls, tc);
 203 }
 204
 205
 206 /**
 207  * The write completion function; called upon writing some data to stream or
 208  * upon error
 209  *
 210  * @param cls the closure from GNUNET_STREAM_write/read
 211  * @param status the status of the stream at the time this function is called
 212  * @param size the number of bytes read or written
 213  */
 214 static void
 215 write_completion (void *cls,
 216                   enum GNUNET_STREAM_Status status,
 217                   size_t size)
 218 {
 219   struct PeerData *peer=cls;
 220
 221   GNUNET_assert (GNUNET_STREAM_OK == status);
 222   GNUNET_assert (size <= strlen (data));
 223   peer->bytes_wrote += size;
 224
 225   if (peer->bytes_wrote < strlen(data)) /* Have more data to send */
 226     {
 227       GNUNET_SCHEDULER_add_now (&stream_write_task, peer);
 228     }
 229   else
 230     {
 231       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 232                   "Writing completed\n");
 233
 234       if (&peer1 == peer)   /* Peer1 has finished writing; should read now */
 235         {
 236           peer->bytes_read = 0;
 237           GNUNET_SCHEDULER_add_now (&stream_read_task, peer);
 238         }
 239       else
 240         {
 241           writing_success = GNUNET_YES;
 242           if (GNUNET_YES == reading_success)
 243             GNUNET_SCHEDULER_add_now (&do_shutdown, NULL);
 244         }
 245     }
 246 }
 247
 248
 249 /**
 250  * Function executed after stream has been established
 251  *
 252  * @param cls the closure from GNUNET_STREAM_open
 253  * @param socket socket to use to communicate with the other side (read/write)
 254  */
 255 static void
 256 stream_open_cb (void *cls,
 257                 struct GNUNET_STREAM_Socket *socket)
 258 {
 259   struct PeerData *peer=cls;
 260
 261   GNUNET_assert (&peer1 == peer);
 262   GNUNET_assert (socket == peer1.socket);
 263   GNUNET_assert (socket == peer->socket);
 264
 265   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Stream established from peer1\n");
 266   peer->bytes_wrote = 0;
 267   GNUNET_SCHEDULER_add_now (&stream_write_task, peer);
 268 }
 269
 270
 271 /**
 272  * Input processor
 273  *
 274  * @param cls the closure from GNUNET_STREAM_write/read
 275  * @param status the status of the stream at the time this function is called
 276  * @param data traffic from the other side
 277  * @param size the number of bytes available in data read
 278  * @return number of bytes of processed from 'data' (any data remaining should be
 279  *         given to the next time the read processor is called).
 280  */
 281 static size_t
 282 input_processor (void *cls,
 283                  enum GNUNET_STREAM_Status status,
 284                  const void *input_data,
 285                  size_t size)
 286 {
 287   struct PeerData *peer = cls;
 288
 289   GNUNET_assert (GNUNET_STREAM_OK == status);
 290   GNUNET_assert (size <= strlen (data));
 291   GNUNET_assert (strncmp ((const char *) data + peer->bytes_read,
 292                           (const char *) input_data,
 293                           size));
 294   peer->bytes_read += size;
 295
 296   if (peer->bytes_read < strlen (data))
 297     {
 298       GNUNET_SCHEDULER_add_now (&stream_read_task, peer);
 299     }
 300   else
 301     {
 302       if (&peer2 == peer)    /* Peer2 has completed reading; should write */
 303         {
 304           peer->bytes_wrote = 0;
 305           GNUNET_SCHEDULER_add_now (&stream_write_task, peer);
 306         }
 307       else                      /* Peer1 has completed reading. End of tests */
 308         {
 309           reading_success = GNUNET_YES;
 310           if (GNUNET_YES == writing_success)
 311             GNUNET_SCHEDULER_add_now (&do_shutdown, NULL);
 312         }
 313     }
 314   return size;
 315 }
 316
 317
 318 /**
 319  * Functions of this type are called upon new stream connection from other peers
 320  *
 321  * @param cls the PeerData of peer2
 322  * @param socket the socket representing the stream
 323  * @param initiator the identity of the peer who wants to establish a stream
 324  *            with us
 325  * @return GNUNET_OK to keep the socket open, GNUNET_SYSERR to close the
 326  *             stream (the socket will be invalid after the call)
 327  */
 328 static int
 329 stream_listen_cb (void *cls,
 330            struct GNUNET_STREAM_Socket *socket,
 331            const struct GNUNET_PeerIdentity *initiator)
 332 {
 333   struct PeerData *peer=cls;
 334   struct GNUNET_PeerIdentity self;
 335
 336   GNUNET_assert (NULL != socket);
 337   GNUNET_assert (socket != peer1.socket);
 338   GNUNET_assert (&peer2 == peer);
 339
 340   /* Get our identity */
 341   GNUNET_assert (GNUNET_OK == GNUNET_TESTING_get_peer_identity (config_peer1,
 342                                                                 &self));
 343   GNUNET_assert (0 == memcmp (&self,
 344                               initiator,
 345                               sizeof (struct GNUNET_PeerIdentity)));
 346
 347   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 348               "Peer connected: %s\n", GNUNET_i2s(initiator));
 349
 350   peer->socket = socket;
 351   peer->bytes_read = 0;
 352   GNUNET_SCHEDULER_add_now (&stream_read_task, &peer2);
 353   return GNUNET_OK;
 354 }
 355
 356
 357 /**
 358  * Testing function
 359  *
 360  * @param cls NULL
 361  * @param tc the task context
 362  */
 363 static void
 364 test (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 365 {
 366   struct GNUNET_PeerIdentity self;
 367
 368   test_task = GNUNET_SCHEDULER_NO_TASK;
 369   /* Get our identity */
 370   GNUNET_assert (GNUNET_OK == GNUNET_TESTING_get_peer_identity (config_peer1,
 371                                                                 &self));
 372
 373   peer2_listen_socket = GNUNET_STREAM_listen (config_peer2,
 374                                               10, /* App port */
 375                                               &stream_listen_cb,
 376                                               &peer2);
 377   GNUNET_assert (NULL != peer2_listen_socket);
 378
 379   /* Connect to stream library */
 380   peer1.socket = GNUNET_STREAM_open (config_peer1,
 381                                      &self,         /* Null for local peer? */
 382                                      10,           /* App port */
 383                                      &stream_open_cb,
 384                                      &peer1);
 385   GNUNET_assert (NULL != peer1.socket);
 386 }
 387
 388 /**
 389  * Initialize framework and start test
 390  */
 391 static void
 392 run (void *cls, char *const *args, const char *cfgfile,
 393      const struct GNUNET_CONFIGURATION_Handle *cfg)
 394 {
 395    GNUNET_log_setup ("test_stream_local",
 396 #if VERBOSE
 397                     "DEBUG",
 398 #else
 399                     "WARNING",
 400 #endif
 401                     NULL);
 402    /* Duplicate the configuration */
 403    config_peer1 = GNUNET_CONFIGURATION_dup (cfg);
 404    config_peer2 = GNUNET_CONFIGURATION_dup (cfg);
 405    arm_pid =
 406      GNUNET_OS_start_process (GNUNET_YES, NULL, NULL, "gnunet-service-arm",
 407                               "gnunet-service-arm",
 408 #if VERBOSE_ARM
 409                               "-L", "DEBUG",
 410 #endif
 411                               "-c", "test_stream_local.conf", NULL);
 412
 413    abort_task =
 414      GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply
 415                                    (GNUNET_TIME_UNIT_SECONDS, 60), &do_abort,
 416                                     NULL);
 417
 418    test_task = GNUNET_SCHEDULER_add_now (&test, NULL);
 419 }
 420
 421 /**
 422  * Main function
 423  */
 424 int main (int argc, char **argv)
 425 {
 426   int ret;
 427
 428   char *const argv2[] = { "test-stream-local",
 429                           "-c", "test_stream_local.conf",
 430 #if VERBOSE
 431                           "-L", "DEBUG",
 432 #endif
 433                           NULL
 434   };
 435
 436   struct GNUNET_GETOPT_CommandLineOption options[] = {
 437     GNUNET_GETOPT_OPTION_END
 438   };
 439
 440   ret =
 441       GNUNET_PROGRAM_run ((sizeof (argv2) / sizeof (char *)) - 1, argv2,
 442                           "test-stream-local", "nohelp", options, &run, NULL);
 443
 444   if (GNUNET_OK != ret)
 445   {
 446     GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "run failed with error code %d\n",
 447                 ret);
 448     return 1;
 449   }
 450   if (GNUNET_SYSERR == result)
 451   {
 452     GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "test failed\n");
 453     return 1;
 454   }
 455   GNUNET_log (GNUNET_ERROR_TYPE_INFO, "test ok\n");
 456   return 0;
 457 }