gnunet-svn
[Top][All Lists]
Advanced

[Date Prev][Date Next][Thread Prev][Thread Next][Date Index][Thread Index]

[GNUnet-SVN] r27004 - in gnunet/src: include set


From: gnunet
Subject: [GNUnet-SVN] r27004 - in gnunet/src: include set
Date: Tue, 30 Apr 2013 02:04:26 +0200

Author: dold
Date: 2013-04-30 02:04:25 +0200 (Tue, 30 Apr 2013)
New Revision: 27004

Modified:
   gnunet/src/include/gnunet_set_service.h
   gnunet/src/set/gnunet-service-set.c
   gnunet/src/set/gnunet-service-set.h
   gnunet/src/set/gnunet-service-set_union.c
   gnunet/src/set/gnunet-set-bug.c
   gnunet/src/set/mq.c
   gnunet/src/set/mq.h
   gnunet/src/set/set_api.c
Log:
implemented most parts of the set service


Modified: gnunet/src/include/gnunet_set_service.h
===================================================================
--- gnunet/src/include/gnunet_set_service.h     2013-04-29 09:09:01 UTC (rev 
27003)
+++ gnunet/src/include/gnunet_set_service.h     2013-04-30 00:04:25 UTC (rev 
27004)
@@ -91,9 +91,10 @@
    */
   GNUNET_SET_STATUS_TIMEOUT,
   /**
-   * The other peer refused to to the operation with us
+   * The other peer refused to to the operation with us,
+   * or something went wrong.
    */
-  GNUNET_SET_STATUS_REFUSED,
+  GNUNET_SET_STATUS_FAILURE,
   /**
    * Success, all elements have been sent.
    */
@@ -258,6 +259,8 @@
  *        makes it harder for an attacker to exploit this
  * @param timeout result_cb will be called with GNUNET_SET_STATUS_TIMEOUT
  *        if the operation is not done after the specified time
+ * @param result_mode specified how results will be returned,
+ *        see 'GNUNET_SET_ResultMode'.
  * @param result_cb called on error or success
  * @param result_cls closure for result_cb
  * @return a handle to cancel the operation
@@ -304,14 +307,18 @@
 GNUNET_SET_listen_cancel (struct GNUNET_SET_ListenHandle *lh);
 
 
+
 /**
- * Accept a request we got via GNUNET_SET_listen
+ * Accept a request we got via GNUNET_SET_listen.
  *
  * @param request request to accept
  * @param set set used for the requested operation 
  * @param timeout timeout for the set operation
+ * @param result_mode specified how results will be returned,
+ *        see 'GNUNET_SET_ResultMode'.
  * @param result_cb callback for the results
- * @param cls closure for result_cb
+ * @param result_cls closure for result_cb
+ * @return a handle to cancel the operation
  */
 struct GNUNET_SET_OperationHandle *
 GNUNET_SET_accept (struct GNUNET_SET_Request *request,
@@ -325,7 +332,7 @@
 /**
  * Cancel the given set operation.
  *
- * @param op set operation to cancel
+ * @param oh set operation to cancel
  */
 void
 GNUNET_SET_operation_cancel (struct GNUNET_SET_OperationHandle *oh);

Modified: gnunet/src/set/gnunet-service-set.c
===================================================================
--- gnunet/src/set/gnunet-service-set.c 2013-04-29 09:09:01 UTC (rev 27003)
+++ gnunet/src/set/gnunet-service-set.c 2013-04-30 00:04:25 UTC (rev 27004)
@@ -73,6 +73,7 @@
 
 /**
  * Counter for allocating unique request IDs for clients.
+ * Used to identify incoming requests from remote peers.
  */
 static uint32_t request_id = 1;
 
@@ -128,7 +129,7 @@
 
 
 /**
- * Get the incoming socket associated with the given id
+ * Get the incoming socket associated with the given id.
  *
  * @param id id to look for
  * @return the incoming socket associated with the id,
@@ -164,13 +165,12 @@
 
 
 /**
- * Handle a request for a set operation for
+ * Handle a request for a set operation from
  * another peer.
  *
  * @param cls the incoming socket
  * @param mh the message
  */
-
 static void
 handle_p2p_operation_request (void *cls, const struct GNUNET_MessageHeader *mh)
 {
@@ -213,12 +213,12 @@
          (htons (msg->operation) != listener->operation) )
       continue;
     mqm = GNUNET_MQ_msg (cmsg, GNUNET_MESSAGE_TYPE_SET_REQUEST);
-    if (GNUNET_OK !=
-        GNUNET_MQ_nest (mqm, context_msg))
+    if (GNUNET_OK != GNUNET_MQ_nest_mh (mqm, context_msg))
     {
       /* FIXME: disconnect the peer */
       GNUNET_MQ_discard (mqm);
       GNUNET_break (0);
+      return;
     }
     incoming->request_id = request_id++;
     cmsg->request_id = htonl (incoming->request_id);
@@ -280,7 +280,7 @@
 
 
 /**
- * Called when a client wants to create a new set.
+ * Called when a client wants to create a new listener.
  *
  * @param cls unused
  * @param client client that sent the message
@@ -311,14 +311,15 @@
 
 
 /**
- * Called when a client wants to create a new set.
+ * Called when a client wants to remove an element
+ * from the set it inhabits.
  *
  * @param cls unused
  * @param client client that sent the message
  * @param m message sent by the client
  */
 static void
-handle_client_add (void *cls,
+handle_client_remove (void *cls,
                       struct GNUNET_SERVER_Client *client,
                       const struct GNUNET_MessageHeader *m)
 {
@@ -348,6 +349,44 @@
 
 
 /**
+ * Called when a client wants to add an element to a
+ * set it inhabits.
+ *
+ * @param cls unused
+ * @param client client that sent the message
+ * @param m message sent by the client
+ */
+static void
+handle_client_add (void *cls,
+                      struct GNUNET_SERVER_Client *client,
+                      const struct GNUNET_MessageHeader *m)
+{
+  struct Set *set;
+
+  set = get_set (client);
+  if (NULL == set)
+  {
+    GNUNET_break (0);
+    _GSS_client_disconnect (client);
+    return;
+  }
+  switch (set->operation)
+  {
+    case GNUNET_SET_OPERATION_UNION:
+      _GSS_union_remove ((struct ElementMessage *) m, set);
+    case GNUNET_SET_OPERATION_INTERSECTION:
+      /* FIXME: cfuchs */
+      break;
+    default:
+      GNUNET_assert (0);
+      break;
+  }
+
+  GNUNET_SERVER_receive_done (client, GNUNET_OK);
+}
+
+
+/**
  * Called when a client wants to evaluate a set operation with another peer.
  *
  * @param cls unused
@@ -423,11 +462,11 @@
 
 /**
  * Handle a request from the client to accept
- * a set operation.
+ * a set operation that came from a remote peer.
  *
  * @param cls unused
  * @param client the client
- * @param m the message
+ * @param mh the message
  */
 static void
 handle_client_accept (void *cls,
@@ -439,7 +478,6 @@
   struct AcceptMessage *msg = (struct AcceptMessage *) mh;
 
   set = get_set (client);
-  
 
   if (NULL == set)
   {
@@ -545,12 +583,14 @@
  * @param cfg configuration to use
  */
 static void
-run (void *cls, struct GNUNET_SERVER_Handle *server, const struct 
GNUNET_CONFIGURATION_Handle *cfg)
+run (void *cls, struct GNUNET_SERVER_Handle *server,
+     const struct GNUNET_CONFIGURATION_Handle *cfg)
 {
   static const struct GNUNET_SERVER_MessageHandler server_handlers[] = {
     {handle_client_create, NULL, GNUNET_MESSAGE_TYPE_SET_CREATE, 0},
     {handle_client_listen, NULL, GNUNET_MESSAGE_TYPE_SET_LISTEN, 0},
     {handle_client_add, NULL, GNUNET_MESSAGE_TYPE_SET_ADD, 0},
+    {handle_client_remove, NULL, GNUNET_MESSAGE_TYPE_SET_ADD, 0},
     {handle_client_cancel, NULL, GNUNET_MESSAGE_TYPE_SET_CANCEL, 0},
     {handle_client_evaluate, NULL, GNUNET_MESSAGE_TYPE_SET_EVALUATE, 0},
     {handle_client_ack, NULL, GNUNET_MESSAGE_TYPE_SET_ACK, 0},

Modified: gnunet/src/set/gnunet-service-set.h
===================================================================
--- gnunet/src/set/gnunet-service-set.h 2013-04-29 09:09:01 UTC (rev 27003)
+++ gnunet/src/set/gnunet-service-set.h 2013-04-30 00:04:25 UTC (rev 27004)
@@ -19,7 +19,9 @@
 */
 
 /**
- * @brief common stuff for the set service
+ * @file set/gnunet-service-set.h
+ * @brief common components for the implementation the different set operations
+ * @author Florian Dold
  */
 
 #ifndef GNUNET_SERVICE_SET_H_PRIVATE
@@ -90,6 +92,10 @@
 };
 
 
+/**
+ * A listener is inhabited by a client, and
+ * waits for evaluation requests from remote peers.
+ */
 struct Listener
 {
   /**
@@ -181,8 +187,8 @@
   enum GNUNET_SET_OperationType operation;
 
   /**
-   * Request id associated with the
-   * request coming from this client
+   * Unique request id for the request from
+   * a remote peer.
    */
   uint32_t request_id;
 };
@@ -217,6 +223,10 @@
 
 
 void
+_GSS_union_remove (struct ElementMessage *m, struct Set *set);
+
+
+void
 _GSS_union_accept (struct AcceptMessage *m, struct Set *set,
                    struct Incoming *incoming);
 

Modified: gnunet/src/set/gnunet-service-set_union.c
===================================================================
--- gnunet/src/set/gnunet-service-set_union.c   2013-04-29 09:09:01 UTC (rev 
27003)
+++ gnunet/src/set/gnunet-service-set_union.c   2013-04-30 00:04:25 UTC (rev 
27004)
@@ -60,15 +60,39 @@
 #define MAX_IBF_ORDER (16)
 
 
-enum UnionOperationState
+/**
+ * Current phase we are in for a union operation
+ */
+enum UnionOperationPhase
 {
-  STATE_EXPECT_SE,
-  STATE_EXPECT_IBF,
-  STATE_EXPECT_IBF_CONT,
-  STATE_EXPECT_ELEMENTS,
-  STATE_EXPECT_ELEMENTS_AND_REQUESTS,
-  STATE_WAIT_SENT_DONE,
-  STATE_FINISHED
+  /**
+   * We sent the request message, and expect a strata estimator
+   */
+  PHASE_EXPECT_SE,
+  /**
+   * We sent the strata estimator, and expect an IBF
+   */
+  PHASE_EXPECT_IBF,
+  /**
+   * We know what type of IBF the other peer wants to send us,
+   * and expect the remaining parts
+   */
+  PHASE_EXPECT_IBF_CONT,
+  /**
+   * We are sending request and elements,
+   * and thus only expect elements from the other peer.
+   */
+  PHASE_EXPECT_ELEMENTS,
+  /**
+   * We are expecting elements and requests, and send
+   * requested elements back to the other peer.
+   */
+  PHASE_EXPECT_ELEMENTS_AND_REQUESTS,
+  /**
+   * The protocol is over.
+   * Results may still have to be sent to the client.
+   */
+  PHASE_FINISHED
 };
 
 
@@ -79,7 +103,7 @@
 struct UnionEvaluateOperation
 {
   /**
-   * Local set the operation is evaluated on
+   * Local set the operation is evaluated on.
    */
   struct Set *set;
 
@@ -116,30 +140,20 @@
   enum GNUNET_SET_OperationType operation;
 
   /**
-   * GNUNET_YES if we started the operation,
-   * GNUNET_NO if the other peer started it.
+   * Request ID to multiplex set operations to
+   * the client inhabiting the set.
    */
-  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;
-
   /**
    * Number of ibf buckets received
    */
   unsigned int ibf_buckets_received;
 
   /**
-   * order of the ibf we receive
+   * Copy of the set's strata estimator at the time of
+   * creation of this operation
    */
-  unsigned int ibf_order;
-
   struct StrataEstimator *se;
 
   /**
@@ -148,29 +162,30 @@
   struct InvertibleBloomFilter *remote_ibf;
 
   /**
-   * Array of IBFs, some of them pre-allocated
+   * IBF of the set's element.
    */
   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
+   * Current state of the operation.
    */
-  enum UnionOperationState state;
+  enum UnionOperationPhase phase;
 
   /**
    * Salt to use for this operation.
    */
   uint16_t salt;
+
+  /**
+   * Generation in which the operation handle
+   * was created.
+   */
+  unsigned int generation_created;
   
   /**
    * Evaluate operations are held in
@@ -185,6 +200,7 @@
   struct UnionEvaluateOperation *prev;
 };
 
+
 /**
  * Information about the element in a set.
  * All elements are stored in a hash-table
@@ -208,38 +224,40 @@
   struct GNUNET_HashCode element_hash;
 
   /**
-   * Generation the element was added.
+   * Generation the element was added by the client.
    * Operations of earlier generations will not consider the element.
    */
-  int generation_add;
-
   unsigned int generation_added;
 
-  unsigned int generation_removed;
+  /**
+   * GNUNET_YES if the element has been removed in some generation.
+   */
+  int removed;
 
-  unsigned int generation_received;
-
   /**
-   * Generation this element was removed.
+   * Generation the element was removed by the client. 
    * Operations of later generations will not consider the element.
+   * Only valid if is_removed is GNUNET_YES.
    */
-  int generation_remove;
+  unsigned int generation_removed;
 
   /**
-   * 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.
+   * GNUNET_YES if the element is a remote element, and does not belong
+   * to the operation's set.
    */
   int remote;
 };
 
+
 /**
  * Information about the element used for 
  * a specific union operation.
  */
 struct KeyEntry
 {
+  /**
+   * IBF key for the entry, derived from the current salt.
+   */
   struct IBF_Key ibf_key;
 
   /**
@@ -254,7 +272,26 @@
   struct KeyEntry *next_colliding;
 };
 
+/**
+ * Used as a closure for sending elements
+ * with a specific IBF key.
+ */
+struct SendElementClosure
+{
+  /**
+   * The IBF key whose matching elements should be
+   * sent.
+   */
+  struct IBF_Key ibf_key;
 
+  /**
+   * Operation for which the elements
+   * should be sent.
+   */
+  struct UnionEvaluateOperation *eo;
+};
+
+
 /**
  * Extra state required for efficient set union.
  */
@@ -263,6 +300,8 @@
   /**
    * The strata estimator is only generated once for
    * each set.
+   * The IBF keys are derived from the element hashes with
+   * salt=0.
    */
   struct StrataEstimator *se;
 
@@ -282,14 +321,67 @@
    * a linked list.
    */
   struct UnionEvaluateOperation *ops_tail;
+
+  /**
+   * Current generation, that is, number of
+   * previously executed operations on this set
+   */
+  unsigned int current_generation;
 };
 
 
+
+
+/**
+ * Destroy a union operation, and free all resources
+ * associated with it.
+ *
+ * @param eo the union operation to destroy
+ */
+static void
+destroy_union_operation (struct UnionEvaluateOperation *eo)
+{
+  GNUNET_CONTAINER_DLL_remove (eo->set->state.u->ops_head,
+                               eo->set->state.u->ops_tail,
+                               eo);
+  GNUNET_free (eo);
+  /* FIXME: free and destroy everything else */
+}
+
+
+/**
+ * Inform the client that the union operation has failed,
+ * and proceed to destroy the evaluate operation.
+ *
+ * @param eo the union operation to fail
+ */
+static void
+fail_union_operation (struct UnionEvaluateOperation *eo)
+{
+  struct GNUNET_MQ_Message *mqm;
+  struct ResultMessage *msg;
+
+  mqm = GNUNET_MQ_msg (msg, GNUNET_MESSAGE_TYPE_SET_RESULT);
+  msg->result_status = htons (GNUNET_SET_STATUS_FAILURE);
+  msg->request_id = eo->request_id;
+  GNUNET_MQ_send (eo->set->client_mq, mqm);
+  destroy_union_operation (eo);
+}
+
+
+/**
+ * Derive the IBF key from a hash code and 
+ * a salt.
+ *
+ * @param src the hash code
+ * @param salt salt to use
+ * @return the derived IBF key
+ */
 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,
@@ -299,23 +391,27 @@
 }
 
 
+/**
+ * Send a request for the evaluate operation to a remote peer
+ *
+ * @param eo operation with the other peer
+ */
 static void
 send_operation_request (struct UnionEvaluateOperation *eo)
 {
   struct GNUNET_MQ_Message *mqm;
   struct OperationRequestMessage *msg;
-  int ret;
 
   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 */
-    _GSS_client_disconnect (eo->set->client);
-    GNUNET_MQ_discard (mqm);
-    GNUNET_break (0);
-    return;
-  }
+  if (NULL != eo->context_msg)
+    if (GNUNET_OK != GNUNET_MQ_nest (mqm, eo->context_msg, ntohs 
(eo->context_msg->size)))
+    {
+      /* the context message is too large */
+      _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);
@@ -369,8 +465,9 @@
   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);
+  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)
   {
@@ -386,8 +483,8 @@
 
 static int
 prepare_ibf_iterator (void *cls,
-                     uint32_t key,
-                     void *value)
+                      uint32_t key,
+                      void *value)
 {
   struct InvertibleBloomFilter *ibf = cls;
   struct KeyEntry *ke = value;
@@ -396,6 +493,7 @@
   return GNUNET_YES;
 }
 
+
 static int
 init_key_to_element_iterator (void *cls,
                               const struct GNUNET_HashCode *key,
@@ -404,10 +502,18 @@
   struct UnionEvaluateOperation *eo = cls;
   struct ElementEntry *e = value;
 
+  /* make sure that the element belongs to the set at the time
+   * of creating the operation */
+  if ( (e->generation_added > eo->generation_created) ||
+       ( (GNUNET_YES == e->removed) &&
+         (e->generation_removed < eo->generation_created)))
+    return GNUNET_YES;
+
   insert_element (eo, e);
   return GNUNET_YES;
 }
 
+
 static void
 prepare_ibf (struct UnionEvaluateOperation *eo, uint16_t size)
 {
@@ -422,14 +528,16 @@
   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);
+  GNUNET_CONTAINER_multihashmap32_iterate (eo->key_to_element,
+                                           prepare_ibf_iterator, 
eo->local_ibf);
 }
 
 
 /**
  * Send an ibf of appropriate size.
  *
- * @param cpi the peer
+ * @param eo the union operation
+ * @param ibf_order order of the ibf to send, size=2^order
  */
 static void
 send_ibf (struct UnionEvaluateOperation *eo, uint16_t ibf_order)
@@ -437,7 +545,7 @@
   unsigned int buckets_sent = 0;
   struct InvertibleBloomFilter *ibf;
 
-  prepare_ibf (eo, ibf_order);
+  prepare_ibf (eo, 1<<ibf_order);
 
   ibf = eo->local_ibf;
 
@@ -462,14 +570,14 @@
     GNUNET_MQ_send (eo->mq, mqm);
   }
 
-  eo->state = STATE_EXPECT_ELEMENTS_AND_REQUESTS;
+  eo->phase = PHASE_EXPECT_ELEMENTS_AND_REQUESTS;
 }
 
 
 /**
- * Send a strata estimator.
+ * Send a strata estimator to the remote peer.
  *
- * @param cpi the peer
+ * @param eo the union operation with the remote peer
  */
 static void
 send_strata_estimator (struct UnionEvaluateOperation *eo)
@@ -482,23 +590,35 @@
                                     GNUNET_MESSAGE_TYPE_SET_P2P_SE);
   strata_estimator_write (eo->set->state.u->se, &strata_msg[1]);
   GNUNET_MQ_send (eo->mq, mqm);
-  eo->state = STATE_EXPECT_IBF;
+  eo->phase = PHASE_EXPECT_IBF;
 }
 
+static unsigned int
+get_order_from_difference (unsigned int diff)
+{
+  unsigned int ibf_order;
 
+  ibf_order = 2;
+  while ((1<<ibf_order) < (2 * diff))
+    ibf_order++;
+  if (ibf_order > MAX_IBF_ORDER)
+    ibf_order = MAX_IBF_ORDER;
+  return ibf_order;
+}
+
+
 static void
 handle_p2p_strata_estimator (void *cls, const struct GNUNET_MessageHeader *mh)
 {
   struct UnionEvaluateOperation *eo = cls;
   struct StrataEstimator *remote_se;
-  int ibf_order;
   int diff;
 
   GNUNET_log (GNUNET_ERROR_TYPE_INFO, "got se\n");
 
-  if (eo->state != STATE_EXPECT_SE)
+  if (eo->phase != PHASE_EXPECT_SE)
   {
-    /* FIXME: handle */
+    fail_union_operation (eo);
     GNUNET_break (0);
     return;
   }
@@ -510,31 +630,77 @@
   strata_estimator_destroy (remote_se);
   strata_estimator_destroy (eo->se);
   eo->se = NULL;
-  /* minimum order */
-  ibf_order = 2;
-  while ((1<<ibf_order) < (2 * diff))
-    ibf_order++;
-  if (ibf_order > MAX_IBF_ORDER)
-    ibf_order = MAX_IBF_ORDER;
-  send_ibf (eo, ibf_order);
+  send_ibf (eo, get_order_from_difference (diff));
 }
 
 
+
+static int
+send_element_iterator (void *cls,
+                      uint32_t key,
+                      void *value)
+{
+  struct SendElementClosure *sec = cls;
+  struct IBF_Key ibf_key = sec->ibf_key;
+  struct UnionEvaluateOperation *eo = sec->eo;
+  struct KeyEntry *ke = value;
+
+  if (ke->ibf_key.key_val != ibf_key.key_val)
+    return GNUNET_YES;
+  while (NULL != ke)
+  {
+    const struct GNUNET_SET_Element *const element = &ke->element->element;
+    struct GNUNET_MQ_Message *mqm;
+
+    GNUNET_assert (ke->ibf_key.key_val == ibf_key.key_val);
+    mqm = GNUNET_MQ_msg_header (GNUNET_MESSAGE_TYPE_SET_P2P_ELEMENTS);
+    if (GNUNET_OK != GNUNET_MQ_nest (mqm, element->data, element->size))
+    {
+      GNUNET_break (0);
+      GNUNET_MQ_discard (mqm);
+      continue;
+    }
+    GNUNET_MQ_send (eo->mq, mqm);
+  }
+  return GNUNET_NO;
+}
+
 /**
- * FIXME
+ * Send all elements that have the specified IBF key
+ * to the remote peer of the union operation
  *
- * @param
+ * @param eo union operation
+ * @param ibf_key IBF key of interest
  */
 static void
-decode (struct UnionEvaluateOperation *eo)
+send_elements_for_key (struct UnionEvaluateOperation *eo, struct IBF_Key 
ibf_key)
 {
+  struct SendElementClosure send_cls;
+
+  send_cls.ibf_key = ibf_key;
+  send_cls.eo = eo;
+  GNUNET_CONTAINER_multihashmap32_get_multiple (eo->key_to_element, (uint32_t) 
ibf_key.key_val,
+                                                send_element_iterator, 
&send_cls);
+}
+
+
+
+/**
+ * Decode which elements are missing on each side, and
+ * send the appropriate elemens and requests
+ *
+ * @param eo union operation
+ */
+static void
+decode_and_send (struct UnionEvaluateOperation *eo)
+{
   struct IBF_Key key;
   int side;
   struct InvertibleBloomFilter *diff_ibf;
 
-  GNUNET_assert (STATE_EXPECT_ELEMENTS == eo->state);
+  GNUNET_assert (PHASE_EXPECT_ELEMENTS == eo->phase);
 
-  prepare_ibf (eo, eo->ibf_order);
+  prepare_ibf (eo, eo->remote_ibf->size);
   diff_ibf = ibf_dup (eo->local_ibf);
   ibf_subtract (diff_ibf, eo->remote_ibf);
 
@@ -545,14 +711,16 @@
     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 */
-      GNUNET_assert (0);
+      GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "decoding failed, sending larger 
ibf (size %u)\n",
+                  diff_ibf->size * 2);
+      send_ibf (eo, diff_ibf->size * 2);
+      ibf_destroy (diff_ibf);
       return;
     }
     if (GNUNET_NO == res)
     {
       struct GNUNET_MQ_Message *mqm;
+
       GNUNET_log (GNUNET_ERROR_TYPE_INFO, "transmitted all values, sending 
DONE\n");
       mqm = GNUNET_MQ_msg_header (GNUNET_MESSAGE_TYPE_SET_P2P_DONE);
       GNUNET_MQ_send (eo->mq, mqm);
@@ -560,11 +728,7 @@
     }
     if (1 == side)
     {
-      //struct ElementEntry *e;
-      /* we have the element(s), send it to the other peer */
-      //GNUNET_CONTAINER_multihashmap32_get_multiple 
(eo->set->state.u->elements,
-      //                                              (uint32_t) key.key_val);
-      /* FIXME */
+      send_elements_for_key (eo, key);
     }
     else
     {
@@ -573,7 +737,8 @@
 
       /* FIXME: before sending the request, check if we may just have the 
element */
       /* FIXME: merge multiple requests */
-      mqm = GNUNET_MQ_msg_header_extra (msg, sizeof (struct IBF_Key), 
GNUNET_MESSAGE_TYPE_SET_P2P_ELEMENT_REQUESTS);
+      mqm = GNUNET_MQ_msg_header_extra (msg, sizeof (struct IBF_Key),
+                                        
GNUNET_MESSAGE_TYPE_SET_P2P_ELEMENT_REQUESTS);
       *(struct IBF_Key *) &msg[1] = key;
       GNUNET_MQ_send (eo->mq, mqm);
     }
@@ -588,31 +753,25 @@
   struct IBFMessage *msg = (struct IBFMessage *) mh;
   unsigned int buckets_in_message;
 
-  if (eo->state == STATE_EXPECT_ELEMENTS_AND_REQUESTS)
+  if ( (eo->phase == PHASE_EXPECT_ELEMENTS_AND_REQUESTS) ||
+       (eo->phase == PHASE_EXPECT_IBF) )
   {
-    /* check that the ibf is a new one / first part */
-    /* clear outgoing messages */
-    GNUNET_assert (0);
-  }
-  else if (eo->state == STATE_EXPECT_IBF)
-  {
-    eo->state = STATE_EXPECT_IBF_CONT;
-    eo->ibf_order = msg->order;
+    eo->phase = PHASE_EXPECT_IBF_CONT;
     GNUNET_assert (NULL == eo->remote_ibf);
     eo->remote_ibf = ibf_create (1<<msg->order, SE_IBF_HASH_NUM);
-    if (ntohs (msg->offset) != 0)
+    if (0 != ntohs (msg->offset))
     {
-      /* FIXME: handle */
-      GNUNET_assert (0);
+      GNUNET_break (0);
+      fail_union_operation (eo);
     }
   }
-  else if (eo->state == STATE_EXPECT_IBF_CONT)
+  else if (eo->phase == PHASE_EXPECT_IBF_CONT)
   {
     if ( (ntohs (msg->offset) != eo->ibf_buckets_received) ||
-         (msg->order != eo->ibf_order) )
+         (1<<msg->order != eo->remote_ibf->size) )
     {
-      /* FIXME: handle */
-      GNUNET_assert (0);
+      GNUNET_break (0);
+      fail_union_operation (eo);
     }
   }
 
@@ -620,33 +779,69 @@
 
   if ((ntohs (msg->header.size) - sizeof *msg) != buckets_in_message * 
IBF_BUCKET_SIZE)
   {
-    /* FIXME: handle, message was malformed */
-    GNUNET_assert (0);
+    GNUNET_break (0);
+    fail_union_operation (eo);
   }
 
   ibf_read_slice (&msg[1], eo->ibf_buckets_received, buckets_in_message, 
eo->remote_ibf);
   eo->ibf_buckets_received += buckets_in_message;
 
-  if (eo->ibf_buckets_received == (1<<eo->ibf_order))
+  if (eo->ibf_buckets_received == eo->remote_ibf->size)
   {
-    eo->state = STATE_EXPECT_ELEMENTS;
-    decode (eo);
+    eo->phase = PHASE_EXPECT_ELEMENTS;
+    decode_and_send (eo);
   }
 }
 
 
+/**
+ * Send an element to the client of the operations's set.
+ *
+ * @param eo union operation
+ * @param element element to send
+ */
 static void
+send_client_element (struct UnionEvaluateOperation *eo,
+                     struct GNUNET_SET_Element *element)
+{
+  struct GNUNET_MQ_Message *mqm;
+  struct ResultMessage *rm;
+
+  GNUNET_assert (0 != eo->request_id);
+  mqm = GNUNET_MQ_msg (rm, GNUNET_MESSAGE_TYPE_SET_RESULT);
+  if (GNUNET_OK != GNUNET_MQ_nest (mqm, element->data, element->size))
+  {
+    GNUNET_MQ_discard (mqm);
+    GNUNET_break (0);
+    return;
+  }
+
+  GNUNET_MQ_send (eo->mq, mqm);
+}
+
+
+static void
 handle_p2p_elements (void *cls, const struct GNUNET_MessageHeader *mh)
 {
   struct UnionEvaluateOperation *eo = cls;
+  struct ElementEntry *ee;
+  uint16_t element_size;
 
-  if ( (eo->state != STATE_EXPECT_ELEMENTS) &&
-       (eo->state != STATE_EXPECT_ELEMENTS_AND_REQUESTS) )
+  if ( (eo->phase != PHASE_EXPECT_ELEMENTS) &&
+       (eo->phase != PHASE_EXPECT_ELEMENTS_AND_REQUESTS) )
   {
-    /* FIXME: handle */
+    fail_union_operation (eo);
     GNUNET_break (0);
     return;
   }
+  element_size = ntohs (mh->size) - sizeof (struct GNUNET_MessageHeader);
+  ee = GNUNET_malloc (sizeof *eo + element_size);
+  ee->element.data = &ee[1];
+  memcpy (ee->element.data, &mh[1], element_size);
+  ee->remote = GNUNET_YES;
+
+  insert_element (eo, ee);
+  send_client_element (eo, &ee->element);
 }
 
 
@@ -654,24 +849,64 @@
 handle_p2p_element_requests (void *cls, const struct GNUNET_MessageHeader *mh)
 {
   struct UnionEvaluateOperation *eo = cls;
+  struct IBF_Key *ibf_key;
+  unsigned int num_keys;
 
   /* look up elements and send them */
-  if (eo->state != STATE_EXPECT_ELEMENTS_AND_REQUESTS)
+  if (eo->phase != PHASE_EXPECT_ELEMENTS_AND_REQUESTS)
   {
-    /* FIXME: handle */
+    fail_union_operation (eo);
     GNUNET_break (0);
     return;
   }
+
+  num_keys = (ntohs (mh->size) - sizeof *mh) / sizeof (struct IBF_Key);
+
+  if ((ntohs (mh->size) - sizeof *mh) != num_keys * sizeof (struct IBF_Key))
+  {
+    fail_union_operation (eo);
+    GNUNET_break (0);
+    return;
+  }
+
+  ibf_key = (struct IBF_Key *) &mh[1];
+  while (0 != num_keys--)
+  {
+    send_elements_for_key (eo, *ibf_key);
+    ibf_key++;
+  }
 }
 
 
 static void
 handle_p2p_done (void *cls, const struct GNUNET_MessageHeader *mh)
 {
+  struct UnionEvaluateOperation *eo = cls;
+
+  if (eo->phase == PHASE_EXPECT_ELEMENTS_AND_REQUESTS)
+  {
+    /* we got all requests, but still have to send our elements as response */
+    struct GNUNET_MQ_Message *mqm;
+
+    eo->phase = PHASE_FINISHED;
+    mqm = GNUNET_MQ_msg_header (GNUNET_MESSAGE_TYPE_SET_P2P_DONE);
+    GNUNET_MQ_send (eo->mq, mqm);
+    return;
+  }
+  if (eo->phase == PHASE_EXPECT_ELEMENTS)
+  {
+    /* it's all over! */
+    eo->phase = PHASE_FINISHED;
+    return;
+  }
   GNUNET_break (0);
+  fail_union_operation (eo);
 }
 
 
+/**
+ * The handlers array, used for both evaluate and accept
+ */
 static const struct GNUNET_MQ_Handler union_handlers[] = {
   {handle_p2p_elements, GNUNET_MESSAGE_TYPE_SET_P2P_ELEMENTS},
   {handle_p2p_strata_estimator, GNUNET_MESSAGE_TYPE_SET_P2P_SE},
@@ -704,8 +939,9 @@
                                               union_handlers, eo);
   /* we started the operation, thus we have to send the operation request */
   send_operation_request (eo);
-  eo->state = STATE_EXPECT_SE;
+  eo->phase = PHASE_EXPECT_SE;
 }
+
        
 
 void
@@ -732,6 +968,7 @@
   struct UnionEvaluateOperation *eo;
 
   eo = GNUNET_new (struct UnionEvaluateOperation);
+  eo->generation_created = set->state.u->current_generation++;
   eo->set = set;
   eo->peer = incoming->peer;
   eo->app_id = incoming->app_id;
@@ -762,7 +999,6 @@
 }
 
 
-
 void
 _GSS_union_add (struct ElementMessage *m, struct Set *set)
 {
@@ -788,3 +1024,34 @@
   strata_estimator_insert (set->state.u->se, get_ibf_key (&ee->element_hash, 
0));
 }
 
+
+/**
+ * Remove the element given in the element message from the set.
+ * Only marks the element as removed, so that older set operations can still 
exchange it.
+ *
+ * @param m message with the element
+ * @param set set to remove the element from
+ */
+void
+_GSS_union_remove (struct ElementMessage *m, struct Set *set)
+{
+  struct GNUNET_HashCode hash;
+  struct ElementEntry *ee;
+
+  GNUNET_CRYPTO_hash (&m[1], ntohs (m->header.size), &hash);
+  
+  ee = GNUNET_CONTAINER_multihashmap_get (set->state.u->elements, &hash);
+  if (NULL == ee)
+  {
+    GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "client tried to remove 
non-existing element\n");
+    return;
+  }
+  if (GNUNET_YES == ee->removed)
+  {
+    GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "client tried to remove element 
twice\n");
+    return;
+  }
+  ee->removed = GNUNET_YES;
+  ee->generation_removed = set->state.u->current_generation;
+}
+

Modified: gnunet/src/set/gnunet-set-bug.c
===================================================================
--- gnunet/src/set/gnunet-set-bug.c     2013-04-29 09:09:01 UTC (rev 27003)
+++ gnunet/src/set/gnunet-set-bug.c     2013-04-30 00:04:25 UTC (rev 27004)
@@ -47,11 +47,21 @@
   return size;
 }
 
-static int listen_cb (void *cls,
-                      struct GNUNET_STREAM_Socket *socket,
-                      const struct 
-                      GNUNET_PeerIdentity *initiator)
+static int
+listen_cb (void *cls,
+           struct GNUNET_STREAM_Socket *socket,
+           const struct 
+           GNUNET_PeerIdentity *initiator)
 {
+  if (NULL == socket)
+  {
+    GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "socket listen failed\n");
+    return GNUNET_NO;
+  }
+
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "socket listen succesful\n");
+  GNUNET_assert (NULL != socket);
+  GNUNET_assert (0 == memcmp (initiator, &local_id, sizeof (*initiator)));
   GNUNET_STREAM_read (socket, GNUNET_TIME_UNIT_FOREVER_REL, 
                       stream_data_processor, NULL);
   return GNUNET_YES;
@@ -100,9 +110,9 @@
    static const struct GNUNET_GETOPT_CommandLineOption options[] = {
       GNUNET_GETOPT_OPTION_END
   };
-  GNUNET_PROGRAM_run2 (argc, argv, "gnunet-set",
+  GNUNET_PROGRAM_run (argc, argv, "gnunet-set",
                      "help",
-                     options, &run, NULL, GNUNET_NO);
+                     options, &run, NULL);
   return 0;
 }
 

Modified: gnunet/src/set/mq.c
===================================================================
--- gnunet/src/set/mq.c 2013-04-29 09:09:01 UTC (rev 27003)
+++ gnunet/src/set/mq.c 2013-04-30 00:04:25 UTC (rev 27004)
@@ -20,12 +20,15 @@
 
 /**
  * @author Florian Dold
- * @file mq/mq.c
+ * @file set/mq.c
  * @brief general purpose request queue
  */
 
 #include "mq.h"
 
+
+#define LOG(kind,...) GNUNET_log_from (kind, "mq",__VA_ARGS__)
+
 /**
  * Signature of functions implementing the
  * sending part of a message queue
@@ -96,6 +99,26 @@
   void *impl_state;
 
   /**
+   * Callback will be called when the message queue is empty
+   */
+  GNUNET_MQ_NotifyCallback empty_cb;
+
+  /**
+   * Closure for empty_cb
+   */
+  void *empty_cls;
+
+  /**
+   * Callback will be called when a read error occurs.
+   */
+  GNUNET_MQ_NotifyCallback read_error_cb;
+
+  /**
+   * Closure for read_error_cb
+   */
+  void *read_error_cls;
+
+  /**
    * Linked list of messages pending to be sent
    */
   struct GNUNET_MQ_Message *msg_head;
@@ -217,21 +240,21 @@
 
 int
 GNUNET_MQ_nest_ (struct GNUNET_MQ_Message **mqmp,
-                 const struct GNUNET_MessageHeader *m)
+                 const void *data, uint16_t len)
 {
   size_t new_size;
   size_t old_size;
 
-  if (NULL == m)
+  if (NULL == data)
     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))
+  if (ntohs ((*mqmp)->mh->size) + len < len)
     return GNUNET_SYSERR;
-  new_size = old_size + ntohs (m->size);
+  new_size = old_size + len;
   *mqmp = GNUNET_realloc (mqmp, sizeof (struct GNUNET_MQ_Message) + new_size);
-  memcpy ((*mqmp)->mh + old_size, m, new_size - old_size);
+  memcpy ((*mqmp)->mh + old_size, data, new_size - old_size);
   (*mqmp)->mh->size = htons (new_size);
   return GNUNET_OK;
 }
@@ -275,7 +298,11 @@
   mqm = mq->msg_head;
   mq->current_msg = mqm;
   if (NULL == mqm)
+  {
+    if (NULL != mq->empty_cb)
+      mq->empty_cb (mq->empty_cls);
     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,
@@ -415,6 +442,8 @@
                                              GNUNET_TIME_UNIT_FOREVER_REL,
                                              &transmit_queued, mq);
   }
+  else if (NULL != mq->empty_cb)
+    mq->empty_cb (mq->empty_cls);
   return msg_size;
 }
 
@@ -490,6 +519,8 @@
                                              GNUNET_TIME_UNIT_FOREVER_REL, 
GNUNET_NO,
                                              
&connection_client_transmit_queued, mq);
   }
+  else if (NULL != mq->empty_cb)
+    mq->empty_cb (mq->empty_cls);
   return msg_size;
 }
 
@@ -530,8 +561,14 @@
                        const struct GNUNET_MessageHeader *msg)
 {
   struct GNUNET_MQ_MessageQueue *mq = cls;
-
-  GNUNET_assert (NULL != msg);
+  
+  if (NULL == msg)
+  {
+    if (NULL == mq->read_error_cb)
+      LOG (GNUNET_ERROR_TYPE_WARNING, "ignoring read error (no handler 
installed)\n");
+    mq->read_error_cb (mq->read_error_cls);
+    return;
+  }
   dispatch_message (mq, msg);
 }
 
@@ -564,7 +601,6 @@
 }
 
 
-
 void
 GNUNET_MQ_replace_handlers (struct GNUNET_MQ_MessageQueue *mq,
                             const struct GNUNET_MQ_Handler *new_handlers,
@@ -575,13 +611,12 @@
 }
 
 
-
 /**
  * Associate the assoc_data in mq with a unique request id.
  *
  * @param mq message queue, id will be unique for the queue
  * @param mqm message to associate
- * @param data to associate
+ * @param assoc_data to associate
  */
 uint32_t
 GNUNET_MQ_assoc_add (struct GNUNET_MQ_MessageQueue *mq,
@@ -639,3 +674,20 @@
   GNUNET_free (mq);
 }
 
+
+/**
+ * Call a callback once all messages queued have been sent,
+ * i.e. the message queue is empty.
+ *
+ * @param mqm the message queue to send the notification for
+ * @param cb the callback to call on an empty queue
+ * @param cls closure for cb
+ */
+void
+GNUNET_MQ_notify_empty (struct GNUNET_MQ_MessageQueue *mqm,
+                        GNUNET_MQ_NotifyCallback cb,
+                        void *cls)
+{
+  mqm->empty_cb = cb;
+  mqm->empty_cls = cls;
+}

Modified: gnunet/src/set/mq.h
===================================================================
--- gnunet/src/set/mq.h 2013-04-29 09:09:01 UTC (rev 27003)
+++ gnunet/src/set/mq.h 2013-04-30 00:04:25 UTC (rev 27004)
@@ -20,7 +20,7 @@
 
 /**
  * @author Florian Dold
- * @file mq/mq.h
+ * @file set/mq.h
  * @brief general purpose request queue
  */
 #ifndef MQ_H
@@ -58,14 +58,40 @@
  */
 #define GNUNET_MQ_msg(mvar, type) GNUNET_MQ_msg_extra(mvar, 0, type)
 
-#define GNUNET_MQ_nest(mqm, mh) GNUNET_MQ_nest_ (&mqm, mh)
+/**
+ * Append data to the end of an existing MQ message.
+ * If the operation is successful, mqm is changed to point to the new MQ 
message,
+ * and GNUNET_OK is returned.
+ * On failure, GNUNET_SYSERR is returned, and the pointer mqm is not changed,
+ * the user of this API must take care of disposing the already allocated 
message
+ * (either by sending it, or by using GNUNET_MQ_discard)
+ *
+ * @param mqm MQ message to augment with additional data
+ * @param src source buffer for the additional data
+ * @param len length of the additional data
+ */
+#define GNUNET_MQ_nest(mqm, src, len) GNUNET_MQ_nest_ (&mqm, src, len)
 
+
+
 /**
+ * Append a message to the end of an existing MQ message.
+ * If the operation is successful, mqm is changed to point to the new MQ 
message,
+ * and GNUNET_OK is returned.
+ * On failure, GNUNET_SYSERR is returned, and the pointer mqm is not changed,
+ * the user of this API must take care of disposing the already allocated 
message
+ * (either by sending it, or by using GNUNET_MQ_discard)
+ *
+ * @param mqm MQ message to augment with additional data
+ * @param mh the message to append, must be of type 'struct 
GNUNET_MessageHeader *'
+ */
+#define GNUNET_MQ_nest_mh(mqm, mh) ((NULL == mh) ? (GNUNET_OK) : 
GNUNET_MQ_nest((mqm), (mh), ntohs ((mh)->size)))
+
+
+/**
  * Allocate a GNUNET_MQ_Message, where the message only consists of a header.
  * The allocated message will already have the type and size field set.
  *
- * @param mvar variable to store the allocated message in;
- *             must have a header field
  * @param type type of the message
  */
 #define GNUNET_MQ_msg_header(type) GNUNET_MQ_msg_ (NULL, sizeof (struct 
GNUNET_MessageHeader), type)
@@ -75,8 +101,7 @@
  * Allocate a GNUNET_MQ_Message, where the message only consists of a header 
and extra space.
  * The allocated message will already have the type and size field set.
  *
- * @param mvar variable to store the allocated message in;
- *             must have a header field
+ * @param mh pointer that will changed to point at to the allocated message 
header
  * @param esize extra space to allocate after the message header
  * @param type type of the message
  */
@@ -86,7 +111,7 @@
 /**
  * End-marker for the handlers array
  */
-#define GNUNET_MQ_HANDLERS_END {NULL, 0}
+#define GNUNET_MQ_HANDLERS_END {NULL, 0, 0}
 
 /**
  * Opaque handle to a message queue
@@ -120,9 +145,17 @@
 
 
   /**
-   * Type of the message we are interested in
+   * Type of the message this handler covers.
    */
   uint16_t type;
+
+  /**
+   * Expected size of messages of this type.  Use 0 for
+   * variable-size.  If non-zero, messages of the given
+   * type will be discarded (and the connection closed)
+   * if they do not have the right size.
+   */
+  uint16_t expected_size;
 };
 
 /**
@@ -146,7 +179,7 @@
 
 int
 GNUNET_MQ_nest_ (struct GNUNET_MQ_Message **mqmp,
-                 const struct GNUNET_MessageHeader *m);
+                 const void *src, uint16_t len);
 
 
 /**
@@ -186,7 +219,7 @@
  *
  * @param mq message queue, id will be unique for the queue
  * @param mqm message to associate
- * @param data to associate
+ * @param assoc_data to associate
  */
 uint32_t
 GNUNET_MQ_assoc_add (struct GNUNET_MQ_MessageQueue *mq,
@@ -234,7 +267,7 @@
 /**
  * Create a message queue for a GNUNET_STREAM_Socket.
  *
- * @param param client the client
+ * @param client the client
  * @return the message queue
  */
 struct GNUNET_MQ_MessageQueue *
@@ -286,8 +319,34 @@
                        GNUNET_MQ_NotifyCallback cb,
                        void *cls);
 
+/**
+ * Call a callback once all messages queued have been sent,
+ * i.e. the message queue is empty.
+ *
+ * @param mqm the message queue to send the notification for
+ * @param cb the callback to call on an empty queue
+ * @param cls closure for cb
+ */
+void
+GNUNET_MQ_notify_empty (struct GNUNET_MQ_MessageQueue *mqm,
+                        GNUNET_MQ_NotifyCallback cb,
+                        void *cls);
 
+
 /**
+ * Call a callback if reading encountered an error.
+ *
+ * @param mqm the message queue to send the notification for
+ * @param cb the callback to call on a read error
+ * @param cls closure for cb
+ */
+void
+GNUNET_MQ_notify_read_error (struct GNUNET_MQ_MessageQueue *mqm,
+                             GNUNET_MQ_NotifyCallback cb,
+                             void *cls);
+
+
+/**
  * Destroy the message queue.
  *
  * @param mq message queue to destroy

Modified: gnunet/src/set/set_api.c
===================================================================
--- gnunet/src/set/set_api.c    2013-04-29 09:09:01 UTC (rev 27003)
+++ gnunet/src/set/set_api.c    2013-04-30 00:04:25 UTC (rev 27004)
@@ -276,9 +276,17 @@
  * Evaluate a set operation with our set and the set of another peer.
  *
  * @param set set to use
+ * @param salt salt for HKDF (explain more here)
  * @param other_peer peer with the other set
  * @param app_id hash for the application using the set
  * @param context_msg additional information for the request
+ * @param salt salt used for the set operation; sometimes set operations
+ *        fail due to hash collisions, using a different salt for each 
operation
+ *        makes it harder for an attacker to exploit this
+ * @param timeout result_cb will be called with GNUNET_SET_STATUS_TIMEOUT
+ *        if the operation is not done after the specified time
+ * @param result_mode specified how results will be returned,
+ *        see 'GNUNET_SET_ResultMode'.
  * @param result_cb called on error or success
  * @param result_cls closure for result_cb
  * @return a handle to cancel the operation
@@ -307,10 +315,11 @@
   msg->request_id = htonl (GNUNET_MQ_assoc_add (set->mq, mqm, oh));
   msg->peer = *other_peer;
   msg->app_id = *app_id;
-
-  if (GNUNET_OK != GNUNET_MQ_nest (mqm, context_msg))
-    GNUNET_assert (0);
   
+  if (NULL != context_msg)
+    if (GNUNET_OK != GNUNET_MQ_nest (mqm, context_msg, ntohs 
(context_msg->size)))
+      GNUNET_assert (0);
+  
   oh->timeout_task = GNUNET_SCHEDULER_add_delayed (timeout, 
operation_timeout_task, oh);
   GNUNET_MQ_send (set->mq, mqm);
 
@@ -378,13 +387,16 @@
 
 
 /**
- * Accept a request we got via GNUNET_SET_listen
+ * Accept a request we got via GNUNET_SET_listen.
  *
  * @param request request to accept
  * @param set set used for the requested operation 
  * @param timeout timeout for the set operation
+ * @param result_mode specified how results will be returned,
+ *        see 'GNUNET_SET_ResultMode'.
  * @param result_cb callback for the results
- * @param cls closure for result_cb
+ * @param result_cls closure for result_cb
+ * @return a handle to cancel the operation
  */
 struct GNUNET_SET_OperationHandle *
 GNUNET_SET_accept (struct GNUNET_SET_Request *request,
@@ -421,18 +433,18 @@
 /**
  * Cancel the given set operation.
  *
- * @param op set operation to cancel
+ * @param oh set operation to cancel
  */
 void
-GNUNET_SET_operation_cancel (struct GNUNET_SET_OperationHandle *h)
+GNUNET_SET_operation_cancel (struct GNUNET_SET_OperationHandle *oh)
 {
   struct GNUNET_MQ_Message *mqm;
   struct GNUNET_SET_OperationHandle *h_assoc;
 
-  h_assoc = GNUNET_MQ_assoc_remove (h->set->mq, h->request_id);
-  GNUNET_assert (h_assoc == h);
+  h_assoc = GNUNET_MQ_assoc_remove (oh->set->mq, oh->request_id);
+  GNUNET_assert (h_assoc == oh);
   mqm = GNUNET_MQ_msg_header (GNUNET_MESSAGE_TYPE_SET_CANCEL);
-  GNUNET_MQ_send (h->set->mq, mqm);
-  GNUNET_free (h);
+  GNUNET_MQ_send (oh->set->mq, mqm);
+  GNUNET_free (oh);
 }
 




reply via email to

[Prev in Thread] Current Thread [Next in Thread]