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 2, 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.
23 #include "gnunet_common.h"
24 #include "gnunet_protocols.h"
25 #include "gnunet_util_lib.h"
26 #include "gnunet_consensus_service.h"
27 #include "gnunet_core_service.h"
28 #include "gnunet_mesh_service.h"
29 #include "consensus.h"
32 struct ConsensusSession;
35 send_next (struct ConsensusSession *session);
39 * An element that is waiting to be transmitted to a client.
44 * Pending elements are kept in a DLL.
46 struct PendingElement *next;
49 * Pending elements are kept in a DLL.
51 struct PendingElement *prev;
56 struct GNUNET_CONSENSUS_Element *element;
61 * A consensus session consists of one local client and the remote authorities.
63 struct ConsensusSession
66 * Consensus sessions are kept in a DLL.
68 struct ConsensusSession *next;
71 * Consensus sessions are kept in a DLL.
73 struct ConsensusSession *prev;
76 * Local consensus identification, chosen by clients.
78 struct GNUNET_HashCode *local_id;
81 * Global consensus identification, computed
82 * from the local id and participating authorities.
84 struct GNUNET_HashCode *global_id;
87 * Corresponding server handle.
89 struct GNUNET_SERVER_Client *client;
92 * Client wants to receive and send updates.
97 * Values in the consensus set of this session,
98 * all of them either have been sent or approved by the client.
100 struct GNUNET_CONTAINER_MultiHashMap *values;
103 * Elements that have not been sent to the client yet.
105 struct PendingElement *transmit_pending_head;
108 * Elements that have not been sent to the client yet.
110 struct PendingElement *transmit_pending_tail;
113 * Elements that have not been sent to the client yet.
115 struct PendingElement *approval_pending_head;
118 * Elements that have not been sent to the client yet.
120 struct PendingElement *approval_pending_tail;
123 * Currently active transmit handle for sending to the client
125 struct GNUNET_SERVER_TransmitHandle *th;
128 * Once conclude_requested is GNUNET_YES, the client may not
129 * insert any more values.
131 int conclude_requested;
134 * Client has been informed about the conclusion.
139 * Number of other peers in the consensus
146 * Linked list of sesstions this peer participates in.
148 static struct ConsensusSession *sessions_head;
151 * Linked list of sesstions this peer participates in.
153 static struct ConsensusSession *sessions_tail;
156 * Configuration of the consensus service.
158 static const struct GNUNET_CONFIGURATION_Handle *cfg;
161 * Handle to the server for this service.
163 static struct GNUNET_SERVER_Handle *srv;
166 * Peer that runs this service
168 static struct GNUNET_PeerIdentity *my_peer;
171 disconnect_client (struct GNUNET_SERVER_Client *client)
177 compute_global_id (struct GNUNET_HashCode *dst,
178 const struct GNUNET_HashCode *local_id,
179 const struct GNUNET_PeerIdentity *peers,
183 struct GNUNET_HashCode tmp;
186 for (i = 0; i < num_peers; ++i)
188 /* FIXME: maybe hash_xor/hash allow aliased source/target, and we can get by without tmp */
189 GNUNET_CRYPTO_hash_xor (dst, &peers[0].hashPubKey, &tmp);
191 GNUNET_CRYPTO_hash (dst, sizeof (struct GNUNET_PeerIdentity), &tmp);
198 transmit_pending (void *cls, size_t size, void *buf)
200 struct GNUNET_CONSENSUS_Element *element;
201 struct GNUNET_CONSENSUS_ElementMessage *msg;
202 struct ConsensusSession *session;
204 session = (struct ConsensusSession *) cls;
205 msg = (struct GNUNET_CONSENSUS_ElementMessage *) buf;
206 element = session->transmit_pending_head->element;
208 GNUNET_assert (NULL != element);
212 msg->element_type = element->type;
213 msg->header.type = htons (GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_RECEIVED_ELEMENT);
214 msg->header.size = htons (sizeof (struct GNUNET_CONSENSUS_ElementMessage) + element->size);
215 memcpy (&msg[1], element->data, element->size);
217 session->transmit_pending_head = session->transmit_pending_head->next;
221 return sizeof (struct GNUNET_CONSENSUS_ElementMessage) + element->size;
226 transmit_conclude_done (void *cls, size_t size, void *buf)
228 struct GNUNET_CONSENSUS_ConcludeDoneMessage *msg;
230 msg = (struct GNUNET_CONSENSUS_ConcludeDoneMessage *) buf;
231 msg->header.type = htons (GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_CONCLUDE_DONE);
232 msg->header.size = htons (sizeof (struct GNUNET_CONSENSUS_ConcludeDoneMessage));
233 msg->num_peers = htons (0);
235 return sizeof (struct GNUNET_CONSENSUS_ConcludeDoneMessage);
240 * Schedule sending the next message (if there is any) to a client.
242 * @param cli the client to send the next message to
245 send_next (struct ConsensusSession *session)
249 GNUNET_assert (NULL != session);
251 if (NULL != session->th)
256 if ((session->conclude_requested == GNUNET_YES) && (session->conclude_sent == GNUNET_NO))
258 /* just the conclude message with no other authorities in the dummy */
259 msize = sizeof (struct GNUNET_CONSENSUS_ConcludeMessage);
261 GNUNET_SERVER_notify_transmit_ready (session->client, msize,
262 GNUNET_TIME_UNIT_FOREVER_REL, &transmit_conclude_done, session);
263 session->conclude_sent = GNUNET_YES;
265 else if (NULL != session->transmit_pending_head)
267 msize = session->transmit_pending_head->element->size + sizeof (struct GNUNET_CONSENSUS_ElementMessage);
269 GNUNET_SERVER_notify_transmit_ready (session->client, msize,
270 GNUNET_TIME_UNIT_FOREVER_REL, &transmit_pending, session);
271 /* TODO: insert into ack pending */
277 * Called when a client wants to join a consensus session.
280 * @param client client that sent the message
281 * @param m message sent by the client
284 client_join (void *cls,
285 struct GNUNET_SERVER_Client *client,
286 const struct GNUNET_MessageHeader *m)
288 struct GNUNET_HashCode global_id;
289 const struct GNUNET_CONSENSUS_JoinMessage *msg;
290 struct ConsensusSession *session;
292 GNUNET_log (GNUNET_ERROR_TYPE_INFO, "client joining\n");
294 msg = (struct GNUNET_CONSENSUS_JoinMessage *) m;
296 compute_global_id (&global_id, &msg->session_id, (struct GNUNET_PeerIdentity *) &m[1], msg->num_peers);
298 session = sessions_head;
299 while (NULL != session)
301 if (client == session->client)
303 GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "client already in session\n");
304 disconnect_client (client);
307 if (0 == memcmp (session->global_id, &global_id, sizeof (struct GNUNET_HashCode)))
309 GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "session already owned by another client\n");
310 disconnect_client (client);
315 GNUNET_SERVER_client_keep (client);
317 /* session does not exist yet, create it */
318 session = GNUNET_malloc (sizeof (struct ConsensusSession));
319 session->local_id = GNUNET_memdup (&msg->session_id, sizeof (struct GNUNET_HashCode));
320 session->global_id = GNUNET_memdup (&global_id, sizeof (struct GNUNET_HashCode));
321 session->values = GNUNET_CONTAINER_multihashmap_create (4, GNUNET_NO);
322 session->client = client;
324 GNUNET_CONTAINER_DLL_insert (sessions_head, sessions_tail, session);
326 GNUNET_log (GNUNET_ERROR_TYPE_INFO, "created new session\n");
328 GNUNET_SERVER_receive_done (client, GNUNET_OK);
333 * Called when a client performs an insert operation.
336 client_insert (void *cls,
337 struct GNUNET_SERVER_Client *client,
338 const struct GNUNET_MessageHeader *m)
340 struct ConsensusSession *session;
341 struct GNUNET_CONSENSUS_ElementMessage *msg;
342 struct GNUNET_CONSENSUS_Element *element;
343 struct GNUNET_HashCode key;
346 GNUNET_log(GNUNET_ERROR_TYPE_INFO, "insert\n");
348 session = sessions_head;
349 while (NULL != session)
351 if (session->client == client)
357 GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "client tried to insert, but client is not in any session\n");
358 GNUNET_SERVER_client_disconnect (client);
362 msg = (struct GNUNET_CONSENSUS_ElementMessage *) m;
363 element_size = msg->header.size - sizeof (struct GNUNET_CONSENSUS_ElementMessage);
365 element = GNUNET_malloc (sizeof (struct GNUNET_CONSENSUS_Element) + element_size);
367 element->type = msg->element_type;
368 element->size = element_size;
369 memcpy (&element[1], &msg[1], element_size);
370 element->data = &element[1];
372 GNUNET_CRYPTO_hash (element, element_size, &key);
374 GNUNET_CONTAINER_multihashmap_put (session->values, &key, element,
375 GNUNET_CONTAINER_MULTIHASHMAPOPTION_REPLACE);
377 GNUNET_SERVER_receive_done (client, GNUNET_OK);
384 * Called when a client wants to begin
387 client_begin (void *cls,
388 struct GNUNET_SERVER_Client *client,
389 const struct GNUNET_MessageHeader *message)
391 struct ConsensusSession *session;
393 GNUNET_log (GNUNET_ERROR_TYPE_INFO, "client requested begin\n");
395 session = sessions_head;
396 while (NULL != session)
398 if (session->client == client)
404 GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "client tried to 'begin', but client is not in any session\n");
405 GNUNET_SERVER_client_disconnect (client);
409 session->begin = GNUNET_YES;
413 GNUNET_SERVER_receive_done (client, GNUNET_OK);
419 * Called when a client performs the conclude operation.
422 client_conclude (void *cls,
423 struct GNUNET_SERVER_Client *client,
424 const struct GNUNET_MessageHeader *message)
426 struct ConsensusSession *session;
428 session = sessions_head;
429 while ((session != NULL) && (session->client != client))
431 session = session->next;
435 GNUNET_SERVER_client_disconnect (client);
438 session->conclude_requested = GNUNET_YES;
440 GNUNET_SERVER_receive_done (client, GNUNET_OK);
445 * Called when a client sends an ack
448 client_ack (void *cls,
449 struct GNUNET_SERVER_Client *client,
450 const struct GNUNET_MessageHeader *message)
452 GNUNET_log (GNUNET_ERROR_TYPE_INFO, "client ack received\n");
456 * Task that disconnects from core.
458 * @param cls core handle
459 * @param tc context information (why was this task triggered now)
462 disconnect_core (void *cls,
463 const struct GNUNET_SCHEDULER_TaskContext *tc)
465 struct GNUNET_CORE_Handle *core;
466 core = (struct GNUNET_CORE_Handle *) cls;
467 GNUNET_CORE_disconnect (core);
469 GNUNET_log (GNUNET_ERROR_TYPE_INFO, "disconnected from core\n");
474 core_startup (void *cls,
475 struct GNUNET_CORE_Handle *core,
476 const struct GNUNET_PeerIdentity *peer)
478 static const struct GNUNET_SERVER_MessageHandler handlers[] = {
479 {&client_join, NULL, GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_JOIN, 0},
480 {&client_insert, NULL, GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_INSERT, 0},
481 {&client_begin, NULL, GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_BEGIN,
482 sizeof (struct GNUNET_MessageHeader)},
483 {&client_conclude, NULL, GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_CONCLUDE,
484 sizeof (struct GNUNET_CONSENSUS_ConcludeMessage)},
485 {&client_ack, NULL, GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_ACK,
486 sizeof (struct GNUNET_CONSENSUS_AckMessage)},
490 GNUNET_SERVER_add_handlers (srv, handlers);
491 my_peer = GNUNET_memdup(peer, sizeof (struct GNUNET_PeerIdentity));
492 GNUNET_SCHEDULER_add_now (&disconnect_core, core);
493 GNUNET_log(GNUNET_ERROR_TYPE_INFO, "connected to core\n");
498 * Process consensus requests.
501 * @param server the initialized server
502 * @param c configuration to use
505 run (void *cls, struct GNUNET_SERVER_Handle *server, const struct GNUNET_CONFIGURATION_Handle *c)
507 struct GNUNET_CORE_Handle *my_core;
508 static const struct GNUNET_CORE_MessageHandler handlers[] = {
512 GNUNET_log(GNUNET_ERROR_TYPE_INFO, "consensus running\n");
516 my_core = GNUNET_CORE_connect (c, NULL, &core_startup, NULL, NULL, NULL, GNUNET_NO, NULL, GNUNET_NO, handlers);
517 GNUNET_assert (NULL != my_core);
522 * The main function for the statistics service.
524 * @param argc number of arguments from the command line
525 * @param argv command line arguments
526 * @return 0 ok, 1 on error
529 main (int argc, char *const *argv)
532 GNUNET_SERVICE_run (argc, argv, "consensus", GNUNET_SERVICE_OPTION_NONE, &run, NULL)) ? 0 : 1;