2 This file is part of GNUnet.
3 (C) 2009 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 2, 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 program that provides the file-sharing service
24 * @author Christian Grothoff
27 * - fix gazillion of minor FIXME's
28 * - possible major issue: we may queue "gazillions" of (K|S)Blocks for the
29 * core to transmit to another peer; need to make sure this is bounded overall...
30 * - randomly delay processing for improved anonymity (can wait)
31 * - content migration (put in local DS) (can wait)
32 * - handle some special cases when forwarding replies based on tracked requests (can wait)
33 * - tracking of success correlations for hot-path routing (can wait)
34 * - various load-based actions (can wait)
35 * - validation of KSBLOCKS (can wait)
36 * - remove on-demand blocks if they keep failing (can wait)
37 * - check that we decrement PIDs always where necessary (can wait)
38 * - find out how to use core-pulling instead of pushing (at least for some cases)
42 #include "gnunet_core_service.h"
43 #include "gnunet_datastore_service.h"
44 #include "gnunet_peer_lib.h"
45 #include "gnunet_protocols.h"
46 #include "gnunet_signatures.h"
47 #include "gnunet_util_lib.h"
52 * In-memory information about indexed files (also available
59 * This is a linked list.
61 struct IndexInfo *next;
64 * Name of the indexed file. Memory allocated
65 * at the end of this struct (do not free).
70 * Context for transmitting confirmation to client,
71 * NULL if we've done this already.
73 struct GNUNET_SERVER_TransmitContext *tc;
76 * Hash of the contents of the file.
78 GNUNET_HashCode file_id;
84 * Signature of a function that is called whenever a datastore
85 * request can be processed (or an entry put on the queue times out).
88 * @param ok GNUNET_OK if DS is ready, GNUNET_SYSERR on timeout
90 typedef void (*RequestFunction)(void *cls,
95 * Doubly-linked list of our requests for the datastore.
97 struct DatastoreRequestQueue
101 * This is a doubly-linked list.
103 struct DatastoreRequestQueue *next;
106 * This is a doubly-linked list.
108 struct DatastoreRequestQueue *prev;
111 * Function to call (will issue the request).
121 * When should this request time-out because we don't care anymore?
123 struct GNUNET_TIME_Absolute timeout;
126 * ID of task used for signaling timeout.
128 GNUNET_SCHEDULER_TaskIdentifier task;
134 * Closure for processing START_SEARCH messages from a client.
136 struct LocalGetContext
140 * This is a doubly-linked list.
142 struct LocalGetContext *next;
145 * This is a doubly-linked list.
147 struct LocalGetContext *prev;
150 * Client that initiated the search.
152 struct GNUNET_SERVER_Client *client;
155 * Array of results that we've already received
158 GNUNET_HashCode *results;
161 * Bloomfilter over all results (for fast query construction);
162 * NULL if we don't have any results.
164 * FIXME: this member is not used, is that OK? If so, it should
167 struct GNUNET_CONTAINER_BloomFilter *results_bf;
170 * DS request associated with this operation.
172 struct DatastoreRequestQueue *req;
175 * Current result message to transmit to client (or NULL).
177 struct ContentMessage *result;
180 * Type of the content that we're looking for.
186 * Desired anonymity level.
188 uint32_t anonymity_level;
191 * Number of results actually stored in the results array.
193 unsigned int results_used;
196 * Size of the results array in memory.
198 unsigned int results_size;
201 * Size (in bytes) of the 'results_bf' bloomfilter.
203 * FIXME: this member is not used, is that OK? If so, it should
206 size_t results_bf_size;
209 * If the request is for a DBLOCK or IBLOCK, this is the identity of
210 * the peer that is known to have a response. Set to all-zeros if
211 * such a target is not known (note that even if OUR anonymity
212 * level is >0 we may happen to know the responder's identity;
213 * nevertheless, we should probably not use it for a DHT-lookup
214 * or similar blunt actions in order to avoid exposing ourselves).
216 struct GNUNET_PeerIdentity target;
219 * If the request is for an SBLOCK, this is the identity of the
220 * pseudonym to which the SBLOCK belongs.
222 GNUNET_HashCode namespace;
225 * Hash of the keyword (aka query) for KBLOCKs; Hash of
226 * the CHK-encoded block for DBLOCKS and IBLOCKS (aka query)
227 * and hash of the identifier XORed with the target for
228 * SBLOCKS (aka query).
230 GNUNET_HashCode query;
236 * Possible routing policies for an FS-GET request.
241 * Simply drop the request.
243 ROUTING_POLICY_NONE = 0,
246 * Answer it if we can from local datastore.
248 ROUTING_POLICY_ANSWER = 1,
251 * Forward the request to other peers (if possible).
253 ROUTING_POLICY_FORWARD = 2,
256 * Forward to other peers, and ask them to route
257 * the response via ourselves.
259 ROUTING_POLICY_INDIRECT = 6,
262 * Do everything we could possibly do (that would
265 ROUTING_POLICY_ALL = 7
270 * Internal context we use for our initial processing
273 struct ProcessGetContext
276 * The search query (used for datastore lookup).
278 GNUNET_HashCode query;
281 * Which peer we should forward the response to.
283 struct GNUNET_PeerIdentity reply_to;
286 * Namespace for the result (only set for SKS requests)
288 GNUNET_HashCode namespace;
291 * Peer that we should forward the query to if possible
292 * (since that peer likely has the content).
294 struct GNUNET_PeerIdentity prime_target;
297 * When did we receive this request?
299 struct GNUNET_TIME_Absolute start_time;
302 * Our entry in the DRQ (non-NULL while we wait for our
303 * turn to interact with the local database).
305 struct DatastoreRequestQueue *drq;
308 * Filter used to eliminate duplicate results. Can be NULL if we
309 * are not yet filtering any results.
311 struct GNUNET_CONTAINER_BloomFilter *bf;
314 * Bitmap describing which of the optional
315 * hash codes / peer identities were given to us.
320 * Desired block type.
325 * Priority of the request.
330 * Size of the 'bf' (in bytes).
335 * In what ways are we going to process
338 enum RoutingPolicy policy;
341 * Time-to-live for the request (value
347 * Number to mingle hashes for bloom-filter
353 * Number of results that were found so far.
355 unsigned int results_found;
360 * Information we keep for each pending reply.
365 * This is a linked list.
367 struct PendingReply *next;
370 * Size of the reply; actual reply message follows
371 * at the end of this struct.
379 * All requests from a client are kept in a doubly-linked list.
381 struct ClientRequestList;
385 * Information we keep for each pending request. We should try to
386 * keep this struct as small as possible since its memory consumption
387 * is key to how many requests we can have pending at once.
389 struct PendingRequest
393 * ID of a client making a request, NULL if this entry is for a
396 struct GNUNET_SERVER_Client *client;
399 * If this request was made by a client,
400 * this is our entry in the client request
401 * list; otherwise NULL.
403 struct ClientRequestList *crl_entry;
406 * If this is a namespace query, pointer to the hash of the public
407 * key of the namespace; otherwise NULL.
409 GNUNET_HashCode *namespace;
412 * Bloomfilter we use to filter out replies that we don't care about
413 * (anymore). NULL as long as we are interested in all replies.
415 struct GNUNET_CONTAINER_BloomFilter *bf;
418 * Replies that we have received but were unable to forward yet
419 * (typically non-null only if we have a pending transmission
420 * request with the client or the respective peer).
422 struct PendingReply *replies_pending;
425 * Pending transmission request with the core service for the target
426 * peer (for processing of 'replies_pending') or Handle for a
427 * pending query-request for P2P-transmission with the core service.
428 * If non-NULL, this request must be cancelled should this struct be
431 struct GNUNET_CORE_TransmitHandle *cth;
434 * Pending transmission request for the target client (for processing of
435 * 'replies_pending').
437 struct GNUNET_CONNECTION_TransmitHandle *th;
440 * Hash code of all replies that we have seen so far (only valid
441 * if client is not NULL since we only track replies like this for
444 GNUNET_HashCode *replies_seen;
447 * When did we first see this request (form this peer), or, if our
448 * client is initiating, when did we last initiate a search?
450 struct GNUNET_TIME_Absolute start_time;
453 * The query that this request is for.
455 GNUNET_HashCode query;
458 * The task responsible for transmitting queries
461 GNUNET_SCHEDULER_TaskIdentifier task;
464 * (Interned) Peer identifier (only valid if "client" is NULL)
465 * that identifies a peer that gave us this request.
467 GNUNET_PEER_Id source_pid;
470 * (Interned) Peer identifier that identifies a preferred target
473 GNUNET_PEER_Id target_pid;
476 * (Interned) Peer identifiers of peers that have already
477 * received our query for this content.
479 GNUNET_PEER_Id *used_pids;
482 * Size of the 'bf' (in bytes).
487 * Desired anonymity level; only valid for requests from a local client.
489 uint32_t anonymity_level;
492 * How many entries in "used_pids" are actually valid?
494 unsigned int used_pids_off;
497 * How long is the "used_pids" array?
499 unsigned int used_pids_size;
502 * How many entries in "replies_seen" are actually valid?
504 unsigned int replies_seen_off;
507 * How long is the "replies_seen" array?
509 unsigned int replies_seen_size;
512 * Priority with which this request was made. If one of our clients
513 * made the request, then this is the current priority that we are
514 * using when initiating the request. This value is used when
515 * we decide to reward other peers with trust for providing a reply.
520 * Priority points left for us to spend when forwarding this request
523 uint32_t remaining_priority;
526 * Number to mingle hashes for bloom-filter
532 * TTL with which we saw this request (or, if we initiated, TTL that
533 * we used for the request).
538 * Type of the content that this request is for.
546 * All requests from a client are kept in a doubly-linked list.
548 struct ClientRequestList
551 * This is a doubly-linked list.
553 struct ClientRequestList *next;
556 * This is a doubly-linked list.
558 struct ClientRequestList *prev;
561 * A request from this client.
563 struct PendingRequest *req;
566 * Client list with the head and tail of this DLL.
568 struct ClientList *cl;
573 * Linked list of all clients that we are
574 * currently processing requests for.
580 * This is a linked list.
582 struct ClientList *next;
585 * What client is this entry for?
587 struct GNUNET_SERVER_Client* client;
590 * Head of the DLL of requests from this client.
592 struct ClientRequestList *head;
595 * Tail of the DLL of requests from this client.
597 struct ClientRequestList *tail;
603 * Closure for "process_reply" function.
605 struct ProcessReplyClosure
608 * The data for the reply.
613 * When the reply expires.
615 struct GNUNET_TIME_Absolute expiration;
623 * Namespace that this reply belongs to
624 * (if it is of type SBLOCK).
626 GNUNET_HashCode namespace;
634 * How much was this reply worth to us?
641 * Information about a peer that we are connected to.
642 * We track data that is useful for determining which
643 * peers should receive our requests.
649 * List of the last clients for which this peer
650 * successfully answered a query.
652 struct GNUNET_SERVER_Client *last_client_replies[CS2P_SUCCESS_LIST_SIZE];
655 * List of the last PIDs for which
656 * this peer successfully answered a query;
657 * We use 0 to indicate no successful reply.
659 GNUNET_PEER_Id last_p2p_replies[P2P_SUCCESS_LIST_SIZE];
662 * Average delay between sending the peer a request and
663 * getting a reply (only calculated over the requests for
664 * which we actually got a reply). Calculated
665 * as a moving average: new_delay = ((n-1)*last_delay+curr_delay) / n
667 struct GNUNET_TIME_Relative avg_delay;
670 * Average priority of successful replies. Calculated
671 * as a moving average: new_avg = ((n-1)*last_avg+curr_prio) / n
676 * The peer's identity.
681 * Number of requests we have currently pending
682 * with this peer (that is, requests that were
683 * transmitted so recently that we would not retransmit
686 unsigned int pending_requests;
689 * Which offset in "last_p2p_replies" will be updated next?
690 * (we go round-robin).
692 unsigned int last_p2p_replies_woff;
695 * Which offset in "last_client_replies" will be updated next?
696 * (we go round-robin).
698 unsigned int last_client_replies_woff;
704 * Our connection to the datastore.
706 static struct GNUNET_DATASTORE_Handle *dsh;
711 static struct GNUNET_SCHEDULER_Handle *sched;
716 const struct GNUNET_CONFIGURATION_Handle *cfg;
719 * Handle to the core service (NULL until we've connected to it).
721 struct GNUNET_CORE_Handle *core;
724 * Head of doubly-linked LGC list.
726 static struct LocalGetContext *lgc_head;
729 * Tail of doubly-linked LGC list.
731 static struct LocalGetContext *lgc_tail;
734 * Head of request queue for the datastore, sorted by timeout.
736 static struct DatastoreRequestQueue *drq_head;
739 * Tail of request queue for the datastore.
741 static struct DatastoreRequestQueue *drq_tail;
744 * Linked list of indexed files.
746 static struct IndexInfo *indexed_files;
749 * Maps hash over content of indexed files to the respective filename.
750 * The filenames are pointers into the indexed_files linked list and
751 * do not need to be freed.
753 static struct GNUNET_CONTAINER_MultiHashMap *ifm;
756 * Map of query hash codes to requests.
758 static struct GNUNET_CONTAINER_MultiHashMap *requests_by_query;
761 * Map of peer IDs to requests (for those requests coming
764 static struct GNUNET_CONTAINER_MultiHashMap *requests_by_peer;
767 * Linked list of all of our clients and their requests.
769 static struct ClientList *clients;
772 * Heap with the request that will expire next at the top. Contains
773 * pointers of type "struct PendingRequest*"; these will *also* be
774 * aliased from the "requests_by_peer" data structures and the
775 * "requests_by_query" table. Note that requests from our clients
776 * don't expire and are thus NOT in the "requests_by_expiration"
777 * (or the "requests_by_peer" tables).
779 static struct GNUNET_CONTAINER_Heap *requests_by_expiration;
782 * Map of peer identifiers to "struct ConnectedPeer" (for that peer).
784 static struct GNUNET_CONTAINER_MultiHashMap *connected_peers;
787 * Maximum number of requests (from other peers) that we're
788 * willing to have pending at any given point in time.
789 * FIXME: set from configuration (and 32 is a tiny value for testing only).
791 static uint64_t max_pending_requests = 32;
794 * Write the current index information list to disk.
799 struct GNUNET_BIO_WriteHandle *wh;
801 struct IndexInfo *pos;
804 GNUNET_CONFIGURATION_get_value_filename (cfg,
809 GNUNET_log (GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
810 _("Configuration option `%s' in section `%s' missing.\n"),
815 wh = GNUNET_BIO_write_open (fn);
818 GNUNET_log (GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
819 _("Could not open `%s'.\n"),
828 GNUNET_BIO_write (wh,
830 sizeof (GNUNET_HashCode))) ||
832 GNUNET_BIO_write_string (wh,
838 GNUNET_BIO_write_close (wh))
840 GNUNET_log (GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
841 _("Error writing `%s'.\n"),
851 * Read index information from disk.
856 struct GNUNET_BIO_ReadHandle *rh;
858 struct IndexInfo *pos;
865 GNUNET_CONFIGURATION_get_value_filename (cfg,
870 GNUNET_log (GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
871 _("Configuration option `%s' in section `%s' missing.\n"),
876 rh = GNUNET_BIO_read_open (fn);
879 GNUNET_log (GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
880 _("Could not open `%s'.\n"),
886 while ( (GNUNET_OK ==
888 "Hash of indexed file",
890 sizeof (GNUNET_HashCode))) &&
892 GNUNET_BIO_read_string (rh,
893 "Name of indexed file",
897 slen = strlen (fname) + 1;
898 pos = GNUNET_malloc (sizeof (struct IndexInfo) + slen);
900 pos->filename = (const char *) &pos[1];
901 memcpy (&pos[1], fname, slen);
903 GNUNET_CONTAINER_multihashmap_put (ifm,
905 (void*) pos->filename,
906 GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY))
912 pos->next = indexed_files;
917 GNUNET_BIO_read_close (rh, &emsg))
924 * We've validated the hash of the file we're about to
925 * index. Signal success to the client and update
926 * our internal data structures.
928 * @param ii the index info entry for the request
931 signal_index_ok (struct IndexInfo *ii)
934 GNUNET_CONTAINER_multihashmap_put (ifm,
936 (void*) ii->filename,
937 GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY))
939 GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
940 _("Index request received for file `%s' is indexed as `%s'. Permitting anyway.\n"),
942 (const char*) GNUNET_CONTAINER_multihashmap_get (ifm,
944 GNUNET_SERVER_transmit_context_append (ii->tc,
946 GNUNET_MESSAGE_TYPE_FS_INDEX_START_OK);
947 GNUNET_SERVER_transmit_context_run (ii->tc,
948 GNUNET_TIME_UNIT_MINUTES);
952 ii->next = indexed_files;
955 GNUNET_SERVER_transmit_context_append (ii->tc,
957 GNUNET_MESSAGE_TYPE_FS_INDEX_START_OK);
958 GNUNET_SERVER_transmit_context_run (ii->tc,
959 GNUNET_TIME_UNIT_MINUTES);
965 * Function called once the hash computation over an
966 * indexed file has completed.
968 * @param cls closure, our publishing context
969 * @param res resulting hash, NULL on error
972 hash_for_index_val (void *cls,
973 const GNUNET_HashCode *
976 struct IndexInfo *ii = cls;
978 if ( (res == NULL) ||
981 sizeof(GNUNET_HashCode))) )
983 GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
984 _("Hash mismatch trying to index file `%s'\n"),
986 GNUNET_SERVER_transmit_context_append (ii->tc,
988 GNUNET_MESSAGE_TYPE_FS_INDEX_START_FAILED);
989 GNUNET_SERVER_transmit_context_run (ii->tc,
990 GNUNET_TIME_UNIT_MINUTES);
994 signal_index_ok (ii);
999 * Handle INDEX_START-message.
1001 * @param cls closure
1002 * @param client identification of the client
1003 * @param message the actual message
1006 handle_index_start (void *cls,
1007 struct GNUNET_SERVER_Client *client,
1008 const struct GNUNET_MessageHeader *message)
1010 const struct IndexStartMessage *ism;
1013 struct IndexInfo *ii;
1020 msize = ntohs(message->size);
1021 if ( (msize <= sizeof (struct IndexStartMessage)) ||
1022 ( ((const char *)message)[msize-1] != '\0') )
1025 GNUNET_SERVER_receive_done (client,
1029 ism = (const struct IndexStartMessage*) message;
1030 fn = (const char*) &ism[1];
1031 dev = ntohl (ism->device);
1032 ino = GNUNET_ntohll (ism->inode);
1033 ism = (const struct IndexStartMessage*) message;
1034 slen = strlen (fn) + 1;
1035 ii = GNUNET_malloc (sizeof (struct IndexInfo) + slen);
1036 ii->filename = (const char*) &ii[1];
1037 memcpy (&ii[1], fn, slen);
1038 ii->file_id = ism->file_id;
1039 ii->tc = GNUNET_SERVER_transmit_context_create (client);
1040 if ( ( (dev != 0) ||
1042 (GNUNET_OK == GNUNET_DISK_file_get_identifiers (fn,
1048 /* fast validation OK! */
1049 signal_index_ok (ii);
1052 /* slow validation, need to hash full file (again) */
1053 GNUNET_CRYPTO_hash_file (sched,
1054 GNUNET_SCHEDULER_PRIORITY_IDLE,
1058 &hash_for_index_val,
1064 * Handle INDEX_LIST_GET-message.
1066 * @param cls closure
1067 * @param client identification of the client
1068 * @param message the actual message
1071 handle_index_list_get (void *cls,
1072 struct GNUNET_SERVER_Client *client,
1073 const struct GNUNET_MessageHeader *message)
1075 struct GNUNET_SERVER_TransmitContext *tc;
1076 struct IndexInfoMessage *iim;
1077 char buf[GNUNET_SERVER_MAX_MESSAGE_SIZE];
1080 struct GNUNET_MessageHeader *msg;
1081 struct IndexInfo *pos;
1083 tc = GNUNET_SERVER_transmit_context_create (client);
1084 iim = (struct IndexInfoMessage*) buf;
1086 pos = indexed_files;
1090 iim->file_id = pos->file_id;
1092 slen = strlen (fn) + 1;
1093 if (slen + sizeof (struct IndexInfoMessage) >
1094 GNUNET_SERVER_MAX_MESSAGE_SIZE)
1099 memcpy (&iim[1], fn, slen);
1100 GNUNET_SERVER_transmit_context_append
1103 sizeof (struct IndexInfoMessage)
1104 - sizeof (struct GNUNET_MessageHeader) + slen,
1105 GNUNET_MESSAGE_TYPE_FS_INDEX_LIST_ENTRY);
1108 GNUNET_SERVER_transmit_context_append (tc,
1110 GNUNET_MESSAGE_TYPE_FS_INDEX_LIST_END);
1111 GNUNET_SERVER_transmit_context_run (tc,
1112 GNUNET_TIME_UNIT_MINUTES);
1117 * Handle UNINDEX-message.
1119 * @param cls closure
1120 * @param client identification of the client
1121 * @param message the actual message
1124 handle_unindex (void *cls,
1125 struct GNUNET_SERVER_Client *client,
1126 const struct GNUNET_MessageHeader *message)
1128 const struct UnindexMessage *um;
1129 struct IndexInfo *pos;
1130 struct IndexInfo *prev;
1131 struct IndexInfo *next;
1132 struct GNUNET_SERVER_TransmitContext *tc;
1135 um = (const struct UnindexMessage*) message;
1138 pos = indexed_files;
1142 if (0 == memcmp (&pos->file_id,
1144 sizeof (GNUNET_HashCode)))
1147 indexed_files = pos->next;
1149 prev->next = pos->next;
1159 if (GNUNET_YES == found)
1160 write_index_list ();
1161 tc = GNUNET_SERVER_transmit_context_create (client);
1162 GNUNET_SERVER_transmit_context_append (tc,
1164 GNUNET_MESSAGE_TYPE_FS_UNINDEX_OK);
1165 GNUNET_SERVER_transmit_context_run (tc,
1166 GNUNET_TIME_UNIT_MINUTES);
1171 * Run the next DS request in our
1172 * queue, we're done with the current one.
1177 struct DatastoreRequestQueue *e;
1179 while (NULL != (e = drq_head))
1181 if (0 != GNUNET_TIME_absolute_get_remaining (e->timeout).value)
1183 if (e->task != GNUNET_SCHEDULER_NO_TASK)
1184 GNUNET_SCHEDULER_cancel (sched, e->task);
1185 GNUNET_CONTAINER_DLL_remove (drq_head, drq_tail, e);
1186 e->req (e->req_cls, GNUNET_NO);
1191 if (e->task != GNUNET_SCHEDULER_NO_TASK)
1192 GNUNET_SCHEDULER_cancel (sched, e->task);
1193 e->task = GNUNET_SCHEDULER_NO_TASK;
1194 e->req (e->req_cls, GNUNET_YES);
1195 GNUNET_CONTAINER_DLL_remove (drq_head, drq_tail, e);
1201 * A datastore request had to be timed out.
1203 * @param cls closure (of type "struct DatastoreRequestQueue*")
1204 * @param tc task context, unused
1207 timeout_ds_request (void *cls,
1208 const struct GNUNET_SCHEDULER_TaskContext *tc)
1210 struct DatastoreRequestQueue *e = cls;
1212 e->task = GNUNET_SCHEDULER_NO_TASK;
1213 GNUNET_CONTAINER_DLL_remove (drq_head, drq_tail, e);
1214 e->req (e->req_cls, GNUNET_NO);
1220 * Queue a request for the datastore.
1222 * @param deadline by when the request should run
1223 * @param fun function to call once the request can be run
1224 * @param fun_cls closure for fun
1226 static struct DatastoreRequestQueue *
1227 queue_ds_request (struct GNUNET_TIME_Relative deadline,
1228 RequestFunction fun,
1231 struct DatastoreRequestQueue *e;
1232 struct DatastoreRequestQueue *bef;
1234 if (drq_head == NULL)
1236 /* no other requests pending, run immediately */
1237 fun (fun_cls, GNUNET_OK);
1240 e = GNUNET_malloc (sizeof (struct DatastoreRequestQueue));
1241 e->timeout = GNUNET_TIME_relative_to_absolute (deadline);
1243 e->req_cls = fun_cls;
1244 if (deadline.value == GNUNET_TIME_UNIT_FOREVER_REL.value)
1246 /* local request, highest prio, put at head of queue
1247 regardless of deadline */
1253 while ( (NULL != bef) &&
1254 (e->timeout.value < bef->timeout.value) )
1257 GNUNET_CONTAINER_DLL_insert_after (drq_head, drq_tail, bef, e);
1258 if (deadline.value == GNUNET_TIME_UNIT_FOREVER_REL.value)
1260 e->task = GNUNET_SCHEDULER_add_delayed (sched,
1262 GNUNET_SCHEDULER_PRIORITY_BACKGROUND,
1263 GNUNET_SCHEDULER_NO_TASK,
1265 &timeout_ds_request,
1272 * Free the state associated with a local get context.
1274 * @param lgc the lgc to free
1277 local_get_context_free (struct LocalGetContext *lgc)
1279 GNUNET_CONTAINER_DLL_remove (lgc_head, lgc_tail, lgc);
1280 GNUNET_SERVER_client_drop (lgc->client);
1281 GNUNET_free_non_null (lgc->results);
1282 if (lgc->results_bf != NULL)
1283 GNUNET_CONTAINER_bloomfilter_free (lgc->results_bf);
1284 if (lgc->req != NULL)
1286 if (lgc->req->task != GNUNET_SCHEDULER_NO_TASK)
1287 GNUNET_SCHEDULER_cancel (sched, lgc->req->task);
1288 GNUNET_CONTAINER_DLL_remove (lgc_head, lgc_tail, lgc);
1289 GNUNET_free (lgc->req);
1296 * We're able to transmit the next (local) result to the client.
1297 * Do it and ask the datastore for more. Or, on error, tell
1298 * the datastore to stop giving us more.
1300 * @param cls our closure (struct LocalGetContext)
1301 * @param max maximum number of bytes we can transmit
1302 * @param buf where to copy our message
1303 * @return number of bytes copied to buf
1306 transmit_local_result (void *cls,
1310 struct LocalGetContext *lgc = cls;
1316 GNUNET_free (lgc->result);
1318 GNUNET_DATASTORE_get_next (dsh, GNUNET_NO);
1321 msize = ntohs (lgc->result->header.size);
1322 GNUNET_assert (max >= msize);
1323 memcpy (buf, lgc->result, msize);
1324 GNUNET_free (lgc->result);
1326 GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
1332 * Continuation called from datastore's remove
1336 * @param success did the deletion work?
1337 * @param msg error message
1340 remove_cont (void *cls,
1344 if (GNUNET_OK != success)
1345 GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
1346 _("Failed to delete bogus block: %s\n"),
1348 GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
1353 * Mingle hash with the mingle_number to
1354 * produce different bits.
1357 mingle_hash (const GNUNET_HashCode * in,
1358 int32_t mingle_number,
1359 GNUNET_HashCode * hc)
1363 GNUNET_CRYPTO_hash (&mingle_number,
1366 GNUNET_CRYPTO_hash_xor (&m, in, hc);
1371 * We've received an on-demand encoded block
1372 * from the datastore. Attempt to do on-demand
1373 * encoding and (if successful), call the
1374 * continuation with the resulting block. On
1375 * error, clean up and ask the datastore for
1378 * @param key key for the content
1379 * @param size number of bytes in data
1380 * @param data content stored
1381 * @param type type of the content
1382 * @param priority priority of the content
1383 * @param anonymity anonymity-level for the content
1384 * @param expiration expiration time for the content
1385 * @param uid unique identifier for the datum;
1386 * maybe 0 if no unique identifier is available
1387 * @param cont function to call with the actual block
1388 * @param cont_cls closure for cont
1391 handle_on_demand_block (const GNUNET_HashCode * key,
1397 struct GNUNET_TIME_Absolute
1398 expiration, uint64_t uid,
1399 GNUNET_DATASTORE_Iterator cont,
1402 const struct OnDemandBlock *odb;
1403 GNUNET_HashCode nkey;
1404 struct GNUNET_CRYPTO_AesSessionKey skey;
1405 struct GNUNET_CRYPTO_AesInitializationVector iv;
1406 GNUNET_HashCode query;
1408 char ndata[DBLOCK_SIZE];
1409 char edata[DBLOCK_SIZE];
1411 struct GNUNET_DISK_FileHandle *fh;
1414 if (size != sizeof (struct OnDemandBlock))
1417 GNUNET_DATASTORE_remove (dsh,
1423 GNUNET_TIME_UNIT_FOREVER_REL);
1426 odb = (const struct OnDemandBlock*) data;
1427 off = GNUNET_ntohll (odb->offset);
1428 fn = (const char*) GNUNET_CONTAINER_multihashmap_get (ifm,
1431 if ( (NULL == fn) ||
1432 (NULL == (fh = GNUNET_DISK_file_open (fn,
1433 GNUNET_DISK_OPEN_READ,
1434 GNUNET_DISK_PERM_NONE))) ||
1436 GNUNET_DISK_file_seek (fh,
1438 GNUNET_DISK_SEEK_SET)) ||
1440 (nsize = GNUNET_DISK_file_read (fh,
1444 GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
1445 _("Could not access indexed file `%s' at offset %llu: %s\n"),
1446 GNUNET_h2s (&odb->file_id),
1447 (unsigned long long) off,
1450 GNUNET_DISK_file_close (fh);
1451 /* FIXME: if this happens often, we need
1452 to remove the OnDemand block from the DS! */
1453 GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
1456 GNUNET_DISK_file_close (fh);
1457 GNUNET_CRYPTO_hash (ndata,
1460 GNUNET_CRYPTO_hash_to_aes_key (&nkey, &skey, &iv);
1461 GNUNET_CRYPTO_aes_encrypt (ndata,
1466 GNUNET_CRYPTO_hash (edata,
1469 if (0 != memcmp (&query,
1471 sizeof (GNUNET_HashCode)))
1473 GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
1474 _("Indexed file `%s' changed at offset %llu\n"),
1476 (unsigned long long) off);
1477 /* FIXME: if this happens often, we need
1478 to remove the OnDemand block from the DS! */
1479 GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
1486 GNUNET_DATASTORE_BLOCKTYPE_DBLOCK,
1495 * How many bytes should a bloomfilter be if we have already seen
1496 * entry_count responses? Note that BLOOMFILTER_K gives us the number
1497 * of bits set per entry. Furthermore, we should not re-size the
1498 * filter too often (to keep it cheap).
1500 * Since other peers will also add entries but not resize the filter,
1501 * we should generally pick a slightly larger size than what the
1502 * strict math would suggest.
1504 * @return must be a power of two and smaller or equal to 2^15.
1507 compute_bloomfilter_size (unsigned int entry_count)
1510 unsigned int ideal = (entry_count * BLOOMFILTER_K) / 4;
1511 uint16_t max = 1 << 15;
1513 if (entry_count > max)
1516 while ((size < max) && (size < ideal))
1525 * Recalculate our bloom filter for filtering replies.
1527 * @param count number of entries we are filtering right now
1528 * @param mingle set to our new mingling value
1529 * @param bf_size set to the size of the bloomfilter
1530 * @param entries the entries to filter
1531 * @return updated bloomfilter, NULL for none
1533 static struct GNUNET_CONTAINER_BloomFilter *
1534 refresh_bloomfilter (unsigned int count,
1537 const GNUNET_HashCode *entries)
1539 struct GNUNET_CONTAINER_BloomFilter *bf;
1542 GNUNET_HashCode mhash;
1546 nsize = compute_bloomfilter_size (count);
1547 *mingle = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, -1);
1549 bf = GNUNET_CONTAINER_bloomfilter_init (NULL,
1552 for (i=0;i<count;i++)
1554 mingle_hash (&entries[i], *mingle, &mhash);
1555 GNUNET_CONTAINER_bloomfilter_add (bf, &mhash);
1562 * Closure used for "target_peer_select_cb".
1564 struct PeerSelectionContext
1567 * The request for which we are selecting
1570 struct PendingRequest *pr;
1573 * Current "prime" target.
1575 struct GNUNET_PeerIdentity target;
1578 * How much do we like this target?
1580 double target_score;
1586 * Function called for each connected peer to determine
1587 * which one(s) would make good targets for forwarding.
1589 * @param cls closure (struct PeerSelectionContext)
1590 * @param key current key code (peer identity)
1591 * @param value value in the hash map (struct ConnectedPeer)
1592 * @return GNUNET_YES if we should continue to
1597 target_peer_select_cb (void *cls,
1598 const GNUNET_HashCode * key,
1601 struct PeerSelectionContext *psc = cls;
1602 struct ConnectedPeer *cp = value;
1603 struct PendingRequest *pr = psc->pr;
1607 /* 1) check if we have already (recently) forwarded to this peer */
1608 for (i=0;i<pr->used_pids_off;i++)
1609 if (pr->used_pids[i] == cp->pid)
1610 return GNUNET_YES; /* skip */
1611 // 2) calculate how much we'd like to forward to this peer
1612 score = 0; // FIXME!
1614 /* store best-fit in closure */
1615 if (score > psc->target_score)
1617 psc->target_score = score;
1618 psc->target.hashPubKey = *key;
1625 * We use a random delay to make the timing of requests
1626 * less predictable. This function returns such a random
1629 * @return random delay to use for some request, between 0 and TTL_DECREMENT ms
1631 static struct GNUNET_TIME_Relative
1632 get_processing_delay ()
1634 return GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MILLISECONDS,
1635 GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
1641 * Task that is run for each request with the
1642 * goal of forwarding the associated query to
1643 * other peers. The task should re-schedule
1644 * itself to be re-run once the TTL has expired.
1645 * (or at a later time if more peers should
1646 * be queried earlier).
1648 * @param cls the requests "struct PendingRequest*"
1649 * @param tc task context (unused)
1652 forward_request_task (void *cls,
1653 const struct GNUNET_SCHEDULER_TaskContext *tc);
1657 * We've selected a peer for forwarding of a query.
1658 * Construct the message and then re-schedule the
1659 * task to forward again to (other) peers.
1661 * @param cls closure
1662 * @param size number of bytes available in buf
1663 * @param buf where the callee should write the message
1664 * @return number of bytes written to buf
1667 transmit_request_cb (void *cls,
1671 struct PendingRequest *pr = cls;
1672 struct GetMessage *gm;
1673 GNUNET_HashCode *ext;
1679 /* (1) check for timeout */
1682 /* timeout, try another peer immediately again */
1683 pr->task = GNUNET_SCHEDULER_add_delayed (sched,
1685 GNUNET_SCHEDULER_PRIORITY_IDLE,
1686 GNUNET_SCHEDULER_NO_TASK,
1687 GNUNET_TIME_UNIT_ZERO,
1688 &forward_request_task,
1692 /* (2) build query message */
1693 k = 0; // FIXME: count hash codes!
1694 msize = sizeof (struct GetMessage) + pr->bf_size + k * sizeof(GNUNET_HashCode);
1695 GNUNET_assert (msize < GNUNET_SERVER_MAX_MESSAGE_SIZE);
1696 gm = (struct GetMessage*) buf;
1697 gm->header.type = htons (GNUNET_MESSAGE_TYPE_FS_GET);
1698 gm->header.size = htons (msize);
1699 gm->type = htonl (pr->type);
1700 pr->remaining_priority /= 2;
1701 gm->priority = htonl (pr->remaining_priority);
1702 gm->ttl = htonl (pr->ttl);
1703 gm->filter_mutator = htonl(pr->mingle);
1704 gm->hash_bitmap = htonl (42);
1705 gm->query = pr->query;
1706 ext = (GNUNET_HashCode*) &gm[1];
1707 // FIXME: setup "ext[0]..[k-1]"
1708 bfdata = (char *) &ext[k];
1710 GNUNET_CONTAINER_bloomfilter_get_raw_data (pr->bf,
1714 /* (3) schedule job to do it again (or another peer, etc.) */
1715 pr->task = GNUNET_SCHEDULER_add_delayed (sched,
1717 GNUNET_SCHEDULER_PRIORITY_IDLE,
1718 GNUNET_SCHEDULER_NO_TASK,
1719 get_processing_delay (), // FIXME!
1720 &forward_request_task,
1728 * Function called after we've tried to reserve
1729 * a certain amount of bandwidth for a reply.
1730 * Check if we succeeded and if so send our query.
1732 * @param cls the requests "struct PendingRequest*"
1733 * @param peer identifies the peer
1734 * @param latency current latency estimate, "FOREVER" if we have been
1736 * @param bpm_in set to the current bandwidth limit (receiving) for this peer
1737 * @param bpm_out set to the current bandwidth limit (sending) for this peer
1738 * @param amount set to the amount that was actually reserved or unreserved
1739 * @param preference current traffic preference for the given peer
1742 target_reservation_cb (void *cls,
1744 GNUNET_PeerIdentity * peer,
1745 unsigned int bpm_in,
1746 unsigned int bpm_out,
1747 struct GNUNET_TIME_Relative
1748 latency, int amount,
1749 unsigned long long preference)
1751 struct PendingRequest *pr = cls;
1754 struct GNUNET_TIME_Relative maxdelay;
1756 GNUNET_assert (peer != NULL);
1757 if ( (amount != DBLOCK_SIZE) ||
1760 /* try again later; FIXME: we may need to un-reserve "amount"? */
1761 pr->task = GNUNET_SCHEDULER_add_delayed (sched,
1763 GNUNET_SCHEDULER_PRIORITY_IDLE,
1764 GNUNET_SCHEDULER_NO_TASK,
1765 get_processing_delay (), // FIXME: longer?
1766 &forward_request_task,
1770 // (2) transmit, update ttl/priority
1771 // FIXME: calculate priority, maxdelay, size properly!
1774 maxdelay = GNUNET_CONSTANTS_SERVICE_TIMEOUT;
1775 pr->cth = GNUNET_CORE_notify_transmit_ready (core,
1780 &transmit_request_cb,
1782 if (pr->cth == NULL)
1784 /* try again later */
1785 pr->task = GNUNET_SCHEDULER_add_delayed (sched,
1787 GNUNET_SCHEDULER_PRIORITY_IDLE,
1788 GNUNET_SCHEDULER_NO_TASK,
1789 get_processing_delay (), // FIXME: longer?
1790 &forward_request_task,
1797 * Task that is run for each request with the
1798 * goal of forwarding the associated query to
1799 * other peers. The task should re-schedule
1800 * itself to be re-run once the TTL has expired.
1801 * (or at a later time if more peers should
1802 * be queried earlier).
1804 * @param cls the requests "struct PendingRequest*"
1805 * @param tc task context (unused)
1808 forward_request_task (void *cls,
1809 const struct GNUNET_SCHEDULER_TaskContext *tc)
1811 struct PendingRequest *pr = cls;
1812 struct PeerSelectionContext psc;
1814 pr->task = GNUNET_SCHEDULER_NO_TASK;
1815 if (pr->cth != NULL)
1817 /* we're busy transmitting a result, wait a bit */
1818 pr->task = GNUNET_SCHEDULER_add_delayed (sched,
1820 GNUNET_SCHEDULER_PRIORITY_IDLE,
1821 GNUNET_SCHEDULER_NO_TASK,
1822 get_processing_delay (),
1823 &forward_request_task,
1827 /* (1) select target */
1829 psc.target_score = DBL_MIN;
1830 GNUNET_CONTAINER_multihashmap_iterate (connected_peers,
1831 &target_peer_select_cb,
1833 if (psc.target_score == DBL_MIN)
1835 /* no possible target found, wait some time */
1836 pr->task = GNUNET_SCHEDULER_add_delayed (sched,
1838 GNUNET_SCHEDULER_PRIORITY_IDLE,
1839 GNUNET_SCHEDULER_NO_TASK,
1840 get_processing_delay (), // FIXME: exponential back-off? or at least wait longer...
1841 &forward_request_task,
1845 /* (2) reserve reply bandwidth */
1846 // FIXME: need a way to cancel; this
1847 // async operation is problematic (segv-problematic)
1848 // if "pr" is destroyed while it happens!
1849 GNUNET_CORE_peer_configure (core,
1851 GNUNET_CONSTANTS_SERVICE_TIMEOUT,
1853 DBLOCK_SIZE, // FIXME: make dependent on type?
1855 &target_reservation_cb,
1861 * We're processing (local) results for a search request
1862 * from a (local) client. Pass applicable results to the
1863 * client and if we are done either clean up (operation
1864 * complete) or switch to P2P search (more results possible).
1866 * @param cls our closure (struct LocalGetContext)
1867 * @param key key for the content
1868 * @param size number of bytes in data
1869 * @param data content stored
1870 * @param type type of the content
1871 * @param priority priority of the content
1872 * @param anonymity anonymity-level for the content
1873 * @param expiration expiration time for the content
1874 * @param uid unique identifier for the datum;
1875 * maybe 0 if no unique identifier is available
1878 process_local_get_result (void *cls,
1879 const GNUNET_HashCode * key,
1885 struct GNUNET_TIME_Absolute
1889 struct LocalGetContext *lgc = cls;
1890 struct PendingRequest *pr;
1891 struct ClientRequestList *crl;
1892 struct ClientList *cl;
1898 /* no further results from datastore; continue
1899 processing further requests from the client and
1900 allow the next task to use the datastore; also,
1901 switch to P2P requests or clean up our state. */
1903 GNUNET_SERVER_receive_done (lgc->client,
1905 if ( (lgc->results_used == 0) ||
1906 (lgc->type == GNUNET_DATASTORE_BLOCKTYPE_KBLOCK) ||
1907 (lgc->type == GNUNET_DATASTORE_BLOCKTYPE_SBLOCK) ||
1908 (lgc->type == GNUNET_DATASTORE_BLOCKTYPE_SKBLOCK) )
1911 while ( (NULL != cl) &&
1912 (cl->client != lgc->client) )
1916 cl = GNUNET_malloc (sizeof (struct ClientList));
1917 cl->client = lgc->client;
1921 crl = GNUNET_malloc (sizeof (struct ClientRequestList));
1923 GNUNET_CONTAINER_DLL_insert (cl->head, cl->tail, crl);
1924 pr = GNUNET_malloc (sizeof (struct PendingRequest));
1925 pr->client = lgc->client;
1926 GNUNET_SERVER_client_keep (pr->client);
1927 pr->crl_entry = crl;
1929 if (lgc->type == GNUNET_DATASTORE_BLOCKTYPE_SBLOCK)
1931 pr->namespace = GNUNET_malloc (sizeof (GNUNET_HashCode));
1932 *pr->namespace = lgc->namespace;
1934 pr->replies_seen = lgc->results;
1935 lgc->results = NULL;
1936 pr->start_time = GNUNET_TIME_absolute_get ();
1937 pr->query = lgc->query;
1938 pr->target_pid = GNUNET_PEER_intern (&lgc->target);
1939 pr->replies_seen_off = lgc->results_used;
1940 pr->replies_seen_size = lgc->results_size;
1941 lgc->results_size = 0;
1942 pr->type = lgc->type;
1943 pr->anonymity_level = lgc->anonymity_level;
1944 pr->bf = refresh_bloomfilter (pr->replies_seen_off,
1948 GNUNET_CONTAINER_multihashmap_put (requests_by_query,
1951 GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
1952 pr->task = GNUNET_SCHEDULER_add_delayed (sched,
1954 GNUNET_SCHEDULER_PRIORITY_IDLE,
1955 GNUNET_SCHEDULER_NO_TASK,
1956 get_processing_delay (),
1957 &forward_request_task,
1959 local_get_context_free (lgc);
1962 /* got all possible results, clean up! */
1963 local_get_context_free (lgc);
1966 if (type == GNUNET_DATASTORE_BLOCKTYPE_ONDEMAND)
1968 handle_on_demand_block (key, size, data, type, priority,
1969 anonymity, expiration, uid,
1970 &process_local_get_result,
1974 if (type != lgc->type)
1976 /* this should be virtually impossible to reach (DBLOCK
1977 query hash being identical to KBLOCK/SBLOCK query hash);
1978 nevertheless, if it happens, the correct thing is to
1979 simply skip the result. */
1980 GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
1983 /* check if this is a result we've alredy
1985 for (i=0;i<lgc->results_used;i++)
1986 if (0 == memcmp (key,
1988 sizeof (GNUNET_HashCode)))
1990 GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
1993 if (lgc->results_used == lgc->results_size)
1994 GNUNET_array_grow (lgc->results,
1996 lgc->results_size * 2 + 2);
1997 GNUNET_CRYPTO_hash (data,
1999 &lgc->results[lgc->results_used++]);
2000 msize = size + sizeof (struct ContentMessage);
2001 GNUNET_assert (msize < GNUNET_SERVER_MAX_MESSAGE_SIZE);
2002 lgc->result = GNUNET_malloc (msize);
2003 lgc->result->header.size = htons (msize);
2004 lgc->result->header.type = htons (GNUNET_MESSAGE_TYPE_FS_CONTENT);
2005 lgc->result->type = htonl (type);
2006 lgc->result->expiration = GNUNET_TIME_absolute_hton (expiration);
2007 memcpy (&lgc->result[1],
2010 GNUNET_SERVER_notify_transmit_ready (lgc->client,
2012 GNUNET_TIME_UNIT_FOREVER_REL,
2013 &transmit_local_result,
2019 * We're processing a search request from a local
2020 * client. Now it is our turn to query the datastore.
2022 * @param cls our closure (struct LocalGetContext)
2026 transmit_local_get (void *cls,
2027 const struct GNUNET_SCHEDULER_TaskContext *tc)
2029 struct LocalGetContext *lgc = cls;
2033 if (type == GNUNET_DATASTORE_BLOCKTYPE_DBLOCK)
2034 type = GNUNET_DATASTORE_BLOCKTYPE_ANY; /* to get on-demand as well */
2035 GNUNET_DATASTORE_get (dsh,
2038 &process_local_get_result,
2040 GNUNET_TIME_UNIT_FOREVER_REL);
2045 * We're processing a search request from a local
2046 * client. Now it is our turn to query the datastore.
2048 * @param cls our closure (struct LocalGetContext)
2049 * @param ok did we succeed to queue for datastore access, should always be GNUNET_OK
2052 transmit_local_get_ready (void *cls,
2055 struct LocalGetContext *lgc = cls;
2057 GNUNET_assert (GNUNET_OK == ok);
2058 GNUNET_SCHEDULER_add_continuation (sched,
2060 &transmit_local_get,
2062 GNUNET_SCHEDULER_REASON_PREREQ_DONE);
2067 * Handle START_SEARCH-message (search request from client).
2069 * @param cls closure
2070 * @param client identification of the client
2071 * @param message the actual message
2074 handle_start_search (void *cls,
2075 struct GNUNET_SERVER_Client *client,
2076 const struct GNUNET_MessageHeader *message)
2078 const struct SearchMessage *sm;
2079 struct LocalGetContext *lgc;
2083 msize = ntohs (message->size);
2084 if ( (msize < sizeof (struct SearchMessage)) ||
2085 (0 != (msize - sizeof (struct SearchMessage)) % sizeof (GNUNET_HashCode)) )
2088 GNUNET_SERVER_receive_done (client,
2092 sc = (msize - sizeof (struct SearchMessage)) / sizeof (GNUNET_HashCode);
2093 sm = (const struct SearchMessage*) message;
2094 GNUNET_SERVER_client_keep (client);
2095 lgc = GNUNET_malloc (sizeof (struct LocalGetContext));
2098 lgc->results_used = sc;
2099 GNUNET_array_grow (lgc->results,
2102 memcpy (lgc->results,
2104 sc * sizeof (GNUNET_HashCode));
2106 lgc->client = client;
2107 lgc->type = ntohl (sm->type);
2108 lgc->anonymity_level = ntohl (sm->anonymity_level);
2111 case GNUNET_DATASTORE_BLOCKTYPE_DBLOCK:
2112 case GNUNET_DATASTORE_BLOCKTYPE_IBLOCK:
2113 lgc->target.hashPubKey = sm->target;
2115 case GNUNET_DATASTORE_BLOCKTYPE_SBLOCK:
2116 lgc->namespace = sm->target;
2121 lgc->query = sm->query;
2122 GNUNET_CONTAINER_DLL_insert (lgc_head, lgc_tail, lgc);
2123 lgc->req = queue_ds_request (GNUNET_TIME_UNIT_FOREVER_REL,
2124 &transmit_local_get_ready,
2130 * List of handlers for the messages understood by this
2133 static struct GNUNET_SERVER_MessageHandler handlers[] = {
2134 {&handle_index_start, NULL,
2135 GNUNET_MESSAGE_TYPE_FS_INDEX_START, 0},
2136 {&handle_index_list_get, NULL,
2137 GNUNET_MESSAGE_TYPE_FS_INDEX_LIST_GET, sizeof(struct GNUNET_MessageHeader) },
2138 {&handle_unindex, NULL, GNUNET_MESSAGE_TYPE_FS_UNINDEX,
2139 sizeof (struct UnindexMessage) },
2140 {&handle_start_search, NULL, GNUNET_MESSAGE_TYPE_FS_START_SEARCH,
2147 * Clean up the memory used by the PendingRequest structure (except
2148 * for the client or peer list that the request may be part of).
2150 * @param pr request to clean up
2153 destroy_pending_request (struct PendingRequest *pr)
2155 struct PendingReply *reply;
2156 struct ClientList *cl;
2158 GNUNET_CONTAINER_multihashmap_remove (requests_by_query,
2161 // FIXME: not sure how this can work (efficiently)
2162 // also, what does the return value mean?
2163 if (pr->client == NULL)
2165 GNUNET_CONTAINER_heap_remove_node (requests_by_expiration,
2170 cl = pr->crl_entry->cl;
2171 GNUNET_CONTAINER_DLL_remove (cl->head,
2175 if (GNUNET_SCHEDULER_NO_TASK != pr->task)
2176 GNUNET_SCHEDULER_cancel (sched, pr->task);
2177 if (NULL != pr->cth)
2178 GNUNET_CORE_notify_transmit_ready_cancel (pr->cth);
2180 GNUNET_CONTAINER_bloomfilter_free (pr->bf);
2182 GNUNET_CONNECTION_notify_transmit_ready_cancel (pr->th);
2183 while (NULL != (reply = pr->replies_pending))
2185 pr->replies_pending = reply->next;
2186 GNUNET_free (reply);
2188 GNUNET_PEER_change_rc (pr->source_pid, -1);
2189 GNUNET_PEER_change_rc (pr->target_pid, -1);
2190 GNUNET_PEER_decrement_rcs (pr->used_pids, pr->used_pids_off);
2191 GNUNET_free_non_null (pr->used_pids);
2192 GNUNET_free_non_null (pr->replies_seen);
2193 GNUNET_free_non_null (pr->namespace);
2199 * A client disconnected. Remove all of its pending queries.
2201 * @param cls closure, NULL
2202 * @param client identification of the client
2205 handle_client_disconnect (void *cls,
2206 struct GNUNET_SERVER_Client
2209 struct LocalGetContext *lgc;
2210 struct ClientList *cpos;
2211 struct ClientList *cprev;
2212 struct ClientRequestList *rl;
2215 while ( (NULL != lgc) &&
2216 (lgc->client != client) )
2219 local_get_context_free (lgc);
2222 while ( (NULL != cpos) &&
2223 (clients->client != client) )
2231 clients = cpos->next;
2233 cprev->next = cpos->next;
2234 while (NULL != (rl = cpos->head))
2236 cpos->head = rl->next;
2237 destroy_pending_request (rl->req);
2246 * Iterator over entries in the "requests_by_query" map
2247 * that frees all the entries.
2249 * @param cls closure, NULL
2250 * @param key current key code (the query, unused)
2251 * @param value value in the hash map, of type "struct PendingRequest*"
2252 * @return GNUNET_YES (we should continue to iterate)
2255 destroy_pending_request_cb (void *cls,
2256 const GNUNET_HashCode * key,
2259 struct PendingRequest *pr = value;
2261 destroy_pending_request (pr);
2267 * Task run during shutdown.
2273 shutdown_task (void *cls,
2274 const struct GNUNET_SCHEDULER_TaskContext *tc)
2276 struct IndexInfo *pos;
2279 GNUNET_CORE_disconnect (core);
2280 GNUNET_DATASTORE_disconnect (dsh,
2283 GNUNET_CONTAINER_multihashmap_iterate (requests_by_query,
2284 &destroy_pending_request_cb,
2286 while (clients != NULL)
2287 handle_client_disconnect (NULL,
2289 GNUNET_CONTAINER_multihashmap_destroy (requests_by_query);
2290 requests_by_query = NULL;
2291 GNUNET_CONTAINER_multihashmap_destroy (requests_by_peer);
2292 requests_by_peer = NULL;
2293 GNUNET_CONTAINER_heap_destroy (requests_by_expiration);
2294 requests_by_expiration = NULL;
2295 // FIXME: iterate over entries and free individually?
2296 // (or do we get disconnect notifications?)
2297 GNUNET_CONTAINER_multihashmap_destroy (connected_peers);
2298 connected_peers = NULL;
2299 GNUNET_CONTAINER_multihashmap_destroy (ifm);
2301 while (NULL != (pos = indexed_files))
2303 indexed_files = pos->next;
2310 * Free (each) request made by the peer.
2312 * @param cls closure, points to peer that the request belongs to
2313 * @param key current key code
2314 * @param value value in the hash map
2315 * @return GNUNET_YES (we should continue to iterate)
2318 destroy_request (void *cls,
2319 const GNUNET_HashCode * key,
2322 const struct GNUNET_PeerIdentity * peer = cls;
2323 struct PendingRequest *pr = value;
2325 GNUNET_CONTAINER_multihashmap_remove (requests_by_peer,
2328 destroy_pending_request (pr);
2335 * Method called whenever a given peer connects.
2337 * @param cls closure, not used
2338 * @param peer peer identity this notification is about
2341 peer_connect_handler (void *cls,
2343 GNUNET_PeerIdentity * peer)
2345 struct ConnectedPeer *cp;
2347 cp = GNUNET_malloc (sizeof (struct ConnectedPeer));
2348 cp->pid = GNUNET_PEER_intern (peer);
2349 GNUNET_CONTAINER_multihashmap_put (connected_peers,
2352 GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY);
2357 * Method called whenever a peer disconnects.
2359 * @param cls closure, not used
2360 * @param peer peer identity this notification is about
2363 peer_disconnect_handler (void *cls,
2365 GNUNET_PeerIdentity * peer)
2367 struct ConnectedPeer *cp;
2369 cp = GNUNET_CONTAINER_multihashmap_get (connected_peers,
2371 GNUNET_PEER_change_rc (cp->pid, -1);
2372 GNUNET_PEER_decrement_rcs (cp->last_p2p_replies, P2P_SUCCESS_LIST_SIZE);
2374 GNUNET_CONTAINER_multihashmap_get_multiple (requests_by_peer,
2382 * We're processing a GET request from
2383 * another peer and have decided to forward
2384 * it to other peers.
2386 * @param cls our "struct ProcessGetContext *"
2390 forward_get_request (void *cls,
2391 const struct GNUNET_SCHEDULER_TaskContext *tc)
2393 struct ProcessGetContext *pgc = cls;
2394 struct PendingRequest *pr;
2395 struct PendingRequest *eer;
2396 struct GNUNET_PeerIdentity target;
2398 pr = GNUNET_malloc (sizeof (struct PendingRequest));
2399 if (GET_MESSAGE_BIT_SKS_NAMESPACE == (GET_MESSAGE_BIT_SKS_NAMESPACE & pgc->bm))
2401 pr->namespace = GNUNET_malloc (sizeof(GNUNET_HashCode));
2402 *pr->namespace = pgc->namespace;
2405 pr->bf_size = pgc->bf_size;
2407 pr->start_time = pgc->start_time;
2408 pr->query = pgc->query;
2409 pr->source_pid = GNUNET_PEER_intern (&pgc->reply_to);
2410 if (GET_MESSAGE_BIT_TRANSMIT_TO == (GET_MESSAGE_BIT_TRANSMIT_TO & pgc->bm))
2411 pr->target_pid = GNUNET_PEER_intern (&pgc->prime_target);
2412 pr->anonymity_level = 1; /* default */
2413 pr->priority = pgc->priority;
2414 pr->remaining_priority = pr->priority;
2415 pr->mingle = pgc->mingle;
2417 pr->type = pgc->type;
2418 GNUNET_CONTAINER_multihashmap_put (requests_by_query,
2421 GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
2422 GNUNET_CONTAINER_multihashmap_put (requests_by_peer,
2423 &pgc->reply_to.hashPubKey,
2425 GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
2426 GNUNET_CONTAINER_heap_insert (requests_by_expiration,
2428 pr->start_time.value + pr->ttl);
2429 if (GNUNET_CONTAINER_heap_get_size (requests_by_expiration) > max_pending_requests)
2431 /* expire oldest request! */
2432 eer = GNUNET_CONTAINER_heap_peek (requests_by_expiration);
2433 GNUNET_PEER_resolve (eer->source_pid,
2435 GNUNET_CONTAINER_multihashmap_remove (requests_by_peer,
2438 destroy_pending_request (eer);
2440 pr->task = GNUNET_SCHEDULER_add_delayed (sched,
2442 GNUNET_SCHEDULER_PRIORITY_IDLE,
2443 GNUNET_SCHEDULER_NO_TASK,
2444 get_processing_delay (),
2445 &forward_request_task,
2450 * Transmit the given message by copying it to
2451 * the target buffer "buf". "buf" will be
2452 * NULL and "size" zero if the socket was closed for
2453 * writing in the meantime. In that case, only
2457 * @param cls closure, pointer to the message
2458 * @param size number of bytes available in buf
2459 * @param buf where the callee should write the message
2460 * @return number of bytes written to buf
2463 transmit_message (void *cls,
2464 size_t size, void *buf)
2466 struct GNUNET_MessageHeader *msg = cls;
2472 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2473 "Dropping reply, core too busy.\n");
2478 msize = ntohs (msg->size);
2479 GNUNET_assert (size >= msize);
2480 memcpy (buf, msg, msize);
2487 * Test if the load on this peer is too high
2488 * to even consider processing the query at
2491 * @return GNUNET_YES if the load is too high, GNUNET_NO otherwise
2494 test_load_too_high ()
2496 return GNUNET_NO; // FIXME
2501 * We're processing (local) results for a search request
2502 * from another peer. Pass applicable results to the
2503 * peer and if we are done either clean up (operation
2504 * complete) or forward to other peers (more results possible).
2506 * @param cls our closure (struct LocalGetContext)
2507 * @param key key for the content
2508 * @param size number of bytes in data
2509 * @param data content stored
2510 * @param type type of the content
2511 * @param priority priority of the content
2512 * @param anonymity anonymity-level for the content
2513 * @param expiration expiration time for the content
2514 * @param uid unique identifier for the datum;
2515 * maybe 0 if no unique identifier is available
2518 process_p2p_get_result (void *cls,
2519 const GNUNET_HashCode * key,
2525 struct GNUNET_TIME_Absolute
2529 struct ProcessGetContext *pgc = cls;
2530 GNUNET_HashCode dhash;
2531 GNUNET_HashCode mhash;
2532 struct PutMessage *reply;
2536 /* no more results */
2537 if ( ( (pgc->policy & ROUTING_POLICY_FORWARD) == ROUTING_POLICY_FORWARD) &&
2538 ( (0 == pgc->results_found) ||
2539 (pgc->type == GNUNET_DATASTORE_BLOCKTYPE_KBLOCK) ||
2540 (pgc->type == GNUNET_DATASTORE_BLOCKTYPE_SBLOCK) ||
2541 (pgc->type == GNUNET_DATASTORE_BLOCKTYPE_SKBLOCK) ) )
2543 GNUNET_SCHEDULER_add_continuation (sched,
2545 &forward_get_request,
2547 GNUNET_SCHEDULER_REASON_PREREQ_DONE);
2551 if (pgc->bf != NULL)
2552 GNUNET_CONTAINER_bloomfilter_free (pgc->bf);
2558 if (type == GNUNET_DATASTORE_BLOCKTYPE_ONDEMAND)
2560 handle_on_demand_block (key, size, data, type, priority,
2561 anonymity, expiration, uid,
2562 &process_p2p_get_result,
2566 /* check for duplicates */
2567 GNUNET_CRYPTO_hash (data, size, &dhash);
2568 mingle_hash (&dhash,
2571 if ( (pgc->bf != NULL) &&
2573 GNUNET_CONTAINER_bloomfilter_test (pgc->bf,
2577 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2578 "Result from datastore filtered by bloomfilter.\n");
2580 GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
2583 pgc->results_found++;
2584 if ( (pgc->type == GNUNET_DATASTORE_BLOCKTYPE_KBLOCK) ||
2585 (pgc->type == GNUNET_DATASTORE_BLOCKTYPE_SBLOCK) ||
2586 (pgc->type == GNUNET_DATASTORE_BLOCKTYPE_SKBLOCK) )
2588 if (pgc->bf == NULL)
2591 pgc->bf = GNUNET_CONTAINER_bloomfilter_init (NULL,
2595 GNUNET_CONTAINER_bloomfilter_add (pgc->bf,
2599 reply = GNUNET_malloc (sizeof (struct PutMessage) + size);
2600 reply->header.size = htons (sizeof (struct PutMessage) + size);
2601 reply->header.type = htons (GNUNET_MESSAGE_TYPE_FS_PUT);
2602 reply->type = htonl (type);
2603 reply->expiration = GNUNET_TIME_relative_hton (GNUNET_TIME_absolute_get_remaining (expiration));
2604 memcpy (&reply[1], data, size);
2605 GNUNET_CORE_notify_transmit_ready (core,
2607 ACCEPTABLE_REPLY_DELAY,
2609 sizeof (struct PutMessage) + size,
2612 if ( (GNUNET_YES == test_load_too_high()) ||
2613 (pgc->results_found > 5 + 2 * pgc->priority) )
2615 GNUNET_DATASTORE_get_next (dsh, GNUNET_NO);
2616 pgc->policy &= ~ ROUTING_POLICY_FORWARD;
2619 GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
2624 * We're processing a GET request from another peer. Give it to our
2627 * @param cls our "struct ProcessGetContext"
2628 * @param ok did we get a datastore slice or not?
2631 ds_get_request (void *cls,
2634 struct ProcessGetContext *pgc = cls;
2636 struct GNUNET_TIME_Relative timeout;
2638 if (GNUNET_OK != ok)
2640 /* no point in doing P2P stuff if we can't even do local */
2645 if (type == GNUNET_DATASTORE_BLOCKTYPE_DBLOCK)
2646 type = GNUNET_DATASTORE_BLOCKTYPE_ANY; /* to get on-demand as well */
2647 timeout = GNUNET_TIME_relative_multiply (BASIC_DATASTORE_REQUEST_DELAY,
2648 (pgc->priority + 1));
2649 GNUNET_DATASTORE_get (dsh,
2652 &process_p2p_get_result,
2659 * The priority level imposes a bound on the maximum
2660 * value for the ttl that can be requested.
2662 * @param ttl_in requested ttl
2663 * @param prio given priority
2664 * @return ttl_in if ttl_in is below the limit,
2665 * otherwise the ttl-limit for the given priority
2668 bound_ttl (int32_t ttl_in, uint32_t prio)
2670 unsigned long long allowed;
2674 allowed = ((unsigned long long) prio) * TTL_DECREMENT / 1000;
2675 if (ttl_in > allowed)
2677 if (allowed >= (1 << 30))
2686 * We've received a request with the specified
2687 * priority. Bound it according to how much
2688 * we trust the given peer.
2690 * @param prio_in requested priority
2691 * @param peer the peer making the request
2692 * @return effective priority
2695 bound_priority (uint32_t prio_in,
2696 const struct GNUNET_PeerIdentity *peer)
2703 * Handle P2P "GET" request.
2705 * @param cls closure, always NULL
2706 * @param other the other peer involved (sender or receiver, NULL
2707 * for loopback messages where we are both sender and receiver)
2708 * @param message the actual message
2709 * @return GNUNET_OK to keep the connection open,
2710 * GNUNET_SYSERR to close it (signal serious error)
2713 handle_p2p_get (void *cls,
2714 const struct GNUNET_PeerIdentity *other,
2715 const struct GNUNET_MessageHeader *message)
2718 const struct GetMessage *gm;
2720 const GNUNET_HashCode *opt;
2721 struct ProcessGetContext *pgc;
2724 uint32_t ttl_decrement;
2729 msize = ntohs(message->size);
2730 if (msize < sizeof (struct GetMessage))
2732 GNUNET_break_op (0);
2733 return GNUNET_SYSERR;
2735 gm = (const struct GetMessage*) message;
2736 bm = ntohl (gm->hash_bitmap);
2744 if (msize < sizeof (struct GetMessage) + bits * sizeof (GNUNET_HashCode))
2746 GNUNET_break_op (0);
2747 return GNUNET_SYSERR;
2749 opt = (const GNUNET_HashCode*) &gm[1];
2750 bfsize = msize - sizeof (struct GetMessage) + bits * sizeof (GNUNET_HashCode);
2751 pgc = GNUNET_malloc (sizeof (struct ProcessGetContext));
2754 pgc->bf = GNUNET_CONTAINER_bloomfilter_init ((const char*) &pgc[1],
2757 pgc->bf_size = bfsize;
2759 pgc->type = ntohl (gm->type);
2760 pgc->bm = ntohl (gm->hash_bitmap);
2761 pgc->mingle = gm->filter_mutator;
2763 if (0 != (pgc->bm & GET_MESSAGE_BIT_RETURN_TO))
2764 pgc->reply_to.hashPubKey = opt[bits++];
2766 pgc->reply_to = *other;
2767 if (0 != (pgc->bm & GET_MESSAGE_BIT_SKS_NAMESPACE))
2768 pgc->namespace = opt[bits++];
2769 else if (pgc->type == GNUNET_DATASTORE_BLOCKTYPE_SBLOCK)
2771 GNUNET_break_op (0);
2773 return GNUNET_SYSERR;
2775 if (0 != (pgc->bm & GET_MESSAGE_BIT_TRANSMIT_TO))
2776 pgc->prime_target.hashPubKey = opt[bits++];
2777 /* note that we can really only check load here since otherwise
2778 peers could find out that we are overloaded by being disconnected
2779 after sending us a malformed query... */
2780 if (GNUNET_YES == test_load_too_high ())
2782 if (NULL != pgc->bf)
2783 GNUNET_CONTAINER_bloomfilter_free (pgc->bf);
2786 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2787 "Dropping query from `%s', this peer is too busy.\n",
2788 GNUNET_h2s (other));
2792 net_load_up = 50; // FIXME
2793 net_load_down = 50; // FIXME
2794 pgc->policy = ROUTING_POLICY_NONE;
2795 if ( (net_load_up < IDLE_LOAD_THRESHOLD) &&
2796 (net_load_down < IDLE_LOAD_THRESHOLD) )
2798 pgc->policy |= ROUTING_POLICY_ALL;
2799 pgc->priority = 0; /* no charge */
2803 pgc->priority = bound_priority (ntohl (gm->priority), other);
2805 IDLE_LOAD_THRESHOLD + pgc->priority * pgc->priority) &&
2807 IDLE_LOAD_THRESHOLD + pgc->priority * pgc->priority) )
2809 pgc->policy |= ROUTING_POLICY_ALL;
2813 // FIXME: is this sound?
2814 if (net_load_up < 90 + 10 * pgc->priority)
2815 pgc->policy |= ROUTING_POLICY_FORWARD;
2816 if (net_load_down < 90 + 10 * pgc->priority)
2817 pgc->policy |= ROUTING_POLICY_ANSWER;
2820 if (pgc->policy == ROUTING_POLICY_NONE)
2823 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2824 "Dropping query from `%s', network saturated.\n",
2825 GNUNET_h2s (other));
2827 if (NULL != pgc->bf)
2828 GNUNET_CONTAINER_bloomfilter_free (pgc->bf);
2830 return GNUNET_OK; /* drop */
2832 if ((pgc->policy & ROUTING_POLICY_INDIRECT) != ROUTING_POLICY_INDIRECT)
2833 pgc->priority = 0; /* kill the priority (we cannot benefit) */
2834 pgc->ttl = bound_ttl (ntohl (gm->ttl), pgc->priority);
2835 /* decrement ttl (always) */
2836 ttl_decrement = 2 * TTL_DECREMENT +
2837 GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
2839 if ( (pgc->ttl < 0) &&
2840 (pgc->ttl - ttl_decrement > 0) )
2843 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2844 "Dropping query from `%s' due to TTL underflow.\n",
2845 GNUNET_h2s (other));
2847 /* integer underflow => drop (should be very rare)! */
2848 if (NULL != pgc->bf)
2849 GNUNET_CONTAINER_bloomfilter_free (pgc->bf);
2853 pgc->ttl -= ttl_decrement;
2854 pgc->start_time = GNUNET_TIME_absolute_get ();
2855 preference = (double) pgc->priority;
2856 if (preference < QUERY_BANDWIDTH_VALUE)
2857 preference = QUERY_BANDWIDTH_VALUE;
2858 // FIXME: also reserve bandwidth for reply?
2859 GNUNET_CORE_peer_configure (core,
2861 GNUNET_TIME_UNIT_FOREVER_REL,
2862 0, 0, preference, NULL, NULL);
2863 if (0 != (pgc->policy & ROUTING_POLICY_ANSWER))
2864 pgc->drq = queue_ds_request (BASIC_DATASTORE_REQUEST_DELAY,
2868 GNUNET_SCHEDULER_add_continuation (sched,
2870 &forward_get_request,
2872 GNUNET_SCHEDULER_REASON_PREREQ_DONE);
2878 * Function called to notify us that we can now transmit a reply to a
2879 * client or peer. "buf" will be NULL and "size" zero if the socket was
2880 * closed for writing in the meantime.
2882 * @param cls closure, points to a "struct PendingRequest*" with
2883 * one or more pending replies
2884 * @param size number of bytes available in buf
2885 * @param buf where the callee should write the message
2886 * @return number of bytes written to buf
2889 transmit_result (void *cls,
2893 struct PendingRequest *pr = cls;
2895 struct PendingReply *reply;
2899 while (NULL != (reply = pr->replies_pending))
2901 if ( (reply->msize + ret < ret) ||
2902 (reply->msize + ret > size) )
2904 pr->replies_pending = reply->next;
2905 memcpy (&cbuf[ret], &reply[1], reply->msize);
2906 ret += reply->msize;
2914 * Iterator over pending requests.
2916 * @param cls response (struct ProcessReplyClosure)
2917 * @param key our query
2918 * @param value value in the hash map (meta-info about the query)
2919 * @return GNUNET_YES (we should continue to iterate)
2922 process_reply (void *cls,
2923 const GNUNET_HashCode * key,
2926 struct ProcessReplyClosure *prq = cls;
2927 struct PendingRequest *pr = value;
2928 struct PendingRequest *eer;
2929 struct PendingReply *reply;
2930 struct PutMessage *pm;
2931 struct ContentMessage *cm;
2932 GNUNET_HashCode chash;
2933 GNUNET_HashCode mhash;
2934 struct GNUNET_PeerIdentity target;
2937 struct GNUNET_TIME_Relative max_delay;
2939 GNUNET_CRYPTO_hash (prq->data,
2944 case GNUNET_DATASTORE_BLOCKTYPE_DBLOCK:
2945 case GNUNET_DATASTORE_BLOCKTYPE_IBLOCK:
2947 case GNUNET_DATASTORE_BLOCKTYPE_SBLOCK:
2948 /* FIXME: does prq->namespace match our expectations? */
2949 /* then: fall-through??? */
2950 case GNUNET_DATASTORE_BLOCKTYPE_KBLOCK:
2953 mingle_hash (&chash, pr->mingle, &mhash);
2954 if (GNUNET_YES == GNUNET_CONTAINER_bloomfilter_test (pr->bf,
2956 return GNUNET_YES; /* duplicate */
2957 GNUNET_CONTAINER_bloomfilter_add (pr->bf,
2961 case GNUNET_DATASTORE_BLOCKTYPE_SKBLOCK:
2962 // FIXME: any checks against duplicates for SKBlocks?
2965 prio = pr->priority;
2966 prq->priority += pr->remaining_priority;
2967 pr->remaining_priority = 0;
2968 if (pr->client != NULL)
2970 if (pr->replies_seen_size == pr->replies_seen_off)
2971 GNUNET_array_grow (pr->replies_seen,
2972 pr->replies_seen_size,
2973 pr->replies_seen_size * 2 + 4);
2974 pr->replies_seen[pr->replies_seen_off++] = chash;
2975 // FIXME: possibly recalculate BF!
2977 if (pr->client == NULL)
2979 msize = sizeof (struct ContentMessage) + prq->size;
2980 reply = GNUNET_malloc (msize + sizeof (struct PendingReply));
2981 reply->msize = msize;
2982 cm = (struct ContentMessage*) &reply[1];
2983 cm->header.type = htons (GNUNET_MESSAGE_TYPE_FS_CONTENT);
2984 cm->header.size = htons (msize);
2985 cm->type = htonl (prq->type);
2986 cm->expiration = GNUNET_TIME_absolute_hton (prq->expiration);
2987 reply->next = pr->replies_pending;
2988 pr->replies_pending = reply;
2989 memcpy (&reply[1], prq->data, prq->size);
2990 if (pr->cth != NULL)
2992 max_delay = GNUNET_TIME_UNIT_FOREVER_REL;
2993 if (GNUNET_CONTAINER_heap_get_size (requests_by_expiration) >= max_pending_requests)
2995 /* estimate expiration time from time difference between
2996 first request that will be discarded and this request */
2997 eer = GNUNET_CONTAINER_heap_peek (requests_by_expiration);
2998 max_delay = GNUNET_TIME_absolute_get_difference (pr->start_time,
3001 GNUNET_PEER_resolve (pr->source_pid,
3003 pr->cth = GNUNET_CORE_notify_transmit_ready (core,
3010 if (NULL == pr->cth)
3012 // FIXME: now what? discard?
3017 msize = sizeof (struct PutMessage) + prq->size;
3018 reply = GNUNET_malloc (msize + sizeof (struct PendingReply));
3019 reply->msize = msize;
3020 reply->next = pr->replies_pending;
3021 pm = (struct PutMessage*) &reply[1];
3022 pm->header.type = htons (GNUNET_MESSAGE_TYPE_FS_PUT);
3023 pm->header.size = htons (msize);
3024 pm->type = htonl (prq->type);
3025 pm->expiration = GNUNET_TIME_relative_hton (GNUNET_TIME_absolute_get_remaining (prq->expiration));
3026 pr->replies_pending = reply;
3027 memcpy (&reply[1], prq->data, prq->size);
3030 pr->th = GNUNET_SERVER_notify_transmit_ready (pr->client,
3032 GNUNET_TIME_UNIT_FOREVER_REL,
3037 // FIXME: need to try again later (not much
3038 // to do here specifically, but we need to
3039 // check somewhere else to handle this case!)
3042 // FIXME: implement hot-path routing statistics keeping!
3048 * Check if the given KBlock is well-formed.
3050 * @param kb the kblock data (or at least "dsize" bytes claiming to be one)
3051 * @param dsize size of "kb" in bytes; check for < sizeof(struct KBlock)!
3052 * @param query where to store the query that this block answers
3053 * @return GNUNET_OK if this is actually a well-formed KBlock
3056 check_kblock (const struct KBlock *kb,
3058 GNUNET_HashCode *query)
3060 if (dsize < sizeof (struct KBlock))
3062 GNUNET_break_op (0);
3063 return GNUNET_SYSERR;
3065 if (dsize - sizeof (struct KBlock) !=
3066 ntohs (kb->purpose.size)
3067 - sizeof (struct GNUNET_CRYPTO_RsaSignaturePurpose)
3068 - sizeof(struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded) )
3070 GNUNET_break_op (0);
3071 return GNUNET_SYSERR;
3074 GNUNET_CRYPTO_rsa_verify (GNUNET_SIGNATURE_PURPOSE_FS_KBLOCK,
3079 GNUNET_break_op (0);
3080 return GNUNET_SYSERR;
3083 GNUNET_CRYPTO_hash (&kb->keyspace,
3084 sizeof (struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded),
3091 * Check if the given SBlock is well-formed.
3093 * @param sb the sblock data (or at least "dsize" bytes claiming to be one)
3094 * @param dsize size of "kb" in bytes; check for < sizeof(struct SBlock)!
3095 * @param query where to store the query that this block answers
3096 * @param namespace where to store the namespace that this block belongs to
3097 * @return GNUNET_OK if this is actually a well-formed SBlock
3100 check_sblock (const struct SBlock *sb,
3102 GNUNET_HashCode *query,
3103 GNUNET_HashCode *namespace)
3105 if (dsize < sizeof (struct SBlock))
3107 GNUNET_break_op (0);
3108 return GNUNET_SYSERR;
3111 ntohs (sb->purpose.size) + sizeof (struct GNUNET_CRYPTO_RsaSignature))
3113 GNUNET_break_op (0);
3114 return GNUNET_SYSERR;
3117 GNUNET_CRYPTO_rsa_verify (GNUNET_SIGNATURE_PURPOSE_FS_SBLOCK,
3122 GNUNET_break_op (0);
3123 return GNUNET_SYSERR;
3126 *query = sb->identifier;
3127 if (namespace != NULL)
3128 GNUNET_CRYPTO_hash (&sb->subspace,
3129 sizeof (struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded),
3136 * Handle P2P "PUT" request.
3138 * @param cls closure, always NULL
3139 * @param other the other peer involved (sender or receiver, NULL
3140 * for loopback messages where we are both sender and receiver)
3141 * @param message the actual message
3142 * @return GNUNET_OK to keep the connection open,
3143 * GNUNET_SYSERR to close it (signal serious error)
3146 handle_p2p_put (void *cls,
3147 const struct GNUNET_PeerIdentity *other,
3148 const struct GNUNET_MessageHeader *message)
3150 const struct PutMessage *put;
3154 struct GNUNET_TIME_Absolute expiration;
3155 GNUNET_HashCode query;
3156 struct ProcessReplyClosure prq;
3158 msize = ntohs (message->size);
3159 if (msize < sizeof (struct PutMessage))
3162 return GNUNET_SYSERR;
3164 put = (const struct PutMessage*) message;
3165 dsize = msize - sizeof (struct PutMessage);
3166 type = ntohl (put->type);
3167 expiration = GNUNET_TIME_relative_to_absolute (GNUNET_TIME_relative_ntoh (put->expiration));
3169 /* first, validate! */
3172 case GNUNET_DATASTORE_BLOCKTYPE_DBLOCK:
3173 case GNUNET_DATASTORE_BLOCKTYPE_IBLOCK:
3174 GNUNET_CRYPTO_hash (&put[1], dsize, &query);
3176 case GNUNET_DATASTORE_BLOCKTYPE_KBLOCK:
3178 check_kblock ((const struct KBlock*) &put[1],
3181 return GNUNET_SYSERR;
3183 case GNUNET_DATASTORE_BLOCKTYPE_SBLOCK:
3185 check_sblock ((const struct SBlock*) &put[1],
3189 return GNUNET_SYSERR;
3191 case GNUNET_DATASTORE_BLOCKTYPE_SKBLOCK:
3192 // FIXME -- validate SKBLOCK!
3196 /* unknown block type */
3197 GNUNET_break_op (0);
3198 return GNUNET_SYSERR;
3201 /* now, lookup 'query' */
3202 prq.data = (const void*) &put[1];
3205 prq.expiration = expiration;
3207 GNUNET_CONTAINER_multihashmap_get_multiple (requests_by_query,
3211 // FIXME: if migration is on and load is low,
3212 // queue to store data in datastore;
3213 // use "prq.priority" for that!
3219 * List of handlers for P2P messages
3220 * that we care about.
3222 static struct GNUNET_CORE_MessageHandler p2p_handlers[] =
3225 GNUNET_MESSAGE_TYPE_FS_GET, 0 },
3227 GNUNET_MESSAGE_TYPE_FS_PUT, 0 },
3233 * Task that will try to initiate a connection with the
3240 core_connect_task (void *cls,
3241 const struct GNUNET_SCHEDULER_TaskContext *tc);
3245 * Function called by the core after we've
3248 * @param cls closure, unused
3249 * @param server handle to the core service
3250 * @param my_identity our peer identity (unused)
3251 * @param publicKey our public key (unused)
3254 core_start_cb (void *cls,
3255 struct GNUNET_CORE_Handle * server,
3256 const struct GNUNET_PeerIdentity *
3259 GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded *
3264 GNUNET_SCHEDULER_add_delayed (sched,
3266 GNUNET_SCHEDULER_PRIORITY_HIGH,
3267 GNUNET_SCHEDULER_NO_TASK,
3268 GNUNET_TIME_UNIT_SECONDS,
3278 * Task that will try to initiate a connection with the
3285 core_connect_task (void *cls,
3286 const struct GNUNET_SCHEDULER_TaskContext *tc)
3288 GNUNET_CORE_connect (sched,
3290 GNUNET_TIME_UNIT_FOREVER_REL,
3293 &peer_connect_handler,
3294 &peer_disconnect_handler,
3303 * Process fs requests.
3305 * @param cls closure
3306 * @param s scheduler to use
3307 * @param server the initialized server
3308 * @param c configuration to use
3312 struct GNUNET_SCHEDULER_Handle *s,
3313 struct GNUNET_SERVER_Handle *server,
3314 const struct GNUNET_CONFIGURATION_Handle *c)
3319 ifm = GNUNET_CONTAINER_multihashmap_create (128);
3320 requests_by_query = GNUNET_CONTAINER_multihashmap_create (128); // FIXME: get size from config
3321 requests_by_peer = GNUNET_CONTAINER_multihashmap_create (128); // FIXME: get size from config
3322 connected_peers = GNUNET_CONTAINER_multihashmap_create (64);
3323 requests_by_expiration = GNUNET_CONTAINER_heap_create (GNUNET_CONTAINER_HEAP_ORDER_MIN);
3325 dsh = GNUNET_DATASTORE_connect (cfg,
3329 GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
3330 _("Failed to connect to datastore service.\n"));
3333 GNUNET_SERVER_disconnect_notify (server,
3334 &handle_client_disconnect,
3336 GNUNET_SERVER_add_handlers (server, handlers);
3337 core_connect_task (NULL, NULL);
3338 GNUNET_SCHEDULER_add_delayed (sched,
3340 GNUNET_SCHEDULER_PRIORITY_IDLE,
3341 GNUNET_SCHEDULER_NO_TASK,
3342 GNUNET_TIME_UNIT_FOREVER_REL,
3349 * The main function for the fs service.
3351 * @param argc number of arguments from the command line
3352 * @param argv command line arguments
3353 * @return 0 ok, 1 on error
3356 main (int argc, char *const *argv)
3358 return (GNUNET_OK ==
3359 GNUNET_SERVICE_run (argc,
3361 "fs", &run, NULL, NULL, NULL)) ? 0 : 1;
3364 /* end of gnunet-service-fs.c */