X-Git-Url: https://git.librecmc.org/?a=blobdiff_plain;f=src%2Fpeerstore%2Fpeerstore_api.c;h=2138b23ff322033757e4aefa6c457157ef90c47c;hb=0f649608b4d8c3f82fd1f240bc0d28c2713c7699;hp=46f170a23f61c48675d56ed55a5a725ec92c9a19;hpb=fb12af8319a84389a536ab2f3735daf3b24f11bd;p=oweals%2Fgnunet.git diff --git a/src/peerstore/peerstore_api.c b/src/peerstore/peerstore_api.c index 46f170a23..2138b23ff 100644 --- a/src/peerstore/peerstore_api.c +++ b/src/peerstore/peerstore_api.c @@ -132,11 +132,6 @@ struct GNUNET_PEERSTORE_StoreContext */ void *value; - /** - * MQ Envelope with store request message - */ - struct GNUNET_MQ_Envelope *ev; - /** * Peer the store is for. */ @@ -180,9 +175,24 @@ struct GNUNET_PEERSTORE_IterateContext struct GNUNET_PEERSTORE_Handle *h; /** - * MQ Envelope with iterate request message + * Which subsystem does the store? */ - struct GNUNET_MQ_Envelope *ev; + char *sub_system; + + /** + * Peer the store is for. + */ + struct GNUNET_PeerIdentity peer; + + /** + * Key for the store operation. + */ + char *key; + + /** + * Operation timeout + */ + struct GNUNET_TIME_Relative timeout; /** * Callback with each matching record @@ -195,8 +205,7 @@ struct GNUNET_PEERSTORE_IterateContext void *callback_cls; /** - * #GNUNET_YES / #GNUNET_NO - * Iterate request has been sent and we are still expecting records + * #GNUNET_YES if we are currently processing records. */ int iterating; @@ -204,7 +213,7 @@ struct GNUNET_PEERSTORE_IterateContext * Task identifier for the function called * on iterate request timeout */ - GNUNET_SCHEDULER_TaskIdentifier timeout_task; + struct GNUNET_SCHEDULER_Task * timeout_task; }; @@ -228,11 +237,6 @@ struct GNUNET_PEERSTORE_WatchContext */ struct GNUNET_PEERSTORE_Handle *h; - /** - * MQ Envelope with watch request message - */ - struct GNUNET_MQ_Envelope *ev; - /** * Callback with each record received */ @@ -248,12 +252,6 @@ struct GNUNET_PEERSTORE_WatchContext */ struct GNUNET_HashCode keyhash; - /** - * #GNUNET_YES / #GNUNET_NO - * if sent, cannot be canceled - */ - int request_sent; - }; /******************************************************************************/ @@ -286,37 +284,6 @@ handle_watch_result (void *cls, const struct GNUNET_MessageHeader *msg); static void reconnect (struct GNUNET_PEERSTORE_Handle *h); -/** - * Callback after MQ envelope is sent - * - * @param cls a 'struct GNUNET_PEERSTORE_WatchContext *' - */ -static void -watch_request_sent (void *cls) -{ - struct GNUNET_PEERSTORE_WatchContext *wc = cls; - - wc->request_sent = GNUNET_YES; - wc->ev = NULL; -} - - -/** - * Callback after MQ envelope is sent - * - * @param cls a `struct GNUNET_PEERSTORE_IterateContext *` - */ -static void -iterate_request_sent (void *cls) -{ - struct GNUNET_PEERSTORE_IterateContext *ic = cls; - - LOG (GNUNET_ERROR_TYPE_DEBUG, - "Iterate request sent to service.\n"); - ic->iterating = GNUNET_YES; - ic->ev = NULL; -} - /** * Callback after MQ envelope is sent @@ -330,7 +297,6 @@ store_request_sent (void *cls) GNUNET_PEERSTORE_Continuation cont; void *cont_cls; - sc->ev = NULL; cont = sc->cont; cont_cls = sc->cont_cls; GNUNET_PEERSTORE_store_cancel (sc); @@ -367,28 +333,49 @@ handle_client_error (void *cls, enum GNUNET_MQ_Error error) /** * Iterator over previous watches to resend them + * + * @param cls the `struct GNUNET_PEERSTORE_Handle` + * @param key key for the watch + * @param value the `struct GNUNET_PEERSTORE_WatchContext *` + * @return #GNUNET_YES (continue to iterate) */ static int -rewatch_it (void *cls, - const struct GNUNET_HashCode *key, - void *value) +rewatch_it (void *cls, const struct GNUNET_HashCode *key, void *value) { struct GNUNET_PEERSTORE_Handle *h = cls; struct GNUNET_PEERSTORE_WatchContext *wc = value; struct StoreKeyHashMessage *hm; + struct GNUNET_MQ_Envelope *ev; - if (GNUNET_YES == wc->request_sent) - { /* Envelope gone, create new one. */ - wc->ev = GNUNET_MQ_msg (hm, GNUNET_MESSAGE_TYPE_PEERSTORE_WATCH); - hm->keyhash = wc->keyhash; - wc->request_sent = GNUNET_NO; - } - GNUNET_MQ_notify_sent (wc->ev, &watch_request_sent, wc); - GNUNET_MQ_send (h->mq, wc->ev); + ev = GNUNET_MQ_msg (hm, GNUNET_MESSAGE_TYPE_PEERSTORE_WATCH); + hm->keyhash = wc->keyhash; + GNUNET_MQ_send (h->mq, ev); return GNUNET_YES; } +/** + * Called when the iterate request is timedout + * + * @param cls a `struct GNUNET_PEERSTORE_IterateContext *` + * @param tc Scheduler task context (unused) + */ +static void +iterate_timeout (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc) +{ + struct GNUNET_PEERSTORE_IterateContext *ic = cls; + GNUNET_PEERSTORE_Processor callback; + void *callback_cls; + + ic->timeout_task = NULL; + callback = ic->callback; + callback_cls = ic->callback_cls; + GNUNET_PEERSTORE_iterate_cancel (ic); + if (NULL != callback) + callback (callback_cls, NULL, _("timeout")); +} + + /** * Close the existing connection to PEERSTORE and reconnect. * @@ -398,19 +385,21 @@ static void reconnect (struct GNUNET_PEERSTORE_Handle *h) { struct GNUNET_PEERSTORE_IterateContext *ic; + struct GNUNET_PEERSTORE_IterateContext *next; GNUNET_PEERSTORE_Processor icb; void *icb_cls; struct GNUNET_PEERSTORE_StoreContext *sc; + struct GNUNET_MQ_Envelope *ev; - LOG (GNUNET_ERROR_TYPE_DEBUG, - "Reconnecting...\n"); - for (sc = h->store_head; NULL != sc; sc = sc->next) + LOG (GNUNET_ERROR_TYPE_DEBUG, "Reconnecting...\n"); + for (ic = h->iterate_head; NULL != ic; ic = next) { - if (NULL != sc->ev) - { - GNUNET_MQ_send_cancel (sc->ev); - sc->ev = NULL; - } + next = ic->next; + icb = ic->callback; + icb_cls = ic->callback_cls; + GNUNET_PEERSTORE_iterate_cancel (ic); + if (NULL != icb) + icb (icb_cls, NULL, _("Iteration canceled due to reconnection.")); } if (NULL != h->mq) { @@ -422,6 +411,7 @@ reconnect (struct GNUNET_PEERSTORE_Handle *h) GNUNET_CLIENT_disconnect (h->client); h->client = NULL; } + h->client = GNUNET_CLIENT_connect ("peerstore", h->cfg); GNUNET_assert (NULL != h->client); h->mq = @@ -430,43 +420,26 @@ reconnect (struct GNUNET_PEERSTORE_Handle *h) LOG (GNUNET_ERROR_TYPE_DEBUG, "Resending pending requests after reconnect.\n"); if (NULL != h->watches) - GNUNET_CONTAINER_multihashmap_iterate (h->watches, - &rewatch_it, - h); + GNUNET_CONTAINER_multihashmap_iterate (h->watches, &rewatch_it, h); for (ic = h->iterate_head; NULL != ic; ic = ic->next) { - if (GNUNET_YES == ic->iterating) - { - icb = ic->callback; - icb_cls = ic->callback_cls; - GNUNET_PEERSTORE_iterate_cancel (ic); - if (NULL != icb) - icb (icb_cls, - NULL, - _("Iteration canceled due to reconnection.")); - } - else - { - GNUNET_MQ_notify_sent (ic->ev, - &iterate_request_sent, - ic); - GNUNET_MQ_send (h->mq, ic->ev); - } + ev = + PEERSTORE_create_record_mq_envelope (ic->sub_system, &ic->peer, ic->key, + NULL, 0, NULL, 0, + GNUNET_MESSAGE_TYPE_PEERSTORE_ITERATE); + GNUNET_MQ_send (h->mq, ev); + ic->timeout_task = + GNUNET_SCHEDULER_add_delayed (ic->timeout, &iterate_timeout, ic); } for (sc = h->store_head; NULL != sc; sc = sc->next) { - sc->ev = PEERSTORE_create_record_mq_envelope (sc->sub_system, - &sc->peer, - sc->key, - sc->value, - sc->size, - &sc->expiry, - sc->options, - GNUNET_MESSAGE_TYPE_PEERSTORE_STORE); - GNUNET_MQ_notify_sent (sc->ev, - &store_request_sent, - sc); - GNUNET_MQ_send (h->mq, sc->ev); + ev = + PEERSTORE_create_record_mq_envelope (sc->sub_system, &sc->peer, sc->key, + sc->value, sc->size, &sc->expiry, + sc->options, + GNUNET_MESSAGE_TYPE_PEERSTORE_STORE); + GNUNET_MQ_notify_sent (ev, &store_request_sent, sc); + GNUNET_MQ_send (h->mq, ev); } } @@ -480,9 +453,7 @@ reconnect (struct GNUNET_PEERSTORE_Handle *h) * @return #GNUNET_YES to continue iteration */ static int -destroy_watch (void *cls, - const struct GNUNET_HashCode *key, - void *value) +destroy_watch (void *cls, const struct GNUNET_HashCode *key, void *value) { struct GNUNET_PEERSTORE_WatchContext *wc = value; @@ -545,8 +516,7 @@ GNUNET_PEERSTORE_connect (const struct GNUNET_CONFIGURATION_Handle *cfg) GNUNET_free (h); return NULL; } - LOG (GNUNET_ERROR_TYPE_DEBUG, - "New connection created\n"); + LOG (GNUNET_ERROR_TYPE_DEBUG, "New connection created\n"); return h; } @@ -567,13 +537,10 @@ GNUNET_PEERSTORE_disconnect (struct GNUNET_PEERSTORE_Handle *h, int sync_first) struct GNUNET_PEERSTORE_StoreContext *sc; struct GNUNET_PEERSTORE_StoreContext *sc_iter; - LOG (GNUNET_ERROR_TYPE_DEBUG, - "Disconnecting.\n"); + LOG (GNUNET_ERROR_TYPE_DEBUG, "Disconnecting.\n"); if (NULL != h->watches) { - GNUNET_CONTAINER_multihashmap_iterate (h->watches, - &destroy_watch, - NULL); + GNUNET_CONTAINER_multihashmap_iterate (h->watches, &destroy_watch, NULL); GNUNET_CONTAINER_multihashmap_destroy (h->watches); h->watches = NULL; } @@ -621,20 +588,12 @@ GNUNET_PEERSTORE_store_cancel (struct GNUNET_PEERSTORE_StoreContext *sc) { struct GNUNET_PEERSTORE_Handle *h = sc->h; - if (NULL != sc->ev) - { - GNUNET_MQ_send_cancel (sc->ev); - sc->ev = NULL; - } - GNUNET_CONTAINER_DLL_remove (sc->h->store_head, - sc->h->store_tail, - sc); + GNUNET_CONTAINER_DLL_remove (sc->h->store_head, sc->h->store_tail, sc); GNUNET_free (sc->sub_system); GNUNET_free (sc->value); GNUNET_free (sc->key); GNUNET_free (sc); - if ( (GNUNET_YES == h->disconnecting) && - (NULL == h->store_head) ) + if ((GNUNET_YES == h->disconnecting) && (NULL == h->store_head)) do_disconnect (h); } @@ -658,14 +617,11 @@ GNUNET_PEERSTORE_store_cancel (struct GNUNET_PEERSTORE_StoreContext *sc) struct GNUNET_PEERSTORE_StoreContext * GNUNET_PEERSTORE_store (struct GNUNET_PEERSTORE_Handle *h, const char *sub_system, - const struct GNUNET_PeerIdentity *peer, - const char *key, - const void *value, - size_t size, + const struct GNUNET_PeerIdentity *peer, const char *key, + const void *value, size_t size, struct GNUNET_TIME_Absolute expiry, enum GNUNET_PEERSTORE_StoreOption options, - GNUNET_PEERSTORE_Continuation cont, - void *cont_cls) + GNUNET_PEERSTORE_Continuation cont, void *cont_cls) { struct GNUNET_MQ_Envelope *ev; struct GNUNET_PEERSTORE_StoreContext *sc; @@ -673,15 +629,11 @@ GNUNET_PEERSTORE_store (struct GNUNET_PEERSTORE_Handle *h, LOG (GNUNET_ERROR_TYPE_DEBUG, "Storing value (size: %lu) for subsytem `%s', peer `%s', key `%s'\n", size, sub_system, GNUNET_i2s (peer), key); - ev = PEERSTORE_create_record_mq_envelope (sub_system, - peer, - key, - value, - size, - &expiry, - options, + ev = PEERSTORE_create_record_mq_envelope (sub_system, peer, key, value, size, + &expiry, options, GNUNET_MESSAGE_TYPE_PEERSTORE_STORE); sc = GNUNET_new (struct GNUNET_PEERSTORE_StoreContext); + sc->sub_system = GNUNET_strdup (sub_system); sc->peer = *peer; sc->key = GNUNET_strdup (key); @@ -693,12 +645,8 @@ GNUNET_PEERSTORE_store (struct GNUNET_PEERSTORE_Handle *h, sc->cont_cls = cont_cls; sc->h = h; - GNUNET_CONTAINER_DLL_insert_tail (h->store_head, - h->store_tail, - sc); - GNUNET_MQ_notify_sent (ev, - &store_request_sent, - sc); + GNUNET_CONTAINER_DLL_insert_tail (h->store_head, h->store_tail, sc); + GNUNET_MQ_notify_sent (ev, &store_request_sent, sc); GNUNET_MQ_send (h->mq, ev); return sc; @@ -716,8 +664,7 @@ GNUNET_PEERSTORE_store (struct GNUNET_PEERSTORE_Handle *h, * @param msg message received, NULL on timeout or fatal error */ static void -handle_iterate_result (void *cls, - const struct GNUNET_MessageHeader *msg) +handle_iterate_result (void *cls, const struct GNUNET_MessageHeader *msg) { struct GNUNET_PEERSTORE_Handle *h = cls; struct GNUNET_PEERSTORE_IterateContext *ic; @@ -735,6 +682,7 @@ handle_iterate_result (void *cls, reconnect (h); return; } + ic->iterating = GNUNET_YES; callback = ic->callback; callback_cls = ic->callback_cls; if (NULL == msg) /* Connection error */ @@ -772,31 +720,6 @@ handle_iterate_result (void *cls, } -/** - * Called when the iterate request is timedout - * - * @param cls a `struct GNUNET_PEERSTORE_IterateContext *` - * @param tc Scheduler task context (unused) - */ -static void -iterate_timeout (void *cls, - const struct GNUNET_SCHEDULER_TaskContext *tc) -{ - struct GNUNET_PEERSTORE_IterateContext *ic = cls; - GNUNET_PEERSTORE_Processor callback; - void *callback_cls; - - ic->timeout_task = GNUNET_SCHEDULER_NO_TASK; - callback = ic->callback; - callback_cls = ic->callback_cls; - GNUNET_PEERSTORE_iterate_cancel (ic); - if (NULL != callback) - callback (callback_cls, - NULL, - _("timeout")); -} - - /** * Cancel an iterate request * Please do not call after the iterate request is done @@ -806,21 +729,17 @@ iterate_timeout (void *cls, void GNUNET_PEERSTORE_iterate_cancel (struct GNUNET_PEERSTORE_IterateContext *ic) { - if (GNUNET_SCHEDULER_NO_TASK != ic->timeout_task) + if (NULL != ic->timeout_task) { GNUNET_SCHEDULER_cancel (ic->timeout_task); - ic->timeout_task = GNUNET_SCHEDULER_NO_TASK; + ic->timeout_task = NULL; } if (GNUNET_NO == ic->iterating) { - if (NULL != ic->ev) - { - GNUNET_MQ_send_cancel (ic->ev); - ic->ev = NULL; - } - GNUNET_CONTAINER_DLL_remove (ic->h->iterate_head, - ic->h->iterate_tail, - ic); + GNUNET_CONTAINER_DLL_remove (ic->h->iterate_head, ic->h->iterate_tail, ic); + GNUNET_free (ic->sub_system); + if (NULL != ic->key) + GNUNET_free (ic->key); GNUNET_free (ic); } else @@ -844,8 +763,7 @@ struct GNUNET_PEERSTORE_IterateContext * GNUNET_PEERSTORE_iterate (struct GNUNET_PEERSTORE_Handle *h, const char *sub_system, const struct GNUNET_PeerIdentity *peer, - const char *key, - struct GNUNET_TIME_Relative timeout, + const char *key, struct GNUNET_TIME_Relative timeout, GNUNET_PEERSTORE_Processor callback, void *callback_cls) { @@ -859,15 +777,18 @@ GNUNET_PEERSTORE_iterate (struct GNUNET_PEERSTORE_Handle *h, ic->callback = callback; ic->callback_cls = callback_cls; - ic->ev = ev; ic->h = h; - ic->iterating = GNUNET_NO; + ic->sub_system = GNUNET_strdup (sub_system); + if (NULL != peer) + ic->peer = *peer; + if (NULL != key) + ic->key = GNUNET_strdup (key); + ic->timeout = timeout; GNUNET_CONTAINER_DLL_insert_tail (h->iterate_head, h->iterate_tail, ic); LOG (GNUNET_ERROR_TYPE_DEBUG, "Sending an iterate request for sub system `%s'\n", sub_system); - GNUNET_MQ_notify_sent (ev, &iterate_request_sent, ic); GNUNET_MQ_send (h->mq, ev); ic->timeout_task = GNUNET_SCHEDULER_add_delayed (timeout, &iterate_timeout, ic); @@ -896,14 +817,15 @@ handle_watch_result (void *cls, const struct GNUNET_MessageHeader *msg) if (NULL == msg) { LOG (GNUNET_ERROR_TYPE_ERROR, - _("Problem receiving a watch response, no way to determine which request.\n")); + _ + ("Problem receiving a watch response, no way to determine which request.\n")); reconnect (h); return; } - LOG (GNUNET_ERROR_TYPE_DEBUG, - "Received a watch record from service.\n"); + LOG (GNUNET_ERROR_TYPE_DEBUG, "Received a watch record from service.\n"); record = PEERSTORE_parse_record_message (msg); PEERSTORE_hash_key (record->sub_system, record->peer, record->key, &keyhash); + // FIXME: what if there are multiple watches for the same key? wc = GNUNET_CONTAINER_multihashmap_get (h->watches, &keyhash); if (NULL == wc) { @@ -931,23 +853,15 @@ GNUNET_PEERSTORE_watch_cancel (struct GNUNET_PEERSTORE_WatchContext *wc) struct GNUNET_MQ_Envelope *ev; struct StoreKeyHashMessage *hm; - LOG (GNUNET_ERROR_TYPE_DEBUG, "Canceling watch.\n"); - if (GNUNET_YES == wc->request_sent) /* If request already sent to service, send a cancel request. */ - { - ev = GNUNET_MQ_msg (hm, GNUNET_MESSAGE_TYPE_PEERSTORE_WATCH_CANCEL); - hm->keyhash = wc->keyhash; - GNUNET_MQ_send (h->mq, ev); - wc->callback = NULL; - wc->callback_cls = NULL; - } - if (NULL != wc->ev) - { - GNUNET_MQ_send_cancel (wc->ev); - wc->ev = NULL; - } - GNUNET_CONTAINER_multihashmap_remove (h->watches, &wc->keyhash, wc); + LOG (GNUNET_ERROR_TYPE_DEBUG, + "Canceling watch.\n"); + ev = GNUNET_MQ_msg (hm, GNUNET_MESSAGE_TYPE_PEERSTORE_WATCH_CANCEL); + hm->keyhash = wc->keyhash; + GNUNET_MQ_send (h->mq, ev); + GNUNET_CONTAINER_multihashmap_remove (h->watches, + &wc->keyhash, + wc); GNUNET_free (wc); - } @@ -977,23 +891,27 @@ GNUNET_PEERSTORE_watch (struct GNUNET_PEERSTORE_Handle *h, GNUNET_assert (NULL != peer); GNUNET_assert (NULL != key); ev = GNUNET_MQ_msg (hm, GNUNET_MESSAGE_TYPE_PEERSTORE_WATCH); - PEERSTORE_hash_key (sub_system, peer, key, &hm->keyhash); + PEERSTORE_hash_key (sub_system, + peer, + key, + &hm->keyhash); wc = GNUNET_new (struct GNUNET_PEERSTORE_WatchContext); - wc->callback = callback; wc->callback_cls = callback_cls; - wc->ev = ev; wc->h = h; - wc->request_sent = GNUNET_NO; wc->keyhash = hm->keyhash; if (NULL == h->watches) h->watches = GNUNET_CONTAINER_multihashmap_create (5, GNUNET_NO); - GNUNET_CONTAINER_multihashmap_put (h->watches, &wc->keyhash, wc, - GNUNET_CONTAINER_MULTIHASHMAPOPTION_REPLACE); + GNUNET_assert (GNUNET_OK == + GNUNET_CONTAINER_multihashmap_put (h->watches, + &wc->keyhash, + wc, + GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE)); LOG (GNUNET_ERROR_TYPE_DEBUG, "Sending a watch request for ss `%s', peer `%s', key `%s'.\n", - sub_system, GNUNET_i2s (peer), key); - GNUNET_MQ_notify_sent (ev, &watch_request_sent, wc); + sub_system, + GNUNET_i2s (peer), + key); GNUNET_MQ_send (h->mq, ev); return wc; }