2 This file is part of GNUnet.
3 (C) 2009, 2010 Christian Grothoff (and other contributing authors)
5 GNUnet is free software; you can redistribute it and/or modify
6 it under the terms of the GNU General Public License as published
7 by the Free Software Foundation; either version 3, or (at your
8 option) any later version.
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 General Public License for more details.
15 You should have received a copy of the GNU General Public License
16 along with GNUnet; see the file COPYING. If not, write to the
17 Free Software Foundation, Inc., 59 Temple Place - Suite 330,
18 Boston, MA 02111-1307, USA.
22 * @file fs/gnunet-service-fs.c
23 * @brief gnunet anonymity protocol implementation
24 * @author Christian Grothoff
31 #include "gnunet_constants.h"
32 #include "gnunet_core_service.h"
33 #include "gnunet_dht_service.h"
34 #include "gnunet_datastore_service.h"
35 #include "gnunet_load_lib.h"
36 #include "gnunet_peer_lib.h"
37 #include "gnunet_protocols.h"
38 #include "gnunet_signatures.h"
39 #include "gnunet_statistics_service.h"
40 #include "gnunet_util_lib.h"
41 #include "gnunet-service-fs_indexing.h"
44 #define DEBUG_FS GNUNET_NO
47 * Should we introduce random latency in processing? Required for proper
48 * implementation of GAP, but can be disabled for performance evaluation of
49 * the basic routing algorithm.
51 * Note that with delays enabled, performance can be significantly lower
52 * (several orders of magnitude in 2-peer test runs); if you want to
53 * measure throughput of other components, set this to NO. Also, you
54 * might want to consider changing 'RETRY_PROBABILITY_INV' to 1 for
55 * a rather wasteful mode of operation (that might still get the highest
56 * throughput overall).
58 #define SUPPORT_DELAYS GNUNET_YES
61 * Size for the hash map for DHT requests from the FS
62 * service. Should be about the number of concurrent
63 * DHT requests we plan to make.
65 #define FS_DHT_HT_SIZE 1024
68 * How often do we flush trust values to disk?
70 #define TRUST_FLUSH_FREQ GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MINUTES, 5)
73 * How often do we at most PUT content into the DHT?
75 #define MAX_DHT_PUT_FREQ GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 5)
78 * Inverse of the probability that we will submit the same query
79 * to the same peer again. If the same peer already got the query
80 * repeatedly recently, the probability is multiplied by the inverse
81 * of this number each time. Note that we only try about every TTL_DECREMENT/2
82 * plus MAX_CORK_DELAY (so roughly every 3.5s).
84 * Note that this factor is a key influence to performance in small
85 * networks (especially test networks of 2 peers) because if there is
86 * only a single peer with the data, this value will determine how
87 * soon we might re-try. For example, a value of 3 can result in
88 * 1.7 MB/s transfer rates for a 10 MB file when a value of 1 would
89 * give us 5 MB/s. OTOH, obviously re-trying the same peer can be
90 * rather inefficient in larger networks, hence picking 1 is in
91 * general not the best choice.
93 #define RETRY_PROBABILITY_INV 3
96 * What is the maximum delay for a P2P FS message (in our interaction
97 * with core)? FS-internal delays are another story. The value is
98 * chosen based on the 32k block size. Given that peers typcially
99 * have at least 1 kb/s bandwidth, 45s waits give us a chance to
100 * transmit one message even to the lowest-bandwidth peers.
102 #define MAX_TRANSMIT_DELAY GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 45)
105 * Maximum number of requests (from other peers, overall) that we're
106 * willing to have pending at any given point in time. Can be changed
107 * via the configuration file (32k is just the default).
109 static unsigned long long max_pending_requests = (32 * 1024);
113 * Information we keep for each pending reply. The
114 * actual message follows at the end of this struct.
116 struct PendingMessage;
119 * Function called upon completion of a transmission.
122 * @param pid ID of receiving peer, 0 on transmission error
124 typedef void (*TransmissionContinuation)(void * cls,
125 GNUNET_PEER_Id tpid);
129 * Information we keep for each pending message (GET/PUT). The
130 * actual message follows at the end of this struct.
132 struct PendingMessage
135 * This is a doubly-linked list of messages to the same peer.
137 struct PendingMessage *next;
140 * This is a doubly-linked list of messages to the same peer.
142 struct PendingMessage *prev;
145 * Entry in pending message list for this pending message.
147 struct PendingMessageList *pml;
150 * Function to call immediately once we have transmitted this
153 TransmissionContinuation cont;
161 * Do not transmit this pending message until this deadline.
163 struct GNUNET_TIME_Absolute delay_until;
166 * Size of the reply; actual reply message follows
167 * at the end of this struct.
172 * How important is this message for us?
180 * Information about a peer that we are connected to.
181 * We track data that is useful for determining which
182 * peers should receive our requests. We also keep
183 * a list of messages to transmit to this peer.
189 * List of the last clients for which this peer successfully
192 struct GNUNET_SERVER_Client *last_client_replies[CS2P_SUCCESS_LIST_SIZE];
195 * List of the last PIDs for which
196 * this peer successfully answered a query;
197 * We use 0 to indicate no successful reply.
199 GNUNET_PEER_Id last_p2p_replies[P2P_SUCCESS_LIST_SIZE];
202 * Average delay between sending the peer a request and
203 * getting a reply (only calculated over the requests for
204 * which we actually got a reply). Calculated
205 * as a moving average: new_delay = ((n-1)*last_delay+curr_delay) / n
207 * FIXME: actually, this is currently the delay between us originally
208 * receiving (not forwarding!) a request and us receiving a reply from
209 * this peer (regardless of when we transmitted this request to this peer!)
211 struct GNUNET_TIME_Relative avg_delay;
214 * Point in time until which this peer does not want us to migrate content
217 struct GNUNET_TIME_Absolute migration_blocked;
220 * Time until when we blocked this peer from migrating
223 struct GNUNET_TIME_Absolute last_migration_block;
226 * Transmission times for the last MAX_QUEUE_PER_PEER
227 * requests for this peer. Used as a ring buffer, current
228 * offset is stored in 'last_request_times_off'. If the
229 * oldest entry is more recent than the 'avg_delay', we should
230 * not send any more requests right now.
232 struct GNUNET_TIME_Absolute last_request_times[MAX_QUEUE_PER_PEER];
235 * Handle for an active request for transmission to this
238 struct GNUNET_CORE_TransmitHandle *cth;
241 * Messages (replies, queries, content migration) we would like to
242 * send to this peer in the near future. Sorted by priority, head.
244 struct PendingMessage *pending_messages_head;
247 * Messages (replies, queries, content migration) we would like to
248 * send to this peer in the near future. Sorted by priority, tail.
250 struct PendingMessage *pending_messages_tail;
253 * How long does it typically take for us to transmit a message
254 * to this peer? (delay between the request being issued and
255 * the callback being invoked).
257 struct GNUNET_LOAD_Value *transmission_delay;
260 * Time when the last transmission request was issued.
262 struct GNUNET_TIME_Absolute last_transmission_request_start;
265 * ID of delay task for scheduling transmission.
267 GNUNET_SCHEDULER_TaskIdentifier delayed_transmission_request_task;
270 * Average priority of successful replies. Calculated
271 * as a moving average: new_avg = ((n-1)*last_avg+curr_prio) / n
276 * Increase in traffic preference still to be submitted
277 * to the core service for this peer.
279 uint64_t inc_preference;
282 * Trust rating for this peer
287 * Trust rating for this peer on disk.
292 * The peer's identity.
297 * Size of the linked list of 'pending_messages'.
299 unsigned int pending_requests;
302 * Which offset in "last_p2p_replies" will be updated next?
303 * (we go round-robin).
305 unsigned int last_p2p_replies_woff;
308 * Which offset in "last_client_replies" will be updated next?
309 * (we go round-robin).
311 unsigned int last_client_replies_woff;
314 * Current offset into 'last_request_times' ring buffer.
316 unsigned int last_request_times_off;
322 * Information we keep for each pending request. We should try to
323 * keep this struct as small as possible since its memory consumption
324 * is key to how many requests we can have pending at once.
326 struct PendingRequest;
330 * Doubly-linked list of requests we are performing
331 * on behalf of the same client.
333 struct ClientRequestList
337 * This is a doubly-linked list.
339 struct ClientRequestList *next;
342 * This is a doubly-linked list.
344 struct ClientRequestList *prev;
347 * Request this entry represents.
349 struct PendingRequest *req;
352 * Client list this request belongs to.
354 struct ClientList *client_list;
360 * Replies to be transmitted to the client. The actual
361 * response message is allocated after this struct.
363 struct ClientResponseMessage
366 * This is a doubly-linked list.
368 struct ClientResponseMessage *next;
371 * This is a doubly-linked list.
373 struct ClientResponseMessage *prev;
376 * Client list entry this response belongs to.
378 struct ClientList *client_list;
381 * Number of bytes in the response.
388 * Linked list of clients we are performing requests
394 * This is a linked list.
396 struct ClientList *next;
399 * ID of a client making a request, NULL if this entry is for a
402 struct GNUNET_SERVER_Client *client;
405 * Head of list of requests performed on behalf
406 * of this client right now.
408 struct ClientRequestList *rl_head;
411 * Tail of list of requests performed on behalf
412 * of this client right now.
414 struct ClientRequestList *rl_tail;
417 * Head of linked list of responses.
419 struct ClientResponseMessage *res_head;
422 * Tail of linked list of responses.
424 struct ClientResponseMessage *res_tail;
427 * Context for sending replies.
429 struct GNUNET_CONNECTION_TransmitHandle *th;
435 * Information about a peer that we have forwarded this
436 * request to already.
438 struct UsedTargetEntry
441 * What was the last time we have transmitted this request to this
444 struct GNUNET_TIME_Absolute last_request_time;
447 * How often have we transmitted this request to this peer?
449 unsigned int num_requests;
452 * PID of the target peer.
463 * Doubly-linked list of messages we are performing
464 * due to a pending request.
466 struct PendingMessageList
470 * This is a doubly-linked list of messages on behalf of the same request.
472 struct PendingMessageList *next;
475 * This is a doubly-linked list of messages on behalf of the same request.
477 struct PendingMessageList *prev;
480 * Message this entry represents.
482 struct PendingMessage *pm;
485 * Request this entry belongs to.
487 struct PendingRequest *req;
490 * Peer this message is targeted for.
492 struct ConnectedPeer *target;
498 * Information we keep for each pending request. We should try to
499 * keep this struct as small as possible since its memory consumption
500 * is key to how many requests we can have pending at once.
502 struct PendingRequest
506 * If this request was made by a client, this is our entry in the
507 * client request list; otherwise NULL.
509 struct ClientRequestList *client_request_list;
512 * Entry of peer responsible for this entry (if this request
513 * was made by a peer).
515 struct ConnectedPeer *cp;
518 * If this is a namespace query, pointer to the hash of the public
519 * key of the namespace; otherwise NULL. Pointer will be to the
520 * end of this struct (so no need to free it).
522 const GNUNET_HashCode *namespace;
525 * Bloomfilter we use to filter out replies that we don't care about
526 * (anymore). NULL as long as we are interested in all replies.
528 struct GNUNET_CONTAINER_BloomFilter *bf;
531 * Context of our GNUNET_CORE_peer_change_preference call.
533 struct GNUNET_CORE_InformationRequestContext *irc;
536 * Reference to DHT get operation for this request (or NULL).
538 struct GNUNET_DHT_GetHandle *dht_get;
541 * Hash code of all replies that we have seen so far (only valid
542 * if client is not NULL since we only track replies like this for
545 GNUNET_HashCode *replies_seen;
548 * Node in the heap representing this entry; NULL
549 * if we have no heap node.
551 struct GNUNET_CONTAINER_HeapNode *hnode;
554 * Head of list of messages being performed on behalf of this
557 struct PendingMessageList *pending_head;
560 * Tail of list of messages being performed on behalf of this
563 struct PendingMessageList *pending_tail;
566 * When did we first see this request (form this peer), or, if our
567 * client is initiating, when did we last initiate a search?
569 struct GNUNET_TIME_Absolute start_time;
572 * The query that this request is for.
574 GNUNET_HashCode query;
577 * The task responsible for transmitting queries
580 GNUNET_SCHEDULER_TaskIdentifier task;
583 * (Interned) Peer identifier that identifies a preferred target
586 GNUNET_PEER_Id target_pid;
589 * (Interned) Peer identifiers of peers that have already
590 * received our query for this content.
592 struct UsedTargetEntry *used_targets;
595 * Our entry in the queue (non-NULL while we wait for our
596 * turn to interact with the local database).
598 struct GNUNET_DATASTORE_QueueEntry *qe;
601 * Size of the 'bf' (in bytes).
606 * Desired anonymity level; only valid for requests from a local client.
608 uint32_t anonymity_level;
611 * How many entries in "used_targets" are actually valid?
613 unsigned int used_targets_off;
616 * How long is the "used_targets" array?
618 unsigned int used_targets_size;
621 * Number of results found for this request.
623 unsigned int results_found;
626 * How many entries in "replies_seen" are actually valid?
628 unsigned int replies_seen_off;
631 * How long is the "replies_seen" array?
633 unsigned int replies_seen_size;
636 * Priority with which this request was made. If one of our clients
637 * made the request, then this is the current priority that we are
638 * using when initiating the request. This value is used when
639 * we decide to reward other peers with trust for providing a reply.
644 * Priority points left for us to spend when forwarding this request
647 uint32_t remaining_priority;
650 * Number to mingle hashes for bloom-filter tests with.
655 * TTL with which we saw this request (or, if we initiated, TTL that
656 * we used for the request).
661 * Type of the content that this request is for.
663 enum GNUNET_BLOCK_Type type;
666 * Remove this request after transmission of the current response.
671 * GNUNET_YES if we should not forward this request to other peers.
676 * GNUNET_YES if we should not forward this request to other peers.
684 * Block that is ready for migration to other peers. Actual data is at the end of the block.
686 struct MigrationReadyBlock
690 * This is a doubly-linked list.
692 struct MigrationReadyBlock *next;
695 * This is a doubly-linked list.
697 struct MigrationReadyBlock *prev;
700 * Query for the block.
702 GNUNET_HashCode query;
705 * When does this block expire?
707 struct GNUNET_TIME_Absolute expiration;
710 * Peers we would consider forwarding this
711 * block to. Zero for empty entries.
713 GNUNET_PEER_Id target_list[MIGRATION_LIST_SIZE];
721 * Number of targets already used.
723 unsigned int used_targets;
728 enum GNUNET_BLOCK_Type type;
733 * Our connection to the datastore.
735 static struct GNUNET_DATASTORE_Handle *dsh;
740 static struct GNUNET_BLOCK_Context *block_ctx;
743 * Our block configuration.
745 static struct GNUNET_CONFIGURATION_Handle *block_cfg;
750 static struct GNUNET_SCHEDULER_Handle *sched;
755 static const struct GNUNET_CONFIGURATION_Handle *cfg;
758 * Map of peer identifiers to "struct ConnectedPeer" (for that peer).
760 static struct GNUNET_CONTAINER_MultiHashMap *connected_peers;
763 * Map of peer identifiers to "struct PendingRequest" (for that peer).
765 static struct GNUNET_CONTAINER_MultiHashMap *peer_request_map;
768 * Map of query identifiers to "struct PendingRequest" (for that query).
770 static struct GNUNET_CONTAINER_MultiHashMap *query_request_map;
773 * Heap with the request that will expire next at the top. Contains
774 * pointers of type "struct PendingRequest*"; these will *also* be
775 * aliased from the "requests_by_peer" data structures and the
776 * "requests_by_query" table. Note that requests from our clients
777 * don't expire and are thus NOT in the "requests_by_expiration"
778 * (or the "requests_by_peer" tables).
780 static struct GNUNET_CONTAINER_Heap *requests_by_expiration_heap;
783 * Handle for reporting statistics.
785 static struct GNUNET_STATISTICS_Handle *stats;
788 * Linked list of clients we are currently processing requests for.
790 static struct ClientList *client_list;
793 * Pointer to handle to the core service (points to NULL until we've
796 static struct GNUNET_CORE_Handle *core;
799 * Head of linked list of blocks that can be migrated.
801 static struct MigrationReadyBlock *mig_head;
804 * Tail of linked list of blocks that can be migrated.
806 static struct MigrationReadyBlock *mig_tail;
809 * Request to datastore for migration (or NULL).
811 static struct GNUNET_DATASTORE_QueueEntry *mig_qe;
814 * Request to datastore for DHT PUTs (or NULL).
816 static struct GNUNET_DATASTORE_QueueEntry *dht_qe;
819 * Type we will request for the next DHT PUT round from the datastore.
821 static enum GNUNET_BLOCK_Type dht_put_type = GNUNET_BLOCK_TYPE_FS_KBLOCK;
824 * Where do we store trust information?
826 static char *trustDirectory;
829 * ID of task that collects blocks for migration.
831 static GNUNET_SCHEDULER_TaskIdentifier mig_task;
834 * ID of task that collects blocks for DHT PUTs.
836 static GNUNET_SCHEDULER_TaskIdentifier dht_task;
839 * What is the maximum frequency at which we are allowed to
840 * poll the datastore for migration content?
842 static struct GNUNET_TIME_Relative min_migration_delay;
845 * Handle for DHT operations.
847 static struct GNUNET_DHT_Handle *dht_handle;
850 * Size of the doubly-linked list of migration blocks.
852 static unsigned int mig_size;
855 * Are we allowed to migrate content to this peer.
857 static int active_migration;
860 * How many entires with zero anonymity do we currently estimate
861 * to have in the database?
863 static unsigned int zero_anonymity_count_estimate;
866 * Typical priorities we're seeing from other peers right now. Since
867 * most priorities will be zero, this value is the weighted average of
868 * non-zero priorities seen "recently". In order to ensure that new
869 * values do not dramatically change the ratio, values are first
870 * "capped" to a reasonable range (+N of the current value) and then
871 * averaged into the existing value by a ratio of 1:N. Hence
872 * receiving the largest possible priority can still only raise our
873 * "current_priorities" by at most 1.
875 static double current_priorities;
878 * Datastore 'GET' load tracking.
880 static struct GNUNET_LOAD_Value *datastore_get_load;
883 * Datastore 'PUT' load tracking.
885 static struct GNUNET_LOAD_Value *datastore_put_load;
888 * How long do requests typically stay in the routing table?
890 static struct GNUNET_LOAD_Value *rt_entry_lifetime;
893 * We've just now completed a datastore request. Update our
894 * datastore load calculations.
896 * @param start time when the datastore request was issued
899 update_datastore_delays (struct GNUNET_TIME_Absolute start)
901 struct GNUNET_TIME_Relative delay;
903 delay = GNUNET_TIME_absolute_get_duration (start);
904 GNUNET_LOAD_update (datastore_get_load,
910 * Get the filename under which we would store the GNUNET_HELLO_Message
911 * for the given host and protocol.
912 * @return filename of the form DIRECTORY/HOSTID
915 get_trust_filename (const struct GNUNET_PeerIdentity *id)
917 struct GNUNET_CRYPTO_HashAsciiEncoded fil;
920 GNUNET_CRYPTO_hash_to_enc (&id->hashPubKey, &fil);
921 GNUNET_asprintf (&fn, "%s%s%s", trustDirectory, DIR_SEPARATOR_STR, &fil);
928 * Transmit messages by copying it to the target buffer
929 * "buf". "buf" will be NULL and "size" zero if the socket was closed
930 * for writing in the meantime. In that case, do nothing
931 * (the disconnect or shutdown handler will take care of the rest).
932 * If we were able to transmit messages and there are still more
933 * pending, ask core again for further calls to this function.
935 * @param cls closure, pointer to the 'struct ConnectedPeer*'
936 * @param size number of bytes available in buf
937 * @param buf where the callee should write the message
938 * @return number of bytes written to buf
941 transmit_to_peer (void *cls,
942 size_t size, void *buf);
945 /* ******************* clean up functions ************************ */
948 * Delete the given migration block.
950 * @param mb block to delete
953 delete_migration_block (struct MigrationReadyBlock *mb)
955 GNUNET_CONTAINER_DLL_remove (mig_head,
958 GNUNET_PEER_decrement_rcs (mb->target_list,
959 MIGRATION_LIST_SIZE);
966 * Compare the distance of two peers to a key.
969 * @param p1 first peer
970 * @param p2 second peer
971 * @return GNUNET_YES if P1 is closer to key than P2
974 is_closer (const GNUNET_HashCode *key,
975 const struct GNUNET_PeerIdentity *p1,
976 const struct GNUNET_PeerIdentity *p2)
978 return GNUNET_CRYPTO_hash_xorcmp (&p1->hashPubKey,
985 * Consider migrating content to a given peer.
987 * @param cls 'struct MigrationReadyBlock*' to select
988 * targets for (or NULL for none)
989 * @param key ID of the peer
990 * @param value 'struct ConnectedPeer' of the peer
991 * @return GNUNET_YES (always continue iteration)
994 consider_migration (void *cls,
995 const GNUNET_HashCode *key,
998 struct MigrationReadyBlock *mb = cls;
999 struct ConnectedPeer *cp = value;
1000 struct MigrationReadyBlock *pos;
1001 struct GNUNET_PeerIdentity cppid;
1002 struct GNUNET_PeerIdentity otherpid;
1003 struct GNUNET_PeerIdentity worstpid;
1008 /* consider 'cp' as a migration target for mb */
1009 if (GNUNET_TIME_absolute_get_remaining (cp->migration_blocked).value > 0)
1010 return GNUNET_YES; /* peer has requested no migration! */
1013 GNUNET_PEER_resolve (cp->pid,
1015 repl = MIGRATION_LIST_SIZE;
1016 for (i=0;i<MIGRATION_LIST_SIZE;i++)
1018 if (mb->target_list[i] == 0)
1020 mb->target_list[i] = cp->pid;
1021 GNUNET_PEER_change_rc (mb->target_list[i], 1);
1022 repl = MIGRATION_LIST_SIZE;
1025 GNUNET_PEER_resolve (mb->target_list[i],
1027 if ( (repl == MIGRATION_LIST_SIZE) &&
1028 is_closer (&mb->query,
1033 worstpid = otherpid;
1035 else if ( (repl != MIGRATION_LIST_SIZE) &&
1036 (is_closer (&mb->query,
1041 worstpid = otherpid;
1044 if (repl != MIGRATION_LIST_SIZE)
1046 GNUNET_PEER_change_rc (mb->target_list[repl], -1);
1047 mb->target_list[repl] = cp->pid;
1048 GNUNET_PEER_change_rc (mb->target_list[repl], 1);
1052 /* consider scheduling transmission to cp for content migration */
1053 if (cp->cth != NULL)
1059 for (i=0;i<MIGRATION_LIST_SIZE;i++)
1061 if (cp->pid == pos->target_list[i])
1066 msize = GNUNET_MIN (msize,
1074 return GNUNET_YES; /* no content available */
1076 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1077 "Trying to migrate at least %u bytes to peer `%s'\n",
1081 if (cp->delayed_transmission_request_task != GNUNET_SCHEDULER_NO_TASK)
1083 GNUNET_SCHEDULER_cancel (sched, cp->delayed_transmission_request_task);
1084 cp->delayed_transmission_request_task = GNUNET_SCHEDULER_NO_TASK;
1087 = GNUNET_CORE_notify_transmit_ready (core,
1088 0, GNUNET_TIME_UNIT_FOREVER_REL,
1089 (const struct GNUNET_PeerIdentity*) key,
1090 msize + sizeof (struct PutMessage),
1098 * Task that is run periodically to obtain blocks for content
1102 * @param tc scheduler context (also unused)
1105 gather_migration_blocks (void *cls,
1106 const struct GNUNET_SCHEDULER_TaskContext *tc);
1112 * Task that is run periodically to obtain blocks for DHT PUTs.
1114 * @param cls type of blocks to gather
1115 * @param tc scheduler context (unused)
1118 gather_dht_put_blocks (void *cls,
1119 const struct GNUNET_SCHEDULER_TaskContext *tc);
1123 * If the migration task is not currently running, consider
1124 * (re)scheduling it with the appropriate delay.
1127 consider_migration_gathering ()
1129 struct GNUNET_TIME_Relative delay;
1135 if (mig_task != GNUNET_SCHEDULER_NO_TASK)
1137 delay = GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS,
1139 delay = GNUNET_TIME_relative_divide (delay,
1140 MAX_MIGRATION_QUEUE);
1141 delay = GNUNET_TIME_relative_max (delay,
1142 min_migration_delay);
1143 mig_task = GNUNET_SCHEDULER_add_delayed (sched,
1145 &gather_migration_blocks,
1151 * If the DHT PUT gathering task is not currently running, consider
1152 * (re)scheduling it with the appropriate delay.
1155 consider_dht_put_gathering (void *cls)
1157 struct GNUNET_TIME_Relative delay;
1163 if (dht_task != GNUNET_SCHEDULER_NO_TASK)
1165 if (zero_anonymity_count_estimate > 0)
1167 delay = GNUNET_TIME_relative_divide (GNUNET_DHT_DEFAULT_REPUBLISH_FREQUENCY,
1168 zero_anonymity_count_estimate);
1169 delay = GNUNET_TIME_relative_min (delay,
1174 /* if we have NO zero-anonymity content yet, wait 5 minutes for some to
1175 (hopefully) appear */
1176 delay = GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MINUTES, 5);
1178 dht_task = GNUNET_SCHEDULER_add_delayed (sched,
1180 &gather_dht_put_blocks,
1186 * Process content offered for migration.
1188 * @param cls closure
1189 * @param key key for the content
1190 * @param size number of bytes in data
1191 * @param data content stored
1192 * @param type type of the content
1193 * @param priority priority of the content
1194 * @param anonymity anonymity-level for the content
1195 * @param expiration expiration time for the content
1196 * @param uid unique identifier for the datum;
1197 * maybe 0 if no unique identifier is available
1200 process_migration_content (void *cls,
1201 const GNUNET_HashCode * key,
1204 enum GNUNET_BLOCK_Type type,
1207 struct GNUNET_TIME_Absolute
1208 expiration, uint64_t uid)
1210 struct MigrationReadyBlock *mb;
1215 if (mig_size < MAX_MIGRATION_QUEUE)
1216 consider_migration_gathering ();
1219 if (type == GNUNET_BLOCK_TYPE_FS_ONDEMAND)
1222 GNUNET_FS_handle_on_demand_block (key, size, data,
1223 type, priority, anonymity,
1225 &process_migration_content,
1228 GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
1233 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1234 "Retrieved block `%s' of type %u for migration\n",
1238 mb = GNUNET_malloc (sizeof (struct MigrationReadyBlock) + size);
1240 mb->expiration = expiration;
1243 memcpy (&mb[1], data, size);
1244 GNUNET_CONTAINER_DLL_insert_after (mig_head,
1249 GNUNET_CONTAINER_multihashmap_iterate (connected_peers,
1250 &consider_migration,
1252 GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
1257 * Function called upon completion of the DHT PUT operation.
1260 dht_put_continuation (void *cls,
1261 const struct GNUNET_SCHEDULER_TaskContext *tc)
1263 GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
1268 * Store content in DHT.
1270 * @param cls closure
1271 * @param key key for the content
1272 * @param size number of bytes in data
1273 * @param data content stored
1274 * @param type type of the content
1275 * @param priority priority of the content
1276 * @param anonymity anonymity-level for the content
1277 * @param expiration expiration time for the content
1278 * @param uid unique identifier for the datum;
1279 * maybe 0 if no unique identifier is available
1282 process_dht_put_content (void *cls,
1283 const GNUNET_HashCode * key,
1286 enum GNUNET_BLOCK_Type type,
1289 struct GNUNET_TIME_Absolute
1290 expiration, uint64_t uid)
1292 static unsigned int counter;
1293 static GNUNET_HashCode last_vhash;
1294 static GNUNET_HashCode vhash;
1299 consider_dht_put_gathering (cls);
1302 /* slightly funky code to estimate the total number of values with zero
1303 anonymity from the maximum observed length of a monotonically increasing
1304 sequence of hashes over the contents */
1305 GNUNET_CRYPTO_hash (data, size, &vhash);
1306 if (GNUNET_CRYPTO_hash_cmp (&vhash, &last_vhash) <= 0)
1308 if (zero_anonymity_count_estimate > 0)
1309 zero_anonymity_count_estimate /= 2;
1315 if (zero_anonymity_count_estimate < (1 << counter))
1316 zero_anonymity_count_estimate = (1 << counter);
1318 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1319 "Retrieved block `%s' of type %u for DHT PUT\n",
1323 GNUNET_DHT_put (dht_handle,
1330 GNUNET_TIME_UNIT_FOREVER_REL,
1331 &dht_put_continuation,
1337 * Task that is run periodically to obtain blocks for content
1341 * @param tc scheduler context (also unused)
1344 gather_migration_blocks (void *cls,
1345 const struct GNUNET_SCHEDULER_TaskContext *tc)
1347 mig_task = GNUNET_SCHEDULER_NO_TASK;
1350 mig_qe = GNUNET_DATASTORE_get_random (dsh, 0, UINT_MAX,
1351 GNUNET_TIME_UNIT_FOREVER_REL,
1352 &process_migration_content, NULL);
1353 GNUNET_assert (mig_qe != NULL);
1359 * Task that is run periodically to obtain blocks for DHT PUTs.
1361 * @param cls type of blocks to gather
1362 * @param tc scheduler context (unused)
1365 gather_dht_put_blocks (void *cls,
1366 const struct GNUNET_SCHEDULER_TaskContext *tc)
1368 dht_task = GNUNET_SCHEDULER_NO_TASK;
1371 if (dht_put_type == GNUNET_BLOCK_TYPE_FS_ONDEMAND)
1372 dht_put_type = GNUNET_BLOCK_TYPE_FS_KBLOCK;
1373 dht_qe = GNUNET_DATASTORE_get_zero_anonymity (dsh, 0, UINT_MAX,
1374 GNUNET_TIME_UNIT_FOREVER_REL,
1376 &process_dht_put_content, NULL);
1377 GNUNET_assert (dht_qe != NULL);
1383 * We're done with a particular message list entry.
1384 * Free all associated resources.
1386 * @param pml entry to destroy
1389 destroy_pending_message_list_entry (struct PendingMessageList *pml)
1391 GNUNET_CONTAINER_DLL_remove (pml->req->pending_head,
1392 pml->req->pending_tail,
1394 GNUNET_CONTAINER_DLL_remove (pml->target->pending_messages_head,
1395 pml->target->pending_messages_tail,
1397 pml->target->pending_requests--;
1398 GNUNET_free (pml->pm);
1404 * Destroy the given pending message (and call the respective
1407 * @param pm message to destroy
1408 * @param tpid id of peer that the message was delivered to, or 0 for none
1411 destroy_pending_message (struct PendingMessage *pm,
1412 GNUNET_PEER_Id tpid)
1414 struct PendingMessageList *pml = pm->pml;
1415 TransmissionContinuation cont;
1419 cont_cls = pm->cont_cls;
1422 GNUNET_assert (pml->pm == pm);
1423 GNUNET_assert ( (tpid == 0) || (tpid == pml->target->pid) );
1424 destroy_pending_message_list_entry (pml);
1431 cont (cont_cls, tpid);
1436 * We're done processing a particular request.
1437 * Free all associated resources.
1439 * @param pr request to destroy
1442 destroy_pending_request (struct PendingRequest *pr)
1444 struct GNUNET_PeerIdentity pid;
1447 if (pr->hnode != NULL)
1449 GNUNET_CONTAINER_heap_remove_node (requests_by_expiration_heap,
1453 if (NULL == pr->client_request_list)
1455 GNUNET_STATISTICS_update (stats,
1456 gettext_noop ("# P2P searches active"),
1462 GNUNET_STATISTICS_update (stats,
1463 gettext_noop ("# client searches active"),
1468 GNUNET_CONTAINER_multihashmap_remove (query_request_map,
1472 GNUNET_LOAD_update (rt_entry_lifetime,
1473 GNUNET_TIME_absolute_get_duration (pr->start_time).value);
1477 GNUNET_DATASTORE_cancel (pr->qe);
1480 if (pr->dht_get != NULL)
1482 GNUNET_DHT_get_stop (pr->dht_get);
1485 if (pr->client_request_list != NULL)
1487 GNUNET_CONTAINER_DLL_remove (pr->client_request_list->client_list->rl_head,
1488 pr->client_request_list->client_list->rl_tail,
1489 pr->client_request_list);
1490 GNUNET_free (pr->client_request_list);
1491 pr->client_request_list = NULL;
1495 GNUNET_PEER_resolve (pr->cp->pid,
1497 (void) GNUNET_CONTAINER_multihashmap_remove (peer_request_map,
1504 GNUNET_CONTAINER_bloomfilter_free (pr->bf);
1507 if (pr->irc != NULL)
1509 GNUNET_CORE_peer_change_preference_cancel (pr->irc);
1512 if (pr->replies_seen != NULL)
1514 GNUNET_free (pr->replies_seen);
1515 pr->replies_seen = NULL;
1517 if (pr->task != GNUNET_SCHEDULER_NO_TASK)
1519 GNUNET_SCHEDULER_cancel (sched,
1521 pr->task = GNUNET_SCHEDULER_NO_TASK;
1523 while (NULL != pr->pending_head)
1524 destroy_pending_message_list_entry (pr->pending_head);
1525 GNUNET_PEER_change_rc (pr->target_pid, -1);
1526 if (pr->used_targets != NULL)
1528 for (i=0;i<pr->used_targets_off;i++)
1529 GNUNET_PEER_change_rc (pr->used_targets[i].pid, -1);
1530 GNUNET_free (pr->used_targets);
1531 pr->used_targets_off = 0;
1532 pr->used_targets_size = 0;
1533 pr->used_targets = NULL;
1540 * Method called whenever a given peer connects.
1542 * @param cls closure, not used
1543 * @param peer peer identity this notification is about
1544 * @param latency reported latency of the connection with 'other'
1545 * @param distance reported distance (DV) to 'other'
1548 peer_connect_handler (void *cls,
1550 GNUNET_PeerIdentity * peer,
1551 struct GNUNET_TIME_Relative latency,
1554 struct ConnectedPeer *cp;
1555 struct MigrationReadyBlock *pos;
1559 cp = GNUNET_malloc (sizeof (struct ConnectedPeer));
1560 cp->transmission_delay = GNUNET_LOAD_value_init ();
1561 cp->pid = GNUNET_PEER_intern (peer);
1563 fn = get_trust_filename (peer);
1564 if ((GNUNET_DISK_file_test (fn) == GNUNET_YES) &&
1565 (sizeof (trust) == GNUNET_DISK_fn_read (fn, &trust, sizeof (trust))))
1566 cp->disk_trust = cp->trust = ntohl (trust);
1569 GNUNET_break (GNUNET_OK ==
1570 GNUNET_CONTAINER_multihashmap_put (connected_peers,
1573 GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
1578 (void) consider_migration (pos, &peer->hashPubKey, cp);
1585 * Increase the host credit by a value.
1587 * @param host which peer to change the trust value on
1588 * @param value is the int value by which the
1589 * host credit is to be increased or decreased
1590 * @returns the actual change in trust (positive or negative)
1593 change_host_trust (struct ConnectedPeer *host, int value)
1595 unsigned int old_trust;
1599 GNUNET_assert (host != NULL);
1600 old_trust = host->trust;
1603 if (host->trust + value < host->trust)
1605 value = UINT32_MAX - host->trust;
1606 host->trust = UINT32_MAX;
1609 host->trust += value;
1613 if (host->trust < -value)
1615 value = -host->trust;
1619 host->trust += value;
1626 * Write host-trust information to a file - flush the buffer entry!
1629 flush_trust (void *cls,
1630 const GNUNET_HashCode *key,
1633 struct ConnectedPeer *host = value;
1636 struct GNUNET_PeerIdentity pid;
1638 if (host->trust == host->disk_trust)
1639 return GNUNET_OK; /* unchanged */
1640 GNUNET_PEER_resolve (host->pid,
1642 fn = get_trust_filename (&pid);
1643 if (host->trust == 0)
1645 if ((0 != UNLINK (fn)) && (errno != ENOENT))
1646 GNUNET_log_strerror_file (GNUNET_ERROR_TYPE_WARNING |
1647 GNUNET_ERROR_TYPE_BULK, "unlink", fn);
1651 trust = htonl (host->trust);
1652 if (sizeof(uint32_t) == GNUNET_DISK_fn_write (fn, &trust,
1654 GNUNET_DISK_PERM_USER_READ | GNUNET_DISK_PERM_USER_WRITE
1655 | GNUNET_DISK_PERM_GROUP_READ | GNUNET_DISK_PERM_OTHER_READ))
1656 host->disk_trust = host->trust;
1663 * Call this method periodically to scan data/hosts for new hosts.
1666 cron_flush_trust (void *cls,
1667 const struct GNUNET_SCHEDULER_TaskContext *tc)
1670 if (NULL == connected_peers)
1672 GNUNET_CONTAINER_multihashmap_iterate (connected_peers,
1677 if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
1679 GNUNET_SCHEDULER_add_delayed (tc->sched,
1680 TRUST_FLUSH_FREQ, &cron_flush_trust, NULL);
1685 * Free (each) request made by the peer.
1687 * @param cls closure, points to peer that the request belongs to
1688 * @param key current key code
1689 * @param value value in the hash map
1690 * @return GNUNET_YES (we should continue to iterate)
1693 destroy_request (void *cls,
1694 const GNUNET_HashCode * key,
1697 const struct GNUNET_PeerIdentity * peer = cls;
1698 struct PendingRequest *pr = value;
1700 GNUNET_break (GNUNET_YES ==
1701 GNUNET_CONTAINER_multihashmap_remove (peer_request_map,
1704 destroy_pending_request (pr);
1710 * Method called whenever a peer disconnects.
1712 * @param cls closure, not used
1713 * @param peer peer identity this notification is about
1716 peer_disconnect_handler (void *cls,
1718 GNUNET_PeerIdentity * peer)
1720 struct ConnectedPeer *cp;
1721 struct PendingMessage *pm;
1723 struct MigrationReadyBlock *pos;
1724 struct MigrationReadyBlock *next;
1726 GNUNET_CONTAINER_multihashmap_get_multiple (peer_request_map,
1730 cp = GNUNET_CONTAINER_multihashmap_get (connected_peers,
1734 for (i=0;i<CS2P_SUCCESS_LIST_SIZE;i++)
1736 if (NULL != cp->last_client_replies[i])
1738 GNUNET_SERVER_client_drop (cp->last_client_replies[i]);
1739 cp->last_client_replies[i] = NULL;
1742 GNUNET_break (GNUNET_YES ==
1743 GNUNET_CONTAINER_multihashmap_remove (connected_peers,
1746 /* remove this peer from migration considerations; schedule
1749 while (NULL != (pos = next))
1752 for (i=0;i<MIGRATION_LIST_SIZE;i++)
1754 if (pos->target_list[i] == cp->pid)
1756 GNUNET_PEER_change_rc (pos->target_list[i], -1);
1757 pos->target_list[i] = 0;
1760 if (pos->used_targets >= GNUNET_CONTAINER_multihashmap_size (connected_peers))
1762 delete_migration_block (pos);
1763 consider_migration_gathering ();
1766 GNUNET_CONTAINER_multihashmap_iterate (connected_peers,
1767 &consider_migration,
1770 GNUNET_PEER_change_rc (cp->pid, -1);
1771 GNUNET_PEER_decrement_rcs (cp->last_p2p_replies, P2P_SUCCESS_LIST_SIZE);
1772 if (NULL != cp->cth)
1774 GNUNET_CORE_notify_transmit_ready_cancel (cp->cth);
1777 if (cp->delayed_transmission_request_task != GNUNET_SCHEDULER_NO_TASK)
1779 GNUNET_SCHEDULER_cancel (sched, cp->delayed_transmission_request_task);
1780 cp->delayed_transmission_request_task = GNUNET_SCHEDULER_NO_TASK;
1782 while (NULL != (pm = cp->pending_messages_head))
1783 destroy_pending_message (pm, 0 /* delivery failed */);
1784 GNUNET_LOAD_value_free (cp->transmission_delay);
1785 GNUNET_break (0 == cp->pending_requests);
1791 * Iterator over hash map entries that removes all occurences
1792 * of the given 'client' from the 'last_client_replies' of the
1793 * given connected peer.
1795 * @param cls closure, the 'struct GNUNET_SERVER_Client*' to remove
1796 * @param key current key code (unused)
1797 * @param value value in the hash map (the 'struct ConnectedPeer*' to change)
1798 * @return GNUNET_YES (we should continue to iterate)
1801 remove_client_from_last_client_replies (void *cls,
1802 const GNUNET_HashCode * key,
1805 struct GNUNET_SERVER_Client *client = cls;
1806 struct ConnectedPeer *cp = value;
1809 for (i=0;i<CS2P_SUCCESS_LIST_SIZE;i++)
1811 if (cp->last_client_replies[i] == client)
1813 GNUNET_SERVER_client_drop (cp->last_client_replies[i]);
1814 cp->last_client_replies[i] = NULL;
1822 * A client disconnected. Remove all of its pending queries.
1824 * @param cls closure, NULL
1825 * @param client identification of the client
1828 handle_client_disconnect (void *cls,
1829 struct GNUNET_SERVER_Client
1832 struct ClientList *pos;
1833 struct ClientList *prev;
1834 struct ClientRequestList *rcl;
1835 struct ClientResponseMessage *creply;
1841 while ( (NULL != pos) &&
1842 (pos->client != client) )
1848 return; /* no requests pending for this client */
1849 while (NULL != (rcl = pos->rl_head))
1851 GNUNET_log (GNUNET_ERROR_TYPE_INFO,
1852 "Destroying pending request `%s' on disconnect\n",
1853 GNUNET_h2s (&rcl->req->query));
1854 destroy_pending_request (rcl->req);
1857 client_list = pos->next;
1859 prev->next = pos->next;
1860 if (pos->th != NULL)
1862 GNUNET_CONNECTION_notify_transmit_ready_cancel (pos->th);
1865 while (NULL != (creply = pos->res_head))
1867 GNUNET_CONTAINER_DLL_remove (pos->res_head,
1870 GNUNET_free (creply);
1872 GNUNET_SERVER_client_drop (pos->client);
1874 GNUNET_CONTAINER_multihashmap_iterate (connected_peers,
1875 &remove_client_from_last_client_replies,
1881 * Iterator to free peer entries.
1883 * @param cls closure, unused
1884 * @param key current key code
1885 * @param value value in the hash map (peer entry)
1886 * @return GNUNET_YES (we should continue to iterate)
1889 clean_peer (void *cls,
1890 const GNUNET_HashCode * key,
1893 peer_disconnect_handler (NULL, (const struct GNUNET_PeerIdentity*) key);
1899 * Task run during shutdown.
1905 shutdown_task (void *cls,
1906 const struct GNUNET_SCHEDULER_TaskContext *tc)
1910 GNUNET_DATASTORE_cancel (mig_qe);
1915 GNUNET_DATASTORE_cancel (dht_qe);
1918 if (GNUNET_SCHEDULER_NO_TASK != mig_task)
1920 GNUNET_SCHEDULER_cancel (sched, mig_task);
1921 mig_task = GNUNET_SCHEDULER_NO_TASK;
1923 if (GNUNET_SCHEDULER_NO_TASK != dht_task)
1925 GNUNET_SCHEDULER_cancel (sched, dht_task);
1926 dht_task = GNUNET_SCHEDULER_NO_TASK;
1928 while (client_list != NULL)
1929 handle_client_disconnect (NULL,
1930 client_list->client);
1931 cron_flush_trust (NULL, NULL);
1932 GNUNET_CONTAINER_multihashmap_iterate (connected_peers,
1935 GNUNET_break (0 == GNUNET_CONTAINER_heap_get_size (requests_by_expiration_heap));
1936 GNUNET_CONTAINER_heap_destroy (requests_by_expiration_heap);
1937 requests_by_expiration_heap = 0;
1938 GNUNET_CONTAINER_multihashmap_destroy (connected_peers);
1939 connected_peers = NULL;
1940 GNUNET_break (0 == GNUNET_CONTAINER_multihashmap_size (query_request_map));
1941 GNUNET_CONTAINER_multihashmap_destroy (query_request_map);
1942 query_request_map = NULL;
1943 GNUNET_LOAD_value_free (rt_entry_lifetime);
1944 rt_entry_lifetime = NULL;
1945 GNUNET_break (0 == GNUNET_CONTAINER_multihashmap_size (peer_request_map));
1946 GNUNET_CONTAINER_multihashmap_destroy (peer_request_map);
1947 peer_request_map = NULL;
1948 GNUNET_assert (NULL != core);
1949 GNUNET_CORE_disconnect (core);
1953 GNUNET_STATISTICS_destroy (stats, GNUNET_NO);
1958 GNUNET_DATASTORE_disconnect (dsh,
1962 while (mig_head != NULL)
1963 delete_migration_block (mig_head);
1964 GNUNET_assert (0 == mig_size);
1965 GNUNET_DHT_disconnect (dht_handle);
1967 GNUNET_LOAD_value_free (datastore_get_load);
1968 datastore_get_load = NULL;
1969 GNUNET_LOAD_value_free (datastore_put_load);
1970 datastore_put_load = NULL;
1971 GNUNET_BLOCK_context_destroy (block_ctx);
1973 GNUNET_CONFIGURATION_destroy (block_cfg);
1977 GNUNET_free_non_null (trustDirectory);
1978 trustDirectory = NULL;
1982 /* ******************* Utility functions ******************** */
1986 * We've had to delay a request for transmission to core, but now
1987 * we should be ready. Run it.
1989 * @param cls the 'struct ConnectedPeer' for which a request was delayed
1990 * @param tc task context (unused)
1993 delayed_transmission_request (void *cls,
1994 const struct GNUNET_SCHEDULER_TaskContext *tc)
1996 struct ConnectedPeer *cp = cls;
1997 struct GNUNET_PeerIdentity pid;
1998 struct PendingMessage *pm;
2000 pm = cp->pending_messages_head;
2001 cp->delayed_transmission_request_task = GNUNET_SCHEDULER_NO_TASK;
2002 GNUNET_assert (cp->cth == NULL);
2005 GNUNET_PEER_resolve (cp->pid,
2007 cp->last_transmission_request_start = GNUNET_TIME_absolute_get ();
2008 cp->cth = GNUNET_CORE_notify_transmit_ready (core,
2010 GNUNET_CONSTANTS_SERVICE_TIMEOUT,
2019 * Transmit messages by copying it to the target buffer
2020 * "buf". "buf" will be NULL and "size" zero if the socket was closed
2021 * for writing in the meantime. In that case, do nothing
2022 * (the disconnect or shutdown handler will take care of the rest).
2023 * If we were able to transmit messages and there are still more
2024 * pending, ask core again for further calls to this function.
2026 * @param cls closure, pointer to the 'struct ConnectedPeer*'
2027 * @param size number of bytes available in buf
2028 * @param buf where the callee should write the message
2029 * @return number of bytes written to buf
2032 transmit_to_peer (void *cls,
2033 size_t size, void *buf)
2035 struct ConnectedPeer *cp = cls;
2037 struct PendingMessage *pm;
2038 struct PendingMessage *next_pm;
2039 struct GNUNET_TIME_Absolute now;
2040 struct GNUNET_TIME_Relative min_delay;
2041 struct MigrationReadyBlock *mb;
2042 struct MigrationReadyBlock *next;
2043 struct PutMessage migm;
2046 struct GNUNET_PeerIdentity pid;
2052 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2053 "Dropping message, core too busy.\n");
2055 GNUNET_LOAD_update (cp->transmission_delay,
2059 GNUNET_LOAD_update (cp->transmission_delay,
2060 GNUNET_TIME_absolute_get_duration (cp->last_transmission_request_start).value);
2061 now = GNUNET_TIME_absolute_get ();
2063 min_delay = GNUNET_TIME_UNIT_FOREVER_REL;
2064 next_pm = cp->pending_messages_head;
2065 while ( (NULL != (pm = next_pm) ) &&
2066 (pm->msize <= size) )
2069 if (pm->delay_until.value > now.value)
2071 min_delay = GNUNET_TIME_relative_min (min_delay,
2072 GNUNET_TIME_absolute_get_remaining (pm->delay_until));
2075 memcpy (&cbuf[msize], &pm[1], pm->msize);
2078 if (NULL == pm->pml)
2080 GNUNET_CONTAINER_DLL_remove (cp->pending_messages_head,
2081 cp->pending_messages_tail,
2083 cp->pending_requests--;
2085 destroy_pending_message (pm, cp->pid);
2088 min_delay = GNUNET_TIME_UNIT_ZERO;
2089 if (NULL != cp->pending_messages_head)
2091 GNUNET_assert (GNUNET_SCHEDULER_NO_TASK == cp->delayed_transmission_request_task);
2092 cp->delayed_transmission_request_task
2093 = GNUNET_SCHEDULER_add_delayed (sched,
2095 &delayed_transmission_request,
2100 GNUNET_PEER_resolve (cp->pid,
2103 while (NULL != (mb = next))
2106 for (i=0;i<MIGRATION_LIST_SIZE;i++)
2108 if ( (cp->pid == mb->target_list[i]) &&
2109 (mb->size + sizeof (migm) <= size) )
2111 GNUNET_PEER_change_rc (mb->target_list[i], -1);
2112 mb->target_list[i] = 0;
2114 memset (&migm, 0, sizeof (migm));
2115 migm.header.size = htons (sizeof (migm) + mb->size);
2116 migm.header.type = htons (GNUNET_MESSAGE_TYPE_FS_PUT);
2117 migm.type = htonl (mb->type);
2118 migm.expiration = GNUNET_TIME_absolute_hton (mb->expiration);
2119 memcpy (&cbuf[msize], &migm, sizeof (migm));
2120 msize += sizeof (migm);
2121 size -= sizeof (migm);
2122 memcpy (&cbuf[msize], &mb[1], mb->size);
2126 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2127 "Pushing migration block `%s' (%u bytes) to `%s'\n",
2128 GNUNET_h2s (&mb->query),
2129 (unsigned int) mb->size,
2137 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2138 "Migration block `%s' (%u bytes) is not on migration list for peer `%s'\n",
2139 GNUNET_h2s (&mb->query),
2140 (unsigned int) mb->size,
2145 if ( (mb->used_targets >= MIGRATION_TARGET_COUNT) ||
2146 (mb->used_targets >= GNUNET_CONTAINER_multihashmap_size (connected_peers)) )
2148 delete_migration_block (mb);
2149 consider_migration_gathering ();
2152 consider_migration (NULL,
2157 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2158 "Transmitting %u bytes to peer with PID %u\n",
2159 (unsigned int) msize,
2160 (unsigned int) cp->pid);
2167 * Add a message to the set of pending messages for the given peer.
2169 * @param cp peer to send message to
2170 * @param pm message to queue
2171 * @param pr request on which behalf this message is being queued
2174 add_to_pending_messages_for_peer (struct ConnectedPeer *cp,
2175 struct PendingMessage *pm,
2176 struct PendingRequest *pr)
2178 struct PendingMessage *pos;
2179 struct PendingMessageList *pml;
2180 struct GNUNET_PeerIdentity pid;
2182 GNUNET_assert (pm->next == NULL);
2183 GNUNET_assert (pm->pml == NULL);
2186 pml = GNUNET_malloc (sizeof (struct PendingMessageList));
2191 GNUNET_CONTAINER_DLL_insert (pr->pending_head,
2195 pos = cp->pending_messages_head;
2196 while ( (pos != NULL) &&
2197 (pm->priority < pos->priority) )
2199 GNUNET_CONTAINER_DLL_insert_after (cp->pending_messages_head,
2200 cp->pending_messages_tail,
2203 cp->pending_requests++;
2204 if (cp->pending_requests > MAX_QUEUE_PER_PEER)
2206 GNUNET_STATISTICS_update (stats,
2207 gettext_noop ("# P2P searches discarded (queue length bound)"),
2210 destroy_pending_message (cp->pending_messages_tail, 0);
2212 GNUNET_PEER_resolve (cp->pid, &pid);
2213 if (NULL != cp->cth)
2215 GNUNET_CORE_notify_transmit_ready_cancel (cp->cth);
2218 if (cp->delayed_transmission_request_task != GNUNET_SCHEDULER_NO_TASK)
2220 GNUNET_SCHEDULER_cancel (sched, cp->delayed_transmission_request_task);
2221 cp->delayed_transmission_request_task = GNUNET_SCHEDULER_NO_TASK;
2223 /* need to schedule transmission */
2224 cp->last_transmission_request_start = GNUNET_TIME_absolute_get ();
2225 cp->cth = GNUNET_CORE_notify_transmit_ready (core,
2226 cp->pending_messages_head->priority,
2229 cp->pending_messages_head->msize,
2232 if (cp->cth == NULL)
2235 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2236 "Failed to schedule transmission with core!\n");
2238 GNUNET_STATISTICS_update (stats,
2239 gettext_noop ("# CORE transmission failures"),
2247 * Test if the DATABASE (GET) load on this peer is too high
2248 * to even consider processing the query at
2251 * @return GNUNET_YES if the load is too high to do anything (load high)
2252 * GNUNET_NO to process normally (load normal)
2253 * GNUNET_SYSERR to process for free (load low)
2256 test_get_load_too_high (uint32_t priority)
2260 ld = GNUNET_LOAD_get_load (datastore_get_load);
2263 GNUNET_STATISTICS_update (stats,
2264 gettext_noop ("# requests done for free (low load)"),
2267 return GNUNET_SYSERR;
2271 GNUNET_STATISTICS_update (stats,
2272 gettext_noop ("# requests done for a price (normal load)"),
2277 GNUNET_STATISTICS_update (stats,
2278 gettext_noop ("# requests dropped due to high load"),
2288 * Test if the DATABASE (PUT) load on this peer is too high
2289 * to even consider processing the query at
2292 * @return GNUNET_YES if the load is too high to do anything (load high)
2293 * GNUNET_NO to process normally (load normal or low)
2296 test_put_load_too_high (uint32_t priority)
2300 if (GNUNET_LOAD_get_average (datastore_put_load) < 50)
2301 return GNUNET_NO; /* very fast */
2302 ld = GNUNET_LOAD_get_load (datastore_put_load);
2303 if (ld < 2.0 * (1 + priority))
2305 GNUNET_STATISTICS_update (stats,
2306 gettext_noop ("# storage requests dropped due to high load"),
2313 /* ******************* Pending Request Refresh Task ******************** */
2318 * We use a random delay to make the timing of requests less
2319 * predictable. This function returns such a random delay. We add a base
2320 * delay of MAX_CORK_DELAY (1s).
2322 * FIXME: make schedule dependent on the specifics of the request?
2323 * Or bandwidth and number of connected peers and load?
2325 * @return random delay to use for some request, between 1s and 1000+TTL_DECREMENT ms
2327 static struct GNUNET_TIME_Relative
2328 get_processing_delay ()
2331 GNUNET_TIME_relative_add (GNUNET_CONSTANTS_MAX_CORK_DELAY,
2332 GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MILLISECONDS,
2333 GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
2339 * We're processing a GET request from another peer and have decided
2340 * to forward it to other peers. This function is called periodically
2341 * and should forward the request to other peers until we have all
2342 * possible replies. If we have transmitted the *only* reply to
2343 * the initiator we should destroy the pending request. If we have
2344 * many replies in the queue to the initiator, we should delay sending
2345 * out more queries until the reply queue has shrunk some.
2347 * @param cls our "struct ProcessGetContext *"
2351 forward_request_task (void *cls,
2352 const struct GNUNET_SCHEDULER_TaskContext *tc);
2356 * Function called after we either failed or succeeded
2357 * at transmitting a query to a peer.
2359 * @param cls the requests "struct PendingRequest*"
2360 * @param tpid ID of receiving peer, 0 on transmission error
2363 transmit_query_continuation (void *cls,
2364 GNUNET_PEER_Id tpid)
2366 struct PendingRequest *pr = cls;
2369 GNUNET_STATISTICS_update (stats,
2370 gettext_noop ("# queries scheduled for forwarding"),
2376 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2377 "Transmission of request failed, will try again later.\n");
2379 if (pr->task == GNUNET_SCHEDULER_NO_TASK)
2380 pr->task = GNUNET_SCHEDULER_add_delayed (sched,
2381 get_processing_delay (),
2382 &forward_request_task,
2387 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2388 "Transmitted query `%s'\n",
2389 GNUNET_h2s (&pr->query));
2391 GNUNET_STATISTICS_update (stats,
2392 gettext_noop ("# queries forwarded"),
2395 for (i=0;i<pr->used_targets_off;i++)
2396 if (pr->used_targets[i].pid == tpid)
2397 break; /* found match! */
2398 if (i == pr->used_targets_off)
2400 /* need to create new entry */
2401 if (pr->used_targets_off == pr->used_targets_size)
2402 GNUNET_array_grow (pr->used_targets,
2403 pr->used_targets_size,
2404 pr->used_targets_size * 2 + 2);
2405 GNUNET_PEER_change_rc (tpid, 1);
2406 pr->used_targets[pr->used_targets_off].pid = tpid;
2407 pr->used_targets[pr->used_targets_off].num_requests = 0;
2408 i = pr->used_targets_off++;
2410 pr->used_targets[i].last_request_time = GNUNET_TIME_absolute_get ();
2411 pr->used_targets[i].num_requests++;
2412 if (pr->task == GNUNET_SCHEDULER_NO_TASK)
2413 pr->task = GNUNET_SCHEDULER_add_delayed (sched,
2414 get_processing_delay (),
2415 &forward_request_task,
2421 * How many bytes should a bloomfilter be if we have already seen
2422 * entry_count responses? Note that BLOOMFILTER_K gives us the number
2423 * of bits set per entry. Furthermore, we should not re-size the
2424 * filter too often (to keep it cheap).
2426 * Since other peers will also add entries but not resize the filter,
2427 * we should generally pick a slightly larger size than what the
2428 * strict math would suggest.
2430 * @return must be a power of two and smaller or equal to 2^15.
2433 compute_bloomfilter_size (unsigned int entry_count)
2436 unsigned int ideal = (entry_count * BLOOMFILTER_K) / 4;
2437 uint16_t max = 1 << 15;
2439 if (entry_count > max)
2442 while ((size < max) && (size < ideal))
2451 * Recalculate our bloom filter for filtering replies. This function
2452 * will create a new bloom filter from scratch, so it should only be
2453 * called if we have no bloomfilter at all (and hence can create a
2454 * fresh one of minimal size without problems) OR if our peer is the
2455 * initiator (in which case we may resize to larger than mimimum size).
2457 * @param pr request for which the BF is to be recomputed
2460 refresh_bloomfilter (struct PendingRequest *pr)
2464 GNUNET_HashCode mhash;
2466 nsize = compute_bloomfilter_size (pr->replies_seen_off);
2467 if (nsize == pr->bf_size)
2468 return; /* size not changed */
2470 GNUNET_CONTAINER_bloomfilter_free (pr->bf);
2471 pr->bf_size = nsize;
2472 pr->mingle = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, -1);
2473 pr->bf = GNUNET_CONTAINER_bloomfilter_init (NULL,
2476 for (i=0;i<pr->replies_seen_off;i++)
2478 GNUNET_BLOCK_mingle_hash (&pr->replies_seen[i],
2481 GNUNET_CONTAINER_bloomfilter_add (pr->bf, &mhash);
2487 * Function called after we've tried to reserve a certain amount of
2488 * bandwidth for a reply. Check if we succeeded and if so send our
2491 * @param cls the requests "struct PendingRequest*"
2492 * @param peer identifies the peer
2493 * @param bpm_in set to the current bandwidth limit (receiving) for this peer
2494 * @param bpm_out set to the current bandwidth limit (sending) for this peer
2495 * @param amount set to the amount that was actually reserved or unreserved
2496 * @param preference current traffic preference for the given peer
2499 target_reservation_cb (void *cls,
2501 GNUNET_PeerIdentity * peer,
2502 struct GNUNET_BANDWIDTH_Value32NBO bpm_in,
2503 struct GNUNET_BANDWIDTH_Value32NBO bpm_out,
2505 uint64_t preference)
2507 struct PendingRequest *pr = cls;
2508 struct ConnectedPeer *cp;
2509 struct PendingMessage *pm;
2510 struct GetMessage *gm;
2511 GNUNET_HashCode *ext;
2522 /* error in communication with core, try again later */
2523 if (pr->task == GNUNET_SCHEDULER_NO_TASK)
2524 pr->task = GNUNET_SCHEDULER_add_delayed (sched,
2525 get_processing_delay (),
2526 &forward_request_task,
2530 /* (3) transmit, update ttl/priority */
2531 cp = GNUNET_CONTAINER_multihashmap_get (connected_peers,
2535 /* Peer must have just left */
2537 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2538 "Selected peer disconnected!\n");
2540 if (pr->task == GNUNET_SCHEDULER_NO_TASK)
2541 pr->task = GNUNET_SCHEDULER_add_delayed (sched,
2542 get_processing_delay (),
2543 &forward_request_task,
2547 no_route = GNUNET_NO;
2553 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2554 "Failed to reserve bandwidth for reply (got %d/%u bytes only)!\n",
2558 GNUNET_STATISTICS_update (stats,
2559 gettext_noop ("# reply bandwidth reservation requests failed"),
2562 if (pr->task == GNUNET_SCHEDULER_NO_TASK)
2563 pr->task = GNUNET_SCHEDULER_add_delayed (sched,
2564 get_processing_delay (),
2565 &forward_request_task,
2567 return; /* this target round failed */
2569 no_route = GNUNET_YES;
2572 GNUNET_STATISTICS_update (stats,
2573 gettext_noop ("# queries scheduled for forwarding"),
2576 for (i=0;i<pr->used_targets_off;i++)
2577 if (pr->used_targets[i].pid == cp->pid)
2579 GNUNET_STATISTICS_update (stats,
2580 gettext_noop ("# queries retransmitted to same target"),
2586 /* build message and insert message into priority queue */
2588 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2589 "Forwarding request `%s' to `%4s'!\n",
2590 GNUNET_h2s (&pr->query),
2595 if (GNUNET_YES == no_route)
2597 bm |= GET_MESSAGE_BIT_RETURN_TO;
2600 if (pr->namespace != NULL)
2602 bm |= GET_MESSAGE_BIT_SKS_NAMESPACE;
2605 if (pr->target_pid != 0)
2607 bm |= GET_MESSAGE_BIT_TRANSMIT_TO;
2610 msize = sizeof (struct GetMessage) + pr->bf_size + k * sizeof(GNUNET_HashCode);
2611 GNUNET_assert (msize < GNUNET_SERVER_MAX_MESSAGE_SIZE);
2612 pm = GNUNET_malloc (sizeof (struct PendingMessage) + msize);
2614 gm = (struct GetMessage*) &pm[1];
2615 gm->header.type = htons (GNUNET_MESSAGE_TYPE_FS_GET);
2616 gm->header.size = htons (msize);
2617 gm->type = htonl (pr->type);
2618 pr->remaining_priority /= 2;
2619 gm->priority = htonl (pr->remaining_priority);
2620 gm->ttl = htonl (pr->ttl);
2621 gm->filter_mutator = htonl(pr->mingle);
2622 gm->hash_bitmap = htonl (bm);
2623 gm->query = pr->query;
2624 ext = (GNUNET_HashCode*) &gm[1];
2626 if (GNUNET_YES == no_route)
2627 GNUNET_PEER_resolve (pr->cp->pid, (struct GNUNET_PeerIdentity*) &ext[k++]);
2628 if (pr->namespace != NULL)
2629 memcpy (&ext[k++], pr->namespace, sizeof (GNUNET_HashCode));
2630 if (pr->target_pid != 0)
2631 GNUNET_PEER_resolve (pr->target_pid, (struct GNUNET_PeerIdentity*) &ext[k++]);
2632 bfdata = (char *) &ext[k];
2634 GNUNET_CONTAINER_bloomfilter_get_raw_data (pr->bf,
2637 pm->cont = &transmit_query_continuation;
2639 cp->last_request_times[(cp->last_request_times_off++) % MAX_QUEUE_PER_PEER] = GNUNET_TIME_absolute_get ();
2640 add_to_pending_messages_for_peer (cp, pm, pr);
2645 * Closure used for "target_peer_select_cb".
2647 struct PeerSelectionContext
2650 * The request for which we are selecting
2653 struct PendingRequest *pr;
2656 * Current "prime" target.
2658 struct GNUNET_PeerIdentity target;
2661 * How much do we like this target?
2663 double target_score;
2669 * Function called for each connected peer to determine
2670 * which one(s) would make good targets for forwarding.
2672 * @param cls closure (struct PeerSelectionContext)
2673 * @param key current key code (peer identity)
2674 * @param value value in the hash map (struct ConnectedPeer)
2675 * @return GNUNET_YES if we should continue to
2680 target_peer_select_cb (void *cls,
2681 const GNUNET_HashCode * key,
2684 struct PeerSelectionContext *psc = cls;
2685 struct ConnectedPeer *cp = value;
2686 struct PendingRequest *pr = psc->pr;
2687 struct GNUNET_TIME_Relative delay;
2692 /* 1) check that this peer is not the initiator */
2696 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2697 "Skipping initiator in forwarding selection\n");
2699 return GNUNET_YES; /* skip */
2702 /* 2) check if we have already (recently) forwarded to this peer */
2703 /* 2a) this particular request */
2705 for (i=0;i<pr->used_targets_off;i++)
2706 if (pr->used_targets[i].pid == cp->pid)
2708 pc = pr->used_targets[i].num_requests;
2709 GNUNET_assert (pc > 0);
2710 if (0 != GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
2711 RETRY_PROBABILITY_INV * pc))
2714 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2715 "NOT re-trying query that was previously transmitted %u times\n",
2718 return GNUNET_YES; /* skip */
2725 GNUNET_log (GNUNET_ERROR_TYPE_INFO,
2726 "Re-trying query that was previously transmitted %u times to this peer\n",
2730 /* 2b) many other requests to this peer */
2731 delay = GNUNET_TIME_absolute_get_duration (cp->last_request_times[cp->last_request_times_off % MAX_QUEUE_PER_PEER]);
2732 if (delay.value <= cp->avg_delay.value)
2735 GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
2736 "NOT sending query since we send %u others to this peer in the last %llums\n",
2738 cp->avg_delay.value);
2740 return GNUNET_YES; /* skip */
2743 /* 3) calculate how much we'd like to forward to this peer,
2744 starting with a random value that is strong enough
2745 to at least give any peer a chance sometimes
2746 (compared to the other factors that come later) */
2747 /* 3a) count successful (recent) routes from cp for same source */
2750 score = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
2751 P2P_SUCCESS_LIST_SIZE);
2752 for (i=0;i<P2P_SUCCESS_LIST_SIZE;i++)
2753 if (cp->last_p2p_replies[i] == pr->cp->pid)
2754 score += 1.0; /* likely successful based on hot path */
2758 score = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
2759 CS2P_SUCCESS_LIST_SIZE);
2760 for (i=0;i<CS2P_SUCCESS_LIST_SIZE;i++)
2761 if (cp->last_client_replies[i] == pr->client_request_list->client_list->client)
2762 score += 1.0; /* likely successful based on hot path */
2764 /* 3b) include latency */
2765 if (cp->avg_delay.value < 4 * TTL_DECREMENT)
2766 score += 1.0; /* likely fast based on latency */
2767 /* 3c) include priorities */
2768 if (cp->avg_priority <= pr->remaining_priority / 2.0)
2769 score += 1.0; /* likely successful based on priorities */
2770 /* 3d) penalize for queue size */
2771 score -= (2.0 * cp->pending_requests / (double) MAX_QUEUE_PER_PEER);
2772 /* 3e) include peer proximity */
2773 score -= (2.0 * (GNUNET_CRYPTO_hash_distance_u32 (key,
2774 &pr->query)) / (double) UINT32_MAX);
2775 /* 4) super-bonus for being the known target */
2776 if (pr->target_pid == cp->pid)
2778 /* store best-fit in closure */
2780 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2781 "Peer `%s' gets score %f for forwarding query, max is %f\n",
2786 score++; /* avoid zero */
2787 if (score > psc->target_score)
2789 psc->target_score = score;
2790 psc->target.hashPubKey = *key;
2797 * The priority level imposes a bound on the maximum
2798 * value for the ttl that can be requested.
2800 * @param ttl_in requested ttl
2801 * @param prio given priority
2802 * @return ttl_in if ttl_in is below the limit,
2803 * otherwise the ttl-limit for the given priority
2806 bound_ttl (int32_t ttl_in, uint32_t prio)
2808 unsigned long long allowed;
2812 allowed = ((unsigned long long) prio) * TTL_DECREMENT / 1000;
2813 if (ttl_in > allowed)
2815 if (allowed >= (1 << 30))
2824 * Iterator called on each result obtained for a DHT
2825 * operation that expects a reply
2827 * @param cls closure
2828 * @param exp when will this value expire
2829 * @param key key of the result
2830 * @param get_path NULL-terminated array of pointers
2831 * to the peers on reverse GET path (or NULL if not recorded)
2832 * @param put_path NULL-terminated array of pointers
2833 * to the peers on the PUT path (or NULL if not recorded)
2834 * @param type type of the result
2835 * @param size number of bytes in data
2836 * @param data pointer to the result data
2839 process_dht_reply (void *cls,
2840 struct GNUNET_TIME_Absolute exp,
2841 const GNUNET_HashCode * key,
2842 const struct GNUNET_PeerIdentity * const *get_path,
2843 const struct GNUNET_PeerIdentity * const *put_path,
2844 enum GNUNET_BLOCK_Type type,
2850 * We're processing a GET request and have decided
2851 * to forward it to other peers. This function is called periodically
2852 * and should forward the request to other peers until we have all
2853 * possible replies. If we have transmitted the *only* reply to
2854 * the initiator we should destroy the pending request. If we have
2855 * many replies in the queue to the initiator, we should delay sending
2856 * out more queries until the reply queue has shrunk some.
2858 * @param cls our "struct ProcessGetContext *"
2862 forward_request_task (void *cls,
2863 const struct GNUNET_SCHEDULER_TaskContext *tc)
2865 struct PendingRequest *pr = cls;
2866 struct PeerSelectionContext psc;
2867 struct ConnectedPeer *cp;
2868 struct GNUNET_TIME_Relative delay;
2870 pr->task = GNUNET_SCHEDULER_NO_TASK;
2871 if (pr->irc != NULL)
2874 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2875 "Forwarding of query `%s' not attempted due to pending local lookup!\n",
2876 GNUNET_h2s (&pr->query));
2878 return; /* already pending */
2880 if (GNUNET_YES == pr->local_only)
2881 return; /* configured to not do P2P search */
2883 if ( (0 == pr->anonymity_level) &&
2884 (GNUNET_YES != pr->forward_only) &&
2885 (pr->type != GNUNET_BLOCK_TYPE_FS_DBLOCK) &&
2886 (pr->type != GNUNET_BLOCK_TYPE_FS_IBLOCK) )
2888 pr->dht_get = GNUNET_DHT_get_start (dht_handle,
2889 GNUNET_TIME_UNIT_FOREVER_REL,
2896 (pr->namespace != NULL) ? sizeof (GNUNET_HashCode) : 0,
2900 /* (1) select target */
2902 psc.target_score = -DBL_MAX;
2903 GNUNET_CONTAINER_multihashmap_iterate (connected_peers,
2904 &target_peer_select_cb,
2906 if (psc.target_score == -DBL_MAX)
2908 delay = get_processing_delay ();
2910 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2911 "No peer selected for forwarding of query `%s', will try again in %llu ms!\n",
2912 GNUNET_h2s (&pr->query),
2915 pr->task = GNUNET_SCHEDULER_add_delayed (sched,
2917 &forward_request_task,
2919 return; /* nobody selected */
2921 /* (3) update TTL/priority */
2922 if (pr->client_request_list != NULL)
2924 /* FIXME: use better algorithm!? */
2925 if (0 == GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
2928 /* bound priority we use by priorities we see from other peers
2929 rounded up (must round up so that we can see non-zero
2930 priorities, but round up as little as possible to make it
2931 plausible that we forwarded another peers request) */
2932 if (pr->priority > current_priorities + 1.0)
2933 pr->priority = (uint32_t) current_priorities + 1.0;
2934 pr->ttl = bound_ttl (pr->ttl + TTL_DECREMENT * 2,
2937 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2938 "Trying query `%s' with priority %u and TTL %d.\n",
2939 GNUNET_h2s (&pr->query),
2945 /* (3) reserve reply bandwidth */
2946 if (GNUNET_NO == pr->forward_only)
2948 cp = GNUNET_CONTAINER_multihashmap_get (connected_peers,
2949 &psc.target.hashPubKey);
2950 GNUNET_assert (NULL != cp);
2951 pr->irc = GNUNET_CORE_peer_change_preference (sched, cfg,
2953 GNUNET_CONSTANTS_SERVICE_TIMEOUT,
2954 GNUNET_BANDWIDTH_value_init (UINT32_MAX),
2957 &target_reservation_cb,
2959 cp->inc_preference = 0;
2963 /* force forwarding */
2964 static struct GNUNET_BANDWIDTH_Value32NBO zerobw;
2965 target_reservation_cb (pr, &psc.target,
2966 zerobw, zerobw, 0, 0.0);
2971 /* **************************** P2P PUT Handling ************************ */
2975 * Function called after we either failed or succeeded
2976 * at transmitting a reply to a peer.
2978 * @param cls the requests "struct PendingRequest*"
2979 * @param tpid ID of receiving peer, 0 on transmission error
2982 transmit_reply_continuation (void *cls,
2983 GNUNET_PEER_Id tpid)
2985 struct PendingRequest *pr = cls;
2989 case GNUNET_BLOCK_TYPE_FS_DBLOCK:
2990 case GNUNET_BLOCK_TYPE_FS_IBLOCK:
2991 /* only one reply expected, done with the request! */
2992 destroy_pending_request (pr);
2994 case GNUNET_BLOCK_TYPE_ANY:
2995 case GNUNET_BLOCK_TYPE_FS_KBLOCK:
2996 case GNUNET_BLOCK_TYPE_FS_SBLOCK:
3006 * Transmit the given message by copying it to the target buffer
3007 * "buf". "buf" will be NULL and "size" zero if the socket was closed
3008 * for writing in the meantime. In that case, do nothing
3009 * (the disconnect or shutdown handler will take care of the rest).
3010 * If we were able to transmit messages and there are still more
3011 * pending, ask core again for further calls to this function.
3013 * @param cls closure, pointer to the 'struct ClientList*'
3014 * @param size number of bytes available in buf
3015 * @param buf where the callee should write the message
3016 * @return number of bytes written to buf
3019 transmit_to_client (void *cls,
3020 size_t size, void *buf)
3022 struct ClientList *cl = cls;
3024 struct ClientResponseMessage *creply;
3031 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
3032 "Not sending reply, client communication problem.\n");
3037 while ( (NULL != (creply = cl->res_head) ) &&
3038 (creply->msize <= size) )
3040 memcpy (&cbuf[msize], &creply[1], creply->msize);
3041 msize += creply->msize;
3042 size -= creply->msize;
3043 GNUNET_CONTAINER_DLL_remove (cl->res_head,
3046 GNUNET_free (creply);
3049 cl->th = GNUNET_SERVER_notify_transmit_ready (cl->client,
3051 GNUNET_TIME_UNIT_FOREVER_REL,
3052 &transmit_to_client,
3055 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
3056 "Transmitted %u bytes to client\n",
3057 (unsigned int) msize);
3064 * Closure for "process_reply" function.
3066 struct ProcessReplyClosure
3069 * The data for the reply.
3074 * Who gave us this reply? NULL for local host (or DHT)
3076 struct ConnectedPeer *sender;
3079 * When the reply expires.
3081 struct GNUNET_TIME_Absolute expiration;
3089 * Type of the block.
3091 enum GNUNET_BLOCK_Type type;
3094 * How much was this reply worth to us?
3099 * Evaluation result (returned).
3101 enum GNUNET_BLOCK_EvaluationResult eval;
3104 * Did we finish processing the associated request?
3109 * Did we find a matching request?
3116 * We have received a reply; handle it!
3118 * @param cls response (struct ProcessReplyClosure)
3119 * @param key our query
3120 * @param value value in the hash map (info about the query)
3121 * @return GNUNET_YES (we should continue to iterate)
3124 process_reply (void *cls,
3125 const GNUNET_HashCode * key,
3128 struct ProcessReplyClosure *prq = cls;
3129 struct PendingRequest *pr = value;
3130 struct PendingMessage *reply;
3131 struct ClientResponseMessage *creply;
3132 struct ClientList *cl;
3133 struct PutMessage *pm;
3134 struct ConnectedPeer *cp;
3135 struct GNUNET_TIME_Relative cur_delay;
3137 struct GNUNET_TIME_Relative art_delay;
3143 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
3144 "Matched result (type %u) for query `%s' with pending request\n",
3145 (unsigned int) prq->type,
3148 GNUNET_STATISTICS_update (stats,
3149 gettext_noop ("# replies received and matched"),
3152 if (prq->sender != NULL)
3154 for (i=0;i<pr->used_targets_off;i++)
3155 if (pr->used_targets[i].pid == prq->sender->pid)
3157 if (i < pr->used_targets_off)
3159 cur_delay = GNUNET_TIME_absolute_get_duration (pr->used_targets[i].last_request_time);
3160 prq->sender->avg_delay.value
3161 = (prq->sender->avg_delay.value *
3162 (RUNAVG_DELAY_N - 1) + cur_delay.value) / RUNAVG_DELAY_N;
3163 prq->sender->avg_priority
3164 = (prq->sender->avg_priority *
3165 (RUNAVG_DELAY_N - 1) + pr->priority) / (double) RUNAVG_DELAY_N;
3169 GNUNET_PEER_change_rc (prq->sender->last_p2p_replies
3170 [prq->sender->last_p2p_replies_woff % P2P_SUCCESS_LIST_SIZE],
3172 GNUNET_PEER_change_rc (pr->cp->pid, 1);
3173 prq->sender->last_p2p_replies
3174 [(prq->sender->last_p2p_replies_woff++) % P2P_SUCCESS_LIST_SIZE]
3179 if (NULL != prq->sender->last_client_replies
3180 [(prq->sender->last_client_replies_woff) % CS2P_SUCCESS_LIST_SIZE])
3181 GNUNET_SERVER_client_drop (prq->sender->last_client_replies
3182 [(prq->sender->last_client_replies_woff) % CS2P_SUCCESS_LIST_SIZE]);
3183 prq->sender->last_client_replies
3184 [(prq->sender->last_client_replies_woff++) % CS2P_SUCCESS_LIST_SIZE]
3185 = pr->client_request_list->client_list->client;
3186 GNUNET_SERVER_client_keep (pr->client_request_list->client_list->client);
3189 prq->eval = GNUNET_BLOCK_evaluate (block_ctx,
3194 pr->namespace, (pr->namespace != NULL) ? sizeof (GNUNET_HashCode) : 0,
3199 case GNUNET_BLOCK_EVALUATION_OK_MORE:
3201 case GNUNET_BLOCK_EVALUATION_OK_LAST:
3202 while (NULL != pr->pending_head)
3203 destroy_pending_message_list_entry (pr->pending_head);
3206 if (pr->client_request_list != NULL)
3207 GNUNET_SERVER_receive_done (pr->client_request_list->client_list->client,
3209 GNUNET_DATASTORE_cancel (pr->qe);
3212 pr->do_remove = GNUNET_YES;
3213 if (pr->task != GNUNET_SCHEDULER_NO_TASK)
3215 GNUNET_SCHEDULER_cancel (sched,
3217 pr->task = GNUNET_SCHEDULER_NO_TASK;
3219 GNUNET_break (GNUNET_YES ==
3220 GNUNET_CONTAINER_multihashmap_remove (query_request_map,
3223 GNUNET_LOAD_update (rt_entry_lifetime,
3224 GNUNET_TIME_absolute_get_duration (pr->start_time).value);
3226 case GNUNET_BLOCK_EVALUATION_OK_DUPLICATE:
3227 GNUNET_STATISTICS_update (stats,
3228 gettext_noop ("# duplicate replies discarded (bloomfilter)"),
3232 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
3233 "Duplicate response `%s', discarding.\n",
3234 GNUNET_h2s (&mhash));
3236 return GNUNET_YES; /* duplicate */
3237 case GNUNET_BLOCK_EVALUATION_RESULT_INVALID:
3238 return GNUNET_YES; /* wrong namespace */
3239 case GNUNET_BLOCK_EVALUATION_REQUEST_VALID:
3242 case GNUNET_BLOCK_EVALUATION_REQUEST_INVALID:
3245 case GNUNET_BLOCK_EVALUATION_TYPE_NOT_SUPPORTED:
3246 GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
3247 _("Unsupported block type %u\n"),
3251 if (pr->client_request_list != NULL)
3253 if (pr->replies_seen_size == pr->replies_seen_off)
3254 GNUNET_array_grow (pr->replies_seen,
3255 pr->replies_seen_size,
3256 pr->replies_seen_size * 2 + 4);
3257 GNUNET_CRYPTO_hash (prq->data,
3259 &pr->replies_seen[pr->replies_seen_off++]);
3260 refresh_bloomfilter (pr);
3262 if (NULL == prq->sender)
3265 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
3266 "Found result for query `%s' in local datastore\n",
3269 GNUNET_STATISTICS_update (stats,
3270 gettext_noop ("# results found locally"),
3274 prq->priority += pr->remaining_priority;
3275 pr->remaining_priority = 0;
3276 pr->results_found++;
3277 prq->request_found = GNUNET_YES;
3278 if (NULL != pr->client_request_list)
3280 GNUNET_STATISTICS_update (stats,
3281 gettext_noop ("# replies received for local clients"),
3284 cl = pr->client_request_list->client_list;
3285 msize = sizeof (struct PutMessage) + prq->size;
3286 creply = GNUNET_malloc (msize + sizeof (struct ClientResponseMessage));
3287 creply->msize = msize;
3288 creply->client_list = cl;
3289 GNUNET_CONTAINER_DLL_insert_after (cl->res_head,
3293 pm = (struct PutMessage*) &creply[1];
3294 pm->header.type = htons (GNUNET_MESSAGE_TYPE_FS_PUT);
3295 pm->header.size = htons (msize);
3296 pm->type = htonl (prq->type);
3297 pm->expiration = GNUNET_TIME_absolute_hton (prq->expiration);
3298 memcpy (&pm[1], prq->data, prq->size);
3302 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
3303 "Transmitting result for query `%s' to client\n",
3306 cl->th = GNUNET_SERVER_notify_transmit_ready (cl->client,
3308 GNUNET_TIME_UNIT_FOREVER_REL,
3309 &transmit_to_client,
3312 GNUNET_break (cl->th != NULL);
3315 prq->finished = GNUNET_YES;
3316 destroy_pending_request (pr);
3323 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
3324 "Transmitting result for query `%s' to other peer (PID=%u)\n",
3326 (unsigned int) cp->pid);
3328 GNUNET_STATISTICS_update (stats,
3329 gettext_noop ("# replies received for other peers"),
3332 msize = sizeof (struct PutMessage) + prq->size;
3333 reply = GNUNET_malloc (msize + sizeof (struct PendingMessage));
3334 reply->cont = &transmit_reply_continuation;
3335 reply->cont_cls = pr;
3337 art_delay = GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MILLISECONDS,
3338 GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
3341 = GNUNET_TIME_relative_to_absolute (art_delay);
3342 GNUNET_STATISTICS_update (stats,
3343 gettext_noop ("cummulative artificial delay introduced (ms)"),
3347 reply->msize = msize;
3348 reply->priority = UINT32_MAX; /* send replies first! */
3349 pm = (struct PutMessage*) &reply[1];
3350 pm->header.type = htons (GNUNET_MESSAGE_TYPE_FS_PUT);
3351 pm->header.size = htons (msize);
3352 pm->type = htonl (prq->type);
3353 pm->expiration = GNUNET_TIME_absolute_hton (prq->expiration);
3354 memcpy (&pm[1], prq->data, prq->size);
3355 add_to_pending_messages_for_peer (cp, reply, pr);
3362 * Iterator called on each result obtained for a DHT
3363 * operation that expects a reply
3365 * @param cls closure
3366 * @param exp when will this value expire
3367 * @param key key of the result
3368 * @param get_path NULL-terminated array of pointers
3369 * to the peers on reverse GET path (or NULL if not recorded)
3370 * @param put_path NULL-terminated array of pointers
3371 * to the peers on the PUT path (or NULL if not recorded)
3372 * @param type type of the result
3373 * @param size number of bytes in data
3374 * @param data pointer to the result data
3377 process_dht_reply (void *cls,
3378 struct GNUNET_TIME_Absolute exp,
3379 const GNUNET_HashCode * key,
3380 const struct GNUNET_PeerIdentity * const *get_path,
3381 const struct GNUNET_PeerIdentity * const *put_path,
3382 enum GNUNET_BLOCK_Type type,
3386 struct PendingRequest *pr = cls;
3387 struct ProcessReplyClosure prq;
3389 memset (&prq, 0, sizeof (prq));
3391 prq.expiration = exp;
3394 process_reply (&prq, key, pr);
3400 * Continuation called to notify client about result of the
3403 * @param cls closure
3404 * @param success GNUNET_SYSERR on failure
3405 * @param msg NULL on success, otherwise an error message
3408 put_migration_continuation (void *cls,
3412 struct GNUNET_TIME_Absolute *start = cls;
3413 struct GNUNET_TIME_Relative delay;
3415 delay = GNUNET_TIME_absolute_get_duration (*start);
3416 GNUNET_free (start);
3417 GNUNET_LOAD_update (datastore_put_load,
3419 if (GNUNET_OK == success)
3421 GNUNET_STATISTICS_update (stats,
3422 gettext_noop ("# datastore 'put' failures"),
3429 * Handle P2P "PUT" message.
3431 * @param cls closure, always NULL
3432 * @param other the other peer involved (sender or receiver, NULL
3433 * for loopback messages where we are both sender and receiver)
3434 * @param message the actual message
3435 * @param latency reported latency of the connection with 'other'
3436 * @param distance reported distance (DV) to 'other'
3437 * @return GNUNET_OK to keep the connection open,
3438 * GNUNET_SYSERR to close it (signal serious error)
3441 handle_p2p_put (void *cls,
3442 const struct GNUNET_PeerIdentity *other,
3443 const struct GNUNET_MessageHeader *message,
3444 struct GNUNET_TIME_Relative latency,
3447 const struct PutMessage *put;
3450 enum GNUNET_BLOCK_Type type;
3451 struct GNUNET_TIME_Absolute expiration;
3452 GNUNET_HashCode query;
3453 struct ProcessReplyClosure prq;
3454 struct GNUNET_TIME_Absolute *start;
3455 struct GNUNET_TIME_Relative block_time;
3457 struct ConnectedPeer *cp;
3458 struct PendingMessage *pm;
3459 struct MigrationStopMessage *msm;
3461 msize = ntohs (message->size);
3462 if (msize < sizeof (struct PutMessage))
3465 return GNUNET_SYSERR;
3467 put = (const struct PutMessage*) message;
3468 dsize = msize - sizeof (struct PutMessage);
3469 type = ntohl (put->type);
3470 expiration = GNUNET_TIME_absolute_ntoh (put->expiration);
3472 if (type == GNUNET_BLOCK_TYPE_FS_ONDEMAND)
3473 return GNUNET_SYSERR;
3475 GNUNET_BLOCK_get_key (block_ctx,
3481 GNUNET_break_op (0);
3482 return GNUNET_SYSERR;
3485 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
3486 "Received result for query `%s' from peer `%4s'\n",
3487 GNUNET_h2s (&query),
3488 GNUNET_i2s (other));
3490 GNUNET_STATISTICS_update (stats,
3491 gettext_noop ("# replies received (overall)"),
3494 /* now, lookup 'query' */
3495 prq.data = (const void*) &put[1];
3497 prq.sender = GNUNET_CONTAINER_multihashmap_get (connected_peers,
3498 &other->hashPubKey);
3503 prq.expiration = expiration;
3505 prq.finished = GNUNET_NO;
3506 prq.request_found = GNUNET_NO;
3507 GNUNET_CONTAINER_multihashmap_get_multiple (query_request_map,
3511 if (prq.sender != NULL)
3513 prq.sender->inc_preference += CONTENT_BANDWIDTH_VALUE + 1000 * prq.priority;
3514 prq.sender->trust += prq.priority;
3516 if ( (GNUNET_YES == active_migration) &&
3517 (GNUNET_NO == test_put_load_too_high (prq.priority)) )
3520 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
3521 "Replicating result for query `%s' with priority %u\n",
3522 GNUNET_h2s (&query),
3525 start = GNUNET_malloc (sizeof (struct GNUNET_TIME_Absolute));
3526 *start = GNUNET_TIME_absolute_get ();
3527 GNUNET_DATASTORE_put (dsh,
3528 0, &query, dsize, &put[1],
3529 type, prq.priority, 1 /* anonymity */,
3531 1 + prq.priority, MAX_DATASTORE_QUEUE,
3532 GNUNET_CONSTANTS_SERVICE_TIMEOUT,
3533 &put_migration_continuation,
3536 putl = GNUNET_LOAD_get_load (datastore_put_load);
3537 if ( (GNUNET_NO == prq.request_found) &&
3538 ( (GNUNET_YES != active_migration) ||
3539 (putl > 2.5 * (1 + prq.priority)) ) )
3541 cp = GNUNET_CONTAINER_multihashmap_get (connected_peers,
3542 &other->hashPubKey);
3543 if (GNUNET_TIME_absolute_get_duration (cp->last_migration_block).value < 5000)
3544 return GNUNET_OK; /* already blocked */
3545 /* We're too busy; send MigrationStop message! */
3546 if (GNUNET_YES != active_migration)
3547 putl = 1.0 + GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, 5);
3548 block_time = GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MILLISECONDS,
3549 5000 + GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
3550 (unsigned int) (60000 * putl * putl)));
3552 cp->last_migration_block = GNUNET_TIME_relative_to_absolute (block_time);
3553 pm = GNUNET_malloc (sizeof (struct PendingMessage) +
3554 sizeof (struct MigrationStopMessage));
3555 pm->msize = sizeof (struct MigrationStopMessage);
3556 pm->priority = UINT32_MAX;
3557 msm = (struct MigrationStopMessage*) &pm[1];
3558 msm->header.size = htons (sizeof (struct MigrationStopMessage));
3559 msm->header.type = htons (GNUNET_MESSAGE_TYPE_FS_MIGRATION_STOP);
3560 msm->duration = GNUNET_TIME_relative_hton (block_time);
3561 add_to_pending_messages_for_peer (cp,
3570 * Handle P2P "MIGRATION_STOP" message.
3572 * @param cls closure, always NULL
3573 * @param other the other peer involved (sender or receiver, NULL
3574 * for loopback messages where we are both sender and receiver)
3575 * @param message the actual message
3576 * @param latency reported latency of the connection with 'other'
3577 * @param distance reported distance (DV) to 'other'
3578 * @return GNUNET_OK to keep the connection open,
3579 * GNUNET_SYSERR to close it (signal serious error)
3582 handle_p2p_migration_stop (void *cls,
3583 const struct GNUNET_PeerIdentity *other,
3584 const struct GNUNET_MessageHeader *message,
3585 struct GNUNET_TIME_Relative latency,
3588 struct ConnectedPeer *cp;
3589 const struct MigrationStopMessage *msm;
3591 msm = (const struct MigrationStopMessage*) message;
3592 cp = GNUNET_CONTAINER_multihashmap_get (connected_peers,
3593 &other->hashPubKey);
3599 cp->migration_blocked = GNUNET_TIME_relative_to_absolute (GNUNET_TIME_relative_ntoh (msm->duration));
3605 /* **************************** P2P GET Handling ************************ */
3609 * Closure for 'check_duplicate_request_{peer,client}'.
3611 struct CheckDuplicateRequestClosure
3614 * The new request we should check if it already exists.
3616 const struct PendingRequest *pr;
3619 * Existing request found by the checker, NULL if none.
3621 struct PendingRequest *have;
3626 * Iterator over entries in the 'query_request_map' that
3627 * tries to see if we have the same request pending from
3628 * the same client already.
3630 * @param cls closure (our 'struct CheckDuplicateRequestClosure')
3631 * @param key current key code (query, ignored, must match)
3632 * @param value value in the hash map (a 'struct PendingRequest'
3633 * that already exists)
3634 * @return GNUNET_YES if we should continue to
3635 * iterate (no match yet)
3636 * GNUNET_NO if not (match found).
3639 check_duplicate_request_client (void *cls,
3640 const GNUNET_HashCode * key,
3643 struct CheckDuplicateRequestClosure *cdc = cls;
3644 struct PendingRequest *have = value;
3646 if (have->client_request_list == NULL)
3648 if ( (cdc->pr->client_request_list->client_list->client == have->client_request_list->client_list->client) &&
3659 * We're processing (local) results for a search request
3660 * from another peer. Pass applicable results to the
3661 * peer and if we are done either clean up (operation
3662 * complete) or forward to other peers (more results possible).
3664 * @param cls our closure (struct LocalGetContext)
3665 * @param key key for the content
3666 * @param size number of bytes in data
3667 * @param data content stored
3668 * @param type type of the content
3669 * @param priority priority of the content
3670 * @param anonymity anonymity-level for the content
3671 * @param expiration expiration time for the content
3672 * @param uid unique identifier for the datum;
3673 * maybe 0 if no unique identifier is available
3676 process_local_reply (void *cls,
3677 const GNUNET_HashCode * key,
3680 enum GNUNET_BLOCK_Type type,
3683 struct GNUNET_TIME_Absolute
3687 struct PendingRequest *pr = cls;
3688 struct ProcessReplyClosure prq;
3689 struct CheckDuplicateRequestClosure cdrc;
3690 GNUNET_HashCode query;
3691 unsigned int old_rf;
3696 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
3697 "Done processing local replies, forwarding request to other peers.\n");
3700 if (pr->client_request_list != NULL)
3702 GNUNET_SERVER_receive_done (pr->client_request_list->client_list->client,
3704 /* Figure out if this is a duplicate request and possibly
3705 merge 'struct PendingRequest' entries */
3708 GNUNET_CONTAINER_multihashmap_get_multiple (query_request_map,
3710 &check_duplicate_request_client,
3712 if (cdrc.have != NULL)
3715 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
3716 "Received request for block `%s' twice from client, will only request once.\n",
3717 GNUNET_h2s (&pr->query));
3720 destroy_pending_request (pr);
3724 if (pr->local_only == GNUNET_YES)
3726 destroy_pending_request (pr);
3729 /* no more results */
3730 if (pr->task == GNUNET_SCHEDULER_NO_TASK)
3731 pr->task = GNUNET_SCHEDULER_add_now (sched,
3732 &forward_request_task,
3737 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
3738 "New local response to `%s' of type %u.\n",
3742 if (type == GNUNET_BLOCK_TYPE_FS_ONDEMAND)
3745 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
3746 "Found ONDEMAND block, performing on-demand encoding\n");
3748 GNUNET_STATISTICS_update (stats,
3749 gettext_noop ("# on-demand blocks matched requests"),
3753 GNUNET_FS_handle_on_demand_block (key, size, data, type, priority,
3754 anonymity, expiration, uid,
3755 &process_local_reply,
3759 GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
3763 old_rf = pr->results_found;
3764 memset (&prq, 0, sizeof (prq));
3766 prq.expiration = expiration;
3769 GNUNET_BLOCK_get_key (block_ctx,
3776 GNUNET_DATASTORE_remove (dsh,
3780 GNUNET_TIME_UNIT_FOREVER_REL,
3782 GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
3786 prq.priority = priority;
3787 prq.finished = GNUNET_NO;
3788 prq.request_found = GNUNET_NO;
3789 if ( (old_rf == 0) &&
3790 (pr->results_found == 0) )
3791 update_datastore_delays (pr->start_time);
3792 process_reply (&prq, key, pr);
3793 if (prq.finished == GNUNET_YES)
3796 return; /* done here */
3797 if (prq.eval == GNUNET_BLOCK_EVALUATION_OK_LAST)
3799 pr->local_only = GNUNET_YES; /* do not forward */
3800 GNUNET_DATASTORE_get_next (dsh, GNUNET_NO);
3803 if ( (pr->client_request_list == NULL) &&
3804 ( (GNUNET_YES == test_get_load_too_high (0)) ||
3805 (pr->results_found > 5 + 2 * pr->priority) ) )
3808 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
3809 "Load too high, done with request\n");
3811 GNUNET_STATISTICS_update (stats,
3812 gettext_noop ("# processing result set cut short due to load"),
3815 /* FIXME: if this is activated, we might stall large downloads
3816 indefinitely since (presumably) the load can never go down again! */
3818 GNUNET_DATASTORE_get_next (dsh, GNUNET_NO);
3822 GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
3827 * We've received a request with the specified priority. Bound it
3828 * according to how much we trust the given peer.
3830 * @param prio_in requested priority
3831 * @param cp the peer making the request
3832 * @return effective priority
3835 bound_priority (uint32_t prio_in,
3836 struct ConnectedPeer *cp)
3838 #define N ((double)128.0)
3843 ld = test_get_load_too_high (0);
3844 if (ld == GNUNET_SYSERR)
3845 return 0; /* excess resources */
3846 ret = change_host_trust (cp, prio_in);
3849 if (ret > current_priorities + N)
3850 rret = current_priorities + N;
3854 = (current_priorities * (N-1) + rret)/N;
3856 if ( (ld == GNUNET_YES) && (ret > 0) )
3858 /* try with charging */
3859 ld = test_get_load_too_high (ret);
3861 if (ld == GNUNET_YES)
3865 change_host_trust (cp, -ret);
3866 return -1; /* not enough resources */
3874 * Iterator over entries in the 'query_request_map' that
3875 * tries to see if we have the same request pending from
3876 * the same peer already.
3878 * @param cls closure (our 'struct CheckDuplicateRequestClosure')
3879 * @param key current key code (query, ignored, must match)
3880 * @param value value in the hash map (a 'struct PendingRequest'
3881 * that already exists)
3882 * @return GNUNET_YES if we should continue to
3883 * iterate (no match yet)
3884 * GNUNET_NO if not (match found).
3887 check_duplicate_request_peer (void *cls,
3888 const GNUNET_HashCode * key,
3891 struct CheckDuplicateRequestClosure *cdc = cls;
3892 struct PendingRequest *have = value;
3894 if (cdc->pr->target_pid == have->target_pid)
3904 * Handle P2P "GET" request.
3906 * @param cls closure, always NULL
3907 * @param other the other peer involved (sender or receiver, NULL
3908 * for loopback messages where we are both sender and receiver)
3909 * @param message the actual message
3910 * @param latency reported latency of the connection with 'other'
3911 * @param distance reported distance (DV) to 'other'
3912 * @return GNUNET_OK to keep the connection open,
3913 * GNUNET_SYSERR to close it (signal serious error)
3916 handle_p2p_get (void *cls,
3917 const struct GNUNET_PeerIdentity *other,
3918 const struct GNUNET_MessageHeader *message,
3919 struct GNUNET_TIME_Relative latency,
3922 struct PendingRequest *pr;
3923 struct ConnectedPeer *cp;
3924 struct ConnectedPeer *cps;
3925 struct CheckDuplicateRequestClosure cdc;
3926 struct GNUNET_TIME_Relative timeout;
3928 const struct GetMessage *gm;
3930 const GNUNET_HashCode *opt;
3933 uint32_t ttl_decrement;
3935 enum GNUNET_BLOCK_Type type;
3938 msize = ntohs(message->size);
3939 if (msize < sizeof (struct GetMessage))
3941 GNUNET_break_op (0);
3942 return GNUNET_SYSERR;
3944 gm = (const struct GetMessage*) message;
3946 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
3947 "Received request for `%s'\n",
3948 GNUNET_h2s (&gm->query));
3950 type = ntohl (gm->type);
3951 bm = ntohl (gm->hash_bitmap);
3959 if (msize < sizeof (struct GetMessage) + bits * sizeof (GNUNET_HashCode))
3961 GNUNET_break_op (0);
3962 return GNUNET_SYSERR;
3964 opt = (const GNUNET_HashCode*) &gm[1];
3965 bfsize = msize - sizeof (struct GetMessage) + bits * sizeof (GNUNET_HashCode);
3966 bm = ntohl (gm->hash_bitmap);
3968 cps = GNUNET_CONTAINER_multihashmap_get (connected_peers,
3969 &other->hashPubKey);
3972 /* peer must have just disconnected */
3973 GNUNET_STATISTICS_update (stats,
3974 gettext_noop ("# requests dropped due to initiator not being connected"),
3977 return GNUNET_SYSERR;
3979 if (0 != (bm & GET_MESSAGE_BIT_RETURN_TO))
3980 cp = GNUNET_CONTAINER_multihashmap_get (connected_peers,
3987 if (0 != (bm & GET_MESSAGE_BIT_RETURN_TO))
3988 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
3989 "Failed to find RETURN-TO peer `%4s' in connection set. Dropping query.\n",
3990 GNUNET_i2s ((const struct GNUNET_PeerIdentity*) &opt[bits-1]));
3993 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
3994 "Failed to find peer `%4s' in connection set. Dropping query.\n",
3995 GNUNET_i2s (other));
3997 GNUNET_STATISTICS_update (stats,
3998 gettext_noop ("# requests dropped due to missing reverse route"),
4001 /* FIXME: try connect? */
4004 /* note that we can really only check load here since otherwise
4005 peers could find out that we are overloaded by not being
4006 disconnected after sending us a malformed query... */
4007 priority = bound_priority (ntohl (gm->priority), cps);
4011 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
4012 "Dropping query from `%s', this peer is too busy.\n",
4013 GNUNET_i2s (other));
4015 GNUNET_STATISTICS_update (stats,
4016 gettext_noop ("# requests dropped due to high load"),
4022 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
4023 "Received request for `%s' of type %u from peer `%4s' with flags %u\n",
4024 GNUNET_h2s (&gm->query),
4025 (unsigned int) type,
4029 have_ns = (0 != (bm & GET_MESSAGE_BIT_SKS_NAMESPACE));
4030 pr = GNUNET_malloc (sizeof (struct PendingRequest) +
4031 (have_ns ? sizeof(GNUNET_HashCode) : 0));
4034 pr->namespace = (GNUNET_HashCode*) &pr[1];
4035 memcpy (&pr[1], &opt[bits++], sizeof (GNUNET_HashCode));
4037 if ( (GNUNET_LOAD_get_load (cp->transmission_delay) > 3 * (1 + priority)) ||
4038 (GNUNET_LOAD_get_average (cp->transmission_delay) >
4039 GNUNET_CONSTANTS_MAX_CORK_DELAY.value * 2 + GNUNET_LOAD_get_average (rt_entry_lifetime)) )
4041 /* don't have BW to send to peer, or would likely take longer than we have for it,
4042 so at best indirect the query */
4044 /* FIXME: if this line is enabled, the 'perf' test for larger files simply "hangs";
4045 the cause seems to be that the load goes up (to the point where we do this)
4046 and then never goes down again... (outch) */
4047 // pr->forward_only = GNUNET_YES;
4050 pr->mingle = ntohl (gm->filter_mutator);
4051 if (0 != (bm & GET_MESSAGE_BIT_TRANSMIT_TO))
4052 pr->target_pid = GNUNET_PEER_intern ((const struct GNUNET_PeerIdentity*) &opt[bits++]);
4053 pr->anonymity_level = 1;
4054 pr->priority = (uint32_t) priority;
4055 pr->ttl = bound_ttl (ntohl (gm->ttl), pr->priority);
4056 pr->query = gm->query;
4057 /* decrement ttl (always) */
4058 ttl_decrement = 2 * TTL_DECREMENT +
4059 GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
4061 if ( (pr->ttl < 0) &&
4062 (((int32_t)(pr->ttl - ttl_decrement)) > 0) )
4065 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
4066 "Dropping query from `%s' due to TTL underflow (%d - %u).\n",
4071 GNUNET_STATISTICS_update (stats,
4072 gettext_noop ("# requests dropped due TTL underflow"),
4075 /* integer underflow => drop (should be very rare)! */
4079 pr->ttl -= ttl_decrement;
4080 pr->start_time = GNUNET_TIME_absolute_get ();
4082 /* get bloom filter */
4085 pr->bf = GNUNET_CONTAINER_bloomfilter_init ((const char*) &opt[bits],
4088 pr->bf_size = bfsize;
4092 GNUNET_CONTAINER_multihashmap_get_multiple (query_request_map,
4094 &check_duplicate_request_peer,
4096 if (cdc.have != NULL)
4098 if (cdc.have->start_time.value + cdc.have->ttl >=
4099 pr->start_time.value + pr->ttl)
4101 /* existing request has higher TTL, drop new one! */
4102 cdc.have->priority += pr->priority;
4103 destroy_pending_request (pr);
4105 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
4106 "Have existing request with higher TTL, dropping new request.\n",
4107 GNUNET_i2s (other));
4109 GNUNET_STATISTICS_update (stats,
4110 gettext_noop ("# requests dropped due to higher-TTL request"),
4117 /* existing request has lower TTL, drop old one! */
4118 pr->priority += cdc.have->priority;
4119 /* Possible optimization: if we have applicable pending
4120 replies in 'cdc.have', we might want to move those over
4121 (this is a really rare special-case, so it is not clear
4122 that this would be worth it) */
4123 destroy_pending_request (cdc.have);
4124 /* keep processing 'pr'! */
4129 GNUNET_break (GNUNET_OK ==
4130 GNUNET_CONTAINER_multihashmap_put (query_request_map,
4133 GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE));
4134 GNUNET_break (GNUNET_OK ==
4135 GNUNET_CONTAINER_multihashmap_put (peer_request_map,
4138 GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE));
4140 pr->hnode = GNUNET_CONTAINER_heap_insert (requests_by_expiration_heap,
4142 pr->start_time.value + pr->ttl);
4144 GNUNET_STATISTICS_update (stats,
4145 gettext_noop ("# P2P searches received"),
4148 GNUNET_STATISTICS_update (stats,
4149 gettext_noop ("# P2P searches active"),
4153 /* calculate change in traffic preference */
4154 cps->inc_preference += pr->priority * 1000 + QUERY_BANDWIDTH_VALUE;
4155 /* process locally */
4156 if (type == GNUNET_BLOCK_TYPE_FS_DBLOCK)
4157 type = GNUNET_BLOCK_TYPE_ANY; /* to get on-demand as well */
4158 timeout = GNUNET_TIME_relative_multiply (BASIC_DATASTORE_REQUEST_DELAY,
4159 (pr->priority + 1));
4160 if (GNUNET_YES != pr->forward_only)
4163 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
4164 "Handing request for `%s' to datastore\n",
4165 GNUNET_h2s (&gm->query));
4167 pr->qe = GNUNET_DATASTORE_get (dsh,
4171 MAX_DATASTORE_QUEUE,
4173 &process_local_reply,
4177 GNUNET_STATISTICS_update (stats,
4178 gettext_noop ("# requests dropped by datastore (queue length limit)"),
4185 GNUNET_STATISTICS_update (stats,
4186 gettext_noop ("# requests forwarded due to high load"),
4191 /* Are multiple results possible (and did we look locally)? If so, start processing remotely now! */
4194 case GNUNET_BLOCK_TYPE_FS_DBLOCK:
4195 case GNUNET_BLOCK_TYPE_FS_IBLOCK:
4196 /* only one result, wait for datastore */
4197 if (GNUNET_YES != pr->forward_only)
4199 GNUNET_STATISTICS_update (stats,
4200 gettext_noop ("# requests not instantly forwarded (waiting for datastore)"),
4206 if (pr->task == GNUNET_SCHEDULER_NO_TASK)
4207 pr->task = GNUNET_SCHEDULER_add_now (sched,
4208 &forward_request_task,
4212 /* make sure we don't track too many requests */
4213 if (GNUNET_CONTAINER_heap_get_size (requests_by_expiration_heap) > max_pending_requests)
4215 pr = GNUNET_CONTAINER_heap_peek (requests_by_expiration_heap);
4216 GNUNET_assert (pr != NULL);
4217 destroy_pending_request (pr);
4223 /* **************************** CS GET Handling ************************ */
4227 * Handle START_SEARCH-message (search request from client).
4229 * @param cls closure
4230 * @param client identification of the client
4231 * @param message the actual message
4234 handle_start_search (void *cls,
4235 struct GNUNET_SERVER_Client *client,
4236 const struct GNUNET_MessageHeader *message)
4238 static GNUNET_HashCode all_zeros;
4239 const struct SearchMessage *sm;
4240 struct ClientList *cl;
4241 struct ClientRequestList *crl;
4242 struct PendingRequest *pr;
4245 enum GNUNET_BLOCK_Type type;
4247 msize = ntohs (message->size);
4248 if ( (msize < sizeof (struct SearchMessage)) ||
4249 (0 != (msize - sizeof (struct SearchMessage)) % sizeof (GNUNET_HashCode)) )
4252 GNUNET_SERVER_receive_done (client,
4256 GNUNET_STATISTICS_update (stats,
4257 gettext_noop ("# client searches received"),
4260 sc = (msize - sizeof (struct SearchMessage)) / sizeof (GNUNET_HashCode);
4261 sm = (const struct SearchMessage*) message;
4262 type = ntohl (sm->type);
4264 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
4265 "Received request for `%s' of type %u from local client\n",
4266 GNUNET_h2s (&sm->query),
4267 (unsigned int) type);
4270 while ( (cl != NULL) &&
4271 (cl->client != client) )
4275 cl = GNUNET_malloc (sizeof (struct ClientList));
4276 cl->client = client;
4277 GNUNET_SERVER_client_keep (client);
4278 cl->next = client_list;
4281 /* detect duplicate KBLOCK requests */
4282 if ( (type == GNUNET_BLOCK_TYPE_FS_KBLOCK) ||
4283 (type == GNUNET_BLOCK_TYPE_FS_NBLOCK) ||
4284 (type == GNUNET_BLOCK_TYPE_ANY) )
4287 while ( (crl != NULL) &&
4288 ( (0 != memcmp (&crl->req->query,
4290 sizeof (GNUNET_HashCode))) ||
4291 (crl->req->type != type) ) )
4296 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
4297 "Have existing request, merging content-seen lists.\n");
4300 /* Duplicate request (used to send long list of
4301 known/blocked results); merge 'pr->replies_seen'
4302 and update bloom filter */
4303 GNUNET_array_grow (pr->replies_seen,
4304 pr->replies_seen_size,
4305 pr->replies_seen_off + sc);
4306 memcpy (&pr->replies_seen[pr->replies_seen_off],
4308 sc * sizeof (GNUNET_HashCode));
4309 pr->replies_seen_off += sc;
4310 refresh_bloomfilter (pr);
4311 GNUNET_STATISTICS_update (stats,
4312 gettext_noop ("# client searches updated (merged content seen list)"),
4315 GNUNET_SERVER_receive_done (client,
4320 GNUNET_STATISTICS_update (stats,
4321 gettext_noop ("# client searches active"),
4324 pr = GNUNET_malloc (sizeof (struct PendingRequest) +
4325 ((type == GNUNET_BLOCK_TYPE_FS_SBLOCK) ? sizeof(GNUNET_HashCode) : 0));
4326 crl = GNUNET_malloc (sizeof (struct ClientRequestList));
4327 memset (crl, 0, sizeof (struct ClientRequestList));
4328 crl->client_list = cl;
4329 GNUNET_CONTAINER_DLL_insert (cl->rl_head,
4334 pr->client_request_list = crl;
4335 GNUNET_array_grow (pr->replies_seen,
4336 pr->replies_seen_size,
4338 memcpy (pr->replies_seen,
4340 sc * sizeof (GNUNET_HashCode));
4341 pr->replies_seen_off = sc;
4342 pr->anonymity_level = ntohl (sm->anonymity_level);
4343 pr->start_time = GNUNET_TIME_absolute_get ();
4344 refresh_bloomfilter (pr);
4345 pr->query = sm->query;
4346 if (0 == (1 & ntohl (sm->options)))
4347 pr->local_only = GNUNET_NO;
4349 pr->local_only = GNUNET_YES;
4352 case GNUNET_BLOCK_TYPE_FS_DBLOCK:
4353 case GNUNET_BLOCK_TYPE_FS_IBLOCK:
4354 if (0 != memcmp (&sm->target,
4356 sizeof (GNUNET_HashCode)))
4357 pr->target_pid = GNUNET_PEER_intern ((const struct GNUNET_PeerIdentity*) &sm->target);
4359 case GNUNET_BLOCK_TYPE_FS_SBLOCK:
4360 pr->namespace = (GNUNET_HashCode*) &pr[1];
4361 memcpy (&pr[1], &sm->target, sizeof (GNUNET_HashCode));
4366 GNUNET_break (GNUNET_OK ==
4367 GNUNET_CONTAINER_multihashmap_put (query_request_map,
4370 GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE));
4371 if (type == GNUNET_BLOCK_TYPE_FS_DBLOCK)
4372 type = GNUNET_BLOCK_TYPE_ANY; /* get on-demand blocks too! */
4373 pr->qe = GNUNET_DATASTORE_get (dsh,
4377 GNUNET_CONSTANTS_SERVICE_TIMEOUT,
4378 &process_local_reply,
4383 /* **************************** Startup ************************ */
4386 * Process fs requests.
4388 * @param s scheduler to use
4389 * @param server the initialized server
4390 * @param c configuration to use
4393 main_init (struct GNUNET_SCHEDULER_Handle *s,
4394 struct GNUNET_SERVER_Handle *server,
4395 const struct GNUNET_CONFIGURATION_Handle *c)
4397 static const struct GNUNET_CORE_MessageHandler p2p_handlers[] =
4400 GNUNET_MESSAGE_TYPE_FS_GET, 0 },
4402 GNUNET_MESSAGE_TYPE_FS_PUT, 0 },
4403 { &handle_p2p_migration_stop,
4404 GNUNET_MESSAGE_TYPE_FS_MIGRATION_STOP,
4405 sizeof (struct MigrationStopMessage) },
4408 static const struct GNUNET_SERVER_MessageHandler handlers[] = {
4409 {&GNUNET_FS_handle_index_start, NULL,
4410 GNUNET_MESSAGE_TYPE_FS_INDEX_START, 0},
4411 {&GNUNET_FS_handle_index_list_get, NULL,
4412 GNUNET_MESSAGE_TYPE_FS_INDEX_LIST_GET, sizeof(struct GNUNET_MessageHeader) },
4413 {&GNUNET_FS_handle_unindex, NULL, GNUNET_MESSAGE_TYPE_FS_UNINDEX,
4414 sizeof (struct UnindexMessage) },
4415 {&handle_start_search, NULL, GNUNET_MESSAGE_TYPE_FS_START_SEARCH,
4419 unsigned long long enc = 128;
4423 stats = GNUNET_STATISTICS_create (sched, "fs", cfg);
4424 min_migration_delay = GNUNET_TIME_UNIT_SECONDS;
4426 GNUNET_CONFIGURATION_get_value_number (cfg,
4428 "MAX_PENDING_REQUESTS",
4429 &max_pending_requests)) ||
4431 GNUNET_CONFIGURATION_get_value_number (cfg,
4433 "EXPECTED_NEIGHBOUR_COUNT",
4436 GNUNET_CONFIGURATION_get_value_time (cfg,
4438 "MIN_MIGRATION_DELAY",
4439 &min_migration_delay)) )
4441 GNUNET_log (GNUNET_ERROR_TYPE_INFO,
4442 _("Configuration fails to specify certain parameters, assuming default values."));
4444 connected_peers = GNUNET_CONTAINER_multihashmap_create (enc);
4445 query_request_map = GNUNET_CONTAINER_multihashmap_create (max_pending_requests);
4446 rt_entry_lifetime = GNUNET_LOAD_value_init ();
4447 peer_request_map = GNUNET_CONTAINER_multihashmap_create (enc);
4448 requests_by_expiration_heap = GNUNET_CONTAINER_heap_create (GNUNET_CONTAINER_HEAP_ORDER_MIN);
4449 core = GNUNET_CORE_connect (sched,
4451 GNUNET_TIME_UNIT_FOREVER_REL,
4454 &peer_connect_handler,
4455 &peer_disconnect_handler,
4462 GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
4463 _("Failed to connect to `%s' service.\n"),
4465 GNUNET_CONTAINER_multihashmap_destroy (connected_peers);
4466 connected_peers = NULL;
4467 GNUNET_CONTAINER_multihashmap_destroy (query_request_map);
4468 query_request_map = NULL;
4469 GNUNET_LOAD_value_free (rt_entry_lifetime);
4470 rt_entry_lifetime = NULL;
4471 GNUNET_CONTAINER_heap_destroy (requests_by_expiration_heap);
4472 requests_by_expiration_heap = NULL;
4473 GNUNET_CONTAINER_multihashmap_destroy (peer_request_map);
4474 peer_request_map = NULL;
4477 GNUNET_DATASTORE_disconnect (dsh, GNUNET_NO);
4480 return GNUNET_SYSERR;
4482 /* FIXME: distinguish between sending and storing in options? */
4483 if (active_migration)
4485 GNUNET_log (GNUNET_ERROR_TYPE_INFO,
4486 _("Content migration is enabled, will start to gather data\n"));
4487 consider_migration_gathering ();
4489 consider_dht_put_gathering (NULL);
4490 GNUNET_SERVER_disconnect_notify (server,
4491 &handle_client_disconnect,
4493 GNUNET_assert (GNUNET_OK ==
4494 GNUNET_CONFIGURATION_get_value_filename (cfg,
4498 GNUNET_DISK_directory_create (trustDirectory);
4499 GNUNET_SCHEDULER_add_with_priority (sched,
4500 GNUNET_SCHEDULER_PRIORITY_HIGH,
4501 &cron_flush_trust, NULL);
4504 GNUNET_SERVER_add_handlers (server, handlers);
4505 GNUNET_SCHEDULER_add_delayed (sched,
4506 GNUNET_TIME_UNIT_FOREVER_REL,
4514 * Process fs requests.
4516 * @param cls closure
4517 * @param sched scheduler to use
4518 * @param server the initialized server
4519 * @param cfg configuration to use
4523 struct GNUNET_SCHEDULER_Handle *sched,
4524 struct GNUNET_SERVER_Handle *server,
4525 const struct GNUNET_CONFIGURATION_Handle *cfg)
4527 active_migration = GNUNET_CONFIGURATION_get_value_yesno (cfg,
4530 dsh = GNUNET_DATASTORE_connect (cfg,
4534 GNUNET_SCHEDULER_shutdown (sched);
4537 datastore_get_load = GNUNET_LOAD_value_init ();
4538 datastore_put_load = GNUNET_LOAD_value_init ();
4539 block_cfg = GNUNET_CONFIGURATION_create ();
4540 GNUNET_CONFIGURATION_set_value_string (block_cfg,
4544 block_ctx = GNUNET_BLOCK_context_create (block_cfg);
4545 GNUNET_assert (NULL != block_ctx);
4546 dht_handle = GNUNET_DHT_connect (sched,
4549 if ( (GNUNET_OK != GNUNET_FS_indexing_init (sched, cfg, dsh)) ||
4550 (GNUNET_OK != main_init (sched, server, cfg)) )
4552 GNUNET_SCHEDULER_shutdown (sched);
4553 GNUNET_DATASTORE_disconnect (dsh, GNUNET_NO);
4555 GNUNET_DHT_disconnect (dht_handle);
4557 GNUNET_BLOCK_context_destroy (block_ctx);
4559 GNUNET_CONFIGURATION_destroy (block_cfg);
4561 GNUNET_LOAD_value_free (datastore_get_load);
4562 datastore_get_load = NULL;
4563 GNUNET_LOAD_value_free (datastore_put_load);
4564 datastore_put_load = NULL;
4571 * The main function for the fs service.
4573 * @param argc number of arguments from the command line
4574 * @param argv command line arguments
4575 * @return 0 ok, 1 on error
4578 main (int argc, char *const *argv)
4580 return (GNUNET_OK ==
4581 GNUNET_SERVICE_run (argc,
4584 GNUNET_SERVICE_OPTION_NONE,
4585 &run, NULL)) ? 0 : 1;
4588 /* end of gnunet-service-fs.c */