2 This file is part of GNUnet
3 (C) 2013 Christian Grothoff (and other contributing authors)
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 2, or (at your
8 option) any later version.
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.
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.
22 * @file set/gnunet-service-set.c
23 * @brief two-peer set operations
24 * @author Florian Dold
28 #include "gnunet-service-set.h"
29 #include "gnunet_container_lib.h"
30 #include "gnunet_crypto_lib.h"
32 #include "strata_estimator.h"
33 #include "set_protocol.h"
38 * Number of IBFs in a strata estimator.
40 #define SE_STRATA_COUNT 32
42 * Size of the IBFs in the strata estimator.
44 #define SE_IBF_SIZE 80
46 * hash num parameter for the difference digests and strata estimators
48 #define SE_IBF_HASH_NUM 3
51 * Number of buckets that can be transmitted in one message.
53 #define MAX_BUCKETS_PER_MESSAGE ((1<<15) / IBF_BUCKET_SIZE)
56 * The maximum size of an ibf we use is 2^(MAX_IBF_ORDER).
57 * Choose this value so that computing the IBF is still cheaper
58 * than transmitting all values.
60 #define MAX_IBF_ORDER (16)
63 enum UnionOperationState
67 STATE_EXPECT_IBF_CONT,
68 STATE_EXPECT_ELEMENTS,
69 STATE_EXPECT_ELEMENTS_AND_REQUESTS,
76 * State of an evaluate operation
79 struct UnionEvaluateOperation
82 * Local set the operation is evaluated on
87 * Peer with the remote set
89 struct GNUNET_PeerIdentity peer;
92 * Application-specific identifier
94 struct GNUNET_HashCode app_id;
97 * Context message, given to us
98 * by the client, may be NULL.
100 struct GNUNET_MessageHeader *context_msg;
103 * Stream socket connected to the other peer
105 struct GNUNET_STREAM_Socket *socket;
108 * Message queue for the peer on the other
111 struct GNUNET_MQ_MessageQueue *mq;
114 * Type of this operation
116 enum GNUNET_SET_OperationType operation;
119 * GNUNET_YES if we started the operation,
120 * GNUNET_NO if the other peer started it.
125 * Request id, so we can use one client handle
126 * for multiple operations
130 /* last difference estimate */
134 * Number of ibf buckets received
136 unsigned int ibf_buckets_received;
139 * Current salt in use, zero unless
140 * we detected a collision
145 * order of the ibf we receive
147 unsigned int ibf_order;
149 struct StrataEstimator *se;
152 * The ibf we currently receive
154 struct InvertibleBloomFilter *remote_ibf;
157 * Array of IBFs, some of them pre-allocated
159 struct InvertibleBloomFilter *local_ibf;
162 * Elements we received from the other peer.
164 struct GNUNET_CONTAINER_MultiHashMap *received_elements;
167 * Maps IBF-Keys (specific to the current salt) to elements.
169 struct GNUNET_CONTAINER_MultiHashMap32 *key_to_element;
172 * Current state of the operation
174 enum UnionOperationState state;
177 * Evaluate operations are held in
180 struct UnionEvaluateOperation *next;
183 * Evaluate operations are held in
186 struct UnionEvaluateOperation *prev;
190 * Information about the element in a set.
191 * All elements are stored in a hash-table
192 * from their hash-code to their 'struct Element',
193 * so that the remove and add operations are reasonably
199 * The actual element. The data for the element
200 * should be allocated at the end of this struct.
202 struct GNUNET_SET_Element element;
205 * Hash of the element.
206 * Will be used to derive the different IBF keys
207 * for different salts.
209 struct GNUNET_HashCode element_hash;
212 * Generation the element was added.
213 * Operations of earlier generations will not consider the element.
218 * Generation this element was removed.
219 * Operations of later generations will not consider the element.
221 int generation_remove;
224 * GNUNET_YES if we received the element from a remote peer, and not
225 * from the local peer. Note that if the local client inserts an
226 * element *after* we got it from a remote peer, the element is
233 * Information about the element used for
234 * a specific union operation.
238 struct IBF_Key ibf_key;
241 * The actual element associated with the key
243 struct ElementEntry *element;
246 * Element that collides with this element
249 struct KeyEntry *next_colliding;
254 * Extra state required for efficient set union.
259 * The strata estimator is only generated once for
262 struct StrataEstimator *se;
265 * Maps 'struct GNUNET_HashCode' to 'struct ElementEntry'.
267 struct GNUNET_CONTAINER_MultiHashMap *elements;
270 * Evaluate operations are held in
273 struct UnionEvaluateOperation *ops_head;
276 * Evaluate operations are held in
279 struct UnionEvaluateOperation *ops_tail;
283 static struct IBF_Key
284 get_ibf_key (struct GNUNET_HashCode *src, uint16_t salt)
288 GNUNET_CRYPTO_hkdf (&key, sizeof (key),
289 GCRY_MD_SHA512, GCRY_MD_SHA256,
291 &salt, sizeof (salt),
298 send_operation_request (struct UnionEvaluateOperation *eo)
300 struct GNUNET_MQ_Message *mqm;
301 struct OperationRequestMessage *msg;
304 mqm = GNUNET_MQ_msg (msg, GNUNET_MESSAGE_TYPE_SET_P2P_OPERATION_REQUEST);
305 ret = GNUNET_MQ_nest (mqm, eo->context_msg);
306 if (GNUNET_OK != ret)
308 /* the context message is too large */
309 _GSS_client_disconnect (eo->set->client);
310 GNUNET_MQ_discard (mqm);
314 msg->operation = eo->operation;
315 msg->app_id = eo->app_id;
316 GNUNET_MQ_send (eo->mq, mqm);
318 GNUNET_log (GNUNET_ERROR_TYPE_INFO, "sent op request\n");
323 * Iterator to create the mapping between ibf keys
324 * and element entries.
327 * @param key current key code
328 * @param value value in the hash map
329 * @return GNUNET_YES if we should continue to
334 insert_element_iterator (void *cls,
338 struct KeyEntry *const new_k = cls;
339 struct KeyEntry *old_k = value;
341 GNUNET_assert (NULL != old_k);
344 if (old_k->ibf_key.key_val == new_k->ibf_key.key_val)
346 new_k->next_colliding = old_k;
347 old_k->next_colliding = new_k;
350 old_k = old_k->next_colliding;
351 } while (NULL != old_k);
357 insert_element (struct UnionEvaluateOperation *eo, struct ElementEntry *ee)
360 struct IBF_Key ibf_key;
363 ibf_key = get_ibf_key (&ee->element_hash, eo->salt);
364 k = GNUNET_new (struct KeyEntry);
366 k->ibf_key = ibf_key;
367 ret = GNUNET_CONTAINER_multihashmap32_get_multiple (eo->key_to_element, (uint32_t) ibf_key.key_val,
368 insert_element_iterator, k);
369 /* was the element inserted into a colliding bucket? */
370 if (GNUNET_SYSERR == ret)
372 GNUNET_assert (NULL != k->next_colliding);
375 GNUNET_CONTAINER_multihashmap32_put (eo->key_to_element, (uint32_t) ibf_key.key_val, k,
376 GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY);
377 if (NULL != eo->local_ibf)
378 ibf_insert (eo->local_ibf, ibf_key);
383 prepare_ibf_iterator (void *cls,
387 struct InvertibleBloomFilter *ibf = cls;
388 struct KeyEntry *ke = value;
390 ibf_insert (ibf, ke->ibf_key);
395 init_key_to_element_iterator (void *cls,
396 const struct GNUNET_HashCode *key,
399 struct UnionEvaluateOperation *eo = cls;
400 struct ElementEntry *e = value;
402 insert_element (eo, e);
407 prepare_ibf (struct UnionEvaluateOperation *eo, uint16_t size)
409 if (NULL == eo->key_to_element)
412 len = GNUNET_CONTAINER_multihashmap_size (eo->set->state.u->elements);
413 eo->key_to_element = GNUNET_CONTAINER_multihashmap32_create (len);
414 GNUNET_CONTAINER_multihashmap_iterate (eo->set->state.u->elements,
415 init_key_to_element_iterator, eo);
417 if (NULL != eo->local_ibf)
418 ibf_destroy (eo->local_ibf);
419 eo->local_ibf = ibf_create (size, SE_IBF_HASH_NUM);
420 GNUNET_CONTAINER_multihashmap32_iterate (eo->key_to_element, prepare_ibf_iterator, eo->local_ibf);
425 * Send an ibf of appropriate size.
427 * @param cpi the peer
430 send_ibf (struct UnionEvaluateOperation *eo, uint16_t ibf_order)
432 unsigned int buckets_sent = 0;
433 struct InvertibleBloomFilter *ibf;
435 prepare_ibf (eo, ibf_order);
439 while (buckets_sent < (1 << ibf_order))
441 unsigned int buckets_in_message;
442 struct GNUNET_MQ_Message *mqm;
443 struct IBFMessage *msg;
445 buckets_in_message = (1 << ibf_order) - buckets_sent;
446 /* limit to maximum */
447 if (buckets_in_message > MAX_BUCKETS_PER_MESSAGE)
448 buckets_in_message = MAX_BUCKETS_PER_MESSAGE;
450 mqm = GNUNET_MQ_msg_extra (msg, buckets_in_message * IBF_BUCKET_SIZE,
451 GNUNET_MESSAGE_TYPE_SET_P2P_IBF);
452 msg->order = htons (ibf_order);
453 msg->offset = htons (buckets_sent);
454 ibf_write_slice (ibf, buckets_sent,
455 buckets_in_message, &msg[1]);
456 buckets_sent += buckets_in_message;
457 GNUNET_MQ_send (eo->mq, mqm);
460 eo->state = STATE_EXPECT_ELEMENTS_AND_REQUESTS;
465 * Send a strata estimator.
467 * @param cpi the peer
470 send_strata_estimator (struct UnionEvaluateOperation *eo)
472 struct GNUNET_MQ_Message *mqm;
473 struct GNUNET_MessageHeader *strata_msg;
475 mqm = GNUNET_MQ_msg_header_extra (strata_msg,
476 SE_STRATA_COUNT * IBF_BUCKET_SIZE * SE_IBF_SIZE,
477 GNUNET_MESSAGE_TYPE_SET_P2P_SE);
478 strata_estimator_write (eo->set->state.u->se, &strata_msg[1]);
479 GNUNET_MQ_send (eo->mq, mqm);
480 eo->state = STATE_EXPECT_IBF;
485 handle_p2p_strata_estimator (void *cls, const struct GNUNET_MessageHeader *mh)
487 struct UnionEvaluateOperation *eo = cls;
488 struct StrataEstimator *remote_se;
492 GNUNET_log (GNUNET_ERROR_TYPE_INFO, "got se\n");
494 if (eo->state != STATE_EXPECT_SE)
500 remote_se = strata_estimator_create (SE_STRATA_COUNT, SE_IBF_SIZE,
502 strata_estimator_read (&mh[1], remote_se);
503 GNUNET_assert (NULL != eo->se);
504 diff = strata_estimator_difference (remote_se, eo->se);
505 strata_estimator_destroy (remote_se);
506 strata_estimator_destroy (eo->se);
510 while ((1<<ibf_order) < (2 * diff))
512 if (ibf_order > MAX_IBF_ORDER)
513 ibf_order = MAX_IBF_ORDER;
514 send_ibf (eo, ibf_order);
524 decode (struct UnionEvaluateOperation *eo)
528 struct InvertibleBloomFilter *diff_ibf;
530 GNUNET_assert (STATE_EXPECT_ELEMENTS == eo->state);
532 prepare_ibf (eo, eo->ibf_order);
533 diff_ibf = ibf_dup (eo->local_ibf);
534 ibf_subtract (diff_ibf, eo->remote_ibf);
540 res = ibf_decode (diff_ibf, &side, &key);
541 if (GNUNET_SYSERR == res)
543 /* decoding failed, we tell the other peer by sending our ibf
544 * with a larger order */
548 if (GNUNET_NO == res)
550 struct GNUNET_MQ_Message *mqm;
551 GNUNET_log (GNUNET_ERROR_TYPE_INFO, "transmitted all values, sending DONE\n");
552 mqm = GNUNET_MQ_msg_header (GNUNET_MESSAGE_TYPE_SET_P2P_DONE);
553 GNUNET_MQ_send (eo->mq, mqm);
558 //struct ElementEntry *e;
559 /* we have the element(s), send it to the other peer */
560 //GNUNET_CONTAINER_multihashmap32_get_multiple (eo->set->state.u->elements,
561 // (uint32_t) key.key_val);
566 struct GNUNET_MQ_Message *mqm;
567 struct GNUNET_MessageHeader *msg;
569 /* FIXME: before sending the request, check if we may just have the element */
570 /* FIXME: merge multiple requests */
571 mqm = GNUNET_MQ_msg_header_extra (msg, sizeof (struct IBF_Key), GNUNET_MESSAGE_TYPE_SET_P2P_ELEMENT_REQUESTS);
572 *(struct IBF_Key *) &msg[1] = key;
573 GNUNET_MQ_send (eo->mq, mqm);
580 handle_p2p_ibf (void *cls, const struct GNUNET_MessageHeader *mh)
582 struct UnionEvaluateOperation *eo = cls;
583 struct IBFMessage *msg = (struct IBFMessage *) mh;
584 unsigned int buckets_in_message;
586 if (eo->state == STATE_EXPECT_ELEMENTS_AND_REQUESTS)
588 /* check that the ibf is a new one / first part */
589 /* clear outgoing messages */
592 else if (eo->state == STATE_EXPECT_IBF)
594 eo->state = STATE_EXPECT_IBF_CONT;
595 eo->ibf_order = msg->order;
596 GNUNET_assert (NULL == eo->remote_ibf);
597 eo->remote_ibf = ibf_create (1<<msg->order, SE_IBF_HASH_NUM);
598 if (ntohs (msg->offset) != 0)
604 else if (eo->state == STATE_EXPECT_IBF_CONT)
606 if ( (ntohs (msg->offset) != eo->ibf_buckets_received) ||
607 (msg->order != eo->ibf_order) )
614 buckets_in_message = (ntohs (msg->header.size) - sizeof *msg) / IBF_BUCKET_SIZE;
616 if ((ntohs (msg->header.size) - sizeof *msg) != buckets_in_message * IBF_BUCKET_SIZE)
618 /* FIXME: handle, message was malformed */
622 ibf_read_slice (&msg[1], eo->ibf_buckets_received, buckets_in_message, eo->remote_ibf);
623 eo->ibf_buckets_received += buckets_in_message;
625 if (eo->ibf_buckets_received == (1<<eo->ibf_order))
627 eo->state = STATE_EXPECT_ELEMENTS;
634 handle_p2p_elements (void *cls, const struct GNUNET_MessageHeader *mh)
636 struct UnionEvaluateOperation *eo = cls;
638 if ( (eo->state != STATE_EXPECT_ELEMENTS) &&
639 (eo->state != STATE_EXPECT_ELEMENTS_AND_REQUESTS) )
649 handle_p2p_element_requests (void *cls, const struct GNUNET_MessageHeader *mh)
651 struct UnionEvaluateOperation *eo = cls;
653 /* look up elements and send them */
654 if (eo->state != STATE_EXPECT_ELEMENTS_AND_REQUESTS)
664 handle_p2p_done (void *cls, const struct GNUNET_MessageHeader *mh)
670 static const struct GNUNET_MQ_Handler union_handlers[] = {
671 {handle_p2p_elements, GNUNET_MESSAGE_TYPE_SET_P2P_ELEMENTS},
672 {handle_p2p_strata_estimator, GNUNET_MESSAGE_TYPE_SET_P2P_SE},
673 {handle_p2p_ibf, GNUNET_MESSAGE_TYPE_SET_P2P_IBF},
674 {handle_p2p_element_requests, GNUNET_MESSAGE_TYPE_SET_P2P_ELEMENT_REQUESTS},
675 {handle_p2p_done, GNUNET_MESSAGE_TYPE_SET_P2P_DONE},
676 GNUNET_MQ_HANDLERS_END
681 * Functions of this type will be called when a stream is established
683 * @param cls the closure from GNUNET_STREAM_open
684 * @param socket socket to use to communicate with the
685 * other side (read/write)
688 stream_open_cb (void *cls,
689 struct GNUNET_STREAM_Socket *socket)
691 struct UnionEvaluateOperation *eo = cls;
693 GNUNET_assert (NULL == eo->mq);
694 GNUNET_assert (socket == eo->socket);
696 GNUNET_log (GNUNET_ERROR_TYPE_INFO, "open cb successful\n");
698 eo->mq = GNUNET_MQ_queue_for_stream_socket (eo->socket,
700 /* we started the operation, thus we have to send the operation request */
701 send_operation_request (eo);
702 eo->state = STATE_EXPECT_SE;
707 _GSS_union_evaluate (struct EvaluateMessage *m, struct Set *set)
709 struct UnionEvaluateOperation *eo;
711 GNUNET_log (GNUNET_ERROR_TYPE_INFO, "evaluating union operation\n");
713 eo = GNUNET_new (struct UnionEvaluateOperation);
717 GNUNET_STREAM_open (configuration, &eo->peer, GNUNET_APPLICATION_TYPE_SET,
719 GNUNET_STREAM_OPTION_END);
724 _GSS_union_accept (struct AcceptMessage *m, struct Set *set,
725 struct Incoming *incoming)
727 struct UnionEvaluateOperation *eo;
729 eo = GNUNET_new (struct UnionEvaluateOperation);
731 eo->peer = incoming->peer;
732 eo->app_id = incoming->app_id;
733 eo->salt = ntohs (incoming->salt);
734 eo->request_id = m->request_id;
736 eo->mq = incoming->mq;
737 /* the peer's socket is now ours, we'll receive all messages */
738 GNUNET_MQ_replace_handlers (eo->mq, union_handlers, eo);
739 /* kick of the operation */
740 send_strata_estimator (eo);
745 _GSS_union_set_create (void)
749 GNUNET_log (GNUNET_ERROR_TYPE_INFO, "set created\n");
751 set = GNUNET_malloc (sizeof (struct Set) + sizeof (struct UnionState));
752 set->state.u = (struct UnionState *) &set[1];
753 set->operation = GNUNET_SET_OPERATION_UNION;
754 set->state.u->se = strata_estimator_create (SE_STRATA_COUNT,
755 SE_IBF_SIZE, SE_IBF_HASH_NUM);
762 _GSS_union_add (struct ElementMessage *m, struct Set *set)
764 struct ElementEntry *ee;
765 struct ElementEntry *ee_dup;
766 uint16_t element_size;
768 element_size = ntohs (m->header.size) - sizeof *m;
769 ee = GNUNET_malloc (element_size + sizeof *ee);
770 ee->element.size = element_size;
771 ee->element.data = &ee[1];
772 memcpy (ee->element.data, &m[1], element_size);
773 GNUNET_CRYPTO_hash (ee->element.data, element_size, &ee->element_hash);
774 ee_dup = GNUNET_CONTAINER_multihashmap_get (set->state.u->elements, &ee->element_hash);
777 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "element inserted twice, ignoring\n");
781 GNUNET_CONTAINER_multihashmap_put (set->state.u->elements, &ee->element_hash, ee,
782 GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY);
783 strata_estimator_insert (set->state.u->se, get_ibf_key (&ee->element_hash, 0));