src/fs/gnunet-service-fs.c

   1 /*
   2      This file is part of GNUnet.
   3      (C) 2009, 2010 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 gnunet anonymity protocol implementation
  24  * @author Christian Grothoff
  25  *
  26  * TODO:
  27  * - have non-zero preference / priority for requests we initiate!
  28  * - track stats for hot-path routing
  29  * - implement hot-path routing decision procedure
  30  * - implement: bound_priority, test_load_too_high, validate_skblock
  31  * - add content migration support (store locally)
  32  * - statistics
  33  */
  34 #include "platform.h"
  35 #include <float.h>
  36 #include "gnunet_constants.h"
  37 #include "gnunet_core_service.h"
  38 #include "gnunet_datastore_service.h"
  39 #include "gnunet_peer_lib.h"
  40 #include "gnunet_protocols.h"
  41 #include "gnunet_signatures.h"
  42 #include "gnunet_util_lib.h"
  43 #include "gnunet-service-fs_drq.h"
  44 #include "gnunet-service-fs_indexing.h"
  45 #include "fs.h"
  46
  47 #define DEBUG_FS GNUNET_YES
  48
  49 /**
  50  * Maximum number of outgoing messages we queue per peer.
  51  * FIXME: set to a tiny value for testing; make configurable.
  52  */
  53 #define MAX_QUEUE_PER_PEER 2
  54
  55
  56 /**
  57  * Maximum number of requests (from other peers) that we're
  58  * willing to have pending at any given point in time.
  59  * FIXME: set from configuration (and 32 is a tiny value for testing only).
  60  */
  61 static uint64_t max_pending_requests = 32;
  62
  63
  64 /**
  65  * Information we keep for each pending reply.  The
  66  * actual message follows at the end of this struct.
  67  */
  68 struct PendingMessage;
  69
  70
  71 /**
  72  * Function called upon completion of a transmission.
  73  *
  74  * @param cls closure
  75  * @param pid ID of receiving peer, 0 on transmission error
  76  */
  77 typedef void (*TransmissionContinuation)(void * cls,
  78                                          GNUNET_PEER_Id tpid);
  79
  80
  81 /**
  82  * Information we keep for each pending message (GET/PUT).  The
  83  * actual message follows at the end of this struct.
  84  */
  85 struct PendingMessage
  86 {
  87   /**
  88    * This is a doubly-linked list of messages to the same peer.
  89    */
  90   struct PendingMessage *next;
  91
  92   /**
  93    * This is a doubly-linked list of messages to the same peer.
  94    */
  95   struct PendingMessage *prev;
  96
  97   /**
  98    * Entry in pending message list for this pending message.
  99    */
 100   struct PendingMessageList *pml;
 101
 102   /**
 103    * Function to call immediately once we have transmitted this
 104    * message.
 105    */
 106   TransmissionContinuation cont;
 107
 108   /**
 109    * Closure for cont.
 110    */
 111   void *cont_cls;
 112
 113   /**
 114    * Size of the reply; actual reply message follows
 115    * at the end of this struct.
 116    */
 117   size_t msize;
 118
 119   /**
 120    * How important is this message for us?
 121    */
 122   uint32_t priority;
 123
 124 };
 125
 126
 127 /**
 128  * Information about a peer that we are connected to.
 129  * We track data that is useful for determining which
 130  * peers should receive our requests.  We also keep
 131  * a list of messages to transmit to this peer.
 132  */
 133 struct ConnectedPeer
 134 {
 135
 136   /**
 137    * List of the last clients for which this peer successfully
 138    * answered a query.
 139    */
 140   struct GNUNET_SERVER_Client *last_client_replies[CS2P_SUCCESS_LIST_SIZE];
 141
 142   /**
 143    * List of the last PIDs for which
 144    * this peer successfully answered a query;
 145    * We use 0 to indicate no successful reply.
 146    */
 147   GNUNET_PEER_Id last_p2p_replies[P2P_SUCCESS_LIST_SIZE];
 148
 149   /**
 150    * Average delay between sending the peer a request and
 151    * getting a reply (only calculated over the requests for
 152    * which we actually got a reply).   Calculated
 153    * as a moving average: new_delay = ((n-1)*last_delay+curr_delay) / n
 154    */
 155   struct GNUNET_TIME_Relative avg_delay;
 156
 157   /**
 158    * Handle for an active request for transmission to this
 159    * peer, or NULL.
 160    */
 161   struct GNUNET_CORE_TransmitHandle *cth;
 162
 163   /**
 164    * Messages (replies, queries, content migration) we would like to
 165    * send to this peer in the near future.  Sorted by priority, head.
 166    */
 167   struct PendingMessage *pending_messages_head;
 168
 169   /**
 170    * Messages (replies, queries, content migration) we would like to
 171    * send to this peer in the near future.  Sorted by priority, tail.
 172    */
 173   struct PendingMessage *pending_messages_tail;
 174
 175   /**
 176    * Average priority of successful replies.  Calculated
 177    * as a moving average: new_avg = ((n-1)*last_avg+curr_prio) / n
 178    */
 179   double avg_priority;
 180
 181   /**
 182    * Increase in traffic preference still to be submitted
 183    * to the core service for this peer. FIXME: double or 'uint64_t'?
 184    */
 185   double inc_preference;
 186
 187   /**
 188    * The peer's identity.
 189    */
 190   GNUNET_PEER_Id pid;
 191
 192   /**
 193    * Size of the linked list of 'pending_messages'.
 194    */
 195   unsigned int pending_requests;
 196
 197   /**
 198    * Which offset in "last_p2p_replies" will be updated next?
 199    * (we go round-robin).
 200    */
 201   unsigned int last_p2p_replies_woff;
 202
 203   /**
 204    * Which offset in "last_client_replies" will be updated next?
 205    * (we go round-robin).
 206    */
 207   unsigned int last_client_replies_woff;
 208
 209 };
 210
 211
 212 /**
 213  * Information we keep for each pending request.  We should try to
 214  * keep this struct as small as possible since its memory consumption
 215  * is key to how many requests we can have pending at once.
 216  */
 217 struct PendingRequest;
 218
 219
 220 /**
 221  * Doubly-linked list of requests we are performing
 222  * on behalf of the same client.
 223  */
 224 struct ClientRequestList
 225 {
 226
 227   /**
 228    * This is a doubly-linked list.
 229    */
 230   struct ClientRequestList *next;
 231
 232   /**
 233    * This is a doubly-linked list.
 234    */
 235   struct ClientRequestList *prev;
 236
 237   /**
 238    * Request this entry represents.
 239    */
 240   struct PendingRequest *req;
 241
 242   /**
 243    * Client list this request belongs to.
 244    */
 245   struct ClientList *client_list;
 246
 247 };
 248
 249
 250 /**
 251  * Replies to be transmitted to the client.  The actual
 252  * response message is allocated after this struct.
 253  */
 254 struct ClientResponseMessage
 255 {
 256   /**
 257    * This is a doubly-linked list.
 258    */
 259   struct ClientResponseMessage *next;
 260
 261   /**
 262    * This is a doubly-linked list.
 263    */
 264   struct ClientResponseMessage *prev;
 265
 266   /**
 267    * Client list entry this response belongs to.
 268    */
 269   struct ClientList *client_list;
 270
 271   /**
 272    * Number of bytes in the response.
 273    */
 274   size_t msize;
 275 };
 276
 277
 278 /**
 279  * Linked list of clients we are performing requests
 280  * for right now.
 281  */
 282 struct ClientList
 283 {
 284   /**
 285    * This is a linked list.
 286    */
 287   struct ClientList *next;
 288
 289   /**
 290    * ID of a client making a request, NULL if this entry is for a
 291    * peer.
 292    */
 293   struct GNUNET_SERVER_Client *client;
 294
 295   /**
 296    * Head of list of requests performed on behalf
 297    * of this client right now.
 298    */
 299   struct ClientRequestList *rl_head;
 300
 301   /**
 302    * Tail of list of requests performed on behalf
 303    * of this client right now.
 304    */
 305   struct ClientRequestList *rl_tail;
 306
 307   /**
 308    * Head of linked list of responses.
 309    */
 310   struct ClientResponseMessage *res_head;
 311
 312   /**
 313    * Tail of linked list of responses.
 314    */
 315   struct ClientResponseMessage *res_tail;
 316
 317   /**
 318    * Context for sending replies.
 319    */
 320   struct GNUNET_CONNECTION_TransmitHandle *th;
 321
 322 };
 323
 324
 325 /**
 326  * Doubly-linked list of messages we are performing
 327  * due to a pending request.
 328  */
 329 struct PendingMessageList
 330 {
 331
 332   /**
 333    * This is a doubly-linked list of messages on behalf of the same request.
 334    */
 335   struct PendingMessageList *next;
 336
 337   /**
 338    * This is a doubly-linked list of messages on behalf of the same request.
 339    */
 340   struct PendingMessageList *prev;
 341
 342   /**
 343    * Message this entry represents.
 344    */
 345   struct PendingMessage *pm;
 346
 347   /**
 348    * Request this entry belongs to.
 349    */
 350   struct PendingRequest *req;
 351
 352   /**
 353    * Peer this message is targeted for.
 354    */
 355   struct ConnectedPeer *target;
 356
 357 };
 358
 359
 360 /**
 361  * Information we keep for each pending request.  We should try to
 362  * keep this struct as small as possible since its memory consumption
 363  * is key to how many requests we can have pending at once.
 364  */
 365 struct PendingRequest
 366 {
 367
 368   /**
 369    * If this request was made by a client, this is our entry in the
 370    * client request list; otherwise NULL.
 371    */
 372   struct ClientRequestList *client_request_list;
 373
 374   /**
 375    * Entry of peer responsible for this entry (if this request
 376    * was made by a peer).
 377    */
 378   struct ConnectedPeer *cp;
 379
 380   /**
 381    * If this is a namespace query, pointer to the hash of the public
 382    * key of the namespace; otherwise NULL.  Pointer will be to the
 383    * end of this struct (so no need to free it).
 384    */
 385   const GNUNET_HashCode *namespace;
 386
 387   /**
 388    * Bloomfilter we use to filter out replies that we don't care about
 389    * (anymore).  NULL as long as we are interested in all replies.
 390    */
 391   struct GNUNET_CONTAINER_BloomFilter *bf;
 392
 393   /**
 394    * Context of our GNUNET_CORE_peer_change_preference call.
 395    */
 396   struct GNUNET_CORE_InformationRequestContext *irc;
 397
 398   /**
 399    * Hash code of all replies that we have seen so far (only valid
 400    * if client is not NULL since we only track replies like this for
 401    * our own clients).
 402    */
 403   GNUNET_HashCode *replies_seen;
 404
 405   /**
 406    * Node in the heap representing this entry; NULL
 407    * if we have no heap node.
 408    */
 409   struct GNUNET_CONTAINER_HeapNode *hnode;
 410
 411   /**
 412    * Head of list of messages being performed on behalf of this
 413    * request.
 414    */
 415   struct PendingMessageList *pending_head;
 416
 417   /**
 418    * Tail of list of messages being performed on behalf of this
 419    * request.
 420    */
 421   struct PendingMessageList *pending_tail;
 422
 423   /**
 424    * When did we first see this request (form this peer), or, if our
 425    * client is initiating, when did we last initiate a search?
 426    */
 427   struct GNUNET_TIME_Absolute start_time;
 428
 429   /**
 430    * The query that this request is for.
 431    */
 432   GNUNET_HashCode query;
 433
 434   /**
 435    * The task responsible for transmitting queries
 436    * for this request.
 437    */
 438   GNUNET_SCHEDULER_TaskIdentifier task;
 439
 440   /**
 441    * (Interned) Peer identifier that identifies a preferred target
 442    * for requests.
 443    */
 444   GNUNET_PEER_Id target_pid;
 445
 446   /**
 447    * (Interned) Peer identifiers of peers that have already
 448    * received our query for this content.
 449    */
 450   GNUNET_PEER_Id *used_pids;
 451
 452   /**
 453    * Our entry in the DRQ (non-NULL while we wait for our
 454    * turn to interact with the local database).
 455    */
 456   struct DatastoreRequestQueue *drq;
 457
 458   /**
 459    * Size of the 'bf' (in bytes).
 460    */
 461   size_t bf_size;
 462
 463   /**
 464    * Desired anonymity level; only valid for requests from a local client.
 465    */
 466   uint32_t anonymity_level;
 467
 468   /**
 469    * How many entries in "used_pids" are actually valid?
 470    */
 471   unsigned int used_pids_off;
 472
 473   /**
 474    * How long is the "used_pids" array?
 475    */
 476   unsigned int used_pids_size;
 477
 478   /**
 479    * Number of results found for this request.
 480    */
 481   unsigned int results_found;
 482
 483   /**
 484    * How many entries in "replies_seen" are actually valid?
 485    */
 486   unsigned int replies_seen_off;
 487
 488   /**
 489    * How long is the "replies_seen" array?
 490    */
 491   unsigned int replies_seen_size;
 492
 493   /**
 494    * Priority with which this request was made.  If one of our clients
 495    * made the request, then this is the current priority that we are
 496    * using when initiating the request.  This value is used when
 497    * we decide to reward other peers with trust for providing a reply.
 498    */
 499   uint32_t priority;
 500
 501   /**
 502    * Priority points left for us to spend when forwarding this request
 503    * to other peers.
 504    */
 505   uint32_t remaining_priority;
 506
 507   /**
 508    * Number to mingle hashes for bloom-filter tests with.
 509    */
 510   int32_t mingle;
 511
 512   /**
 513    * TTL with which we saw this request (or, if we initiated, TTL that
 514    * we used for the request).
 515    */
 516   int32_t ttl;
 517
 518   /**
 519    * Type of the content that this request is for.
 520    */
 521   uint32_t type;
 522
 523 };
 524
 525
 526 /**
 527  * Our scheduler.
 528  */
 529 static struct GNUNET_SCHEDULER_Handle *sched;
 530
 531 /**
 532  * Our configuration.
 533  */
 534 static const struct GNUNET_CONFIGURATION_Handle *cfg;
 535
 536 /**
 537  * Map of peer identifiers to "struct ConnectedPeer" (for that peer).
 538  */
 539 static struct GNUNET_CONTAINER_MultiHashMap *connected_peers;
 540
 541 /**
 542  * Map of peer identifiers to "struct PendingRequest" (for that peer).
 543  */
 544 static struct GNUNET_CONTAINER_MultiHashMap *peer_request_map;
 545
 546 /**
 547  * Map of query identifiers to "struct PendingRequest" (for that query).
 548  */
 549 static struct GNUNET_CONTAINER_MultiHashMap *query_request_map;
 550
 551 /**
 552  * Heap with the request that will expire next at the top.  Contains
 553  * pointers of type "struct PendingRequest*"; these will *also* be
 554  * aliased from the "requests_by_peer" data structures and the
 555  * "requests_by_query" table.  Note that requests from our clients
 556  * don't expire and are thus NOT in the "requests_by_expiration"
 557  * (or the "requests_by_peer" tables).
 558  */
 559 static struct GNUNET_CONTAINER_Heap *requests_by_expiration_heap;
 560
 561 /**
 562  * Linked list of clients we are currently processing requests for.
 563  */
 564 struct ClientList *client_list;
 565
 566 /**
 567  * Pointer to handle to the core service (points to NULL until we've
 568  * connected to it).
 569  */
 570 struct GNUNET_CORE_Handle *core;
 571
 572
 573 /* ******************* clean up functions ************************ */
 574
 575
 576 /**
 577  * We're done with a particular message list entry.
 578  * Free all associated resources.
 579  *
 580  * @param pml entry to destroy
 581  */
 582 static void
 583 destroy_pending_message_list_entry (struct PendingMessageList *pml)
 584 {
 585   GNUNET_CONTAINER_DLL_remove (pml->req->pending_head,
 586                                pml->req->pending_tail,
 587                                pml);
 588   GNUNET_CONTAINER_DLL_remove (pml->target->pending_messages_head,
 589                                pml->target->pending_messages_tail,
 590                                pml->pm);
 591   pml->target->pending_requests--;
 592   GNUNET_free (pml->pm);
 593   GNUNET_free (pml);
 594 }
 595
 596
 597 /**
 598  * Destroy the given pending message (and call the respective
 599  * continuation).
 600  *
 601  * @param pm message to destroy
 602  * @param tpid id of peer that the message was delivered to, or 0 for none
 603  */
 604 static void
 605 destroy_pending_message (struct PendingMessage *pm,
 606                          GNUNET_PEER_Id tpid)
 607 {
 608   struct PendingMessageList *pml = pm->pml;
 609   TransmissionContinuation cont;
 610   void *cont_cls;
 611
 612   GNUNET_assert (pml->pm == pm);
 613   GNUNET_assert ( (tpid == 0) || (tpid == pml->target->pid) );
 614   cont = pm->cont;
 615   cont_cls = pm->cont_cls;
 616   destroy_pending_message_list_entry (pml);
 617   cont (cont_cls, 0);
 618 }
 619
 620
 621 /**
 622  * We're done processing a particular request.
 623  * Free all associated resources.
 624  *
 625  * @param pr request to destroy
 626  */
 627 static void
 628 destroy_pending_request (struct PendingRequest *pr)
 629 {
 630   struct GNUNET_PeerIdentity pid;
 631
 632   if (pr->hnode != NULL)
 633     {
 634       GNUNET_CONTAINER_heap_remove_node (requests_by_expiration_heap,
 635                                          pr->hnode);
 636       pr->hnode = NULL;
 637     }
 638   /* might have already been removed from map in 'process_reply' (if
 639      there was a unique reply) or never inserted if it was a
 640      duplicate; hence ignore the return value here */
 641   (void) GNUNET_CONTAINER_multihashmap_remove (query_request_map,
 642                                                &pr->query,
 643                                                pr);
 644   if (pr->drq != NULL)
 645     {
 646       GNUNET_FS_drq_get_cancel (pr->drq);
 647       pr->drq = NULL;
 648     }
 649   if (pr->client_request_list != NULL)
 650     {
 651       GNUNET_CONTAINER_DLL_remove (pr->client_request_list->client_list->rl_head,
 652                                    pr->client_request_list->client_list->rl_tail,
 653                                    pr->client_request_list);
 654       GNUNET_free (pr->client_request_list);
 655       pr->client_request_list = NULL;
 656     }
 657   if (pr->cp != NULL)
 658     {
 659       GNUNET_PEER_resolve (pr->cp->pid,
 660                            &pid);
 661       (void) GNUNET_CONTAINER_multihashmap_remove (peer_request_map,
 662                                                    &pid.hashPubKey,
 663                                                    pr);
 664       pr->cp = NULL;
 665     }
 666   if (pr->bf != NULL)
 667     {
 668       GNUNET_CONTAINER_bloomfilter_free (pr->bf);
 669       pr->bf = NULL;
 670     }
 671   if (pr->irc != NULL)
 672     {
 673       GNUNET_CORE_peer_change_preference_cancel (pr->irc);
 674       pr->irc = NULL;
 675     }
 676   if (pr->replies_seen != NULL)
 677     {
 678       GNUNET_free (pr->replies_seen);
 679       pr->replies_seen = NULL;
 680     }
 681   if (pr->task != GNUNET_SCHEDULER_NO_TASK)
 682     {
 683       GNUNET_SCHEDULER_cancel (sched,
 684                                pr->task);
 685       pr->task = GNUNET_SCHEDULER_NO_TASK;
 686     }
 687   while (NULL != pr->pending_head)
 688     destroy_pending_message_list_entry (pr->pending_head);
 689   GNUNET_PEER_change_rc (pr->target_pid, -1);
 690   if (pr->used_pids != NULL)
 691     {
 692       GNUNET_PEER_decrement_rcs (pr->used_pids, pr->used_pids_off);
 693       GNUNET_free (pr->used_pids);
 694       pr->used_pids_off = 0;
 695       pr->used_pids_size = 0;
 696       pr->used_pids = NULL;
 697     }
 698   GNUNET_free (pr);
 699 }
 700
 701
 702 /**
 703  * Method called whenever a given peer connects.
 704  *
 705  * @param cls closure, not used
 706  * @param peer peer identity this notification is about
 707  * @param latency reported latency of the connection with 'other'
 708  * @param distance reported distance (DV) to 'other'
 709  */
 710 static void
 711 peer_connect_handler (void *cls,
 712                       const struct
 713                       GNUNET_PeerIdentity * peer,
 714                       struct GNUNET_TIME_Relative latency,
 715                       uint32_t distance)
 716 {
 717   struct ConnectedPeer *cp;
 718
 719   cp = GNUNET_malloc (sizeof (struct ConnectedPeer));
 720   cp->pid = GNUNET_PEER_intern (peer);
 721   GNUNET_CONTAINER_multihashmap_put (connected_peers,
 722                                      &peer->hashPubKey,
 723                                      cp,
 724                                      GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY);
 725 }
 726
 727
 728 /**
 729  * Free (each) request made by the peer.
 730  *
 731  * @param cls closure, points to peer that the request belongs to
 732  * @param key current key code
 733  * @param value value in the hash map
 734  * @return GNUNET_YES (we should continue to iterate)
 735  */
 736 static int
 737 destroy_request (void *cls,
 738                  const GNUNET_HashCode * key,
 739                  void *value)
 740 {
 741   const struct GNUNET_PeerIdentity * peer = cls;
 742   struct PendingRequest *pr = value;
 743
 744   GNUNET_CONTAINER_multihashmap_remove (peer_request_map,
 745                                         &peer->hashPubKey,
 746                                         pr);
 747   destroy_pending_request (pr);
 748   return GNUNET_YES;
 749 }
 750
 751
 752 /**
 753  * Method called whenever a peer disconnects.
 754  *
 755  * @param cls closure, not used
 756  * @param peer peer identity this notification is about
 757  */
 758 static void
 759 peer_disconnect_handler (void *cls,
 760                          const struct
 761                          GNUNET_PeerIdentity * peer)
 762 {
 763   struct ConnectedPeer *cp;
 764   struct PendingMessage *pm;
 765   unsigned int i;
 766
 767   GNUNET_CONTAINER_multihashmap_get_multiple (peer_request_map,
 768                                               &peer->hashPubKey,
 769                                               &destroy_request,
 770                                               (void*) peer);
 771   cp = GNUNET_CONTAINER_multihashmap_get (connected_peers,
 772                                           &peer->hashPubKey);
 773   if (cp == NULL)
 774     return;
 775   for (i=0;i<CS2P_SUCCESS_LIST_SIZE;i++)
 776     {
 777       if (NULL != cp->last_client_replies[i])
 778         {
 779           GNUNET_SERVER_client_drop (cp->last_client_replies[i]);
 780           cp->last_client_replies[i] = NULL;
 781         }
 782     }
 783   GNUNET_CONTAINER_multihashmap_remove (connected_peers,
 784                                         &peer->hashPubKey,
 785                                         cp);
 786   GNUNET_PEER_change_rc (cp->pid, -1);
 787   GNUNET_PEER_decrement_rcs (cp->last_p2p_replies, P2P_SUCCESS_LIST_SIZE);
 788   if (NULL != cp->cth)
 789     GNUNET_CORE_notify_transmit_ready_cancel (cp->cth);
 790   while (NULL != (pm = cp->pending_messages_head))
 791     destroy_pending_message (pm, 0 /* delivery failed */);
 792   GNUNET_break (0 == cp->pending_requests);
 793   GNUNET_free (cp);
 794 }
 795
 796
 797 /**
 798  * Iterator over hash map entries that removes all occurences
 799  * of the given 'client' from the 'last_client_replies' of the
 800  * given connected peer.
 801  *
 802  * @param cls closure, the 'struct GNUNET_SERVER_Client*' to remove
 803  * @param key current key code (unused)
 804  * @param value value in the hash map (the 'struct ConnectedPeer*' to change)
 805  * @return GNUNET_YES (we should continue to iterate)
 806  */
 807 static int
 808 remove_client_from_last_client_replies (void *cls,
 809                                         const GNUNET_HashCode * key,
 810                                         void *value)
 811 {
 812   struct GNUNET_SERVER_Client *client = cls;
 813   struct ConnectedPeer *cp = value;
 814   unsigned int i;
 815
 816   for (i=0;i<CS2P_SUCCESS_LIST_SIZE;i++)
 817     {
 818       if (cp->last_client_replies[i] == client)
 819         {
 820           GNUNET_SERVER_client_drop (cp->last_client_replies[i]);
 821           cp->last_client_replies[i] = NULL;
 822         }
 823     }
 824   return GNUNET_YES;
 825 }
 826
 827
 828 /**
 829  * A client disconnected.  Remove all of its pending queries.
 830  *
 831  * @param cls closure, NULL
 832  * @param client identification of the client
 833  */
 834 static void
 835 handle_client_disconnect (void *cls,
 836                           struct GNUNET_SERVER_Client
 837                           * client)
 838 {
 839   struct ClientList *pos;
 840   struct ClientList *prev;
 841   struct ClientRequestList *rcl;
 842   struct ClientResponseMessage *creply;
 843
 844   if (client == NULL)
 845     return;
 846   prev = NULL;
 847   pos = client_list;
 848   while ( (NULL != pos) &&
 849           (pos->client != client) )
 850     {
 851       prev = pos;
 852       pos = pos->next;
 853     }
 854   if (pos == NULL)
 855     return; /* no requests pending for this client */
 856   while (NULL != (rcl = pos->rl_head))
 857     destroy_pending_request (rcl->req);
 858   if (prev == NULL)
 859     client_list = pos->next;
 860   else
 861     prev->next = pos->next;
 862   if (pos->th != NULL)
 863     {
 864       GNUNET_CONNECTION_notify_transmit_ready_cancel (pos->th);
 865       pos->th = NULL;
 866     }
 867   while (NULL != (creply = pos->res_head))
 868     {
 869       GNUNET_CONTAINER_DLL_remove (pos->res_head,
 870                                    pos->res_tail,
 871                                    creply);
 872       GNUNET_free (creply);
 873     }
 874   GNUNET_SERVER_client_drop (pos->client);
 875   GNUNET_free (pos);
 876   GNUNET_CONTAINER_multihashmap_iterate (connected_peers,
 877                                          &remove_client_from_last_client_replies,
 878                                          client);
 879 }
 880
 881
 882 /**
 883  * Iterator to free peer entries.
 884  *
 885  * @param cls closure, unused
 886  * @param key current key code
 887  * @param value value in the hash map (peer entry)
 888  * @return GNUNET_YES (we should continue to iterate)
 889  */
 890 static int
 891 clean_peer (void *cls,
 892             const GNUNET_HashCode * key,
 893             void *value)
 894 {
 895   peer_disconnect_handler (NULL, (const struct GNUNET_PeerIdentity*) key);
 896   return GNUNET_YES;
 897 }
 898
 899
 900 /**
 901  * Task run during shutdown.
 902  *
 903  * @param cls unused
 904  * @param tc unused
 905  */
 906 static void
 907 shutdown_task (void *cls,
 908                const struct GNUNET_SCHEDULER_TaskContext *tc)
 909 {
 910   while (client_list != NULL)
 911     handle_client_disconnect (NULL,
 912                               client_list->client);
 913   GNUNET_CONTAINER_multihashmap_iterate (connected_peers,
 914                                          &clean_peer,
 915                                          NULL);
 916   GNUNET_break (0 == GNUNET_CONTAINER_heap_get_size (requests_by_expiration_heap));
 917   GNUNET_CONTAINER_heap_destroy (requests_by_expiration_heap);
 918   requests_by_expiration_heap = 0;
 919   GNUNET_CONTAINER_multihashmap_destroy (connected_peers);
 920   connected_peers = NULL;
 921   GNUNET_break (0 == GNUNET_CONTAINER_multihashmap_size (query_request_map));
 922   GNUNET_CONTAINER_multihashmap_destroy (query_request_map);
 923   query_request_map = NULL;
 924   GNUNET_break (0 == GNUNET_CONTAINER_multihashmap_size (peer_request_map));
 925   GNUNET_CONTAINER_multihashmap_destroy (peer_request_map);
 926   peer_request_map = NULL;
 927   GNUNET_assert (NULL != core);
 928   GNUNET_CORE_disconnect (core);
 929   core = NULL;
 930   sched = NULL;
 931   cfg = NULL;
 932 }
 933
 934
 935 /* ******************* Utility functions  ******************** */
 936
 937
 938 /**
 939  * Transmit the given message by copying it to the target buffer
 940  * "buf".  "buf" will be NULL and "size" zero if the socket was closed
 941  * for writing in the meantime.  In that case, do nothing
 942  * (the disconnect or shutdown handler will take care of the rest).
 943  * If we were able to transmit messages and there are still more
 944  * pending, ask core again for further calls to this function.
 945  *
 946  * @param cls closure, pointer to the 'struct ConnectedPeer*'
 947  * @param size number of bytes available in buf
 948  * @param buf where the callee should write the message
 949  * @return number of bytes written to buf
 950  */
 951 static size_t
 952 transmit_to_peer (void *cls,
 953                   size_t size, void *buf)
 954 {
 955   struct ConnectedPeer *cp = cls;
 956   char *cbuf = buf;
 957   struct GNUNET_PeerIdentity pid;
 958   struct PendingMessage *pm;
 959   size_t msize;
 960
 961   cp->cth = NULL;
 962   if (NULL == buf)
 963     {
 964 #if DEBUG_FS
 965       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 966                   "Dropping message, core too busy.\n");
 967 #endif
 968       return 0;
 969     }
 970   msize = 0;
 971   while ( (NULL != (pm = cp->pending_messages_head) ) &&
 972           (pm->msize <= size) )
 973     {
 974       memcpy (&cbuf[msize], &pm[1], pm->msize);
 975       msize += pm->msize;
 976       size -= pm->msize;
 977       destroy_pending_message (pm, cp->pid);
 978     }
 979   if (NULL != pm)
 980     {
 981       GNUNET_PEER_resolve (cp->pid,
 982                            &pid);
 983       cp->cth = GNUNET_CORE_notify_transmit_ready (core,
 984                                                    pm->priority,
 985                                                    GNUNET_CONSTANTS_SERVICE_TIMEOUT,
 986                                                    &pid,
 987                                                    pm->msize,
 988                                                    &transmit_to_peer,
 989                                                    pm);
 990     }
 991 #if DEBUG_FS > 2
 992   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 993               "Transmitting %u bytes to peer %u.\n",
 994               msize,
 995               cp->pid);
 996 #endif
 997   return msize;
 998 }
 999
1000
1001 /**
1002  * Add a message to the set of pending messages for the given peer.
1003  *
1004  * @param cp peer to send message to
1005  * @param pm message to queue
1006  * @param pr request on which behalf this message is being queued
1007  */
1008 static void
1009 add_to_pending_messages_for_peer (struct ConnectedPeer *cp,
1010                                   struct PendingMessage *pm,
1011                                   struct PendingRequest *pr)
1012 {
1013   struct PendingMessage *pos;
1014   struct PendingMessageList *pml;
1015   struct GNUNET_PeerIdentity pid;
1016
1017   GNUNET_assert (pm->next == NULL);
1018   GNUNET_assert (pm->pml == NULL);
1019   pml = GNUNET_malloc (sizeof (struct PendingMessageList));
1020   pml->req = pr;
1021   pml->target = cp;
1022   pml->pm = pm;
1023   pm->pml = pml;
1024   GNUNET_CONTAINER_DLL_insert (pr->pending_head,
1025                                pr->pending_tail,
1026                                pml);
1027   pos = cp->pending_messages_head;
1028   while ( (pos != NULL) &&
1029           (pm->priority < pos->priority) )
1030     pos = pos->next;
1031   GNUNET_CONTAINER_DLL_insert_after (cp->pending_messages_head,
1032                                      cp->pending_messages_tail,
1033                                      pos,
1034                                      pm);
1035   cp->pending_requests++;
1036   if (cp->pending_requests > MAX_QUEUE_PER_PEER)
1037     destroy_pending_message (cp->pending_messages_tail, 0);
1038   if (cp->cth == NULL)
1039     {
1040       /* need to schedule transmission */
1041       GNUNET_PEER_resolve (cp->pid, &pid);
1042       cp->cth = GNUNET_CORE_notify_transmit_ready (core,
1043                                                    cp->pending_messages_head->priority,
1044                                                    GNUNET_TIME_UNIT_FOREVER_REL,
1045                                                    &pid,
1046                                                    cp->pending_messages_head->msize,
1047                                                    &transmit_to_peer,
1048                                                    cp);
1049     }
1050   if (cp->cth == NULL)
1051     {
1052 #if DEBUG_FS
1053       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1054                   "Failed to schedule transmission with core!\n");
1055 #endif
1056       /* FIXME: call stats (rare, bad case) */
1057     }
1058 }
1059
1060
1061 /**
1062  * Mingle hash with the mingle_number to produce different bits.
1063  */
1064 static void
1065 mingle_hash (const GNUNET_HashCode * in,
1066              int32_t mingle_number,
1067              GNUNET_HashCode * hc)
1068 {
1069   GNUNET_HashCode m;
1070
1071   GNUNET_CRYPTO_hash (&mingle_number,
1072                       sizeof (int32_t),
1073                       &m);
1074   GNUNET_CRYPTO_hash_xor (&m, in, hc);
1075 }
1076
1077
1078 /**
1079  * Test if the load on this peer is too high
1080  * to even consider processing the query at
1081  * all.
1082  *
1083  * @return GNUNET_YES if the load is too high, GNUNET_NO otherwise
1084  */
1085 static int
1086 test_load_too_high ()
1087 {
1088   return GNUNET_NO; // FIXME
1089 }
1090
1091
1092 /* ******************* Pending Request Refresh Task ******************** */
1093
1094
1095 /**
1096  * Function called after we either failed or succeeded
1097  * at transmitting a query to a peer.
1098  *
1099  * @param cls the requests "struct PendingRequest*"
1100  * @param tpid ID of receiving peer, 0 on transmission error
1101  */
1102 static void
1103 transmit_query_continuation (void *cls,
1104                              GNUNET_PEER_Id tpid)
1105 {
1106   struct PendingRequest *pr = cls;
1107
1108   if (tpid == 0)
1109     return;
1110   GNUNET_PEER_change_rc (tpid, 1);
1111   if (pr->used_pids_off == pr->used_pids_size)
1112     GNUNET_array_grow (pr->used_pids,
1113                        pr->used_pids_size,
1114                        pr->used_pids_size * 2 + 2);
1115   pr->used_pids[pr->used_pids_off++] = tpid;
1116 }
1117
1118
1119 /**
1120  * How many bytes should a bloomfilter be if we have already seen
1121  * entry_count responses?  Note that BLOOMFILTER_K gives us the number
1122  * of bits set per entry.  Furthermore, we should not re-size the
1123  * filter too often (to keep it cheap).
1124  *
1125  * Since other peers will also add entries but not resize the filter,
1126  * we should generally pick a slightly larger size than what the
1127  * strict math would suggest.
1128  *
1129  * @return must be a power of two and smaller or equal to 2^15.
1130  */
1131 static size_t
1132 compute_bloomfilter_size (unsigned int entry_count)
1133 {
1134   size_t size;
1135   unsigned int ideal = (entry_count * BLOOMFILTER_K) / 4;
1136   uint16_t max = 1 << 15;
1137
1138   if (entry_count > max)
1139     return max;
1140   size = 8;
1141   while ((size < max) && (size < ideal))
1142     size *= 2;
1143   if (size > max)
1144     return max;
1145   return size;
1146 }
1147
1148
1149 /**
1150  * Recalculate our bloom filter for filtering replies.
1151  *
1152  * @param count number of entries we are filtering right now
1153  * @param mingle set to our new mingling value
1154  * @param bf_size set to the size of the bloomfilter
1155  * @param entries the entries to filter
1156  * @return updated bloomfilter, NULL for none
1157  */
1158 static struct GNUNET_CONTAINER_BloomFilter *
1159 refresh_bloomfilter (unsigned int count,
1160                      int32_t * mingle,
1161                      size_t *bf_size,
1162                      const GNUNET_HashCode *entries)
1163 {
1164   struct GNUNET_CONTAINER_BloomFilter *bf;
1165   size_t nsize;
1166   unsigned int i;
1167   GNUNET_HashCode mhash;
1168
1169   if (0 == count)
1170     return NULL;
1171   nsize = compute_bloomfilter_size (count);
1172   *mingle = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, -1);
1173   *bf_size = nsize;
1174   bf = GNUNET_CONTAINER_bloomfilter_init (NULL,
1175                                           nsize,
1176                                           BLOOMFILTER_K);
1177   for (i=0;i<count;i++)
1178     {
1179       mingle_hash (&entries[i], *mingle, &mhash);
1180       GNUNET_CONTAINER_bloomfilter_add (bf, &mhash);
1181     }
1182   return bf;
1183 }
1184
1185
1186 /**
1187  * We use a random delay to make the timing of requests less
1188  * predictable.  This function returns such a random delay.
1189  *
1190  * FIXME: make schedule dependent on the specifics of the request?
1191  * Or bandwidth and number of connected peers and load?
1192  *
1193  * @return random delay to use for some request, between 0 and TTL_DECREMENT ms
1194  */
1195 static struct GNUNET_TIME_Relative
1196 get_processing_delay ()
1197 {
1198   return GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MILLISECONDS,
1199                                         GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
1200                                                                   TTL_DECREMENT));
1201 }
1202
1203
1204 /**
1205  * Function called after we've tried to reserve a certain amount of
1206  * bandwidth for a reply.  Check if we succeeded and if so send our
1207  * query.
1208  *
1209  * @param cls the requests "struct PendingRequest*"
1210  * @param peer identifies the peer
1211  * @param bpm_in set to the current bandwidth limit (receiving) for this peer
1212  * @param bpm_out set to the current bandwidth limit (sending) for this peer
1213  * @param amount set to the amount that was actually reserved or unreserved
1214  * @param preference current traffic preference for the given peer
1215  */
1216 static void
1217 target_reservation_cb (void *cls,
1218                        const struct
1219                        GNUNET_PeerIdentity * peer,
1220                        unsigned int bpm_in,
1221                        unsigned int bpm_out,
1222                        int amount,
1223                        uint64_t preference)
1224 {
1225   struct PendingRequest *pr = cls;
1226   struct ConnectedPeer *cp;
1227   struct PendingMessage *pm;
1228   struct GetMessage *gm;
1229   GNUNET_HashCode *ext;
1230   char *bfdata;
1231   size_t msize;
1232   unsigned int k;
1233   int no_route;
1234   uint32_t bm;
1235
1236   pr->irc = NULL;
1237   GNUNET_assert (peer != NULL);
1238   // (3) transmit, update ttl/priority
1239   cp = GNUNET_CONTAINER_multihashmap_get (connected_peers,
1240                                           &peer->hashPubKey);
1241   if (cp == NULL)
1242     {
1243       /* Peer must have just left */
1244 #if DEBUG_FS
1245       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1246                   "Selected peer disconnected!\n");
1247 #endif
1248       return;
1249     }
1250   no_route = GNUNET_NO;
1251   /* FIXME: check against DBLOCK_SIZE and possibly return
1252      amount to reserve; however, this also needs to work
1253      with testcases which currently start out with a far
1254      too low per-peer bw limit, so they would never send
1255      anything.  Big issue. */
1256   if (amount == 0)
1257     {
1258       if (pr->cp == NULL)
1259         {
1260 #if DEBUG_FS > 1
1261           GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1262                       "Failed to reserve bandwidth for reply (got %d/%u bytes only)!\n",
1263                       amount,
1264                       DBLOCK_SIZE);
1265 #endif
1266           return;  /* this target round failed */
1267         }
1268       /* FIXME: if we are "quite" busy, we may still want to skip
1269          this round; need more load detection code! */
1270       no_route = GNUNET_YES;
1271     }
1272
1273   /* build message and insert message into priority queue */
1274 #if DEBUG_FS
1275   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1276               "Forwarding request `%s' to `%4s'!\n",
1277               GNUNET_h2s (&pr->query),
1278               GNUNET_i2s (peer));
1279 #endif
1280   k = 0;
1281   bm = 0;
1282   if (GNUNET_YES == no_route)
1283     {
1284       bm |= GET_MESSAGE_BIT_RETURN_TO;
1285       k++;
1286     }
1287   if (pr->namespace != NULL)
1288     {
1289       bm |= GET_MESSAGE_BIT_SKS_NAMESPACE;
1290       k++;
1291     }
1292   if (pr->target_pid != 0)
1293     {
1294       bm |= GET_MESSAGE_BIT_TRANSMIT_TO;
1295       k++;
1296     }
1297   msize = sizeof (struct GetMessage) + pr->bf_size + k * sizeof(GNUNET_HashCode);
1298   GNUNET_assert (msize < GNUNET_SERVER_MAX_MESSAGE_SIZE);
1299   pm = GNUNET_malloc (sizeof (struct PendingMessage) + msize);
1300   pm->msize = msize;
1301   gm = (struct GetMessage*) &pm[1];
1302   gm->header.type = htons (GNUNET_MESSAGE_TYPE_FS_GET);
1303   gm->header.size = htons (msize);
1304   gm->type = htonl (pr->type);
1305   pr->remaining_priority /= 2;
1306   gm->priority = htonl (pr->remaining_priority);
1307   gm->ttl = htonl (pr->ttl);
1308   gm->filter_mutator = htonl(pr->mingle);
1309   gm->hash_bitmap = htonl (bm);
1310   gm->query = pr->query;
1311   ext = (GNUNET_HashCode*) &gm[1];
1312   k = 0;
1313   if (GNUNET_YES == no_route)
1314     GNUNET_PEER_resolve (pr->cp->pid, (struct GNUNET_PeerIdentity*) &ext[k++]);
1315   if (pr->namespace != NULL)
1316     memcpy (&ext[k++], pr->namespace, sizeof (GNUNET_HashCode));
1317   if (pr->target_pid != 0)
1318     GNUNET_PEER_resolve (pr->target_pid, (struct GNUNET_PeerIdentity*) &ext[k++]);
1319   bfdata = (char *) &ext[k];
1320   if (pr->bf != NULL)
1321     GNUNET_CONTAINER_bloomfilter_get_raw_data (pr->bf,
1322                                                bfdata,
1323                                                pr->bf_size);
1324   pm->cont = &transmit_query_continuation;
1325   pm->cont_cls = pr;
1326   add_to_pending_messages_for_peer (cp, pm, pr);
1327 }
1328
1329
1330 /**
1331  * Closure used for "target_peer_select_cb".
1332  */
1333 struct PeerSelectionContext
1334 {
1335   /**
1336    * The request for which we are selecting
1337    * peers.
1338    */
1339   struct PendingRequest *pr;
1340
1341   /**
1342    * Current "prime" target.
1343    */
1344   struct GNUNET_PeerIdentity target;
1345
1346   /**
1347    * How much do we like this target?
1348    */
1349   double target_score;
1350
1351 };
1352
1353
1354 /**
1355  * Function called for each connected peer to determine
1356  * which one(s) would make good targets for forwarding.
1357  *
1358  * @param cls closure (struct PeerSelectionContext)
1359  * @param key current key code (peer identity)
1360  * @param value value in the hash map (struct ConnectedPeer)
1361  * @return GNUNET_YES if we should continue to
1362  *         iterate,
1363  *         GNUNET_NO if not.
1364  */
1365 static int
1366 target_peer_select_cb (void *cls,
1367                        const GNUNET_HashCode * key,
1368                        void *value)
1369 {
1370   struct PeerSelectionContext *psc = cls;
1371   struct ConnectedPeer *cp = value;
1372   struct PendingRequest *pr = psc->pr;
1373   double score;
1374   unsigned int i;
1375
1376   /* 1) check if we have already (recently) forwarded to this peer */
1377   for (i=0;i<pr->used_pids_off;i++)
1378     if (pr->used_pids[i] == cp->pid)
1379       return GNUNET_YES; /* skip */
1380   // 2) calculate how much we'd like to forward to this peer
1381   score = 42; // FIXME!
1382   // FIXME: also need API to gather data on responsiveness
1383   // of this peer (we have fields for that in 'cp', but
1384   // they are never set!)
1385
1386   /* store best-fit in closure */
1387   if (score > psc->target_score)
1388     {
1389       psc->target_score = score;
1390       psc->target.hashPubKey = *key;
1391     }
1392   return GNUNET_YES;
1393 }
1394
1395
1396 /**
1397  * We're processing a GET request from another peer and have decided
1398  * to forward it to other peers.  This function is called periodically
1399  * and should forward the request to other peers until we have all
1400  * possible replies.  If we have transmitted the *only* reply to
1401  * the initiator we should destroy the pending request.  If we have
1402  * many replies in the queue to the initiator, we should delay sending
1403  * out more queries until the reply queue has shrunk some.
1404  *
1405  * @param cls our "struct ProcessGetContext *"
1406  * @param tc unused
1407  */
1408 static void
1409 forward_request_task (void *cls,
1410                      const struct GNUNET_SCHEDULER_TaskContext *tc)
1411 {
1412   struct PendingRequest *pr = cls;
1413   struct PeerSelectionContext psc;
1414   struct ConnectedPeer *cp;
1415
1416   pr->task = GNUNET_SCHEDULER_add_delayed (sched,
1417                                            get_processing_delay (),
1418                                            &forward_request_task,
1419                                            pr);
1420   if (pr->irc != NULL)
1421     return; /* previous request still pending */
1422   /* (1) select target */
1423   psc.pr = pr;
1424   psc.target_score = DBL_MIN;
1425   GNUNET_CONTAINER_multihashmap_iterate (connected_peers,
1426                                          &target_peer_select_cb,
1427                                          &psc);
1428   if (psc.target_score == DBL_MIN)
1429     {
1430 #if DEBUG_FS
1431       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1432                   "No peer selected for forwarding!\n");
1433 #endif
1434       return; /* nobody selected */
1435     }
1436
1437   /* (2) reserve reply bandwidth */
1438   cp = GNUNET_CONTAINER_multihashmap_get (connected_peers,
1439                                           &psc.target.hashPubKey);
1440   pr->irc = GNUNET_CORE_peer_change_preference (sched, cfg,
1441                                                 &psc.target,
1442                                                 GNUNET_CONSTANTS_SERVICE_TIMEOUT,
1443                                                 (uint32_t) -1 /* no limit */,
1444                                                 DBLOCK_SIZE,
1445                                                 (uint64_t) cp->inc_preference,
1446                                                 &target_reservation_cb,
1447                                                 pr);
1448   cp->inc_preference = 0.0;
1449 }
1450
1451
1452 /* **************************** P2P PUT Handling ************************ */
1453
1454
1455 /**
1456  * Function called after we either failed or succeeded
1457  * at transmitting a reply to a peer.
1458  *
1459  * @param cls the requests "struct PendingRequest*"
1460  * @param tpid ID of receiving peer, 0 on transmission error
1461  */
1462 static void
1463 transmit_reply_continuation (void *cls,
1464                              GNUNET_PEER_Id tpid)
1465 {
1466   struct PendingRequest *pr = cls;
1467
1468   switch (pr->type)
1469     {
1470     case GNUNET_DATASTORE_BLOCKTYPE_DBLOCK:
1471     case GNUNET_DATASTORE_BLOCKTYPE_IBLOCK:
1472       /* only one reply expected, done with the request! */
1473       destroy_pending_request (pr);
1474       break;
1475     case GNUNET_DATASTORE_BLOCKTYPE_KBLOCK:
1476     case GNUNET_DATASTORE_BLOCKTYPE_SBLOCK:
1477       break;
1478     default:
1479       GNUNET_break (0);
1480       break;
1481     }
1482 }
1483
1484
1485 /**
1486  * Check if the given KBlock is well-formed.
1487  *
1488  * @param kb the kblock data (or at least "dsize" bytes claiming to be one)
1489  * @param dsize size of "kb" in bytes; check for < sizeof(struct KBlock)!
1490  * @param query where to store the query that this block answers
1491  * @return GNUNET_OK if this is actually a well-formed KBlock
1492  */
1493 static int
1494 check_kblock (const struct KBlock *kb,
1495               size_t dsize,
1496               GNUNET_HashCode *query)
1497 {
1498   if (dsize < sizeof (struct KBlock))
1499     {
1500       GNUNET_break_op (0);
1501       return GNUNET_SYSERR;
1502     }
1503   if (dsize - sizeof (struct KBlock) !=
1504       ntohs (kb->purpose.size)
1505       - sizeof (struct GNUNET_CRYPTO_RsaSignaturePurpose)
1506       - sizeof(struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded) )
1507     {
1508       GNUNET_break_op (0);
1509       return GNUNET_SYSERR;
1510     }
1511   if (GNUNET_OK !=
1512       GNUNET_CRYPTO_rsa_verify (GNUNET_SIGNATURE_PURPOSE_FS_KBLOCK,
1513                                 &kb->purpose,
1514                                 &kb->signature,
1515                                 &kb->keyspace))
1516     {
1517       GNUNET_break_op (0);
1518       return GNUNET_SYSERR;
1519     }
1520   if (query != NULL)
1521     GNUNET_CRYPTO_hash (&kb->keyspace,
1522                         sizeof (struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded),
1523                         query);
1524   return GNUNET_OK;
1525 }
1526
1527
1528 /**
1529  * Check if the given SBlock is well-formed.
1530  *
1531  * @param sb the sblock data (or at least "dsize" bytes claiming to be one)
1532  * @param dsize size of "kb" in bytes; check for < sizeof(struct SBlock)!
1533  * @param query where to store the query that this block answers
1534  * @param namespace where to store the namespace that this block belongs to
1535  * @return GNUNET_OK if this is actually a well-formed SBlock
1536  */
1537 static int
1538 check_sblock (const struct SBlock *sb,
1539               size_t dsize,
1540               GNUNET_HashCode *query,
1541               GNUNET_HashCode *namespace)
1542 {
1543   if (dsize < sizeof (struct SBlock))
1544     {
1545       GNUNET_break_op (0);
1546       return GNUNET_SYSERR;
1547     }
1548   if (dsize !=
1549       ntohs (sb->purpose.size) + sizeof (struct GNUNET_CRYPTO_RsaSignature))
1550     {
1551       GNUNET_break_op (0);
1552       return GNUNET_SYSERR;
1553     }
1554   if (GNUNET_OK !=
1555       GNUNET_CRYPTO_rsa_verify (GNUNET_SIGNATURE_PURPOSE_FS_SBLOCK,
1556                                 &sb->purpose,
1557                                 &sb->signature,
1558                                 &sb->subspace))
1559     {
1560       GNUNET_break_op (0);
1561       return GNUNET_SYSERR;
1562     }
1563   if (query != NULL)
1564     *query = sb->identifier;
1565   if (namespace != NULL)
1566     GNUNET_CRYPTO_hash (&sb->subspace,
1567                         sizeof (struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded),
1568                         namespace);
1569   return GNUNET_OK;
1570 }
1571
1572
1573 /**
1574  * Transmit the given message by copying it to the target buffer
1575  * "buf".  "buf" will be NULL and "size" zero if the socket was closed
1576  * for writing in the meantime.  In that case, do nothing
1577  * (the disconnect or shutdown handler will take care of the rest).
1578  * If we were able to transmit messages and there are still more
1579  * pending, ask core again for further calls to this function.
1580  *
1581  * @param cls closure, pointer to the 'struct ClientList*'
1582  * @param size number of bytes available in buf
1583  * @param buf where the callee should write the message
1584  * @return number of bytes written to buf
1585  */
1586 static size_t
1587 transmit_to_client (void *cls,
1588                   size_t size, void *buf)
1589 {
1590   struct ClientList *cl = cls;
1591   char *cbuf = buf;
1592   struct ClientResponseMessage *creply;
1593   size_t msize;
1594
1595   cl->th = NULL;
1596   if (NULL == buf)
1597     {
1598 #if DEBUG_FS
1599       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1600                   "Not sending reply, client communication problem.\n");
1601 #endif
1602       return 0;
1603     }
1604   msize = 0;
1605   while ( (NULL != (creply = cl->res_head) ) &&
1606           (creply->msize <= size) )
1607     {
1608       memcpy (&cbuf[msize], &creply[1], creply->msize);
1609       msize += creply->msize;
1610       size -= creply->msize;
1611       GNUNET_CONTAINER_DLL_remove (cl->res_head,
1612                                    cl->res_tail,
1613                                    creply);
1614       GNUNET_free (creply);
1615     }
1616   if (NULL != creply)
1617     cl->th = GNUNET_SERVER_notify_transmit_ready (cl->client,
1618                                                   creply->msize,
1619                                                   GNUNET_TIME_UNIT_FOREVER_REL,
1620                                                   &transmit_to_client,
1621                                                   cl);
1622 #if DEBUG_FS
1623   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1624               "Transmitted %u bytes to client\n",
1625               (unsigned int) msize);
1626 #endif
1627   return msize;
1628 }
1629
1630
1631 /**
1632  * Closure for "process_reply" function.
1633  */
1634 struct ProcessReplyClosure
1635 {
1636   /**
1637    * The data for the reply.
1638    */
1639   const void *data;
1640
1641   // FIXME: add 'struct ConnectedPeer' to track 'last_xxx_replies' here!
1642
1643   /**
1644    * When the reply expires.
1645    */
1646   struct GNUNET_TIME_Absolute expiration;
1647
1648   /**
1649    * Size of data.
1650    */
1651   size_t size;
1652
1653   /**
1654    * Namespace that this reply belongs to
1655    * (if it is of type SBLOCK).
1656    */
1657   GNUNET_HashCode namespace;
1658
1659   /**
1660    * Type of the block.
1661    */
1662   uint32_t type;
1663
1664   /**
1665    * How much was this reply worth to us?
1666    */
1667   uint32_t priority;
1668 };
1669
1670
1671 /**
1672  * We have received a reply; handle it!
1673  *
1674  * @param cls response (struct ProcessReplyClosure)
1675  * @param key our query
1676  * @param value value in the hash map (info about the query)
1677  * @return GNUNET_YES (we should continue to iterate)
1678  */
1679 static int
1680 process_reply (void *cls,
1681                const GNUNET_HashCode * key,
1682                void *value)
1683 {
1684   struct ProcessReplyClosure *prq = cls;
1685   struct PendingRequest *pr = value;
1686   struct PendingMessage *reply;
1687   struct ClientResponseMessage *creply;
1688   struct ClientList *cl;
1689   struct PutMessage *pm;
1690   struct ConnectedPeer *cp;
1691   GNUNET_HashCode chash;
1692   GNUNET_HashCode mhash;
1693   size_t msize;
1694   uint32_t prio;
1695   int do_remove;
1696
1697 #if DEBUG_FS
1698   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1699               "Matched result (type %u) for query `%s' with pending request\n",
1700               (unsigned int) prq->type,
1701               GNUNET_h2s (key));
1702 #endif
1703   do_remove = GNUNET_NO;
1704   GNUNET_CRYPTO_hash (prq->data,
1705                       prq->size,
1706                       &chash);
1707   switch (prq->type)
1708     {
1709     case GNUNET_DATASTORE_BLOCKTYPE_DBLOCK:
1710     case GNUNET_DATASTORE_BLOCKTYPE_IBLOCK:
1711       /* only possible reply, stop requesting! */
1712       while (NULL != pr->pending_head)
1713         destroy_pending_message_list_entry (pr->pending_head);
1714       if (pr->drq != NULL)
1715         {
1716           if (pr->client_request_list != NULL)
1717             GNUNET_SERVER_receive_done (pr->client_request_list->client_list->client,
1718                                         GNUNET_YES);
1719           GNUNET_FS_drq_get_cancel (pr->drq);
1720           pr->drq = NULL;
1721         }
1722       do_remove = GNUNET_YES;
1723       break;
1724     case GNUNET_DATASTORE_BLOCKTYPE_SBLOCK:
1725       if (0 != memcmp (pr->namespace,
1726                        &prq->namespace,
1727                        sizeof (GNUNET_HashCode)))
1728         return GNUNET_YES; /* wrong namespace */
1729       /* then: fall-through! */
1730     case GNUNET_DATASTORE_BLOCKTYPE_KBLOCK:
1731       if (pr->bf != NULL)
1732         {
1733           mingle_hash (&chash, pr->mingle, &mhash);
1734           if (GNUNET_YES == GNUNET_CONTAINER_bloomfilter_test (pr->bf,
1735                                                                &mhash))
1736             return GNUNET_YES; /* duplicate */
1737           GNUNET_CONTAINER_bloomfilter_add (pr->bf,
1738                                             &mhash);
1739         }
1740       if (pr->client_request_list != NULL)
1741         {
1742           if (pr->replies_seen_size == pr->replies_seen_off)
1743             {
1744               GNUNET_array_grow (pr->replies_seen,
1745                                  pr->replies_seen_size,
1746                                  pr->replies_seen_size * 2 + 4);
1747               if (pr->bf != NULL)
1748                 GNUNET_CONTAINER_bloomfilter_free (pr->bf);
1749               pr->bf = refresh_bloomfilter (pr->replies_seen_off,
1750                                             &pr->mingle,
1751                                             &pr->bf_size,
1752                                             pr->replies_seen);
1753             }
1754             pr->replies_seen[pr->replies_seen_off++] = chash;
1755
1756         }
1757       break;
1758     case GNUNET_DATASTORE_BLOCKTYPE_SKBLOCK:
1759       // FIXME: any checks against duplicates for SKBlocks?
1760       break;
1761     default:
1762       GNUNET_break (0);
1763       return GNUNET_YES;
1764     }
1765   prio = pr->priority;
1766   prq->priority += pr->remaining_priority;
1767   pr->remaining_priority = 0;
1768   if (pr->client_request_list != NULL)
1769     {
1770 #if DEBUG_FS
1771       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1772                   "Transmitting result for query `%s' to local client\n",
1773                   GNUNET_h2s (key));
1774 #endif
1775       cl = pr->client_request_list->client_list;
1776       msize = sizeof (struct PutMessage) + prq->size;
1777       creply = GNUNET_malloc (msize + sizeof (struct ClientResponseMessage));
1778       creply->msize = msize;
1779       creply->client_list = cl;
1780       GNUNET_CONTAINER_DLL_insert_after (cl->res_head,
1781                                          cl->res_tail,
1782                                          cl->res_tail,
1783                                          creply);
1784       pm = (struct PutMessage*) &creply[1];
1785       pm->header.type = htons (GNUNET_MESSAGE_TYPE_FS_PUT);
1786       pm->header.size = htons (msize);
1787       pm->type = htonl (prq->type);
1788       pm->expiration = GNUNET_TIME_absolute_hton (prq->expiration);
1789       memcpy (&pm[1], prq->data, prq->size);
1790       if (NULL == cl->th)
1791         cl->th = GNUNET_SERVER_notify_transmit_ready (cl->client,
1792                                                       msize,
1793                                                       GNUNET_TIME_UNIT_FOREVER_REL,
1794                                                       &transmit_to_client,
1795                                                       cl);
1796       GNUNET_break (cl->th != NULL);
1797     }
1798   else
1799     {
1800       cp = pr->cp;
1801 #if DEBUG_FS
1802       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1803                   "Transmitting result for query `%s' to other peer (PID=%u)\n",
1804                   GNUNET_h2s (key),
1805                   (unsigned int) cp->pid);
1806 #endif
1807       msize = sizeof (struct PutMessage) + prq->size;
1808       reply = GNUNET_malloc (msize + sizeof (struct PendingMessage));
1809       reply->cont = &transmit_reply_continuation;
1810       reply->cont_cls = pr;
1811       reply->msize = msize;
1812       reply->priority = (uint32_t) -1; /* send replies first! */
1813       pm = (struct PutMessage*) &reply[1];
1814       pm->header.type = htons (GNUNET_MESSAGE_TYPE_FS_PUT);
1815       pm->header.size = htons (msize);
1816       pm->type = htonl (prq->type);
1817       pm->expiration = GNUNET_TIME_absolute_hton (prq->expiration);
1818       memcpy (&pm[1], prq->data, prq->size);
1819       add_to_pending_messages_for_peer (cp, reply, pr);
1820     }
1821   if (GNUNET_YES == do_remove)
1822     {
1823       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1824                   "Removing request `%s' from request map (has been satisfied)\n",
1825                   GNUNET_h2s (key));
1826       GNUNET_break (GNUNET_YES ==
1827                     GNUNET_CONTAINER_multihashmap_remove (query_request_map,
1828                                                           key,
1829                                                           pr));
1830       // FIXME: request somehow does not fully
1831       // disappear; how to fix?
1832       // destroy_pending_request (pr); (not like this!)
1833     }
1834
1835   // FIXME: implement hot-path routing statistics keeping!
1836   return GNUNET_YES;
1837 }
1838
1839
1840 /**
1841  * Handle P2P "PUT" message.
1842  *
1843  * @param cls closure, always NULL
1844  * @param other the other peer involved (sender or receiver, NULL
1845  *        for loopback messages where we are both sender and receiver)
1846  * @param message the actual message
1847  * @param latency reported latency of the connection with 'other'
1848  * @param distance reported distance (DV) to 'other'
1849  * @return GNUNET_OK to keep the connection open,
1850  *         GNUNET_SYSERR to close it (signal serious error)
1851  */
1852 static int
1853 handle_p2p_put (void *cls,
1854                 const struct GNUNET_PeerIdentity *other,
1855                 const struct GNUNET_MessageHeader *message,
1856                 struct GNUNET_TIME_Relative latency,
1857                 uint32_t distance)
1858 {
1859   const struct PutMessage *put;
1860   uint16_t msize;
1861   size_t dsize;
1862   uint32_t type;
1863   struct GNUNET_TIME_Absolute expiration;
1864   GNUNET_HashCode query;
1865   struct ProcessReplyClosure prq;
1866
1867   msize = ntohs (message->size);
1868   if (msize < sizeof (struct PutMessage))
1869     {
1870       GNUNET_break_op(0);
1871       return GNUNET_SYSERR;
1872     }
1873   put = (const struct PutMessage*) message;
1874   dsize = msize - sizeof (struct PutMessage);
1875   type = ntohl (put->type);
1876   expiration = GNUNET_TIME_absolute_ntoh (put->expiration);
1877
1878   /* first, validate! */
1879   switch (type)
1880     {
1881     case GNUNET_DATASTORE_BLOCKTYPE_DBLOCK:
1882     case GNUNET_DATASTORE_BLOCKTYPE_IBLOCK:
1883       GNUNET_CRYPTO_hash (&put[1], dsize, &query);
1884       break;
1885     case GNUNET_DATASTORE_BLOCKTYPE_KBLOCK:
1886       if (GNUNET_OK !=
1887           check_kblock ((const struct KBlock*) &put[1],
1888                         dsize,
1889                         &query))
1890         return GNUNET_SYSERR;
1891       break;
1892     case GNUNET_DATASTORE_BLOCKTYPE_SBLOCK:
1893       if (GNUNET_OK !=
1894           check_sblock ((const struct SBlock*) &put[1],
1895                         dsize,
1896                         &query,
1897                         &prq.namespace))
1898         return GNUNET_SYSERR;
1899       break;
1900     case GNUNET_DATASTORE_BLOCKTYPE_SKBLOCK:
1901       // FIXME -- validate SKBLOCK!
1902       GNUNET_break (0);
1903       return GNUNET_OK;
1904     default:
1905       /* unknown block type */
1906       GNUNET_break_op (0);
1907       return GNUNET_SYSERR;
1908     }
1909
1910 #if DEBUG_FS
1911   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1912               "Received result for query `%s' from peer `%4s'\n",
1913               GNUNET_h2s (&query),
1914               GNUNET_i2s (other));
1915 #endif
1916   /* now, lookup 'query' */
1917   prq.data = (const void*) &put[1];
1918   prq.size = dsize;
1919   prq.type = type;
1920   prq.expiration = expiration;
1921   prq.priority = 0;
1922   GNUNET_CONTAINER_multihashmap_get_multiple (query_request_map,
1923                                               &query,
1924                                               &process_reply,
1925                                               &prq);
1926   // FIXME: if migration is on and load is low,
1927   // queue to store data in datastore;
1928   // use "prq.priority" for that!
1929   return GNUNET_OK;
1930 }
1931
1932
1933 /* **************************** P2P GET Handling ************************ */
1934
1935
1936 /**
1937  * Closure for 'check_duplicate_request_{peer,client}'.
1938  */
1939 struct CheckDuplicateRequestClosure
1940 {
1941   /**
1942    * The new request we should check if it already exists.
1943    */
1944   const struct PendingRequest *pr;
1945
1946   /**
1947    * Existing request found by the checker, NULL if none.
1948    */
1949   struct PendingRequest *have;
1950 };
1951
1952
1953 /**
1954  * Iterator over entries in the 'query_request_map' that
1955  * tries to see if we have the same request pending from
1956  * the same client already.
1957  *
1958  * @param cls closure (our 'struct CheckDuplicateRequestClosure')
1959  * @param key current key code (query, ignored, must match)
1960  * @param value value in the hash map (a 'struct PendingRequest'
1961  *              that already exists)
1962  * @return GNUNET_YES if we should continue to
1963  *         iterate (no match yet)
1964  *         GNUNET_NO if not (match found).
1965  */
1966 static int
1967 check_duplicate_request_client (void *cls,
1968                                 const GNUNET_HashCode * key,
1969                                 void *value)
1970 {
1971   struct CheckDuplicateRequestClosure *cdc = cls;
1972   struct PendingRequest *have = value;
1973
1974   if (have->client_request_list == NULL)
1975     return GNUNET_YES;
1976   if ( (cdc->pr->client_request_list->client_list->client == have->client_request_list->client_list->client) &&
1977        (cdc->pr != have) )
1978     {
1979       cdc->have = have;
1980       return GNUNET_NO;
1981     }
1982   return GNUNET_YES;
1983 }
1984
1985
1986 /**
1987  * We're processing (local) results for a search request
1988  * from another peer.  Pass applicable results to the
1989  * peer and if we are done either clean up (operation
1990  * complete) or forward to other peers (more results possible).
1991  *
1992  * @param cls our closure (struct LocalGetContext)
1993  * @param key key for the content
1994  * @param size number of bytes in data
1995  * @param data content stored
1996  * @param type type of the content
1997  * @param priority priority of the content
1998  * @param anonymity anonymity-level for the content
1999  * @param expiration expiration time for the content
2000  * @param uid unique identifier for the datum;
2001  *        maybe 0 if no unique identifier is available
2002  */
2003 static void
2004 process_local_reply (void *cls,
2005                      const GNUNET_HashCode * key,
2006                      uint32_t size,
2007                      const void *data,
2008                      uint32_t type,
2009                      uint32_t priority,
2010                      uint32_t anonymity,
2011                      struct GNUNET_TIME_Absolute
2012                      expiration,
2013                      uint64_t uid)
2014 {
2015   struct PendingRequest *pr = cls;
2016   struct ProcessReplyClosure prq;
2017   struct CheckDuplicateRequestClosure cdrc;
2018   GNUNET_HashCode dhash;
2019   GNUNET_HashCode mhash;
2020   GNUNET_HashCode query;
2021
2022   if (NULL == key)
2023     {
2024 #if DEBUG_FS > 1
2025       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2026                   "Done processing local replies, forwarding request to other peers.\n");
2027 #endif
2028       pr->drq = NULL;
2029       if (pr->client_request_list != NULL)
2030         {
2031           GNUNET_SERVER_receive_done (pr->client_request_list->client_list->client,
2032                                       GNUNET_YES);
2033           /* Figure out if this is a duplicate request and possibly
2034              merge 'struct PendingRequest' entries */
2035           cdrc.have = NULL;
2036           cdrc.pr = pr;
2037           GNUNET_CONTAINER_multihashmap_get_multiple (query_request_map,
2038                                                       &pr->query,
2039                                                       &check_duplicate_request_client,
2040                                                       &cdrc);
2041           if (cdrc.have != NULL)
2042             {
2043 #if DEBUG_FS
2044               GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2045                           "Received request for block `%s' twice from client, will only request once.\n",
2046                           GNUNET_h2s (&pr->query));
2047 #endif
2048
2049               destroy_pending_request (pr);
2050               return;
2051             }
2052         }
2053
2054       /* no more results */
2055       if (pr->task == GNUNET_SCHEDULER_NO_TASK)
2056         pr->task = GNUNET_SCHEDULER_add_now (sched,
2057                                              &forward_request_task,
2058                                              pr);
2059       return;
2060     }
2061   if (type == GNUNET_DATASTORE_BLOCKTYPE_ONDEMAND)
2062     {
2063 #if DEBUG_FS
2064       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2065                   "Found ONDEMAND block, performing on-demand encoding\n");
2066 #endif
2067       if (GNUNET_OK !=
2068           GNUNET_FS_handle_on_demand_block (key, size, data, type, priority,
2069                                             anonymity, expiration, uid,
2070                                             &process_local_reply,
2071                                             pr))
2072         GNUNET_FS_drq_get_next (GNUNET_YES);
2073       return;
2074     }
2075   /* check for duplicates */
2076   GNUNET_CRYPTO_hash (data, size, &dhash);
2077   mingle_hash (&dhash,
2078                pr->mingle,
2079                &mhash);
2080   if ( (pr->bf != NULL) &&
2081        (GNUNET_YES ==
2082         GNUNET_CONTAINER_bloomfilter_test (pr->bf,
2083                                            &mhash)) )
2084     {
2085 #if DEBUG_FS
2086       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2087                   "Result from datastore filtered by bloomfilter (duplicate).\n");
2088 #endif
2089       GNUNET_FS_drq_get_next (GNUNET_YES);
2090       return;
2091     }
2092 #if DEBUG_FS
2093   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2094               "Found result for query `%s' in local datastore\n",
2095               GNUNET_h2s (key));
2096 #endif
2097   pr->results_found++;
2098   if ( (pr->type == GNUNET_DATASTORE_BLOCKTYPE_KBLOCK) ||
2099        (pr->type == GNUNET_DATASTORE_BLOCKTYPE_SBLOCK) ||
2100        (pr->type == GNUNET_DATASTORE_BLOCKTYPE_SKBLOCK) )
2101     {
2102       if (pr->bf == NULL)
2103         {
2104           pr->bf_size = 32;
2105           pr->bf = GNUNET_CONTAINER_bloomfilter_init (NULL,
2106                                                       pr->bf_size,
2107                                                       BLOOMFILTER_K);
2108         }
2109       GNUNET_CONTAINER_bloomfilter_add (pr->bf,
2110                                         &mhash);
2111     }
2112   memset (&prq, 0, sizeof (prq));
2113   prq.data = data;
2114   prq.expiration = expiration;
2115   prq.size = size;
2116   if ( (type == GNUNET_DATASTORE_BLOCKTYPE_SBLOCK) &&
2117        (GNUNET_OK != check_sblock ((const struct SBlock*) data,
2118                                    size,
2119                                    &query,
2120                                    &prq.namespace)) )
2121     {
2122       GNUNET_break (0);
2123       /* FIXME: consider removing the block? */
2124       GNUNET_FS_drq_get_next (GNUNET_YES);
2125       return;
2126     }
2127   prq.type = type;
2128   prq.priority = priority;
2129   process_reply (&prq, key, pr);
2130
2131   if ( ( (pr->client_request_list == NULL) &&
2132          ( (GNUNET_YES == test_load_too_high()) ||
2133            (pr->results_found > 5 + 2 * pr->priority) ) ) ||
2134        (type == GNUNET_DATASTORE_BLOCKTYPE_DBLOCK) )
2135     {
2136 #if DEBUG_FS > 2
2137       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2138                   "Unique reply found or load too high, done with request\n");
2139 #endif
2140       GNUNET_FS_drq_get_next (GNUNET_NO);
2141       return;
2142     }
2143   GNUNET_FS_drq_get_next (GNUNET_YES);
2144 }
2145
2146
2147 /**
2148  * The priority level imposes a bound on the maximum
2149  * value for the ttl that can be requested.
2150  *
2151  * @param ttl_in requested ttl
2152  * @param prio given priority
2153  * @return ttl_in if ttl_in is below the limit,
2154  *         otherwise the ttl-limit for the given priority
2155  */
2156 static int32_t
2157 bound_ttl (int32_t ttl_in, uint32_t prio)
2158 {
2159   unsigned long long allowed;
2160
2161   if (ttl_in <= 0)
2162     return ttl_in;
2163   allowed = ((unsigned long long) prio) * TTL_DECREMENT / 1000;
2164   if (ttl_in > allowed)
2165     {
2166       if (allowed >= (1 << 30))
2167         return 1 << 30;
2168       return allowed;
2169     }
2170   return ttl_in;
2171 }
2172
2173
2174 /**
2175  * We've received a request with the specified priority.  Bound it
2176  * according to how much we trust the given peer.
2177  *
2178  * @param prio_in requested priority
2179  * @param cp the peer making the request
2180  * @return effective priority
2181  */
2182 static uint32_t
2183 bound_priority (uint32_t prio_in,
2184                 struct ConnectedPeer *cp)
2185 {
2186   return 0; // FIXME!
2187 }
2188
2189
2190 /**
2191  * Iterator over entries in the 'query_request_map' that
2192  * tries to see if we have the same request pending from
2193  * the same peer already.
2194  *
2195  * @param cls closure (our 'struct CheckDuplicateRequestClosure')
2196  * @param key current key code (query, ignored, must match)
2197  * @param value value in the hash map (a 'struct PendingRequest'
2198  *              that already exists)
2199  * @return GNUNET_YES if we should continue to
2200  *         iterate (no match yet)
2201  *         GNUNET_NO if not (match found).
2202  */
2203 static int
2204 check_duplicate_request_peer (void *cls,
2205                               const GNUNET_HashCode * key,
2206                               void *value)
2207 {
2208   struct CheckDuplicateRequestClosure *cdc = cls;
2209   struct PendingRequest *have = value;
2210
2211   if (cdc->pr->target_pid == have->target_pid)
2212     {
2213       cdc->have = have;
2214       return GNUNET_NO;
2215     }
2216   return GNUNET_YES;
2217 }
2218
2219
2220 /**
2221  * Handle P2P "GET" request.
2222  *
2223  * @param cls closure, always NULL
2224  * @param other the other peer involved (sender or receiver, NULL
2225  *        for loopback messages where we are both sender and receiver)
2226  * @param message the actual message
2227  * @param latency reported latency of the connection with 'other'
2228  * @param distance reported distance (DV) to 'other'
2229  * @return GNUNET_OK to keep the connection open,
2230  *         GNUNET_SYSERR to close it (signal serious error)
2231  */
2232 static int
2233 handle_p2p_get (void *cls,
2234                 const struct GNUNET_PeerIdentity *other,
2235                 const struct GNUNET_MessageHeader *message,
2236                 struct GNUNET_TIME_Relative latency,
2237                 uint32_t distance)
2238 {
2239   struct PendingRequest *pr;
2240   struct ConnectedPeer *cp;
2241   struct ConnectedPeer *cps;
2242   struct CheckDuplicateRequestClosure cdc;
2243   struct GNUNET_TIME_Relative timeout;
2244   uint16_t msize;
2245   const struct GetMessage *gm;
2246   unsigned int bits;
2247   const GNUNET_HashCode *opt;
2248   uint32_t bm;
2249   size_t bfsize;
2250   uint32_t ttl_decrement;
2251   uint32_t type;
2252   double preference;
2253   int have_ns;
2254
2255   msize = ntohs(message->size);
2256   if (msize < sizeof (struct GetMessage))
2257     {
2258       GNUNET_break_op (0);
2259       return GNUNET_SYSERR;
2260     }
2261   gm = (const struct GetMessage*) message;
2262   type = ntohl (gm->type);
2263   switch (type)
2264     {
2265     case GNUNET_DATASTORE_BLOCKTYPE_DBLOCK:
2266     case GNUNET_DATASTORE_BLOCKTYPE_IBLOCK:
2267     case GNUNET_DATASTORE_BLOCKTYPE_KBLOCK:
2268     case GNUNET_DATASTORE_BLOCKTYPE_SBLOCK:
2269       break;
2270     default:
2271       GNUNET_break_op (0);
2272       return GNUNET_SYSERR;
2273     }
2274   bm = ntohl (gm->hash_bitmap);
2275   bits = 0;
2276   while (bm > 0)
2277     {
2278       if (1 == (bm & 1))
2279         bits++;
2280       bm >>= 1;
2281     }
2282   if (msize < sizeof (struct GetMessage) + bits * sizeof (GNUNET_HashCode))
2283     {
2284       GNUNET_break_op (0);
2285       return GNUNET_SYSERR;
2286     }
2287   opt = (const GNUNET_HashCode*) &gm[1];
2288   bfsize = msize - sizeof (struct GetMessage) + bits * sizeof (GNUNET_HashCode);
2289   bm = ntohl (gm->hash_bitmap);
2290   if ( (0 != (bm & GET_MESSAGE_BIT_SKS_NAMESPACE)) &&
2291        (ntohl (gm->type) == GNUNET_DATASTORE_BLOCKTYPE_SBLOCK) )
2292     {
2293       GNUNET_break_op (0);
2294       return GNUNET_SYSERR;
2295     }
2296   bits = 0;
2297   cps = GNUNET_CONTAINER_multihashmap_get (connected_peers,
2298                                            &other->hashPubKey);
2299   GNUNET_assert (NULL != cps);
2300   if (0 != (bm & GET_MESSAGE_BIT_RETURN_TO))
2301     cp = GNUNET_CONTAINER_multihashmap_get (connected_peers,
2302                                             &opt[bits++]);
2303   else
2304     cp = cps;
2305   if (cp == NULL)
2306     {
2307 #if DEBUG_FS
2308       if (0 != (bm & GET_MESSAGE_BIT_RETURN_TO))
2309         GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2310                     "Failed to find RETURN-TO peer `%4s' in connection set. Dropping query.\n",
2311                     GNUNET_i2s ((const struct GNUNET_PeerIdentity*) &opt[bits-1]));
2312
2313       else
2314         GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2315                     "Failed to find peer `%4s' in connection set. Dropping query.\n",
2316                     GNUNET_i2s (other));
2317 #endif
2318      /* FIXME: try connect? */
2319       return GNUNET_OK;
2320     }
2321   /* note that we can really only check load here since otherwise
2322      peers could find out that we are overloaded by not being
2323      disconnected after sending us a malformed query... */
2324   if (GNUNET_YES == test_load_too_high ())
2325     {
2326 #if DEBUG_FS
2327       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2328                   "Dropping query from `%s', this peer is too busy.\n",
2329                   GNUNET_i2s (other));
2330 #endif
2331       return GNUNET_OK;
2332     }
2333
2334 #if DEBUG_FS
2335   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2336               "Received request for `%s' of type %u from peer `%4s'\n",
2337               GNUNET_h2s (&gm->query),
2338               (unsigned int) ntohl (gm->type),
2339               GNUNET_i2s (other));
2340 #endif
2341   have_ns = (0 != (bm & GET_MESSAGE_BIT_SKS_NAMESPACE));
2342   pr = GNUNET_malloc (sizeof (struct PendingRequest) +
2343                       (have_ns ? sizeof(GNUNET_HashCode) : 0));
2344   if (have_ns)
2345     pr->namespace = (GNUNET_HashCode*) &pr[1];
2346   pr->type = type;
2347   pr->mingle = ntohl (gm->filter_mutator);
2348   if (0 != (bm & GET_MESSAGE_BIT_SKS_NAMESPACE))
2349     memcpy (&pr[1], &opt[bits++], sizeof (GNUNET_HashCode));
2350   else if (pr->type == GNUNET_DATASTORE_BLOCKTYPE_SBLOCK)
2351     {
2352       GNUNET_break_op (0);
2353       GNUNET_free (pr);
2354       return GNUNET_SYSERR;
2355     }
2356   if (0 != (bm & GET_MESSAGE_BIT_TRANSMIT_TO))
2357     pr->target_pid = GNUNET_PEER_intern ((const struct GNUNET_PeerIdentity*) &opt[bits++]);
2358
2359   pr->anonymity_level = 1;
2360   pr->priority = bound_priority (ntohl (gm->priority), cps);
2361   pr->ttl = bound_ttl (ntohl (gm->ttl), pr->priority);
2362   pr->query = gm->query;
2363   /* decrement ttl (always) */
2364   ttl_decrement = 2 * TTL_DECREMENT +
2365     GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
2366                               TTL_DECREMENT);
2367   if ( (pr->ttl < 0) &&
2368        (pr->ttl - ttl_decrement > 0) )
2369     {
2370 #if DEBUG_FS
2371       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2372                   "Dropping query from `%s' due to TTL underflow.\n",
2373                   GNUNET_i2s (other));
2374 #endif
2375       /* integer underflow => drop (should be very rare)! */
2376       GNUNET_free (pr);
2377       return GNUNET_OK;
2378     }
2379   pr->ttl -= ttl_decrement;
2380   pr->start_time = GNUNET_TIME_absolute_get ();
2381
2382   /* get bloom filter */
2383   if (bfsize > 0)
2384     {
2385       pr->bf = GNUNET_CONTAINER_bloomfilter_init ((const char*) &opt[bits],
2386                                                   bfsize,
2387                                                   BLOOMFILTER_K);
2388       pr->bf_size = bfsize;
2389     }
2390
2391   cdc.have = NULL;
2392   cdc.pr = pr;
2393   GNUNET_CONTAINER_multihashmap_get_multiple (query_request_map,
2394                                               &gm->query,
2395                                               &check_duplicate_request_peer,
2396                                               &cdc);
2397   if (cdc.have != NULL)
2398     {
2399       if (cdc.have->start_time.value + cdc.have->ttl >=
2400           pr->start_time.value + pr->ttl)
2401         {
2402           /* existing request has higher TTL, drop new one! */
2403           cdc.have->priority += pr->priority;
2404           destroy_pending_request (pr);
2405 #if DEBUG_FS
2406           GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2407                       "Have existing request with higher TTL, dropping new request.\n",
2408                       GNUNET_i2s (other));
2409 #endif
2410           return GNUNET_OK;
2411         }
2412       else
2413         {
2414           /* existing request has lower TTL, drop old one! */
2415           pr->priority += cdc.have->priority;
2416           /* Possible optimization: if we have applicable pending
2417              replies in 'cdc.have', we might want to move those over
2418              (this is a really rare special-case, so it is not clear
2419              that this would be worth it) */
2420           destroy_pending_request (cdc.have);
2421           /* keep processing 'pr'! */
2422         }
2423     }
2424
2425   pr->cp = cp;
2426   GNUNET_CONTAINER_multihashmap_put (query_request_map,
2427                                      &gm->query,
2428                                      pr,
2429                                      GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
2430   GNUNET_CONTAINER_multihashmap_put (peer_request_map,
2431                                      &other->hashPubKey,
2432                                      pr,
2433                                      GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
2434
2435   pr->hnode = GNUNET_CONTAINER_heap_insert (requests_by_expiration_heap,
2436                                             pr,
2437                                             pr->start_time.value + pr->ttl);
2438
2439
2440   /* calculate change in traffic preference */
2441   preference = (double) pr->priority;
2442   if (preference < QUERY_BANDWIDTH_VALUE)
2443     preference = QUERY_BANDWIDTH_VALUE;
2444   cps->inc_preference += preference;
2445
2446   /* process locally */
2447   if (type == GNUNET_DATASTORE_BLOCKTYPE_DBLOCK)
2448     type = GNUNET_DATASTORE_BLOCKTYPE_ANY; /* to get on-demand as well */
2449   timeout = GNUNET_TIME_relative_multiply (BASIC_DATASTORE_REQUEST_DELAY,
2450                                            (pr->priority + 1));
2451   pr->drq = GNUNET_FS_drq_get (&gm->query,
2452                                pr->type,
2453                                &process_local_reply,
2454                                pr,
2455                                timeout,
2456                                GNUNET_NO);
2457
2458   /* Are multiple results possible?  If so, start processing remotely now! */
2459   switch (pr->type)
2460     {
2461     case GNUNET_DATASTORE_BLOCKTYPE_DBLOCK:
2462     case GNUNET_DATASTORE_BLOCKTYPE_IBLOCK:
2463       /* only one result, wait for datastore */
2464       break;
2465     default:
2466       pr->task = GNUNET_SCHEDULER_add_now (sched,
2467                                            &forward_request_task,
2468                                            pr);
2469     }
2470
2471   /* make sure we don't track too many requests */
2472   if (GNUNET_CONTAINER_heap_get_size (requests_by_expiration_heap) > max_pending_requests)
2473     {
2474       pr = GNUNET_CONTAINER_heap_peek (requests_by_expiration_heap);
2475       destroy_pending_request (pr);
2476     }
2477   return GNUNET_OK;
2478 }
2479
2480
2481 /* **************************** CS GET Handling ************************ */
2482
2483
2484 /**
2485  * Handle START_SEARCH-message (search request from client).
2486  *
2487  * @param cls closure
2488  * @param client identification of the client
2489  * @param message the actual message
2490  */
2491 static void
2492 handle_start_search (void *cls,
2493                      struct GNUNET_SERVER_Client *client,
2494                      const struct GNUNET_MessageHeader *message)
2495 {
2496   static GNUNET_HashCode all_zeros;
2497   const struct SearchMessage *sm;
2498   struct ClientList *cl;
2499   struct ClientRequestList *crl;
2500   struct PendingRequest *pr;
2501   uint16_t msize;
2502   unsigned int sc;
2503   uint32_t type;
2504
2505   msize = ntohs (message->size);
2506   if ( (msize < sizeof (struct SearchMessage)) ||
2507        (0 != (msize - sizeof (struct SearchMessage)) % sizeof (GNUNET_HashCode)) )
2508     {
2509       GNUNET_break (0);
2510       GNUNET_SERVER_receive_done (client,
2511                                   GNUNET_SYSERR);
2512       return;
2513     }
2514   sc = (msize - sizeof (struct SearchMessage)) / sizeof (GNUNET_HashCode);
2515   sm = (const struct SearchMessage*) message;
2516
2517   cl = client_list;
2518   while ( (cl != NULL) &&
2519           (cl->client != client) )
2520     cl = cl->next;
2521   if (cl == NULL)
2522     {
2523       cl = GNUNET_malloc (sizeof (struct ClientList));
2524       cl->client = client;
2525       GNUNET_SERVER_client_keep (client);
2526       cl->next = client_list;
2527       client_list = cl;
2528     }
2529   type = ntohl (sm->type);
2530   switch (type)
2531     {
2532     case GNUNET_DATASTORE_BLOCKTYPE_ANY:
2533     case GNUNET_DATASTORE_BLOCKTYPE_DBLOCK:
2534     case GNUNET_DATASTORE_BLOCKTYPE_IBLOCK:
2535     case GNUNET_DATASTORE_BLOCKTYPE_KBLOCK:
2536     case GNUNET_DATASTORE_BLOCKTYPE_SBLOCK:
2537       break;
2538     default:
2539       GNUNET_break (0);
2540       GNUNET_SERVER_receive_done (client,
2541                                   GNUNET_SYSERR);
2542       return;
2543     }
2544 #if DEBUG_FS
2545   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2546               "Received request for `%s' of type %u from local client\n",
2547               GNUNET_h2s (&sm->query),
2548               (unsigned int) type);
2549 #endif
2550
2551   /* detect duplicate KBLOCK requests */
2552   if (type == GNUNET_DATASTORE_BLOCKTYPE_KBLOCK)
2553     {
2554       crl = cl->rl_head;
2555       while ( (crl != NULL) &&
2556               ( (0 != memcmp (&crl->req->query,
2557                               &sm->query,
2558                               sizeof (GNUNET_HashCode))) ||
2559                 (crl->req->type != type) ) )
2560         crl = crl->next;
2561       if (crl != NULL)
2562         {
2563 #if DEBUG_FS
2564           GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2565                       "Have existing request, merging content-seen lists.\n");
2566 #endif
2567           pr = crl->req;
2568           /* Duplicate request (used to send long list of
2569              known/blocked results); merge 'pr->replies_seen'
2570              and update bloom filter */
2571           GNUNET_array_grow (pr->replies_seen,
2572                              pr->replies_seen_size,
2573                              pr->replies_seen_off + sc);
2574           memcpy (&pr->replies_seen[pr->replies_seen_off],
2575                   &sm[1],
2576                   sc * sizeof (GNUNET_HashCode));
2577           pr->replies_seen_off += sc;
2578           if (pr->bf != NULL)
2579             GNUNET_CONTAINER_bloomfilter_free (pr->bf);
2580           pr->bf = refresh_bloomfilter (pr->replies_seen_off,
2581                                         &pr->mingle,
2582                                         &pr->bf_size,
2583                                         pr->replies_seen);
2584           GNUNET_SERVER_receive_done (client,
2585                                       GNUNET_OK);
2586           return;
2587         }
2588     }
2589   pr = GNUNET_malloc (sizeof (struct PendingRequest) +
2590                       ((type == GNUNET_DATASTORE_BLOCKTYPE_SBLOCK)?sizeof(GNUNET_HashCode):0));
2591   crl = GNUNET_malloc (sizeof (struct ClientRequestList));
2592   memset (crl, 0, sizeof (struct ClientRequestList));
2593   crl->client_list = cl;
2594   GNUNET_CONTAINER_DLL_insert (cl->rl_head,
2595                                cl->rl_tail,
2596                                crl);
2597   crl->req = pr;
2598   pr->type = type;
2599   pr->client_request_list = crl;
2600   GNUNET_array_grow (pr->replies_seen,
2601                      pr->replies_seen_size,
2602                      sc);
2603   memcpy (pr->replies_seen,
2604           &sm[1],
2605           sc * sizeof (GNUNET_HashCode));
2606   pr->replies_seen_off = sc;
2607   pr->anonymity_level = ntohl (sm->anonymity_level);
2608   pr->bf = refresh_bloomfilter (pr->replies_seen_off,
2609                                 &pr->mingle,
2610                                 &pr->bf_size,
2611                                 pr->replies_seen);
2612  pr->query = sm->query;
2613   switch (type)
2614     {
2615     case GNUNET_DATASTORE_BLOCKTYPE_DBLOCK:
2616     case GNUNET_DATASTORE_BLOCKTYPE_IBLOCK:
2617       if (0 != memcmp (&sm->target,
2618                        &all_zeros,
2619                        sizeof (GNUNET_HashCode)))
2620         pr->target_pid = GNUNET_PEER_intern ((const struct GNUNET_PeerIdentity*) &sm->target);
2621       break;
2622     case GNUNET_DATASTORE_BLOCKTYPE_SBLOCK:
2623       pr->namespace = (GNUNET_HashCode*) &pr[1];
2624       memcpy (&pr[1], &sm->target, sizeof (GNUNET_HashCode));
2625       break;
2626     default:
2627       break;
2628     }
2629   GNUNET_CONTAINER_multihashmap_put (query_request_map,
2630                                      &sm->query,
2631                                      pr,
2632                                      GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
2633   if (type == GNUNET_DATASTORE_BLOCKTYPE_DBLOCK)
2634     type = GNUNET_DATASTORE_BLOCKTYPE_ANY; /* get on-demand blocks too! */
2635   pr->drq = GNUNET_FS_drq_get (&sm->query,
2636                                type,
2637                                &process_local_reply,
2638                                pr,
2639                                GNUNET_TIME_UNIT_FOREVER_REL,
2640                                GNUNET_YES);
2641 }
2642
2643
2644 /* **************************** Startup ************************ */
2645
2646
2647 /**
2648  * List of handlers for P2P messages
2649  * that we care about.
2650  */
2651 static struct GNUNET_CORE_MessageHandler p2p_handlers[] =
2652   {
2653     { &handle_p2p_get,
2654       GNUNET_MESSAGE_TYPE_FS_GET, 0 },
2655     { &handle_p2p_put,
2656       GNUNET_MESSAGE_TYPE_FS_PUT, 0 },
2657     { NULL, 0, 0 }
2658   };
2659
2660
2661 /**
2662  * List of handlers for the messages understood by this
2663  * service.
2664  */
2665 static struct GNUNET_SERVER_MessageHandler handlers[] = {
2666   {&GNUNET_FS_handle_index_start, NULL,
2667    GNUNET_MESSAGE_TYPE_FS_INDEX_START, 0},
2668   {&GNUNET_FS_handle_index_list_get, NULL,
2669    GNUNET_MESSAGE_TYPE_FS_INDEX_LIST_GET, sizeof(struct GNUNET_MessageHeader) },
2670   {&GNUNET_FS_handle_unindex, NULL, GNUNET_MESSAGE_TYPE_FS_UNINDEX,
2671    sizeof (struct UnindexMessage) },
2672   {&handle_start_search, NULL, GNUNET_MESSAGE_TYPE_FS_START_SEARCH,
2673    0 },
2674   {NULL, NULL, 0, 0}
2675 };
2676
2677
2678 /**
2679  * Process fs requests.
2680  *
2681  * @param s scheduler to use
2682  * @param server the initialized server
2683  * @param c configuration to use
2684  */
2685 static int
2686 main_init (struct GNUNET_SCHEDULER_Handle *s,
2687            struct GNUNET_SERVER_Handle *server,
2688            const struct GNUNET_CONFIGURATION_Handle *c)
2689 {
2690   sched = s;
2691   cfg = c;
2692   connected_peers = GNUNET_CONTAINER_multihashmap_create (128); // FIXME: get size from config
2693   query_request_map = GNUNET_CONTAINER_multihashmap_create (128); // FIXME: get size from config
2694   peer_request_map = GNUNET_CONTAINER_multihashmap_create (128); // FIXME: get size from config
2695   requests_by_expiration_heap = GNUNET_CONTAINER_heap_create (GNUNET_CONTAINER_HEAP_ORDER_MIN);
2696   core = GNUNET_CORE_connect (sched,
2697                               cfg,
2698                               GNUNET_TIME_UNIT_FOREVER_REL,
2699                               NULL,
2700                               NULL,
2701                               NULL,
2702                               &peer_connect_handler,
2703                               &peer_disconnect_handler,
2704                               NULL, GNUNET_NO,
2705                               NULL, GNUNET_NO,
2706                               p2p_handlers);
2707   if (NULL == core)
2708     {
2709       GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
2710                   _("Failed to connect to `%s' service.\n"),
2711                   "core");
2712       GNUNET_CONTAINER_multihashmap_destroy (connected_peers);
2713       connected_peers = NULL;
2714       GNUNET_CONTAINER_multihashmap_destroy (query_request_map);
2715       query_request_map = NULL;
2716       GNUNET_CONTAINER_heap_destroy (requests_by_expiration_heap);
2717       requests_by_expiration_heap = NULL;
2718       GNUNET_CONTAINER_multihashmap_destroy (peer_request_map);
2719       peer_request_map = NULL;
2720
2721       return GNUNET_SYSERR;
2722     }
2723   GNUNET_SERVER_disconnect_notify (server,
2724                                    &handle_client_disconnect,
2725                                    NULL);
2726   GNUNET_SERVER_add_handlers (server, handlers);
2727   GNUNET_SCHEDULER_add_delayed (sched,
2728                                 GNUNET_TIME_UNIT_FOREVER_REL,
2729                                 &shutdown_task,
2730                                 NULL);
2731   return GNUNET_OK;
2732 }
2733
2734
2735 /**
2736  * Process fs requests.
2737  *
2738  * @param cls closure
2739  * @param sched scheduler to use
2740  * @param server the initialized server
2741  * @param cfg configuration to use
2742  */
2743 static void
2744 run (void *cls,
2745      struct GNUNET_SCHEDULER_Handle *sched,
2746      struct GNUNET_SERVER_Handle *server,
2747      const struct GNUNET_CONFIGURATION_Handle *cfg)
2748 {
2749   if ( (GNUNET_OK != GNUNET_FS_drq_init (sched, cfg)) ||
2750        (GNUNET_OK != GNUNET_FS_indexing_init (sched, cfg)) ||
2751        (GNUNET_OK != main_init (sched, server, cfg)) )
2752     {
2753       GNUNET_SCHEDULER_shutdown (sched);
2754       return;
2755     }
2756 }
2757
2758
2759 /**
2760  * The main function for the fs service.
2761  *
2762  * @param argc number of arguments from the command line
2763  * @param argv command line arguments
2764  * @return 0 ok, 1 on error
2765  */
2766 int
2767 main (int argc, char *const *argv)
2768 {
2769   return (GNUNET_OK ==
2770           GNUNET_SERVICE_run (argc,
2771                               argv,
2772                               "fs",
2773                               GNUNET_SERVICE_OPTION_NONE,
2774                               &run, NULL)) ? 0 : 1;
2775 }
2776
2777 /* end of gnunet-service-fs.c */