src/consensus/consensus_api.c

   1 /*
   2      This file is part of GNUnet.
   3      Copyright (C) 2012, 2016 GNUnet e.V.
   4
   5      GNUnet is free software: you can redistribute it and/or modify it
   6      under the terms of the GNU Affero General Public License as published
   7      by the Free Software Foundation, either version 3 of the License,
   8      or (at your 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      Affero General Public License for more details.
  14 */
  15
  16 /**
  17  * @file consensus/consensus_api.c
  18  * @brief
  19  * @author Florian Dold
  20  */
  21 #include "platform.h"
  22 #include "gnunet_util_lib.h"
  23 #include "gnunet_protocols.h"
  24 #include "gnunet_client_lib.h"
  25 #include "gnunet_consensus_service.h"
  26 #include "consensus.h"
  27
  28
  29 #define LOG(kind,...) GNUNET_log_from (kind, "consensus-api",__VA_ARGS__)
  30
  31
  32 /**
  33  * Handle for the service.
  34  */
  35 struct GNUNET_CONSENSUS_Handle
  36 {
  37   /**
  38    * Configuration to use.
  39    */
  40   const struct GNUNET_CONFIGURATION_Handle *cfg;
  41
  42   /**
  43    * Callback for new elements. Not called for elements added locally.
  44    */
  45   GNUNET_CONSENSUS_ElementCallback new_element_cb;
  46
  47   /**
  48    * Closure for @e new_element_cb
  49    */
  50   void *new_element_cls;
  51
  52   /**
  53    * The (local) session identifier for the consensus session.
  54    */
  55   struct GNUNET_HashCode session_id;
  56
  57   /**
  58    * #GNUNET_YES iff the join message has been sent to the service.
  59    */
  60   int joined;
  61
  62   /**
  63    * Called when the conclude operation finishes or fails.
  64    */
  65   GNUNET_CONSENSUS_ConcludeCallback conclude_cb;
  66
  67   /**
  68    * Closure for the @e conclude_cb callback.
  69    */
  70   void *conclude_cls;
  71
  72   /**
  73    * Deadline for the conclude operation.
  74    */
  75   struct GNUNET_TIME_Absolute conclude_deadline;
  76
  77   /**
  78    * Message queue for the client.
  79    */
  80   struct GNUNET_MQ_Handle *mq;
  81 };
  82
  83
  84 /**
  85  * FIXME: this should not bee necessary when the API
  86  * issue has been fixed
  87  */
  88 struct InsertDoneInfo
  89 {
  90   GNUNET_CONSENSUS_InsertDoneCallback idc;
  91   void *cls;
  92 };
  93
  94
  95 /**
  96  * Called when the server has sent is a new element
  97  *
  98  * @param cls consensus handle
  99  * @param msg element message
 100  */
 101 static int
 102 check_new_element (void *cls,
 103                    const struct GNUNET_CONSENSUS_ElementMessage *msg)
 104 {
 105   /* any size is fine, elements are variable-size */
 106   return GNUNET_OK;
 107 }
 108
 109
 110 /**
 111  * Called when the server has sent is a new element
 112  *
 113  * @param cls consensus handle
 114  * @param msg element message
 115  */
 116 static void
 117 handle_new_element (void *cls,
 118                     const struct GNUNET_CONSENSUS_ElementMessage *msg)
 119 {
 120   struct GNUNET_CONSENSUS_Handle *consensus = cls;
 121   struct GNUNET_SET_Element element;
 122
 123   LOG (GNUNET_ERROR_TYPE_DEBUG,
 124        "received new element\n");
 125   element.element_type = msg->element_type;
 126   element.size = ntohs (msg->header.size) - sizeof (struct GNUNET_CONSENSUS_ElementMessage);
 127   element.data = &msg[1];
 128   consensus->new_element_cb (consensus->new_element_cls,
 129                              &element);
 130 }
 131
 132
 133 /**
 134  * Called when the server has announced
 135  * that the conclusion is over.
 136  *
 137  * @param cls consensus handle
 138  * @param msg conclude done message
 139  */
 140 static void
 141 handle_conclude_done (void *cls,
 142                       const struct GNUNET_MessageHeader *msg)
 143 {
 144   struct GNUNET_CONSENSUS_Handle *consensus = cls;
 145   GNUNET_CONSENSUS_ConcludeCallback cc;
 146
 147   GNUNET_MQ_destroy (consensus->mq);
 148   consensus->mq = NULL;
 149   GNUNET_assert (NULL != (cc = consensus->conclude_cb));
 150   consensus->conclude_cb = NULL;
 151   cc (consensus->conclude_cls);
 152 }
 153
 154
 155 /**
 156  * Generic error handler, called with the appropriate
 157  * error code and the same closure specified at the creation of
 158  * the message queue.
 159  * Not every message queue implementation supports an error handler.
 160  *
 161  * @param cls closure, same closure as for the message handlers
 162  * @param error error code
 163  */
 164 static void
 165 mq_error_handler (void *cls,
 166                   enum GNUNET_MQ_Error error)
 167 {
 168   LOG (GNUNET_ERROR_TYPE_WARNING,
 169        "consensus service disconnected us\n");
 170 }
 171
 172
 173 /**
 174  * Create a consensus session.
 175  *
 176  * @param cfg configuration to use for connecting to the consensus service
 177  * @param num_peers number of peers in the peers array
 178  * @param peers array of peers participating in this consensus session
 179  *              Inclusion of the local peer is optional.
 180  * @param session_id session identifier
 181  *                   Allows a group of peers to have more than consensus session.
 182  * @param start start time of the consensus, conclude should be called before
 183  *              the start time.
 184  * @param deadline time when the consensus should have concluded
 185  * @param new_element_cb callback, called when a new element is added to the set by
 186  *                    another peer
 187  * @param new_element_cls closure for new_element
 188  * @return handle to use, NULL on error
 189  */
 190 struct GNUNET_CONSENSUS_Handle *
 191 GNUNET_CONSENSUS_create (const struct GNUNET_CONFIGURATION_Handle *cfg,
 192                          unsigned int num_peers,
 193                          const struct GNUNET_PeerIdentity *peers,
 194                          const struct GNUNET_HashCode *session_id,
 195                          struct GNUNET_TIME_Absolute start,
 196                          struct GNUNET_TIME_Absolute deadline,
 197                          GNUNET_CONSENSUS_ElementCallback new_element_cb,
 198                          void *new_element_cls)
 199 {
 200   struct GNUNET_CONSENSUS_Handle *consensus
 201     = GNUNET_new (struct GNUNET_CONSENSUS_Handle);
 202   struct GNUNET_MQ_MessageHandler mq_handlers[] = {
 203     GNUNET_MQ_hd_var_size (new_element,
 204                            GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_RECEIVED_ELEMENT,
 205                            struct GNUNET_CONSENSUS_ElementMessage,
 206                            consensus),
 207     GNUNET_MQ_hd_fixed_size (conclude_done,
 208                              GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_CONCLUDE_DONE,
 209                              struct GNUNET_MessageHeader,
 210                              consensus),
 211     GNUNET_MQ_handler_end ()
 212   };
 213   struct GNUNET_CONSENSUS_JoinMessage *join_msg;
 214   struct GNUNET_MQ_Envelope *ev;
 215
 216   consensus->cfg = cfg;
 217   consensus->new_element_cb = new_element_cb;
 218   consensus->new_element_cls = new_element_cls;
 219   consensus->session_id = *session_id;
 220   consensus->mq = GNUNET_CLIENT_connect (cfg,
 221                                          "consensus",
 222                                          mq_handlers,
 223                                          &mq_error_handler,
 224                                          consensus);
 225   if (NULL == consensus->mq)
 226   {
 227     GNUNET_free (consensus);
 228     return NULL;
 229   }
 230   ev = GNUNET_MQ_msg_extra (join_msg,
 231                             (num_peers * sizeof (struct GNUNET_PeerIdentity)),
 232                             GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_JOIN);
 233
 234   join_msg->session_id = consensus->session_id;
 235   join_msg->start = GNUNET_TIME_absolute_hton (start);
 236   join_msg->deadline = GNUNET_TIME_absolute_hton (deadline);
 237   join_msg->num_peers = htonl (num_peers);
 238   GNUNET_memcpy(&join_msg[1],
 239          peers,
 240          num_peers * sizeof (struct GNUNET_PeerIdentity));
 241
 242   GNUNET_MQ_send (consensus->mq, ev);
 243   return consensus;
 244 }
 245
 246
 247 static void
 248 idc_adapter (void *cls)
 249 {
 250   struct InsertDoneInfo *i = cls;
 251   i->idc (i->cls, GNUNET_OK);
 252   GNUNET_free (i);
 253 }
 254
 255 /**
 256  * Insert an element in the set being reconsiled.  Must not be called after
 257  * "GNUNET_CONSENSUS_conclude".
 258  *
 259  * @param consensus handle for the consensus session
 260  * @param element the element to be inserted
 261  * @param idc function called when we are done with this element and it
 262  *            is thus allowed to call GNUNET_CONSENSUS_insert again
 263  * @param idc_cls closure for 'idc'
 264  */
 265 void
 266 GNUNET_CONSENSUS_insert (struct GNUNET_CONSENSUS_Handle *consensus,
 267                          const struct GNUNET_SET_Element *element,
 268                          GNUNET_CONSENSUS_InsertDoneCallback idc,
 269                          void *idc_cls)
 270 {
 271   struct GNUNET_CONSENSUS_ElementMessage *element_msg;
 272   struct GNUNET_MQ_Envelope *ev;
 273   struct InsertDoneInfo *i;
 274
 275   LOG (GNUNET_ERROR_TYPE_DEBUG, "inserting, size=%llu\n", element->size);
 276
 277   ev = GNUNET_MQ_msg_extra (element_msg, element->size,
 278                             GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_INSERT);
 279
 280   GNUNET_memcpy (&element_msg[1], element->data, element->size);
 281
 282   if (NULL != idc)
 283   {
 284     i = GNUNET_new (struct InsertDoneInfo);
 285     i->idc = idc;
 286     i->cls = idc_cls;
 287     GNUNET_MQ_notify_sent (ev, idc_adapter, i);
 288   }
 289   GNUNET_MQ_send (consensus->mq, ev);
 290 }
 291
 292
 293 /**
 294  * We are done with inserting new elements into the consensus;
 295  * try to conclude the consensus within a given time window.
 296  * After conclude has been called, no further elements may be
 297  * inserted by the client.
 298  *
 299  * @param consensus consensus session
 300  * @param deadline deadline after which the conculde callback
 301  *                must be called
 302  * @param conclude called when the conclusion was successful
 303  * @param conclude_cls closure for the conclude callback
 304  */
 305 void
 306 GNUNET_CONSENSUS_conclude (struct GNUNET_CONSENSUS_Handle *consensus,
 307                            GNUNET_CONSENSUS_ConcludeCallback conclude,
 308                            void *conclude_cls)
 309 {
 310   struct GNUNET_MQ_Envelope *ev;
 311
 312   GNUNET_assert (NULL != conclude);
 313   GNUNET_assert (NULL == consensus->conclude_cb);
 314
 315   consensus->conclude_cls = conclude_cls;
 316   consensus->conclude_cb = conclude;
 317
 318   ev = GNUNET_MQ_msg_header (GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_CONCLUDE);
 319   GNUNET_MQ_send (consensus->mq, ev);
 320 }
 321
 322
 323 /**
 324  * Destroy a consensus handle (free all state associated with
 325  * it, no longer call any of the callbacks).
 326  *
 327  * @param consensus handle to destroy
 328  */
 329 void
 330 GNUNET_CONSENSUS_destroy (struct GNUNET_CONSENSUS_Handle *consensus)
 331 {
 332   if (NULL != consensus->mq)
 333   {
 334     GNUNET_MQ_destroy (consensus->mq);
 335     consensus->mq = NULL;
 336   }
 337   GNUNET_free (consensus);
 338 }
 339
 340 /* end of consensus_api.c */