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