X-Git-Url: https://git.librecmc.org/?a=blobdiff_plain;f=src%2Ffs%2Fgnunet-service-fs_cp.c;h=7e5ffb45e02e495d74b83699bc63aa666e2202fa;hb=89c5d0047e261ed7e5fabb2e771123f1707e4670;hp=bca07db328459d798ccb46dc46ca10995ef7141f;hpb=b4a203c2fd2cdebc74c37215455ec2545c8bc99e;p=oweals%2Fgnunet.git diff --git a/src/fs/gnunet-service-fs_cp.c b/src/fs/gnunet-service-fs_cp.c index bca07db32..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,10 +14,9 @@ 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' @@ -86,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. @@ -155,7 +154,7 @@ struct GSF_DelayedHandle /** * Task for the delay. */ - GNUNET_SCHEDULER_TaskIdentifier delay_task; + struct GNUNET_SCHEDULER_Task *delay_task; /** * Size of the message. @@ -184,7 +183,7 @@ struct PeerRequest /** * Task for asynchronous stopping of this request. */ - GNUNET_SCHEDULER_TaskIdentifier kill_task; + struct GNUNET_SCHEDULER_Task *kill_task; }; @@ -209,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 @@ -248,7 +247,7 @@ struct GSF_ConnectedPeer /** * Task scheduled if we need to retry bandwidth reservation later. */ - GNUNET_SCHEDULER_TaskIdentifier rc_delay_task; + struct GNUNET_SCHEDULER_Task *rc_delay_task; /** * Active requests from this neighbour, map of query to 'struct PeerRequest'. @@ -275,31 +274,36 @@ struct GSF_ConnectedPeer */ 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_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; @@ -322,7 +326,7 @@ struct GSF_ConnectedPeer /** - * Map from peer identities to 'struct GSF_ConnectPeer' entries. + * Map from peer identities to `struct GSF_ConnectPeer` entries. */ static struct GNUNET_CONTAINER_MultiPeerMap *cp_map; @@ -347,8 +351,8 @@ GSF_update_peer_latency_ (const struct GNUNET_PeerIdentity *id, 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...) */ + GNUNET_LOAD_value_set_decline (cp->ppd.transmission_delay, + latency); } @@ -368,19 +372,21 @@ 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); +peer_transmit_ready_cb (void *cls, + 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 amount set to the amount that was actually reserved or unreserved; * either the full requested amount or zero (no partial reservations) @@ -388,8 +394,10 @@ peer_transmit_ready_cb (void *cls, size_t size, void *buf); * long should the client wait until re-trying? */ static void -ats_reserve_callback (void *cls, const struct GNUNET_PeerIdentity *peer, - int32_t amount, struct GNUNET_TIME_Relative res_delay); +ats_reserve_callback (void *cls, + const struct GNUNET_PeerIdentity *peer, + int32_t amount, + struct GNUNET_TIME_Relative res_delay); /** @@ -412,13 +420,16 @@ schedule_transmission (struct GSF_PeerTransmitHandle *pth) 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); + 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)) + if ( (GNUNET_YES == pth->is_query) && + (GNUNET_YES != pth->was_reserved) ) { /* query, need reservation */ if (GNUNET_YES != cp->did_reserve) @@ -426,18 +437,22 @@ schedule_transmission (struct GSF_PeerTransmitHandle *pth) 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); + 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_notify_transmit_ready (GSF_core, + GNUNET_YES, GNUNET_CORE_PRIO_BACKGROUND, - GNUNET_TIME_absolute_get_remaining - (pth->timeout), &target, pth->size, + 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--); @@ -447,13 +462,15 @@ schedule_transmission (struct GSF_PeerTransmitHandle *pth) /** * 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) +peer_transmit_ready_cb (void *cls, + size_t size, + void *buf) { struct GSF_ConnectedPeer *cp = cls; struct GSF_PeerTransmitHandle *pth = cp->pth_head; @@ -468,17 +485,19 @@ peer_transmit_ready_cb (void *cls, size_t size, void *buf) schedule_transmission (pth); return 0; } - if (GNUNET_SCHEDULER_NO_TASK != pth->timeout_task) + if (NULL != pth->timeout_task) { GNUNET_SCHEDULER_cancel (pth->timeout_task); - pth->timeout_task = GNUNET_SCHEDULER_NO_TASK; + pth->timeout_task = NULL; } - GNUNET_CONTAINER_DLL_remove (cp->pth_head, cp->pth_tail, pth); + GNUNET_CONTAINER_DLL_remove (cp->pth_head, + 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_TIME_absolute_get (); GNUNET_assert (0 < cp->ppd.pending_queries--); } else if (GNUNET_NO == pth->is_query) @@ -502,19 +521,22 @@ peer_transmit_ready_cb (void *cls, size_t size, void *buf) /** * (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) +retry_reservation (void *cls, + const struct GNUNET_SCHEDULER_TaskContext *tc) { struct GSF_ConnectedPeer *cp = cls; struct GNUNET_PeerIdentity target; GNUNET_PEER_resolve (cp->ppd.pid, &target); - cp->rc_delay_task = GNUNET_SCHEDULER_NO_TASK; + cp->rc_delay_task = NULL; cp->rc = - GNUNET_ATS_reserve_bandwidth (GSF_ats, &target, DBLOCK_SIZE, + GNUNET_ATS_reserve_bandwidth (GSF_ats, + &target, + DBLOCK_SIZE, &ats_reserve_callback, cp); } @@ -522,7 +544,7 @@ retry_reservation (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc) /** * 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 amount set to the amount that was actually reserved or unreserved; * either the full requested amount or zero (no partial reservations) @@ -530,8 +552,10 @@ retry_reservation (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc) * long should the client wait until re-trying? */ static void -ats_reserve_callback (void *cls, const struct GNUNET_PeerIdentity *peer, - int32_t amount, struct GNUNET_TIME_Relative res_delay) +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; @@ -544,26 +568,34 @@ ats_reserve_callback (void *cls, const struct GNUNET_PeerIdentity *peer, if (0 == amount) { cp->rc_delay_task = - GNUNET_SCHEDULER_add_delayed (res_delay, &retry_reservation, cp); + GNUNET_SCHEDULER_add_delayed (res_delay, + &retry_reservation, + cp); return; } cp->did_reserve = GNUNET_YES; pth = cp->pth_head; - if ((NULL != pth) && (NULL == cp->cth) && (0 == cp->cth_in_progress)) + if ( (NULL != pth) && + (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_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_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 * @@ -573,10 +605,13 @@ ats_reserve_callback (void *cls, const struct GNUNET_PeerIdentity *peer, * @return #GNUNET_NO to stop iterating since we only expect 0 or 1 records */ static int -peer_respect_cb (void *cls, struct GNUNET_PEERSTORE_Record *record, char *emsg) +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); @@ -586,6 +621,7 @@ peer_respect_cb (void *cls, struct GNUNET_PEERSTORE_Record *record, char *emsg) return GNUNET_NO; } + /** * A peer connected to us. Setup the connected peer * records. @@ -601,7 +637,8 @@ GSF_peer_connect_handler_ (const struct GNUNET_PeerIdentity *peer, { struct GSF_ConnectedPeer *cp; - GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Connected to peer %s\n", + 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); @@ -631,16 +668,17 @@ 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) +revive_migration (void *cls, + 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_us) { @@ -675,8 +713,8 @@ 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 - * @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, @@ -703,11 +741,13 @@ GSF_handle_p2p_migration_stop_ (void *cls, GNUNET_i2s (other), GNUNET_STRINGS_relative_time_to_string (bt, GNUNET_YES)); cp->ppd.migration_blocked_until = GNUNET_TIME_relative_to_absolute (bt); - if (GNUNET_SCHEDULER_NO_TASK == cp->mig_revive_task) + 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); + GNUNET_SCHEDULER_add_delayed (bt, + &revive_migration, cp); } return GNUNET_OK; } @@ -716,13 +756,15 @@ 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 -copy_reply (void *cls, size_t buf_size, void *buf) +copy_reply (void *cls, + size_t buf_size, + void *buf) { struct PutMessage *pm = cls; size_t size; @@ -733,14 +775,16 @@ copy_reply (void *cls, size_t buf_size, void *buf) size = ntohs (pm->header.size); memcpy (buf, pm, size); GNUNET_STATISTICS_update (GSF_stats, - gettext_noop - ("# replies transmitted to other peers"), 1, + gettext_noop ("# replies transmitted to other peers"), + 1, GNUNET_NO); } else { size = 0; - GNUNET_STATISTICS_update (GSF_stats, gettext_noop ("# replies dropped"), 1, + GNUNET_STATISTICS_update (GSF_stats, + gettext_noop ("# replies dropped"), + 1, GNUNET_NO); } GNUNET_free (pm); @@ -760,10 +804,10 @@ free_pending_request (struct PeerRequest *peerreq, { struct GSF_ConnectedPeer *cp = peerreq->cp; - if (GNUNET_SCHEDULER_NO_TASK != peerreq->kill_task) + if (NULL != peerreq->kill_task) { GNUNET_SCHEDULER_cancel (peerreq->kill_task); - peerreq->kill_task = GNUNET_SCHEDULER_NO_TASK; + peerreq->kill_task = NULL; } GNUNET_STATISTICS_update (GSF_stats, gettext_noop ("# P2P searches active"), -1, GNUNET_NO); @@ -779,11 +823,13 @@ free_pending_request (struct PeerRequest *peerreq, * * @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 struct GNUNET_HashCode * query, void *value) +cancel_pending_request (void *cls, + const struct GNUNET_HashCode *query, + void *value) { struct PeerRequest *peerreq = value; struct GSF_PendingRequest *pr = peerreq->pr; @@ -803,13 +849,14 @@ cancel_pending_request (void *cls, const struct GNUNET_HashCode * query, void *v * @param tc task context */ static void -peer_request_destroy (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc) +peer_request_destroy (void *cls, + 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); } @@ -818,24 +865,33 @@ peer_request_destroy (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc) /** * 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) +transmit_delayed_now (void *cls, + 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); + GNUNET_CONTAINER_DLL_remove (cp->delayed_head, + 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); + (void) GSF_peer_transmit_ (cp, + GNUNET_NO, + UINT32_MAX, + REPLY_TIMEOUT, + dh->msize, + ©_reply, + dh->pm); GNUNET_free (dh); } @@ -873,22 +929,25 @@ get_randomized_delay () * 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, +handle_p2p_reply (void *cls, + 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, + enum GNUNET_BLOCK_Type type, + const void *data, size_t data_len) { struct PeerRequest *peerreq = cls; @@ -946,8 +1005,9 @@ handle_p2p_reply (void *cls, enum GNUNET_BLOCK_EvaluationResult eval, pm->type = htonl (type); pm->expiration = GNUNET_TIME_absolute_hton (expiration); memcpy (&pm[1], data, data_len); - if ((UINT32_MAX != reply_anonymity_level) && (0 != reply_anonymity_level) && - (GNUNET_YES == GSF_enable_randomized_delays)) + if ( (UINT32_MAX != reply_anonymity_level) && + (0 != reply_anonymity_level) && + (GNUNET_YES == GSF_enable_randomized_delays) ) { struct GSF_DelayedHandle *dh; @@ -955,19 +1015,28 @@ handle_p2p_reply (void *cls, enum GNUNET_BLOCK_EvaluationResult eval, dh->cp = cp; dh->pm = pm; dh->msize = msize; - GNUNET_CONTAINER_DLL_insert (cp->delayed_head, cp->delayed_tail, dh); + 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); + &transmit_delayed_now, + dh); } else { - (void) GSF_peer_transmit_ (cp, GNUNET_NO, UINT32_MAX, REPLY_TIMEOUT, msize, - ©_reply, pm); + (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) + if (NULL == peerreq->kill_task) { GNUNET_STATISTICS_update (GSF_stats, gettext_noop @@ -1143,7 +1212,6 @@ GSF_handle_p2p_query_ (const struct GNUNET_PeerIdentity *other, enum GNUNET_BLOCK_Type type; GNUNET_PEER_Id spid; - GNUNET_assert (other != NULL); msize = ntohs (message->size); if (msize < sizeof (struct GetMessage)) { @@ -1152,7 +1220,8 @@ GSF_handle_p2p_query_ (const struct GNUNET_PeerIdentity *other, } GNUNET_STATISTICS_update (GSF_stats, gettext_noop - ("# GET requests received (from other peers)"), 1, + ("# GET requests received (from other peers)"), + 1, GNUNET_NO); gm = (const struct GetMessage *) message; type = ntohl (gm->type); @@ -1198,25 +1267,36 @@ GSF_handle_p2p_query_ (const struct GNUNET_PeerIdentity *other, { 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", + "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 `%4s' in connection set. Dropping query.\n", + "Failed to find peer `%s' in connection set. Dropping query.\n", GNUNET_i2s (other)); -#if INSANE_STATISTICS GNUNET_STATISTICS_update (GSF_stats, gettext_noop ("# requests dropped due to missing reverse route"), - 1, GNUNET_NO); -#endif + 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); + priority = bound_priority (ntohl (gm->priority), + cps); if (priority < 0) { GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, @@ -1225,7 +1305,7 @@ GSF_handle_p2p_query_ (const struct GNUNET_PeerIdentity *other, return NULL; } GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, - "Received request for `%s' of type %u from peer `%4s' with flags %u\n", + "Received request for `%s' of type %u from peer `%s' with flags %u\n", GNUNET_h2s (&gm->query), (unsigned int) type, GNUNET_i2s (other), @@ -1267,8 +1347,9 @@ GSF_handle_p2p_query_ (const struct GNUNET_PeerIdentity *other, ttl -= ttl_decrement; /* test if the request already exists */ - peerreq = GNUNET_CONTAINER_multihashmap_get (cp->request_map, &gm->query); - if (peerreq != NULL) + peerreq = GNUNET_CONTAINER_multihashmap_get (cp->request_map, + &gm->query); + if (NULL != peerreq) { pr = peerreq->pr; prd = GSF_pending_request_get_data_ (pr); @@ -1296,13 +1377,21 @@ GSF_handle_p2p_query_ (const struct GNUNET_PeerIdentity *other, peerreq = GNUNET_new (struct PeerRequest); peerreq->cp = cp; - pr = GSF_pending_request_create_ (options, type, &gm->query, + pr = GSF_pending_request_create_ (options, + 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 */ + (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; @@ -1324,20 +1413,23 @@ 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) +peer_transmit_timeout (void *cls, + const struct GNUNET_SCHEDULER_TaskContext *tc) { struct GSF_PeerTransmitHandle *pth = cls; struct GSF_ConnectedPeer *cp; GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Timeout trying to transmit to other peer\n"); - pth->timeout_task = GNUNET_SCHEDULER_NO_TASK; + pth->timeout_task = NULL; cp = pth->cp; - GNUNET_CONTAINER_DLL_remove (cp->pth_head, cp->pth_tail, pth); + 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) @@ -1369,9 +1461,12 @@ peer_transmit_timeout (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc) * @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) +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) { struct GSF_PeerTransmitHandle *pth; struct GSF_PeerTransmitHandle *pos; @@ -1394,13 +1489,18 @@ GSF_peer_transmit_ (struct GSF_ConnectedPeer *cp, int is_query, prev = pos; pos = pos->next; } - GNUNET_CONTAINER_DLL_insert_after (cp->pth_head, cp->pth_tail, prev, pth); + 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; } @@ -1416,13 +1516,15 @@ GSF_peer_transmit_cancel_ (struct GSF_PeerTransmitHandle *pth) { struct GSF_ConnectedPeer *cp; - if (GNUNET_SCHEDULER_NO_TASK != pth->timeout_task) + if (NULL != pth->timeout_task) { GNUNET_SCHEDULER_cancel (pth->timeout_task); - pth->timeout_task = GNUNET_SCHEDULER_NO_TASK; + pth->timeout_task = NULL; } cp = pth->cp; - GNUNET_CONTAINER_DLL_remove (cp->pth_head, cp->pth_tail, pth); + 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) @@ -1492,6 +1594,34 @@ GSF_peer_update_responder_peer_ (struct GSF_ConnectedPeer *cp, } +/** + * Write peer-respect information to a file - flush the buffer entry! + * + * @param cls unused + * @param key peer identity + * @param value the `struct GSF_ConnectedPeer` to flush + * @return #GNUNET_OK to continue iteration + */ +static int +flush_respect (void *cls, + const struct GNUNET_PeerIdentity *key, + void *value) +{ + struct GSF_ConnectedPeer *cp = value; + struct GNUNET_PeerIdentity pid; + + 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; +} + + /** * A peer disconnected from us. Tear down the connected peer * record. @@ -1500,7 +1630,8 @@ GSF_peer_update_responder_peer_ (struct GSF_ConnectedPeer *cp, * @param peer identity of peer that connected */ void -GSF_peer_disconnect_handler_ (void *cls, const struct GNUNET_PeerIdentity *peer) +GSF_peer_disconnect_handler_ (void *cls, + const struct GNUNET_PeerIdentity *peer) { struct GSF_ConnectedPeer *cp; struct GSF_PeerTransmitHandle *pth; @@ -1510,6 +1641,7 @@ GSF_peer_disconnect_handler_ (void *cls, const struct GNUNET_PeerIdentity *peer) if (NULL == 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, @@ -1532,10 +1664,10 @@ GSF_peer_disconnect_handler_ (void *cls, const struct GNUNET_PeerIdentity *peer) GNUNET_ATS_reserve_bandwidth_cancel (cp->rc); cp->rc = NULL; } - if (GNUNET_SCHEDULER_NO_TASK != cp->rc_delay_task) + if (NULL != cp->rc_delay_task) { GNUNET_SCHEDULER_cancel (cp->rc_delay_task); - cp->rc_delay_task = GNUNET_SCHEDULER_NO_TASK; + cp->rc_delay_task = NULL; } GNUNET_CONTAINER_multihashmap_iterate (cp->request_map, &cancel_pending_request, cp); @@ -1554,34 +1686,45 @@ GSF_peer_disconnect_handler_ (void *cls, const struct GNUNET_PeerIdentity *peer) GNUNET_assert (0 == cp->cth_in_progress); while (NULL != (pth = cp->pth_head)) { - if (pth->timeout_task != GNUNET_SCHEDULER_NO_TASK) + if (pth->timeout_task != NULL) { GNUNET_SCHEDULER_cancel (pth->timeout_task); - pth->timeout_task = GNUNET_SCHEDULER_NO_TASK; + pth->timeout_task = NULL; } - GNUNET_CONTAINER_DLL_remove (cp->pth_head, cp->pth_tail, pth); + 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_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); - if (GNUNET_SCHEDULER_NO_TASK != cp->mig_revive_task) + if (NULL != cp->mig_revive_task) { GNUNET_SCHEDULER_cancel (cp->mig_revive_task); - cp->mig_revive_task = GNUNET_SCHEDULER_NO_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 { @@ -1591,7 +1734,7 @@ struct IterationContext GSF_ConnectedPeerIterator it; /** - * Closure for 'it'. + * Closure for @e it. */ void *it_cls; }; @@ -1606,12 +1749,16 @@ struct IterationContext * @return #GNUNET_YES to continue iteration */ static int -call_iterator (void *cls, const struct GNUNET_PeerIdentity * key, void *value) +call_iterator (void *cls, + 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); + ic->it (ic->it_cls, + key, cp, + &cp->ppd); return GNUNET_YES; } @@ -1623,13 +1770,16 @@ call_iterator (void *cls, const struct GNUNET_PeerIdentity * key, void *value) * @param it_cls closure for @a it */ void -GSF_iterate_connected_peers_ (GSF_ConnectedPeerIterator it, void *it_cls) +GSF_iterate_connected_peers_ (GSF_ConnectedPeerIterator it, + void *it_cls) { struct IterationContext ic; ic.it = it; ic.it_cls = it_cls; - GNUNET_CONTAINER_multipeermap_iterate (cp_map, &call_iterator, &ic); + GNUNET_CONTAINER_multipeermap_iterate (cp_map, + &call_iterator, + &ic); } @@ -1665,13 +1815,15 @@ GSF_connected_peer_get_identity2_ (const struct GSF_ConnectedPeer *cp) /** * 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) +create_migration_stop_message (void *cls, + size_t size, + void *buf) { struct GSF_ConnectedPeer *cp = cls; struct MigrationStopMessage msm; @@ -1727,32 +1879,6 @@ GSF_block_peer_migration_ (struct GSF_ConnectedPeer *cp, } -/** - * Write peer-respect information to a file - flush the buffer entry! - * - * @param cls unused - * @param key peer identity - * @param value the 'struct GSF_ConnectedPeer' to flush - * @return GNUNET_OK to continue iteration - */ -static int -flush_respect (void *cls, const struct GNUNET_PeerIdentity * key, void *value) -{ - struct GSF_ConnectedPeer *cp = value; - struct GNUNET_PeerIdentity pid; - - 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; -} - - /** * Notify core about a preference we have for the given peer * (to allocate more resources towards it). The change will @@ -1777,12 +1903,13 @@ GSF_connected_peer_change_preference_ (struct GSF_ConnectedPeer *cp, * @param tc task context, not used */ static void -cron_flush_respect (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_multipeermap_iterate (cp_map, &flush_respect, NULL); + GNUNET_CONTAINER_multipeermap_iterate (cp_map, + &flush_respect, NULL); if (NULL == tc) return; if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN)) @@ -1831,7 +1958,8 @@ void GSF_connected_peer_done_ () { cron_flush_respect (NULL, NULL); - GNUNET_CONTAINER_multipeermap_iterate (cp_map, &clean_peer, NULL); + GNUNET_CONTAINER_multipeermap_iterate (cp_map, + &clean_peer, NULL); GNUNET_CONTAINER_multipeermap_destroy (cp_map); cp_map = NULL; GNUNET_PEERSTORE_disconnect (peerstore, GNUNET_YES); @@ -1841,7 +1969,7 @@ GSF_connected_peer_done_ () /** * 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)