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