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