src/set/gnunet-service-set_intersection.c

   1 /*
   2       This file is part of GNUnet
   3       (C) 2013 Christian Grothoff (and other contributing authors)
   4
   5       GNUnet is free software; you can redistribute it and/or modify
   6       it under the terms of the GNU General Public License as published
   7       by the Free Software Foundation; either version 3, or (at your
   8       option) any later version.
   9
  10       GNUnet is distributed in the hope that it will be useful, but
  11       WITHOUT ANY WARRANTY; without even the implied warranty of
  12       MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13       General Public License for more details.
  14
  15       You should have received a copy of the GNU General Public License
  16       along with GNUnet; see the file COPYING.  If not, write to the
  17       Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  18       Boston, MA 02111-1307, USA.
  19 */
  20
  21 /**
  22  * @file set/gnunet-service-set_intersection.c
  23  * @brief two-peer set intersection
  24  * @author Christian Fuchs
  25  */
  26 #include "platform.h"
  27 #include "gnunet_util_lib.h"
  28 #include "gnunet-service-set.h"
  29 #include "gnunet_block_lib.h"
  30 #include "set_protocol.h"
  31 #include <gcrypt.h>
  32
  33 #define BLOOMFILTER_SIZE GNUNET_CRYPTO_HASH_LENGTH
  34 /**
  35  * Current phase we are in for a intersection operation.
  36  */
  37 enum IntersectionOperationPhase
  38 {
  39   /**
  40    * Alices has suggested an operation to bob,
  41    * and is waiting for a bf or session end.
  42    */
  43   PHASE_INITIAL,
  44   /**
  45    * Bob has accepted the operation, Bob and Alice are now exchanging bfs
  46    * until one notices the their element count is equal
  47    */
  48   PHASE_BF_EXCHANGE,
  49   /**
  50    * if both peers have an equal peercount, they enter this state for
  51    * one more turn, to see if they actually have agreed on a correct set.
  52    * if a peer finds the same element count after the next iteration,
  53    * it ends the the session
  54    */
  55   PHASE_MAYBE_FINISHED,
  56   /**
  57    * The protocol is over.
  58    * Results may still have to be sent to the client.
  59    */
  60   PHASE_FINISHED
  61 };
  62
  63
  64 /**
  65  * State of an evaluate operation
  66  * with another peer.
  67  */
  68 struct OperationState
  69 {
  70   /**
  71    * The bf we currently receive
  72    */
  73   struct GNUNET_CONTAINER_BloomFilter *remote_bf;
  74
  75   /**
  76    * BF of the set's element.
  77    */
  78   struct GNUNET_CONTAINER_BloomFilter *local_bf;
  79
  80   /**
  81    * for multipart msgs we have to store the bloomfilter-data until we fully sent it.
  82    */
  83   char * local_bf_data;
  84
  85   /**
  86    * for multipart msgs we have to store the bloomfilter-data until we fully sent it.
  87    */
  88   uint32_t local_bf_data_size;
  89
  90   /**
  91    * Current state of the operation.
  92    */
  93   enum IntersectionOperationPhase phase;
  94
  95   /**
  96    * Generation in which the operation handle
  97    * was created.
  98    */
  99   unsigned int generation_created;
 100
 101   /**
 102    * Maps element-id-hashes to 'elements in our set'.
 103    */
 104   struct GNUNET_CONTAINER_MultiHashMap *my_elements;
 105
 106   /**
 107    * Current element count contained within contained_elements
 108    */
 109   uint32_t my_element_count;
 110
 111   /**
 112    * Iterator for sending elements on the key to element mapping to the client.
 113    */
 114   struct GNUNET_CONTAINER_MultiHashMapIterator *full_result_iter;
 115
 116   /**
 117    * Evaluate operations are held in
 118    * a linked list.
 119    */
 120   struct OperationState *next;
 121
 122    /**
 123     * Evaluate operations are held in
 124     * a linked list.
 125     */
 126   struct OperationState *prev;
 127
 128   /**
 129    * Did we send the client that we are done?
 130    */
 131   int client_done_sent;
 132 };
 133
 134
 135 /**
 136  * Extra state required for efficient set intersection.
 137  */
 138 struct SetState
 139 {
 140   /**
 141    * Number of currently valid elements in the set which have not been removed
 142    */
 143   uint32_t current_set_element_count;
 144 };
 145
 146
 147 /**
 148  * Alice's version:
 149  *
 150  * fills the contained-elements hashmap with all relevant
 151  * elements and adds their mutated hashes to our local bloomfilter with mutator+1
 152  *
 153  * @param cls closure
 154  * @param key current key code
 155  * @param value value in the hash map
 156  * @return #GNUNET_YES if we should continue to
 157  *         iterate,
 158  *         #GNUNET_NO if not.
 159  */
 160 static int
 161 iterator_initialization_by_alice (void *cls,
 162                                   const struct GNUNET_HashCode *key,
 163                                   void *value)
 164 {
 165   struct ElementEntry *ee = value;
 166   struct Operation *op = cls;
 167   struct GNUNET_HashCode mutated_hash;
 168
 169   //only consider this element, if it is valid for us
 170   if ((op->generation_created >= ee->generation_removed)
 171        || (op->generation_created < ee->generation_added))
 172     return GNUNET_YES;
 173
 174   // not contained according to bob's bloomfilter
 175   GNUNET_BLOCK_mingle_hash(&ee->element_hash,
 176                            op->spec->salt,
 177                            &mutated_hash);
 178   if (GNUNET_NO == GNUNET_CONTAINER_bloomfilter_test (op->state->remote_bf,
 179                                                       &mutated_hash))
 180     return GNUNET_YES;
 181
 182   op->state->my_element_count++;
 183   GNUNET_assert (GNUNET_YES ==
 184                  GNUNET_CONTAINER_multihashmap_put (op->state->my_elements,
 185                                                     &ee->element_hash, ee,
 186                                                     GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
 187
 188   /* create our own bloomfilter with salt+1 */
 189   GNUNET_BLOCK_mingle_hash (&ee->element_hash,
 190                             op->spec->salt + 1,
 191                             &mutated_hash);
 192   GNUNET_CONTAINER_bloomfilter_add (op->state->local_bf,
 193                                     &mutated_hash);
 194
 195   return GNUNET_YES;
 196 }
 197
 198 /**
 199  * fills the contained-elements hashmap with all relevant
 200  * elements and adds their mutated hashes to our local bloomfilter
 201  *
 202  * @param cls closure
 203  * @param key current key code
 204  * @param value value in the hash map
 205  * @return #GNUNET_YES if we should continue to
 206  *         iterate,
 207  *         #GNUNET_NO if not.
 208  */
 209 static int
 210 iterator_initialization (void *cls,
 211                          const struct GNUNET_HashCode *key,
 212                          void *value)
 213 {
 214   struct ElementEntry *ee = value;
 215   struct Operation *op = cls;
 216   struct GNUNET_HashCode mutated_hash;
 217
 218   //only consider this element, if it is valid for us
 219   if ((op->generation_created >= ee->generation_removed)
 220        || (op->generation_created < ee->generation_added))
 221     return GNUNET_YES;
 222
 223   GNUNET_assert (GNUNET_YES ==
 224                  GNUNET_CONTAINER_multihashmap_put (op->state->my_elements,
 225                                                     &ee->element_hash, ee,
 226                                                     GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
 227   GNUNET_BLOCK_mingle_hash (&ee->element_hash,
 228                             op->spec->salt,
 229                             &mutated_hash);
 230   GNUNET_CONTAINER_bloomfilter_add (op->state->local_bf,
 231                                     &mutated_hash);
 232   return GNUNET_YES;
 233 }
 234
 235
 236 /**
 237  * removes element from a hashmap if it is not contained within the
 238  * provided remote bloomfilter. Then, fill our new bloomfilter.
 239  *
 240  * @param cls closure
 241  * @param key current key code
 242  * @param value value in the hash map
 243  * @return #GNUNET_YES if we should continue to
 244  *         iterate,
 245  *         #GNUNET_NO if not.
 246  */
 247 static int
 248 iterator_bf_round (void *cls,
 249                    const struct GNUNET_HashCode *key,
 250                    void *value)
 251 {
 252   struct ElementEntry *ee = value;
 253   struct Operation *op = cls;
 254   struct GNUNET_HashCode mutated_hash;
 255
 256   GNUNET_BLOCK_mingle_hash(&ee->element_hash, op->spec->salt, &mutated_hash);
 257
 258   if (GNUNET_NO ==
 259       GNUNET_CONTAINER_bloomfilter_test (op->state->remote_bf,
 260                                          &mutated_hash))
 261   {
 262     op->state->my_element_count--;
 263     GNUNET_assert (GNUNET_YES ==
 264                    GNUNET_CONTAINER_multihashmap_remove (op->state->my_elements,
 265                                                          &ee->element_hash,
 266                                                          ee));
 267     return GNUNET_YES;
 268   }
 269   GNUNET_BLOCK_mingle_hash(&ee->element_hash,
 270                            op->spec->salt+1,
 271                            &mutated_hash);
 272   GNUNET_CONTAINER_bloomfilter_add (op->state->local_bf,
 273                                     &mutated_hash);
 274   return GNUNET_YES;
 275 }
 276
 277
 278 /**
 279  * Inform the client that the union operation has failed,
 280  * and proceed to destroy the evaluate operation.
 281  *
 282  * @param op the intersection operation to fail
 283  */
 284 static void
 285 fail_intersection_operation (struct Operation *op)
 286 {
 287   struct GNUNET_MQ_Envelope *ev;
 288   struct GNUNET_SET_ResultMessage *msg;
 289
 290   if (op->state->my_elements)
 291     GNUNET_CONTAINER_multihashmap_destroy(op->state->my_elements);
 292
 293   GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "intersection operation failed\n");
 294
 295   ev = GNUNET_MQ_msg (msg, GNUNET_MESSAGE_TYPE_SET_RESULT);
 296   msg->result_status = htons (GNUNET_SET_STATUS_FAILURE);
 297   msg->request_id = htonl (op->spec->client_request_id);
 298   msg->element_type = htons (0);
 299   GNUNET_MQ_send (op->spec->set->client_mq, ev);
 300   _GSS_operation_destroy (op);
 301 }
 302
 303
 304 /**
 305  * Send a request for the evaluate operation to a remote peer
 306  *
 307  * @param op operation with the other peer
 308  */
 309 static void
 310 send_operation_request (struct Operation *op)
 311 {
 312   struct GNUNET_MQ_Envelope *ev;
 313   struct OperationRequestMessage *msg;
 314
 315   ev = GNUNET_MQ_msg_nested_mh (msg, GNUNET_MESSAGE_TYPE_SET_P2P_OPERATION_REQUEST,
 316                                 op->spec->context_msg);
 317
 318   if (NULL == ev)
 319   {
 320     /* the context message is too large */
 321     GNUNET_break (0);
 322     GNUNET_SERVER_client_disconnect (op->spec->set->client);
 323     return;
 324   }
 325   msg->operation = htonl (GNUNET_SET_OPERATION_INTERSECTION);
 326   msg->app_id = op->spec->app_id;
 327   msg->salt = htonl (op->spec->salt);
 328   msg->element_count = htonl(op->state->my_element_count);
 329
 330   GNUNET_MQ_send (op->mq, ev);
 331
 332   if (NULL != op->spec->context_msg)
 333     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "sent op request with context message\n");
 334   else
 335     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "sent op request without context message\n");
 336
 337   if (NULL != op->spec->context_msg)
 338   {
 339     GNUNET_free (op->spec->context_msg);
 340     op->spec->context_msg = NULL;
 341   }
 342 }
 343
 344 static void
 345 send_bloomfilter_multipart (struct Operation *op, uint32_t offset)
 346 {
 347   struct GNUNET_MQ_Envelope *ev;
 348   struct BFMessage *msg;
 349   uint32_t chunk_size = (GNUNET_SERVER_MAX_MESSAGE_SIZE - sizeof(struct BFMessage));
 350   uint32_t todo_size = op->state->local_bf_data_size - offset;
 351
 352   if (todo_size < chunk_size)
 353     // we probably need many chunks, thus we assume a maximum packet size by default
 354     chunk_size = todo_size;
 355
 356   ev = GNUNET_MQ_msg_extra (msg, chunk_size, GNUNET_MESSAGE_TYPE_SET_INTERSECTION_P2P_BF);
 357
 358   msg->reserved = 0;
 359   msg->bloomfilter_total_length = htonl (op->state->local_bf_data_size);
 360   msg->bloomfilter_length = htonl (chunk_size);
 361   msg->bloomfilter_offset = htonl (offset);
 362   memcpy(&msg[1], op->state->local_bf, chunk_size);
 363
 364   GNUNET_MQ_send (op->mq, ev);
 365
 366   if (op->state->local_bf_data_size == offset + chunk_size)
 367   {
 368     // done
 369     GNUNET_CONTAINER_bloomfilter_free (op->state->local_bf);
 370     GNUNET_free(op->state->local_bf_data);
 371     op->state->local_bf_data = NULL;
 372     op->state->local_bf = NULL;
 373     return;
 374   }
 375
 376   send_bloomfilter_multipart (op, offset + chunk_size);
 377 }
 378
 379 /**
 380  * Send a bloomfilter to our peer.
 381  * that the operation is over.
 382  * After the result done message has been sent to the client,
 383  * destroy the evaluate operation.
 384  *
 385  * @param op intersection operation
 386  */
 387 static void
 388 send_bloomfilter (struct Operation *op)
 389 {
 390   struct GNUNET_MQ_Envelope *ev;
 391   struct BFMessage *msg;
 392   uint32_t bf_size;
 393
 394   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "sending bf of size %u\n");
 395
 396   // send our bloomfilter
 397   bf_size = GNUNET_CONTAINER_bloomfilter_get_size (op->state->local_bf);
 398   if ( GNUNET_SERVER_MAX_MESSAGE_SIZE <= bf_size + sizeof(struct BFMessage))
 399   {
 400     // singlepart
 401     ev = GNUNET_MQ_msg_extra (msg, bf_size, GNUNET_MESSAGE_TYPE_SET_INTERSECTION_P2P_BF);
 402
 403     GNUNET_CONTAINER_bloomfilter_free (op->state->local_bf);
 404     op->state->local_bf = NULL;
 405
 406     msg->reserved = 0;
 407     msg->sender_element_count = htonl (op->state->my_element_count);
 408     msg->bloomfilter_length = htonl (bf_size);
 409     msg->bloomfilter_offset = htonl (0);
 410     msg->sender_mutator = htonl (op->spec->salt);
 411
 412     GNUNET_MQ_send (op->mq, ev);
 413   }
 414   else {
 415     //multipart
 416     op->state->local_bf_data = (char *)GNUNET_malloc(bf_size);
 417     GNUNET_assert (GNUNET_SYSERR !=
 418                  GNUNET_CONTAINER_bloomfilter_get_raw_data (op->state->local_bf,
 419                                                             op->state->local_bf_data,
 420                                                             bf_size));
 421     op->state->local_bf_data_size = bf_size;
 422     send_bloomfilter_multipart (op, 0);
 423   }
 424 }
 425
 426
 427 /**
 428  * Signal to the client that the operation has finished and
 429  * destroy the operation.
 430  *
 431  * @param cls operation to destroy
 432  */
 433 static void
 434 send_client_done_and_destroy (void *cls)
 435 {
 436   struct Operation *op = cls;
 437   struct GNUNET_MQ_Envelope *ev;
 438   struct GNUNET_SET_ResultMessage *rm;
 439   ev = GNUNET_MQ_msg (rm, GNUNET_MESSAGE_TYPE_SET_RESULT);
 440   rm->request_id = htonl (op->spec->client_request_id);
 441   rm->result_status = htons (GNUNET_SET_STATUS_DONE);
 442   rm->element_type = htons (0);
 443   GNUNET_MQ_send (op->spec->set->client_mq, ev);
 444   _GSS_operation_destroy (op);
 445 }
 446
 447
 448 /**
 449  * Send all elements in the full result iterator.
 450  *
 451  * @param cls operation
 452  */
 453 static void
 454 send_remaining_elements (void *cls)
 455 {
 456   struct Operation *op = cls;
 457   struct ElementEntry *remaining; //TODO rework this, key entry does not exist here
 458   struct GNUNET_MQ_Envelope *ev;
 459   struct GNUNET_SET_ResultMessage *rm;
 460   struct GNUNET_SET_Element *element;
 461   int res;
 462
 463   res = GNUNET_CONTAINER_multihashmap_iterator_next (op->state->full_result_iter, NULL, (const void **) &remaining);
 464   if (GNUNET_NO == res) {
 465     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "sending done and destroy because iterator ran out\n");
 466     send_client_done_and_destroy (op);
 467     return;
 468   }
 469
 470   element = &remaining->element;
 471   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "sending element (size %u) to client (full set)\n", element->size);
 472   GNUNET_assert (0 != op->spec->client_request_id);
 473
 474   ev = GNUNET_MQ_msg_extra (rm, element->size, GNUNET_MESSAGE_TYPE_SET_RESULT);
 475   GNUNET_assert (NULL != ev);
 476
 477   rm->result_status = htons (GNUNET_SET_STATUS_OK);
 478   rm->request_id = htonl (op->spec->client_request_id);
 479   rm->element_type = element->type;
 480   memcpy (&rm[1], element->data, element->size);
 481
 482   GNUNET_MQ_notify_sent (ev, send_remaining_elements, op);
 483   GNUNET_MQ_send (op->spec->set->client_mq, ev);
 484 }
 485
 486
 487 /**
 488  * Inform the peer that this operation is complete.
 489  *
 490  * @param op the intersection operation to fail
 491  */
 492 static void
 493 send_peer_done (struct Operation *op)
 494 {
 495   struct GNUNET_MQ_Envelope *ev;
 496
 497   op->state->phase = PHASE_FINISHED;
 498   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Intersection succeeded, sending DONE\n");
 499   GNUNET_CONTAINER_bloomfilter_free (op->state->local_bf);
 500   op->state->local_bf = NULL;
 501
 502   ev = GNUNET_MQ_msg_header (GNUNET_MESSAGE_TYPE_SET_P2P_DONE);
 503   GNUNET_MQ_send (op->mq, ev);
 504 }
 505
 506
 507 /**
 508  * Handle an BF message from a remote peer.
 509  *
 510  * @param cls the intersection operation
 511  * @param mh the header of the message
 512  */
 513 static void
 514 handle_p2p_bf (void *cls, const struct GNUNET_MessageHeader *mh)
 515 {
 516   struct Operation *op = cls;
 517   const struct BFMessage *msg = (const struct BFMessage *) mh;
 518   uint32_t old_elements;
 519   uint32_t peer_elements;
 520
 521   old_elements = op->state->my_element_count;
 522   op->spec->salt = ntohl (msg->sender_mutator);
 523
 524   op->state->remote_bf = GNUNET_CONTAINER_bloomfilter_init ((const char*) &msg[1],
 525                                                             BLOOMFILTER_SIZE,
 526                                                             ntohl (msg->bloomfilter_length));
 527   op->state->local_bf = GNUNET_CONTAINER_bloomfilter_init (NULL,
 528                                                            BLOOMFILTER_SIZE,
 529                                                            GNUNET_CONSTANTS_BLOOMFILTER_K);
 530   switch (op->state->phase)
 531   {
 532   case PHASE_INITIAL:
 533     // If we are ot our first msg
 534     op->state->my_elements = GNUNET_CONTAINER_multihashmap_create (op->state->my_element_count, GNUNET_YES);
 535
 536     GNUNET_CONTAINER_multihashmap_iterate (op->spec->set->elements,
 537                                            &iterator_initialization_by_alice,
 538                                            op);
 539     break;
 540   case PHASE_BF_EXCHANGE:
 541   case PHASE_MAYBE_FINISHED:
 542     // if we are bob or alice and are continuing operation
 543     GNUNET_CONTAINER_multihashmap_iterate (op->spec->set->elements,
 544                                            &iterator_bf_round,
 545                                            op);
 546     break;
 547   default:
 548     GNUNET_break_op (0);
 549     fail_intersection_operation(op);
 550   }
 551   // the iterators created a new BF with salt+1
 552   // the peer needs this information for decoding the next BF
 553   // this behavior can be modified at will later on.
 554   op->spec->salt++;
 555
 556   GNUNET_CONTAINER_bloomfilter_free (op->state->remote_bf);
 557   op->state->remote_bf = NULL;
 558
 559   peer_elements = ntohl(msg->sender_element_count);
 560   if ((op->state->phase == PHASE_MAYBE_FINISHED)
 561        && (old_elements == op->state->my_element_count)
 562        && (op->state->my_element_count == peer_elements)){
 563     // In the last round we though we were finished, we now know this is correct
 564     send_peer_done(op);
 565     return;
 566   }
 567
 568   op->state->phase = PHASE_BF_EXCHANGE;
 569   // maybe we are finished, but we do one more round to make certain
 570   // we don't have false positives ...
 571   if (op->state->my_element_count == peer_elements)
 572       op->state->phase = PHASE_MAYBE_FINISHED;
 573
 574   send_bloomfilter (op);
 575 }
 576
 577
 578 /**
 579  * Handle an BF message from a remote peer.
 580  *
 581  * @param cls the intersection operation
 582  * @param mh the header of the message
 583  */
 584 static void
 585 handle_p2p_element_info (void *cls, const struct GNUNET_MessageHeader *mh)
 586 {
 587   struct Operation *op = cls;
 588   struct BFMessage *msg = (struct BFMessage *) mh;
 589
 590   op->spec->remote_element_count = ntohl(msg->sender_element_count);
 591   if ((op->state->phase != PHASE_INITIAL)
 592       || (op->state->my_element_count > op->spec->remote_element_count)){
 593     GNUNET_break_op (0);
 594     fail_intersection_operation(op);
 595   }
 596
 597   op->state->phase = PHASE_BF_EXCHANGE;
 598   op->state->my_elements = GNUNET_CONTAINER_multihashmap_create (1, GNUNET_YES);
 599
 600   op->state->local_bf = GNUNET_CONTAINER_bloomfilter_init (NULL,
 601                                                            BLOOMFILTER_SIZE,
 602                                                            GNUNET_CONSTANTS_BLOOMFILTER_K);
 603   GNUNET_CONTAINER_multihashmap_iterate (op->spec->set->elements,
 604                                          &iterator_initialization,
 605                                          op);
 606
 607   GNUNET_CONTAINER_bloomfilter_free (op->state->remote_bf);
 608   op->state->remote_bf = NULL;
 609
 610   if (op->state->my_element_count == ntohl (msg->sender_element_count))
 611     op->state->phase = PHASE_MAYBE_FINISHED;
 612
 613   send_bloomfilter (op);
 614 }
 615
 616
 617 /**
 618  * Send our element to the peer, in case our element count is lower than his
 619  *
 620  * @param op intersection operation
 621  */
 622 static void
 623 send_element_count (struct Operation *op)
 624 {
 625   struct GNUNET_MQ_Envelope *ev;
 626   struct BFMessage *msg;
 627
 628   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "sending element count (bf_msg)\n");
 629
 630   // just send our element count, as the other peer must start
 631   ev = GNUNET_MQ_msg (msg, GNUNET_MESSAGE_TYPE_SET_INTERSECTION_P2P_ELEMENT_INFO);
 632   msg->reserved = 0;
 633   msg->sender_element_count = htonl (op->state->my_element_count);
 634   msg->bloomfilter_length = htonl (0);
 635   msg->sender_mutator = htonl (0);
 636
 637   GNUNET_MQ_send (op->mq, ev);
 638 }
 639
 640
 641 /**
 642  * Send a result message to the client indicating
 643  * that the operation is over.
 644  * After the result done message has been sent to the client,
 645  * destroy the evaluate operation.
 646  *
 647  * @param op intersection operation
 648  */
 649 static void
 650 finish_and_destroy (struct Operation *op)
 651 {
 652   GNUNET_assert (GNUNET_NO == op->state->client_done_sent);
 653
 654   if (GNUNET_SET_RESULT_FULL == op->spec->result_mode)
 655   {
 656     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "sending full result set\n");
 657     op->state->full_result_iter =
 658         GNUNET_CONTAINER_multihashmap_iterator_create (op->state->my_elements);
 659     send_remaining_elements (op);
 660     return;
 661   }
 662   send_client_done_and_destroy (op);
 663 }
 664
 665
 666 /**
 667  * Handle a done message from a remote peer
 668  *
 669  * @param cls the union operation
 670  * @param mh the message
 671  */
 672 static void
 673 handle_p2p_done (void *cls,
 674                  const struct GNUNET_MessageHeader *mh)
 675 {
 676   struct Operation *op = cls;
 677
 678   if ((op->state->phase = PHASE_FINISHED) || (op->state->phase = PHASE_MAYBE_FINISHED)){
 679     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "got final DONE\n");
 680
 681     finish_and_destroy (op);
 682     return;
 683   }
 684
 685   GNUNET_break_op (0);
 686   fail_intersection_operation (op);
 687 }
 688
 689
 690 /**
 691  * Evaluate a union operation with
 692  * a remote peer.
 693  *
 694  * @param op operation to evaluate
 695  */
 696 static void
 697 intersection_evaluate (struct Operation *op)
 698 {
 699   op->state = GNUNET_new (struct OperationState);
 700   /* we started the operation, thus we have to send the operation request */
 701   op->state->phase = PHASE_INITIAL;
 702   op->state->my_elements = GNUNET_CONTAINER_multihashmap_create(1, GNUNET_YES);
 703   op->state->my_element_count = op->spec->set->state->current_set_element_count;
 704
 705   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 706               "evaluating intersection operation");
 707   send_operation_request (op);
 708 }
 709
 710
 711 /**
 712  * Accept an union operation request from a remote peer.
 713  * Only initializes the private operation state.
 714  *
 715  * @param op operation that will be accepted as a union operation
 716  */
 717 static void
 718 intersection_accept (struct Operation *op)
 719 {
 720   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "accepting set union operation\n");
 721   op->state = GNUNET_new (struct OperationState);
 722   op->state->my_elements = GNUNET_CONTAINER_multihashmap_create(1, GNUNET_YES);
 723   op->state->my_element_count = op->spec->set->state->current_set_element_count;
 724
 725   // if Alice (the peer) has more elements than Bob (us), she should start
 726   if (op->spec->remote_element_count < op->state->my_element_count){
 727     op->state->phase = PHASE_INITIAL;
 728     send_element_count(op);
 729     return;
 730   }
 731   // create a new bloomfilter in case we have fewer elements
 732   op->state->phase = PHASE_BF_EXCHANGE;
 733   op->state->local_bf = GNUNET_CONTAINER_bloomfilter_init (NULL,
 734                                                            BLOOMFILTER_SIZE,
 735                                                            GNUNET_CONSTANTS_BLOOMFILTER_K);
 736   GNUNET_CONTAINER_multihashmap_iterate (op->spec->set->elements,
 737                                          &iterator_initialization,
 738                                          op);
 739   send_bloomfilter (op);
 740 }
 741
 742
 743 /**
 744  * Create a new set supporting the intersection operation
 745  *
 746  * @return the newly created set
 747  */
 748 static struct SetState *
 749 intersection_set_create ()
 750 {
 751   struct SetState *set_state;
 752
 753   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 754               "intersection set created\n");
 755   set_state = GNUNET_new (struct SetState);
 756   set_state->current_set_element_count = 0;
 757
 758   return set_state;
 759 }
 760
 761
 762 /**
 763  * Add the element from the given element message to the set.
 764  *
 765  * @param set_state state of the set want to add to
 766  * @param ee the element to add to the set
 767  */
 768 static void
 769 intersection_add (struct SetState *set_state,
 770                   struct ElementEntry *ee)
 771 {
 772   GNUNET_assert(0 < set_state->current_set_element_count);
 773   set_state->current_set_element_count++;
 774 }
 775
 776
 777 /**
 778  * Destroy a set that supports the intersection operation
 779  *
 780  * @param set_state the set to destroy
 781  */
 782 static void
 783 intersection_set_destroy (struct SetState *set_state)
 784 {
 785   GNUNET_free (set_state);
 786 }
 787
 788
 789 /**
 790  * Remove the element given in the element message from the set.
 791  *
 792  * @param set_state state of the set to remove from
 793  * @param element set element to remove
 794  */
 795 static void
 796 intersection_remove (struct SetState *set_state,
 797                      struct ElementEntry *element)
 798 {
 799   GNUNET_assert(0 < set_state->current_set_element_count);
 800   set_state->current_set_element_count--;
 801 }
 802
 803
 804 /**
 805  * Dispatch messages for a intersection operation.
 806  *
 807  * @param op the state of the intersection evaluate operation
 808  * @param mh the received message
 809  * @return #GNUNET_SYSERR if the tunnel should be disconnected,
 810  *         #GNUNET_OK otherwise
 811  */
 812 static int
 813 intersection_handle_p2p_message (struct Operation *op,
 814                                  const struct GNUNET_MessageHeader *mh)
 815 {
 816   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "received p2p message (t: %u, s: %u)\n",
 817               ntohs (mh->type), ntohs (mh->size));
 818   switch (ntohs (mh->type))
 819   {
 820     /* this message handler is not active until after we received an
 821      * operation request message, thus the ops request is not handled here
 822      */
 823   case GNUNET_MESSAGE_TYPE_SET_INTERSECTION_P2P_ELEMENT_INFO:
 824     handle_p2p_element_info (op, mh);
 825     break;
 826   case GNUNET_MESSAGE_TYPE_SET_INTERSECTION_P2P_BF:
 827     handle_p2p_bf (op, mh);
 828     break;
 829   case GNUNET_MESSAGE_TYPE_SET_P2P_DONE:
 830     handle_p2p_done (op, mh);
 831     break;
 832   default:
 833     /* something wrong with mesh's message handlers? */
 834     GNUNET_assert (0);
 835   }
 836   return GNUNET_OK;
 837 }
 838
 839
 840 /**
 841  * handler for peer-disconnects, notifies the client about the aborted operation
 842  *
 843  * @param op the destroyed operation
 844  */
 845 static void
 846 intersection_peer_disconnect (struct Operation *op)
 847 {
 848   if (PHASE_FINISHED != op->state->phase)
 849   {
 850     struct GNUNET_MQ_Envelope *ev;
 851     struct GNUNET_SET_ResultMessage *msg;
 852
 853     ev = GNUNET_MQ_msg (msg, GNUNET_MESSAGE_TYPE_SET_RESULT);
 854     msg->request_id = htonl (op->spec->client_request_id);
 855     msg->result_status = htons (GNUNET_SET_STATUS_FAILURE);
 856     msg->element_type = htons (0);
 857     GNUNET_MQ_send (op->spec->set->client_mq, ev);
 858     GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "other peer disconnected prematurely\n");
 859     _GSS_operation_destroy (op);
 860     return;
 861   }
 862   // else: the session has already been concluded
 863   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "other peer disconnected (finished)\n");
 864   if (GNUNET_NO == op->state->client_done_sent)
 865     finish_and_destroy (op);
 866 }
 867
 868
 869 /**
 870  * Destroy the union operation.  Only things specific to the union operation are destroyed.
 871  *
 872  * @param op union operation to destroy
 873  */
 874 static void
 875 intersection_op_cancel (struct Operation *op)
 876 {
 877   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "destroying intersection op\n");
 878   /* check if the op was canceled twice */
 879   GNUNET_assert (NULL != op->state);
 880   if (NULL != op->state->remote_bf)
 881   {
 882     GNUNET_CONTAINER_bloomfilter_free (op->state->remote_bf);
 883     op->state->remote_bf = NULL;
 884   }
 885   if (NULL != op->state->local_bf)
 886   {
 887     GNUNET_CONTAINER_bloomfilter_free (op->state->local_bf);
 888     op->state->local_bf = NULL;
 889   }
 890   if (NULL != op->state->my_elements)
 891   {
 892     // no need to free the elements, they are still part of the set
 893     GNUNET_CONTAINER_multihashmap_destroy (op->state->my_elements);
 894     op->state->my_elements = NULL;
 895   }
 896   GNUNET_free (op->state);
 897   op->state = NULL;
 898   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "destroying intersection op done\n");
 899 }
 900
 901 const struct SetVT *
 902 _GSS_intersection_vt ()
 903 {
 904   static const struct SetVT intersection_vt = {
 905     .create = &intersection_set_create,
 906     .msg_handler = &intersection_handle_p2p_message,
 907     .add = &intersection_add,
 908     .remove = &intersection_remove,
 909     .destroy_set = &intersection_set_destroy,
 910     .evaluate = &intersection_evaluate,
 911     .accept = &intersection_accept,
 912     .peer_disconnect = &intersection_peer_disconnect,
 913     .cancel = &intersection_op_cancel,
 914   };
 915
 916   return &intersection_vt;
 917 }