src/consensus/consensus_api.c

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