hacking on fs
[oweals/gnunet.git] / src / fs / gnunet-service-fs.c
1 /*
2      This file is part of GNUnet.
3      (C) 2009 Christian Grothoff (and other contributing authors)
4
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.
9
10      GNUnet is distributed in the hope that it will be useful, but
11      WITHOUT ANY WARRANTY; without even the implied warranty of
12      MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
13      General Public License for more details.
14
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.
19 */
20
21 /**
22  * @file fs/gnunet-service-fs.c
23  * @brief program that provides the file-sharing service
24  * @author Christian Grothoff
25  *
26  * TODO:
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)
39  */
40 #include "platform.h"
41 #include <float.h>
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"
48 #include "fs.h"
49
50 #define DEBUG_FS GNUNET_NO
51
52
53 /**
54  * In-memory information about indexed files (also available
55  * on-disk).
56  */
57 struct IndexInfo
58 {
59   
60   /**
61    * This is a linked list.
62    */
63   struct IndexInfo *next;
64
65   /**
66    * Name of the indexed file.  Memory allocated
67    * at the end of this struct (do not free).
68    */
69   const char *filename;
70
71   /**
72    * Context for transmitting confirmation to client,
73    * NULL if we've done this already.
74    */
75   struct GNUNET_SERVER_TransmitContext *tc;
76   
77   /**
78    * Hash of the contents of the file.
79    */
80   GNUNET_HashCode file_id;
81
82 };
83
84
85 /**
86  * Signature of a function that is called whenever a datastore
87  * request can be processed (or an entry put on the queue times out).
88  *
89  * @param cls closure
90  * @param ok GNUNET_OK if DS is ready, GNUNET_SYSERR on timeout
91  */
92 typedef void (*RequestFunction)(void *cls,
93                                 int ok);
94
95
96 /**
97  * Doubly-linked list of our requests for the datastore.
98  */
99 struct DatastoreRequestQueue
100 {
101
102   /**
103    * This is a doubly-linked list.
104    */
105   struct DatastoreRequestQueue *next;
106
107   /**
108    * This is a doubly-linked list.
109    */
110   struct DatastoreRequestQueue *prev;
111
112   /**
113    * Function to call (will issue the request).
114    */
115   RequestFunction req;
116
117   /**
118    * Closure for req.
119    */
120   void *req_cls;
121
122   /**
123    * When should this request time-out because we don't care anymore?
124    */
125   struct GNUNET_TIME_Absolute timeout;
126     
127   /**
128    * ID of task used for signaling timeout.
129    */
130   GNUNET_SCHEDULER_TaskIdentifier task;
131
132 };
133
134
135 /**
136  * Closure for processing START_SEARCH messages from a client.
137  */
138 struct LocalGetContext
139 {
140
141   /**
142    * This is a doubly-linked list.
143    */
144   struct LocalGetContext *next;
145
146   /**
147    * This is a doubly-linked list.
148    */
149   struct LocalGetContext *prev;
150
151   /**
152    * Client that initiated the search.
153    */
154   struct GNUNET_SERVER_Client *client;
155
156   /**
157    * Array of results that we've already received 
158    * (can be NULL).
159    */
160   GNUNET_HashCode *results; 
161
162   /**
163    * Bloomfilter over all results (for fast query construction);
164    * NULL if we don't have any results.
165    *
166    * FIXME: this member is not used, is that OK? If so, it should
167    * be removed!
168    */
169   struct GNUNET_CONTAINER_BloomFilter *results_bf; 
170
171   /**
172    * DS request associated with this operation.
173    */
174   struct DatastoreRequestQueue *req;
175
176   /**
177    * Current result message to transmit to client (or NULL).
178    */
179   struct ContentMessage *result;
180   
181   /**
182    * Type of the content that we're looking for.
183    * 0 for any.
184    */
185   uint32_t type;
186
187   /**
188    * Desired anonymity level.
189    */
190   uint32_t anonymity_level;
191
192   /**
193    * Number of results actually stored in the results array.
194    */
195   unsigned int results_used;
196   
197   /**
198    * Size of the results array in memory.
199    */
200   unsigned int results_size;
201   
202   /**
203    * Size (in bytes) of the 'results_bf' bloomfilter.
204    *
205    * FIXME: this member is not used, is that OK? If so, it should
206    * be removed!
207    */
208   size_t results_bf_size;
209
210   /**
211    * If the request is for a DBLOCK or IBLOCK, this is the identity of
212    * the peer that is known to have a response.  Set to all-zeros if
213    * such a target is not known (note that even if OUR anonymity
214    * level is >0 we may happen to know the responder's identity;
215    * nevertheless, we should probably not use it for a DHT-lookup
216    * or similar blunt actions in order to avoid exposing ourselves).
217    */
218   struct GNUNET_PeerIdentity target;
219
220   /**
221    * If the request is for an SBLOCK, this is the identity of the
222    * pseudonym to which the SBLOCK belongs. 
223    */
224   GNUNET_HashCode namespace;
225
226   /**
227    * Hash of the keyword (aka query) for KBLOCKs; Hash of
228    * the CHK-encoded block for DBLOCKS and IBLOCKS (aka query)
229    * and hash of the identifier XORed with the target for
230    * SBLOCKS (aka query).
231    */
232   GNUNET_HashCode query;
233
234 };
235
236
237 /**
238  * Possible routing policies for an FS-GET request.
239  */
240 enum RoutingPolicy
241   {
242     /**
243      * Simply drop the request.
244      */
245     ROUTING_POLICY_NONE = 0,
246     
247     /**
248      * Answer it if we can from local datastore.
249      */
250     ROUTING_POLICY_ANSWER = 1,
251
252     /**
253      * Forward the request to other peers (if possible).
254      */
255     ROUTING_POLICY_FORWARD = 2,
256
257     /**
258      * Forward to other peers, and ask them to route
259      * the response via ourselves.
260      */
261     ROUTING_POLICY_INDIRECT = 6,
262     
263     /**
264      * Do everything we could possibly do (that would
265      * make sense).
266      */
267     ROUTING_POLICY_ALL = 7
268   };
269
270
271 /**
272  * Internal context we use for our initial processing
273  * of a GET request.
274  */
275 struct ProcessGetContext
276 {
277   /**
278    * The search query (used for datastore lookup).
279    */
280   GNUNET_HashCode query;
281   
282   /**
283    * Which peer we should forward the response to.
284    */
285   struct GNUNET_PeerIdentity reply_to;
286
287   /**
288    * Namespace for the result (only set for SKS requests)
289    */
290   GNUNET_HashCode namespace;
291
292   /**
293    * Peer that we should forward the query to if possible
294    * (since that peer likely has the content).
295    */
296   struct GNUNET_PeerIdentity prime_target;
297
298   /**
299    * When did we receive this request?
300    */
301   struct GNUNET_TIME_Absolute start_time;
302
303   /**
304    * Our entry in the DRQ (non-NULL while we wait for our
305    * turn to interact with the local database).
306    */
307   struct DatastoreRequestQueue *drq;
308
309   /**
310    * Filter used to eliminate duplicate results.  Can be NULL if we
311    * are not yet filtering any results.
312    */
313   struct GNUNET_CONTAINER_BloomFilter *bf;
314
315   /**
316    * Bitmap describing which of the optional
317    * hash codes / peer identities were given to us.
318    */
319   uint32_t bm;
320
321   /**
322    * Desired block type.
323    */
324   uint32_t type;
325
326   /**
327    * Priority of the request.
328    */
329   uint32_t priority;
330
331   /**
332    * Size of the 'bf' (in bytes).
333    */
334   size_t bf_size;
335
336   /**
337    * In what ways are we going to process
338    * the request?
339    */
340   enum RoutingPolicy policy;
341
342   /**
343    * Time-to-live for the request (value
344    * we use).
345    */
346   int32_t ttl;
347
348   /**
349    * Number to mingle hashes for bloom-filter
350    * tests with.
351    */
352   int32_t mingle;
353
354   /**
355    * Number of results that were found so far.
356    */
357   unsigned int results_found;
358 };
359
360
361 /**
362  * Information we keep for each pending reply.
363  */
364 struct PendingReply
365 {
366   /**
367    * This is a linked list.
368    */
369   struct PendingReply *next;
370
371   /**
372    * Size of the reply; actual reply message follows
373    * at the end of this struct.
374    */
375   size_t msize;
376
377 };
378
379
380 /**
381  * All requests from a client are kept in a doubly-linked list.
382  */
383 struct ClientRequestList;
384
385
386 /**
387  * Information we keep for each pending request.  We should try to
388  * keep this struct as small as possible since its memory consumption
389  * is key to how many requests we can have pending at once.
390  */
391 struct PendingRequest
392 {
393
394   /**
395    * ID of a client making a request, NULL if this entry is for a
396    * peer.
397    */
398   struct GNUNET_SERVER_Client *client;
399
400   /**
401    * If this request was made by a client,
402    * this is our entry in the client request
403    * list; otherwise NULL.
404    */
405   struct ClientRequestList *crl_entry;
406
407   /**
408    * If this is a namespace query, pointer to the hash of the public
409    * key of the namespace; otherwise NULL.
410    */
411   GNUNET_HashCode *namespace;
412
413   /**
414    * Bloomfilter we use to filter out replies that we don't care about
415    * (anymore).  NULL as long as we are interested in all replies.
416    */
417   struct GNUNET_CONTAINER_BloomFilter *bf;
418
419   /**
420    * Replies that we have received but were unable to forward yet
421    * (typically non-null only if we have a pending transmission
422    * request with the client or the respective peer).
423    */
424   struct PendingReply *replies_pending;
425
426   /**
427    * Pending transmission request with the core service for the target
428    * peer (for processing of 'replies_pending') or Handle for a
429    * pending query-request for P2P-transmission with the core service.
430    * If non-NULL, this request must be cancelled should this struct be
431    * destroyed!
432    */
433   struct GNUNET_CORE_TransmitHandle *cth;
434
435   /**
436    * Pending transmission request for the target client (for processing of
437    * 'replies_pending').
438    */
439   struct GNUNET_CONNECTION_TransmitHandle *th;
440
441   /**
442    * Hash code of all replies that we have seen so far (only valid
443    * if client is not NULL since we only track replies like this for
444    * our own clients).
445    */
446   GNUNET_HashCode *replies_seen;
447
448   /**
449    * When did we first see this request (form this peer), or, if our
450    * client is initiating, when did we last initiate a search?
451    */
452   struct GNUNET_TIME_Absolute start_time;
453
454   /**
455    * The query that this request is for.
456    */
457   GNUNET_HashCode query;
458
459   /**
460    * The task responsible for transmitting queries
461    * for this request.
462    */
463   GNUNET_SCHEDULER_TaskIdentifier task;
464
465   /**
466    * (Interned) Peer identifier (only valid if "client" is NULL)
467    * that identifies a peer that gave us this request.
468    */
469   GNUNET_PEER_Id source_pid;
470
471   /**
472    * (Interned) Peer identifier that identifies a preferred target
473    * for requests.
474    */
475   GNUNET_PEER_Id target_pid;
476
477   /**
478    * (Interned) Peer identifiers of peers that have already
479    * received our query for this content.
480    */
481   GNUNET_PEER_Id *used_pids;
482
483   /**
484    * Size of the 'bf' (in bytes).
485    */
486   size_t bf_size;
487
488   /**
489    * Desired anonymity level; only valid for requests from a local client.
490    */
491   uint32_t anonymity_level;
492
493   /**
494    * How many entries in "used_pids" are actually valid?
495    */
496   unsigned int used_pids_off;
497
498   /**
499    * How long is the "used_pids" array?
500    */
501   unsigned int used_pids_size;
502
503   /**
504    * How many entries in "replies_seen" are actually valid?
505    */
506   unsigned int replies_seen_off;
507
508   /**
509    * How long is the "replies_seen" array?
510    */
511   unsigned int replies_seen_size;
512   
513   /**
514    * Priority with which this request was made.  If one of our clients
515    * made the request, then this is the current priority that we are
516    * using when initiating the request.  This value is used when
517    * we decide to reward other peers with trust for providing a reply.
518    */
519   uint32_t priority;
520
521   /**
522    * Priority points left for us to spend when forwarding this request
523    * to other peers.
524    */
525   uint32_t remaining_priority;
526
527   /**
528    * Number to mingle hashes for bloom-filter
529    * tests with.
530    */
531   int32_t mingle;
532
533   /**
534    * TTL with which we saw this request (or, if we initiated, TTL that
535    * we used for the request).
536    */
537   int32_t ttl;
538   
539   /**
540    * Type of the content that this request is for.
541    */
542   uint32_t type;
543
544 };
545
546
547 /**
548  * All requests from a client are kept in a doubly-linked list.
549  */
550 struct ClientRequestList
551 {
552   /**
553    * This is a doubly-linked list.
554    */
555   struct ClientRequestList *next;
556
557   /**
558    * This is a doubly-linked list.
559    */ 
560   struct ClientRequestList *prev;
561
562   /**
563    * A request from this client.
564    */
565   struct PendingRequest *req;
566
567   /**
568    * Client list with the head and tail of this DLL.
569    */
570   struct ClientList *cl;
571 };
572
573
574 /**
575  * Linked list of all clients that we are 
576  * currently processing requests for.
577  */
578 struct ClientList
579 {
580
581   /**
582    * This is a linked list.
583    */
584   struct ClientList *next;
585
586   /**
587    * What client is this entry for?
588    */
589   struct GNUNET_SERVER_Client* client;
590
591   /**
592    * Head of the DLL of requests from this client.
593    */
594   struct ClientRequestList *head;
595
596   /**
597    * Tail of the DLL of requests from this client.
598    */
599   struct ClientRequestList *tail;
600
601 };
602
603
604 /**
605  * Closure for "process_reply" function.
606  */
607 struct ProcessReplyClosure
608 {
609   /**
610    * The data for the reply.
611    */
612   const void *data;
613
614   /**
615    * When the reply expires.
616    */
617   struct GNUNET_TIME_Absolute expiration;
618
619   /**
620    * Size of data.
621    */
622   size_t size;
623
624   /**
625    * Namespace that this reply belongs to
626    * (if it is of type SBLOCK).
627    */
628   GNUNET_HashCode namespace;
629
630   /**
631    * Type of the block.
632    */
633   uint32_t type;
634
635   /**
636    * How much was this reply worth to us?
637    */
638   uint32_t priority;
639 };
640
641
642 /**
643  * Information about a peer that we are connected to.
644  * We track data that is useful for determining which
645  * peers should receive our requests.
646  */
647 struct ConnectedPeer
648 {
649
650   /**
651    * List of the last clients for which this peer
652    * successfully answered a query. 
653    */
654   struct GNUNET_SERVER_Client *last_client_replies[CS2P_SUCCESS_LIST_SIZE];
655
656   /**
657    * List of the last PIDs for which
658    * this peer successfully answered a query;
659    * We use 0 to indicate no successful reply.
660    */
661   GNUNET_PEER_Id last_p2p_replies[P2P_SUCCESS_LIST_SIZE];
662
663   /**
664    * Average delay between sending the peer a request and
665    * getting a reply (only calculated over the requests for
666    * which we actually got a reply).   Calculated
667    * as a moving average: new_delay = ((n-1)*last_delay+curr_delay) / n
668    */ 
669   struct GNUNET_TIME_Relative avg_delay;
670
671   /**
672    * Average priority of successful replies.  Calculated
673    * as a moving average: new_avg = ((n-1)*last_avg+curr_prio) / n
674    */
675   double avg_priority;
676
677   /**
678    * The peer's identity.
679    */
680   GNUNET_PEER_Id pid;  
681
682   /**
683    * Number of requests we have currently pending
684    * with this peer (that is, requests that were
685    * transmitted so recently that we would not retransmit
686    * them right now).
687    */
688   unsigned int pending_requests;
689
690   /**
691    * Which offset in "last_p2p_replies" will be updated next?
692    * (we go round-robin).
693    */
694   unsigned int last_p2p_replies_woff;
695
696   /**
697    * Which offset in "last_client_replies" will be updated next?
698    * (we go round-robin).
699    */
700   unsigned int last_client_replies_woff;
701
702 };
703
704
705 /**
706  * Our connection to the datastore.
707  */
708 static struct GNUNET_DATASTORE_Handle *dsh;
709
710 /**
711  * Our scheduler.
712  */
713 static struct GNUNET_SCHEDULER_Handle *sched;
714
715 /**
716  * Our configuration.
717  */
718 const struct GNUNET_CONFIGURATION_Handle *cfg;
719
720 /**
721  * Handle to the core service (NULL until we've connected to it).
722  */
723 struct GNUNET_CORE_Handle *core;
724
725 /**
726  * Head of doubly-linked LGC list.
727  */
728 static struct LocalGetContext *lgc_head;
729
730 /**
731  * Tail of doubly-linked LGC list.
732  */
733 static struct LocalGetContext *lgc_tail;
734
735 /**
736  * Head of request queue for the datastore, sorted by timeout.
737  */
738 static struct DatastoreRequestQueue *drq_head;
739
740 /**
741  * Tail of request queue for the datastore.
742  */
743 static struct DatastoreRequestQueue *drq_tail;
744
745 /**
746  * Linked list of indexed files.
747  */
748 static struct IndexInfo *indexed_files;
749
750 /**
751  * Maps hash over content of indexed files to the respective filename.
752  * The filenames are pointers into the indexed_files linked list and
753  * do not need to be freed.
754  */
755 static struct GNUNET_CONTAINER_MultiHashMap *ifm;
756
757 /**
758  * Map of query hash codes to requests.
759  */
760 static struct GNUNET_CONTAINER_MultiHashMap *requests_by_query;
761
762 /**
763  * Map of peer IDs to requests (for those requests coming
764  * from other peers).
765  */
766 static struct GNUNET_CONTAINER_MultiHashMap *requests_by_peer;
767
768 /**
769  * Linked list of all of our clients and their requests.
770  */
771 static struct ClientList *clients;
772
773 /**
774  * Heap with the request that will expire next at the top.  Contains
775  * pointers of type "struct PendingRequest*"; these will *also* be
776  * aliased from the "requests_by_peer" data structures and the
777  * "requests_by_query" table.  Note that requests from our clients
778  * don't expire and are thus NOT in the "requests_by_expiration"
779  * (or the "requests_by_peer" tables).
780  */
781 static struct GNUNET_CONTAINER_Heap *requests_by_expiration;
782
783 /**
784  * Map of peer identifiers to "struct ConnectedPeer" (for that peer).
785  */
786 static struct GNUNET_CONTAINER_MultiHashMap *connected_peers;
787
788 /**
789  * Maximum number of requests (from other peers) that we're
790  * willing to have pending at any given point in time.
791  * FIXME: set from configuration (and 32 is a tiny value for testing only).
792  */
793 static uint64_t max_pending_requests = 32;
794
795 /**
796  * Write the current index information list to disk.
797  */ 
798 static void
799 write_index_list ()
800 {
801   struct GNUNET_BIO_WriteHandle *wh;
802   char *fn;
803   struct IndexInfo *pos;  
804
805   if (GNUNET_OK !=
806       GNUNET_CONFIGURATION_get_value_filename (cfg,
807                                                "FS",
808                                                "INDEXDB",
809                                                &fn))
810     {
811       GNUNET_log (GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
812                   _("Configuration option `%s' in section `%s' missing.\n"),
813                   "INDEXDB",
814                   "FS");
815       return;
816     }
817   wh = GNUNET_BIO_write_open (fn);
818   if (NULL == wh)
819     {
820       GNUNET_log (GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
821                   _("Could not open `%s'.\n"),
822                   fn);
823       GNUNET_free (fn);
824       return;
825     }
826   pos = indexed_files;
827   while (pos != NULL)
828     {
829       if ( (GNUNET_OK !=
830             GNUNET_BIO_write (wh,
831                               &pos->file_id,
832                               sizeof (GNUNET_HashCode))) ||
833            (GNUNET_OK !=
834             GNUNET_BIO_write_string (wh,
835                                      pos->filename)) )
836         break;
837       pos = pos->next;
838     }
839   if (GNUNET_OK != 
840       GNUNET_BIO_write_close (wh))
841     {
842       GNUNET_log (GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
843                   _("Error writing `%s'.\n"),
844                   fn);
845       GNUNET_free (fn);
846       return;
847     }
848   GNUNET_free (fn);
849 }
850
851
852 /**
853  * Read index information from disk.
854  */
855 static void
856 read_index_list ()
857 {
858   struct GNUNET_BIO_ReadHandle *rh;
859   char *fn;
860   struct IndexInfo *pos;  
861   char *fname;
862   GNUNET_HashCode hc;
863   size_t slen;
864   char *emsg;
865
866   if (GNUNET_OK !=
867       GNUNET_CONFIGURATION_get_value_filename (cfg,
868                                                "FS",
869                                                "INDEXDB",
870                                                &fn))
871     {
872       GNUNET_log (GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
873                   _("Configuration option `%s' in section `%s' missing.\n"),
874                   "INDEXDB",
875                   "FS");
876       return;
877     }
878   if (GNUNET_NO == GNUNET_DISK_file_test (fn))
879     {
880       /* no index info  yet */
881       GNUNET_free (fn);
882       return;
883     }
884   rh = GNUNET_BIO_read_open (fn);
885   if (NULL == rh)
886     {
887       GNUNET_log (GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
888                   _("Could not open `%s'.\n"),
889                   fn);
890       GNUNET_free (fn);
891       return;
892     }
893
894   while ( (GNUNET_OK ==
895            GNUNET_BIO_read (rh,
896                             "Hash of indexed file",
897                             &hc,
898                             sizeof (GNUNET_HashCode))) &&
899           (GNUNET_OK ==
900            GNUNET_BIO_read_string (rh, 
901                                    "Name of indexed file",
902                                    &fname,
903                                    1024 * 16)) )
904     {
905       slen = strlen (fname) + 1;
906       pos = GNUNET_malloc (sizeof (struct IndexInfo) + slen);
907       pos->file_id = hc;
908       pos->filename = (const char *) &pos[1];
909       memcpy (&pos[1], fname, slen);
910       if (GNUNET_SYSERR ==
911           GNUNET_CONTAINER_multihashmap_put (ifm,
912                                              &hc,
913                                              (void*) pos->filename,
914                                              GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY))
915         {
916           GNUNET_free (pos);
917         }
918       else
919         {
920           pos->next = indexed_files;
921           indexed_files = pos;
922         }
923       GNUNET_free (fname);
924     }
925   if (GNUNET_OK != 
926       GNUNET_BIO_read_close (rh, &emsg))
927     GNUNET_free (emsg);
928   GNUNET_free (fn);
929 }
930
931
932 /**
933  * We've validated the hash of the file we're about to
934  * index.  Signal success to the client and update
935  * our internal data structures.
936  *
937  * @param ii the index info entry for the request
938  */
939 static void
940 signal_index_ok (struct IndexInfo *ii)
941 {
942   if (GNUNET_SYSERR ==
943       GNUNET_CONTAINER_multihashmap_put (ifm,
944                                          &ii->file_id,
945                                          (void*) ii->filename,
946                                          GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY))
947     {
948       GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
949                   _("Index request received for file `%s' is indexed as `%s'.  Permitting anyway.\n"),
950                   ii->filename,
951                   (const char*) GNUNET_CONTAINER_multihashmap_get (ifm,
952                                                                    &ii->file_id));
953       GNUNET_SERVER_transmit_context_append (ii->tc,
954                                              NULL, 0,
955                                              GNUNET_MESSAGE_TYPE_FS_INDEX_START_OK);
956       GNUNET_SERVER_transmit_context_run (ii->tc,
957                                           GNUNET_TIME_UNIT_MINUTES);
958       GNUNET_free (ii);
959       return;
960     }
961   ii->next = indexed_files;
962   indexed_files = ii;
963   write_index_list ();
964   GNUNET_SERVER_transmit_context_append (ii->tc,
965                                          NULL, 0,
966                                          GNUNET_MESSAGE_TYPE_FS_INDEX_START_OK);
967   GNUNET_SERVER_transmit_context_run (ii->tc,
968                                       GNUNET_TIME_UNIT_MINUTES);
969   ii->tc = NULL;
970 }
971
972
973 /**
974  * Function called once the hash computation over an
975  * indexed file has completed.
976  *
977  * @param cls closure, our publishing context
978  * @param res resulting hash, NULL on error
979  */
980 static void 
981 hash_for_index_val (void *cls,
982                     const GNUNET_HashCode *
983                     res)
984 {
985   struct IndexInfo *ii = cls;
986   
987   if ( (res == NULL) ||
988        (0 != memcmp (res,
989                      &ii->file_id,
990                      sizeof(GNUNET_HashCode))) )
991     {
992       GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
993                   _("Hash mismatch trying to index file `%s'\n"),
994                   ii->filename);
995       GNUNET_SERVER_transmit_context_append (ii->tc,
996                                              NULL, 0,
997                                              GNUNET_MESSAGE_TYPE_FS_INDEX_START_FAILED);
998       GNUNET_SERVER_transmit_context_run (ii->tc,
999                                           GNUNET_TIME_UNIT_MINUTES);
1000       GNUNET_free (ii);
1001       return;
1002     }
1003   signal_index_ok (ii);
1004 }
1005
1006
1007 /**
1008  * Handle INDEX_START-message.
1009  *
1010  * @param cls closure
1011  * @param client identification of the client
1012  * @param message the actual message
1013  */
1014 static void
1015 handle_index_start (void *cls,
1016                     struct GNUNET_SERVER_Client *client,
1017                     const struct GNUNET_MessageHeader *message)
1018 {
1019   const struct IndexStartMessage *ism;
1020   const char *fn;
1021   uint16_t msize;
1022   struct IndexInfo *ii;
1023   size_t slen;
1024   uint32_t dev;
1025   uint64_t ino;
1026   uint32_t mydev;
1027   uint64_t myino;
1028
1029   msize = ntohs(message->size);
1030   if ( (msize <= sizeof (struct IndexStartMessage)) ||
1031        ( ((const char *)message)[msize-1] != '\0') )
1032     {
1033       GNUNET_break (0);
1034       GNUNET_SERVER_receive_done (client,
1035                                   GNUNET_SYSERR);
1036       return;
1037     }
1038   ism = (const struct IndexStartMessage*) message;
1039   fn = (const char*) &ism[1];
1040   dev = ntohl (ism->device);
1041   ino = GNUNET_ntohll (ism->inode);
1042   ism = (const struct IndexStartMessage*) message;
1043   slen = strlen (fn) + 1;
1044   ii = GNUNET_malloc (sizeof (struct IndexInfo) + slen);
1045   ii->filename = (const char*) &ii[1];
1046   memcpy (&ii[1], fn, slen);
1047   ii->file_id = ism->file_id;  
1048   ii->tc = GNUNET_SERVER_transmit_context_create (client);
1049   if ( ( (dev != 0) ||
1050          (ino != 0) ) &&
1051        (GNUNET_OK == GNUNET_DISK_file_get_identifiers (fn,
1052                                                        &mydev,
1053                                                        &myino)) &&
1054        ( (dev == mydev) &&
1055          (ino == myino) ) )
1056     {      
1057       /* fast validation OK! */
1058       signal_index_ok (ii);
1059       return;
1060     }
1061   /* slow validation, need to hash full file (again) */
1062   GNUNET_CRYPTO_hash_file (sched,
1063                            GNUNET_SCHEDULER_PRIORITY_IDLE,
1064                            GNUNET_NO,
1065                            fn,
1066                            HASHING_BLOCKSIZE,
1067                            &hash_for_index_val,
1068                            ii);
1069 }
1070
1071
1072 /**
1073  * Handle INDEX_LIST_GET-message.
1074  *
1075  * @param cls closure
1076  * @param client identification of the client
1077  * @param message the actual message
1078  */
1079 static void
1080 handle_index_list_get (void *cls,
1081                        struct GNUNET_SERVER_Client *client,
1082                        const struct GNUNET_MessageHeader *message)
1083 {
1084   struct GNUNET_SERVER_TransmitContext *tc;
1085   struct IndexInfoMessage *iim;
1086   char buf[GNUNET_SERVER_MAX_MESSAGE_SIZE];
1087   size_t slen;
1088   const char *fn;
1089   struct GNUNET_MessageHeader *msg;
1090   struct IndexInfo *pos;
1091
1092   tc = GNUNET_SERVER_transmit_context_create (client);
1093   iim = (struct IndexInfoMessage*) buf;
1094   msg = &iim->header;
1095   pos = indexed_files;
1096   while (NULL != pos)
1097     {
1098       iim->reserved = 0;
1099       iim->file_id = pos->file_id;
1100       fn = pos->filename;
1101       slen = strlen (fn) + 1;
1102       if (slen + sizeof (struct IndexInfoMessage) > 
1103           GNUNET_SERVER_MAX_MESSAGE_SIZE)
1104         {
1105           GNUNET_break (0);
1106           break;
1107         }
1108       memcpy (&iim[1], fn, slen);
1109       GNUNET_SERVER_transmit_context_append
1110         (tc,
1111          &msg[1],
1112          sizeof (struct IndexInfoMessage) 
1113          - sizeof (struct GNUNET_MessageHeader) + slen,
1114          GNUNET_MESSAGE_TYPE_FS_INDEX_LIST_ENTRY);
1115       pos = pos->next;
1116     }
1117   GNUNET_SERVER_transmit_context_append (tc,
1118                                          NULL, 0,
1119                                          GNUNET_MESSAGE_TYPE_FS_INDEX_LIST_END);
1120   GNUNET_SERVER_transmit_context_run (tc,
1121                                       GNUNET_TIME_UNIT_MINUTES);
1122 }
1123
1124
1125 /**
1126  * Handle UNINDEX-message.
1127  *
1128  * @param cls closure
1129  * @param client identification of the client
1130  * @param message the actual message
1131  */
1132 static void
1133 handle_unindex (void *cls,
1134                 struct GNUNET_SERVER_Client *client,
1135                 const struct GNUNET_MessageHeader *message)
1136 {
1137   const struct UnindexMessage *um;
1138   struct IndexInfo *pos;
1139   struct IndexInfo *prev;
1140   struct IndexInfo *next;
1141   struct GNUNET_SERVER_TransmitContext *tc;
1142   int found;
1143   
1144   um = (const struct UnindexMessage*) message;
1145   found = GNUNET_NO;
1146   prev = NULL;
1147   pos = indexed_files;
1148   while (NULL != pos)
1149     {
1150       next = pos->next;
1151       if (0 == memcmp (&pos->file_id,
1152                        &um->file_id,
1153                        sizeof (GNUNET_HashCode)))
1154         {
1155           if (prev == NULL)
1156             indexed_files = pos->next;
1157           else
1158             prev->next = pos->next;
1159           GNUNET_free (pos);
1160           found = GNUNET_YES;
1161         }
1162       else
1163         {
1164           prev = pos;
1165         }
1166       pos = next;
1167     }
1168   if (GNUNET_YES == found)
1169     write_index_list ();
1170   tc = GNUNET_SERVER_transmit_context_create (client);
1171   GNUNET_SERVER_transmit_context_append (tc,
1172                                          NULL, 0,
1173                                          GNUNET_MESSAGE_TYPE_FS_UNINDEX_OK);
1174   GNUNET_SERVER_transmit_context_run (tc,
1175                                       GNUNET_TIME_UNIT_MINUTES);
1176 }
1177
1178
1179 /**
1180  * Run the next DS request in our
1181  * queue, we're done with the current one.
1182  */
1183 static void
1184 next_ds_request ()
1185 {
1186   struct DatastoreRequestQueue *e;
1187   
1188   while (NULL != (e = drq_head))
1189     {
1190       if (0 != GNUNET_TIME_absolute_get_remaining (e->timeout).value)
1191         break;
1192       if (e->task != GNUNET_SCHEDULER_NO_TASK)
1193         GNUNET_SCHEDULER_cancel (sched, e->task);
1194       GNUNET_CONTAINER_DLL_remove (drq_head, drq_tail, e);
1195       e->req (e->req_cls, GNUNET_NO);
1196       GNUNET_free (e);  
1197     }
1198   if (e == NULL)
1199     return;
1200   if (e->task != GNUNET_SCHEDULER_NO_TASK)
1201     GNUNET_SCHEDULER_cancel (sched, e->task);
1202   e->task = GNUNET_SCHEDULER_NO_TASK;
1203   e->req (e->req_cls, GNUNET_YES);
1204   GNUNET_CONTAINER_DLL_remove (drq_head, drq_tail, e);
1205   GNUNET_free (e);  
1206 }
1207
1208
1209 /**
1210  * A datastore request had to be timed out. 
1211  *
1212  * @param cls closure (of type "struct DatastoreRequestQueue*")
1213  * @param tc task context, unused
1214  */
1215 static void
1216 timeout_ds_request (void *cls,
1217                     const struct GNUNET_SCHEDULER_TaskContext *tc)
1218 {
1219   struct DatastoreRequestQueue *e = cls;
1220
1221   e->task = GNUNET_SCHEDULER_NO_TASK;
1222   GNUNET_CONTAINER_DLL_remove (drq_head, drq_tail, e);
1223   e->req (e->req_cls, GNUNET_NO);
1224   GNUNET_free (e);  
1225 }
1226
1227
1228 /**
1229  * Queue a request for the datastore.
1230  *
1231  * @param deadline by when the request should run
1232  * @param fun function to call once the request can be run
1233  * @param fun_cls closure for fun
1234  */
1235 static struct DatastoreRequestQueue *
1236 queue_ds_request (struct GNUNET_TIME_Relative deadline,
1237                   RequestFunction fun,
1238                   void *fun_cls)
1239 {
1240   struct DatastoreRequestQueue *e;
1241   struct DatastoreRequestQueue *bef;
1242
1243   if (drq_head == NULL)
1244     {
1245       /* no other requests pending, run immediately */
1246       fun (fun_cls, GNUNET_OK);
1247       return NULL;
1248     }
1249   e = GNUNET_malloc (sizeof (struct DatastoreRequestQueue));
1250   e->timeout = GNUNET_TIME_relative_to_absolute (deadline);
1251   e->req = fun;
1252   e->req_cls = fun_cls;
1253   if (deadline.value == GNUNET_TIME_UNIT_FOREVER_REL.value)
1254     {
1255       /* local request, highest prio, put at head of queue
1256          regardless of deadline */
1257       bef = NULL;
1258     }
1259   else
1260     {
1261       bef = drq_tail;
1262       while ( (NULL != bef) &&
1263               (e->timeout.value < bef->timeout.value) )
1264         bef = bef->prev;
1265     }
1266   GNUNET_CONTAINER_DLL_insert_after (drq_head, drq_tail, bef, e);
1267   if (deadline.value == GNUNET_TIME_UNIT_FOREVER_REL.value)
1268     return e;
1269   e->task = GNUNET_SCHEDULER_add_delayed (sched,
1270                                           GNUNET_NO,
1271                                           GNUNET_SCHEDULER_PRIORITY_BACKGROUND,
1272                                           GNUNET_SCHEDULER_NO_TASK,
1273                                           deadline,
1274                                           &timeout_ds_request,
1275                                           e);
1276   return e;                                    
1277 }
1278
1279
1280 /**
1281  * Free the state associated with a local get context.
1282  *
1283  * @param lgc the lgc to free
1284  */
1285 static void
1286 local_get_context_free (struct LocalGetContext *lgc) 
1287 {
1288   GNUNET_CONTAINER_DLL_remove (lgc_head, lgc_tail, lgc);
1289   GNUNET_SERVER_client_drop (lgc->client); 
1290   GNUNET_free_non_null (lgc->results);
1291   if (lgc->results_bf != NULL)
1292     GNUNET_CONTAINER_bloomfilter_free (lgc->results_bf);
1293   if (lgc->req != NULL)
1294     {
1295       if (lgc->req->task != GNUNET_SCHEDULER_NO_TASK)
1296         GNUNET_SCHEDULER_cancel (sched, lgc->req->task);
1297       GNUNET_CONTAINER_DLL_remove (lgc_head, lgc_tail, lgc);
1298       GNUNET_free (lgc->req);
1299     }
1300   GNUNET_free (lgc);
1301 }
1302
1303
1304 /**
1305  * We're able to transmit the next (local) result to the client.
1306  * Do it and ask the datastore for more.  Or, on error, tell
1307  * the datastore to stop giving us more.
1308  *
1309  * @param cls our closure (struct LocalGetContext)
1310  * @param max maximum number of bytes we can transmit
1311  * @param buf where to copy our message
1312  * @return number of bytes copied to buf
1313  */
1314 static size_t
1315 transmit_local_result (void *cls,
1316                        size_t max,
1317                        void *buf)
1318 {
1319   struct LocalGetContext *lgc = cls;  
1320   uint16_t msize;
1321
1322   if (NULL == buf)
1323     {
1324 #if DEBUG_FS
1325       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1326                   "Failed to transmit result to local client, aborting datastore iteration.\n");
1327 #endif
1328       /* error, abort! */
1329       GNUNET_free (lgc->result);
1330       lgc->result = NULL;
1331       GNUNET_DATASTORE_get_next (dsh, GNUNET_NO);
1332       return 0;
1333     }
1334   msize = ntohs (lgc->result->header.size);
1335 #if DEBUG_FS
1336   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1337               "Transmitting %u bytes of result to local client.\n",
1338               msize);
1339 #endif
1340   GNUNET_assert (max >= msize);
1341   memcpy (buf, lgc->result, msize);
1342   GNUNET_free (lgc->result);
1343   lgc->result = NULL;
1344   GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
1345   return msize;
1346 }
1347
1348
1349 /**
1350  * Continuation called from datastore's remove
1351  * function.
1352  *
1353  * @param cls unused
1354  * @param success did the deletion work?
1355  * @param msg error message
1356  */
1357 static void
1358 remove_cont (void *cls,
1359              int success,
1360              const char *msg)
1361 {
1362   if (GNUNET_OK != success)
1363     GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
1364                 _("Failed to delete bogus block: %s\n"),
1365                 msg);
1366   GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
1367 }
1368
1369
1370 /**
1371  * Mingle hash with the mingle_number to
1372  * produce different bits.
1373  */
1374 static void
1375 mingle_hash (const GNUNET_HashCode * in,
1376              int32_t mingle_number, 
1377              GNUNET_HashCode * hc)
1378 {
1379   GNUNET_HashCode m;
1380
1381   GNUNET_CRYPTO_hash (&mingle_number, 
1382                       sizeof (int32_t), 
1383                       &m);
1384   GNUNET_CRYPTO_hash_xor (&m, in, hc);
1385 }
1386
1387
1388 /**
1389  * We've received an on-demand encoded block
1390  * from the datastore.  Attempt to do on-demand
1391  * encoding and (if successful), call the 
1392  * continuation with the resulting block.  On
1393  * error, clean up and ask the datastore for
1394  * more results.
1395  *
1396  * @param key key for the content
1397  * @param size number of bytes in data
1398  * @param data content stored
1399  * @param type type of the content
1400  * @param priority priority of the content
1401  * @param anonymity anonymity-level for the content
1402  * @param expiration expiration time for the content
1403  * @param uid unique identifier for the datum;
1404  *        maybe 0 if no unique identifier is available
1405  * @param cont function to call with the actual block
1406  * @param cont_cls closure for cont
1407  */
1408 static void
1409 handle_on_demand_block (const GNUNET_HashCode * key,
1410                         uint32_t size,
1411                         const void *data,
1412                         uint32_t type,
1413                         uint32_t priority,
1414                         uint32_t anonymity,
1415                         struct GNUNET_TIME_Absolute
1416                         expiration, uint64_t uid,
1417                         GNUNET_DATASTORE_Iterator cont,
1418                         void *cont_cls)
1419 {
1420   const struct OnDemandBlock *odb;
1421   GNUNET_HashCode nkey;
1422   struct GNUNET_CRYPTO_AesSessionKey skey;
1423   struct GNUNET_CRYPTO_AesInitializationVector iv;
1424   GNUNET_HashCode query;
1425   ssize_t nsize;
1426   char ndata[DBLOCK_SIZE];
1427   char edata[DBLOCK_SIZE];
1428   const char *fn;
1429   struct GNUNET_DISK_FileHandle *fh;
1430   uint64_t off;
1431
1432   if (size != sizeof (struct OnDemandBlock))
1433     {
1434       GNUNET_break (0);
1435       GNUNET_DATASTORE_remove (dsh, 
1436                                key,
1437                                size,
1438                                data,
1439                                &remove_cont,
1440                                NULL,
1441                                GNUNET_TIME_UNIT_FOREVER_REL);     
1442       return;
1443     }
1444   odb = (const struct OnDemandBlock*) data;
1445   off = GNUNET_ntohll (odb->offset);
1446   fn = (const char*) GNUNET_CONTAINER_multihashmap_get (ifm,
1447                                                         &odb->file_id);
1448   fh = NULL;
1449   if ( (NULL == fn) ||
1450        (NULL == (fh = GNUNET_DISK_file_open (fn, 
1451                                              GNUNET_DISK_OPEN_READ,
1452                                              GNUNET_DISK_PERM_NONE))) ||
1453        (off !=
1454         GNUNET_DISK_file_seek (fh,
1455                                off,
1456                                GNUNET_DISK_SEEK_SET)) ||
1457        (-1 ==
1458         (nsize = GNUNET_DISK_file_read (fh,
1459                                         ndata,
1460                                         sizeof (ndata)))) )
1461     {
1462       GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
1463                   _("Could not access indexed file `%s' at offset %llu: %s\n"),
1464                   GNUNET_h2s (&odb->file_id),
1465                   (unsigned long long) off,
1466                   STRERROR (errno));
1467       if (fh != NULL)
1468         GNUNET_DISK_file_close (fh);
1469       /* FIXME: if this happens often, we need
1470          to remove the OnDemand block from the DS! */
1471       GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);        
1472       return;
1473     }
1474   GNUNET_DISK_file_close (fh);
1475   GNUNET_CRYPTO_hash (ndata,
1476                       nsize,
1477                       &nkey);
1478   GNUNET_CRYPTO_hash_to_aes_key (&nkey, &skey, &iv);
1479   GNUNET_CRYPTO_aes_encrypt (ndata,
1480                              nsize,
1481                              &skey,
1482                              &iv,
1483                              edata);
1484   GNUNET_CRYPTO_hash (edata,
1485                       nsize,
1486                       &query);
1487   if (0 != memcmp (&query, 
1488                    key,
1489                    sizeof (GNUNET_HashCode)))
1490     {
1491       GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
1492                   _("Indexed file `%s' changed at offset %llu\n"),
1493                   fn,
1494                   (unsigned long long) off);
1495       /* FIXME: if this happens often, we need
1496          to remove the OnDemand block from the DS! */
1497       GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
1498       return;
1499     }
1500   cont (cont_cls,
1501         key,
1502         nsize,
1503         edata,
1504         GNUNET_DATASTORE_BLOCKTYPE_DBLOCK,
1505         priority,
1506         anonymity,
1507         expiration,
1508         uid);
1509 }
1510
1511
1512 /**
1513  * How many bytes should a bloomfilter be if we have already seen
1514  * entry_count responses?  Note that BLOOMFILTER_K gives us the number
1515  * of bits set per entry.  Furthermore, we should not re-size the
1516  * filter too often (to keep it cheap).
1517  *
1518  * Since other peers will also add entries but not resize the filter,
1519  * we should generally pick a slightly larger size than what the
1520  * strict math would suggest.
1521  *
1522  * @return must be a power of two and smaller or equal to 2^15.
1523  */
1524 static size_t
1525 compute_bloomfilter_size (unsigned int entry_count)
1526 {
1527   size_t size;
1528   unsigned int ideal = (entry_count * BLOOMFILTER_K) / 4;
1529   uint16_t max = 1 << 15;
1530
1531   if (entry_count > max)
1532     return max;
1533   size = 8;
1534   while ((size < max) && (size < ideal))
1535     size *= 2;
1536   if (size > max)
1537     return max;
1538   return size;
1539 }
1540
1541
1542 /**
1543  * Recalculate our bloom filter for filtering replies.
1544  *
1545  * @param count number of entries we are filtering right now
1546  * @param mingle set to our new mingling value
1547  * @param bf_size set to the size of the bloomfilter
1548  * @param entries the entries to filter
1549  * @return updated bloomfilter, NULL for none
1550  */
1551 static struct GNUNET_CONTAINER_BloomFilter *
1552 refresh_bloomfilter (unsigned int count,
1553                      int32_t * mingle,
1554                      size_t *bf_size,
1555                      const GNUNET_HashCode *entries)
1556 {
1557   struct GNUNET_CONTAINER_BloomFilter *bf;
1558   size_t nsize;
1559   unsigned int i;
1560   GNUNET_HashCode mhash;
1561
1562   if (0 == count)
1563     return NULL;
1564   nsize = compute_bloomfilter_size (count);
1565   *mingle = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, -1);
1566   *bf_size = nsize;
1567   bf = GNUNET_CONTAINER_bloomfilter_init (NULL, 
1568                                           nsize,
1569                                           BLOOMFILTER_K);
1570   for (i=0;i<count;i++)
1571     {
1572       mingle_hash (&entries[i], *mingle, &mhash);
1573       GNUNET_CONTAINER_bloomfilter_add (bf, &mhash);
1574     }
1575   return bf;
1576 }
1577
1578
1579 /**
1580  * Closure used for "target_peer_select_cb".
1581  */
1582 struct PeerSelectionContext 
1583 {
1584   /**
1585    * The request for which we are selecting
1586    * peers.
1587    */
1588   struct PendingRequest *pr;
1589
1590   /**
1591    * Current "prime" target.
1592    */
1593   struct GNUNET_PeerIdentity target;
1594
1595   /**
1596    * How much do we like this target?
1597    */
1598   double target_score;
1599
1600 };
1601
1602
1603 /**
1604  * Function called for each connected peer to determine
1605  * which one(s) would make good targets for forwarding.
1606  *
1607  * @param cls closure (struct PeerSelectionContext)
1608  * @param key current key code (peer identity)
1609  * @param value value in the hash map (struct ConnectedPeer)
1610  * @return GNUNET_YES if we should continue to
1611  *         iterate,
1612  *         GNUNET_NO if not.
1613  */
1614 static int
1615 target_peer_select_cb (void *cls,
1616                        const GNUNET_HashCode * key,
1617                        void *value)
1618 {
1619   struct PeerSelectionContext *psc = cls;
1620   struct ConnectedPeer *cp = value;
1621   struct PendingRequest *pr = psc->pr;
1622   double score;
1623   unsigned int i;
1624
1625   /* 1) check if we have already (recently) forwarded to this peer */
1626   for (i=0;i<pr->used_pids_off;i++)
1627     if (pr->used_pids[i] == cp->pid)
1628       return GNUNET_YES; /* skip */
1629   // 2) calculate how much we'd like to forward to this peer
1630   score = 0; // FIXME!
1631   
1632   /* store best-fit in closure */
1633   if (score > psc->target_score)
1634     {
1635       psc->target_score = score;
1636       psc->target.hashPubKey = *key; 
1637     }
1638   return GNUNET_YES;
1639 }
1640
1641
1642 /**
1643  * We use a random delay to make the timing of requests
1644  * less predictable.  This function returns such a random
1645  * delay.
1646  *
1647  * @return random delay to use for some request, between 0 and TTL_DECREMENT ms
1648  */
1649 static struct GNUNET_TIME_Relative
1650 get_processing_delay ()
1651 {
1652   return GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MILLISECONDS,
1653                                         GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
1654                                                                   TTL_DECREMENT));
1655 }
1656
1657
1658 /**
1659  * Task that is run for each request with the
1660  * goal of forwarding the associated query to
1661  * other peers.  The task should re-schedule
1662  * itself to be re-run once the TTL has expired.
1663  * (or at a later time if more peers should
1664  * be queried earlier).
1665  *
1666  * @param cls the requests "struct PendingRequest*"
1667  * @param tc task context (unused)
1668  */
1669 static void
1670 forward_request_task (void *cls,
1671                       const struct GNUNET_SCHEDULER_TaskContext *tc);
1672
1673
1674 /**
1675  * We've selected a peer for forwarding of a query.
1676  * Construct the message and then re-schedule the
1677  * task to forward again to (other) peers.
1678  *
1679  * @param cls closure
1680  * @param size number of bytes available in buf
1681  * @param buf where the callee should write the message
1682  * @return number of bytes written to buf
1683  */
1684 static size_t
1685 transmit_request_cb (void *cls,
1686                      size_t size, 
1687                      void *buf)
1688 {
1689   struct PendingRequest *pr = cls;
1690   struct GetMessage *gm;
1691   GNUNET_HashCode *ext;
1692   char *bfdata;
1693   size_t msize;
1694   unsigned int k;
1695
1696   pr->cth = NULL;
1697   /* (1) check for timeout */
1698   if (NULL == buf)
1699     {
1700       /* timeout, try another peer immediately again */
1701       pr->task = GNUNET_SCHEDULER_add_delayed (sched,
1702                                                GNUNET_NO,
1703                                                GNUNET_SCHEDULER_PRIORITY_IDLE,
1704                                                GNUNET_SCHEDULER_NO_TASK,
1705                                                GNUNET_TIME_UNIT_ZERO,
1706                                                &forward_request_task,
1707                                                pr);
1708       return 0;
1709     }
1710   /* (2) build query message */
1711   k = 0; // FIXME: count hash codes!
1712   msize = sizeof (struct GetMessage) + pr->bf_size + k * sizeof(GNUNET_HashCode);
1713   GNUNET_assert (msize < GNUNET_SERVER_MAX_MESSAGE_SIZE);
1714   gm = (struct GetMessage*) buf;
1715   gm->header.type = htons (GNUNET_MESSAGE_TYPE_FS_GET);
1716   gm->header.size = htons (msize);
1717   gm->type = htonl (pr->type);
1718   pr->remaining_priority /= 2;
1719   gm->priority = htonl (pr->remaining_priority);
1720   gm->ttl = htonl (pr->ttl);
1721   gm->filter_mutator = htonl(pr->mingle);
1722   gm->hash_bitmap = htonl (42);
1723   gm->query = pr->query;
1724   ext = (GNUNET_HashCode*) &gm[1];
1725   // FIXME: setup "ext[0]..[k-1]"
1726   bfdata = (char *) &ext[k];
1727   if (pr->bf != NULL)
1728     GNUNET_CONTAINER_bloomfilter_get_raw_data (pr->bf,
1729                                                bfdata,
1730                                                pr->bf_size);
1731   
1732   /* (3) schedule job to do it again (or another peer, etc.) */
1733   pr->task = GNUNET_SCHEDULER_add_delayed (sched,
1734                                            GNUNET_NO,
1735                                            GNUNET_SCHEDULER_PRIORITY_IDLE,
1736                                            GNUNET_SCHEDULER_NO_TASK,
1737                                            get_processing_delay (), // FIXME!
1738                                            &forward_request_task,
1739                                            pr);
1740
1741   return msize;
1742 }
1743
1744
1745 /**
1746  * Function called after we've tried to reserve
1747  * a certain amount of bandwidth for a reply.
1748  * Check if we succeeded and if so send our query.
1749  *
1750  * @param cls the requests "struct PendingRequest*"
1751  * @param peer identifies the peer
1752  * @param latency current latency estimate, "FOREVER" if we have been
1753  *                disconnected
1754  * @param bpm_in set to the current bandwidth limit (receiving) for this peer
1755  * @param bpm_out set to the current bandwidth limit (sending) for this peer
1756  * @param amount set to the amount that was actually reserved or unreserved
1757  * @param preference current traffic preference for the given peer
1758  */
1759 static void
1760 target_reservation_cb (void *cls,
1761                        const struct
1762                        GNUNET_PeerIdentity * peer,
1763                        unsigned int bpm_in,
1764                        unsigned int bpm_out,
1765                        struct GNUNET_TIME_Relative
1766                        latency, int amount,
1767                        unsigned long long preference)
1768 {
1769   struct PendingRequest *pr = cls;
1770   uint32_t priority;
1771   uint16_t size;
1772   struct GNUNET_TIME_Relative maxdelay;
1773
1774   GNUNET_assert (peer != NULL);
1775   if ( (amount != DBLOCK_SIZE) ||
1776        (pr->cth != NULL) )
1777     {
1778       /* try again later; FIXME: we may need to un-reserve "amount"? */
1779       pr->task = GNUNET_SCHEDULER_add_delayed (sched,
1780                                                GNUNET_NO,
1781                                                GNUNET_SCHEDULER_PRIORITY_IDLE,
1782                                                GNUNET_SCHEDULER_NO_TASK,
1783                                                get_processing_delay (), // FIXME: longer?
1784                                                &forward_request_task,
1785                                                pr);
1786       return;
1787     }
1788   // (2) transmit, update ttl/priority
1789   // FIXME: calculate priority, maxdelay, size properly!
1790   priority = 0;
1791   size = 60000;
1792   maxdelay = GNUNET_CONSTANTS_SERVICE_TIMEOUT;
1793   pr->cth = GNUNET_CORE_notify_transmit_ready (core,
1794                                                priority,
1795                                                maxdelay,
1796                                                peer,
1797                                                size,
1798                                                &transmit_request_cb,
1799                                                pr);
1800   if (pr->cth == NULL)
1801     {
1802       /* try again later */
1803       pr->task = GNUNET_SCHEDULER_add_delayed (sched,
1804                                                GNUNET_NO,
1805                                                GNUNET_SCHEDULER_PRIORITY_IDLE,
1806                                                GNUNET_SCHEDULER_NO_TASK,
1807                                                get_processing_delay (), // FIXME: longer?
1808                                                &forward_request_task,
1809                                                pr);
1810     }
1811 }
1812
1813
1814 /**
1815  * Task that is run for each request with the
1816  * goal of forwarding the associated query to
1817  * other peers.  The task should re-schedule
1818  * itself to be re-run once the TTL has expired.
1819  * (or at a later time if more peers should
1820  * be queried earlier).
1821  *
1822  * @param cls the requests "struct PendingRequest*"
1823  * @param tc task context (unused)
1824  */
1825 static void
1826 forward_request_task (void *cls,
1827                       const struct GNUNET_SCHEDULER_TaskContext *tc)
1828 {
1829   struct PendingRequest *pr = cls;
1830   struct PeerSelectionContext psc;
1831
1832   pr->task = GNUNET_SCHEDULER_NO_TASK;
1833   if (pr->cth != NULL) 
1834     {
1835       /* we're busy transmitting a result, wait a bit */
1836       pr->task = GNUNET_SCHEDULER_add_delayed (sched,
1837                                                GNUNET_NO,
1838                                                GNUNET_SCHEDULER_PRIORITY_IDLE,
1839                                                GNUNET_SCHEDULER_NO_TASK,
1840                                                get_processing_delay (), 
1841                                                &forward_request_task,
1842                                                pr);
1843       return;
1844     }
1845   /* (1) select target */
1846   psc.pr = pr;
1847   psc.target_score = DBL_MIN;
1848   GNUNET_CONTAINER_multihashmap_iterate (connected_peers,
1849                                          &target_peer_select_cb,
1850                                          &psc);
1851   if (psc.target_score == DBL_MIN)
1852     {
1853       /* no possible target found, wait some time */
1854       pr->task = GNUNET_SCHEDULER_add_delayed (sched,
1855                                                GNUNET_NO,
1856                                                GNUNET_SCHEDULER_PRIORITY_IDLE,
1857                                                GNUNET_SCHEDULER_NO_TASK,
1858                                                get_processing_delay (), // FIXME: exponential back-off? or at least wait longer...
1859                                                &forward_request_task,
1860                                                pr);
1861       return;
1862     }
1863   /* (2) reserve reply bandwidth */
1864   // FIXME: need a way to cancel; this
1865   // async operation is problematic (segv-problematic)
1866   // if "pr" is destroyed while it happens!
1867   GNUNET_CORE_peer_configure (core,
1868                               &psc.target,
1869                               GNUNET_CONSTANTS_SERVICE_TIMEOUT, 
1870                               -1,
1871                               DBLOCK_SIZE, // FIXME: make dependent on type?
1872                               0,
1873                               &target_reservation_cb,
1874                               pr);
1875 }
1876
1877
1878 /**
1879  * We're processing (local) results for a search request
1880  * from a (local) client.  Pass applicable results to the
1881  * client and if we are done either clean up (operation
1882  * complete) or switch to P2P search (more results possible).
1883  *
1884  * @param cls our closure (struct LocalGetContext)
1885  * @param key key for the content
1886  * @param size number of bytes in data
1887  * @param data content stored
1888  * @param type type of the content
1889  * @param priority priority of the content
1890  * @param anonymity anonymity-level for the content
1891  * @param expiration expiration time for the content
1892  * @param uid unique identifier for the datum;
1893  *        maybe 0 if no unique identifier is available
1894  */
1895 static void
1896 process_local_get_result (void *cls,
1897                           const GNUNET_HashCode * key,
1898                           uint32_t size,
1899                           const void *data,
1900                           uint32_t type,
1901                           uint32_t priority,
1902                           uint32_t anonymity,
1903                           struct GNUNET_TIME_Absolute
1904                           expiration, 
1905                           uint64_t uid)
1906 {
1907   struct LocalGetContext *lgc = cls;
1908   struct PendingRequest *pr;
1909   struct ClientRequestList *crl;
1910   struct ClientList *cl;
1911   size_t msize;
1912   unsigned int i;
1913
1914   if (key == NULL)
1915     {
1916 #if DEBUG_FS
1917       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1918                   "Received last result for `%s' from local datastore, deciding what to do next.\n",
1919                   GNUNET_h2s (&lgc->query));
1920 #endif
1921       /* no further results from datastore; continue
1922          processing further requests from the client and
1923          allow the next task to use the datastore; also,
1924          switch to P2P requests or clean up our state. */
1925       next_ds_request ();
1926       GNUNET_SERVER_receive_done (lgc->client,
1927                                   GNUNET_OK);
1928       if ( (lgc->results_used == 0) ||
1929            (lgc->type == GNUNET_DATASTORE_BLOCKTYPE_KBLOCK) ||
1930            (lgc->type == GNUNET_DATASTORE_BLOCKTYPE_SBLOCK) ||
1931            (lgc->type == GNUNET_DATASTORE_BLOCKTYPE_SKBLOCK) )
1932         {
1933 #if DEBUG_FS
1934           GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1935                       "Forwarding query for `%s' to network.\n",
1936                       GNUNET_h2s (&lgc->query));
1937 #endif
1938           cl = clients;
1939           while ( (NULL != cl) &&
1940                   (cl->client != lgc->client) )
1941             cl = cl->next;
1942           if (cl == NULL)
1943             {
1944               cl = GNUNET_malloc (sizeof (struct ClientList));
1945               cl->client = lgc->client;
1946               cl->next = clients;
1947               clients = cl;
1948             }
1949           crl = GNUNET_malloc (sizeof (struct ClientRequestList));
1950           crl->cl = cl;
1951           GNUNET_CONTAINER_DLL_insert (cl->head, cl->tail, crl);
1952           pr = GNUNET_malloc (sizeof (struct PendingRequest));
1953           pr->client = lgc->client;
1954           GNUNET_SERVER_client_keep (pr->client);
1955           pr->crl_entry = crl;
1956           crl->req = pr;
1957           if (lgc->type == GNUNET_DATASTORE_BLOCKTYPE_SBLOCK)
1958             {
1959               pr->namespace = GNUNET_malloc (sizeof (GNUNET_HashCode));
1960               *pr->namespace = lgc->namespace;
1961             }
1962           pr->replies_seen = lgc->results;
1963           lgc->results = NULL;
1964           pr->start_time = GNUNET_TIME_absolute_get ();
1965           pr->query = lgc->query;
1966           pr->target_pid = GNUNET_PEER_intern (&lgc->target);
1967           pr->replies_seen_off = lgc->results_used;
1968           pr->replies_seen_size = lgc->results_size;
1969           lgc->results_size = 0;
1970           pr->type = lgc->type;
1971           pr->anonymity_level = lgc->anonymity_level;
1972           pr->bf = refresh_bloomfilter (pr->replies_seen_off,
1973                                         &pr->mingle,
1974                                         &pr->bf_size,
1975                                         pr->replies_seen);
1976           GNUNET_CONTAINER_multihashmap_put (requests_by_query,
1977                                              &pr->query,
1978                                              pr,
1979                                              GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
1980           pr->task = GNUNET_SCHEDULER_add_delayed (sched,
1981                                                    GNUNET_NO,
1982                                                    GNUNET_SCHEDULER_PRIORITY_IDLE,
1983                                                    GNUNET_SCHEDULER_NO_TASK,
1984                                                    get_processing_delay (),
1985                                                    &forward_request_task,
1986                                                    pr);
1987           local_get_context_free (lgc);
1988           return;
1989         }
1990       /* got all possible results, clean up! */
1991 #if DEBUG_FS
1992       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1993                   "Found all possible results for query for `%s', done!\n",
1994                   GNUNET_h2s (&lgc->query));
1995 #endif
1996       local_get_context_free (lgc);
1997       return;
1998     }
1999   if (type == GNUNET_DATASTORE_BLOCKTYPE_ONDEMAND)
2000     {
2001 #if DEBUG_FS
2002       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2003                   "Received on-demand block for `%s' from local datastore, fetching data.\n",
2004                   GNUNET_h2s (&lgc->query));
2005 #endif
2006       handle_on_demand_block (key, size, data, type, priority, 
2007                               anonymity, expiration, uid,
2008                               &process_local_get_result,
2009                               lgc);
2010       return;
2011     }
2012   if ( (type != lgc->type) &&
2013        (lgc->type != GNUNET_DATASTORE_BLOCKTYPE_ANY) )
2014     {
2015       /* this should be virtually impossible to reach (DBLOCK 
2016          query hash being identical to KBLOCK/SBLOCK query hash);
2017          nevertheless, if it happens, the correct thing is to
2018          simply skip the result. */
2019 #if DEBUG_FS
2020       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2021                   "Received block of unexpected type (%u, want %u) for `%s' from local datastore, ignoring.\n",
2022                   type,
2023                   lgc->type,
2024                   GNUNET_h2s (&lgc->query));
2025 #endif
2026       GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);        
2027       return;
2028     }
2029   /* check if this is a result we've alredy
2030      received */
2031   for (i=0;i<lgc->results_used;i++)
2032     if (0 == memcmp (key,
2033                      &lgc->results[i],
2034                      sizeof (GNUNET_HashCode)))
2035       {
2036 #if DEBUG_FS
2037         GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2038                     "Received duplicate result for `%s' from local datastore, ignoring.\n",
2039                     GNUNET_h2s (&lgc->query));
2040 #endif
2041         GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
2042         return; 
2043       }
2044   if (lgc->results_used == lgc->results_size)
2045     GNUNET_array_grow (lgc->results,
2046                        lgc->results_size,
2047                        lgc->results_size * 2 + 2);
2048   GNUNET_CRYPTO_hash (data, 
2049                       size, 
2050                       &lgc->results[lgc->results_used++]);    
2051   msize = size + sizeof (struct ContentMessage);
2052   GNUNET_assert (msize < GNUNET_SERVER_MAX_MESSAGE_SIZE);
2053   lgc->result = GNUNET_malloc (msize);
2054   lgc->result->header.size = htons (msize);
2055   lgc->result->header.type = htons (GNUNET_MESSAGE_TYPE_FS_CONTENT);
2056   lgc->result->type = htonl (type);
2057   lgc->result->expiration = GNUNET_TIME_absolute_hton (expiration);
2058   memcpy (&lgc->result[1],
2059           data,
2060           size);
2061 #if DEBUG_FS
2062   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2063               "Received new result for `%s' from local datastore, passing to client.\n",
2064               GNUNET_h2s (&lgc->query));
2065 #endif
2066   GNUNET_SERVER_notify_transmit_ready (lgc->client,
2067                                        msize,
2068                                        GNUNET_TIME_UNIT_FOREVER_REL,
2069                                        &transmit_local_result,
2070                                        lgc);
2071 }
2072
2073
2074 /**
2075  * We're processing a search request from a local
2076  * client.  Now it is our turn to query the datastore.
2077  * 
2078  * @param cls our closure (struct LocalGetContext)
2079  * @param tc unused
2080  */
2081 static void
2082 transmit_local_get (void *cls,
2083                     const struct GNUNET_SCHEDULER_TaskContext *tc)
2084 {
2085   struct LocalGetContext *lgc = cls;
2086   uint32_t type;
2087   
2088   type = lgc->type;
2089   if (type == GNUNET_DATASTORE_BLOCKTYPE_DBLOCK)
2090     type = GNUNET_DATASTORE_BLOCKTYPE_ANY; /* to get on-demand as well */
2091   GNUNET_DATASTORE_get (dsh,
2092                         &lgc->query,
2093                         type,
2094                         &process_local_get_result,
2095                         lgc,
2096                         GNUNET_TIME_UNIT_FOREVER_REL);
2097 }
2098
2099
2100 /**
2101  * We're processing a search request from a local
2102  * client.  Now it is our turn to query the datastore.
2103  * 
2104  * @param cls our closure (struct LocalGetContext)
2105  * @param ok did we succeed to queue for datastore access, should always be GNUNET_OK
2106  */
2107 static void 
2108 transmit_local_get_ready (void *cls,
2109                           int ok)
2110 {
2111   struct LocalGetContext *lgc = cls;
2112
2113   GNUNET_assert (GNUNET_OK == ok);
2114   GNUNET_SCHEDULER_add_continuation (sched,
2115                                      GNUNET_NO,
2116                                      &transmit_local_get,
2117                                      lgc,
2118                                      GNUNET_SCHEDULER_REASON_PREREQ_DONE);
2119 }
2120
2121
2122 /**
2123  * Handle START_SEARCH-message (search request from client).
2124  *
2125  * @param cls closure
2126  * @param client identification of the client
2127  * @param message the actual message
2128  */
2129 static void
2130 handle_start_search (void *cls,
2131                      struct GNUNET_SERVER_Client *client,
2132                      const struct GNUNET_MessageHeader *message)
2133 {
2134   const struct SearchMessage *sm;
2135   struct LocalGetContext *lgc;
2136   uint16_t msize;
2137   unsigned int sc;
2138   
2139   msize = ntohs (message->size);
2140   if ( (msize < sizeof (struct SearchMessage)) ||
2141        (0 != (msize - sizeof (struct SearchMessage)) % sizeof (GNUNET_HashCode)) )
2142     {
2143       GNUNET_break (0);
2144       GNUNET_SERVER_receive_done (client,
2145                                   GNUNET_SYSERR);
2146       return;
2147     }
2148   sc = (msize - sizeof (struct SearchMessage)) / sizeof (GNUNET_HashCode);
2149   sm = (const struct SearchMessage*) message;
2150   GNUNET_SERVER_client_keep (client);
2151   lgc = GNUNET_malloc (sizeof (struct LocalGetContext));
2152   if  (sc > 0)
2153     {
2154       lgc->results_used = sc;
2155       GNUNET_array_grow (lgc->results,
2156                          lgc->results_size,
2157                          sc * 2);
2158       memcpy (lgc->results,
2159               &sm[1],
2160               sc * sizeof (GNUNET_HashCode));
2161     }
2162   lgc->client = client;
2163   lgc->type = ntohl (sm->type);
2164   lgc->anonymity_level = ntohl (sm->anonymity_level);
2165   switch (lgc->type)
2166     {
2167     case GNUNET_DATASTORE_BLOCKTYPE_DBLOCK:
2168     case GNUNET_DATASTORE_BLOCKTYPE_IBLOCK:
2169       lgc->target.hashPubKey = sm->target;
2170       break;
2171     case GNUNET_DATASTORE_BLOCKTYPE_SBLOCK:
2172       lgc->namespace = sm->target;
2173       break;
2174     default:
2175       break;
2176     }
2177   lgc->query = sm->query;
2178   GNUNET_CONTAINER_DLL_insert (lgc_head, lgc_tail, lgc);
2179   lgc->req = queue_ds_request (GNUNET_TIME_UNIT_FOREVER_REL,
2180                                &transmit_local_get_ready,
2181                                lgc);
2182 }
2183
2184
2185 /**
2186  * List of handlers for the messages understood by this
2187  * service.
2188  */
2189 static struct GNUNET_SERVER_MessageHandler handlers[] = {
2190   {&handle_index_start, NULL, 
2191    GNUNET_MESSAGE_TYPE_FS_INDEX_START, 0},
2192   {&handle_index_list_get, NULL, 
2193    GNUNET_MESSAGE_TYPE_FS_INDEX_LIST_GET, sizeof(struct GNUNET_MessageHeader) },
2194   {&handle_unindex, NULL, GNUNET_MESSAGE_TYPE_FS_UNINDEX, 
2195    sizeof (struct UnindexMessage) },
2196   {&handle_start_search, NULL, GNUNET_MESSAGE_TYPE_FS_START_SEARCH, 
2197    0 },
2198   {NULL, NULL, 0, 0}
2199 };
2200
2201
2202 /**
2203  * Clean up the memory used by the PendingRequest structure (except
2204  * for the client or peer list that the request may be part of).
2205  *
2206  * @param pr request to clean up
2207  */
2208 static void
2209 destroy_pending_request (struct PendingRequest *pr)
2210 {
2211   struct PendingReply *reply;
2212   struct ClientList *cl;
2213
2214   GNUNET_CONTAINER_multihashmap_remove (requests_by_query,
2215                                         &pr->query,
2216                                         pr);
2217   // FIXME: not sure how this can work (efficiently)
2218   // also, what does the return value mean?
2219   if (pr->client == NULL)
2220     {
2221       GNUNET_CONTAINER_heap_remove_node (requests_by_expiration,
2222                                          pr);
2223     }
2224   else
2225     {
2226       cl = pr->crl_entry->cl;
2227       GNUNET_CONTAINER_DLL_remove (cl->head,
2228                                    cl->tail,
2229                                    pr->crl_entry);
2230     }
2231   if (GNUNET_SCHEDULER_NO_TASK != pr->task)
2232     GNUNET_SCHEDULER_cancel (sched, pr->task);
2233   if (NULL != pr->cth)
2234     GNUNET_CORE_notify_transmit_ready_cancel (pr->cth);
2235   if (NULL != pr->bf)
2236     GNUNET_CONTAINER_bloomfilter_free (pr->bf);
2237   if (NULL != pr->th)
2238     GNUNET_CONNECTION_notify_transmit_ready_cancel (pr->th);
2239   while (NULL != (reply = pr->replies_pending))
2240     {
2241       pr->replies_pending = reply->next;
2242       GNUNET_free (reply);
2243     }
2244   GNUNET_PEER_change_rc (pr->source_pid, -1);
2245   GNUNET_PEER_change_rc (pr->target_pid, -1);
2246   GNUNET_PEER_decrement_rcs (pr->used_pids, pr->used_pids_off);
2247   GNUNET_free_non_null (pr->used_pids);
2248   GNUNET_free_non_null (pr->replies_seen);
2249   GNUNET_free_non_null (pr->namespace);
2250   GNUNET_free (pr);
2251 }
2252
2253
2254 /**
2255  * A client disconnected.  Remove all of its pending queries.
2256  *
2257  * @param cls closure, NULL
2258  * @param client identification of the client
2259  */
2260 static void
2261 handle_client_disconnect (void *cls,
2262                           struct GNUNET_SERVER_Client
2263                           * client)
2264 {
2265   struct LocalGetContext *lgc;
2266   struct ClientList *cpos;
2267   struct ClientList *cprev;
2268   struct ClientRequestList *rl;
2269
2270   lgc = lgc_head;
2271   while ( (NULL != lgc) &&
2272           (lgc->client != client) )
2273     lgc = lgc->next;
2274   if (lgc != NULL)
2275     local_get_context_free (lgc);
2276   cprev = NULL;
2277   cpos = clients;
2278   while ( (NULL != cpos) &&
2279           (clients->client != client) )
2280     {
2281       cprev = cpos;
2282       cpos = cpos->next;
2283     }
2284   if (cpos != NULL)
2285     {
2286       if (cprev == NULL)
2287         clients = cpos->next;
2288       else
2289         cprev->next = cpos->next;
2290       while (NULL != (rl = cpos->head))
2291         {
2292           cpos->head = rl->next;
2293           destroy_pending_request (rl->req);
2294           GNUNET_free (rl);
2295         }
2296       GNUNET_free (cpos);
2297     }
2298 }
2299
2300
2301 /**
2302  * Iterator over entries in the "requests_by_query" map
2303  * that frees all the entries.
2304  *
2305  * @param cls closure, NULL
2306  * @param key current key code (the query, unused) 
2307  * @param value value in the hash map, of type "struct PendingRequest*"
2308  * @return GNUNET_YES (we should continue to  iterate)
2309  */
2310 static int 
2311 destroy_pending_request_cb (void *cls,
2312                             const GNUNET_HashCode * key,
2313                             void *value)
2314 {
2315   struct PendingRequest *pr = value;
2316
2317   destroy_pending_request (pr);
2318   return GNUNET_YES;
2319 }
2320
2321
2322 /**
2323  * Task run during shutdown.
2324  *
2325  * @param cls unused
2326  * @param tc unused
2327  */
2328 static void
2329 shutdown_task (void *cls,
2330                const struct GNUNET_SCHEDULER_TaskContext *tc)
2331 {
2332   struct IndexInfo *pos;  
2333
2334   if (NULL != core)
2335     GNUNET_CORE_disconnect (core);
2336   GNUNET_DATASTORE_disconnect (dsh,
2337                                GNUNET_NO);
2338   dsh = NULL;
2339   GNUNET_CONTAINER_multihashmap_iterate (requests_by_query,
2340                                          &destroy_pending_request_cb,
2341                                          NULL);
2342   while (clients != NULL)
2343     handle_client_disconnect (NULL,
2344                               clients->client);
2345   GNUNET_CONTAINER_multihashmap_destroy (requests_by_query);
2346   requests_by_query = NULL;
2347   GNUNET_CONTAINER_multihashmap_destroy (requests_by_peer);
2348   requests_by_peer = NULL;
2349   GNUNET_CONTAINER_heap_destroy (requests_by_expiration);
2350   requests_by_expiration = NULL;
2351   // FIXME: iterate over entries and free individually?
2352   // (or do we get disconnect notifications?)
2353   GNUNET_CONTAINER_multihashmap_destroy (connected_peers);
2354   connected_peers = NULL;
2355   GNUNET_CONTAINER_multihashmap_destroy (ifm);
2356   ifm = NULL;
2357   while (NULL != (pos = indexed_files))
2358     {
2359       indexed_files = pos->next;
2360       GNUNET_free (pos);
2361     }
2362 }
2363
2364
2365 /**
2366  * Free (each) request made by the peer.
2367  *
2368  * @param cls closure, points to peer that the request belongs to
2369  * @param key current key code
2370  * @param value value in the hash map
2371  * @return GNUNET_YES (we should continue to iterate)
2372  */
2373 static int
2374 destroy_request (void *cls,
2375                  const GNUNET_HashCode * key,
2376                  void *value)
2377 {
2378   const struct GNUNET_PeerIdentity * peer = cls;
2379   struct PendingRequest *pr = value;
2380   
2381   GNUNET_CONTAINER_multihashmap_remove (requests_by_peer,
2382                                         &peer->hashPubKey,
2383                                         pr);
2384   destroy_pending_request (pr);
2385   return GNUNET_YES;
2386 }
2387
2388
2389
2390 /**
2391  * Method called whenever a given peer connects.
2392  *
2393  * @param cls closure, not used
2394  * @param peer peer identity this notification is about
2395  */
2396 static void 
2397 peer_connect_handler (void *cls,
2398                       const struct
2399                       GNUNET_PeerIdentity * peer)
2400 {
2401   struct ConnectedPeer *cp;
2402
2403   cp = GNUNET_malloc (sizeof (struct ConnectedPeer));
2404   cp->pid = GNUNET_PEER_intern (peer);
2405   GNUNET_CONTAINER_multihashmap_put (connected_peers,
2406                                      &peer->hashPubKey,
2407                                      cp,
2408                                      GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY);
2409 }
2410
2411
2412 /**
2413  * Method called whenever a peer disconnects.
2414  *
2415  * @param cls closure, not used
2416  * @param peer peer identity this notification is about
2417  */
2418 static void
2419 peer_disconnect_handler (void *cls,
2420                          const struct
2421                          GNUNET_PeerIdentity * peer)
2422 {
2423   struct ConnectedPeer *cp;
2424
2425   cp = GNUNET_CONTAINER_multihashmap_get (connected_peers,
2426                                           &peer->hashPubKey);
2427   GNUNET_PEER_change_rc (cp->pid, -1);
2428   GNUNET_PEER_decrement_rcs (cp->last_p2p_replies, P2P_SUCCESS_LIST_SIZE);
2429   GNUNET_free (cp);
2430   GNUNET_CONTAINER_multihashmap_get_multiple (requests_by_peer,
2431                                               &peer->hashPubKey,
2432                                               &destroy_request,
2433                                               (void*) peer);
2434 }
2435
2436
2437 /**
2438  * We're processing a GET request from
2439  * another peer and have decided to forward
2440  * it to other peers.
2441  *
2442  * @param cls our "struct ProcessGetContext *"
2443  * @param tc unused
2444  */
2445 static void
2446 forward_get_request (void *cls,
2447                      const struct GNUNET_SCHEDULER_TaskContext *tc)
2448 {
2449   struct ProcessGetContext *pgc = cls;
2450   struct PendingRequest *pr;
2451   struct PendingRequest *eer;
2452   struct GNUNET_PeerIdentity target;
2453
2454   pr = GNUNET_malloc (sizeof (struct PendingRequest));
2455   if (GET_MESSAGE_BIT_SKS_NAMESPACE == (GET_MESSAGE_BIT_SKS_NAMESPACE & pgc->bm))
2456     {
2457       pr->namespace = GNUNET_malloc (sizeof(GNUNET_HashCode));
2458       *pr->namespace = pgc->namespace;
2459     }
2460   pr->bf = pgc->bf;
2461   pr->bf_size = pgc->bf_size;
2462   pgc->bf = NULL;
2463   pr->start_time = pgc->start_time;
2464   pr->query = pgc->query;
2465   pr->source_pid = GNUNET_PEER_intern (&pgc->reply_to);
2466   if (GET_MESSAGE_BIT_TRANSMIT_TO == (GET_MESSAGE_BIT_TRANSMIT_TO & pgc->bm))
2467     pr->target_pid = GNUNET_PEER_intern (&pgc->prime_target);
2468   pr->anonymity_level = 1; /* default */
2469   pr->priority = pgc->priority;
2470   pr->remaining_priority = pr->priority;
2471   pr->mingle = pgc->mingle;
2472   pr->ttl = pgc->ttl; 
2473   pr->type = pgc->type;
2474   GNUNET_CONTAINER_multihashmap_put (requests_by_query,
2475                                      &pr->query,
2476                                      pr,
2477                                      GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
2478   GNUNET_CONTAINER_multihashmap_put (requests_by_peer,
2479                                      &pgc->reply_to.hashPubKey,
2480                                      pr,
2481                                      GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
2482   GNUNET_CONTAINER_heap_insert (requests_by_expiration,
2483                                 pr,
2484                                 pr->start_time.value + pr->ttl);
2485   if (GNUNET_CONTAINER_heap_get_size (requests_by_expiration) > max_pending_requests)
2486     {
2487       /* expire oldest request! */
2488       eer = GNUNET_CONTAINER_heap_peek (requests_by_expiration);
2489       GNUNET_PEER_resolve (eer->source_pid,
2490                            &target);    
2491       GNUNET_CONTAINER_multihashmap_remove (requests_by_peer,
2492                                             &target.hashPubKey,
2493                                             eer);
2494       destroy_pending_request (eer);     
2495     }
2496   pr->task = GNUNET_SCHEDULER_add_delayed (sched,
2497                                            GNUNET_NO,
2498                                            GNUNET_SCHEDULER_PRIORITY_IDLE,
2499                                            GNUNET_SCHEDULER_NO_TASK,
2500                                            get_processing_delay (),
2501                                            &forward_request_task,
2502                                            pr);
2503   GNUNET_free (pgc); 
2504 }
2505 /**
2506  * Transmit the given message by copying it to
2507  * the target buffer "buf".  "buf" will be
2508  * NULL and "size" zero if the socket was closed for
2509  * writing in the meantime.  In that case, only
2510
2511  * free the message
2512  *
2513  * @param cls closure, pointer to the message
2514  * @param size number of bytes available in buf
2515  * @param buf where the callee should write the message
2516  * @return number of bytes written to buf
2517  */
2518 static size_t
2519 transmit_message (void *cls,
2520                   size_t size, void *buf)
2521 {
2522   struct GNUNET_MessageHeader *msg = cls;
2523   uint16_t msize;
2524   
2525   if (NULL == buf)
2526     {
2527 #if DEBUG_FS
2528       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2529                   "Dropping reply, core too busy.\n");
2530 #endif
2531       GNUNET_free (msg);
2532       return 0;
2533     }
2534   msize = ntohs (msg->size);
2535   GNUNET_assert (size >= msize);
2536   memcpy (buf, msg, msize);
2537   GNUNET_free (msg);
2538   return msize;
2539 }
2540
2541
2542 /**
2543  * Test if the load on this peer is too high
2544  * to even consider processing the query at
2545  * all.
2546  * 
2547  * @return GNUNET_YES if the load is too high, GNUNET_NO otherwise
2548  */
2549 static int
2550 test_load_too_high ()
2551 {
2552   return GNUNET_NO; // FIXME
2553 }
2554
2555
2556 /**
2557  * We're processing (local) results for a search request
2558  * from another peer.  Pass applicable results to the
2559  * peer and if we are done either clean up (operation
2560  * complete) or forward to other peers (more results possible).
2561  *
2562  * @param cls our closure (struct LocalGetContext)
2563  * @param key key for the content
2564  * @param size number of bytes in data
2565  * @param data content stored
2566  * @param type type of the content
2567  * @param priority priority of the content
2568  * @param anonymity anonymity-level for the content
2569  * @param expiration expiration time for the content
2570  * @param uid unique identifier for the datum;
2571  *        maybe 0 if no unique identifier is available
2572  */
2573 static void
2574 process_p2p_get_result (void *cls,
2575                         const GNUNET_HashCode * key,
2576                         uint32_t size,
2577                         const void *data,
2578                         uint32_t type,
2579                         uint32_t priority,
2580                         uint32_t anonymity,
2581                         struct GNUNET_TIME_Absolute
2582                         expiration, 
2583                         uint64_t uid)
2584 {
2585   struct ProcessGetContext *pgc = cls;
2586   GNUNET_HashCode dhash;
2587   GNUNET_HashCode mhash;
2588   struct PutMessage *reply;
2589   
2590   if (NULL == key)
2591     {
2592       /* no more results */
2593       if ( ( (pgc->policy & ROUTING_POLICY_FORWARD) ==  ROUTING_POLICY_FORWARD) &&
2594            ( (0 == pgc->results_found) ||
2595              (pgc->type == GNUNET_DATASTORE_BLOCKTYPE_KBLOCK) ||
2596              (pgc->type == GNUNET_DATASTORE_BLOCKTYPE_SBLOCK) ||
2597              (pgc->type == GNUNET_DATASTORE_BLOCKTYPE_SKBLOCK) ) )
2598         {
2599           GNUNET_SCHEDULER_add_continuation (sched,
2600                                              GNUNET_NO,
2601                                              &forward_get_request,
2602                                              pgc,
2603                                              GNUNET_SCHEDULER_REASON_PREREQ_DONE);
2604         }
2605       else
2606         {
2607           if (pgc->bf != NULL)
2608             GNUNET_CONTAINER_bloomfilter_free (pgc->bf);
2609           GNUNET_free (pgc); 
2610         }
2611       next_ds_request ();
2612       return;
2613     }
2614   if (type == GNUNET_DATASTORE_BLOCKTYPE_ONDEMAND)
2615     {
2616       handle_on_demand_block (key, size, data, type, priority, 
2617                               anonymity, expiration, uid,
2618                               &process_p2p_get_result,
2619                               pgc);
2620       return;
2621     }
2622   /* check for duplicates */
2623   GNUNET_CRYPTO_hash (data, size, &dhash);
2624   mingle_hash (&dhash, 
2625                pgc->mingle,
2626                &mhash);
2627   if ( (pgc->bf != NULL) &&
2628        (GNUNET_YES ==
2629         GNUNET_CONTAINER_bloomfilter_test (pgc->bf,
2630                                            &mhash)) )
2631     {      
2632 #if DEBUG_FS
2633       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2634                   "Result from datastore filtered by bloomfilter.\n");
2635 #endif
2636       GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
2637       return;
2638     }
2639   pgc->results_found++;
2640   if ( (pgc->type == GNUNET_DATASTORE_BLOCKTYPE_KBLOCK) ||
2641        (pgc->type == GNUNET_DATASTORE_BLOCKTYPE_SBLOCK) ||
2642        (pgc->type == GNUNET_DATASTORE_BLOCKTYPE_SKBLOCK) )
2643     {
2644       if (pgc->bf == NULL)
2645         {
2646           pgc->bf_size = 32;
2647           pgc->bf = GNUNET_CONTAINER_bloomfilter_init (NULL,
2648                                                        pgc->bf_size, 
2649                                                        BLOOMFILTER_K);
2650         }
2651       GNUNET_CONTAINER_bloomfilter_add (pgc->bf, 
2652                                         &mhash);
2653     }
2654
2655   reply = GNUNET_malloc (sizeof (struct PutMessage) + size);
2656   reply->header.size = htons (sizeof (struct PutMessage) + size);
2657   reply->header.type = htons (GNUNET_MESSAGE_TYPE_FS_PUT);
2658   reply->type = htonl (type);
2659   reply->expiration = GNUNET_TIME_relative_hton (GNUNET_TIME_absolute_get_remaining (expiration));
2660   memcpy (&reply[1], data, size);
2661   GNUNET_CORE_notify_transmit_ready (core,
2662                                      pgc->priority,
2663                                      ACCEPTABLE_REPLY_DELAY,
2664                                      &pgc->reply_to,
2665                                      sizeof (struct PutMessage) + size,
2666                                      &transmit_message,
2667                                      reply);
2668   if ( (GNUNET_YES == test_load_too_high()) ||
2669        (pgc->results_found > 5 + 2 * pgc->priority) )
2670     {
2671       GNUNET_DATASTORE_get_next (dsh, GNUNET_NO);
2672       pgc->policy &= ~ ROUTING_POLICY_FORWARD;
2673       return;
2674     }
2675   GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
2676 }
2677   
2678
2679 /**
2680  * We're processing a GET request from another peer.  Give it to our
2681  * local datastore.
2682  *
2683  * @param cls our "struct ProcessGetContext"
2684  * @param ok did we get a datastore slice or not?
2685  */
2686 static void
2687 ds_get_request (void *cls, 
2688                 int ok)
2689 {
2690   struct ProcessGetContext *pgc = cls;
2691   uint32_t type;
2692   struct GNUNET_TIME_Relative timeout;
2693
2694   if (GNUNET_OK != ok)
2695     {
2696       /* no point in doing P2P stuff if we can't even do local */
2697       GNUNET_free (dsh);
2698       return;
2699     }
2700   type = pgc->type;
2701   if (type == GNUNET_DATASTORE_BLOCKTYPE_DBLOCK)
2702     type = GNUNET_DATASTORE_BLOCKTYPE_ANY; /* to get on-demand as well */
2703   timeout = GNUNET_TIME_relative_multiply (BASIC_DATASTORE_REQUEST_DELAY,
2704                                            (pgc->priority + 1));
2705   GNUNET_DATASTORE_get (dsh,
2706                         &pgc->query,
2707                         type,
2708                         &process_p2p_get_result,
2709                         pgc,
2710                         timeout);
2711 }
2712
2713
2714 /**
2715  * The priority level imposes a bound on the maximum
2716  * value for the ttl that can be requested.
2717  *
2718  * @param ttl_in requested ttl
2719  * @param prio given priority
2720  * @return ttl_in if ttl_in is below the limit,
2721  *         otherwise the ttl-limit for the given priority
2722  */
2723 static int32_t
2724 bound_ttl (int32_t ttl_in, uint32_t prio)
2725 {
2726   unsigned long long allowed;
2727
2728   if (ttl_in <= 0)
2729     return ttl_in;
2730   allowed = ((unsigned long long) prio) * TTL_DECREMENT / 1000; 
2731   if (ttl_in > allowed)      
2732     {
2733       if (allowed >= (1 << 30))
2734         return 1 << 30;
2735       return allowed;
2736     }
2737   return ttl_in;
2738 }
2739
2740
2741 /**
2742  * We've received a request with the specified
2743  * priority.  Bound it according to how much
2744  * we trust the given peer.
2745  * 
2746  * @param prio_in requested priority
2747  * @param peer the peer making the request
2748  * @return effective priority
2749  */
2750 static uint32_t
2751 bound_priority (uint32_t prio_in,
2752                 const struct GNUNET_PeerIdentity *peer)
2753 {
2754   return 0; // FIXME!
2755 }
2756
2757
2758 /**
2759  * Handle P2P "GET" request.
2760  *
2761  * @param cls closure, always NULL
2762  * @param other the other peer involved (sender or receiver, NULL
2763  *        for loopback messages where we are both sender and receiver)
2764  * @param message the actual message
2765  * @return GNUNET_OK to keep the connection open,
2766  *         GNUNET_SYSERR to close it (signal serious error)
2767  */
2768 static int
2769 handle_p2p_get (void *cls,
2770                 const struct GNUNET_PeerIdentity *other,
2771                 const struct GNUNET_MessageHeader *message)
2772 {
2773   uint16_t msize;
2774   const struct GetMessage *gm;
2775   unsigned int bits;
2776   const GNUNET_HashCode *opt;
2777   struct ProcessGetContext *pgc;
2778   uint32_t bm;
2779   size_t bfsize;
2780   uint32_t ttl_decrement;
2781   double preference;
2782   int net_load_up;
2783   int net_load_down;
2784
2785   msize = ntohs(message->size);
2786   if (msize < sizeof (struct GetMessage))
2787     {
2788       GNUNET_break_op (0);
2789       return GNUNET_SYSERR;
2790     }
2791   gm = (const struct GetMessage*) message;
2792   bm = ntohl (gm->hash_bitmap);
2793   bits = 0;
2794   while (bm > 0)
2795     {
2796       if (1 == (bm & 1))
2797         bits++;
2798       bm >>= 1;
2799     }
2800   if (msize < sizeof (struct GetMessage) + bits * sizeof (GNUNET_HashCode))
2801     {
2802       GNUNET_break_op (0);
2803       return GNUNET_SYSERR;
2804     }  
2805   opt = (const GNUNET_HashCode*) &gm[1];
2806   bfsize = msize - sizeof (struct GetMessage) + bits * sizeof (GNUNET_HashCode);
2807   pgc = GNUNET_malloc (sizeof (struct ProcessGetContext));
2808   if (bfsize > 0)
2809     {
2810       pgc->bf = GNUNET_CONTAINER_bloomfilter_init ((const char*) &pgc[1],
2811                                                    bfsize,
2812                                                    BLOOMFILTER_K);
2813       pgc->bf_size = bfsize;
2814     }
2815   pgc->type = ntohl (gm->type);
2816   pgc->bm = ntohl (gm->hash_bitmap);
2817   pgc->mingle = gm->filter_mutator;
2818   bits = 0;
2819   if (0 != (pgc->bm & GET_MESSAGE_BIT_RETURN_TO))
2820     pgc->reply_to.hashPubKey = opt[bits++];
2821   else
2822     pgc->reply_to = *other;
2823   if (0 != (pgc->bm & GET_MESSAGE_BIT_SKS_NAMESPACE))
2824     pgc->namespace = opt[bits++];
2825   else if (pgc->type == GNUNET_DATASTORE_BLOCKTYPE_SBLOCK)
2826     {
2827       GNUNET_break_op (0);
2828       GNUNET_free (pgc);
2829       return GNUNET_SYSERR;
2830     }
2831   if (0 != (pgc->bm & GET_MESSAGE_BIT_TRANSMIT_TO))
2832     pgc->prime_target.hashPubKey = opt[bits++];
2833   /* note that we can really only check load here since otherwise
2834      peers could find out that we are overloaded by being disconnected
2835      after sending us a malformed query... */
2836   if (GNUNET_YES == test_load_too_high ())
2837     {
2838       if (NULL != pgc->bf)
2839         GNUNET_CONTAINER_bloomfilter_free (pgc->bf);
2840       GNUNET_free (pgc);
2841 #if DEBUG_FS
2842       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2843                   "Dropping query from `%s', this peer is too busy.\n",
2844                   GNUNET_i2s (other));
2845 #endif
2846       return GNUNET_OK;
2847     }
2848   net_load_up = 50; // FIXME
2849   net_load_down = 50; // FIXME
2850   pgc->policy = ROUTING_POLICY_NONE;
2851   if ( (net_load_up < IDLE_LOAD_THRESHOLD) &&
2852        (net_load_down < IDLE_LOAD_THRESHOLD) )
2853     {
2854       pgc->policy |= ROUTING_POLICY_ALL;
2855       pgc->priority = 0; /* no charge */
2856     }
2857   else
2858     {
2859       pgc->priority = bound_priority (ntohl (gm->priority), other);
2860       if ( (net_load_up < 
2861             IDLE_LOAD_THRESHOLD + pgc->priority * pgc->priority) &&
2862            (net_load_down < 
2863             IDLE_LOAD_THRESHOLD + pgc->priority * pgc->priority) )
2864         {
2865           pgc->policy |= ROUTING_POLICY_ALL;
2866         }
2867       else
2868         {
2869           // FIXME: is this sound?
2870           if (net_load_up < 90 + 10 * pgc->priority)
2871             pgc->policy |= ROUTING_POLICY_FORWARD;
2872           if (net_load_down < 90 + 10 * pgc->priority)
2873             pgc->policy |= ROUTING_POLICY_ANSWER;
2874         }
2875     }
2876   if (pgc->policy == ROUTING_POLICY_NONE)
2877     {
2878 #if DEBUG_FS
2879       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2880                   "Dropping query from `%s', network saturated.\n",
2881                   GNUNET_i2s (other));
2882 #endif
2883       if (NULL != pgc->bf)
2884         GNUNET_CONTAINER_bloomfilter_free (pgc->bf);
2885       GNUNET_free (pgc);
2886       return GNUNET_OK;     /* drop */
2887     }
2888   if ((pgc->policy & ROUTING_POLICY_INDIRECT) != ROUTING_POLICY_INDIRECT)
2889     pgc->priority = 0;  /* kill the priority (we cannot benefit) */
2890   pgc->ttl = bound_ttl (ntohl (gm->ttl), pgc->priority);
2891   /* decrement ttl (always) */
2892   ttl_decrement = 2 * TTL_DECREMENT +
2893     GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
2894                               TTL_DECREMENT);
2895   if ( (pgc->ttl < 0) &&
2896        (pgc->ttl - ttl_decrement > 0) )
2897     {
2898 #if DEBUG_FS
2899       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2900                   "Dropping query from `%s' due to TTL underflow.\n",
2901                   GNUNET_i2s (other));
2902 #endif
2903       /* integer underflow => drop (should be very rare)! */
2904       if (NULL != pgc->bf)
2905         GNUNET_CONTAINER_bloomfilter_free (pgc->bf);
2906       GNUNET_free (pgc);
2907       return GNUNET_OK;
2908     }
2909   pgc->ttl -= ttl_decrement;
2910   pgc->start_time = GNUNET_TIME_absolute_get ();
2911   preference = (double) pgc->priority;
2912   if (preference < QUERY_BANDWIDTH_VALUE)
2913     preference = QUERY_BANDWIDTH_VALUE;
2914   // FIXME: also reserve bandwidth for reply?
2915   GNUNET_CORE_peer_configure (core,
2916                               other,
2917                               GNUNET_TIME_UNIT_FOREVER_REL,
2918                               0, 0, preference, NULL, NULL);
2919   if (0 != (pgc->policy & ROUTING_POLICY_ANSWER))
2920     pgc->drq = queue_ds_request (BASIC_DATASTORE_REQUEST_DELAY,
2921                                  &ds_get_request,
2922                                  pgc);
2923   else
2924     GNUNET_SCHEDULER_add_continuation (sched,
2925                                        GNUNET_NO,
2926                                        &forward_get_request,
2927                                        pgc,
2928                                        GNUNET_SCHEDULER_REASON_PREREQ_DONE);
2929   return GNUNET_OK;
2930 }
2931
2932
2933 /**
2934  * Function called to notify us that we can now transmit a reply to a
2935  * client or peer.  "buf" will be NULL and "size" zero if the socket was
2936  * closed for writing in the meantime.
2937  *
2938  * @param cls closure, points to a "struct PendingRequest*" with
2939  *            one or more pending replies
2940  * @param size number of bytes available in buf
2941  * @param buf where the callee should write the message
2942  * @return number of bytes written to buf
2943  */
2944 static size_t
2945 transmit_result (void *cls,
2946                  size_t size, 
2947                  void *buf)
2948 {
2949   struct PendingRequest *pr = cls;
2950   char *cbuf = buf;
2951   struct PendingReply *reply;
2952   size_t ret;
2953
2954   ret = 0;
2955   while (NULL != (reply = pr->replies_pending))
2956     {
2957       if ( (reply->msize + ret < ret) ||
2958            (reply->msize + ret > size) )
2959         break;
2960       pr->replies_pending = reply->next;
2961       memcpy (&cbuf[ret], &reply[1], reply->msize);
2962       ret += reply->msize;
2963       GNUNET_free (reply);
2964     }
2965   return ret;
2966 }
2967
2968
2969 /**
2970  * Iterator over pending requests.
2971  *
2972  * @param cls response (struct ProcessReplyClosure)
2973  * @param key our query
2974  * @param value value in the hash map (meta-info about the query)
2975  * @return GNUNET_YES (we should continue to iterate)
2976  */
2977 static int
2978 process_reply (void *cls,
2979                const GNUNET_HashCode * key,
2980                void *value)
2981 {
2982   struct ProcessReplyClosure *prq = cls;
2983   struct PendingRequest *pr = value;
2984   struct PendingRequest *eer;
2985   struct PendingReply *reply;
2986   struct PutMessage *pm;
2987   struct ContentMessage *cm;
2988   GNUNET_HashCode chash;
2989   GNUNET_HashCode mhash;
2990   struct GNUNET_PeerIdentity target;
2991   size_t msize;
2992   uint32_t prio;
2993   struct GNUNET_TIME_Relative max_delay;
2994   
2995   GNUNET_CRYPTO_hash (prq->data,
2996                       prq->size,
2997                       &chash);
2998   switch (prq->type)
2999     {
3000     case GNUNET_DATASTORE_BLOCKTYPE_DBLOCK:
3001     case GNUNET_DATASTORE_BLOCKTYPE_IBLOCK:
3002       break;
3003     case GNUNET_DATASTORE_BLOCKTYPE_SBLOCK:
3004       /* FIXME: does prq->namespace match our expectations? */
3005       /* then: fall-through??? */
3006     case GNUNET_DATASTORE_BLOCKTYPE_KBLOCK:
3007       if (pr->bf != NULL) 
3008         {
3009           mingle_hash (&chash, pr->mingle, &mhash);
3010           if (GNUNET_YES == GNUNET_CONTAINER_bloomfilter_test (pr->bf,
3011                                                                &mhash))
3012             return GNUNET_YES; /* duplicate */
3013           GNUNET_CONTAINER_bloomfilter_add (pr->bf,
3014                                             &mhash);
3015         }
3016       break;
3017     case GNUNET_DATASTORE_BLOCKTYPE_SKBLOCK:
3018       // FIXME: any checks against duplicates for SKBlocks?
3019       break;
3020     }
3021   prio = pr->priority;
3022   prq->priority += pr->remaining_priority;
3023   pr->remaining_priority = 0;
3024   if (pr->client != NULL)
3025     {
3026       if (pr->replies_seen_size == pr->replies_seen_off)
3027         GNUNET_array_grow (pr->replies_seen,
3028                            pr->replies_seen_size,
3029                            pr->replies_seen_size * 2 + 4);
3030       pr->replies_seen[pr->replies_seen_off++] = chash;
3031       // FIXME: possibly recalculate BF!
3032     }
3033   if (pr->client == NULL)
3034     {
3035       msize = sizeof (struct ContentMessage) + prq->size;
3036       reply = GNUNET_malloc (msize + sizeof (struct PendingReply));
3037       reply->msize = msize;
3038       cm = (struct ContentMessage*) &reply[1];
3039       cm->header.type = htons (GNUNET_MESSAGE_TYPE_FS_CONTENT);
3040       cm->header.size = htons (msize);
3041       cm->type = htonl (prq->type);
3042       cm->expiration = GNUNET_TIME_absolute_hton (prq->expiration);
3043       reply->next = pr->replies_pending;
3044       pr->replies_pending = reply;
3045       memcpy (&reply[1], prq->data, prq->size);
3046       if (pr->cth != NULL)
3047         return GNUNET_YES;
3048       max_delay = GNUNET_TIME_UNIT_FOREVER_REL;
3049       if (GNUNET_CONTAINER_heap_get_size (requests_by_expiration) >= max_pending_requests)
3050         {
3051           /* estimate expiration time from time difference between
3052              first request that will be discarded and this request */
3053           eer = GNUNET_CONTAINER_heap_peek (requests_by_expiration);
3054           max_delay = GNUNET_TIME_absolute_get_difference (pr->start_time,
3055                                                            eer->start_time);
3056         }
3057       GNUNET_PEER_resolve (pr->source_pid,
3058                            &target);
3059       pr->cth = GNUNET_CORE_notify_transmit_ready (core,
3060                                                    prio,
3061                                                    max_delay,
3062                                                    &target,
3063                                                    msize,
3064                                                    &transmit_result,
3065                                                    pr);
3066       if (NULL == pr->cth)
3067         {
3068           // FIXME: now what? discard?
3069         }
3070     }
3071   else
3072     {
3073       msize = sizeof (struct PutMessage) + prq->size;
3074       reply = GNUNET_malloc (msize + sizeof (struct PendingReply));
3075       reply->msize = msize;
3076       reply->next = pr->replies_pending;
3077       pm = (struct PutMessage*) &reply[1];
3078       pm->header.type = htons (GNUNET_MESSAGE_TYPE_FS_PUT);
3079       pm->header.size = htons (msize);
3080       pm->type = htonl (prq->type);
3081       pm->expiration = GNUNET_TIME_relative_hton (GNUNET_TIME_absolute_get_remaining (prq->expiration));
3082       pr->replies_pending = reply;
3083       memcpy (&reply[1], prq->data, prq->size);
3084       if (pr->th != NULL)
3085         return GNUNET_YES;
3086       pr->th = GNUNET_SERVER_notify_transmit_ready (pr->client,
3087                                                     msize,
3088                                                     GNUNET_TIME_UNIT_FOREVER_REL,
3089                                                     &transmit_result,
3090                                                     pr);
3091       if (pr->th == NULL)
3092         {
3093           // FIXME: need to try again later (not much
3094           // to do here specifically, but we need to
3095           // check somewhere else to handle this case!)
3096         }
3097     }
3098   // FIXME: implement hot-path routing statistics keeping!
3099   return GNUNET_YES;
3100 }
3101
3102
3103 /**
3104  * Check if the given KBlock is well-formed.
3105  *
3106  * @param kb the kblock data (or at least "dsize" bytes claiming to be one)
3107  * @param dsize size of "kb" in bytes; check for < sizeof(struct KBlock)!
3108  * @param query where to store the query that this block answers
3109  * @return GNUNET_OK if this is actually a well-formed KBlock
3110  */
3111 static int
3112 check_kblock (const struct KBlock *kb,
3113               size_t dsize,
3114               GNUNET_HashCode *query)
3115 {
3116   if (dsize < sizeof (struct KBlock))
3117     {
3118       GNUNET_break_op (0);
3119       return GNUNET_SYSERR;
3120     }
3121   if (dsize - sizeof (struct KBlock) !=
3122       ntohs (kb->purpose.size) 
3123       - sizeof (struct GNUNET_CRYPTO_RsaSignaturePurpose) 
3124       - sizeof(struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded) ) 
3125     {
3126       GNUNET_break_op (0);
3127       return GNUNET_SYSERR;
3128     }
3129   if (GNUNET_OK !=
3130       GNUNET_CRYPTO_rsa_verify (GNUNET_SIGNATURE_PURPOSE_FS_KBLOCK,
3131                                 &kb->purpose,
3132                                 &kb->signature,
3133                                 &kb->keyspace)) 
3134     {
3135       GNUNET_break_op (0);
3136       return GNUNET_SYSERR;
3137     }
3138   if (query != NULL)
3139     GNUNET_CRYPTO_hash (&kb->keyspace,
3140                         sizeof (struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded),
3141                         query);
3142   return GNUNET_OK;
3143 }
3144
3145
3146 /**
3147  * Check if the given SBlock is well-formed.
3148  *
3149  * @param sb the sblock data (or at least "dsize" bytes claiming to be one)
3150  * @param dsize size of "kb" in bytes; check for < sizeof(struct SBlock)!
3151  * @param query where to store the query that this block answers
3152  * @param namespace where to store the namespace that this block belongs to
3153  * @return GNUNET_OK if this is actually a well-formed SBlock
3154  */
3155 static int
3156 check_sblock (const struct SBlock *sb,
3157               size_t dsize,
3158               GNUNET_HashCode *query,   
3159               GNUNET_HashCode *namespace)
3160 {
3161   if (dsize < sizeof (struct SBlock))
3162     {
3163       GNUNET_break_op (0);
3164       return GNUNET_SYSERR;
3165     }
3166   if (dsize !=
3167       ntohs (sb->purpose.size) + sizeof (struct GNUNET_CRYPTO_RsaSignature))
3168     {
3169       GNUNET_break_op (0);
3170       return GNUNET_SYSERR;
3171     }
3172   if (GNUNET_OK !=
3173       GNUNET_CRYPTO_rsa_verify (GNUNET_SIGNATURE_PURPOSE_FS_SBLOCK,
3174                                 &sb->purpose,
3175                                 &sb->signature,
3176                                 &sb->subspace)) 
3177     {
3178       GNUNET_break_op (0);
3179       return GNUNET_SYSERR;
3180     }
3181   if (query != NULL)
3182     *query = sb->identifier;
3183   if (namespace != NULL)
3184     GNUNET_CRYPTO_hash (&sb->subspace,
3185                         sizeof (struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded),
3186                         namespace);
3187   return GNUNET_OK;
3188 }
3189
3190
3191 /**
3192  * Handle P2P "PUT" request.
3193  *
3194  * @param cls closure, always NULL
3195  * @param other the other peer involved (sender or receiver, NULL
3196  *        for loopback messages where we are both sender and receiver)
3197  * @param message the actual message
3198  * @return GNUNET_OK to keep the connection open,
3199  *         GNUNET_SYSERR to close it (signal serious error)
3200  */
3201 static int
3202 handle_p2p_put (void *cls,
3203                 const struct GNUNET_PeerIdentity *other,
3204                 const struct GNUNET_MessageHeader *message)
3205 {
3206   const struct PutMessage *put;
3207   uint16_t msize;
3208   size_t dsize;
3209   uint32_t type;
3210   struct GNUNET_TIME_Absolute expiration;
3211   GNUNET_HashCode query;
3212   struct ProcessReplyClosure prq;
3213
3214   msize = ntohs (message->size);
3215   if (msize < sizeof (struct PutMessage))
3216     {
3217       GNUNET_break_op(0);
3218       return GNUNET_SYSERR;
3219     }
3220   put = (const struct PutMessage*) message;
3221   dsize = msize - sizeof (struct PutMessage);
3222   type = ntohl (put->type);
3223   expiration = GNUNET_TIME_relative_to_absolute (GNUNET_TIME_relative_ntoh (put->expiration));
3224
3225   /* first, validate! */
3226   switch (type)
3227     {
3228     case GNUNET_DATASTORE_BLOCKTYPE_DBLOCK:
3229     case GNUNET_DATASTORE_BLOCKTYPE_IBLOCK:
3230       GNUNET_CRYPTO_hash (&put[1], dsize, &query);
3231       break;
3232     case GNUNET_DATASTORE_BLOCKTYPE_KBLOCK:
3233       if (GNUNET_OK !=
3234           check_kblock ((const struct KBlock*) &put[1],
3235                         dsize,
3236                         &query))
3237         return GNUNET_SYSERR;
3238       break;
3239     case GNUNET_DATASTORE_BLOCKTYPE_SBLOCK:
3240       if (GNUNET_OK !=
3241           check_sblock ((const struct SBlock*) &put[1],
3242                         dsize,
3243                         &query,
3244                         &prq.namespace))
3245         return GNUNET_SYSERR;
3246       break;
3247     case GNUNET_DATASTORE_BLOCKTYPE_SKBLOCK:
3248       // FIXME -- validate SKBLOCK!
3249       GNUNET_break (0);
3250       return GNUNET_OK;
3251     default:
3252       /* unknown block type */
3253       GNUNET_break_op (0);
3254       return GNUNET_SYSERR;
3255     }
3256
3257   /* now, lookup 'query' */
3258   prq.data = (const void*) &put[1];
3259   prq.size = dsize;
3260   prq.type = type;
3261   prq.expiration = expiration;
3262   prq.priority = 0;
3263   GNUNET_CONTAINER_multihashmap_get_multiple (requests_by_query,
3264                                               &query,
3265                                               &process_reply,
3266                                               &prq);
3267   // FIXME: if migration is on and load is low,
3268   // queue to store data in datastore;
3269   // use "prq.priority" for that!
3270   return GNUNET_OK;
3271 }
3272
3273
3274 /**
3275  * List of handlers for P2P messages
3276  * that we care about.
3277  */
3278 static struct GNUNET_CORE_MessageHandler p2p_handlers[] =
3279   {
3280     { &handle_p2p_get, 
3281       GNUNET_MESSAGE_TYPE_FS_GET, 0 },
3282     { &handle_p2p_put, 
3283       GNUNET_MESSAGE_TYPE_FS_PUT, 0 },
3284     { NULL, 0, 0 }
3285   };
3286
3287
3288 /**
3289  * Task that will try to initiate a connection with the
3290  * core service.
3291  * 
3292  * @param cls unused
3293  * @param tc unused
3294  */
3295 static void
3296 core_connect_task (void *cls,
3297                    const struct GNUNET_SCHEDULER_TaskContext *tc);
3298
3299
3300 /**
3301  * Function called by the core after we've
3302  * connected.
3303  *
3304  * @param cls closure, unused
3305  * @param server handle to the core service
3306  * @param my_identity our peer identity (unused)
3307  * @param publicKey our public key (unused)
3308  */
3309 static void
3310 core_start_cb (void *cls,
3311                struct GNUNET_CORE_Handle * server,
3312                const struct GNUNET_PeerIdentity *
3313                my_identity,
3314                const struct
3315                GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded *
3316                publicKey)
3317 {
3318   if (server == NULL)
3319     {
3320       GNUNET_SCHEDULER_add_delayed (sched,
3321                                     GNUNET_NO,
3322                                     GNUNET_SCHEDULER_PRIORITY_HIGH,
3323                                     GNUNET_SCHEDULER_NO_TASK,
3324                                     GNUNET_TIME_UNIT_SECONDS,
3325                                     &core_connect_task,
3326                                     NULL);
3327       return;
3328     }
3329   core = server;
3330 }
3331
3332
3333 /**
3334  * Task that will try to initiate a connection with the
3335  * core service.
3336  * 
3337  * @param cls unused
3338  * @param tc unused
3339  */
3340 static void
3341 core_connect_task (void *cls,
3342                    const struct GNUNET_SCHEDULER_TaskContext *tc)
3343 {
3344   GNUNET_CORE_connect (sched,
3345                        cfg,
3346                        GNUNET_TIME_UNIT_FOREVER_REL,
3347                        NULL,
3348                        &core_start_cb,
3349                        &peer_connect_handler,
3350                        &peer_disconnect_handler,
3351                        NULL, 
3352                        NULL, GNUNET_NO,
3353                        NULL, GNUNET_NO,
3354                        p2p_handlers);
3355 }
3356
3357
3358 /**
3359  * Process fs requests.
3360  *
3361  * @param cls closure
3362  * @param s scheduler to use
3363  * @param server the initialized server
3364  * @param c configuration to use
3365  */
3366 static void
3367 run (void *cls,
3368      struct GNUNET_SCHEDULER_Handle *s,
3369      struct GNUNET_SERVER_Handle *server,
3370      const struct GNUNET_CONFIGURATION_Handle *c)
3371 {
3372   sched = s;
3373   cfg = c;
3374
3375   ifm = GNUNET_CONTAINER_multihashmap_create (128);
3376   requests_by_query = GNUNET_CONTAINER_multihashmap_create (128); // FIXME: get size from config
3377   requests_by_peer = GNUNET_CONTAINER_multihashmap_create (128); // FIXME: get size from config
3378   connected_peers = GNUNET_CONTAINER_multihashmap_create (64);
3379   requests_by_expiration = GNUNET_CONTAINER_heap_create (GNUNET_CONTAINER_HEAP_ORDER_MIN); 
3380   read_index_list ();
3381   dsh = GNUNET_DATASTORE_connect (cfg,
3382                                   sched);
3383   if (NULL == dsh)
3384     {
3385       GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
3386                   _("Failed to connect to datastore service.\n"));
3387       return;
3388     }
3389   GNUNET_SERVER_disconnect_notify (server, 
3390                                    &handle_client_disconnect,
3391                                    NULL);
3392   GNUNET_SERVER_add_handlers (server, handlers);
3393   core_connect_task (NULL, NULL);
3394   GNUNET_SCHEDULER_add_delayed (sched,
3395                                 GNUNET_YES,
3396                                 GNUNET_SCHEDULER_PRIORITY_IDLE,
3397                                 GNUNET_SCHEDULER_NO_TASK,
3398                                 GNUNET_TIME_UNIT_FOREVER_REL,
3399                                 &shutdown_task,
3400                                 NULL);
3401 }
3402
3403
3404 /**
3405  * The main function for the fs service.
3406  *
3407  * @param argc number of arguments from the command line
3408  * @param argv command line arguments
3409  * @return 0 ok, 1 on error
3410  */
3411 int
3412 main (int argc, char *const *argv)
3413 {
3414   return (GNUNET_OK ==
3415           GNUNET_SERVICE_run (argc,
3416                               argv,
3417                               "fs", &run, NULL, NULL, NULL)) ? 0 : 1;
3418 }
3419
3420 /* end of gnunet-service-fs.c */