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