- gnunet-consensus now profiling tool

- work on service implementation, not working yet

11 files changed, 1383 insertions, 567 deletions

diff --git a/src/consensus/Makefile.am b/src/consensus/Makefile.am
index 1beaa0c62..f5a5c5cdc 100644
--- a/src/consensus/Makefile.am
+++ b/src/consensus/Makefile.am
@@ -31,6 +31,7 @@ gnunet_consensus_SOURCES = \
 gnunet_consensus_LDADD = \
   $(top_builddir)/src/util/libgnunetutil.la \
   $(top_builddir)/src/consensus/libgnunetconsensus.la \
+  $(top_builddir)/src/testbed/libgnunettestbed.la \
   $(GN_LIBINTL)
 gnunet_consensus_DEPENDENCIES = \
   libgnunetconsensus.la
@@ -53,10 +54,12 @@ gnunet_consensus_ibf_LDADD = \
   $(GN_LIBINTL)
 
 gnunet_service_consensus_SOURCES = \
- gnunet-service-consensus.c
+ gnunet-service-consensus.c \
+ ibf.c
 gnunet_service_consensus_LDADD = \
   $(top_builddir)/src/util/libgnunetutil.la \
   $(top_builddir)/src/core/libgnunetcore.la \
+  $(top_builddir)/src/stream/libgnunetstream.la \
   $(top_builddir)/src/mesh/libgnunetmesh.la \
   $(GN_LIBINTL)
 
diff --git a/src/consensus/consensus.h b/src/consensus/consensus.h
index 75b90b0f9..8436364b6 100644
--- a/src/consensus/consensus.h
+++ b/src/consensus/consensus.h
@@ -71,6 +71,10 @@ struct GNUNET_CONSENSUS_ConcludeDoneMessage
    */
   struct GNUNET_MessageHeader header;
 
+  uint32_t group_id;
+
+  uint32_t num_elements;
+
   uint16_t num_peers;
 
   /** PeerIdentity[num_peers] */
diff --git a/src/consensus/consensus_api.c b/src/consensus/consensus_api.c
index 25c76b358..5c0494254 100644
--- a/src/consensus/consensus_api.c
+++ b/src/consensus/consensus_api.c
@@ -54,11 +54,6 @@ struct QueuedMessage
   struct GNUNET_MessageHeader *msg;
 
   /**
-   * Size of the message in msg.
-   */
-  size_t size;
-
-  /**
    * Will be called after transmit, if not NULL
    */
   GNUNET_CONSENSUS_InsertDoneCallback idc;
@@ -154,7 +149,7 @@ struct GNUNET_CONSENSUS_Handle
  * @param consensus consensus handle
  */
 static void
-schedule_transmit (struct GNUNET_CONSENSUS_Handle *consensus);
+send_next (struct GNUNET_CONSENSUS_Handle *consensus);
 
 
 /**
@@ -168,16 +163,17 @@ schedule_transmit (struct GNUNET_CONSENSUS_Handle *consensus);
  * @param buf where the callee should write the message
  * @return number of bytes written to buf
  */
-static size_t transmit_queued (void *cls, size_t size,
-                               void *buf)
+static size_t
+transmit_queued (void *cls, size_t size,
+                 void *buf)
 {
   struct GNUNET_CONSENSUS_Handle *consensus;
   struct QueuedMessage *qmsg;
-  size_t ret_size;
-
-  printf("transmitting queued\n");
+  size_t msg_size;
 
   consensus = (struct GNUNET_CONSENSUS_Handle *) cls;
+  consensus->th = NULL;
+
   qmsg = consensus->messages_head;
   GNUNET_CONTAINER_DLL_remove (consensus->messages_head, consensus->messages_tail, qmsg);
   GNUNET_assert (qmsg);
@@ -188,10 +184,14 @@ static size_t transmit_queued (void *cls, size_t size,
     {
       qmsg->idc (qmsg->idc_cls, GNUNET_YES);
     }
+    return 0;
   }
 
-  memcpy (buf, qmsg->msg, qmsg->size);
-  ret_size = qmsg->size;
+  msg_size = ntohs (qmsg->msg->size);
+
+  GNUNET_assert (size >= msg_size);
+
+  memcpy (buf, qmsg->msg, msg_size);
   if (NULL != qmsg->idc)
   {
     qmsg->idc (qmsg->idc_cls, GNUNET_YES);
@@ -199,9 +199,9 @@ static size_t transmit_queued (void *cls, size_t size,
   GNUNET_free (qmsg->msg);
   GNUNET_free (qmsg);
 
-  schedule_transmit (consensus);
+  send_next (consensus);
 
-  return ret_size;
+  return msg_size;
 }
 
 
@@ -211,7 +211,7 @@ static size_t transmit_queued (void *cls, size_t size,
  * @param consensus consensus handle
  */
 static void
-schedule_transmit (struct GNUNET_CONSENSUS_Handle *consensus)
+send_next (struct GNUNET_CONSENSUS_Handle *consensus)
 {
   if (NULL != consensus->th)
     return;
@@ -219,9 +219,10 @@ schedule_transmit (struct GNUNET_CONSENSUS_Handle *consensus)
   if (NULL != consensus->messages_head)
   {
     LOG (GNUNET_ERROR_TYPE_INFO, "scheduling queued\n");
-    GNUNET_CLIENT_notify_transmit_ready (consensus->client, consensus->messages_head->size, 
-                                         GNUNET_TIME_UNIT_FOREVER_REL,
-                                         GNUNET_NO, &transmit_queued, consensus);
+    consensus->th = 
+        GNUNET_CLIENT_notify_transmit_ready (consensus->client, ntohs (consensus->messages_head->msg->size),
+                                             GNUNET_TIME_UNIT_FOREVER_REL,
+                                             GNUNET_NO, &transmit_queued, consensus);
   }
 }
 
@@ -270,8 +271,7 @@ handle_conclude_done (struct GNUNET_CONSENSUS_Handle *consensus,
                      struct GNUNET_CONSENSUS_ConcludeDoneMessage *msg)
 {
   GNUNET_assert (NULL != consensus->conclude_cb);
-  consensus->conclude_cb (consensus->conclude_cls,
-                         0, NULL);
+  consensus->conclude_cb (consensus->conclude_cls, NULL);
   consensus->conclude_cb = NULL;
 }
 
@@ -356,7 +356,7 @@ transmit_join (void *cls, size_t size, void *buf)
            consensus->peers,
            consensus->num_peers * sizeof (struct GNUNET_PeerIdentity));
 
-  schedule_transmit (consensus);
+  send_next (consensus);
 
   GNUNET_CLIENT_receive (consensus->client, &message_handler, consensus,
                          GNUNET_TIME_UNIT_FOREVER_REL);
@@ -454,13 +454,12 @@ GNUNET_CONSENSUS_insert (struct GNUNET_CONSENSUS_Handle *consensus,
 
   qmsg = GNUNET_malloc (sizeof (struct QueuedMessage));
   qmsg->msg = (struct GNUNET_MessageHeader *) element_msg;
-  qmsg->size = element_msg_size;
   qmsg->idc = idc;
   qmsg->idc_cls = idc_cls;
 
   GNUNET_CONTAINER_DLL_insert_tail (consensus->messages_head, consensus->messages_tail, qmsg);
 
-  schedule_transmit (consensus);
+  send_next (consensus);
 }
 
 
@@ -500,11 +499,10 @@ GNUNET_CONSENSUS_conclude (struct GNUNET_CONSENSUS_Handle *consensus,
 
   qmsg = GNUNET_malloc (sizeof (struct QueuedMessage));
   qmsg->msg = (struct GNUNET_MessageHeader *) conclude_msg;
-  qmsg->size = sizeof (struct GNUNET_CONSENSUS_ConcludeMessage);
 
   GNUNET_CONTAINER_DLL_insert_tail (consensus->messages_head, consensus->messages_tail, qmsg);
 
-  schedule_transmit (consensus);
+  send_next (consensus);
 }
 
 
diff --git a/src/consensus/consensus_protocol.h b/src/consensus/consensus_protocol.h
new file mode 100644
index 000000000..105708ee9
--- /dev/null
+++ b/src/consensus/consensus_protocol.h
@@ -0,0 +1,71 @@
+/*
+      This file is part of GNUnet
+      (C) 2012 Christian Grothoff (and other contributing authors)
+
+      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
+      option) any later version.
+
+      GNUnet is distributed in the hope that it will be useful, but
+      WITHOUT ANY WARRANTY; without even the implied warranty of
+      MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+      General Public License for more details.
+
+      You should have received a copy of the GNU General Public License
+      along with GNUnet; see the file COPYING.  If not, write to the
+      Free Software Foundation, Inc., 59 Temple Place - Suite 330,
+      Boston, MA 02111-1307, USA.
+*/
+
+
+/**
+ * @file consensus/consensus_protocol.h
+ * @brief p2p message definitions for consensus
+ * @author Florian Dold
+ */
+
+#ifndef GNUNET_CONSENSUS_PROTOCOL_H
+#define GNUNET_CONSENSUS_PROTOCOL_H
+
+#include "platform.h"
+#include "gnunet_common.h"
+#include "gnunet_protocols.h"
+
+
+GNUNET_NETWORK_STRUCT_BEGIN
+
+struct StrataMessage
+{
+  struct GNUNET_MessageHeader header;
+  /**
+   * Number of strata in this estimator.
+   */
+  uint16_t num_strata;
+  /* struct GNUNET_HashCode hash_buckets[ibf_size*num_strata] */
+  /* struct GNUNET_HashCode id_buckets[ibf_size*num_strata] */
+  /* uint8_t count_buckets[ibf_size*num_strata] */
+};
+
+struct DifferenceDigest
+{
+
+  struct GNUNET_MessageHeader header;
+};
+
+struct Element
+{
+  struct GNUNET_MessageHeader header;
+};
+
+struct ConsensusHello
+{
+  struct GNUNET_MessageHeader header;
+  struct GNUNET_HashCode global_id;
+  uint8_t round;
+};
+
+
+GNUNET_NETWORK_STRUCT_END
+
+#endif
diff --git a/src/consensus/gnunet-consensus-start-peers.c b/src/consensus/gnunet-consensus-start-peers.c
index 8b516393f..fb7f047ae 100644
--- a/src/consensus/gnunet-consensus-start-peers.c
+++ b/src/consensus/gnunet-consensus-start-peers.c
@@ -147,9 +147,6 @@ run (void *cls, char *const *args, const char *cfgfile,
                                   NULL,
                                   test_master,
                                   NULL);
-
-
-  printf("hello there!\n");
 }
 
 
diff --git a/src/consensus/gnunet-consensus.c b/src/consensus/gnunet-consensus.c
index cd267f5ec..c8a5593f1 100644
--- a/src/consensus/gnunet-consensus.c
+++ b/src/consensus/gnunet-consensus.c
@@ -20,210 +20,283 @@
 
 /**
  * @file consensus/gnunet-consensus.c
- * @brief 
+ * @brief profiling tool for gnunet-consensus
  * @author Florian Dold
  */
 #include "platform.h"
+#include "gnunet_common.h"
 #include "gnunet_util_lib.h"
 #include "gnunet_consensus_service.h"
+#include "gnunet_testbed_service.h"
 
+static unsigned int num_peers = 2;
 
+static unsigned int replication = 1;
 
-/**
- * Handle to the consensus service
- */
-static struct GNUNET_CONSENSUS_Handle *consensus;
-/**
- * Session id
- */
-static char *session_id_str;
+static unsigned int num_values = 5;
 
-/**
- * File handle to STDIN
- */
-static struct GNUNET_DISK_FileHandle *stdin_fh;
+static struct GNUNET_TIME_Relative conclude_timeout;
 
-/**
- * Task for reading from stdin
- */
-static GNUNET_SCHEDULER_TaskIdentifier stdin_tid = GNUNET_SCHEDULER_NO_TASK;
+static struct GNUNET_CONSENSUS_Handle **consensus_handles;
+
+static unsigned int num_connected_handles;
+
+static struct GNUNET_TESTBED_Peer **peers;
+
+static struct GNUNET_PeerIdentity *peer_ids;
 
+static unsigned int num_retrieved_peer_ids;
 
+static struct GNUNET_HashCode session_id;
+
+
+/**
+ * Signature of the event handler function called by the
+ * respective event controller.
+ *
+ * @param cls closure
+ * @param event information about the event
+ */
 static void
-stdin_cb (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc);
+controller_cb(void *cls,
+              const struct GNUNET_TESTBED_EventInformation *event)
+{
+  GNUNET_assert (0);
+}
 
 
 /**
  * Called when a conclusion was successful.
  *
  * @param cls
- * @param num_peers_in_consensus
- * @param peers_in_consensus
+ * @param group
+ * @return GNUNET_YES if more consensus groups should be offered, GNUNET_NO if not
  */
+static int
+conclude_cb (void *cls, const struct GNUNET_CONSENSUS_Group *group)
+{
+  return GNUNET_NO;
+}
+
+
+
 static void
-conclude_cb (void *cls, 
-             unsigned int consensus_group_count,
-             const struct GNUNET_CONSENSUS_Group *groups)
+generate_indices (int *indices)
 {
-  printf("reached conclusion\n");
-  GNUNET_SCHEDULER_shutdown ();
+  int j;
+  j = 0;
+  while (j < replication)
+  {
+    int n;
+    int k;
+    int repeat;
+    n = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, num_peers);
+    repeat = GNUNET_NO;
+    for (k = 0; k < j; k++)
+      if (indices[k] == n)
+      {
+        repeat = GNUNET_YES;
+        break;
+      }
+    if (GNUNET_NO == repeat)
+      indices[j++] = n;
+  }
 }
 
 
 static void
-insert_done_cb (void *cls,
-                int success)
+do_consensus ()
 {
-  struct GNUNET_CONSENSUS_Element *element = cls;
+  int unique_indices[replication];
+  int i;
 
-  GNUNET_free (element);
-  if (GNUNET_YES != success)
+  for (i = 0; i < num_values; i++)
   {
-    printf ("insert failed\n");
-    GNUNET_SCHEDULER_shutdown ();
-    return;
+    int j;
+    struct GNUNET_HashCode *val;
+    struct GNUNET_CONSENSUS_Element *element;
+    generate_indices(unique_indices);
+
+    val = GNUNET_malloc (sizeof *val);
+    GNUNET_CRYPTO_hash_create_random (GNUNET_CRYPTO_QUALITY_WEAK, val);
+
+    element = GNUNET_malloc (sizeof *element);
+    element->data = val;
+    element->size = sizeof *val;
+
+    for (j = 0; j < replication; j++)
+    {
+      int cid;
+      cid = unique_indices[j];
+      GNUNET_CONSENSUS_insert (consensus_handles[cid], element, NULL, NULL);
+    }
   }
-  GNUNET_assert (GNUNET_SCHEDULER_NO_TASK == stdin_tid);
-  stdin_tid = GNUNET_SCHEDULER_add_read_file (GNUNET_TIME_UNIT_FOREVER_REL, stdin_fh,
-                                              &stdin_cb, NULL);    
+
+  for (i = 0; i < num_peers; i++)
+    GNUNET_CONSENSUS_conclude (consensus_handles[i], conclude_timeout, 0, conclude_cb, consensus_handles[i]);
 }
 
 
 /**
- * Called whenever we can read stdin non-blocking 
+ * Callback to be called when a service connect operation is completed
  *
- * @param cls unused
- * @param tc scheduler context 
+ * @param cls the callback closure from functions generating an operation
+ * @param op the operation that has been finished
+ * @param ca_result the service handle returned from GNUNET_TESTBED_ConnectAdapter()
+ * @param emsg error message in case the operation has failed; will be NULL if
+ *          operation has executed successfully.
  */
 static void
-stdin_cb (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
+connect_complete (void *cls,
+                  struct GNUNET_TESTBED_Operation *op,
+                  void *ca_result,
+                  const char *emsg)
 {
-  char buf[1024];
-  char *ret;
-  struct GNUNET_CONSENSUS_Element *element;
-
-  stdin_tid = GNUNET_SCHEDULER_NO_TASK;
-  if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
-    return; /* we're done here */
-  ret = fgets (buf, 1024, stdin);
-  if (NULL == ret)
+  struct GNUNET_CONSENSUS_Handle **chp;
+
+  if (NULL != emsg)
   {
-    if (feof (stdin))
-    {
-      printf ("concluding ...\n");
-      GNUNET_CONSENSUS_conclude (consensus, GNUNET_TIME_UNIT_FOREVER_REL, 0, conclude_cb, NULL);
-    }
-    return;
+    GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "testbed connect emsg: %s\n", emsg);
+    GNUNET_assert (0);
   }
 
-  printf("read: %s", buf);
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "connect complete\n");
+
+  chp = (struct GNUNET_CONSENSUS_Handle **) cls;
+  *chp = (struct GNUNET_CONSENSUS_Handle *) ca_result;
+  num_connected_handles++;
 
-  element = GNUNET_malloc (sizeof (struct GNUNET_CONSENSUS_Element) + strlen(buf) + 1);
-  element->type = 0;
-  element->size = strlen(buf) + 1;
-  element->data = &element[1];
-  strcpy ((char *) &element[1], buf);
-  GNUNET_CONSENSUS_insert (consensus, element, &insert_done_cb, element); 
+  if (num_connected_handles == num_peers)
+  {
+    do_consensus ();
+  }
+}
+
+
+static int
+new_element_cb (void *cls,
+                struct GNUNET_CONSENSUS_Element *element)
+{
+  return GNUNET_YES;
 }
 
 
 /**
- * Called when a new element was received from another peer, or an error occured.
- *
- * May deliver duplicate values.
- *
- * Elements given to a consensus operation by the local peer are NOT given
- * to this callback.
+ * Adapter function called to establish a connection to
+ * a service.
  *
  * @param cls closure
- * @param element new element, NULL on error
- * @return GNUNET_OK if the valid is well-formed and should be added to the consensus,
- *         GNUNET_SYSERR if the element should be ignored and not be propagated
+ * @param cfg configuration of the peer to connect to; will be available until
+ *          GNUNET_TESTBED_operation_done() is called on the operation returned
+ *          from GNUNET_TESTBED_service_connect()
+ * @return service handle to return in 'op_result', NULL on error
  */
-static int
-cb (void *cls,
-    struct GNUNET_CONSENSUS_Element *element)
+static void *
+connect_adapter (void *cls,
+                 const struct GNUNET_CONFIGURATION_Handle *cfg)
 {
-  if (NULL == element)
-  {
-    printf("error receiving from consensus\n");
-    GNUNET_SCHEDULER_shutdown ();
-    return GNUNET_NO;
-  }
-  printf("got element\n");
-  return GNUNET_YES;
+  struct GNUNET_CONSENSUS_Handle *consensus;
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "connect adapter, %d peers\n", num_peers);
+  consensus = GNUNET_CONSENSUS_create (cfg, num_peers, peer_ids, &session_id, new_element_cb, NULL);
+  GNUNET_assert (NULL != consensus);
+  return consensus;
 }
 
 
 /**
- * Function run on shutdown to clean up.
+ * Adapter function called to destroy a connection to
+ * a service.
  *
- * @param cls the statistics handle
- * @param tc scheduler context
+ * @param cls closure
+ * @param op_result service handle returned from the connect adapter
  */
 static void
-shutdown_task (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
+disconnect_adapter(void *cls, void *op_result)
 {
-  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "shutting down\n");
-  if (NULL != consensus)
-  {
-    GNUNET_CONSENSUS_destroy (consensus);
-    consensus = NULL;
-  }
+  /* FIXME: what to do here? */
 }
 
 
+/**
+ * Callback to be called when the requested peer information is available
+ *
+ * @param cb_cls the closure from GNUNET_TETSBED_peer_get_information()
+ * @param op the operation this callback corresponds to
+ * @param pinfo the result; will be NULL if the operation has failed
+ * @param emsg error message if the operation has failed; will be NULL if the
+ *          operation is successfull
+ */
 static void
-run (void *cls, char *const *args, const char *cfgfile,
-     const struct GNUNET_CONFIGURATION_Handle *cfg)
+peer_info_cb (void *cb_cls,
+              struct GNUNET_TESTBED_Operation *op,
+              const struct GNUNET_TESTBED_PeerInformation *pinfo,
+              const char *emsg)
 {
-  struct GNUNET_HashCode sid;
-  struct GNUNET_PeerIdentity *pids;
-  int count;
+  struct GNUNET_PeerIdentity *p;
   int i;
 
-  if (NULL == session_id_str)
-  {
-    GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "no session id given (missing -s/--session-id)\n");
-    return;
-  }
+  GNUNET_assert (NULL == emsg);
 
-  GNUNET_CRYPTO_hash (session_id_str, strlen (session_id_str), &sid);
+  p = (struct GNUNET_PeerIdentity *) cb_cls;
 
-  for (count = 0; NULL != args[count]; count++);
- 
-  if (0 != count)
-  { 
-    pids = GNUNET_malloc (count * sizeof (struct GNUNET_PeerIdentity));
+  if (pinfo->pit == GNUNET_TESTBED_PIT_IDENTITY)
+  {
+    *p = *pinfo->result.id;
+    num_retrieved_peer_ids++;
+    if (num_retrieved_peer_ids == num_peers)
+      for (i = 0; i < num_peers; i++)
+        GNUNET_TESTBED_service_connect (NULL, peers[i], "consensus", connect_complete, &consensus_handles[i],
+                                        connect_adapter, disconnect_adapter, NULL);
   }
   else
   {
-    pids = NULL;
+    GNUNET_assert (0);
   }
+}
 
-  for (i = 0; i < count; i++)
-  {
-    int ret;
-    ret = GNUNET_CRYPTO_hash_from_string (args[i], &pids[i].hashPubKey);
-    if (GNUNET_OK != ret)
-    {
-      GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "peer identity '%s' is malformed\n", args[i]);
-      return;
-    }
-  }
 
-  GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_FOREVER_REL,
-                                &shutdown_task, NULL);
-  
-  consensus = 
-      GNUNET_CONSENSUS_create (cfg,
-                               count, pids,
-                               &sid,
-                               &cb, NULL);
-
-  stdin_fh = GNUNET_DISK_get_handle_from_native (stdin);
-  stdin_tid = GNUNET_SCHEDULER_add_read_file (GNUNET_TIME_UNIT_FOREVER_REL, stdin_fh,
-                                        &stdin_cb, NULL);
+static void
+test_master (void *cls,
+             unsigned int num_peers,
+             struct GNUNET_TESTBED_Peer **started_peers)
+{
+  int i;
+
+
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "test master\n");
+
+  peers = started_peers;
+
+  peer_ids = GNUNET_malloc (num_peers * sizeof (struct GNUNET_PeerIdentity));
+
+  consensus_handles = GNUNET_malloc (num_peers * sizeof (struct ConsensusHandle *));
+
+  for (i = 0; i < num_peers; i++)
+    GNUNET_TESTBED_peer_get_information (peers[i],
+                                         GNUNET_TESTBED_PIT_IDENTITY,
+                                         peer_info_cb,
+                                         &peer_ids[i]);
+}
+
+static void
+run (void *cls, char *const *args, const char *cfgfile,
+     const struct GNUNET_CONFIGURATION_Handle *cfg)
+{
+  static char *session_str = "gnunet-consensus/test";
+
+
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "running gnunet-consensus\n");
+
+  GNUNET_CRYPTO_hash (session_str, strlen(session_str), &session_id);
+
+  (void) GNUNET_TESTBED_test_run ("gnunet-consensus",
+                                  cfgfile,
+                                  num_peers,
+                                  0,
+                                  controller_cb,
+                                  NULL,
+                                  test_master,
+                                  NULL);
 }
 
 
@@ -231,13 +304,24 @@ int
 main (int argc, char **argv)
 {
    static const struct GNUNET_GETOPT_CommandLineOption options[] = {
-      { 's', "session-id", "ID",
-        gettext_noop ("session identifier"),
-        GNUNET_YES, &GNUNET_GETOPT_set_string, &session_id_str },
-        GNUNET_GETOPT_OPTION_END
-   };
-  GNUNET_PROGRAM_run (argc, argv, "gnunet-consensus",
+      { 'n', "num-peers", NULL,
+        gettext_noop ("number of peers in consensus"),
+        GNUNET_YES, &GNUNET_GETOPT_set_uint, &num_peers },
+      { 'k', "value-replication", NULL,
+        gettext_noop ("how many peers receive one value?"),
+        GNUNET_YES, &GNUNET_GETOPT_set_uint, &replication },
+      { 'x', "num-values", NULL,
+        gettext_noop ("number of values"),
+        GNUNET_YES, &GNUNET_GETOPT_set_uint, &num_values },
+      { 't', "timeout", NULL,
+        gettext_noop ("consensus timeout"),
+        GNUNET_YES, &GNUNET_GETOPT_set_relative_time, &conclude_timeout },
+      GNUNET_GETOPT_OPTION_END
+  };
+  conclude_timeout = GNUNET_TIME_UNIT_SECONDS;
+  GNUNET_PROGRAM_run2 (argc, argv, "gnunet-consensus",
                       "help",
-                      options, &run, NULL);
+                      options, &run, NULL, GNUNET_YES);
   return 0;
 }
+
diff --git a/src/consensus/gnunet-service-consensus.c b/src/consensus/gnunet-service-consensus.c
index 1b394db19..ad0266954 100644
--- a/src/consensus/gnunet-service-consensus.c
+++ b/src/consensus/gnunet-service-consensus.c
@@ -32,15 +32,45 @@
 #include "gnunet_util_lib.h"
 #include "gnunet_consensus_service.h"
 #include "gnunet_core_service.h"
-#include "gnunet_mesh_service.h"
+#include "gnunet_stream_lib.h"
+#include "consensus_protocol.h"
+#include "ibf.h"
 #include "consensus.h"
 
 
+/**
+ * Number of IBFs in a strata estimator.
+ */
+#define STRATA_COUNT 32
+/**
+ * Number of buckets per IBF.
+ */
+#define STRATA_IBF_BUCKETS 80
+/**
+ * hash num parameter of the IBF
+ */
+#define STRATA_HASH_NUM 3
+/**
+ * Number of strata that can be transmitted in one message.
+ */
+#define STRATA_PER_MESSAGE ((1<<15) / (IBF_BUCKET_SIZE * STRATA_IBF_BUCKETS))
+
+
+
+/* forward declarations */
+
 struct ConsensusSession;
+struct IncomingSocket;
 
 static void
 send_next (struct ConsensusSession *session);
 
+static void 
+write_strata (void *cls, enum GNUNET_STREAM_Status status, size_t size);
+
+static int
+get_peer_idx (const struct GNUNET_PeerIdentity *peer, const struct ConsensusSession *session);
+
 
 /**
  * An element that is waiting to be transmitted to a client.
@@ -63,22 +93,75 @@ struct PendingElement
   struct GNUNET_CONSENSUS_Element *element;
 };
 
-
-/*
- * A peer that is also in a consensus session.
- * Note that 'this' peer is not in the list.
- */
-struct ConsensusPeer
+struct ConsensusPeerInformation
 {
-  struct GNUNET_PeerIdentity *peer_id;
+  struct GNUNET_STREAM_Socket *socket;
+
+  /**
+   * Is socket's connection established, i.e. can we write to it?
+   * Only relevent on outgoing cpi.
+   */
+  int is_connected;
+
+  /**
+   * Type of the peer in the all-to-all rounds,
+   * GNUNET_YES if we initiate reconciliation.
+   */
+  int is_outgoing;
+
+  /**
+   * Did we receive/send a consensus hello?
+   */
+  int hello;
+
+  /**
+   * Handle for currently active read
+   */
+  struct GNUNET_STREAM_ReadHandle *rh;
+
+  /**
+   * Handle for currently active read
+   */
+  struct GNUNET_STREAM_WriteHandle *wh;
+
+  /**
+   * How many of the strate in the ibf were
+   * sent or received in this round?
+   */
+  int strata_counter;
+
+  struct InvertibleBloomFilter *my_ibf;
+
+  int my_ibf_bucket_counter;
+
+  struct InvertibleBloomFilter *peer_ibf;
+
+  int peer_ibf_bucket_counter;
 
   /**
-   * Incoming tunnel from the peer.
+   * Strata estimator of the peer, NULL if our peer
+   * initiated the reconciliation.
    */
-  struct GNUNET_MESH_Tunnel *incoming_tunnel;
+  struct InvertibleBloomFilter **strata;
 
-  struct InvertibleBloomFilter *last_ibf;
+  struct GNUNET_SERVER_MessageStreamTokenizer *mst;
+
+  struct ConsensusSession *session;
+};
+
+struct QueuedMessage
+{
+  struct GNUNET_MessageHeader *msg;
+
+  /**
+   * Queued messages are stored in a doubly linked list.
+   */
+  struct QueuedMessage *next;
 
+  /**
+   * Queued messages are stored in a doubly linked list.
+   */
+  struct QueuedMessage *prev;
 };
 
 
@@ -98,15 +181,17 @@ struct ConsensusSession
   struct ConsensusSession *prev;
 
   /**
-   * Local consensus identification, chosen by clients.
+   * Join message. Used to initialize the session later,
+   * if the identity of the local peer is not yet known.
+   * NULL if the session has been fully initialized.
    */
-  struct GNUNET_HashCode *local_id;
- 
+  struct GNUNET_CONSENSUS_JoinMessage *join_msg;
+
   /**
   * Global consensus identification, computed
   * from the local id and participating authorities.
   */
-  struct GNUNET_HashCode *global_id;
+  struct GNUNET_HashCode global_id;
 
   /**
    * Local client in this consensus session.
@@ -140,6 +225,10 @@ struct ConsensusSession
    */
   struct PendingElement *approval_pending_tail;
 
+  struct QueuedMessage *client_messages_head;
+
+  struct QueuedMessage *client_messages_tail;
+
   /**
    * Currently active transmit handle for sending to the client
    */
@@ -152,11 +241,6 @@ struct ConsensusSession
   int conclude_requested;
 
   /**
-   * Client has been informed about the conclusion.
-   */
-  int conclude_sent;
-
-  /**
    * Minimum number of peers to form a consensus group
    */
   int conclude_group_min;
@@ -178,30 +262,74 @@ struct ConsensusSession
    */
   unsigned int num_peers;
 
-  /**
-   * Other peers in the consensus, array of ConsensusPeer
-   */
-  struct ConsensusPeer *peers;
+  struct ConsensusPeerInformation *info;
 
   /**
-   * Tunnel for broadcasting to all other authorities
+   * Sorted array of peer identities in this consensus session,
+   * includes the local peer.
    */
-  struct GNUNET_MESH_Tunnel *broadcast_tunnel;
+  struct GNUNET_PeerIdentity *peers;
 
   /**
-   * Time limit for one round of pairwise exchange.
-   * FIXME: should not actually be a constant
+   * Index of the local peer in the peers array
    */
-  struct GNUNET_TIME_Relative round_time;
+  int local_peer_idx;
 
   /**
    * Task identifier for the round timeout task
    */
   GNUNET_SCHEDULER_TaskIdentifier round_timeout_tid;
+
+  struct InvertibleBloomFilter **strata;
 };
 
 
 /**
+ * Sockets from other peers who want to communicate with us.
+ * It may not be known yet which consensus session they belong to.
+ */
+struct IncomingSocket
+{
+  /**
+   * Incoming sockets are kept in a double linked list.
+   */
+  struct IncomingSocket *next;
+
+  /**
+   * Incoming sockets are kept in a double linked list.
+   */
+  struct IncomingSocket *prev;
+
+  /**
+   * The actual socket.
+   */
+  struct GNUNET_STREAM_Socket *socket;
+
+  /**
+   * Handle for currently active read
+   */
+  struct GNUNET_STREAM_ReadHandle *rh;
+
+  /**
+   * Peer that connected to us with the socket.
+   */
+  struct GNUNET_PeerIdentity *peer;
+
+  /**
+   * Message stream tokenizer for this socket.
+   */
+  struct GNUNET_SERVER_MessageStreamTokenizer *mst;
+
+  /**
+   * Peer-in-session this socket belongs to, once known, otherwise NULL.
+   */
+  struct ConsensusPeerInformation *cpi;
+};
+
+static struct IncomingSocket *incoming_sockets_head;
+static struct IncomingSocket *incoming_sockets_tail;
+
+/**
  * Linked list of sesstions this peer participates in.
  */
 static struct ConsensusSession *sessions_head;
@@ -222,32 +350,349 @@ static const struct GNUNET_CONFIGURATION_Handle *cfg;
 static struct GNUNET_SERVER_Handle *srv;
 
 /**
- * Peer that runs this service
+ * Peer that runs this service.
  */
 static struct GNUNET_PeerIdentity *my_peer;
 
 /**
- * Handle to the mesh service.
+ * Handle to the core service. Only used during service startup, will be NULL after that.
  */
-static struct GNUNET_MESH_Handle *mesh;
+static struct GNUNET_CORE_Handle *core;
 
 /**
- * Handle to the core service. Only used during service startup, will be NULL after that.
+ * Listener for sockets from peers that want to reconcile with us.
  */
-static struct GNUNET_CORE_Handle *core;
+static struct GNUNET_STREAM_ListenSocket *listener;
 
+
+static int
+estimate_difference (struct InvertibleBloomFilter** strata1,
+                     struct InvertibleBloomFilter** strata2)
+{
+  int i;
+  int count;
+  count = 0;
+  for (i = STRATA_COUNT - 1; i >= 0; i--)
+  {
+    struct InvertibleBloomFilter *diff;
+    int ibf_count;
+    int more;
+    ibf_count = 0;
+    diff = ibf_dup (strata1[i]);
+    ibf_subtract (diff, strata2[i]);
+    for (;;)
+    {
+      more = ibf_decode (diff, NULL, NULL);
+      if (GNUNET_NO == more)
+      {
+        count += ibf_count;
+        break;
+      }
+      if (GNUNET_SYSERR == more)
+      {
+        return count * (1 << (i + 1));
+      }
+      ibf_count++;
+    }
+    ibf_destroy (diff);
+  }
+  return count;
+}
+
+
+/**
+ * Functions of this signature are called whenever data is available from the
+ * stream.
+ *
+ * @param cls the closure from GNUNET_STREAM_read
+ * @param status the status of the stream at the time this function is called
+ * @param data traffic from the other side
+ * @param size the number of bytes available in data read; will be 0 on timeout 
+ * @return number of bytes of processed from 'data' (any data remaining should be
+ *         given to the next time the read processor is called).
+ */
+static size_t
+stream_data_processor (void *cls,
+                       enum GNUNET_STREAM_Status status,
+                       const void *data,
+                       size_t size)
+{
+  struct IncomingSocket *incoming;
+  int ret;
+
+  GNUNET_assert (GNUNET_STREAM_OK == status);
+
+  incoming = (struct IncomingSocket *) cls;
+
+  ret = GNUNET_SERVER_mst_receive (incoming->mst, incoming, data, size, GNUNET_NO, GNUNET_NO);
+  if (GNUNET_SYSERR == ret)
+  {
+    /* FIXME: handle this correctly */
+    GNUNET_assert (0);
+  }
+
+  /* read again */
+  incoming->rh = GNUNET_STREAM_read (incoming->socket, GNUNET_TIME_UNIT_FOREVER_REL,
+                                     &stream_data_processor, incoming);
+
+  /* we always read all data */
+  return size;
+}
+
+static int
+handle_p2p_strata (struct ConsensusPeerInformation *cpi, const struct StrataMessage *strata_msg)
+{
+  int i;
+  int num_strata;
+  struct GNUNET_HashCode *hash_src;
+  uint8_t *count_src;
+
+  GNUNET_assert (GNUNET_NO == cpi->is_outgoing);
+
+  if (NULL == cpi->strata)
+  {
+    cpi->strata = GNUNET_malloc (STRATA_COUNT * sizeof (struct InvertibleBloomFilter *));
+    for (i = 0; i < STRATA_COUNT; i++)
+      cpi->strata[i] = ibf_create (STRATA_IBF_BUCKETS, STRATA_HASH_NUM, 0);
+  }
+
+  num_strata = ntohs (strata_msg->num_strata);
+
+  /* for correct message alignment, copy bucket types seperately */
+  hash_src = (struct GNUNET_HashCode *) &strata_msg[1];
+
+  for (i = 0; i < num_strata; i++)
+  {
+    memcpy (cpi->strata[cpi->strata_counter+i]->hash_sum, hash_src, STRATA_IBF_BUCKETS * sizeof *hash_src);
+    hash_src += STRATA_IBF_BUCKETS;
+  }
+
+  for (i = 0; i < num_strata; i++)
+  {
+    memcpy (cpi->strata[cpi->strata_counter+i]->id_sum, hash_src, STRATA_IBF_BUCKETS * sizeof *hash_src);
+    hash_src += STRATA_IBF_BUCKETS;
+  }
+
+  count_src = (uint8_t *) hash_src;
+
+  for (i = 0; i < num_strata; i++)
+  {
+    uint8_t zero[STRATA_IBF_BUCKETS];
+    memset (zero, 0, STRATA_IBF_BUCKETS);
+    memcpy (cpi->strata[cpi->strata_counter+i]->count, count_src, STRATA_IBF_BUCKETS);
+    count_src += STRATA_IBF_BUCKETS;
+  }
+
+  GNUNET_assert (count_src == (((uint8_t *) &strata_msg[1]) + STRATA_IBF_BUCKETS * num_strata * IBF_BUCKET_SIZE));
+
+  cpi->strata_counter += num_strata;
+
+  if (STRATA_COUNT == cpi->strata_counter)
+  {
+    int diff;
+    diff = estimate_difference (cpi->session->strata, cpi->strata);
+    GNUNET_log (GNUNET_ERROR_TYPE_INFO, "diff=%d\n", diff);
+  }
+
+  return GNUNET_YES;
+}
+
+
+static int
+handle_p2p_ibf (struct ConsensusPeerInformation *cpi, const struct DifferenceDigest *strata)
+{
+  return GNUNET_YES;
+}
+
+
+static int
+handle_p2p_element (struct ConsensusPeerInformation *cpi, const struct Element *strata)
+{
+  return GNUNET_YES;
+}
+
+
+static int
+handle_p2p_hello (struct IncomingSocket *inc, const struct ConsensusHello *hello)
+{
+  struct ConsensusSession *session;
+  session = sessions_head;
+  while (NULL != session)
+  {
+    if (0 == GNUNET_CRYPTO_hash_cmp (&session->global_id, &hello->global_id))
+    {
+      int idx;
+      GNUNET_log (GNUNET_ERROR_TYPE_INFO, "peer helloed session\n");
+      idx = get_peer_idx (inc->peer, session);
+      GNUNET_assert (-1 != idx);
+      GNUNET_log (GNUNET_ERROR_TYPE_INFO, "idx is %d\n", idx);
+      inc->cpi = &session->info[idx];
+      GNUNET_assert (GNUNET_NO == inc->cpi->is_outgoing);
+      inc->cpi->mst = inc->mst;
+      inc->cpi->hello = GNUNET_YES;
+      inc->cpi->socket = inc->socket;
+      return GNUNET_YES;
+    }
+    session = session->next;
+  }
+  GNUNET_assert (0);
+  return GNUNET_NO;
+}
+
+
+/**
+ * Functions with this signature are called whenever a
+ * complete message is received by the tokenizer.
+ *
+ * Do not call GNUNET_SERVER_mst_destroy in callback
+ *
+ * @param cls closure
+ * @param client identification of the client
+ * @param message the actual message
+ *
+ * @return GNUNET_OK on success, GNUNET_SYSERR to stop further processing
+ */
+static int
+mst_session_callback (void *cls, void *client, const struct GNUNET_MessageHeader *message)
+{
+  struct ConsensusPeerInformation *cpi;
+  cpi = (struct ConsensusPeerInformation *) cls;
+  switch (ntohs( message->type))
+  {
+    case GNUNET_MESSAGE_TYPE_CONSENSUS_P2P_DELTA_ESTIMATE:
+      return handle_p2p_strata (cpi, (struct StrataMessage *) message);
+    case GNUNET_MESSAGE_TYPE_CONSENSUS_P2P_DIFFERENCE_DIGEST:
+      return handle_p2p_ibf (cpi, (struct DifferenceDigest *) message);
+    case GNUNET_MESSAGE_TYPE_CONSENSUS_P2P_ELEMENTS:
+      return handle_p2p_element (cpi, (struct Element *) message);
+    default:
+      /* FIXME: handle correctly */
+      GNUNET_assert (0);
+  }
+  return GNUNET_OK;
+}
+
+
+/**
+ * Handle tokenized messages from stream sockets.
+ * Delegate them if the socket belongs to a session,
+ * handle hello messages otherwise.
+ *
+ * Do not call GNUNET_SERVER_mst_destroy in callback
+ *
+ * @param cls closure, unused
+ * @param client incoming socket this message comes from
+ * @param message the actual message
+ *
+ * @return GNUNET_OK on success, GNUNET_SYSERR to stop further processing
+ */
+static int
+mst_incoming_callback (void *cls, void *client, const struct GNUNET_MessageHeader *message)
+{
+  struct IncomingSocket *inc;
+  inc = (struct IncomingSocket *) client;
+  switch (ntohs( message->type))
+  {
+    case GNUNET_MESSAGE_TYPE_CONSENSUS_P2P_HELLO:
+      return handle_p2p_hello (inc, (struct ConsensusHello *) message);
+    default:
+      if (NULL != inc->cpi)
+        return mst_session_callback (inc->cpi, client, message);
+      /* FIXME: disconnect peer properly */
+      GNUNET_assert (0);
+  }
+  return GNUNET_OK;
+}
+
+
+/**
+ * Functions of this type are called upon new stream connection from other peers
+ * or upon binding error which happen when the app_port given in
+ * GNUNET_STREAM_listen() is already taken.
+ *
+ * @param cls the closure from GNUNET_STREAM_listen
+ * @param socket the socket representing the stream; NULL on binding error
+ * @param initiator the identity of the peer who wants to establish a stream
+ *            with us; NULL on binding error
+ * @return GNUNET_OK to keep the socket open, GNUNET_SYSERR to close the
+ *             stream (the socket will be invalid after the call)
+ */
+static int
+listen_cb (void *cls,
+           struct GNUNET_STREAM_Socket *socket,
+           const struct GNUNET_PeerIdentity *initiator)
+{
+  struct IncomingSocket *incoming;
+
+  GNUNET_assert (NULL != socket);
+
+  incoming = GNUNET_malloc (sizeof *incoming);
+
+  incoming->socket = socket;
+  incoming->peer = GNUNET_memdup (initiator, sizeof *initiator);
+
+  incoming->rh = GNUNET_STREAM_read (socket, GNUNET_TIME_UNIT_FOREVER_REL,
+                                     &stream_data_processor, incoming);
+
+
+  incoming->mst = GNUNET_SERVER_mst_create (mst_incoming_callback, incoming);
+
+  GNUNET_CONTAINER_DLL_insert_tail (incoming_sockets_head, incoming_sockets_tail, incoming);
+
+  return GNUNET_OK;
+}
+
+
+static void
+destroy_session (struct ConsensusSession *session)
+{
+  /* FIXME: more stuff to free! */
+  GNUNET_CONTAINER_DLL_remove (sessions_head, sessions_tail, session);
+  GNUNET_SERVER_client_drop (session->client);
+  GNUNET_free (session);
+}
+
+
+/**
+ * Disconnect a client, and destroy all sessions associated with it.
+ *
+ * @param client the client to disconnect
+ */
 static void
 disconnect_client (struct GNUNET_SERVER_Client *client)
 {
+  struct ConsensusSession *session;
   GNUNET_SERVER_client_disconnect (client);
-  /* FIXME: free data structures that this client owns */
+  
+  /* if the client owns a session, remove it */
+  session = sessions_head;
+  while (NULL != session)
+  {
+    if (client == session->client)
+    {
+      destroy_session (session);
+      break;
+    }
+    session = session->next;
+  }
 }
 
+
+/**
+ * Compute a global, (hopefully) unique consensus session id,
+ * from the local id of the consensus session, and the identities of all participants.
+ * Thus, if the local id of two consensus sessions coincide, but are not comprised of
+ * exactly the same peers, the global id will be different.
+ *
+ * @param local_id local id of the consensus session
+ * @param peers array of all peers participating in the consensus session
+ * @param num_peers number of elements in the peers array
+ * @param dst where the result is stored, may not be NULL
+ */
 static void
-compute_global_id (struct GNUNET_HashCode *dst,
-                   const struct GNUNET_HashCode *local_id,
-                   const struct GNUNET_PeerIdentity *peers,
-                   int num_peers)
+compute_global_id (const struct GNUNET_HashCode *local_id,
+                   const struct GNUNET_PeerIdentity *peers, int num_peers, 
+                   struct GNUNET_HashCode *dst)
 {
   int i;
   struct GNUNET_HashCode tmp;
@@ -263,45 +708,50 @@ compute_global_id (struct GNUNET_HashCode *dst,
 }
 
 
+/**
+ * Function called to notify a client about the connection
+ * begin ready to queue more data.  "buf" will be
+ * NULL and "size" zero if the connection was closed for
+ * writing in the meantime.
+ *
+ * @param cls consensus session
+ * @param size number of bytes available in buf
+ * @param buf where the callee should write the message
+ * @return number of bytes written to buf
+ */
 static size_t
-transmit_pending (void *cls, size_t size, void *buf)
+transmit_queued (void *cls, size_t size,
+                 void *buf)
 {
-  struct GNUNET_CONSENSUS_Element *element;
-  struct GNUNET_CONSENSUS_ElementMessage *msg;
   struct ConsensusSession *session;
+  struct QueuedMessage *qmsg;
+  size_t msg_size;
 
   session = (struct ConsensusSession *) cls;
-  msg = (struct GNUNET_CONSENSUS_ElementMessage *) buf;
-  element = session->transmit_pending_head->element;
-
-  GNUNET_assert (NULL != element);
-
   session->th = NULL;
 
-  msg->element_type = element->type;
-  msg->header.type = htons (GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_RECEIVED_ELEMENT);
-  msg->header.size = htons (sizeof (struct GNUNET_CONSENSUS_ElementMessage) + element->size);
-  memcpy (&msg[1], element->data, element->size);
 
-  session->transmit_pending_head = session->transmit_pending_head->next;
+  qmsg = session->client_messages_head;
+  GNUNET_CONTAINER_DLL_remove (session->client_messages_head, session->client_messages_tail, qmsg);
+  GNUNET_assert (qmsg);
 
-  send_next (session);
+  if (NULL == buf)
+  {
+    destroy_session (session);
+    return 0;
+  }
 
-  return sizeof (struct GNUNET_CONSENSUS_ElementMessage) + element->size;
-}
+  msg_size = ntohs (qmsg->msg->size);
 
+  GNUNET_assert (size >= msg_size);
 
-static size_t
-transmit_conclude_done (void *cls, size_t size, void *buf)
-{
-  struct GNUNET_CONSENSUS_ConcludeDoneMessage *msg;
+  memcpy (buf, qmsg->msg, msg_size);
+  GNUNET_free (qmsg->msg);
+  GNUNET_free (qmsg);
 
-  msg = (struct GNUNET_CONSENSUS_ConcludeDoneMessage *) buf;
-  msg->header.type = htons (GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_CONCLUDE_DONE);
-  msg->header.size = htons (sizeof (struct GNUNET_CONSENSUS_ConcludeDoneMessage));
-  msg->num_peers = htons (0);
+  send_next (session);
 
-  return sizeof (struct GNUNET_CONSENSUS_ConcludeDoneMessage);
+  return msg_size;
 }
 
 
@@ -313,65 +763,253 @@ transmit_conclude_done (void *cls, size_t size, void *buf)
 static void
 send_next (struct ConsensusSession *session)
 {
-  int msize;
 
   GNUNET_assert (NULL != session);
 
   if (NULL != session->th)
-  {
     return;
-  }
 
-  if ((session->conclude_requested == GNUNET_YES) && (session->conclude_sent == GNUNET_NO))
+  if (NULL != session->client_messages_head)
   {
-    /* FIXME */
-    msize = sizeof (struct GNUNET_CONSENSUS_ConcludeMessage);
-    session->th =
-        GNUNET_SERVER_notify_transmit_ready (session->client, msize,
-                                             GNUNET_TIME_UNIT_FOREVER_REL, &transmit_conclude_done, session);
-    session->conclude_sent = GNUNET_YES;
+    int msize;
+    msize = ntohs (session->client_messages_head->msg->size);
+    GNUNET_log (GNUNET_ERROR_TYPE_INFO, "scheduling queued\n");
+    session->th = GNUNET_SERVER_notify_transmit_ready (session->client, msize, 
+                                                       GNUNET_TIME_UNIT_FOREVER_REL,
+                                                       &transmit_queued, session);
   }
-  else if (NULL != session->transmit_pending_head)
+}
+
+
+/**
+ * Although GNUNET_CRYPTO_hash_cmp exisits, it does not have
+ * the correct signature to be used with e.g. qsort.
+ * We use this function instead.
+ *
+ * @param h1 some hash code
+ * @param h2 some hash code
+ * @return 1 if h1 > h2, -1 if h1 < h2 and 0 if h1 == h2.
+ */
+static int
+hash_cmp (const void *a, const void *b)
+{
+  return GNUNET_CRYPTO_hash_cmp ((struct GNUNET_HashCode *) a, (struct GNUNET_HashCode *) b);
+}
+
+
+/**
+ * Search peer in the list of peers in session.
+ *
+ * @param peer peer to find
+ * @param session session with peer
+ * @return index of peer, -1 if peer is not in session
+ */
+static int
+get_peer_idx (const struct GNUNET_PeerIdentity *peer, const struct ConsensusSession *session)
+{
+  const struct GNUNET_PeerIdentity *needle;
+  needle = bsearch (peer, session->peers, session->num_peers, sizeof (struct GNUNET_PeerIdentity), &hash_cmp);
+  if (NULL == needle)
+    return -1;
+  return needle - session->peers;
+}
+
+
+
+static void
+hello_cont (void *cls, enum GNUNET_STREAM_Status status, size_t size)
+{
+  struct ConsensusPeerInformation *cpi;
+
+  cpi = (struct ConsensusPeerInformation *) cls;
+  cpi->hello = GNUNET_YES;
+  
+  GNUNET_assert (GNUNET_STREAM_OK == status);
+
+  cpi = (struct ConsensusPeerInformation *) cls;
+
+  if (cpi->session->conclude_requested)
   {
-    msize = session->transmit_pending_head->element->size + sizeof (struct GNUNET_CONSENSUS_ElementMessage);
-    session->th =
-        GNUNET_SERVER_notify_transmit_ready (session->client, msize,
-                                             GNUNET_TIME_UNIT_FOREVER_REL, &transmit_pending, session);
-    /* TODO: insert into ack pending */
+    write_strata (cpi, GNUNET_STREAM_OK, 0);  
   }
 }
 
 
 /**
- * Method called whenever a peer has disconnected from the tunnel.
- * Implementations of this callback must NOT call
- * GNUNET_MESH_tunnel_destroy immediately, but instead schedule those
- * to run in some other task later.  However, calling 
- * "GNUNET_MESH_notify_transmit_ready_cancel" is allowed.
+ * Functions of this type will be called when a stream is established
  *
- * @param cls closure
- * @param peer peer identity the tunnel stopped working with
+ * @param cls the closure from GNUNET_STREAM_open
+ * @param socket socket to use to communicate with the other side (read/write)
  */
 static void
-disconnect_handler (void *cls, const struct GNUNET_PeerIdentity *peer)
+open_cb (void *cls, struct GNUNET_STREAM_Socket *socket)
+{
+  struct ConsensusPeerInformation *cpi;
+  struct ConsensusHello *hello;
+
+
+  cpi = (struct ConsensusPeerInformation *) cls;
+  cpi->is_connected = GNUNET_YES;
+
+  hello = GNUNET_malloc (sizeof *hello);
+  hello->header.size = htons (sizeof *hello);
+  hello->header.type = htons (GNUNET_MESSAGE_TYPE_CONSENSUS_P2P_HELLO);
+  memcpy (&hello->global_id, &cpi->session->global_id, sizeof (struct GNUNET_HashCode));
+
+
+  cpi->wh =
+      GNUNET_STREAM_write (socket, hello, sizeof *hello, GNUNET_TIME_UNIT_FOREVER_REL, hello_cont, cpi);
+
+}
+
+
+static void
+initialize_session_info (struct ConsensusSession *session)
+{
+  int i;
+  int last;
+
+  for (i = 0; i < session->num_peers; ++i)
+  {
+    /* initialize back-references, so consensus peer information can
+     * be used as closure */
+    session->info[i].session = session;
+
+  }
+
+  last = (session->local_peer_idx + (session->num_peers / 2)) % session->num_peers;
+  i = (session->local_peer_idx + 1) % session->num_peers;
+  while (i != last)
+  {
+    session->info[i].is_outgoing = GNUNET_YES;
+    session->info[i].socket = GNUNET_STREAM_open (cfg, &session->peers[i], GNUNET_APPLICATION_TYPE_CONSENSUS,
+                                                  open_cb, &session->info[i], GNUNET_STREAM_OPTION_END);
+    session->info[i].mst = GNUNET_SERVER_mst_create (mst_session_callback, session);
+    i = (i + 1) % session->num_peers;
+  }
+  // tie-breaker for even number of peers
+  if (((session->num_peers % 2) == 0) && (session->local_peer_idx < last))
+  {
+    session->info[last].is_outgoing = GNUNET_YES;
+    session->info[last].socket = GNUNET_STREAM_open (cfg, &session->peers[last], GNUNET_APPLICATION_TYPE_CONSENSUS,
+                                                     open_cb, &session->info[last], GNUNET_STREAM_OPTION_END);
+  }
+}
+
+
+/**
+ * Create the sorted list of peers for the session,
+ * add the local peer if not in the join message.
+ */
+static void
+initialize_session_peer_list (struct ConsensusSession *session)
+{
+  int local_peer_in_list;
+  int listed_peers;
+  const struct GNUNET_PeerIdentity *msg_peers;
+  unsigned int i;
+
+  GNUNET_assert (NULL != session->join_msg);
+
+  /* peers in the join message, may or may not include the local peer */
+  listed_peers = ntohs (session->join_msg->num_peers);
+  
+  session->num_peers = listed_peers;
+
+  msg_peers = (struct GNUNET_PeerIdentity *) &session->join_msg[1];
+
+  local_peer_in_list = GNUNET_NO;
+  for (i = 0; i < listed_peers; i++)
+  {
+    if (0 == memcmp (&msg_peers[i], my_peer, sizeof (struct GNUNET_PeerIdentity)))
+    {
+      local_peer_in_list = GNUNET_YES;
+      break;
+    }
+  }
+
+  if (GNUNET_NO == local_peer_in_list)
+    session->num_peers++;
+
+  session->peers = GNUNET_malloc (session->num_peers * sizeof (struct GNUNET_PeerIdentity));
+
+  if (GNUNET_NO == local_peer_in_list)
+    session->peers[session->num_peers - 1] = *my_peer;
+
+  memcpy (session->peers, msg_peers, listed_peers * sizeof (struct GNUNET_PeerIdentity));
+  qsort (session->peers, session->num_peers, sizeof (struct GNUNET_PeerIdentity), &hash_cmp);
+}
+
+
+static void
+strata_insert (struct InvertibleBloomFilter **strata, struct GNUNET_HashCode *key)
 {
-  /* FIXME: how do we handle this */
+  uint32_t v;
+  int i;
+  v = key->bits[0];
+  /* count trailing '1'-bits of v */
+  for (i = 0; v & 1; v>>=1, i++);
+
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "strata insert at %d\n", i);
+
+  ibf_insert (strata[i], key);
 }
 
 
 /**
- * Method called whenever a peer has connected to the tunnel.
+ * Initialize the session, continue receiving messages from the owning client
  *
- * @param cls closure
- * @param peer peer identity the tunnel was created to, NULL on timeout
- * @param atsi performance data for the connection
+ * @param session the session to initialize
  */
 static void
-connect_handler (void *cls,
-                 const struct GNUNET_PeerIdentity *peer,
-                 const struct GNUNET_ATS_Information *atsi)
+initialize_session (struct ConsensusSession *session)
 {
-  /* not much we can do here, now we know the other peer has been added to our broadcast tunnel */
+  const struct ConsensusSession *other_session;
+  int i;
+
+  GNUNET_assert (NULL != session->join_msg);
+
+  initialize_session_peer_list (session);
+
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "session with %u peers\n", session->num_peers);
+
+  compute_global_id (&session->join_msg->session_id, session->peers, session->num_peers, &session->global_id);
+
+  /* Check if some local client already owns the session. */
+  other_session = sessions_head;
+  while (NULL != other_session)
+  {
+    if ((other_session != session) && 
+        (0 == GNUNET_CRYPTO_hash_cmp (&session->global_id, &other_session->global_id)))
+    {
+      /* session already owned by another client */
+      GNUNET_break (0);
+      disconnect_client (session->client);
+      return;
+    }
+    other_session = other_session->next;
+  }
+
+  session->values = GNUNET_CONTAINER_multihashmap_create (256, GNUNET_NO);
+
+  session->local_peer_idx = get_peer_idx (my_peer, session);
+  GNUNET_assert (-1 != session->local_peer_idx);
+
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "%d is the local peer\n", session->local_peer_idx);
+
+  session->strata = GNUNET_malloc (STRATA_COUNT * sizeof (struct InvertibleBloomFilter *));
+  for (i = 0; i < STRATA_COUNT; i++)
+    session->strata[i] = ibf_create (STRATA_IBF_BUCKETS, STRATA_HASH_NUM, 0);
+
+  session->info = GNUNET_malloc (session->num_peers * sizeof (struct ConsensusPeerInformation));
+
+  initialize_session_info (session);
+
+  GNUNET_free (session->join_msg);
+  session->join_msg = NULL;
+
+  GNUNET_SERVER_receive_done (session->client, GNUNET_OK);
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "session %s initialized\n", GNUNET_h2s (&session->global_id));
 }
 
 
@@ -387,88 +1025,49 @@ client_join (void *cls,
              struct GNUNET_SERVER_Client *client,
              const struct GNUNET_MessageHeader *m)
 {
-  struct GNUNET_HashCode global_id;
-  const struct GNUNET_CONSENSUS_JoinMessage *msg;
   struct ConsensusSession *session;
-  unsigned int i;
-
-  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "client joining\n");
-
-  msg = (struct GNUNET_CONSENSUS_JoinMessage *) m;
-
-  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "session id is %s\n", GNUNET_h2s (&msg->session_id));
 
-  compute_global_id (&global_id, &msg->session_id, (struct GNUNET_PeerIdentity *) &m[1], msg->num_peers);
-
-  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "computed global id is %s\n", GNUNET_h2s (&global_id));
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "join received\n");
 
+  // make sure the client has not already joined a session
   session = sessions_head;
   while (NULL != session)
   {
-    if (client == session->client)
-    {
-
-      GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "client already in session\n");
-      disconnect_client (client);
-      return;
-    }
-    if (0 == memcmp (session->global_id, &global_id, sizeof (struct GNUNET_HashCode)))
+    if (session->client == client)
     {
-      GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "session already owned by another client\n");
+      GNUNET_break (0);
       disconnect_client (client);
       return;
     }
     session = session->next;
   }
 
-  GNUNET_SERVER_client_keep (client);
-
-  /* session does not exist yet, create it */
   session = GNUNET_malloc (sizeof (struct ConsensusSession));
-  session->local_id = GNUNET_memdup (&msg->session_id, sizeof (struct GNUNET_HashCode));
-  session->global_id = GNUNET_memdup (&global_id, sizeof (struct GNUNET_HashCode));
-  session->values = GNUNET_CONTAINER_multihashmap_create (4, GNUNET_NO);
+  session->join_msg = (struct GNUNET_CONSENSUS_JoinMessage *) GNUNET_copy_message (m);
   session->client = client;
-  /* FIXME: should not be a constant, but chosen adaptively */
-  session->round_time = GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 5);
-
-  session->broadcast_tunnel = GNUNET_MESH_tunnel_create (mesh, session, connect_handler, disconnect_handler, session);
-
-  session->num_peers = 0;
-
-  /* count the peers that are not the local peer */
-  for (i = 0; i < msg->num_peers; i++)
-  {
-    struct GNUNET_PeerIdentity *peers;
-    peers = (struct GNUNET_PeerIdentity *) &msg[1];
-    if (0 != memcmp (&peers[i], my_peer, sizeof (struct GNUNET_PeerIdentity)))
-      session->num_peers++;
-  }
+  GNUNET_SERVER_client_keep (client);
 
-  session->peers = GNUNET_malloc (session->num_peers * sizeof (struct ConsensusPeer));
+  GNUNET_CONTAINER_DLL_insert (sessions_head, sessions_tail, session);
 
-  /* copy the peer identities and add peers to broadcast tunnel */
-  for (i = 0; i < msg->num_peers; i++)
+  // Initialize session later if local peer identity is not known yet.
+  if (NULL == my_peer)
   {
-    struct GNUNET_PeerIdentity *peers;
-    peers = (struct GNUNET_PeerIdentity *) &msg[1];
-    if (0 != memcmp (&peers[i], my_peer, sizeof (struct GNUNET_PeerIdentity)))
-    {
-      *session->peers->peer_id = peers[i];
-      GNUNET_MESH_peer_request_connect_add (session->broadcast_tunnel, &peers[i]);
-    }
+    GNUNET_SERVER_disable_receive_done_warning (client);
+    GNUNET_log (GNUNET_ERROR_TYPE_INFO, "session init delayed\n");
+    return;
   }
 
-  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "created new session\n");
-
-  GNUNET_CONTAINER_DLL_insert (sessions_head, sessions_tail, session);
-
-  GNUNET_SERVER_receive_done (client, GNUNET_OK);
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "session init now\n");
+  initialize_session (session);
 }
 
 
 /**
  * Called when a client performs an insert operation.
+ *
+ * @param cls (unused)
+ * @param client client handle
+ * @param message message sent by the client
  */
 void
 client_insert (void *cls,
@@ -510,7 +1109,9 @@ client_insert (void *cls,
   GNUNET_CRYPTO_hash (element, element_size, &key);
 
   GNUNET_CONTAINER_multihashmap_put (session->values, &key, element,
-                                     GNUNET_CONTAINER_MULTIHASHMAPOPTION_REPLACE);
+                                     GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
+
+  strata_insert (session->strata, &key);
 
   GNUNET_SERVER_receive_done (client, GNUNET_OK);
 
@@ -518,31 +1119,145 @@ client_insert (void *cls,
 }
 
 
+
+/**
+ * Functions of this signature are called whenever writing operations
+ * on a stream are executed
+ *
+ * @param cls the closure from GNUNET_STREAM_write
+ * @param status the status of the stream at the time this function is called;
+ *          GNUNET_STREAM_OK if writing to stream was completed successfully;
+ *          GNUNET_STREAM_TIMEOUT if the given data is not sent successfully
+ *          (this doesn't mean that the data is never sent, the receiver may
+ *          have read the data but its ACKs may have been lost);
+ *          GNUNET_STREAM_SHUTDOWN if the stream is shutdown for writing in the
+ *          mean time; GNUNET_STREAM_SYSERR if the stream is broken and cannot
+ *          be processed.
+ * @param size the number of bytes written
+ */
+static void 
+write_strata (void *cls, enum GNUNET_STREAM_Status status, size_t size)
+{
+  struct ConsensusPeerInformation *cpi;
+  struct StrataMessage *strata_msg;
+  size_t msize;
+  int i;
+  struct GNUNET_HashCode *hash_dst;
+  uint8_t *count_dst;
+  int num_strata;
+
+  cpi = (struct ConsensusPeerInformation *) cls;
+
+  GNUNET_assert (GNUNET_YES == cpi->is_outgoing);
+
+  /* FIXME: handle this */
+  GNUNET_assert (GNUNET_STREAM_OK == status);
+
+  if (STRATA_COUNT == cpi->strata_counter)
+  {
+    /* strata have been written, wait for other side's IBF */
+    return;
+  }
+
+  if ((STRATA_COUNT - cpi->strata_counter) < STRATA_PER_MESSAGE)
+    num_strata = (STRATA_COUNT - cpi->strata_counter);
+  else
+    num_strata = STRATA_PER_MESSAGE;
+
+
+  msize = (sizeof *strata_msg) + (num_strata * IBF_BUCKET_SIZE * STRATA_IBF_BUCKETS);
+
+  strata_msg = GNUNET_malloc (msize);
+  strata_msg->header.size = htons (msize);
+  strata_msg->header.type = htons (GNUNET_MESSAGE_TYPE_CONSENSUS_P2P_DELTA_ESTIMATE);
+  strata_msg->num_strata = htons (num_strata);
+
+  /* for correct message alignment, copy bucket types seperately */
+  hash_dst = (struct GNUNET_HashCode *) &strata_msg[1];
+
+  for (i = 0; i < num_strata; i++)
+  {
+    memcpy (hash_dst, cpi->session->strata[cpi->strata_counter+i]->hash_sum, STRATA_IBF_BUCKETS * sizeof *hash_dst);
+    hash_dst += STRATA_IBF_BUCKETS;
+  }
+
+  for (i = 0; i < num_strata; i++)
+  {
+    memcpy (hash_dst, cpi->session->strata[cpi->strata_counter+i]->id_sum, STRATA_IBF_BUCKETS * sizeof *hash_dst);
+    hash_dst += STRATA_IBF_BUCKETS;
+  }
+
+  count_dst = (uint8_t *) hash_dst;
+
+  for (i = 0; i < num_strata; i++)
+  {
+    memcpy (count_dst, cpi->session->strata[cpi->strata_counter+i]->count, STRATA_IBF_BUCKETS);
+    count_dst += STRATA_IBF_BUCKETS;
+  }
+
+  cpi->strata_counter += num_strata;
+
+  cpi->wh = GNUNET_STREAM_write (cpi->socket, strata_msg, msize, GNUNET_TIME_UNIT_FOREVER_REL,
+                                 write_strata, cpi);
+
+  GNUNET_assert (NULL != cpi->wh);
+}
+
+
 /**
- * Do one round of the conclusion.
- * Start by broadcasting the set difference estimator (IBF strata).
+ * Functions of this signature are called whenever writing operations
+ * on a stream are executed
  *
+ * @param cls the closure from GNUNET_STREAM_write
+ * @param status the status of the stream at the time this function is called;
+ *          GNUNET_STREAM_OK if writing to stream was completed successfully;
+ *          GNUNET_STREAM_TIMEOUT if the given data is not sent successfully
+ *          (this doesn't mean that the data is never sent, the receiver may
+ *          have read the data but its ACKs may have been lost);
+ *          GNUNET_STREAM_SHUTDOWN if the stream is shutdown for writing in the
+ *          mean time; GNUNET_STREAM_SYSERR if the stream is broken and cannot
+ *          be processed.
+ * @param size the number of bytes written
  */
-void
-conclude_do_round (struct ConsensusSession *session)
+static void 
+write_ibf (void *cls, enum GNUNET_STREAM_Status status, size_t size)
 {
-  /* FIXME */
+  struct ConsensusPeerInformation *cpi;
+
+  cpi = (struct ConsensusPeerInformation *) cls;
 }
 
 
 /**
- * Cancel the current round if necessary, decide to run another round or
- * terminate.
+ * Functions of this signature are called whenever writing operations
+ * on a stream are executed
+ *
+ * @param cls the closure from GNUNET_STREAM_write
+ * @param status the status of the stream at the time this function is called;
+ *          GNUNET_STREAM_OK if writing to stream was completed successfully;
+ *          GNUNET_STREAM_TIMEOUT if the given data is not sent successfully
+ *          (this doesn't mean that the data is never sent, the receiver may
+ *          have read the data but its ACKs may have been lost);
+ *          GNUNET_STREAM_SHUTDOWN if the stream is shutdown for writing in the
+ *          mean time; GNUNET_STREAM_SYSERR if the stream is broken and cannot
+ *          be processed.
+ * @param size the number of bytes written
  */
-void
-conclude_round_done (struct ConsensusSession *session)
+static void 
+write_values (void *cls, enum GNUNET_STREAM_Status status, size_t size)
 {
-  /* FIXME */
+  struct ConsensusPeerInformation *cpi;
+
+  cpi = (struct ConsensusPeerInformation *) cls;
 }
 
 
 /**
  * Called when a client performs the conclude operation.
+ *
+ * @param cls (unused)
+ * @param client client handle
+ * @param message message sent by the client
  */
 void
 client_conclude (void *cls,
@@ -550,40 +1265,55 @@ client_conclude (void *cls,
              const struct GNUNET_MessageHeader *message)
 {
   struct ConsensusSession *session;
+  int i;
 
   GNUNET_log (GNUNET_ERROR_TYPE_INFO, "conclude requested\n");
 
   session = sessions_head;
   while ((session != NULL) && (session->client != client))
-  {
     session = session->next;
-  }
   if (NULL == session)
   {
-    GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "client not found\n");
+    /* client not found */
+    GNUNET_break (0);
     GNUNET_SERVER_client_disconnect (client);
     return;
   }
 
   if (GNUNET_YES == session->conclude_requested)
   {
-    GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "client requested conclude twice\n");
-    GNUNET_SERVER_client_disconnect (client);
+    /* client requested conclude twice */
+    GNUNET_break (0);
+    disconnect_client (client);
     return;
   }
 
   session->conclude_requested = GNUNET_YES;
 
-  conclude_do_round (session);
+  /* FIXME: write to already connected sockets */
 
-  GNUNET_SERVER_receive_done (client, GNUNET_OK);
+  for (i = 0; i < session->num_peers; i++)
+  {
+    if ( (GNUNET_YES == session->info[i].is_outgoing) &&
+         (GNUNET_YES == session->info[i].hello) )
+    {
+      /* kick off transmitting strata by calling the write continuation */
+      write_strata (&session->info[i], GNUNET_STREAM_OK, 0);
+    }
+  }
+  
 
+  GNUNET_SERVER_receive_done (client, GNUNET_OK);
   send_next (session);
 }
 
 
 /**
  * Called when a client sends an ack
+ *
+ * @param cls (unused)
+ * @param client client handle
+ * @param message message sent by the client
  */
 void
 client_ack (void *cls,
@@ -614,71 +1344,23 @@ core_startup (void *cls,
               struct GNUNET_CORE_Handle *core,
               const struct GNUNET_PeerIdentity *peer)
 {
-  static const struct GNUNET_SERVER_MessageHandler handlers[] = {
-    {&client_join, NULL, GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_JOIN, 0},
-    {&client_insert, NULL, GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_INSERT, 0},
-    {&client_conclude, NULL, GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_CONCLUDE,
-        sizeof (struct GNUNET_CONSENSUS_ConcludeMessage)},
-    {&client_ack, NULL, GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_ACK,
-        sizeof (struct GNUNET_CONSENSUS_AckMessage)},
-    {NULL, NULL, 0, 0}
-  };
+  struct ConsensusSession *session;
 
-  GNUNET_SERVER_add_handlers (srv, handlers);
   my_peer = GNUNET_memdup(peer, sizeof (struct GNUNET_PeerIdentity));
   /* core can't be disconnected directly in the core startup callback, schedule a task to do it! */
   GNUNET_SCHEDULER_add_now (&disconnect_core, core);
   GNUNET_log(GNUNET_ERROR_TYPE_INFO, "connected to core\n");
-}
-
-
-
-/**
- * Method called whenever another peer has added us to a tunnel
- * the other peer initiated.
- * Only called (once) upon reception of data with a message type which was
- * subscribed to in GNUNET_MESH_connect. A call to GNUNET_MESH_tunnel_destroy
- * causes te tunnel to be ignored and no further notifications are sent about
- * the same tunnel.
- *
- * @param cls closure
- * @param tunnel new handle to the tunnel
- * @param initiator peer that started the tunnel
- * @param atsi performance information for the tunnel
- * @return initial tunnel context for the tunnel
- *         (can be NULL -- that's not an error)
- */
-static void *
-new_tunnel (void *cls,
-            struct GNUNET_MESH_Tunnel *tunnel,
-            const struct GNUNET_PeerIdentity *initiator,
-            const struct GNUNET_ATS_Information *atsi)
-{
-  /* there's nothing we can do here, as we don't have the global consensus id yet */
-  return NULL;
-}
 
-
-/**
- * Function called whenever an inbound tunnel is destroyed.  Should clean up
- * any associated state.  This function is NOT called if the client has
- * explicitly asked for the tunnel to be destroyed using
- * GNUNET_MESH_tunnel_destroy. It must NOT call GNUNET_MESH_tunnel_destroy on
- * the tunnel.
- *
- * @param cls closure (set from GNUNET_MESH_connect)
- * @param tunnel connection to the other end (henceforth invalid)
- * @param tunnel_ctx place where local state associated
- *                   with the tunnel is stored
- */
-static void
-cleaner (void *cls, const struct GNUNET_MESH_Tunnel *tunnel, void *tunnel_ctx)
-{
-  /* FIXME: what to do here? */
+  session = sessions_head;
+  while (NULL != session)
+  {
+    if (NULL != session->join_msg)
+      initialize_session (session);
+    session = session->next;
+  }
 }
 
 
-
 /**
  * Called to clean up, after a shutdown has been requested.
  *
@@ -689,105 +1371,27 @@ static void
 shutdown_task (void *cls,
                const struct GNUNET_SCHEDULER_TaskContext *tc)
 {
-  /* mesh requires all the tunnels to be destroyed manually */
   while (NULL != sessions_head)
   {
     struct ConsensusSession *session;
     session = sessions_head;
-    GNUNET_MESH_tunnel_destroy (sessions_head->broadcast_tunnel);
     sessions_head = sessions_head->next;
     GNUNET_free (session);
   }
 
-  if (NULL != mesh)
-  {
-    GNUNET_MESH_disconnect (mesh);
-    mesh = NULL;
-  }
   if (NULL != core)
   {
     GNUNET_CORE_disconnect (core);
     core = NULL;
   }
-}
-
-
-
-/**
- * Functions with this signature are called whenever a message is
- * received.
- *
- * @param cls closure (set from GNUNET_MESH_connect)
- * @param tunnel connection to the other end
- * @param tunnel_ctx place to store local state associated with the tunnel
- * @param sender who sent the message
- * @param message the actual message
- * @param atsi performance data for the connection
- * @return GNUNET_OK to keep the connection open,
- *         GNUNET_SYSERR to close it (signal serious error)
- */
-static int
-p2p_delta_estimate (void *cls,
-                    struct GNUNET_MESH_Tunnel * tunnel,
-                    void **tunnel_ctx,
-                    const struct GNUNET_PeerIdentity *sender,
-                    const struct GNUNET_MessageHeader *message,
-                    const struct GNUNET_ATS_Information *atsi)
-{
-  /* FIXME */
-  return GNUNET_OK;
-}
-
-
-/**
- * Functions with this signature are called whenever a message is
- * received.
- *
- * @param cls closure (set from GNUNET_MESH_connect)
- * @param tunnel connection to the other end
- * @param tunnel_ctx place to store local state associated with the tunnel
- * @param sender who sent the message
- * @param message the actual message
- * @param atsi performance data for the connection
- * @return GNUNET_OK to keep the connection open,
- *         GNUNET_SYSERR to close it (signal serious error)
- */
-static int
-p2p_difference_digest (void *cls,
-                       struct GNUNET_MESH_Tunnel * tunnel,
-                       void **tunnel_ctx,
-                       const struct GNUNET_PeerIdentity *sender,
-                       const struct GNUNET_MessageHeader *message,
-                       const struct GNUNET_ATS_Information *atsi)
-{
-  /* FIXME */
-  return GNUNET_OK;
-}
 
+  if (NULL != listener)
+  {
+    GNUNET_STREAM_listen_close (listener);
+    listener = NULL;
+  } 
 
-/**
- * Functions with this signature are called whenever a message is
- * received.
- *
- * @param cls closure (set from GNUNET_MESH_connect)
- * @param tunnel connection to the other end
- * @param tunnel_ctx place to store local state associated with the tunnel
- * @param sender who sent the message
- * @param message the actual message
- * @param atsi performance data for the connection
- * @return GNUNET_OK to keep the connection open,
- *         GNUNET_SYSERR to close it (signal serious error)
- */
-static int
-p2p_elements_and_requests (void *cls,
-                           struct GNUNET_MESH_Tunnel * tunnel,
-                           void **tunnel_ctx,
-                           const struct GNUNET_PeerIdentity *sender,
-                           const struct GNUNET_MessageHeader *message,
-                           const struct GNUNET_ATS_Information *atsi)
-{
-  /* FIXME */
-  return GNUNET_OK;
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "handled shutdown request\n");
 }
 
 
@@ -801,33 +1405,38 @@ p2p_elements_and_requests (void *cls,
 static void
 run (void *cls, struct GNUNET_SERVER_Handle *server, const struct GNUNET_CONFIGURATION_Handle *c)
 {
-  static const struct GNUNET_CORE_MessageHandler handlers[] = {
+  static const struct GNUNET_CORE_MessageHandler core_handlers[] = {
     {NULL, 0, 0}
   };
-  static const struct GNUNET_MESH_MessageHandler mesh_handlers[] = {
-    {p2p_delta_estimate, GNUNET_MESSAGE_TYPE_CONSENSUS_P2P_DELTA_ESTIMATE, 0},
-    {p2p_difference_digest, GNUNET_MESSAGE_TYPE_CONSENSUS_P2P_DIFFERENCE_DIGEST, 0},
-    {p2p_elements_and_requests, GNUNET_MESSAGE_TYPE_CONSENSUS_P2P_ELEMENTS_AND_REQUESTS, 0},
-    {NULL, 0, 0}
-  };
-  static const GNUNET_MESH_ApplicationType app_types[] = { 
-    GNUNET_APPLICATION_TYPE_CONSENSUS,
-    GNUNET_APPLICATION_TYPE_END
+  static const struct GNUNET_SERVER_MessageHandler server_handlers[] = {
+    {&client_join, NULL, GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_JOIN, 0},
+    {&client_insert, NULL, GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_INSERT, 0},
+    {&client_conclude, NULL, GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_CONCLUDE,
+        sizeof (struct GNUNET_CONSENSUS_ConcludeMessage)},
+    {&client_ack, NULL, GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_ACK,
+        sizeof (struct GNUNET_CONSENSUS_AckMessage)},
+    {NULL, NULL, 0, 0}
   };
 
-  GNUNET_log(GNUNET_ERROR_TYPE_INFO, "consensus running\n");
-
   cfg = c;
   srv = server;
 
+  GNUNET_SERVER_add_handlers (server, server_handlers);
+
   GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_FOREVER_REL, &shutdown_task, NULL);
 
-  mesh = GNUNET_MESH_connect (cfg, NULL, new_tunnel, cleaner, mesh_handlers, app_types);
-  GNUNET_assert (NULL != mesh);
+
+  listener = GNUNET_STREAM_listen (cfg, GNUNET_APPLICATION_TYPE_CONSENSUS,
+                                   listen_cb, NULL,
+                                   GNUNET_STREAM_OPTION_END);
+
 
   /* we have to wait for the core_startup callback before proceeding with the consensus service startup */
-  core = GNUNET_CORE_connect (c, NULL, &core_startup, NULL, NULL, NULL, GNUNET_NO, NULL, GNUNET_NO, handlers);
+  core = GNUNET_CORE_connect (c, NULL, &core_startup, NULL, NULL, NULL, GNUNET_NO, NULL, GNUNET_NO, core_handlers);
   GNUNET_assert (NULL != core);
+
+  GNUNET_log(GNUNET_ERROR_TYPE_INFO, "consensus running\n");
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "strata per msg: %d\n", STRATA_PER_MESSAGE);
 }
 
 
@@ -843,6 +1452,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");
   return (GNUNET_OK == ret) ? 0 : 1;
 }
 
diff --git a/src/consensus/ibf.c b/src/consensus/ibf.c
index 4f1aca939..2d06fc29b 100644
--- a/src/consensus/ibf.c
+++ b/src/consensus/ibf.c
@@ -25,50 +25,8 @@
  * @author Florian Dold
  */
 
-#include "ibf.h"
-
-
-/**
- * Opaque handle to an invertible bloom filter (IBF).
- *
- * An IBF is a counting bloom filter that has the ability to restore
- * the hashes of its stored elements with high probability.
- */
-struct InvertibleBloomFilter
-{
-  /**
-   * How many cells does this IBF have?
-   */
-  unsigned int size;
 
-  /**
-   * In how many cells do we hash one element?
-   * Usually 4 or 3.
-   */
-  unsigned int hash_num;
-
-  /**
-   * Salt for mingling hashes
-   */
-  uint32_t salt;
-
-  /**
-   * How many times has a bucket been hit?
-   * Can be negative, as a result of IBF subtraction.
-   */
-  int8_t *count;
-
-  /**
-   * xor sums of the elements' hash codes, used to identify the elements.
-   */
-  struct GNUNET_HashCode *id_sum;
-
-  /**
-   * xor sums of the "hash of the hash".
-   */
-  struct GNUNET_HashCode *hash_sum;
-
-};
+#include "ibf.h"
 
 
 /**
@@ -152,6 +110,8 @@ ibf_insert_on_side (struct InvertibleBloomFilter *ibf,
       used_buckets[i] = bucket;
       
       ibf->count[bucket] += side;
+
+      GNUNET_log_from(GNUNET_ERROR_TYPE_INFO, "ibf", "inserting in bucket %d \n", bucket);
       
       GNUNET_CRYPTO_hash_xor (&key_copy, &ibf->id_sum[bucket],
                               &ibf->id_sum[bucket]);
@@ -214,8 +174,6 @@ ibf_decode (struct InvertibleBloomFilter *ibf,
   int i;
 
   GNUNET_assert (NULL != ibf);
-  GNUNET_assert (NULL != ret_id);
-  GNUNET_assert (NULL != ret_side);
 
   for (i = 0; i < ibf->size; i++)
   {
@@ -227,8 +185,10 @@ ibf_decode (struct InvertibleBloomFilter *ibf,
     if (0 != memcmp (&hash, &ibf->hash_sum[i], sizeof (struct GNUNET_HashCode)))
       continue;
 
-    *ret_side = ibf->count[i];
-    *ret_id = ibf->id_sum[i];
+    if (NULL != ret_side)
+      *ret_side = ibf->count[i];
+    if (NULL != ret_id)
+      *ret_id = ibf->id_sum[i];
 
     /* insert on the opposite side, effectively removing the element */
     ibf_insert_on_side (ibf, &ibf->id_sum[i], -ibf->count[i]);
@@ -269,3 +229,36 @@ ibf_subtract (struct InvertibleBloomFilter *ibf1, struct InvertibleBloomFilter *
   }
 }
 
+/**
+ * Create a copy of an IBF, the copy has to be destroyed properly.
+ *
+ * @param ibf the IBF to copy
+ */
+struct InvertibleBloomFilter *
+ibf_dup (struct InvertibleBloomFilter *ibf)
+{
+  struct InvertibleBloomFilter *copy;
+  copy = GNUNET_malloc (sizeof *copy);
+  copy->hash_num = ibf->hash_num;
+  copy->salt = ibf->salt;
+  copy->size = ibf->size;
+  copy->hash_sum = GNUNET_memdup (ibf->hash_sum, ibf->size * sizeof (struct GNUNET_HashCode));
+  copy->id_sum = GNUNET_memdup (ibf->id_sum, ibf->size * sizeof (struct GNUNET_HashCode));
+  copy->count = GNUNET_memdup (ibf->count, ibf->size * sizeof (uint8_t));
+  return copy;
+}
+
+/**
+ * Destroy all resources associated with the invertible bloom filter.
+ * No more ibf_*-functions may be called on ibf after calling destroy.
+ *
+ * @param ibf the intertible bloom filter to destroy
+ */
+void
+ibf_destroy (struct InvertibleBloomFilter *ibf)
+{
+  GNUNET_free (ibf->hash_sum);
+  GNUNET_free (ibf->id_sum);
+  GNUNET_free (ibf->count);
+  GNUNET_free (ibf);
+}
diff --git a/src/consensus/ibf.h b/src/consensus/ibf.h
index e43d7c5fe..26f101905 100644
--- a/src/consensus/ibf.h
+++ b/src/consensus/ibf.h
@@ -39,14 +39,52 @@ extern "C"
 #endif
 #endif
 
+/**
+ * Size of one ibf bucket in bytes
+ */
+#define IBF_BUCKET_SIZE (64+64+1)
+
 
 /**
- * Opaque handle to an invertible bloom filter (IBF).
+ * Invertible bloom filter (IBF).
  *
  * An IBF is a counting bloom filter that has the ability to restore
  * the hashes of its stored elements with high probability.
  */
-struct InvertibleBloomFilter;
+struct InvertibleBloomFilter
+{
+  /**
+   * How many cells does this IBF have?
+   */
+  unsigned int size;
+
+  /**
+   * In how many cells do we hash one element?
+   * Usually 4 or 3.
+   */
+  unsigned int hash_num;
+
+  /**
+   * Salt for mingling hashes
+   */
+  uint32_t salt;
+
+  /**
+   * xor sums of the elements' hash codes, used to identify the elements.
+   */
+  struct GNUNET_HashCode *id_sum;
+
+  /**
+   * xor sums of the "hash of the hash".
+   */
+  struct GNUNET_HashCode *hash_sum;
+
+  /**
+   * How many times has a bucket been hit?
+   * Can be negative, as a result of IBF subtraction.
+   */
+  int8_t *count;
+};
 
 
 /**
@@ -106,15 +144,6 @@ ibf_decode (struct InvertibleBloomFilter *ibf, int *side, struct GNUNET_HashCode
 struct InvertibleBloomFilter *
 ibf_dup (struct InvertibleBloomFilter *ibf);
 
-
-/*
-ibf_hton ();
-
-ibf_ntoh ();
-
-ibf_get_nbo_size ();
-*/
-
 /**
  * Destroy all resources associated with the invertible bloom filter.
  * No more ibf_*-functions may be called on ibf after calling destroy.
diff --git a/src/consensus/test_consensus.conf b/src/consensus/test_consensus.conf
index 86dfadb9b..01266c2a9 100644
--- a/src/consensus/test_consensus.conf
+++ b/src/consensus/test_consensus.conf
@@ -1,6 +1,6 @@
 [consensus]
 AUTOSTART = YES
-PORT = 2103
+PORT = 2110
 HOSTNAME = localhost
 HOME = $SERVICEHOME
 BINARY = gnunet-service-consensus
diff --git a/src/consensus/test_consensus_api.c b/src/consensus/test_consensus_api.c
index e752983c2..c79a6221c 100644
--- a/src/consensus/test_consensus_api.c
+++ b/src/consensus/test_consensus_api.c
@@ -17,6 +17,7 @@
      Free Software Foundation, Inc., 59 Temple Place - Suite 330,
      Boston, MA 02111-1307, USA.
 */
+
 /**
  * @file consensus/test_consensus_api.c
  * @brief testcase for consensus_api.c
@@ -29,18 +30,20 @@
 
 static struct GNUNET_CONSENSUS_Handle *consensus;
 
-static int insert;
-
 static struct GNUNET_HashCode session_id;
 
 
-static void conclude_done (void *cls, 
-                           unsigned int num_peers_in_consensus,
-                           const struct GNUNET_PeerIdentity *peers_in_consensus)
+static int
+conclude_done (void *cls, const struct GNUNET_CONSENSUS_Group *group)
 {
-  struct GNUNET_CONSENSUS_Handle *consensus;
-  consensus = (struct GNUNET_CONSENSUS_Handle *) cls;
-  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "concluded\n");
+  if (NULL == group)
+  {
+    GNUNET_log (GNUNET_ERROR_TYPE_INFO, "conclude over\n");
+    GNUNET_SCHEDULER_shutdown ();
+    return GNUNET_NO;
+  }
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "concluded\n");
+  return GNUNET_YES;
 }
 
 static int
@@ -54,7 +57,30 @@ on_new_element (void *cls,
 static void
 insert_done (void *cls, int success)
 {
+  /* make sure cb is only called once */
+  static int called = GNUNET_NO;
+  GNUNET_assert (GNUNET_NO == called);
+  called = GNUNET_YES;
   GNUNET_log (GNUNET_ERROR_TYPE_INFO, "insert done\n");
+  GNUNET_CONSENSUS_conclude (consensus, GNUNET_TIME_UNIT_SECONDS, 0, &conclude_done, NULL);
+}
+
+
+/**
+ * Signature of the main function of a task.
+ *
+ * @param cls closure
+ * @param tc context information (why was this task triggered now)
+ */
+static void
+on_shutdown (void *cls,
+          const struct GNUNET_SCHEDULER_TaskContext *tc)
+{
+  if (NULL != consensus)
+  {
+    GNUNET_CONSENSUS_destroy (consensus);
+    consensus = NULL;
+  }
 }
 
 
@@ -69,18 +95,19 @@ run (void *cls,
   struct GNUNET_CONSENSUS_Element el2 = {"bar", 4, 0};
 
   GNUNET_log_setup ("test_consensus_api",
-                    "DEBUG",
+                    "INFO",
                     NULL);
 
   GNUNET_log (GNUNET_ERROR_TYPE_INFO, "testing consensus api\n");
 
+  GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_FOREVER_REL, &on_shutdown, NULL);
+
   GNUNET_CRYPTO_hash (str, strlen (str), &session_id);
   consensus = GNUNET_CONSENSUS_create (cfg, 0, NULL, &session_id, on_new_element, &consensus);
   GNUNET_assert (consensus != NULL);
-  /*
-  GNUNET_CONSENSUS_insert (consensus1, &el1, &insert_done, &consensus1);
-  GNUNET_CONSENSUS_insert (consensus2, &el2, &insert_done, &consensus2);
-  */
+
+  GNUNET_CONSENSUS_insert (consensus, &el1, NULL, &consensus);
+  GNUNET_CONSENSUS_insert (consensus, &el2, &insert_done, &consensus);
 }