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) &&
 210       (GNUNET_TESTBED_BARRIERSTATUS_CROSSED == status))
 211     GNUNET_TESTBED_queue_message_ (c, GNUNET_copy_message (&msg->header));
 212   barrier->cb (barrier->cls, name, barrier, status, emsg);
 213   if (GNUNET_TESTBED_BARRIERSTATUS_INITIALISED == status)
 214     return;           /* just initialised; skip cleanup */
 215
 216  cleanup:
 217   GNUNET_free_non_null (emsg);
 218   if (NULL != barrier)
 219     barrier_remove (barrier);
 220 }
 221
 222
 223 /**
 224  * Initialise a barrier and call the given callback when the required percentage
 225  * of peers (quorum) reach the barrier OR upon error.
 226  *
 227  * @param controller the handle to the controller
 228  * @param name identification name of the barrier
 229  * @param quorum the percentage of peers that is required to reach the barrier.
 230  *   Peers signal reaching a barrier by calling
 231  *   GNUNET_TESTBED_barrier_reached().
 232  * @param cb the callback to call when the barrier is reached or upon error.
 233  *   Cannot be NULL.
 234  * @param cls closure for the above callback
 235  * @param echo GNUNET_YES to echo the barrier crossed status message back to the
 236  *   controller
 237  * @return barrier handle; NULL upon error
 238  */
 239 struct GNUNET_TESTBED_Barrier *
 240 GNUNET_TESTBED_barrier_init_ (struct GNUNET_TESTBED_Controller *controller,
 241                               const char *name,
 242                               unsigned int quorum,
 243                               GNUNET_TESTBED_barrier_status_cb cb, void *cls,
 244                               int echo)
 245 {
 246   struct GNUNET_TESTBED_BarrierInit *msg;
 247   struct GNUNET_TESTBED_Barrier *barrier;
 248   struct GNUNET_HashCode key;
 249   size_t name_len;
 250   uint16_t msize;
 251
 252   GNUNET_assert (quorum <= 100);
 253   GNUNET_assert (NULL != cb);
 254   name_len = strlen (name);
 255   GNUNET_assert (0 < name_len);
 256   GNUNET_CRYPTO_hash (name, name_len, &key);
 257   if (NULL == barrier_map)
 258     barrier_map = GNUNET_CONTAINER_multihashmap_create (3, GNUNET_YES);
 259   if (GNUNET_YES ==
 260       GNUNET_CONTAINER_multihashmap_contains (barrier_map, &key))
 261   {
 262     GNUNET_break (0);
 263     return NULL;
 264   }
 265   LOG_DEBUG ("Initialising barrier `%s'\n", name);
 266   barrier = GNUNET_new (struct GNUNET_TESTBED_Barrier);
 267   barrier->c = controller;
 268   barrier->name = GNUNET_strdup (name);
 269   barrier->cb = cb;
 270   barrier->cls = cls;
 271   barrier->echo = echo;
 272   (void) memcpy (&barrier->key, &key, sizeof (struct GNUNET_HashCode));
 273   GNUNET_assert (GNUNET_OK ==
 274                  GNUNET_CONTAINER_multihashmap_put (barrier_map,
 275                                                     &barrier->key,
 276                                                     barrier,
 277                                                     GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST));
 278   msize = name_len + sizeof (struct GNUNET_TESTBED_BarrierInit);
 279   msg = GNUNET_malloc (msize);
 280   msg->header.size = htons (msize);
 281   msg->header.type = htons (GNUNET_MESSAGE_TYPE_TESTBED_BARRIER_INIT);
 282   msg->quorum = (uint8_t) quorum;
 283   (void) memcpy (msg->name, barrier->name, name_len);
 284   GNUNET_TESTBED_queue_message_ (barrier->c, &msg->header);
 285   return barrier;
 286 }
 287
 288
 289 /**
 290  * Initialise a barrier and call the given callback when the required percentage
 291  * of peers (quorum) reach the barrier OR upon error.
 292  *
 293  * @param controller the handle to the controller
 294  * @param name identification name of the barrier
 295  * @param quorum the percentage of peers that is required to reach the barrier.
 296  *   Peers signal reaching a barrier by calling
 297  *   GNUNET_TESTBED_barrier_reached().
 298  * @param cb the callback to call when the barrier is reached or upon error.
 299  *   Cannot be NULL.
 300  * @param cls closure for the above callback
 301  * @return barrier handle; NULL upon error
 302  */
 303 struct GNUNET_TESTBED_Barrier *
 304 GNUNET_TESTBED_barrier_init (struct GNUNET_TESTBED_Controller *controller,
 305                              const char *name,
 306                              unsigned int quorum,
 307                              GNUNET_TESTBED_barrier_status_cb cb, void *cls)
 308 {
 309   return GNUNET_TESTBED_barrier_init_ (controller,
 310                                        name, quorum, cb, cls, GNUNET_YES);
 311 }
 312
 313
 314 /**
 315  * Cancel a barrier.
 316  *
 317  * @param barrier the barrier handle
 318  */
 319 void
 320 GNUNET_TESTBED_barrier_cancel (struct GNUNET_TESTBED_Barrier *barrier)
 321 {
 322   struct GNUNET_TESTBED_BarrierCancel *msg;
 323   uint16_t msize;
 324
 325   msize = sizeof (struct GNUNET_TESTBED_BarrierCancel) + strlen (barrier->name);
 326   msg = GNUNET_malloc (msize);
 327   msg->header.size = htons (msize);
 328   msg->header.type = htons (GNUNET_MESSAGE_TYPE_TESTBED_BARRIER_CANCEL);
 329   (void) memcpy (msg->name, barrier->name, strlen (barrier->name));
 330   GNUNET_TESTBED_queue_message_ (barrier->c, &msg->header);
 331   barrier_remove (barrier);
 332 }
 333
 334
 335 /**
 336  * Barrier wait handle
 337  */
 338 struct GNUNET_TESTBED_BarrierWaitHandle
 339 {
 340   /**
 341    * The name of the barrier
 342    */
 343   char *name;
 344
 345   /**
 346    * Then configuration used for the client connection
 347    */
 348   struct GNUNET_CONFIGURATION_Handle *cfg;
 349
 350   /**
 351    * The client connection
 352    */
 353   struct GNUNET_CLIENT_Connection *conn;
 354
 355   /**
 356    * Transmit handle
 357    */
 358   struct GNUNET_CLIENT_TransmitHandle *tx;
 359
 360   /**
 361    * The message to transmit with tx
 362    */
 363   struct GNUNET_MessageHeader *msg;
 364
 365   /**
 366    * The barrier wait callback
 367    */
 368   GNUNET_TESTBED_barrier_wait_cb cb;
 369
 370   /**
 371    * The closure for the above callback
 372    */
 373   void *cls;
 374 };
 375
 376
 377 /**
 378  * Function to destroy barrier wait handle
 379  *
 380  * @param h the handle to destroy
 381  */
 382 static void
 383 destroy_handle (struct GNUNET_TESTBED_BarrierWaitHandle *h)
 384 {
 385   GNUNET_free (h->name);
 386   if (NULL != h->tx)
 387     GNUNET_CLIENT_notify_transmit_ready_cancel (h->tx);
 388   if (NULL != h->conn)
 389     GNUNET_CLIENT_disconnect (h->conn);
 390   if (NULL != h->msg)
 391     GNUNET_free (h->msg);
 392   GNUNET_CONFIGURATION_destroy (h->cfg);
 393   GNUNET_free (h);
 394 }
 395
 396
 397 /**
 398  * Type of a function to call when we receive a message
 399  * from the service.
 400  *
 401  * @param cls closure
 402  * @param message received message; NULL on timeout or fatal error
 403  */
 404 static void
 405 receive_handler (void *cls,
 406                  const struct GNUNET_MessageHeader *message)
 407 {
 408   struct GNUNET_TESTBED_BarrierWaitHandle *h = cls;
 409   const struct GNUNET_TESTBED_BarrierStatusMsg *msg;
 410   uint16_t msize;
 411
 412   if (NULL == message)
 413   {
 414     GNUNET_break_op (0);
 415     goto fail;
 416   }
 417   if (GNUNET_MESSAGE_TYPE_TESTBED_BARRIER_STATUS != ntohs (message->type))
 418   {
 419     GNUNET_break_op (0);
 420     goto fail;
 421   }
 422   msize = ntohs (message->size);
 423   if (msize <= sizeof (struct GNUNET_TESTBED_BarrierStatusMsg))
 424   {
 425     GNUNET_break_op (0);
 426     goto fail;
 427   }
 428   msg = (const struct GNUNET_TESTBED_BarrierStatusMsg *) message;
 429   switch (ntohs (msg->status))
 430   {
 431   case GNUNET_TESTBED_BARRIERSTATUS_ERROR:
 432     goto fail;
 433   case GNUNET_TESTBED_BARRIERSTATUS_INITIALISED:
 434     GNUNET_break (0);           /* FIXME */
 435     goto destroy;
 436   case GNUNET_TESTBED_BARRIERSTATUS_CROSSED:
 437     h->cb (h->cls, h->name, GNUNET_OK);
 438     goto destroy;
 439   default:
 440     GNUNET_break_op (0);
 441   }
 442
 443  fail:
 444   h->cb (h->cls, h->name, GNUNET_SYSERR);
 445
 446  destroy:
 447   destroy_handle (h);
 448 }
 449
 450
 451 /**
 452  * Function called to notify a client about the connection
 453  * begin ready to queue more data.  "buf" will be
 454  * NULL and "size" zero if the connection was closed for
 455  * writing in the meantime.
 456  *
 457  * @param cls closure
 458  * @param size number of bytes available in buf
 459  * @param buf where the callee should write the message
 460  * @return number of bytes written to buf
 461  */
 462 static size_t
 463 transmit_notify (void *cls, size_t size, void *buf)
 464 {
 465   struct GNUNET_TESTBED_BarrierWaitHandle *h = cls;
 466   uint16_t msize;
 467
 468   h->tx = NULL;
 469   if ((0 == size) || (NULL == buf))
 470   {
 471     destroy_handle (h);
 472     return 0;
 473   }
 474   msize = htons (h->msg->size);
 475   GNUNET_assert (msize <= size);
 476   (void) memcpy (buf, h->msg, msize);
 477   GNUNET_free (h->msg);
 478   h->msg = NULL;
 479   GNUNET_CLIENT_receive (h->conn, &receive_handler, h, GNUNET_TIME_UNIT_FOREVER_REL);
 480   return msize;
 481 }
 482
 483
 484 /**
 485  * Wait for a barrier to be crossed.  This function should be called by the
 486  * peers which have been started by the testbed.  If the peer is not started by
 487  * testbed this function may return error
 488  *
 489  * @param name the name of the barrier
 490  * @param cb the barrier wait callback
 491  * @param cls the closure for the above callback
 492  * @return barrier wait handle which can be used to cancel the waiting at
 493  *   anytime before the callback is called.  NULL upon error.
 494  */
 495 struct GNUNET_TESTBED_BarrierWaitHandle *
 496 GNUNET_TESTBED_barrier_wait (const char *name,
 497                              GNUNET_TESTBED_barrier_wait_cb cb,
 498                              void *cls)
 499 {
 500   struct GNUNET_TESTBED_BarrierWait *msg;
 501   struct GNUNET_CONFIGURATION_Handle *cfg;
 502   struct GNUNET_TESTBED_BarrierWaitHandle *h;
 503   char *cfg_filename;
 504   size_t name_len;
 505   uint16_t msize;
 506
 507   GNUNET_assert (NULL != cb);
 508   GNUNET_assert (NULL != name);
 509   cfg_filename = getenv (ENV_TESTBED_CONFIG);
 510   if (NULL == cfg_filename)
 511   {
 512     LOG (GNUNET_ERROR_TYPE_ERROR, "Are you running under testbed?\n");
 513     return NULL;
 514   }
 515   cfg = GNUNET_CONFIGURATION_create ();
 516   if (GNUNET_OK != GNUNET_CONFIGURATION_load (cfg, cfg_filename))
 517   {
 518     LOG (GNUNET_ERROR_TYPE_ERROR, "Unable to load configuration from file `%s'\n",
 519          cfg_filename);
 520     GNUNET_CONFIGURATION_destroy (cfg);
 521     return NULL;
 522   }
 523   h = GNUNET_new (struct GNUNET_TESTBED_BarrierWaitHandle);
 524   h->name = GNUNET_strdup (name);
 525   h->cfg = cfg;
 526   h->conn = GNUNET_CLIENT_connect ("testbed-barrier", h->cfg);
 527   h->cb = cb;
 528   h->cls = cls;
 529   if (NULL == h->conn)
 530   {
 531     LOG (GNUNET_ERROR_TYPE_ERROR,
 532          "Unable to connect to local testbed-barrier service\n");
 533     destroy_handle (h);
 534     return NULL;
 535   }
 536   name_len = strlen (name);
 537   msize = sizeof (struct GNUNET_TESTBED_BarrierWait) + name_len;
 538   msg = GNUNET_malloc (msize);
 539   msg->header.type = htons (GNUNET_MESSAGE_TYPE_TESTBED_BARRIER_WAIT);
 540   msg->header.size = htons (msize);
 541   (void) memcpy (msg->name, name, name_len);
 542   h->msg = &msg->header;
 543   h->tx =
 544       GNUNET_CLIENT_notify_transmit_ready (h->conn, msize,
 545                                            GNUNET_TIME_UNIT_FOREVER_REL,
 546                                            GNUNET_NO,
 547                                            &transmit_notify,
 548                                            h);
 549   return h;
 550 }
 551
 552
 553 /**
 554  * Cancel a barrier wait handle
 555  *
 556  * @param h the barrier wait handle
 557  */
 558 void
 559 GNUNET_TESTBED_barrier_wait_cancel (struct GNUNET_TESTBED_BarrierWaitHandle *h)
 560 {
 561   destroy_handle (h);
 562 }
 563
 564 /* end of testbed_api_barriers.c */