src/set/gnunet-service-set_union.c

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