2 This file is part of GNUnet.
3 (C) 2012 Christian Grothoff (and other contributing authors)
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.
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.
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.
22 * @file consensus/consensus_api.c
24 * @author Florian Dold
27 #include "gnunet_protocols.h"
28 #include "gnunet_client_lib.h"
29 #include "gnunet_consensus_service.h"
30 #include "consensus.h"
33 #define LOG(kind,...) GNUNET_log_from (kind, "consensus-api",__VA_ARGS__)
37 * Handle for the service.
39 struct GNUNET_CONSENSUS_Handle
42 * Configuration to use.
44 const struct GNUNET_CONFIGURATION_Handle *cfg;
47 * Socket (if available).
49 struct GNUNET_CLIENT_Connection *client;
52 * Callback for new elements. Not called for elements added locally.
54 GNUNET_CONSENSUS_NewElementCallback new_element_cb;
57 * Closure for new_element_cb
59 void *new_element_cls;
62 * Session identifier for the consensus session.
64 struct GNUNET_HashCode session_id;
67 * Number of peers in the consensus. Optionally includes the local peer.
72 * Peer identities of peers in the consensus. Optionally includes the local peer.
74 struct GNUNET_PeerIdentity *peers;
77 * Currently active transmit request.
79 struct GNUNET_CLIENT_TransmitHandle *th;
82 * GNUNES_YES iff the join message has been sent to the service.
87 * Called when the current insertion operation finishes.
88 * NULL if there is no insert operation active.
90 GNUNET_CONSENSUS_InsertDoneCallback idc;
93 * Closure for the insert done callback.
98 * An element that was requested to be inserted.
100 struct GNUNET_CONSENSUS_Element *insert_element;
103 * Called when the conclude operation finishes or fails.
105 GNUNET_CONSENSUS_ConcludeCallback conclude_cb;
108 * Closure for the conclude callback.
113 * Deadline for the conclude operation.
115 struct GNUNET_TIME_Absolute conclude_deadline;
120 handle_new_element(struct GNUNET_CONSENSUS_Handle *consensus,
121 struct GNUNET_CONSENSUS_ElementMessage *msg)
123 struct GNUNET_CONSENSUS_Element element;
124 element.type = msg->element_type;
125 element.size = msg->header.size - sizeof (struct GNUNET_CONSENSUS_ElementMessage);
126 element.data = &msg[1];
127 consensus->new_element_cb (consensus->new_element_cls, &element);
131 handle_conclude_done(struct GNUNET_CONSENSUS_Handle *consensus,
132 struct GNUNET_CONSENSUS_ConcludeDoneMessage *msg)
134 GNUNET_assert (NULL != consensus->conclude_cb);
135 consensus->conclude_cb(consensus->conclude_cls,
137 (struct GNUNET_PeerIdentity *) &msg[1]);
138 consensus->conclude_cb = NULL;
144 * Type of a function to call when we receive a message
148 * @param msg message received, NULL on timeout or fatal error
151 message_handler (void *cls, const struct GNUNET_MessageHeader *msg)
153 struct GNUNET_CONSENSUS_Handle *consensus = cls;
155 LOG (GNUNET_ERROR_TYPE_INFO, "received message from consensus service\n");
159 /* Error, timeout, death */
160 LOG (GNUNET_ERROR_TYPE_ERROR, "error receiving\n");
161 GNUNET_CLIENT_disconnect (consensus->client);
162 consensus->client = NULL;
163 consensus->new_element_cb (NULL, NULL);
164 if (NULL != consensus->idc)
166 consensus->idc(consensus->idc_cls, GNUNET_NO);
167 consensus->idc = NULL;
168 consensus->idc_cls = NULL;
173 switch (ntohs(msg->type))
175 case GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_RECEIVED_ELEMENT:
176 handle_new_element (consensus, (struct GNUNET_CONSENSUS_ElementMessage *) msg);
178 case GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_CONCLUDE_DONE:
179 handle_conclude_done (consensus, (struct GNUNET_CONSENSUS_ConcludeDoneMessage *) msg);
182 LOG(GNUNET_ERROR_TYPE_WARNING, "did not understand message type sent by service, ignoring");
184 GNUNET_CLIENT_receive (consensus->client, &message_handler, consensus,
185 GNUNET_TIME_UNIT_FOREVER_REL);
192 * Function called to notify a client about the connection
193 * begin ready to queue more data. "buf" will be
194 * NULL and "size" zero if the connection was closed for
195 * writing in the meantime.
198 * @param size number of bytes available in buf
199 * @param buf where the callee should write the message
200 * @return number of bytes written to buf
203 transmit_insert (void *cls, size_t size, void *buf)
205 struct GNUNET_CONSENSUS_ElementMessage *msg;
206 struct GNUNET_CONSENSUS_Handle *consensus;
207 GNUNET_CONSENSUS_InsertDoneCallback idc;
211 GNUNET_assert (NULL != buf);
215 GNUNET_assert (NULL != consensus->insert_element);
217 consensus->th = NULL;
221 msize = sizeof (struct GNUNET_CONSENSUS_ElementMessage) +
222 consensus->insert_element->size;
224 msg->header.type = htons (GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_INSERT);
225 msg->header.size = htons (msize);
227 consensus->insert_element->data,
228 consensus->insert_element->size);
231 idc = consensus->idc;
232 consensus->idc = NULL;
233 idc_cls = consensus->idc_cls;
234 consensus->idc_cls = NULL;
235 idc (idc_cls, GNUNET_YES);
242 * Function called to notify a client about the connection
243 * begin ready to queue more data. "buf" will be
244 * NULL and "size" zero if the connection was closed for
245 * writing in the meantime.
248 * @param size number of bytes available in buf
249 * @param buf where the callee should write the message
250 * @return number of bytes written to buf
253 transmit_join (void *cls, size_t size, void *buf)
255 struct GNUNET_CONSENSUS_JoinMessage *msg;
256 struct GNUNET_CONSENSUS_Handle *consensus;
259 GNUNET_assert (NULL != buf);
261 LOG (GNUNET_ERROR_TYPE_DEBUG, "transmitting join message\n");
264 consensus->th = NULL;
265 consensus->joined = 1;
269 msize = sizeof (struct GNUNET_CONSENSUS_JoinMessage) +
270 consensus->num_peers * sizeof (struct GNUNET_PeerIdentity);
272 msg->header.type = htons (GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_JOIN);
273 msg->header.size = htons (msize);
274 msg->session_id = consensus->session_id;
275 msg->num_peers = htons (consensus->num_peers);
278 consensus->num_peers * sizeof (struct GNUNET_PeerIdentity));
280 if (consensus->insert_element != NULL)
283 GNUNET_CLIENT_notify_transmit_ready (consensus->client,
285 GNUNET_TIME_UNIT_FOREVER_REL,
286 GNUNET_NO, &transmit_insert, consensus);
289 GNUNET_CLIENT_receive (consensus->client, &message_handler, consensus,
290 GNUNET_TIME_UNIT_FOREVER_REL);
297 * Function called to notify a client about the connection
298 * begin ready to queue more data. "buf" will be
299 * NULL and "size" zero if the connection was closed for
300 * writing in the meantime.
303 * @param size number of bytes available in buf
304 * @param buf where the callee should write the message
305 * @return number of bytes written to buf
308 transmit_conclude (void *cls, size_t size, void *buf)
310 struct GNUNET_CONSENSUS_ConcludeMessage *msg;
311 struct GNUNET_CONSENSUS_Handle *consensus;
314 GNUNET_assert (NULL != buf);
317 consensus->th = NULL;
321 msize = sizeof (struct GNUNET_CONSENSUS_ConcludeMessage);
323 msg->header.type = htons (GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_CONCLUDE);
324 msg->header.size = htons (msize);
326 GNUNET_TIME_relative_hton (GNUNET_TIME_absolute_get_remaining(consensus->conclude_deadline));
333 * Function called to notify a client about the connection
334 * begin ready to queue more data. "buf" will be
335 * NULL and "size" zero if the connection was closed for
336 * writing in the meantime.
338 * @param cls the consensus handle
339 * @param size number of bytes available in buf
340 * @param buf where the callee should write the message
341 * @return number of bytes written to buf
344 transmit_begin (void *cls, size_t size, void *buf)
346 struct GNUNET_MessageHeader *msg;
347 struct GNUNET_CONSENSUS_Handle *consensus;
350 GNUNET_assert (NULL != buf);
353 consensus->th = NULL;
357 msize = sizeof (struct GNUNET_MessageHeader);
359 msg->type = htons (GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_BEGIN);
360 msg->size = htons (msize);
367 * Create a consensus session.
371 * @param peers array of peers participating in this consensus session
372 * Inclusion of the local peer is optional.
373 * @param session_id session identifier
374 * Allows a group of peers to have more than consensus session.
375 * @param new_element_cb callback, called when a new element is added to the set by
377 * @param new_element_cls closure for new_element
378 * @return handle to use, NULL on error
380 struct GNUNET_CONSENSUS_Handle *
381 GNUNET_CONSENSUS_create (const struct GNUNET_CONFIGURATION_Handle *cfg,
382 unsigned int num_peers,
383 const struct GNUNET_PeerIdentity *peers,
384 const struct GNUNET_HashCode *session_id,
385 GNUNET_CONSENSUS_NewElementCallback new_element_cb,
386 void *new_element_cls)
388 struct GNUNET_CONSENSUS_Handle *consensus;
389 size_t join_message_size;
391 consensus = GNUNET_malloc (sizeof (struct GNUNET_CONSENSUS_Handle));
392 consensus->cfg = cfg;
393 consensus->new_element_cb = new_element_cb;
394 consensus->new_element_cls = new_element_cls;
395 consensus->num_peers = num_peers;
396 consensus->session_id = *session_id;
400 consensus->peers = NULL;
402 else if (num_peers > 0)
404 consensus->peers = GNUNET_memdup (peers, num_peers * sizeof (struct GNUNET_PeerIdentity));
411 consensus->client = GNUNET_CLIENT_connect ("consensus", cfg);
413 GNUNET_assert (consensus->client != NULL);
415 join_message_size = (sizeof (struct GNUNET_CONSENSUS_JoinMessage)) +
416 (num_peers * sizeof (struct GNUNET_PeerIdentity));
419 GNUNET_CLIENT_notify_transmit_ready (consensus->client,
421 GNUNET_TIME_UNIT_FOREVER_REL,
422 GNUNET_NO, &transmit_join, consensus);
424 GNUNET_assert (consensus->th != NULL);
432 * Insert an element in the set being reconsiled. Must not be called after
433 * "GNUNET_CONSENSUS_conclude".
435 * @param consensus handle for the consensus session
436 * @param element the element to be inserted
437 * @param idc function called when we are done with this element and it
438 * is thus allowed to call GNUNET_CONSENSUS_insert again
439 * @param idc_cls closure for 'idc'
442 GNUNET_CONSENSUS_insert (struct GNUNET_CONSENSUS_Handle *consensus,
443 const struct GNUNET_CONSENSUS_Element *element,
444 GNUNET_CONSENSUS_InsertDoneCallback idc,
448 GNUNET_assert (NULL == consensus->idc);
449 GNUNET_assert (NULL == consensus->insert_element);
451 consensus->idc = idc;
452 consensus->idc_cls = idc_cls;
453 consensus->insert_element = GNUNET_memdup(element, sizeof (struct GNUNET_CONSENSUS_Element) + element->size);
455 if (consensus->joined == 0)
457 GNUNET_assert (NULL != consensus->th);
461 GNUNET_assert (NULL == consensus->th);
464 GNUNET_CLIENT_notify_transmit_ready (consensus->client,
465 element->size + sizeof (struct GNUNET_CONSENSUS_ElementMessage),
466 GNUNET_TIME_UNIT_FOREVER_REL,
467 GNUNET_NO, &transmit_insert, consensus);
472 * Begin reconciling elements with other peers.
474 * @param consensus handle for the consensus session
477 GNUNET_CONSENSUS_begin (struct GNUNET_CONSENSUS_Handle *consensus)
479 GNUNET_assert (NULL == consensus->idc);
480 GNUNET_assert (NULL == consensus->insert_element);
483 GNUNET_CLIENT_notify_transmit_ready (consensus->client,
484 sizeof (struct GNUNET_MessageHeader),
485 GNUNET_TIME_UNIT_FOREVER_REL,
486 GNUNET_NO, &transmit_begin, consensus);
491 * We are finished inserting new elements into the consensus;
492 * try to conclude the consensus within a given time window.
494 * @param consensus consensus session
495 * @param timeout timeout after which the conculde callback
497 * @param conclude called when the conclusion was successful
498 * @param conclude_cls closure for the conclude callback
501 GNUNET_CONSENSUS_conclude (struct GNUNET_CONSENSUS_Handle *consensus,
502 struct GNUNET_TIME_Relative timeout,
503 GNUNET_CONSENSUS_ConcludeCallback conclude,
506 GNUNET_assert (NULL == consensus->th);
507 GNUNET_assert (NULL == consensus->conclude_cb);
509 consensus->conclude_cls = conclude_cls;
510 consensus->conclude_cb = conclude;
511 consensus->conclude_deadline = GNUNET_TIME_relative_to_absolute(timeout);
514 GNUNET_CLIENT_notify_transmit_ready (consensus->client,
515 sizeof (struct GNUNET_CONSENSUS_ConcludeMessage),
517 GNUNET_NO, &transmit_conclude, consensus);
518 if (NULL == consensus->th)
520 conclude(conclude_cls, 0, NULL);
526 * Destroy a consensus handle (free all state associated with
527 * it, no longer call any of the callbacks).
529 * @param consensus handle to destroy
532 GNUNET_CONSENSUS_destroy (struct GNUNET_CONSENSUS_Handle *consensus)
534 if (consensus->client != NULL)
536 GNUNET_CLIENT_disconnect (consensus->client);
537 consensus->client = NULL;
539 GNUNET_free (consensus->peers);
540 GNUNET_free (consensus);