src/consensus/gnunet-service-consensus.c

   1 /*
   2       This file is part of GNUnet
   3       (C) 2012, 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 2, 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/gnunet-service-consensus.c
  23  * @brief multi-peer set reconciliation
  24  * @author Florian Dold
  25  */
  26
  27 #include "platform.h"
  28 #include "gnunet_common.h"
  29 #include "gnunet_protocols.h"
  30 #include "gnunet_applications.h"
  31 #include "gnunet_util_lib.h"
  32 #include "gnunet_set_service.h"
  33 #include "gnunet_consensus_service.h"
  34 #include "consensus_protocol.h"
  35 #include "consensus.h"
  36
  37
  38 /**
  39  * Log macro that prefixes the local peer and the peer we are in contact with.
  40  */
  41 #define LOG_PP(kind, cpi, m,...) GNUNET_log (kind, "P%d for P%d: " m, \
  42    cpi->session->local_peer_idx, (int) (cpi - cpi->session->info),##__VA_ARGS__)
  43
  44
  45 /**
  46  * Number of exponential rounds, used in the exp and completion round.
  47  */
  48 #define NUM_EXP_ROUNDS 4
  49
  50 /* forward declarations */
  51
  52 /* mutual recursion with struct ConsensusSession */
  53 struct ConsensusPeerInformation;
  54
  55 /* mutual recursion with round_over */
  56 static void
  57 subround_over (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc);
  58
  59
  60 /**
  61  * Describes the current round a consensus session is in.
  62  */
  63 enum ConsensusRound
  64 {
  65   /**
  66    * Not started the protocol yet.
  67    */
  68   CONSENSUS_ROUND_BEGIN=0,
  69   /**
  70    * Distribution of elements with the exponential scheme.
  71    */
  72   CONSENSUS_ROUND_EXCHANGE,
  73   /**
  74    * Exchange which elements each peer has, but don't
  75    * transmit the element's data, only their SHA-512 hashes.
  76    * This round uses the all-to-all scheme.
  77    */
  78   CONSENSUS_ROUND_INVENTORY,
  79   /**
  80    * Collect and distribute missing values with the exponential scheme.
  81    */
  82   CONSENSUS_ROUND_COMPLETION,
  83   /**
  84    * Consensus concluded. After timeout and finished communication with client,
  85    * consensus session will be destroyed.
  86    */
  87   CONSENSUS_ROUND_FINISH
  88 };
  89
  90
  91 /**
  92  * Complete information about the current round and all
  93  * subrounds.
  94  */
  95 struct RoundInfo
  96 {
  97   /**
  98    * The current main round.
  99    */
 100   enum ConsensusRound round;
 101   /**
 102    * The current exp round, valid if
 103    * the main round is an exp round.
 104    */
 105   uint32_t exp_round;
 106   /**
 107    * The current exp subround, valid if
 108    * the main round is an exp round.
 109    */
 110   uint32_t exp_subround;
 111 };
 112
 113
 114 /**
 115  * A consensus session consists of one local client and the remote authorities.
 116  */
 117 struct ConsensusSession
 118 {
 119   /**
 120    * Consensus sessions are kept in a DLL.
 121    */
 122   struct ConsensusSession *next;
 123
 124   /**
 125    * Consensus sessions are kept in a DLL.
 126    */
 127   struct ConsensusSession *prev;
 128
 129   /**
 130   * Global consensus identification, computed
 131   * from the session id and participating authorities.
 132   */
 133   struct GNUNET_HashCode global_id;
 134
 135   /**
 136    * Client that inhabits the session
 137    */
 138   struct GNUNET_SERVER_Client *client;
 139
 140   /**
 141    * Queued messages to the client.
 142    */
 143   struct GNUNET_MQ_Handle *client_mq;
 144
 145   /**
 146    * Timeout for all rounds together, single rounds will schedule a timeout task
 147    * with a fraction of the conclude timeout.
 148    * Only valid once the current round is not CONSENSUS_ROUND_BEGIN.
 149    */
 150   struct GNUNET_TIME_Relative conclude_timeout;
 151
 152   /**
 153    * Timeout task identifier for the current round.
 154    */
 155   GNUNET_SCHEDULER_TaskIdentifier round_timeout_tid;
 156
 157   /**
 158    * Number of other peers in the consensus.
 159    */
 160   unsigned int num_peers;
 161
 162   /**
 163    * Information about the other peers,
 164    * their state, etc.
 165    */
 166   struct ConsensusPeerInformation *info;
 167
 168   /**
 169    * Index of the local peer in the peers array
 170    */
 171   unsigned int local_peer_idx;
 172
 173   /**
 174    * Current round
 175    */
 176   enum ConsensusRound current_round;
 177
 178   /**
 179    * Permutation of peers for the current round,
 180    * maps logical index (for current round) to physical index (location in info array)
 181    */
 182   uint32_t *shuffle;
 183
 184   /**
 185    * Current round of the exponential scheme.
 186    */
 187   uint32_t exp_round;
 188
 189   /**
 190    * Current sub-round of the exponential scheme.
 191    */
 192   uint32_t exp_subround;
 193
 194   /**
 195    * The partner for the current exp-round
 196    */
 197   struct ConsensusPeerInformation *partner_outgoing;
 198
 199   /**
 200    * The partner for the current exp-round
 201    */
 202   struct ConsensusPeerInformation *partner_incoming;
 203
 204   /**
 205    * The consensus set of this session.
 206    */
 207   struct GNUNET_SET_Handle *element_set;
 208
 209   /**
 210    * Listener for requests from other peers.
 211    * Uses the session's global id as app id.
 212    */
 213   struct GNUNET_SET_ListenHandle *set_listener;
 214 };
 215
 216
 217 /**
 218  * Information about a peer that is in a consensus session.
 219  */
 220 struct ConsensusPeerInformation
 221 {
 222   /**
 223    * Peer identitty of the peer in the consensus session
 224    */
 225   struct GNUNET_PeerIdentity peer_id;
 226
 227   /**
 228    * Back-reference to the consensus session,
 229    * to that ConsensusPeerInformation can be used as a closure
 230    */
 231   struct ConsensusSession *session;
 232
 233   /**
 234    * We have finishes the exp-subround with the peer.
 235    */
 236   int exp_subround_finished;
 237
 238   /**
 239    * Set operation we are currently executing with this peer.
 240    */
 241   struct GNUNET_SET_OperationHandle *set_op;
 242
 243   /**
 244    * Set operation we are planning on executing with this peer.
 245    */
 246   struct GNUNET_SET_OperationHandle *delayed_set_op;
 247
 248   /**
 249    * Info about the round of the delayed set operation.
 250    */
 251   struct RoundInfo delayed_round_info;
 252 };
 253
 254
 255 /**
 256  * Linked list of sessions this peer participates in.
 257  */
 258 static struct ConsensusSession *sessions_head;
 259
 260 /**
 261  * Linked list of sessions this peer participates in.
 262  */
 263 static struct ConsensusSession *sessions_tail;
 264
 265 /**
 266  * Configuration of the consensus service.
 267  */
 268 static const struct GNUNET_CONFIGURATION_Handle *cfg;
 269
 270 /**
 271  * Handle to the server for this service.
 272  */
 273 static struct GNUNET_SERVER_Handle *srv;
 274
 275 /**
 276  * Peer that runs this service.
 277  */
 278 static struct GNUNET_PeerIdentity my_peer;
 279
 280
 281 static int
 282 have_exp_subround_finished (const struct ConsensusSession *session)
 283 {
 284   int not_finished;
 285   not_finished = 0;
 286   if ( (NULL != session->partner_outgoing) &&
 287        (GNUNET_NO == session->partner_outgoing->exp_subround_finished) )
 288     not_finished++;
 289   if ( (NULL != session->partner_incoming) &&
 290        (GNUNET_NO == session->partner_incoming->exp_subround_finished) )
 291     not_finished++;
 292   if (0 == not_finished)
 293     return GNUNET_YES;
 294   return GNUNET_NO;
 295 }
 296
 297
 298 /**
 299  * Destroy a session, free all resources associated with it.
 300  *
 301  * @param session the session to destroy
 302  */
 303 static void
 304 destroy_session (struct ConsensusSession *session)
 305 {
 306   int i;
 307
 308   GNUNET_CONTAINER_DLL_remove (sessions_head, sessions_tail, session);
 309   if (NULL != session->element_set)
 310   {
 311     GNUNET_SET_destroy (session->element_set);
 312     session->element_set = NULL;
 313   }
 314   if (NULL != session->set_listener)
 315   {
 316     GNUNET_SET_listen_cancel (session->set_listener);
 317     session->set_listener = NULL;
 318   }
 319   if (NULL != session->client_mq)
 320   {
 321     GNUNET_MQ_destroy (session->client_mq);
 322     session->client_mq = NULL;
 323   }
 324   if (NULL != session->client)
 325   {
 326     GNUNET_SERVER_client_disconnect (session->client);
 327     session->client = NULL;
 328   }
 329   if (NULL != session->shuffle)
 330   {
 331     GNUNET_free (session->shuffle);
 332     session->shuffle = NULL;
 333   }
 334   if (NULL != session->info)
 335   {
 336     for (i = 0; i < session->num_peers; i++)
 337     {
 338       struct ConsensusPeerInformation *cpi;
 339       cpi = &session->info[i];
 340       if (NULL != cpi->set_op)
 341       {
 342         GNUNET_SET_operation_cancel (cpi->set_op);
 343         cpi->set_op = NULL;
 344       }
 345     }
 346     GNUNET_free (session->info);
 347     session->info = NULL;
 348   }
 349   GNUNET_free (session);
 350 }
 351
 352
 353 /**
 354  * Iterator for set elements.
 355  *
 356  * @param cls closure
 357  * @param element the current element, NULL if all elements have been
 358  *        iterated over
 359  * @return GNUNET_YES to continue iterating, GNUNET_NO to stop.
 360  */
 361 static int
 362 send_to_client_iter (void *cls,
 363                      const struct GNUNET_SET_Element *element)
 364 {
 365   struct ConsensusSession *session = cls;
 366   struct GNUNET_MQ_Envelope *ev;
 367
 368   if (NULL != element)
 369   {
 370     struct GNUNET_CONSENSUS_ElementMessage *m;
 371
 372     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "P%d: got element for client\n",
 373                 session->local_peer_idx);
 374
 375     ev = GNUNET_MQ_msg_extra (m, element->size, GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_RECEIVED_ELEMENT);
 376     m->element_type = htons (element->type);
 377     memcpy (&m[1], element->data, element->size);
 378     GNUNET_MQ_send (session->client_mq, ev);
 379   }
 380   else
 381   {
 382     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "P%d: finished iterating elements for client\n",
 383                 session->local_peer_idx);
 384     ev = GNUNET_MQ_msg_header (GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_CONCLUDE_DONE);
 385     GNUNET_MQ_send (session->client_mq, ev);
 386   }
 387   return GNUNET_YES;
 388 }
 389
 390
 391 /**
 392  * Start the next round.
 393  * This function can be invoked as a timeout task, or called manually (tc will be NULL then).
 394  *
 395  * @param cls the session
 396  * @param tc task context, for when this task is invoked by the scheduler,
 397  *           NULL if invoked for another reason
 398  */
 399 static void
 400 round_over (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 401 {
 402   struct ConsensusSession *session;
 403
 404   /* don't kick off next round if we're shutting down */
 405   if ((NULL != tc) && (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
 406     return;
 407
 408   session = cls;
 409   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "P%d: round over\n", session->local_peer_idx);
 410
 411   if (session->round_timeout_tid != GNUNET_SCHEDULER_NO_TASK)
 412   {
 413     GNUNET_SCHEDULER_cancel (session->round_timeout_tid);
 414     session->round_timeout_tid = GNUNET_SCHEDULER_NO_TASK;
 415   }
 416
 417   switch (session->current_round)
 418   {
 419     case CONSENSUS_ROUND_BEGIN:
 420       session->current_round = CONSENSUS_ROUND_EXCHANGE;
 421       session->exp_round = 0;
 422       subround_over (session, NULL);
 423       break;
 424     case CONSENSUS_ROUND_EXCHANGE:
 425       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "P%d: finished, sending elements to client\n",
 426                   session->local_peer_idx);
 427       session->current_round = CONSENSUS_ROUND_FINISH;
 428       GNUNET_SET_iterate (session->element_set, send_to_client_iter, session);
 429       break;
 430     default:
 431       GNUNET_assert (0);
 432   }
 433 }
 434
 435
 436 /**
 437  * Create a new permutation for the session's peers in session->shuffle.
 438  * Uses a Fisher-Yates shuffle with pseudo-randomness coming from
 439  * both the global session id and the current round index.
 440  *
 441  * @param session the session to create the new permutation for
 442  */
 443 static void
 444 shuffle (struct ConsensusSession *session)
 445 {
 446   uint32_t i;
 447   uint32_t randomness[session->num_peers-1];
 448
 449   if (NULL == session->shuffle)
 450     session->shuffle = GNUNET_malloc (session->num_peers * sizeof (*session->shuffle));
 451
 452   GNUNET_CRYPTO_kdf (randomness, sizeof (randomness),
 453                      &session->exp_round, sizeof (uint32_t),
 454                      &session->global_id, sizeof (struct GNUNET_HashCode),
 455                      NULL);
 456
 457   for (i = 0; i < session->num_peers; i++)
 458     session->shuffle[i] = i;
 459
 460   for (i = session->num_peers - 1; i > 0; i--)
 461   {
 462     uint32_t x;
 463     uint32_t tmp;
 464     x = randomness[i-1] % session->num_peers;
 465     tmp = session->shuffle[x];
 466     session->shuffle[x] = session->shuffle[i];
 467     session->shuffle[i] = tmp;
 468   }
 469 }
 470
 471
 472 /**
 473  * Find and set the partner_incoming and partner_outgoing of our peer,
 474  * one of them may not exist (and thus set to NULL) if the number of peers
 475  * in the session is not a power of two.
 476  *
 477  * @param session the consensus session
 478  */
 479 static void
 480 find_partners (struct ConsensusSession *session)
 481 {
 482   unsigned int arc;
 483   unsigned int num_ghosts;
 484   unsigned int largest_arc;
 485   int partner_idx;
 486
 487   /* shuffled local index */
 488   int my_idx = session->shuffle[session->local_peer_idx];
 489
 490   /* distance to neighboring peer in current subround */
 491   arc = 1 << session->exp_subround;
 492   largest_arc = 1;
 493   while (largest_arc < session->num_peers)
 494     largest_arc <<= 1;
 495   num_ghosts = largest_arc - session->num_peers;
 496
 497   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "num ghosts: %d\n", num_ghosts);
 498
 499   if (0 == (my_idx & arc))
 500   {
 501     /* we are outgoing */
 502     partner_idx = (my_idx + arc) % session->num_peers;
 503     session->partner_outgoing = &session->info[session->shuffle[partner_idx]];
 504     session->partner_outgoing->exp_subround_finished = GNUNET_NO;
 505     /* are we a 'ghost' of a peer that would exist if
 506      * the number of peers was a power of two, and thus have to partner
 507      * with an additional peer?
 508      */
 509     if (my_idx < num_ghosts)
 510     {
 511       int ghost_partner_idx;
 512       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "my index %d, arc %d, peers %u\n", my_idx, arc, session->num_peers);
 513       ghost_partner_idx = (my_idx - (int) arc) % (int) session->num_peers;
 514       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "ghost partner is before %d\n", ghost_partner_idx);
 515       /* platform dependent; modulo sometimes returns negative values */
 516       if (ghost_partner_idx < 0)
 517         ghost_partner_idx += session->num_peers;
 518       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "ghost partner is after %d\n", ghost_partner_idx);
 519       session->partner_incoming = &session->info[session->shuffle[ghost_partner_idx]];
 520       session->partner_incoming->exp_subround_finished = GNUNET_NO;
 521       return;
 522     }
 523     session->partner_incoming = NULL;
 524     return;
 525   }
 526   partner_idx = (my_idx - (int) arc) % (int) session->num_peers;
 527   if (partner_idx < 0)
 528     partner_idx += session->num_peers;
 529   session->partner_outgoing = NULL;
 530   session->partner_incoming = &session->info[session->shuffle[partner_idx]];
 531   session->partner_incoming->exp_subround_finished = GNUNET_NO;
 532 }
 533
 534
 535 /**
 536  * Callback for set operation results. Called for each element
 537  * in the result set.
 538  *
 539  * @param cls closure
 540  * @param element a result element, only valid if status is GNUNET_SET_STATUS_OK
 541  * @param status see enum GNUNET_SET_Status
 542  */
 543 static void
 544 set_result_cb (void *cls,
 545                const struct GNUNET_SET_Element *element,
 546                enum GNUNET_SET_Status status)
 547 {
 548   struct ConsensusPeerInformation *cpi = cls;
 549   unsigned int remote_idx = cpi - cpi->session->info;
 550   unsigned int local_idx = cpi->session->local_peer_idx;
 551
 552   GNUNET_assert ((cpi == cpi->session->partner_outgoing) ||
 553                  (cpi == cpi->session->partner_incoming));
 554
 555   switch (status)
 556   {
 557     case GNUNET_SET_STATUS_OK:
 558       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "P%u: set result from P%u: element\n",
 559                   local_idx, remote_idx);
 560       break;
 561     case GNUNET_SET_STATUS_FAILURE:
 562       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "P%u: set result from P%u: failure\n",
 563                   local_idx, remote_idx);
 564       cpi->set_op = NULL;
 565       return;
 566     case GNUNET_SET_STATUS_HALF_DONE:
 567     case GNUNET_SET_STATUS_DONE:
 568       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "P%u: set result from P%u: done\n",
 569                   local_idx, remote_idx);
 570       cpi->exp_subround_finished = GNUNET_YES;
 571       cpi->set_op = NULL;
 572       if (have_exp_subround_finished (cpi->session) == GNUNET_YES)
 573       {
 574         GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "P%u: all reconciliations of subround done\n",
 575                     local_idx);
 576         subround_over (cpi->session, NULL);
 577       }
 578       else
 579       {
 580         GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "P%u: waiting for further set results\n",
 581                     local_idx);
 582       }
 583       return;
 584     default:
 585       GNUNET_break (0);
 586       return;
 587   }
 588
 589   switch (cpi->session->current_round)
 590   {
 591     case CONSENSUS_ROUND_EXCHANGE:
 592       GNUNET_SET_add_element (cpi->session->element_set, element, NULL, NULL);
 593       break;
 594     default:
 595       GNUNET_break (0);
 596       return;
 597   }
 598 }
 599
 600
 601 /**
 602  * Compare the round the session is in with the round of the given context message.
 603  *
 604  * @param session a consensus session
 605  * @param round a round context message
 606  * @return 0 if it's the same round, -1 if the session is in an earlier round,
 607  *         1 if the session is in a later round
 608  */
 609 static int
 610 rounds_compare (struct ConsensusSession *session,
 611                 struct RoundInfo* ri)
 612 {
 613   if (session->current_round < ri->round)
 614     return -1;
 615   if (session->current_round > ri->round)
 616     return 1;
 617   if (session->current_round == CONSENSUS_ROUND_EXCHANGE)
 618   {
 619     if (session->exp_round < ri->exp_round)
 620       return -1;
 621     if (session->exp_round > ri->exp_round)
 622       return 1;
 623     if (session->exp_subround < ri->exp_subround)
 624       return -1;
 625     if (session->exp_subround < ri->exp_subround)
 626       return 1;
 627     return 0;
 628   }
 629   /* comparing rounds when we are not in a exp round */
 630   GNUNET_assert (0);
 631 }
 632
 633
 634 /**
 635  * Do the next subround in the exp-scheme.
 636  * This function can be invoked as a timeout task, or called manually (tc will be NULL then).
 637  *
 638  * @param cls the session
 639  * @param tc task context, for when this task is invoked by the scheduler,
 640  *           NULL if invoked for another reason
 641  */
 642 static void
 643 subround_over (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 644 {
 645   struct ConsensusSession *session;
 646   int i;
 647
 648   /* don't kick off next subround if we're shutting down */
 649   if ((NULL != tc) && (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
 650     return;
 651   session = cls;
 652   /* cancel timeout */
 653   if (session->round_timeout_tid != GNUNET_SCHEDULER_NO_TASK)
 654   {
 655     GNUNET_SCHEDULER_cancel (session->round_timeout_tid);
 656     session->round_timeout_tid = GNUNET_SCHEDULER_NO_TASK;
 657   }
 658
 659   if (session->exp_round >= NUM_EXP_ROUNDS)
 660   {
 661     round_over (session, NULL);
 662     return;
 663   }
 664
 665   if (session->exp_round == 0)
 666   {
 667     /* initialize everything for the log-rounds */
 668     session->exp_round = 1;
 669     session->exp_subround = 0;
 670     if (NULL == session->shuffle)
 671       session->shuffle = GNUNET_malloc ((sizeof (int)) * session->num_peers);
 672     for (i = 0; i < session->num_peers; i++)
 673       session->shuffle[i] = i;
 674   }
 675   else if (session->exp_subround + 1 >= (int) ceil (log2 (session->num_peers)))
 676   {
 677     /* subrounds done, start new log-round */
 678     session->exp_round++;
 679     session->exp_subround = 0;
 680     //shuffle (session);
 681   }
 682   else
 683   {
 684     session->exp_subround++;
 685   }
 686
 687   /* determine the incoming and outgoing partner */
 688   find_partners (session);
 689
 690   GNUNET_assert (session->partner_outgoing != &session->info[session->local_peer_idx]);
 691   GNUNET_assert (session->partner_incoming != &session->info[session->local_peer_idx]);
 692
 693   /* initiate set operation with the outgoing partner */
 694   if (NULL != session->partner_outgoing)
 695   {
 696     struct GNUNET_CONSENSUS_RoundContextMessage *msg;
 697     msg = GNUNET_new (struct GNUNET_CONSENSUS_RoundContextMessage);
 698     msg->header.type = htons (GNUNET_MESSAGE_TYPE_CONSENSUS_P2P_ROUND_CONTEXT);
 699     msg->header.size = htons (sizeof *msg);
 700     msg->round = htonl (session->current_round);
 701     msg->exp_round = htonl (session->exp_round);
 702     msg->exp_subround = htonl (session->exp_subround);
 703
 704     if (NULL != session->partner_outgoing->set_op)
 705     {
 706       GNUNET_SET_operation_cancel (session->partner_outgoing->set_op);
 707     }
 708     session->partner_outgoing->set_op =
 709         GNUNET_SET_prepare (&session->partner_outgoing->peer_id,
 710                             &session->global_id,
 711                             (struct GNUNET_MessageHeader *) msg,
 712                             0, /* FIXME: salt */
 713                             GNUNET_SET_RESULT_ADDED,
 714                             set_result_cb, session->partner_outgoing);
 715     GNUNET_free (msg);
 716     GNUNET_SET_commit (session->partner_outgoing->set_op, session->element_set);
 717   }
 718
 719   /* commit to the delayed set operation */
 720   if ((NULL != session->partner_incoming) && (NULL != session->partner_incoming->delayed_set_op))
 721   {
 722     int cmp = rounds_compare (session, &session->partner_incoming->delayed_round_info);
 723
 724     if (NULL != session->partner_incoming->set_op)
 725     {
 726       GNUNET_SET_operation_cancel (session->partner_incoming->set_op);
 727       session->partner_incoming->set_op = NULL;
 728     }
 729     if (cmp == 0)
 730     {
 731       GNUNET_SET_commit (session->partner_incoming->delayed_set_op, session->element_set);
 732       session->partner_incoming->set_op = session->partner_incoming->delayed_set_op;
 733       session->partner_incoming->delayed_set_op = NULL;
 734       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "P%d resumed delayed round with P%d\n",
 735                   session->local_peer_idx, (int) (session->partner_incoming - session->info));
 736     }
 737     else
 738     {
 739       /* this should not happen -- a round has been skipped! */
 740       GNUNET_break_op (0);
 741     }
 742   }
 743
 744 #ifdef GNUNET_EXTRA_LOGGING
 745   {
 746     int in;
 747     int out;
 748     if (session->partner_outgoing == NULL)
 749       out = -1;
 750     else
 751       out = (int) (session->partner_outgoing - session->info);
 752     if (session->partner_incoming == NULL)
 753       in = -1;
 754     else
 755       in = (int) (session->partner_incoming - session->info);
 756     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "P%u: doing exp-round, r=%d, sub=%d, in: %d, out: %d\n", session->local_peer_idx,
 757                 session->exp_round, session->exp_subround, in, out);
 758   }
 759 #endif /* GNUNET_EXTRA_LOGGING */
 760
 761 }
 762
 763
 764 /**
 765  * Search peer in the list of peers in session.
 766  *
 767  * @param peer peer to find
 768  * @param session session with peer
 769  * @return index of peer, -1 if peer is not in session
 770  */
 771 static int
 772 get_peer_idx (const struct GNUNET_PeerIdentity *peer, const struct ConsensusSession *session)
 773 {
 774   int i;
 775   for (i = 0; i < session->num_peers; i++)
 776     if (0 == memcmp (peer, &session->info[i].peer_id, sizeof *peer))
 777       return i;
 778   return -1;
 779 }
 780
 781
 782 /**
 783  * Compute a global, (hopefully) unique consensus session id,
 784  * from the local id of the consensus session, and the identities of all participants.
 785  * Thus, if the local id of two consensus sessions coincide, but are not comprised of
 786  * exactly the same peers, the global id will be different.
 787  *
 788  * @param session session to generate the global id for
 789  * @param session_id local id of the consensus session
 790  */
 791 static void
 792 compute_global_id (struct ConsensusSession *session, const struct GNUNET_HashCode *session_id)
 793 {
 794   int i;
 795   struct GNUNET_HashCode tmp;
 796
 797   /* FIXME: use kdf? */
 798
 799   session->global_id = *session_id;
 800   for (i = 0; i < session->num_peers; ++i)
 801   {
 802     GNUNET_CRYPTO_hash_xor (&session->global_id, &session->info[i].peer_id.hashPubKey, &tmp);
 803     session->global_id = tmp;
 804     GNUNET_CRYPTO_hash (&session->global_id, sizeof (struct GNUNET_PeerIdentity), &tmp);
 805     session->global_id = tmp;
 806   }
 807 }
 808
 809
 810 /**
 811  * Although GNUNET_CRYPTO_hash_cmp exisits, it does not have
 812  * the correct signature to be used with e.g. qsort.
 813  * We use this function instead.
 814  *
 815  * @param h1 some hash code
 816  * @param h2 some hash code
 817  * @return 1 if h1 > h2, -1 if h1 < h2 and 0 if h1 == h2.
 818  */
 819 static int
 820 hash_cmp (const void *h1, const void *h2)
 821 {
 822   return GNUNET_CRYPTO_hash_cmp ((struct GNUNET_HashCode *) h1, (struct GNUNET_HashCode *) h2);
 823 }
 824
 825
 826 /**
 827  * Create the sorted list of peers for the session,
 828  * add the local peer if not in the join message.
 829  */
 830 static void
 831 initialize_session_peer_list (struct ConsensusSession *session,
 832                               struct GNUNET_CONSENSUS_JoinMessage *join_msg)
 833 {
 834   unsigned int local_peer_in_list;
 835   uint32_t listed_peers;
 836   const struct GNUNET_PeerIdentity *msg_peers;
 837   struct GNUNET_PeerIdentity *peers;
 838   unsigned int i;
 839
 840   GNUNET_assert (NULL != join_msg);
 841
 842   /* peers in the join message, may or may not include the local peer */
 843   listed_peers = ntohl (join_msg->num_peers);
 844
 845   session->num_peers = listed_peers;
 846
 847   msg_peers = (struct GNUNET_PeerIdentity *) &join_msg[1];
 848
 849   local_peer_in_list = GNUNET_NO;
 850   for (i = 0; i < listed_peers; i++)
 851   {
 852     if (0 == memcmp (&msg_peers[i], &my_peer, sizeof (struct GNUNET_PeerIdentity)))
 853     {
 854       local_peer_in_list = GNUNET_YES;
 855       break;
 856     }
 857   }
 858
 859   if (GNUNET_NO == local_peer_in_list)
 860     session->num_peers++;
 861
 862   peers = GNUNET_malloc (session->num_peers * sizeof (struct GNUNET_PeerIdentity));
 863
 864   if (GNUNET_NO == local_peer_in_list)
 865     peers[session->num_peers - 1] = my_peer;
 866
 867   memcpy (peers, msg_peers, listed_peers * sizeof (struct GNUNET_PeerIdentity));
 868   qsort (peers, session->num_peers, sizeof (struct GNUNET_PeerIdentity), &hash_cmp);
 869
 870   session->info = GNUNET_malloc (session->num_peers * sizeof (struct ConsensusPeerInformation));
 871
 872   for (i = 0; i < session->num_peers; ++i)
 873   {
 874     /* initialize back-references, so consensus peer information can
 875      * be used as closure */
 876     session->info[i].session = session;
 877     session->info[i].peer_id = peers[i];
 878   }
 879
 880   GNUNET_free (peers);
 881 }
 882
 883
 884 /**
 885  * Called when another peer wants to do a set operation with the
 886  * local peer.
 887  *
 888  * @param cls closure
 889  * @param other_peer the other peer
 890  * @param context_msg message with application specific information from
 891  *        the other peer
 892  * @param request request from the other peer, use GNUNET_SET_accept
 893  *        to accept it, otherwise the request will be refused
 894  *        Note that we don't use a return value here, as it is also
 895  *        necessary to specify the set we want to do the operation with,
 896  *        whith sometimes can be derived from the context message.
 897  *        Also necessary to specify the timeout.
 898  */
 899 static void
 900 set_listen_cb (void *cls,
 901                const struct GNUNET_PeerIdentity *other_peer,
 902                const struct GNUNET_MessageHeader *context_msg,
 903                struct GNUNET_SET_Request *request)
 904 {
 905   struct ConsensusSession *session = cls;
 906   struct GNUNET_CONSENSUS_RoundContextMessage *msg = (struct GNUNET_CONSENSUS_RoundContextMessage *) context_msg;
 907   struct ConsensusPeerInformation *cpi;
 908   struct GNUNET_SET_OperationHandle *set_op;
 909   struct RoundInfo round_info;
 910   int index;
 911   int cmp;
 912
 913   if (NULL == context_msg)
 914   {
 915     GNUNET_break_op (0);
 916     return;
 917   }
 918
 919   index = get_peer_idx (other_peer, session);
 920
 921   if (index < 0)
 922   {
 923     GNUNET_break_op (0);
 924     return;
 925   }
 926
 927   round_info.round = ntohl (msg->round);
 928   round_info.exp_round = ntohl (msg->exp_round);
 929   round_info.exp_subround = ntohl (msg->exp_subround);
 930
 931   cpi = &session->info[index];
 932
 933   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "P%d got set request from P%d\n", session->local_peer_idx, index);
 934
 935   switch (session->current_round)
 936   {
 937     case CONSENSUS_ROUND_EXCHANGE:
 938       cmp = rounds_compare (session, &round_info);
 939       if (cmp > 0)
 940       {
 941         /* the other peer is too late */
 942         GNUNET_break_op (0);
 943         return;
 944       }
 945       /* kill old request, if any. this is legal,
 946        * as the other peer would not make a new request if it would want to
 947        * complete the old one! */
 948       if (NULL != cpi->set_op)
 949       {
 950         GNUNET_SET_operation_cancel (cpi->set_op);
 951         cpi->set_op = NULL;
 952       }
 953       set_op = GNUNET_SET_accept (request, GNUNET_SET_RESULT_ADDED,
 954                                        set_result_cb, &session->info[index]);
 955       if (cmp == 0)
 956       {
 957         cpi->set_op = set_op;
 958         GNUNET_SET_commit (set_op, session->element_set);
 959         GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "P%d commited to set request from P%d\n", session->local_peer_idx, index);
 960       }
 961       else
 962       {
 963         /* if there's a exp subround running, mark it as finished, as the set op has been canceled! */
 964         cpi->delayed_set_op = set_op;
 965         cpi->delayed_round_info = round_info;
 966         cpi->exp_subround_finished = GNUNET_YES;
 967         GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "P%d delaying set request from P%d\n", session->local_peer_idx, index);
 968       }
 969       break;
 970     default:
 971       GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "P%d got unexpected set request in round %d from P%d\n",
 972                   session->local_peer_idx, session->current_round, index);
 973       GNUNET_break_op (0);
 974       return;
 975   }
 976 }
 977
 978
 979 /**
 980  * Initialize the session, continue receiving messages from the owning client
 981  *
 982  * @param session the session to initialize
 983  * @param join_msg the join message from the client
 984  */
 985 static void
 986 initialize_session (struct ConsensusSession *session,
 987                     struct GNUNET_CONSENSUS_JoinMessage *join_msg)
 988 {
 989   struct ConsensusSession *other_session;
 990
 991   initialize_session_peer_list (session, join_msg);
 992   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "session with %u peers\n", session->num_peers);
 993   compute_global_id (session, &join_msg->session_id);
 994
 995   /* check if some local client already owns the session.
 996    * it is only legal to have a session with an existing global id
 997    * if all other sessions with this global id are finished.*/
 998   other_session = sessions_head;
 999   while (NULL != other_session)
1000   {
1001     if ((other_session != session) &&
1002         (0 == GNUNET_CRYPTO_hash_cmp (&session->global_id, &other_session->global_id)))
1003     {
1004       if (CONSENSUS_ROUND_FINISH != other_session->current_round)
1005       {
1006         GNUNET_break (0);
1007         destroy_session (session);
1008         return;
1009       }
1010       break;
1011     }
1012     other_session = other_session->next;
1013   }
1014
1015   session->local_peer_idx = get_peer_idx (&my_peer, session);
1016   GNUNET_assert (-1 != session->local_peer_idx);
1017   session->element_set = GNUNET_SET_create (cfg, GNUNET_SET_OPERATION_UNION);
1018   GNUNET_assert (NULL != session->element_set);
1019   session->set_listener = GNUNET_SET_listen (cfg, GNUNET_SET_OPERATION_UNION,
1020                                              &session->global_id,
1021                                              set_listen_cb, session);
1022   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "%d is the local peer\n", session->local_peer_idx);
1023   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "session %s initialized\n", GNUNET_h2s (&session->global_id));
1024 }
1025
1026
1027 static struct ConsensusSession *
1028 get_session_by_client (struct GNUNET_SERVER_Client *client)
1029 {
1030   struct ConsensusSession *session;
1031
1032   session = sessions_head;
1033   while (NULL != session)
1034   {
1035     if (session->client == client)
1036       return session;
1037     session = session->next;
1038   }
1039   return NULL;
1040 }
1041
1042
1043 /**
1044  * Called when a client wants to join a consensus session.
1045  *
1046  * @param cls unused
1047  * @param client client that sent the message
1048  * @param m message sent by the client
1049  */
1050 static void
1051 client_join (void *cls,
1052              struct GNUNET_SERVER_Client *client,
1053              const struct GNUNET_MessageHeader *m)
1054 {
1055   struct ConsensusSession *session;
1056
1057   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "join message sent by client\n");
1058
1059   session = get_session_by_client (client);
1060   if (NULL != session)
1061   {
1062     GNUNET_break (0);
1063     GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
1064     return;
1065   }
1066   session = GNUNET_new (struct ConsensusSession);
1067   session->client = client;
1068   session->client_mq = GNUNET_MQ_queue_for_server_client (client);
1069   GNUNET_CONTAINER_DLL_insert (sessions_head, sessions_tail, session);
1070   initialize_session (session, (struct GNUNET_CONSENSUS_JoinMessage *) m);
1071   GNUNET_SERVER_receive_done (client, GNUNET_OK);
1072
1073   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "join done\n");
1074 }
1075
1076
1077 /**
1078  * Called when a client performs an insert operation.
1079  *
1080  * @param cls (unused)
1081  * @param client client handle
1082  * @param m message sent by the client
1083  */
1084 void
1085 client_insert (void *cls,
1086                struct GNUNET_SERVER_Client *client,
1087                const struct GNUNET_MessageHeader *m)
1088 {
1089   struct ConsensusSession *session;
1090   struct GNUNET_CONSENSUS_ElementMessage *msg;
1091   struct GNUNET_SET_Element *element;
1092   ssize_t element_size;
1093
1094   session = get_session_by_client (client);
1095
1096   if (NULL == session)
1097   {
1098     GNUNET_break (0);
1099     GNUNET_SERVER_client_disconnect (client);
1100     return;
1101   }
1102
1103   if (CONSENSUS_ROUND_BEGIN != session->current_round)
1104   {
1105     GNUNET_break (0);
1106     GNUNET_SERVER_client_disconnect (client);
1107     return;
1108   }
1109
1110   msg = (struct GNUNET_CONSENSUS_ElementMessage *) m;
1111   element_size = ntohs (msg->header.size) - sizeof (struct GNUNET_CONSENSUS_ElementMessage);
1112   if (element_size < 0)
1113   {
1114     GNUNET_break (0);
1115     return;
1116   }
1117
1118   element = GNUNET_malloc (sizeof (struct GNUNET_SET_Element) + element_size);
1119   element->type = msg->element_type;
1120   element->size = element_size;
1121   memcpy (&element[1], &msg[1], element_size);
1122   element->data = &element[1];
1123   GNUNET_SET_add_element (session->element_set, element, NULL, NULL);
1124   GNUNET_free (element);
1125   GNUNET_SERVER_receive_done (client, GNUNET_OK);
1126
1127   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "P%u: element added\n", session->local_peer_idx);
1128 }
1129
1130
1131 /**
1132  * Called when a client performs the conclude operation.
1133  *
1134  * @param cls (unused)
1135  * @param client client handle
1136  * @param message message sent by the client
1137  */
1138 static void
1139 client_conclude (void *cls,
1140                  struct GNUNET_SERVER_Client *client,
1141                  const struct GNUNET_MessageHeader *message)
1142 {
1143   struct ConsensusSession *session;
1144   struct GNUNET_CONSENSUS_ConcludeMessage *cmsg;
1145
1146
1147   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "conclude requested\n");
1148   cmsg = (struct GNUNET_CONSENSUS_ConcludeMessage *) message;
1149   session = get_session_by_client (client);
1150   if (NULL == session)
1151   {
1152     /* client not found */
1153     GNUNET_break (0);
1154     GNUNET_SERVER_client_disconnect (client);
1155     return;
1156   }
1157   if (CONSENSUS_ROUND_BEGIN != session->current_round)
1158   {
1159     /* client requested conclude twice */
1160     GNUNET_break (0);
1161     return;
1162   }
1163   if (session->num_peers <= 1)
1164   {
1165     /* FIXME: what to do here? */
1166     //send_client_conclude_done (session);
1167   }
1168   else
1169   {
1170     session->conclude_timeout = GNUNET_TIME_relative_ntoh (cmsg->timeout);
1171     /* the 'begin' round is over, start with the next, actual round */
1172     round_over (session, NULL);
1173   }
1174
1175   GNUNET_assert (CONSENSUS_ROUND_BEGIN != session->current_round);
1176   GNUNET_SERVER_receive_done (client, GNUNET_OK);
1177 }
1178
1179
1180 /**
1181  * Called to clean up, after a shutdown has been requested.
1182  *
1183  * @param cls closure
1184  * @param tc context information (why was this task triggered now)
1185  */
1186 static void
1187 shutdown_task (void *cls,
1188                const struct GNUNET_SCHEDULER_TaskContext *tc)
1189 {
1190   while (NULL != sessions_head)
1191     destroy_session (sessions_head);
1192
1193   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "handled shutdown request\n");
1194 }
1195
1196
1197 /**
1198  * Clean up after a client after it is
1199  * disconnected (either by us or by itself)
1200  *
1201  * @param cls closure, unused
1202  * @param client the client to clean up after
1203  */
1204 void
1205 handle_client_disconnect (void *cls, struct GNUNET_SERVER_Client *client)
1206 {
1207   struct ConsensusSession *session;
1208
1209   session = get_session_by_client (client);
1210   if (NULL == session)
1211     return;
1212   if ((CONSENSUS_ROUND_BEGIN == session->current_round) ||
1213       (CONSENSUS_ROUND_FINISH == session->current_round))
1214     destroy_session (session);
1215   else
1216     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "client disconnected, but waiting for consensus to finish\n");
1217 }
1218
1219
1220 /**
1221  * Start processing consensus requests.
1222  *
1223  * @param cls closure
1224  * @param server the initialized server
1225  * @param c configuration to use
1226  */
1227 static void
1228 run (void *cls, struct GNUNET_SERVER_Handle *server,
1229      const struct GNUNET_CONFIGURATION_Handle *c)
1230 {
1231   static const struct GNUNET_SERVER_MessageHandler server_handlers[] = {
1232     {&client_conclude, NULL, GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_CONCLUDE,
1233         sizeof (struct GNUNET_CONSENSUS_ConcludeMessage)},
1234     {&client_insert, NULL, GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_INSERT, 0},
1235     {&client_join, NULL, GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_JOIN, 0},
1236     {NULL, NULL, 0, 0}
1237   };
1238
1239   cfg = c;
1240   srv = server;
1241   if (GNUNET_OK != GNUNET_CRYPTO_get_host_identity (cfg, &my_peer))
1242   {
1243     GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "could not retrieve host identity\n");
1244     GNUNET_break (0);
1245     GNUNET_SCHEDULER_shutdown ();
1246     return;
1247   }
1248   GNUNET_SERVER_add_handlers (server, server_handlers);
1249   GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_FOREVER_REL, &shutdown_task, NULL);
1250   GNUNET_SERVER_disconnect_notify (server, handle_client_disconnect, NULL);
1251   GNUNET_log (GNUNET_ERROR_TYPE_INFO, "consensus running\n");
1252 }
1253
1254
1255 /**
1256  * The main function for the consensus service.
1257  *
1258  * @param argc number of arguments from the command line
1259  * @param argv command line arguments
1260  * @return 0 ok, 1 on error
1261  */
1262 int
1263 main (int argc, char *const *argv)
1264 {
1265   int ret;
1266   ret = GNUNET_SERVICE_run (argc, argv, "consensus", GNUNET_SERVICE_OPTION_NONE, &run, NULL);
1267   GNUNET_log (GNUNET_ERROR_TYPE_INFO, "exit (%d)\n", GNUNET_OK != ret);
1268   return (GNUNET_OK == ret) ? 0 : 1;
1269 }
1270