first steps towards MQ
[oweals/gnunet.git] / src / testbed / testbed_api_barriers.c
1 /*
2   This file is part of GNUnet.
3   Copyright (C) 2008--2013 GNUnet e.V.
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., 51 Franklin Street, Fifth Floor,
18   Boston, MA 02110-1301, USA.
19 */
20
21 /**
22  * @file testbed/testbed_api_barriers.c
23  * @brief API implementation for testbed barriers
24  * @author Sree Harsha Totakura <sreeharsha@totakura.in>
25  */
26
27 #include "platform.h"
28 #include "gnunet_testbed_service.h"
29 #include "testbed_api.h"
30 #include "testbed_api_barriers.h"
31
32 /**
33  * Logging shorthand
34  */
35 #define LOG(type, ...)                          \
36   GNUNET_log_from (type, "testbed-api-barriers", __VA_ARGS__);
37
38 /**
39  * Debug logging shorthand
40  */
41 #define LOG_DEBUG(...)                          \
42   LOG (GNUNET_ERROR_TYPE_DEBUG, __VA_ARGS__);
43
44 /**
45  * Handle for barrier
46  */
47 struct GNUNET_TESTBED_Barrier
48 {
49   /**
50    * hashcode identifying this barrier in the hashmap
51    */
52   struct GNUNET_HashCode key;
53
54   /**
55    * The controller handle given while initiliasing this barrier
56    */
57   struct GNUNET_TESTBED_Controller *c;
58
59   /**
60    * The name of the barrier
61    */
62   char *name;
63
64   /**
65    * The continuation callback to call when we have a status update on this
66    */
67   GNUNET_TESTBED_barrier_status_cb cb;
68
69   /**
70    * the closure for the above callback
71    */
72   void *cls;
73
74   /**
75    * Should the barrier crossed status message be echoed back to the controller?
76    */
77   int echo;
78 };
79
80
81 /**
82  * handle for hashtable of barrier handles
83  */
84 static struct GNUNET_CONTAINER_MultiHashMap *barrier_map;
85
86
87 /**
88  * Remove a barrier and it was the last one in the barrier hash map, destroy the
89  * hash map
90  *
91  * @param barrier the barrier to remove
92  */
93 static void
94 barrier_remove (struct GNUNET_TESTBED_Barrier *barrier)
95 {
96   GNUNET_assert (NULL != barrier_map); /* No barriers present */
97   GNUNET_assert (GNUNET_OK ==
98                  GNUNET_CONTAINER_multihashmap_remove (barrier_map,
99                                                        &barrier->key,
100                                                        barrier));
101   GNUNET_free (barrier->name);
102   GNUNET_free (barrier);
103   if (0 == GNUNET_CONTAINER_multihashmap_size (barrier_map))
104   {
105     GNUNET_CONTAINER_multihashmap_destroy (barrier_map);
106     barrier_map = NULL;
107   }
108 }
109
110
111 /**
112  * Validate #GNUNET_MESSAGE_TYPE_TESTBED_BARRIER_STATUS message.
113  *
114  * @param cls the controller handle to determine the connection this message
115  *   belongs to
116  * @param msg the barrier status message
117  * @return #GNUNET_OK if the message is valid; #GNUNET_SYSERR to tear it
118  *   down signalling an error (message malformed)
119  */
120 int
121 check_barrier_status_ (struct GNUNET_TESTBED_Controller *c,
122                        const struct GNUNET_TESTBED_BarrierStatusMsg *msg)
123 {
124   uint16_t msize;
125   uint16_t name_len;
126   int status;
127   const char *name;
128   size_t emsg_len;
129
130   msize = ntohs (msg->header.size);
131   name = msg->data;
132   name_len = ntohs (msg->name_len);
133
134   if (sizeof (struct GNUNET_TESTBED_BarrierStatusMsg) + name_len + 1 > msize)
135   {
136     GNUNET_break_op (0);
137     return GNUNET_SYSERR;
138   }
139   if ('\0' != name[name_len])
140   {
141     GNUNET_break_op (0);
142     return GNUNET_SYSERR;
143   }
144   status = ntohs (msg->status);
145   if (GNUNET_TESTBED_BARRIERSTATUS_ERROR == status)
146   {
147     emsg_len = msize - (sizeof (struct GNUNET_TESTBED_BarrierStatusMsg) + name_len
148                         + 1); /* +1!? */
149     if (0 == emsg_len)
150     {
151       GNUNET_break_op (0);
152       return GNUNET_SYSERR;
153     }
154   }
155   return GNUNET_OK;
156 }
157
158
159 /**
160  * Handler for #GNUNET_MESSAGE_TYPE_TESTBED_BARRIER_STATUS messages
161  *
162  * @param c the controller handle to determine the connection this message
163  *   belongs to
164  * @param msg the barrier status message
165  */
166 void
167 handle_barrier_status_ (struct GNUNET_TESTBED_Controller *c,
168                         const struct GNUNET_TESTBED_BarrierStatusMsg *msg)
169 {
170   struct GNUNET_TESTBED_Barrier *barrier;
171   char *emsg;
172   const char *name;
173   struct GNUNET_HashCode key;
174   size_t emsg_len;
175   int status;
176   uint16_t msize;
177   uint16_t name_len;
178
179   emsg = NULL;
180   barrier = NULL;
181   msize = ntohs (msg->header.size);
182   name = msg->data;
183   name_len = ntohs (msg->name_len);
184   LOG_DEBUG ("Received BARRIER_STATUS msg\n");
185   status = ntohs (msg->status);
186   if (GNUNET_TESTBED_BARRIERSTATUS_ERROR == status)
187   {
188     status = -1;
189     emsg_len = msize - (sizeof (struct GNUNET_TESTBED_BarrierStatusMsg) + name_len
190                         + 1);
191     emsg = GNUNET_malloc (emsg_len + 1);
192     memcpy (emsg,
193             msg->data + name_len + 1,
194             emsg_len);
195   }
196   if (NULL == barrier_map)
197   {
198     GNUNET_break_op (0);
199     goto cleanup;
200   }
201   GNUNET_CRYPTO_hash (name, name_len, &key);
202   barrier = GNUNET_CONTAINER_multihashmap_get (barrier_map, &key);
203   if (NULL == barrier)
204   {
205     GNUNET_break_op (0);
206     goto cleanup;
207   }
208   GNUNET_assert (NULL != barrier->cb);
209   if ((GNUNET_YES == barrier->echo) && (GNUNET_TESTBED_BARRIERSTATUS_CROSSED == status))
210     GNUNET_TESTBED_queue_message_ (c, GNUNET_copy_message (&msg->header));
211   barrier->cb (barrier->cls, name, barrier, status, emsg);
212   if (GNUNET_TESTBED_BARRIERSTATUS_INITIALISED == status)
213     return;           /* just initialised; skip cleanup */
214
215  cleanup:
216   GNUNET_free_non_null (emsg);
217   if (NULL != barrier)
218     barrier_remove (barrier);
219 }
220
221
222 /**
223  * Initialise a barrier and call the given callback when the required percentage
224  * of peers (quorum) reach the barrier OR upon error.
225  *
226  * @param controller the handle to the controller
227  * @param name identification name of the barrier
228  * @param quorum the percentage of peers that is required to reach the barrier.
229  *   Peers signal reaching a barrier by calling
230  *   GNUNET_TESTBED_barrier_reached().
231  * @param cb the callback to call when the barrier is reached or upon error.
232  *   Cannot be NULL.
233  * @param cls closure for the above callback
234  * @param echo GNUNET_YES to echo the barrier crossed status message back to the
235  *   controller
236  * @return barrier handle; NULL upon error
237  */
238 struct GNUNET_TESTBED_Barrier *
239 GNUNET_TESTBED_barrier_init_ (struct GNUNET_TESTBED_Controller *controller,
240                               const char *name,
241                               unsigned int quorum,
242                               GNUNET_TESTBED_barrier_status_cb cb, void *cls,
243                               int echo)
244 {
245   struct GNUNET_TESTBED_BarrierInit *msg;
246   struct GNUNET_TESTBED_Barrier *barrier;
247   struct GNUNET_HashCode key;
248   size_t name_len;
249   uint16_t msize;
250
251   GNUNET_assert (quorum <= 100);
252   GNUNET_assert (NULL != cb);
253   name_len = strlen (name);
254   GNUNET_assert (0 < name_len);
255   GNUNET_CRYPTO_hash (name, name_len, &key);
256   if (NULL == barrier_map)
257     barrier_map = GNUNET_CONTAINER_multihashmap_create (3, GNUNET_YES);
258   if (GNUNET_YES ==
259       GNUNET_CONTAINER_multihashmap_contains (barrier_map, &key))
260   {
261     GNUNET_break (0);
262     return NULL;
263   }
264   LOG_DEBUG ("Initialising barrier `%s'\n", name);
265   barrier = GNUNET_new (struct GNUNET_TESTBED_Barrier);
266   barrier->c = controller;
267   barrier->name = GNUNET_strdup (name);
268   barrier->cb = cb;
269   barrier->cls = cls;
270   barrier->echo = echo;
271   (void) memcpy (&barrier->key, &key, sizeof (struct GNUNET_HashCode));
272   GNUNET_assert (GNUNET_OK ==
273                  GNUNET_CONTAINER_multihashmap_put (barrier_map, &barrier->key,
274                                                     barrier,
275                                                     GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST));
276   msize = name_len + sizeof (struct GNUNET_TESTBED_BarrierInit);
277   msg = GNUNET_malloc (msize);
278   msg->header.size = htons (msize);
279   msg->header.type = htons (GNUNET_MESSAGE_TYPE_TESTBED_BARRIER_INIT);
280   msg->quorum = (uint8_t) quorum;
281   (void) memcpy (msg->name, barrier->name, name_len);
282   GNUNET_TESTBED_queue_message_ (barrier->c, &msg->header);
283   return barrier;
284 }
285
286
287 /**
288  * Initialise a barrier and call the given callback when the required percentage
289  * of peers (quorum) reach the barrier OR upon error.
290  *
291  * @param controller the handle to the controller
292  * @param name identification name of the barrier
293  * @param quorum the percentage of peers that is required to reach the barrier.
294  *   Peers signal reaching a barrier by calling
295  *   GNUNET_TESTBED_barrier_reached().
296  * @param cb the callback to call when the barrier is reached or upon error.
297  *   Cannot be NULL.
298  * @param cls closure for the above callback
299  * @return barrier handle; NULL upon error
300  */
301 struct GNUNET_TESTBED_Barrier *
302 GNUNET_TESTBED_barrier_init (struct GNUNET_TESTBED_Controller *controller,
303                              const char *name,
304                              unsigned int quorum,
305                              GNUNET_TESTBED_barrier_status_cb cb, void *cls)
306 {
307   return GNUNET_TESTBED_barrier_init_ (controller, name, quorum, cb, cls, GNUNET_YES);
308 }
309
310
311 /**
312  * Cancel a barrier.
313  *
314  * @param barrier the barrier handle
315  */
316 void
317 GNUNET_TESTBED_barrier_cancel (struct GNUNET_TESTBED_Barrier *barrier)
318 {
319   struct GNUNET_TESTBED_BarrierCancel *msg;
320   uint16_t msize;
321
322   msize = sizeof (struct GNUNET_TESTBED_BarrierCancel) + strlen (barrier->name);
323   msg = GNUNET_malloc (msize);
324   msg->header.size = htons (msize);
325   msg->header.type = htons (GNUNET_MESSAGE_TYPE_TESTBED_BARRIER_CANCEL);
326   (void) memcpy (msg->name, barrier->name, strlen (barrier->name));
327   GNUNET_TESTBED_queue_message_ (barrier->c, &msg->header);
328   barrier_remove (barrier);
329 }
330
331
332 /**
333  * Barrier wait handle
334  */
335 struct GNUNET_TESTBED_BarrierWaitHandle
336 {
337   /**
338    * The name of the barrier
339    */
340   char *name;
341
342   /**
343    * Then configuration used for the client connection
344    */
345   struct GNUNET_CONFIGURATION_Handle *cfg;
346
347   /**
348    * The client connection
349    */
350   struct GNUNET_CLIENT_Connection *conn;
351
352   /**
353    * Transmit handle
354    */
355   struct GNUNET_CLIENT_TransmitHandle *tx;
356
357   /**
358    * The message to transmit with tx
359    */
360   struct GNUNET_MessageHeader *msg;
361
362   /**
363    * The barrier wait callback
364    */
365   GNUNET_TESTBED_barrier_wait_cb cb;
366
367   /**
368    * The closure for the above callback
369    */
370   void *cls;
371 };
372
373
374 /**
375  * Function to destroy barrier wait handle
376  *
377  * @param h the handle to destroy
378  */
379 static void
380 destroy_handle (struct GNUNET_TESTBED_BarrierWaitHandle *h)
381 {
382   GNUNET_free (h->name);
383   if (NULL != h->tx)
384     GNUNET_CLIENT_notify_transmit_ready_cancel (h->tx);
385   if (NULL != h->conn)
386     GNUNET_CLIENT_disconnect (h->conn);
387   if (NULL != h->msg)
388     GNUNET_free (h->msg);
389   GNUNET_CONFIGURATION_destroy (h->cfg);
390   GNUNET_free (h);
391 }
392
393
394 /**
395  * Type of a function to call when we receive a message
396  * from the service.
397  *
398  * @param cls closure
399  * @param message received message; NULL on timeout or fatal error
400  */
401 static void
402 receive_handler (void *cls, const struct GNUNET_MessageHeader *message)
403 {
404   struct GNUNET_TESTBED_BarrierWaitHandle *h = cls;
405   const struct GNUNET_TESTBED_BarrierStatusMsg *msg;
406   uint16_t msize;
407
408   if (NULL == message)
409   {
410     GNUNET_break_op (0);
411     goto fail;
412   }
413   if (GNUNET_MESSAGE_TYPE_TESTBED_BARRIER_STATUS != ntohs (message->type))
414   {
415     GNUNET_break_op (0);
416     goto fail;
417   }
418   msize = ntohs (message->size);
419   if (msize <= sizeof (struct GNUNET_TESTBED_BarrierStatusMsg))
420   {
421     GNUNET_break_op (0);
422     goto fail;
423   }
424   msg = (const struct GNUNET_TESTBED_BarrierStatusMsg *) message;
425   switch (ntohs (msg->status))
426   {
427   case GNUNET_TESTBED_BARRIERSTATUS_ERROR:
428     goto fail;
429   case GNUNET_TESTBED_BARRIERSTATUS_INITIALISED:
430     GNUNET_break (0);           /* FIXME */
431     goto destroy;
432   case GNUNET_TESTBED_BARRIERSTATUS_CROSSED:
433     h->cb (h->cls, h->name, GNUNET_OK);
434     goto destroy;
435   default:
436     GNUNET_break_op (0);
437   }
438
439  fail:
440   h->cb (h->cls, h->name, GNUNET_SYSERR);
441
442  destroy:
443   destroy_handle (h);
444 }
445
446
447 /**
448  * Function called to notify a client about the connection
449  * begin ready to queue more data.  "buf" will be
450  * NULL and "size" zero if the connection was closed for
451  * writing in the meantime.
452  *
453  * @param cls closure
454  * @param size number of bytes available in buf
455  * @param buf where the callee should write the message
456  * @return number of bytes written to buf
457  */
458 static size_t
459 transmit_notify (void *cls, size_t size, void *buf)
460 {
461   struct GNUNET_TESTBED_BarrierWaitHandle *h = cls;
462   uint16_t msize;
463
464   h->tx = NULL;
465   if ((0 == size) || (NULL == buf))
466   {
467     destroy_handle (h);
468     return 0;
469   }
470   msize = htons (h->msg->size);
471   GNUNET_assert (msize <= size);
472   (void) memcpy (buf, h->msg, msize);
473   GNUNET_free (h->msg);
474   h->msg = NULL;
475   GNUNET_CLIENT_receive (h->conn, &receive_handler, h, GNUNET_TIME_UNIT_FOREVER_REL);
476   return msize;
477 }
478
479
480 /**
481  * Wait for a barrier to be crossed.  This function should be called by the
482  * peers which have been started by the testbed.  If the peer is not started by
483  * testbed this function may return error
484  *
485  * @param name the name of the barrier
486  * @param cb the barrier wait callback
487  * @param cls the closure for the above callback
488  * @return barrier wait handle which can be used to cancel the waiting at
489  *   anytime before the callback is called.  NULL upon error.
490  */
491 struct GNUNET_TESTBED_BarrierWaitHandle *
492 GNUNET_TESTBED_barrier_wait (const char *name,
493                              GNUNET_TESTBED_barrier_wait_cb cb,
494                              void *cls)
495 {
496   struct GNUNET_TESTBED_BarrierWait *msg;
497   struct GNUNET_CONFIGURATION_Handle *cfg;
498   struct GNUNET_TESTBED_BarrierWaitHandle *h;
499   char *cfg_filename;
500   size_t name_len;
501   uint16_t msize;
502
503   GNUNET_assert (NULL != cb);
504   GNUNET_assert (NULL != name);
505   cfg_filename = getenv (ENV_TESTBED_CONFIG);
506   if (NULL == cfg_filename)
507   {
508     LOG (GNUNET_ERROR_TYPE_ERROR, "Are you running under testbed?\n");
509     return NULL;
510   }
511   cfg = GNUNET_CONFIGURATION_create ();
512   if (GNUNET_OK != GNUNET_CONFIGURATION_load (cfg, cfg_filename))
513   {
514     LOG (GNUNET_ERROR_TYPE_ERROR, "Unable to load configuration from file `%s'\n",
515          cfg_filename);
516     GNUNET_CONFIGURATION_destroy (cfg);
517     return NULL;
518   }
519   h = GNUNET_new (struct GNUNET_TESTBED_BarrierWaitHandle);
520   h->name = GNUNET_strdup (name);
521   h->cfg = cfg;
522   h->conn = GNUNET_CLIENT_connect ("testbed-barrier", h->cfg);
523   h->cb = cb;
524   h->cls = cls;
525   if (NULL == h->conn)
526   {
527     LOG (GNUNET_ERROR_TYPE_ERROR, "Unable to connect to local testbed-barrier service\n");
528     destroy_handle (h);
529     return NULL;
530   }
531   name_len = strlen (name);
532   msize = sizeof (struct GNUNET_TESTBED_BarrierWait) + name_len;
533   msg = GNUNET_malloc (msize);
534   msg->header.type = htons (GNUNET_MESSAGE_TYPE_TESTBED_BARRIER_WAIT);
535   msg->header.size = htons (msize);
536   (void) memcpy (msg->name, name, name_len);
537   h->msg = &msg->header;
538   h->tx =
539       GNUNET_CLIENT_notify_transmit_ready (h->conn, msize,
540                                            GNUNET_TIME_UNIT_FOREVER_REL,
541                                            GNUNET_NO,
542                                            &transmit_notify,
543                                            h);
544   return h;
545 }
546
547
548 /**
549  * Cancel a barrier wait handle
550  *
551  * @param h the barrier wait handle
552  */
553 void
554 GNUNET_TESTBED_barrier_wait_cancel (struct GNUNET_TESTBED_BarrierWaitHandle *h)
555 {
556   destroy_handle (h);
557 }
558
559 /* end of testbed_api_barriers.c */