src/testbed/gnunet-service-testbed_barriers.c

   1 /*
   2   This file is part of GNUnet.
   3   (C) 2008--2013 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 testbed/gnunet-service-testbed_barriers.c
  23  * @brief barrier handling at the testbed controller
  24  * @author Sree Harsha Totakura <sreeharsha@totakura.in>
  25  */
  26
  27 #include "gnunet-service-testbed.h"
  28 #include "gnunet-service-testbed_barriers.h"
  29 #include "testbed_api_barriers.h"
  30
  31
  32 /**
  33  * timeout for outgoing message transmissions in seconds
  34  */
  35 #define MESSAGE_SEND_TIMEOUT(s) \
  36   GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, s)
  37
  38
  39 /**
  40  * Test to see if local peers have reached the required quorum of a barrier
  41  */
  42 #define LOCAL_QUORUM_REACHED(barrier)           \
  43   ((barrier->quorum * GST_num_local_peers) <= (barrier->nreached * 100))
  44
  45
  46 #ifdef LOG
  47 #undef LOG
  48 #endif
  49
  50 /**
  51  * Logging shorthand
  52  */
  53 #define LOG(kind,...)                                           \
  54   GNUNET_log_from (kind, "testbed-barriers", __VA_ARGS__)
  55
  56
  57 /**
  58  * Barrier
  59  */
  60 struct Barrier;
  61
  62
  63 /**
  64  * Message queue for transmitting messages
  65  */
  66 struct MessageQueue
  67 {
  68   /**
  69    * next pointer for DLL
  70    */
  71   struct MessageQueue *next;
  72
  73   /**
  74    * prev pointer for DLL
  75    */
  76   struct MessageQueue *prev;
  77
  78   /**
  79    * The message to be sent
  80    */
  81   struct GNUNET_MessageHeader *msg;
  82 };
  83
  84
  85 /**
  86  * Context to be associated with each client
  87  */
  88 struct ClientCtx
  89 {
  90   /**
  91    * The barrier this client is waiting for
  92    */
  93   struct Barrier *barrier;
  94
  95   /**
  96    * DLL next ptr
  97    */
  98   struct ClientCtx *next;
  99
 100   /**
 101    * DLL prev ptr
 102    */
 103   struct ClientCtx *prev;
 104
 105   /**
 106    * The client handle
 107    */
 108   struct GNUNET_SERVER_Client *client;
 109
 110   /**
 111    * the transmission handle
 112    */
 113   struct GNUNET_SERVER_TransmitHandle *tx;
 114
 115   /**
 116    * message queue head
 117    */
 118   struct MessageQueue *mq_head;
 119
 120   /**
 121    * message queue tail
 122    */
 123   struct MessageQueue *mq_tail;
 124 };
 125
 126
 127 /**
 128  * Wrapper around Barrier handle
 129  */
 130 struct WBarrier
 131 {
 132   /**
 133    * DLL next pointer
 134    */
 135   struct WBarrier *next;
 136
 137   /**
 138    * DLL prev pointer
 139    */
 140   struct WBarrier *prev;
 141
 142   /**
 143    * The local barrier associated with the creation of this wrapper
 144    */
 145   struct Barrier *barrier;
 146
 147   /**
 148    * The barrier handle from API
 149    */
 150   struct GNUNET_TESTBED_Barrier *hbarrier;
 151
 152   /**
 153    * Has this barrier been crossed?
 154    */
 155   uint8_t reached;
 156 };
 157
 158
 159 /**
 160  * Barrier
 161  */
 162 struct Barrier
 163 {
 164   /**
 165    * The hashcode of the barrier name
 166    */
 167   struct GNUNET_HashCode hash;
 168
 169   /**
 170    * The client handle to the master controller
 171    */
 172   struct GNUNET_SERVER_Client *mc;
 173
 174   /**
 175    * The name of the barrier
 176    */
 177   char *name;
 178
 179   /**
 180    * DLL head for the list of clients waiting for this barrier
 181    */
 182   struct ClientCtx *head;
 183
 184   /**
 185    * DLL tail for the list of clients waiting for this barrier
 186    */
 187   struct ClientCtx *tail;
 188
 189   /**
 190    * DLL head for the list of barrier handles
 191    */
 192   struct WBarrier *whead;
 193
 194   /**
 195    * DLL tail for the list of barrier handles
 196    */
 197   struct WBarrier *wtail;
 198
 199   /**
 200    * Identifier for the timeout task
 201    */
 202   GNUNET_SCHEDULER_TaskIdentifier tout_task;
 203
 204   /**
 205    * The status of this barrier
 206    */
 207   enum GNUNET_TESTBED_BarrierStatus status;
 208
 209   /**
 210    * Number of barriers wrapped in the above DLL
 211    */
 212   unsigned int num_wbarriers;
 213
 214   /**
 215    * Number of wrapped barriers reached so far
 216    */
 217   unsigned int num_wbarriers_reached;
 218
 219   /**
 220    * Number of wrapped barrier initialised so far
 221    */
 222   unsigned int num_wbarriers_inited;
 223
 224   /**
 225    * Number of peers which have reached this barrier
 226    */
 227   unsigned int nreached;
 228
 229   /**
 230    * Number of slaves we have initialised this barrier
 231    */
 232   unsigned int nslaves;
 233
 234   /**
 235    * Quorum percentage to be reached
 236    */
 237   uint8_t quorum;
 238
 239 };
 240
 241
 242 /**
 243  * Hashtable handle for storing initialised barriers
 244  */
 245 static struct GNUNET_CONTAINER_MultiHashMap *barrier_map;
 246
 247 /**
 248  * Service context
 249  */
 250 static struct GNUNET_SERVICE_Context *ctx;
 251
 252
 253 /**
 254  * Function called to notify a client about the connection
 255  * begin ready to queue more data.  "buf" will be
 256  * NULL and "size" zero if the connection was closed for
 257  * writing in the meantime.
 258  *
 259  * @param cls client context
 260  * @param size number of bytes available in buf
 261  * @param buf where the callee should write the message
 262  * @return number of bytes written to buf
 263  */
 264 static size_t
 265 transmit_ready_cb (void *cls, size_t size, void *buf)
 266 {
 267   struct ClientCtx *ctx = cls;
 268   struct GNUNET_SERVER_Client *client = ctx->client;
 269   struct MessageQueue *mq;
 270   struct GNUNET_MessageHeader *msg;
 271   size_t wrote;
 272
 273   ctx->tx = NULL;
 274   wrote = 0;
 275   if ((0 == size) || (NULL == buf))
 276   {
 277     GNUNET_assert (NULL != ctx->client);
 278     GNUNET_SERVER_client_drop (ctx->client);
 279     ctx->client = NULL;
 280     return 0;
 281   }
 282   mq = ctx->mq_head;
 283   msg = mq->msg;
 284   wrote = ntohs (msg->size);
 285   GNUNET_assert (size >= wrote);
 286   (void) memcpy (buf, msg, wrote);
 287   GNUNET_CONTAINER_DLL_remove (ctx->mq_head, ctx->mq_tail, mq);
 288   GNUNET_free (mq->msg);
 289   GNUNET_free (mq);
 290   if (NULL != (mq = ctx->mq_head))
 291     ctx->tx = GNUNET_SERVER_notify_transmit_ready (client, ntohs (msg->size),
 292                                                   MESSAGE_SEND_TIMEOUT (30),
 293                                                   &transmit_ready_cb, ctx);
 294   return wrote;
 295 }
 296
 297
 298 /**
 299  * Queue a message into a clients message queue
 300  *
 301  * @param ctx the context associated with the client
 302  * @param msg the message to queue.  Will be consumed
 303  */
 304 static void
 305 queue_message (struct ClientCtx *ctx, struct GNUNET_MessageHeader *msg)
 306 {
 307   struct MessageQueue *mq;
 308   struct GNUNET_SERVER_Client *client = ctx->client;
 309
 310   mq = GNUNET_malloc (sizeof (struct MessageQueue));
 311   mq->msg = msg;
 312   LOG_DEBUG ("Queueing message of type %u, size %u for sending\n",
 313              ntohs (msg->type), ntohs (msg->size));
 314   GNUNET_CONTAINER_DLL_insert_tail (ctx->mq_head, ctx->mq_tail, mq);
 315   if (NULL == ctx->tx)
 316    ctx->tx = GNUNET_SERVER_notify_transmit_ready (client, ntohs (msg->size),
 317                                                   MESSAGE_SEND_TIMEOUT (30),
 318                                                   &transmit_ready_cb, ctx);
 319 }
 320
 321
 322 /**
 323  * Function to cleanup client context data structure
 324  *
 325  * @param ctx the client context data structure
 326  */
 327 static void
 328 cleanup_clientctx (struct ClientCtx *ctx)
 329 {
 330   struct MessageQueue *mq;
 331
 332   if (NULL != ctx->client)
 333     GNUNET_SERVER_client_drop (ctx->client);
 334   if (NULL != ctx->tx)
 335     GNUNET_SERVER_notify_transmit_ready_cancel (ctx->tx);
 336   if (NULL != (mq = ctx->mq_head))
 337   {
 338     GNUNET_CONTAINER_DLL_remove (ctx->mq_head, ctx->mq_tail, mq);
 339     GNUNET_free (mq->msg);
 340     GNUNET_free (mq);
 341   }
 342   GNUNET_free (ctx);
 343 }
 344
 345
 346 /**
 347  * Function to remove a barrier from the barrier map and cleanup resources
 348  * occupied by a barrier
 349  *
 350  * @param barrier the barrier handle
 351  */
 352 static void
 353 remove_barrier (struct Barrier *barrier)
 354 {
 355   struct ClientCtx *ctx;
 356
 357   GNUNET_assert (GNUNET_YES == GNUNET_CONTAINER_multihashmap_remove (barrier_map,
 358                                                                     &barrier->hash,
 359                                                                     barrier));
 360   while (NULL != (ctx = barrier->head))
 361   {
 362     GNUNET_CONTAINER_DLL_remove (barrier->head, barrier->tail, ctx);
 363     cleanup_clientctx (ctx);
 364   }
 365   GNUNET_free (barrier->name);
 366   GNUNET_SERVER_client_drop (barrier->mc);
 367   GNUNET_free (barrier);
 368 }
 369
 370
 371 /**
 372  * Cancels all subcontroller barrier handles
 373  *
 374  * @param barrier the local barrier
 375  */
 376 static void
 377 cancel_wrappers (struct Barrier *barrier)
 378 {
 379   struct WBarrier *wrapper;
 380
 381   while (NULL != (wrapper = barrier->whead))
 382   {
 383     GNUNET_TESTBED_barrier_cancel (wrapper->hbarrier);
 384     GNUNET_CONTAINER_DLL_remove (barrier->whead, barrier->wtail, wrapper);
 385     GNUNET_free (wrapper);
 386   }
 387 }
 388
 389
 390 /**
 391  * Send a status message about a barrier to the given client
 392  *
 393  * @param client the client to send the message to
 394  * @param name the barrier name
 395  * @param status the status of the barrier
 396  * @param emsg the error message; should be non-NULL for
 397  *   status=GNUNET_TESTBED_BARRIERSTATUS_ERROR
 398  */
 399 static void
 400 send_client_status_msg (struct GNUNET_SERVER_Client *client,
 401                         const char *name,
 402                         enum GNUNET_TESTBED_BarrierStatus status,
 403                         const char *emsg)
 404 {
 405   struct GNUNET_TESTBED_BarrierStatusMsg *msg;
 406   size_t name_len;
 407   uint16_t msize;
 408
 409   GNUNET_assert ((NULL == emsg) || (GNUNET_TESTBED_BARRIERSTATUS_ERROR == status));
 410   name_len = strlen (name);
 411   msize = sizeof (struct GNUNET_TESTBED_BarrierStatusMsg)
 412       + (name_len + 1)
 413       + ((NULL == emsg) ? 0 : (strlen (emsg) + 1));
 414   msg = GNUNET_malloc (msize);
 415   msg->header.size = htons (msize);
 416   msg->header.type = htons (GNUNET_MESSAGE_TYPE_TESTBED_BARRIER_STATUS);
 417   msg->status = htons (status);
 418   msg->name_len = htons ((uint16_t) name_len);
 419   (void) memcpy (msg->data, name, name_len);
 420   if (NULL != emsg)
 421     (void) memcpy (msg->data + name_len + 1, emsg, strlen (emsg));
 422   GST_queue_message (client, &msg->header);
 423 }
 424
 425
 426 /**
 427  * Sends a barrier failed message
 428  *
 429  * @param barrier the corresponding barrier
 430  * @param emsg the error message; should be non-NULL for
 431  *   status=GNUNET_TESTBED_BARRIERSTATUS_ERROR
 432  */
 433 static void
 434 send_barrier_status_msg (struct Barrier *barrier, const char *emsg)
 435 {
 436   GNUNET_assert (0 != barrier->status);
 437   send_client_status_msg (barrier->mc, barrier->name, barrier->status, emsg);
 438 }
 439
 440
 441 /**
 442  * Message handler for GNUNET_MESSAGE_TYPE_TESTBED_BARRIER_WAIT messages.  This
 443  * message should come from peers or a shared helper service using the
 444  * testbed-barrier client API (@see gnunet_testbed_barrier_service.h)
 445  *
 446  * This handler is queued in the main service and will handle the messages sent
 447  * either from the testbed driver or from a high level controller
 448  *
 449  * @param cls NULL
 450  * @param client identification of the client
 451  * @param message the actual message
 452  */
 453 static void
 454 handle_barrier_wait (void *cls, struct GNUNET_SERVER_Client *client,
 455                      const struct GNUNET_MessageHeader *message)
 456 {
 457   const struct GNUNET_TESTBED_BarrierWait *msg;
 458   struct Barrier *barrier;
 459   char *name;
 460   struct ClientCtx *client_ctx;
 461   struct GNUNET_HashCode key;
 462   size_t name_len;
 463   uint16_t msize;
 464
 465   msize = ntohs (message->size);
 466   if (msize <= sizeof (struct GNUNET_TESTBED_BarrierWait))
 467   {
 468     GNUNET_break_op (0);
 469     GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
 470     return;
 471   }
 472   if (NULL == barrier_map)
 473   {
 474     GNUNET_break (0);
 475     GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
 476     return;
 477   }
 478   msg = (const struct GNUNET_TESTBED_BarrierWait *) message;
 479   name_len = msize - sizeof (struct GNUNET_TESTBED_BarrierWait);
 480   name = GNUNET_malloc (name_len + 1);
 481   name[name_len] = '\0';
 482   (void) memcpy (name, msg->name, name_len);
 483   LOG_DEBUG ("Received BARRIER_WAIT for barrier `%s'\n", name);
 484   GNUNET_CRYPTO_hash (name, name_len, &key);
 485   if (NULL == (barrier = GNUNET_CONTAINER_multihashmap_get (barrier_map, &key)))
 486   {
 487     GNUNET_break (0);
 488     GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
 489     GNUNET_free (name);
 490     return;
 491   }
 492   client_ctx = GNUNET_SERVER_client_get_user_context (client, struct ClientCtx);
 493   if (NULL == client_ctx)
 494   {
 495     client_ctx = GNUNET_malloc (sizeof (struct ClientCtx));
 496     client_ctx->client = client;
 497     GNUNET_SERVER_client_keep (client);
 498     client_ctx->barrier = barrier;
 499     GNUNET_CONTAINER_DLL_insert_tail (barrier->head, barrier->tail, client_ctx);
 500     GNUNET_SERVER_client_set_user_context (client, client_ctx);
 501   }
 502   barrier->nreached++;
 503   if ((barrier->num_wbarriers_reached == barrier->num_wbarriers)
 504         && (LOCAL_QUORUM_REACHED (barrier)))
 505   {
 506     barrier->status = GNUNET_TESTBED_BARRIERSTATUS_CROSSED;
 507     send_barrier_status_msg (barrier, NULL);
 508   }
 509   GNUNET_SERVER_receive_done (client, GNUNET_OK);
 510 }
 511
 512
 513 /**
 514  * Functions with this signature are called whenever a client
 515  * is disconnected on the network level.
 516  *
 517  * @param cls closure
 518  * @param client identification of the client; NULL
 519  *        for the last call when the server is destroyed
 520  */
 521 static void
 522 disconnect_cb (void *cls, struct GNUNET_SERVER_Client *client)
 523 {
 524   struct ClientCtx *client_ctx;
 525
 526   if (NULL == client)
 527     return;
 528   client_ctx = GNUNET_SERVER_client_get_user_context (client, struct ClientCtx);
 529   if (NULL == client_ctx)
 530     return;
 531   cleanup_clientctx (client_ctx);
 532 }
 533
 534
 535 /**
 536  * Function to initialise barrriers component
 537  *
 538  * @param cfg the configuration to use for initialisation
 539  */
 540 void
 541 GST_barriers_init (struct GNUNET_CONFIGURATION_Handle *cfg)
 542 {
 543   static const struct GNUNET_SERVER_MessageHandler message_handlers[] = {
 544     {&handle_barrier_wait, NULL, GNUNET_MESSAGE_TYPE_TESTBED_BARRIER_WAIT, 0},
 545     {NULL, NULL, 0, 0}
 546   };
 547   struct GNUNET_SERVER_Handle *srv;
 548
 549   barrier_map = GNUNET_CONTAINER_multihashmap_create (3, GNUNET_YES);
 550   ctx = GNUNET_SERVICE_start ("testbed-barrier", cfg,
 551                               GNUNET_SERVICE_OPTION_MANUAL_SHUTDOWN);
 552   srv = GNUNET_SERVICE_get_server (ctx);
 553   GNUNET_SERVER_add_handlers (srv, message_handlers);
 554   GNUNET_SERVER_disconnect_notify (srv, &disconnect_cb, NULL);
 555 }
 556
 557
 558 /**
 559  * Function to stop the barrier service
 560  */
 561 void
 562 GST_barriers_destroy ()
 563 {
 564   GNUNET_assert (NULL != barrier_map);
 565   GNUNET_CONTAINER_multihashmap_destroy (barrier_map);
 566   GNUNET_assert (NULL != ctx);
 567   GNUNET_SERVICE_stop (ctx);
 568 }
 569
 570
 571 /**
 572  * Functions of this type are to be given as callback argument to
 573  * GNUNET_TESTBED_barrier_init().  The callback will be called when status
 574  * information is available for the barrier.
 575  *
 576  * @param cls the closure given to GNUNET_TESTBED_barrier_init()
 577  * @param name the name of the barrier
 578  * @param b_ the barrier handle
 579  * @param status status of the barrier; GNUNET_OK if the barrier is crossed;
 580  *   GNUNET_SYSERR upon error
 581  * @param emsg if the status were to be GNUNET_SYSERR, this parameter has the
 582  *   error messsage
 583  */
 584 static void
 585 wbarrier_status_cb (void *cls, const char *name,
 586                     struct GNUNET_TESTBED_Barrier *b_,
 587                     enum GNUNET_TESTBED_BarrierStatus status,
 588                     const char *emsg)
 589 {
 590   struct WBarrier *wrapper = cls;
 591   struct Barrier *barrier = wrapper->barrier;
 592
 593   GNUNET_assert (b_ == wrapper->hbarrier);
 594   wrapper->hbarrier = NULL;
 595   GNUNET_CONTAINER_DLL_remove (barrier->whead, barrier->wtail, wrapper);
 596   GNUNET_free (wrapper);
 597   switch (status)
 598   {
 599   case GNUNET_TESTBED_BARRIERSTATUS_ERROR:
 600     LOG (GNUNET_ERROR_TYPE_ERROR,
 601          "Initialising barrier `%s' failed at a sub-controller: %s\n",
 602          barrier->name, (NULL != emsg) ? emsg : "NULL");
 603     cancel_wrappers (barrier);
 604     if (NULL == emsg)
 605       emsg = "Initialisation failed at a sub-controller";
 606     barrier->status = GNUNET_TESTBED_BARRIERSTATUS_ERROR;
 607     send_barrier_status_msg (barrier, emsg);
 608     return;
 609   case GNUNET_TESTBED_BARRIERSTATUS_CROSSED:
 610     if (GNUNET_TESTBED_BARRIERSTATUS_INITIALISED != barrier->status)
 611     {
 612       GNUNET_break_op (0);
 613       return;
 614     }
 615     barrier->num_wbarriers_reached++;
 616     if ((barrier->num_wbarriers_reached == barrier->num_wbarriers)
 617         && (LOCAL_QUORUM_REACHED (barrier)))
 618     {
 619       barrier->status = GNUNET_TESTBED_BARRIERSTATUS_CROSSED;
 620       send_barrier_status_msg (barrier, NULL);
 621     }
 622     return;
 623   case GNUNET_TESTBED_BARRIERSTATUS_INITIALISED:
 624     if (0 != barrier->status)
 625     {
 626       GNUNET_break_op (0);
 627       return;
 628     }
 629     barrier->num_wbarriers_inited++;
 630     if (barrier->num_wbarriers_inited == barrier->num_wbarriers)
 631     {
 632       barrier->status = GNUNET_TESTBED_BARRIERSTATUS_INITIALISED;
 633       send_barrier_status_msg (barrier, NULL);
 634     }
 635     return;
 636   }
 637 }
 638
 639
 640 /**
 641  * Function called upon timeout while waiting for a response from the
 642  * subcontrollers to barrier init message
 643  *
 644  * @param cls barrier
 645  * @param tc scheduler task context
 646  */
 647 static void
 648 fwd_tout_barrier_init (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 649 {
 650   struct Barrier *barrier = cls;
 651
 652   cancel_wrappers (barrier);
 653   barrier->status = GNUNET_TESTBED_BARRIERSTATUS_ERROR;
 654   send_barrier_status_msg (barrier,
 655                            "Timedout while propagating barrier initialisation\n");
 656   remove_barrier (barrier);
 657 }
 658
 659
 660 /**
 661  * Message handler for GNUNET_MESSAGE_TYPE_TESTBED_BARRIER_INIT messages.  This
 662  * message should always come from a parent controller or the testbed API if we
 663  * are the root controller.
 664  *
 665  * This handler is queued in the main service and will handle the messages sent
 666  * either from the testbed driver or from a high level controller
 667  *
 668  * @param cls NULL
 669  * @param client identification of the client
 670  * @param message the actual message
 671  */
 672 void
 673 GST_handle_barrier_init (void *cls, struct GNUNET_SERVER_Client *client,
 674                          const struct GNUNET_MessageHeader *message)
 675 {
 676   const struct GNUNET_TESTBED_BarrierInit *msg;
 677   char *name;
 678   struct Barrier *barrier;
 679   struct Slave *slave;
 680   struct WBarrier *wrapper;
 681   struct GNUNET_HashCode hash;
 682   size_t name_len;
 683   unsigned int cnt;
 684   uint16_t msize;
 685
 686   if (NULL == GST_context)
 687   {
 688     GNUNET_break_op (0);
 689     GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
 690     return;
 691   }
 692   if (client != GST_context->client)
 693   {
 694     GNUNET_break_op (0);
 695     GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
 696     return;
 697   }
 698   msize = ntohs (message->size);
 699   if (msize <= sizeof (struct GNUNET_TESTBED_BarrierInit))
 700   {
 701     GNUNET_break_op (0);
 702     GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
 703     return;
 704   }
 705   msg = (const struct GNUNET_TESTBED_BarrierInit *) message;
 706   name_len = (size_t) msize - sizeof (struct GNUNET_TESTBED_BarrierInit);
 707   name = GNUNET_malloc (name_len + 1);
 708   (void) memcpy (name, msg->name, name_len);
 709   GNUNET_CRYPTO_hash (name, name_len, &hash);
 710   LOG_DEBUG ("Received BARRIER_INIT for barrier `%s'\n", name);
 711   if (GNUNET_YES == GNUNET_CONTAINER_multihashmap_contains (barrier_map, &hash))
 712   {
 713
 714     send_client_status_msg (client, name, GNUNET_TESTBED_BARRIERSTATUS_ERROR,
 715                             "A barrier with the same name already exists");
 716     GNUNET_free (name);
 717     GNUNET_SERVER_receive_done (client, GNUNET_OK);
 718     return;
 719   }
 720   barrier = GNUNET_malloc (sizeof (struct Barrier));
 721   (void) memcpy (&barrier->hash, &hash, sizeof (struct GNUNET_HashCode));
 722   barrier->quorum = msg->quorum;
 723   barrier->name = name;
 724   barrier->mc = client;
 725   GNUNET_SERVER_client_keep (client);
 726   GNUNET_assert (GNUNET_OK ==
 727                  GNUNET_CONTAINER_multihashmap_put (barrier_map,
 728                                                     &barrier->hash,
 729                                                     barrier,
 730                                                     GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST));
 731   GNUNET_SERVER_receive_done (client, GNUNET_OK);
 732   /* Propagate barrier init to subcontrollers */
 733   for (cnt = 0; cnt < GST_slave_list_size; cnt++)
 734   {
 735     if (NULL == (slave = GST_slave_list[cnt]))
 736       continue;
 737     if (NULL == slave->controller)
 738     {
 739       GNUNET_break (0);/* May happen when we are connecting to the controller */
 740       continue;
 741     }
 742     wrapper = GNUNET_malloc (sizeof (struct WBarrier));
 743     wrapper->barrier = barrier;
 744     GNUNET_CONTAINER_DLL_insert_tail (barrier->whead, barrier->wtail, wrapper);
 745     wrapper->hbarrier = GNUNET_TESTBED_barrier_init_ (slave->controller,
 746                                                       barrier->name,
 747                                                       barrier->quorum,
 748                                                       &wbarrier_status_cb,
 749                                                       wrapper,
 750                                                       GNUNET_NO);
 751   }
 752   if (NULL == barrier->whead)   /* No further propagation */
 753   {
 754     barrier->status = GNUNET_TESTBED_BARRIERSTATUS_INITIALISED;
 755     LOG_DEBUG ("Sending GNUNET_TESTBED_BARRIERSTATUS_INITIALISED for barrier `%s'\n",
 756                barrier->name);
 757     send_barrier_status_msg (barrier, NULL);
 758   }else
 759     barrier->tout_task = GNUNET_SCHEDULER_add_delayed (MESSAGE_SEND_TIMEOUT (30),
 760                                                        &fwd_tout_barrier_init,
 761                                                        barrier);
 762 }
 763
 764
 765 /**
 766  * Message handler for GNUNET_MESSAGE_TYPE_TESTBED_BARRIER_CANCEL messages.  This
 767  * message should always come from a parent controller or the testbed API if we
 768  * are the root controller.
 769  *
 770  * This handler is queued in the main service and will handle the messages sent
 771  * either from the testbed driver or from a high level controller
 772  *
 773  * @param cls NULL
 774  * @param client identification of the client
 775  * @param message the actual message
 776  */
 777 void
 778 GST_handle_barrier_cancel (void *cls, struct GNUNET_SERVER_Client *client,
 779                            const struct GNUNET_MessageHeader *message)
 780 {
 781   const struct GNUNET_TESTBED_BarrierCancel *msg;
 782   char *name;
 783   struct Barrier *barrier;
 784   struct GNUNET_HashCode hash;
 785   size_t name_len;
 786   uint16_t msize;
 787
 788   if (NULL == GST_context)
 789   {
 790     GNUNET_break_op (0);
 791     GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
 792     return;
 793   }
 794   if (client != GST_context->client)
 795   {
 796     GNUNET_break_op (0);
 797     GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
 798     return;
 799   }
 800   msize = ntohs (message->size);
 801   if (msize <= sizeof (struct GNUNET_TESTBED_BarrierCancel))
 802   {
 803     GNUNET_break_op (0);
 804     GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
 805     return;
 806   }
 807   msg = (const struct GNUNET_TESTBED_BarrierCancel *) message;
 808   name_len = msize - sizeof (struct GNUNET_TESTBED_BarrierCancel);
 809   name = GNUNET_malloc (name_len + 1);
 810   (void) memcpy (name, msg->name, name_len);
 811   GNUNET_CRYPTO_hash (name, name_len, &hash);
 812   if (GNUNET_NO == GNUNET_CONTAINER_multihashmap_contains (barrier_map, &hash))
 813   {
 814     GNUNET_break_op (0);
 815     GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
 816     return;
 817   }
 818   barrier = GNUNET_CONTAINER_multihashmap_get (barrier_map, &hash);
 819   GNUNET_assert (NULL != barrier);
 820   cancel_wrappers (barrier);
 821   remove_barrier (barrier);
 822   GNUNET_SERVER_receive_done (client, GNUNET_OK);
 823 }
 824
 825
 826 /**
 827  * Message handler for GNUNET_MESSAGE_TYPE_TESTBED_BARRIER_STATUS messages.
 828  * This handler is queued in the main service and will handle the messages sent
 829  * either from the testbed driver or from a high level controller
 830  *
 831  * @param cls NULL
 832  * @param client identification of the client
 833  * @param message the actual message
 834  */
 835 void
 836 GST_handle_barrier_status (void *cls, struct GNUNET_SERVER_Client *client,
 837                            const struct GNUNET_MessageHeader *message)
 838 {
 839   const struct GNUNET_TESTBED_BarrierStatusMsg *msg;
 840   struct Barrier *barrier;
 841   struct ClientCtx *client_ctx;
 842   const char *name;
 843   struct GNUNET_HashCode key;
 844   enum GNUNET_TESTBED_BarrierStatus status;
 845   uint16_t msize;
 846   uint16_t name_len;
 847
 848   if (NULL == GST_context)
 849   {
 850     GNUNET_break_op (0);
 851     GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
 852     return;
 853   }
 854   if (client != GST_context->client)
 855   {
 856     GNUNET_break_op (0);
 857     GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
 858     return;
 859   }
 860   msize = ntohs (message->size);
 861   if (msize <= sizeof (struct GNUNET_TESTBED_BarrierStatusMsg))
 862   {
 863     GNUNET_break_op (0);
 864     GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
 865     return;
 866   }
 867   msg = (const struct GNUNET_TESTBED_BarrierStatusMsg *) message;
 868   status = ntohs (msg->status);
 869   if (GNUNET_TESTBED_BARRIERSTATUS_CROSSED != status)
 870   {
 871     GNUNET_break_op (0);        /* current we only expect BARRIER_CROSSED
 872                                    status message this way */
 873     GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
 874     return;
 875   }
 876   name = msg->data;
 877   name_len = ntohs (msg->name_len);
 878   if ((sizeof (struct GNUNET_TESTBED_BarrierStatusMsg) + name_len + 1) != msize)
 879   {
 880     GNUNET_break_op (0);
 881     GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
 882     return;
 883   }
 884   if ('\0' != name[name_len])
 885   {
 886     GNUNET_break_op (0);
 887     GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
 888     return;
 889   }
 890   GNUNET_CRYPTO_hash (name, name_len, &key);
 891   barrier = GNUNET_CONTAINER_multihashmap_get (barrier_map, &key);
 892   if (NULL == barrier)
 893   {
 894     GNUNET_break_op (0);
 895     GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
 896     return;
 897   }
 898   GNUNET_SERVER_receive_done (client, GNUNET_OK);
 899   while (NULL != (client_ctx = barrier->head)) /* Notify peers */
 900   {
 901     queue_message (client_ctx, GNUNET_copy_message (message));
 902     GNUNET_CONTAINER_DLL_remove (barrier->head, barrier->tail, client_ctx);
 903   }
 904 }
 905
 906 /* end of gnunet-service-testbed_barriers.c */