Add missing include
[oweals/gnunet.git] / src / consensus / gnunet-service-consensus.c
index cb7ab2c6835f8477f0fa7aef88b91e80df6b6b0c..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"
@@ -148,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.
    */
@@ -177,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.
    */
@@ -283,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) &&
@@ -297,7 +300,7 @@ 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
@@ -331,6 +334,11 @@ destroy_session (struct ConsensusSession *session)
     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++)
@@ -369,7 +377,7 @@ send_to_client_iter (void *cls,
   {
     struct GNUNET_CONSENSUS_ElementMessage *m;
 
-    GNUNET_log (GNUNET_ERROR_TYPE_INFO, "P%d: got element for client\n",
+    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);
@@ -379,7 +387,7 @@ send_to_client_iter (void *cls,
   }
   else
   {
-    GNUNET_log (GNUNET_ERROR_TYPE_INFO, "P%d: finished iterating elements for client\n",
+    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);
@@ -396,7 +404,7 @@ send_to_client_iter (void *cls,
  * @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;
@@ -406,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)
   {
@@ -422,7 +430,7 @@ round_over (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
       subround_over (session, NULL);
       break;
     case CONSENSUS_ROUND_EXCHANGE:
-      GNUNET_log (GNUNET_ERROR_TYPE_INFO, "P%d: finished, sending elements to client\n",
+      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);
@@ -448,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);
@@ -466,6 +476,10 @@ shuffle (struct ConsensusSession *session)
     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;
 }
 
 
@@ -493,14 +507,15 @@ find_partners (struct ConsensusSession *session)
   while (largest_arc < session->num_peers)
     largest_arc <<= 1;
   num_ghosts = largest_arc - session->num_peers;
-
-  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "num ghosts: %d\n", num_ghosts);
+  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);
 
   if (0 == (my_idx & arc))
   {
     /* we are outgoing */
     partner_idx = (my_idx + arc) % session->num_peers;
-    session->partner_outgoing = &session->info[session->shuffle[partner_idx]];
+    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
@@ -509,25 +524,29 @@ find_partners (struct ConsensusSession *session)
     if (my_idx < num_ghosts)
     {
       int ghost_partner_idx;
-      GNUNET_log (GNUNET_ERROR_TYPE_INFO, "my index %d, arc %d, peers %u\n", my_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;
-      GNUNET_log (GNUNET_ERROR_TYPE_INFO, "ghost partner is before %d\n", ghost_partner_idx);
       /* platform dependent; modulo sometimes returns negative values */
       if (ghost_partner_idx < 0)
         ghost_partner_idx += session->num_peers;
-      GNUNET_log (GNUNET_ERROR_TYPE_INFO, "ghost partner is after %d\n", ghost_partner_idx);
-      session->partner_incoming = &session->info[session->shuffle[ghost_partner_idx]];
-      session->partner_incoming->exp_subround_finished = GNUNET_NO;
-      return;
+      /* 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;
   }
+  /* 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[partner_idx]];
+  session->partner_incoming = &session->info[session->shuffle_inv[partner_idx]];
   session->partner_incoming->exp_subround_finished = GNUNET_NO;
 }
 
@@ -540,7 +559,7 @@ 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)
@@ -555,29 +574,29 @@ set_result_cb (void *cls,
   switch (status)
   {
     case GNUNET_SET_STATUS_OK:
-      GNUNET_log (GNUNET_ERROR_TYPE_INFO, "P%u: set result from P%u: 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, "P%u: set result from P%u: failure\n",
+      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, "P%u: set result from P%u: 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_INFO, "P%u: all reconciliations of subround done\n",
+        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_INFO, "P%u: waiting for further set results\n",
+        GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "P%u: waiting for further set results\n",
                     local_idx);
       }
       return;
@@ -602,7 +621,7 @@ 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 round a round context message
+ * @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
  */
@@ -655,7 +674,7 @@ 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);
@@ -669,17 +688,19 @@ 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)))
   {
     /* subrounds done, start new log-round */
     session->exp_round++;
     session->exp_subround = 0;
-    //shuffle (session);
+    shuffle (session);
   }
-  else 
+  else
   {
     session->exp_subround++;
   }
@@ -731,7 +752,7 @@ subround_over (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
       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_INFO, "P%d resumed delayed round with P%d\n",
+      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
@@ -753,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 */
@@ -789,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;
@@ -841,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];
@@ -930,10 +954,13 @@ set_listen_cb (void *cls,
 
   cpi = &session->info[index];
 
-  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "P%d got set request 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:
       cmp = rounds_compare (session, &round_info);
       if (cmp > 0)
@@ -956,7 +983,7 @@ set_listen_cb (void *cls,
       {
         cpi->set_op = set_op;
         GNUNET_SET_commit (set_op, session->element_set);
-        GNUNET_log (GNUNET_ERROR_TYPE_INFO, "P%d commited to set request from P%d\n", session->local_peer_idx, index);
+        GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "P%d commited to set request from P%d\n", session->local_peer_idx, index);
       }
       else
       {
@@ -964,10 +991,12 @@ set_listen_cb (void *cls,
         cpi->delayed_set_op = set_op;
         cpi->delayed_round_info = round_info;
         cpi->exp_subround_finished = GNUNET_YES;
-        GNUNET_log (GNUNET_ERROR_TYPE_INFO, "P%d delaying set request from P%d\n", session->local_peer_idx, index);
+        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;
   }
@@ -987,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.
@@ -996,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)
@@ -1017,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));
 }
 
 
@@ -1052,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)
@@ -1068,7 +1097,7 @@ client_join (void *cls,
   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");
 }
 
 
@@ -1122,7 +1151,7 @@ client_insert (void *cls,
   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);
 }
 
 
@@ -1142,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)
@@ -1160,8 +1189,8 @@ client_conclude (void *cls,
   }
   if (session->num_peers <= 1)
   {
-    /* FIXME: what to do here? */
-    //send_client_conclude_done (session);
+    session->current_round = CONSENSUS_ROUND_FINISH;
+    GNUNET_SET_iterate (session->element_set, send_to_client_iter, session);
   }
   else
   {
@@ -1188,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");
 }
 
 
@@ -1211,7 +1240,7 @@ handle_client_disconnect (void *cls, struct GNUNET_SERVER_Client *client)
       (CONSENSUS_ROUND_FINISH == session->current_round))
     destroy_session (session);
   else
-    GNUNET_log (GNUNET_ERROR_TYPE_INFO, "client disconnected, but waiting for consensus to finish\n");
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "client disconnected, but waiting for consensus to finish\n");
 }
 
 
@@ -1236,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);