src/set/gnunet-service-set_union.c

   1 /*
   2       This file is part of GNUnet
   3       Copyright (C) 2013-2017 GNUnet e.V.
   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., 51 Franklin Street, Fifth Floor,
  18       Boston, MA 02110-1301, USA.
  19 */
  20 /**
  21  * @file set/gnunet-service-set_union.c
  22  * @brief two-peer set operations
  23  * @author Florian Dold
  24  * @author Christian Grothoff
  25  */
  26 #include "platform.h"
  27 #include "gnunet_util_lib.h"
  28 #include "gnunet_statistics_service.h"
  29 #include "gnunet-service-set.h"
  30 #include "ibf.h"
  31 #include "gnunet-service-set_union.h"
  32 #include "gnunet-service-set_union_strata_estimator.h"
  33 #include "gnunet-service-set_protocol.h"
  34 #include <gcrypt.h>
  35
  36
  37 #define LOG(kind,...) GNUNET_log_from (kind, "set-union",__VA_ARGS__)
  38
  39
  40 /**
  41  * Number of IBFs in a strata estimator.
  42  */
  43 #define SE_STRATA_COUNT 32
  44
  45 /**
  46  * Size of the IBFs in the strata estimator.
  47  */
  48 #define SE_IBF_SIZE 80
  49
  50 /**
  51  * The hash num parameter for the difference digests and strata estimators.
  52  */
  53 #define SE_IBF_HASH_NUM 4
  54
  55 /**
  56  * Number of buckets that can be transmitted in one message.
  57  */
  58 #define MAX_BUCKETS_PER_MESSAGE ((1<<15) / IBF_BUCKET_SIZE)
  59
  60 /**
  61  * The maximum size of an ibf we use is 2^(MAX_IBF_ORDER).
  62  * Choose this value so that computing the IBF is still cheaper
  63  * than transmitting all values.
  64  */
  65 #define MAX_IBF_ORDER (20)
  66
  67 /**
  68  * Number of buckets used in the ibf per estimated
  69  * difference.
  70  */
  71 #define IBF_ALPHA 4
  72
  73
  74 /**
  75  * Current phase we are in for a union operation.
  76  */
  77 enum UnionOperationPhase
  78 {
  79   /**
  80    * We sent the request message, and expect a strata estimator.
  81    */
  82   PHASE_EXPECT_SE,
  83
  84   /**
  85    * We sent the strata estimator, and expect an IBF. This phase is entered once
  86    * upon initialization and later via #PHASE_EXPECT_ELEMENTS_AND_REQUESTS.
  87    *
  88    * XXX: could use better wording.
  89    * XXX: repurposed to also expect a "request full set" message, should be renamed
  90    *
  91    * After receiving the complete IBF, we enter #PHASE_EXPECT_ELEMENTS
  92    */
  93   PHASE_EXPECT_IBF,
  94
  95   /**
  96    * Continuation for multi part IBFs.
  97    */
  98   PHASE_EXPECT_IBF_CONT,
  99
 100   /**
 101    * We are decoding an IBF.
 102    */
 103   PHASE_INVENTORY_ACTIVE,
 104
 105   /**
 106    * The other peer is decoding the IBF we just sent.
 107    */
 108   PHASE_INVENTORY_PASSIVE,
 109
 110   /**
 111    * The protocol is almost finished, but we still have to flush our message
 112    * queue and/or expect some elements.
 113    */
 114   PHASE_FINISH_CLOSING,
 115
 116   /**
 117    * In the penultimate phase,
 118    * we wait until all our demands
 119    * are satisfied.  Then we send a done
 120    * message, and wait for another done message.
 121    */
 122   PHASE_FINISH_WAITING,
 123
 124   /**
 125    * In the ultimate phase, we wait until
 126    * our demands are satisfied and then
 127    * quit (sending another DONE message).
 128    */
 129   PHASE_DONE,
 130
 131   /**
 132    * After sending the full set, wait for responses with the elements
 133    * that the local peer is missing.
 134    */
 135   PHASE_FULL_SENDING,
 136 };
 137
 138
 139 /**
 140  * State of an evaluate operation with another peer.
 141  */
 142 struct OperationState
 143 {
 144   /**
 145    * Copy of the set's strata estimator at the time of
 146    * creation of this operation.
 147    */
 148   struct StrataEstimator *se;
 149
 150   /**
 151    * The IBF we currently receive.
 152    */
 153   struct InvertibleBloomFilter *remote_ibf;
 154
 155   /**
 156    * The IBF with the local set's element.
 157    */
 158   struct InvertibleBloomFilter *local_ibf;
 159
 160   /**
 161    * Maps unsalted IBF-Keys to elements.
 162    * Used as a multihashmap, the keys being the lower 32bit of the IBF-Key.
 163    * Colliding IBF-Keys are linked.
 164    */
 165   struct GNUNET_CONTAINER_MultiHashMap32 *key_to_element;
 166
 167   /**
 168    * Current state of the operation.
 169    */
 170   enum UnionOperationPhase phase;
 171
 172   /**
 173    * Did we send the client that we are done?
 174    */
 175   int client_done_sent;
 176
 177   /**
 178    * Number of ibf buckets already received into the @a remote_ibf.
 179    */
 180   unsigned int ibf_buckets_received;
 181
 182   /**
 183    * Hashes for elements that we have demanded from the other peer.
 184    */
 185   struct GNUNET_CONTAINER_MultiHashMap *demanded_hashes;
 186
 187   /**
 188    * Salt that we're using for sending IBFs
 189    */
 190   uint32_t salt_send;
 191
 192   /**
 193    * Salt for the IBF we've received and that we're currently decoding.
 194    */
 195   uint32_t salt_receive;
 196
 197   /**
 198    * Number of elements we received from the other peer
 199    * that were not in the local set yet.
 200    */
 201   uint32_t received_fresh;
 202
 203   /**
 204    * Total number of elements received from the other peer.
 205    */
 206   uint32_t received_total;
 207
 208   /**
 209    * Initial size of our set, just before
 210    * the operation started.
 211    */
 212   uint64_t initial_size;
 213 };
 214
 215
 216 /**
 217  * The key entry is used to associate an ibf key with an element.
 218  */
 219 struct KeyEntry
 220 {
 221   /**
 222    * IBF key for the entry, derived from the current salt.
 223    */
 224   struct IBF_Key ibf_key;
 225
 226   /**
 227    * The actual element associated with the key.
 228    *
 229    * Only owned by the union operation if element->operation
 230    * is #GNUNET_YES.
 231    */
 232   struct ElementEntry *element;
 233
 234   /**
 235    * Did we receive this element?
 236    * Even if element->is_foreign is false, we might
 237    * have received the element, so this indicates that
 238    * the other peer has it.
 239    */
 240   int received;
 241 };
 242
 243
 244 /**
 245  * Used as a closure for sending elements
 246  * with a specific IBF key.
 247  */
 248 struct SendElementClosure
 249 {
 250   /**
 251    * The IBF key whose matching elements should be
 252    * sent.
 253    */
 254   struct IBF_Key ibf_key;
 255
 256   /**
 257    * Operation for which the elements
 258    * should be sent.
 259    */
 260   struct Operation *op;
 261 };
 262
 263
 264 /**
 265  * Extra state required for efficient set union.
 266  */
 267 struct SetState
 268 {
 269   /**
 270    * The strata estimator is only generated once for
 271    * each set.
 272    * The IBF keys are derived from the element hashes with
 273    * salt=0.
 274    */
 275   struct StrataEstimator *se;
 276 };
 277
 278
 279 /**
 280  * Iterator over hash map entries, called to
 281  * destroy the linked list of colliding ibf key entries.
 282  *
 283  * @param cls closure
 284  * @param key current key code
 285  * @param value value in the hash map
 286  * @return #GNUNET_YES if we should continue to iterate,
 287  *         #GNUNET_NO if not.
 288  */
 289 static int
 290 destroy_key_to_element_iter (void *cls,
 291                              uint32_t key,
 292                              void *value)
 293 {
 294   struct KeyEntry *k = value;
 295
 296   GNUNET_assert (NULL != k);
 297   if (GNUNET_YES == k->element->remote)
 298   {
 299     GNUNET_free (k->element);
 300     k->element = NULL;
 301   }
 302   GNUNET_free (k);
 303   return GNUNET_YES;
 304 }
 305
 306
 307 /**
 308  * Destroy the union operation.  Only things specific to the union
 309  * operation are destroyed.
 310  *
 311  * @param op union operation to destroy
 312  */
 313 static void
 314 union_op_cancel (struct Operation *op)
 315 {
 316   LOG (GNUNET_ERROR_TYPE_DEBUG,
 317        "destroying union op\n");
 318   /* check if the op was canceled twice */
 319   GNUNET_assert (NULL != op->state);
 320   if (NULL != op->state->remote_ibf)
 321   {
 322     ibf_destroy (op->state->remote_ibf);
 323     op->state->remote_ibf = NULL;
 324   }
 325   if (NULL != op->state->demanded_hashes)
 326   {
 327     GNUNET_CONTAINER_multihashmap_destroy (op->state->demanded_hashes);
 328     op->state->demanded_hashes = NULL;
 329   }
 330   if (NULL != op->state->local_ibf)
 331   {
 332     ibf_destroy (op->state->local_ibf);
 333     op->state->local_ibf = NULL;
 334   }
 335   if (NULL != op->state->se)
 336   {
 337     strata_estimator_destroy (op->state->se);
 338     op->state->se = NULL;
 339   }
 340   if (NULL != op->state->key_to_element)
 341   {
 342     GNUNET_CONTAINER_multihashmap32_iterate (op->state->key_to_element,
 343                                              &destroy_key_to_element_iter,
 344                                              NULL);
 345     GNUNET_CONTAINER_multihashmap32_destroy (op->state->key_to_element);
 346     op->state->key_to_element = NULL;
 347   }
 348   GNUNET_free (op->state);
 349   op->state = NULL;
 350   LOG (GNUNET_ERROR_TYPE_DEBUG,
 351        "destroying union op done\n");
 352 }
 353
 354
 355 /**
 356  * Inform the client that the union operation has failed,
 357  * and proceed to destroy the evaluate operation.
 358  *
 359  * @param op the union operation to fail
 360  */
 361 static void
 362 fail_union_operation (struct Operation *op)
 363 {
 364   struct GNUNET_MQ_Envelope *ev;
 365   struct GNUNET_SET_ResultMessage *msg;
 366
 367   LOG (GNUNET_ERROR_TYPE_WARNING,
 368        "union operation failed\n");
 369   ev = GNUNET_MQ_msg (msg, GNUNET_MESSAGE_TYPE_SET_RESULT);
 370   msg->result_status = htons (GNUNET_SET_STATUS_FAILURE);
 371   msg->request_id = htonl (op->client_request_id);
 372   msg->element_type = htons (0);
 373   GNUNET_MQ_send (op->set->cs->mq,
 374                   ev);
 375   _GSS_operation_destroy (op, GNUNET_YES);
 376 }
 377
 378
 379 /**
 380  * Derive the IBF key from a hash code and
 381  * a salt.
 382  *
 383  * @param src the hash code
 384  * @return the derived IBF key
 385  */
 386 static struct IBF_Key
 387 get_ibf_key (const struct GNUNET_HashCode *src)
 388 {
 389   struct IBF_Key key;
 390   uint16_t salt = 0;
 391
 392   GNUNET_assert (GNUNET_OK ==
 393                  GNUNET_CRYPTO_kdf (&key, sizeof (key),
 394                                     src, sizeof *src,
 395                                     &salt, sizeof (salt),
 396                                     NULL, 0));
 397   return key;
 398 }
 399
 400
 401 /**
 402  * Context for #op_get_element_iterator
 403  */
 404 struct GetElementContext
 405 {
 406   /**
 407    * FIXME.
 408    */
 409   struct GNUNET_HashCode hash;
 410
 411   /**
 412    * FIXME.
 413    */
 414   struct KeyEntry *k;
 415 };
 416
 417
 418 /**
 419  * Iterator over the mapping from IBF keys to element entries.  Checks if we
 420  * have an element with a given GNUNET_HashCode.
 421  *
 422  * @param cls closure
 423  * @param key current key code
 424  * @param value value in the hash map
 425  * @return #GNUNET_YES if we should search further,
 426  *         #GNUNET_NO if we've found the element.
 427  */
 428 static int
 429 op_get_element_iterator (void *cls,
 430                          uint32_t key,
 431                          void *value)
 432 {
 433   struct GetElementContext *ctx = cls;
 434   struct KeyEntry *k = value;
 435
 436   GNUNET_assert (NULL != k);
 437   if (0 == GNUNET_CRYPTO_hash_cmp (&k->element->element_hash,
 438                                    &ctx->hash))
 439   {
 440     ctx->k = k;
 441     return GNUNET_NO;
 442   }
 443   return GNUNET_YES;
 444 }
 445
 446
 447 /**
 448  * Determine whether the given element is already in the operation's element
 449  * set.
 450  *
 451  * @param op operation that should be tested for 'element_hash'
 452  * @param element_hash hash of the element to look for
 453  * @return #GNUNET_YES if the element has been found, #GNUNET_NO otherwise
 454  */
 455 static struct KeyEntry *
 456 op_get_element (struct Operation *op,
 457                 const struct GNUNET_HashCode *element_hash)
 458 {
 459   int ret;
 460   struct IBF_Key ibf_key;
 461   struct GetElementContext ctx = {{{ 0 }} , 0};
 462
 463   ctx.hash = *element_hash;
 464
 465   ibf_key = get_ibf_key (element_hash);
 466   ret = GNUNET_CONTAINER_multihashmap32_get_multiple (op->state->key_to_element,
 467                                                       (uint32_t) ibf_key.key_val,
 468                                                       op_get_element_iterator,
 469                                                       &ctx);
 470
 471   /* was the iteration aborted because we found the element? */
 472   if (GNUNET_SYSERR == ret)
 473   {
 474     GNUNET_assert (NULL != ctx.k);
 475     return ctx.k;
 476   }
 477   return NULL;
 478 }
 479
 480
 481 /**
 482  * Insert an element into the union operation's
 483  * key-to-element mapping. Takes ownership of 'ee'.
 484  * Note that this does not insert the element in the set,
 485  * only in the operation's key-element mapping.
 486  * This is done to speed up re-tried operations, if some elements
 487  * were transmitted, and then the IBF fails to decode.
 488  *
 489  * XXX: clarify ownership, doesn't sound right.
 490  *
 491  * @param op the union operation
 492  * @param ee the element entry
 493  * @parem received was this element received from the remote peer?
 494  */
 495 static void
 496 op_register_element (struct Operation *op,
 497                      struct ElementEntry *ee,
 498                      int received)
 499 {
 500   struct IBF_Key ibf_key;
 501   struct KeyEntry *k;
 502
 503   ibf_key = get_ibf_key (&ee->element_hash);
 504   k = GNUNET_new (struct KeyEntry);
 505   k->element = ee;
 506   k->ibf_key = ibf_key;
 507   k->received = received;
 508   GNUNET_assert (GNUNET_OK ==
 509                  GNUNET_CONTAINER_multihashmap32_put (op->state->key_to_element,
 510                                                       (uint32_t) ibf_key.key_val,
 511                                                       k,
 512                                                       GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE));
 513 }
 514
 515
 516 /**
 517  * FIXME.
 518  */
 519 static void
 520 salt_key (const struct IBF_Key *k_in,
 521           uint32_t salt,
 522           struct IBF_Key *k_out)
 523 {
 524   int s = salt % 64;
 525   uint64_t x = k_in->key_val;
 526   /* rotate ibf key */
 527   x = (x >> s) | (x << (64 - s));
 528   k_out->key_val = x;
 529 }
 530
 531
 532 /**
 533  * FIXME.
 534  */
 535 static void
 536 unsalt_key (const struct IBF_Key *k_in,
 537             uint32_t salt,
 538             struct IBF_Key *k_out)
 539 {
 540   int s = salt % 64;
 541   uint64_t x = k_in->key_val;
 542   x = (x << s) | (x >> (64 - s));
 543   k_out->key_val = x;
 544 }
 545
 546
 547 /**
 548  * Insert a key into an ibf.
 549  *
 550  * @param cls the ibf
 551  * @param key unused
 552  * @param value the key entry to get the key from
 553  */
 554 static int
 555 prepare_ibf_iterator (void *cls,
 556                       uint32_t key,
 557                       void *value)
 558 {
 559   struct Operation *op = cls;
 560   struct KeyEntry *ke = value;
 561   struct IBF_Key salted_key;
 562
 563   LOG (GNUNET_ERROR_TYPE_DEBUG,
 564        "[OP %x] inserting %lx (hash %s) into ibf\n",
 565        (void *) op,
 566        (unsigned long) ke->ibf_key.key_val,
 567        GNUNET_h2s (&ke->element->element_hash));
 568   salt_key (&ke->ibf_key,
 569             op->state->salt_send,
 570             &salted_key);
 571   ibf_insert (op->state->local_ibf, salted_key);
 572   return GNUNET_YES;
 573 }
 574
 575
 576 /**
 577  * Iterator for initializing the
 578  * key-to-element mapping of a union operation
 579  *
 580  * @param cls the union operation `struct Operation *`
 581  * @param key unused
 582  * @param value the `struct ElementEntry *` to insert
 583  *        into the key-to-element mapping
 584  * @return #GNUNET_YES (to continue iterating)
 585  */
 586 static int
 587 init_key_to_element_iterator (void *cls,
 588                               const struct GNUNET_HashCode *key,
 589                               void *value)
 590 {
 591   struct Operation *op = cls;
 592   struct ElementEntry *ee = value;
 593
 594   /* make sure that the element belongs to the set at the time
 595    * of creating the operation */
 596   if (GNUNET_NO ==
 597       _GSS_is_element_of_operation (ee,
 598                                     op))
 599     return GNUNET_YES;
 600   GNUNET_assert (GNUNET_NO == ee->remote);
 601   op_register_element (op,
 602                        ee,
 603                        GNUNET_NO);
 604   return GNUNET_YES;
 605 }
 606
 607
 608 /**
 609  * Initialize the IBF key to element mapping local to this set
 610  * operation.
 611  *
 612  * @param op the set union operation
 613  */
 614 static void
 615 initialize_key_to_element (struct Operation *op)
 616 {
 617   unsigned int len;
 618
 619   GNUNET_assert (NULL == op->state->key_to_element);
 620   len = GNUNET_CONTAINER_multihashmap_size (op->set->content->elements);
 621   op->state->key_to_element = GNUNET_CONTAINER_multihashmap32_create (len + 1);
 622   GNUNET_CONTAINER_multihashmap_iterate (op->set->content->elements,
 623                                          &init_key_to_element_iterator,
 624                                          op);
 625 }
 626
 627
 628 /**
 629  * Create an ibf with the operation's elements
 630  * of the specified size
 631  *
 632  * @param op the union operation
 633  * @param size size of the ibf to create
 634  * @return #GNUNET_OK on success, #GNUNET_SYSERR on failure
 635  */
 636 static int
 637 prepare_ibf (struct Operation *op,
 638              uint32_t size)
 639 {
 640   GNUNET_assert (NULL != op->state->key_to_element);
 641
 642   if (NULL != op->state->local_ibf)
 643     ibf_destroy (op->state->local_ibf);
 644   op->state->local_ibf = ibf_create (size, SE_IBF_HASH_NUM);
 645   if (NULL == op->state->local_ibf)
 646   {
 647     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 648                 "Failed to allocate local IBF\n");
 649     return GNUNET_SYSERR;
 650   }
 651   GNUNET_CONTAINER_multihashmap32_iterate (op->state->key_to_element,
 652                                            &prepare_ibf_iterator,
 653                                            op);
 654   return GNUNET_OK;
 655 }
 656
 657
 658 /**
 659  * Send an ibf of appropriate size.
 660  *
 661  * Fragments the IBF into multiple messages if necessary.
 662  *
 663  * @param op the union operation
 664  * @param ibf_order order of the ibf to send, size=2^order
 665  * @return #GNUNET_OK on success, #GNUNET_SYSERR on failure
 666  */
 667 static int
 668 send_ibf (struct Operation *op,
 669           uint16_t ibf_order)
 670 {
 671   unsigned int buckets_sent = 0;
 672   struct InvertibleBloomFilter *ibf;
 673
 674   if (GNUNET_OK !=
 675       prepare_ibf (op, 1<<ibf_order))
 676   {
 677     /* allocation failed */
 678     return GNUNET_SYSERR;
 679   }
 680
 681   LOG (GNUNET_ERROR_TYPE_DEBUG,
 682        "sending ibf of size %u\n",
 683        1<<ibf_order);
 684
 685   {
 686     char name[64] = { 0 };
 687     snprintf (name, sizeof (name), "# sent IBF (order %u)", ibf_order);
 688     GNUNET_STATISTICS_update (_GSS_statistics, name, 1, GNUNET_NO);
 689   }
 690
 691   ibf = op->state->local_ibf;
 692
 693   while (buckets_sent < (1 << ibf_order))
 694   {
 695     unsigned int buckets_in_message;
 696     struct GNUNET_MQ_Envelope *ev;
 697     struct IBFMessage *msg;
 698
 699     buckets_in_message = (1 << ibf_order) - buckets_sent;
 700     /* limit to maximum */
 701     if (buckets_in_message > MAX_BUCKETS_PER_MESSAGE)
 702       buckets_in_message = MAX_BUCKETS_PER_MESSAGE;
 703
 704     ev = GNUNET_MQ_msg_extra (msg,
 705                               buckets_in_message * IBF_BUCKET_SIZE,
 706                               GNUNET_MESSAGE_TYPE_SET_UNION_P2P_IBF);
 707     msg->reserved1 = 0;
 708     msg->reserved2 = 0;
 709     msg->order = ibf_order;
 710     msg->offset = htonl (buckets_sent);
 711     msg->salt = htonl (op->state->salt_send);
 712     ibf_write_slice (ibf, buckets_sent,
 713                      buckets_in_message, &msg[1]);
 714     buckets_sent += buckets_in_message;
 715     LOG (GNUNET_ERROR_TYPE_DEBUG,
 716          "ibf chunk size %u, %u/%u sent\n",
 717          buckets_in_message,
 718          buckets_sent,
 719          1<<ibf_order);
 720     GNUNET_MQ_send (op->mq, ev);
 721   }
 722
 723   /* The other peer must decode the IBF, so
 724    * we're passive. */
 725   op->state->phase = PHASE_INVENTORY_PASSIVE;
 726   return GNUNET_OK;
 727 }
 728
 729
 730 /**
 731  * Compute the necessary order of an ibf
 732  * from the size of the symmetric set difference.
 733  *
 734  * @param diff the difference
 735  * @return the required size of the ibf
 736  */
 737 static unsigned int
 738 get_order_from_difference (unsigned int diff)
 739 {
 740   unsigned int ibf_order;
 741
 742   ibf_order = 2;
 743   while ( ( (1<<ibf_order) < (IBF_ALPHA * diff) ||
 744             ((1<<ibf_order) < SE_IBF_HASH_NUM) ) &&
 745           (ibf_order < MAX_IBF_ORDER) )
 746     ibf_order++;
 747   // add one for correction
 748   return ibf_order + 1;
 749 }
 750
 751
 752 /**
 753  * Send a set element.
 754  *
 755  * @param cls the union operation `struct Operation *`
 756  * @param key unused
 757  * @param value the `struct ElementEntry *` to insert
 758  *        into the key-to-element mapping
 759  * @return #GNUNET_YES (to continue iterating)
 760  */
 761 static int
 762 send_full_element_iterator (void *cls,
 763                        const struct GNUNET_HashCode *key,
 764                        void *value)
 765 {
 766   struct Operation *op = cls;
 767   struct GNUNET_SET_ElementMessage *emsg;
 768   struct ElementEntry *ee = value;
 769   struct GNUNET_SET_Element *el = &ee->element;
 770   struct GNUNET_MQ_Envelope *ev;
 771
 772   LOG (GNUNET_ERROR_TYPE_DEBUG,
 773        "Sending element %s\n",
 774        GNUNET_h2s (key));
 775   ev = GNUNET_MQ_msg_extra (emsg,
 776                             el->size,
 777                             GNUNET_MESSAGE_TYPE_SET_UNION_P2P_FULL_ELEMENT);
 778   emsg->element_type = htons (el->element_type);
 779   GNUNET_memcpy (&emsg[1],
 780                  el->data,
 781                  el->size);
 782   GNUNET_MQ_send (op->mq,
 783                   ev);
 784   return GNUNET_YES;
 785 }
 786
 787
 788 /**
 789  * Switch to full set transmission for @a op.
 790  *
 791  * @param op operation to switch to full set transmission.
 792  */
 793 static void
 794 send_full_set (struct Operation *op)
 795 {
 796   struct GNUNET_MQ_Envelope *ev;
 797
 798   op->state->phase = PHASE_FULL_SENDING;
 799   LOG (GNUNET_ERROR_TYPE_DEBUG,
 800        "Dedicing to transmit the full set\n");
 801   /* FIXME: use a more memory-friendly way of doing this with an
 802      iterator, just as we do in the non-full case! */
 803   (void) GNUNET_CONTAINER_multihashmap_iterate (op->set->content->elements,
 804                                                 &send_full_element_iterator,
 805                                                 op);
 806   ev = GNUNET_MQ_msg_header (GNUNET_MESSAGE_TYPE_SET_UNION_P2P_FULL_DONE);
 807   GNUNET_MQ_send (op->mq,
 808                   ev);
 809 }
 810
 811
 812 /**
 813  * Handle a strata estimator from a remote peer
 814  *
 815  * @param cls the union operation
 816  * @param msg the message
 817  */
 818 int
 819 check_union_p2p_strata_estimator (void *cls,
 820                                   const struct StrataEstimatorMessage *msg)
 821 {
 822   struct Operation *op = cls;
 823   int is_compressed;
 824   size_t len;
 825
 826   if (op->state->phase != PHASE_EXPECT_SE)
 827   {
 828     GNUNET_break (0);
 829     return GNUNET_SYSERR;
 830   }
 831   is_compressed = (GNUNET_MESSAGE_TYPE_SET_UNION_P2P_SEC == htons (msg->header.type));
 832   len = ntohs (msg->header.size) - sizeof (struct StrataEstimatorMessage);
 833   if ( (GNUNET_NO == is_compressed) &&
 834        (len != SE_STRATA_COUNT * SE_IBF_SIZE * IBF_BUCKET_SIZE) )
 835   {
 836     GNUNET_break (0);
 837     return GNUNET_SYSERR;
 838   }
 839   return GNUNET_OK;
 840 }
 841
 842
 843 /**
 844  * Handle a strata estimator from a remote peer
 845  *
 846  * @param cls the union operation
 847  * @param msg the message
 848  */
 849 void
 850 handle_union_p2p_strata_estimator (void *cls,
 851                                    const struct StrataEstimatorMessage *msg)
 852 {
 853   struct Operation *op = cls;
 854   struct StrataEstimator *remote_se;
 855   unsigned int diff;
 856   uint64_t other_size;
 857   size_t len;
 858   int is_compressed;
 859
 860   is_compressed = (GNUNET_MESSAGE_TYPE_SET_UNION_P2P_SEC == htons (msg->header.type));
 861   GNUNET_STATISTICS_update (_GSS_statistics,
 862                             "# bytes of SE received",
 863                             ntohs (msg->header.size),
 864                             GNUNET_NO);
 865   len = ntohs (msg->header.size) - sizeof (struct StrataEstimatorMessage);
 866   other_size = GNUNET_ntohll (msg->set_size);
 867   remote_se = strata_estimator_create (SE_STRATA_COUNT,
 868                                        SE_IBF_SIZE,
 869                                        SE_IBF_HASH_NUM);
 870   if (NULL == remote_se)
 871   {
 872     /* insufficient resources, fail */
 873     fail_union_operation (op);
 874     return;
 875   }
 876   if (GNUNET_OK !=
 877       strata_estimator_read (&msg[1],
 878                              len,
 879                              is_compressed,
 880                              remote_se))
 881   {
 882     /* decompression failed */
 883     strata_estimator_destroy (remote_se);
 884     fail_union_operation (op);
 885     return;
 886   }
 887   GNUNET_assert (NULL != op->state->se);
 888   diff = strata_estimator_difference (remote_se,
 889                                       op->state->se);
 890
 891   if (diff > 200)
 892     diff = diff * 3 / 2;
 893
 894   strata_estimator_destroy (remote_se);
 895   strata_estimator_destroy (op->state->se);
 896   op->state->se = NULL;
 897   LOG (GNUNET_ERROR_TYPE_DEBUG,
 898        "got se diff=%d, using ibf size %d\n",
 899        diff,
 900        1U << get_order_from_difference (diff));
 901
 902   {
 903     char *set_debug;
 904
 905     set_debug = getenv ("GNUNET_SET_BENCHMARK");
 906     if ( (NULL != set_debug) &&
 907          (0 == strcmp (set_debug, "1")) )
 908     {
 909       FILE *f = fopen ("set.log", "a");
 910       fprintf (f, "%llu\n", (unsigned long long) diff);
 911       fclose (f);
 912     }
 913   }
 914
 915   if ( (GNUNET_YES == op->byzantine) &&
 916        (other_size < op->byzantine_lower_bound) )
 917   {
 918     GNUNET_break (0);
 919     fail_union_operation (op);
 920     return;
 921   }
 922
 923   if ( (GNUNET_YES == op->force_full) ||
 924        (diff > op->state->initial_size / 4) ||
 925        (0 == other_size) )
 926   {
 927     LOG (GNUNET_ERROR_TYPE_DEBUG,
 928          "Deciding to go for full set transmission (diff=%d, own set=%u)\n",
 929          diff,
 930          op->state->initial_size);
 931     GNUNET_STATISTICS_update (_GSS_statistics,
 932                               "# of full sends",
 933                               1,
 934                               GNUNET_NO);
 935     if ( (op->state->initial_size <= other_size) ||
 936          (0 == other_size) )
 937     {
 938       send_full_set (op);
 939     }
 940     else
 941     {
 942       struct GNUNET_MQ_Envelope *ev;
 943
 944       LOG (GNUNET_ERROR_TYPE_DEBUG,
 945            "Telling other peer that we expect its full set\n");
 946       op->state->phase = PHASE_EXPECT_IBF;
 947       ev = GNUNET_MQ_msg_header (GNUNET_MESSAGE_TYPE_SET_UNION_P2P_REQUEST_FULL);
 948       GNUNET_MQ_send (op->mq,
 949                       ev);
 950     }
 951   }
 952   else
 953   {
 954     GNUNET_STATISTICS_update (_GSS_statistics,
 955                               "# of ibf sends",
 956                               1,
 957                               GNUNET_NO);
 958     if (GNUNET_OK !=
 959         send_ibf (op,
 960                   get_order_from_difference (diff)))
 961     {
 962       /* Internal error, best we can do is shut the connection */
 963       GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 964                   "Failed to send IBF, closing connection\n");
 965       fail_union_operation (op);
 966       return;
 967     }
 968   }
 969   GNUNET_CADET_receive_done (op->channel);
 970 }
 971
 972
 973 /**
 974  * Iterator to send elements to a remote peer
 975  *
 976  * @param cls closure with the element key and the union operation
 977  * @param key ignored
 978  * @param value the key entry
 979  */
 980 static int
 981 send_offers_iterator (void *cls,
 982                       uint32_t key,
 983                       void *value)
 984 {
 985   struct SendElementClosure *sec = cls;
 986   struct Operation *op = sec->op;
 987   struct KeyEntry *ke = value;
 988   struct GNUNET_MQ_Envelope *ev;
 989   struct GNUNET_MessageHeader *mh;
 990
 991   /* Detect 32-bit key collision for the 64-bit IBF keys. */
 992   if (ke->ibf_key.key_val != sec->ibf_key.key_val)
 993     return GNUNET_YES;
 994
 995   ev = GNUNET_MQ_msg_header_extra (mh,
 996                                    sizeof (struct GNUNET_HashCode),
 997                                    GNUNET_MESSAGE_TYPE_SET_UNION_P2P_OFFER);
 998
 999   GNUNET_assert (NULL != ev);
1000   *(struct GNUNET_HashCode *) &mh[1] = ke->element->element_hash;
1001   LOG (GNUNET_ERROR_TYPE_DEBUG,
1002        "[OP %x] sending element offer (%s) to peer\n",
1003        (void *) op,
1004        GNUNET_h2s (&ke->element->element_hash));
1005   GNUNET_MQ_send (op->mq, ev);
1006   return GNUNET_YES;
1007 }
1008
1009
1010 /**
1011  * Send offers (in the form of GNUNET_Hash-es) to the remote peer for the given IBF key.
1012  *
1013  * @param op union operation
1014  * @param ibf_key IBF key of interest
1015  */
1016 static void
1017 send_offers_for_key (struct Operation *op,
1018                      struct IBF_Key ibf_key)
1019 {
1020   struct SendElementClosure send_cls;
1021
1022   send_cls.ibf_key = ibf_key;
1023   send_cls.op = op;
1024   (void) GNUNET_CONTAINER_multihashmap32_get_multiple (op->state->key_to_element,
1025                                                        (uint32_t) ibf_key.key_val,
1026                                                        &send_offers_iterator,
1027                                                        &send_cls);
1028 }
1029
1030
1031 /**
1032  * Decode which elements are missing on each side, and
1033  * send the appropriate offers and inquiries.
1034  *
1035  * @param op union operation
1036  * @return #GNUNET_OK on success, #GNUNET_SYSERR on failure
1037  */
1038 static int
1039 decode_and_send (struct Operation *op)
1040 {
1041   struct IBF_Key key;
1042   struct IBF_Key last_key;
1043   int side;
1044   unsigned int num_decoded;
1045   struct InvertibleBloomFilter *diff_ibf;
1046
1047   GNUNET_assert (PHASE_INVENTORY_ACTIVE == op->state->phase);
1048
1049   if (GNUNET_OK !=
1050       prepare_ibf (op,
1051                    op->state->remote_ibf->size))
1052   {
1053     GNUNET_break (0);
1054     /* allocation failed */
1055     return GNUNET_SYSERR;
1056   }
1057   diff_ibf = ibf_dup (op->state->local_ibf);
1058   ibf_subtract (diff_ibf,
1059                 op->state->remote_ibf);
1060
1061   ibf_destroy (op->state->remote_ibf);
1062   op->state->remote_ibf = NULL;
1063
1064   LOG (GNUNET_ERROR_TYPE_DEBUG,
1065        "decoding IBF (size=%u)\n",
1066        diff_ibf->size);
1067
1068   num_decoded = 0;
1069   key.key_val = 0; /* just to avoid compiler thinking we use undef'ed variable */
1070
1071   while (1)
1072   {
1073     int res;
1074     int cycle_detected = GNUNET_NO;
1075
1076     last_key = key;
1077
1078     res = ibf_decode (diff_ibf, &side, &key);
1079     if (res == GNUNET_OK)
1080     {
1081       LOG (GNUNET_ERROR_TYPE_DEBUG,
1082            "decoded ibf key %lx\n",
1083            (unsigned long) key.key_val);
1084       num_decoded += 1;
1085       if ( (num_decoded > diff_ibf->size) ||
1086            ( (num_decoded > 1) &&
1087              (last_key.key_val == key.key_val) ) )
1088       {
1089         LOG (GNUNET_ERROR_TYPE_DEBUG,
1090              "detected cyclic ibf (decoded %u/%u)\n",
1091              num_decoded,
1092              diff_ibf->size);
1093         cycle_detected = GNUNET_YES;
1094       }
1095     }
1096     if ( (GNUNET_SYSERR == res) ||
1097          (GNUNET_YES == cycle_detected) )
1098     {
1099       int next_order;
1100       next_order = 0;
1101       while (1<<next_order < diff_ibf->size)
1102         next_order++;
1103       next_order++;
1104       if (next_order <= MAX_IBF_ORDER)
1105       {
1106         LOG (GNUNET_ERROR_TYPE_DEBUG,
1107              "decoding failed, sending larger ibf (size %u)\n",
1108              1<<next_order);
1109         GNUNET_STATISTICS_update (_GSS_statistics,
1110                                   "# of IBF retries",
1111                                   1,
1112                                   GNUNET_NO);
1113         op->state->salt_send++;
1114         if (GNUNET_OK !=
1115             send_ibf (op, next_order))
1116         {
1117           /* Internal error, best we can do is shut the connection */
1118           GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
1119                       "Failed to send IBF, closing connection\n");
1120           fail_union_operation (op);
1121           ibf_destroy (diff_ibf);
1122           return GNUNET_SYSERR;
1123         }
1124       }
1125       else
1126       {
1127         GNUNET_STATISTICS_update (_GSS_statistics,
1128                                   "# of failed union operations (too large)",
1129                                   1,
1130                                   GNUNET_NO);
1131         // XXX: Send the whole set, element-by-element
1132         LOG (GNUNET_ERROR_TYPE_ERROR,
1133              "set union failed: reached ibf limit\n");
1134         fail_union_operation (op);
1135         ibf_destroy (diff_ibf);
1136         return GNUNET_SYSERR;
1137       }
1138       break;
1139     }
1140     if (GNUNET_NO == res)
1141     {
1142       struct GNUNET_MQ_Envelope *ev;
1143
1144       LOG (GNUNET_ERROR_TYPE_DEBUG,
1145            "transmitted all values, sending DONE\n");
1146       ev = GNUNET_MQ_msg_header (GNUNET_MESSAGE_TYPE_SET_UNION_P2P_DONE);
1147       GNUNET_MQ_send (op->mq, ev);
1148       /* We now wait until we get a DONE message back
1149        * and then wait for our MQ to be flushed and all our
1150        * demands be delivered. */
1151       break;
1152     }
1153     if (1 == side)
1154     {
1155       struct IBF_Key unsalted_key;
1156
1157       unsalt_key (&key,
1158                   op->state->salt_receive,
1159                   &unsalted_key);
1160       send_offers_for_key (op,
1161                            unsalted_key);
1162     }
1163     else if (-1 == side)
1164     {
1165       struct GNUNET_MQ_Envelope *ev;
1166       struct InquiryMessage *msg;
1167
1168       /* It may be nice to merge multiple requests, but with CADET's corking it is not worth
1169        * the effort additional complexity. */
1170       ev = GNUNET_MQ_msg_extra (msg,
1171                                 sizeof (struct IBF_Key),
1172                                 GNUNET_MESSAGE_TYPE_SET_UNION_P2P_INQUIRY);
1173       msg->salt = htonl (op->state->salt_receive);
1174       GNUNET_memcpy (&msg[1],
1175               &key,
1176               sizeof (struct IBF_Key));
1177       LOG (GNUNET_ERROR_TYPE_DEBUG,
1178            "sending element inquiry for IBF key %lx\n",
1179            (unsigned long) key.key_val);
1180       GNUNET_MQ_send (op->mq, ev);
1181     }
1182     else
1183     {
1184       GNUNET_assert (0);
1185     }
1186   }
1187   ibf_destroy (diff_ibf);
1188   return GNUNET_OK;
1189 }
1190
1191
1192 /**
1193  * Check an IBF message from a remote peer.
1194  *
1195  * Reassemble the IBF from multiple pieces, and
1196  * process the whole IBF once possible.
1197  *
1198  * @param cls the union operation
1199  * @param msg the header of the message
1200  * @return #GNUNET_OK if @a msg is well-formed
1201  */
1202 int
1203 check_union_p2p_ibf (void *cls,
1204                      const struct IBFMessage *msg)
1205 {
1206   struct Operation *op = cls;
1207   unsigned int buckets_in_message;
1208
1209   if (GNUNET_SET_OPERATION_UNION != op->set->operation)
1210   {
1211     GNUNET_break_op (0);
1212     return GNUNET_SYSERR;
1213   }
1214   buckets_in_message = (ntohs (msg->header.size) - sizeof *msg) / IBF_BUCKET_SIZE;
1215   if (0 == buckets_in_message)
1216   {
1217     GNUNET_break_op (0);
1218     return GNUNET_SYSERR;
1219   }
1220   if ((ntohs (msg->header.size) - sizeof *msg) != buckets_in_message * IBF_BUCKET_SIZE)
1221   {
1222     GNUNET_break_op (0);
1223     return GNUNET_SYSERR;
1224   }
1225   if (op->state->phase == PHASE_EXPECT_IBF_CONT)
1226   {
1227     if (ntohl (msg->offset) != op->state->ibf_buckets_received)
1228     {
1229       GNUNET_break_op (0);
1230       return GNUNET_SYSERR;
1231     }
1232     if (1<<msg->order != op->state->remote_ibf->size)
1233     {
1234       GNUNET_break_op (0);
1235       return GNUNET_SYSERR;
1236     }
1237     if (ntohl (msg->salt) != op->state->salt_receive)
1238     {
1239       GNUNET_break_op (0);
1240       return GNUNET_SYSERR;
1241     }
1242   }
1243   else if ( (op->state->phase != PHASE_INVENTORY_PASSIVE) &&
1244             (op->state->phase != PHASE_EXPECT_IBF) )
1245   {
1246     GNUNET_break_op (0);
1247     return GNUNET_SYSERR;
1248   }
1249
1250   return GNUNET_OK;
1251 }
1252
1253
1254 /**
1255  * Handle an IBF message from a remote peer.
1256  *
1257  * Reassemble the IBF from multiple pieces, and
1258  * process the whole IBF once possible.
1259  *
1260  * @param cls the union operation
1261  * @param msg the header of the message
1262  */
1263 void
1264 handle_union_p2p_ibf (void *cls,
1265                       const struct IBFMessage *msg)
1266 {
1267   struct Operation *op = cls;
1268   unsigned int buckets_in_message;
1269
1270   buckets_in_message = (ntohs (msg->header.size) - sizeof *msg) / IBF_BUCKET_SIZE;
1271   if ( (op->state->phase == PHASE_INVENTORY_PASSIVE) ||
1272        (op->state->phase == PHASE_EXPECT_IBF) )
1273   {
1274     op->state->phase = PHASE_EXPECT_IBF_CONT;
1275     GNUNET_assert (NULL == op->state->remote_ibf);
1276     LOG (GNUNET_ERROR_TYPE_DEBUG,
1277          "Creating new ibf of size %u\n",
1278          1 << msg->order);
1279     op->state->remote_ibf = ibf_create (1<<msg->order, SE_IBF_HASH_NUM);
1280     op->state->salt_receive = ntohl (msg->salt);
1281     LOG (GNUNET_ERROR_TYPE_DEBUG,
1282          "Receiving new IBF with salt %u\n",
1283          op->state->salt_receive);
1284     if (NULL == op->state->remote_ibf)
1285     {
1286       GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
1287                   "Failed to parse remote IBF, closing connection\n");
1288       fail_union_operation (op);
1289       return;
1290     }
1291     op->state->ibf_buckets_received = 0;
1292     if (0 != ntohl (msg->offset))
1293     {
1294       GNUNET_break_op (0);
1295       fail_union_operation (op);
1296       return;
1297     }
1298   }
1299   else
1300   {
1301     GNUNET_assert (op->state->phase == PHASE_EXPECT_IBF_CONT);
1302     LOG (GNUNET_ERROR_TYPE_DEBUG,
1303          "Received more of IBF\n");
1304   }
1305   GNUNET_assert (NULL != op->state->remote_ibf);
1306
1307   ibf_read_slice (&msg[1],
1308                   op->state->ibf_buckets_received,
1309                   buckets_in_message,
1310                   op->state->remote_ibf);
1311   op->state->ibf_buckets_received += buckets_in_message;
1312
1313   if (op->state->ibf_buckets_received == op->state->remote_ibf->size)
1314   {
1315     LOG (GNUNET_ERROR_TYPE_DEBUG,
1316          "received full ibf\n");
1317     op->state->phase = PHASE_INVENTORY_ACTIVE;
1318     if (GNUNET_OK !=
1319         decode_and_send (op))
1320     {
1321       /* Internal error, best we can do is shut down */
1322       GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
1323                   "Failed to decode IBF, closing connection\n");
1324       fail_union_operation (op);
1325       return;
1326     }
1327   }
1328   GNUNET_CADET_receive_done (op->channel);
1329 }
1330
1331
1332 /**
1333  * Send a result message to the client indicating
1334  * that there is a new element.
1335  *
1336  * @param op union operation
1337  * @param element element to send
1338  * @param status status to send with the new element
1339  */
1340 static void
1341 send_client_element (struct Operation *op,
1342                      struct GNUNET_SET_Element *element,
1343                      int status)
1344 {
1345   struct GNUNET_MQ_Envelope *ev;
1346   struct GNUNET_SET_ResultMessage *rm;
1347
1348   LOG (GNUNET_ERROR_TYPE_DEBUG,
1349        "sending element (size %u) to client\n",
1350        element->size);
1351   GNUNET_assert (0 != op->client_request_id);
1352   ev = GNUNET_MQ_msg_extra (rm, element->size, GNUNET_MESSAGE_TYPE_SET_RESULT);
1353   if (NULL == ev)
1354   {
1355     GNUNET_MQ_discard (ev);
1356     GNUNET_break (0);
1357     return;
1358   }
1359   rm->result_status = htons (status);
1360   rm->request_id = htonl (op->client_request_id);
1361   rm->element_type = htons (element->element_type);
1362   rm->current_size = GNUNET_htonll (GNUNET_CONTAINER_multihashmap32_size (op->state->key_to_element));
1363   GNUNET_memcpy (&rm[1],
1364                  element->data,
1365                  element->size);
1366   GNUNET_MQ_send (op->set->cs->mq,
1367                   ev);
1368 }
1369
1370
1371 /**
1372  * Destroy remote channel.
1373  *
1374  * @param op operation
1375  */
1376 void destroy_channel (struct Operation *op)
1377 {
1378   struct GNUNET_CADET_Channel *channel;
1379
1380   if (NULL != (channel = op->channel))
1381   {
1382     /* This will free op; called conditionally as this helper function
1383        is also called from within the channel disconnect handler. */
1384     op->channel = NULL;
1385     GNUNET_CADET_channel_destroy (channel);
1386   }
1387 }
1388
1389
1390 /**
1391  * Signal to the client that the operation has finished and
1392  * destroy the operation.
1393  *
1394  * @param cls operation to destroy
1395  */
1396 static void
1397 send_client_done (void *cls)
1398 {
1399   struct Operation *op = cls;
1400   struct GNUNET_MQ_Envelope *ev;
1401   struct GNUNET_SET_ResultMessage *rm;
1402
1403   if (GNUNET_YES == op->state->client_done_sent) {
1404     return;
1405   }
1406
1407   if (PHASE_DONE != op->state->phase) {
1408     LOG (GNUNET_ERROR_TYPE_WARNING,
1409          "union operation failed\n");
1410     ev = GNUNET_MQ_msg (rm, GNUNET_MESSAGE_TYPE_SET_RESULT);
1411     rm->result_status = htons (GNUNET_SET_STATUS_FAILURE);
1412     rm->request_id = htonl (op->client_request_id);
1413     rm->element_type = htons (0);
1414     GNUNET_MQ_send (op->set->cs->mq,
1415                     ev);
1416     return;
1417   }
1418
1419   op->state->client_done_sent = GNUNET_YES;
1420
1421   LOG (GNUNET_ERROR_TYPE_INFO,
1422        "Signalling client that union operation is done\n");
1423   ev = GNUNET_MQ_msg (rm,
1424                       GNUNET_MESSAGE_TYPE_SET_RESULT);
1425   rm->request_id = htonl (op->client_request_id);
1426   rm->result_status = htons (GNUNET_SET_STATUS_DONE);
1427   rm->element_type = htons (0);
1428   rm->current_size = GNUNET_htonll (GNUNET_CONTAINER_multihashmap32_size (op->state->key_to_element));
1429   GNUNET_MQ_send (op->set->cs->mq,
1430                   ev);
1431 }
1432
1433
1434 /**
1435  * Tests if the operation is finished, and if so notify.
1436  *
1437  * @param op operation to check
1438  */
1439 static void
1440 maybe_finish (struct Operation *op)
1441 {
1442   unsigned int num_demanded;
1443
1444   num_demanded = GNUNET_CONTAINER_multihashmap_size (op->state->demanded_hashes);
1445
1446   if (PHASE_FINISH_WAITING == op->state->phase)
1447   {
1448     LOG (GNUNET_ERROR_TYPE_DEBUG,
1449          "In PHASE_FINISH_WAITING, pending %u demands\n",
1450          num_demanded);
1451     if (0 == num_demanded)
1452     {
1453       struct GNUNET_MQ_Envelope *ev;
1454
1455       op->state->phase = PHASE_DONE;
1456       ev = GNUNET_MQ_msg_header (GNUNET_MESSAGE_TYPE_SET_UNION_P2P_DONE);
1457       GNUNET_MQ_send (op->mq,
1458                       ev);
1459       /* We now wait until the other peer sends P2P_OVER
1460        * after it got all elements from us. */
1461     }
1462   }
1463   if (PHASE_FINISH_CLOSING == op->state->phase)
1464   {
1465     LOG (GNUNET_ERROR_TYPE_DEBUG,
1466          "In PHASE_FINISH_CLOSING, pending %u demands\n",
1467          num_demanded);
1468     if (0 == num_demanded)
1469     {
1470       op->state->phase = PHASE_DONE;
1471       send_client_done (op);
1472       destroy_channel (op);
1473     }
1474   }
1475 }
1476
1477
1478 /**
1479  * Check an element message from a remote peer.
1480  *
1481  * @param cls the union operation
1482  * @param emsg the message
1483  */
1484 int
1485 check_union_p2p_elements (void *cls,
1486                           const struct GNUNET_SET_ElementMessage *emsg)
1487 {
1488   struct Operation *op = cls;
1489
1490   if (GNUNET_SET_OPERATION_UNION != op->set->operation)
1491   {
1492     GNUNET_break_op (0);
1493     return GNUNET_SYSERR;
1494   }
1495   if (0 == GNUNET_CONTAINER_multihashmap_size (op->state->demanded_hashes))
1496   {
1497     GNUNET_break_op (0);
1498     return GNUNET_SYSERR;
1499   }
1500   return GNUNET_OK;
1501 }
1502
1503
1504 /**
1505  * Handle an element message from a remote peer.
1506  * Sent by the other peer either because we decoded an IBF and placed a demand,
1507  * or because the other peer switched to full set transmission.
1508  *
1509  * @param cls the union operation
1510  * @param emsg the message
1511  */
1512 void
1513 handle_union_p2p_elements (void *cls,
1514                            const struct GNUNET_SET_ElementMessage *emsg)
1515 {
1516   struct Operation *op = cls;
1517   struct ElementEntry *ee;
1518   struct KeyEntry *ke;
1519   uint16_t element_size;
1520
1521   element_size = ntohs (emsg->header.size) - sizeof (struct GNUNET_SET_ElementMessage);
1522   ee = GNUNET_malloc (sizeof (struct ElementEntry) + element_size);
1523   GNUNET_memcpy (&ee[1],
1524                  &emsg[1],
1525                  element_size);
1526   ee->element.size = element_size;
1527   ee->element.data = &ee[1];
1528   ee->element.element_type = ntohs (emsg->element_type);
1529   ee->remote = GNUNET_YES;
1530   GNUNET_SET_element_hash (&ee->element,
1531                            &ee->element_hash);
1532   if (GNUNET_NO ==
1533       GNUNET_CONTAINER_multihashmap_remove (op->state->demanded_hashes,
1534                                             &ee->element_hash,
1535                                             NULL))
1536   {
1537     /* We got something we didn't demand, since it's not in our map. */
1538     GNUNET_break_op (0);
1539     fail_union_operation (op);
1540     return;
1541   }
1542
1543   LOG (GNUNET_ERROR_TYPE_DEBUG,
1544        "Got element (size %u, hash %s) from peer\n",
1545        (unsigned int) element_size,
1546        GNUNET_h2s (&ee->element_hash));
1547
1548   GNUNET_STATISTICS_update (_GSS_statistics,
1549                             "# received elements",
1550                             1,
1551                             GNUNET_NO);
1552   GNUNET_STATISTICS_update (_GSS_statistics,
1553                             "# exchanged elements",
1554                             1,
1555                             GNUNET_NO);
1556
1557   op->state->received_total++;
1558
1559   ke = op_get_element (op, &ee->element_hash);
1560   if (NULL != ke)
1561   {
1562     /* Got repeated element.  Should not happen since
1563      * we track demands. */
1564     GNUNET_STATISTICS_update (_GSS_statistics,
1565                               "# repeated elements",
1566                               1,
1567                               GNUNET_NO);
1568     ke->received = GNUNET_YES;
1569     GNUNET_free (ee);
1570   }
1571   else
1572   {
1573     LOG (GNUNET_ERROR_TYPE_DEBUG,
1574          "Registering new element from remote peer\n");
1575     op->state->received_fresh++;
1576     op_register_element (op, ee, GNUNET_YES);
1577     /* only send results immediately if the client wants it */
1578     switch (op->result_mode)
1579     {
1580       case GNUNET_SET_RESULT_ADDED:
1581         send_client_element (op, &ee->element, GNUNET_SET_STATUS_OK);
1582         break;
1583       case GNUNET_SET_RESULT_SYMMETRIC:
1584         send_client_element (op, &ee->element, GNUNET_SET_STATUS_ADD_LOCAL);
1585         break;
1586       default:
1587         /* Result mode not supported, should have been caught earlier. */
1588         GNUNET_break (0);
1589         break;
1590     }
1591   }
1592
1593   if ( (op->state->received_total > 8) &&
1594        (op->state->received_fresh < op->state->received_total / 3) )
1595   {
1596     /* The other peer gave us lots of old elements, there's something wrong. */
1597     GNUNET_break_op (0);
1598     fail_union_operation (op);
1599     return;
1600   }
1601   GNUNET_CADET_receive_done (op->channel);
1602   maybe_finish (op);
1603 }
1604
1605
1606 /**
1607  * Check a full element message from a remote peer.
1608  *
1609  * @param cls the union operation
1610  * @param emsg the message
1611  */
1612 int
1613 check_union_p2p_full_element (void *cls,
1614                               const struct GNUNET_SET_ElementMessage *emsg)
1615 {
1616   struct Operation *op = cls;
1617
1618   if (GNUNET_SET_OPERATION_UNION != op->set->operation)
1619   {
1620     GNUNET_break_op (0);
1621     return GNUNET_SYSERR;
1622   }
1623   // FIXME: check that we expect full elements here?
1624   return GNUNET_OK;
1625 }
1626
1627
1628 /**
1629  * Handle an element message from a remote peer.
1630  *
1631  * @param cls the union operation
1632  * @param emsg the message
1633  */
1634 void
1635 handle_union_p2p_full_element (void *cls,
1636                                const struct GNUNET_SET_ElementMessage *emsg)
1637 {
1638   struct Operation *op = cls;
1639   struct ElementEntry *ee;
1640   struct KeyEntry *ke;
1641   uint16_t element_size;
1642
1643   element_size = ntohs (emsg->header.size) - sizeof (struct GNUNET_SET_ElementMessage);
1644   ee = GNUNET_malloc (sizeof (struct ElementEntry) + element_size);
1645   GNUNET_memcpy (&ee[1], &emsg[1], element_size);
1646   ee->element.size = element_size;
1647   ee->element.data = &ee[1];
1648   ee->element.element_type = ntohs (emsg->element_type);
1649   ee->remote = GNUNET_YES;
1650   GNUNET_SET_element_hash (&ee->element, &ee->element_hash);
1651
1652   LOG (GNUNET_ERROR_TYPE_DEBUG,
1653        "Got element (full diff, size %u, hash %s) from peer\n",
1654        (unsigned int) element_size,
1655        GNUNET_h2s (&ee->element_hash));
1656
1657   GNUNET_STATISTICS_update (_GSS_statistics,
1658                             "# received elements",
1659                             1,
1660                             GNUNET_NO);
1661   GNUNET_STATISTICS_update (_GSS_statistics,
1662                             "# exchanged elements",
1663                             1,
1664                             GNUNET_NO);
1665
1666   op->state->received_total++;
1667
1668   ke = op_get_element (op, &ee->element_hash);
1669   if (NULL != ke)
1670   {
1671     /* Got repeated element.  Should not happen since
1672      * we track demands. */
1673     GNUNET_STATISTICS_update (_GSS_statistics,
1674                               "# repeated elements",
1675                               1,
1676                               GNUNET_NO);
1677     ke->received = GNUNET_YES;
1678     GNUNET_free (ee);
1679   }
1680   else
1681   {
1682     LOG (GNUNET_ERROR_TYPE_DEBUG,
1683          "Registering new element from remote peer\n");
1684     op->state->received_fresh++;
1685     op_register_element (op, ee, GNUNET_YES);
1686     /* only send results immediately if the client wants it */
1687     switch (op->result_mode)
1688     {
1689       case GNUNET_SET_RESULT_ADDED:
1690         send_client_element (op, &ee->element, GNUNET_SET_STATUS_OK);
1691         break;
1692       case GNUNET_SET_RESULT_SYMMETRIC:
1693         send_client_element (op, &ee->element, GNUNET_SET_STATUS_ADD_LOCAL);
1694         break;
1695       default:
1696         /* Result mode not supported, should have been caught earlier. */
1697         GNUNET_break (0);
1698         break;
1699     }
1700   }
1701
1702   if ( (GNUNET_YES == op->byzantine) &&
1703        (op->state->received_total > 384 + op->state->received_fresh * 4) &&
1704        (op->state->received_fresh < op->state->received_total / 6) )
1705   {
1706     /* The other peer gave us lots of old elements, there's something wrong. */
1707     LOG (GNUNET_ERROR_TYPE_ERROR,
1708          "Other peer sent only %llu/%llu fresh elements, failing operation\n",
1709          (unsigned long long) op->state->received_fresh,
1710          (unsigned long long) op->state->received_total);
1711     GNUNET_break_op (0);
1712     fail_union_operation (op);
1713     return;
1714   }
1715   GNUNET_CADET_receive_done (op->channel);
1716 }
1717
1718
1719 /**
1720  * Send offers (for GNUNET_Hash-es) in response
1721  * to inquiries (for IBF_Key-s).
1722  *
1723  * @param cls the union operation
1724  * @param msg the message
1725  */
1726 int
1727 check_union_p2p_inquiry (void *cls,
1728                          const struct InquiryMessage *msg)
1729 {
1730   struct Operation *op = cls;
1731   unsigned int num_keys;
1732
1733   if (GNUNET_SET_OPERATION_UNION != op->set->operation)
1734   {
1735     GNUNET_break_op (0);
1736     return GNUNET_SYSERR;
1737   }
1738   if (op->state->phase != PHASE_INVENTORY_PASSIVE)
1739   {
1740     GNUNET_break_op (0);
1741     return GNUNET_SYSERR;
1742   }
1743   num_keys = (ntohs (msg->header.size) - sizeof (struct InquiryMessage))
1744     / sizeof (struct IBF_Key);
1745   if ((ntohs (msg->header.size) - sizeof (struct InquiryMessage))
1746       != num_keys * sizeof (struct IBF_Key))
1747   {
1748     GNUNET_break_op (0);
1749     return GNUNET_SYSERR;
1750   }
1751   return GNUNET_OK;
1752 }
1753
1754
1755 /**
1756  * Send offers (for GNUNET_Hash-es) in response
1757  * to inquiries (for IBF_Key-s).
1758  *
1759  * @param cls the union operation
1760  * @param msg the message
1761  */
1762 void
1763 handle_union_p2p_inquiry (void *cls,
1764                           const struct InquiryMessage *msg)
1765 {
1766   struct Operation *op = cls;
1767   const struct IBF_Key *ibf_key;
1768   unsigned int num_keys;
1769
1770   LOG (GNUNET_ERROR_TYPE_DEBUG,
1771        "Received union inquiry\n");
1772   num_keys = (ntohs (msg->header.size) - sizeof (struct InquiryMessage))
1773     / sizeof (struct IBF_Key);
1774   ibf_key = (const struct IBF_Key *) &msg[1];
1775   while (0 != num_keys--)
1776   {
1777     struct IBF_Key unsalted_key;
1778
1779     unsalt_key (ibf_key,
1780                 ntohl (msg->salt),
1781                 &unsalted_key);
1782     send_offers_for_key (op,
1783                          unsalted_key);
1784     ibf_key++;
1785   }
1786   GNUNET_CADET_receive_done (op->channel);
1787 }
1788
1789
1790 /**
1791  * Iterator over hash map entries, called to
1792  * destroy the linked list of colliding ibf key entries.
1793  *
1794  * @param cls closure
1795  * @param key current key code
1796  * @param value value in the hash map
1797  * @return #GNUNET_YES if we should continue to iterate,
1798  *         #GNUNET_NO if not.
1799  */
1800 static int
1801 send_missing_full_elements_iter (void *cls,
1802                                  uint32_t key,
1803                                  void *value)
1804 {
1805   struct Operation *op = cls;
1806   struct KeyEntry *ke = value;
1807   struct GNUNET_MQ_Envelope *ev;
1808   struct GNUNET_SET_ElementMessage *emsg;
1809   struct ElementEntry *ee = ke->element;
1810
1811   if (GNUNET_YES == ke->received)
1812     return GNUNET_YES;
1813   ev = GNUNET_MQ_msg_extra (emsg,
1814                             ee->element.size,
1815                             GNUNET_MESSAGE_TYPE_SET_UNION_P2P_FULL_ELEMENT);
1816   GNUNET_memcpy (&emsg[1],
1817                  ee->element.data,
1818                  ee->element.size);
1819   emsg->element_type = htons (ee->element.element_type);
1820   GNUNET_MQ_send (op->mq,
1821                   ev);
1822   return GNUNET_YES;
1823 }
1824
1825
1826 /**
1827  * Handle a request for full set transmission.
1828  *
1829  * @parem cls closure, a set union operation
1830  * @param mh the demand message
1831  */
1832 void
1833 handle_union_p2p_request_full (void *cls,
1834                                const struct GNUNET_MessageHeader *mh)
1835 {
1836   struct Operation *op = cls;
1837
1838   LOG (GNUNET_ERROR_TYPE_DEBUG,
1839        "Received request for full set transmission\n");
1840   if (GNUNET_SET_OPERATION_UNION != op->set->operation)
1841   {
1842     GNUNET_break_op (0);
1843     fail_union_operation (op);
1844     return;
1845   }
1846   if (PHASE_EXPECT_IBF != op->state->phase)
1847   {
1848     GNUNET_break_op (0);
1849     fail_union_operation (op);
1850     return;
1851   }
1852
1853   // FIXME: we need to check that our set is larger than the
1854   // byzantine_lower_bound by some threshold
1855   send_full_set (op);
1856   GNUNET_CADET_receive_done (op->channel);
1857 }
1858
1859
1860 /**
1861  * Handle a "full done" message.
1862  *
1863  * @parem cls closure, a set union operation
1864  * @param mh the demand message
1865  */
1866 void
1867 handle_union_p2p_full_done (void *cls,
1868                             const struct GNUNET_MessageHeader *mh)
1869 {
1870   struct Operation *op = cls;
1871
1872   switch (op->state->phase)
1873   {
1874   case PHASE_EXPECT_IBF:
1875     {
1876       struct GNUNET_MQ_Envelope *ev;
1877
1878       LOG (GNUNET_ERROR_TYPE_DEBUG,
1879            "got FULL DONE, sending elements that other peer is missing\n");
1880
1881       /* send all the elements that did not come from the remote peer */
1882       GNUNET_CONTAINER_multihashmap32_iterate (op->state->key_to_element,
1883                                                &send_missing_full_elements_iter,
1884                                                op);
1885
1886       ev = GNUNET_MQ_msg_header (GNUNET_MESSAGE_TYPE_SET_UNION_P2P_FULL_DONE);
1887       GNUNET_MQ_send (op->mq,
1888                       ev);
1889       op->state->phase = PHASE_DONE;
1890       /* we now wait until the other peer sends us the OVER message*/
1891     }
1892     break;
1893   case PHASE_FULL_SENDING:
1894     {
1895       LOG (GNUNET_ERROR_TYPE_DEBUG,
1896            "got FULL DONE, finishing\n");
1897       /* We sent the full set, and got the response for that.  We're done. */
1898       op->state->phase = PHASE_DONE;
1899       GNUNET_CADET_receive_done (op->channel);
1900       send_client_done (op);
1901       destroy_channel (op);
1902       return;
1903     }
1904     break;
1905   default:
1906     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1907                 "Handle full done phase is %u\n",
1908                 (unsigned) op->state->phase);
1909     GNUNET_break_op (0);
1910     fail_union_operation (op);
1911     return;
1912   }
1913   GNUNET_CADET_receive_done (op->channel);
1914 }
1915
1916
1917 /**
1918  * Check a demand by the other peer for elements based on a list
1919  * of `struct GNUNET_HashCode`s.
1920  *
1921  * @parem cls closure, a set union operation
1922  * @param mh the demand message
1923  * @return #GNUNET_OK if @a mh is well-formed
1924  */
1925 int
1926 check_union_p2p_demand (void *cls,
1927                         const struct GNUNET_MessageHeader *mh)
1928 {
1929   struct Operation *op = cls;
1930   unsigned int num_hashes;
1931
1932   if (GNUNET_SET_OPERATION_UNION != op->set->operation)
1933   {
1934     GNUNET_break_op (0);
1935     return GNUNET_SYSERR;
1936   }
1937   num_hashes = (ntohs (mh->size) - sizeof (struct GNUNET_MessageHeader))
1938     / sizeof (struct GNUNET_HashCode);
1939   if ((ntohs (mh->size) - sizeof (struct GNUNET_MessageHeader))
1940       != num_hashes * sizeof (struct GNUNET_HashCode))
1941   {
1942     GNUNET_break_op (0);
1943     return GNUNET_SYSERR;
1944   }
1945   return GNUNET_OK;
1946 }
1947
1948
1949 /**
1950  * Handle a demand by the other peer for elements based on a list
1951  * of `struct GNUNET_HashCode`s.
1952  *
1953  * @parem cls closure, a set union operation
1954  * @param mh the demand message
1955  */
1956 void
1957 handle_union_p2p_demand (void *cls,
1958                          const struct GNUNET_MessageHeader *mh)
1959 {
1960   struct Operation *op = cls;
1961   struct ElementEntry *ee;
1962   struct GNUNET_SET_ElementMessage *emsg;
1963   const struct GNUNET_HashCode *hash;
1964   unsigned int num_hashes;
1965   struct GNUNET_MQ_Envelope *ev;
1966
1967   num_hashes = (ntohs (mh->size) - sizeof (struct GNUNET_MessageHeader))
1968     / sizeof (struct GNUNET_HashCode);
1969   for (hash = (const struct GNUNET_HashCode *) &mh[1];
1970        num_hashes > 0;
1971        hash++, num_hashes--)
1972   {
1973     ee = GNUNET_CONTAINER_multihashmap_get (op->set->content->elements,
1974                                             hash);
1975     if (NULL == ee)
1976     {
1977       /* Demand for non-existing element. */
1978       GNUNET_break_op (0);
1979       fail_union_operation (op);
1980       return;
1981     }
1982     if (GNUNET_NO == _GSS_is_element_of_operation (ee, op))
1983     {
1984       /* Probably confused lazily copied sets. */
1985       GNUNET_break_op (0);
1986       fail_union_operation (op);
1987       return;
1988     }
1989     ev = GNUNET_MQ_msg_extra (emsg, ee->element.size, GNUNET_MESSAGE_TYPE_SET_P2P_ELEMENTS);
1990     GNUNET_memcpy (&emsg[1], ee->element.data, ee->element.size);
1991     emsg->reserved = htons (0);
1992     emsg->element_type = htons (ee->element.element_type);
1993     LOG (GNUNET_ERROR_TYPE_DEBUG,
1994          "[OP %x] Sending demanded element (size %u, hash %s) to peer\n",
1995          (void *) op,
1996          (unsigned int) ee->element.size,
1997          GNUNET_h2s (&ee->element_hash));
1998     GNUNET_MQ_send (op->mq, ev);
1999     GNUNET_STATISTICS_update (_GSS_statistics,
2000                               "# exchanged elements",
2001                               1,
2002                               GNUNET_NO);
2003
2004     switch (op->result_mode)
2005     {
2006       case GNUNET_SET_RESULT_ADDED:
2007         /* Nothing to do. */
2008         break;
2009       case GNUNET_SET_RESULT_SYMMETRIC:
2010         send_client_element (op, &ee->element, GNUNET_SET_STATUS_ADD_REMOTE);
2011         break;
2012       default:
2013         /* Result mode not supported, should have been caught earlier. */
2014         GNUNET_break (0);
2015         break;
2016     }
2017   }
2018   GNUNET_CADET_receive_done (op->channel);
2019 }
2020
2021
2022 /**
2023  * Check offer (of `struct GNUNET_HashCode`s).
2024  *
2025  * @param cls the union operation
2026  * @param mh the message
2027  * @return #GNUNET_OK if @a mh is well-formed
2028  */
2029 int
2030 check_union_p2p_offer (void *cls,
2031                         const struct GNUNET_MessageHeader *mh)
2032 {
2033   struct Operation *op = cls;
2034   unsigned int num_hashes;
2035
2036   if (GNUNET_SET_OPERATION_UNION != op->set->operation)
2037   {
2038     GNUNET_break_op (0);
2039     return GNUNET_SYSERR;
2040   }
2041   /* look up elements and send them */
2042   if ( (op->state->phase != PHASE_INVENTORY_PASSIVE) &&
2043        (op->state->phase != PHASE_INVENTORY_ACTIVE))
2044   {
2045     GNUNET_break_op (0);
2046     return GNUNET_SYSERR;
2047   }
2048   num_hashes = (ntohs (mh->size) - sizeof (struct GNUNET_MessageHeader))
2049     / sizeof (struct GNUNET_HashCode);
2050   if ((ntohs (mh->size) - sizeof (struct GNUNET_MessageHeader)) !=
2051       num_hashes * sizeof (struct GNUNET_HashCode))
2052   {
2053     GNUNET_break_op (0);
2054     return GNUNET_SYSERR;
2055   }
2056   return GNUNET_OK;
2057 }
2058
2059
2060 /**
2061  * Handle offers (of `struct GNUNET_HashCode`s) and
2062  * respond with demands (of `struct GNUNET_HashCode`s).
2063  *
2064  * @param cls the union operation
2065  * @param mh the message
2066  */
2067 void
2068 handle_union_p2p_offer (void *cls,
2069                         const struct GNUNET_MessageHeader *mh)
2070 {
2071   struct Operation *op = cls;
2072   const struct GNUNET_HashCode *hash;
2073   unsigned int num_hashes;
2074
2075   num_hashes = (ntohs (mh->size) - sizeof (struct GNUNET_MessageHeader))
2076     / sizeof (struct GNUNET_HashCode);
2077   for (hash = (const struct GNUNET_HashCode *) &mh[1];
2078        num_hashes > 0;
2079        hash++, num_hashes--)
2080   {
2081     struct ElementEntry *ee;
2082     struct GNUNET_MessageHeader *demands;
2083     struct GNUNET_MQ_Envelope *ev;
2084
2085     ee = GNUNET_CONTAINER_multihashmap_get (op->set->content->elements,
2086                                             hash);
2087     if (NULL != ee)
2088       if (GNUNET_YES == _GSS_is_element_of_operation (ee, op))
2089         continue;
2090
2091     if (GNUNET_YES ==
2092         GNUNET_CONTAINER_multihashmap_contains (op->state->demanded_hashes,
2093                                                 hash))
2094     {
2095       LOG (GNUNET_ERROR_TYPE_DEBUG,
2096            "Skipped sending duplicate demand\n");
2097       continue;
2098     }
2099
2100     GNUNET_assert (GNUNET_OK ==
2101                    GNUNET_CONTAINER_multihashmap_put (op->state->demanded_hashes,
2102                                                       hash,
2103                                                       NULL,
2104                                                       GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST));
2105
2106     LOG (GNUNET_ERROR_TYPE_DEBUG,
2107          "[OP %x] Requesting element (hash %s)\n",
2108          (void *) op, GNUNET_h2s (hash));
2109     ev = GNUNET_MQ_msg_header_extra (demands,
2110                                      sizeof (struct GNUNET_HashCode),
2111                                      GNUNET_MESSAGE_TYPE_SET_UNION_P2P_DEMAND);
2112     GNUNET_memcpy (&demands[1],
2113                    hash,
2114                    sizeof (struct GNUNET_HashCode));
2115     GNUNET_MQ_send (op->mq, ev);
2116   }
2117   GNUNET_CADET_receive_done (op->channel);
2118 }
2119
2120
2121 /**
2122  * Handle a done message from a remote peer
2123  *
2124  * @param cls the union operation
2125  * @param mh the message
2126  */
2127 void
2128 handle_union_p2p_done (void *cls,
2129                        const struct GNUNET_MessageHeader *mh)
2130 {
2131   struct Operation *op = cls;
2132
2133   if (GNUNET_SET_OPERATION_UNION != op->set->operation)
2134   {
2135     GNUNET_break_op (0);
2136     fail_union_operation (op);
2137     return;
2138   }
2139   switch (op->state->phase)
2140   {
2141   case PHASE_INVENTORY_PASSIVE:
2142     /* We got all requests, but still have to send our elements in response. */
2143     op->state->phase = PHASE_FINISH_WAITING;
2144
2145     LOG (GNUNET_ERROR_TYPE_DEBUG,
2146          "got DONE (as passive partner), waiting for our demands to be satisfied\n");
2147     /* The active peer is done sending offers
2148      * and inquiries.  This means that all
2149      * our responses to that (demands and offers)
2150      * must be in flight (queued or in mesh).
2151      *
2152      * We should notify the active peer once
2153      * all our demands are satisfied, so that the active
2154      * peer can quit if we gave him everything.
2155      */
2156     GNUNET_CADET_receive_done (op->channel);
2157     maybe_finish (op);
2158     return;
2159   case PHASE_INVENTORY_ACTIVE:
2160     LOG (GNUNET_ERROR_TYPE_DEBUG,
2161          "got DONE (as active partner), waiting to finish\n");
2162     /* All demands of the other peer are satisfied,
2163      * and we processed all offers, thus we know
2164      * exactly what our demands must be.
2165      *
2166      * We'll close the channel
2167      * to the other peer once our demands are met.
2168      */
2169     op->state->phase = PHASE_FINISH_CLOSING;
2170     GNUNET_CADET_receive_done (op->channel);
2171     maybe_finish (op);
2172     return;
2173   default:
2174     GNUNET_break_op (0);
2175     fail_union_operation (op);
2176     return;
2177   }
2178 }
2179
2180 /**
2181  * Handle a over message from a remote peer
2182  *
2183  * @param cls the union operation
2184  * @param mh the message
2185  */
2186 void
2187 handle_union_p2p_over (void *cls,
2188                        const struct GNUNET_MessageHeader *mh)
2189 {
2190   send_client_done (cls);
2191 }
2192
2193
2194 /**
2195  * Initiate operation to evaluate a set union with a remote peer.
2196  *
2197  * @param op operation to perform (to be initialized)
2198  * @param opaque_context message to be transmitted to the listener
2199  *        to convince him to accept, may be NULL
2200  */
2201 static struct OperationState *
2202 union_evaluate (struct Operation *op,
2203                 const struct GNUNET_MessageHeader *opaque_context)
2204 {
2205   struct OperationState *state;
2206   struct GNUNET_MQ_Envelope *ev;
2207   struct OperationRequestMessage *msg;
2208
2209   ev = GNUNET_MQ_msg_nested_mh (msg,
2210                                 GNUNET_MESSAGE_TYPE_SET_P2P_OPERATION_REQUEST,
2211                                 opaque_context);
2212   if (NULL == ev)
2213   {
2214     /* the context message is too large */
2215     GNUNET_break (0);
2216     return NULL;
2217   }
2218   state = GNUNET_new (struct OperationState);
2219   state->demanded_hashes = GNUNET_CONTAINER_multihashmap_create (32,
2220                                                                  GNUNET_NO);
2221   /* copy the current generation's strata estimator for this operation */
2222   state->se = strata_estimator_dup (op->set->state->se);
2223   /* we started the operation, thus we have to send the operation request */
2224   state->phase = PHASE_EXPECT_SE;
2225   state->salt_receive = state->salt_send = 42; // FIXME?????
2226   LOG (GNUNET_ERROR_TYPE_DEBUG,
2227        "Initiating union operation evaluation\n");
2228   GNUNET_STATISTICS_update (_GSS_statistics,
2229                             "# of total union operations",
2230                             1,
2231                             GNUNET_NO);
2232   GNUNET_STATISTICS_update (_GSS_statistics,
2233                             "# of initiated union operations",
2234                             1,
2235                             GNUNET_NO);
2236   msg->operation = htonl (GNUNET_SET_OPERATION_UNION);
2237   GNUNET_MQ_send (op->mq,
2238                   ev);
2239
2240   if (NULL != opaque_context)
2241     LOG (GNUNET_ERROR_TYPE_DEBUG,
2242          "sent op request with context message\n");
2243   else
2244     LOG (GNUNET_ERROR_TYPE_DEBUG,
2245          "sent op request without context message\n");
2246
2247   op->state = state;
2248   initialize_key_to_element (op);
2249   state->initial_size = GNUNET_CONTAINER_multihashmap32_size (state->key_to_element);
2250   return state;
2251 }
2252
2253
2254 /**
2255  * Accept an union operation request from a remote peer.
2256  * Only initializes the private operation state.
2257  *
2258  * @param op operation that will be accepted as a union operation
2259  */
2260 static struct OperationState *
2261 union_accept (struct Operation *op)
2262 {
2263   struct OperationState *state;
2264   const struct StrataEstimator *se;
2265   struct GNUNET_MQ_Envelope *ev;
2266   struct StrataEstimatorMessage *strata_msg;
2267   char *buf;
2268   size_t len;
2269   uint16_t type;
2270
2271   LOG (GNUNET_ERROR_TYPE_DEBUG,
2272        "accepting set union operation\n");
2273   GNUNET_STATISTICS_update (_GSS_statistics,
2274                             "# of accepted union operations",
2275                             1,
2276                             GNUNET_NO);
2277   GNUNET_STATISTICS_update (_GSS_statistics,
2278                             "# of total union operations",
2279                             1,
2280                             GNUNET_NO);
2281
2282   state = GNUNET_new (struct OperationState);
2283   state->se = strata_estimator_dup (op->set->state->se);
2284   state->demanded_hashes = GNUNET_CONTAINER_multihashmap_create (32,
2285                                                                  GNUNET_NO);
2286   state->salt_receive = state->salt_send = 42; // FIXME?????
2287   op->state = state;
2288   initialize_key_to_element (op);
2289   state->initial_size = GNUNET_CONTAINER_multihashmap32_size (state->key_to_element);
2290
2291   /* kick off the operation */
2292   se = state->se;
2293   buf = GNUNET_malloc (se->strata_count * IBF_BUCKET_SIZE * se->ibf_size);
2294   len = strata_estimator_write (se,
2295                                 buf);
2296   if (len < se->strata_count * IBF_BUCKET_SIZE * se->ibf_size)
2297     type = GNUNET_MESSAGE_TYPE_SET_UNION_P2P_SEC;
2298   else
2299     type = GNUNET_MESSAGE_TYPE_SET_UNION_P2P_SE;
2300   ev = GNUNET_MQ_msg_extra (strata_msg,
2301                             len,
2302                             type);
2303   GNUNET_memcpy (&strata_msg[1],
2304                  buf,
2305                  len);
2306   GNUNET_free (buf);
2307   strata_msg->set_size
2308     = GNUNET_htonll (GNUNET_CONTAINER_multihashmap_size (op->set->content->elements));
2309   GNUNET_MQ_send (op->mq,
2310                   ev);
2311   state->phase = PHASE_EXPECT_IBF;
2312   return state;
2313 }
2314
2315
2316 /**
2317  * Create a new set supporting the union operation
2318  *
2319  * We maintain one strata estimator per set and then manipulate it over the
2320  * lifetime of the set, as recreating a strata estimator would be expensive.
2321  *
2322  * @return the newly created set, NULL on error
2323  */
2324 static struct SetState *
2325 union_set_create (void)
2326 {
2327   struct SetState *set_state;
2328
2329   LOG (GNUNET_ERROR_TYPE_DEBUG,
2330        "union set created\n");
2331   set_state = GNUNET_new (struct SetState);
2332   set_state->se = strata_estimator_create (SE_STRATA_COUNT,
2333                                            SE_IBF_SIZE, SE_IBF_HASH_NUM);
2334   if (NULL == set_state->se)
2335   {
2336     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
2337                 "Failed to allocate strata estimator\n");
2338     GNUNET_free (set_state);
2339     return NULL;
2340   }
2341   return set_state;
2342 }
2343
2344
2345 /**
2346  * Add the element from the given element message to the set.
2347  *
2348  * @param set_state state of the set want to add to
2349  * @param ee the element to add to the set
2350  */
2351 static void
2352 union_add (struct SetState *set_state,
2353            struct ElementEntry *ee)
2354 {
2355   strata_estimator_insert (set_state->se,
2356                            get_ibf_key (&ee->element_hash));
2357 }
2358
2359
2360 /**
2361  * Remove the element given in the element message from the set.
2362  * Only marks the element as removed, so that older set operations can still exchange it.
2363  *
2364  * @param set_state state of the set to remove from
2365  * @param ee set element to remove
2366  */
2367 static void
2368 union_remove (struct SetState *set_state,
2369               struct ElementEntry *ee)
2370 {
2371   strata_estimator_remove (set_state->se,
2372                            get_ibf_key (&ee->element_hash));
2373 }
2374
2375
2376 /**
2377  * Destroy a set that supports the union operation.
2378  *
2379  * @param set_state the set to destroy
2380  */
2381 static void
2382 union_set_destroy (struct SetState *set_state)
2383 {
2384   if (NULL != set_state->se)
2385   {
2386     strata_estimator_destroy (set_state->se);
2387     set_state->se = NULL;
2388   }
2389   GNUNET_free (set_state);
2390 }
2391
2392
2393 /**
2394  * Copy union-specific set state.
2395  *
2396  * @param state source state for copying the union state
2397  * @return a copy of the union-specific set state
2398  */
2399 static struct SetState *
2400 union_copy_state (struct SetState *state)
2401 {
2402   struct SetState *new_state;
2403
2404   GNUNET_assert ( (NULL != state) &&
2405                   (NULL != state->se) );
2406   new_state = GNUNET_new (struct SetState);
2407   new_state->se = strata_estimator_dup (state->se);
2408
2409   return new_state;
2410 }
2411
2412
2413 /**
2414  * Handle case where channel went down for an operation.
2415  *
2416  * @param op operation that lost the channel
2417  */
2418 static void
2419 union_channel_death (struct Operation *op)
2420 {
2421   send_client_done (op);
2422   _GSS_operation_destroy (op,
2423                           GNUNET_YES);
2424 }
2425
2426
2427 /**
2428  * Get the table with implementing functions for
2429  * set union.
2430  *
2431  * @return the operation specific VTable
2432  */
2433 const struct SetVT *
2434 _GSS_union_vt ()
2435 {
2436   static const struct SetVT union_vt = {
2437     .create = &union_set_create,
2438     .add = &union_add,
2439     .remove = &union_remove,
2440     .destroy_set = &union_set_destroy,
2441     .evaluate = &union_evaluate,
2442     .accept = &union_accept,
2443     .cancel = &union_op_cancel,
2444     .copy_state = &union_copy_state,
2445     .channel_death = &union_channel_death
2446   };
2447
2448   return &union_vt;
2449 }