4 files changed, 60 insertions, 9 deletions

diff --git a/src/consensus/gnunet-service-consensus.c b/src/consensus/gnunet-service-consensus.c
index 472e96706..567480e58 100644
--- a/src/consensus/gnunet-service-consensus.c
+++ b/src/consensus/gnunet-service-consensus.c
@@ -293,7 +293,11 @@ destroy_session (struct ConsensusSession *session)
     {
       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;
@@ -315,17 +319,27 @@ 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_MQ_Envelope *ev;
     struct GNUNET_CONSENSUS_ElementMessage *m;
 
+    GNUNET_log (GNUNET_ERROR_TYPE_INFO, "P%d: got element for client\n",
+                session->local_peer_idx);
+
     ev = GNUNET_MQ_msg (m, 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_INFO, "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;
 }
 
@@ -368,6 +382,7 @@ round_over (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
                   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);
   }
@@ -425,9 +440,12 @@ find_partners (struct ConsensusSession *session)
   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;
+  partner_idx = (my_idx + arc) % session->num_peers;
   largest_arc = 1;
   while (largest_arc < session->num_peers)
     largest_arc <<= 1;
@@ -435,7 +453,7 @@ find_partners (struct ConsensusSession *session)
 
   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]];
@@ -443,10 +461,10 @@ find_partners (struct ConsensusSession *session)
      * 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;
+      ghost_partner_idx = (my_idx - arc) % session->num_peers;
       /* platform dependent; modulo sometimes returns negative values */
       if (ghost_partner_idx < 0)
         ghost_partner_idx += arc;
@@ -487,11 +505,13 @@ set_result_cb (void *cls,
       break;
     case GNUNET_SET_STATUS_FAILURE:
       GNUNET_log (GNUNET_ERROR_TYPE_INFO, "set result: failure\n");
+      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");
       cpi->exp_subround_finished = GNUNET_YES;
+      cpi->set_op = NULL;
       if (have_exp_subround_finished (cpi->session) == GNUNET_YES)
         subround_over (cpi->session, NULL);
       return;
@@ -536,6 +556,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 */
@@ -752,6 +779,12 @@ set_listen_cb (void *cls,
   struct ConsensusPeerInformation *cpi;
   int index;
 
+  /* FIXME: should this even happen? */
+  /*
+  if (NULL == request)
+    return;
+  */
+
   if (NULL == context_msg)
   {
     GNUNET_break_op (0);
@@ -900,6 +933,7 @@ client_join (void *cls,
   }
   session = GNUNET_new (struct ConsensusSession);
   session->client = client;
+  session->client_mq = GNUNET_MQ_queue_for_server_client (client);
   GNUNET_SERVER_client_keep (client);
   GNUNET_CONTAINER_DLL_insert (sessions_head, sessions_tail, session);
   initialize_session (session, (struct GNUNET_CONSENSUS_JoinMessage *) m);
diff --git a/src/consensus/test_consensus.conf b/src/consensus/test_consensus.conf
index 37facf84e..d0190f8ba 100644
--- a/src/consensus/test_consensus.conf
+++ b/src/consensus/test_consensus.conf
@@ -28,3 +28,18 @@ PREFIX = valgrind
 
 [testbed]
 OVERLAY_TOPOLOGY = CLIQUE
+
+[hostlist]
+SERVERS = 
+
+
+[nat]
+# Use addresses from the local network interfaces (inluding loopback, but also others)
+USE_LOCALADDR = YES
+
+# Disable IPv6 support
+DISABLEV6 = NO
+
+# Do we use addresses from localhost address ranges? (::1, 127.0.0.0/8)
+RETURN_LOCAL_ADDRESSES = YES
+
diff --git a/src/set/gnunet-service-set_union.c b/src/set/gnunet-service-set_union.c
index 8a6bfa448..644f975b6 100644
--- a/src/set/gnunet-service-set_union.c
+++ b/src/set/gnunet-service-set_union.c
@@ -873,8 +873,9 @@ decode_and_send (struct OperationState *eo)
       last_key = key;
 
     res = ibf_decode (diff_ibf, &side, &key);
-    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "decoded ibf key %lx\n",
-                key.key_val);
+    if (res == GNUNET_OK)
+      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "decoded ibf key %lx\n",
+                  key.key_val);
     num_decoded += 1;
     if (num_decoded > diff_ibf->size || (num_decoded > 1 && last_key.key_val == key.key_val))
       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "detected cyclic ibf (decoded %u/%u)\n",
diff --git a/src/set/set_api.c b/src/set/set_api.c
index c2e88e65f..b2e62cb4d 100644
--- a/src/set/set_api.c
+++ b/src/set/set_api.c
@@ -564,8 +564,9 @@ GNUNET_SET_listen (const struct GNUNET_CONFIGURATION_Handle *cfg,
 void
 GNUNET_SET_listen_cancel (struct GNUNET_SET_ListenHandle *lh)
 {
-  GNUNET_CLIENT_disconnect (lh->client);
+  LOG (GNUNET_ERROR_TYPE_DEBUG, "canceling listener\n");
   GNUNET_MQ_destroy (lh->mq);
+  GNUNET_CLIENT_disconnect (lh->client);
   GNUNET_free (lh);
 }