X-Git-Url: https://git.librecmc.org/?a=blobdiff_plain;f=src%2Fset%2Fset_api.c;h=5b5b1b8eecfdd1d97d2d7eaf92c66a9cd100a08c;hb=237dd9f9c93cabd29015d0f5e3c8ceff8af3f071;hp=90cba446cbc1ada0924cb629ad9df68d35c4d92f;hpb=ce9b32618b6ee488352ef0eb506c744868145f82;p=oweals%2Fgnunet.git diff --git a/src/set/set_api.c b/src/set/set_api.c index 90cba446c..5b5b1b8ee 100644 --- a/src/set/set_api.c +++ b/src/set/set_api.c @@ -1,6 +1,6 @@ /* This file is part of GNUnet. - (C) 2012, 2013 Christian Grothoff (and other contributing authors) + Copyright (C) 2012-2016 GNUnet e.V. GNUnet is free software; you can redistribute it and/or modify it under the terms of the GNU General Public License as published @@ -14,37 +14,42 @@ You should have received a copy of the GNU General Public License along with GNUnet; see the file COPYING. If not, write to the - Free Software Foundation, Inc., 59 Temple Place - Suite 330, - Boston, MA 02111-1307, USA. + Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, + Boston, MA 02110-1301, USA. */ - /** * @file set/set_api.c * @brief api for the set service * @author Florian Dold + * @author Christian Grothoff */ #include "platform.h" #include "gnunet_util_lib.h" #include "gnunet_protocols.h" -#include "gnunet_client_lib.h" #include "gnunet_set_service.h" #include "set.h" #define LOG(kind,...) GNUNET_log_from (kind, "set-api",__VA_ARGS__) +struct SetCopyRequest +{ + struct SetCopyRequest *next; + + struct SetCopyRequest *prev; + + void *cls; + + GNUNET_SET_CopyReadyCallback cb; +}; + /** * Opaque handle to a set. */ struct GNUNET_SET_Handle { /** - * Client connected to the set service. - */ - struct GNUNET_CLIENT_Connection *client; - - /** - * Message queue for 'client'. + * Message queue for @e client. */ struct GNUNET_MQ_Handle *mq; @@ -58,6 +63,17 @@ struct GNUNET_SET_Handle */ struct GNUNET_SET_OperationHandle *ops_tail; + /** + * Callback for the current iteration over the set, + * NULL if no iterator is active. + */ + GNUNET_SET_ElementIterator iterator; + + /** + * Closure for @e iterator + */ + void *iterator_cls; + /** * Should the set be destroyed once all operations are gone? */ @@ -69,20 +85,30 @@ struct GNUNET_SET_Handle int invalid; /** - * Callback for the current iteration over the set, - * NULL if no iterator is active. + * Both client and service count the number of iterators + * created so far to match replies with iterators. */ - GNUNET_SET_ElementIterator iterator; + uint16_t iteration_id; /** - * Closure for 'iterator' + * Configuration, needed when creating (lazy) copies. */ - void *iterator_cls; + const struct GNUNET_CONFIGURATION_Handle *cfg; + + /** + * Doubly linked list of copy requests. + */ + struct SetCopyRequest *copy_req_head; + + /** + * Doubly linked list of copy requests. + */ + struct SetCopyRequest *copy_req_tail; }; /** - * Opaque handle to a set operation request from another peer. + * Handle for a set operation request from another peer. */ struct GNUNET_SET_Request { @@ -94,15 +120,14 @@ struct GNUNET_SET_Request /** * Has the request been accepted already? - * GNUNET_YES/GNUNET_NO + * #GNUNET_YES/#GNUNET_NO */ int accepted; }; /** - * Handle to an operation. - * Only known to the service after commiting + * Handle to an operation. Only known to the service after committing * the handle with a set. */ struct GNUNET_SET_OperationHandle @@ -114,7 +139,7 @@ struct GNUNET_SET_OperationHandle GNUNET_SET_ResultIterator result_cb; /** - * Closure for result_cb. + * Closure for @e result_cb. */ void *result_cls; @@ -124,11 +149,6 @@ struct GNUNET_SET_OperationHandle */ struct GNUNET_SET_Handle *set; - /** - * Request ID to identify the operation within the set. - */ - uint32_t request_id; - /** * Message sent to the server on calling conclude, * NULL if conclude has been called. @@ -150,6 +170,11 @@ struct GNUNET_SET_OperationHandle * Handles are kept in a linked list. */ struct GNUNET_SET_OperationHandle *next; + + /** + * Request ID to identify the operation within the set. + */ + uint32_t request_id; }; @@ -158,10 +183,6 @@ struct GNUNET_SET_OperationHandle */ struct GNUNET_SET_ListenHandle { - /** - * Connection to the service. - */ - struct GNUNET_CLIENT_Connection *client; /** * Message queue for the client. @@ -182,15 +203,10 @@ struct GNUNET_SET_ListenHandle GNUNET_SET_ListenCallback listen_cb; /** - * Closure for listen_cb. + * Closure for @e listen_cb. */ void *listen_cls; - /** - * Operation we listen for. - */ - enum GNUNET_SET_OperationType operation; - /** * Application ID we listen for. */ @@ -204,60 +220,151 @@ struct GNUNET_SET_ListenHandle /** * Task for reconnecting when the listener fails. */ - GNUNET_SCHEDULER_TaskIdentifier reconnect_task; + struct GNUNET_SCHEDULER_Task *reconnect_task; + + /** + * Operation we listen for. + */ + enum GNUNET_SET_OperationType operation; }; -/* forward declaration */ +/* mutual recursion with handle_copy_lazy */ +static struct GNUNET_SET_Handle * +create_internal (const struct GNUNET_CONFIGURATION_Handle *cfg, + enum GNUNET_SET_OperationType op, + const uint32_t *cookie); + + +/** + * Handle element for iteration over the set. Notifies the + * iterator and sends an acknowledgement to the service. + * + * @param cls the `struct GNUNET_SET_Handle *` + * @param msg the message + */ static void -listen_connect (void *cls, - const struct GNUNET_SCHEDULER_TaskContext *tc); +handle_copy_lazy (void *cls, + const struct GNUNET_SET_CopyLazyResponseMessage *msg) +{ + struct GNUNET_SET_Handle *set = cls; + struct SetCopyRequest *req; + struct GNUNET_SET_Handle *new_set; + + req = set->copy_req_head; + if (NULL == req) + { + /* Service sent us unsolicited lazy copy response */ + GNUNET_break (0); + return; + } + + LOG (GNUNET_ERROR_TYPE_DEBUG, + "Handling response to lazy copy\n"); + GNUNET_CONTAINER_DLL_remove (set->copy_req_head, + set->copy_req_tail, + req); + // We pass none as operation here, since it doesn't matter when + // cloning. + new_set = create_internal (set->cfg, + GNUNET_SET_OPERATION_NONE, + &msg->cookie); + req->cb (req->cls, new_set); + GNUNET_free (req); +} /** - * Handle element for iteration over the set. + * Check that the given @a msg is well-formed. * - * @param cls the set + * @param cls closure + * @param msg message to check + * @return #GNUNET_OK if message is well-formed + */ +static int +check_iter_element (void *cls, + const struct GNUNET_SET_IterResponseMessage *msg) +{ + /* minimum size was already checked, everything else is OK! */ + return GNUNET_OK; +} + + +/** + * Handle element for iteration over the set. Notifies the + * iterator and sends an acknowledgement to the service. + * + * @param cls the `struct GNUNET_SET_Handle *` * @param mh the message */ static void -handle_iter_element (void *cls, const struct GNUNET_MessageHeader *mh) +handle_iter_element (void *cls, + const struct GNUNET_SET_IterResponseMessage *msg) { struct GNUNET_SET_Handle *set = cls; + GNUNET_SET_ElementIterator iter = set->iterator; struct GNUNET_SET_Element element; - const struct GNUNET_SET_IterResponseMessage *msg = - (const struct GNUNET_SET_IterResponseMessage *) mh; struct GNUNET_SET_IterAckMessage *ack_msg; struct GNUNET_MQ_Envelope *ev; + uint16_t msize; - if (NULL == set->iterator) - return; - - element.size = ntohs (mh->size) - sizeof (struct GNUNET_SET_IterResponseMessage); - element.element_type = htons (msg->element_type); - element.data = &msg[1]; - set->iterator (set->iterator_cls, &element); - ev = GNUNET_MQ_msg (ack_msg, GNUNET_MESSAGE_TYPE_SET_ITER_ACK); - ack_msg->send_more = htonl (1); + msize = ntohs (msg->header.size); + if (set->iteration_id != ntohs (msg->iteration_id)) + { + /* element from a previous iteration, skip! */ + iter = NULL; + } + if (NULL != iter) + { + element.size = msize - sizeof (struct GNUNET_SET_IterResponseMessage); + element.element_type = ntohs (msg->element_type); + element.data = &msg[1]; + iter (set->iterator_cls, + &element); + } + ev = GNUNET_MQ_msg (ack_msg, + GNUNET_MESSAGE_TYPE_SET_ITER_ACK); + ack_msg->send_more = htonl ((NULL != iter)); GNUNET_MQ_send (set->mq, ev); } /** - * Handle element for iteration over the set. + * Handle message signalling conclusion of iteration over the set. + * Notifies the iterator that we are done. * * @param cls the set * @param mh the message */ static void -handle_iter_done (void *cls, const struct GNUNET_MessageHeader *mh) +handle_iter_done (void *cls, + const struct GNUNET_MessageHeader *mh) { struct GNUNET_SET_Handle *set = cls; + GNUNET_SET_ElementIterator iter = set->iterator; - if (NULL == set->iterator) + if (NULL == iter) return; + set->iterator = NULL; + set->iteration_id++; + iter (set->iterator_cls, + NULL); +} - set->iterator (set->iterator_cls, NULL); + +/** + * Check that the given @a msg is well-formed. + * + * @param cls closure + * @param msg message to check + * @return #GNUNET_OK if message is well-formed + */ +static int +check_result (void *cls, + const struct GNUNET_SET_ResultMessage *msg) +{ + /* minimum size was already checked, everything else is OK! */ + return GNUNET_OK; } @@ -268,185 +375,114 @@ handle_iter_done (void *cls, const struct GNUNET_MessageHeader *mh) * @param mh the message */ static void -handle_result (void *cls, const struct GNUNET_MessageHeader *mh) +handle_result (void *cls, + const struct GNUNET_SET_ResultMessage *msg) { - const struct GNUNET_SET_ResultMessage *msg; struct GNUNET_SET_Handle *set = cls; struct GNUNET_SET_OperationHandle *oh; struct GNUNET_SET_Element e; enum GNUNET_SET_Status result_status; + int destroy_set; - msg = (const struct GNUNET_SET_ResultMessage *) mh; - GNUNET_assert (NULL != set); GNUNET_assert (NULL != set->mq); - result_status = ntohs (msg->result_status); + LOG (GNUNET_ERROR_TYPE_DEBUG, + "Got result message with status %d\n", + result_status); - oh = GNUNET_MQ_assoc_get (set->mq, ntohl (msg->request_id)); - // 'oh' can be NULL if we canceled the operation, but the service - // did not get the cancel message yet. + oh = GNUNET_MQ_assoc_get (set->mq, + ntohl (msg->request_id)); if (NULL == oh) { - GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "ignoring result from canceled operation\n"); + /* 'oh' can be NULL if we canceled the operation, but the service + did not get the cancel message yet. */ + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, + "Ignoring result from canceled operation\n"); return; } - /* status is not STATUS_OK => there's no attached element, - * and this is the last result message we get */ - if (GNUNET_SET_STATUS_OK != result_status) + + switch (result_status) { - GNUNET_MQ_assoc_remove (set->mq, ntohl (msg->request_id)); - GNUNET_CONTAINER_DLL_remove (oh->set->ops_head, oh->set->ops_tail, oh); - if (GNUNET_YES == oh->set->destroy_requested) - GNUNET_SET_destroy (oh->set); - if (NULL != oh->result_cb) - oh->result_cb (oh->result_cls, NULL, result_status); - GNUNET_free (oh); - return; + case GNUNET_SET_STATUS_OK: + case GNUNET_SET_STATUS_ADD_LOCAL: + case GNUNET_SET_STATUS_ADD_REMOTE: + goto do_element; + case GNUNET_SET_STATUS_FAILURE: + case GNUNET_SET_STATUS_DONE: + goto do_final; + case GNUNET_SET_STATUS_HALF_DONE: + /* not used anymore */ + GNUNET_assert (0); } - e.data = &msg[1]; - e.size = ntohs (mh->size) - sizeof (struct GNUNET_SET_ResultMessage); - e.element_type = msg->element_type; - if (NULL != oh->result_cb) - oh->result_cb (oh->result_cls, &e, result_status); -} - - -/** - * Handle request message for a listen operation - * - * @param cls the listen handle - * @param mh the message - */ -static void -handle_request (void *cls, - const struct GNUNET_MessageHeader *mh) -{ - const struct GNUNET_SET_RequestMessage *msg = (const struct GNUNET_SET_RequestMessage *) mh; - struct GNUNET_SET_ListenHandle *lh = cls; - struct GNUNET_SET_Request *req; - const struct GNUNET_MessageHeader *context_msg; - +do_final: LOG (GNUNET_ERROR_TYPE_DEBUG, - "processing operation request\n"); - req = GNUNET_new (struct GNUNET_SET_Request); - req->accept_id = ntohl (msg->accept_id); - context_msg = GNUNET_MQ_extract_nested_mh (msg); - /* calling #GNUNET_SET_accept() in the listen cb will set req->accepted */ - lh->listen_cb (lh->listen_cls, &msg->peer_id, context_msg, req); - - /* we got another request => reset the backoff */ - lh->reconnect_backoff = GNUNET_TIME_UNIT_MILLISECONDS; - - if (GNUNET_NO == req->accepted) + "Treating result as final status\n"); + GNUNET_MQ_assoc_remove (set->mq, + ntohl (msg->request_id)); + GNUNET_CONTAINER_DLL_remove (set->ops_head, + set->ops_tail, + oh); + /* Need to do this calculation _before_ the result callback, + as IF the application still has a valid set handle, it + may trigger destruction of the set during the callback. */ + destroy_set = (GNUNET_YES == set->destroy_requested) && + (NULL == set->ops_head); + if (NULL != oh->result_cb) + { + oh->result_cb (oh->result_cls, + NULL, + GNUNET_ntohll (msg->current_size), + result_status); + } + else { - struct GNUNET_MQ_Envelope *mqm; - struct GNUNET_SET_RejectMessage *rmsg; - - mqm = GNUNET_MQ_msg (rmsg, - GNUNET_MESSAGE_TYPE_SET_REJECT); - rmsg->accept_reject_id = msg->accept_id; - GNUNET_MQ_send (lh->mq, mqm); LOG (GNUNET_ERROR_TYPE_DEBUG, - "rejecting request\n"); + "No callback for final status\n"); } - GNUNET_free (req); - - LOG (GNUNET_ERROR_TYPE_DEBUG, - "processed op request from service\n"); - - /* the accept-case is handled in GNUNET_SET_accept, - * as we have the accept message available there */ -} - - -static void -handle_client_listener_error (void *cls, - enum GNUNET_MQ_Error error) -{ - struct GNUNET_SET_ListenHandle *lh = cls; + if (destroy_set) + GNUNET_SET_destroy (set); + GNUNET_free (oh); + return; +do_element: LOG (GNUNET_ERROR_TYPE_DEBUG, - "listener broke down, re-connecting\n"); - GNUNET_CLIENT_disconnect (lh->client); - lh->client = NULL; - GNUNET_MQ_destroy (lh->mq); - lh->mq = NULL; - lh->reconnect_task = GNUNET_SCHEDULER_add_delayed (lh->reconnect_backoff, - &listen_connect, lh); - lh->reconnect_backoff = GNUNET_TIME_STD_BACKOFF (lh->reconnect_backoff); -} - - -/** - * Destroy the set handle if no operations are left, mark the set - * for destruction otherwise. - * - * @param set set handle to destroy - */ -static int -set_destroy (struct GNUNET_SET_Handle *set) -{ - if (NULL != set->ops_head) - { - set->destroy_requested = GNUNET_YES; - return GNUNET_NO; - } - LOG (GNUNET_ERROR_TYPE_DEBUG, "Really destroying set\n"); - GNUNET_CLIENT_disconnect (set->client); - set->client = NULL; - GNUNET_MQ_destroy (set->mq); - set->mq = NULL; - GNUNET_free (set); - return GNUNET_YES; + "Treating result as element\n"); + e.data = &msg[1]; + e.size = ntohs (msg->header.size) - sizeof (struct GNUNET_SET_ResultMessage); + e.element_type = ntohs (msg->element_type); + if (NULL != oh->result_cb) + oh->result_cb (oh->result_cls, + &e, + GNUNET_ntohll (msg->current_size), + result_status); } /** - * Cancel the given set operation. We need to send an explicit cancel message, - * as all operations one one set communicate using one handle. - * - * In contrast to #GNUNET_SET_operation_cancel(), this function indicates whether - * the set of the operation has been destroyed because all operations are done and - * the set's destruction was requested before. + * Destroy the given set operation. * - * @param oh set operation to cancel - * @return #GNUNET_YES if the set of the operation was destroyed + * @param oh set operation to destroy */ -static int -set_operation_cancel (struct GNUNET_SET_OperationHandle *oh) +static void +set_operation_destroy (struct GNUNET_SET_OperationHandle *oh) { - int ret = GNUNET_NO; + struct GNUNET_SET_Handle *set = oh->set; + struct GNUNET_SET_OperationHandle *h_assoc; if (NULL != oh->conclude_mqm) GNUNET_MQ_discard (oh->conclude_mqm); - /* is the operation already commited? */ - if (NULL != oh->set) + if (NULL != set) { - struct GNUNET_SET_OperationHandle *h_assoc; - struct GNUNET_SET_CancelMessage *m; - struct GNUNET_MQ_Envelope *mqm; - - GNUNET_CONTAINER_DLL_remove (oh->set->ops_head, - oh->set->ops_tail, + GNUNET_CONTAINER_DLL_remove (set->ops_head, + set->ops_tail, oh); - h_assoc = GNUNET_MQ_assoc_remove (oh->set->mq, + h_assoc = GNUNET_MQ_assoc_remove (set->mq, oh->request_id); - GNUNET_assert ((h_assoc == NULL) || (h_assoc == oh)); - mqm = GNUNET_MQ_msg (m, GNUNET_MESSAGE_TYPE_SET_CANCEL); - m->request_id = htonl (oh->request_id); - GNUNET_MQ_send (oh->set->mq, mqm); - - if (GNUNET_YES == oh->set->destroy_requested) - { - LOG (GNUNET_ERROR_TYPE_DEBUG, - "Destroying set after operation cancel\n"); - ret = set_destroy (oh->set); - } + GNUNET_assert ((NULL == h_assoc) || (h_assoc == oh)); } GNUNET_free (oh); - return ret; } @@ -460,27 +496,123 @@ set_operation_cancel (struct GNUNET_SET_OperationHandle *oh) void GNUNET_SET_operation_cancel (struct GNUNET_SET_OperationHandle *oh) { - (void) set_operation_cancel (oh); + struct GNUNET_SET_Handle *set = oh->set; + struct GNUNET_SET_CancelMessage *m; + struct GNUNET_MQ_Envelope *mqm; + + if (NULL != set) + { + mqm = GNUNET_MQ_msg (m, GNUNET_MESSAGE_TYPE_SET_CANCEL); + m->request_id = htonl (oh->request_id); + GNUNET_MQ_send (set->mq, mqm); + } + set_operation_destroy (oh); + if ( (NULL != set) && + (GNUNET_YES == set->destroy_requested) && + (NULL == set->ops_head) ) + { + LOG (GNUNET_ERROR_TYPE_DEBUG, + "Destroying set after operation cancel\n"); + GNUNET_SET_destroy (set); + } } +/** + * We encountered an error communicating with the set service while + * performing a set operation. Report to the application. + * + * @param cls the `struct GNUNET_SET_Handle` + * @param error error code + */ static void -handle_client_set_error (void *cls, enum GNUNET_MQ_Error error) +handle_client_set_error (void *cls, + enum GNUNET_MQ_Error error) { struct GNUNET_SET_Handle *set = cls; + GNUNET_SET_ElementIterator iter = set->iterator; - LOG (GNUNET_ERROR_TYPE_DEBUG, - "handling client set error\n"); - + LOG (GNUNET_ERROR_TYPE_ERROR, + "Handling client set error %d\n", + error); while (NULL != set->ops_head) { if (NULL != set->ops_head->result_cb) - set->ops_head->result_cb (set->ops_head->result_cls, NULL, + set->ops_head->result_cb (set->ops_head->result_cls, + NULL, + 0, GNUNET_SET_STATUS_FAILURE); - if (GNUNET_YES == set_operation_cancel (set->ops_head)) - return; /* stop if the set is destroyed */ + set_operation_destroy (set->ops_head); } + set->iterator = NULL; + set->iteration_id++; set->invalid = GNUNET_YES; + if (NULL != iter) + iter (set->iterator_cls, + NULL); +} + + +static struct GNUNET_SET_Handle * +create_internal (const struct GNUNET_CONFIGURATION_Handle *cfg, + enum GNUNET_SET_OperationType op, + const uint32_t *cookie) +{ + struct GNUNET_SET_Handle *set = GNUNET_new (struct GNUNET_SET_Handle); + struct GNUNET_MQ_MessageHandler mq_handlers[] = { + GNUNET_MQ_hd_var_size (result, + GNUNET_MESSAGE_TYPE_SET_RESULT, + struct GNUNET_SET_ResultMessage, + set), + GNUNET_MQ_hd_var_size (iter_element, + GNUNET_MESSAGE_TYPE_SET_ITER_ELEMENT, + struct GNUNET_SET_IterResponseMessage, + set), + GNUNET_MQ_hd_fixed_size (iter_done, + GNUNET_MESSAGE_TYPE_SET_ITER_DONE, + struct GNUNET_MessageHeader, + set), + GNUNET_MQ_hd_fixed_size (copy_lazy, + GNUNET_MESSAGE_TYPE_SET_COPY_LAZY_RESPONSE, + struct GNUNET_SET_CopyLazyResponseMessage, + set), + GNUNET_MQ_handler_end () + }; + struct GNUNET_MQ_Envelope *mqm; + struct GNUNET_SET_CreateMessage *create_msg; + struct GNUNET_SET_CopyLazyConnectMessage *copy_msg; + + set->cfg = cfg; + set->mq = GNUNET_CLIENT_connect (cfg, + "set", + mq_handlers, + &handle_client_set_error, + set); + if (NULL == set->mq) + { + GNUNET_free (set); + return NULL; + } + if (NULL == cookie) + { + LOG (GNUNET_ERROR_TYPE_DEBUG, + "Creating new set (operation %u)\n", + op); + mqm = GNUNET_MQ_msg (create_msg, + GNUNET_MESSAGE_TYPE_SET_CREATE); + create_msg->operation = htonl (op); + } + else + { + LOG (GNUNET_ERROR_TYPE_DEBUG, + "Creating new set (lazy copy)\n", + op); + mqm = GNUNET_MQ_msg (copy_msg, + GNUNET_MESSAGE_TYPE_SET_COPY_LAZY_CONNECT); + copy_msg->cookie = *cookie; + } + GNUNET_MQ_send (set->mq, mqm); + return set; } @@ -499,35 +631,15 @@ struct GNUNET_SET_Handle * GNUNET_SET_create (const struct GNUNET_CONFIGURATION_Handle *cfg, enum GNUNET_SET_OperationType op) { - static const struct GNUNET_MQ_MessageHandler mq_handlers[] = { - {handle_result, GNUNET_MESSAGE_TYPE_SET_RESULT, 0}, - {handle_iter_element, GNUNET_MESSAGE_TYPE_SET_ITER_ELEMENT, 0}, - {handle_iter_done, GNUNET_MESSAGE_TYPE_SET_ITER_DONE, 0}, - GNUNET_MQ_HANDLERS_END - }; - struct GNUNET_SET_Handle *set; - struct GNUNET_MQ_Envelope *mqm; - struct GNUNET_SET_CreateMessage *msg; - - set = GNUNET_new (struct GNUNET_SET_Handle); - set->client = GNUNET_CLIENT_connect ("set", cfg); - LOG (GNUNET_ERROR_TYPE_DEBUG, "set client created\n"); - GNUNET_assert (NULL != set->client); - set->mq = GNUNET_MQ_queue_for_connection_client (set->client, mq_handlers, - handle_client_set_error, set); - GNUNET_assert (NULL != set->mq); - mqm = GNUNET_MQ_msg (msg, GNUNET_MESSAGE_TYPE_SET_CREATE); - msg->operation = htonl (op); - GNUNET_MQ_send (set->mq, mqm); - return set; + return create_internal (cfg, op, NULL); } /** - * Add an element to the given set. - * After the element has been added (in the sense of being - * transmitted to the set service), cont will be called. - * Calls to add_element can be queued + * Add an element to the given set. After the element has been added + * (in the sense of being transmitted to the set service), @a cont + * will be called. Multiple calls to GNUNET_SET_add_element() can be + * queued. * * @param set set to add element to * @param element element to add to the set @@ -545,27 +657,33 @@ GNUNET_SET_add_element (struct GNUNET_SET_Handle *set, struct GNUNET_MQ_Envelope *mqm; struct GNUNET_SET_ElementMessage *msg; + LOG (GNUNET_ERROR_TYPE_INFO, "adding element of type %u\n", (unsigned) element->element_type); + if (GNUNET_YES == set->invalid) { if (NULL != cont) cont (cont_cls); return GNUNET_SYSERR; } - - mqm = GNUNET_MQ_msg_extra (msg, element->size, GNUNET_MESSAGE_TYPE_SET_ADD); - msg->element_type = element->element_type; - memcpy (&msg[1], element->data, element->size); - GNUNET_MQ_notify_sent (mqm, cont, cont_cls); + mqm = GNUNET_MQ_msg_extra (msg, + element->size, + GNUNET_MESSAGE_TYPE_SET_ADD); + msg->element_type = htons (element->element_type); + GNUNET_memcpy (&msg[1], + element->data, + element->size); + GNUNET_MQ_notify_sent (mqm, + cont, cont_cls); GNUNET_MQ_send (set->mq, mqm); return GNUNET_OK; } /** - * Remove an element to the given set. - * After the element has been removed (in the sense of the - * request being transmitted to the set service), cont will be called. - * Calls to remove_element can be queued + * Remove an element to the given set. After the element has been + * removed (in the sense of the request being transmitted to the set + * service), @a cont will be called. Multiple calls to + * GNUNET_SET_remove_element() can be queued * * @param set set to remove element from * @param element element to remove from the set @@ -589,27 +707,48 @@ GNUNET_SET_remove_element (struct GNUNET_SET_Handle *set, cont (cont_cls); return GNUNET_SYSERR; } - mqm = GNUNET_MQ_msg_extra (msg, element->size, GNUNET_MESSAGE_TYPE_SET_REMOVE); - msg->element_type = element->element_type; - memcpy (&msg[1], element->data, element->size); - GNUNET_MQ_notify_sent (mqm, cont, cont_cls); + msg->element_type = htons (element->element_type); + GNUNET_memcpy (&msg[1], + element->data, + element->size); + GNUNET_MQ_notify_sent (mqm, + cont, cont_cls); GNUNET_MQ_send (set->mq, mqm); return GNUNET_OK; } /** - * Destroy the set handle, and free all associated resources. + * Destroy the set handle if no operations are left, mark the set + * for destruction otherwise. * * @param set set handle to destroy */ void GNUNET_SET_destroy (struct GNUNET_SET_Handle *set) { - (void) set_destroy (set); + /* destroying set while iterator is active is currently + not supported; we should expand the API to allow + clients to explicitly cancel the iteration! */ + GNUNET_assert (NULL == set->iterator); + if (NULL != set->ops_head) + { + LOG (GNUNET_ERROR_TYPE_DEBUG, + "Set operations are pending, delaying set destruction\n"); + set->destroy_requested = GNUNET_YES; + return; + } + LOG (GNUNET_ERROR_TYPE_DEBUG, + "Really destroying set\n"); + if (NULL != set->mq) + { + GNUNET_MQ_destroy (set->mq); + set->mq = NULL; + } + GNUNET_free (set); } @@ -632,13 +771,18 @@ GNUNET_SET_prepare (const struct GNUNET_PeerIdentity *other_peer, const struct GNUNET_HashCode *app_id, const struct GNUNET_MessageHeader *context_msg, enum GNUNET_SET_ResultMode result_mode, + struct GNUNET_SET_Option options[], GNUNET_SET_ResultIterator result_cb, void *result_cls) { struct GNUNET_MQ_Envelope *mqm; struct GNUNET_SET_OperationHandle *oh; struct GNUNET_SET_EvaluateMessage *msg; + struct GNUNET_SET_Option *opt; + LOG (GNUNET_ERROR_TYPE_DEBUG, + "Client prepares set operation (%d)\n", + result_mode); oh = GNUNET_new (struct GNUNET_SET_OperationHandle); oh->result_cb = result_cb; oh->result_cls = result_cls; @@ -648,6 +792,25 @@ GNUNET_SET_prepare (const struct GNUNET_PeerIdentity *other_peer, msg->app_id = *app_id; msg->result_mode = htonl (result_mode); msg->target_peer = *other_peer; + for (opt = options; opt->type != 0; opt++) + { + switch (opt->type) + { + case GNUNET_SET_OPTION_BYZANTINE: + msg->byzantine = GNUNET_YES; + msg->byzantine_lower_bound = opt->v.num; + break; + case GNUNET_SET_OPTION_FORCE_FULL: + msg->force_full = GNUNET_YES; + break; + case GNUNET_SET_OPTION_FORCE_DELTA: + msg->force_delta = GNUNET_YES; + break; + default: + LOG (GNUNET_ERROR_TYPE_ERROR, + "Option with type %d not recognized\n", (int) opt->type); + } + } oh->conclude_mqm = mqm; oh->request_id_addr = &msg->request_id; @@ -656,47 +819,137 @@ GNUNET_SET_prepare (const struct GNUNET_PeerIdentity *other_peer, /** - * Connect to the set service in order to listen - * for request. + * Connect to the set service in order to listen for requests. * - * @param cls the listen handle to connect - * @param tc task context if invoked as a task, NULL otherwise + * @param cls the `struct GNUNET_SET_ListenHandle *` to connect */ static void -listen_connect (void *cls, - const struct GNUNET_SCHEDULER_TaskContext *tc) +listen_connect (void *cls); + + +/** + * Check validity of request message for a listen operation + * + * @param cls the listen handle + * @param msg the message + * @return #GNUNET_OK if the message is well-formed + */ +static int +check_request (void *cls, + const struct GNUNET_SET_RequestMessage *msg) { - struct GNUNET_MQ_Envelope *mqm; - struct GNUNET_SET_ListenMessage *msg; - struct GNUNET_SET_ListenHandle *lh = cls; - static const struct GNUNET_MQ_MessageHandler mq_handlers[] = { - {handle_request, GNUNET_MESSAGE_TYPE_SET_REQUEST}, - GNUNET_MQ_HANDLERS_END - }; + const struct GNUNET_MessageHeader *context_msg; - if ((tc != NULL) &&(tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN) != 0) + if (ntohs (msg->header.size) == sizeof (*msg)) + return GNUNET_OK; /* no context message is OK */ + context_msg = GNUNET_MQ_extract_nested_mh (msg); + if (NULL == context_msg) { - LOG (GNUNET_ERROR_TYPE_DEBUG, "listener not reconnecting due to shutdown\n"); - return; + /* malformed context message is NOT ok */ + GNUNET_break_op (0); + return GNUNET_SYSERR; } + return GNUNET_OK; +} - lh->reconnect_task = GNUNET_SCHEDULER_NO_TASK; - GNUNET_assert (NULL == lh->client); - lh->client = GNUNET_CLIENT_connect ("set", lh->cfg); - if (NULL == lh->client) - { - LOG (GNUNET_ERROR_TYPE_ERROR, - "could not connect to set (wrong configuration?), giving up listening\n"); - return; - } +/** + * Handle request message for a listen operation + * + * @param cls the listen handle + * @param msg the message + */ +static void +handle_request (void *cls, + const struct GNUNET_SET_RequestMessage *msg) +{ + struct GNUNET_SET_ListenHandle *lh = cls; + struct GNUNET_SET_Request req; + const struct GNUNET_MessageHeader *context_msg; + struct GNUNET_MQ_Envelope *mqm; + struct GNUNET_SET_RejectMessage *rmsg; + + LOG (GNUNET_ERROR_TYPE_DEBUG, + "Processing incoming operation request\n"); + /* we got another valid request => reset the backoff */ + lh->reconnect_backoff = GNUNET_TIME_UNIT_MILLISECONDS; + req.accept_id = ntohl (msg->accept_id); + req.accepted = GNUNET_NO; + context_msg = GNUNET_MQ_extract_nested_mh (msg); + /* calling #GNUNET_SET_accept() in the listen cb will set req->accepted */ + lh->listen_cb (lh->listen_cls, + &msg->peer_id, + context_msg, + &req); + if (GNUNET_YES == req.accepted) + return; /* the accept-case is handled in #GNUNET_SET_accept() */ + LOG (GNUNET_ERROR_TYPE_DEBUG, + "Rejecting request\n"); + mqm = GNUNET_MQ_msg (rmsg, + GNUNET_MESSAGE_TYPE_SET_REJECT); + rmsg->accept_reject_id = msg->accept_id; + GNUNET_MQ_send (lh->mq, mqm); +} + + +/** + * Our connection with the set service encountered an error, + * re-initialize with exponential back-off. + * + * @param cls the `struct GNUNET_SET_ListenHandle *` + * @param error reason for the disconnect + */ +static void +handle_client_listener_error (void *cls, + enum GNUNET_MQ_Error error) +{ + struct GNUNET_SET_ListenHandle *lh = cls; + + LOG (GNUNET_ERROR_TYPE_DEBUG, + "Listener broke down (%d), re-connecting\n", + (int) error); + GNUNET_MQ_destroy (lh->mq); + lh->mq = NULL; + lh->reconnect_task = GNUNET_SCHEDULER_add_delayed (lh->reconnect_backoff, + &listen_connect, + lh); + lh->reconnect_backoff = GNUNET_TIME_STD_BACKOFF (lh->reconnect_backoff); +} + + +/** + * Connect to the set service in order to listen for requests. + * + * @param cls the `struct GNUNET_SET_ListenHandle *` to connect + */ +static void +listen_connect (void *cls) +{ + struct GNUNET_SET_ListenHandle *lh = cls; + struct GNUNET_MQ_MessageHandler mq_handlers[] = { + GNUNET_MQ_hd_var_size (request, + GNUNET_MESSAGE_TYPE_SET_REQUEST, + struct GNUNET_SET_RequestMessage, + lh), + GNUNET_MQ_handler_end () + }; + struct GNUNET_MQ_Envelope *mqm; + struct GNUNET_SET_ListenMessage *msg; + + lh->reconnect_task = NULL; GNUNET_assert (NULL == lh->mq); - lh->mq = GNUNET_MQ_queue_for_connection_client (lh->client, mq_handlers, - handle_client_listener_error, lh); + lh->mq = GNUNET_CLIENT_connect (lh->cfg, + "set", + mq_handlers, + &handle_client_listener_error, + lh); + if (NULL == lh->mq) + return; mqm = GNUNET_MQ_msg (msg, GNUNET_MESSAGE_TYPE_SET_LISTEN); msg->operation = htonl (lh->operation); msg->app_id = lh->app_id; - GNUNET_MQ_send (lh->mq, mqm); + GNUNET_MQ_send (lh->mq, + mqm); } @@ -709,7 +962,7 @@ listen_connect (void *cls, * @param app_id id of the application that handles set operation requests * @param listen_cb called for each incoming request matching the operation * and application id - * @param listen_cls handle for listen_cb + * @param listen_cls handle for @a listen_cb * @return a handle that can be used to cancel the listen operation */ struct GNUNET_SET_ListenHandle * @@ -728,7 +981,12 @@ GNUNET_SET_listen (const struct GNUNET_CONFIGURATION_Handle *cfg, lh->operation = operation; lh->app_id = *app_id; lh->reconnect_backoff = GNUNET_TIME_UNIT_MILLISECONDS; - listen_connect (lh, NULL); + listen_connect (lh); + if (NULL == lh->mq) + { + GNUNET_free (lh); + return NULL; + } return lh; } @@ -741,22 +999,17 @@ GNUNET_SET_listen (const struct GNUNET_CONFIGURATION_Handle *cfg, void GNUNET_SET_listen_cancel (struct GNUNET_SET_ListenHandle *lh) { - LOG (GNUNET_ERROR_TYPE_DEBUG, "canceling listener\n"); - /* listener's connection may have failed, thus mq/client could be NULL */ + LOG (GNUNET_ERROR_TYPE_DEBUG, + "Canceling listener\n"); if (NULL != lh->mq) { GNUNET_MQ_destroy (lh->mq); lh->mq = NULL; } - if (NULL != lh->client) - { - GNUNET_CLIENT_disconnect (lh->client); - lh->client = NULL; - } - if (GNUNET_SCHEDULER_NO_TASK != lh->reconnect_task) + if (NULL != lh->reconnect_task) { GNUNET_SCHEDULER_cancel (lh->reconnect_task); - lh->reconnect_task = GNUNET_SCHEDULER_NO_TASK; + lh->reconnect_task = NULL; } GNUNET_free (lh); } @@ -779,6 +1032,7 @@ GNUNET_SET_listen_cancel (struct GNUNET_SET_ListenHandle *lh) struct GNUNET_SET_OperationHandle * GNUNET_SET_accept (struct GNUNET_SET_Request *request, enum GNUNET_SET_ResultMode result_mode, + struct GNUNET_SET_Option options[], GNUNET_SET_ResultIterator result_cb, void *result_cls) { @@ -786,21 +1040,19 @@ GNUNET_SET_accept (struct GNUNET_SET_Request *request, struct GNUNET_SET_OperationHandle *oh; struct GNUNET_SET_AcceptMessage *msg; - GNUNET_assert (NULL != request); GNUNET_assert (GNUNET_NO == request->accepted); + LOG (GNUNET_ERROR_TYPE_DEBUG, + "Client accepts set operation (%d)\n", + result_mode); request->accepted = GNUNET_YES; - - oh = GNUNET_new (struct GNUNET_SET_OperationHandle); - oh->result_cb = result_cb; - oh->result_cls = result_cls; - mqm = GNUNET_MQ_msg (msg, GNUNET_MESSAGE_TYPE_SET_ACCEPT); msg->accept_reject_id = htonl (request->accept_id); msg->result_mode = htonl (result_mode); - + oh = GNUNET_new (struct GNUNET_SET_OperationHandle); + oh->result_cb = result_cb; + oh->result_cls = result_cls; oh->conclude_mqm = mqm; oh->request_id_addr = &msg->request_id; - return oh; } @@ -822,17 +1074,27 @@ int GNUNET_SET_commit (struct GNUNET_SET_OperationHandle *oh, struct GNUNET_SET_Handle *set) { - GNUNET_assert (NULL == oh->set); + if (NULL != oh->set) + { + /* Some other set was already commited for this + * operation, there is a logic bug in the client of this API */ + GNUNET_break (0); + return GNUNET_OK; + } if (GNUNET_YES == set->invalid) return GNUNET_SYSERR; + LOG (GNUNET_ERROR_TYPE_DEBUG, + "Client commits to SET\n"); GNUNET_assert (NULL != oh->conclude_mqm); oh->set = set; GNUNET_CONTAINER_DLL_insert (set->ops_head, set->ops_tail, oh); - oh->request_id = GNUNET_MQ_assoc_add (set->mq, oh); + oh->request_id = GNUNET_MQ_assoc_add (set->mq, + oh); *oh->request_id_addr = htonl (oh->request_id); - GNUNET_MQ_send (set->mq, oh->conclude_mqm); + GNUNET_MQ_send (set->mq, + oh->conclude_mqm); oh->conclude_mqm = NULL; oh->request_id_addr = NULL; return GNUNET_OK; @@ -840,9 +1102,9 @@ GNUNET_SET_commit (struct GNUNET_SET_OperationHandle *oh, /** - * Iterate over all elements in the given set. - * Note that this operation involves transferring every element of the set - * from the service to the client, and is thus costly. + * Iterate over all elements in the given set. Note that this + * operation involves transferring every element of the set from the + * service to the client, and is thus costly. * * @param set the set to iterate over * @param iter the iterator to call for each element @@ -858,15 +1120,13 @@ GNUNET_SET_iterate (struct GNUNET_SET_Handle *set, { struct GNUNET_MQ_Envelope *ev; - GNUNET_assert (NULL != iter); - if (GNUNET_YES == set->invalid) return GNUNET_SYSERR; if (NULL != set->iterator) return GNUNET_NO; LOG (GNUNET_ERROR_TYPE_DEBUG, - "iterating set\n"); + "Iterating over set\n"); set->iterator = iter; set->iterator_cls = iter_cls; ev = GNUNET_MQ_msg_header (GNUNET_MESSAGE_TYPE_SET_ITER_REQUEST); @@ -874,4 +1134,69 @@ GNUNET_SET_iterate (struct GNUNET_SET_Handle *set, return GNUNET_YES; } + +void +GNUNET_SET_copy_lazy (struct GNUNET_SET_Handle *set, + GNUNET_SET_CopyReadyCallback cb, + void *cls) +{ + struct GNUNET_MQ_Envelope *ev; + struct SetCopyRequest *req; + + ev = GNUNET_MQ_msg_header (GNUNET_MESSAGE_TYPE_SET_COPY_LAZY_PREPARE); + GNUNET_MQ_send (set->mq, ev); + + req = GNUNET_new (struct SetCopyRequest); + req->cb = cb; + req->cls = cls; + GNUNET_CONTAINER_DLL_insert (set->copy_req_head, + set->copy_req_tail, + req); +} + + +/** + * Create a copy of an element. The copy + * must be GNUNET_free-d by the caller. + * + * @param element the element to copy + * @return the copied element + */ +struct GNUNET_SET_Element * +GNUNET_SET_element_dup (const struct GNUNET_SET_Element *element) +{ + struct GNUNET_SET_Element *copy; + + copy = GNUNET_malloc (element->size + sizeof (struct GNUNET_SET_Element)); + copy->size = element->size; + copy->element_type = element->element_type; + copy->data = ©[1]; + GNUNET_memcpy (©[1], + element->data, + copy->size); + return copy; +} + + +/** + * Hash a set element. + * + * @param element the element that should be hashed + * @param[out] ret_hash a pointer to where the hash of @a element + * should be stored + */ +void +GNUNET_SET_element_hash (const struct GNUNET_SET_Element *element, + struct GNUNET_HashCode *ret_hash) +{ + struct GNUNET_HashContext *ctx = GNUNET_CRYPTO_hash_context_start (); + + /* It's not guaranteed that the element data is always after the element header, + so we need to hash the chunks separately. */ + GNUNET_CRYPTO_hash_context_read (ctx, &element->size, sizeof (uint16_t)); + GNUNET_CRYPTO_hash_context_read (ctx, &element->element_type, sizeof (uint16_t)); + GNUNET_CRYPTO_hash_context_read (ctx, element->data, element->size); + GNUNET_CRYPTO_hash_context_finish (ctx, ret_hash); +} + /* end of set_api.c */