aboutsummaryrefslogtreecommitdiff
path: root/src/fs/gnunet-service-fs_cp.c
diff options
context:
space:
mode:
Diffstat (limited to 'src/fs/gnunet-service-fs_cp.c')
-rw-r--r--src/fs/gnunet-service-fs_cp.c1795
1 files changed, 0 insertions, 1795 deletions
diff --git a/src/fs/gnunet-service-fs_cp.c b/src/fs/gnunet-service-fs_cp.c
deleted file mode 100644
index 29e4c5910..000000000
--- a/src/fs/gnunet-service-fs_cp.c
+++ /dev/null
@@ -1,1795 +0,0 @@
1/*
2 This file is part of GNUnet.
3 Copyright (C) 2011, 2016 GNUnet e.V.
4
5 GNUnet is free software: you can redistribute it and/or modify it
6 under the terms of the GNU Affero General Public License as published
7 by the Free Software Foundation, either version 3 of the License,
8 or (at your option) any later version.
9
10 GNUnet is distributed in the hope that it will be useful, but
11 WITHOUT ANY WARRANTY; without even the implied warranty of
12 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
13 Affero General Public License for more details.
14
15 You should have received a copy of the GNU Affero General Public License
16 along with this program. If not, see <http://www.gnu.org/licenses/>.
17
18 SPDX-License-Identifier: AGPL3.0-or-later
19 */
20/**
21 * @file fs/gnunet-service-fs_cp.c
22 * @brief API to handle 'connected peers'
23 * @author Christian Grothoff
24 */
25#include "platform.h"
26#include "gnunet_util_lib.h"
27#include "gnunet_load_lib.h"
28#include "gnunet-service-fs.h"
29#include "gnunet-service-fs_cp.h"
30#include "gnunet-service-fs_pe.h"
31#include "gnunet-service-fs_pr.h"
32#include "gnunet-service-fs_push.h"
33#include "gnunet_peerstore_service.h"
34
35
36/**
37 * Ratio for moving average delay calculation. The previous
38 * average goes in with a factor of (n-1) into the calculation.
39 * Must be > 0.
40 */
41#define RUNAVG_DELAY_N 16
42
43/**
44 * How often do we flush respect values to disk?
45 */
46#define RESPECT_FLUSH_FREQ GNUNET_TIME_relative_multiply ( \
47 GNUNET_TIME_UNIT_MINUTES, 5)
48
49/**
50 * After how long do we discard a reply?
51 */
52#define REPLY_TIMEOUT GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MINUTES, \
53 2)
54
55/**
56 * Collect an instance number of statistics? May cause excessive IPC.
57 */
58#define INSANE_STATISTICS GNUNET_NO
59
60
61/**
62 * Handle to cancel a transmission request.
63 */
64struct GSF_PeerTransmitHandle
65{
66 /**
67 * Kept in a doubly-linked list.
68 */
69 struct GSF_PeerTransmitHandle *next;
70
71 /**
72 * Kept in a doubly-linked list.
73 */
74 struct GSF_PeerTransmitHandle *prev;
75
76 /**
77 * Time when this transmission request was issued.
78 */
79 struct GNUNET_TIME_Absolute transmission_request_start_time;
80
81 /**
82 * Envelope with the actual message.
83 */
84 struct GNUNET_MQ_Envelope *env;
85
86 /**
87 * Peer this request targets.
88 */
89 struct GSF_ConnectedPeer *cp;
90
91 /**
92 * #GNUNET_YES if this is a query, #GNUNET_NO for content.
93 */
94 int is_query;
95
96 /**
97 * Did we get a reservation already?
98 */
99 int was_reserved;
100
101 /**
102 * Priority of this request.
103 */
104 uint32_t priority;
105};
106
107
108/**
109 * Handle for an entry in our delay list.
110 */
111struct GSF_DelayedHandle
112{
113 /**
114 * Kept in a doubly-linked list.
115 */
116 struct GSF_DelayedHandle *next;
117
118 /**
119 * Kept in a doubly-linked list.
120 */
121 struct GSF_DelayedHandle *prev;
122
123 /**
124 * Peer this transmission belongs to.
125 */
126 struct GSF_ConnectedPeer *cp;
127
128 /**
129 * Envelope of the message that was delayed.
130 */
131 struct GNUNET_MQ_Envelope *env;
132
133 /**
134 * Task for the delay.
135 */
136 struct GNUNET_SCHEDULER_Task *delay_task;
137
138 /**
139 * Size of the message.
140 */
141 size_t msize;
142};
143
144
145/**
146 * Information per peer and request.
147 */
148struct PeerRequest
149{
150 /**
151 * Handle to generic request (generic: from peer or local client).
152 */
153 struct GSF_PendingRequest *pr;
154
155 /**
156 * Which specific peer issued this request?
157 */
158 struct GSF_ConnectedPeer *cp;
159
160 /**
161 * Task for asynchronous stopping of this request.
162 */
163 struct GNUNET_SCHEDULER_Task *kill_task;
164};
165
166
167/**
168 * A connected peer.
169 */
170struct GSF_ConnectedPeer
171{
172 /**
173 * Performance data for this peer.
174 */
175 struct GSF_PeerPerformanceData ppd;
176
177 /**
178 * Time until when we blocked this peer from migrating
179 * data to us.
180 */
181 struct GNUNET_TIME_Absolute last_migration_block;
182
183 /**
184 * Task scheduled to revive migration to this peer.
185 */
186 struct GNUNET_SCHEDULER_Task *mig_revive_task;
187
188 /**
189 * Messages (replies, queries, content migration) we would like to
190 * send to this peer in the near future. Sorted by priority, head.
191 */
192 struct GSF_PeerTransmitHandle *pth_head;
193
194 /**
195 * Messages (replies, queries, content migration) we would like to
196 * send to this peer in the near future. Sorted by priority, tail.
197 */
198 struct GSF_PeerTransmitHandle *pth_tail;
199
200 /**
201 * Messages (replies, queries, content migration) we would like to
202 * send to this peer in the near future. Sorted by priority, head.
203 */
204 struct GSF_DelayedHandle *delayed_head;
205
206 /**
207 * Messages (replies, queries, content migration) we would like to
208 * send to this peer in the near future. Sorted by priority, tail.
209 */
210 struct GSF_DelayedHandle *delayed_tail;
211
212 /**
213 * Context of our GNUNET_ATS_reserve_bandwidth call (or NULL).
214 */
215 struct GNUNET_ATS_ReservationContext *rc;
216
217 /**
218 * Task scheduled if we need to retry bandwidth reservation later.
219 */
220 struct GNUNET_SCHEDULER_Task *rc_delay_task;
221
222 /**
223 * Active requests from this neighbour, map of query to `struct PeerRequest`.
224 */
225 struct GNUNET_CONTAINER_MultiHashMap *request_map;
226
227 /**
228 * Handle for an active request for transmission to this
229 * peer.
230 */
231 struct GNUNET_MQ_Handle *mq;
232
233 /**
234 * Increase in traffic preference still to be submitted
235 * to the core service for this peer.
236 */
237 uint64_t inc_preference;
238
239 /**
240 * Number of entries in @e delayed_head DLL.
241 */
242 unsigned int delay_queue_size;
243
244 /**
245 * Respect rating for this peer on disk.
246 */
247 uint32_t disk_respect;
248
249 /**
250 * Which offset in @e last_p2p_replies will be updated next?
251 * (we go round-robin).
252 */
253 unsigned int last_p2p_replies_woff;
254
255 /**
256 * Which offset in @e last_client_replies will be updated next?
257 * (we go round-robin).
258 */
259 unsigned int last_client_replies_woff;
260
261 /**
262 * Current offset into @e last_request_times ring buffer.
263 */
264 unsigned int last_request_times_off;
265
266 /**
267 * #GNUNET_YES if we did successfully reserve 32k bandwidth,
268 * #GNUNET_NO if not.
269 */
270 int did_reserve;
271
272 /**
273 * Handle to the PEERSTORE iterate request for peer respect value
274 */
275 struct GNUNET_PEERSTORE_IterateContext *respect_iterate_req;
276};
277
278
279/**
280 * Map from peer identities to `struct GSF_ConnectPeer` entries.
281 */
282static struct GNUNET_CONTAINER_MultiPeerMap *cp_map;
283
284/**
285 * Handle to peerstore service.
286 */
287static struct GNUNET_PEERSTORE_Handle *peerstore;
288
289/**
290 * Task used to flush respect values to disk.
291 */
292static struct GNUNET_SCHEDULER_Task *fr_task;
293
294
295/**
296 * Update the latency information kept for the given peer.
297 *
298 * @param id peer record to update
299 * @param latency current latency value
300 */
301void
302GSF_update_peer_latency_ (const struct GNUNET_PeerIdentity *id,
303 struct GNUNET_TIME_Relative latency)
304{
305 struct GSF_ConnectedPeer *cp;
306
307 cp = GSF_peer_get_ (id);
308 if (NULL == cp)
309 return; /* we're not yet connected at the core level, ignore */
310 GNUNET_LOAD_value_set_decline (cp->ppd.transmission_delay,
311 latency);
312}
313
314
315/**
316 * Return the performance data record for the given peer
317 *
318 * @param cp peer to query
319 * @return performance data record for the peer
320 */
321struct GSF_PeerPerformanceData *
322GSF_get_peer_performance_data_ (struct GSF_ConnectedPeer *cp)
323{
324 return &cp->ppd;
325}
326
327
328/**
329 * Core is ready to transmit to a peer, get the message.
330 *
331 * @param cp which peer to send a message to
332 */
333static void
334peer_transmit (struct GSF_ConnectedPeer *cp);
335
336
337/**
338 * Function called by core upon success or failure of our bandwidth reservation request.
339 *
340 * @param cls the `struct GSF_ConnectedPeer` of the peer for which we made the request
341 * @param peer identifies the peer
342 * @param amount set to the amount that was actually reserved or unreserved;
343 * either the full requested amount or zero (no partial reservations)
344 * @param res_delay if the reservation could not be satisfied (amount was 0), how
345 * long should the client wait until re-trying?
346 */
347static void
348ats_reserve_callback (void *cls,
349 const struct GNUNET_PeerIdentity *peer,
350 int32_t amount,
351 struct GNUNET_TIME_Relative res_delay);
352
353
354/**
355 * If ready (bandwidth reserved), try to schedule transmission via
356 * core for the given handle.
357 *
358 * @param pth transmission handle to schedule
359 */
360static void
361schedule_transmission (struct GSF_PeerTransmitHandle *pth)
362{
363 struct GSF_ConnectedPeer *cp;
364 struct GNUNET_PeerIdentity target;
365
366 cp = pth->cp;
367 GNUNET_assert (0 != cp->ppd.pid);
368 GNUNET_PEER_resolve (cp->ppd.pid, &target);
369
370 if (0 != cp->inc_preference)
371 {
372 GNUNET_ATS_performance_change_preference (GSF_ats,
373 &target,
374 GNUNET_ATS_PREFERENCE_BANDWIDTH,
375 (double) cp->inc_preference,
376 GNUNET_ATS_PREFERENCE_END);
377 cp->inc_preference = 0;
378 }
379
380 if ((GNUNET_YES == pth->is_query) &&
381 (GNUNET_YES != pth->was_reserved))
382 {
383 /* query, need reservation */
384 if (GNUNET_YES != cp->did_reserve)
385 return; /* not ready */
386 cp->did_reserve = GNUNET_NO;
387 /* reservation already done! */
388 pth->was_reserved = GNUNET_YES;
389 cp->rc = GNUNET_ATS_reserve_bandwidth (GSF_ats,
390 &target,
391 DBLOCK_SIZE,
392 &ats_reserve_callback,
393 cp);
394 return;
395 }
396 peer_transmit (cp);
397}
398
399
400/**
401 * Core is ready to transmit to a peer, get the message.
402 *
403 * @param cp which peer to send a message to
404 */
405static void
406peer_transmit (struct GSF_ConnectedPeer *cp)
407{
408 struct GSF_PeerTransmitHandle *pth = cp->pth_head;
409 struct GSF_PeerTransmitHandle *pos;
410
411 if (NULL == pth)
412 return;
413 GNUNET_CONTAINER_DLL_remove (cp->pth_head,
414 cp->pth_tail,
415 pth);
416 if (GNUNET_YES == pth->is_query)
417 {
418 cp->ppd.last_request_times[(cp->last_request_times_off++)
419 % MAX_QUEUE_PER_PEER] =
420 GNUNET_TIME_absolute_get ();
421 GNUNET_assert (0 < cp->ppd.pending_queries--);
422 }
423 else if (GNUNET_NO == pth->is_query)
424 {
425 GNUNET_assert (0 < cp->ppd.pending_replies--);
426 }
427 GNUNET_LOAD_update (cp->ppd.transmission_delay,
428 GNUNET_TIME_absolute_get_duration
429 (pth->transmission_request_start_time).rel_value_us);
430 GNUNET_MQ_send (cp->mq,
431 pth->env);
432 GNUNET_free (pth);
433 if (NULL != (pos = cp->pth_head))
434 {
435 GNUNET_assert (pos != pth);
436 schedule_transmission (pos);
437 }
438}
439
440
441/**
442 * (re)try to reserve bandwidth from the given peer.
443 *
444 * @param cls the `struct GSF_ConnectedPeer` to reserve from
445 */
446static void
447retry_reservation (void *cls)
448{
449 struct GSF_ConnectedPeer *cp = cls;
450 struct GNUNET_PeerIdentity target;
451
452 GNUNET_PEER_resolve (cp->ppd.pid, &target);
453 cp->rc_delay_task = NULL;
454 cp->rc =
455 GNUNET_ATS_reserve_bandwidth (GSF_ats,
456 &target,
457 DBLOCK_SIZE,
458 &ats_reserve_callback, cp);
459}
460
461
462/**
463 * Function called by core upon success or failure of our bandwidth reservation request.
464 *
465 * @param cls the `struct GSF_ConnectedPeer` of the peer for which we made the request
466 * @param peer identifies the peer
467 * @param amount set to the amount that was actually reserved or unreserved;
468 * either the full requested amount or zero (no partial reservations)
469 * @param res_delay if the reservation could not be satisfied (amount was 0), how
470 * long should the client wait until re-trying?
471 */
472static void
473ats_reserve_callback (void *cls,
474 const struct GNUNET_PeerIdentity *peer,
475 int32_t amount,
476 struct GNUNET_TIME_Relative res_delay)
477{
478 struct GSF_ConnectedPeer *cp = cls;
479 struct GSF_PeerTransmitHandle *pth;
480
481 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
482 "Reserved %d bytes / need to wait %s for reservation\n",
483 (int) amount,
484 GNUNET_STRINGS_relative_time_to_string (res_delay, GNUNET_YES));
485 cp->rc = NULL;
486 if (0 == amount)
487 {
488 cp->rc_delay_task =
489 GNUNET_SCHEDULER_add_delayed (res_delay,
490 &retry_reservation,
491 cp);
492 return;
493 }
494 cp->did_reserve = GNUNET_YES;
495 pth = cp->pth_head;
496 if (NULL != pth)
497 {
498 /* reservation success, try transmission now! */
499 peer_transmit (cp);
500 }
501}
502
503
504/**
505 * Function called by PEERSTORE with peer respect record
506 *
507 * @param cls handle to connected peer entry
508 * @param record peerstore record information
509 * @param emsg error message, or NULL if no errors
510 */
511static void
512peer_respect_cb (void *cls,
513 const struct GNUNET_PEERSTORE_Record *record,
514 const char *emsg)
515{
516 struct GSF_ConnectedPeer *cp = cls;
517
518 GNUNET_assert (NULL != cp->respect_iterate_req);
519 if ((NULL != record) &&
520 (sizeof(cp->disk_respect) == record->value_size))
521 {
522 cp->disk_respect = *((uint32_t *) record->value);
523 cp->ppd.respect += *((uint32_t *) record->value);
524 }
525 GSF_push_start_ (cp);
526 if (NULL != record)
527 GNUNET_PEERSTORE_iterate_cancel (cp->respect_iterate_req);
528 cp->respect_iterate_req = NULL;
529}
530
531
532/**
533 * Function called for each pending request whenever a new
534 * peer connects, giving us a chance to decide about submitting
535 * the existing request to the new peer.
536 *
537 * @param cls the `struct GSF_ConnectedPeer` of the new peer
538 * @param key query for the request
539 * @param pr handle to the pending request
540 * @return #GNUNET_YES to continue to iterate
541 */
542static int
543consider_peer_for_forwarding (void *cls,
544 const struct GNUNET_HashCode *key,
545 struct GSF_PendingRequest *pr)
546{
547 struct GSF_ConnectedPeer *cp = cls;
548 struct GNUNET_PeerIdentity pid;
549
550 if (GNUNET_YES !=
551 GSF_pending_request_test_active_ (pr))
552 return GNUNET_YES; /* request is not actually active, skip! */
553 GSF_connected_peer_get_identity_ (cp, &pid);
554 if (GNUNET_YES !=
555 GSF_pending_request_test_target_ (pr, &pid))
556 {
557 GNUNET_STATISTICS_update (GSF_stats,
558 gettext_noop ("# Loopback routes suppressed"),
559 1,
560 GNUNET_NO);
561 return GNUNET_YES;
562 }
563 GSF_plan_add_ (cp, pr);
564 return GNUNET_YES;
565}
566
567
568void *
569GSF_peer_connect_handler (void *cls,
570 const struct GNUNET_PeerIdentity *peer,
571 struct GNUNET_MQ_Handle *mq)
572{
573 struct GSF_ConnectedPeer *cp;
574
575 if (0 ==
576 GNUNET_memcmp (&GSF_my_id,
577 peer))
578 return NULL;
579 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
580 "Connected to peer %s\n",
581 GNUNET_i2s (peer));
582 cp = GNUNET_new (struct GSF_ConnectedPeer);
583 cp->ppd.pid = GNUNET_PEER_intern (peer);
584 cp->ppd.peer = peer;
585 cp->mq = mq;
586 cp->ppd.transmission_delay = GNUNET_LOAD_value_init (GNUNET_TIME_UNIT_ZERO);
587 cp->rc =
588 GNUNET_ATS_reserve_bandwidth (GSF_ats,
589 peer,
590 DBLOCK_SIZE,
591 &ats_reserve_callback, cp);
592 cp->request_map = GNUNET_CONTAINER_multihashmap_create (128,
593 GNUNET_YES);
594 GNUNET_break (GNUNET_OK ==
595 GNUNET_CONTAINER_multipeermap_put (cp_map,
596 GSF_connected_peer_get_identity2_ (
597 cp),
598 cp,
599 GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
600 GNUNET_STATISTICS_set (GSF_stats,
601 gettext_noop ("# peers connected"),
602 GNUNET_CONTAINER_multipeermap_size (cp_map),
603 GNUNET_NO);
604 cp->respect_iterate_req
605 = GNUNET_PEERSTORE_iterate (peerstore,
606 "fs",
607 peer,
608 "respect",
609 &peer_respect_cb,
610 cp);
611 GSF_iterate_pending_requests_ (&consider_peer_for_forwarding,
612 cp);
613 return cp;
614}
615
616
617/**
618 * It may be time to re-start migrating content to this
619 * peer. Check, and if so, restart migration.
620 *
621 * @param cls the `struct GSF_ConnectedPeer`
622 */
623static void
624revive_migration (void *cls)
625{
626 struct GSF_ConnectedPeer *cp = cls;
627 struct GNUNET_TIME_Relative bt;
628
629 cp->mig_revive_task = NULL;
630 bt = GNUNET_TIME_absolute_get_remaining (cp->ppd.migration_blocked_until);
631 if (0 != bt.rel_value_us)
632 {
633 /* still time left... */
634 cp->mig_revive_task =
635 GNUNET_SCHEDULER_add_delayed (bt, &revive_migration, cp);
636 return;
637 }
638 GSF_push_start_ (cp);
639}
640
641
642struct GSF_ConnectedPeer *
643GSF_peer_get_ (const struct GNUNET_PeerIdentity *peer)
644{
645 if (NULL == cp_map)
646 return NULL;
647 return GNUNET_CONTAINER_multipeermap_get (cp_map, peer);
648}
649
650
651/**
652 * Handle P2P #GNUNET_MESSAGE_TYPE_FS_MIGRATION_STOP message.
653 *
654 * @param cls closure, the `struct GSF_ConnectedPeer`
655 * @param msm the actual message
656 */
657void
658handle_p2p_migration_stop (void *cls,
659 const struct MigrationStopMessage *msm)
660{
661 struct GSF_ConnectedPeer *cp = cls;
662 struct GNUNET_TIME_Relative bt;
663
664 GNUNET_STATISTICS_update (GSF_stats,
665 gettext_noop ("# migration stop messages received"),
666 1, GNUNET_NO);
667 bt = GNUNET_TIME_relative_ntoh (msm->duration);
668 GNUNET_log (GNUNET_ERROR_TYPE_INFO,
669 _ ("Migration of content to peer `%s' blocked for %s\n"),
670 GNUNET_i2s (cp->ppd.peer),
671 GNUNET_STRINGS_relative_time_to_string (bt, GNUNET_YES));
672 cp->ppd.migration_blocked_until = GNUNET_TIME_relative_to_absolute (bt);
673 if ((NULL == cp->mig_revive_task) &&
674 (NULL == cp->respect_iterate_req))
675 {
676 GSF_push_stop_ (cp);
677 cp->mig_revive_task =
678 GNUNET_SCHEDULER_add_delayed (bt,
679 &revive_migration, cp);
680 }
681}
682
683
684/**
685 * Free resources associated with the given peer request.
686 *
687 * @param peerreq request to free
688 */
689static void
690free_pending_request (struct PeerRequest *peerreq)
691{
692 struct GSF_ConnectedPeer *cp = peerreq->cp;
693 struct GSF_PendingRequestData *prd;
694
695 prd = GSF_pending_request_get_data_ (peerreq->pr);
696 if (NULL != peerreq->kill_task)
697 {
698 GNUNET_SCHEDULER_cancel (peerreq->kill_task);
699 peerreq->kill_task = NULL;
700 }
701 GNUNET_STATISTICS_update (GSF_stats,
702 gettext_noop ("# P2P searches active"),
703 -1,
704 GNUNET_NO);
705 GNUNET_break (GNUNET_YES ==
706 GNUNET_CONTAINER_multihashmap_remove (cp->request_map,
707 &prd->query,
708 peerreq));
709 GNUNET_free (peerreq);
710}
711
712
713/**
714 * Cancel all requests associated with the peer.
715 *
716 * @param cls unused
717 * @param query hash code of the request
718 * @param value the `struct GSF_PendingRequest`
719 * @return #GNUNET_YES (continue to iterate)
720 */
721static int
722cancel_pending_request (void *cls,
723 const struct GNUNET_HashCode *query,
724 void *value)
725{
726 struct PeerRequest *peerreq = value;
727 struct GSF_PendingRequest *pr = peerreq->pr;
728
729 free_pending_request (peerreq);
730 GSF_pending_request_cancel_ (pr,
731 GNUNET_NO);
732 return GNUNET_OK;
733}
734
735
736/**
737 * Free the given request.
738 *
739 * @param cls the request to free
740 */
741static void
742peer_request_destroy (void *cls)
743{
744 struct PeerRequest *peerreq = cls;
745 struct GSF_PendingRequest *pr = peerreq->pr;
746 struct GSF_PendingRequestData *prd;
747
748 peerreq->kill_task = NULL;
749 prd = GSF_pending_request_get_data_ (pr);
750 cancel_pending_request (NULL,
751 &prd->query,
752 peerreq);
753}
754
755
756/**
757 * The artificial delay is over, transmit the message now.
758 *
759 * @param cls the `struct GSF_DelayedHandle` with the message
760 */
761static void
762transmit_delayed_now (void *cls)
763{
764 struct GSF_DelayedHandle *dh = cls;
765 struct GSF_ConnectedPeer *cp = dh->cp;
766
767 GNUNET_CONTAINER_DLL_remove (cp->delayed_head,
768 cp->delayed_tail,
769 dh);
770 cp->delay_queue_size--;
771 GSF_peer_transmit_ (cp,
772 GNUNET_NO,
773 UINT32_MAX,
774 dh->env);
775 GNUNET_free (dh);
776}
777
778
779/**
780 * Get the randomized delay a response should be subjected to.
781 *
782 * @return desired delay
783 */
784static struct GNUNET_TIME_Relative
785get_randomized_delay ()
786{
787 struct GNUNET_TIME_Relative ret;
788
789 ret =
790 GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MILLISECONDS,
791 GNUNET_CRYPTO_random_u32
792 (GNUNET_CRYPTO_QUALITY_WEAK,
793 2 * GSF_avg_latency.rel_value_us + 1));
794#if INSANE_STATISTICS
795 GNUNET_STATISTICS_update (GSF_stats,
796 gettext_noop
797 ("# artificial delays introduced (ms)"),
798 ret.rel_value_us / 1000LL, GNUNET_NO);
799#endif
800 return ret;
801}
802
803
804/**
805 * Handle a reply to a pending request. Also called if a request
806 * expires (then with data == NULL). The handler may be called
807 * many times (depending on the request type), but will not be
808 * called during or after a call to GSF_pending_request_cancel
809 * and will also not be called anymore after a call signalling
810 * expiration.
811 *
812 * @param cls `struct PeerRequest` this is an answer for
813 * @param eval evaluation of the result
814 * @param pr handle to the original pending request
815 * @param reply_anonymity_level anonymity level for the reply, UINT32_MAX for "unknown"
816 * @param expiration when does @a data expire?
817 * @param last_transmission when did we last transmit a request for this block
818 * @param type type of the block
819 * @param data response data, NULL on request expiration
820 * @param data_len number of bytes in @a data
821 */
822static void
823handle_p2p_reply (void *cls,
824 enum GNUNET_BLOCK_ReplyEvaluationResult eval,
825 struct GSF_PendingRequest *pr,
826 uint32_t reply_anonymity_level,
827 struct GNUNET_TIME_Absolute expiration,
828 struct GNUNET_TIME_Absolute last_transmission,
829 enum GNUNET_BLOCK_Type type,
830 const void *data,
831 size_t data_len)
832{
833 struct PeerRequest *peerreq = cls;
834 struct GSF_ConnectedPeer *cp = peerreq->cp;
835 struct GSF_PendingRequestData *prd;
836 struct GNUNET_MQ_Envelope *env;
837 struct PutMessage *pm;
838 size_t msize;
839
840 GNUNET_assert (data_len + sizeof(struct PutMessage) <
841 GNUNET_MAX_MESSAGE_SIZE);
842 GNUNET_assert (peerreq->pr == pr);
843 prd = GSF_pending_request_get_data_ (pr);
844 if (NULL == data)
845 {
846 free_pending_request (peerreq);
847 return;
848 }
849 GNUNET_break (GNUNET_BLOCK_TYPE_ANY != type);
850 if ( (prd->type != type) &&
851 (GNUNET_BLOCK_TYPE_ANY != prd->type) )
852 {
853 GNUNET_STATISTICS_update (GSF_stats,
854 "# replies dropped due to type mismatch",
855 1, GNUNET_NO);
856 return;
857 }
858 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
859 "Transmitting result for query `%s' to peer\n",
860 GNUNET_h2s (&prd->query));
861 GNUNET_STATISTICS_update (GSF_stats,
862 "# replies received for other peers",
863 1,
864 GNUNET_NO);
865 msize = sizeof(struct PutMessage) + data_len;
866 if (msize >= GNUNET_MAX_MESSAGE_SIZE)
867 {
868 GNUNET_break (0);
869 return;
870 }
871 if ( (UINT32_MAX != reply_anonymity_level) &&
872 (reply_anonymity_level > 1) )
873 {
874 if (reply_anonymity_level - 1 > GSF_cover_content_count)
875 {
876 GNUNET_STATISTICS_update (GSF_stats,
877 "# replies dropped due to insufficient cover traffic",
878 1, GNUNET_NO);
879 return;
880 }
881 GSF_cover_content_count -= (reply_anonymity_level - 1);
882 }
883
884 env = GNUNET_MQ_msg_extra (pm,
885 data_len,
886 GNUNET_MESSAGE_TYPE_FS_PUT);
887 pm->type = htonl (type);
888 pm->expiration = GNUNET_TIME_absolute_hton (expiration);
889 GNUNET_memcpy (&pm[1],
890 data,
891 data_len);
892 if ((UINT32_MAX != reply_anonymity_level) &&
893 (0 != reply_anonymity_level) &&
894 (GNUNET_YES == GSF_enable_randomized_delays))
895 {
896 struct GSF_DelayedHandle *dh;
897
898 dh = GNUNET_new (struct GSF_DelayedHandle);
899 dh->cp = cp;
900 dh->env = env;
901 dh->msize = msize;
902 GNUNET_CONTAINER_DLL_insert (cp->delayed_head,
903 cp->delayed_tail,
904 dh);
905 cp->delay_queue_size++;
906 dh->delay_task =
907 GNUNET_SCHEDULER_add_delayed (get_randomized_delay (),
908 &transmit_delayed_now,
909 dh);
910 }
911 else
912 {
913 GSF_peer_transmit_ (cp,
914 GNUNET_NO,
915 UINT32_MAX,
916 env);
917 }
918 if (GNUNET_BLOCK_REPLY_OK_LAST != eval)
919 return;
920 if (NULL == peerreq->kill_task)
921 {
922 GNUNET_STATISTICS_update (GSF_stats,
923 "# P2P searches destroyed due to ultimate reply",
924 1,
925 GNUNET_NO);
926 peerreq->kill_task =
927 GNUNET_SCHEDULER_add_now (&peer_request_destroy,
928 peerreq);
929 }
930}
931
932
933/**
934 * Increase the peer's respect by a value.
935 *
936 * @param cp which peer to change the respect value on
937 * @param value is the int value by which the
938 * peer's credit is to be increased or decreased
939 * @returns the actual change in respect (positive or negative)
940 */
941static int
942change_peer_respect (struct GSF_ConnectedPeer *cp, int value)
943{
944 if (0 == value)
945 return 0;
946 GNUNET_assert (NULL != cp);
947 if (value > 0)
948 {
949 if (cp->ppd.respect + value < cp->ppd.respect)
950 {
951 value = UINT32_MAX - cp->ppd.respect;
952 cp->ppd.respect = UINT32_MAX;
953 }
954 else
955 cp->ppd.respect += value;
956 }
957 else
958 {
959 if (cp->ppd.respect < -value)
960 {
961 value = -cp->ppd.respect;
962 cp->ppd.respect = 0;
963 }
964 else
965 cp->ppd.respect += value;
966 }
967 return value;
968}
969
970
971/**
972 * We've received a request with the specified priority. Bound it
973 * according to how much we respect the given peer.
974 *
975 * @param prio_in requested priority
976 * @param cp the peer making the request
977 * @return effective priority
978 */
979static int32_t
980bound_priority (uint32_t prio_in,
981 struct GSF_ConnectedPeer *cp)
982{
983#define N ((double) 128.0)
984 uint32_t ret;
985 double rret;
986 int ld;
987
988 ld = GSF_test_get_load_too_high_ (0);
989 if (GNUNET_SYSERR == ld)
990 {
991#if INSANE_STATISTICS
992 GNUNET_STATISTICS_update (GSF_stats,
993 gettext_noop
994 ("# requests done for free (low load)"), 1,
995 GNUNET_NO);
996#endif
997 return 0; /* excess resources */
998 }
999 if (prio_in > INT32_MAX)
1000 prio_in = INT32_MAX;
1001 ret = -change_peer_respect (cp, -(int) prio_in);
1002 if (ret > 0)
1003 {
1004 if (ret > GSF_current_priorities + N)
1005 rret = GSF_current_priorities + N;
1006 else
1007 rret = ret;
1008 GSF_current_priorities = (GSF_current_priorities * (N - 1) + rret) / N;
1009 }
1010 if ((GNUNET_YES == ld) && (ret > 0))
1011 {
1012 /* try with charging */
1013 ld = GSF_test_get_load_too_high_ (ret);
1014 }
1015 if (GNUNET_YES == ld)
1016 {
1017 GNUNET_STATISTICS_update (GSF_stats,
1018 gettext_noop
1019 ("# request dropped, priority insufficient"), 1,
1020 GNUNET_NO);
1021 /* undo charge */
1022 change_peer_respect (cp, (int) ret);
1023 return -1; /* not enough resources */
1024 }
1025 else
1026 {
1027 GNUNET_STATISTICS_update (GSF_stats,
1028 gettext_noop
1029 ("# requests done for a price (normal load)"),
1030 1,
1031 GNUNET_NO);
1032 }
1033#undef N
1034 return ret;
1035}
1036
1037
1038/**
1039 * The priority level imposes a bound on the maximum
1040 * value for the ttl that can be requested.
1041 *
1042 * @param ttl_in requested ttl
1043 * @param prio given priority
1044 * @return @a ttl_in if @a ttl_in is below the limit,
1045 * otherwise the ttl-limit for the given @a prio
1046 */
1047static int32_t
1048bound_ttl (int32_t ttl_in,
1049 uint32_t prio)
1050{
1051 unsigned long long allowed;
1052
1053 if (ttl_in <= 0)
1054 return ttl_in;
1055 allowed = ((unsigned long long) prio) * TTL_DECREMENT / 1000;
1056 if (ttl_in > allowed)
1057 {
1058 if (allowed >= (1 << 30))
1059 return 1 << 30;
1060 return allowed;
1061 }
1062 return ttl_in;
1063}
1064
1065
1066/**
1067 * Closure for #test_exist_cb().
1068 */
1069struct TestExistClosure
1070{
1071 /**
1072 * Priority of the incoming request.
1073 */
1074 int32_t priority;
1075
1076 /**
1077 * Relative TTL of the incoming request.
1078 */
1079 int32_t ttl;
1080
1081 /**
1082 * Type of the incoming request.
1083 */
1084 enum GNUNET_BLOCK_Type type;
1085
1086 /**
1087 * Set to #GNUNET_YES if we are done handling the query.
1088 */
1089 int finished;
1090};
1091
1092
1093/**
1094 * Test if the query already exists. If so, merge it, otherwise
1095 * keep `finished` at #GNUNET_NO.
1096 *
1097 * @param cls our `struct TestExistClosure`
1098 * @param hc the key of the query
1099 * @param value the existing `struct PeerRequest`.
1100 * @return #GNUNET_YES to continue to iterate,
1101 * #GNUNET_NO if we successfully merged
1102 */
1103static int
1104test_exist_cb (void *cls,
1105 const struct GNUNET_HashCode *hc,
1106 void *value)
1107{
1108 struct TestExistClosure *tec = cls;
1109 struct PeerRequest *peerreq = value;
1110 struct GSF_PendingRequest *pr;
1111 struct GSF_PendingRequestData *prd;
1112
1113 pr = peerreq->pr;
1114 prd = GSF_pending_request_get_data_ (pr);
1115 if (prd->type != tec->type)
1116 return GNUNET_YES;
1117 if (prd->ttl.abs_value_us >=
1118 GNUNET_TIME_absolute_get ().abs_value_us + tec->ttl * 1000LL)
1119 {
1120 /* existing request has higher TTL, drop new one! */
1121 prd->priority += tec->priority;
1122 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1123 "Have existing request with higher TTL, dropping new request.\n");
1124 GNUNET_STATISTICS_update (GSF_stats,
1125 gettext_noop
1126 ("# requests dropped due to higher-TTL request"),
1127 1, GNUNET_NO);
1128 tec->finished = GNUNET_YES;
1129 return GNUNET_NO;
1130 }
1131 /* existing request has lower TTL, drop old one! */
1132 tec->priority += prd->priority;
1133 free_pending_request (peerreq);
1134 GSF_pending_request_cancel_ (pr,
1135 GNUNET_YES);
1136 return GNUNET_NO;
1137}
1138
1139
1140/**
1141 * Handle P2P "QUERY" message. Creates the pending request entry
1142 * and sets up all of the data structures to that we will
1143 * process replies properly. Does not initiate forwarding or
1144 * local database lookups.
1145 *
1146 * @param cls the other peer involved (sender of the message)
1147 * @param gm the GET message
1148 */
1149void
1150handle_p2p_get (void *cls,
1151 const struct GetMessage *gm)
1152{
1153 struct GSF_ConnectedPeer *cps = cls;
1154 struct PeerRequest *peerreq;
1155 struct GSF_PendingRequest *pr;
1156 struct GSF_ConnectedPeer *cp;
1157 const struct GNUNET_PeerIdentity *target;
1158 enum GSF_PendingRequestOptions options;
1159 uint16_t msize;
1160 unsigned int bits;
1161 const struct GNUNET_PeerIdentity *opt;
1162 uint32_t bm;
1163 size_t bfsize;
1164 uint32_t ttl_decrement;
1165 struct TestExistClosure tec;
1166 GNUNET_PEER_Id spid;
1167 const struct GSF_PendingRequestData *prd;
1168
1169 msize = ntohs (gm->header.size);
1170 tec.type = ntohl (gm->type);
1171 bm = ntohl (gm->hash_bitmap);
1172 bits = 0;
1173 while (bm > 0)
1174 {
1175 if (1 == (bm & 1))
1176 bits++;
1177 bm >>= 1;
1178 }
1179 opt = (const struct GNUNET_PeerIdentity *) &gm[1];
1180 bfsize = msize - sizeof(struct GetMessage) - bits * sizeof(struct
1181 GNUNET_PeerIdentity);
1182 GNUNET_STATISTICS_update (GSF_stats,
1183 gettext_noop
1184 ("# GET requests received (from other peers)"),
1185 1,
1186 GNUNET_NO);
1187 GSF_cover_query_count++;
1188 bm = ntohl (gm->hash_bitmap);
1189 bits = 0;
1190 if (0 != (bm & GET_MESSAGE_BIT_RETURN_TO))
1191 cp = GSF_peer_get_ (&opt[bits++]);
1192 else
1193 cp = cps;
1194 if (NULL == cp)
1195 {
1196 if (0 != (bm & GET_MESSAGE_BIT_RETURN_TO))
1197 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1198 "Failed to find RETURN-TO peer `%s' in connection set. Dropping query.\n",
1199 GNUNET_i2s (&opt[bits - 1]));
1200
1201 else
1202 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1203 "Failed to find peer `%s' in connection set. Dropping query.\n",
1204 GNUNET_i2s (cps->ppd.peer));
1205 GNUNET_STATISTICS_update (GSF_stats,
1206 gettext_noop
1207 (
1208 "# requests dropped due to missing reverse route"),
1209 1,
1210 GNUNET_NO);
1211 return;
1212 }
1213 unsigned int queue_size = GNUNET_MQ_get_length (cp->mq);
1214 queue_size += cp->ppd.pending_replies + cp->delay_queue_size;
1215 if (queue_size > MAX_QUEUE_PER_PEER)
1216 {
1217 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1218 "Peer `%s' has too many replies queued already. Dropping query.\n",
1219 GNUNET_i2s (cps->ppd.peer));
1220 GNUNET_STATISTICS_update (GSF_stats,
1221 gettext_noop (
1222 "# requests dropped due to full reply queue"),
1223 1,
1224 GNUNET_NO);
1225 return;
1226 }
1227 /* note that we can really only check load here since otherwise
1228 * peers could find out that we are overloaded by not being
1229 * disconnected after sending us a malformed query... */
1230 tec.priority = bound_priority (ntohl (gm->priority),
1231 cps);
1232 if (tec.priority < 0)
1233 {
1234 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1235 "Dropping query from `%s', this peer is too busy.\n",
1236 GNUNET_i2s (cps->ppd.peer));
1237 return;
1238 }
1239 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1240 "Received request for `%s' of type %u from peer `%s' with flags %u\n",
1241 GNUNET_h2s (&gm->query),
1242 (unsigned int) tec.type,
1243 GNUNET_i2s (cps->ppd.peer),
1244 (unsigned int) bm);
1245 target =
1246 (0 !=
1247 (bm & GET_MESSAGE_BIT_TRANSMIT_TO)) ? (&opt[bits++]) : NULL;
1248 options = GSF_PRO_DEFAULTS;
1249 spid = 0;
1250 if ((GNUNET_LOAD_get_load (cp->ppd.transmission_delay) > 3 * (1
1251 + tec.priority))
1252 || (GNUNET_LOAD_get_average (cp->ppd.transmission_delay) >
1253 GNUNET_CONSTANTS_MAX_CORK_DELAY.rel_value_us * 2
1254 + GNUNET_LOAD_get_average (GSF_rt_entry_lifetime)))
1255 {
1256 /* don't have BW to send to peer, or would likely take longer than we have for it,
1257 * so at best indirect the query */
1258 tec.priority = 0;
1259 options |= GSF_PRO_FORWARD_ONLY;
1260 spid = GNUNET_PEER_intern (cps->ppd.peer);
1261 GNUNET_assert (0 != spid);
1262 }
1263 tec.ttl = bound_ttl (ntohl (gm->ttl),
1264 tec.priority);
1265 /* decrement ttl (always) */
1266 ttl_decrement =
1267 2 * TTL_DECREMENT + GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
1268 TTL_DECREMENT);
1269 if ((tec.ttl < 0) &&
1270 (((int32_t) (tec.ttl - ttl_decrement)) > 0))
1271 {
1272 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1273 "Dropping query from `%s' due to TTL underflow (%d - %u).\n",
1274 GNUNET_i2s (cps->ppd.peer),
1275 tec.ttl,
1276 ttl_decrement);
1277 GNUNET_STATISTICS_update (GSF_stats,
1278 gettext_noop
1279 ("# requests dropped due TTL underflow"), 1,
1280 GNUNET_NO);
1281 /* integer underflow => drop (should be very rare)! */
1282 return;
1283 }
1284 tec.ttl -= ttl_decrement;
1285
1286 /* test if the request already exists */
1287 tec.finished = GNUNET_NO;
1288 GNUNET_CONTAINER_multihashmap_get_multiple (cp->request_map,
1289 &gm->query,
1290 &test_exist_cb,
1291 &tec);
1292 if (GNUNET_YES == tec.finished)
1293 return; /* merged into existing request, we're done */
1294
1295 peerreq = GNUNET_new (struct PeerRequest);
1296 peerreq->cp = cp;
1297 pr = GSF_pending_request_create_ (options,
1298 tec.type,
1299 &gm->query,
1300 target,
1301 (bfsize > 0)
1302 ? (const char *) &opt[bits]
1303 : NULL,
1304 bfsize,
1305 1 /* anonymity */,
1306 (uint32_t) tec.priority,
1307 tec.ttl,
1308 spid,
1309 GNUNET_PEER_intern (cps->ppd.peer),
1310 NULL, 0, /* replies_seen */
1311 &handle_p2p_reply,
1312 peerreq);
1313 GNUNET_assert (NULL != pr);
1314 prd = GSF_pending_request_get_data_ (pr);
1315 peerreq->pr = pr;
1316 GNUNET_break (GNUNET_OK ==
1317 GNUNET_CONTAINER_multihashmap_put (cp->request_map,
1318 &prd->query,
1319 peerreq,
1320 GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE));
1321 GNUNET_STATISTICS_update (GSF_stats,
1322 gettext_noop (
1323 "# P2P query messages received and processed"),
1324 1,
1325 GNUNET_NO);
1326 GNUNET_STATISTICS_update (GSF_stats,
1327 gettext_noop ("# P2P searches active"),
1328 1,
1329 GNUNET_NO);
1330 GSF_pending_request_get_data_ (pr)->has_started = GNUNET_YES;
1331 GSF_local_lookup_ (pr,
1332 &GSF_consider_forwarding,
1333 NULL);
1334}
1335
1336
1337/**
1338 * Transmit a message to the given peer as soon as possible.
1339 * If the peer disconnects before the transmission can happen,
1340 * the callback is invoked with a `NULL` @a buffer.
1341 *
1342 * @param cp target peer
1343 * @param is_query is this a query (#GNUNET_YES) or content (#GNUNET_NO) or neither (#GNUNET_SYSERR)
1344 * @param priority how important is this request?
1345 * @param env message to send
1346 */
1347void
1348GSF_peer_transmit_ (struct GSF_ConnectedPeer *cp,
1349 int is_query,
1350 uint32_t priority,
1351 struct GNUNET_MQ_Envelope *env)
1352{
1353 struct GSF_PeerTransmitHandle *pth;
1354 struct GSF_PeerTransmitHandle *pos;
1355 struct GSF_PeerTransmitHandle *prev;
1356
1357 pth = GNUNET_new (struct GSF_PeerTransmitHandle);
1358 pth->transmission_request_start_time = GNUNET_TIME_absolute_get ();
1359 pth->env = env;
1360 pth->is_query = is_query;
1361 pth->priority = priority;
1362 pth->cp = cp;
1363 /* insertion sort (by priority, descending) */
1364 prev = NULL;
1365 pos = cp->pth_head;
1366 while ((NULL != pos) && (pos->priority > priority))
1367 {
1368 prev = pos;
1369 pos = pos->next;
1370 }
1371 GNUNET_CONTAINER_DLL_insert_after (cp->pth_head,
1372 cp->pth_tail,
1373 prev,
1374 pth);
1375 if (GNUNET_YES == is_query)
1376 cp->ppd.pending_queries++;
1377 else if (GNUNET_NO == is_query)
1378 cp->ppd.pending_replies++;
1379 schedule_transmission (pth);
1380}
1381
1382
1383/**
1384 * Report on receiving a reply; update the performance record of the given peer.
1385 *
1386 * @param cp responding peer (will be updated)
1387 * @param request_time time at which the original query was transmitted
1388 * @param request_priority priority of the original request
1389 */
1390void
1391GSF_peer_update_performance_ (struct GSF_ConnectedPeer *cp,
1392 struct GNUNET_TIME_Absolute request_time,
1393 uint32_t request_priority)
1394{
1395 struct GNUNET_TIME_Relative delay;
1396
1397 delay = GNUNET_TIME_absolute_get_duration (request_time);
1398 cp->ppd.avg_reply_delay.rel_value_us =
1399 (cp->ppd.avg_reply_delay.rel_value_us * (RUNAVG_DELAY_N - 1)
1400 + delay.rel_value_us) / RUNAVG_DELAY_N;
1401 cp->ppd.avg_priority =
1402 (cp->ppd.avg_priority * (RUNAVG_DELAY_N - 1)
1403 + request_priority) / RUNAVG_DELAY_N;
1404}
1405
1406
1407/**
1408 * Report on receiving a reply in response to an initiating client.
1409 * Remember that this peer is good for this client.
1410 *
1411 * @param cp responding peer (will be updated)
1412 * @param initiator_client local client on responsible for query
1413 */
1414void
1415GSF_peer_update_responder_client_ (struct GSF_ConnectedPeer *cp,
1416 struct GSF_LocalClient *initiator_client)
1417{
1418 cp->ppd.last_client_replies[cp->last_client_replies_woff++
1419 % CS2P_SUCCESS_LIST_SIZE] = initiator_client;
1420}
1421
1422
1423/**
1424 * Report on receiving a reply in response to an initiating peer.
1425 * Remember that this peer is good for this initiating peer.
1426 *
1427 * @param cp responding peer (will be updated)
1428 * @param initiator_peer other peer responsible for query
1429 */
1430void
1431GSF_peer_update_responder_peer_ (struct GSF_ConnectedPeer *cp,
1432 const struct GSF_ConnectedPeer *initiator_peer)
1433{
1434 unsigned int woff;
1435
1436 woff = cp->last_p2p_replies_woff % P2P_SUCCESS_LIST_SIZE;
1437 GNUNET_PEER_change_rc (cp->ppd.last_p2p_replies[woff], -1);
1438 cp->ppd.last_p2p_replies[woff] = initiator_peer->ppd.pid;
1439 GNUNET_PEER_change_rc (initiator_peer->ppd.pid, 1);
1440 cp->last_p2p_replies_woff = (woff + 1) % P2P_SUCCESS_LIST_SIZE;
1441}
1442
1443
1444/**
1445 * Write peer-respect information to a file - flush the buffer entry!
1446 *
1447 * @param cls unused
1448 * @param key peer identity
1449 * @param value the `struct GSF_ConnectedPeer` to flush
1450 * @return #GNUNET_OK to continue iteration
1451 */
1452static int
1453flush_respect (void *cls,
1454 const struct GNUNET_PeerIdentity *key,
1455 void *value)
1456{
1457 struct GSF_ConnectedPeer *cp = value;
1458 struct GNUNET_PeerIdentity pid;
1459
1460 if (cp->ppd.respect == cp->disk_respect)
1461 return GNUNET_OK; /* unchanged */
1462 GNUNET_assert (0 != cp->ppd.pid);
1463 GNUNET_PEER_resolve (cp->ppd.pid, &pid);
1464 GNUNET_PEERSTORE_store (peerstore, "fs", &pid, "respect", &cp->ppd.respect,
1465 sizeof(cp->ppd.respect),
1466 GNUNET_TIME_UNIT_FOREVER_ABS,
1467 GNUNET_PEERSTORE_STOREOPTION_REPLACE,
1468 NULL,
1469 NULL);
1470 return GNUNET_OK;
1471}
1472
1473
1474void
1475GSF_peer_disconnect_handler (void *cls,
1476 const struct GNUNET_PeerIdentity *peer,
1477 void *internal_cls)
1478{
1479 struct GSF_ConnectedPeer *cp = internal_cls;
1480 struct GSF_PeerTransmitHandle *pth;
1481 struct GSF_DelayedHandle *dh;
1482
1483 if (NULL == cp)
1484 return; /* must have been disconnect from core with
1485 * 'peer' == my_id, ignore */
1486 flush_respect (NULL,
1487 peer,
1488 cp);
1489 GNUNET_assert (GNUNET_YES ==
1490 GNUNET_CONTAINER_multipeermap_remove (cp_map,
1491 peer,
1492 cp));
1493 GNUNET_STATISTICS_set (GSF_stats,
1494 gettext_noop ("# peers connected"),
1495 GNUNET_CONTAINER_multipeermap_size (cp_map),
1496 GNUNET_NO);
1497 if (NULL != cp->respect_iterate_req)
1498 {
1499 GNUNET_PEERSTORE_iterate_cancel (cp->respect_iterate_req);
1500 cp->respect_iterate_req = NULL;
1501 }
1502 if (NULL != cp->rc)
1503 {
1504 GNUNET_ATS_reserve_bandwidth_cancel (cp->rc);
1505 cp->rc = NULL;
1506 }
1507 if (NULL != cp->rc_delay_task)
1508 {
1509 GNUNET_SCHEDULER_cancel (cp->rc_delay_task);
1510 cp->rc_delay_task = NULL;
1511 }
1512 GNUNET_CONTAINER_multihashmap_iterate (cp->request_map,
1513 &cancel_pending_request,
1514 cp);
1515 GNUNET_CONTAINER_multihashmap_destroy (cp->request_map);
1516 cp->request_map = NULL;
1517 GSF_plan_notify_peer_disconnect_ (cp);
1518 GNUNET_LOAD_value_free (cp->ppd.transmission_delay);
1519 GNUNET_PEER_decrement_rcs (cp->ppd.last_p2p_replies,
1520 P2P_SUCCESS_LIST_SIZE);
1521 memset (cp->ppd.last_p2p_replies,
1522 0,
1523 sizeof(cp->ppd.last_p2p_replies));
1524 GSF_push_stop_ (cp);
1525 while (NULL != (pth = cp->pth_head))
1526 {
1527 GNUNET_CONTAINER_DLL_remove (cp->pth_head,
1528 cp->pth_tail,
1529 pth);
1530 if (GNUNET_YES == pth->is_query)
1531 GNUNET_assert (0 < cp->ppd.pending_queries--);
1532 else if (GNUNET_NO == pth->is_query)
1533 GNUNET_assert (0 < cp->ppd.pending_replies--);
1534 GNUNET_free (pth);
1535 }
1536 while (NULL != (dh = cp->delayed_head))
1537 {
1538 GNUNET_CONTAINER_DLL_remove (cp->delayed_head,
1539 cp->delayed_tail,
1540 dh);
1541 GNUNET_MQ_discard (dh->env);
1542 cp->delay_queue_size--;
1543 GNUNET_SCHEDULER_cancel (dh->delay_task);
1544 GNUNET_free (dh);
1545 }
1546 GNUNET_PEER_change_rc (cp->ppd.pid, -1);
1547 if (NULL != cp->mig_revive_task)
1548 {
1549 GNUNET_SCHEDULER_cancel (cp->mig_revive_task);
1550 cp->mig_revive_task = NULL;
1551 }
1552 GNUNET_break (0 == cp->ppd.pending_queries);
1553 GNUNET_break (0 == cp->ppd.pending_replies);
1554 GNUNET_free (cp);
1555}
1556
1557
1558/**
1559 * Closure for #call_iterator().
1560 */
1561struct IterationContext
1562{
1563 /**
1564 * Function to call on each entry.
1565 */
1566 GSF_ConnectedPeerIterator it;
1567
1568 /**
1569 * Closure for @e it.
1570 */
1571 void *it_cls;
1572};
1573
1574
1575/**
1576 * Function that calls the callback for each peer.
1577 *
1578 * @param cls the `struct IterationContext *`
1579 * @param key identity of the peer
1580 * @param value the `struct GSF_ConnectedPeer *`
1581 * @return #GNUNET_YES to continue iteration
1582 */
1583static int
1584call_iterator (void *cls,
1585 const struct GNUNET_PeerIdentity *key,
1586 void *value)
1587{
1588 struct IterationContext *ic = cls;
1589 struct GSF_ConnectedPeer *cp = value;
1590
1591 ic->it (ic->it_cls,
1592 key, cp,
1593 &cp->ppd);
1594 return GNUNET_YES;
1595}
1596
1597
1598void
1599GSF_iterate_connected_peers_ (GSF_ConnectedPeerIterator it,
1600 void *it_cls)
1601{
1602 struct IterationContext ic;
1603
1604 ic.it = it;
1605 ic.it_cls = it_cls;
1606 GNUNET_CONTAINER_multipeermap_iterate (cp_map,
1607 &call_iterator,
1608 &ic);
1609}
1610
1611
1612/**
1613 * Obtain the identity of a connected peer.
1614 *
1615 * @param cp peer to get identity of
1616 * @param id identity to set (written to)
1617 */
1618void
1619GSF_connected_peer_get_identity_ (const struct GSF_ConnectedPeer *cp,
1620 struct GNUNET_PeerIdentity *id)
1621{
1622 GNUNET_assert (0 != cp->ppd.pid);
1623 GNUNET_PEER_resolve (cp->ppd.pid, id);
1624}
1625
1626
1627/**
1628 * Obtain the identity of a connected peer.
1629 *
1630 * @param cp peer to get identity of
1631 * @return reference to peer identity, valid until peer disconnects (!)
1632 */
1633const struct GNUNET_PeerIdentity *
1634GSF_connected_peer_get_identity2_ (const struct GSF_ConnectedPeer *cp)
1635{
1636 GNUNET_assert (0 != cp->ppd.pid);
1637 return GNUNET_PEER_resolve2 (cp->ppd.pid);
1638}
1639
1640
1641/**
1642 * Ask a peer to stop migrating data to us until the given point
1643 * in time.
1644 *
1645 * @param cp peer to ask
1646 * @param block_time until when to block
1647 */
1648void
1649GSF_block_peer_migration_ (struct GSF_ConnectedPeer *cp,
1650 struct GNUNET_TIME_Absolute block_time)
1651{
1652 struct GNUNET_MQ_Envelope *env;
1653 struct MigrationStopMessage *msm;
1654
1655 if (cp->last_migration_block.abs_value_us > block_time.abs_value_us)
1656 {
1657 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1658 "Migration already blocked for another %s\n",
1659 GNUNET_STRINGS_relative_time_to_string (
1660 GNUNET_TIME_absolute_get_remaining
1661 (cp->
1662 last_migration_block), GNUNET_YES));
1663 return; /* already blocked */
1664 }
1665 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Asking to stop migration for %s\n",
1666 GNUNET_STRINGS_relative_time_to_string (
1667 GNUNET_TIME_absolute_get_remaining (block_time),
1668 GNUNET_YES));
1669 cp->last_migration_block = block_time;
1670 env = GNUNET_MQ_msg (msm,
1671 GNUNET_MESSAGE_TYPE_FS_MIGRATION_STOP);
1672 msm->reserved = htonl (0);
1673 msm->duration
1674 = GNUNET_TIME_relative_hton (GNUNET_TIME_absolute_get_remaining
1675 (cp->last_migration_block));
1676 GNUNET_STATISTICS_update (GSF_stats,
1677 gettext_noop ("# migration stop messages sent"),
1678 1,
1679 GNUNET_NO);
1680 GSF_peer_transmit_ (cp,
1681 GNUNET_SYSERR,
1682 UINT32_MAX,
1683 env);
1684}
1685
1686
1687/**
1688 * Notify core about a preference we have for the given peer
1689 * (to allocate more resources towards it). The change will
1690 * be communicated the next time we reserve bandwidth with
1691 * core (not instantly).
1692 *
1693 * @param cp peer to reserve bandwidth from
1694 * @param pref preference change
1695 */
1696void
1697GSF_connected_peer_change_preference_ (struct GSF_ConnectedPeer *cp,
1698 uint64_t pref)
1699{
1700 cp->inc_preference += pref;
1701}
1702
1703
1704/**
1705 * Call this method periodically to flush respect information to disk.
1706 *
1707 * @param cls closure, not used
1708 */
1709static void
1710cron_flush_respect (void *cls)
1711{
1712 fr_task = NULL;
1713 GNUNET_CONTAINER_multipeermap_iterate (cp_map,
1714 &flush_respect,
1715 NULL);
1716 fr_task = GNUNET_SCHEDULER_add_delayed_with_priority (RESPECT_FLUSH_FREQ,
1717 GNUNET_SCHEDULER_PRIORITY_HIGH,
1718 &cron_flush_respect,
1719 NULL);
1720}
1721
1722
1723/**
1724 * Initialize peer management subsystem.
1725 */
1726void
1727GSF_connected_peer_init_ ()
1728{
1729 cp_map = GNUNET_CONTAINER_multipeermap_create (128, GNUNET_YES);
1730 peerstore = GNUNET_PEERSTORE_connect (GSF_cfg);
1731 fr_task = GNUNET_SCHEDULER_add_with_priority (GNUNET_SCHEDULER_PRIORITY_HIGH,
1732 &cron_flush_respect, NULL);
1733}
1734
1735
1736/**
1737 * Shutdown peer management subsystem.
1738 */
1739void
1740GSF_connected_peer_done_ ()
1741{
1742 GNUNET_CONTAINER_multipeermap_iterate (cp_map,
1743 &flush_respect,
1744 NULL);
1745 GNUNET_SCHEDULER_cancel (fr_task);
1746 fr_task = NULL;
1747 GNUNET_CONTAINER_multipeermap_destroy (cp_map);
1748 cp_map = NULL;
1749 GNUNET_PEERSTORE_disconnect (peerstore,
1750 GNUNET_YES);
1751}
1752
1753
1754/**
1755 * Iterator to remove references to LC entry.
1756 *
1757 * @param cls the `struct GSF_LocalClient *` to look for
1758 * @param key current key code
1759 * @param value value in the hash map (peer entry)
1760 * @return #GNUNET_YES (we should continue to iterate)
1761 */
1762static int
1763clean_local_client (void *cls,
1764 const struct GNUNET_PeerIdentity *key,
1765 void *value)
1766{
1767 const struct GSF_LocalClient *lc = cls;
1768 struct GSF_ConnectedPeer *cp = value;
1769 unsigned int i;
1770
1771 for (i = 0; i < CS2P_SUCCESS_LIST_SIZE; i++)
1772 if (cp->ppd.last_client_replies[i] == lc)
1773 cp->ppd.last_client_replies[i] = NULL;
1774 return GNUNET_YES;
1775}
1776
1777
1778/**
1779 * Notification that a local client disconnected. Clean up all of our
1780 * references to the given handle.
1781 *
1782 * @param lc handle to the local client (henceforth invalid)
1783 */
1784void
1785GSF_handle_local_client_disconnect_ (const struct GSF_LocalClient *lc)
1786{
1787 if (NULL == cp_map)
1788 return; /* already cleaned up */
1789 GNUNET_CONTAINER_multipeermap_iterate (cp_map,
1790 &clean_local_client,
1791 (void *) lc);
1792}
1793
1794
1795/* end of gnunet-service-fs_cp.c */
generated by cgit v1.2.3 (git 2.39.1) at 2024-06-05 21:06:08 +0000