work on gnunet-set, isolated bug in stream

8 files changed, 476 insertions, 377 deletions

diff --git a/src/set/Makefile.am b/src/set/Makefile.am
index c1639823e..a4c4fa6be 100644
--- a/src/set/Makefile.am
+++ b/src/set/Makefile.am
@@ -16,7 +16,7 @@ if USE_COVERAGE
 endif
 
 bin_PROGRAMS = \
- gnunet-set
+ gnunet-set gnunet-set-bug
 
 libexec_PROGRAMS = \
  gnunet-service-set
@@ -35,6 +35,17 @@ gnunet_set_LDADD = \
 gnunet_set_DEPENDENCIES = \
   libgnunetset.la
 
+gnunet_set_bug_SOURCES = \
+ gnunet-set-bug.c \
+ mq.c
+gnunet_set_bug_LDADD = \
+  $(top_builddir)/src/util/libgnunetutil.la \
+  $(top_builddir)/src/stream/libgnunetstream.la \
+  $(GN_LIBINTL)
+# hack for mq.c, see automake Objects ‘created with both libtool and without’
+# remove once GNUNET_MQ is in util/
+gnunet_set_bug_CFLAGS = $(AM_CFLAGS)
+
 gnunet_service_set_SOURCES = \
  gnunet-service-set.c \
  gnunet-service-set_union.c \
diff --git a/src/set/gnunet-service-set.c b/src/set/gnunet-service-set.c
index d6258aa78..314e7719d 100644
--- a/src/set/gnunet-service-set.c
+++ b/src/set/gnunet-service-set.c
@@ -84,7 +84,7 @@ static uint32_t request_id = 1;
  * @param client the client to disconnect
  */
 void
-client_disconnect (struct GNUNET_SERVER_Client *client)
+_GSS_client_disconnect (struct GNUNET_SERVER_Client *client)
 {
   /* FIXME: clean up any data structures belonging to the client */
   GNUNET_SERVER_client_disconnect (client);
@@ -170,6 +170,7 @@ destroy_incoming (struct Incoming *incoming)
  * @param cls the incoming socket
  * @param mh the message
  */
+
 static void
 handle_p2p_operation_request (void *cls, const struct GNUNET_MessageHeader *mh)
 {
@@ -180,6 +181,8 @@ handle_p2p_operation_request (void *cls, const struct GNUNET_MessageHeader *mh)
   struct Listener *listener;
   const struct GNUNET_MessageHeader *context_msg;
 
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "got operation request\n");
+
   if (ntohs (mh->size) < sizeof *msg)
   {
     GNUNET_break (0);
@@ -201,18 +204,28 @@ handle_p2p_operation_request (void *cls, const struct GNUNET_MessageHeader *mh)
       return;
     }
   }
-
-  for (listener = listeners_head; listener != NULL; listener = listener->next)
+  /* find the appropriate listener */
+  for (listener = listeners_head;
+       listener != NULL;
+       listener = listener->next)
   {
     if ( (0 != GNUNET_CRYPTO_hash_cmp (&msg->app_id, &listener->app_id)) ||
          (htons (msg->operation) != listener->operation) )
       continue;
-    mqm = GNUNET_MQ_msg_concat (cmsg, context_msg, GNUNET_MESSAGE_TYPE_SET_REQUEST);
+    mqm = GNUNET_MQ_msg (cmsg, GNUNET_MESSAGE_TYPE_SET_REQUEST);
+    if (GNUNET_OK !=
+        GNUNET_MQ_nest (mqm, context_msg))
+    {
+      /* FIXME: disconnect the peer */
+      GNUNET_MQ_discard (mqm);
+      GNUNET_break (0);
+    }
     incoming->request_id = request_id++;
     cmsg->request_id = htonl (incoming->request_id);
     GNUNET_MQ_send (listener->client_mq, mqm);
     return;
   }
+  /* FIXME: send a reject message */
 }
 
 
@@ -249,7 +262,7 @@ handle_client_create (void *cls,
       GNUNET_assert (0);
       break;
     case GNUNET_SET_OPERATION_UNION:
-      set = union_set_create ();
+      set = _GSS_union_set_create ();
       break;
     default:
       GNUNET_free (set);
@@ -261,7 +274,7 @@ handle_client_create (void *cls,
   set->client = client;
   set->client_mq = GNUNET_MQ_queue_for_server_client (client);
   GNUNET_CONTAINER_DLL_insert (sets_head, sets_tail, set);
-
+  
   GNUNET_SERVER_receive_done (client, GNUNET_OK);
 }
 
@@ -292,6 +305,8 @@ handle_client_listen (void *cls,
   listener->app_id = msg->app_id;
   listener->operation = msg->operation;
   GNUNET_CONTAINER_DLL_insert_tail (listeners_head, listeners_tail, listener);
+
+  GNUNET_SERVER_receive_done (client, GNUNET_OK);
 }
 
 
@@ -313,14 +328,13 @@ handle_client_add (void *cls,
   if (NULL == set)
   {
     GNUNET_break (0);
-    client_disconnect (client);
+    _GSS_client_disconnect (client);
     return;
   }
   switch (set->operation)
   {
     case GNUNET_SET_OPERATION_UNION:
-      union_add (set, (struct ElementMessage *) m);
-      break;
+      _GSS_union_add ((struct ElementMessage *) m, set);
     case GNUNET_SET_OPERATION_INTERSECTION:
       /* FIXME: cfuchs */
       break;
@@ -328,6 +342,8 @@ handle_client_add (void *cls,
       GNUNET_assert (0);
       break;
   }
+
+  GNUNET_SERVER_receive_done (client, GNUNET_OK);
 }
 
 
@@ -344,24 +360,15 @@ handle_client_evaluate (void *cls,
                         const struct GNUNET_MessageHeader *m)
 {
   struct Set *set;
-  struct EvaluateMessage *msg = (struct EvaluateMessage *) m;
-  struct EvaluateOperation *eo;
 
   set = get_set (client);
-
   if (NULL == set)
   {
     GNUNET_break (0);
-    client_disconnect (client);
+    _GSS_client_disconnect (client);
     return;
   }
 
-  eo = GNUNET_new (struct EvaluateOperation);
-  eo->peer = msg->peer;
-  eo->app_id = msg->app_id;
-  eo->request_id = msg->request_id;
-  eo->context_msg = GNUNET_copy_message (&msg[1].header);
-  eo->set = set;
 
   switch (set->operation)
   {
@@ -369,12 +376,14 @@ handle_client_evaluate (void *cls,
       /* FIXME: cfuchs */
       break;
     case GNUNET_SET_OPERATION_UNION:
-      union_evaluate (eo);
+      _GSS_union_evaluate ((struct EvaluateMessage *) m, set);
       break;
     default:
       GNUNET_assert (0);
       break;
   }
+
+  GNUNET_SERVER_receive_done (client, GNUNET_OK);
 }
 
 
@@ -391,6 +400,7 @@ handle_client_cancel (void *cls,
                       const struct GNUNET_MessageHeader *m)
 {
   /* FIXME: implement */
+  GNUNET_SERVER_receive_done (client, GNUNET_OK);
 }
 
 
@@ -407,6 +417,7 @@ handle_client_ack (void *cls,
                    const struct GNUNET_MessageHeader *m)
 {
   /* FIXME: implement */
+  GNUNET_SERVER_receive_done (client, GNUNET_OK);
 }
 
 
@@ -421,19 +432,18 @@ handle_client_ack (void *cls,
 static void
 handle_client_accept (void *cls,
                       struct GNUNET_SERVER_Client *client,
-                      const struct GNUNET_MessageHeader *m)
+                      const struct GNUNET_MessageHeader *mh)
 {
-  struct AcceptMessage *msg = (struct AcceptMessage *) m;
   struct Set *set;
   struct Incoming *incoming;
-  struct EvaluateOperation *eo;
+  struct AcceptMessage *msg = (struct AcceptMessage *) mh;
 
   set = get_set (client);
 
   if (NULL == set)
   {
     GNUNET_break (0);
-    client_disconnect (client);
+    _GSS_client_disconnect (client);
     return;
   }
 
@@ -443,16 +453,10 @@ handle_client_accept (void *cls,
        (incoming->operation != set->operation) )
   {
     GNUNET_break (0);
-    client_disconnect (client);
+    _GSS_client_disconnect (client);
     return;
   }
 
-  eo = GNUNET_new (struct EvaluateOperation);
-  eo->peer = incoming->peer;
-  eo->app_id = incoming->app_id;
-  eo->request_id = msg->request_id;
-  eo->set = set;
-
   switch (set->operation)
   {
     case GNUNET_SET_OPERATION_INTERSECTION:
@@ -460,12 +464,15 @@ handle_client_accept (void *cls,
       GNUNET_assert (0);
       break;
     case GNUNET_SET_OPERATION_UNION:
-      union_accept (eo, incoming);
+      _GSS_union_accept (msg, set, incoming);
       break;
     default:
       GNUNET_assert (0);
       break;
   }
+  /* FIXME: destroy incoming */
+
+  GNUNET_SERVER_receive_done (client, GNUNET_OK);
 }
 
 
@@ -522,7 +529,7 @@ shutdown_task (void *cls,
   {
     GNUNET_STREAM_listen_close (stream_listen_socket);
     stream_listen_socket = NULL;
-  } 
+  }
 
   GNUNET_log (GNUNET_ERROR_TYPE_INFO, "handled shutdown request\n");
 }
@@ -575,4 +582,3 @@ main (int argc, char *const *argv)
   return (GNUNET_OK == ret) ? 0 : 1;
 }
 
-
diff --git a/src/set/gnunet-service-set.h b/src/set/gnunet-service-set.h
index 685fc47a4..2d0d6595d 100644
--- a/src/set/gnunet-service-set.h
+++ b/src/set/gnunet-service-set.h
@@ -86,83 +86,7 @@ struct Set
   union {
     struct IntersectionState *i;
     struct UnionState *u;
-  } extra;
-};
-
-
-/**
- * State for an evaluate operation for a set that
- * supports set union.
- */
-struct UnionEvaluateOperation;
-
-
-/* FIXME: cfuchs */
-struct IntersectionEvaluateOperation
-{
-  /* FIXME: cfuchs */
-};
-
-
-/**
- * State of evaluation a set operation with
- * another peer
- */
-struct EvaluateOperation
-{
-  /**
-   * Local set the operation is evaluated on
-   */
-  struct Set *set;
-
-  /**
-   * Peer with the remote set
-   */
-  struct GNUNET_PeerIdentity peer;
-
-  /**
-   * Application-specific identifier
-   */
-  struct GNUNET_HashCode app_id;
-
-  /**
-   * Context message, given to us
-   * by the client, may be NULL.
-   */
-  struct GNUNET_MessageHeader *context_msg;
-
-  /**
-   * Stream socket connected to the other peer
-   */
-  struct GNUNET_STREAM_Socket *socket;
-
-  /**
-   * Message queue for the peer on the other
-   * end
-   */
-  struct GNUNET_MQ_MessageQueue *mq;
-
-  /**
-   * Type of this operation
-   */
-  enum GNUNET_SET_OperationType operation;
-
-  /**
-   * GNUNET_YES if we started the operation,
-   * GNUNET_NO if the other peer started it.
-   */
-  int is_outgoing;
-
-  /**
-   * Request id, so we can use one client handle
-   * for multiple operations
-   */
-  uint32_t request_id;
-
-  union {
-    struct UnionEvaluateOperation *u;
-    struct IntersectionEvaluateOperation *i;
-  } extra;
+  } state;
 };
 
 
@@ -246,6 +170,12 @@ struct Incoming
   struct GNUNET_MessageHeader *context_msg;
 
   /**
+   * Salt the peer has requested to use for the
+   * operation
+   */
+  uint16_t salt;
+
+  /**
    * Operation the other peer wants to do
    */
   enum GNUNET_SET_OperationType operation;
@@ -271,23 +201,24 @@ extern const struct GNUNET_CONFIGURATION_Handle *configuration;
  * @param client the client to disconnect
  */
 void
-client_disconnect (struct GNUNET_SERVER_Client *client);
+_GSS_client_disconnect (struct GNUNET_SERVER_Client *client);
 
 
 struct Set *
-union_set_create (void);
+_GSS_union_set_create (void);
 
 
 void
-union_evaluate (struct EvaluateOperation *eo);
+_GSS_union_evaluate (struct EvaluateMessage *m, struct Set *set);
 
 
 void
-union_add (struct Set *set, struct ElementMessage *m);
+_GSS_union_add (struct ElementMessage *m, struct Set *set);
 
 
 void
-union_accept (struct EvaluateOperation *eo, struct Incoming *incoming);
+_GSS_union_accept (struct AcceptMessage *m, struct Set *set,
+                   struct Incoming *incoming);
 
 
 #endif
diff --git a/src/set/gnunet-service-set_union.c b/src/set/gnunet-service-set_union.c
index e65452a54..4903ce605 100644
--- a/src/set/gnunet-service-set_union.c
+++ b/src/set/gnunet-service-set_union.c
@@ -26,9 +26,12 @@
 
 
 #include "gnunet-service-set.h"
-#include "set_protocol.h"
+#include "gnunet_container_lib.h"
+#include "gnunet_crypto_lib.h"
 #include "ibf.h"
 #include "strata_estimator.h"
+#include "set_protocol.h"
+#include <gcrypt.h>
 
 
 /**
@@ -49,7 +52,6 @@
  */
 #define MAX_BUCKETS_PER_MESSAGE ((1<<15) / IBF_BUCKET_SIZE)
 
-
 /**
  * The maximum size of an ibf we use is 2^(MAX_IBF_ORDER).
  * Choose this value so that computing the IBF is still cheaper
@@ -76,6 +78,55 @@ enum UnionOperationState
  */
 struct UnionEvaluateOperation
 {
+  /**
+   * Local set the operation is evaluated on
+   */
+  struct Set *set;
+
+  /**
+   * Peer with the remote set
+   */
+  struct GNUNET_PeerIdentity peer;
+
+  /**
+   * Application-specific identifier
+   */
+  struct GNUNET_HashCode app_id;
+
+  /**
+   * Context message, given to us
+   * by the client, may be NULL.
+   */
+  struct GNUNET_MessageHeader *context_msg;
+
+  /**
+   * Stream socket connected to the other peer
+   */
+  struct GNUNET_STREAM_Socket *socket;
+
+  /**
+   * Message queue for the peer on the other
+   * end
+   */
+  struct GNUNET_MQ_MessageQueue *mq;
+
+  /**
+   * Type of this operation
+   */
+  enum GNUNET_SET_OperationType operation;
+
+  /**
+   * GNUNET_YES if we started the operation,
+   * GNUNET_NO if the other peer started it.
+   */
+  int is_outgoing;
+
+  /**
+   * Request id, so we can use one client handle
+   * for multiple operations
+   */
+  uint32_t request_id;
+
   /* last difference estimate */
   unsigned int diff;
 
@@ -95,51 +146,109 @@ struct UnionEvaluateOperation
    */
   unsigned int ibf_order;
 
+  struct StrataEstimator *se;
+
   /**
    * The ibf we currently receive
    */
-  struct InvertibleBloomFilter *ibf_received;
+  struct InvertibleBloomFilter *remote_ibf;
 
-  struct StrataEstimator *se;
+  /**
+   * Array of IBFs, some of them pre-allocated
+   */
+  struct InvertibleBloomFilter *local_ibf;
+
+  /**
+   * Elements we received from the other peer.
+   */
+  struct GNUNET_CONTAINER_MultiHashMap *received_elements;
+
+  /**
+   * Maps IBF-Keys (specific to the current salt) to elements.
+   */
+  struct GNUNET_CONTAINER_MultiHashMap32 *key_to_element;
 
   /**
    * Current state of the operation
    */
   enum UnionOperationState state;
+  
+  /**
+   * Evaluate operations are held in
+   * a linked list.
+   */
+  struct UnionEvaluateOperation *next;
+  
+   /**
+   * Evaluate operations are held in
+   * a linked list.
+   */
+  struct UnionEvaluateOperation *prev;
 };
 
-
 /**
- * Element entry, stored in the hash maps from
- * partial IBF keys to elements.
+ * Information about the element in a set.
+ * All elements are stored in a hash-table
+ * from their hash-code to their 'struct Element',
+ * so that the remove and add operations are reasonably
+ * fast.
  */
 struct ElementEntry
 {
   /**
-   * The actual element
+   * The actual element. The data for the element
+   * should be allocated at the end of this struct.
    */
-  struct GNUNET_SET_Element *element;
+  struct GNUNET_SET_Element element;
 
   /**
-   * Actual ibf key of the element entry
+   * Hash of the element.
+   * Will be used to derive the different IBF keys
+   * for different salts.
    */
-  struct IBF_Key ibf_key;
+  struct GNUNET_HashCode element_hash;
 
   /**
-   * Linked list, note that the next element
-   * has to have an ibf_key that is lexicographically
-   * equal or larger.
+   * Generation the element was added.
+   * Operations of earlier generations will not consider the element.
    */
-  struct ElementEntry *next;
+  int generation_add;
 
   /**
-   * GNUNET_YES if the element was received from
-   * the remote peer, and the local peer did not previously
-   * have it
+   * Generation this element was removed.
+   * Operations of later generations will not consider the element.
+   */
+  int generation_remove;
+
+  /**
+   * GNUNET_YES if we received the element from a remote peer, and not
+   * from the local peer.  Note that if the local client inserts an
+   * element *after* we got it from a remote peer, the element is
+   * considered local.
    */
   int remote;
 };
 
+/**
+ * Information about the element used for 
+ * a specific union operation.
+ */
+struct KeyEntry
+{
+  struct IBF_Key ibf_key;
+
+  /**
+   * The actual element associated with the key
+   */
+  struct ElementEntry *element;
+
+  /**
+   * Element that collides with this element
+   * on the ibf key
+   */
+  struct KeyEntry *next_colliding;
+};
+
 
 /**
  * Extra state required for efficient set union.
@@ -147,47 +256,72 @@ struct ElementEntry
 struct UnionState
 {
   /**
-   * Strate estimator of the set we currently have,
-   * used for estimation of the symmetric difference
+   * The strata estimator is only generated once for
+   * each set.
    */
   struct StrataEstimator *se;
 
   /**
-   * Array of IBFs, some of them pre-allocated
+   * Maps 'struct GNUNET_HashCode' to 'struct ElementEntry'.
+   */
+  struct GNUNET_CONTAINER_MultiHashMap *elements;
+
+  /**
+   * Evaluate operations are held in
+   * a linked list.
    */
-  struct InvertibleBloomFilter **ibfs;
+  struct UnionEvaluateOperation *ops_head;
 
   /**
-   * Maps the first 32 bits of the ibf-key to
-   * elements.
+   * Evaluate operations are held in
+   * a linked list.
    */
-  struct GNUNET_CONTAINER_MultiHashMap32 *elements;
+  struct UnionEvaluateOperation *ops_tail;
 };
 
 
+static struct IBF_Key
+get_ibf_key (struct GNUNET_HashCode *src, uint16_t salt)
+{
+   
+  struct IBF_Key key;
+  GNUNET_CRYPTO_hkdf (&key, sizeof (key),
+                      GCRY_MD_SHA512, GCRY_MD_SHA256,
+                      src, sizeof *src,
+                      &salt, sizeof (salt),
+                      NULL, 0);
+  return key;
+}
+
+
 static void
-send_operation_request (struct EvaluateOperation *eo)
+send_operation_request (struct UnionEvaluateOperation *eo)
 {
   struct GNUNET_MQ_Message *mqm;
   struct OperationRequestMessage *msg;
+  int ret;
 
-  mqm = GNUNET_MQ_msg_concat (msg, eo->context_msg,
-                              GNUNET_MESSAGE_TYPE_SET_P2P_OPERATION_REQUEST);
-  if (NULL == mqm)
+  mqm = GNUNET_MQ_msg (msg, GNUNET_MESSAGE_TYPE_SET_P2P_OPERATION_REQUEST);
+  ret = GNUNET_MQ_nest (mqm, eo->context_msg);
+  if (GNUNET_OK != ret)
   {
     /* the context message is too large */
-    client_disconnect (eo->set->client);
+    _GSS_client_disconnect (eo->set->client);
+    GNUNET_MQ_discard (mqm);
     GNUNET_break (0);
     return;
   }
   msg->operation = eo->operation;
   msg->app_id = eo->app_id;
   GNUNET_MQ_send (eo->mq, mqm);
+
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "sent op request\n");
 }
 
 
 /**
- * Iterator to insert values into an ibf.
+ * Iterator to create the mapping between ibf keys
+ * and element entries.
  *
  * @param cls closure
  * @param key current key code
@@ -197,54 +331,93 @@ send_operation_request (struct EvaluateOperation *eo)
  *         GNUNET_NO if not.
  */
 static int
-ibf_insert_iterator (void *cls,
-                     uint32_t key,
-                     void *value)
+insert_element_iterator (void *cls,
+                         uint32_t key,
+                         void *value)
 {
-  struct InvertibleBloomFilter *ibf = cls;
-  struct ElementEntry *e = value;
-  struct IBF_Key ibf_key;
-
-  GNUNET_assert (NULL != e);
-  ibf_key = e->ibf_key;
-  ibf_insert (ibf, ibf_key);
-  e = e->next;
+  struct KeyEntry *const new_k = cls;
+  struct KeyEntry *old_k = value;
 
-  while (NULL != e)
+  GNUNET_assert (NULL != old_k);
+  do
   {
-    /* only insert keys we haven't seen yet */
-    if (0 != memcmp (&e->ibf_key, &ibf_key, sizeof ibf_key))
+    if (old_k->ibf_key.key_val == new_k->ibf_key.key_val)
     {
-      ibf_key = e->ibf_key;
-      ibf_insert (ibf, ibf_key);
+      new_k->next_colliding = old_k;
+      old_k->next_colliding = new_k;
+      return GNUNET_NO;
     }
-    e = e->next; 
+    old_k = old_k->next_colliding;
+  } while (NULL != old_k);
+  return GNUNET_YES;
+}
+
+
+static void
+insert_element (struct UnionEvaluateOperation *eo, struct ElementEntry *ee)
+{
+  int ret;
+  struct IBF_Key ibf_key;
+  struct KeyEntry *k;
+
+  ibf_key = get_ibf_key (&ee->element_hash, eo->salt);
+  k = GNUNET_new (struct KeyEntry);
+  k->element = ee;
+  k->ibf_key = ibf_key;
+  ret = GNUNET_CONTAINER_multihashmap32_get_multiple (eo->key_to_element, (uint32_t) ibf_key.key_val,
+                                                    insert_element_iterator, k);
+  /* was the element inserted into a colliding bucket? */
+  if (GNUNET_SYSERR == ret)
+  {
+    GNUNET_assert (NULL != k->next_colliding);
+    return;
   }
+  GNUNET_CONTAINER_multihashmap32_put (eo->key_to_element, (uint32_t) ibf_key.key_val, k,
+                                       GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY);
+  if (NULL != eo->local_ibf)
+    ibf_insert (eo->local_ibf, ibf_key);
+}
+
+
+static int
+prepare_ibf_iterator (void *cls,
+                     uint32_t key,
+                     void *value)
+{
+  struct InvertibleBloomFilter *ibf = cls;
+  struct KeyEntry *ke = value;
 
+  ibf_insert (ibf, ke->ibf_key);
   return GNUNET_YES;
 }
 
-
-/**
- * Create and populate an IBF for the specified peer,
- * if it does not already exist.
- *
- * @param cpi peer to create the ibf for
- */
-static struct InvertibleBloomFilter *
-prepare_ibf (struct EvaluateOperation *eo, uint16_t order)
+static int
+init_key_to_element_iterator (void *cls,
+                              const struct GNUNET_HashCode *key,
+                              void *value)
 {
-  struct UnionState *us = eo->set->extra.u;
+  struct UnionEvaluateOperation *eo = cls;
+  struct ElementEntry *e = value;
+
+  insert_element (eo, e);
+  return GNUNET_YES;
+}
 
-  GNUNET_assert (order <= MAX_IBF_ORDER);
-  if (NULL == us->ibfs)
-    us->ibfs = GNUNET_malloc (MAX_IBF_ORDER * sizeof (struct InvertibleBloomFilter *));
-  if (NULL == us->ibfs[order])
+static void
+prepare_ibf (struct UnionEvaluateOperation *eo, uint16_t size)
+{
+  if (NULL == eo->key_to_element)
   {
-    us->ibfs[order] = ibf_create (1 << order, SE_IBF_HASH_NUM);
-    GNUNET_CONTAINER_multihashmap32_iterate (us->elements, ibf_insert_iterator, us->ibfs[order]);
+    unsigned int len;
+    len = GNUNET_CONTAINER_multihashmap_size (eo->set->state.u->elements);
+    eo->key_to_element = GNUNET_CONTAINER_multihashmap32_create (len);
+    GNUNET_CONTAINER_multihashmap_iterate (eo->set->state.u->elements,
+                                             init_key_to_element_iterator, eo);
   }
-  return us->ibfs[order];
+  if (NULL != eo->local_ibf)
+    ibf_destroy (eo->local_ibf);
+  eo->local_ibf = ibf_create (size, SE_IBF_HASH_NUM);
+  GNUNET_CONTAINER_multihashmap32_iterate (eo->key_to_element, prepare_ibf_iterator, eo->local_ibf);
 }
 
 
@@ -254,12 +427,14 @@ prepare_ibf (struct EvaluateOperation *eo, uint16_t order)
  * @param cpi the peer
  */
 static void
-send_ibf (struct EvaluateOperation *eo, uint16_t ibf_order)
+send_ibf (struct UnionEvaluateOperation *eo, uint16_t ibf_order)
 {
   unsigned int buckets_sent = 0;
   struct InvertibleBloomFilter *ibf;
 
-  ibf = prepare_ibf (eo, ibf_order);
+  prepare_ibf (eo, ibf_order);
+
+  ibf = eo->local_ibf;
 
   while (buckets_sent < (1 << ibf_order))
   {
@@ -282,7 +457,7 @@ send_ibf (struct EvaluateOperation *eo, uint16_t ibf_order)
     GNUNET_MQ_send (eo->mq, mqm);
   }
 
-  eo->extra.u->state = STATE_EXPECT_ELEMENTS_AND_REQUESTS;
+  eo->state = STATE_EXPECT_ELEMENTS_AND_REQUESTS;
 }
 
 
@@ -292,7 +467,7 @@ send_ibf (struct EvaluateOperation *eo, uint16_t ibf_order)
  * @param cpi the peer
  */
 static void
-send_strata_estimator (struct EvaluateOperation *eo)
+send_strata_estimator (struct UnionEvaluateOperation *eo)
 {
   struct GNUNET_MQ_Message *mqm;
   struct GNUNET_MessageHeader *strata_msg;
@@ -300,31 +475,36 @@ send_strata_estimator (struct EvaluateOperation *eo)
   mqm = GNUNET_MQ_msg_header_extra (strata_msg,
                                     SE_STRATA_COUNT * IBF_BUCKET_SIZE * SE_IBF_SIZE,
                                     GNUNET_MESSAGE_TYPE_SET_P2P_SE);
-  strata_estimator_write (eo->set->extra.u->se, &strata_msg[1]);
+  strata_estimator_write (eo->set->state.u->se, &strata_msg[1]);
   GNUNET_MQ_send (eo->mq, mqm);
-
-  eo->extra.u->state = STATE_EXPECT_IBF;
+  eo->state = STATE_EXPECT_IBF;
 }
 
 
 static void
 handle_p2p_strata_estimator (void *cls, const struct GNUNET_MessageHeader *mh)
 {
-  struct EvaluateOperation *eo = cls;
+  struct UnionEvaluateOperation *eo = cls;
+  struct StrataEstimator *remote_se;
   int ibf_order;
   int diff;
 
-  if (eo->extra.u->state != STATE_EXPECT_SE)
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "got se\n");
+
+  if (eo->state != STATE_EXPECT_SE)
   {
     /* FIXME: handle */
     GNUNET_break (0);
     return;
   }
-  GNUNET_assert (NULL == eo->extra.u->se);
-  eo->extra.u->se = strata_estimator_create (SE_STRATA_COUNT, SE_IBF_SIZE, SE_IBF_HASH_NUM);
-  strata_estimator_read (&mh[1], eo->extra.u->se);
-  GNUNET_assert (NULL != eo->set->extra.u->se);
-  diff = strata_estimator_difference (eo->set->extra.u->se, eo->extra.u->se);
+  remote_se = strata_estimator_create (SE_STRATA_COUNT, SE_IBF_SIZE,
+                                       SE_IBF_HASH_NUM);
+  strata_estimator_read (&mh[1], remote_se);
+  GNUNET_assert (NULL != eo->se);
+  diff = strata_estimator_difference (remote_se, eo->se);
+  strata_estimator_destroy (remote_se);
+  strata_estimator_destroy (eo->se);
+  eo->se = NULL;
   /* minimum order */
   ibf_order = 2;
   while ((1<<ibf_order) < (2 * diff))
@@ -341,16 +521,17 @@ handle_p2p_strata_estimator (void *cls, const struct GNUNET_MessageHeader *mh)
  * @param
  */
 static void
-decode (struct EvaluateOperation *eo)
+decode (struct UnionEvaluateOperation *eo)
 {
   struct IBF_Key key;
   int side;
   struct InvertibleBloomFilter *diff_ibf;
 
-  GNUNET_assert (STATE_EXPECT_ELEMENTS == eo->extra.u->state);
+  GNUNET_assert (STATE_EXPECT_ELEMENTS == eo->state);
 
-  diff_ibf = ibf_dup (prepare_ibf (eo, eo->extra.u->ibf_order));
-  ibf_subtract (diff_ibf, eo->extra.u->ibf_received);
+  prepare_ibf (eo, eo->ibf_order);
+  diff_ibf = ibf_dup (eo->local_ibf);
+  ibf_subtract (diff_ibf, eo->remote_ibf);
 
   while (1)
   {
@@ -359,7 +540,8 @@ decode (struct EvaluateOperation *eo)
     res = ibf_decode (diff_ibf, &side, &key);
     if (GNUNET_SYSERR == res)
     {
-      /* decoding failed, we tell the other peer by sending our ibf with a larger order */
+      /* decoding failed, we tell the other peer by sending our ibf
+       * with a larger order */
       GNUNET_assert (0);
       return;
     }
@@ -373,20 +555,11 @@ decode (struct EvaluateOperation *eo)
     }
     if (1 == side)
     {
-      struct ElementEntry *e;
+      //struct ElementEntry *e;
       /* we have the element(s), send it to the other peer */
-      e = GNUNET_CONTAINER_multihashmap32_get (eo->set->extra.u->elements, (uint32_t) key.key_val);
-      if (NULL == e)
-      {
-        /* FIXME */
-        GNUNET_assert (0);
-        return;
-      }
-      while (NULL != e)
-      {
-        /* FIXME: send element */
-        e = e->next;
-      }
+      //GNUNET_CONTAINER_multihashmap32_get_multiple (eo->set->state.u->elements,
+      //                                              (uint32_t) key.key_val);
+      /* FIXME */
     }
     else
     {
@@ -403,37 +576,35 @@ decode (struct EvaluateOperation *eo)
 }
 
 
-
 static void
 handle_p2p_ibf (void *cls, const struct GNUNET_MessageHeader *mh)
 {
-  struct EvaluateOperation *eo = cls;
-  struct UnionEvaluateOperation *ueo = eo->extra.u;
+  struct UnionEvaluateOperation *eo = cls;
   struct IBFMessage *msg = (struct IBFMessage *) mh;
   unsigned int buckets_in_message;
 
-  if (ueo->state == STATE_EXPECT_ELEMENTS_AND_REQUESTS)
+  if (eo->state == STATE_EXPECT_ELEMENTS_AND_REQUESTS)
   {
     /* check that the ibf is a new one / first part */
     /* clear outgoing messages */
     GNUNET_assert (0);
   }
-  else if (ueo->state == STATE_EXPECT_IBF)
+  else if (eo->state == STATE_EXPECT_IBF)
   {
-    ueo->state = STATE_EXPECT_IBF_CONT;
-    ueo->ibf_order = msg->order;
-    GNUNET_assert (NULL == ueo->ibf_received);
-    ueo->ibf_received = ibf_create (1<<msg->order, SE_IBF_HASH_NUM);
+    eo->state = STATE_EXPECT_IBF_CONT;
+    eo->ibf_order = msg->order;
+    GNUNET_assert (NULL == eo->remote_ibf);
+    eo->remote_ibf = ibf_create (1<<msg->order, SE_IBF_HASH_NUM);
     if (ntohs (msg->offset) != 0)
     {
       /* FIXME: handle */
       GNUNET_assert (0);
     }
   }
-  else if (ueo->state == STATE_EXPECT_IBF_CONT)
+  else if (eo->state == STATE_EXPECT_IBF_CONT)
   {
-    if ( (ntohs (msg->offset) != ueo->ibf_buckets_received) ||
-         (msg->order != ueo->ibf_order) )
+    if ( (ntohs (msg->offset) != eo->ibf_buckets_received) ||
+         (msg->order != eo->ibf_order) )
     {
       /* FIXME: handle */
       GNUNET_assert (0);
@@ -448,12 +619,12 @@ handle_p2p_ibf (void *cls, const struct GNUNET_MessageHeader *mh)
     GNUNET_assert (0);
   }
 
-  ibf_read_slice (&msg[1], ueo->ibf_buckets_received, buckets_in_message, ueo->ibf_received);
-  ueo->ibf_buckets_received += buckets_in_message;
+  ibf_read_slice (&msg[1], eo->ibf_buckets_received, buckets_in_message, eo->remote_ibf);
+  eo->ibf_buckets_received += buckets_in_message;
 
-  if (ueo->ibf_buckets_received == (1<<ueo->ibf_order))
+  if (eo->ibf_buckets_received == (1<<eo->ibf_order))
   {
-    ueo->state = STATE_EXPECT_ELEMENTS;
+    eo->state = STATE_EXPECT_ELEMENTS;
     decode (eo);
   }
 }
@@ -462,10 +633,10 @@ handle_p2p_ibf (void *cls, const struct GNUNET_MessageHeader *mh)
 static void
 handle_p2p_elements (void *cls, const struct GNUNET_MessageHeader *mh)
 {
-  struct EvaluateOperation *eo = cls;
+  struct UnionEvaluateOperation *eo = cls;
 
-  if ( (eo->extra.u->state != STATE_EXPECT_ELEMENTS) &&
-       (eo->extra.u->state != STATE_EXPECT_ELEMENTS_AND_REQUESTS) )
+  if ( (eo->state != STATE_EXPECT_ELEMENTS) &&
+       (eo->state != STATE_EXPECT_ELEMENTS_AND_REQUESTS) )
   {
     /* FIXME: handle */
     GNUNET_break (0);
@@ -477,10 +648,10 @@ handle_p2p_elements (void *cls, const struct GNUNET_MessageHeader *mh)
 static void
 handle_p2p_element_requests (void *cls, const struct GNUNET_MessageHeader *mh)
 {
-  struct EvaluateOperation *eo = cls;
+  struct UnionEvaluateOperation *eo = cls;
 
   /* look up elements and send them */
-  if (eo->extra.u->state != STATE_EXPECT_ELEMENTS_AND_REQUESTS)
+  if (eo->state != STATE_EXPECT_ELEMENTS_AND_REQUESTS)
   {
     /* FIXME: handle */
     GNUNET_break (0);
@@ -508,141 +679,107 @@ static const struct GNUNET_MQ_Handler union_handlers[] = {
 
 /**
  * Functions of this type will be called when a stream is established
- *
+ * 
  * @param cls the closure from GNUNET_STREAM_open
- * @param socket socket to use to communicate with the other side (read/write)
+ * @param socket socket to use to communicate with the
+ *        other side (read/write)
  */
 static void
 stream_open_cb (void *cls,
                 struct GNUNET_STREAM_Socket *socket)
 {
-  struct EvaluateOperation *eo = cls;
+  struct UnionEvaluateOperation *eo = cls;
 
   GNUNET_assert (NULL == eo->mq);
   GNUNET_assert (socket == eo->socket);
 
-  eo->mq = GNUNET_MQ_queue_for_stream_socket (eo->socket, union_handlers, eo);
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "open cb successful\n");
+
+  eo->mq = GNUNET_MQ_queue_for_stream_socket (eo->socket,
+                                              union_handlers, eo);
+  /* we started the operation, thus we have to send the operation request */
   send_operation_request (eo);
+  eo->state = STATE_EXPECT_SE;
 }
         
 
 void
-union_evaluate (struct EvaluateOperation *eo)
+_GSS_union_evaluate (struct EvaluateMessage *m, struct Set *set)
 {
-  GNUNET_assert (GNUNET_SET_OPERATION_UNION == eo->set->operation);
-  eo->socket = 
-      GNUNET_STREAM_open (configuration, &eo->peer, GNUNET_APPLICATION_TYPE_SET,
-                          stream_open_cb, GNUNET_STREAM_OPTION_END);
-}
-
-
-static void
-insert_ibf_key_unchecked (struct UnionState *us, struct IBF_Key ibf_key)
-{
-  int i;
-
-  strata_estimator_insert (us->se, ibf_key);
-  for (i = 0; i <= MAX_IBF_ORDER; i++)
-  {
-    if (NULL == us->ibfs)
-      break;
-    if (NULL == us->ibfs[i])
-      continue;
-    ibf_insert (us->ibfs[i], ibf_key);
-  }
-}
+  struct UnionEvaluateOperation *eo;
 
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "evaluating union operation\n");
 
-/**
- * Insert an element into the consensus set of the specified session.
- * The element will not be copied, and freed when destroying the session.
- *
- * @param session session for new element
- * @param element element to insert
- */
-static void
-insert_element (struct Set *set, struct GNUNET_SET_Element *element)
-{
-  struct UnionState *us = set->extra.u;
-  struct GNUNET_HashCode hash;
-  struct ElementEntry *e;
-  struct ElementEntry *e_old;
-
-  e = GNUNET_new (struct ElementEntry);
-  e->element = element;
-  GNUNET_CRYPTO_hash (e->element->data, e->element->size, &hash);
-  e->ibf_key = ibf_key_from_hashcode (&hash);
-
-  e_old = GNUNET_CONTAINER_multihashmap32_get (us->elements, (uint32_t) e->ibf_key.key_val);
-  if (NULL == e_old)
-  {
-    GNUNET_CONTAINER_multihashmap32_put (us->elements, (uint32_t) e->ibf_key.key_val, e,
-                                         GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY);
-    return;
-  }
-
-  while (NULL != e_old)
-  {
-    int cmp = memcmp (&e->ibf_key, &e_old->ibf_key, sizeof (struct IBF_Key));
-    if (cmp < 0)
-    {
-      if (NULL == e_old->next)
-      {
-        e_old->next = e;
-        insert_ibf_key_unchecked (us, e->ibf_key);
-        return;
-      }
-      e_old = e_old->next;
-    }
-    else if (cmp == 0)
-    {
-      e->next = e_old->next;
-      e_old->next = e;
-      return;
-    }
-    else
-    {
-      e->next = e_old;
-      insert_ibf_key_unchecked (us, e->ibf_key);
-      return;
-    }
-  } 
+  eo = GNUNET_new (struct UnionEvaluateOperation);
+  eo->peer = m->peer;
+  eo->set = set;
+  eo->socket = 
+      GNUNET_STREAM_open (configuration, &eo->peer, GNUNET_APPLICATION_TYPE_SET,
+                          stream_open_cb, eo,
+                          GNUNET_STREAM_OPTION_END);
 }
 
 
 void
-union_accept (struct EvaluateOperation *eo, struct Incoming *incoming)
+_GSS_union_accept (struct AcceptMessage *m, struct Set *set,
+                   struct Incoming *incoming)
 {
-  GNUNET_assert (NULL != incoming->mq); 
+  struct UnionEvaluateOperation *eo;
+
+  eo = GNUNET_new (struct UnionEvaluateOperation);
+  eo->set = set;
+  eo->peer = incoming->peer;
+  eo->app_id = incoming->app_id;
+  eo->salt = ntohs (incoming->salt);
+  eo->request_id = m->request_id;
+  eo->set = set;
   eo->mq = incoming->mq;
+  /* the peer's socket is now ours, we'll receive all messages */
   GNUNET_MQ_replace_handlers (eo->mq, union_handlers, eo);
-
+  /* kick of the operation */
   send_strata_estimator (eo);
 }
 
 
 struct Set *
-union_set_create ()
+_GSS_union_set_create (void)
 {
   struct Set *set;
+
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "set created\n");
+  
   set = GNUNET_malloc (sizeof (struct Set) + sizeof (struct UnionState));
-  set->extra.u = (struct UnionState *) &set[1];
+  set->state.u = (struct UnionState *) &set[1];
   set->operation = GNUNET_SET_OPERATION_UNION;
-  set->extra.u->se = strata_estimator_create (SE_STRATA_COUNT, SE_IBF_SIZE, SE_IBF_HASH_NUM);
+  set->state.u->se = strata_estimator_create (SE_STRATA_COUNT,
+                                              SE_IBF_SIZE, SE_IBF_HASH_NUM);  
   return set;
 }
 
 
+
 void
-union_add (struct Set *set, struct ElementMessage *m)
+_GSS_union_add (struct ElementMessage *m, struct Set *set)
 {
-  struct GNUNET_SET_Element *element;
+  struct ElementEntry *ee;
+  struct ElementEntry *ee_dup;
   uint16_t element_size;
+  
   element_size = ntohs (m->header.size) - sizeof *m;
-  element = GNUNET_malloc (sizeof *element + element_size);
-  element->size = element_size;
-  element->data = &element[1];
-  memcpy (element->data, &m[1], element_size);
-  insert_element (set, element);
+  ee = GNUNET_malloc (element_size + sizeof *ee);
+  ee->element.size = element_size;
+  ee->element.data = &ee[1];
+  memcpy (ee->element.data, &m[1], element_size);
+  GNUNET_CRYPTO_hash (ee->element.data, element_size, &ee->element_hash);
+  ee_dup = GNUNET_CONTAINER_multihashmap_get (set->state.u->elements, &ee->element_hash);
+  if (NULL != ee_dup)
+  {
+    GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "element inserted twice, ignoring\n");
+    GNUNET_free (ee);
+    return;
+  }
+  GNUNET_CONTAINER_multihashmap_put (set->state.u->elements, &ee->element_hash, ee,
+                                     GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY);
+  strata_estimator_insert (set->state.u->se, get_ibf_key (&ee->element_hash, 0));
 }
 
diff --git a/src/set/gnunet-set.c b/src/set/gnunet-set.c
index 7de687611..c49b60dfd 100644
--- a/src/set/gnunet-set.c
+++ b/src/set/gnunet-set.c
@@ -30,6 +30,8 @@
 #include "gnunet_set_service.h"
 
 
+static struct GNUNET_PeerIdentity local_id;
+
 static struct GNUNET_HashCode app_id;
 static struct GNUNET_SET_Handle *set1;
 static struct GNUNET_SET_Handle *set2;
@@ -45,6 +47,13 @@ listen_cb (void *cls,
   GNUNET_log (GNUNET_ERROR_TYPE_INFO, "listen cb called\n");
 }
 
+static void
+result_cb (void *cls, struct GNUNET_SET_Element *element,
+           enum GNUNET_SET_Status status)
+{
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "got result\n");
+}
+
 
 /**
  * Main function that will be run.
@@ -60,12 +69,19 @@ run (void *cls, char *const *args,
      const struct GNUNET_CONFIGURATION_Handle *cfg)
 {
   static const char* app_str = "gnunet-set";
+  
   GNUNET_CRYPTO_hash (app_str, strlen (app_str), &app_id);
 
+  GNUNET_CRYPTO_get_host_identity (cfg, &local_id);
+
   set1 = GNUNET_SET_create (cfg, GNUNET_SET_OPERATION_UNION);
   set2 = GNUNET_SET_create (cfg, GNUNET_SET_OPERATION_UNION);
-  listen_handle = GNUNET_SET_listen (cfg, GNUNET_SET_OPERATION_UNION, &app_id,
-                                     listen_cb, NULL);
+  listen_handle = GNUNET_SET_listen (cfg, GNUNET_SET_OPERATION_UNION,
+                                     &app_id, listen_cb, NULL);
+
+  GNUNET_SET_evaluate (set1, &local_id, &app_id, NULL, 42,
+                       GNUNET_TIME_UNIT_FOREVER_REL, GNUNET_SET_RESULT_ADDED,
+                       result_cb, NULL);
 }
 
 
diff --git a/src/set/mq.c b/src/set/mq.c
index 236a692d4..92120a607 100644
--- a/src/set/mq.c
+++ b/src/set/mq.c
@@ -215,21 +215,25 @@ GNUNET_MQ_msg_ (struct GNUNET_MessageHeader **mhp, uint16_t size, uint16_t type)
 }
 
 
-struct GNUNET_MQ_Message *
-GNUNET_MQ_msg_concat_ (struct GNUNET_MessageHeader **mhp, uint16_t base_size, struct GNUNET_MessageHeader *m, uint16_t type)
+int
+GNUNET_MQ_nest_ (struct GNUNET_MQ_Message **mqmp,
+                 const struct GNUNET_MessageHeader *m)
 {
-  struct GNUNET_MQ_Message *mq;
+  size_t new_size;
+  size_t old_size;
 
-  GNUNET_assert (NULL != mhp);
   if (NULL == m)
-    return GNUNET_MQ_msg_ (mhp, base_size, type);
-  GNUNET_assert (ntohs (m->size >= sizeof (struct GNUNET_MessageHeader)));
-  /* check for overflow */
-  if (base_size + ntohs (m->size) <= base_size)
-    return NULL; 
-  mq = GNUNET_MQ_msg_ (mhp, base_size + ntohs (m->size), type);
-  memcpy (((void *) *mhp) + base_size, m, ntohs (m->size));
-  return mq;
+    return GNUNET_OK;
+  GNUNET_assert (NULL != mqmp);
+  old_size = ntohs ((*mqmp)->mh->size);
+  /* message too large to concatenate? */
+  if (ntohs ((*mqmp)->mh->size) + ntohs (m->size) < ntohs (m->size))
+    return GNUNET_SYSERR;
+  new_size = old_size + ntohs (m->size);
+  *mqmp = GNUNET_realloc (mqmp, sizeof (struct GNUNET_MQ_Message) + new_size);
+  memcpy ((*mqmp)->mh + old_size, m, new_size - old_size);
+  (*mqmp)->mh->size = htons (new_size);
+  return GNUNET_OK;
 }
 
 
@@ -274,20 +278,26 @@ stream_write_queued (void *cls, enum GNUNET_STREAM_Status status, size_t size)
     return;
   GNUNET_CONTAINER_DLL_remove (mq->msg_head, mq->msg_tail, mqm);
   mss->wh = GNUNET_STREAM_write (mss->socket, mqm->mh, ntohs (mqm->mh->size),
-                                 GNUNET_TIME_UNIT_FOREVER_REL, stream_write_queued, cls);
+                                 GNUNET_TIME_UNIT_FOREVER_REL,
+                                 stream_write_queued, mq);
   GNUNET_assert (NULL != mss->wh);
 }
 
 
 static void
-stream_socket_send_impl (struct GNUNET_MQ_MessageQueue *mq, struct GNUNET_MQ_Message *mqm)
+stream_socket_send_impl (struct GNUNET_MQ_MessageQueue *mq,
+                         struct GNUNET_MQ_Message *mqm)
 {
+  struct MessageStreamState *mss = (struct MessageStreamState *) mq->impl_state;
   if (NULL != mq->current_msg)
   {
     GNUNET_CONTAINER_DLL_insert_tail (mq->msg_head, mq->msg_tail, mqm);
     return;
   }
-  stream_write_queued (mq, GNUNET_STREAM_OK, 0);
+  mq->current_msg = mqm;
+  mss->wh = GNUNET_STREAM_write (mss->socket, mqm->mh, ntohs (mqm->mh->size),
+                                 GNUNET_TIME_UNIT_FOREVER_REL,
+                                 stream_write_queued, mq);
 }
 
 
@@ -304,7 +314,8 @@ stream_socket_send_impl (struct GNUNET_MQ_MessageQueue *mq, struct GNUNET_MQ_Mes
  * @return GNUNET_OK on success, GNUNET_SYSERR to stop further processing
  */
 static int
-stream_mst_callback (void *cls, void *client, const struct GNUNET_MessageHeader *message)
+stream_mst_callback (void *cls, void *client,
+                     const struct GNUNET_MessageHeader *message)
 {
   struct GNUNET_MQ_MessageQueue *mq = cls;
 
@@ -334,12 +345,14 @@ stream_data_processor (void *cls,
   struct GNUNET_MQ_MessageQueue *mq = cls;
   struct MessageStreamState *mss;
   int ret;
+  
   mss = (struct MessageStreamState *) mq->impl_state;
-
   GNUNET_assert (GNUNET_STREAM_OK == status);
   ret = GNUNET_SERVER_mst_receive (mss->mst, NULL, data, size, GNUNET_NO, GNUNET_NO);
   GNUNET_assert (GNUNET_OK == ret);
   /* we always read all data */
+    mss->rh = GNUNET_STREAM_read (mss->socket, GNUNET_TIME_UNIT_FOREVER_REL, 
+                                  stream_data_processor, mq);
   return size;
 }
 
@@ -369,8 +382,7 @@ GNUNET_MQ_queue_for_stream_socket (struct GNUNET_STREAM_Socket *socket,
 }
 
 
-/**
- * Transmit a queued message to the session's client.
+/*** Transmit a queued message to the session's client.
  *
  * @param cls consensus session
  * @param size number of bytes available in buf
@@ -474,7 +486,7 @@ connection_client_transmit_queued (void *cls, size_t size,
     mq->current_msg = mq->msg_head;
     GNUNET_CONTAINER_DLL_remove (mq->msg_head, mq->msg_tail, mq->current_msg);
     state->th = 
-        GNUNET_CLIENT_notify_transmit_ready (state->connection, msg_size, 
+      GNUNET_CLIENT_notify_transmit_ready (state->connection, htons (mq->current_msg->mh->size), 
                                              GNUNET_TIME_UNIT_FOREVER_REL, GNUNET_NO,
                                              &connection_client_transmit_queued, mq);
   }
@@ -483,7 +495,8 @@ connection_client_transmit_queued (void *cls, size_t size,
 
 
 static void
-connection_client_send_impl (struct GNUNET_MQ_MessageQueue *mq, struct GNUNET_MQ_Message *mqm)
+connection_client_send_impl (struct GNUNET_MQ_MessageQueue *mq,
+                             struct GNUNET_MQ_Message *mqm)
 {
   struct ClientConnectionState *state = mq->impl_state;
   int msize;
@@ -519,7 +532,6 @@ handle_client_message (void *cls,
   struct GNUNET_MQ_MessageQueue *mq = cls;
 
   GNUNET_assert (NULL != msg);
-
   dispatch_message (mq, msg);
 }
 
diff --git a/src/set/mq.h b/src/set/mq.h
index 3d8be789d..371bb5846 100644
--- a/src/set/mq.h
+++ b/src/set/mq.h
@@ -58,20 +58,7 @@
  */
 #define GNUNET_MQ_msg(mvar, type) GNUNET_MQ_msg_extra(mvar, 0, type)
 
-/**
- * Allocate a GNUNET_MQ_Message, and concatenate another message
- * after the space needed by the message struct.
- * // nest?
- *
- * @param mvar variable to store the allocated message in;
- *             must have a header field
- * @param mc message to concatenate, can be NULL
- * @param type type of the message
- * @return the MQ message, NULL if mc is to large to be concatenated
- */
-#define GNUNET_MQ_msg_concat(mvar, mc, t) GNUNET_MQ_msg_concat_(((void) mvar->header, (struct GNUNET_MessageHeader **) &(mvar)), \
-      sizeof *mvar, (struct GNUNET_MessageHeader *) mc, t)
-
+#define GNUNET_MQ_nest(mqm, mh) GNUNET_MQ_nest_ (&mqm, mh)
 
 /**
  * Allocate a GNUNET_MQ_Message, where the message only consists of a header.
@@ -157,14 +144,9 @@ struct GNUNET_MQ_Message *
 GNUNET_MQ_msg_ (struct GNUNET_MessageHeader **mhp, uint16_t size, uint16_t type);
 
 
-/**
- * Create a new message for MQ, by concatenating another message
- * after a message of the specified type.
- *
- * @retrn the allocated MQ message
- */
-struct GNUNET_MQ_Message *
-GNUNET_MQ_msg_concat_ (struct GNUNET_MessageHeader **mhp, uint16_t base_size, struct GNUNET_MessageHeader *m, uint16_t type);
+int
+GNUNET_MQ_nest_ (struct GNUNET_MQ_Message **mqmp,
+                 const struct GNUNET_MessageHeader *m);
 
 
 /**
diff --git a/src/set/set_api.c b/src/set/set_api.c
index b2491afe7..daa15c081 100644
--- a/src/set/set_api.c
+++ b/src/set/set_api.c
@@ -288,6 +288,7 @@ GNUNET_SET_evaluate (struct GNUNET_SET_Handle *set,
                      const struct GNUNET_PeerIdentity *other_peer,
                      const struct GNUNET_HashCode *app_id,
                      const struct GNUNET_MessageHeader *context_msg,
+                     uint16_t salt,
                      struct GNUNET_TIME_Relative timeout,
                      enum GNUNET_SET_ResultMode result_mode,
                      GNUNET_SET_ResultIterator result_cb,
@@ -302,11 +303,14 @@ GNUNET_SET_evaluate (struct GNUNET_SET_Handle *set,
   oh->result_cls = result_cls;
   oh->set = set;
 
-  mqm = GNUNET_MQ_msg_extra (msg, htons(context_msg->size), GNUNET_MESSAGE_TYPE_SET_EVALUATE);
+  mqm = GNUNET_MQ_msg (msg, GNUNET_MESSAGE_TYPE_SET_EVALUATE);
   msg->request_id = htonl (GNUNET_MQ_assoc_add (set->mq, mqm, oh));
   msg->peer = *other_peer;
   msg->app_id = *app_id;
-  memcpy (&msg[1], context_msg, htons (context_msg->size));
+
+  if (GNUNET_OK != GNUNET_MQ_nest (mqm, context_msg))
+    GNUNET_assert (0);
+  
   oh->timeout_task = GNUNET_SCHEDULER_add_delayed (timeout, operation_timeout_task, oh);
   GNUNET_MQ_send (set->mq, mqm);