GNUnet  0.11.x
gnunet-service-set_intersection.c
Go to the documentation of this file.
1 /*
2  This file is part of GNUnet
3  Copyright (C) 2013-2017 GNUnet e.V.
4 
5  GNUnet is free software: you can redistribute it and/or modify it
6  under the terms of the GNU Affero General Public License as published
7  by the Free Software Foundation, either version 3 of the License,
8  or (at your 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  Affero General Public License for more details.
14 
15  You should have received a copy of the GNU Affero General Public License
16  along with this program. If not, see <http://www.gnu.org/licenses/>.
17 
18  SPDX-License-Identifier: AGPL3.0-or-later
19  */
26 #include "platform.h"
27 #include "gnunet_util_lib.h"
29 #include "gnunet-service-set.h"
30 #include "gnunet_block_lib.h"
33 #include <gcrypt.h>
34 
35 
40 {
45 
51 
58 
64 
70 
76 };
77 
78 
83 {
88 
93 
99 
104 
109 
114 
119  char *bf_data;
120 
126  struct GNUNET_HashCode my_xor;
127 
133  struct GNUNET_HashCode other_xor;
134 
138  uint32_t bf_data_offset;
139 
145 
149  uint32_t bf_data_size;
150 
155 
160  uint32_t salt;
161 
166 
171  unsigned int generation_created;
172 
177 
184 };
185 
186 
191 struct SetState
192 {
198 };
199 
200 
208 static void
210  struct GNUNET_SET_Element *element)
211 {
212  struct GNUNET_MQ_Envelope *ev;
213  struct GNUNET_SET_ResultMessage *rm;
214 
215  if (GNUNET_SET_RESULT_REMOVED != op->result_mode)
216  return; /* Wrong mode for transmitting removed elements */
218  "Sending removed element (size %u) to client\n",
219  element->size);
221  "# Element removed messages sent",
222  1,
223  GNUNET_NO);
224  GNUNET_assert (0 != op->client_request_id);
225  ev = GNUNET_MQ_msg_extra (rm,
226  element->size,
228  if (NULL == ev)
229  {
230  GNUNET_break (0);
231  return;
232  }
233  rm->result_status = htons (GNUNET_SET_STATUS_OK);
234  rm->request_id = htonl (op->client_request_id);
235  rm->element_type = element->element_type;
236  GNUNET_memcpy (&rm[1],
237  element->data,
238  element->size);
239  GNUNET_MQ_send (op->set->cs->mq,
240  ev);
241 }
242 
243 
252 static int
254  const struct GNUNET_HashCode *key,
255  void *value)
256 {
257  struct Operation *op = cls;
258  struct ElementEntry *ee = value;
259  struct GNUNET_HashCode mutated_hash;
260 
261 
263  "FIMA called for %s:%u\n",
264  GNUNET_h2s (&ee->element_hash),
265  ee->element.size);
266 
268  {
270  "Reduced initialization, not starting with %s:%u (wrong generation)\n",
271  GNUNET_h2s (&ee->element_hash),
272  ee->element.size);
273  return GNUNET_YES; /* element not valid in our operation's generation */
274  }
275 
276  /* Test if element is in other peer's bloomfilter */
278  op->state->salt,
279  &mutated_hash);
281  "Testing mingled hash %s with salt %u\n",
282  GNUNET_h2s (&mutated_hash),
283  op->state->salt);
284  if (GNUNET_NO ==
285  GNUNET_CONTAINER_bloomfilter_test (op->state->remote_bf,
286  &mutated_hash))
287  {
288  /* remove this element */
290  &ee->element);
292  "Reduced initialization, not starting with %s:%u\n",
293  GNUNET_h2s (&ee->element_hash),
294  ee->element.size);
295  return GNUNET_YES;
296  }
297  op->state->my_element_count++;
298  GNUNET_CRYPTO_hash_xor (&op->state->my_xor,
299  &ee->element_hash,
300  &op->state->my_xor);
302  "Filtered initialization of my_elements, adding %s:%u\n",
303  GNUNET_h2s (&ee->element_hash),
304  ee->element.size);
306  GNUNET_CONTAINER_multihashmap_put (op->state->my_elements,
307  &ee->element_hash,
308  ee,
310 
311  return GNUNET_YES;
312 }
313 
314 
324 static int
326  const struct GNUNET_HashCode *key,
327  void *value)
328 {
329  struct Operation *op = cls;
330  struct ElementEntry *ee = value;
331  struct GNUNET_HashCode mutated_hash;
332 
334  op->state->salt,
335  &mutated_hash);
337  "Testing mingled hash %s with salt %u\n",
338  GNUNET_h2s (&mutated_hash),
339  op->state->salt);
340  if (GNUNET_NO ==
341  GNUNET_CONTAINER_bloomfilter_test (op->state->remote_bf,
342  &mutated_hash))
343  {
344  GNUNET_break (0 < op->state->my_element_count);
345  op->state->my_element_count--;
346  GNUNET_CRYPTO_hash_xor (&op->state->my_xor,
347  &ee->element_hash,
348  &op->state->my_xor);
350  "Bloom filter reduction of my_elements, removing %s:%u\n",
351  GNUNET_h2s (&ee->element_hash),
352  ee->element.size);
354  GNUNET_CONTAINER_multihashmap_remove (op->state->my_elements,
355  &ee->element_hash,
356  ee));
358  &ee->element);
359  }
360  else
361  {
363  "Bloom filter reduction of my_elements, keeping %s:%u\n",
364  GNUNET_h2s (&ee->element_hash),
365  ee->element.size);
366  }
367  return GNUNET_YES;
368 }
369 
370 
379 static int
381  const struct GNUNET_HashCode *key,
382  void *value)
383 {
384  struct Operation *op = cls;
385  struct ElementEntry *ee = value;
386  struct GNUNET_HashCode mutated_hash;
387 
389  op->state->salt,
390  &mutated_hash);
392  "Initializing BF with hash %s with salt %u\n",
393  GNUNET_h2s (&mutated_hash),
394  op->state->salt);
395  GNUNET_CONTAINER_bloomfilter_add (op->state->local_bf,
396  &mutated_hash);
397  return GNUNET_YES;
398 }
399 
400 
407 static void
409 {
410  struct GNUNET_MQ_Envelope *ev;
412 
414  "Intersection operation failed\n");
416  "# Intersection operations failed",
417  1,
418  GNUNET_NO);
419  if (NULL != op->state->my_elements)
420  {
421  GNUNET_CONTAINER_multihashmap_destroy (op->state->my_elements);
422  op->state->my_elements = NULL;
423  }
424  ev = GNUNET_MQ_msg (msg,
426  msg->result_status = htons (GNUNET_SET_STATUS_FAILURE);
427  msg->request_id = htonl (op->client_request_id);
428  msg->element_type = htons (0);
429  GNUNET_MQ_send (op->set->cs->mq,
430  ev);
432  GNUNET_YES);
433 }
434 
435 
442 static void
444 {
445  struct GNUNET_MQ_Envelope *ev;
446  struct BFMessage *msg;
447  uint32_t bf_size;
448  uint32_t bf_elementbits;
449  uint32_t chunk_size;
450  char *bf_data;
451  uint32_t offset;
452 
453  /* We consider the ratio of the set sizes to determine
454  the number of bits per element, as the smaller set
455  should use more bits to maximize its set reduction
456  potential and minimize overall bandwidth consumption. */
457  bf_elementbits = 2 + ceil (log2 ((double)
458  (op->remote_element_count
459  / (double) op->state->my_element_count)));
460  if (bf_elementbits < 1)
461  bf_elementbits = 1; /* make sure k is not 0 */
462  /* optimize BF-size to ~50% of bits set */
463  bf_size = ceil ((double) (op->state->my_element_count
464  * bf_elementbits / log (2)));
466  "Sending Bloom filter (%u) of size %u bytes\n",
467  (unsigned int) bf_elementbits,
468  (unsigned int) bf_size);
469  op->state->local_bf = GNUNET_CONTAINER_bloomfilter_init (NULL,
470  bf_size,
471  bf_elementbits);
473  UINT32_MAX);
474  GNUNET_CONTAINER_multihashmap_iterate (op->state->my_elements,
476  op);
477 
478  /* send our Bloom filter */
480  "# Intersection Bloom filters sent",
481  1,
482  GNUNET_NO);
483  chunk_size = 60 * 1024 - sizeof(struct BFMessage);
484  if (bf_size <= chunk_size)
485  {
486  /* singlepart */
487  chunk_size = bf_size;
488  ev = GNUNET_MQ_msg_extra (msg,
489  chunk_size,
493  op->state->local_bf,
494  (char *) &msg[1],
495  bf_size));
496  msg->sender_element_count = htonl (op->state->my_element_count);
497  msg->bloomfilter_total_length = htonl (bf_size);
498  msg->bits_per_element = htonl (bf_elementbits);
499  msg->sender_mutator = htonl (op->state->salt);
500  msg->element_xor_hash = op->state->my_xor;
501  GNUNET_MQ_send (op->mq, ev);
502  }
503  else
504  {
505  /* multipart */
506  bf_data = GNUNET_malloc (bf_size);
509  op->state->local_bf,
510  bf_data,
511  bf_size));
512  offset = 0;
513  while (offset < bf_size)
514  {
515  if (bf_size - chunk_size < offset)
516  chunk_size = bf_size - offset;
517  ev = GNUNET_MQ_msg_extra (msg,
518  chunk_size,
520  GNUNET_memcpy (&msg[1],
521  &bf_data[offset],
522  chunk_size);
523  offset += chunk_size;
524  msg->sender_element_count = htonl (op->state->my_element_count);
525  msg->bloomfilter_total_length = htonl (bf_size);
526  msg->bits_per_element = htonl (bf_elementbits);
527  msg->sender_mutator = htonl (op->state->salt);
528  msg->element_xor_hash = op->state->my_xor;
529  GNUNET_MQ_send (op->mq, ev);
530  }
531  GNUNET_free (bf_data);
532  }
533  GNUNET_CONTAINER_bloomfilter_free (op->state->local_bf);
534  op->state->local_bf = NULL;
535 }
536 
537 
544 static void
546 {
547  struct Operation *op = cls;
548  struct GNUNET_MQ_Envelope *ev;
549  struct GNUNET_SET_ResultMessage *rm;
550 
552  "Intersection succeeded, sending DONE to local client\n");
554  "# Intersection operations succeeded",
555  1,
556  GNUNET_NO);
557  ev = GNUNET_MQ_msg (rm,
559  rm->request_id = htonl (op->client_request_id);
561  rm->element_type = htons (0);
562  GNUNET_MQ_send (op->set->cs->mq,
563  ev);
565  GNUNET_YES);
566 }
567 
568 
577 static void
579 {
580  struct Operation *op = cls;
581 
583  "DONE sent to other peer, now waiting for other end to close the channel\n");
584  op->state->phase = PHASE_FINISHED;
585  op->state->channel_death_expected = GNUNET_YES;
586 }
587 
588 
596 static void
598 {
599  struct GNUNET_MQ_Envelope *ev;
600  struct IntersectionDoneMessage *idm;
601 
602  GNUNET_assert (PHASE_MUST_SEND_DONE == op->state->phase);
603  GNUNET_assert (GNUNET_NO == op->state->channel_death_expected);
604  ev = GNUNET_MQ_msg (idm,
606  idm->final_element_count = htonl (op->state->my_element_count);
607  idm->element_xor_hash = op->state->my_xor;
610  op);
611  GNUNET_MQ_send (op->mq,
612  ev);
613 }
614 
615 
621 static void
623 {
624  struct Operation *op = cls;
625  const void *nxt;
626  const struct ElementEntry *ee;
627  struct GNUNET_MQ_Envelope *ev;
628  struct GNUNET_SET_ResultMessage *rm;
629  const struct GNUNET_SET_Element *element;
630  int res;
631 
633  op->state->full_result_iter,
634  NULL,
635  &nxt);
636  if (GNUNET_NO == res)
637  {
639  "Sending done and destroy because iterator ran out\n");
641  op->state->full_result_iter);
642  op->state->full_result_iter = NULL;
643  if (PHASE_DONE_RECEIVED == op->state->phase)
644  {
645  op->state->phase = PHASE_FINISHED;
647  }
648  else if (PHASE_MUST_SEND_DONE == op->state->phase)
649  {
650  send_p2p_done (op);
651  }
652  else
653  {
654  GNUNET_assert (0);
655  }
656  return;
657  }
658  ee = nxt;
659  element = &ee->element;
661  "Sending element %s:%u to client (full set)\n",
662  GNUNET_h2s (&ee->element_hash),
663  element->size);
664  GNUNET_assert (0 != op->client_request_id);
665  ev = GNUNET_MQ_msg_extra (rm,
666  element->size,
668  GNUNET_assert (NULL != ev);
669  rm->result_status = htons (GNUNET_SET_STATUS_OK);
670  rm->request_id = htonl (op->client_request_id);
671  rm->element_type = element->element_type;
672  GNUNET_memcpy (&rm[1],
673  element->data,
674  element->size);
677  op);
678  GNUNET_MQ_send (op->set->cs->mq,
679  ev);
680 }
681 
682 
692 static int
694  const struct GNUNET_HashCode *key,
695  void *value)
696 {
697  struct ElementEntry *ee = value;
698  struct Operation *op = cls;
699 
701  return GNUNET_YES; /* element not live in operation's generation */
702  GNUNET_CRYPTO_hash_xor (&op->state->my_xor,
703  &ee->element_hash,
704  &op->state->my_xor);
706  "Initial full initialization of my_elements, adding %s:%u\n",
707  GNUNET_h2s (&ee->element_hash),
708  ee->element.size);
710  GNUNET_CONTAINER_multihashmap_put (op->state->my_elements,
711  &ee->element_hash,
712  ee,
714  return GNUNET_YES;
715 }
716 
717 
724 static void
726 {
727  struct GNUNET_MQ_Envelope *ev;
729 
731  "Sending our element count (%u)\n",
732  op->state->my_element_count);
733  ev = GNUNET_MQ_msg (msg,
735  msg->sender_element_count = htonl (op->state->my_element_count);
736  GNUNET_MQ_send (op->mq, ev);
737 }
738 
739 
746 static void
748 {
749  op->state->phase = PHASE_BF_EXCHANGE;
750  GNUNET_CONTAINER_multihashmap_iterate (op->set->content->elements,
752  op);
754 }
755 
756 
764 void
766  const struct
768 {
769  struct Operation *op = cls;
770 
771  if (GNUNET_SET_OPERATION_INTERSECTION != op->set->operation)
772  {
773  GNUNET_break_op (0);
775  return;
776  }
777  op->remote_element_count = ntohl (msg->sender_element_count);
779  "Received remote element count (%u), I have %u\n",
780  op->remote_element_count,
781  op->state->my_element_count);
782  if (((PHASE_INITIAL != op->state->phase) &&
783  (PHASE_COUNT_SENT != op->state->phase)) ||
784  (op->state->my_element_count > op->remote_element_count) ||
785  (0 == op->state->my_element_count) ||
786  (0 == op->remote_element_count))
787  {
788  GNUNET_break_op (0);
790  return;
791  }
792  GNUNET_break (NULL == op->state->remote_bf);
794  GNUNET_CADET_receive_done (op->channel);
795 }
796 
797 
803 static void
805 {
807  "Received BF in phase %u, foreign count is %u, my element count is %u/%u\n",
808  op->state->phase,
809  op->remote_element_count,
810  op->state->my_element_count,
811  GNUNET_CONTAINER_multihashmap_size (op->set->content->elements));
812  switch (op->state->phase)
813  {
814  case PHASE_INITIAL:
815  GNUNET_break_op (0);
817  return;
818 
819  case PHASE_COUNT_SENT:
820  /* This is the first BF being sent, build our initial map with
821  filtering in place */
822  op->state->my_element_count = 0;
823  GNUNET_CONTAINER_multihashmap_iterate (op->set->content->elements,
825  op);
826  break;
827 
828  case PHASE_BF_EXCHANGE:
829  /* Update our set by reduction */
830  GNUNET_CONTAINER_multihashmap_iterate (op->state->my_elements,
832  op);
833  break;
834 
836  GNUNET_break_op (0);
838  return;
839 
840  case PHASE_DONE_RECEIVED:
841  GNUNET_break_op (0);
843  return;
844 
845  case PHASE_FINISHED:
846  GNUNET_break_op (0);
848  return;
849  }
850  GNUNET_CONTAINER_bloomfilter_free (op->state->remote_bf);
851  op->state->remote_bf = NULL;
852 
853  if ((0 == op->state->my_element_count) || /* fully disjoint */
854  ((op->state->my_element_count == op->remote_element_count) &&
855  (0 == GNUNET_memcmp (&op->state->my_xor,
856  &op->state->other_xor))))
857  {
858  /* we are done */
859  op->state->phase = PHASE_MUST_SEND_DONE;
861  "Intersection succeeded, sending DONE to other peer\n");
862  GNUNET_CONTAINER_bloomfilter_free (op->state->local_bf);
863  op->state->local_bf = NULL;
864  if (GNUNET_SET_RESULT_FULL == op->result_mode)
865  {
867  "Sending full result set (%u elements)\n",
868  GNUNET_CONTAINER_multihashmap_size (op->state->my_elements));
869  op->state->full_result_iter
871  op->state->my_elements);
873  return;
874  }
875  send_p2p_done (op);
876  return;
877  }
878  op->state->phase = PHASE_BF_EXCHANGE;
880 }
881 
882 
890 int
892  const struct BFMessage *msg)
893 {
894  struct Operation *op = cls;
895 
896  if (GNUNET_SET_OPERATION_INTERSECTION != op->set->operation)
897  {
898  GNUNET_break_op (0);
899  return GNUNET_SYSERR;
900  }
901  return GNUNET_OK;
902 }
903 
904 
911 void
913  const struct BFMessage *msg)
914 {
915  struct Operation *op = cls;
916  uint32_t bf_size;
917  uint32_t chunk_size;
918  uint32_t bf_bits_per_element;
919 
920  switch (op->state->phase)
921  {
922  case PHASE_INITIAL:
923  GNUNET_break_op (0);
925  return;
926 
927  case PHASE_COUNT_SENT:
928  case PHASE_BF_EXCHANGE:
929  bf_size = ntohl (msg->bloomfilter_total_length);
930  bf_bits_per_element = ntohl (msg->bits_per_element);
931  chunk_size = htons (msg->header.size) - sizeof(struct BFMessage);
932  op->state->other_xor = msg->element_xor_hash;
933  if (bf_size == chunk_size)
934  {
935  if (NULL != op->state->bf_data)
936  {
937  GNUNET_break_op (0);
939  return;
940  }
941  /* single part, done here immediately */
942  op->state->remote_bf
943  = GNUNET_CONTAINER_bloomfilter_init ((const char *) &msg[1],
944  bf_size,
945  bf_bits_per_element);
946  op->state->salt = ntohl (msg->sender_mutator);
947  op->remote_element_count = ntohl (msg->sender_element_count);
948  process_bf (op);
949  break;
950  }
951  /* multipart chunk */
952  if (NULL == op->state->bf_data)
953  {
954  /* first chunk, initialize */
955  op->state->bf_data = GNUNET_malloc (bf_size);
956  op->state->bf_data_size = bf_size;
957  op->state->bf_bits_per_element = bf_bits_per_element;
958  op->state->bf_data_offset = 0;
959  op->state->salt = ntohl (msg->sender_mutator);
960  op->remote_element_count = ntohl (msg->sender_element_count);
961  }
962  else
963  {
964  /* increment */
965  if ((op->state->bf_data_size != bf_size) ||
966  (op->state->bf_bits_per_element != bf_bits_per_element) ||
967  (op->state->bf_data_offset + chunk_size > bf_size) ||
968  (op->state->salt != ntohl (msg->sender_mutator)) ||
969  (op->remote_element_count != ntohl (msg->sender_element_count)))
970  {
971  GNUNET_break_op (0);
973  return;
974  }
975  }
976  GNUNET_memcpy (&op->state->bf_data[op->state->bf_data_offset],
977  (const char *) &msg[1],
978  chunk_size);
979  op->state->bf_data_offset += chunk_size;
980  if (op->state->bf_data_offset == bf_size)
981  {
982  /* last chunk, run! */
983  op->state->remote_bf
984  = GNUNET_CONTAINER_bloomfilter_init (op->state->bf_data,
985  bf_size,
986  bf_bits_per_element);
987  GNUNET_free (op->state->bf_data);
988  op->state->bf_data = NULL;
989  op->state->bf_data_size = 0;
990  process_bf (op);
991  }
992  break;
993 
994  default:
995  GNUNET_break_op (0);
997  return;
998  }
999  GNUNET_CADET_receive_done (op->channel);
1000 }
1001 
1002 
1011 static int
1012 filter_all (void *cls,
1013  const struct GNUNET_HashCode *key,
1014  void *value)
1015 {
1016  struct Operation *op = cls;
1017  struct ElementEntry *ee = value;
1018 
1019  GNUNET_break (0 < op->state->my_element_count);
1020  op->state->my_element_count--;
1021  GNUNET_CRYPTO_hash_xor (&op->state->my_xor,
1022  &ee->element_hash,
1023  &op->state->my_xor);
1025  "Final reduction of my_elements, removing %s:%u\n",
1026  GNUNET_h2s (&ee->element_hash),
1027  ee->element.size);
1029  GNUNET_CONTAINER_multihashmap_remove (op->state->my_elements,
1030  &ee->element_hash,
1031  ee));
1033  &ee->element);
1034  return GNUNET_YES;
1035 }
1036 
1037 
1044 void
1046  const struct IntersectionDoneMessage *idm)
1047 {
1048  struct Operation *op = cls;
1049 
1050  if (GNUNET_SET_OPERATION_INTERSECTION != op->set->operation)
1051  {
1052  GNUNET_break_op (0);
1054  return;
1055  }
1056  if (PHASE_BF_EXCHANGE != op->state->phase)
1057  {
1058  /* wrong phase to conclude? FIXME: Or should we allow this
1059  if the other peer has _initially_ already an empty set? */
1060  GNUNET_break_op (0);
1062  return;
1063  }
1064  if (0 == ntohl (idm->final_element_count))
1065  {
1066  /* other peer determined empty set is the intersection,
1067  remove all elements */
1068  GNUNET_CONTAINER_multihashmap_iterate (op->state->my_elements,
1069  &filter_all,
1070  op);
1071  }
1072  if ((op->state->my_element_count != ntohl (idm->final_element_count)) ||
1073  (0 != GNUNET_memcmp (&op->state->my_xor,
1074  &idm->element_xor_hash)))
1075  {
1076  /* Other peer thinks we are done, but we disagree on the result! */
1077  GNUNET_break_op (0);
1079  return;
1080  }
1082  "Got IntersectionDoneMessage, have %u elements in intersection\n",
1083  op->state->my_element_count);
1084  op->state->phase = PHASE_DONE_RECEIVED;
1085  GNUNET_CADET_receive_done (op->channel);
1086 
1087  GNUNET_assert (GNUNET_NO == op->state->client_done_sent);
1088  if (GNUNET_SET_RESULT_FULL == op->result_mode)
1089  {
1091  "Sending full result set to client (%u elements)\n",
1092  GNUNET_CONTAINER_multihashmap_size (op->state->my_elements));
1093  op->state->full_result_iter
1094  = GNUNET_CONTAINER_multihashmap_iterator_create (op->state->my_elements);
1096  return;
1097  }
1098  op->state->phase = PHASE_FINISHED;
1100 }
1101 
1102 
1112 static struct OperationState *
1114  const struct GNUNET_MessageHeader *opaque_context)
1115 {
1116  struct OperationState *state;
1117  struct GNUNET_MQ_Envelope *ev;
1118  struct OperationRequestMessage *msg;
1119 
1122  opaque_context);
1123  if (NULL == ev)
1124  {
1125  /* the context message is too large!? */
1126  GNUNET_break (0);
1127  return NULL;
1128  }
1130  "Initiating intersection operation evaluation\n");
1131  state = GNUNET_new (struct OperationState);
1132  /* we started the operation, thus we have to send the operation request */
1133  state->phase = PHASE_INITIAL;
1134  state->my_element_count = op->set->state->current_set_element_count;
1135  state->my_elements
1136  = GNUNET_CONTAINER_multihashmap_create (state->my_element_count,
1137  GNUNET_YES);
1138 
1139  msg->operation = htonl (GNUNET_SET_OPERATION_INTERSECTION);
1140  msg->element_count = htonl (state->my_element_count);
1141  GNUNET_MQ_send (op->mq,
1142  ev);
1143  state->phase = PHASE_COUNT_SENT;
1144  if (NULL != opaque_context)
1146  "Sent op request with context message\n");
1147  else
1149  "Sent op request without context message\n");
1150  return state;
1151 }
1152 
1153 
1160 static struct OperationState *
1162 {
1163  struct OperationState *state;
1164 
1166  "Accepting set intersection operation\n");
1167  state = GNUNET_new (struct OperationState);
1168  state->phase = PHASE_INITIAL;
1169  state->my_element_count
1170  = op->set->state->current_set_element_count;
1171  state->my_elements
1173  op->remote_element_count),
1174  GNUNET_YES);
1175  op->state = state;
1176  if (op->remote_element_count < state->my_element_count)
1177  {
1178  /* If the other peer (Alice) has fewer elements than us (Bob),
1179  we just send the count as Alice should send the first BF */
1181  state->phase = PHASE_COUNT_SENT;
1182  return state;
1183  }
1184  /* We have fewer elements, so we start with the BF */
1186  return state;
1187 }
1188 
1189 
1196 static void
1198 {
1199  /* check if the op was canceled twice */
1200  GNUNET_assert (NULL != op->state);
1201  if (NULL != op->state->remote_bf)
1202  {
1203  GNUNET_CONTAINER_bloomfilter_free (op->state->remote_bf);
1204  op->state->remote_bf = NULL;
1205  }
1206  if (NULL != op->state->local_bf)
1207  {
1208  GNUNET_CONTAINER_bloomfilter_free (op->state->local_bf);
1209  op->state->local_bf = NULL;
1210  }
1211  if (NULL != op->state->my_elements)
1212  {
1213  GNUNET_CONTAINER_multihashmap_destroy (op->state->my_elements);
1214  op->state->my_elements = NULL;
1215  }
1216  if (NULL != op->state->full_result_iter)
1217  {
1219  op->state->full_result_iter);
1220  op->state->full_result_iter = NULL;
1221  }
1222  GNUNET_free (op->state);
1223  op->state = NULL;
1225  "Destroying intersection op state done\n");
1226 }
1227 
1228 
1234 static struct SetState *
1236 {
1237  struct SetState *set_state;
1238 
1240  "Intersection set created\n");
1241  set_state = GNUNET_new (struct SetState);
1242  set_state->current_set_element_count = 0;
1243 
1244  return set_state;
1245 }
1246 
1247 
1254 static void
1255 intersection_add (struct SetState *set_state,
1256  struct ElementEntry *ee)
1257 {
1258  set_state->current_set_element_count++;
1259 }
1260 
1261 
1267 static void
1269 {
1270  GNUNET_free (set_state);
1271 }
1272 
1273 
1280 static void
1281 intersection_remove (struct SetState *set_state,
1282  struct ElementEntry *element)
1283 {
1284  GNUNET_assert (0 < set_state->current_set_element_count);
1285  set_state->current_set_element_count--;
1286 }
1287 
1288 
1294 static void
1296 {
1297  if (GNUNET_YES == op->state->channel_death_expected)
1298  {
1299  /* oh goodie, we are done! */
1301  }
1302  else
1303  {
1304  /* sorry, channel went down early, too bad. */
1306  GNUNET_YES);
1307  }
1308 }
1309 
1310 
1316 const struct SetVT *
1318 {
1319  static const struct SetVT intersection_vt = {
1321  .add = &intersection_add,
1322  .remove = &intersection_remove,
1323  .destroy_set = &intersection_set_destroy,
1324  .evaluate = &intersection_evaluate,
1325  .accept = &intersection_accept,
1326  .cancel = &intersection_op_cancel,
1327  .channel_death = &intersection_channel_death,
1328  };
1329 
1330  return &intersection_vt;
1331 }
struct GNUNET_MessageHeader * msg
Definition: 005.c:2
static struct GNUNET_ARM_Operation * op
Current operation.
Definition: gnunet-arm.c:144
static int res
struct GNUNET_HashCode key
The key used in the DHT.
static char * value
Value of the record to add/remove.
enum State state
current state of profiling
int _GSS_is_element_of_operation(struct ElementEntry *ee, struct Operation *op)
Is element ee part of the set used by op?
struct GNUNET_STATISTICS_Handle * _GSS_statistics
Statistics handle.
void _GSS_operation_destroy(struct Operation *op, int gc)
Destroy the given operation.
common components for the implementation the different set operations
static void intersection_remove(struct SetState *set_state, struct ElementEntry *element)
Remove the element given in the element message from the set.
void handle_intersection_p2p_bf(void *cls, const struct BFMessage *msg)
Handle an BF message from a remote peer.
void handle_intersection_p2p_element_info(void *cls, const struct IntersectionElementInfoMessage *msg)
Handle the initial struct IntersectionElementInfoMessage from a remote peer.
IntersectionOperationPhase
Current phase we are in for a intersection operation.
@ PHASE_INITIAL
We are just starting.
@ PHASE_DONE_RECEIVED
We have received the P2P DONE message, and must finish with the local client before terminating the c...
@ PHASE_FINISHED
The protocol is over.
@ PHASE_BF_EXCHANGE
We have initialized our set and are now reducing it by exchanging Bloom filters until one party notic...
@ PHASE_COUNT_SENT
We have send the number of our elements to the other peer, but did not setup our element set yet.
@ PHASE_MUST_SEND_DONE
We must next send the P2P DONE message (after finishing mostly with the local client).
static int iterator_bf_reduce(void *cls, const struct GNUNET_HashCode *key, void *value)
Removes elements from our hashmap if they are not contained within the provided remote bloomfilter.
static void begin_bf_exchange(struct Operation *op)
We go first, initialize our map with all elements and send the first Bloom filter.
static void intersection_set_destroy(struct SetState *set_state)
Destroy a set that supports the intersection operation.
static int filter_all(void *cls, const struct GNUNET_HashCode *key, void *value)
Remove all elements from our hashmap.
static int filtered_map_initialization(void *cls, const struct GNUNET_HashCode *key, void *value)
Fills the "my_elements" hashmap with all relevant elements.
static void send_p2p_done(struct Operation *op)
Notify the other peer that we are done.
static void send_remaining_elements(void *cls)
Send all elements in the full result iterator.
void handle_intersection_p2p_done(void *cls, const struct IntersectionDoneMessage *idm)
Handle a done message from a remote peer.
static void fail_intersection_operation(struct Operation *op)
Inform the client that the intersection operation has failed, and proceed to destroy the evaluate ope...
const struct SetVT * _GSS_intersection_vt()
Get the table with implementing functions for set intersection.
static void finished_local_operations(void *cls)
Remember that we are done dealing with the local client AND have sent the other peer our message that...
int check_intersection_p2p_bf(void *cls, const struct BFMessage *msg)
Check an BF message from a remote peer.
static void send_client_done_and_destroy(void *cls)
Signal to the client that the operation has finished and destroy the operation.
static struct SetState * intersection_set_create()
Create a new set supporting the intersection operation.
static int initialize_map_unfiltered(void *cls, const struct GNUNET_HashCode *key, void *value)
Fills the "my_elements" hashmap with the initial set of (non-deleted) elements from the set of the sp...
static void send_bloomfilter(struct Operation *op)
Send a bloomfilter to our peer.
static void intersection_op_cancel(struct Operation *op)
Destroy the intersection operation.
static void process_bf(struct Operation *op)
Process a Bloomfilter once we got all the chunks.
static void send_element_count(struct Operation *op)
Send our element count to the peer, in case our element count is lower than theirs.
static void intersection_add(struct SetState *set_state, struct ElementEntry *ee)
Add the element from the given element message to the set.
static void intersection_channel_death(struct Operation *op)
Callback for channel death for the intersection operation.
static void send_client_removed_element(struct Operation *op, struct GNUNET_SET_Element *element)
If applicable in the current operation mode, send a result message to the client indicating we remove...
static int iterator_bf_create(void *cls, const struct GNUNET_HashCode *key, void *value)
Create initial bloomfilter based on all the elements given.
static struct OperationState * intersection_accept(struct Operation *op)
Accept an intersection operation request from a remote peer.
static struct OperationState * intersection_evaluate(struct Operation *op, const struct GNUNET_MessageHeader *opaque_context)
Initiate a set intersection operation with a remote peer.
two-peer set operations
Peer-to-Peer messages for gnunet set.
#define GNUNET_log(kind,...)
#define GNUNET_memcmp(a, b)
Compare memory in a and b, where both must be of the same pointer type.
#define GNUNET_memcpy(dst, src, n)
Call memcpy() but check for n being 0 first.
@ GNUNET_OK
Definition: gnunet_common.h:95
@ GNUNET_YES
Definition: gnunet_common.h:97
@ GNUNET_NO
Definition: gnunet_common.h:94
@ GNUNET_SYSERR
Definition: gnunet_common.h:93
#define GNUNET_MIN(a, b)
void GNUNET_BLOCK_mingle_hash(const struct GNUNET_HashCode *in, uint32_t mingle_number, struct GNUNET_HashCode *hc)
Mingle hash with the mingle_number to produce different bits.
Definition: block.c:81
int GNUNET_CONTAINER_bloomfilter_get_raw_data(const struct GNUNET_CONTAINER_BloomFilter *bf, char *data, size_t size)
Copy the raw data of this Bloom filter into the given data array.
void GNUNET_CONTAINER_bloomfilter_add(struct GNUNET_CONTAINER_BloomFilter *bf, const struct GNUNET_HashCode *e)
Add an element to the filter.
int GNUNET_CONTAINER_bloomfilter_test(const struct GNUNET_CONTAINER_BloomFilter *bf, const struct GNUNET_HashCode *e)
Test if an element is in the filter.
struct GNUNET_CONTAINER_BloomFilter * GNUNET_CONTAINER_bloomfilter_init(const char *data, size_t size, unsigned int k)
Create a Bloom filter from raw bits.
void GNUNET_CONTAINER_bloomfilter_free(struct GNUNET_CONTAINER_BloomFilter *bf)
Free the space associated with a filter in memory, flush to drive if needed (do not free the space on...
void GNUNET_CADET_receive_done(struct GNUNET_CADET_Channel *channel)
Send an ack on the channel to confirm the processing of a message.
Definition: cadet_api.c:888
uint32_t GNUNET_CRYPTO_random_u32(enum GNUNET_CRYPTO_Quality mode, uint32_t i)
Produce a random value.
@ GNUNET_CRYPTO_QUALITY_NONCE
Randomness for IVs etc.
void GNUNET_CRYPTO_hash_xor(const struct GNUNET_HashCode *a, const struct GNUNET_HashCode *b, struct GNUNET_HashCode *result)
compute result = a ^ b
Definition: crypto_hash.c:134
int GNUNET_CONTAINER_multihashmap_remove(struct GNUNET_CONTAINER_MultiHashMap *map, const struct GNUNET_HashCode *key, const void *value)
Remove the given key-value pair from the map.
int GNUNET_CONTAINER_multihashmap_iterator_next(struct GNUNET_CONTAINER_MultiHashMapIterator *iter, struct GNUNET_HashCode *key, const void **value)
Retrieve the next element from the hash map at the iterator's position.
struct GNUNET_CONTAINER_MultiHashMap * GNUNET_CONTAINER_multihashmap_create(unsigned int len, int do_not_copy_keys)
Create a multi hash map.
int GNUNET_CONTAINER_multihashmap_put(struct GNUNET_CONTAINER_MultiHashMap *map, const struct GNUNET_HashCode *key, void *value, enum GNUNET_CONTAINER_MultiHashMapOption opt)
Store a key-value pair in the map.
unsigned int GNUNET_CONTAINER_multihashmap_size(const struct GNUNET_CONTAINER_MultiHashMap *map)
Get the number of key-value pairs in the map.
int GNUNET_CONTAINER_multihashmap_iterate(struct GNUNET_CONTAINER_MultiHashMap *map, GNUNET_CONTAINER_MulitHashMapIteratorCallback it, void *it_cls)
Iterate over all entries in the map.
void GNUNET_CONTAINER_multihashmap_destroy(struct GNUNET_CONTAINER_MultiHashMap *map)
Destroy a hash map.
struct GNUNET_CONTAINER_MultiHashMapIterator * GNUNET_CONTAINER_multihashmap_iterator_create(const struct GNUNET_CONTAINER_MultiHashMap *map)
Create an iterator for a multihashmap.
void GNUNET_CONTAINER_multihashmap_iterator_destroy(struct GNUNET_CONTAINER_MultiHashMapIterator *iter)
Destroy a multihashmap iterator.
@ GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY
There must only be one value per key; storing a value should fail if a value under the same key alrea...
#define GNUNET_break_op(cond)
Use this for assertion violations caused by other peers (i.e.
#define GNUNET_assert(cond)
Use this for fatal errors that cannot be handled.
#define GNUNET_break(cond)
Use this for internal assertion violations that are not fatal (can be handled) but should not occur.
const char * GNUNET_h2s(const struct GNUNET_HashCode *hc)
Convert a hash value to a string (for printing debug messages).
@ GNUNET_ERROR_TYPE_WARNING
@ GNUNET_ERROR_TYPE_DEBUG
#define GNUNET_new(type)
Allocate a struct or union of the given type.
#define GNUNET_malloc(size)
Wrapper around malloc.
#define GNUNET_free(ptr)
Wrapper around free.
void GNUNET_MQ_send(struct GNUNET_MQ_Handle *mq, struct GNUNET_MQ_Envelope *ev)
Send a message with the given message queue.
Definition: mq.c:355
#define GNUNET_MQ_msg_extra(mvar, esize, type)
Allocate an envelope, with extra space allocated after the space needed by the message struct.
Definition: gnunet_mq_lib.h:52
#define GNUNET_MQ_msg_nested_mh(mvar, type, mh)
Allocate a GNUNET_MQ_Envelope, and append a payload message after the given message struct.
#define GNUNET_MQ_msg(mvar, type)
Allocate a GNUNET_MQ_Envelope.
Definition: gnunet_mq_lib.h:67
void GNUNET_MQ_notify_sent(struct GNUNET_MQ_Envelope *ev, GNUNET_SCHEDULER_TaskCallback cb, void *cb_cls)
Call a callback once the envelope has been sent, that is, sending it can not be canceled anymore.
Definition: mq.c:787
#define GNUNET_MESSAGE_TYPE_SET_P2P_OPERATION_REQUEST
Request a set operation from a remote peer.
#define GNUNET_MESSAGE_TYPE_SET_INTERSECTION_P2P_DONE
Intersection operation is done.
#define GNUNET_MESSAGE_TYPE_SET_RESULT
Create an empty set.
#define GNUNET_MESSAGE_TYPE_SET_INTERSECTION_P2P_ELEMENT_INFO
Information about the element count for intersection.
#define GNUNET_MESSAGE_TYPE_SET_INTERSECTION_P2P_BF
Bloom filter message for intersection exchange started by Bob.
@ GNUNET_SET_STATUS_OK
Everything went ok, we are transmitting an element of the result (in set, or to be removed from set,...
@ GNUNET_SET_STATUS_FAILURE
The other peer refused to to the operation with us, or something went wrong.
@ GNUNET_SET_STATUS_DONE
Success, all elements have been sent (and received).
@ GNUNET_SET_RESULT_REMOVED
Client gets only elements that have been removed from the set.
@ GNUNET_SET_RESULT_FULL
Client gets every element in the resulting set.
@ GNUNET_SET_OPERATION_INTERSECTION
Set intersection, only return elements that are in both sets.
void GNUNET_STATISTICS_update(struct GNUNET_STATISTICS_Handle *handle, const char *name, int64_t delta, int make_persistent)
Set statistic value for the peer.
Bloom filter messages exchanged for set intersection calculation.
Information about an element element in the set.
struct GNUNET_SET_Element element
The actual element.
struct GNUNET_HashCode element_hash
Hash of the element.
Internal representation of the hash map.
A 512-bit hashcode.
Header for all communications.
uint16_t size
The length of the struct (in bytes, including the length field itself), in big-endian format.
Element stored in a set.
uint16_t size
Number of bytes in the buffer pointed to by data.
const void * data
Actual data of the element.
uint16_t element_type
Application-specific element type.
Message sent by the service to the client to indicate an element that is removed (set intersection) o...
Definition: set.h:245
uint32_t request_id
id the result belongs to
Definition: set.h:259
uint16_t result_status
Was the evaluation successful? Contains an enum GNUNET_SET_Status in NBO.
Definition: set.h:265
uint16_t element_type
Type of the element attached to the message, if any.
Definition: set.h:270
Last message, send to confirm the final set.
uint32_t final_element_count
Final number of elements in intersection.
struct GNUNET_HashCode element_xor_hash
XOR of all hashes over all elements remaining in the set.
During intersection, the first (and possibly second) message send it the number of elements in the se...
State of an evaluate operation with another peer.
uint32_t bf_bits_per_element
size of the bloomfilter
struct GNUNET_CONTAINER_MultiHashMap * my_elements
Remaining elements in the intersection operation.
uint32_t salt
Salt currently used for BF construction (by us or the other peer, depending on where we are in the co...
struct GNUNET_CONTAINER_BloomFilter * local_bf
BF of the set's element.
struct OperationState * next
Evaluate operations are held in a linked list.
unsigned int generation_created
Generation in which the operation handle was created.
uint32_t bf_data_offset
How many bytes of bf_data are valid?
struct GNUNET_HashCode my_xor
XOR of the keys of all of the elements (remaining) in my set.
enum IntersectionOperationPhase phase
Current state of the operation.
char * bf_data
For multipart BF transmissions, we have to store the bloomfilter-data until we fully received it.
struct GNUNET_HashCode other_xor
XOR of the keys of all of the elements (remaining) in the other peer's set.
uint32_t my_element_count
Current element count contained within my_elements.
struct GNUNET_CONTAINER_MultiHashMapIterator * full_result_iter
Iterator for sending the final set of my_elements to the client.
int channel_death_expected
Set whenever we reach the state where the death of the channel is perfectly find and should NOT resul...
uint32_t bf_data_size
size of the bloomfilter in bf_data.
struct GNUNET_CONTAINER_BloomFilter * remote_bf
The bf we currently receive.
struct OperationState * prev
Evaluate operations are held in a linked list.
int client_done_sent
Did we send the client that we are done?
Operation context used to execute a set operation.
uint32_t bf_bits_per_element
size of the bloomfilter
Extra state required for efficient set intersection.
uint32_t current_set_element_count
Number of currently valid elements in the set which have not been removed.
Dispatch table for a specific set operation.
SetCreateImpl create
Callback for the set creation.