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