-fixing misc issues and bugs, including better termination logic for intersection...
[oweals/gnunet.git] / src / set / gnunet-service-set_union.c
index c0072801552c1b4096159eab199d3147bcf97a6b..459996eca81c43fc8f14ecd5b3414117f1ce5af4 100644 (file)
@@ -17,9 +17,8 @@
       Free Software Foundation, Inc., 59 Temple Place - Suite 330,
       Boston, MA 02111-1307, USA.
 */
-
 /**
- * @file set/gnunet-service-set.c
+ * @file set/gnunet-service-set_union.c
  * @brief two-peer set operations
  * @author Florian Dold
  */
@@ -27,8 +26,8 @@
 #include "gnunet_util_lib.h"
 #include "gnunet-service-set.h"
 #include "ibf.h"
-#include "strata_estimator.h"
-#include "set_protocol.h"
+#include "gnunet-service-set_union_strata_estimator.h"
+#include "gnunet-service-set_protocol.h"
 #include <gcrypt.h>
 
 
@@ -76,9 +75,9 @@ enum UnionOperationPhase
 
   /**
    * We sent the strata estimator, and expect an IBF. This phase is entered once
-   * upon initialization and later via PHASE_EXPECT_ELEMENTS_AND_REQUESTS.
+   * upon initialization and later via #PHASE_EXPECT_ELEMENTS_AND_REQUESTS.
    *
-   * After receiving the complete IBF, we enter PHASE_EXPECT_ELEMENTS
+   * After receiving the complete IBF, we enter #PHASE_EXPECT_ELEMENTS
    */
   PHASE_EXPECT_IBF,
 
@@ -93,7 +92,7 @@ enum UnionOperationPhase
    *
    * We are currently decoding an IBF until it can no longer be decoded,
    * we currently send requests and expect elements
-   * The remote peer is in PHASE_EXPECT_ELEMENTS_AND_REQUESTS
+   * The remote peer is in #PHASE_EXPECT_ELEMENTS_AND_REQUESTS
    */
   PHASE_EXPECT_ELEMENTS,
 
@@ -103,10 +102,10 @@ enum UnionOperationPhase
    *
    * We are in this phase if we have SENT an IBF for the remote peer to decode.
    * We expect requests, send elements or could receive an new IBF, which takes
-   * us via PHASE_EXPECT_IBF to phase PHASE_EXPECT_ELEMENTS
+   * us via #PHASE_EXPECT_IBF to phase #PHASE_EXPECT_ELEMENTS
    *
    * The remote peer is thus in:
-   * PHASE_EXPECT_ELEMENTS
+   * #PHASE_EXPECT_ELEMENTS
    */
   PHASE_EXPECT_ELEMENTS_AND_REQUESTS,
 
@@ -171,8 +170,7 @@ struct OperationState
 
 
 /**
- * The key entry is used to associate an ibf key with
- * an element.
+ * The key entry is used to associate an ibf key with an element.
  */
 struct KeyEntry
 {
@@ -230,7 +228,8 @@ struct SetState
 
 
 /**
- * Iterator over hash map entries.
+ * Iterator over hash map entries, called to
+ * destroy the linked list of colliding ibf key entries.
  *
  * @param cls closure
  * @param key current key code
@@ -245,7 +244,7 @@ destroy_key_to_element_iter (void *cls,
                              void *value)
 {
   struct KeyEntry *k = value;
-  /* destroy the linked list of colliding ibf key entries */
+
   while (NULL != k)
   {
     struct KeyEntry *k_tmp = k;
@@ -262,14 +261,16 @@ destroy_key_to_element_iter (void *cls,
 
 
 /**
- * Destroy the union operation.  Only things specific to the union operation are destroyed.
+ * Destroy the union operation.  Only things specific to the union
+ * operation are destroyed.
  *
  * @param op union operation to destroy
  */
 static void
 union_op_cancel (struct Operation *op)
 {
-  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "destroying union op\n");
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "destroying union op\n");
   /* check if the op was canceled twice */
   GNUNET_assert (NULL != op->state);
   if (NULL != op->state->remote_ibf)
@@ -289,13 +290,16 @@ union_op_cancel (struct Operation *op)
   }
   if (NULL != op->state->key_to_element)
   {
-    GNUNET_CONTAINER_multihashmap32_iterate (op->state->key_to_element, destroy_key_to_element_iter, NULL);
+    GNUNET_CONTAINER_multihashmap32_iterate (op->state->key_to_element,
+                                             &destroy_key_to_element_iter,
+                                             NULL);
     GNUNET_CONTAINER_multihashmap32_destroy (op->state->key_to_element);
     op->state->key_to_element = NULL;
   }
   GNUNET_free (op->state);
   op->state = NULL;
-  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "destroying union op done\n");
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "destroying union op done\n");
 }
 
 
@@ -311,8 +315,8 @@ fail_union_operation (struct Operation *op)
   struct GNUNET_MQ_Envelope *ev;
   struct GNUNET_SET_ResultMessage *msg;
 
-  GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "union operation failed\n");
-
+  GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+              "union operation failed\n");
   ev = GNUNET_MQ_msg (msg, GNUNET_MESSAGE_TYPE_SET_RESULT);
   msg->result_status = htons (GNUNET_SET_STATUS_FAILURE);
   msg->request_id = htonl (op->spec->client_request_id);
@@ -331,7 +335,8 @@ fail_union_operation (struct Operation *op)
  * @return the derived IBF key
  */
 static struct IBF_Key
-get_ibf_key (const struct GNUNET_HashCode *src, uint16_t salt)
+get_ibf_key (const struct GNUNET_HashCode *src,
+             uint16_t salt)
 {
   struct IBF_Key key;
 
@@ -344,45 +349,6 @@ get_ibf_key (const struct GNUNET_HashCode *src, uint16_t salt)
 }
 
 
-/**
- * Send a request for the evaluate operation to a remote peer
- *
- * @param op operation with the other peer
- */
-static void
-send_operation_request (struct Operation *op)
-{
-  struct GNUNET_MQ_Envelope *ev;
-  struct OperationRequestMessage *msg;
-
-  ev = GNUNET_MQ_msg_nested_mh (msg, GNUNET_MESSAGE_TYPE_SET_P2P_OPERATION_REQUEST,
-                                op->spec->context_msg);
-
-  if (NULL == ev)
-  {
-    /* the context message is too large */
-    GNUNET_break (0);
-    GNUNET_SERVER_client_disconnect (op->spec->set->client);
-    return;
-  }
-  msg->operation = htonl (GNUNET_SET_OPERATION_UNION);
-  msg->app_id = op->spec->app_id;
-  msg->salt = htonl (op->spec->salt);
-  GNUNET_MQ_send (op->mq, ev);
-
-  if (NULL != op->spec->context_msg)
-    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "sent op request with context message\n");
-  else
-    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "sent op request without context message\n");
-
-  if (NULL != op->spec->context_msg)
-  {
-    GNUNET_free (op->spec->context_msg);
-    op->spec->context_msg = NULL;
-  }
-}
-
-
 /**
  * Iterator to create the mapping between ibf keys
  * and element entries.
@@ -390,8 +356,7 @@ send_operation_request (struct Operation *op)
  * @param cls closure
  * @param key current key code
  * @param value value in the hash map
- * @return #GNUNET_YES if we should continue to
- *         iterate,
+ * @return #GNUNET_YES if we should continue to iterate,
  *         #GNUNET_NO if not.
  */
 static int
@@ -423,9 +388,7 @@ op_register_element_iterator (void *cls,
  * @param cls closure
  * @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.
+ * @return #GNUNET_YES (we should continue to iterate)
  */
 static int
 op_has_element_iterator (void *cls,
@@ -438,7 +401,8 @@ op_has_element_iterator (void *cls,
   GNUNET_assert (NULL != k);
   while (NULL != k)
   {
-    if (0 == GNUNET_CRYPTO_hash_cmp (&k->element->element_hash, element_hash))
+    if (0 == GNUNET_CRYPTO_hash_cmp (&k->element->element_hash,
+                                     element_hash))
       return GNUNET_NO;
     k = k->next_colliding;
   }
@@ -455,7 +419,8 @@ op_has_element_iterator (void *cls,
  * @return #GNUNET_YES if the element has been found, #GNUNET_NO otherwise
  */
 static int
-op_has_element (struct Operation *op, const struct GNUNET_HashCode *element_hash)
+op_has_element (struct Operation *op,
+                const struct GNUNET_HashCode *element_hash)
 {
   int ret;
   struct IBF_Key ibf_key;
@@ -463,7 +428,8 @@ op_has_element (struct Operation *op, const struct GNUNET_HashCode *element_hash
   ibf_key = get_ibf_key (element_hash, op->spec->salt);
   ret = GNUNET_CONTAINER_multihashmap32_get_multiple (op->state->key_to_element,
                                                       (uint32_t) ibf_key.key_val,
-                                                      op_has_element_iterator, (void *) element_hash);
+                                                      op_has_element_iterator,
+                                                      (void *) element_hash);
 
   /* was the iteration aborted because we found the element? */
   if (GNUNET_SYSERR == ret)
@@ -497,13 +463,15 @@ op_register_element (struct Operation *op,
   k->ibf_key = ibf_key;
   ret = GNUNET_CONTAINER_multihashmap32_get_multiple (op->state->key_to_element,
                                                       (uint32_t) ibf_key.key_val,
-                                                      op_register_element_iterator, k);
+                                                      op_register_element_iterator,
+                                                      k);
 
   /* was the element inserted into a colliding bucket? */
   if (GNUNET_SYSERR == ret)
     return;
-
-  GNUNET_CONTAINER_multihashmap32_put (op->state->key_to_element, (uint32_t) ibf_key.key_val, k,
+  GNUNET_CONTAINER_multihashmap32_put (op->state->key_to_element,
+                                       (uint32_t) ibf_key.key_val,
+                                       k,
                                        GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
 }
 
@@ -523,8 +491,9 @@ prepare_ibf_iterator (void *cls,
   struct InvertibleBloomFilter *ibf = cls;
   struct KeyEntry *ke = value;
 
-  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "inserting %x into ibf\n", ke->ibf_key.key_val);
-
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "inserting %x into ibf\n",
+              ke->ibf_key.key_val);
   ibf_insert (ibf, ke->ibf_key);
   return GNUNET_YES;
 }
@@ -534,12 +503,11 @@ prepare_ibf_iterator (void *cls,
  * Iterator for initializing the
  * key-to-element mapping of a union operation
  *
- * @param cls the union operation
- * @param key unised
- * @param value the element entry to insert
+ * @param cls the union operation `struct Operation *`
+ * @param key unused
+ * @param value the `struct ElementEntry *` to insert
  *        into the key-to-element mapping
- * @return GNUNET_YES to continue iterating,
- *         GNUNET_NO to stop
+ * @return #GNUNET_YES (to continue iterating)
  */
 static int
 init_key_to_element_iterator (void *cls,
@@ -571,7 +539,8 @@ init_key_to_element_iterator (void *cls,
  * @param size size of the ibf to create
  */
 static void
-prepare_ibf (struct Operation *op, uint16_t size)
+prepare_ibf (struct Operation *op,
+             uint16_t size)
 {
   if (NULL == op->state->key_to_element)
   {
@@ -585,7 +554,8 @@ prepare_ibf (struct Operation *op, uint16_t size)
     ibf_destroy (op->state->local_ibf);
   op->state->local_ibf = ibf_create (size, SE_IBF_HASH_NUM);
   GNUNET_CONTAINER_multihashmap32_iterate (op->state->key_to_element,
-                                           prepare_ibf_iterator, op->state->local_ibf);
+                                           &prepare_ibf_iterator,
+                                           op->state->local_ibf);
 }
 
 
@@ -596,14 +566,17 @@ prepare_ibf (struct Operation *op, uint16_t size)
  * @param ibf_order order of the ibf to send, size=2^order
  */
 static void
-send_ibf (struct Operation *op, uint16_t ibf_order)
+send_ibf (struct Operation *op,
+          uint16_t ibf_order)
 {
   unsigned int buckets_sent = 0;
   struct InvertibleBloomFilter *ibf;
 
   prepare_ibf (op, 1<<ibf_order);
 
-  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "sending ibf of size %u\n", 1<<ibf_order);
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "sending ibf of size %u\n",
+              1<<ibf_order);
 
   ibf = op->state->local_ibf;
 
@@ -626,8 +599,11 @@ send_ibf (struct Operation *op, uint16_t ibf_order)
     ibf_write_slice (ibf, buckets_sent,
                      buckets_in_message, &msg[1]);
     buckets_sent += buckets_in_message;
-    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "ibf chunk size %u, %u/%u sent\n",
-                buckets_in_message, buckets_sent, 1<<ibf_order);
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                "ibf chunk size %u, %u/%u sent\n",
+                buckets_in_message,
+                buckets_sent,
+                1<<ibf_order);
     GNUNET_MQ_send (op->mq, ev);
   }
 
@@ -652,7 +628,8 @@ send_strata_estimator (struct Operation *op)
   strata_estimator_write (op->state->se, &strata_msg[1]);
   GNUNET_MQ_send (op->mq, ev);
   op->state->phase = PHASE_EXPECT_IBF;
-  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "sent SE, expecting IBF\n");
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "sent SE, expecting IBF\n");
 }
 
 
@@ -669,7 +646,8 @@ get_order_from_difference (unsigned int diff)
   unsigned int ibf_order;
 
   ibf_order = 2;
-  while ((1<<ibf_order) < (IBF_ALPHA * diff) || (1<<ibf_order) < SE_IBF_HASH_NUM)
+  while ( (1<<ibf_order) < (IBF_ALPHA * diff) ||
+          ((1<<ibf_order) < SE_IBF_HASH_NUM) )
     ibf_order++;
   if (ibf_order > MAX_IBF_ORDER)
     ibf_order = MAX_IBF_ORDER;
@@ -684,7 +662,8 @@ get_order_from_difference (unsigned int diff)
  * @param mh the message
  */
 static void
-handle_p2p_strata_estimator (void *cls, const struct GNUNET_MessageHeader *mh)
+handle_p2p_strata_estimator (void *cls,
+                             const struct GNUNET_MessageHeader *mh)
 {
   struct Operation *op = cls;
   struct StrataEstimator *remote_se;
@@ -704,9 +683,12 @@ handle_p2p_strata_estimator (void *cls, const struct GNUNET_MessageHeader *mh)
   strata_estimator_destroy (remote_se);
   strata_estimator_destroy (op->state->se);
   op->state->se = NULL;
-  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "got se diff=%d, using ibf size %d\n",
-              diff, 1<<get_order_from_difference (diff));
-  send_ibf (op, get_order_from_difference (diff));
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "got se diff=%d, using ibf size %d\n",
+              diff,
+              1<<get_order_from_difference (diff));
+  send_ibf (op,
+            get_order_from_difference (diff));
 }
 
 
@@ -737,15 +719,20 @@ send_element_iterator (void *cls,
     struct GNUNET_MessageHeader *mh;
 
     GNUNET_assert (ke->ibf_key.key_val == ibf_key.key_val);
-    ev = GNUNET_MQ_msg_header_extra (mh, element->size, GNUNET_MESSAGE_TYPE_SET_P2P_ELEMENTS);
+    ev = GNUNET_MQ_msg_header_extra (mh,
+                                     element->size,
+                                     GNUNET_MESSAGE_TYPE_SET_P2P_ELEMENTS);
     if (NULL == ev)
     {
       /* element too large */
       GNUNET_break (0);
       continue;
     }
-    memcpy (&mh[1], element->data, element->size);
-    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "sending element (%s) to peer\n",
+    memcpy (&mh[1],
+            element->data,
+            element->size);
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                "sending element (%s) to peer\n",
                 GNUNET_h2s (&ke->element->element_hash));
     GNUNET_MQ_send (op->mq, ev);
     ke = ke->next_colliding;
@@ -753,6 +740,7 @@ send_element_iterator (void *cls,
   return GNUNET_NO;
 }
 
+
 /**
  * Send all elements that have the specified IBF key
  * to the remote peer of the union operation
@@ -797,7 +785,9 @@ decode_and_send (struct Operation *op)
   ibf_destroy (op->state->remote_ibf);
   op->state->remote_ibf = NULL;
 
-  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "decoding IBF (size=%u)\n", diff_ibf->size);
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "decoding IBF (size=%u)\n",
+              diff_ibf->size);
 
   num_decoded = 0;
   last_key.key_val = 0;
@@ -812,17 +802,22 @@ decode_and_send (struct Operation *op)
     res = ibf_decode (diff_ibf, &side, &key);
     if (res == GNUNET_OK)
     {
-      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "decoded ibf key %lx\n",
+      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                  "decoded ibf key %lx\n",
                   key.key_val);
       num_decoded += 1;
-      if (num_decoded > diff_ibf->size || (num_decoded > 1 && last_key.key_val == key.key_val))
+      if ( (num_decoded > diff_ibf->size) ||
+           (num_decoded > 1 && last_key.key_val == key.key_val) )
       {
-        GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "detected cyclic ibf (decoded %u/%u)\n",
-                    num_decoded, diff_ibf->size);
+        GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                    "detected cyclic ibf (decoded %u/%u)\n",
+                    num_decoded,
+                    diff_ibf->size);
         cycle_detected = GNUNET_YES;
       }
     }
-    if ((GNUNET_SYSERR == res) || (GNUNET_YES == cycle_detected))
+    if ( (GNUNET_SYSERR == res) ||
+         (GNUNET_YES == cycle_detected) )
     {
       int next_order;
       next_order = 0;
@@ -847,8 +842,9 @@ decode_and_send (struct Operation *op)
     {
       struct GNUNET_MQ_Envelope *ev;
 
-      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "transmitted all values, sending DONE\n");
-      ev = GNUNET_MQ_msg_header (GNUNET_MESSAGE_TYPE_SET_P2P_DONE);
+      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                  "transmitted all values, sending DONE\n");
+      ev = GNUNET_MQ_msg_header (GNUNET_MESSAGE_TYPE_SET_UNION_P2P_DONE);
       GNUNET_MQ_send (op->mq, ev);
       break;
     }
@@ -867,7 +863,8 @@ decode_and_send (struct Operation *op)
                                         GNUNET_MESSAGE_TYPE_SET_P2P_ELEMENT_REQUESTS);
 
       *(struct IBF_Key *) &msg[1] = key;
-      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "sending element request\n");
+      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                  "sending element request\n");
       GNUNET_MQ_send (op->mq, ev);
     }
     else
@@ -897,7 +894,9 @@ handle_p2p_ibf (void *cls, const struct GNUNET_MessageHeader *mh)
   {
     op->state->phase = PHASE_EXPECT_IBF_CONT;
     GNUNET_assert (NULL == op->state->remote_ibf);
-    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "creating new ibf of size %u\n", 1<<msg->order);
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                "creating new ibf of size %u\n",
+                1<<msg->order);
     op->state->remote_ibf = ibf_create (1<<msg->order, SE_IBF_HASH_NUM);
     op->state->ibf_buckets_received = 0;
     if (0 != ntohs (msg->offset))
@@ -934,12 +933,16 @@ handle_p2p_ibf (void *cls, const struct GNUNET_MessageHeader *mh)
     return;
   }
 
-  ibf_read_slice (&msg[1], op->state->ibf_buckets_received, buckets_in_message, op->state->remote_ibf);
+  ibf_read_slice (&msg[1],
+                  op->state->ibf_buckets_received,
+                  buckets_in_message,
+                  op->state->remote_ibf);
   op->state->ibf_buckets_received += buckets_in_message;
 
   if (op->state->ibf_buckets_received == op->state->remote_ibf->size)
   {
-    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "received full ibf\n");
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                "received full ibf\n");
     op->state->phase = PHASE_EXPECT_ELEMENTS;
     decode_and_send (op);
   }
@@ -960,7 +963,9 @@ send_client_element (struct Operation *op,
   struct GNUNET_MQ_Envelope *ev;
   struct GNUNET_SET_ResultMessage *rm;
 
-  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "sending element (size %u) to client\n", element->size);
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "sending element (size %u) to client\n",
+              element->size);
   GNUNET_assert (0 != op->spec->client_request_id);
   ev = GNUNET_MQ_msg_extra (rm, element->size, GNUNET_MESSAGE_TYPE_SET_RESULT);
   if (NULL == ev)
@@ -1014,16 +1019,18 @@ send_remaining_elements (void *cls)
   struct KeyEntry *ke;
   int res;
 
-  res = GNUNET_CONTAINER_multihashmap32_iterator_next (op->state->full_result_iter, NULL, (const void **) &ke);
+  res = GNUNET_CONTAINER_multihashmap32_iterator_next (op->state->full_result_iter,
+                                                       NULL,
+                                                       (const void **) &ke);
   if (GNUNET_NO == res)
   {
-    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "sending done and destroy because iterator ran out\n");
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                "sending done and destroy because iterator ran out\n");
     send_done_and_destroy (op);
     return;
   }
-
-  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "sending elements from key entry\n");
-
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "sending elements from key entry\n");
   while (1)
   {
     struct GNUNET_MQ_Envelope *ev;
@@ -1031,7 +1038,9 @@ send_remaining_elements (void *cls)
     struct GNUNET_SET_Element *element;
     element = &ke->element->element;
 
-    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "sending element (size %u) to client (full set)\n", element->size);
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                "sending element (size %u) to client (full set)\n",
+                element->size);
     GNUNET_assert (0 != op->spec->client_request_id);
     ev = GNUNET_MQ_msg_extra (rm, element->size, GNUNET_MESSAGE_TYPE_SET_RESULT);
     if (NULL == ev)
@@ -1044,7 +1053,7 @@ send_remaining_elements (void *cls)
     rm->request_id = htonl (op->spec->client_request_id);
     rm->element_type = element->element_type;
     memcpy (&rm[1], element->data, element->size);
-    if (ke->next_colliding == NULL)
+    if (NULL == ke->next_colliding)
     {
       GNUNET_MQ_notify_sent (ev, send_remaining_elements, op);
       GNUNET_MQ_send (op->spec->set->client_mq, ev);
@@ -1073,7 +1082,8 @@ finish_and_destroy (struct Operation *op)
   {
     /* prevent that the op is free'd by the tunnel end handler */
     op->keep = GNUNET_YES;
-    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "sending full result set\n");
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                "sending full result set\n");
     GNUNET_assert (NULL == op->state->full_result_iter);
     op->state->full_result_iter =
         GNUNET_CONTAINER_multihashmap32_iterator_create (op->state->key_to_element);
@@ -1091,7 +1101,8 @@ finish_and_destroy (struct Operation *op)
  * @param mh the message
  */
 static void
-handle_p2p_elements (void *cls, const struct GNUNET_MessageHeader *mh)
+handle_p2p_elements (void *cls,
+                     const struct GNUNET_MessageHeader *mh)
 {
   struct Operation *op = cls;
   struct ElementEntry *ee;
@@ -1188,15 +1199,17 @@ handle_p2p_done (void *cls, const struct GNUNET_MessageHeader *mh)
   {
     /* we got all requests, but still have to send our elements as response */
 
-    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "got DONE, sending final DONE after elements\n");
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                "got DONE, sending final DONE after elements\n");
     op->state->phase = PHASE_FINISHED;
-    ev = GNUNET_MQ_msg_header (GNUNET_MESSAGE_TYPE_SET_P2P_DONE);
+    ev = GNUNET_MQ_msg_header (GNUNET_MESSAGE_TYPE_SET_UNION_P2P_DONE);
     GNUNET_MQ_send (op->mq, ev);
     return;
   }
   if (op->state->phase == PHASE_EXPECT_ELEMENTS)
   {
-    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "got final DONE\n");
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                "got final DONE\n");
     op->state->phase = PHASE_FINISHED;
     finish_and_destroy (op);
     return;
@@ -1207,21 +1220,46 @@ handle_p2p_done (void *cls, const struct GNUNET_MessageHeader *mh)
 
 
 /**
- * Evaluate a union operation with
- * a remote peer.
+ * Initiate operation to evaluate a set union with a remote peer.
  *
- * @param op operation to evaluate
+ * @param op operation to perform (to be initialized)
+ * @param opaque_context message to be transmitted to the listener
+ *        to convince him to accept, may be NULL
  */
 static void
-union_evaluate (struct Operation *op)
+union_evaluate (struct Operation *op,
+                const struct GNUNET_MessageHeader *opaque_context)
 {
+  struct GNUNET_MQ_Envelope *ev;
+  struct OperationRequestMessage *msg;
+
   op->state = GNUNET_new (struct OperationState);
-  // copy the current generation's strata estimator for this operation
+  /* copy the current generation's strata estimator for this operation */
   op->state->se = strata_estimator_dup (op->spec->set->state->se);
   /* we started the operation, thus we have to send the operation request */
   op->state->phase = PHASE_EXPECT_SE;
-  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "evaluating union operation\n");
-  send_operation_request (op);
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "Initiating union operation evaluation\n");
+  ev = GNUNET_MQ_msg_nested_mh (msg,
+                                GNUNET_MESSAGE_TYPE_SET_P2P_OPERATION_REQUEST,
+                                opaque_context);
+  if (NULL == ev)
+  {
+    /* the context message is too large */
+    GNUNET_break (0);
+    GNUNET_SERVER_client_disconnect (op->spec->set->client);
+    return;
+  }
+  msg->operation = htonl (GNUNET_SET_OPERATION_UNION);
+  msg->app_id = op->spec->app_id;
+  GNUNET_MQ_send (op->mq, ev);
+
+  if (NULL != opaque_context)
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                "sent op request with context message\n");
+  else
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                "sent op request without context message\n");
 }
 
 
@@ -1234,7 +1272,8 @@ union_evaluate (struct Operation *op)
 static void
 union_accept (struct Operation *op)
 {
-  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "accepting set union operation\n");
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "accepting set union operation\n");
   op->state = GNUNET_new (struct OperationState);
   op->state->se = strata_estimator_dup (op->spec->set->state->se);
   /* kick off the operation */
@@ -1255,11 +1294,11 @@ union_set_create (void)
 {
   struct SetState *set_state;
 
-  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "union set created\n");
-
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "union set created\n");
   set_state = GNUNET_new (struct SetState);
   set_state->se = strata_estimator_create (SE_STRATA_COUNT,
-                                              SE_IBF_SIZE, SE_IBF_HASH_NUM);
+                                           SE_IBF_SIZE, SE_IBF_HASH_NUM);
   return set_state;
 }
 
@@ -1273,7 +1312,8 @@ union_set_create (void)
 static void
 union_add (struct SetState *set_state, struct ElementEntry *ee)
 {
-  strata_estimator_insert (set_state->se, get_ibf_key (&ee->element_hash, 0));
+  strata_estimator_insert (set_state->se,
+                           get_ibf_key (&ee->element_hash, 0));
 }
 
 
@@ -1287,7 +1327,8 @@ union_add (struct SetState *set_state, struct ElementEntry *ee)
 static void
 union_remove (struct SetState *set_state, struct ElementEntry *ee)
 {
-  strata_estimator_remove (set_state->se, get_ibf_key (&ee->element_hash, 0));
+  strata_estimator_remove (set_state->se,
+                           get_ibf_key (&ee->element_hash, 0));
 }
 
 
@@ -1320,8 +1361,10 @@ int
 union_handle_p2p_message (struct Operation *op,
                           const struct GNUNET_MessageHeader *mh)
 {
-  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "received p2p message (t: %u, s: %u)\n",
-              ntohs (mh->type), ntohs (mh->size));
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "received p2p message (t: %u, s: %u)\n",
+              ntohs (mh->type),
+              ntohs (mh->size));
   switch (ntohs (mh->type))
   {
     case GNUNET_MESSAGE_TYPE_SET_UNION_P2P_IBF:
@@ -1336,7 +1379,7 @@ union_handle_p2p_message (struct Operation *op,
     case GNUNET_MESSAGE_TYPE_SET_P2P_ELEMENT_REQUESTS:
       handle_p2p_element_requests (op, mh);
       break;
-    case GNUNET_MESSAGE_TYPE_SET_P2P_DONE:
+    case GNUNET_MESSAGE_TYPE_SET_UNION_P2P_DONE:
       handle_p2p_done (op, mh);
       break;
     default:
@@ -1371,7 +1414,8 @@ union_peer_disconnect (struct Operation *op)
     return;
   }
   // else: the session has already been concluded
-  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "other peer disconnected (finished)\n");
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "other peer disconnected (finished)\n");
   if (GNUNET_NO == op->state->client_done_sent)
     finish_and_destroy (op);
 }