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