src/set/gnunet-service-set_union.c

   1 /*
   2       This file is part of GNUnet
   3       (C) 2013 Christian Grothoff (and other contributing authors)
   4
   5       GNUnet is free software; you can redistribute it and/or modify
   6       it under the terms of the GNU General Public License as published
   7       by the Free Software Foundation; either version 3, or (at your
   8       option) any later version.
   9
  10       GNUnet is distributed in the hope that it will be useful, but
  11       WITHOUT ANY WARRANTY; without even the implied warranty of
  12       MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13       General Public License for more details.
  14
  15       You should have received a copy of the GNU General Public License
  16       along with GNUnet; see the file COPYING.  If not, write to the
  17       Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  18       Boston, MA 02111-1307, USA.
  19 */
  20
  21 /**
  22  * @file set/gnunet-service-set.c
  23  * @brief two-peer set operations
  24  * @author Florian Dold
  25  */
  26 #include "platform.h"
  27 #include "gnunet_util_lib.h"
  28 #include "gnunet-service-set.h"
  29 #include "ibf.h"
  30 #include "strata_estimator.h"
  31 #include "set_protocol.h"
  32 #include <gcrypt.h>
  33
  34
  35 /**
  36  * Number of IBFs in a strata estimator.
  37  */
  38 #define SE_STRATA_COUNT 32
  39 /**
  40  * Size of the IBFs in the strata estimator.
  41  */
  42 #define SE_IBF_SIZE 80
  43 /**
  44  * hash num parameter for the difference digests and strata estimators
  45  */
  46 #define SE_IBF_HASH_NUM 4
  47
  48 /**
  49  * Number of buckets that can be transmitted in one message.
  50  */
  51 #define MAX_BUCKETS_PER_MESSAGE ((1<<15) / IBF_BUCKET_SIZE)
  52
  53 /**
  54  * The maximum size of an ibf we use is 2^(MAX_IBF_ORDER).
  55  * Choose this value so that computing the IBF is still cheaper
  56  * than transmitting all values.
  57  */
  58 #define MAX_IBF_ORDER (16)
  59
  60 /**
  61  * Number of buckets used in the ibf per estimated
  62  * difference.
  63  */
  64 #define IBF_ALPHA 4
  65
  66
  67 /**
  68  * Current phase we are in for a union operation.
  69  */
  70 enum UnionOperationPhase
  71 {
  72   /**
  73    * We sent the request message, and expect a strata estimator
  74    */
  75   PHASE_EXPECT_SE,
  76   /**
  77    * We sent the strata estimator, and expect an IBF
  78    */
  79   PHASE_EXPECT_IBF,
  80   /**
  81    * We know what type of IBF the other peer wants to send us,
  82    * and expect the remaining parts
  83    */
  84   PHASE_EXPECT_IBF_CONT,
  85   /**
  86    * We are sending request and elements,
  87    * and thus only expect elements from the other peer.
  88    */
  89   PHASE_EXPECT_ELEMENTS,
  90   /**
  91    * We are expecting elements and requests, and send
  92    * requested elements back to the other peer.
  93    */
  94   PHASE_EXPECT_ELEMENTS_AND_REQUESTS,
  95   /**
  96    * The protocol is over.
  97    * Results may still have to be sent to the client.
  98    */
  99   PHASE_FINISHED
 100 };
 101
 102
 103 /**
 104  * State of an evaluate operation
 105  * with another peer.
 106  */
 107 struct OperationState
 108 {
 109   /**
 110    * Tunnel to the remote peer.
 111    */
 112   struct GNUNET_MESH_Tunnel *tunnel;
 113
 114   /**
 115    * Message queue for the peer.
 116    */
 117   struct GNUNET_MQ_Handle *mq;
 118
 119   /**
 120    * Number of ibf buckets received
 121    */
 122   unsigned int ibf_buckets_received;
 123
 124   /**
 125    * Copy of the set's strata estimator at the time of
 126    * creation of this operation
 127    */
 128   struct StrataEstimator *se;
 129
 130   /**
 131    * The ibf we currently receive
 132    */
 133   struct InvertibleBloomFilter *remote_ibf;
 134
 135   /**
 136    * IBF of the set's element.
 137    */
 138   struct InvertibleBloomFilter *local_ibf;
 139
 140   /**
 141    * Maps IBF-Keys (specific to the current salt) to elements.
 142    * Used as a multihashmap, the keys being the lower 32bit of the IBF-Key.
 143    * Colliding IBF-Keys are linked.
 144    */
 145   struct GNUNET_CONTAINER_MultiHashMap32 *key_to_element;
 146
 147   /**
 148    * Iterator for sending elements on the key to element mapping to the client.
 149    */
 150   struct GNUNET_CONTAINER_MultiHashMap32Iterator *full_result_iter;
 151
 152   /**
 153    * Current state of the operation.
 154    */
 155   enum UnionOperationPhase phase;
 156
 157   /**
 158    * Did we send the client that we are done?
 159    */
 160   int client_done_sent;
 161 };
 162
 163
 164 /**
 165  * The key entry is used to associate an ibf key with
 166  * an element.
 167  */
 168 struct KeyEntry
 169 {
 170   /**
 171    * IBF key for the entry, derived from the current salt.
 172    */
 173   struct IBF_Key ibf_key;
 174
 175   /**
 176    * The actual element associated with the key.
 177    */
 178   struct ElementEntry *element;
 179
 180   /**
 181    * Element that collides with this element
 182    * on the ibf key. All colliding entries must have the same ibf key.
 183    */
 184   struct KeyEntry *next_colliding;
 185 };
 186
 187
 188 /**
 189  * Used as a closure for sending elements
 190  * with a specific IBF key.
 191  */
 192 struct SendElementClosure
 193 {
 194   /**
 195    * The IBF key whose matching elements should be
 196    * sent.
 197    */
 198   struct IBF_Key ibf_key;
 199
 200   /**
 201    * Operation for which the elements
 202    * should be sent.
 203    */
 204   struct Operation *op;
 205 };
 206
 207
 208 /**
 209  * Extra state required for efficient set union.
 210  */
 211 struct SetState
 212 {
 213   /**
 214    * The strata estimator is only generated once for
 215    * each set.
 216    * The IBF keys are derived from the element hashes with
 217    * salt=0.
 218    */
 219   struct StrataEstimator *se;
 220 };
 221
 222
 223 /**
 224  * Iterator over hash map entries.
 225  *
 226  * @param cls closure
 227  * @param key current key code
 228  * @param value value in the hash map
 229  * @return #GNUNET_YES if we should continue to
 230  *         iterate,
 231  *         #GNUNET_NO if not.
 232  */
 233 static int
 234 destroy_key_to_element_iter (void *cls,
 235                              uint32_t key,
 236                              void *value)
 237 {
 238   struct KeyEntry *k = value;
 239   /* destroy the linked list of colliding ibf key entries */
 240   while (NULL != k)
 241   {
 242     struct KeyEntry *k_tmp = k;
 243     k = k->next_colliding;
 244     if (GNUNET_YES == k_tmp->element->remote)
 245     {
 246       GNUNET_free (k_tmp->element);
 247       k_tmp->element = NULL;
 248     }
 249     GNUNET_free (k_tmp);
 250   }
 251   return GNUNET_YES;
 252 }
 253
 254
 255 /**
 256  * Destroy the union operation.  Only things specific to the union operation are destroyed.
 257  *
 258  * @param op union operation to destroy
 259  */
 260 static void
 261 union_op_cancel (struct Operation *op)
 262 {
 263   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "destroying union op\n");
 264   /* check if the op was canceled twice */
 265   GNUNET_assert (NULL != op->state);
 266   if (NULL != op->state->remote_ibf)
 267   {
 268     ibf_destroy (op->state->remote_ibf);
 269     op->state->remote_ibf = NULL;
 270   }
 271   if (NULL != op->state->local_ibf)
 272   {
 273     ibf_destroy (op->state->local_ibf);
 274     op->state->local_ibf = NULL;
 275   }
 276   if (NULL != op->state->se)
 277   {
 278     strata_estimator_destroy (op->state->se);
 279     op->state->se = NULL;
 280   }
 281   if (NULL != op->state->key_to_element)
 282   {
 283     GNUNET_CONTAINER_multihashmap32_iterate (op->state->key_to_element, destroy_key_to_element_iter, NULL);
 284     GNUNET_CONTAINER_multihashmap32_destroy (op->state->key_to_element);
 285     op->state->key_to_element = NULL;
 286   }
 287   GNUNET_free (op->state);
 288   op->state = NULL;
 289   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "destroying union op done\n");
 290 }
 291
 292
 293 /**
 294  * Inform the client that the union operation has failed,
 295  * and proceed to destroy the evaluate operation.
 296  *
 297  * @param op the union operation to fail
 298  */
 299 static void
 300 fail_union_operation (struct Operation *op)
 301 {
 302   struct GNUNET_MQ_Envelope *ev;
 303   struct GNUNET_SET_ResultMessage *msg;
 304
 305   GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "union operation failed\n");
 306
 307   ev = GNUNET_MQ_msg (msg, GNUNET_MESSAGE_TYPE_SET_RESULT);
 308   msg->result_status = htons (GNUNET_SET_STATUS_FAILURE);
 309   msg->request_id = htonl (op->spec->client_request_id);
 310   msg->element_type = htons (0);
 311   GNUNET_MQ_send (op->spec->set->client_mq, ev);
 312   _GSS_operation_destroy (op);
 313 }
 314
 315
 316 /**
 317  * Derive the IBF key from a hash code and
 318  * a salt.
 319  *
 320  * @param src the hash code
 321  * @param salt salt to use
 322  * @return the derived IBF key
 323  */
 324 static struct IBF_Key
 325 get_ibf_key (const struct GNUNET_HashCode *src, uint16_t salt)
 326 {
 327   struct IBF_Key key;
 328
 329   GNUNET_CRYPTO_hkdf (&key, sizeof (key),
 330                       GCRY_MD_SHA512, GCRY_MD_SHA256,
 331                       src, sizeof *src,
 332                       &salt, sizeof (salt),
 333                       NULL, 0);
 334   return key;
 335 }
 336
 337
 338 /**
 339  * Send a request for the evaluate operation to a remote peer
 340  *
 341  * @param op operation with the other peer
 342  */
 343 static void
 344 send_operation_request (struct Operation *op)
 345 {
 346   struct GNUNET_MQ_Envelope *ev;
 347   struct OperationRequestMessage *msg;
 348
 349   ev = GNUNET_MQ_msg_nested_mh (msg, GNUNET_MESSAGE_TYPE_SET_P2P_OPERATION_REQUEST,
 350                                 op->spec->context_msg);
 351
 352   if (NULL == ev)
 353   {
 354     /* the context message is too large */
 355     GNUNET_break (0);
 356     GNUNET_SERVER_client_disconnect (op->spec->set->client);
 357     return;
 358   }
 359   msg->operation = htonl (GNUNET_SET_OPERATION_UNION);
 360   msg->app_id = op->spec->app_id;
 361   msg->salt = htonl (op->spec->salt);
 362   GNUNET_MQ_send (op->mq, ev);
 363
 364   if (NULL != op->spec->context_msg)
 365     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "sent op request with context message\n");
 366   else
 367     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "sent op request without context message\n");
 368
 369   if (NULL != op->spec->context_msg)
 370   {
 371     GNUNET_free (op->spec->context_msg);
 372     op->spec->context_msg = NULL;
 373   }
 374 }
 375
 376
 377 /**
 378  * Iterator to create the mapping between ibf keys
 379  * and element entries.
 380  *
 381  * @param cls closure
 382  * @param key current key code
 383  * @param value value in the hash map
 384  * @return #GNUNET_YES if we should continue to
 385  *         iterate,
 386  *         #GNUNET_NO if not.
 387  */
 388 static int
 389 op_register_element_iterator (void *cls,
 390                               uint32_t key,
 391                               void *value)
 392 {
 393   struct KeyEntry *const new_k = cls;
 394   struct KeyEntry *old_k = value;
 395
 396   GNUNET_assert (NULL != old_k);
 397   /* check if our ibf key collides with the ibf key in the existing entry */
 398   if (old_k->ibf_key.key_val == new_k->ibf_key.key_val)
 399   {
 400     /* insert the the new key in the collision chain */
 401     new_k->next_colliding = old_k->next_colliding;
 402     old_k->next_colliding = new_k;
 403     /* signal to the caller that we were able to insert into a colliding bucket */
 404     return GNUNET_NO;
 405   }
 406   return GNUNET_YES;
 407 }
 408
 409
 410 /**
 411  * Iterator to create the mapping between ibf keys
 412  * and element entries.
 413  *
 414  * @param cls closure
 415  * @param key current key code
 416  * @param value value in the hash map
 417  * @return #GNUNET_YES if we should continue to
 418  *         iterate,
 419  *         #GNUNET_NO if not.
 420  */
 421 static int
 422 op_has_element_iterator (void *cls,
 423                          uint32_t key,
 424                          void *value)
 425 {
 426   struct GNUNET_HashCode *element_hash = cls;
 427   struct KeyEntry *k = value;
 428
 429   GNUNET_assert (NULL != k);
 430   while (NULL != k)
 431   {
 432     if (0 == GNUNET_CRYPTO_hash_cmp (&k->element->element_hash, element_hash))
 433       return GNUNET_NO;
 434     k = k->next_colliding;
 435   }
 436   return GNUNET_YES;
 437 }
 438
 439
 440 /**
 441  * Determine whether the given element is already in the operation's element
 442  * set.
 443  *
 444  * @param op operation that should be tested for 'element_hash'
 445  * @param element_hash hash of the element to look for
 446  * @return #GNUNET_YES if the element has been found, #GNUNET_NO otherwise
 447  */
 448 static int
 449 op_has_element (struct Operation *op, const struct GNUNET_HashCode *element_hash)
 450 {
 451   int ret;
 452   struct IBF_Key ibf_key;
 453
 454   ibf_key = get_ibf_key (element_hash, op->spec->salt);
 455   ret = GNUNET_CONTAINER_multihashmap32_get_multiple (op->state->key_to_element,
 456                                                       (uint32_t) ibf_key.key_val,
 457                                                       op_has_element_iterator, (void *) element_hash);
 458
 459   /* was the iteration aborted because we found the element? */
 460   if (GNUNET_SYSERR == ret)
 461     return GNUNET_YES;
 462   return GNUNET_NO;
 463 }
 464
 465
 466 /**
 467  * Insert an element into the union operation's
 468  * key-to-element mapping. Takes ownership of 'ee'.
 469  * Note that this does not insert the element in the set,
 470  * only in the operation's key-element mapping.
 471  * This is done to speed up re-tried operations, if some elements
 472  * were transmitted, and then the IBF fails to decode.
 473  *
 474  * @param op the union operation
 475  * @param ee the element entry
 476  */
 477 static void
 478 op_register_element (struct Operation *op, struct ElementEntry *ee)
 479 {
 480   int ret;
 481   struct IBF_Key ibf_key;
 482   struct KeyEntry *k;
 483
 484   ibf_key = get_ibf_key (&ee->element_hash, op->spec->salt);
 485   k = GNUNET_new (struct KeyEntry);
 486   k->element = ee;
 487   k->ibf_key = ibf_key;
 488   ret = GNUNET_CONTAINER_multihashmap32_get_multiple (op->state->key_to_element,
 489                                                       (uint32_t) ibf_key.key_val,
 490                                                       op_register_element_iterator, k);
 491
 492   /* was the element inserted into a colliding bucket? */
 493   if (GNUNET_SYSERR == ret)
 494     return;
 495
 496   GNUNET_CONTAINER_multihashmap32_put (op->state->key_to_element, (uint32_t) ibf_key.key_val, k,
 497                                        GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
 498 }
 499
 500
 501 /**
 502  * Insert a key into an ibf.
 503  *
 504  * @param cls the ibf
 505  * @param key unused
 506  * @param value the key entry to get the key from
 507  */
 508 static int
 509 prepare_ibf_iterator (void *cls,
 510                       uint32_t key,
 511                       void *value)
 512 {
 513   struct InvertibleBloomFilter *ibf = cls;
 514   struct KeyEntry *ke = value;
 515
 516   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "inserting %x into ibf\n", ke->ibf_key.key_val);
 517
 518   ibf_insert (ibf, ke->ibf_key);
 519   return GNUNET_YES;
 520 }
 521
 522
 523 /**
 524  * Iterator for initializing the
 525  * key-to-element mapping of a union operation
 526  *
 527  * @param cls the union operation
 528  * @param key unised
 529  * @param value the element entry to insert
 530  *        into the key-to-element mapping
 531  * @return GNUNET_YES to continue iterating,
 532  *         GNUNET_NO to stop
 533  */
 534 static int
 535 init_key_to_element_iterator (void *cls,
 536                               const struct GNUNET_HashCode *key,
 537                               void *value)
 538 {
 539   struct Operation *op = cls;
 540   struct ElementEntry *e = value;
 541
 542   /* make sure that the element belongs to the set at the time
 543    * of creating the operation */
 544   if ( (e->generation_added > op->generation_created) ||
 545        ( (GNUNET_YES == e->removed) &&
 546          (e->generation_removed < op->generation_created)))
 547     return GNUNET_YES;
 548
 549   GNUNET_assert (GNUNET_NO == e->remote);
 550
 551   op_register_element (op, e);
 552   return GNUNET_YES;
 553 }
 554
 555
 556 /**
 557  * Create an ibf with the operation's elements
 558  * of the specified size
 559  *
 560  * @param op the union operation
 561  * @param size size of the ibf to create
 562  */
 563 static void
 564 prepare_ibf (struct Operation *op, uint16_t size)
 565 {
 566   if (NULL == op->state->key_to_element)
 567   {
 568     unsigned int len;
 569     len = GNUNET_CONTAINER_multihashmap_size (op->spec->set->elements);
 570     op->state->key_to_element = GNUNET_CONTAINER_multihashmap32_create (len + 1);
 571     GNUNET_CONTAINER_multihashmap_iterate (op->spec->set->elements,
 572                                            init_key_to_element_iterator, op);
 573   }
 574   if (NULL != op->state->local_ibf)
 575     ibf_destroy (op->state->local_ibf);
 576   op->state->local_ibf = ibf_create (size, SE_IBF_HASH_NUM);
 577   GNUNET_CONTAINER_multihashmap32_iterate (op->state->key_to_element,
 578                                            prepare_ibf_iterator, op->state->local_ibf);
 579 }
 580
 581
 582 /**
 583  * Send an ibf of appropriate size.
 584  *
 585  * @param op the union operation
 586  * @param ibf_order order of the ibf to send, size=2^order
 587  */
 588 static void
 589 send_ibf (struct Operation *op, uint16_t ibf_order)
 590 {
 591   unsigned int buckets_sent = 0;
 592   struct InvertibleBloomFilter *ibf;
 593
 594   prepare_ibf (op, 1<<ibf_order);
 595
 596   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "sending ibf of size %u\n", 1<<ibf_order);
 597
 598   ibf = op->state->local_ibf;
 599
 600   while (buckets_sent < (1 << ibf_order))
 601   {
 602     unsigned int buckets_in_message;
 603     struct GNUNET_MQ_Envelope *ev;
 604     struct IBFMessage *msg;
 605
 606     buckets_in_message = (1 << ibf_order) - buckets_sent;
 607     /* limit to maximum */
 608     if (buckets_in_message > MAX_BUCKETS_PER_MESSAGE)
 609       buckets_in_message = MAX_BUCKETS_PER_MESSAGE;
 610
 611     ev = GNUNET_MQ_msg_extra (msg, buckets_in_message * IBF_BUCKET_SIZE,
 612                                GNUNET_MESSAGE_TYPE_SET_UNION_P2P_IBF);
 613     msg->reserved = 0;
 614     msg->order = ibf_order;
 615     msg->offset = htons (buckets_sent);
 616     ibf_write_slice (ibf, buckets_sent,
 617                      buckets_in_message, &msg[1]);
 618     buckets_sent += buckets_in_message;
 619     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "ibf chunk size %u, %u/%u sent\n",
 620                 buckets_in_message, buckets_sent, 1<<ibf_order);
 621     GNUNET_MQ_send (op->mq, ev);
 622   }
 623
 624   op->state->phase = PHASE_EXPECT_ELEMENTS_AND_REQUESTS;
 625 }
 626
 627
 628 /**
 629  * Send a strata estimator to the remote peer.
 630  *
 631  * @param op the union operation with the remote peer
 632  */
 633 static void
 634 send_strata_estimator (struct Operation *op)
 635 {
 636   struct GNUNET_MQ_Envelope *ev;
 637   struct GNUNET_MessageHeader *strata_msg;
 638
 639   ev = GNUNET_MQ_msg_header_extra (strata_msg,
 640                                    SE_STRATA_COUNT * IBF_BUCKET_SIZE * SE_IBF_SIZE,
 641                                    GNUNET_MESSAGE_TYPE_SET_UNION_P2P_SE);
 642   strata_estimator_write (op->state->se, &strata_msg[1]);
 643   GNUNET_MQ_send (op->mq, ev);
 644   op->state->phase = PHASE_EXPECT_IBF;
 645   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "sent SE, expecting IBF\n");
 646 }
 647
 648
 649 /**
 650  * Compute the necessary order of an ibf
 651  * from the size of the symmetric set difference.
 652  *
 653  * @param diff the difference
 654  * @return the required size of the ibf
 655  */
 656 static unsigned int
 657 get_order_from_difference (unsigned int diff)
 658 {
 659   unsigned int ibf_order;
 660
 661   ibf_order = 2;
 662   while ((1<<ibf_order) < (IBF_ALPHA * diff) || (1<<ibf_order) < SE_IBF_HASH_NUM)
 663     ibf_order++;
 664   if (ibf_order > MAX_IBF_ORDER)
 665     ibf_order = MAX_IBF_ORDER;
 666   return ibf_order;
 667 }
 668
 669
 670 /**
 671  * Handle a strata estimator from a remote peer
 672  *
 673  * @param cls the union operation
 674  * @param mh the message
 675  */
 676 static void
 677 handle_p2p_strata_estimator (void *cls, const struct GNUNET_MessageHeader *mh)
 678 {
 679   struct Operation *op = cls;
 680   struct StrataEstimator *remote_se;
 681   int diff;
 682
 683   if (op->state->phase != PHASE_EXPECT_SE)
 684   {
 685     fail_union_operation (op);
 686     GNUNET_break (0);
 687     return;
 688   }
 689   remote_se = strata_estimator_create (SE_STRATA_COUNT, SE_IBF_SIZE,
 690                                        SE_IBF_HASH_NUM);
 691   strata_estimator_read (&mh[1], remote_se);
 692   GNUNET_assert (NULL != op->state->se);
 693   diff = strata_estimator_difference (remote_se, op->state->se);
 694   strata_estimator_destroy (remote_se);
 695   strata_estimator_destroy (op->state->se);
 696   op->state->se = NULL;
 697   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "got se diff=%d, using ibf size %d\n",
 698               diff, 1<<get_order_from_difference (diff));
 699   send_ibf (op, get_order_from_difference (diff));
 700 }
 701
 702
 703
 704 /**
 705  * Iterator to send elements to a remote peer
 706  *
 707  * @param cls closure with the element key and the union operation
 708  * @param key ignored
 709  * @param value the key entry
 710  */
 711 static int
 712 send_element_iterator (void *cls,
 713                        uint32_t key,
 714                        void *value)
 715 {
 716   struct SendElementClosure *sec = cls;
 717   struct IBF_Key ibf_key = sec->ibf_key;
 718   struct Operation *op = sec->op;
 719   struct KeyEntry *ke = value;
 720
 721   if (ke->ibf_key.key_val != ibf_key.key_val)
 722     return GNUNET_YES;
 723   while (NULL != ke)
 724   {
 725     const struct GNUNET_SET_Element *const element = &ke->element->element;
 726     struct GNUNET_MQ_Envelope *ev;
 727     struct GNUNET_MessageHeader *mh;
 728
 729     GNUNET_assert (ke->ibf_key.key_val == ibf_key.key_val);
 730     ev = GNUNET_MQ_msg_header_extra (mh, element->size, GNUNET_MESSAGE_TYPE_SET_P2P_ELEMENTS);
 731     if (NULL == ev)
 732     {
 733       /* element too large */
 734       GNUNET_break (0);
 735       continue;
 736     }
 737     memcpy (&mh[1], element->data, element->size);
 738     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "sending element (%s) to peer\n",
 739                 GNUNET_h2s (&ke->element->element_hash));
 740     GNUNET_MQ_send (op->mq, ev);
 741     ke = ke->next_colliding;
 742   }
 743   return GNUNET_NO;
 744 }
 745
 746 /**
 747  * Send all elements that have the specified IBF key
 748  * to the remote peer of the union operation
 749  *
 750  * @param op union operation
 751  * @param ibf_key IBF key of interest
 752  */
 753 static void
 754 send_elements_for_key (struct Operation *op, struct IBF_Key ibf_key)
 755 {
 756   struct SendElementClosure send_cls;
 757
 758   send_cls.ibf_key = ibf_key;
 759   send_cls.op = op;
 760   GNUNET_CONTAINER_multihashmap32_get_multiple (op->state->key_to_element, (uint32_t) ibf_key.key_val,
 761                                                 &send_element_iterator, &send_cls);
 762 }
 763
 764
 765 /**
 766  * Decode which elements are missing on each side, and
 767  * send the appropriate elemens and requests
 768  *
 769  * @param op union operation
 770  */
 771 static void
 772 decode_and_send (struct Operation *op)
 773 {
 774   struct IBF_Key key;
 775   struct IBF_Key last_key;
 776   int side;
 777   unsigned int num_decoded;
 778   struct InvertibleBloomFilter *diff_ibf;
 779
 780   GNUNET_assert (PHASE_EXPECT_ELEMENTS == op->state->phase);
 781
 782   prepare_ibf (op, op->state->remote_ibf->size);
 783   diff_ibf = ibf_dup (op->state->local_ibf);
 784   ibf_subtract (diff_ibf, op->state->remote_ibf);
 785
 786   ibf_destroy (op->state->remote_ibf);
 787   op->state->remote_ibf = NULL;
 788
 789   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "decoding IBF (size=%u)\n", diff_ibf->size);
 790
 791   num_decoded = 0;
 792   last_key.key_val = 0;
 793
 794   while (1)
 795   {
 796     int res;
 797     int cycle_detected = GNUNET_NO;
 798
 799     last_key = key;
 800
 801     res = ibf_decode (diff_ibf, &side, &key);
 802     if (res == GNUNET_OK)
 803     {
 804       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "decoded ibf key %lx\n",
 805                   key.key_val);
 806       num_decoded += 1;
 807       if (num_decoded > diff_ibf->size || (num_decoded > 1 && last_key.key_val == key.key_val))
 808       {
 809         GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "detected cyclic ibf (decoded %u/%u)\n",
 810                     num_decoded, diff_ibf->size);
 811         cycle_detected = GNUNET_YES;
 812       }
 813     }
 814     if ((GNUNET_SYSERR == res) || (GNUNET_YES == cycle_detected))
 815     {
 816       int next_order;
 817       next_order = 0;
 818       while (1<<next_order < diff_ibf->size)
 819         next_order++;
 820       next_order++;
 821       if (next_order <= MAX_IBF_ORDER)
 822       {
 823         GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 824                     "decoding failed, sending larger ibf (size %u)\n",
 825                     1<<next_order);
 826         send_ibf (op, next_order);
 827       }
 828       else
 829       {
 830         GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 831                     "set union failed: reached ibf limit\n");
 832       }
 833       break;
 834     }
 835     if (GNUNET_NO == res)
 836     {
 837       struct GNUNET_MQ_Envelope *ev;
 838
 839       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "transmitted all values, sending DONE\n");
 840       ev = GNUNET_MQ_msg_header (GNUNET_MESSAGE_TYPE_SET_P2P_DONE);
 841       GNUNET_MQ_send (op->mq, ev);
 842       break;
 843     }
 844     if (1 == side)
 845     {
 846       send_elements_for_key (op, key);
 847     }
 848     else if (-1 == side)
 849     {
 850       struct GNUNET_MQ_Envelope *ev;
 851       struct GNUNET_MessageHeader *msg;
 852
 853       /* It may be nice to merge multiple requests, but with mesh's corking it is not worth
 854        * the effort additional complexity. */
 855       ev = GNUNET_MQ_msg_header_extra (msg, sizeof (struct IBF_Key),
 856                                         GNUNET_MESSAGE_TYPE_SET_P2P_ELEMENT_REQUESTS);
 857
 858       *(struct IBF_Key *) &msg[1] = key;
 859       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "sending element request\n");
 860       GNUNET_MQ_send (op->mq, ev);
 861     }
 862     else
 863     {
 864       GNUNET_assert (0);
 865     }
 866   }
 867   ibf_destroy (diff_ibf);
 868 }
 869
 870
 871 /**
 872  * Handle an IBF message from a remote peer.
 873  *
 874  * @param cls the union operation
 875  * @param mh the header of the message
 876  */
 877 static void
 878 handle_p2p_ibf (void *cls, const struct GNUNET_MessageHeader *mh)
 879 {
 880   struct Operation *op = cls;
 881   struct IBFMessage *msg = (struct IBFMessage *) mh;
 882   unsigned int buckets_in_message;
 883
 884   if ( (op->state->phase == PHASE_EXPECT_ELEMENTS_AND_REQUESTS) ||
 885        (op->state->phase == PHASE_EXPECT_IBF) )
 886   {
 887     op->state->phase = PHASE_EXPECT_IBF_CONT;
 888     GNUNET_assert (NULL == op->state->remote_ibf);
 889     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "creating new ibf of size %u\n", 1<<msg->order);
 890     op->state->remote_ibf = ibf_create (1<<msg->order, SE_IBF_HASH_NUM);
 891     op->state->ibf_buckets_received = 0;
 892     if (0 != ntohs (msg->offset))
 893     {
 894       GNUNET_break (0);
 895       fail_union_operation (op);
 896       return;
 897     }
 898   }
 899   else if (op->state->phase == PHASE_EXPECT_IBF_CONT)
 900   {
 901     if ( (ntohs (msg->offset) != op->state->ibf_buckets_received) ||
 902          (1<<msg->order != op->state->remote_ibf->size) )
 903     {
 904       GNUNET_break (0);
 905       fail_union_operation (op);
 906       return;
 907     }
 908   }
 909
 910   buckets_in_message = (ntohs (msg->header.size) - sizeof *msg) / IBF_BUCKET_SIZE;
 911
 912   if (0 == buckets_in_message)
 913   {
 914     GNUNET_break_op (0);
 915     fail_union_operation (op);
 916     return;
 917   }
 918
 919   if ((ntohs (msg->header.size) - sizeof *msg) != buckets_in_message * IBF_BUCKET_SIZE)
 920   {
 921     GNUNET_break (0);
 922     fail_union_operation (op);
 923     return;
 924   }
 925
 926   ibf_read_slice (&msg[1], op->state->ibf_buckets_received, buckets_in_message, op->state->remote_ibf);
 927   op->state->ibf_buckets_received += buckets_in_message;
 928
 929   if (op->state->ibf_buckets_received == op->state->remote_ibf->size)
 930   {
 931     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "received full ibf\n");
 932     op->state->phase = PHASE_EXPECT_ELEMENTS;
 933     decode_and_send (op);
 934   }
 935 }
 936
 937
 938 /**
 939  * Send a result message to the client indicating
 940  * that there is a new element.
 941  *
 942  * @param op union operation
 943  * @param element element to send
 944  */
 945 static void
 946 send_client_element (struct Operation *op,
 947                      struct GNUNET_SET_Element *element)
 948 {
 949   struct GNUNET_MQ_Envelope *ev;
 950   struct GNUNET_SET_ResultMessage *rm;
 951
 952   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "sending element (size %u) to client\n", element->size);
 953   GNUNET_assert (0 != op->spec->client_request_id);
 954   ev = GNUNET_MQ_msg_extra (rm, element->size, GNUNET_MESSAGE_TYPE_SET_RESULT);
 955   if (NULL == ev)
 956   {
 957     GNUNET_MQ_discard (ev);
 958     GNUNET_break (0);
 959     return;
 960   }
 961   rm->result_status = htons (GNUNET_SET_STATUS_OK);
 962   rm->request_id = htonl (op->spec->client_request_id);
 963   rm->element_type = element->type;
 964   memcpy (&rm[1], element->data, element->size);
 965   GNUNET_MQ_send (op->spec->set->client_mq, ev);
 966 }
 967
 968
 969 /**
 970  * Signal to the client that the operation has finished and
 971  * destroy the operation.
 972  *
 973  * @param cls operation to destroy
 974  */
 975 static void
 976 send_done_and_destroy (void *cls)
 977 {
 978   struct Operation *op = cls;
 979   struct GNUNET_MQ_Envelope *ev;
 980   struct GNUNET_SET_ResultMessage *rm;
 981   ev = GNUNET_MQ_msg (rm, GNUNET_MESSAGE_TYPE_SET_RESULT);
 982   rm->request_id = htonl (op->spec->client_request_id);
 983   rm->result_status = htons (GNUNET_SET_STATUS_DONE);
 984   rm->element_type = htons (0);
 985   GNUNET_MQ_send (op->spec->set->client_mq, ev);
 986   _GSS_operation_destroy (op);
 987 }
 988
 989
 990 /**
 991  * Send all remaining elements in the full result iterator.
 992  *
 993  * @param cls operation
 994  */
 995 static void
 996 send_remaining_elements (void *cls)
 997 {
 998   struct Operation *op = cls;
 999   struct KeyEntry *ke;
1000   int res;
1001
1002   res = GNUNET_CONTAINER_multihashmap32_iterator_next (op->state->full_result_iter, NULL, (const void **) &ke);
1003   res = GNUNET_NO;
1004   if (GNUNET_NO == res)
1005   {
1006     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "sending done and destroy because iterator ran out\n");
1007     send_done_and_destroy (op);
1008     return;
1009   }
1010
1011   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "sending elements from key entry\n");
1012
1013   while (1)
1014   {
1015     struct GNUNET_MQ_Envelope *ev;
1016     struct GNUNET_SET_ResultMessage *rm;
1017     struct GNUNET_SET_Element *element;
1018     element = &ke->element->element;
1019
1020     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "sending element (size %u) to client (full set)\n", element->size);
1021     GNUNET_assert (0 != op->spec->client_request_id);
1022     ev = GNUNET_MQ_msg_extra (rm, element->size, GNUNET_MESSAGE_TYPE_SET_RESULT);
1023     if (NULL == ev)
1024     {
1025       GNUNET_MQ_discard (ev);
1026       GNUNET_break (0);
1027       continue;
1028     }
1029     rm->result_status = htons (GNUNET_SET_STATUS_OK);
1030     rm->request_id = htonl (op->spec->client_request_id);
1031     rm->element_type = element->type;
1032     memcpy (&rm[1], element->data, element->size);
1033     if (ke->next_colliding == NULL)
1034     {
1035       GNUNET_MQ_notify_sent (ev, send_remaining_elements, op);
1036       GNUNET_MQ_send (op->spec->set->client_mq, ev);
1037       break;
1038     }
1039     GNUNET_MQ_send (op->spec->set->client_mq, ev);
1040     ke = ke->next_colliding;
1041   }
1042 }
1043
1044
1045 /**
1046  * Send a result message to the client indicating
1047  * that the operation is over.
1048  * After the result done message has been sent to the client,
1049  * destroy the evaluate operation.
1050  *
1051  * @param op union operation
1052  */
1053 static void
1054 finish_and_destroy (struct Operation *op)
1055 {
1056   GNUNET_assert (GNUNET_NO == op->state->client_done_sent);
1057
1058   if (GNUNET_SET_RESULT_FULL == op->spec->result_mode)
1059   {
1060     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "sending full result set\n");
1061     GNUNET_assert (NULL == op->state->full_result_iter);
1062     op->state->full_result_iter =
1063         GNUNET_CONTAINER_multihashmap32_iterator_create (op->state->key_to_element);
1064     send_remaining_elements (op);
1065     return;
1066   }
1067   send_done_and_destroy (op);
1068 }
1069
1070
1071 /**
1072  * Handle an element message from a remote peer.
1073  *
1074  * @param cls the union operation
1075  * @param mh the message
1076  */
1077 static void
1078 handle_p2p_elements (void *cls, const struct GNUNET_MessageHeader *mh)
1079 {
1080   struct Operation *op = cls;
1081   struct ElementEntry *ee;
1082   uint16_t element_size;
1083
1084   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "got element from peer\n");
1085
1086   if ( (op->state->phase != PHASE_EXPECT_ELEMENTS) &&
1087        (op->state->phase != PHASE_EXPECT_ELEMENTS_AND_REQUESTS) )
1088   {
1089     fail_union_operation (op);
1090     GNUNET_break (0);
1091     return;
1092   }
1093   element_size = ntohs (mh->size) - sizeof (struct GNUNET_MessageHeader);
1094   ee = GNUNET_malloc (sizeof *ee + element_size);
1095   memcpy (&ee[1], &mh[1], element_size);
1096   ee->element.size = element_size;
1097   ee->element.data = &ee[1];
1098   ee->remote = GNUNET_YES;
1099   GNUNET_CRYPTO_hash (ee->element.data, ee->element.size, &ee->element_hash);
1100
1101   if (GNUNET_YES == op_has_element (op, &ee->element_hash))
1102   {
1103     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "got existing element from peer\n");
1104     GNUNET_free (ee);
1105     return;
1106   }
1107
1108   op_register_element (op, ee);
1109   /* only send results immediately if the client wants it */
1110   if (GNUNET_SET_RESULT_ADDED == op->spec->result_mode)
1111     send_client_element (op, &ee->element);
1112 }
1113
1114
1115 /**
1116  * Handle an element request from a remote peer.
1117  *
1118  * @param cls the union operation
1119  * @param mh the message
1120  */
1121 static void
1122 handle_p2p_element_requests (void *cls, const struct GNUNET_MessageHeader *mh)
1123 {
1124   struct Operation *op = cls;
1125   struct IBF_Key *ibf_key;
1126   unsigned int num_keys;
1127
1128   /* look up elements and send them */
1129   if (op->state->phase != PHASE_EXPECT_ELEMENTS_AND_REQUESTS)
1130   {
1131     GNUNET_break (0);
1132     fail_union_operation (op);
1133     return;
1134   }
1135
1136   num_keys = (ntohs (mh->size) - sizeof *mh) / sizeof (struct IBF_Key);
1137
1138   if ((ntohs (mh->size) - sizeof *mh) != num_keys * sizeof (struct IBF_Key))
1139   {
1140     GNUNET_break (0);
1141     fail_union_operation (op);
1142     return;
1143   }
1144
1145   ibf_key = (struct IBF_Key *) &mh[1];
1146   while (0 != num_keys--)
1147   {
1148     send_elements_for_key (op, *ibf_key);
1149     ibf_key++;
1150   }
1151 }
1152
1153
1154 /**
1155  * Handle a done message from a remote peer
1156  *
1157  * @param cls the union operation
1158  * @param mh the message
1159  */
1160 static void
1161 handle_p2p_done (void *cls, const struct GNUNET_MessageHeader *mh)
1162 {
1163   struct Operation *op = cls;
1164   struct GNUNET_MQ_Envelope *ev;
1165
1166   if (op->state->phase == PHASE_EXPECT_ELEMENTS_AND_REQUESTS)
1167   {
1168     /* we got all requests, but still have to send our elements as response */
1169
1170     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "got DONE, sending final DONE after elements\n");
1171     op->state->phase = PHASE_FINISHED;
1172     ev = GNUNET_MQ_msg_header (GNUNET_MESSAGE_TYPE_SET_P2P_DONE);
1173     GNUNET_MQ_send (op->mq, ev);
1174     return;
1175   }
1176   if (op->state->phase == PHASE_EXPECT_ELEMENTS)
1177   {
1178     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "got final DONE\n");
1179     op->state->phase = PHASE_FINISHED;
1180     finish_and_destroy (op);
1181     return;
1182   }
1183   GNUNET_break (0);
1184   fail_union_operation (op);
1185 }
1186
1187
1188 /**
1189  * Evaluate a union operation with
1190  * a remote peer.
1191  *
1192  * @param op operation to evaluate
1193  */
1194 static void
1195 union_evaluate (struct Operation *op)
1196 {
1197   op->state = GNUNET_new (struct OperationState);
1198   op->state->se = strata_estimator_dup (op->spec->set->state->se);
1199   /* we started the operation, thus we have to send the operation request */
1200   op->state->phase = PHASE_EXPECT_SE;
1201   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "evaluating union operation");
1202   send_operation_request (op);
1203 }
1204
1205
1206 /**
1207  * Accept an union operation request from a remote peer.
1208  * Only initializes the private operation state.
1209  *
1210  * @param op operation that will be accepted as a union operation
1211  */
1212 static void
1213 union_accept (struct Operation *op)
1214 {
1215   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "accepting set union operation\n");
1216   op->state = GNUNET_new (struct OperationState);
1217   op->state->se = strata_estimator_dup (op->spec->set->state->se);
1218   /* kick off the operation */
1219   send_strata_estimator (op);
1220 }
1221
1222
1223 /**
1224  * Create a new set supporting the union operation
1225  *
1226  * @return the newly created set
1227  */
1228 static struct SetState *
1229 union_set_create (void)
1230 {
1231   struct SetState *set_state;
1232
1233   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "union set created\n");
1234
1235   set_state = GNUNET_new (struct SetState);
1236   set_state->se = strata_estimator_create (SE_STRATA_COUNT,
1237                                               SE_IBF_SIZE, SE_IBF_HASH_NUM);
1238   return set_state;
1239 }
1240
1241
1242 /**
1243  * Add the element from the given element message to the set.
1244  *
1245  * @param set_state state of the set want to add to
1246  * @param ee the element to add to the set
1247  */
1248 static void
1249 union_add (struct SetState *set_state, struct ElementEntry *ee)
1250 {
1251   strata_estimator_insert (set_state->se, get_ibf_key (&ee->element_hash, 0));
1252 }
1253
1254
1255 /**
1256  * Remove the element given in the element message from the set.
1257  * Only marks the element as removed, so that older set operations can still exchange it.
1258  *
1259  * @param set_state state of the set to remove from
1260  * @param ee set element to remove
1261  */
1262 static void
1263 union_remove (struct SetState *set_state, struct ElementEntry *ee)
1264 {
1265   strata_estimator_remove (set_state->se, get_ibf_key (&ee->element_hash, 0));
1266 }
1267
1268
1269 /**
1270  * Destroy a set that supports the union operation.
1271  *
1272  * @param set_state the set to destroy
1273  */
1274 static void
1275 union_set_destroy (struct SetState *set_state)
1276 {
1277   if (NULL != set_state->se)
1278   {
1279     strata_estimator_destroy (set_state->se);
1280     set_state->se = NULL;
1281   }
1282   GNUNET_free (set_state);
1283 }
1284
1285
1286 /**
1287  * Dispatch messages for a union operation.
1288  *
1289  * @param op the state of the union evaluate operation
1290  * @param mh the received message
1291  * @return GNUNET_SYSERR if the tunnel should be disconnected,
1292  *         GNUNET_OK otherwise
1293  */
1294 int
1295 union_handle_p2p_message (struct Operation *op,
1296                           const struct GNUNET_MessageHeader *mh)
1297 {
1298   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "received p2p message (t: %u, s: %u)\n",
1299               ntohs (mh->type), ntohs (mh->size));
1300   switch (ntohs (mh->type))
1301   {
1302     case GNUNET_MESSAGE_TYPE_SET_UNION_P2P_IBF:
1303       handle_p2p_ibf (op, mh);
1304       break;
1305     case GNUNET_MESSAGE_TYPE_SET_UNION_P2P_SE:
1306       handle_p2p_strata_estimator (op, mh);
1307       break;
1308     case GNUNET_MESSAGE_TYPE_SET_P2P_ELEMENTS:
1309       handle_p2p_elements (op, mh);
1310       break;
1311     case GNUNET_MESSAGE_TYPE_SET_P2P_ELEMENT_REQUESTS:
1312       handle_p2p_element_requests (op, mh);
1313       break;
1314     case GNUNET_MESSAGE_TYPE_SET_P2P_DONE:
1315       handle_p2p_done (op, mh);
1316       break;
1317     default:
1318       /* something wrong with mesh's message handlers? */
1319       GNUNET_assert (0);
1320   }
1321   return GNUNET_OK;
1322 }
1323
1324
1325 static void
1326 union_peer_disconnect (struct Operation *op)
1327 {
1328   if (PHASE_FINISHED != op->state->phase)
1329   {
1330     struct GNUNET_MQ_Envelope *ev;
1331     struct GNUNET_SET_ResultMessage *msg;
1332
1333     ev = GNUNET_MQ_msg (msg, GNUNET_MESSAGE_TYPE_SET_RESULT);
1334     msg->request_id = htonl (op->spec->client_request_id);
1335     msg->result_status = htons (GNUNET_SET_STATUS_FAILURE);
1336     msg->element_type = htons (0);
1337     GNUNET_MQ_send (op->spec->set->client_mq, ev);
1338     GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "other peer disconnected prematurely\n");
1339     _GSS_operation_destroy (op);
1340     return;
1341   }
1342   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "other peer disconnected (finished)\n");
1343   if (GNUNET_NO == op->state->client_done_sent)
1344     finish_and_destroy (op);
1345 }
1346
1347
1348 const struct SetVT *
1349 _GSS_union_vt ()
1350 {
1351   static const struct SetVT union_vt = {
1352     .create = &union_set_create,
1353     .msg_handler = &union_handle_p2p_message,
1354     .add = &union_add,
1355     .remove = &union_remove,
1356     .destroy_set = &union_set_destroy,
1357     .evaluate = &union_evaluate,
1358     .accept = &union_accept,
1359     .peer_disconnect = &union_peer_disconnect,
1360     .cancel = &union_op_cancel,
1361   };
1362
1363   return &union_vt;
1364 }