/*
This file is part of GNUnet
- (C) 2013 Christian Grothoff (and other contributing authors)
+ (C) 2013, 2014 Christian Grothoff (and other contributing authors)
GNUnet is free software; you can redistribute it and/or modify
it under the terms of the GNU General Public License as published
Free Software Foundation, Inc., 59 Temple Place - Suite 330,
Boston, MA 02111-1307, USA.
*/
-
/**
* @file set/gnunet-service-set_intersection.c
* @brief two-peer set intersection
#define CALCULATE_BF_SIZE(A, B, s, k) \
do { \
k = ceil(1 + log2((double) (2*B / (double) A)));\
+ if (k<1) k=1; /* k can be calculated as 0 */\
s = ceil((double) (A * k / log(2))); \
} while (0)
struct GNUNET_CONTAINER_BloomFilter *local_bf;
/**
- * for multipart msgs we have to store the bloomfilter-data until we fully sent it.
+ * Iterator for sending elements on the key to element mapping to the client.
*/
- char * bf_data;
+ struct GNUNET_CONTAINER_MultiHashMapIterator *full_result_iter;
/**
- * size of the bloomfilter
+ * Evaluate operations are held in a linked list.
*/
- uint32_t bf_data_size;
-
- /**
- * size of the bloomfilter
- */
- uint32_t bf_bits_per_element;
-
+ struct OperationState *next;
+
/**
- * Current state of the operation.
+ * Evaluate operations are held in a linked list.
*/
- enum IntersectionOperationPhase phase;
+ struct OperationState *prev;
/**
- * Generation in which the operation handle
- * was created.
+ * for multipart msgs we have to store the bloomfilter-data until we fully sent it.
*/
- unsigned int generation_created;
+ char *bf_data;
/**
* Maps element-id-hashes to 'elements in our set'.
struct GNUNET_CONTAINER_MultiHashMap *my_elements;
/**
- * Current element count contained within contained_elements
+ * Current element count contained within @e my_elements
*/
uint32_t my_element_count;
/**
- * Iterator for sending elements on the key to element mapping to the client.
+ * size of the bloomfilter in @e bf_data.
*/
- struct GNUNET_CONTAINER_MultiHashMapIterator *full_result_iter;
+ uint32_t bf_data_size;
/**
- * Evaluate operations are held in
- * a linked list.
+ * size of the bloomfilter
*/
- struct OperationState *next;
+ uint32_t bf_bits_per_element;
- /**
- * Evaluate operations are held in
- * a linked list.
- */
- struct OperationState *prev;
+ /**
+ * Current state of the operation.
+ */
+ enum IntersectionOperationPhase phase;
+
+ /**
+ * Generation in which the operation handle
+ * was created.
+ */
+ unsigned int generation_created;
/**
* Did we send the client that we are done?
};
+/**
+ * Send a result message to the client indicating
+ * we removed an element
+ *
+ * @param op union operation
+ * @param element element to send
+ */
+static void
+send_client_element (struct Operation *op,
+ struct GNUNET_SET_Element *element)
+{
+ struct GNUNET_MQ_Envelope *ev;
+ struct GNUNET_SET_ResultMessage *rm;
+
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "sending removed element (size %u) to client\n",
+ element->size);
+ GNUNET_assert (0 != op->spec->client_request_id);
+ ev = GNUNET_MQ_msg_extra (rm, element->size, GNUNET_MESSAGE_TYPE_SET_RESULT);
+ if (NULL == ev)
+ {
+ GNUNET_MQ_discard (ev);
+ GNUNET_break (0);
+ return;
+ }
+ rm->result_status = htons (GNUNET_SET_STATUS_OK);
+ rm->request_id = htonl (op->spec->client_request_id);
+ rm->element_type = element->type;
+ memcpy (&rm[1], element->data, element->size);
+ GNUNET_MQ_send (op->spec->set->client_mq, ev);
+}
+
+
/**
* Alice's version:
*
struct GNUNET_HashCode mutated_hash;
//only consider this element, if it is valid for us
- if ((op->generation_created >= ee->generation_removed)
- || (op->generation_created < ee->generation_added))
+ if ((op->generation_created < ee->generation_removed)
+ && (op->generation_created >= ee->generation_added))
return GNUNET_YES;
// not contained according to bob's bloomfilter
op->spec->salt,
&mutated_hash);
if (GNUNET_NO == GNUNET_CONTAINER_bloomfilter_test (op->state->remote_bf,
- &mutated_hash))
+ &mutated_hash)){
+ if (GNUNET_SET_RESULT_REMOVED == op->spec->result_mode)
+ send_client_element (op, &ee->element);
return GNUNET_YES;
+ }
op->state->my_element_count++;
GNUNET_assert (GNUNET_YES ==
return GNUNET_YES;
}
+
/**
* fills the contained-elements hashmap with all relevant
* elements and adds their mutated hashes to our local bloomfilter
struct Operation *op = cls;
//only consider this element, if it is valid for us
- if ((op->generation_created >= ee->generation_removed)
- || (op->generation_created < ee->generation_added))
+ if ((op->generation_created < ee->generation_removed)
+ && (op->generation_created >= ee->generation_added))
return GNUNET_YES;
GNUNET_assert (GNUNET_YES ==
GNUNET_CONTAINER_multihashmap_remove (op->state->my_elements,
&ee->element_hash,
ee));
+ if (GNUNET_SET_RESULT_REMOVED == op->spec->result_mode)
+ send_client_element (op, &ee->element);
}
return GNUNET_YES;
}
+
/**
- * create a bloomfilter based on the elements given
+ * Create a bloomfilter based on the elements given
*
* @param cls closure
* @param key current key code
*/
static int
iterator_bf_create (void *cls,
- const struct GNUNET_HashCode *key,
- void *value)
+ const struct GNUNET_HashCode *key,
+ void *value)
{
struct ElementEntry *ee = value;
struct Operation *op = cls;
struct GNUNET_HashCode mutated_hash;
- GNUNET_BLOCK_mingle_hash(&ee->element_hash, op->spec->salt, &mutated_hash);
-
+ GNUNET_BLOCK_mingle_hash (&ee->element_hash,
+ op->spec->salt,
+ &mutated_hash);
GNUNET_CONTAINER_bloomfilter_add (op->state->local_bf,
&mutated_hash);
return GNUNET_YES;
}
+
/**
* Inform the client that the union operation has failed,
* and proceed to destroy the evaluate operation.
struct GNUNET_SET_ResultMessage *msg;
if (op->state->my_elements)
+ {
GNUNET_CONTAINER_multihashmap_destroy(op->state->my_elements);
-
- GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "intersection operation failed\n");
+ op->state->my_elements = NULL;
+ }
+ GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+ "intersection operation failed\n");
ev = GNUNET_MQ_msg (msg, GNUNET_MESSAGE_TYPE_SET_RESULT);
msg->result_status = htons (GNUNET_SET_STATUS_FAILURE);
msg->request_id = htonl (op->spec->client_request_id);
msg->element_type = htons (0);
GNUNET_MQ_send (op->spec->set->client_mq, ev);
- _GSS_operation_destroy (op);
+ _GSS_operation_destroy (op, GNUNET_YES);
}
GNUNET_MQ_send (op->mq, ev);
if (NULL != op->spec->context_msg)
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "sent op request with context message\n");
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "sent op request with context message\n");
else
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "sent op request without context message\n");
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "sent op request without context message\n");
if (NULL != op->spec->context_msg)
{
}
}
+
static void
-send_bloomfilter_multipart (struct Operation *op, uint32_t offset)
+send_bloomfilter_multipart (struct Operation *op,
+ uint32_t offset)
{
struct GNUNET_MQ_Envelope *ev;
struct BFPart *msg;
if (todo_size < chunk_size)
chunk_size = todo_size;
- ev = GNUNET_MQ_msg_extra (msg, chunk_size, GNUNET_MESSAGE_TYPE_SET_INTERSECTION_P2P_BF_PART);
+ ev = GNUNET_MQ_msg_extra (msg,
+ chunk_size,
+ GNUNET_MESSAGE_TYPE_SET_INTERSECTION_P2P_BF_PART);
msg->chunk_length = htonl (chunk_size);
msg->chunk_offset = htonl (offset);
op->state->bf_data = NULL;
return;
}
-
send_bloomfilter_multipart (op, offset + chunk_size);
}
+
/**
* Send a bloomfilter to our peer.
* that the operation is over.
uint32_t chunk_size;
struct GNUNET_CONTAINER_BloomFilter * local_bf;
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "sending bf of size %u\n");
-
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "sending bf of size %u\n");
+
CALCULATE_BF_SIZE(op->state->my_element_count,
op->spec->remote_element_count,
bf_size,
bf_elementbits);
op->spec->salt++;
- GNUNET_CONTAINER_multihashmap_iterate (op->spec->set->elements,
+ GNUNET_CONTAINER_multihashmap_iterate (op->state->my_elements,
&iterator_bf_create,
op);
// send our bloomfilter
- if (GNUNET_SERVER_MAX_MESSAGE_SIZE > bf_size + sizeof (struct BFMessage)) {
+ if (GNUNET_SERVER_MAX_MESSAGE_SIZE > bf_size + sizeof (struct BFMessage))
+ {
// singlepart
chunk_size = bf_size;
ev = GNUNET_MQ_msg_extra (msg, chunk_size, GNUNET_MESSAGE_TYPE_SET_INTERSECTION_P2P_BF);
(char*)&msg[1],
bf_size));
}
- else {
+ else
+ {
//multipart
chunk_size = GNUNET_SERVER_MAX_MESSAGE_SIZE - 1 - sizeof (struct BFMessage);
ev = GNUNET_MQ_msg_extra (msg, chunk_size, GNUNET_MESSAGE_TYPE_SET_INTERSECTION_P2P_BF);
struct Operation *op = cls;
struct GNUNET_MQ_Envelope *ev;
struct GNUNET_SET_ResultMessage *rm;
+
ev = GNUNET_MQ_msg (rm, GNUNET_MESSAGE_TYPE_SET_RESULT);
rm->request_id = htonl (op->spec->client_request_id);
rm->result_status = htons (GNUNET_SET_STATUS_DONE);
rm->element_type = htons (0);
GNUNET_MQ_send (op->spec->set->client_mq, ev);
- _GSS_operation_destroy (op);
+ _GSS_operation_destroy (op, GNUNET_YES);
}
struct GNUNET_SET_Element *element;
int res;
- res = GNUNET_CONTAINER_multihashmap_iterator_next (op->state->full_result_iter, NULL, (const void **) &remaining);
- if (GNUNET_NO == res) {
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "sending done and destroy because iterator ran out\n");
+ res = GNUNET_CONTAINER_multihashmap_iterator_next (op->state->full_result_iter,
+ NULL,
+ (const void **) &remaining);
+ if (GNUNET_NO == res)
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "sending done and destroy because iterator ran out\n");
send_client_done_and_destroy (op);
return;
}
element = &remaining->element;
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "sending element (size %u) to client (full set)\n", element->size);
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "sending element (size %u) to client (full set)\n",
+ element->size);
GNUNET_assert (0 != op->spec->client_request_id);
ev = GNUNET_MQ_msg_extra (rm, element->size, GNUNET_MESSAGE_TYPE_SET_RESULT);
struct GNUNET_MQ_Envelope *ev;
op->state->phase = PHASE_FINISHED;
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Intersection succeeded, sending DONE\n");
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Intersection succeeded, sending DONE\n");
GNUNET_CONTAINER_bloomfilter_free (op->state->local_bf);
op->state->local_bf = NULL;
* @param op the intersection operation
*/
static void
-process_bf (struct Operation *op){
+process_bf (struct Operation *op)
+{
uint32_t old_elements;
uint32_t peer_elements;
-
+
old_elements = op->state->my_element_count;
peer_elements = op->spec->remote_element_count;
switch (op->state->phase)
case PHASE_BF_EXCHANGE:
case PHASE_MAYBE_FINISHED:
// if we are bob or alice and are continuing operation
- GNUNET_CONTAINER_multihashmap_iterate (op->spec->set->elements,
+ GNUNET_CONTAINER_multihashmap_iterate (op->state->my_elements,
&iterator_bf_reduce,
op);
break;
if ((0 == op->state->my_element_count) // fully disjoint
|| ((op->state->phase == PHASE_MAYBE_FINISHED) // we agree on a shared set of elements
&& (old_elements == op->state->my_element_count)
- && (op->state->my_element_count == peer_elements))) {
+ && (op->state->my_element_count == peer_elements)))
+ {
// In the last round we though we were finished, we now know this is correct
send_peer_done (op);
return;
const struct BFPart *msg = (const struct BFPart *) mh;
uint32_t chunk_size;
uint32_t chunk_offset;
-
+
chunk_size = ntohl(msg->chunk_length);
chunk_offset = ntohl(msg->chunk_offset);
-
- if ((NULL == op->state->bf_data)
- || (op->state->bf_data_size < chunk_size + chunk_offset)){
+
+ if ((NULL == op->state->bf_data)
+ || (op->state->bf_data_size < chunk_size + chunk_offset))
+ {
// unexpected multipart chunk
GNUNET_break_op (0);
fail_intersection_operation(op);
return;
}
-
+
memcpy (&op->state->bf_data[chunk_offset], (const char*) &msg[1], chunk_size);
if (op->state->bf_data_size != chunk_offset + chunk_size)
op->state->remote_bf = GNUNET_CONTAINER_bloomfilter_init ((const char*) &msg[1],
op->state->bf_data_size,
op->state->bf_bits_per_element);
-
+
GNUNET_free (op->state->bf_data);
op->state->bf_data = NULL;
-
+
process_bf (op);
}
* @param mh the header of the message
*/
static void
-handle_p2p_bf (void *cls, const struct GNUNET_MessageHeader *mh)
+handle_p2p_bf (void *cls,
+ const struct GNUNET_MessageHeader *mh)
{
struct Operation *op = cls;
const struct BFMessage *msg = (const struct BFMessage *) mh;
case PHASE_INITIAL:
case PHASE_BF_EXCHANGE:
case PHASE_MAYBE_FINISHED:
- if (NULL == op->state->bf_data) {
+ if (NULL == op->state->bf_data)
+ {
// no colliding multipart transaction going on currently
op->spec->salt = ntohl (msg->sender_mutator);
bf_size = ntohl (msg->bloomfilter_total_length);
bf_bits_per_element = ntohl (msg->bits_per_element);
chunk_size = ntohl (msg->bloomfilter_length);
op->spec->remote_element_count = ntohl(msg->sender_element_count);
- if (bf_size == chunk_size) {
+ if (bf_size == chunk_size)
+ {
// single part, done here
op->state->remote_bf = GNUNET_CONTAINER_bloomfilter_init ((const char*) &msg[1],
bf_size,
* @param mh the header of the message
*/
static void
-handle_p2p_element_info (void *cls, const struct GNUNET_MessageHeader *mh)
+handle_p2p_element_info (void *cls,
+ const struct GNUNET_MessageHeader *mh)
{
struct Operation *op = cls;
- struct BFMessage *msg = (struct BFMessage *) mh;
+ const struct BFMessage *msg = (const struct BFMessage *) mh;
op->spec->remote_element_count = ntohl(msg->sender_element_count);
if ((op->state->phase != PHASE_INITIAL)
- || (op->state->my_element_count > op->spec->remote_element_count)){
+ || (op->state->my_element_count > op->spec->remote_element_count)
+ || (0 == op->state->my_element_count)
+ || (0 == op->spec->remote_element_count))
+ {
GNUNET_break_op (0);
fail_intersection_operation(op);
+ return;
}
op->state->phase = PHASE_BF_EXCHANGE;
op->state->my_elements = GNUNET_CONTAINER_multihashmap_create (1, GNUNET_YES);
-
+
GNUNET_CONTAINER_multihashmap_iterate (op->spec->set->elements,
&iterator_initialization,
op);
/**
- * Send our element to the peer, in case our element count is lower than his
+ * Send our element count to the peer, in case our element count is lower than his
*
* @param op intersection operation
*/
struct GNUNET_MQ_Envelope *ev;
struct BFMessage *msg;
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "sending element count (bf_msg)\n");
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "sending element count (bf_msg)\n");
// just send our element count, as the other peer must start
ev = GNUNET_MQ_msg (msg, GNUNET_MESSAGE_TYPE_SET_INTERSECTION_P2P_ELEMENT_INFO);
if (GNUNET_SET_RESULT_FULL == op->spec->result_mode)
{
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "sending full result set\n");
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "sending full result set\n");
op->state->full_result_iter =
GNUNET_CONTAINER_multihashmap_iterator_create (op->state->my_elements);
send_remaining_elements (op);
{
struct Operation *op = cls;
- if ((op->state->phase = PHASE_FINISHED) || (op->state->phase = PHASE_MAYBE_FINISHED)){
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "got final DONE\n");
+ if ( (op->state->phase = PHASE_FINISHED) ||
+ (op->state->phase = PHASE_MAYBE_FINISHED) )
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "got final DONE\n");
finish_and_destroy (op);
return;
static void
intersection_accept (struct Operation *op)
{
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "accepting set union operation\n");
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "accepting set union operation\n");
op->state = GNUNET_new (struct OperationState);
op->state->my_elements = GNUNET_CONTAINER_multihashmap_create(1, GNUNET_YES);
op->state->my_element_count = op->spec->set->state->current_set_element_count;
intersection_add (struct SetState *set_state,
struct ElementEntry *ee)
{
- GNUNET_assert(0 < set_state->current_set_element_count);
set_state->current_set_element_count++;
}
intersection_handle_p2p_message (struct Operation *op,
const struct GNUNET_MessageHeader *mh)
{
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "received p2p message (t: %u, s: %u)\n",
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "received p2p message (t: %u, s: %u)\n",
ntohs (mh->type), ntohs (mh->size));
switch (ntohs (mh->type))
{
handle_p2p_done (op, mh);
break;
default:
- /* something wrong with mesh's message handlers? */
+ /* something wrong with cadet's message handlers? */
GNUNET_assert (0);
}
return GNUNET_OK;
msg->result_status = htons (GNUNET_SET_STATUS_FAILURE);
msg->element_type = htons (0);
GNUNET_MQ_send (op->spec->set->client_mq, ev);
- GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "other peer disconnected prematurely\n");
- _GSS_operation_destroy (op);
+ GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+ "other peer disconnected prematurely\n");
+ _GSS_operation_destroy (op, GNUNET_YES);
return;
}
// else: the session has already been concluded
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "other peer disconnected (finished)\n");
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "other peer disconnected (finished)\n");
if (GNUNET_NO == op->state->client_done_sent)
finish_and_destroy (op);
}
static void
intersection_op_cancel (struct Operation *op)
{
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "destroying intersection op\n");
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "destroying intersection op\n");
/* check if the op was canceled twice */
GNUNET_assert (NULL != op->state);
if (NULL != op->state->remote_bf)
GNUNET_CONTAINER_bloomfilter_free (op->state->local_bf);
op->state->local_bf = NULL;
}
- if (NULL != op->state->my_elements)
+/* if (NULL != op->state->my_elements)
{
// no need to free the elements, they are still part of the set
GNUNET_CONTAINER_multihashmap_destroy (op->state->my_elements);
op->state->my_elements = NULL;
- }
+ }*/
GNUNET_free (op->state);
op->state = NULL;
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "destroying intersection op done\n");
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "destroying intersection op done\n");
}
+
+/**
+ * Get the table with implementing functions for set intersection.
+ *
+ * @return the operation specific VTable
+ */
const struct SetVT *
_GSS_intersection_vt ()
{
projects / oweals / gnunet.git / blobdiff