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 "set_protocol.h"
31 #include "strata_estimator.h"
35 * Number of IBFs in a strata estimator.
37 #define SE_STRATA_COUNT 32
39 * Size of the IBFs in the strata estimator.
41 #define SE_IBF_SIZE 80
43 * hash num parameter for the difference digests and strata estimators
45 #define SE_IBF_HASH_NUM 3
48 * Number of buckets that can be transmitted in one message.
50 #define MAX_BUCKETS_PER_MESSAGE ((1<<15) / IBF_BUCKET_SIZE)
54 * The maximum size of an ibf we use is 2^(MAX_IBF_ORDER).
55 * Choose this value so that computing the IBF is still cheaper
56 * than transmitting all values.
58 #define MAX_IBF_ORDER (16)
61 enum UnionOperationState
65 STATE_EXPECT_IBF_CONT,
66 STATE_EXPECT_ELEMENTS,
67 STATE_EXPECT_ELEMENTS_AND_REQUESTS,
74 * State of an evaluate operation
77 struct UnionEvaluateOperation
79 /* last difference estimate */
83 * Number of ibf buckets received
85 unsigned int ibf_buckets_received;
88 * Current salt in use, zero unless
89 * we detected a collision
94 * order of the ibf we receive
96 unsigned int ibf_order;
99 * The ibf we currently receive
101 struct InvertibleBloomFilter *ibf_received;
103 struct StrataEstimator *se;
106 * Current state of the operation
108 enum UnionOperationState state;
113 * Element entry, stored in the hash maps from
114 * partial IBF keys to elements.
121 struct GNUNET_SET_Element *element;
124 * Actual ibf key of the element entry
126 struct IBF_Key ibf_key;
129 * Linked list, note that the next element
130 * has to have an ibf_key that is lexicographically
133 struct ElementEntry *next;
136 * GNUNET_YES if the element was received from
137 * the remote peer, and the local peer did not previously
145 * Extra state required for efficient set union.
150 * Strate estimator of the set we currently have,
151 * used for estimation of the symmetric difference
153 struct StrataEstimator *se;
156 * Array of IBFs, some of them pre-allocated
158 struct InvertibleBloomFilter **ibfs;
161 * Maps the first 32 bits of the ibf-key to
164 struct GNUNET_CONTAINER_MultiHashMap32 *elements;
169 send_operation_request (struct EvaluateOperation *eo)
171 struct GNUNET_MQ_Message *mqm;
172 struct OperationRequestMessage *msg;
174 mqm = GNUNET_MQ_msg_concat (msg, eo->context_msg,
175 GNUNET_MESSAGE_TYPE_SET_P2P_OPERATION_REQUEST);
178 /* the context message is too large */
179 client_disconnect (eo->set->client);
183 msg->operation = eo->operation;
184 msg->app_id = eo->app_id;
185 GNUNET_MQ_send (eo->mq, mqm);
190 * Iterator to insert values into an ibf.
193 * @param key current key code
194 * @param value value in the hash map
195 * @return GNUNET_YES if we should continue to
200 ibf_insert_iterator (void *cls,
204 struct InvertibleBloomFilter *ibf = cls;
205 struct ElementEntry *e = value;
206 struct IBF_Key ibf_key;
208 GNUNET_assert (NULL != e);
209 ibf_key = e->ibf_key;
210 ibf_insert (ibf, ibf_key);
215 /* only insert keys we haven't seen yet */
216 if (0 != memcmp (&e->ibf_key, &ibf_key, sizeof ibf_key))
218 ibf_key = e->ibf_key;
219 ibf_insert (ibf, ibf_key);
229 * Create and populate an IBF for the specified peer,
230 * if it does not already exist.
232 * @param cpi peer to create the ibf for
234 static struct InvertibleBloomFilter *
235 prepare_ibf (struct EvaluateOperation *eo, uint16_t order)
237 struct UnionState *us = eo->set->extra.u;
239 GNUNET_assert (order <= MAX_IBF_ORDER);
240 if (NULL == us->ibfs)
241 us->ibfs = GNUNET_malloc (MAX_IBF_ORDER * sizeof (struct InvertibleBloomFilter *));
242 if (NULL == us->ibfs[order])
244 us->ibfs[order] = ibf_create (1 << order, SE_IBF_HASH_NUM);
245 GNUNET_CONTAINER_multihashmap32_iterate (us->elements, ibf_insert_iterator, us->ibfs[order]);
247 return us->ibfs[order];
252 * Send an ibf of appropriate size.
254 * @param cpi the peer
257 send_ibf (struct EvaluateOperation *eo, uint16_t ibf_order)
259 unsigned int buckets_sent = 0;
260 struct InvertibleBloomFilter *ibf;
262 ibf = prepare_ibf (eo, ibf_order);
264 while (buckets_sent < (1 << ibf_order))
266 unsigned int buckets_in_message;
267 struct GNUNET_MQ_Message *mqm;
268 struct IBFMessage *msg;
270 buckets_in_message = (1 << ibf_order) - buckets_sent;
271 /* limit to maximum */
272 if (buckets_in_message > MAX_BUCKETS_PER_MESSAGE)
273 buckets_in_message = MAX_BUCKETS_PER_MESSAGE;
275 mqm = GNUNET_MQ_msg_extra (msg, buckets_in_message * IBF_BUCKET_SIZE,
276 GNUNET_MESSAGE_TYPE_SET_P2P_IBF);
277 msg->order = htons (ibf_order);
278 msg->offset = htons (buckets_sent);
279 ibf_write_slice (ibf, buckets_sent,
280 buckets_in_message, &msg[1]);
281 buckets_sent += buckets_in_message;
282 GNUNET_MQ_send (eo->mq, mqm);
285 eo->extra.u->state = STATE_EXPECT_ELEMENTS_AND_REQUESTS;
290 * Send a strata estimator.
292 * @param cpi the peer
295 send_strata_estimator (struct EvaluateOperation *eo)
297 struct GNUNET_MQ_Message *mqm;
298 struct GNUNET_MessageHeader *strata_msg;
300 mqm = GNUNET_MQ_msg_header_extra (strata_msg,
301 SE_STRATA_COUNT * IBF_BUCKET_SIZE * SE_IBF_SIZE,
302 GNUNET_MESSAGE_TYPE_SET_P2P_SE);
303 strata_estimator_write (eo->set->extra.u->se, &strata_msg[1]);
304 GNUNET_MQ_send (eo->mq, mqm);
306 eo->extra.u->state = STATE_EXPECT_IBF;
311 handle_p2p_strata_estimator (void *cls, const struct GNUNET_MessageHeader *mh)
313 struct EvaluateOperation *eo = cls;
317 if (eo->extra.u->state != STATE_EXPECT_SE)
323 GNUNET_assert (NULL == eo->extra.u->se);
324 eo->extra.u->se = strata_estimator_create (SE_STRATA_COUNT, SE_IBF_SIZE, SE_IBF_HASH_NUM);
325 strata_estimator_read (&mh[1], eo->extra.u->se);
326 GNUNET_assert (NULL != eo->set->extra.u->se);
327 diff = strata_estimator_difference (eo->set->extra.u->se, eo->extra.u->se);
330 while ((1<<ibf_order) < (2 * diff))
332 if (ibf_order > MAX_IBF_ORDER)
333 ibf_order = MAX_IBF_ORDER;
334 send_ibf (eo, ibf_order);
344 decode (struct EvaluateOperation *eo)
348 struct InvertibleBloomFilter *diff_ibf;
350 GNUNET_assert (STATE_EXPECT_ELEMENTS == eo->extra.u->state);
352 diff_ibf = ibf_dup (prepare_ibf (eo, eo->extra.u->ibf_order));
353 ibf_subtract (diff_ibf, eo->extra.u->ibf_received);
359 res = ibf_decode (diff_ibf, &side, &key);
360 if (GNUNET_SYSERR == res)
362 /* decoding failed, we tell the other peer by sending our ibf with a larger order */
366 if (GNUNET_NO == res)
368 struct GNUNET_MQ_Message *mqm;
369 GNUNET_log (GNUNET_ERROR_TYPE_INFO, "transmitted all values, sending DONE\n");
370 mqm = GNUNET_MQ_msg_header (GNUNET_MESSAGE_TYPE_SET_P2P_DONE);
371 GNUNET_MQ_send (eo->mq, mqm);
376 struct ElementEntry *e;
377 /* we have the element(s), send it to the other peer */
378 e = GNUNET_CONTAINER_multihashmap32_get (eo->set->extra.u->elements, (uint32_t) key.key_val);
387 /* FIXME: send element */
393 struct GNUNET_MQ_Message *mqm;
394 struct GNUNET_MessageHeader *msg;
396 /* FIXME: before sending the request, check if we may just have the element */
397 /* FIXME: merge multiple requests */
398 mqm = GNUNET_MQ_msg_header_extra (msg, sizeof (struct IBF_Key), GNUNET_MESSAGE_TYPE_SET_P2P_ELEMENT_REQUESTS);
399 *(struct IBF_Key *) &msg[1] = key;
400 GNUNET_MQ_send (eo->mq, mqm);
408 handle_p2p_ibf (void *cls, const struct GNUNET_MessageHeader *mh)
410 struct EvaluateOperation *eo = cls;
411 struct UnionEvaluateOperation *ueo = eo->extra.u;
412 struct IBFMessage *msg = (struct IBFMessage *) mh;
413 unsigned int buckets_in_message;
415 if (ueo->state == STATE_EXPECT_ELEMENTS_AND_REQUESTS)
417 /* check that the ibf is a new one / first part */
418 /* clear outgoing messages */
421 else if (ueo->state == STATE_EXPECT_IBF)
423 ueo->state = STATE_EXPECT_IBF_CONT;
424 ueo->ibf_order = msg->order;
425 GNUNET_assert (NULL == ueo->ibf_received);
426 ueo->ibf_received = ibf_create (1<<msg->order, SE_IBF_HASH_NUM);
427 if (ntohs (msg->offset) != 0)
433 else if (ueo->state == STATE_EXPECT_IBF_CONT)
435 if ( (ntohs (msg->offset) != ueo->ibf_buckets_received) ||
436 (msg->order != ueo->ibf_order) )
443 buckets_in_message = (ntohs (msg->header.size) - sizeof *msg) / IBF_BUCKET_SIZE;
445 if ((ntohs (msg->header.size) - sizeof *msg) != buckets_in_message * IBF_BUCKET_SIZE)
447 /* FIXME: handle, message was malformed */
451 ibf_read_slice (&msg[1], ueo->ibf_buckets_received, buckets_in_message, ueo->ibf_received);
452 ueo->ibf_buckets_received += buckets_in_message;
454 if (ueo->ibf_buckets_received == (1<<ueo->ibf_order))
456 ueo->state = STATE_EXPECT_ELEMENTS;
463 handle_p2p_elements (void *cls, const struct GNUNET_MessageHeader *mh)
465 struct EvaluateOperation *eo = cls;
467 if ( (eo->extra.u->state != STATE_EXPECT_ELEMENTS) &&
468 (eo->extra.u->state != STATE_EXPECT_ELEMENTS_AND_REQUESTS) )
478 handle_p2p_element_requests (void *cls, const struct GNUNET_MessageHeader *mh)
480 struct EvaluateOperation *eo = cls;
482 /* look up elements and send them */
483 if (eo->extra.u->state != STATE_EXPECT_ELEMENTS_AND_REQUESTS)
493 handle_p2p_done (void *cls, const struct GNUNET_MessageHeader *mh)
499 static const struct GNUNET_MQ_Handler union_handlers[] = {
500 {handle_p2p_elements, GNUNET_MESSAGE_TYPE_SET_P2P_ELEMENTS},
501 {handle_p2p_strata_estimator, GNUNET_MESSAGE_TYPE_SET_P2P_SE},
502 {handle_p2p_ibf, GNUNET_MESSAGE_TYPE_SET_P2P_IBF},
503 {handle_p2p_element_requests, GNUNET_MESSAGE_TYPE_SET_P2P_ELEMENT_REQUESTS},
504 {handle_p2p_done, GNUNET_MESSAGE_TYPE_SET_P2P_DONE},
505 GNUNET_MQ_HANDLERS_END
510 * Functions of this type will be called when a stream is established
512 * @param cls the closure from GNUNET_STREAM_open
513 * @param socket socket to use to communicate with the other side (read/write)
516 stream_open_cb (void *cls,
517 struct GNUNET_STREAM_Socket *socket)
519 struct EvaluateOperation *eo = cls;
521 GNUNET_assert (NULL == eo->mq);
522 GNUNET_assert (socket == eo->socket);
524 eo->mq = GNUNET_MQ_queue_for_stream_socket (eo->socket, union_handlers, eo);
525 send_operation_request (eo);
530 union_evaluate (struct EvaluateOperation *eo)
532 GNUNET_assert (GNUNET_SET_OPERATION_UNION == eo->set->operation);
534 GNUNET_STREAM_open (configuration, &eo->peer, GNUNET_APPLICATION_TYPE_SET,
535 stream_open_cb, GNUNET_STREAM_OPTION_END);
540 insert_ibf_key_unchecked (struct UnionState *us, struct IBF_Key ibf_key)
544 strata_estimator_insert (us->se, ibf_key);
545 for (i = 0; i <= MAX_IBF_ORDER; i++)
547 if (NULL == us->ibfs)
549 if (NULL == us->ibfs[i])
551 ibf_insert (us->ibfs[i], ibf_key);
557 * Insert an element into the consensus set of the specified session.
558 * The element will not be copied, and freed when destroying the session.
560 * @param session session for new element
561 * @param element element to insert
564 insert_element (struct Set *set, struct GNUNET_SET_Element *element)
566 struct UnionState *us = set->extra.u;
567 struct GNUNET_HashCode hash;
568 struct ElementEntry *e;
569 struct ElementEntry *e_old;
571 e = GNUNET_new (struct ElementEntry);
572 e->element = element;
573 GNUNET_CRYPTO_hash (e->element->data, e->element->size, &hash);
574 e->ibf_key = ibf_key_from_hashcode (&hash);
576 e_old = GNUNET_CONTAINER_multihashmap32_get (us->elements, (uint32_t) e->ibf_key.key_val);
579 GNUNET_CONTAINER_multihashmap32_put (us->elements, (uint32_t) e->ibf_key.key_val, e,
580 GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY);
584 while (NULL != e_old)
586 int cmp = memcmp (&e->ibf_key, &e_old->ibf_key, sizeof (struct IBF_Key));
589 if (NULL == e_old->next)
592 insert_ibf_key_unchecked (us, e->ibf_key);
599 e->next = e_old->next;
606 insert_ibf_key_unchecked (us, e->ibf_key);
614 union_accept (struct EvaluateOperation *eo, struct Incoming *incoming)
616 GNUNET_assert (NULL != incoming->mq);
617 eo->mq = incoming->mq;
618 GNUNET_MQ_replace_handlers (eo->mq, union_handlers, eo);
620 send_strata_estimator (eo);
628 set = GNUNET_malloc (sizeof (struct Set) + sizeof (struct UnionState));
629 set->extra.u = (struct UnionState *) &set[1];
630 set->operation = GNUNET_SET_OPERATION_UNION;
631 set->extra.u->se = strata_estimator_create (SE_STRATA_COUNT, SE_IBF_SIZE, SE_IBF_HASH_NUM);
637 union_add (struct Set *set, struct ElementMessage *m)
639 struct GNUNET_SET_Element *element;
640 uint16_t element_size;
641 element_size = ntohs (m->header.size) - sizeof *m;
642 element = GNUNET_malloc (sizeof *element + element_size);
643 element->size = element_size;
644 element->data = &element[1];
645 memcpy (element->data, &m[1], element_size);
646 insert_element (set, element);