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 #include <sys/socket.h>         /* For SHUT_RD, SHUT_WR */
  29
  30 #include "platform.h"
  31 #include "gnunet_util_lib.h"
  32 #include "gnunet_mesh_service.h"
  33 #include "gnunet_stream_lib.h"
  34
  35 #define VERBOSE 1
  36
  37 static struct GNUNET_OS_Process *arm_pid;
  38 static struct GNUNET_STREAM_Socket *peer1_socket;
  39 static struct GNUNET_STREAM_ListenSocket *peer2_listen_socket;
  40 static struct GNUNET_STREAM_Socket *peer2_socket;
  41
  42 static GNUNET_SCHEDULER_TaskIdentifier abort_task;
  43 static GNUNET_SCHEDULER_TaskIdentifier test_task;
  44 static GNUNET_SCHEDULER_TaskIdentifier read_task;
  45
  46 static GNUNET_STREAM_IOHandle *peer1_IOHandle;
  47 static GNUNET_STREAM_IOHandle *peer2_IOHandle;
  48
  49 static char *data = "ABCD";
  50 static unsigned int data_pointer;
  51 static unsigned int read_pointer;
  52 static int result;
  53
  54 static int peer1_write_pass;
  55 static int peer2_read_pass;
  56 static int peer1_half_closed_write_pass;
  57 static int peer2_half_closed_read_pass;
  58 static int peer1_write_shutdown_pass;
  59 static int peer1_read_shutdown_pass;
  60
  61
  62 /**
  63  * Shutdown nicely
  64  */
  65 static void
  66 do_shutdown (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
  67 {
  68   GNUNET_STREAM_close (peer1_socket);
  69   GNUNET_STREAM_close (peer2_socket);
  70   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "test: shutdown\n");
  71   if (0 != abort_task)
  72   {
  73     GNUNET_SCHEDULER_cancel (abort_task);
  74   }
  75   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "test: arm\n");
  76   if (0 != GNUNET_OS_process_kill (arm_pid, SIGTERM))
  77   {
  78     GNUNET_log_strerror (GNUNET_ERROR_TYPE_WARNING, "kill");
  79   }
  80   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "test: Wait\n");
  81   GNUNET_assert (GNUNET_OK == GNUNET_OS_process_wait (arm_pid));
  82   GNUNET_OS_process_close (arm_pid);
  83 }
  84
  85
  86 /**
  87  * Something went wrong and timed out. Kill everything and set error flag
  88  */
  89 static void
  90 do_abort (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
  91 {
  92   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "test: ABORT\n");
  93   if (0 != test_task)
  94   {
  95     GNUNET_SCHEDULER_cancel (test_task);
  96   }
  97   if (0 != read_task)
  98     {
  99       GNUNET_SCHEDULER_cancel (read_task);
 100     }
 101   result = GNUNET_SYSERR;
 102   abort_task = 0;
 103   do_shutdown (cls, tc);
 104 }
 105
 106
 107 /**
 108  * The write completion function; called upon writing some data to stream or
 109  * upon error
 110  *
 111  * @param cls the closure from GNUNET_STREAM_write/read
 112  * @param status the status of the stream at the time this function is called
 113  * @param size the number of bytes read or written
 114  */
 115 void write_completion (void *cls,
 116                        enum GNUNET_STREAM_Status status,
 117                        size_t size)
 118 {
 119
 120   if (peer1_write_shutdown_pass) /* Called for peer2's write operation */
 121     {
 122       /* peer1 has shutdown reading */
 123       GNUNET_assert (GNUNET_STREAM_SHUTDOWN == status);
 124       GNUNET_assert (0 == size);
 125       peer1_read_shutdown_pass = 1;
 126
 127       GNUNET_SCHEDULER_add_now (&do_shutdown, NULL);
 128     }
 129   GNUNET_assert (GNUNET_STREAM_OK == status);
 130   if (data_pointer + size != strlen(data)) /* Have more data to send */
 131     {
 132       data_pointer += size;
 133       peer1_IOHandle = GNUNET_STREAM_write (peer1_socket,
 134                                             (void *) data,
 135                                             strlen(data) - data_pointer,
 136                                             GNUNET_TIME_relative_multiply
 137                                             (GNUNET_TIME_UNIT_SECONDS, 5),
 138                                             &write_completion,
 139                                             NULL);
 140       GNUNET_assert (NULL != peer1_IOHandle);
 141     }
 142   else{
 143     peer1_write_pass = 1;
 144     /* If we are here and peer2_read_pass == 1 => we have send the data twice */
 145     if (peer2_read_pass) peer1_half_closed_write_pass = 1;
 146
 147     if (! peer1_half_closed_write_pass)
 148       {
 149         GNUNET_STREAM_shutdown (peer1_socket, SHUT_RD);
 150         /* Half closed write */
 151         data_pointer = 0;
 152         peer1_IOHandle = GNUNET_STREAM_write (peer1_socket,
 153                                               (void *) data,
 154                                               strlen(data),
 155                                               GNUNET_TIME_relative_multiply
 156                                               (GNUNET_TIME_UNIT_SECONDS, 5),
 157                                               &write_completion,
 158                                               NULL);
 159       }
 160     else
 161       {
 162         GNUNET_STREAM_shutdown (peer1_socket, SHUT_WR);
 163       }
 164   }
 165 }
 166
 167
 168 /**
 169  * Function executed after stream has been established
 170  *
 171  * @param cls the closure from GNUNET_STREAM_open
 172  * @param socket socket to use to communicate with the other side (read/write)
 173  */
 174 static void
 175 stream_open_cb (void *cls,
 176                 struct GNUNET_STREAM_Socket
 177                 *socket)
 178 {
 179   data_pointer = 0;
 180   GNUNET_assert (socket == peer1_socket);
 181   peer1_IOHandle = GNUNET_STREAM_write (socket, /* socket */
 182                                         (void *) data, /* data */
 183                                         strlen(data),
 184                                         GNUNET_TIME_relative_multiply
 185                                         (GNUNET_TIME_UNIT_SECONDS, 5),
 186                                         &write_completion,
 187                                         NULL);
 188   GNUNET_assert (NULL != peer1_IOHandle);
 189
 190 }
 191
 192
 193 /**
 194  * Input processor
 195  *
 196  * @param cls the closure from GNUNET_STREAM_write/read
 197  * @param status the status of the stream at the time this function is called
 198  * @param data traffic from the other side
 199  * @param size the number of bytes available in data read
 200  * @return number of bytes of processed from 'data' (any data remaining should be
 201  *         given to the next time the read processor is called).
 202  */
 203 static size_t
 204 input_processor (void *cls,
 205                  enum GNUNET_STREAM_Status status,
 206                  const void *input_data,
 207                  size_t size)
 208 {
 209
 210   if (peer2_half_closed_read_pass)
 211     {
 212       GNUNET_assert (GNUNET_STREAM_SHUTDOWN == status);
 213       GNUNET_assert (0 == size);
 214       peer1_write_shutdown_pass = 1;
 215       /* Now this should result in STREAM_SHUTDOWN */
 216       peer2_IOHandle = GNUNET_STREAM_write (peer2_socket,
 217                                             (void *) data,
 218                                             strlen(data),
 219                                             GNUNET_TIME_relative_multiply
 220                                             (GNUNET_TIME_UNIT_SECONDS, 5),
 221                                             &write_completion,
 222                                             (void *) peer2_socket);
 223
 224       return 0;
 225     }
 226
 227   GNUNET_assert (GNUNET_STERAM_OK == status);
 228   GNUNET_assert (size < strlen (data));
 229   GNUNET_assert (strncmp ((const char *) data,
 230                           (const char *) input_data,
 231                           size));
 232   read_pointer += size;
 233
 234   if (read_pointer < strlen (data))
 235     {
 236       peer2_IOHandle = GNUNET_STREAM_read ((struct GNUNET_STREAM_Socket *) cls,
 237                                        GNUNET_TIME_relative_multiply
 238                                        (GNUNET_TIME_UNIT_SECONDS, 5),
 239                                        &input_processor,
 240                                        NULL);
 241       GNUNET_assert (NULL != peer2_IOHandle);
 242     }
 243   else {
 244     /* If we are here and peer2_read_pass => we have finished reading twice */
 245     if (peer1_write_pass && peer2_read_pass) peer2_half_closed_read_pass = 1;
 246     if (peer1_write_pass) peer2_read_pass = 1;
 247
 248     /* Half closed read */
 249     read_pointer = 0;
 250     peer2_IOHandle = GNUNET_STREAM_read ((struct GNUNET_STREAM_Socket *) cls,
 251                                          GNUNET_TIME_relative_multiply
 252                                          (GNUNET_TIME_UNIT_SECONDS, 5),
 253                                          &input_processor,
 254                                          NULL);
 255
 256
 257   }
 258
 259   return size;
 260 }
 261
 262
 263 /**
 264  * Scheduler call back; to be executed when a new stream is connected
 265  */
 266 static void
 267 stream_read (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 268 {
 269   read_task = GNUNET_SCHEDULER_NO_TASK;
 270   GNUNET_assert (NULL != cls);
 271   read_pointer = 0;
 272   GNUNET_STREAM_listen_close (peer2_listen_socket); /* Close listen socket */
 273   peer2_IOHandle = GNUNET_STREAM_read ((struct GNUNET_STREAM_Socket *) cls,
 274                                        GNUNET_TIME_relative_multiply
 275                                        (GNUNET_TIME_UNIT_SECONDS, 5),
 276                                        &input_processor,
 277                                        NULL);
 278   GNUNET_assert (NULL != peer2_IOHandle);
 279 }
 280
 281
 282 /**
 283  * Functions of this type are called upon new stream connection from other peers
 284  *
 285  * @param cls the closure from GNUNET_STREAM_listen
 286  * @param socket the socket representing the stream
 287  * @param initiator the identity of the peer who wants to establish a stream
 288  *            with us
 289  * @return GNUNET_OK to keep the socket open, GNUNET_SYSERR to close the
 290  *             stream (the socket will be invalid after the call)
 291  */
 292 static int
 293 stream_listen_cb (void *cls,
 294            struct GNUNET_STREAM_Socket *socket,
 295            const struct GNUNET_PeerIdentity *initiator)
 296 {
 297   GNUNET_assert (NULL != socket);
 298   GNUNET_assert (NULL == initiator);   /* Local peer=NULL? */
 299   GNUNET_assert (socket != peer1_socket);
 300
 301   peer2_socket = socket;
 302   read_task = GNUNET_SCHEDULER_add_now (&stream_read, (void *) socket);
 303   return GNUNET_OK;
 304 }
 305
 306
 307 /**
 308  * Testing function
 309  */
 310 static void
 311 test (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 312 {
 313   test_task = GNUNET_SCHEDULER_NO_TASK;
 314
 315   /* Connect to stream library */
 316   peer1_socket = GNUNET_STREAM_open (NULL,         /* Null for local peer? */
 317                                      10,           /* App port */
 318                                      open_cb);
 319   GNUNET_assert (NULL != peer1_socket);
 320   peer2_listen_socket = GNUNET_STREAM_listen (10 /* App port */
 321                                               &stream_listen_cb,
 322                                               NULL);
 323   GNUNET_assert (NULL != peer2_listen_socket);
 324
 325 }
 326
 327 /**
 328  * Initialize framework and start test
 329  */
 330 static void
 331 run (void *cls, char *const *args, const char *cfgfile,
 332      const struct GNUNET_CONFIGURATION_Handle *cfg)
 333 {
 334    GNUNET_log_setup ("test_stream_local",
 335 #if VERBOSE
 336                     "DEBUG",
 337 #else
 338                     "WARNING",
 339 #endif
 340                     NULL);
 341    arm_pid =
 342      GNUNET_OS_start_process (NULL, NULL, "gnunet-service-arm",
 343                               "gnunet-service-arm",
 344 #if VERBOSE_ARM
 345                               "-L", "DEBUG",
 346 #endif
 347                               "-c", "test_stream_local.conf", NULL);
 348
 349    abort_task =
 350      GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply
 351                                    (GNUNET_TIME_UNIT_SECONDS, 20), &do_abort,
 352                                     NULL);
 353
 354    test_task = GNUNET_SCHEDULER_add_now (&test, (void *) cfg);
 355
 356 }
 357
 358 /**
 359  * Main function
 360  */
 361 int main (int argc, char **argv)
 362 {
 363   int ret;
 364
 365   char *const argv2[] = { "test-stream-local",
 366                           "-c", "test_stream.conf",
 367 #if VERBOSE
 368                           "-L", "DEBUG",
 369 #endif
 370                           NULL
 371   };
 372
 373   struct GNUNET_GETOPT_CommandLineOption options[] = {
 374     GNUNET_GETOPT_OPTION_END
 375   };
 376
 377   ret =
 378       GNUNET_PROGRAM_run ((sizeof (argv2) / sizeof (char *)) - 1, argv2,
 379                           "test-stream-local", "nohelp", options, &run, NULL);
 380
 381   if (GNUNET_OK != ret)
 382   {
 383     GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "run failed with error code %d\n",
 384                 ret);
 385     return 1;
 386   }
 387   if (GNUNET_SYSERR == result)
 388   {
 389     GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "test failed\n");
 390     return 1;
 391   }
 392   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "test ok\n");
 393   return 0;
 394 }