consensus api, consensus service (local), peer driver and ibf sketch

git-svn-id: https://gnunet.org/svn/gnunet@25275 140774ce-b5e7-0310-ab8b-a85725594a96

9 files changed, 1043 insertions, 248 deletions

diff --git a/src/consensus/Makefile.am b/src/consensus/Makefile.am
index 10b22cc877..29c466901a 100644
--- a/src/consensus/Makefile.am
+++ b/src/consensus/Makefile.am
@@ -16,7 +16,8 @@ if USE_COVERAGE
 endif
 
 bin_PROGRAMS = \
- gnunet-consensus 
+ gnunet-consensus \
+ gnunet-consensus-start-peers
 
 libexec_PROGRAMS = \
  gnunet-service-consensus
@@ -31,6 +32,14 @@ gnunet_consensus_LDADD = \
   $(top_builddir)/src/consensus/libgnunetconsensus.la \
   $(GN_LIBINTL)
 
+gnunet_consensus_start_peers_SOURCES = \
+ gnunet-consensus-start-peers.c
+gnunet_consensus_start_peers_LDADD = \
+  $(top_builddir)/src/util/libgnunetutil.la \
+  $(top_builddir)/src/testbed/libgnunettestbed.la \
+  $(top_builddir)/src/consensus/libgnunetconsensus.la \
+  $(GN_LIBINTL)
+
 gnunet_service_consensus_SOURCES = \
  gnunet-service-consensus.c
 gnunet_service_consensus_LDADD = \
diff --git a/src/consensus/consensus.h b/src/consensus/consensus.h
index 2762e8ff44..d76c6b769e 100644
--- a/src/consensus/consensus.h
+++ b/src/consensus/consensus.h
@@ -90,6 +90,20 @@ struct GNUNET_CONSENSUS_ElementMessage
   /* rest: element data */
 };
 
+struct GNUNET_CONSENSUS_AckMessage
+{
+  /**
+   * Type: GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_ACK
+   */
+  struct GNUNET_MessageHeader header;
+
+  /**
+   * Do we want to keep and propagate the element?
+   */
+  uint8_t keep;
+
+};
+
 GNUNET_NETWORK_STRUCT_END
 
 #endif
diff --git a/src/consensus/consensus_api.c b/src/consensus/consensus_api.c
index 90b0fdf16d..2479c019c5 100644
--- a/src/consensus/consensus_api.c
+++ b/src/consensus/consensus_api.c
@@ -24,6 +24,7 @@
  * @author Florian Dold
  */
 #include "platform.h"
+#include "gnunet_util_lib.h"
 #include "gnunet_protocols.h"
 #include "gnunet_client_lib.h"
 #include "gnunet_consensus_service.h"
@@ -32,6 +33,13 @@
 
 #define LOG(kind,...) GNUNET_log_from (kind, "consensus-api",__VA_ARGS__)
 
+struct ElementAck
+{
+  struct ElementAck *next;
+  struct ElementAck *prev;
+  int keep;
+  struct GNUNET_CONSENSUS_Element *element;
+};
 
 /**
  * Handle for the service.
@@ -113,20 +121,138 @@ struct GNUNET_CONSENSUS_Handle
    * Deadline for the conclude operation.
    */
   struct GNUNET_TIME_Absolute conclude_deadline;
+
+  struct ElementAck *ack_head;
+  struct ElementAck *ack_tail;
+
+  /**
+   * Set to GNUNET_YES if the begin message has been transmitted to the service
+   */
+  int begin_sent;
+
+  /**
+   * Set to GNUNET_YES it the begin message should be transmitted to the service
+   */
+  int begin_requested;
 };
 
 
+static size_t
+transmit_ack (void *cls, size_t size, void *buf);
+
+static size_t
+transmit_insert (void *cls, size_t size, void *buf);
+
+static size_t
+transmit_conclude (void *cls, size_t size, void *buf);
+
+static size_t
+transmit_begin (void *cls, size_t size, void *buf);
+
+
+/**
+ * Call notify_transmit_ready for ack if necessary and possible.
+ */
+static void
+ntr_ack (struct GNUNET_CONSENSUS_Handle *consensus)
+{
+  if ((NULL == consensus->th) && (NULL != consensus->ack_head))
+  {
+    consensus->th =
+        GNUNET_CLIENT_notify_transmit_ready (consensus->client,
+                                             sizeof (struct GNUNET_CONSENSUS_AckMessage),
+                                             GNUNET_TIME_UNIT_FOREVER_REL,
+                                             GNUNET_NO, &transmit_ack, consensus);
+  }
+}
+
+
+/**
+ * Call notify_transmit_ready for ack if necessary and possible.
+ */
+static void
+ntr_insert (struct GNUNET_CONSENSUS_Handle *consensus)
+{
+  if ((NULL == consensus->th) && (NULL != consensus->insert_element))
+  {
+    consensus->th =
+        GNUNET_CLIENT_notify_transmit_ready (consensus->client,
+                                             sizeof (struct GNUNET_CONSENSUS_ElementMessage) + 
+                                                consensus->insert_element->size,
+                                             GNUNET_TIME_UNIT_FOREVER_REL,
+                                             GNUNET_NO, &transmit_insert, consensus);
+  }
+}
+
+
+/**
+ * Call notify_transmit_ready for ack if necessary and possible.
+ */
+static void
+ntr_conclude (struct GNUNET_CONSENSUS_Handle *consensus)
+{
+  if ((NULL == consensus->th) && (NULL != consensus->conclude_cb))
+  {
+    consensus->th =
+        GNUNET_CLIENT_notify_transmit_ready (consensus->client,
+                                             sizeof (struct GNUNET_CONSENSUS_ConcludeMessage),
+                                             GNUNET_TIME_absolute_get_remaining (consensus->conclude_deadline),
+                                             GNUNET_NO, &transmit_conclude, consensus);
+  }
+}
+
+
+/**
+ * Call notify_transmit_ready for ack if necessary and possible.
+ */
+static void
+ntr_begin (struct GNUNET_CONSENSUS_Handle *consensus)
+{
+  if ((NULL == consensus->th) && (GNUNET_YES == consensus->begin_requested) &&
+      (GNUNET_NO == consensus->begin_sent))
+  {
+    consensus->th =
+        GNUNET_CLIENT_notify_transmit_ready (consensus->client,
+                                             sizeof (struct GNUNET_MessageHeader),
+                                             GNUNET_TIME_UNIT_FOREVER_REL,
+                                             GNUNET_NO, &transmit_begin, consensus);
+  }
+}
+
+/**
+ * Called when the server has sent is a new element
+ * 
+ * @param consensus consensus handle
+ * @param msg element message
+ */
 static void
 handle_new_element(struct GNUNET_CONSENSUS_Handle *consensus,
                    struct GNUNET_CONSENSUS_ElementMessage *msg)
 {
   struct GNUNET_CONSENSUS_Element element;
+  struct ElementAck *ack;
+  int ret;
+
   element.type = msg->element_type;
   element.size = msg->header.size - sizeof (struct GNUNET_CONSENSUS_ElementMessage);
   element.data = &msg[1];
-  consensus->new_element_cb (consensus->new_element_cls, &element);
+
+  ret = consensus->new_element_cb (consensus->new_element_cls, &element);
+  ack = GNUNET_malloc (sizeof (struct ElementAck));
+  ack->keep = ret;
+  GNUNET_CONTAINER_DLL_insert_tail (consensus->ack_head, consensus->ack_tail,ack);
+
+  ntr_ack (consensus);
 }
 
+
+/**
+ * Called when the server has announced
+ * that the conclusion is over.
+ * 
+ * @param consensus consensus handle
+ * @param msg conclude done message
+ */
 static void
 handle_conclude_done(struct GNUNET_CONSENSUS_Handle *consensus,
                      struct GNUNET_CONSENSUS_ConcludeDoneMessage *msg)
@@ -170,7 +296,7 @@ message_handler (void *cls, const struct GNUNET_MessageHeader *msg)
     return;
   }
 
-  switch (ntohs(msg->type))
+  switch (ntohs (msg->type))
   {
     case GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_RECEIVED_ELEMENT:
       handle_new_element (consensus, (struct GNUNET_CONSENSUS_ElementMessage *) msg);
@@ -200,6 +326,43 @@ message_handler (void *cls, const struct GNUNET_MessageHeader *msg)
  * @return number of bytes written to buf
  */
 static size_t
+transmit_ack (void *cls, size_t size, void *buf)
+{
+  struct GNUNET_CONSENSUS_AckMessage *msg;
+  struct GNUNET_CONSENSUS_Handle *consensus;
+
+  consensus = (struct GNUNET_CONSENSUS_Handle *) cls;
+
+  GNUNET_assert (NULL != consensus->ack_head);
+
+  msg = (struct GNUNET_CONSENSUS_AckMessage *) buf;
+  msg->keep = consensus->ack_head->keep;
+  msg->header.type = htons (GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_ACK);
+  msg->header.size = htons (sizeof (struct GNUNET_CONSENSUS_AckMessage));
+
+  consensus->ack_head = consensus->ack_head->next;
+
+  consensus->th = NULL;
+
+  ntr_insert (consensus);
+  ntr_ack (consensus);
+  ntr_conclude (consensus);
+
+  return sizeof (struct GNUNET_CONSENSUS_AckMessage);
+}
+
+/**
+ * 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 closure
+ * @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_insert (void *cls, size_t size, void *buf)
 {
   struct GNUNET_CONSENSUS_ElementMessage *msg;
@@ -227,6 +390,7 @@ transmit_insert (void *cls, size_t size, void *buf)
           consensus->insert_element->data,
           consensus->insert_element->size);
 
+  consensus->insert_element = NULL;
 
   idc = consensus->idc;
   consensus->idc = NULL;
@@ -234,6 +398,11 @@ transmit_insert (void *cls, size_t size, void *buf)
   consensus->idc_cls = NULL;
   idc (idc_cls, GNUNET_YES);
 
+
+  ntr_ack (consensus);
+  ntr_insert (consensus);
+  ntr_conclude (consensus);
+
   return msize;
 }
 
@@ -273,18 +442,14 @@ transmit_join (void *cls, size_t size, void *buf)
   msg->header.size = htons (msize);
   msg->session_id = consensus->session_id;
   msg->num_peers = htons (consensus->num_peers);
-  memcpy(&msg[1],
-         consensus->peers,
-         consensus->num_peers * sizeof (struct GNUNET_PeerIdentity));
+  if (0 != msg->num_peers)
+    memcpy(&msg[1],
+           consensus->peers,
+           consensus->num_peers * sizeof (struct GNUNET_PeerIdentity));
 
-  if (consensus->insert_element != NULL)
-  {
-    consensus->th =
-        GNUNET_CLIENT_notify_transmit_ready (consensus->client,
-                                             msize,
-                                             GNUNET_TIME_UNIT_FOREVER_REL,
-                                             GNUNET_NO, &transmit_insert, consensus);
-  }
+  ntr_insert (consensus);
+  ntr_begin (consensus);
+  ntr_conclude (consensus);
 
   GNUNET_CLIENT_receive (consensus->client, &message_handler, consensus,
                          GNUNET_TIME_UNIT_FOREVER_REL);
@@ -325,6 +490,8 @@ transmit_conclude (void *cls, size_t size, void *buf)
   msg->timeout =
       GNUNET_TIME_relative_hton (GNUNET_TIME_absolute_get_remaining(consensus->conclude_deadline));
 
+  ntr_ack (consensus);
+
   return msize;
 }
 
@@ -359,6 +526,10 @@ transmit_begin (void *cls, size_t size, void *buf)
   msg->type = htons (GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_BEGIN);
   msg->size = htons (msize);
 
+  ntr_ack (consensus);
+  ntr_insert (consensus);
+  ntr_conclude (consensus);
+
   return msize;
 }
 
@@ -421,8 +592,8 @@ GNUNET_CONSENSUS_create (const struct GNUNET_CONFIGURATION_Handle *cfg,
                                            GNUNET_TIME_UNIT_FOREVER_REL,
                                            GNUNET_NO, &transmit_join, consensus);
 
-  GNUNET_assert (consensus->th != NULL);
 
+  GNUNET_assert (consensus->th != NULL);
   return consensus;
 }
 
@@ -444,9 +615,9 @@ GNUNET_CONSENSUS_insert (struct GNUNET_CONSENSUS_Handle *consensus,
 			 GNUNET_CONSENSUS_InsertDoneCallback idc,
 			 void *idc_cls)
 {
-
   GNUNET_assert (NULL == consensus->idc);
   GNUNET_assert (NULL == consensus->insert_element);
+  GNUNET_assert (NULL == consensus->conclude_cb);
 
   consensus->idc = idc;
   consensus->idc_cls = idc_cls;
@@ -454,17 +625,10 @@ GNUNET_CONSENSUS_insert (struct GNUNET_CONSENSUS_Handle *consensus,
 
   if (consensus->joined == 0)
   {
-    GNUNET_assert (NULL != consensus->th);
     return;
   }
 
-  GNUNET_assert (NULL == consensus->th);
-
-  consensus->th =
-      GNUNET_CLIENT_notify_transmit_ready (consensus->client,
-                                           element->size + sizeof (struct GNUNET_CONSENSUS_ElementMessage),
-                                           GNUNET_TIME_UNIT_FOREVER_REL,
-                                           GNUNET_NO, &transmit_insert, consensus);
+  ntr_insert (consensus);
 }
 
 
@@ -478,12 +642,12 @@ GNUNET_CONSENSUS_begin (struct GNUNET_CONSENSUS_Handle *consensus)
 {
   GNUNET_assert (NULL == consensus->idc);
   GNUNET_assert (NULL == consensus->insert_element);
+  GNUNET_assert (GNUNET_NO == consensus->begin_requested);
+  GNUNET_assert (GNUNET_NO == consensus->begin_sent);
 
-  consensus->th =
-      GNUNET_CLIENT_notify_transmit_ready (consensus->client,
-                                           sizeof (struct GNUNET_MessageHeader),
-                                           GNUNET_TIME_UNIT_FOREVER_REL,
-                                           GNUNET_NO, &transmit_begin, consensus);
+  consensus->begin_requested = GNUNET_YES;
+
+  ntr_begin (consensus);
 }
 
 
@@ -503,22 +667,17 @@ GNUNET_CONSENSUS_conclude (struct GNUNET_CONSENSUS_Handle *consensus,
 			   GNUNET_CONSENSUS_ConcludeCallback conclude,
 			   void *conclude_cls)
 {
-  GNUNET_assert (NULL == consensus->th);
+  GNUNET_assert (NULL != conclude);
   GNUNET_assert (NULL == consensus->conclude_cb);
 
   consensus->conclude_cls = conclude_cls;
   consensus->conclude_cb = conclude;
   consensus->conclude_deadline = GNUNET_TIME_relative_to_absolute(timeout);
 
-  consensus->th =
-      GNUNET_CLIENT_notify_transmit_ready (consensus->client,
-                                           sizeof (struct GNUNET_CONSENSUS_ConcludeMessage),
-                                           timeout,
-                                           GNUNET_NO, &transmit_conclude, consensus);
-  if (NULL == consensus->th)
-  {
-    conclude(conclude_cls, 0, NULL);
-  }
+
+  /* if transmitting the conclude message is not possible right now, transmit_join
+   * or transmit_ack will handle it */
+  ntr_conclude (consensus);
 }
 
 
@@ -536,7 +695,8 @@ GNUNET_CONSENSUS_destroy (struct GNUNET_CONSENSUS_Handle *consensus)
     GNUNET_CLIENT_disconnect (consensus->client);
     consensus->client = NULL;
   }
-  GNUNET_free (consensus->peers);
+  if (NULL != consensus->peers)
+    GNUNET_free (consensus->peers);
   GNUNET_free (consensus);
 }
 
diff --git a/src/consensus/gnunet-consensus-start-peers.c b/src/consensus/gnunet-consensus-start-peers.c
new file mode 100644
index 0000000000..19eec7744a
--- /dev/null
+++ b/src/consensus/gnunet-consensus-start-peers.c
@@ -0,0 +1,172 @@
+
+/*
+      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/gnunet-consensus-start-peers.c
+ * @brief Starts peers with testebed on localhost,
+ *        prints their configuration files and waits for ^C.
+ * @author Florian Dold
+ */
+#include "platform.h"
+#include "gnunet_util_lib.h"
+#include "gnunet_testbed_service.h"
+
+
+static char *config_template_file;
+static unsigned int num_peers_requested = 2;
+static struct GNUNET_TESTBED_Peer **peers;
+
+
+/**
+ * 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
+peer_info_cb (void *cb_cls,
+              struct GNUNET_TESTBED_Operation
+              *op,
+              const struct
+              GNUNET_TESTBED_PeerInformation
+              *pinfo,
+              const char *emsg)
+{
+  GNUNET_assert (NULL == emsg);
+  if (pinfo->pit == GNUNET_TESTBED_PIT_IDENTITY)
+  {
+    struct GNUNET_CRYPTO_HashAsciiEncoded enc;
+    GNUNET_CRYPTO_hash_to_enc (&pinfo->result.id->hashPubKey, &enc);
+    printf("peer %td identity:\n", ((struct GNUNET_TESTBED_Peer **) cb_cls) - &peers[0]);
+    printf("%s\n", (char *)&enc);
+  }
+  else if (pinfo->pit == GNUNET_TESTBED_PIT_CONFIGURATION)
+  {
+    char *tmpfilename;
+    if (NULL == (tmpfilename = GNUNET_DISK_mktemp ("gnunet-consensus")))
+    {
+      GNUNET_break (0);
+      GNUNET_SCHEDULER_shutdown ();
+      return;
+    }
+    if (GNUNET_SYSERR == 
+        GNUNET_CONFIGURATION_write (pinfo->result.cfg,
+                                    tmpfilename))
+    {
+      GNUNET_break (0);
+      return;
+    }
+    printf("peer %td config file:\n", ((struct GNUNET_TESTBED_Peer **) cb_cls) - &peers[0]);
+    printf("%s\n", tmpfilename);
+  }
+  else
+  {
+    GNUNET_assert (0);
+  }
+}
+
+
+
+/**
+ * Signature of the event handler function called by the
+ * respective event controller.
+ *
+ * @param cls closure
+ * @param event information about the event
+ */
+static void
+controller_cb(void *cls,
+              const struct GNUNET_TESTBED_EventInformation *event)
+{
+  GNUNET_assert (0);
+}
+
+
+
+
+static void
+test_master (void *cls,
+             unsigned int num_peers,
+             struct GNUNET_TESTBED_Peer **started_peers)
+{
+  int i;
+
+  printf("started %d peers\n", num_peers);
+  peers = started_peers;
+
+  for (i = 0; i < num_peers; i++)
+  {
+    GNUNET_TESTBED_peer_get_information (peers[i],
+                                         GNUNET_TESTBED_PIT_IDENTITY,
+                                         peer_info_cb,
+                                         &peers[i]);
+    GNUNET_TESTBED_peer_get_information (peers[i],
+                                         GNUNET_TESTBED_PIT_CONFIGURATION,
+                                         peer_info_cb,
+                                         &peers[i]);
+  }
+}
+
+
+static void
+run (void *cls, char *const *args, const char *cfgfile,
+     const struct GNUNET_CONFIGURATION_Handle *config)
+{
+  if (NULL == config_template_file)
+  {
+    GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "no template file specified\n");
+    return;
+  }
+
+  GNUNET_TESTBED_test_run ("gnunet-consensus-start-peers",
+                           config_template_file,
+                           num_peers_requested,
+                           0,
+                           controller_cb,
+                           NULL,
+                           test_master,
+                           NULL);
+}
+
+
+int
+main (int argc, char **argv)
+{
+  static const struct GNUNET_GETOPT_CommandLineOption options[] = {
+      { 't', "config-template", "TEMPLATE",
+        gettext_noop ("start peers with the given template configuration"),
+        GNUNET_YES, &GNUNET_GETOPT_set_string, &config_template_file },
+      { 'n', "num-peers", "NUM",
+        gettext_noop ("number of peers to start"),
+        GNUNET_YES, &GNUNET_GETOPT_set_uint, &num_peers_requested },
+      GNUNET_GETOPT_OPTION_END
+   };
+
+  /* run without scheduler, as test_run already does this */
+  GNUNET_PROGRAM_run2 (argc, argv, "gnunet-consensus-start-peers",
+		      "help",
+		      options, &run, NULL, GNUNET_YES);
+  return 0;
+}
+
diff --git a/src/consensus/gnunet-consensus.c b/src/consensus/gnunet-consensus.c
index ef067ea34c..12d0965e94 100644
--- a/src/consensus/gnunet-consensus.c
+++ b/src/consensus/gnunet-consensus.c
@@ -29,6 +29,112 @@
 
 
 
+/**
+ * Handle to the consensus service
+ */
+static struct GNUNET_CONSENSUS_Handle *consensus;
+/**
+ * Session id
+ */
+static char *session_id_str;
+
+/**
+ * File handle to STDIN
+ */
+static struct GNUNET_DISK_FileHandle *stdin_fh;
+
+/**
+ * Task for reading from stdin
+ */
+static GNUNET_SCHEDULER_TaskIdentifier stdin_tid = GNUNET_SCHEDULER_NO_TASK;
+
+/**
+ * Element currently being sent to the service
+ */
+static struct GNUNET_CONSENSUS_Element *element;
+
+
+
+static void
+stdin_cb (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc);
+
+
+/**
+ * Called when a conclusion was successful.
+ *
+ * @param cls
+ * @param num_peers_in_consensus
+ * @param peers_in_consensus
+ */
+static void
+conclude_cb (void *cls, 
+             unsigned int num_peers_in_consensus,
+             const struct GNUNET_PeerIdentity *peers_in_consensus)
+{
+  printf("reached conclusion with %d peers\n", num_peers_in_consensus);
+  GNUNET_SCHEDULER_shutdown ();
+}
+
+
+
+static void
+insert_done_cb (void *cls,
+                int success)
+{
+  if (GNUNET_YES != success)
+  {
+    printf ("insert failed\n");
+    GNUNET_SCHEDULER_shutdown ();
+  }
+
+  GNUNET_free (element);
+
+  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);    
+}
+
+
+/**
+ * Called whenever we can read stdin non-blocking 
+ *
+ * @param cls unused
+ * @param tc scheduler context 
+ */
+static void
+stdin_cb (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
+{
+  char buf[1024];
+  char *ret;
+  ret = fgets (buf, 1024, stdin);
+
+  stdin_tid = GNUNET_SCHEDULER_NO_TASK;
+
+  if (NULL == ret)
+  {
+    if (feof (stdin))
+    {
+      printf ("concluding ...\n");
+      GNUNET_CONSENSUS_conclude (consensus, GNUNET_TIME_UNIT_FOREVER_REL, conclude_cb, NULL);
+    }
+    else
+    {
+      GNUNET_SCHEDULER_shutdown ();
+    }
+    return;
+  }
+
+  printf("read: %s", buf);
+
+  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, NULL);
+}
 
 /**
  * Called when a new element was received from another peer, or an error occured.
@@ -47,23 +153,82 @@ static int
 cb (void *cls,
     struct GNUNET_CONSENSUS_Element *element)
 {
-  return 0;
+  printf("got element\n");
+  return GNUNET_YES;
 }
 
+/**
+ * Function run on shutdown to clean up.
+ *
+ * @param cls the statistics handle
+ * @param tc scheduler context
+ */
+static void
+shutdown_task (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
+{
+
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "shutting down\n");
+  if (NULL == consensus)
+  {
+    return;
+  }
+
+  GNUNET_CONSENSUS_destroy (consensus);
+}
 
 
 static void
 run (void *cls, char *const *args, const char *cfgfile,
      const struct GNUNET_CONFIGURATION_Handle *cfg)
 {
-  static struct GNUNET_PeerIdentity pid;
-  static struct GNUNET_HashCode sid;
-  
-  GNUNET_CONSENSUS_create (cfg,
-			   1, &pid,
-			   &sid,
-			   &cb, NULL);
+  struct GNUNET_HashCode sid;
+  struct GNUNET_PeerIdentity *pids;
+  int count;
+  int i;
+
+  if (NULL == session_id_str)
+  {
+    GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "no session id given\n");
+    return;
+  }
+
+  for (count = 0; NULL != args[count]; count++);
+ 
+  if (0 != count)
+  { 
+    pids = GNUNET_malloc (count * sizeof (struct GNUNET_PeerIdentity));
+  }
+  else
+  {
+    pids = NULL;
+  }
+
+  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);
+
+  GNUNET_CONSENSUS_begin (consensus);
+
+
+  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);
 }
 
 
@@ -71,6 +236,9 @@ 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",
diff --git a/src/consensus/gnunet-service-consensus.c b/src/consensus/gnunet-service-consensus.c
index b733a0aec4..195efc681e 100644
--- a/src/consensus/gnunet-service-consensus.c
+++ b/src/consensus/gnunet-service-consensus.c
@@ -20,19 +20,19 @@
 
 
 #include "platform.h"
-#include "gnunet_protocols.h"
 #include "gnunet_common.h"
-#include "gnunet_service_lib.h"
+#include "gnunet_protocols.h"
+#include "gnunet_util_lib.h"
 #include "gnunet_consensus_service.h"
 #include "gnunet_core_service.h"
-#include "gnunet_container_lib.h"
+#include "gnunet_mesh_service.h"
 #include "consensus.h"
 
 
-struct ConsensusClient;
+struct ConsensusSession;
 
 static void
-send_next (struct ConsensusClient *cli);
+send_next (struct ConsensusSession *session);
 
 
 /**
@@ -58,8 +58,7 @@ struct PendingElement
 
 
 /**
- * A consensus session consists of one or more local clients,
- * as well as zero or more remote authorities.
+ * A consensus session consists of one local client and the remote authorities.
  */
 struct ConsensusSession
 {
@@ -74,18 +73,8 @@ struct ConsensusSession
   struct ConsensusSession *prev;
 
   /**
-   * Consensus clients are kept in a DLL.
-   */
-  struct ConsensusClient *clients_head;
-
-  /**
-   * Consensus clients are kept in a DLL.
+   * Local consensus identification, chosen by clients.
    */
-  struct ConsensusClient *clients_tail;
-
-  /**
-  * Local consensus identification, chosen by clients.
-  */
   struct GNUNET_HashCode *local_id;
  
   /**
@@ -95,24 +84,6 @@ struct ConsensusSession
   struct GNUNET_HashCode *global_id;
 
   /**
-   * Values in the consensus set of this session.
-   */
-  struct GNUNET_CONTAINER_MultiHashMap *values;
-};
-
-
-struct ConsensusClient
-{
-  /**
-   * Consensus clients are kept in a DLL.
-   */
-  struct ConsensusClient *next;
-  /**
-   * Consensus clients are kept in a DLL.
-   */
-  struct ConsensusClient *prev;
-
-  /**
    * Corresponding server handle.
    */
   struct GNUNET_SERVER_Client *client;
@@ -123,24 +94,30 @@ struct ConsensusClient
   int begin;
 
   /**
-   * Session this client belongs to
+   * Values in the consensus set of this session,
+   * all of them either have been sent or approved by the client.
    */
-  struct ConsensusSession *session;
+  struct GNUNET_CONTAINER_MultiHashMap *values;
 
   /**
-   * Values in the consensus set of this client.
-   * Includes pending elements.
+   * Elements that have not been sent to the client yet.
    */
-  struct GNUNET_CONTAINER_MultiHashMap *values;
+  struct PendingElement *transmit_pending_head;
 
   /**
-   * Elements that have not been set to the client yet.
+   * Elements that have not been sent to the client yet.
    */
-  struct PendingElement *pending_head;
+  struct PendingElement *transmit_pending_tail;
+
   /**
-   * Elements that have not been set to the client yet.
+   * Elements that have not been sent to the client yet.
    */
-  struct PendingElement *pending_tail;
+  struct PendingElement *approval_pending_head;
+
+  /**
+   * Elements that have not been sent to the client yet.
+   */
+  struct PendingElement *approval_pending_tail;
 
   /**
    * Currently active transmit handle for sending to the client
@@ -157,6 +134,11 @@ struct ConsensusClient
    * Client has been informed about the conclusion.
    */
   int conclude_sent;
+
+  /**
+   * Number of other peers in the consensus
+   */
+  int num_peers;
 };
 
 
@@ -185,30 +167,6 @@ static struct GNUNET_SERVER_Handle *srv;
  */
 static struct GNUNET_PeerIdentity *my_peer;
 
-
-struct ConsensusClient *
-find_client (const struct GNUNET_SERVER_Client *srv_client)
-{
-  struct ConsensusSession *session;
-  struct ConsensusClient *client;
-
-  session = sessions_head;
-  while (NULL != session)
-  {
-    client = session->clients_head;
-    while (NULL != client)
-    {
-      if (client->client == srv_client)
-      {
-        return client;
-      }
-      client = client->next;
-    }
-    session = session->next;
-  }
-  return NULL;
-}
-
 static void
 disconnect_client (struct GNUNET_SERVER_Client *client)
 {
@@ -221,73 +179,44 @@ compute_global_id (struct GNUNET_HashCode *dst,
                    const struct GNUNET_PeerIdentity *peers,
                    int num_peers)
 {
-  *dst = *local_id;
+  int i;
+  struct GNUNET_HashCode tmp;
 
-  /* FIXME: hash other peers into global id */
-}
-
-
-
-/**
- * Iterator over hash map entries.
- *
- * @param cls closure, the client
- * @param key current key code
- * @param value value in the hash map
- * @return GNUNET_YES if we should continue to
- *         iterate,
- *         GNUNET_NO if not.
- */
-int
-update_pending (void *cls,
-                 const struct GNUNET_HashCode *key,
-                 void *value)
-{
-  struct ConsensusClient *cli;
-  struct GNUNET_CONSENSUS_Element *element;
-  struct PendingElement *pending_element;
-
-  cli = (struct ConsensusClient *) cls;
-  element = (struct GNUNET_CONSENSUS_Element *) value;
-  pending_element = GNUNET_malloc (sizeof (struct PendingElement));
-  pending_element->element = element;
-
-  if (GNUNET_NO == GNUNET_CONTAINER_multihashmap_contains (cli->values, key))
+  *dst = *local_id;
+  for (i = 0; i < num_peers; ++i)
   {
-    GNUNET_CONTAINER_DLL_insert_tail (cli->pending_head, cli->pending_tail, pending_element);
-    GNUNET_CONTAINER_multihashmap_put (cli->values, key, element, GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY);
+    /* FIXME: maybe hash_xor/hash allow aliased source/target, and we can get by without tmp */
+    GNUNET_CRYPTO_hash_xor (dst, &peers[0].hashPubKey, &tmp);
+    *dst = tmp;
+    GNUNET_CRYPTO_hash (dst, sizeof (struct GNUNET_PeerIdentity), &tmp);
+    *dst = tmp;
   }
-  
-  return GNUNET_YES;
 }
 
 
-
-
 static size_t
 transmit_pending (void *cls, size_t size, void *buf)
 {
   struct GNUNET_CONSENSUS_Element *element;
   struct GNUNET_CONSENSUS_ElementMessage *msg;
-  struct ConsensusClient *cli;
+  struct ConsensusSession *session;
 
-  cli = (struct ConsensusClient *) cls;
+  session = (struct ConsensusSession *) cls;
   msg = (struct GNUNET_CONSENSUS_ElementMessage *) buf;
-  element = cli->pending_head->element;
+  element = session->transmit_pending_head->element;
 
   GNUNET_assert (NULL != element);
 
-  cli->th = NULL;
+  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;
 
-  cli->pending_head = cli->pending_head->next;
-
-  send_next (cli);
+  send_next (session);
 
   return sizeof (struct GNUNET_CONSENSUS_ElementMessage) + element->size;
 }
@@ -299,7 +228,7 @@ transmit_conclude_done (void *cls, size_t size, void *buf)
   struct GNUNET_CONSENSUS_ConcludeDoneMessage *msg;
 
   msg = (struct GNUNET_CONSENSUS_ConcludeDoneMessage *) buf;
-  msg->header.type = htons (GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_CONCLUDE);
+  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);
 
@@ -313,38 +242,43 @@ transmit_conclude_done (void *cls, size_t size, void *buf)
  * @param cli the client to send the next message to
  */
 static void
-send_next (struct ConsensusClient *cli)
+send_next (struct ConsensusSession *session)
 {
   int msize;
 
-  GNUNET_assert (NULL != cli);
+  GNUNET_assert (NULL != session);
 
-  if (NULL != cli->th)
+  if (NULL != session->th)
   {
     return;
   }
 
-  if ((cli->conclude_requested == GNUNET_YES) && (cli->conclude_sent == GNUNET_NO))
+  if ((session->conclude_requested == GNUNET_YES) && (session->conclude_sent == GNUNET_NO))
   {
     /* just the conclude message with no other authorities in the dummy */
     msize = sizeof (struct GNUNET_CONSENSUS_ConcludeMessage);
-    cli->th =
-        GNUNET_SERVER_notify_transmit_ready (cli->client, msize,
-                                             GNUNET_TIME_UNIT_FOREVER_REL, &transmit_conclude_done, cli);
-    cli->conclude_sent = GNUNET_YES;
+    session->th =
+        GNUNET_SERVER_notify_transmit_ready (session->client, msize,
+                                             GNUNET_TIME_UNIT_FOREVER_REL, &transmit_conclude_done, session);
+    session->conclude_sent = GNUNET_YES;
   }
-  else if (NULL != cli->pending_head)
+  else if (NULL != session->transmit_pending_head)
   {
-    msize = cli->pending_head->element->size + sizeof (struct GNUNET_CONSENSUS_ElementMessage);
-    cli->th =
-        GNUNET_SERVER_notify_transmit_ready (cli->client, msize,
-                                             GNUNET_TIME_UNIT_FOREVER_REL, &transmit_pending, cli);
+    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 */
   }
 }
 
 
 /**
  * Called when a client wants to join a consensus session.
+ *
+ * @param cls unused
+ * @param client client that sent the message
+ * @param m message sent by the client
  */
 static void
 client_join (void *cls,
@@ -354,58 +288,42 @@ client_join (void *cls,
   struct GNUNET_HashCode global_id;
   const struct GNUNET_CONSENSUS_JoinMessage *msg;
   struct ConsensusSession *session;
-  struct ConsensusClient *consensus_client;
 
-  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "join\n");
-
-  fprintf(stderr, "foobar\n");
-
-  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "client joined\n");
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "client joining\n");
 
   msg = (struct GNUNET_CONSENSUS_JoinMessage *) m;
-  
-  /* kill the client if it already is in a session */
-  if (NULL != find_client (client))
-  {
-    GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "client tried to join twice\n");
-    disconnect_client (client);
-    return;
-  }
-
-  consensus_client = GNUNET_malloc (sizeof (struct ConsensusClient));
-  consensus_client->client = client;
-  consensus_client->begin = GNUNET_NO;
-  consensus_client->values = GNUNET_CONTAINER_multihashmap_create (4, GNUNET_NO);
-
-  GNUNET_SERVER_client_keep (client);
-
-  GNUNET_assert (NULL != consensus_client->values);
 
   compute_global_id (&global_id, &msg->session_id, (struct GNUNET_PeerIdentity *) &m[1], msg->num_peers);
 
-  /* look if we already have a session for this local id */
   session = sessions_head;
   while (NULL != session)
   {
-    if (0 == memcmp(&global_id, session->global_id, sizeof (struct GNUNET_HashCode)))
+    if (client == session->client)
     {
-      GNUNET_CONTAINER_DLL_insert (session->clients_head, session->clients_tail, consensus_client);
-      GNUNET_SERVER_receive_done (client, GNUNET_OK);
+      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)))
+    {
+      GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "session already owned by another client\n");
+      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->client = client;
 
   GNUNET_CONTAINER_DLL_insert (sessions_head, sessions_tail, session);
-  GNUNET_CONTAINER_DLL_insert (session->clients_head, session->clients_tail, consensus_client);
 
-  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "created new session");
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "created new session\n");
 
   GNUNET_SERVER_receive_done (client, GNUNET_OK);
 }
@@ -419,18 +337,22 @@ client_insert (void *cls,
              struct GNUNET_SERVER_Client *client,
              const struct GNUNET_MessageHeader *m)
 {
-  struct ConsensusClient *consensus_client;
+  struct ConsensusSession *session;
   struct GNUNET_CONSENSUS_ElementMessage *msg;
   struct GNUNET_CONSENSUS_Element *element;
-  struct PendingElement *pending_element;
   struct GNUNET_HashCode key;
   int element_size;
 
   GNUNET_log(GNUNET_ERROR_TYPE_INFO, "insert\n");
 
-  consensus_client = find_client (client);
+  session = sessions_head;
+  while (NULL != session)
+  {
+    if (session->client == client)
+      break;
+  }
 
-  if (NULL == consensus_client)
+  if (NULL == session)
   {
     GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "client tried to insert, but client is not in any session\n");
     GNUNET_SERVER_client_disconnect (client);
@@ -449,28 +371,12 @@ client_insert (void *cls,
 
   GNUNET_CRYPTO_hash (element, element_size, &key);
 
-  GNUNET_CONTAINER_multihashmap_put (consensus_client->session->values, &key, element,
+  GNUNET_CONTAINER_multihashmap_put (session->values, &key, element,
                                      GNUNET_CONTAINER_MULTIHASHMAPOPTION_REPLACE);
-  GNUNET_CONTAINER_multihashmap_put (consensus_client->values, &key, element,
-                                     GNUNET_CONTAINER_MULTIHASHMAPOPTION_REPLACE);
-
-  /* send the new value to all clients that don't have it */
-
-  consensus_client = consensus_client->session->clients_head;
-  while (NULL != consensus_client)
-  {
-    if (GNUNET_NO == GNUNET_CONTAINER_multihashmap_contains (consensus_client->values, &key))
-    {
-      pending_element = GNUNET_malloc (sizeof (struct PendingElement));
-      pending_element->element = element;
-      GNUNET_CONTAINER_DLL_insert_tail (consensus_client->pending_head, consensus_client->pending_tail, pending_element);
-      GNUNET_CONTAINER_multihashmap_put (consensus_client->values, &key, element,
-                                         GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY);
-      send_next (consensus_client);
-    }
-  }
 
   GNUNET_SERVER_receive_done (client, GNUNET_OK);
+
+  send_next (session);
 }
 
 
@@ -482,20 +388,27 @@ client_begin (void *cls,
              struct GNUNET_SERVER_Client *client,
              const struct GNUNET_MessageHeader *message)
 {
-  struct ConsensusClient *consensus_client;
+  struct ConsensusSession *session;
 
-  consensus_client = find_client (client);
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "client requested begin\n");
+
+  session = sessions_head;
+  while (NULL != session)
+  {
+    if (session->client == client)
+      break;
+  }
 
-  if (NULL == consensus_client)
+  if (NULL == session)
   {
+    GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "client tried to 'begin', but client is not in any session\n");
     GNUNET_SERVER_client_disconnect (client);
     return;
   }
 
-  consensus_client->begin = GNUNET_YES;
+  session->begin = GNUNET_YES;
 
-  GNUNET_CONTAINER_multihashmap_iterate (consensus_client->session->values, &update_pending, NULL);
-  send_next (consensus_client);
+  send_next (session);
 
   GNUNET_SERVER_receive_done (client, GNUNET_OK);
 }
@@ -510,20 +423,35 @@ client_conclude (void *cls,
              struct GNUNET_SERVER_Client *client,
              const struct GNUNET_MessageHeader *message)
 {
-  struct ConsensusClient *consensus_client;
+  struct ConsensusSession *session;
 
-  consensus_client = find_client (client);
-  if (NULL == consensus_client)
+  session = sessions_head;
+  while ((session != NULL) && (session->client != client))
+  {
+    session = session->next;
+  }
+  if (NULL == session)
   {
     GNUNET_SERVER_client_disconnect (client);
     return;
   }
-  consensus_client->conclude_requested = GNUNET_YES;
-  send_next (consensus_client);
-
+  session->conclude_requested = GNUNET_YES;
+  send_next (session);
   GNUNET_SERVER_receive_done (client, GNUNET_OK);
 }
 
+
+/**
+ * Called when a client sends an ack
+ */
+void
+client_ack (void *cls,
+             struct GNUNET_SERVER_Client *client,
+             const struct GNUNET_MessageHeader *message)
+{
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "client ack received\n");
+}
+
 /**
  * Task that disconnects from core.
  *
@@ -538,7 +466,7 @@ disconnect_core (void *cls,
   core = (struct GNUNET_CORE_Handle *) cls;
   GNUNET_CORE_disconnect (core);
 
-  GNUNET_log(GNUNET_ERROR_TYPE_INFO, "disconnected from core\n");
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "disconnected from core\n");
 }
 
 
@@ -554,16 +482,14 @@ core_startup (void *cls,
         sizeof (struct GNUNET_MessageHeader)},
     {&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_SERVER_add_handlers (srv, handlers);
-
   my_peer = GNUNET_memdup(peer, sizeof (struct GNUNET_PeerIdentity));
-
   GNUNET_SCHEDULER_add_now (&disconnect_core, core);
-
   GNUNET_log(GNUNET_ERROR_TYPE_INFO, "connected to core\n");
 }
 
@@ -583,7 +509,7 @@ run (void *cls, struct GNUNET_SERVER_Handle *server, const struct GNUNET_CONFIGU
     {NULL, 0, 0}
   };
 
-  GNUNET_log(GNUNET_ERROR_TYPE_INFO, "run\n");
+  GNUNET_log(GNUNET_ERROR_TYPE_INFO, "consensus  running\n");
 
   cfg = c;
   srv = server;
diff --git a/src/consensus/ibf.c b/src/consensus/ibf.c
new file mode 100644
index 0000000000..d0cb218cc3
--- /dev/null
+++ b/src/consensus/ibf.c
@@ -0,0 +1,244 @@
+/*
+      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/ibf.c
+ * @brief implementation of the invertible bloom filter
+ * @author Florian Dold
+ */
+
+#include "platform.h"
+#include "gnunet_common.h"
+#include "ibf.h"
+
+
+struct PureCells
+{
+  int index;
+  struct PureCells *next;
+  struct PureCells *prev;
+};
+
+struct InvertibleBloomFilter
+{
+  /**
+   * How many cells does this IBF have?
+   */
+  int size;
+
+  /**
+   * In how many cells do we hash one element?
+   * Usually 4 or 3.
+   */
+  int hash_num;
+
+  /**
+   * Salt for mingling hashes
+   */
+  int 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.
+   */
+  GNUNET_HashCode *id_sum;
+
+  /**
+   * xor sums of the "hash of the hash".
+   */
+  GNUNET_HashCode *hash_sum;
+
+  struct PureCells *pure_head;
+  struct PureCells *pure_tail;
+
+  /**
+   * GNUNET_YES: fresh list is deprecated
+   * GNUNET_NO: fresh list is up to date
+   */
+  int pure_fresh;
+};
+
+
+/**
+ * Create an invertible bloom filter.
+ */
+struct InvertibleBloomFilter *
+ibf_create(int size, int hash_num)
+{
+  struct InvertibleBloomFilter *ibf;
+
+  ibf = GNUNET_malloc (sizeof (struct InvertibleBloomFilter));
+  ibf->count = GNUNET_malloc (size * sizeof uint8_t);
+  ibf->id_sum = GNUNET_malloc (size * sizeof (struct GNUNET_HashCode));
+  ibf->hash_sum = GNUNET_malloc (size * sizeof (struct GNUNET_HashCode));
+  ibf->size = size;
+  ibf->hash_num = hash_num;
+}
+
+
+/**
+ * Insert an element into an IBF.
+ */
+void
+ibf_insert (struct InvertibleBloomFilter *ibf, struct GNUNET_HashCode *id)
+{
+  struct GNUNET_HashCode key;
+  struct GNUNET_HashCode id_hash;
+  int i;
+
+  key = *id;
+  GNUNET_hash (id, sizeof (struct GNUNET_HashCode), &id_hash);
+
+  for (i = 0; i < ibf->hash_num; i++)
+  {
+    int bucket;
+    int j;
+    if ((i != 0) && (i % 16) == 0)
+    {
+      GNUNET_hash (&key, sizeof (struct GNUNET_HashCode), &key);
+    }
+    bucket = hash.bits[i%16] % ibf->size;
+
+    /* count<0 can happen after ibf subtraction, but then no insert should be done */
+    GNUNET_assert (ibf->count[bucket] >= 0);
+
+    ibf->count[bucket]++;
+
+    for (j=0; j < 16; j++)
+    {
+      ibf->id_sum.bits[j] ^= &id;
+      ibf->hash_sum.bits[j] ^= &id_hash;
+    }
+
+  }
+}
+
+
+/**
+ * Update the linked list of pure cells, if not fresh anymore
+ */
+void
+update_pure (struct InvertibleBloomFilter *ibf)
+{
+  if (GNUNET_YES == ibf->pure_fresh)
+  {
+    return;
+  }
+
+  ibf->pure_fresh = GNUNET_YES;
+}
+
+/**
+ * Decode and remove an element from the IBF, if possible.
+ *
+ * @param ibf the invertible bloom filter to decode
+ * @param ret_id the hash code of the decoded element, if successful
+ * @param side sign of the cell's count where the decoded element came from.
+ *             A negative sign indicates that the element was recovered resides in an IBF
+ *             that was previously subtracted from.
+ * @return GNUNET_YES if decoding an element was successful, GNUNET_NO if the IBF is empty,
+ *         GNUNET_SYSERR if the decoding has faile
+ */
+int
+ibf_decode (struct InvertibleBloomFilter *ibf, int *ret_side, struct GNUNET_HashCode *ret_id)
+{
+  struct GNUNET_HashCode hash;
+  struct PureCells *pure;
+  int count;
+
+  GNUNET_assert (NULL != ibf);
+  GNUNET_assert (NULL != red_id);
+  GNUNET_assert (NULL != side);
+
+  update_pure (ibf);
+
+  pure = ibf->pure_head;
+  ibf->pure_head = pure->next;
+
+  if (NULL == pure)
+  {
+    int i;
+    for (i = 0; i < ibf->size; i++)
+    {
+      int j;
+      if (0 != ibf->count[i])
+        return GNUNET_SYSERR;
+      for (j = 0; j < 16; ++j)
+        if ((0 != ibf->hash_sum[i].bits[j]) || (0 != ibf->id_sum[i].bits[j]))
+          return GNUNET_SYSERR;
+      return GNUNET_NO;
+    }
+  }
+
+  GNUNET_CRYPTO_hash (ibf->id_sum[pure->idx], sizeof (struct GNUNET_HashCode), &hash);
+
+  if (0 == memcmp (&hash, ibf->hash_sum[pure->idx]))
+  {
+    struct GNUNET_HashCode key;
+    int i;
+
+    *ret_side = ibf->count[pure->index];
+    *ret_id = ibf->id_sum[pure->index];
+
+    key = *ibf->id_sum[pure->index];
+
+    /* delete the item from all buckets */
+    for (i = 0; i < ibf->hash_num; i++)
+    {
+      int bucket;
+      int j;
+      if ((i != 0) && (i % 16) == 0)
+      {
+        GNUNET_hash (&key, sizeof (struct GNUNET_HashCode), &key);
+      }
+      bucket = hash.bits[i%16] % ibf->size;
+
+      ibf->count[bucket] -= count;
+
+      for (j=0; j < 16; j++)
+      {
+        ibf->id_sum.bits[j] ^= &id;
+        ibf->hash_sum.bits[j] ^= &id_hash;
+      }
+      return GNUNET_YES;
+  }
+  return GNUNET_SYSERR;
+}
+
+
+
+/**
+ * Subtract ibf2 from ibf1, storing the result in ibf1.
+ * The two IBF's must have the same parameters size and hash_num.
+ *
+ * @return a newly allocated invertible bloom filter
+ */
+void
+ibf_subtract (struct InvertibleBloomFilter *ibf1, struct InvertibleBloomFilter *ibf2)
+{
+  /* FIXME */
+}
+
diff --git a/src/consensus/ibf.h b/src/consensus/ibf.h
new file mode 100644
index 0000000000..2c9931e695
--- /dev/null
+++ b/src/consensus/ibf.h
@@ -0,0 +1,98 @@
+/*
+      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/ibf.h
+ * @brief invertible bloom filter
+ * @author Florian Dold
+ */
+
+
+/**
+ * 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
+
+/**
+ * Create an invertible bloom filter.
+ *
+ * @param size number of IBF buckets
+ * @param salt salt for mingling hashes, different salt may
+ *        result in less (or more) collisions
+ * @param hash_num number of buckets one element is hashed in
+ * @return the newly created invertible bloom filter
+ */
+struct InvertibleBloomFilter *
+ibf_create(int size, int salt, int hash_num);
+
+/**
+ * Insert an element into an IBF.
+ *
+ * @param ibf the IBF
+ * @param id the element's hash code
+ */
+void
+ibf_insert (struct InvertibleBloomFilter *ibf, GNUNET_HashCode *id);
+
+/**
+ * Subtract ibf2 from ibf1, storing the result in ibf1.
+ * The two IBF's must have the same parameters size and hash_num.
+ */
+void
+ibf_subtract (struct InvertibleBloomFilter *ibf1, struct InvertibleBloomFilter *ibf2);
+
+/**
+ * Decode and remove an element from the IBF, if possible.
+ *
+ * @param ibf the invertible bloom filter
+ * @param the id of the element is written to this hash code
+ * @return GNUNET_YES if decoding an element was successful, GNUNET_NO if it failed to decode
+ */
+int
+ibf_decode (struct InvertibleBloomFilter *ibf, struct GNUNET_HashCode *ret_id);
+
+
+/**
+ * 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);
+
+/*
+ibf_hton();
+
+ibf_ntoh();
+*/
+
+/**
+ * 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);
+
diff --git a/src/consensus/test_consensus.conf b/src/consensus/test_consensus.conf
index 5bd57631b4..86dfadb9b6 100644
--- a/src/consensus/test_consensus.conf
+++ b/src/consensus/test_consensus.conf
@@ -10,7 +10,7 @@ ACCEPT_FROM6 = ::1;
 UNIXPATH = /tmp/gnunet-service-consensus.sock
 UNIX_MATCH_UID = YES
 UNIX_MATCH_GID = YES
-OPTIONS = -LDEBUG
+OPTIONS = -L INFO
 
 
 [transport]
@@ -18,4 +18,8 @@ OPTIONS = -LERROR
 
 
 [arm]
-DEFAULTSERVICES = core
+DEFAULTSERVICES = core consensus
+
+
+[testbed]
+OVERLAY_TOPOLOGY = CLIQUE