-adding missing break statements
[oweals/gnunet.git] / src / consensus / gnunet-service-consensus.c
index 1c2c78422cdb052e522460c17becf0f61f3c8ce0..914943f114e6d9143d5414592590698cd78214b9 100644 (file)
@@ -4,7 +4,7 @@
 
       GNUnet is free software; you can redistribute it and/or modify
       it under the terms of the GNU General Public License as published
-      by the Free Software Foundation; either version 2, or (at your
+      by the Free Software Foundation; either version 3, or (at your
       option) any later version.
 
       GNUnet is distributed in the hope that it will be useful, but
  */
 
 #include "platform.h"
-#include "gnunet_common.h"
+#include "gnunet_util_lib.h"
 #include "gnunet_protocols.h"
 #include "gnunet_applications.h"
-#include "gnunet_util_lib.h"
 #include "gnunet_set_service.h"
 #include "gnunet_consensus_service.h"
 #include "consensus_protocol.h"
@@ -87,6 +86,30 @@ enum ConsensusRound
   CONSENSUS_ROUND_FINISH
 };
 
+
+/**
+ * Complete information about the current round and all
+ * subrounds.
+ */
+struct RoundInfo
+{
+  /**
+   * The current main round.
+   */
+  enum ConsensusRound round;
+  /**
+   * The current exp round, valid if
+   * the main round is an exp round.
+   */
+  uint32_t exp_round;
+  /**
+   * The current exp subround, valid if
+   * the main round is an exp round.
+   */
+  uint32_t exp_subround;
+};
+
+
 /**
  * A consensus session consists of one local client and the remote authorities.
  */
@@ -124,7 +147,7 @@ struct ConsensusSession
    * Only valid once the current round is not CONSENSUS_ROUND_BEGIN.
    */
   struct GNUNET_TIME_Relative conclude_timeout;
-  
+
   /**
    * Timeout task identifier for the current round.
    */
@@ -153,10 +176,14 @@ struct ConsensusSession
 
   /**
    * Permutation of peers for the current round,
-   * maps logical index (for current round) to physical index (location in info array)
    */
   uint32_t *shuffle;
 
+  /**
+   * Inverse permutation of peers for the current round,
+   */
+  uint32_t *shuffle_inv;
+
   /**
    * Current round of the exponential scheme.
    */
@@ -217,9 +244,14 @@ struct ConsensusPeerInformation
   struct GNUNET_SET_OperationHandle *set_op;
 
   /**
-   * Has commit been called on the set_op?
+   * Set operation we are planning on executing with this peer.
+   */
+  struct GNUNET_SET_OperationHandle *delayed_set_op;
+
+  /**
+   * Info about the round of the delayed set operation.
    */
-  int set_op_commited;
+  struct RoundInfo delayed_round_info;
 };
 
 
@@ -254,7 +286,7 @@ have_exp_subround_finished (const struct ConsensusSession *session)
 {
   int not_finished;
   not_finished = 0;
-  if ( (NULL != session->partner_outgoing) && 
+  if ( (NULL != session->partner_outgoing) &&
        (GNUNET_NO == session->partner_outgoing->exp_subround_finished) )
     not_finished++;
   if ( (NULL != session->partner_incoming) &&
@@ -266,11 +298,9 @@ have_exp_subround_finished (const struct ConsensusSession *session)
 }
 
 
-
-
 /**
  * Destroy a session, free all resources associated with it.
- * 
+ *
  * @param session the session to destroy
  */
 static void
@@ -279,23 +309,47 @@ destroy_session (struct ConsensusSession *session)
   int i;
 
   GNUNET_CONTAINER_DLL_remove (sessions_head, sessions_tail, session);
+  if (NULL != session->element_set)
+  {
+    GNUNET_SET_destroy (session->element_set);
+    session->element_set = NULL;
+  }
+  if (NULL != session->set_listener)
+  {
+    GNUNET_SET_listen_cancel (session->set_listener);
+    session->set_listener = NULL;
+  }
   if (NULL != session->client_mq)
   {
     GNUNET_MQ_destroy (session->client_mq);
     session->client_mq = NULL;
   }
+  if (NULL != session->client)
+  {
+    GNUNET_SERVER_client_disconnect (session->client);
+    session->client = NULL;
+  }
   if (NULL != session->shuffle)
   {
     GNUNET_free (session->shuffle);
     session->shuffle = NULL;
   }
+  if (NULL != session->shuffle_inv)
+  {
+    GNUNET_free (session->shuffle_inv);
+    session->shuffle_inv = NULL;
+  }
   if (NULL != session->info)
   {
     for (i = 0; i < session->num_peers; i++)
     {
       struct ConsensusPeerInformation *cpi;
       cpi = &session->info[i];
-      GNUNET_free (cpi);
+      if (NULL != cpi->set_op)
+      {
+        GNUNET_SET_operation_cancel (cpi->set_op);
+        cpi->set_op = NULL;
+      }
     }
     GNUNET_free (session->info);
     session->info = NULL;
@@ -304,6 +358,44 @@ destroy_session (struct ConsensusSession *session)
 }
 
 
+/**
+ * Iterator for set elements.
+ *
+ * @param cls closure
+ * @param element the current element, NULL if all elements have been
+ *        iterated over
+ * @return GNUNET_YES to continue iterating, GNUNET_NO to stop.
+ */
+static int
+send_to_client_iter (void *cls,
+                     const struct GNUNET_SET_Element *element)
+{
+  struct ConsensusSession *session = cls;
+  struct GNUNET_MQ_Envelope *ev;
+
+  if (NULL != element)
+  {
+    struct GNUNET_CONSENSUS_ElementMessage *m;
+
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "P%d: got element for client\n",
+                session->local_peer_idx);
+
+    ev = GNUNET_MQ_msg_extra (m, element->size, GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_RECEIVED_ELEMENT);
+    m->element_type = htons (element->type);
+    memcpy (&m[1], element->data, element->size);
+    GNUNET_MQ_send (session->client_mq, ev);
+  }
+  else
+  {
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "P%d: finished iterating elements for client\n",
+                session->local_peer_idx);
+    ev = GNUNET_MQ_msg_header (GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_CONCLUDE_DONE);
+    GNUNET_MQ_send (session->client_mq, ev);
+  }
+  return GNUNET_YES;
+}
+
+
 /**
  * Start the next round.
  * This function can be invoked as a timeout task, or called manually (tc will be NULL then).
@@ -312,7 +404,7 @@ destroy_session (struct ConsensusSession *session)
  * @param tc task context, for when this task is invoked by the scheduler,
  *           NULL if invoked for another reason
  */
-static void 
+static void
 round_over (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 {
   struct ConsensusSession *session;
@@ -322,7 +414,7 @@ round_over (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
     return;
 
   session = cls;
-  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "P%d: round over\n", session->local_peer_idx);
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "P%d: round over\n", session->local_peer_idx);
 
   if (session->round_timeout_tid != GNUNET_SCHEDULER_NO_TASK)
   {
@@ -338,8 +430,11 @@ round_over (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
       subround_over (session, NULL);
       break;
     case CONSENSUS_ROUND_EXCHANGE:
-      /* FIXME: send all elements to client */
+      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "P%d: finished, sending elements to client\n",
+                  session->local_peer_idx);
       session->current_round = CONSENSUS_ROUND_FINISH;
+      GNUNET_SET_iterate (session->element_set, send_to_client_iter, session);
+      break;
     default:
       GNUNET_assert (0);
   }
@@ -361,8 +456,10 @@ shuffle (struct ConsensusSession *session)
 
   if (NULL == session->shuffle)
     session->shuffle = GNUNET_malloc (session->num_peers * sizeof (*session->shuffle));
+  if (NULL == session->shuffle_inv)
+    session->shuffle_inv = GNUNET_malloc (session->num_peers * sizeof (*session->shuffle_inv));
 
-  GNUNET_CRYPTO_kdf (randomness, sizeof (randomness), 
+  GNUNET_CRYPTO_kdf (randomness, sizeof (randomness),
                     &session->exp_round, sizeof (uint32_t),
                      &session->global_id, sizeof (struct GNUNET_HashCode),
                     NULL);
@@ -374,11 +471,15 @@ shuffle (struct ConsensusSession *session)
   {
     uint32_t x;
     uint32_t tmp;
-    x = randomness[i-1];
+    x = randomness[i-1] % session->num_peers;
     tmp = session->shuffle[x];
     session->shuffle[x] = session->shuffle[i];
     session->shuffle[i] = tmp;
   }
+
+  /* create the inverse */
+  for (i = 0; i < session->num_peers; i++)
+    session->shuffle_inv[session->shuffle[i]] = i;
 }
 
 
@@ -392,48 +493,61 @@ shuffle (struct ConsensusSession *session)
 static void
 find_partners (struct ConsensusSession *session)
 {
-  int arc;
+  unsigned int arc;
+  unsigned int num_ghosts;
+  unsigned int largest_arc;
   int partner_idx;
-  int largest_arc;
-  int num_ghosts;
+
+  /* shuffled local index */
+  int my_idx = session->shuffle[session->local_peer_idx];
 
   /* distance to neighboring peer in current subround */
   arc = 1 << session->exp_subround;
-  partner_idx = (session->local_peer_idx + arc) % session->num_peers;
   largest_arc = 1;
   while (largest_arc < session->num_peers)
     largest_arc <<= 1;
   num_ghosts = largest_arc - session->num_peers;
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "largest arc: %u\n", largest_arc);
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "arc: %u\n", arc);
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "num ghosts: %u\n", num_ghosts);
 
-  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "num ghosts: %d\n", num_ghosts);
-
-  if (0 == (session->local_peer_idx & arc))
+  if (0 == (my_idx & arc))
   {
     /* we are outgoing */
-    session->partner_outgoing = &session->info[session->shuffle[partner_idx]];
+    partner_idx = (my_idx + arc) % session->num_peers;
+    session->partner_outgoing = &session->info[session->shuffle_inv[partner_idx]];
+    session->partner_outgoing->exp_subround_finished = GNUNET_NO;
     /* are we a 'ghost' of a peer that would exist if
      * the number of peers was a power of two, and thus have to partner
      * with an additional peer?
      */
-    if (session->local_peer_idx < num_ghosts)
+    if (my_idx < num_ghosts)
     {
       int ghost_partner_idx;
-      ghost_partner_idx = (session->local_peer_idx - arc) % session->num_peers;
+      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "my index %d, arc %d, peers %u\n", my_idx, arc, session->num_peers);
+      ghost_partner_idx = (my_idx - (int) arc) % (int) session->num_peers;
       /* platform dependent; modulo sometimes returns negative values */
       if (ghost_partner_idx < 0)
-        ghost_partner_idx += arc;
-      session->partner_incoming = &session->info[session->shuffle[ghost_partner_idx]];
-    }
-    else
-    {
-      session->partner_incoming = NULL;
+        ghost_partner_idx += session->num_peers;
+      /* we only need to have a ghost partner if the partner is outgoing */
+      if (0 == (ghost_partner_idx & arc))
+      {
+        GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "ghost partner is %d\n", ghost_partner_idx);
+        session->partner_incoming = &session->info[session->shuffle_inv[ghost_partner_idx]];
+        session->partner_incoming->exp_subround_finished = GNUNET_NO;
+        return;
+      }
     }
+    session->partner_incoming = NULL;
+    return;
   }
-  else
-  {
-    session->partner_outgoing = NULL;
-    session->partner_incoming = &session->info[session->shuffle[partner_idx]];
-  }
+  /* we only have an incoming connection */
+  partner_idx = (my_idx - (int) arc) % (int) session->num_peers;
+  if (partner_idx < 0)
+    partner_idx += session->num_peers;
+  session->partner_outgoing = NULL;
+  session->partner_incoming = &session->info[session->shuffle_inv[partner_idx]];
+  session->partner_incoming->exp_subround_finished = GNUNET_NO;
 }
 
 
@@ -445,27 +559,46 @@ find_partners (struct ConsensusSession *session)
  * @param element a result element, only valid if status is GNUNET_SET_STATUS_OK
  * @param status see enum GNUNET_SET_Status
  */
-static void 
+static void
 set_result_cb (void *cls,
                const struct GNUNET_SET_Element *element,
                enum GNUNET_SET_Status status)
 {
   struct ConsensusPeerInformation *cpi = cls;
+  unsigned int remote_idx = cpi - cpi->session->info;
+  unsigned int local_idx = cpi->session->local_peer_idx;
+
+  GNUNET_assert ((cpi == cpi->session->partner_outgoing) ||
+                 (cpi == cpi->session->partner_incoming));
 
   switch (status)
   {
     case GNUNET_SET_STATUS_OK:
-      GNUNET_log (GNUNET_ERROR_TYPE_INFO, "set result: element\n");
+      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "P%u: set result from P%u: element\n",
+                  local_idx, remote_idx);
       break;
     case GNUNET_SET_STATUS_FAILURE:
-      GNUNET_log (GNUNET_ERROR_TYPE_INFO, "set result: failure\n");
-      break;
+      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "P%u: set result from P%u: failure\n",
+                  local_idx, remote_idx);
+      cpi->set_op = NULL;
+      return;
     case GNUNET_SET_STATUS_HALF_DONE:
     case GNUNET_SET_STATUS_DONE:
-      GNUNET_log (GNUNET_ERROR_TYPE_INFO, "set result: done\n");
+      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "P%u: set result from P%u: done\n",
+                  local_idx, remote_idx);
       cpi->exp_subround_finished = GNUNET_YES;
+      cpi->set_op = NULL;
       if (have_exp_subround_finished (cpi->session) == GNUNET_YES)
+      {
+        GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "P%u: all reconciliations of subround done\n",
+                    local_idx);
         subround_over (cpi->session, NULL);
+      }
+      else
+      {
+        GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "P%u: waiting for further set results\n",
+                    local_idx);
+      }
       return;
     default:
       GNUNET_break (0);
@@ -484,6 +617,39 @@ set_result_cb (void *cls,
 }
 
 
+/**
+ * Compare the round the session is in with the round of the given context message.
+ *
+ * @param session a consensus session
+ * @param ri a round context message
+ * @return 0 if it's the same round, -1 if the session is in an earlier round,
+ *         1 if the session is in a later round
+ */
+static int
+rounds_compare (struct ConsensusSession *session,
+                struct RoundInfo* ri)
+{
+  if (session->current_round < ri->round)
+    return -1;
+  if (session->current_round > ri->round)
+    return 1;
+  if (session->current_round == CONSENSUS_ROUND_EXCHANGE)
+  {
+    if (session->exp_round < ri->exp_round)
+      return -1;
+    if (session->exp_round > ri->exp_round)
+      return 1;
+    if (session->exp_subround < ri->exp_subround)
+      return -1;
+    if (session->exp_subround < ri->exp_subround)
+      return 1;
+    return 0;
+  }
+  /* comparing rounds when we are not in a exp round */
+  GNUNET_assert (0);
+}
+
+
 /**
  * Do the next subround in the exp-scheme.
  * This function can be invoked as a timeout task, or called manually (tc will be NULL then).
@@ -508,6 +674,13 @@ subround_over (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
     GNUNET_SCHEDULER_cancel (session->round_timeout_tid);
     session->round_timeout_tid = GNUNET_SCHEDULER_NO_TASK;
   }
+
+  if (session->exp_round >= NUM_EXP_ROUNDS)
+  {
+    round_over (session, NULL);
+    return;
+  }
+
   if (session->exp_round == 0)
   {
     /* initialize everything for the log-rounds */
@@ -515,8 +688,10 @@ subround_over (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
     session->exp_subround = 0;
     if (NULL == session->shuffle)
       session->shuffle = GNUNET_malloc ((sizeof (int)) * session->num_peers);
+    if (NULL == session->shuffle_inv)
+      session->shuffle_inv = GNUNET_malloc ((sizeof (int)) * session->num_peers);
     for (i = 0; i < session->num_peers; i++)
-      session->shuffle[i] = i;
+      session->shuffle[i] = session->shuffle_inv[i] = i;
   }
   else if (session->exp_subround + 1 >= (int) ceil (log2 (session->num_peers)))
   {
@@ -525,7 +700,7 @@ subround_over (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
     session->exp_subround = 0;
     shuffle (session);
   }
-  else 
+  else
   {
     session->exp_subround++;
   }
@@ -533,6 +708,10 @@ subround_over (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
   /* determine the incoming and outgoing partner */
   find_partners (session);
 
+  GNUNET_assert (session->partner_outgoing != &session->info[session->local_peer_idx]);
+  GNUNET_assert (session->partner_incoming != &session->info[session->local_peer_idx]);
+
+  /* initiate set operation with the outgoing partner */
   if (NULL != session->partner_outgoing)
   {
     struct GNUNET_CONSENSUS_RoundContextMessage *msg;
@@ -551,11 +730,36 @@ subround_over (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
         GNUNET_SET_prepare (&session->partner_outgoing->peer_id,
                             &session->global_id,
                             (struct GNUNET_MessageHeader *) msg,
-                            0, /* FIXME */
+                            0, /* FIXME: salt */
                             GNUNET_SET_RESULT_ADDED,
                             set_result_cb, session->partner_outgoing);
+    GNUNET_free (msg);
     GNUNET_SET_commit (session->partner_outgoing->set_op, session->element_set);
-    session->partner_outgoing->set_op_commited = GNUNET_YES;
+  }
+
+  /* commit to the delayed set operation */
+  if ((NULL != session->partner_incoming) && (NULL != session->partner_incoming->delayed_set_op))
+  {
+    int cmp = rounds_compare (session, &session->partner_incoming->delayed_round_info);
+
+    if (NULL != session->partner_incoming->set_op)
+    {
+      GNUNET_SET_operation_cancel (session->partner_incoming->set_op);
+      session->partner_incoming->set_op = NULL;
+    }
+    if (cmp == 0)
+    {
+      GNUNET_SET_commit (session->partner_incoming->delayed_set_op, session->element_set);
+      session->partner_incoming->set_op = session->partner_incoming->delayed_set_op;
+      session->partner_incoming->delayed_set_op = NULL;
+      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "P%d resumed delayed round with P%d\n",
+                  session->local_peer_idx, (int) (session->partner_incoming - session->info));
+    }
+    else
+    {
+      /* this should not happen -- a round has been skipped! */
+      GNUNET_break_op (0);
+    }
   }
 
 #ifdef GNUNET_EXTRA_LOGGING
@@ -570,7 +774,7 @@ subround_over (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
       in = -1;
     else
       in = (int) (session->partner_incoming - session->info);
-    GNUNET_log (GNUNET_ERROR_TYPE_INFO, "P%u: doing exp-round, r=%d, sub=%d, in: %d, out: %d\n", session->local_peer_idx,
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "P%u: doing exp-round, r=%d, sub=%d, in: %d, out: %d\n", session->local_peer_idx,
                 session->exp_round, session->exp_subround, in, out);
   }
 #endif /* GNUNET_EXTRA_LOGGING */
@@ -606,17 +810,20 @@ get_peer_idx (const struct GNUNET_PeerIdentity *peer, const struct ConsensusSess
  * @param session_id local id of the consensus session
  */
 static void
-compute_global_id (struct ConsensusSession *session, const struct GNUNET_HashCode *session_id)
+compute_global_id (struct ConsensusSession *session,
+                  const struct GNUNET_HashCode *session_id)
 {
   int i;
   struct GNUNET_HashCode tmp;
+  struct GNUNET_HashCode phash;
 
   /* FIXME: use kdf? */
 
   session->global_id = *session_id;
   for (i = 0; i < session->num_peers; ++i)
   {
-    GNUNET_CRYPTO_hash_xor (&session->global_id, &session->info[i].peer_id.hashPubKey, &tmp);
+    GNUNET_CRYPTO_hash (&session->info[i].peer_id, sizeof (struct GNUNET_PeerIdentity), &phash);
+    GNUNET_CRYPTO_hash_xor (&session->global_id, &phash, &tmp);
     session->global_id = tmp;
     GNUNET_CRYPTO_hash (&session->global_id, sizeof (struct GNUNET_PeerIdentity), &tmp);
     session->global_id = tmp;
@@ -658,7 +865,7 @@ initialize_session_peer_list (struct ConsensusSession *session,
 
   /* peers in the join message, may or may not include the local peer */
   listed_peers = ntohl (join_msg->num_peers);
-  
+
   session->num_peers = listed_peers;
 
   msg_peers = (struct GNUNET_PeerIdentity *) &join_msg[1];
@@ -702,6 +909,7 @@ initialize_session_peer_list (struct ConsensusSession *session,
  * Called when another peer wants to do a set operation with the
  * local peer.
  *
+ * @param cls closure
  * @param other_peer the other peer
  * @param context_msg message with application specific information from
  *        the other peer
@@ -721,7 +929,10 @@ set_listen_cb (void *cls,
   struct ConsensusSession *session = cls;
   struct GNUNET_CONSENSUS_RoundContextMessage *msg = (struct GNUNET_CONSENSUS_RoundContextMessage *) context_msg;
   struct ConsensusPeerInformation *cpi;
+  struct GNUNET_SET_OperationHandle *set_op;
+  struct RoundInfo round_info;
   int index;
+  int cmp;
 
   if (NULL == context_msg)
   {
@@ -731,51 +942,61 @@ set_listen_cb (void *cls,
 
   index = get_peer_idx (other_peer, session);
 
-  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "result from %s\n", GNUNET_h2s (&other_peer->hashPubKey));
-
   if (index < 0)
   {
     GNUNET_break_op (0);
     return;
   }
 
+  round_info.round = ntohl (msg->round);
+  round_info.exp_round = ntohl (msg->exp_round);
+  round_info.exp_subround = ntohl (msg->exp_subround);
+
   cpi = &session->info[index];
 
-  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "P%d got result from P%d\n", session->local_peer_idx, index);
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "P%d got set request from P%d\n", session->local_peer_idx, index);
 
   switch (session->current_round)
   {
+    case CONSENSUS_ROUND_BEGIN:
+      /* we're in the begin round, so requests for the exchange round may
+       * come in, they will be delayed for now! */
     case CONSENSUS_ROUND_EXCHANGE:
-      if (ntohl (msg->round) != CONSENSUS_ROUND_EXCHANGE)
-      {
-        GNUNET_break_op (0);
-        return;
-      }
-      if (ntohl (msg->exp_round) < session->exp_round)
-      {
-        GNUNET_break_op (0);
-        return;
-      }
-      if (ntohl (msg->exp_subround) < session->exp_subround)
+      cmp = rounds_compare (session, &round_info);
+      if (cmp > 0)
       {
+        /* the other peer is too late */
         GNUNET_break_op (0);
         return;
       }
+      /* kill old request, if any. this is legal,
+       * as the other peer would not make a new request if it would want to
+       * complete the old one! */
       if (NULL != cpi->set_op)
+      {
         GNUNET_SET_operation_cancel (cpi->set_op);
-      cpi->set_op = GNUNET_SET_accept (request, GNUNET_SET_RESULT_ADDED,
+        cpi->set_op = NULL;
+      }
+      set_op = GNUNET_SET_accept (request, GNUNET_SET_RESULT_ADDED,
                                        set_result_cb, &session->info[index]);
-      if (ntohl (msg->exp_subround) == session->exp_subround)
+      if (cmp == 0)
       {
-        cpi->set_op_commited = GNUNET_YES;
-        GNUNET_SET_commit (cpi->set_op, session->element_set);
+        cpi->set_op = set_op;
+        GNUNET_SET_commit (set_op, session->element_set);
+        GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "P%d commited to set request from P%d\n", session->local_peer_idx, index);
       }
       else
       {
-        cpi->set_op_commited = GNUNET_NO;
+        /* if there's a exp subround running, mark it as finished, as the set op has been canceled! */
+        cpi->delayed_set_op = set_op;
+        cpi->delayed_round_info = round_info;
+        cpi->exp_subround_finished = GNUNET_YES;
+        GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "P%d delaying set request from P%d\n", session->local_peer_idx, index);
       }
       break;
     default:
+      GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "P%d got unexpected set request in round %d from P%d\n",
+                  session->local_peer_idx, session->current_round, index);
       GNUNET_break_op (0);
       return;
   }
@@ -786,6 +1007,7 @@ set_listen_cb (void *cls,
  * Initialize the session, continue receiving messages from the owning client
  *
  * @param session the session to initialize
+ * @param join_msg the join message from the client
  */
 static void
 initialize_session (struct ConsensusSession *session,
@@ -794,7 +1016,7 @@ initialize_session (struct ConsensusSession *session,
   struct ConsensusSession *other_session;
 
   initialize_session_peer_list (session, join_msg);
-  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "session with %u peers\n", session->num_peers);
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "session with %u peers\n", session->num_peers);
   compute_global_id (session, &join_msg->session_id);
 
   /* check if some local client already owns the session.
@@ -803,7 +1025,7 @@ initialize_session (struct ConsensusSession *session,
   other_session = sessions_head;
   while (NULL != other_session)
   {
-    if ((other_session != session) && 
+    if ((other_session != session) &&
         (0 == GNUNET_CRYPTO_hash_cmp (&session->global_id, &other_session->global_id)))
     {
       if (CONSENSUS_ROUND_FINISH != other_session->current_round)
@@ -824,8 +1046,8 @@ initialize_session (struct ConsensusSession *session,
   session->set_listener = GNUNET_SET_listen (cfg, GNUNET_SET_OPERATION_UNION,
                                              &session->global_id,
                                              set_listen_cb, session);
-  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "%d is the local peer\n", session->local_peer_idx);
-  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "session %s initialized\n", GNUNET_h2s (&session->global_id));
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "%d is the local peer\n", session->local_peer_idx);
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "session %s initialized\n", GNUNET_h2s (&session->global_id));
 }
 
 
@@ -859,7 +1081,7 @@ client_join (void *cls,
 {
   struct ConsensusSession *session;
 
-  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "join message sent by client\n");
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "join message sent by client\n");
 
   session = get_session_by_client (client);
   if (NULL != session)
@@ -870,12 +1092,12 @@ client_join (void *cls,
   }
   session = GNUNET_new (struct ConsensusSession);
   session->client = client;
-  GNUNET_SERVER_client_keep (client);
+  session->client_mq = GNUNET_MQ_queue_for_server_client (client);
   GNUNET_CONTAINER_DLL_insert (sessions_head, sessions_tail, session);
   initialize_session (session, (struct GNUNET_CONSENSUS_JoinMessage *) m);
   GNUNET_SERVER_receive_done (client, GNUNET_OK);
 
-  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "join done\n");
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "join done\n");
 }
 
 
@@ -926,9 +1148,10 @@ client_insert (void *cls,
   memcpy (&element[1], &msg[1], element_size);
   element->data = &element[1];
   GNUNET_SET_add_element (session->element_set, element, NULL, NULL);
+  GNUNET_free (element);
   GNUNET_SERVER_receive_done (client, GNUNET_OK);
 
-  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "P%u: element added\n", session->local_peer_idx);
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "P%u: element added\n", session->local_peer_idx);
 }
 
 
@@ -948,7 +1171,7 @@ client_conclude (void *cls,
   struct GNUNET_CONSENSUS_ConcludeMessage *cmsg;
 
 
-  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "conclude requested\n");
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "conclude requested\n");
   cmsg = (struct GNUNET_CONSENSUS_ConcludeMessage *) message;
   session = get_session_by_client (client);
   if (NULL == session)
@@ -966,7 +1189,8 @@ client_conclude (void *cls,
   }
   if (session->num_peers <= 1)
   {
-    //send_client_conclude_done (session);
+    session->current_round = CONSENSUS_ROUND_FINISH;
+    GNUNET_SET_iterate (session->element_set, send_to_client_iter, session);
   }
   else
   {
@@ -993,7 +1217,7 @@ shutdown_task (void *cls,
   while (NULL != sessions_head)
     destroy_session (sessions_head);
 
-  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "handled shutdown request\n");
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "handled shutdown request\n");
 }
 
 
@@ -1041,7 +1265,7 @@ run (void *cls, struct GNUNET_SERVER_Handle *server,
 
   cfg = c;
   srv = server;
-  if (GNUNET_OK != GNUNET_CRYPTO_get_host_identity (cfg, &my_peer))
+  if (GNUNET_OK != GNUNET_CRYPTO_get_peer_identity (cfg, &my_peer))
   {
     GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "could not retrieve host identity\n");
     GNUNET_break (0);
@@ -1067,7 +1291,7 @@ main (int argc, char *const *argv)
 {
   int ret;
   ret = GNUNET_SERVICE_run (argc, argv, "consensus", GNUNET_SERVICE_OPTION_NONE, &run, NULL);
-  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "exit\n");
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "exit (%d)\n", GNUNET_OK != ret);
   return (GNUNET_OK == ret) ? 0 : 1;
 }