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
  29 /**
  30  * timeout for outgoing message transmissions in seconds
  31  */
  32 #define MESSAGE_SEND_TIMEOUT(s) \
  33   GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, s)
  34
  35
  36 /**
  37  * Barrier
  38  */
  39 struct Barrier;
  40
  41
  42 /**
  43  * Message queue for transmitting messages
  44  */
  45 struct MessageQueue
  46 {
  47   /**
  48    * next pointer for DLL
  49    */
  50   struct MessageQueue *next;
  51
  52   /**
  53    * prev pointer for DLL
  54    */
  55   struct MessageQueue *prev;
  56
  57   /**
  58    * The message to be sent
  59    */
  60   struct GNUNET_MessageHeader *msg;
  61 };
  62
  63 /**
  64  * Context to be associated with each client
  65  */
  66 struct ClientCtx
  67 {
  68   /**
  69    * The barrier this client is waiting for
  70    */
  71   struct Barrier *barrier;
  72
  73   /**
  74    * DLL next ptr
  75    */
  76   struct ClientCtx *next;
  77
  78   /**
  79    * DLL prev ptr
  80    */
  81   struct ClientCtx *prev;
  82
  83   /**
  84    * The client handle
  85    */
  86   struct GNUNET_SERVER_Client *client;
  87
  88   /**
  89    * the transmission handle
  90    */
  91   struct GNUNET_SERVER_TransmitHandle *tx;
  92
  93   /**
  94    * message queue head
  95    */
  96   struct MessageQueue *mq_head;
  97
  98   /**
  99    * message queue tail
 100    */
 101   struct MessageQueue *mq_tail;
 102 };
 103
 104
 105 /**
 106  * Barrier
 107  */
 108 struct Barrier
 109 {
 110   /**
 111    * The hashcode of the barrier name
 112    */
 113   struct GNUNET_HashCode hash;
 114
 115   /**
 116    * The name of the barrier
 117    */
 118   char *name;
 119
 120   /**
 121    * DLL head for the list of clients waiting for this barrier
 122    */
 123   struct ClientCtx *head;
 124
 125   /**
 126    * DLL tail for the list of clients waiting for this barrier
 127    */
 128   struct ClientCtx *tail;
 129
 130   /**
 131    * Number of peers which have reached this barrier
 132    */
 133   unsigned int nreached;
 134
 135   /**
 136    * Number of slaves we have initialised this barrier
 137    */
 138   unsigned int nslaves;
 139
 140   /**
 141    * Quorum percentage to be reached
 142    */
 143   uint8_t quorum;
 144
 145   /**
 146    * Was there a timeout while propagating initialisation
 147    */
 148   uint8_t timedout;
 149 };
 150
 151
 152 /**
 153  * Hashtable handle for storing initialised barriers
 154  */
 155 static struct GNUNET_CONTAINER_MultiHashMap *barrier_map;
 156
 157 /**
 158  * Service context
 159  */
 160 static struct GNUNET_SERVICE_Context *ctx;
 161
 162
 163 /**
 164  * Function called to notify a client about the connection
 165  * begin ready to queue more data.  "buf" will be
 166  * NULL and "size" zero if the connection was closed for
 167  * writing in the meantime.
 168  *
 169  * @param cls client context
 170  * @param size number of bytes available in buf
 171  * @param buf where the callee should write the message
 172  * @return number of bytes written to buf
 173  */
 174 static size_t
 175 transmit_ready_cb (void *cls, size_t size, void *buf)
 176 {
 177   struct ClientCtx *ctx = cls;
 178   struct GNUNET_SERVER_Client *client = ctx->client;
 179   struct MessageQueue *mq;
 180   struct GNUNET_MessageHeader *msg;
 181   size_t wrote;
 182
 183   ctx->tx = NULL;
 184   wrote = 0;
 185   if ((0 == size) || (NULL == buf))
 186   {
 187     GNUNET_assert (NULL != ctx->client);
 188     GNUNET_SERVER_client_drop (ctx->client);
 189     ctx->client = NULL;
 190     return 0;
 191   }
 192   mq = ctx->mq_head;
 193   msg = mq->msg;
 194   wrote = ntohs (msg->size);
 195   GNUNET_assert (size >= wrote);
 196   (void) memcpy (buf, msg, wrote);
 197   GNUNET_CONTAINER_DLL_remove (ctx->mq_head, ctx->mq_tail, mq);
 198   GNUNET_free (mq->msg);
 199   GNUNET_free (mq);
 200   if (NULL != (mq = ctx->mq_head))
 201     ctx->tx = GNUNET_SERVER_notify_transmit_ready (client, ntohs (msg->size),
 202                                                   MESSAGE_SEND_TIMEOUT (30),
 203                                                   &transmit_ready_cb, ctx);
 204   return wrote;
 205 }
 206
 207
 208 /**
 209  * Queue a message into a clients message queue
 210  *
 211  * @param ctx the context associated with the client
 212  * @param msg the message to queue.  Will be consumed
 213  */
 214 static void
 215 queue_message (struct ClientCtx *ctx, struct GNUNET_MessageHeader *msg)
 216 {
 217   struct MessageQueue *mq;
 218   struct GNUNET_SERVER_Client *client = ctx->client;
 219
 220   mq = GNUNET_malloc (sizeof (struct MessageQueue));
 221   mq->msg = msg;
 222   GNUNET_CONTAINER_DLL_insert_tail (ctx->mq_head, ctx->mq_tail, mq);
 223   if (NULL == ctx->tx)
 224    ctx->tx = GNUNET_SERVER_notify_transmit_ready (client, ntohs (msg->size),
 225                                                   MESSAGE_SEND_TIMEOUT (30),
 226                                                   &transmit_ready_cb, ctx);
 227 }
 228
 229
 230 #if 0
 231 /**
 232  * Function to remove a barrier from the barrier map and cleanup resources
 233  * occupied by a barrier
 234  *
 235  * @param barrier the barrier handle
 236  */
 237 static void
 238 remove_barrier (struct Barrier *barrier)
 239 {
 240   GNUNET_assert (GNUNET_YES == GNUNET_CONTAINER_multihashmap_remove (barrier_map,
 241                                                                     &barrier->hash,
 242                                                                     barrier));
 243   GNUNET_free (barrier->name);
 244   GNUNET_free (barrier);
 245 }
 246
 247
 248 /**
 249  * Function called upon timeout while waiting for a response from the
 250  * subcontrollers to barrier init message
 251  *
 252  * @param
 253  * @return
 254  */
 255 static void
 256 fwd_tout_barrier_init (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 257 {
 258   struct ForwardedOperationContext *foctx = cls;
 259   struct Barrier *barrier = foctx->cls;
 260
 261   barrier->nslaves--;
 262   barrier->timedout = GNUNET_YES;
 263   if (0 == barrier->nslaves)
 264   {
 265     GST_send_operation_fail_msg (foctx->client, foctx->operation_id,
 266                                  "Timeout while contacting a slave controller");
 267     remove_barrier (barrier);
 268   }
 269 }
 270 #endif
 271
 272 /**
 273  * Task for sending barrier crossed notifications to waiting client
 274  *
 275  * @param cls the barrier which is crossed
 276  * @param tc scheduler task context
 277  */
 278 static void
 279 notify_task_cb (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 280 {
 281   struct Barrier *barrier = cls;
 282   struct ClientCtx *client_ctx;
 283   struct GNUNET_TESTBED_BarrierStatus *msg;
 284   struct GNUNET_MessageHeader *dup_msg;
 285   uint16_t name_len;
 286   uint16_t msize;
 287
 288   name_len = strlen (barrier->name) + 1;
 289   msize = sizeof (struct GNUNET_TESTBED_BarrierStatus) + name_len;
 290   msg = GNUNET_malloc (msize);
 291   msg->header.size = htons (msize);
 292   msg->header.type = htons (GNUNET_MESSAGE_TYPE_TESTBED_BARRIER_STATUS);
 293   msg->status = 0;
 294   msg->name_len = htons (name_len);
 295   (void) memcpy (msg->data, barrier->name, name_len);
 296   msg->data[name_len] = '\0';
 297   while (NULL != (client_ctx = barrier->head))
 298   {
 299     dup_msg = GNUNET_copy_message (&msg->header);
 300     queue_message (client_ctx, dup_msg);
 301     GNUNET_CONTAINER_DLL_remove (barrier->head, barrier->tail, client_ctx);
 302     GNUNET_SERVER_client_set_user_context_ (client_ctx->client, NULL, 0);
 303     GNUNET_free (client_ctx);
 304   }
 305 }
 306
 307
 308 /**
 309  * Message handler for GNUNET_MESSAGE_TYPE_TESTBED_BARRIER_WAIT messages.  This
 310  * message should come from peers or a shared helper service using the
 311  * testbed-barrier client API (@see gnunet_testbed_barrier_service.h)
 312  *
 313  * This handler is queued in the main service and will handle the messages sent
 314  * either from the testbed driver or from a high level controller
 315  *
 316  * @param cls NULL
 317  * @param client identification of the client
 318  * @param message the actual message
 319  */
 320 static void
 321 handle_barrier_wait (void *cls, struct GNUNET_SERVER_Client *client,
 322                      const struct GNUNET_MessageHeader *message)
 323 {
 324   const struct GNUNET_TESTBED_BarrierWait *msg;
 325   struct Barrier *barrier;
 326   char *name;
 327   struct ClientCtx *client_ctx;
 328   struct GNUNET_HashCode key;
 329   size_t name_len;
 330   uint16_t msize;
 331
 332   msize = ntohs (message->size);
 333   if (msize <= sizeof (struct GNUNET_TESTBED_BarrierWait))
 334   {
 335     GNUNET_break_op (0);
 336     GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
 337     return;
 338   }
 339   if (NULL == barrier_map)
 340   {
 341     GNUNET_break (0);
 342     GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
 343     return;
 344   }
 345   msg = (const struct GNUNET_TESTBED_BarrierWait *) message;
 346   name_len = msize - sizeof (struct GNUNET_TESTBED_BarrierWait);
 347   name = GNUNET_malloc (name_len + 1);
 348   name[name_len] = '\0';
 349   (void) memcpy (name, msg->name, name_len);
 350   GNUNET_CRYPTO_hash (name, name_len - 1, &key);
 351   if (NULL == (barrier = GNUNET_CONTAINER_multihashmap_get (barrier_map, &key)))
 352   {
 353     GNUNET_break (0);
 354     GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
 355     GNUNET_free (name);
 356     return;
 357   }
 358   client_ctx = GNUNET_SERVER_client_get_user_context (client, struct ClientCtx);
 359   if (NULL == client_ctx)
 360   {
 361     client_ctx = GNUNET_malloc (sizeof (struct ClientCtx));
 362     client_ctx->client = client;
 363     GNUNET_SERVER_client_keep (client);
 364     client_ctx->barrier = barrier;
 365     GNUNET_CONTAINER_DLL_insert_tail (barrier->head, barrier->tail, client_ctx);
 366     barrier->nreached++;
 367     if ((barrier->quorum * GST_num_local_peers) <= (barrier->nreached * 100))
 368       notify_task_cb (barrier, NULL);
 369   }
 370   GNUNET_SERVER_receive_done (client, GNUNET_OK);
 371 }
 372
 373
 374 /**
 375  * Functions with this signature are called whenever a client
 376  * is disconnected on the network level.
 377  *
 378  * @param cls closure
 379  * @param client identification of the client; NULL
 380  *        for the last call when the server is destroyed
 381  */
 382 static void
 383 disconnect_cb (void *cls, struct GNUNET_SERVER_Client *client)
 384 {
 385   struct ClientCtx *client_ctx;
 386   struct Barrier *barrier;
 387
 388   client_ctx = GNUNET_SERVER_client_get_user_context (client, struct ClientCtx);
 389   if (NULL == client_ctx)
 390     return;
 391   barrier = client_ctx->barrier;
 392   GNUNET_CONTAINER_DLL_remove (barrier->head, barrier->tail, client_ctx);
 393   if (NULL != client_ctx->tx)
 394     GNUNET_SERVER_notify_transmit_ready_cancel (client_ctx->tx);
 395
 396 }
 397
 398
 399 /**
 400  * Function to initialise barrriers component
 401  */
 402 void
 403 GST_barriers_init (struct GNUNET_CONFIGURATION_Handle *cfg)
 404 {
 405   static const struct GNUNET_SERVER_MessageHandler message_handlers[] = {
 406     {&handle_barrier_wait, NULL, GNUNET_MESSAGE_TYPE_TESTBED_BARRIER_WAIT, 0},
 407     {NULL, NULL, 0, 0}
 408   };
 409   struct GNUNET_SERVER_Handle *srv;
 410
 411   barrier_map = GNUNET_CONTAINER_multihashmap_create (3, GNUNET_YES);
 412   ctx = GNUNET_SERVICE_start ("testbed-barrier", cfg,
 413                               GNUNET_SERVICE_OPTION_MANUAL_SHUTDOWN);
 414   srv = GNUNET_SERVICE_get_server (ctx);
 415   GNUNET_SERVER_add_handlers (srv, message_handlers);
 416   GNUNET_SERVER_disconnect_notify (srv, &disconnect_cb, NULL);
 417 }
 418
 419
 420 /**
 421  * Function to stop the barrier service
 422  */
 423 void
 424 GST_barriers_stop ()
 425 {
 426   GNUNET_assert (NULL != ctx);
 427   GNUNET_SERVICE_stop (ctx);
 428 }
 429
 430
 431 /**
 432  * Message handler for GNUNET_MESSAGE_TYPE_TESTBED_BARRIER_INIT messages.  This
 433  * message should always come from a parent controller or the testbed API if we
 434  * are the root controller.
 435  *
 436  * This handler is queued in the main service and will handle the messages sent
 437  * either from the testbed driver or from a high level controller
 438  *
 439  * @param cls NULL
 440  * @param client identification of the client
 441  * @param message the actual message
 442  */
 443 void
 444 GST_handle_barrier_init (void *cls, struct GNUNET_SERVER_Client *client,
 445                          const struct GNUNET_MessageHeader *message)
 446 {
 447   const struct GNUNET_TESTBED_BarrierInit *msg;
 448   const char *name;
 449   struct Barrier *barrier;
 450   struct Slave *slave;
 451   struct GNUNET_HashCode hash;
 452   size_t name_len;
 453   uint64_t op_id;
 454   unsigned int cnt;
 455   uint16_t msize;
 456
 457   if (NULL == GST_context)
 458   {
 459     GNUNET_break_op (0);
 460     GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
 461     return;
 462   }
 463   if (client != GST_context->client)
 464   {
 465     GNUNET_break_op (0);
 466     GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
 467     return;
 468   }
 469   msize = ntohs (message->size);
 470   if (msize <= sizeof (struct GNUNET_TESTBED_BarrierInit))
 471   {
 472     GNUNET_break_op (0);
 473     GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
 474     return;
 475   }
 476   msg = (const struct GNUNET_TESTBED_BarrierInit *) message;
 477   op_id = GNUNET_ntohll (msg->op_id);
 478   name = msg->name;
 479   name_len = (size_t) msize - sizeof (struct GNUNET_TESTBED_BarrierInit);
 480   GNUNET_CRYPTO_hash (name, name_len, &hash);
 481   if (GNUNET_YES == GNUNET_CONTAINER_multihashmap_contains (barrier_map, &hash))
 482   {
 483     GST_send_operation_fail_msg (client, op_id, "Barrier already initialised");
 484     GNUNET_SERVER_receive_done (client, GNUNET_OK);
 485     return;
 486   }
 487   barrier = GNUNET_malloc (sizeof (struct Barrier));
 488   (void) memcpy (&barrier->hash, &hash, sizeof (struct GNUNET_HashCode));
 489   barrier->quorum = msg->quorum;
 490   barrier->name = GNUNET_malloc (name_len + 1);
 491   barrier->name[name_len] = '\0';
 492   (void) memcpy (barrier->name, name, name_len);
 493   GNUNET_assert (GNUNET_OK ==
 494                  GNUNET_CONTAINER_multihashmap_put (barrier_map,
 495                                                     &barrier->hash,
 496                                                     barrier,
 497                                                     GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST));
 498   GNUNET_SERVER_receive_done (client, GNUNET_OK);
 499   /* Propagate barrier init to subcontrollers */
 500   for (cnt = 0; cnt < GST_slave_list_size; cnt++)
 501   {
 502     if (NULL == (slave = GST_slave_list[cnt]))
 503       continue;
 504     if (NULL == slave->controller)
 505     {
 506       GNUNET_break (0);/* May happen when we are connecting to the controller */
 507       continue;
 508     }
 509     GNUNET_break (0);           /* FIXME */
 510   }
 511 }