X-Git-Url: https://git.librecmc.org/?a=blobdiff_plain;f=src%2Ffs%2Fgnunet-service-fs_pe.c;h=ed59c59e603e3e7d6f2066764c83bf85939decc1;hb=da712d081fd688d09a683a8faeb5c84c5c07aa9f;hp=d0e30f02530a2a5503237ec8ff8503e660dcf3d2;hpb=ff74c423c7528e4920a7591e0e64b0b282a81bec;p=oweals%2Fgnunet.git diff --git a/src/fs/gnunet-service-fs_pe.c b/src/fs/gnunet-service-fs_pe.c index d0e30f025..ed59c59e6 100644 --- a/src/fs/gnunet-service-fs_pe.c +++ b/src/fs/gnunet-service-fs_pe.c @@ -30,6 +30,72 @@ #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') @@ -55,9 +121,14 @@ 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; + + /** + * Tail of list of associated pending requests. + */ + struct PendingRequestList *prl_tail; /** * Earliest time we'd be happy to (re)transmit this request. @@ -119,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. @@ -127,35 +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) { struct GSF_PendingRequestData *prd; - - prd = GSF_pending_request_get_data_ (rp->pr); - // FIXME: calculate 'rp->earliest_transmission'! - // fIXME: claculate 'rp->priority'! - - if (GNUNET_TIME_absolute_get_remaining (rp->earliest_transmission).rel_value == 0) - rp->hn = GNUNET_CONTAINER_heap_insert (pp->priority_heap, - rp, - rp->priority); + struct GNUNET_TIME_Relative delay; + + 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 - rp->hn = GNUNET_CONTAINER_heap_insert (pp->delay_heap, - rp, - rp->earliest_transmission.abs_value); + 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; +} /** @@ -166,10 +290,8 @@ 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; @@ -177,36 +299,40 @@ transmit_message_callback (void *cls, pp->pth = NULL; if (NULL == buf) - { - /* failed, try again... */ - pp->task = GNUNET_SCHEDULER_add_now (&schedule_peer_transmission, pp); - return 0; - } + { + /* 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_ (rp->pr, buf_size, buf); + { + 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->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 + "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; } @@ -219,54 +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; size_t msize; pp->task = GNUNET_SCHEDULER_NO_TASK; - GNUNET_assert (NULL == pp->pth); + 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); - } + 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) { - /* priority heap (still) empty, check for delay... */ - rp = GNUNET_CONTAINER_heap_peek (pp->delay_heap); - if (NULL == rp) - return; /* both queues empty */ - pp->task = GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_absolute_get_remaining (rp->earliest_transmission), - &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 */ } +#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 - GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, - "Executing query plan %p\n", - rp); -#endif +#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_ (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); + 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. * @@ -274,67 +475,59 @@ schedule_peer_transmission (void *cls, * @param pr request with the entry */ void -GSF_plan_add_ (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->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); - } + { + 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); - rp = GNUNET_malloc (sizeof (struct GSF_RequestPlan)); #if DEBUG_FS GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, - "Planning transmission of query `%s' to peer `%s' (%p)\n", - GNUNET_h2s (&prd->query), - GNUNET_i2s (&id), - rp); -#endif - rp->pr = pr; - GNUNET_CONTAINER_DLL_insert (prd->rp_head, - prd->rp_tail, - rp); + "Planning transmission of query `%s' to peer `%s'\n", + GNUNET_h2s (&prd->query), GNUNET_i2s (&id)); +#endif + rp = GNUNET_malloc (sizeof (struct GSF_RequestPlan)); + 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 (0 == GNUNET_CONTAINER_heap_get_size (pp->priority_heap)) - { - /* no request that should be done immediately, figure out delay */ - if (rp != GNUNET_CONTAINER_heap_peek (pp->delay_heap)) - return; /* did not change delay heap top, no need to do anything */ - GNUNET_assert (NULL == pp->pth); - if (GNUNET_SCHEDULER_NO_TASK != pp->task) - GNUNET_SCHEDULER_cancel (pp->task); - pp->task = GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_absolute_get_remaining (rp->earliest_transmission), - &schedule_peer_transmission, - pp); - return; - } - - if (pp->pth != NULL) - { - if (rp != GNUNET_CONTAINER_heap_peek (pp->priority_heap)) - return; /* did not change priority heap top, no need to do anyhing */ - 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); } @@ -342,7 +535,7 @@ GSF_plan_add_ (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) @@ -351,36 +544,50 @@ 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); + pp = GNUNET_CONTAINER_multihashmap_get (plans, &id.hashPubKey); if (NULL == pp) - return; /* nothing was ever planned for this peer */ - GNUNET_CONTAINER_multihashmap_remove (plans, - &id.hashPubKey, - 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); + 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_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)) { - 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_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); } @@ -397,16 +604,25 @@ GSF_plan_notify_request_done_ (struct GSF_PendingRequest *pr) { struct GSF_RequestPlan *rp; struct GSF_PendingRequestData *prd; + struct GSF_RequestPlanReference *rpr; prd = GSF_pending_request_get_data_ (pr); - while (NULL != (rp = prd->rp_head)) + 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) { 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); } @@ -426,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); }