X-Git-Url: https://git.librecmc.org/?a=blobdiff_plain;f=src%2Ffs%2Fgnunet-service-fs_cp.c;h=7e5ffb45e02e495d74b83699bc63aa666e2202fa;hb=89c5d0047e261ed7e5fabb2e771123f1707e4670;hp=c111751e4d1d3d2ff9b83f0417840087328c341e;hpb=0983dfa8fb93783e3f1893f633a2a690dbdb258c;p=oweals%2Fgnunet.git diff --git a/src/fs/gnunet-service-fs_cp.c b/src/fs/gnunet-service-fs_cp.c index c111751e4..7e5ffb45e 100644 --- a/src/fs/gnunet-service-fs_cp.c +++ b/src/fs/gnunet-service-fs_cp.c @@ -1,6 +1,6 @@ /* This file is part of GNUnet. - (C) 2011 Christian Grothoff (and other contributing authors) + Copyright (C) 2011 Christian Grothoff (and other contributing authors) GNUnet is free software; you can redistribute it and/or modify it under the terms of the GNU General Public License as published @@ -14,33 +14,47 @@ 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 fs/gnunet-service-fs_cp.c * @brief API to handle 'connected peers' * @author Christian Grothoff */ #include "platform.h" +#include "gnunet_util_lib.h" #include "gnunet_load_lib.h" #include "gnunet-service-fs.h" #include "gnunet-service-fs_cp.h" #include "gnunet-service-fs_pe.h" #include "gnunet-service-fs_pr.h" #include "gnunet-service-fs_push.h" +#include "gnunet_peerstore_service.h" + + +/** + * Ratio for moving average delay calculation. The previous + * average goes in with a factor of (n-1) into the calculation. + * Must be > 0. + */ +#define RUNAVG_DELAY_N 16 /** - * How often do we flush trust values to disk? + * How often do we flush respect values to disk? */ -#define TRUST_FLUSH_FREQ GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MINUTES, 5) +#define RESPECT_FLUSH_FREQ GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MINUTES, 5) /** * After how long do we discard a reply? */ #define REPLY_TIMEOUT GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MINUTES, 2) +/** + * Collect an instane number of statistics? May cause excessive IPC. + */ +#define INSANE_STATISTICS GNUNET_NO + /** * Handle to cancel a transmission request. @@ -58,12 +72,6 @@ struct GSF_PeerTransmitHandle */ struct GSF_PeerTransmitHandle *prev; - /** - * Handle for an active request for transmission to this - * peer, or NULL (if core queue was full). - */ - struct GNUNET_CORE_TransmitHandle *cth; - /** * Time when this transmission request was issued. */ @@ -77,7 +85,7 @@ struct GSF_PeerTransmitHandle /** * Task called on timeout, or 0 for none. */ - GNUNET_SCHEDULER_TaskIdentifier timeout_task; + struct GNUNET_SCHEDULER_Task *timeout_task; /** * Function to call to get the actual message. @@ -90,7 +98,7 @@ struct GSF_PeerTransmitHandle struct GSF_ConnectedPeer *cp; /** - * Closure for 'gmc'. + * Closure for @e gmc. */ void *gmc_cls; @@ -100,15 +108,7 @@ struct GSF_PeerTransmitHandle size_t size; /** - * Set to 1 if we're currently in the process of calling - * 'GNUNET_CORE_notify_transmit_ready' (so while cth is - * NULL, we should not call notify_transmit_ready for this - * handle right now). - */ - unsigned int cth_in_progress; - - /** - * GNUNET_YES if this is a query, GNUNET_NO for content. + * #GNUNET_YES if this is a query, #GNUNET_NO for content. */ int is_query; @@ -134,7 +134,7 @@ struct GSF_DelayedHandle /** * Kept in a doubly-linked list. */ - struct GSF_DelayedHandle *next; + struct GSF_DelayedHandle *next; /** * Kept in a doubly-linked list. @@ -154,13 +154,13 @@ struct GSF_DelayedHandle /** * Task for the delay. */ - GNUNET_SCHEDULER_TaskIdentifier delay_task; + struct GNUNET_SCHEDULER_Task *delay_task; /** * Size of the message. */ size_t msize; - + }; @@ -174,7 +174,7 @@ struct PeerRequest * Handle to generic request. */ struct GSF_PendingRequest *pr; - + /** * Handle to specific peer. */ @@ -183,7 +183,7 @@ struct PeerRequest /** * Task for asynchronous stopping of this request. */ - GNUNET_SCHEDULER_TaskIdentifier kill_task; + struct GNUNET_SCHEDULER_Task *kill_task; }; @@ -191,7 +191,7 @@ struct PeerRequest /** * A connected peer. */ -struct GSF_ConnectedPeer +struct GSF_ConnectedPeer { /** @@ -208,7 +208,7 @@ struct GSF_ConnectedPeer /** * Task scheduled to revive migration to this peer. */ - GNUNET_SCHEDULER_TaskIdentifier mig_revive_task; + struct GNUNET_SCHEDULER_Task *mig_revive_task; /** * Messages (replies, queries, content migration) we would like to @@ -240,20 +240,26 @@ struct GSF_ConnectedPeer struct GSF_PeerTransmitHandle *migration_pth; /** - * Context of our GNUNET_CORE_peer_change_preference call (or NULL). + * Context of our GNUNET_ATS_reserve_bandwidth call (or NULL). */ - struct GNUNET_CORE_InformationRequestContext *irc; + struct GNUNET_ATS_ReservationContext *rc; /** * Task scheduled if we need to retry bandwidth reservation later. */ - GNUNET_SCHEDULER_TaskIdentifier irc_delay_task; + struct GNUNET_SCHEDULER_Task *rc_delay_task; /** * Active requests from this neighbour, map of query to 'struct PeerRequest'. */ struct GNUNET_CONTAINER_MultiHashMap *request_map; + /** + * Handle for an active request for transmission to this + * peer, or NULL (if core queue was full). + */ + struct GNUNET_CORE_TransmitHandle *cth; + /** * Increase in traffic preference still to be submitted * to the core service for this peer. @@ -261,111 +267,98 @@ struct GSF_ConnectedPeer uint64_t inc_preference; /** - * Trust rating for this peer on disk. + * Set to 1 if we're currently in the process of calling + * #GNUNET_CORE_notify_transmit_ready() (so while @e cth is + * NULL, we should not call notify_transmit_ready for this + * handle right now). + */ + unsigned int cth_in_progress; + + /** + * Number of entries in @e delayed_head DLL. + */ + unsigned int delay_queue_size; + + /** + * Respect rating for this peer on disk. */ - uint32_t disk_trust; + uint32_t disk_respect; /** - * Which offset in "last_p2p_replies" will be updated next? + * Which offset in @e last_p2p_replies will be updated next? * (we go round-robin). */ unsigned int last_p2p_replies_woff; /** - * Which offset in "last_client_replies" will be updated next? + * Which offset in @e last_client_replies will be updated next? * (we go round-robin). */ unsigned int last_client_replies_woff; /** - * Current offset into 'last_request_times' ring buffer. + * Current offset into @e last_request_times ring buffer. */ unsigned int last_request_times_off; /** - * GNUNET_YES if we did successfully reserve 32k bandwidth, - * GNUNET_NO if not. + * #GNUNET_YES if we did successfully reserve 32k bandwidth, + * #GNUNET_NO if not. */ int did_reserve; -}; + /** + * Function called when the creation of this record is complete. + */ + GSF_ConnectedPeerCreationCallback creation_cb; + /** + * Closure for @e creation_cb + */ + void *creation_cb_cls; -/** - * Map from peer identities to 'struct GSF_ConnectPeer' entries. - */ -static struct GNUNET_CONTAINER_MultiHashMap *cp_map; + /** + * Handle to the PEERSTORE iterate request for peer respect value + */ + struct GNUNET_PEERSTORE_IterateContext *respect_iterate_req; -/** - * Where do we store trust information? - */ -static char *trustDirectory; +}; /** - * Get the filename under which we would store the GNUNET_HELLO_Message - * for the given host and protocol. - * @return filename of the form DIRECTORY/HOSTID + * Map from peer identities to `struct GSF_ConnectPeer` entries. */ -static char * -get_trust_filename (const struct GNUNET_PeerIdentity *id) -{ - struct GNUNET_CRYPTO_HashAsciiEncoded fil; - char *fn; - - GNUNET_CRYPTO_hash_to_enc (&id->hashPubKey, &fil); - GNUNET_asprintf (&fn, "%s%s%s", trustDirectory, DIR_SEPARATOR_STR, &fil); - return fn; -} - +static struct GNUNET_CONTAINER_MultiPeerMap *cp_map; /** - * Find latency information in 'atsi'. - * - * @param atsi performance data - * @return connection latency + * Handle to peerstore service. */ -static struct GNUNET_TIME_Relative -get_latency (const struct GNUNET_TRANSPORT_ATS_Information *atsi) -{ - if (atsi == NULL) - return GNUNET_TIME_UNIT_SECONDS; - while ( (ntohl (atsi->type) != GNUNET_TRANSPORT_ATS_ARRAY_TERMINATOR) && - (ntohl (atsi->type) != GNUNET_TRANSPORT_ATS_QUALITY_NET_DELAY) ) - atsi++; - if (ntohl (atsi->type) == GNUNET_TRANSPORT_ATS_ARRAY_TERMINATOR) - { - GNUNET_break (0); - /* how can we not have latency data? */ - return GNUNET_TIME_UNIT_SECONDS; - } - return GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MILLISECONDS, - ntohl (atsi->value)); -} +static struct GNUNET_PEERSTORE_Handle *peerstore; /** - * Update the performance information kept for the given peer. + * Update the latency information kept for the given peer. * - * @param cp peer record to update - * @param atsi transport performance data + * @param id peer record to update + * @param latency current latency value */ -static void -update_atsi (struct GSF_ConnectedPeer *cp, - const struct GNUNET_TRANSPORT_ATS_Information *atsi) +void +GSF_update_peer_latency_ (const struct GNUNET_PeerIdentity *id, + struct GNUNET_TIME_Relative latency) { - struct GNUNET_TIME_Relative latency; + struct GSF_ConnectedPeer *cp; - latency = get_latency (atsi); + cp = GSF_peer_get_ (id); + if (NULL == cp) + return; /* we're not yet connected at the core level, ignore */ GNUNET_LOAD_value_set_decline (cp->ppd.transmission_delay, - latency); - /* LATER: merge atsi into cp's performance data (if we ever care...) */ + latency); } /** * Return the performance data record for the given peer - * + * * @param cp peer to query * @return performance data record for the peer */ @@ -379,38 +372,32 @@ GSF_get_peer_performance_data_ (struct GSF_ConnectedPeer *cp) /** * Core is ready to transmit to a peer, get the message. * - * @param cls the 'struct GSF_PeerTransmitHandle' of the message + * @param cls the `struct GSF_PeerTransmitHandle` of the message * @param size number of bytes core is willing to take * @param buf where to copy the message - * @return number of bytes copied to buf + * @return number of bytes copied to @a buf */ static size_t peer_transmit_ready_cb (void *cls, - size_t size, - void *buf); - - + size_t size, + void *buf); /** * Function called by core upon success or failure of our bandwidth reservation request. * - * @param cls the 'struct GSF_ConnectedPeer' of the peer for which we made the request + * @param cls the `struct GSF_ConnectedPeer` of the peer for which we made the request * @param peer identifies the peer - * @param bandwidth_out available amount of outbound bandwidth * @param amount set to the amount that was actually reserved or unreserved; * either the full requested amount or zero (no partial reservations) * @param res_delay if the reservation could not be satisfied (amount was 0), how * long should the client wait until re-trying? - * @param preference current traffic preference for the given peer */ static void -core_reserve_callback (void *cls, - const struct GNUNET_PeerIdentity *peer, - struct GNUNET_BANDWIDTH_Value32NBO bandwidth_out, - int32_t amount, - struct GNUNET_TIME_Relative res_delay, - uint64_t preference); +ats_reserve_callback (void *cls, + const struct GNUNET_PeerIdentity *peer, + int32_t amount, + struct GNUNET_TIME_Relative res_delay); /** @@ -424,99 +411,108 @@ schedule_transmission (struct GSF_PeerTransmitHandle *pth) { struct GSF_ConnectedPeer *cp; struct GNUNET_PeerIdentity target; - uint64_t ip; - if ( (NULL != pth->cth) || - (0 != pth->cth_in_progress) ) - return; /* already done */ cp = pth->cp; - GNUNET_PEER_resolve (cp->ppd.pid, - &target); + if ((NULL != cp->cth) || (0 != cp->cth_in_progress)) + return; /* already done */ + GNUNET_assert (0 != cp->ppd.pid); + GNUNET_PEER_resolve (cp->ppd.pid, &target); + + if (0 != cp->inc_preference) + { + GNUNET_ATS_performance_change_preference (GSF_ats, + &target, + GNUNET_ATS_PREFERENCE_BANDWIDTH, + (double) cp->inc_preference, + GNUNET_ATS_PREFERENCE_END); + cp->inc_preference = 0; + } + if ( (GNUNET_YES == pth->is_query) && (GNUNET_YES != pth->was_reserved) ) - { - /* query, need reservation */ - if (GNUNET_YES != cp->did_reserve) - return; /* not ready */ - cp->did_reserve = GNUNET_NO; - /* reservation already done! */ - pth->was_reserved = GNUNET_YES; - ip = cp->inc_preference; - cp->inc_preference = 0; - cp->irc = GNUNET_CORE_peer_change_preference (GSF_core, - &target, - GNUNET_TIME_UNIT_FOREVER_REL, - GNUNET_BANDWIDTH_VALUE_MAX, - DBLOCK_SIZE, - ip, - &core_reserve_callback, - cp); - } - GNUNET_assert (pth->cth == NULL); - pth->cth_in_progress++; - pth->cth = GNUNET_CORE_notify_transmit_ready (GSF_core, - GNUNET_YES, - pth->priority, - GNUNET_TIME_absolute_get_remaining (pth->timeout), - &target, - pth->size, - &peer_transmit_ready_cb, - pth); - GNUNET_assert (0 < pth->cth_in_progress--); + { + /* query, need reservation */ + if (GNUNET_YES != cp->did_reserve) + return; /* not ready */ + cp->did_reserve = GNUNET_NO; + /* reservation already done! */ + pth->was_reserved = GNUNET_YES; + cp->rc = GNUNET_ATS_reserve_bandwidth (GSF_ats, + &target, + DBLOCK_SIZE, + &ats_reserve_callback, + cp); + return; + } + GNUNET_assert (NULL == cp->cth); + cp->cth_in_progress++; + cp->cth = + GNUNET_CORE_notify_transmit_ready (GSF_core, + GNUNET_YES, + GNUNET_CORE_PRIO_BACKGROUND, + GNUNET_TIME_absolute_get_remaining (pth->timeout), + &target, + pth->size, + &peer_transmit_ready_cb, cp); + GNUNET_assert (NULL != cp->cth); + GNUNET_assert (0 < cp->cth_in_progress--); } /** * Core is ready to transmit to a peer, get the message. * - * @param cls the 'struct GSF_PeerTransmitHandle' of the message + * @param cls the `struct GSF_PeerTransmitHandle` of the message * @param size number of bytes core is willing to take * @param buf where to copy the message - * @return number of bytes copied to buf + * @return number of bytes copied to @a buf */ static size_t peer_transmit_ready_cb (void *cls, - size_t size, - void *buf) + size_t size, + void *buf) { - struct GSF_PeerTransmitHandle *pth = cls; + struct GSF_ConnectedPeer *cp = cls; + struct GSF_PeerTransmitHandle *pth = cp->pth_head; struct GSF_PeerTransmitHandle *pos; - struct GSF_ConnectedPeer *cp; size_t ret; - GNUNET_assert ( (NULL == buf) || - (pth->size <= size) ); - pth->cth = NULL; - if (pth->timeout_task != GNUNET_SCHEDULER_NO_TASK) - { - GNUNET_SCHEDULER_cancel (pth->timeout_task); - pth->timeout_task = GNUNET_SCHEDULER_NO_TASK; - } - cp = pth->cp; + cp->cth = NULL; + if (NULL == pth) + return 0; + if (pth->size > size) + { + schedule_transmission (pth); + return 0; + } + if (NULL != pth->timeout_task) + { + GNUNET_SCHEDULER_cancel (pth->timeout_task); + pth->timeout_task = NULL; + } GNUNET_CONTAINER_DLL_remove (cp->pth_head, - cp->pth_tail, - pth); + cp->pth_tail, + pth); if (GNUNET_YES == pth->is_query) - { - cp->ppd.last_request_times[(cp->last_request_times_off++) % MAX_QUEUE_PER_PEER] = GNUNET_TIME_absolute_get (); - GNUNET_assert (0 < cp->ppd.pending_queries--); - } + { + cp->ppd.last_request_times[(cp->last_request_times_off++) % + MAX_QUEUE_PER_PEER] = + GNUNET_TIME_absolute_get (); + GNUNET_assert (0 < cp->ppd.pending_queries--); + } else if (GNUNET_NO == pth->is_query) - { - GNUNET_assert (0 < cp->ppd.pending_replies--); - } + { + GNUNET_assert (0 < cp->ppd.pending_replies--); + } GNUNET_LOAD_update (cp->ppd.transmission_delay, - GNUNET_TIME_absolute_get_duration (pth->transmission_request_start_time).rel_value); - ret = pth->gmc (pth->gmc_cls, - size, buf); - GNUNET_assert (NULL == pth->cth); - for (pos = cp->pth_head; pos != NULL; pos = pos->next) - { - GNUNET_assert (pos != pth); - schedule_transmission (pos); - } - GNUNET_assert (pth->cth == NULL); - GNUNET_assert (pth->cth_in_progress == 0); + GNUNET_TIME_absolute_get_duration + (pth->transmission_request_start_time).rel_value_us); + ret = pth->gmc (pth->gmc_cls, size, buf); + if (NULL != (pos = cp->pth_head)) + { + GNUNET_assert (pos != pth); + schedule_transmission (pos); + } GNUNET_free (pth); return ret; } @@ -525,81 +521,104 @@ peer_transmit_ready_cb (void *cls, /** * (re)try to reserve bandwidth from the given peer. * - * @param cls the 'struct GSF_ConnectedPeer' to reserve from + * @param cls the `struct GSF_ConnectedPeer` to reserve from * @param tc scheduler context */ static void retry_reservation (void *cls, - const struct GNUNET_SCHEDULER_TaskContext *tc) + const struct GNUNET_SCHEDULER_TaskContext *tc) { struct GSF_ConnectedPeer *cp = cls; - uint64_t ip; struct GNUNET_PeerIdentity target; - GNUNET_PEER_resolve (cp->ppd.pid, - &target); - cp->irc_delay_task = GNUNET_SCHEDULER_NO_TASK; - ip = cp->inc_preference; - cp->inc_preference = 0; - cp->irc = GNUNET_CORE_peer_change_preference (GSF_core, - &target, - GNUNET_TIME_UNIT_FOREVER_REL, - GNUNET_BANDWIDTH_VALUE_MAX, - DBLOCK_SIZE, - ip, - &core_reserve_callback, - cp); + GNUNET_PEER_resolve (cp->ppd.pid, &target); + cp->rc_delay_task = NULL; + cp->rc = + GNUNET_ATS_reserve_bandwidth (GSF_ats, + &target, + DBLOCK_SIZE, + &ats_reserve_callback, cp); } /** * Function called by core upon success or failure of our bandwidth reservation request. * - * @param cls the 'struct GSF_ConnectedPeer' of the peer for which we made the request + * @param cls the `struct GSF_ConnectedPeer` of the peer for which we made the request * @param peer identifies the peer - * @param bandwidth_out available amount of outbound bandwidth * @param amount set to the amount that was actually reserved or unreserved; * either the full requested amount or zero (no partial reservations) * @param res_delay if the reservation could not be satisfied (amount was 0), how * long should the client wait until re-trying? - * @param preference current traffic preference for the given peer */ static void -core_reserve_callback (void *cls, - const struct GNUNET_PeerIdentity *peer, - struct GNUNET_BANDWIDTH_Value32NBO bandwidth_out, - int32_t amount, - struct GNUNET_TIME_Relative res_delay, - uint64_t preference) +ats_reserve_callback (void *cls, + const struct GNUNET_PeerIdentity *peer, + int32_t amount, + struct GNUNET_TIME_Relative res_delay) { struct GSF_ConnectedPeer *cp = cls; struct GSF_PeerTransmitHandle *pth; - cp->irc = NULL; + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, + "Reserved %d bytes / need to wait %s for reservation\n", + (int) amount, + GNUNET_STRINGS_relative_time_to_string (res_delay, GNUNET_YES)); + cp->rc = NULL; if (0 == amount) - { - cp->irc_delay_task = GNUNET_SCHEDULER_add_delayed (res_delay, - &retry_reservation, - cp); - return; - } + { + cp->rc_delay_task = + GNUNET_SCHEDULER_add_delayed (res_delay, + &retry_reservation, + cp); + return; + } cp->did_reserve = GNUNET_YES; pth = cp->pth_head; if ( (NULL != pth) && - (NULL == pth->cth) ) - { - /* reservation success, try transmission now! */ - pth->cth_in_progress++; - pth->cth = GNUNET_CORE_notify_transmit_ready (GSF_core, - GNUNET_YES, - pth->priority, - GNUNET_TIME_absolute_get_remaining (pth->timeout), - peer, - pth->size, - &peer_transmit_ready_cb, - pth); - GNUNET_assert (0 < pth->cth_in_progress--); - } + (NULL == cp->cth) && + (0 == cp->cth_in_progress) ) + { + /* reservation success, try transmission now! */ + cp->cth_in_progress++; + cp->cth = + GNUNET_CORE_notify_transmit_ready (GSF_core, + GNUNET_YES, + GNUNET_CORE_PRIO_BACKGROUND, + GNUNET_TIME_absolute_get_remaining (pth->timeout), + peer, + pth->size, + &peer_transmit_ready_cb, + cp); + GNUNET_assert (NULL != cp->cth); + GNUNET_assert (0 < cp->cth_in_progress--); + } +} + + +/** + * Function called by PEERSTORE with peer respect record + * + * @param cls handle to connected peer entry + * @param record peerstore record information + * @param emsg error message, or NULL if no errors + * @return #GNUNET_NO to stop iterating since we only expect 0 or 1 records + */ +static int +peer_respect_cb (void *cls, + const struct GNUNET_PEERSTORE_Record *record, + const char *emsg) +{ + struct GSF_ConnectedPeer *cp = cls; + + GNUNET_assert (NULL != cp->respect_iterate_req); + cp->respect_iterate_req = NULL; + if ((NULL != record) && (sizeof (cp->disk_respect) == record->value_size)) + cp->disk_respect = cp->ppd.respect = *((uint32_t *)record->value); + GSF_push_start_ (cp); + if (NULL != cp->creation_cb) + cp->creation_cb (cp->creation_cb_cls, cp); + return GNUNET_NO; } @@ -608,42 +627,40 @@ core_reserve_callback (void *cls, * records. * * @param peer identity of peer that connected - * @param atsi performance data for the connection - * @return handle to connected peer entry + * @param creation_cb callback function when the record is created. + * @param creation_cb_cls closure for @creation_cb */ -struct GSF_ConnectedPeer * +void GSF_peer_connect_handler_ (const struct GNUNET_PeerIdentity *peer, - const struct GNUNET_TRANSPORT_ATS_Information *atsi) + GSF_ConnectedPeerCreationCallback creation_cb, + void *creation_cb_cls) { struct GSF_ConnectedPeer *cp; - char *fn; - uint32_t trust; - cp = GNUNET_malloc (sizeof (struct GSF_ConnectedPeer)); + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, + "Connected to peer %s\n", + GNUNET_i2s (peer)); + cp = GNUNET_new (struct GSF_ConnectedPeer); cp->ppd.pid = GNUNET_PEER_intern (peer); cp->ppd.transmission_delay = GNUNET_LOAD_value_init (GNUNET_TIME_UNIT_ZERO); - cp->irc = GNUNET_CORE_peer_change_preference (GSF_core, - peer, - GNUNET_TIME_UNIT_FOREVER_REL, - GNUNET_BANDWIDTH_VALUE_MAX, - DBLOCK_SIZE, - 0, - &core_reserve_callback, - cp); - fn = get_trust_filename (peer); - if ((GNUNET_DISK_file_test (fn) == GNUNET_YES) && - (sizeof (trust) == GNUNET_DISK_fn_read (fn, &trust, sizeof (trust)))) - cp->disk_trust = cp->ppd.trust = ntohl (trust); - GNUNET_free (fn); - cp->request_map = GNUNET_CONTAINER_multihashmap_create (128); + cp->rc = + GNUNET_ATS_reserve_bandwidth (GSF_ats, peer, DBLOCK_SIZE, + &ats_reserve_callback, cp); + cp->request_map = GNUNET_CONTAINER_multihashmap_create (128, GNUNET_NO); GNUNET_break (GNUNET_OK == - GNUNET_CONTAINER_multihashmap_put (cp_map, - &peer->hashPubKey, - cp, - GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY)); - update_atsi (cp, atsi); - GSF_push_start_ (cp); - return cp; + GNUNET_CONTAINER_multipeermap_put (cp_map, + GSF_connected_peer_get_identity2_ (cp), + cp, + GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY)); + GNUNET_STATISTICS_set (GSF_stats, gettext_noop ("# peers connected"), + GNUNET_CONTAINER_multipeermap_size (cp_map), + GNUNET_NO); + cp->creation_cb = creation_cb; + cp->creation_cb_cls = creation_cb_cls; + cp->respect_iterate_req = + GNUNET_PEERSTORE_iterate (peerstore, "fs", peer, "respect", + GNUNET_TIME_UNIT_FOREVER_REL, &peer_respect_cb, + cp); } @@ -651,27 +668,25 @@ GSF_peer_connect_handler_ (const struct GNUNET_PeerIdentity *peer, * It may be time to re-start migrating content to this * peer. Check, and if so, restart migration. * - * @param cls the 'struct GSF_ConnectedPeer' + * @param cls the `struct GSF_ConnectedPeer` * @param tc scheduler context */ static void revive_migration (void *cls, - const struct GNUNET_SCHEDULER_TaskContext *tc) + const struct GNUNET_SCHEDULER_TaskContext *tc) { struct GSF_ConnectedPeer *cp = cls; struct GNUNET_TIME_Relative bt; - - cp->mig_revive_task = GNUNET_SCHEDULER_NO_TASK; + + cp->mig_revive_task = NULL; bt = GNUNET_TIME_absolute_get_remaining (cp->ppd.migration_blocked_until); - if (0 != bt.rel_value) - { - /* still time left... */ - cp->mig_revive_task - = GNUNET_SCHEDULER_add_delayed (bt, - &revive_migration, - cp); - return; - } + if (0 != bt.rel_value_us) + { + /* still time left... */ + cp->mig_revive_task = + GNUNET_SCHEDULER_add_delayed (bt, &revive_migration, cp); + return; + } GSF_push_start_ (cp); } @@ -685,8 +700,9 @@ revive_migration (void *cls, struct GSF_ConnectedPeer * GSF_peer_get_ (const struct GNUNET_PeerIdentity *peer) { - return GNUNET_CONTAINER_multihashmap_get (cp_map, - &peer->hashPubKey); + if (NULL == cp_map) + return NULL; + return GNUNET_CONTAINER_multipeermap_get (cp_map, peer); } @@ -697,43 +713,42 @@ GSF_peer_get_ (const struct GNUNET_PeerIdentity *peer) * @param other the other peer involved (sender or receiver, NULL * for loopback messages where we are both sender and receiver) * @param message the actual message - * @param atsi performance information - * @return GNUNET_OK to keep the connection open, - * GNUNET_SYSERR to close it (signal serious error) + * @return #GNUNET_OK to keep the connection open, + * #GNUNET_SYSERR to close it (signal serious error) */ int GSF_handle_p2p_migration_stop_ (void *cls, - const struct GNUNET_PeerIdentity *other, - const struct GNUNET_MessageHeader *message, - const struct GNUNET_TRANSPORT_ATS_Information *atsi) + const struct GNUNET_PeerIdentity *other, + const struct GNUNET_MessageHeader *message) { - struct GSF_ConnectedPeer *cp; + struct GSF_ConnectedPeer *cp; const struct MigrationStopMessage *msm; struct GNUNET_TIME_Relative bt; - msm = (const struct MigrationStopMessage*) message; - cp = GNUNET_CONTAINER_multihashmap_get (cp_map, - &other->hashPubKey); - if (cp == NULL) - { - GNUNET_break (0); - return GNUNET_OK; - } + msm = (const struct MigrationStopMessage *) message; + cp = GSF_peer_get_ (other); + if (NULL == cp) + { + GNUNET_break (0); + return GNUNET_OK; + } + GNUNET_STATISTICS_update (GSF_stats, + gettext_noop ("# migration stop messages received"), + 1, GNUNET_NO); bt = GNUNET_TIME_relative_ntoh (msm->duration); GNUNET_log (GNUNET_ERROR_TYPE_INFO, - _("Migration of content to peer `%s' blocked for %llu ms\n"), - GNUNET_i2s (other), - (unsigned long long) bt.rel_value); + _("Migration of content to peer `%s' blocked for %s\n"), + GNUNET_i2s (other), + GNUNET_STRINGS_relative_time_to_string (bt, GNUNET_YES)); cp->ppd.migration_blocked_until = GNUNET_TIME_relative_to_absolute (bt); - if (cp->mig_revive_task == GNUNET_SCHEDULER_NO_TASK) - { - GSF_push_stop_ (cp); - cp->mig_revive_task - = GNUNET_SCHEDULER_add_delayed (bt, - &revive_migration, - cp); - } - update_atsi (cp, atsi); + if ( (NULL == cp->mig_revive_task) && + (NULL == cp->respect_iterate_req) ) + { + GSF_push_stop_ (cp); + cp->mig_revive_task = + GNUNET_SCHEDULER_add_delayed (bt, + &revive_migration, cp); + } return GNUNET_OK; } @@ -741,76 +756,88 @@ GSF_handle_p2p_migration_stop_ (void *cls, /** * Copy reply and free put message. * - * @param cls the 'struct PutMessage' - * @param buf_size number of bytes available in buf + * @param cls the `struct PutMessage` + * @param buf_size number of bytes available in @a buf * @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) + * @return number of bytes copied to @a buf, can be 0 (without indicating an error) */ -static size_t +static size_t copy_reply (void *cls, - size_t buf_size, - void *buf) + size_t buf_size, + void *buf) { struct PutMessage *pm = cls; size_t size; - if (buf != NULL) - { - GNUNET_assert (buf_size >= ntohs (pm->header.size)); - size = ntohs (pm->header.size); - memcpy (buf, pm, size); - GNUNET_STATISTICS_update (GSF_stats, - gettext_noop ("# replies transmitted to other peers"), - 1, - GNUNET_NO); - } + if (NULL != buf) + { + GNUNET_assert (buf_size >= ntohs (pm->header.size)); + size = ntohs (pm->header.size); + memcpy (buf, pm, size); + GNUNET_STATISTICS_update (GSF_stats, + gettext_noop ("# replies transmitted to other peers"), + 1, + GNUNET_NO); + } else - { - size = 0; - GNUNET_STATISTICS_update (GSF_stats, - gettext_noop ("# replies dropped"), - 1, - GNUNET_NO); - } + { + size = 0; + GNUNET_STATISTICS_update (GSF_stats, + gettext_noop ("# replies dropped"), + 1, + GNUNET_NO); + } GNUNET_free (pm); return size; } +/** + * Free resources associated with the given peer request. + * + * @param peerreq request to free + * @param query associated key for the request + */ +static void +free_pending_request (struct PeerRequest *peerreq, + const struct GNUNET_HashCode *query) +{ + struct GSF_ConnectedPeer *cp = peerreq->cp; + + if (NULL != peerreq->kill_task) + { + GNUNET_SCHEDULER_cancel (peerreq->kill_task); + peerreq->kill_task = NULL; + } + GNUNET_STATISTICS_update (GSF_stats, gettext_noop ("# P2P searches active"), + -1, GNUNET_NO); + GNUNET_break (GNUNET_YES == + GNUNET_CONTAINER_multihashmap_remove (cp->request_map, + query, peerreq)); + GNUNET_free (peerreq); +} + + /** * Cancel all requests associated with the peer. * * @param cls unused * @param query hash code of the request - * @param value the 'struct GSF_PendingRequest' - * @return GNUNET_YES (continue to iterate) + * @param value the `struct GSF_PendingRequest` + * @return #GNUNET_YES (continue to iterate) */ static int cancel_pending_request (void *cls, - const GNUNET_HashCode *query, - void *value) + const struct GNUNET_HashCode *query, + void *value) { struct PeerRequest *peerreq = value; struct GSF_PendingRequest *pr = peerreq->pr; - struct GSF_ConnectedPeer *cp = peerreq->cp; struct GSF_PendingRequestData *prd; - if (peerreq->kill_task != GNUNET_SCHEDULER_NO_TASK) - { - GNUNET_SCHEDULER_cancel (peerreq->kill_task); - peerreq->kill_task = GNUNET_SCHEDULER_NO_TASK; - } - GNUNET_STATISTICS_update (GSF_stats, - gettext_noop ("# P2P searches active"), - -1, - GNUNET_NO); prd = GSF_pending_request_get_data_ (pr); - GNUNET_break (GNUNET_OK == - GNUNET_CONTAINER_multihashmap_remove (cp->request_map, - &prd->query, - peerreq)); - GSF_pending_request_cancel_ (pr); - GNUNET_free (peerreq); + GSF_pending_request_cancel_ (pr, GNUNET_NO); + free_pending_request (peerreq, &prd->query); return GNUNET_OK; } @@ -820,58 +847,58 @@ cancel_pending_request (void *cls, * * @param cls the request to free * @param tc task context - */ + */ static void peer_request_destroy (void *cls, - const struct GNUNET_SCHEDULER_TaskContext *tc) + const struct GNUNET_SCHEDULER_TaskContext *tc) { struct PeerRequest *peerreq = cls; struct GSF_PendingRequest *pr = peerreq->pr; struct GSF_PendingRequestData *prd; - peerreq->kill_task = GNUNET_SCHEDULER_NO_TASK; + peerreq->kill_task = NULL; prd = GSF_pending_request_get_data_ (pr); - cancel_pending_request (NULL, - &prd->query, - peerreq); + cancel_pending_request (NULL, &prd->query, peerreq); } /** * The artificial delay is over, transmit the message now. * - * @param cls the 'struct GSF_DelayedHandle' with the message + * @param cls the `struct GSF_DelayedHandle` with the message * @param tc scheduler context */ static void transmit_delayed_now (void *cls, - const struct GNUNET_SCHEDULER_TaskContext *tc) + const struct GNUNET_SCHEDULER_TaskContext *tc) { struct GSF_DelayedHandle *dh = cls; struct GSF_ConnectedPeer *cp = dh->cp; GNUNET_CONTAINER_DLL_remove (cp->delayed_head, - cp->delayed_tail, - dh); + cp->delayed_tail, + dh); + cp->delay_queue_size--; if (0 != (GNUNET_SCHEDULER_REASON_SHUTDOWN & tc->reason)) - { - GNUNET_free (dh->pm); - GNUNET_free (dh); - return; - } - (void) GSF_peer_transmit_ (cp, GNUNET_NO, - UINT32_MAX, - REPLY_TIMEOUT, - dh->msize, - ©_reply, - dh->pm); + { + GNUNET_free (dh->pm); + GNUNET_free (dh); + return; + } + (void) GSF_peer_transmit_ (cp, + GNUNET_NO, + UINT32_MAX, + REPLY_TIMEOUT, + dh->msize, + ©_reply, + dh->pm); GNUNET_free (dh); } /** * Get the randomized delay a response should be subjected to. - * + * * @return desired delay */ static struct GNUNET_TIME_Relative @@ -879,15 +906,17 @@ get_randomized_delay () { struct GNUNET_TIME_Relative ret; - /* FIXME: replace 5000 with something relating to current observed P2P message latency */ - ret = GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MILLISECONDS, - GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, - 5000)); + ret = + GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MILLISECONDS, + GNUNET_CRYPTO_random_u32 + (GNUNET_CRYPTO_QUALITY_WEAK, + 2 * GSF_avg_latency.rel_value_us + 1)); +#if INSANE_STATISTICS GNUNET_STATISTICS_update (GSF_stats, - gettext_noop ("# artificial delays introduced (ms)"), - ret.rel_value, - GNUNET_NO); - + gettext_noop + ("# artificial delays introduced (ms)"), + ret.rel_value_us / 1000LL, GNUNET_NO); +#endif return ret; } @@ -896,28 +925,30 @@ get_randomized_delay () * Handle a reply to a pending request. Also called if a request * expires (then with data == NULL). The handler may be called * many times (depending on the request type), but will not be - * called during or after a call to GSF_pending_request_cancel + * called during or after a call to GSF_pending_request_cancel * and will also not be called anymore after a call signalling * expiration. * - * @param cls 'struct PeerRequest' this is an answer for + * @param cls `struct PeerRequest` this is an answer for * @param eval evaluation of the result * @param pr handle to the original pending request * @param reply_anonymity_level anonymity level for the reply, UINT32_MAX for "unknown" - * @param expiration when does 'data' expire? + * @param expiration when does @a data expire? + * @param last_transmission when did we last transmit a request for this block * @param type type of the block * @param data response data, NULL on request expiration - * @param data_len number of bytes in data + * @param data_len number of bytes in @a data */ static void handle_p2p_reply (void *cls, - enum GNUNET_BLOCK_EvaluationResult eval, - struct GSF_PendingRequest *pr, - uint32_t reply_anonymity_level, - struct GNUNET_TIME_Absolute expiration, - enum GNUNET_BLOCK_Type type, - const void *data, - size_t data_len) + enum GNUNET_BLOCK_EvaluationResult eval, + struct GSF_PendingRequest *pr, + uint32_t reply_anonymity_level, + struct GNUNET_TIME_Absolute expiration, + struct GNUNET_TIME_Absolute last_transmission, + enum GNUNET_BLOCK_Type type, + const void *data, + size_t data_len) { struct PeerRequest *peerreq = cls; struct GSF_ConnectedPeer *cp = peerreq->cp; @@ -925,153 +956,146 @@ handle_p2p_reply (void *cls, struct PutMessage *pm; size_t msize; - GNUNET_assert (data_len + sizeof (struct PutMessage) < GNUNET_SERVER_MAX_MESSAGE_SIZE); + GNUNET_assert (data_len + sizeof (struct PutMessage) < + GNUNET_SERVER_MAX_MESSAGE_SIZE); GNUNET_assert (peerreq->pr == pr); prd = GSF_pending_request_get_data_ (pr); if (NULL == data) - { - GNUNET_STATISTICS_update (GSF_stats, - gettext_noop ("# P2P searches active"), - -1, - GNUNET_NO); - GNUNET_break (GNUNET_OK == - GNUNET_CONTAINER_multihashmap_remove (cp->request_map, - &prd->query, - peerreq)); - GNUNET_free (peerreq); - return; - } - GNUNET_break (type != GNUNET_BLOCK_TYPE_ANY); - if ( (prd->type != type) && - (prd->type != GNUNET_BLOCK_TYPE_ANY) ) - { - GNUNET_break (0); - return; - } -#if DEBUG_FS + { + free_pending_request (peerreq, &prd->query); + return; + } + GNUNET_break (GNUNET_BLOCK_TYPE_ANY != type); + if ((prd->type != type) && (GNUNET_BLOCK_TYPE_ANY != prd->type)) + { + GNUNET_STATISTICS_update (GSF_stats, + gettext_noop + ("# replies dropped due to type mismatch"), + 1, GNUNET_NO); + return; + } GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, - "Transmitting result for query `%s' to peer\n", - GNUNET_h2s (&prd->query)); -#endif + "Transmitting result for query `%s' to peer\n", + GNUNET_h2s (&prd->query)); GNUNET_STATISTICS_update (GSF_stats, - gettext_noop ("# replies received for other peers"), - 1, - GNUNET_NO); + gettext_noop ("# replies received for other peers"), + 1, GNUNET_NO); msize = sizeof (struct PutMessage) + data_len; if (msize >= GNUNET_SERVER_MAX_MESSAGE_SIZE) + { + GNUNET_break (0); + return; + } + if ((UINT32_MAX != reply_anonymity_level) && (reply_anonymity_level > 1)) + { + if (reply_anonymity_level - 1 > GSF_cover_content_count) { - GNUNET_break (0); + GNUNET_STATISTICS_update (GSF_stats, + gettext_noop + ("# replies dropped due to insufficient cover traffic"), + 1, GNUNET_NO); return; } - if ( (reply_anonymity_level != UINT32_MAX) && - (reply_anonymity_level > 1) ) - { - if (reply_anonymity_level - 1 > GSF_cover_content_count) - { - GNUNET_STATISTICS_update (GSF_stats, - gettext_noop ("# replies dropped due to insufficient cover traffic"), - 1, - GNUNET_NO); - return; - } - GSF_cover_content_count -= (reply_anonymity_level - 1); - } - + GSF_cover_content_count -= (reply_anonymity_level - 1); + } + pm = GNUNET_malloc (msize); pm->header.type = htons (GNUNET_MESSAGE_TYPE_FS_PUT); pm->header.size = htons (msize); pm->type = htonl (type); pm->expiration = GNUNET_TIME_absolute_hton (expiration); memcpy (&pm[1], data, data_len); - if ( (reply_anonymity_level != UINT32_MAX) && - (reply_anonymity_level != 0) && - (GSF_enable_randomized_delays == GNUNET_YES) ) - { - struct GSF_DelayedHandle *dh; - - dh = GNUNET_malloc (sizeof (struct GSF_DelayedHandle)); - dh->cp = cp; - dh->pm = pm; - dh->msize = msize; - GNUNET_CONTAINER_DLL_insert (cp->delayed_head, - cp->delayed_tail, - dh); - dh->delay_task = GNUNET_SCHEDULER_add_delayed (get_randomized_delay (), - &transmit_delayed_now, - dh); - } + if ( (UINT32_MAX != reply_anonymity_level) && + (0 != reply_anonymity_level) && + (GNUNET_YES == GSF_enable_randomized_delays) ) + { + struct GSF_DelayedHandle *dh; + + dh = GNUNET_new (struct GSF_DelayedHandle); + dh->cp = cp; + dh->pm = pm; + dh->msize = msize; + GNUNET_CONTAINER_DLL_insert (cp->delayed_head, + cp->delayed_tail, + dh); + cp->delay_queue_size++; + dh->delay_task = + GNUNET_SCHEDULER_add_delayed (get_randomized_delay (), + &transmit_delayed_now, + dh); + } else - { - (void) GSF_peer_transmit_ (cp, GNUNET_NO, - UINT32_MAX, - REPLY_TIMEOUT, - msize, - ©_reply, - pm); - } - if (eval != GNUNET_BLOCK_EVALUATION_OK_LAST) + { + (void) GSF_peer_transmit_ (cp, + GNUNET_NO, + UINT32_MAX, + REPLY_TIMEOUT, + msize, + ©_reply, + pm); + } + if (GNUNET_BLOCK_EVALUATION_OK_LAST != eval) return; - if (GNUNET_SCHEDULER_NO_TASK == peerreq->kill_task) - { - GNUNET_STATISTICS_update (GSF_stats, - gettext_noop ("# P2P searches destroyed due to ultimate reply"), - 1, - GNUNET_NO); - peerreq->kill_task = GNUNET_SCHEDULER_add_now (&peer_request_destroy, - peerreq); - } + if (NULL == peerreq->kill_task) + { + GNUNET_STATISTICS_update (GSF_stats, + gettext_noop + ("# P2P searches destroyed due to ultimate reply"), + 1, GNUNET_NO); + peerreq->kill_task = + GNUNET_SCHEDULER_add_now (&peer_request_destroy, peerreq); + } } /** - * Increase the host credit by a value. + * Increase the peer's respect by a value. * - * @param cp which peer to change the trust value on + * @param cp which peer to change the respect value on * @param value is the int value by which the - * host credit is to be increased or decreased - * @returns the actual change in trust (positive or negative) + * peer's credit is to be increased or decreased + * @returns the actual change in respect (positive or negative) */ static int -change_host_trust (struct GSF_ConnectedPeer *cp, int value) +change_peer_respect (struct GSF_ConnectedPeer *cp, int value) { - if (value == 0) + if (0 == value) return 0; - GNUNET_assert (cp != NULL); + GNUNET_assert (NULL != cp); if (value > 0) + { + if (cp->ppd.respect + value < cp->ppd.respect) { - if (cp->ppd.trust + value < cp->ppd.trust) - { - value = UINT32_MAX - cp->ppd.trust; - cp->ppd.trust = UINT32_MAX; - } - else - cp->ppd.trust += value; + value = UINT32_MAX - cp->ppd.respect; + cp->ppd.respect = UINT32_MAX; } + else + cp->ppd.respect += value; + } else + { + if (cp->ppd.respect < -value) { - if (cp->ppd.trust < -value) - { - value = -cp->ppd.trust; - cp->ppd.trust = 0; - } - else - cp->ppd.trust += value; + value = -cp->ppd.respect; + cp->ppd.respect = 0; } + else + cp->ppd.respect += value; + } return value; } /** * We've received a request with the specified priority. Bound it - * according to how much we trust the given peer. - * + * according to how much we respect the given peer. + * * @param prio_in requested priority * @param cp the peer making the request * @return effective priority */ static int32_t -bound_priority (uint32_t prio_in, - struct GSF_ConnectedPeer *cp) +bound_priority (uint32_t prio_in, struct GSF_ConnectedPeer *cp) { #define N ((double)128.0) uint32_t ret; @@ -1079,48 +1103,49 @@ bound_priority (uint32_t prio_in, int ld; ld = GSF_test_get_load_too_high_ (0); - if (ld == GNUNET_SYSERR) - { - GNUNET_STATISTICS_update (GSF_stats, - gettext_noop ("# requests done for free (low load)"), - 1, - GNUNET_NO); - return 0; /* excess resources */ - } + if (GNUNET_SYSERR == ld) + { +#if INSANE_STATISTICS + GNUNET_STATISTICS_update (GSF_stats, + gettext_noop + ("# requests done for free (low load)"), 1, + GNUNET_NO); +#endif + return 0; /* excess resources */ + } if (prio_in > INT32_MAX) prio_in = INT32_MAX; - ret = - change_host_trust (cp, - (int) prio_in); + ret = -change_peer_respect (cp, -(int) prio_in); if (ret > 0) - { - if (ret > GSF_current_priorities + N) - rret = GSF_current_priorities + N; - else - rret = ret; - GSF_current_priorities - = (GSF_current_priorities * (N-1) + rret)/N; - } - if ( (ld == GNUNET_YES) && (ret > 0) ) - { - /* try with charging */ - ld = GSF_test_get_load_too_high_ (ret); - } - if (ld == GNUNET_YES) - { - GNUNET_STATISTICS_update (GSF_stats, - gettext_noop ("# request dropped, priority insufficient"), - 1, - GNUNET_NO); - /* undo charge */ - change_host_trust (cp, (int) ret); - return -1; /* not enough resources */ - } + { + if (ret > GSF_current_priorities + N) + rret = GSF_current_priorities + N; + else + rret = ret; + GSF_current_priorities = (GSF_current_priorities * (N - 1) + rret) / N; + } + if ((GNUNET_YES == ld) && (ret > 0)) + { + /* try with charging */ + ld = GSF_test_get_load_too_high_ (ret); + } + if (GNUNET_YES == ld) + { + GNUNET_STATISTICS_update (GSF_stats, + gettext_noop + ("# request dropped, priority insufficient"), 1, + GNUNET_NO); + /* undo charge */ + change_peer_respect (cp, (int) ret); + return -1; /* not enough resources */ + } else - { - GNUNET_STATISTICS_update (GSF_stats, - gettext_noop ("# requests done for a price (normal load)"), - 1, - GNUNET_NO); - } + { + GNUNET_STATISTICS_update (GSF_stats, + gettext_noop + ("# requests done for a price (normal load)"), 1, + GNUNET_NO); + } #undef N return ret; } @@ -1142,13 +1167,13 @@ bound_ttl (int32_t ttl_in, uint32_t prio) if (ttl_in <= 0) return ttl_in; - allowed = ((unsigned long long) prio) * TTL_DECREMENT / 1000; - if (ttl_in > allowed) - { - if (allowed >= (1 << 30)) - return 1 << 30; - return allowed; - } + allowed = ((unsigned long long) prio) * TTL_DECREMENT / 1000; + if (ttl_in > allowed) + { + if (allowed >= (1 << 30)) + return 1 << 30; + return allowed; + } return ttl_in; } @@ -1166,20 +1191,19 @@ bound_ttl (int32_t ttl_in, uint32_t prio) */ struct GSF_PendingRequest * GSF_handle_p2p_query_ (const struct GNUNET_PeerIdentity *other, - const struct GNUNET_MessageHeader *message) + const struct GNUNET_MessageHeader *message) { struct PeerRequest *peerreq; struct GSF_PendingRequest *pr; struct GSF_PendingRequestData *prd; struct GSF_ConnectedPeer *cp; struct GSF_ConnectedPeer *cps; - const GNUNET_HashCode *namespace; const struct GNUNET_PeerIdentity *target; - enum GSF_PendingRequestOptions options; + enum GSF_PendingRequestOptions options; uint16_t msize; const struct GetMessage *gm; unsigned int bits; - const GNUNET_HashCode *opt; + const struct GNUNET_PeerIdentity *opt; uint32_t bm; size_t bfsize; uint32_t ttl_decrement; @@ -1188,228 +1212,199 @@ GSF_handle_p2p_query_ (const struct GNUNET_PeerIdentity *other, enum GNUNET_BLOCK_Type type; GNUNET_PEER_Id spid; - msize = ntohs(message->size); + msize = ntohs (message->size); if (msize < sizeof (struct GetMessage)) - { - GNUNET_break_op (0); - return NULL; - } + { + GNUNET_break_op (0); + return NULL; + } GNUNET_STATISTICS_update (GSF_stats, - gettext_noop ("# GET requests received (from other peers)"), - 1, - GNUNET_NO); - gm = (const struct GetMessage*) message; + gettext_noop + ("# GET requests received (from other peers)"), + 1, + GNUNET_NO); + gm = (const struct GetMessage *) message; type = ntohl (gm->type); bm = ntohl (gm->hash_bitmap); bits = 0; while (bm > 0) - { - if (1 == (bm & 1)) - bits++; - bm >>= 1; - } - if (msize < sizeof (struct GetMessage) + bits * sizeof (GNUNET_HashCode)) - { - GNUNET_break_op (0); - return NULL; - } - opt = (const GNUNET_HashCode*) &gm[1]; - bfsize = msize - sizeof (struct GetMessage) - bits * sizeof (GNUNET_HashCode); + { + if (1 == (bm & 1)) + bits++; + bm >>= 1; + } + if (msize < sizeof (struct GetMessage) + bits * sizeof (struct GNUNET_PeerIdentity)) + { + GNUNET_break_op (0); + return NULL; + } + opt = (const struct GNUNET_PeerIdentity *) &gm[1]; + bfsize = msize - sizeof (struct GetMessage) - bits * sizeof (struct GNUNET_PeerIdentity); /* bfsize must be power of 2, check! */ - if (0 != ( (bfsize - 1) & bfsize)) - { - GNUNET_break_op (0); - return NULL; - } + if (0 != ((bfsize - 1) & bfsize)) + { + GNUNET_break_op (0); + return NULL; + } GSF_cover_query_count++; bm = ntohl (gm->hash_bitmap); bits = 0; - cps = GNUNET_CONTAINER_multihashmap_get (cp_map, - &other->hashPubKey); + cps = GSF_peer_get_ (other); if (NULL == cps) - { - /* peer must have just disconnected */ - GNUNET_STATISTICS_update (GSF_stats, - gettext_noop ("# requests dropped due to initiator not being connected"), - 1, - GNUNET_NO); - return NULL; - } + { + /* peer must have just disconnected */ + GNUNET_STATISTICS_update (GSF_stats, + gettext_noop + ("# requests dropped due to initiator not being connected"), + 1, GNUNET_NO); + return NULL; + } if (0 != (bm & GET_MESSAGE_BIT_RETURN_TO)) - cp = GNUNET_CONTAINER_multihashmap_get (cp_map, - &opt[bits++]); + cp = GSF_peer_get_ (&opt[bits++]); else cp = cps; - if (cp == NULL) - { -#if DEBUG_FS - if (0 != (bm & GET_MESSAGE_BIT_RETURN_TO)) - GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, - "Failed to find RETURN-TO peer `%4s' in connection set. Dropping query.\n", - GNUNET_i2s ((const struct GNUNET_PeerIdentity*) &opt[bits-1])); - - else - GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, - "Failed to find peer `%4s' in connection set. Dropping query.\n", - GNUNET_i2s (other)); -#endif - GNUNET_STATISTICS_update (GSF_stats, - gettext_noop ("# requests dropped due to missing reverse route"), - 1, - GNUNET_NO); - return NULL; - } + if (NULL == cp) + { + if (0 != (bm & GET_MESSAGE_BIT_RETURN_TO)) + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, + "Failed to find RETURN-TO peer `%s' in connection set. Dropping query.\n", + GNUNET_i2s (&opt[bits - 1])); + + else + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, + "Failed to find peer `%s' in connection set. Dropping query.\n", + GNUNET_i2s (other)); + GNUNET_STATISTICS_update (GSF_stats, + gettext_noop + ("# requests dropped due to missing reverse route"), + 1, + GNUNET_NO); + return NULL; + } + if (cp->ppd.pending_replies + cp->delay_queue_size > MAX_QUEUE_PER_PEER) + { + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, + "Peer `%s' has too many replies queued already. Dropping query.\n", + GNUNET_i2s (other)); + GNUNET_STATISTICS_update (GSF_stats, + gettext_noop ("# requests dropped due to full reply queue"), + 1, + GNUNET_NO); + return NULL; + } /* note that we can really only check load here since otherwise - peers could find out that we are overloaded by not being - disconnected after sending us a malformed query... */ - priority = bound_priority (ntohl (gm->priority), cps); + * peers could find out that we are overloaded by not being + * disconnected after sending us a malformed query... */ + priority = bound_priority (ntohl (gm->priority), + cps); if (priority < 0) - { -#if DEBUG_FS - GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, - "Dropping query from `%s', this peer is too busy.\n", - GNUNET_i2s (other)); -#endif - return NULL; - } -#if DEBUG_FS + { + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, + "Dropping query from `%s', this peer is too busy.\n", + GNUNET_i2s (other)); + return NULL; + } GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, - "Received request for `%s' of type %u from peer `%4s' with flags %u\n", - GNUNET_h2s (&gm->query), - (unsigned int) type, - GNUNET_i2s (other), - (unsigned int) bm); -#endif - namespace = (0 != (bm & GET_MESSAGE_BIT_SKS_NAMESPACE)) ? &opt[bits++] : NULL; - if ( (type == GNUNET_BLOCK_TYPE_FS_SBLOCK) && - (namespace == NULL) ) - { - GNUNET_break_op (0); - return NULL; - } - if ( (type != GNUNET_BLOCK_TYPE_FS_SBLOCK) && - (namespace != NULL) ) - { - GNUNET_break_op (0); - return NULL; - } - target = (0 != (bm & GET_MESSAGE_BIT_TRANSMIT_TO)) ? ((const struct GNUNET_PeerIdentity*) &opt[bits++]) : NULL; - options = 0; + "Received request for `%s' of type %u from peer `%s' with flags %u\n", + GNUNET_h2s (&gm->query), + (unsigned int) type, + GNUNET_i2s (other), + (unsigned int) bm); + target = + (0 != + (bm & GET_MESSAGE_BIT_TRANSMIT_TO)) ? (&opt[bits++]) : NULL; + options = GSF_PRO_DEFAULTS; spid = 0; - if ( (GNUNET_LOAD_get_load (cp->ppd.transmission_delay) > 3 * (1 + priority)) || - (GNUNET_LOAD_get_average (cp->ppd.transmission_delay) > - GNUNET_CONSTANTS_MAX_CORK_DELAY.rel_value * 2 + GNUNET_LOAD_get_average (GSF_rt_entry_lifetime)) ) - { - /* don't have BW to send to peer, or would likely take longer than we have for it, - so at best indirect the query */ - priority = 0; - options |= GSF_PRO_FORWARD_ONLY; - spid = GNUNET_PEER_intern (other); - } + if ((GNUNET_LOAD_get_load (cp->ppd.transmission_delay) > 3 * (1 + priority)) + || (GNUNET_LOAD_get_average (cp->ppd.transmission_delay) > + GNUNET_CONSTANTS_MAX_CORK_DELAY.rel_value_us * 2 + + GNUNET_LOAD_get_average (GSF_rt_entry_lifetime))) + { + /* don't have BW to send to peer, or would likely take longer than we have for it, + * so at best indirect the query */ + priority = 0; + options |= GSF_PRO_FORWARD_ONLY; + spid = GNUNET_PEER_intern (other); + GNUNET_assert (0 != spid); + } ttl = bound_ttl (ntohl (gm->ttl), priority); /* decrement ttl (always) */ - ttl_decrement = 2 * TTL_DECREMENT + - GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, - TTL_DECREMENT); - if ( (ttl < 0) && - (((int32_t)(ttl - ttl_decrement)) > 0) ) - { -#if DEBUG_FS - GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, - "Dropping query from `%s' due to TTL underflow (%d - %u).\n", - GNUNET_i2s (other), - ttl, - ttl_decrement); -#endif - GNUNET_STATISTICS_update (GSF_stats, - gettext_noop ("# requests dropped due TTL underflow"), - 1, - GNUNET_NO); - /* integer underflow => drop (should be very rare)! */ - return NULL; - } + ttl_decrement = + 2 * TTL_DECREMENT + GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, + TTL_DECREMENT); + if ((ttl < 0) && (((int32_t) (ttl - ttl_decrement)) > 0)) + { + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, + "Dropping query from `%s' due to TTL underflow (%d - %u).\n", + GNUNET_i2s (other), ttl, ttl_decrement); + GNUNET_STATISTICS_update (GSF_stats, + gettext_noop + ("# requests dropped due TTL underflow"), 1, + GNUNET_NO); + /* integer underflow => drop (should be very rare)! */ + return NULL; + } ttl -= ttl_decrement; /* test if the request already exists */ peerreq = GNUNET_CONTAINER_multihashmap_get (cp->request_map, - &gm->query); - if (peerreq != NULL) - { - pr = peerreq->pr; - prd = GSF_pending_request_get_data_ (pr); - if ( (prd->type == type) && - ( (type != GNUNET_BLOCK_TYPE_FS_SBLOCK) || - (0 == memcmp (&prd->namespace, - namespace, - sizeof (GNUNET_HashCode))) ) ) - { - if (prd->ttl.abs_value >= GNUNET_TIME_absolute_get().abs_value + ttl) - { - /* existing request has higher TTL, drop new one! */ - prd->priority += priority; -#if DEBUG_FS - GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, - "Have existing request with higher TTL, dropping new request.\n", - GNUNET_i2s (other)); -#endif - GNUNET_STATISTICS_update (GSF_stats, - gettext_noop ("# requests dropped due to higher-TTL request"), - 1, - GNUNET_NO); - return NULL; - } - /* existing request has lower TTL, drop old one! */ - GNUNET_STATISTICS_update (GSF_stats, - gettext_noop ("# P2P searches active"), - -1, - GNUNET_NO); - priority += prd->priority; - GSF_pending_request_cancel_ (pr); - GNUNET_assert (GNUNET_YES == - GNUNET_CONTAINER_multihashmap_remove (cp->request_map, - &gm->query, - peerreq)); - if (peerreq->kill_task != GNUNET_SCHEDULER_NO_TASK) - { - GNUNET_SCHEDULER_cancel (peerreq->kill_task); - peerreq->kill_task = GNUNET_SCHEDULER_NO_TASK; - } - GNUNET_free (peerreq); - } + &gm->query); + if (NULL != peerreq) + { + pr = peerreq->pr; + prd = GSF_pending_request_get_data_ (pr); + if (prd->type == type) + { + if (prd->ttl.abs_value_us >= GNUNET_TIME_absolute_get ().abs_value_us + ttl * 1000LL) + { + /* existing request has higher TTL, drop new one! */ + prd->priority += priority; + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, + "Have existing request with higher TTL, dropping new request.\n", + GNUNET_i2s (other)); + GNUNET_STATISTICS_update (GSF_stats, + gettext_noop + ("# requests dropped due to higher-TTL request"), + 1, GNUNET_NO); + return NULL; + } + /* existing request has lower TTL, drop old one! */ + priority += prd->priority; + GSF_pending_request_cancel_ (pr, GNUNET_YES); + free_pending_request (peerreq, &gm->query); } - - peerreq = GNUNET_malloc (sizeof (struct PeerRequest)); - peerreq->cp = cp; + } + + peerreq = GNUNET_new (struct PeerRequest); + peerreq->cp = cp; pr = GSF_pending_request_create_ (options, - type, - &gm->query, - namespace, - target, - (bfsize > 0) ? (const char*)&opt[bits] : NULL, - bfsize, - ntohl (gm->filter_mutator), - 1 /* anonymity */, - (uint32_t) priority, - ttl, - spid, - NULL, 0, /* replies_seen */ - &handle_p2p_reply, - peerreq); + type, + &gm->query, + target, + (bfsize > 0) + ? (const char *) &opt[bits] + : NULL, + bfsize, + ntohl (gm->filter_mutator), + 1 /* anonymity */, + (uint32_t) priority, + ttl, + spid, + GNUNET_PEER_intern (other), + NULL, 0, /* replies_seen */ + &handle_p2p_reply, peerreq); GNUNET_assert (NULL != pr); peerreq->pr = pr; GNUNET_break (GNUNET_OK == - GNUNET_CONTAINER_multihashmap_put (cp->request_map, - &gm->query, - peerreq, - GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE)); + GNUNET_CONTAINER_multihashmap_put (cp->request_map, &gm->query, + peerreq, + GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE)); GNUNET_STATISTICS_update (GSF_stats, - gettext_noop ("# P2P query messages received and processed"), - 1, - GNUNET_NO); - GNUNET_STATISTICS_update (GSF_stats, - gettext_noop ("# P2P searches active"), - 1, - GNUNET_NO); + gettext_noop + ("# P2P query messages received and processed"), 1, + GNUNET_NO); + GNUNET_STATISTICS_update (GSF_stats, gettext_noop ("# P2P searches active"), + 1, GNUNET_NO); return pr; } @@ -1418,39 +1413,35 @@ GSF_handle_p2p_query_ (const struct GNUNET_PeerIdentity *other, * Function called if there has been a timeout trying to satisfy * a transmission request. * - * @param cls the 'struct GSF_PeerTransmitHandle' of the request + * @param cls the `struct GSF_PeerTransmitHandle` of the request * @param tc scheduler context */ static void peer_transmit_timeout (void *cls, - const struct GNUNET_SCHEDULER_TaskContext *tc) + const struct GNUNET_SCHEDULER_TaskContext *tc) { struct GSF_PeerTransmitHandle *pth = cls; struct GSF_ConnectedPeer *cp; -#if DEBUG_FS GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, - "Timeout trying to transmit to other peer\n"); -#endif - pth->timeout_task = GNUNET_SCHEDULER_NO_TASK; + "Timeout trying to transmit to other peer\n"); + pth->timeout_task = NULL; cp = pth->cp; GNUNET_CONTAINER_DLL_remove (cp->pth_head, - cp->pth_tail, - pth); + cp->pth_tail, + pth); if (GNUNET_YES == pth->is_query) - GNUNET_assert (0 < cp->ppd.pending_queries--); + GNUNET_assert (0 < cp->ppd.pending_queries--); else if (GNUNET_NO == pth->is_query) GNUNET_assert (0 < cp->ppd.pending_replies--); - GNUNET_LOAD_update (cp->ppd.transmission_delay, - UINT64_MAX); - if (NULL != pth->cth) - { - GNUNET_CORE_notify_transmit_ready_cancel (pth->cth); - pth->cth = NULL; - } - pth->gmc (pth->gmc_cls, - 0, NULL); - GNUNET_assert (0 == pth->cth_in_progress); + GNUNET_LOAD_update (cp->ppd.transmission_delay, UINT64_MAX); + if (NULL != cp->cth) + { + GNUNET_CORE_notify_transmit_ready_cancel (cp->cth); + cp->cth = NULL; + } + pth->gmc (pth->gmc_cls, 0, NULL); + GNUNET_assert (0 == cp->cth_in_progress); GNUNET_free (pth); } @@ -1458,31 +1449,30 @@ peer_transmit_timeout (void *cls, /** * Transmit a message to the given peer as soon as possible. * If the peer disconnects before the transmission can happen, - * the callback is invoked with a 'NULL' buffer. + * the callback is invoked with a `NULL` @a buffer. * * @param cp target peer - * @param is_query is this a query (GNUNET_YES) or content (GNUNET_NO) or neither (GNUNET_SYSERR) + * @param is_query is this a query (#GNUNET_YES) or content (#GNUNET_NO) or neither (#GNUNET_SYSERR) * @param priority how important is this request? * @param timeout when does this request timeout (call gmc with error) * @param size number of bytes we would like to send to the peer * @param gmc function to call to get the message - * @param gmc_cls closure for gmc + * @param gmc_cls closure for @a gmc * @return handle to cancel request */ struct GSF_PeerTransmitHandle * GSF_peer_transmit_ (struct GSF_ConnectedPeer *cp, - int is_query, - uint32_t priority, - struct GNUNET_TIME_Relative timeout, - size_t size, - GSF_GetMessageCallback gmc, - void *gmc_cls) + int is_query, + uint32_t priority, + struct GNUNET_TIME_Relative timeout, + size_t size, + GSF_GetMessageCallback gmc, void *gmc_cls) { struct GSF_PeerTransmitHandle *pth; struct GSF_PeerTransmitHandle *pos; struct GSF_PeerTransmitHandle *prev; - pth = GNUNET_malloc (sizeof (struct GSF_PeerTransmitHandle)); + pth = GNUNET_new (struct GSF_PeerTransmitHandle); pth->transmission_request_start_time = GNUNET_TIME_absolute_get (); pth->timeout = GNUNET_TIME_relative_to_absolute (timeout); pth->gmc = gmc; @@ -1494,28 +1484,23 @@ GSF_peer_transmit_ (struct GSF_ConnectedPeer *cp, /* insertion sort (by priority, descending) */ prev = NULL; pos = cp->pth_head; - while ( (pos != NULL) && - (pos->priority > priority) ) - { - prev = pos; - pos = pos->next; - } - if (prev == NULL) - GNUNET_CONTAINER_DLL_insert (cp->pth_head, - cp->pth_tail, - pth); - else - GNUNET_CONTAINER_DLL_insert_after (cp->pth_head, - cp->pth_tail, - prev, - pth); + while ((NULL != pos) && (pos->priority > priority)) + { + prev = pos; + pos = pos->next; + } + GNUNET_CONTAINER_DLL_insert_after (cp->pth_head, + cp->pth_tail, + prev, + pth); if (GNUNET_YES == is_query) cp->ppd.pending_queries++; else if (GNUNET_NO == is_query) cp->ppd.pending_replies++; - pth->timeout_task = GNUNET_SCHEDULER_add_delayed (timeout, - &peer_transmit_timeout, - pth); + pth->timeout_task + = GNUNET_SCHEDULER_add_delayed (timeout, + &peer_transmit_timeout, + pth); schedule_transmission (pth); return pth; } @@ -1531,25 +1516,19 @@ GSF_peer_transmit_cancel_ (struct GSF_PeerTransmitHandle *pth) { struct GSF_ConnectedPeer *cp; - if (pth->timeout_task != GNUNET_SCHEDULER_NO_TASK) - { - GNUNET_SCHEDULER_cancel (pth->timeout_task); - pth->timeout_task = GNUNET_SCHEDULER_NO_TASK; - } - if (NULL != pth->cth) - { - GNUNET_CORE_notify_transmit_ready_cancel (pth->cth); - pth->cth = NULL; - } + if (NULL != pth->timeout_task) + { + GNUNET_SCHEDULER_cancel (pth->timeout_task); + pth->timeout_task = NULL; + } cp = pth->cp; GNUNET_CONTAINER_DLL_remove (cp->pth_head, - cp->pth_tail, - pth); + cp->pth_tail, + pth); if (GNUNET_YES == pth->is_query) - GNUNET_assert (0 < cp->ppd.pending_queries--); + GNUNET_assert (0 < cp->ppd.pending_queries--); else if (GNUNET_NO == pth->is_query) GNUNET_assert (0 < cp->ppd.pending_replies--); - GNUNET_assert (0 == pth->cth_in_progress); GNUNET_free (pth); } @@ -1563,14 +1542,18 @@ GSF_peer_transmit_cancel_ (struct GSF_PeerTransmitHandle *pth) */ void GSF_peer_update_performance_ (struct GSF_ConnectedPeer *cp, - struct GNUNET_TIME_Absolute request_time, - uint32_t request_priority) + struct GNUNET_TIME_Absolute request_time, + uint32_t request_priority) { struct GNUNET_TIME_Relative delay; - delay = GNUNET_TIME_absolute_get_duration (request_time); - cp->ppd.avg_reply_delay.rel_value = (cp->ppd.avg_reply_delay.rel_value * (RUNAVG_DELAY_N-1) + delay.rel_value) / RUNAVG_DELAY_N; - cp->ppd.avg_priority = (cp->ppd.avg_priority * (RUNAVG_DELAY_N-1) + request_priority) / RUNAVG_DELAY_N; + delay = GNUNET_TIME_absolute_get_duration (request_time); + cp->ppd.avg_reply_delay.rel_value_us = + (cp->ppd.avg_reply_delay.rel_value_us * (RUNAVG_DELAY_N - 1) + + delay.rel_value_us) / RUNAVG_DELAY_N; + cp->ppd.avg_priority = + (cp->ppd.avg_priority * (RUNAVG_DELAY_N - 1) + + request_priority) / RUNAVG_DELAY_N; } @@ -1583,9 +1566,10 @@ GSF_peer_update_performance_ (struct GSF_ConnectedPeer *cp, */ void GSF_peer_update_responder_client_ (struct GSF_ConnectedPeer *cp, - struct GSF_LocalClient *initiator_client) + struct GSF_LocalClient *initiator_client) { - cp->ppd.last_client_replies[cp->last_client_replies_woff++ % CS2P_SUCCESS_LIST_SIZE] = initiator_client; + cp->ppd.last_client_replies[cp->last_client_replies_woff++ % + CS2P_SUCCESS_LIST_SIZE] = initiator_client; } @@ -1598,39 +1582,43 @@ GSF_peer_update_responder_client_ (struct GSF_ConnectedPeer *cp, */ void GSF_peer_update_responder_peer_ (struct GSF_ConnectedPeer *cp, - const struct GSF_ConnectedPeer *initiator_peer) + const struct GSF_ConnectedPeer *initiator_peer) { - GNUNET_PEER_change_rc (cp->ppd.last_p2p_replies[cp->last_p2p_replies_woff % P2P_SUCCESS_LIST_SIZE], -1); - cp->ppd.last_p2p_replies[cp->last_p2p_replies_woff++ % P2P_SUCCESS_LIST_SIZE] = initiator_peer->ppd.pid; + unsigned int woff; + + woff = cp->last_p2p_replies_woff % P2P_SUCCESS_LIST_SIZE; + GNUNET_PEER_change_rc (cp->ppd.last_p2p_replies[woff], -1); + cp->ppd.last_p2p_replies[woff] = initiator_peer->ppd.pid; GNUNET_PEER_change_rc (initiator_peer->ppd.pid, 1); + cp->last_p2p_replies_woff = (woff + 1) % P2P_SUCCESS_LIST_SIZE; } /** - * Method called whenever a given peer has a status change. + * Write peer-respect information to a file - flush the buffer entry! * - * @param cls closure - * @param peer peer identity this notification is about - * @param bandwidth_in available amount of inbound bandwidth - * @param bandwidth_out available amount of outbound bandwidth - * @param timeout absolute time when this peer will time out - * unless we see some further activity from it - * @param atsi status information + * @param cls unused + * @param key peer identity + * @param value the `struct GSF_ConnectedPeer` to flush + * @return #GNUNET_OK to continue iteration */ -void -GSF_peer_status_handler_ (void *cls, - const struct GNUNET_PeerIdentity *peer, - struct GNUNET_BANDWIDTH_Value32NBO bandwidth_in, - struct GNUNET_BANDWIDTH_Value32NBO bandwidth_out, - struct GNUNET_TIME_Absolute timeout, - const struct GNUNET_TRANSPORT_ATS_Information *atsi) +static int +flush_respect (void *cls, + const struct GNUNET_PeerIdentity *key, + void *value) { - struct GSF_ConnectedPeer *cp; + struct GSF_ConnectedPeer *cp = value; + struct GNUNET_PeerIdentity pid; - cp = GNUNET_CONTAINER_multihashmap_get (cp_map, - &peer->hashPubKey); - GNUNET_assert (NULL != cp); - update_atsi (cp, atsi); + if (cp->ppd.respect == cp->disk_respect) + return GNUNET_OK; /* unchanged */ + GNUNET_assert (0 != cp->ppd.pid); + GNUNET_PEER_resolve (cp->ppd.pid, &pid); + GNUNET_PEERSTORE_store (peerstore, "fs", &pid, "respect", &cp->ppd.respect, + sizeof (cp->ppd.respect), + GNUNET_TIME_UNIT_FOREVER_ABS, + GNUNET_PEERSTORE_STOREOPTION_REPLACE, NULL, NULL); + return GNUNET_OK; } @@ -1643,83 +1631,100 @@ GSF_peer_status_handler_ (void *cls, */ void GSF_peer_disconnect_handler_ (void *cls, - const struct GNUNET_PeerIdentity *peer) + const struct GNUNET_PeerIdentity *peer) { struct GSF_ConnectedPeer *cp; struct GSF_PeerTransmitHandle *pth; struct GSF_DelayedHandle *dh; - cp = GNUNET_CONTAINER_multihashmap_get (cp_map, - &peer->hashPubKey); + cp = GSF_peer_get_ (peer); if (NULL == cp) - return; /* must have been disconnect from core with - 'peer' == my_id, ignore */ - GNUNET_CONTAINER_multihashmap_remove (cp_map, - &peer->hashPubKey, - cp); + return; /* must have been disconnect from core with + * 'peer' == my_id, ignore */ + flush_respect (NULL, peer, cp); + GNUNET_assert (GNUNET_YES == + GNUNET_CONTAINER_multipeermap_remove (cp_map, + peer, + cp)); + GNUNET_STATISTICS_set (GSF_stats, gettext_noop ("# peers connected"), + GNUNET_CONTAINER_multipeermap_size (cp_map), + GNUNET_NO); + if (NULL != cp->respect_iterate_req) + { + GNUNET_PEERSTORE_iterate_cancel (cp->respect_iterate_req); + cp->respect_iterate_req = NULL; + } if (NULL != cp->migration_pth) - { - GSF_peer_transmit_cancel_ (cp->migration_pth); - cp->migration_pth = NULL; - } - if (NULL != cp->irc) - { - GNUNET_CORE_peer_change_preference_cancel (cp->irc); - cp->irc = NULL; - } - if (GNUNET_SCHEDULER_NO_TASK != cp->irc_delay_task) - { - GNUNET_SCHEDULER_cancel (cp->irc_delay_task); - cp->irc_delay_task = GNUNET_SCHEDULER_NO_TASK; - } + { + GSF_peer_transmit_cancel_ (cp->migration_pth); + cp->migration_pth = NULL; + } + if (NULL != cp->rc) + { + GNUNET_ATS_reserve_bandwidth_cancel (cp->rc); + cp->rc = NULL; + } + if (NULL != cp->rc_delay_task) + { + GNUNET_SCHEDULER_cancel (cp->rc_delay_task); + cp->rc_delay_task = NULL; + } GNUNET_CONTAINER_multihashmap_iterate (cp->request_map, - &cancel_pending_request, - cp); + &cancel_pending_request, cp); GNUNET_CONTAINER_multihashmap_destroy (cp->request_map); cp->request_map = NULL; GSF_plan_notify_peer_disconnect_ (cp); GNUNET_LOAD_value_free (cp->ppd.transmission_delay); GNUNET_PEER_decrement_rcs (cp->ppd.last_p2p_replies, P2P_SUCCESS_LIST_SIZE); + memset (cp->ppd.last_p2p_replies, 0, sizeof (cp->ppd.last_p2p_replies)); + GSF_push_stop_ (cp); + if (NULL != cp->cth) + { + GNUNET_CORE_notify_transmit_ready_cancel (cp->cth); + cp->cth = NULL; + } + GNUNET_assert (0 == cp->cth_in_progress); while (NULL != (pth = cp->pth_head)) + { + if (pth->timeout_task != NULL) { - if (NULL != pth->cth) - { - GNUNET_CORE_notify_transmit_ready_cancel (pth->cth); - pth->cth = NULL; - } - if (pth->timeout_task != GNUNET_SCHEDULER_NO_TASK) - { - GNUNET_SCHEDULER_cancel (pth->timeout_task); - pth->timeout_task = GNUNET_SCHEDULER_NO_TASK; - } - GNUNET_CONTAINER_DLL_remove (cp->pth_head, - cp->pth_tail, - pth); - GNUNET_assert (0 == pth->cth_in_progress); - GNUNET_free (pth); + GNUNET_SCHEDULER_cancel (pth->timeout_task); + pth->timeout_task = NULL; } + GNUNET_CONTAINER_DLL_remove (cp->pth_head, + cp->pth_tail, + pth); + if (GNUNET_YES == pth->is_query) + GNUNET_assert (0 < cp->ppd.pending_queries--); + else if (GNUNET_NO == pth->is_query) + GNUNET_assert (0 < cp->ppd.pending_replies--); + pth->gmc (pth->gmc_cls, 0, NULL); + GNUNET_free (pth); + } while (NULL != (dh = cp->delayed_head)) - { - GNUNET_CONTAINER_DLL_remove (cp->delayed_head, - cp->delayed_tail, - dh); - GNUNET_SCHEDULER_cancel (dh->delay_task); - GNUNET_free (dh->pm); - GNUNET_free (dh); - } + { + GNUNET_CONTAINER_DLL_remove (cp->delayed_head, + cp->delayed_tail, + dh); + cp->delay_queue_size--; + GNUNET_SCHEDULER_cancel (dh->delay_task); + GNUNET_free (dh->pm); + GNUNET_free (dh); + } GNUNET_PEER_change_rc (cp->ppd.pid, -1); - GSF_push_stop_ (cp); - if (GNUNET_SCHEDULER_NO_TASK != cp->mig_revive_task) - { - GNUNET_SCHEDULER_cancel (cp->mig_revive_task); - cp->mig_revive_task = GNUNET_SCHEDULER_NO_TASK; - } + if (NULL != cp->mig_revive_task) + { + GNUNET_SCHEDULER_cancel (cp->mig_revive_task); + cp->mig_revive_task = NULL; + } + GNUNET_break (0 == cp->ppd.pending_queries); + GNUNET_break (0 == cp->ppd.pending_replies); GNUNET_free (cp); } /** - * Closure for 'call_iterator'. + * Closure for #call_iterator(). */ struct IterationContext { @@ -1729,7 +1734,7 @@ struct IterationContext GSF_ConnectedPeerIterator it; /** - * Closure for 'it'. + * Closure for @e it. */ void *it_cls; }; @@ -1738,23 +1743,22 @@ struct IterationContext /** * Function that calls the callback for each peer. * - * @param cls the 'struct IterationContext*' + * @param cls the `struct IterationContext *` * @param key identity of the peer - * @param value the 'struct GSF_ConnectedPeer*' - * @return GNUNET_YES to continue iteration + * @param value the `struct GSF_ConnectedPeer *` + * @return #GNUNET_YES to continue iteration */ static int call_iterator (void *cls, - const GNUNET_HashCode *key, - void *value) + const struct GNUNET_PeerIdentity *key, + void *value) { struct IterationContext *ic = cls; struct GSF_ConnectedPeer *cp = value; - + ic->it (ic->it_cls, - (const struct GNUNET_PeerIdentity*) key, - cp, - &cp->ppd); + key, cp, + &cp->ppd); return GNUNET_YES; } @@ -1763,49 +1767,63 @@ call_iterator (void *cls, * Iterate over all connected peers. * * @param it function to call for each peer - * @param it_cls closure for it + * @param it_cls closure for @a it */ void GSF_iterate_connected_peers_ (GSF_ConnectedPeerIterator it, - void *it_cls) + void *it_cls) { struct IterationContext ic; ic.it = it; ic.it_cls = it_cls; - GNUNET_CONTAINER_multihashmap_iterate (cp_map, - &call_iterator, - &ic); + GNUNET_CONTAINER_multipeermap_iterate (cp_map, + &call_iterator, + &ic); } /** * Obtain the identity of a connected peer. * - * @param cp peer to reserve bandwidth from + * @param cp peer to get identity of * @param id identity to set (written to) */ void GSF_connected_peer_get_identity_ (const struct GSF_ConnectedPeer *cp, - struct GNUNET_PeerIdentity *id) + struct GNUNET_PeerIdentity *id) { - GNUNET_PEER_resolve (cp->ppd.pid, - id); + GNUNET_assert (0 != cp->ppd.pid); + GNUNET_PEER_resolve (cp->ppd.pid, id); +} + + +/** + * Obtain the identity of a connected peer. + * + * @param cp peer to get identity of + * @return reference to peer identity, valid until peer disconnects (!) + */ +const struct GNUNET_PeerIdentity * +GSF_connected_peer_get_identity2_ (const struct GSF_ConnectedPeer *cp) +{ + GNUNET_assert (0 != cp->ppd.pid); + return GNUNET_PEER_resolve2 (cp->ppd.pid); } /** * Assemble a migration stop message for transmission. * - * @param cls the 'struct GSF_ConnectedPeer' to use - * @param size number of bytes we're allowed to write to buf + * @param cls the `struct GSF_ConnectedPeer` to use + * @param size number of bytes we're allowed to write to @a buf * @param buf where to copy the message - * @return number of bytes copied to buf + * @return number of bytes copied to @a buf */ static size_t create_migration_stop_message (void *cls, - size_t size, - void *buf) + size_t size, + void *buf) { struct GSF_ConnectedPeer *cp = cls; struct MigrationStopMessage msm; @@ -1817,8 +1835,13 @@ create_migration_stop_message (void *cls, msm.header.size = htons (sizeof (struct MigrationStopMessage)); msm.header.type = htons (GNUNET_MESSAGE_TYPE_FS_MIGRATION_STOP); msm.reserved = htonl (0); - msm.duration = GNUNET_TIME_relative_hton (GNUNET_TIME_absolute_get_remaining (cp->last_migration_block)); + msm.duration = + GNUNET_TIME_relative_hton (GNUNET_TIME_absolute_get_remaining + (cp->last_migration_block)); memcpy (buf, &msm, sizeof (struct MigrationStopMessage)); + GNUNET_STATISTICS_update (GSF_stats, + gettext_noop ("# migration stop messages sent"), + 1, GNUNET_NO); return sizeof (struct MigrationStopMessage); } @@ -1826,82 +1849,33 @@ create_migration_stop_message (void *cls, /** * Ask a peer to stop migrating data to us until the given point * in time. - * + * * @param cp peer to ask * @param block_time until when to block */ void GSF_block_peer_migration_ (struct GSF_ConnectedPeer *cp, - struct GNUNET_TIME_Relative block_time) + struct GNUNET_TIME_Absolute block_time) { - if (GNUNET_TIME_absolute_get_remaining (cp->last_migration_block).rel_value > block_time.rel_value) - { -#if DEBUG_FS && 0 - GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, - "Migration already blocked for another %llu ms\n", - (unsigned long long) GNUNET_TIME_absolute_get_remaining (cp->last_migration_block).rel_value); -#endif - return; /* already blocked */ - } -#if DEBUG_FS && 0 - GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, - "Asking to stop migration for %llu ms\n", - (unsigned long long) block_time.rel_value); -#endif - cp->last_migration_block = GNUNET_TIME_relative_to_absolute (block_time); - if (cp->migration_pth != NULL) + if (cp->last_migration_block.abs_value_us > block_time.abs_value_us) + { + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, + "Migration already blocked for another %s\n", + GNUNET_STRINGS_relative_time_to_string (GNUNET_TIME_absolute_get_remaining + (cp->last_migration_block), GNUNET_YES)); + return; /* already blocked */ + } + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Asking to stop migration for %s\n", + GNUNET_STRINGS_relative_time_to_string (GNUNET_TIME_absolute_get_remaining (block_time), + GNUNET_YES)); + cp->last_migration_block = block_time; + if (NULL != cp->migration_pth) GSF_peer_transmit_cancel_ (cp->migration_pth); - cp->migration_pth - = GSF_peer_transmit_ (cp, - GNUNET_SYSERR, - UINT32_MAX, - GNUNET_TIME_UNIT_FOREVER_REL, - sizeof (struct MigrationStopMessage), - &create_migration_stop_message, - cp); -} - - -/** - * Write host-trust information to a file - flush the buffer entry! - * - * @param cls closure, not used - * @param key host identity - * @param value the 'struct GSF_ConnectedPeer' to flush - * @return GNUNET_OK to continue iteration - */ -static int -flush_trust (void *cls, - const GNUNET_HashCode *key, - void *value) -{ - struct GSF_ConnectedPeer *cp = value; - char *fn; - uint32_t trust; - struct GNUNET_PeerIdentity pid; - - if (cp->ppd.trust == cp->disk_trust) - return GNUNET_OK; /* unchanged */ - GNUNET_PEER_resolve (cp->ppd.pid, - &pid); - fn = get_trust_filename (&pid); - if (cp->ppd.trust == 0) - { - if ((0 != UNLINK (fn)) && (errno != ENOENT)) - GNUNET_log_strerror_file (GNUNET_ERROR_TYPE_WARNING | - GNUNET_ERROR_TYPE_BULK, "unlink", fn); - } - else - { - trust = htonl (cp->ppd.trust); - if (sizeof(uint32_t) == GNUNET_DISK_fn_write (fn, &trust, - sizeof(uint32_t), - GNUNET_DISK_PERM_USER_READ | GNUNET_DISK_PERM_USER_WRITE - | GNUNET_DISK_PERM_GROUP_READ | GNUNET_DISK_PERM_OTHER_READ)) - cp->disk_trust = cp->ppd.trust; - } - GNUNET_free (fn); - return GNUNET_OK; + cp->migration_pth = + GSF_peer_transmit_ (cp, GNUNET_SYSERR, UINT32_MAX, + GNUNET_TIME_UNIT_FOREVER_REL, + sizeof (struct MigrationStopMessage), + &create_migration_stop_message, cp); } @@ -1916,35 +1890,33 @@ flush_trust (void *cls, */ void GSF_connected_peer_change_preference_ (struct GSF_ConnectedPeer *cp, - uint64_t pref) + uint64_t pref) { cp->inc_preference += pref; } /** - * Call this method periodically to flush trust information to disk. + * Call this method periodically to flush respect information to disk. * * @param cls closure, not used * @param tc task context, not used */ static void -cron_flush_trust (void *cls, - const struct GNUNET_SCHEDULER_TaskContext *tc) +cron_flush_respect (void *cls, + const struct GNUNET_SCHEDULER_TaskContext *tc) { - if (NULL == cp_map) return; - GNUNET_CONTAINER_multihashmap_iterate (cp_map, - &flush_trust, - NULL); + GNUNET_CONTAINER_multipeermap_iterate (cp_map, + &flush_respect, NULL); if (NULL == tc) return; if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN)) return; - GNUNET_SCHEDULER_add_delayed (TRUST_FLUSH_FREQ, - &cron_flush_trust, - NULL); + GNUNET_SCHEDULER_add_delayed_with_priority (RESPECT_FLUSH_FREQ, + GNUNET_SCHEDULER_PRIORITY_HIGH, + &cron_flush_respect, NULL); } @@ -1954,15 +1926,10 @@ cron_flush_trust (void *cls, void GSF_connected_peer_init_ () { - cp_map = GNUNET_CONTAINER_multihashmap_create (128); - GNUNET_assert (GNUNET_OK == - GNUNET_CONFIGURATION_get_value_filename (GSF_cfg, - "fs", - "TRUST", - &trustDirectory)); - GNUNET_DISK_directory_create (trustDirectory); + cp_map = GNUNET_CONTAINER_multipeermap_create (128, GNUNET_YES); + peerstore = GNUNET_PEERSTORE_connect (GSF_cfg); GNUNET_SCHEDULER_add_with_priority (GNUNET_SCHEDULER_PRIORITY_HIGH, - &cron_flush_trust, NULL); + &cron_flush_respect, NULL); } @@ -1972,15 +1939,14 @@ GSF_connected_peer_init_ () * @param cls closure, unused * @param key current key code * @param value value in the hash map (peer entry) - * @return GNUNET_YES (we should continue to iterate) + * @return #GNUNET_YES (we should continue to iterate) */ -static int +static int clean_peer (void *cls, - const GNUNET_HashCode * key, + const struct GNUNET_PeerIdentity *key, void *value) { - GSF_peer_disconnect_handler_ (NULL, - (const struct GNUNET_PeerIdentity*) key); + GSF_peer_disconnect_handler_ (NULL, key); return GNUNET_YES; } @@ -1991,35 +1957,33 @@ clean_peer (void *cls, void GSF_connected_peer_done_ () { - cron_flush_trust (NULL, NULL); - GNUNET_CONTAINER_multihashmap_iterate (cp_map, - &clean_peer, - NULL); - GNUNET_CONTAINER_multihashmap_destroy (cp_map); + cron_flush_respect (NULL, NULL); + GNUNET_CONTAINER_multipeermap_iterate (cp_map, + &clean_peer, NULL); + GNUNET_CONTAINER_multipeermap_destroy (cp_map); cp_map = NULL; - GNUNET_free (trustDirectory); - trustDirectory = NULL; + GNUNET_PEERSTORE_disconnect (peerstore, GNUNET_YES); } /** * Iterator to remove references to LC entry. * - * @param cls the 'struct GSF_LocalClient*' to look for + * @param cls the `struct GSF_LocalClient *` to look for * @param key current key code * @param value value in the hash map (peer entry) - * @return GNUNET_YES (we should continue to iterate) + * @return #GNUNET_YES (we should continue to iterate) */ -static int +static int clean_local_client (void *cls, - const GNUNET_HashCode * key, + const struct GNUNET_PeerIdentity *key, void *value) { const struct GSF_LocalClient *lc = cls; struct GSF_ConnectedPeer *cp = value; unsigned int i; - for (i=0;ippd.last_client_replies[i] == lc) cp->ppd.last_client_replies[i] = NULL; return GNUNET_YES; @@ -2036,10 +2000,9 @@ void GSF_handle_local_client_disconnect_ (const struct GSF_LocalClient *lc) { if (NULL == cp_map) - return; /* already cleaned up */ - GNUNET_CONTAINER_multihashmap_iterate (cp_map, - &clean_local_client, - (void*) lc); + return; /* already cleaned up */ + GNUNET_CONTAINER_multipeermap_iterate (cp_map, &clean_local_client, + (void *) lc); }