X-Git-Url: https://git.librecmc.org/?a=blobdiff_plain;f=src%2Ffs%2Fgnunet-service-fs_pe.c;h=ed59c59e603e3e7d6f2066764c83bf85939decc1;hb=da712d081fd688d09a683a8faeb5c84c5c07aa9f;hp=af20c4bf30f493e18cd22c6219e059e390722287;hpb=b65e5b414fdfbf53f989c77aa4c5b27a658fada2;p=oweals%2Fgnunet.git diff --git a/src/fs/gnunet-service-fs_pe.c b/src/fs/gnunet-service-fs_pe.c index af20c4bf3..ed59c59e6 100644 --- a/src/fs/gnunet-service-fs_pe.c +++ b/src/fs/gnunet-service-fs_pe.c @@ -24,11 +24,78 @@ * @author Christian Grothoff */ #include "platform.h" +#include "gnunet-service-fs.h" #include "gnunet-service-fs_cp.h" #include "gnunet-service-fs_pe.h" #include "gnunet-service-fs_pr.h" +/** + * List of GSF_PendingRequests this request plan + * participates with. + */ +struct PendingRequestList; + + +/** + * DLL of request plans a particular pending request is + * involved with. + */ +struct GSF_RequestPlanReference +{ + + /** + * This is a doubly-linked list. + */ + struct GSF_RequestPlanReference *next; + + /** + * This is a doubly-linked list. + */ + struct GSF_RequestPlanReference *prev; + + /** + * Associated request plan. + */ + struct GSF_RequestPlan *rp; + + /** + * Corresponding PendingRequestList. + */ + struct PendingRequestList *prl; +}; + + +/** + * List of GSF_PendingRequests this request plan + * participates with. + */ +struct PendingRequestList +{ + + /** + * This is a doubly-linked list. + */ + struct PendingRequestList *next; + + /** + * This is a doubly-linked list. + */ + struct PendingRequestList *prev; + + /** + * Array of associated pending requests. + */ + struct GSF_PendingRequest *pr; + + /** + * Corresponding GSF_RequestPlanReference. + */ + struct GSF_RequestPlanReference *rpr; + +}; + + /** * Information we keep per request per peer. This is a doubly-linked * list (with head and tail in the 'struct GSF_PendingRequestData') @@ -54,19 +121,34 @@ struct GSF_RequestPlan struct GNUNET_CONTAINER_HeapNode *hn; /** - * Associated pending request. + * Head of list of associated pending requests. */ - struct GSF_PendingRequest *pr; + struct PendingRequestList *prl_head; /** - * Earliest time we'd be happy to transmit this request. + * Tail of list of associated pending requests. + */ + struct PendingRequestList *prl_tail; + + /** + * Earliest time we'd be happy to (re)transmit this request. */ struct GNUNET_TIME_Absolute earliest_transmission; /** - * Priority for this request for this target. + * When was the last time we transmitted this request to this peer? 0 for never. + */ + struct GNUNET_TIME_Absolute last_transmission; + + /** + * Current priority for this request for this target. */ - uint32_t priority; + uint64_t priority; + + /** + * How often did we transmit this request to this peer? + */ + unsigned int transmission_counter; }; @@ -77,9 +159,14 @@ struct GSF_RequestPlan struct PeerPlan { /** - * Heap with pending queries (struct GSF_RequestPlan), smaller weights mean higher priority. + * Heap with pending queries (struct GSF_RequestPlan), higher weights mean higher priority. + */ + struct GNUNET_CONTAINER_Heap *priority_heap; + + /** + * Heap with pending queries (struct GSF_RequestPlan), by transmission time, lowest first. */ - struct GNUNET_CONTAINER_Heap *heap; + struct GNUNET_CONTAINER_Heap *delay_heap; /** * Current transmission request handle. @@ -103,6 +190,27 @@ struct PeerPlan */ static struct GNUNET_CONTAINER_MultiHashMap *plans; +/** + * Sum of all transmission counters (equals total delay for all plan entries). + */ +static unsigned long long total_delay; + +/** + * Number of plan entries. + */ +static unsigned long long plan_count; + + +/** + * Figure out when and how to transmit to the given peer. + * + * @param cls the 'struct GSF_ConnectedPeer' for transmission + * @param tc scheduler context + */ +static void +schedule_peer_transmission (void *cls, + const struct GNUNET_SCHEDULER_TaskContext *tc); + /** * Insert the given request plan into the heap with the appropriate weight. @@ -111,31 +219,67 @@ static struct GNUNET_CONTAINER_MultiHashMap *plans; * @param rp request to plan */ static void -plan (struct PeerPlan *pp, - struct GSF_RequestPlan *rp) +plan (struct PeerPlan *pp, struct GSF_RequestPlan *rp) { - GNUNET_CONTAINER_HeapCostType weight; struct GSF_PendingRequestData *prd; + struct GNUNET_TIME_Relative delay; - prd = GSF_pending_request_get_data_ (rp->pr); - weight = 0; // FIXME: calculate real weight! - // FIXME: calculate 'rp->earliest_transmission'! - // fIXME: claculate 'rp->priority'! - rp->hn = GNUNET_CONTAINER_heap_insert (pp->heap, - rp, - weight); + GNUNET_STATISTICS_set (GSF_stats, + gettext_noop ("# average retransmission delay (ms)"), + total_delay * 1000LL / plan_count, GNUNET_NO); + prd = GSF_pending_request_get_data_ (rp->prl_head->pr); + // FIXME: calculate 'rp->priority'! + if (rp->transmission_counter < 32) + delay = + GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, + 1LL << rp->transmission_counter); + else + delay = GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, UINT_MAX); + rp->earliest_transmission = GNUNET_TIME_relative_to_absolute (delay); +#if DEBUG_FS + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, + "Earliest (re)transmission for `%s' in %us\n", + GNUNET_h2s (&prd->query), rp->transmission_counter); +#endif + + GNUNET_assert (rp->hn == NULL); + if (GNUNET_TIME_absolute_get_remaining (rp->earliest_transmission).rel_value + == 0) + rp->hn = GNUNET_CONTAINER_heap_insert (pp->priority_heap, rp, rp->priority); + else + rp->hn = + GNUNET_CONTAINER_heap_insert (pp->delay_heap, rp, + rp->earliest_transmission.abs_value); + if (GNUNET_SCHEDULER_NO_TASK != pp->task) + GNUNET_SCHEDULER_cancel (pp->task); + pp->task = GNUNET_SCHEDULER_add_now (&schedule_peer_transmission, pp); } /** - * Figure out when and how to transmit to the given peer. + * Get the pending request with the highest TTL from the given plan. * - * @param cls the 'struct GSF_ConnectedPeer' for transmission - * @param tc scheduler context + * @param rp plan to investigate + * @return pending request with highest TTL */ -static void -schedule_peer_transmission (void *cls, - const struct GNUNET_SCHEDULER_TaskContext *tc); +struct GSF_PendingRequest * +get_latest (const struct GSF_RequestPlan *rp) +{ + struct GSF_PendingRequest *ret; + struct PendingRequestList *prl; + + prl = rp->prl_head; + ret = prl->pr; + prl = prl->next; + while (NULL != prl) + { + if (GSF_pending_request_get_data_ (prl->pr)->ttl.abs_value > + GSF_pending_request_get_data_ (ret)->ttl.abs_value) + ret = prl->pr; + prl = prl->next; + } + return ret; +} /** @@ -146,33 +290,49 @@ schedule_peer_transmission (void *cls, * @param buf where to copy the message, NULL on error (peer disconnect) * @return number of bytes copied to 'buf', can be 0 (without indicating an error) */ -static size_t -transmit_message_callback (void *cls, - size_t buf_size, - void *buf) +static size_t +transmit_message_callback (void *cls, size_t buf_size, void *buf) { struct PeerPlan *pp = cls; struct GSF_RequestPlan *rp; size_t msize; + pp->pth = NULL; if (NULL == buf) - { - /* failed, try again... */ - pp->task = GNUNET_SCHEDULER_add_now (&schedule_peer_transmission, pp); - return 0; - } - rp = GNUNET_CONTAINER_heap_peek (pp->heap); - msize = GSF_pending_request_get_message_ (rp->pr, buf_size, buf); + { + /* failed, try again... */ + pp->task = GNUNET_SCHEDULER_add_now (&schedule_peer_transmission, pp); + return 0; + } + rp = GNUNET_CONTAINER_heap_peek (pp->priority_heap); + if (NULL == rp) + { + pp->task = GNUNET_SCHEDULER_add_now (&schedule_peer_transmission, pp); + return 0; + } + msize = GSF_pending_request_get_message_ (get_latest (rp), buf_size, buf); if (msize > buf_size) - { - /* buffer to small (message changed), try again */ - pp->task = GNUNET_SCHEDULER_add_now (&schedule_peer_transmission, pp); - return 0; - } + { + /* buffer to small (message changed), try again */ + pp->task = GNUNET_SCHEDULER_add_now (&schedule_peer_transmission, pp); + return 0; + } /* remove from root, add again elsewhere... */ - GNUNET_assert (rp == GNUNET_CONTAINER_heap_remove_root (pp->heap)); + GNUNET_assert (rp == GNUNET_CONTAINER_heap_remove_root (pp->priority_heap)); rp->hn = NULL; + rp->last_transmission = GNUNET_TIME_absolute_get (); + rp->transmission_counter++; + total_delay++; +#if DEBUG_FS + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, + "Executing plan %p executed %u times, planning retransmission\n", + rp, rp->transmission_counter); +#endif plan (pp, rp); + GNUNET_STATISTICS_update (GSF_stats, + gettext_noop + ("# queries messages sent to other peers"), 1, + GNUNET_NO); return msize; } @@ -185,42 +345,129 @@ transmit_message_callback (void *cls, */ static void schedule_peer_transmission (void *cls, - const struct GNUNET_SCHEDULER_TaskContext *tc) + const struct GNUNET_SCHEDULER_TaskContext *tc) { struct PeerPlan *pp = cls; struct GSF_RequestPlan *rp; - struct GSF_PendingRequestData *prd; size_t msize; - struct GNUNET_TIME_Relative delay; pp->task = GNUNET_SCHEDULER_NO_TASK; - if (NULL == pp->heap) - return; - if (0 == GNUNET_CONTAINER_heap_get_size (pp->heap)) - return; - GNUNET_assert (NULL == pp->pth); - rp = GNUNET_CONTAINER_heap_peek (pp->heap); - prd = GSF_pending_request_get_data_ (rp->pr); - delay = GNUNET_TIME_absolute_get_remaining (rp->earliest_transmission); - if (delay.rel_value > 0) + if (pp->pth != NULL) + { + GSF_peer_transmit_cancel_ (pp->pth); + pp->pth = NULL; + } + /* move ready requests to priority queue */ + while ((NULL != (rp = GNUNET_CONTAINER_heap_peek (pp->delay_heap))) && + (GNUNET_TIME_absolute_get_remaining + (rp->earliest_transmission).rel_value == 0)) + { + GNUNET_assert (rp == GNUNET_CONTAINER_heap_remove_root (pp->delay_heap)); + rp->hn = GNUNET_CONTAINER_heap_insert (pp->priority_heap, rp, rp->priority); + } + if (0 == GNUNET_CONTAINER_heap_get_size (pp->priority_heap)) + { + /* priority heap (still) empty, check for delay... */ + rp = GNUNET_CONTAINER_heap_peek (pp->delay_heap); + if (NULL == rp) { - pp->task = GNUNET_SCHEDULER_add_delayed (delay, - &schedule_peer_transmission, - pp); - return; +#if DEBUG_FS + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "No active requests for plan %p.\n", + pp); +#endif + return; /* both queues empty */ } - msize = GSF_pending_request_get_message_ (rp->pr, 0, NULL); - pp->pth = GSF_peer_transmit_ (pp->cp, - GNUNET_YES, - rp->priority, - GNUNET_TIME_UNIT_FOREVER_REL, - msize, - &transmit_message_callback, - pp); +#if DEBUG_FS + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, + "Sleeping for %llu ms before retrying requests on plan %p.\n", + (unsigned long long) + GNUNET_TIME_absolute_get_remaining + (rp->earliest_transmission).rel_value, pp); +#endif + pp->task = + GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_absolute_get_remaining + (rp->earliest_transmission), + &schedule_peer_transmission, pp); + return; + } + /* process from priority heap */ + rp = GNUNET_CONTAINER_heap_peek (pp->priority_heap); +#if DEBUG_FS > 1 + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Executing query plan %p\n", rp); +#endif + GNUNET_assert (NULL != rp); + msize = GSF_pending_request_get_message_ (get_latest (rp), 0, NULL); + pp->pth = + GSF_peer_transmit_ (pp->cp, GNUNET_YES, rp->priority, + GNUNET_TIME_UNIT_FOREVER_REL, msize, + &transmit_message_callback, pp); GNUNET_assert (NULL != pp->pth); } +/** + * Closure for 'merge_pr'. + */ +struct MergeContext +{ + + struct GSF_PendingRequest *pr; + + int merged; + +}; + + +/** + * Iterator that checks if an equivalent request is already + * present for this peer. + * + * @param cls closure + * @param node internal node of the heap (ignored) + * @param element request plan stored at the node + * @param cost cost associated with the node (ignored) + * @return GNUNET_YES if we should continue to iterate, + * GNUNET_NO if not (merge success) + */ +static int +merge_pr (void *cls, struct GNUNET_CONTAINER_HeapNode *node, void *element, + GNUNET_CONTAINER_HeapCostType cost) +{ + struct MergeContext *mpr = cls; + struct GSF_RequestPlan *rp = element; + struct GSF_PendingRequestData *prd; + struct GSF_RequestPlanReference *rpr; + struct PendingRequestList *prl; + struct GSF_PendingRequest *latest; + + if (GNUNET_OK != + GSF_pending_request_is_compatible_ (mpr->pr, rp->prl_head->pr)) + return GNUNET_YES; + /* merge new request with existing request plan */ + rpr = GNUNET_malloc (sizeof (struct GSF_RequestPlanReference)); + prl = GNUNET_malloc (sizeof (struct PendingRequestList)); + rpr->rp = rp; + rpr->prl = prl; + prl->rpr = rpr; + prl->pr = mpr->pr; + prd = GSF_pending_request_get_data_ (mpr->pr); + GNUNET_CONTAINER_DLL_insert (prd->rpr_head, prd->rpr_tail, rpr); + GNUNET_CONTAINER_DLL_insert (rp->prl_head, rp->prl_tail, prl); + mpr->merged = GNUNET_YES; + GNUNET_STATISTICS_update (GSF_stats, gettext_noop ("# requests merged"), 1, + GNUNET_NO); + latest = get_latest (rp); + if (GSF_pending_request_get_data_ (latest)->ttl.abs_value < + prd->ttl.abs_value) + { + GNUNET_STATISTICS_update (GSF_stats, gettext_noop ("# requests refreshed"), + 1, GNUNET_NO); + rp->transmission_counter = 0; /* reset */ + } + return GNUNET_NO; +} + + /** * Create a new query plan entry. * @@ -228,44 +475,59 @@ schedule_peer_transmission (void *cls, * @param pr request with the entry */ void -GSF_plan_add_ (const struct GSF_ConnectedPeer *cp, - struct GSF_PendingRequest *pr) +GSF_plan_add_ (struct GSF_ConnectedPeer *cp, struct GSF_PendingRequest *pr) { struct GNUNET_PeerIdentity id; struct PeerPlan *pp; struct GSF_PendingRequestData *prd; struct GSF_RequestPlan *rp; - + struct GSF_RequestPlanReference *rpr; + struct PendingRequestList *prl; + struct MergeContext mpc; + + GNUNET_assert (NULL != cp); GSF_connected_peer_get_identity_ (cp, &id); - pp = GNUNET_CONTAINER_multihashmap_get (plans, - &id.hashPubKey); + pp = GNUNET_CONTAINER_multihashmap_get (plans, &id.hashPubKey); if (NULL == pp) - { - pp = GNUNET_malloc (sizeof (struct PeerPlan)); - pp->heap = GNUNET_CONTAINER_heap_create (GNUNET_CONTAINER_HEAP_ORDER_MIN); - GNUNET_CONTAINER_multihashmap_put (plans, - &id.hashPubKey, - pp, - GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY); - } + { + pp = GNUNET_malloc (sizeof (struct PeerPlan)); + pp->priority_heap = + GNUNET_CONTAINER_heap_create (GNUNET_CONTAINER_HEAP_ORDER_MAX); + pp->delay_heap = + GNUNET_CONTAINER_heap_create (GNUNET_CONTAINER_HEAP_ORDER_MIN); + pp->cp = cp; + GNUNET_CONTAINER_multihashmap_put (plans, &id.hashPubKey, pp, + GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY); + } + mpc.merged = GNUNET_NO; + mpc.pr = pr; + /* FIXME: O(n) call here, LRN reports this is a performance + problem. Try using hash map!? */ + GNUNET_CONTAINER_heap_iterate (pp->priority_heap, &merge_pr, &mpc); + if (mpc.merged != GNUNET_NO) + return; + GNUNET_CONTAINER_heap_iterate (pp->delay_heap, &merge_pr, &mpc); + if (mpc.merged != GNUNET_NO) + return; + plan_count++; + GNUNET_STATISTICS_update (GSF_stats, gettext_noop ("# query plan entries"), 1, + GNUNET_NO); prd = GSF_pending_request_get_data_ (pr); +#if DEBUG_FS + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, + "Planning transmission of query `%s' to peer `%s'\n", + GNUNET_h2s (&prd->query), GNUNET_i2s (&id)); +#endif rp = GNUNET_malloc (sizeof (struct GSF_RequestPlan)); - rp->pr = pr; - GNUNET_CONTAINER_DLL_insert (prd->rp_head, - prd->rp_tail, - rp); + rpr = GNUNET_malloc (sizeof (struct GSF_RequestPlanReference)); + prl = GNUNET_malloc (sizeof (struct PendingRequestList)); + rpr->rp = rp; + rpr->prl = prl; + prl->rpr = rpr; + prl->pr = pr; + GNUNET_CONTAINER_DLL_insert (prd->rpr_head, prd->rpr_tail, rpr); + GNUNET_CONTAINER_DLL_insert (rp->prl_head, rp->prl_tail, prl); plan (pp, rp); - if (pp->pth != NULL) - { - if (rp != GNUNET_CONTAINER_heap_peek (pp->heap)) - return; - GSF_peer_transmit_cancel_ (pp->pth); - pp->pth = NULL; - } - if (GNUNET_SCHEDULER_NO_TASK != pp->task) - GNUNET_SCHEDULER_cancel (pp->task); - pp->task = GNUNET_SCHEDULER_add_now (&schedule_peer_transmission, - pp); } @@ -273,7 +535,7 @@ GSF_plan_add_ (const struct GSF_ConnectedPeer *cp, * Notify the plan about a peer being no longer available; * destroy all entries associated with this peer. * - * @param cp connected peer + * @param cp connected peer */ void GSF_plan_notify_peer_disconnect_ (const struct GSF_ConnectedPeer *cp) @@ -282,26 +544,51 @@ GSF_plan_notify_peer_disconnect_ (const struct GSF_ConnectedPeer *cp) struct PeerPlan *pp; struct GSF_RequestPlan *rp; struct GSF_PendingRequestData *prd; + struct PendingRequestList *prl; GSF_connected_peer_get_identity_ (cp, &id); - pp = GNUNET_CONTAINER_multihashmap_get (plans, - &id.hashPubKey); - GNUNET_CONTAINER_multihashmap_remove (plans, - &id.hashPubKey, - pp); + pp = GNUNET_CONTAINER_multihashmap_get (plans, &id.hashPubKey); + if (NULL == pp) + return; /* nothing was ever planned for this peer */ + GNUNET_assert (GNUNET_YES == + GNUNET_CONTAINER_multihashmap_remove (plans, &id.hashPubKey, + pp)); if (NULL != pp->pth) GSF_peer_transmit_cancel_ (pp->pth); if (GNUNET_SCHEDULER_NO_TASK != pp->task) + { GNUNET_SCHEDULER_cancel (pp->task); - while (NULL != (rp = GNUNET_CONTAINER_heap_remove_root (pp->heap))) + pp->task = GNUNET_SCHEDULER_NO_TASK; + } + while (NULL != (rp = GNUNET_CONTAINER_heap_remove_root (pp->priority_heap))) + { + while (NULL != (prl = rp->prl_head)) { - prd = GSF_pending_request_get_data_ (rp->pr); - GNUNET_CONTAINER_DLL_remove (prd->rp_head, - prd->rp_tail, - rp); - GNUNET_free (rp); + GNUNET_CONTAINER_DLL_remove (rp->prl_head, rp->prl_tail, prl); + prd = GSF_pending_request_get_data_ (prl->pr); + GNUNET_CONTAINER_DLL_remove (prd->rpr_head, prd->rpr_tail, prl->rpr); + GNUNET_free (prl->rpr); + GNUNET_free (prl); } - GNUNET_CONTAINER_heap_destroy (pp->heap); + GNUNET_free (rp); + } + GNUNET_CONTAINER_heap_destroy (pp->priority_heap); + while (NULL != (rp = GNUNET_CONTAINER_heap_remove_root (pp->delay_heap))) + { + while (NULL != (prl = rp->prl_head)) + { + GNUNET_CONTAINER_DLL_remove (rp->prl_head, rp->prl_tail, prl); + prd = GSF_pending_request_get_data_ (prl->pr); + GNUNET_CONTAINER_DLL_remove (prd->rpr_head, prd->rpr_tail, prl->rpr); + GNUNET_free (prl->rpr); + GNUNET_free (prl); + } + GNUNET_free (rp); + } + GNUNET_STATISTICS_set (GSF_stats, gettext_noop ("# query plan entries"), + plan_count, GNUNET_NO); + + GNUNET_CONTAINER_heap_destroy (pp->delay_heap); GNUNET_free (pp); } @@ -313,20 +600,29 @@ GSF_plan_notify_peer_disconnect_ (const struct GSF_ConnectedPeer *cp) * @param pr request that is done */ void -GSF_plan_notify_request_done_ (const struct GSF_PendingRequest *pr) +GSF_plan_notify_request_done_ (struct GSF_PendingRequest *pr) { struct GSF_RequestPlan *rp; struct GSF_PendingRequestData *prd; + struct GSF_RequestPlanReference *rpr; - while (NULL != (rp = prd->rp_head)) + prd = GSF_pending_request_get_data_ (pr); + while (NULL != (rpr = prd->rpr_head)) + { + GNUNET_CONTAINER_DLL_remove (prd->rpr_head, prd->rpr_tail, rpr); + rp = rpr->rp; + GNUNET_CONTAINER_DLL_remove (rp->prl_head, rp->prl_tail, rpr->prl); + GNUNET_free (rpr->prl); + GNUNET_free (rpr); + if (rp->prl_head == 0) { - prd = GSF_pending_request_get_data_ (rp->pr); GNUNET_CONTAINER_heap_remove_node (rp->hn); - GNUNET_CONTAINER_DLL_remove (prd->rp_head, - prd->rp_tail, - rp); + plan_count--; GNUNET_free (rp); } + } + GNUNET_STATISTICS_set (GSF_stats, gettext_noop ("# query plan entries"), + plan_count, GNUNET_NO); } @@ -346,8 +642,7 @@ GSF_plan_init () void GSF_plan_done () { - GNUNET_assert (0 == - GNUNET_CONTAINER_multihashmap_size (plans)); + GNUNET_assert (0 == GNUNET_CONTAINER_multihashmap_size (plans)); GNUNET_CONTAINER_multihashmap_destroy (plans); }