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_ERROR,
 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_CRYPTO_kdf (&key, sizeof (key),
 393                      src, sizeof *src,
 394                      &salt, sizeof (salt),
 395                      NULL, 0);
 396   return key;
 397 }
 398
 399
 400 /**
 401  * Context for #op_get_element_iterator
 402  */
 403 struct GetElementContext
 404 {
 405   /**
 406    * FIXME.
 407    */
 408   struct GNUNET_HashCode hash;
 409
 410   /**
 411    * FIXME.
 412    */
 413   struct KeyEntry *k;
 414 };
 415
 416
 417 /**
 418  * Iterator over the mapping from IBF keys to element entries.  Checks if we
 419  * have an element with a given GNUNET_HashCode.
 420  *
 421  * @param cls closure
 422  * @param key current key code
 423  * @param value value in the hash map
 424  * @return #GNUNET_YES if we should search further,
 425  *         #GNUNET_NO if we've found the element.
 426  */
 427 static int
 428 op_get_element_iterator (void *cls,
 429                          uint32_t key,
 430                          void *value)
 431 {
 432   struct GetElementContext *ctx = cls;
 433   struct KeyEntry *k = value;
 434
 435   GNUNET_assert (NULL != k);
 436   if (0 == GNUNET_CRYPTO_hash_cmp (&k->element->element_hash,
 437                                    &ctx->hash))
 438   {
 439     ctx->k = k;
 440     return GNUNET_NO;
 441   }
 442   return GNUNET_YES;
 443 }
 444
 445
 446 /**
 447  * Determine whether the given element is already in the operation's element
 448  * set.
 449  *
 450  * @param op operation that should be tested for 'element_hash'
 451  * @param element_hash hash of the element to look for
 452  * @return #GNUNET_YES if the element has been found, #GNUNET_NO otherwise
 453  */
 454 static struct KeyEntry *
 455 op_get_element (struct Operation *op,
 456                 const struct GNUNET_HashCode *element_hash)
 457 {
 458   int ret;
 459   struct IBF_Key ibf_key;
 460   struct GetElementContext ctx = {{{ 0 }} , 0};
 461
 462   ctx.hash = *element_hash;
 463
 464   ibf_key = get_ibf_key (element_hash);
 465   ret = GNUNET_CONTAINER_multihashmap32_get_multiple (op->state->key_to_element,
 466                                                       (uint32_t) ibf_key.key_val,
 467                                                       op_get_element_iterator,
 468                                                       &ctx);
 469
 470   /* was the iteration aborted because we found the element? */
 471   if (GNUNET_SYSERR == ret)
 472   {
 473     GNUNET_assert (NULL != ctx.k);
 474     return ctx.k;
 475   }
 476   return NULL;
 477 }
 478
 479
 480 /**
 481  * Insert an element into the union operation's
 482  * key-to-element mapping. Takes ownership of 'ee'.
 483  * Note that this does not insert the element in the set,
 484  * only in the operation's key-element mapping.
 485  * This is done to speed up re-tried operations, if some elements
 486  * were transmitted, and then the IBF fails to decode.
 487  *
 488  * XXX: clarify ownership, doesn't sound right.
 489  *
 490  * @param op the union operation
 491  * @param ee the element entry
 492  * @parem received was this element received from the remote peer?
 493  */
 494 static void
 495 op_register_element (struct Operation *op,
 496                      struct ElementEntry *ee,
 497                      int received)
 498 {
 499   struct IBF_Key ibf_key;
 500   struct KeyEntry *k;
 501
 502   ibf_key = get_ibf_key (&ee->element_hash);
 503   k = GNUNET_new (struct KeyEntry);
 504   k->element = ee;
 505   k->ibf_key = ibf_key;
 506   k->received = received;
 507   GNUNET_assert (GNUNET_OK ==
 508                  GNUNET_CONTAINER_multihashmap32_put (op->state->key_to_element,
 509                                                       (uint32_t) ibf_key.key_val,
 510                                                       k,
 511                                                       GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE));
 512 }
 513
 514
 515 /**
 516  * FIXME.
 517  */
 518 static void
 519 salt_key (const struct IBF_Key *k_in,
 520           uint32_t salt,
 521           struct IBF_Key *k_out)
 522 {
 523   int s = salt % 64;
 524   uint64_t x = k_in->key_val;
 525   /* rotate ibf key */
 526   x = (x >> s) | (x << (64 - s));
 527   k_out->key_val = x;
 528 }
 529
 530
 531 /**
 532  * FIXME.
 533  */
 534 static void
 535 unsalt_key (const struct IBF_Key *k_in,
 536             uint32_t salt,
 537             struct IBF_Key *k_out)
 538 {
 539   int s = salt % 64;
 540   uint64_t x = k_in->key_val;
 541   x = (x << s) | (x >> (64 - s));
 542   k_out->key_val = x;
 543 }
 544
 545
 546 /**
 547  * Insert a key into an ibf.
 548  *
 549  * @param cls the ibf
 550  * @param key unused
 551  * @param value the key entry to get the key from
 552  */
 553 static int
 554 prepare_ibf_iterator (void *cls,
 555                       uint32_t key,
 556                       void *value)
 557 {
 558   struct Operation *op = cls;
 559   struct KeyEntry *ke = value;
 560   struct IBF_Key salted_key;
 561
 562   LOG (GNUNET_ERROR_TYPE_DEBUG,
 563        "[OP %x] inserting %lx (hash %s) into ibf\n",
 564        (void *) op,
 565        (unsigned long) ke->ibf_key.key_val,
 566        GNUNET_h2s (&ke->element->element_hash));
 567   salt_key (&ke->ibf_key,
 568             op->state->salt_send,
 569             &salted_key);
 570   ibf_insert (op->state->local_ibf, salted_key);
 571   return GNUNET_YES;
 572 }
 573
 574
 575 /**
 576  * Iterator for initializing the
 577  * key-to-element mapping of a union operation
 578  *
 579  * @param cls the union operation `struct Operation *`
 580  * @param key unused
 581  * @param value the `struct ElementEntry *` to insert
 582  *        into the key-to-element mapping
 583  * @return #GNUNET_YES (to continue iterating)
 584  */
 585 static int
 586 init_key_to_element_iterator (void *cls,
 587                               const struct GNUNET_HashCode *key,
 588                               void *value)
 589 {
 590   struct Operation *op = cls;
 591   struct ElementEntry *ee = value;
 592
 593   /* make sure that the element belongs to the set at the time
 594    * of creating the operation */
 595   if (GNUNET_NO ==
 596       _GSS_is_element_of_operation (ee,
 597                                     op))
 598     return GNUNET_YES;
 599   GNUNET_assert (GNUNET_NO == ee->remote);
 600   op_register_element (op,
 601                        ee,
 602                        GNUNET_NO);
 603   return GNUNET_YES;
 604 }
 605
 606
 607 /**
 608  * Initialize the IBF key to element mapping local to this set
 609  * operation.
 610  *
 611  * @param op the set union operation
 612  */
 613 static void
 614 initialize_key_to_element (struct Operation *op)
 615 {
 616   unsigned int len;
 617
 618   GNUNET_assert (NULL == op->state->key_to_element);
 619   len = GNUNET_CONTAINER_multihashmap_size (op->set->content->elements);
 620   op->state->key_to_element = GNUNET_CONTAINER_multihashmap32_create (len + 1);
 621   GNUNET_CONTAINER_multihashmap_iterate (op->set->content->elements,
 622                                          &init_key_to_element_iterator,
 623                                          op);
 624 }
 625
 626
 627 /**
 628  * Create an ibf with the operation's elements
 629  * of the specified size
 630  *
 631  * @param op the union operation
 632  * @param size size of the ibf to create
 633  * @return #GNUNET_OK on success, #GNUNET_SYSERR on failure
 634  */
 635 static int
 636 prepare_ibf (struct Operation *op,
 637              uint32_t size)
 638 {
 639   GNUNET_assert (NULL != op->state->key_to_element);
 640
 641   if (NULL != op->state->local_ibf)
 642     ibf_destroy (op->state->local_ibf);
 643   op->state->local_ibf = ibf_create (size, SE_IBF_HASH_NUM);
 644   if (NULL == op->state->local_ibf)
 645   {
 646     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 647                 "Failed to allocate local IBF\n");
 648     return GNUNET_SYSERR;
 649   }
 650   GNUNET_CONTAINER_multihashmap32_iterate (op->state->key_to_element,
 651                                            &prepare_ibf_iterator,
 652                                            op);
 653   return GNUNET_OK;
 654 }
 655
 656
 657 /**
 658  * Send an ibf of appropriate size.
 659  *
 660  * Fragments the IBF into multiple messages if necessary.
 661  *
 662  * @param op the union operation
 663  * @param ibf_order order of the ibf to send, size=2^order
 664  * @return #GNUNET_OK on success, #GNUNET_SYSERR on failure
 665  */
 666 static int
 667 send_ibf (struct Operation *op,
 668           uint16_t ibf_order)
 669 {
 670   unsigned int buckets_sent = 0;
 671   struct InvertibleBloomFilter *ibf;
 672
 673   if (GNUNET_OK !=
 674       prepare_ibf (op, 1<<ibf_order))
 675   {
 676     /* allocation failed */
 677     return GNUNET_SYSERR;
 678   }
 679
 680   LOG (GNUNET_ERROR_TYPE_DEBUG,
 681        "sending ibf of size %u\n",
 682        1<<ibf_order);
 683
 684   {
 685     char name[64] = { 0 };
 686     snprintf (name, sizeof (name), "# sent IBF (order %u)", ibf_order);
 687     GNUNET_STATISTICS_update (_GSS_statistics, name, 1, GNUNET_NO);
 688   }
 689
 690   ibf = op->state->local_ibf;
 691
 692   while (buckets_sent < (1 << ibf_order))
 693   {
 694     unsigned int buckets_in_message;
 695     struct GNUNET_MQ_Envelope *ev;
 696     struct IBFMessage *msg;
 697
 698     buckets_in_message = (1 << ibf_order) - buckets_sent;
 699     /* limit to maximum */
 700     if (buckets_in_message > MAX_BUCKETS_PER_MESSAGE)
 701       buckets_in_message = MAX_BUCKETS_PER_MESSAGE;
 702
 703     ev = GNUNET_MQ_msg_extra (msg,
 704                               buckets_in_message * IBF_BUCKET_SIZE,
 705                               GNUNET_MESSAGE_TYPE_SET_UNION_P2P_IBF);
 706     msg->reserved1 = 0;
 707     msg->reserved2 = 0;
 708     msg->order = ibf_order;
 709     msg->offset = htonl (buckets_sent);
 710     msg->salt = htonl (op->state->salt_send);
 711     ibf_write_slice (ibf, buckets_sent,
 712                      buckets_in_message, &msg[1]);
 713     buckets_sent += buckets_in_message;
 714     LOG (GNUNET_ERROR_TYPE_DEBUG,
 715          "ibf chunk size %u, %u/%u sent\n",
 716          buckets_in_message,
 717          buckets_sent,
 718          1<<ibf_order);
 719     GNUNET_MQ_send (op->mq, ev);
 720   }
 721
 722   /* The other peer must decode the IBF, so
 723    * we're passive. */
 724   op->state->phase = PHASE_INVENTORY_PASSIVE;
 725   return GNUNET_OK;
 726 }
 727
 728
 729 /**
 730  * Compute the necessary order of an ibf
 731  * from the size of the symmetric set difference.
 732  *
 733  * @param diff the difference
 734  * @return the required size of the ibf
 735  */
 736 static unsigned int
 737 get_order_from_difference (unsigned int diff)
 738 {
 739   unsigned int ibf_order;
 740
 741   ibf_order = 2;
 742   while ( (1<<ibf_order) < (IBF_ALPHA * diff) ||
 743           ((1<<ibf_order) < SE_IBF_HASH_NUM) )
 744     ibf_order++;
 745   if (ibf_order > MAX_IBF_ORDER)
 746     ibf_order = MAX_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_INFO,
 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_INFO,
 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_INFO,
 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_INFO,
 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_INFO,
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  * Signal to the client that the operation has finished and
1373  * destroy the operation.
1374  *
1375  * @param cls operation to destroy
1376  */
1377 static void
1378 send_done_and_destroy (void *cls)
1379 {
1380   struct Operation *op = cls;
1381   struct GNUNET_MQ_Envelope *ev;
1382   struct GNUNET_SET_ResultMessage *rm;
1383
1384   LOG (GNUNET_ERROR_TYPE_INFO,
1385        "Signalling client that union operation is done\n");
1386   ev = GNUNET_MQ_msg (rm,
1387                       GNUNET_MESSAGE_TYPE_SET_RESULT);
1388   rm->request_id = htonl (op->client_request_id);
1389   rm->result_status = htons (GNUNET_SET_STATUS_DONE);
1390   rm->element_type = htons (0);
1391   rm->current_size = GNUNET_htonll (GNUNET_CONTAINER_multihashmap32_size (op->state->key_to_element));
1392   GNUNET_MQ_send (op->set->cs->mq,
1393                   ev);
1394   /* Will also call the union-specific cancel function. */
1395   _GSS_operation_destroy (op,
1396                           GNUNET_YES);
1397 }
1398
1399
1400 /**
1401  * Tests if the operation is finished, and if so notify.
1402  *
1403  * @param op operation to check
1404  */
1405 static void
1406 maybe_finish (struct Operation *op)
1407 {
1408   unsigned int num_demanded;
1409
1410   num_demanded = GNUNET_CONTAINER_multihashmap_size (op->state->demanded_hashes);
1411
1412   if (PHASE_FINISH_WAITING == op->state->phase)
1413   {
1414     LOG (GNUNET_ERROR_TYPE_DEBUG,
1415          "In PHASE_FINISH_WAITING, pending %u demands\n",
1416          num_demanded);
1417     if (0 == num_demanded)
1418     {
1419       struct GNUNET_MQ_Envelope *ev;
1420
1421       op->state->phase = PHASE_DONE;
1422       ev = GNUNET_MQ_msg_header (GNUNET_MESSAGE_TYPE_SET_UNION_P2P_DONE);
1423       GNUNET_MQ_send (op->mq,
1424                       ev);
1425       /* We now wait until the other peer closes the channel
1426        * after it got all elements from us. */
1427     }
1428   }
1429   if (PHASE_FINISH_CLOSING == op->state->phase)
1430   {
1431     LOG (GNUNET_ERROR_TYPE_DEBUG,
1432          "In PHASE_FINISH_CLOSING, pending %u demands\n",
1433          num_demanded);
1434     if (0 == num_demanded)
1435     {
1436       op->state->phase = PHASE_DONE;
1437       send_done_and_destroy (op);
1438     }
1439   }
1440 }
1441
1442
1443 /**
1444  * Check an element message from a remote peer.
1445  *
1446  * @param cls the union operation
1447  * @param emsg the message
1448  */
1449 int
1450 check_union_p2p_elements (void *cls,
1451                           const struct GNUNET_SET_ElementMessage *emsg)
1452 {
1453   struct Operation *op = cls;
1454
1455   if (GNUNET_SET_OPERATION_UNION != op->set->operation)
1456   {
1457     GNUNET_break_op (0);
1458     return GNUNET_SYSERR;
1459   }
1460   if (0 == GNUNET_CONTAINER_multihashmap_size (op->state->demanded_hashes))
1461   {
1462     GNUNET_break_op (0);
1463     return GNUNET_SYSERR;
1464   }
1465   return GNUNET_OK;
1466 }
1467
1468
1469 /**
1470  * Handle an element message from a remote peer.
1471  * Sent by the other peer either because we decoded an IBF and placed a demand,
1472  * or because the other peer switched to full set transmission.
1473  *
1474  * @param cls the union operation
1475  * @param emsg the message
1476  */
1477 void
1478 handle_union_p2p_elements (void *cls,
1479                            const struct GNUNET_SET_ElementMessage *emsg)
1480 {
1481   struct Operation *op = cls;
1482   struct ElementEntry *ee;
1483   struct KeyEntry *ke;
1484   uint16_t element_size;
1485
1486   element_size = ntohs (emsg->header.size) - sizeof (struct GNUNET_SET_ElementMessage);
1487   ee = GNUNET_malloc (sizeof (struct ElementEntry) + element_size);
1488   GNUNET_memcpy (&ee[1],
1489                  &emsg[1],
1490                  element_size);
1491   ee->element.size = element_size;
1492   ee->element.data = &ee[1];
1493   ee->element.element_type = ntohs (emsg->element_type);
1494   ee->remote = GNUNET_YES;
1495   GNUNET_SET_element_hash (&ee->element,
1496                            &ee->element_hash);
1497   if (GNUNET_NO ==
1498       GNUNET_CONTAINER_multihashmap_remove (op->state->demanded_hashes,
1499                                             &ee->element_hash,
1500                                             NULL))
1501   {
1502     /* We got something we didn't demand, since it's not in our map. */
1503     GNUNET_break_op (0);
1504     fail_union_operation (op);
1505     return;
1506   }
1507
1508   LOG (GNUNET_ERROR_TYPE_DEBUG,
1509        "Got element (size %u, hash %s) from peer\n",
1510        (unsigned int) element_size,
1511        GNUNET_h2s (&ee->element_hash));
1512
1513   GNUNET_STATISTICS_update (_GSS_statistics,
1514                             "# received elements",
1515                             1,
1516                             GNUNET_NO);
1517   GNUNET_STATISTICS_update (_GSS_statistics,
1518                             "# exchanged elements",
1519                             1,
1520                             GNUNET_NO);
1521
1522   op->state->received_total++;
1523
1524   ke = op_get_element (op, &ee->element_hash);
1525   if (NULL != ke)
1526   {
1527     /* Got repeated element.  Should not happen since
1528      * we track demands. */
1529     GNUNET_STATISTICS_update (_GSS_statistics,
1530                               "# repeated elements",
1531                               1,
1532                               GNUNET_NO);
1533     ke->received = GNUNET_YES;
1534     GNUNET_free (ee);
1535   }
1536   else
1537   {
1538     LOG (GNUNET_ERROR_TYPE_DEBUG,
1539          "Registering new element from remote peer\n");
1540     op->state->received_fresh++;
1541     op_register_element (op, ee, GNUNET_YES);
1542     /* only send results immediately if the client wants it */
1543     switch (op->result_mode)
1544     {
1545       case GNUNET_SET_RESULT_ADDED:
1546         send_client_element (op, &ee->element, GNUNET_SET_STATUS_OK);
1547         break;
1548       case GNUNET_SET_RESULT_SYMMETRIC:
1549         send_client_element (op, &ee->element, GNUNET_SET_STATUS_ADD_LOCAL);
1550         break;
1551       default:
1552         /* Result mode not supported, should have been caught earlier. */
1553         GNUNET_break (0);
1554         break;
1555     }
1556   }
1557
1558   if ( (op->state->received_total > 8) &&
1559        (op->state->received_fresh < op->state->received_total / 3) )
1560   {
1561     /* The other peer gave us lots of old elements, there's something wrong. */
1562     GNUNET_break_op (0);
1563     fail_union_operation (op);
1564     return;
1565   }
1566   GNUNET_CADET_receive_done (op->channel);
1567   maybe_finish (op);
1568 }
1569
1570
1571 /**
1572  * Check a full element message from a remote peer.
1573  *
1574  * @param cls the union operation
1575  * @param emsg the message
1576  */
1577 int
1578 check_union_p2p_full_element (void *cls,
1579                               const struct GNUNET_SET_ElementMessage *emsg)
1580 {
1581   struct Operation *op = cls;
1582
1583   if (GNUNET_SET_OPERATION_UNION != op->set->operation)
1584   {
1585     GNUNET_break_op (0);
1586     return GNUNET_SYSERR;
1587   }
1588   // FIXME: check that we expect full elements here?
1589   return GNUNET_OK;
1590 }
1591
1592
1593 /**
1594  * Handle an element message from a remote peer.
1595  *
1596  * @param cls the union operation
1597  * @param emsg the message
1598  */
1599 void
1600 handle_union_p2p_full_element (void *cls,
1601                                const struct GNUNET_SET_ElementMessage *emsg)
1602 {
1603   struct Operation *op = cls;
1604   struct ElementEntry *ee;
1605   struct KeyEntry *ke;
1606   uint16_t element_size;
1607
1608   element_size = ntohs (emsg->header.size) - sizeof (struct GNUNET_SET_ElementMessage);
1609   ee = GNUNET_malloc (sizeof (struct ElementEntry) + element_size);
1610   GNUNET_memcpy (&ee[1], &emsg[1], element_size);
1611   ee->element.size = element_size;
1612   ee->element.data = &ee[1];
1613   ee->element.element_type = ntohs (emsg->element_type);
1614   ee->remote = GNUNET_YES;
1615   GNUNET_SET_element_hash (&ee->element, &ee->element_hash);
1616
1617   LOG (GNUNET_ERROR_TYPE_DEBUG,
1618        "Got element (full diff, size %u, hash %s) from peer\n",
1619        (unsigned int) element_size,
1620        GNUNET_h2s (&ee->element_hash));
1621
1622   GNUNET_STATISTICS_update (_GSS_statistics,
1623                             "# received elements",
1624                             1,
1625                             GNUNET_NO);
1626   GNUNET_STATISTICS_update (_GSS_statistics,
1627                             "# exchanged elements",
1628                             1,
1629                             GNUNET_NO);
1630
1631   op->state->received_total++;
1632
1633   ke = op_get_element (op, &ee->element_hash);
1634   if (NULL != ke)
1635   {
1636     /* Got repeated element.  Should not happen since
1637      * we track demands. */
1638     GNUNET_STATISTICS_update (_GSS_statistics,
1639                               "# repeated elements",
1640                               1,
1641                               GNUNET_NO);
1642     ke->received = GNUNET_YES;
1643     GNUNET_free (ee);
1644   }
1645   else
1646   {
1647     LOG (GNUNET_ERROR_TYPE_DEBUG,
1648          "Registering new element from remote peer\n");
1649     op->state->received_fresh++;
1650     op_register_element (op, ee, GNUNET_YES);
1651     /* only send results immediately if the client wants it */
1652     switch (op->result_mode)
1653     {
1654       case GNUNET_SET_RESULT_ADDED:
1655         send_client_element (op, &ee->element, GNUNET_SET_STATUS_OK);
1656         break;
1657       case GNUNET_SET_RESULT_SYMMETRIC:
1658         send_client_element (op, &ee->element, GNUNET_SET_STATUS_ADD_LOCAL);
1659         break;
1660       default:
1661         /* Result mode not supported, should have been caught earlier. */
1662         GNUNET_break (0);
1663         break;
1664     }
1665   }
1666
1667   if ( (GNUNET_YES == op->byzantine) &&
1668        (op->state->received_total > 384 + op->state->received_fresh * 4) &&
1669        (op->state->received_fresh < op->state->received_total / 6) )
1670   {
1671     /* The other peer gave us lots of old elements, there's something wrong. */
1672     LOG (GNUNET_ERROR_TYPE_ERROR,
1673          "Other peer sent only %llu/%llu fresh elements, failing operation\n",
1674          (unsigned long long) op->state->received_fresh,
1675          (unsigned long long) op->state->received_total);
1676     GNUNET_break_op (0);
1677     fail_union_operation (op);
1678     return;
1679   }
1680   GNUNET_CADET_receive_done (op->channel);
1681 }
1682
1683
1684 /**
1685  * Send offers (for GNUNET_Hash-es) in response
1686  * to inquiries (for IBF_Key-s).
1687  *
1688  * @param cls the union operation
1689  * @param msg the message
1690  */
1691 int
1692 check_union_p2p_inquiry (void *cls,
1693                          const struct InquiryMessage *msg)
1694 {
1695   struct Operation *op = cls;
1696   unsigned int num_keys;
1697
1698   if (GNUNET_SET_OPERATION_UNION != op->set->operation)
1699   {
1700     GNUNET_break_op (0);
1701     return GNUNET_SYSERR;
1702   }
1703   if (op->state->phase != PHASE_INVENTORY_PASSIVE)
1704   {
1705     GNUNET_break_op (0);
1706     return GNUNET_SYSERR;
1707   }
1708   num_keys = (ntohs (msg->header.size) - sizeof (struct InquiryMessage))
1709     / sizeof (struct IBF_Key);
1710   if ((ntohs (msg->header.size) - sizeof (struct InquiryMessage))
1711       != num_keys * sizeof (struct IBF_Key))
1712   {
1713     GNUNET_break_op (0);
1714     return GNUNET_SYSERR;
1715   }
1716   return GNUNET_OK;
1717 }
1718
1719
1720 /**
1721  * Send offers (for GNUNET_Hash-es) in response
1722  * to inquiries (for IBF_Key-s).
1723  *
1724  * @param cls the union operation
1725  * @param msg the message
1726  */
1727 void
1728 handle_union_p2p_inquiry (void *cls,
1729                           const struct InquiryMessage *msg)
1730 {
1731   struct Operation *op = cls;
1732   const struct IBF_Key *ibf_key;
1733   unsigned int num_keys;
1734
1735   LOG (GNUNET_ERROR_TYPE_INFO,
1736        "Received union inquiry\n");
1737   num_keys = (ntohs (msg->header.size) - sizeof (struct InquiryMessage))
1738     / sizeof (struct IBF_Key);
1739   ibf_key = (const struct IBF_Key *) &msg[1];
1740   while (0 != num_keys--)
1741   {
1742     struct IBF_Key unsalted_key;
1743
1744     unsalt_key (ibf_key,
1745                 ntohl (msg->salt),
1746                 &unsalted_key);
1747     send_offers_for_key (op,
1748                          unsalted_key);
1749     ibf_key++;
1750   }
1751   GNUNET_CADET_receive_done (op->channel);
1752 }
1753
1754
1755 /**
1756  * Iterator over hash map entries, called to
1757  * destroy the linked list of colliding ibf key entries.
1758  *
1759  * @param cls closure
1760  * @param key current key code
1761  * @param value value in the hash map
1762  * @return #GNUNET_YES if we should continue to iterate,
1763  *         #GNUNET_NO if not.
1764  */
1765 static int
1766 send_missing_full_elements_iter (void *cls,
1767                                  uint32_t key,
1768                                  void *value)
1769 {
1770   struct Operation *op = cls;
1771   struct KeyEntry *ke = value;
1772   struct GNUNET_MQ_Envelope *ev;
1773   struct GNUNET_SET_ElementMessage *emsg;
1774   struct ElementEntry *ee = ke->element;
1775
1776   if (GNUNET_YES == ke->received)
1777     return GNUNET_YES;
1778   ev = GNUNET_MQ_msg_extra (emsg,
1779                             ee->element.size,
1780                             GNUNET_MESSAGE_TYPE_SET_UNION_P2P_FULL_ELEMENT);
1781   GNUNET_memcpy (&emsg[1],
1782                  ee->element.data,
1783                  ee->element.size);
1784   emsg->element_type = htons (ee->element.element_type);
1785   GNUNET_MQ_send (op->mq,
1786                   ev);
1787   return GNUNET_YES;
1788 }
1789
1790
1791 /**
1792  * Handle a request for full set transmission.
1793  *
1794  * @parem cls closure, a set union operation
1795  * @param mh the demand message
1796  */
1797 void
1798 handle_union_p2p_request_full (void *cls,
1799                                const struct GNUNET_MessageHeader *mh)
1800 {
1801   struct Operation *op = cls;
1802
1803   LOG (GNUNET_ERROR_TYPE_INFO,
1804        "Received request for full set transmission\n");
1805   if (GNUNET_SET_OPERATION_UNION != op->set->operation)
1806   {
1807     GNUNET_break_op (0);
1808     fail_union_operation (op);
1809     return;
1810   }
1811   if (PHASE_EXPECT_IBF != op->state->phase)
1812   {
1813     GNUNET_break_op (0);
1814     fail_union_operation (op);
1815     return;
1816   }
1817
1818   // FIXME: we need to check that our set is larger than the
1819   // byzantine_lower_bound by some threshold
1820   send_full_set (op);
1821   GNUNET_CADET_receive_done (op->channel);
1822 }
1823
1824
1825 /**
1826  * Handle a "full done" message.
1827  *
1828  * @parem cls closure, a set union operation
1829  * @param mh the demand message
1830  */
1831 void
1832 handle_union_p2p_full_done (void *cls,
1833                             const struct GNUNET_MessageHeader *mh)
1834 {
1835   struct Operation *op = cls;
1836
1837   switch (op->state->phase)
1838   {
1839   case PHASE_EXPECT_IBF:
1840     {
1841       struct GNUNET_MQ_Envelope *ev;
1842
1843       LOG (GNUNET_ERROR_TYPE_DEBUG,
1844            "got FULL DONE, sending elements that other peer is missing\n");
1845
1846       /* send all the elements that did not come from the remote peer */
1847       GNUNET_CONTAINER_multihashmap32_iterate (op->state->key_to_element,
1848                                                &send_missing_full_elements_iter,
1849                                                op);
1850
1851       ev = GNUNET_MQ_msg_header (GNUNET_MESSAGE_TYPE_SET_UNION_P2P_FULL_DONE);
1852       GNUNET_MQ_notify_sent (ev,
1853                              &send_done_and_destroy,
1854                              op);
1855       GNUNET_MQ_send (op->mq,
1856                       ev);
1857       op->state->phase = PHASE_DONE;
1858       /* we now wait until the other peer shuts the tunnel down*/
1859     }
1860     break;
1861   case PHASE_FULL_SENDING:
1862     {
1863       LOG (GNUNET_ERROR_TYPE_DEBUG,
1864            "got FULL DONE, finishing\n");
1865       /* We sent the full set, and got the response for that.  We're done. */
1866       op->state->phase = PHASE_DONE;
1867       GNUNET_CADET_receive_done (op->channel);
1868       send_done_and_destroy (op);
1869       return;
1870     }
1871     break;
1872   default:
1873     GNUNET_log (GNUNET_ERROR_TYPE_INFO,
1874                 "Handle full done phase is %u\n",
1875                 (unsigned) op->state->phase);
1876     GNUNET_break_op (0);
1877     fail_union_operation (op);
1878     return;
1879   }
1880   GNUNET_CADET_receive_done (op->channel);
1881 }
1882
1883
1884 /**
1885  * Check a demand by the other peer for elements based on a list
1886  * of `struct GNUNET_HashCode`s.
1887  *
1888  * @parem cls closure, a set union operation
1889  * @param mh the demand message
1890  * @return #GNUNET_OK if @a mh is well-formed
1891  */
1892 int
1893 check_union_p2p_demand (void *cls,
1894                         const struct GNUNET_MessageHeader *mh)
1895 {
1896   struct Operation *op = cls;
1897   unsigned int num_hashes;
1898
1899   if (GNUNET_SET_OPERATION_UNION != op->set->operation)
1900   {
1901     GNUNET_break_op (0);
1902     return GNUNET_SYSERR;
1903   }
1904   num_hashes = (ntohs (mh->size) - sizeof (struct GNUNET_MessageHeader))
1905     / sizeof (struct GNUNET_HashCode);
1906   if ((ntohs (mh->size) - sizeof (struct GNUNET_MessageHeader))
1907       != num_hashes * sizeof (struct GNUNET_HashCode))
1908   {
1909     GNUNET_break_op (0);
1910     return GNUNET_SYSERR;
1911   }
1912   return GNUNET_OK;
1913 }
1914
1915
1916 /**
1917  * Handle a demand by the other peer for elements based on a list
1918  * of `struct GNUNET_HashCode`s.
1919  *
1920  * @parem cls closure, a set union operation
1921  * @param mh the demand message
1922  */
1923 void
1924 handle_union_p2p_demand (void *cls,
1925                          const struct GNUNET_MessageHeader *mh)
1926 {
1927   struct Operation *op = cls;
1928   struct ElementEntry *ee;
1929   struct GNUNET_SET_ElementMessage *emsg;
1930   const struct GNUNET_HashCode *hash;
1931   unsigned int num_hashes;
1932   struct GNUNET_MQ_Envelope *ev;
1933
1934   num_hashes = (ntohs (mh->size) - sizeof (struct GNUNET_MessageHeader))
1935     / sizeof (struct GNUNET_HashCode);
1936   for (hash = (const struct GNUNET_HashCode *) &mh[1];
1937        num_hashes > 0;
1938        hash++, num_hashes--)
1939   {
1940     ee = GNUNET_CONTAINER_multihashmap_get (op->set->content->elements,
1941                                             hash);
1942     if (NULL == ee)
1943     {
1944       /* Demand for non-existing element. */
1945       GNUNET_break_op (0);
1946       fail_union_operation (op);
1947       return;
1948     }
1949     if (GNUNET_NO == _GSS_is_element_of_operation (ee, op))
1950     {
1951       /* Probably confused lazily copied sets. */
1952       GNUNET_break_op (0);
1953       fail_union_operation (op);
1954       return;
1955     }
1956     ev = GNUNET_MQ_msg_extra (emsg, ee->element.size, GNUNET_MESSAGE_TYPE_SET_P2P_ELEMENTS);
1957     GNUNET_memcpy (&emsg[1], ee->element.data, ee->element.size);
1958     emsg->reserved = htons (0);
1959     emsg->element_type = htons (ee->element.element_type);
1960     LOG (GNUNET_ERROR_TYPE_DEBUG,
1961          "[OP %x] Sending demanded element (size %u, hash %s) to peer\n",
1962          (void *) op,
1963          (unsigned int) ee->element.size,
1964          GNUNET_h2s (&ee->element_hash));
1965     GNUNET_MQ_send (op->mq, ev);
1966     GNUNET_STATISTICS_update (_GSS_statistics,
1967                               "# exchanged elements",
1968                               1,
1969                               GNUNET_NO);
1970
1971     switch (op->result_mode)
1972     {
1973       case GNUNET_SET_RESULT_ADDED:
1974         /* Nothing to do. */
1975         break;
1976       case GNUNET_SET_RESULT_SYMMETRIC:
1977         send_client_element (op, &ee->element, GNUNET_SET_STATUS_ADD_REMOTE);
1978         break;
1979       default:
1980         /* Result mode not supported, should have been caught earlier. */
1981         GNUNET_break (0);
1982         break;
1983     }
1984   }
1985   GNUNET_CADET_receive_done (op->channel);
1986 }
1987
1988
1989 /**
1990  * Check offer (of `struct GNUNET_HashCode`s).
1991  *
1992  * @param cls the union operation
1993  * @param mh the message
1994  * @return #GNUNET_OK if @a mh is well-formed
1995  */
1996 int
1997 check_union_p2p_offer (void *cls,
1998                         const struct GNUNET_MessageHeader *mh)
1999 {
2000   struct Operation *op = cls;
2001   unsigned int num_hashes;
2002
2003   if (GNUNET_SET_OPERATION_UNION != op->set->operation)
2004   {
2005     GNUNET_break_op (0);
2006     return GNUNET_SYSERR;
2007   }
2008   /* look up elements and send them */
2009   if ( (op->state->phase != PHASE_INVENTORY_PASSIVE) &&
2010        (op->state->phase != PHASE_INVENTORY_ACTIVE))
2011   {
2012     GNUNET_break_op (0);
2013     return GNUNET_SYSERR;
2014   }
2015   num_hashes = (ntohs (mh->size) - sizeof (struct GNUNET_MessageHeader))
2016     / sizeof (struct GNUNET_HashCode);
2017   if ((ntohs (mh->size) - sizeof (struct GNUNET_MessageHeader)) !=
2018       num_hashes * sizeof (struct GNUNET_HashCode))
2019   {
2020     GNUNET_break_op (0);
2021     return GNUNET_SYSERR;
2022   }
2023   return GNUNET_OK;
2024 }
2025
2026
2027 /**
2028  * Handle offers (of `struct GNUNET_HashCode`s) and
2029  * respond with demands (of `struct GNUNET_HashCode`s).
2030  *
2031  * @param cls the union operation
2032  * @param mh the message
2033  */
2034 void
2035 handle_union_p2p_offer (void *cls,
2036                         const struct GNUNET_MessageHeader *mh)
2037 {
2038   struct Operation *op = cls;
2039   const struct GNUNET_HashCode *hash;
2040   unsigned int num_hashes;
2041
2042   num_hashes = (ntohs (mh->size) - sizeof (struct GNUNET_MessageHeader))
2043     / sizeof (struct GNUNET_HashCode);
2044   for (hash = (const struct GNUNET_HashCode *) &mh[1];
2045        num_hashes > 0;
2046        hash++, num_hashes--)
2047   {
2048     struct ElementEntry *ee;
2049     struct GNUNET_MessageHeader *demands;
2050     struct GNUNET_MQ_Envelope *ev;
2051
2052     ee = GNUNET_CONTAINER_multihashmap_get (op->set->content->elements,
2053                                             hash);
2054     if (NULL != ee)
2055       if (GNUNET_YES == _GSS_is_element_of_operation (ee, op))
2056         continue;
2057
2058     if (GNUNET_YES ==
2059         GNUNET_CONTAINER_multihashmap_contains (op->state->demanded_hashes,
2060                                                 hash))
2061     {
2062       LOG (GNUNET_ERROR_TYPE_DEBUG,
2063            "Skipped sending duplicate demand\n");
2064       continue;
2065     }
2066
2067     GNUNET_assert (GNUNET_OK ==
2068                    GNUNET_CONTAINER_multihashmap_put (op->state->demanded_hashes,
2069                                                       hash,
2070                                                       NULL,
2071                                                       GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST));
2072
2073     LOG (GNUNET_ERROR_TYPE_DEBUG,
2074          "[OP %x] Requesting element (hash %s)\n",
2075          (void *) op, GNUNET_h2s (hash));
2076     ev = GNUNET_MQ_msg_header_extra (demands,
2077                                      sizeof (struct GNUNET_HashCode),
2078                                      GNUNET_MESSAGE_TYPE_SET_UNION_P2P_DEMAND);
2079     GNUNET_memcpy (&demands[1],
2080                    hash,
2081                    sizeof (struct GNUNET_HashCode));
2082     GNUNET_MQ_send (op->mq, ev);
2083   }
2084   GNUNET_CADET_receive_done (op->channel);
2085 }
2086
2087
2088 /**
2089  * Handle a done message from a remote peer
2090  *
2091  * @param cls the union operation
2092  * @param mh the message
2093  */
2094 void
2095 handle_union_p2p_done (void *cls,
2096                        const struct GNUNET_MessageHeader *mh)
2097 {
2098   struct Operation *op = cls;
2099
2100   if (GNUNET_SET_OPERATION_UNION != op->set->operation)
2101   {
2102     GNUNET_break_op (0);
2103     fail_union_operation (op);
2104     return;
2105   }
2106   switch (op->state->phase)
2107   {
2108   case PHASE_INVENTORY_PASSIVE:
2109     /* We got all requests, but still have to send our elements in response. */
2110     op->state->phase = PHASE_FINISH_WAITING;
2111
2112     LOG (GNUNET_ERROR_TYPE_DEBUG,
2113          "got DONE (as passive partner), waiting for our demands to be satisfied\n");
2114     /* The active peer is done sending offers
2115      * and inquiries.  This means that all
2116      * our responses to that (demands and offers)
2117      * must be in flight (queued or in mesh).
2118      *
2119      * We should notify the active peer once
2120      * all our demands are satisfied, so that the active
2121      * peer can quit if we gave him everything.
2122      */
2123     GNUNET_CADET_receive_done (op->channel);
2124     maybe_finish (op);
2125     return;
2126   case PHASE_INVENTORY_ACTIVE:
2127     LOG (GNUNET_ERROR_TYPE_DEBUG,
2128          "got DONE (as active partner), waiting to finish\n");
2129     /* All demands of the other peer are satisfied,
2130      * and we processed all offers, thus we know
2131      * exactly what our demands must be.
2132      *
2133      * We'll close the channel
2134      * to the other peer once our demands are met.
2135      */
2136     op->state->phase = PHASE_FINISH_CLOSING;
2137     GNUNET_CADET_receive_done (op->channel);
2138     maybe_finish (op);
2139     return;
2140   default:
2141     GNUNET_break_op (0);
2142     fail_union_operation (op);
2143     return;
2144   }
2145 }
2146
2147
2148 /**
2149  * Initiate operation to evaluate a set union with a remote peer.
2150  *
2151  * @param op operation to perform (to be initialized)
2152  * @param opaque_context message to be transmitted to the listener
2153  *        to convince him to accept, may be NULL
2154  */
2155 static struct OperationState *
2156 union_evaluate (struct Operation *op,
2157                 const struct GNUNET_MessageHeader *opaque_context)
2158 {
2159   struct OperationState *state;
2160   struct GNUNET_MQ_Envelope *ev;
2161   struct OperationRequestMessage *msg;
2162
2163   ev = GNUNET_MQ_msg_nested_mh (msg,
2164                                 GNUNET_MESSAGE_TYPE_SET_P2P_OPERATION_REQUEST,
2165                                 opaque_context);
2166   if (NULL == ev)
2167   {
2168     /* the context message is too large */
2169     GNUNET_break (0);
2170     return NULL;
2171   }
2172   state = GNUNET_new (struct OperationState);
2173   state->demanded_hashes = GNUNET_CONTAINER_multihashmap_create (32,
2174                                                                  GNUNET_NO);
2175   /* copy the current generation's strata estimator for this operation */
2176   state->se = strata_estimator_dup (op->set->state->se);
2177   /* we started the operation, thus we have to send the operation request */
2178   state->phase = PHASE_EXPECT_SE;
2179   state->salt_receive = state->salt_send = 42; // FIXME?????
2180   LOG (GNUNET_ERROR_TYPE_DEBUG,
2181        "Initiating union operation evaluation\n");
2182   GNUNET_STATISTICS_update (_GSS_statistics,
2183                             "# of total union operations",
2184                             1,
2185                             GNUNET_NO);
2186   GNUNET_STATISTICS_update (_GSS_statistics,
2187                             "# of initiated union operations",
2188                             1,
2189                             GNUNET_NO);
2190   msg->operation = htonl (GNUNET_SET_OPERATION_UNION);
2191   GNUNET_MQ_send (op->mq,
2192                   ev);
2193
2194   if (NULL != opaque_context)
2195     LOG (GNUNET_ERROR_TYPE_DEBUG,
2196          "sent op request with context message\n");
2197   else
2198     LOG (GNUNET_ERROR_TYPE_DEBUG,
2199          "sent op request without context message\n");
2200
2201   op->state = state;
2202   initialize_key_to_element (op);
2203   state->initial_size = GNUNET_CONTAINER_multihashmap32_size (state->key_to_element);
2204   return state;
2205 }
2206
2207
2208 /**
2209  * Accept an union operation request from a remote peer.
2210  * Only initializes the private operation state.
2211  *
2212  * @param op operation that will be accepted as a union operation
2213  */
2214 static struct OperationState *
2215 union_accept (struct Operation *op)
2216 {
2217   struct OperationState *state;
2218   const struct StrataEstimator *se;
2219   struct GNUNET_MQ_Envelope *ev;
2220   struct StrataEstimatorMessage *strata_msg;
2221   char *buf;
2222   size_t len;
2223   uint16_t type;
2224
2225   LOG (GNUNET_ERROR_TYPE_DEBUG,
2226        "accepting set union operation\n");
2227   GNUNET_STATISTICS_update (_GSS_statistics,
2228                             "# of accepted union operations",
2229                             1,
2230                             GNUNET_NO);
2231   GNUNET_STATISTICS_update (_GSS_statistics,
2232                             "# of total union operations",
2233                             1,
2234                             GNUNET_NO);
2235
2236   state = GNUNET_new (struct OperationState);
2237   state->se = strata_estimator_dup (op->set->state->se);
2238   state->demanded_hashes = GNUNET_CONTAINER_multihashmap_create (32,
2239                                                                  GNUNET_NO);
2240   state->salt_receive = state->salt_send = 42; // FIXME?????
2241   op->state = state;
2242   initialize_key_to_element (op);
2243   state->initial_size = GNUNET_CONTAINER_multihashmap32_size (state->key_to_element);
2244
2245   /* kick off the operation */
2246   se = state->se;
2247   buf = GNUNET_malloc (se->strata_count * IBF_BUCKET_SIZE * se->ibf_size);
2248   len = strata_estimator_write (se,
2249                                 buf);
2250   if (len < se->strata_count * IBF_BUCKET_SIZE * se->ibf_size)
2251     type = GNUNET_MESSAGE_TYPE_SET_UNION_P2P_SEC;
2252   else
2253     type = GNUNET_MESSAGE_TYPE_SET_UNION_P2P_SE;
2254   ev = GNUNET_MQ_msg_extra (strata_msg,
2255                             len,
2256                             type);
2257   GNUNET_memcpy (&strata_msg[1],
2258                  buf,
2259                  len);
2260   GNUNET_free (buf);
2261   strata_msg->set_size
2262     = GNUNET_htonll (GNUNET_CONTAINER_multihashmap_size (op->set->content->elements));
2263   GNUNET_MQ_send (op->mq,
2264                   ev);
2265   state->phase = PHASE_EXPECT_IBF;
2266   return state;
2267 }
2268
2269
2270 /**
2271  * Create a new set supporting the union operation
2272  *
2273  * We maintain one strata estimator per set and then manipulate it over the
2274  * lifetime of the set, as recreating a strata estimator would be expensive.
2275  *
2276  * @return the newly created set, NULL on error
2277  */
2278 static struct SetState *
2279 union_set_create (void)
2280 {
2281   struct SetState *set_state;
2282
2283   LOG (GNUNET_ERROR_TYPE_DEBUG,
2284        "union set created\n");
2285   set_state = GNUNET_new (struct SetState);
2286   set_state->se = strata_estimator_create (SE_STRATA_COUNT,
2287                                            SE_IBF_SIZE, SE_IBF_HASH_NUM);
2288   if (NULL == set_state->se)
2289   {
2290     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
2291                 "Failed to allocate strata estimator\n");
2292     GNUNET_free (set_state);
2293     return NULL;
2294   }
2295   return set_state;
2296 }
2297
2298
2299 /**
2300  * Add the element from the given element message to the set.
2301  *
2302  * @param set_state state of the set want to add to
2303  * @param ee the element to add to the set
2304  */
2305 static void
2306 union_add (struct SetState *set_state,
2307            struct ElementEntry *ee)
2308 {
2309   strata_estimator_insert (set_state->se,
2310                            get_ibf_key (&ee->element_hash));
2311 }
2312
2313
2314 /**
2315  * Remove the element given in the element message from the set.
2316  * Only marks the element as removed, so that older set operations can still exchange it.
2317  *
2318  * @param set_state state of the set to remove from
2319  * @param ee set element to remove
2320  */
2321 static void
2322 union_remove (struct SetState *set_state,
2323               struct ElementEntry *ee)
2324 {
2325   strata_estimator_remove (set_state->se,
2326                            get_ibf_key (&ee->element_hash));
2327 }
2328
2329
2330 /**
2331  * Destroy a set that supports the union operation.
2332  *
2333  * @param set_state the set to destroy
2334  */
2335 static void
2336 union_set_destroy (struct SetState *set_state)
2337 {
2338   if (NULL != set_state->se)
2339   {
2340     strata_estimator_destroy (set_state->se);
2341     set_state->se = NULL;
2342   }
2343   GNUNET_free (set_state);
2344 }
2345
2346
2347 /**
2348  * Copy union-specific set state.
2349  *
2350  * @param state source state for copying the union state
2351  * @return a copy of the union-specific set state
2352  */
2353 static struct SetState *
2354 union_copy_state (struct SetState *state)
2355 {
2356   struct SetState *new_state;
2357
2358   GNUNET_assert ( (NULL != state) &&
2359                   (NULL != state->se) );
2360   new_state = GNUNET_new (struct SetState);
2361   new_state->se = strata_estimator_dup (state->se);
2362
2363   return new_state;
2364 }
2365
2366
2367 /**
2368  * Handle case where channel went down for an operation.
2369  *
2370  * @param op operation that lost the channel
2371  */
2372 static void
2373 union_channel_death (struct Operation *op)
2374 {
2375   _GSS_operation_destroy (op,
2376                           GNUNET_YES);
2377 }
2378
2379
2380 /**
2381  * Get the table with implementing functions for
2382  * set union.
2383  *
2384  * @return the operation specific VTable
2385  */
2386 const struct SetVT *
2387 _GSS_union_vt ()
2388 {
2389   static const struct SetVT union_vt = {
2390     .create = &union_set_create,
2391     .add = &union_add,
2392     .remove = &union_remove,
2393     .destroy_set = &union_set_destroy,
2394     .evaluate = &union_evaluate,
2395     .accept = &union_accept,
2396     .cancel = &union_op_cancel,
2397     .copy_state = &union_copy_state,
2398     .channel_death = &union_channel_death
2399   };
2400
2401   return &union_vt;
2402 }