src/datastore/gnunet-service-datastore.c

   1 /*
   2      This file is part of GNUnet
   3      (C) 2004, 2005, 2006, 2007, 2009 Christian Grothoff (and other contributing authors)
   4
   5      GNUnet is free software; you can redistribute it and/or modify
   6      it under the terms of the GNU General Public License as published
   7      by the Free Software Foundation; either version 2, or (at your
   8      option) any later version.
   9
  10      GNUnet is distributed in the hope that it will be useful, but
  11      WITHOUT ANY WARRANTY; without even the implied warranty of
  12      MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13      General Public License for more details.
  14
  15      You should have received a copy of the GNU General Public License
  16      along with GNUnet; see the file COPYING.  If not, write to the
  17      Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  18      Boston, MA 02111-1307, USA.
  19 */
  20
  21 /**
  22  * @file datastore/gnunet-service-datastore.c
  23  * @brief Management for the datastore for files stored on a GNUnet node
  24  * @author Christian Grothoff
  25  */
  26
  27 #include "platform.h"
  28 #include "gnunet_util_lib.h"
  29 #include "gnunet_protocols.h"
  30 #include "gnunet_statistics_service.h"
  31 #include "plugin_datastore.h"
  32 #include "datastore.h"
  33
  34 /**
  35  * How many messages do we queue at most per client?
  36  */
  37 #define MAX_PENDING 1024
  38
  39 /**
  40  * How long are we at most keeping "expired" content
  41  * past the expiration date in the database?
  42  */
  43 #define MAX_EXPIRE_DELAY GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MINUTES, 15)
  44
  45
  46
  47 /**
  48  * Our datastore plugin.
  49  */
  50 struct DatastorePlugin
  51 {
  52
  53   /**
  54    * API of the transport as returned by the plugin's
  55    * initialization function.
  56    */
  57   struct GNUNET_DATASTORE_PluginFunctions *api;
  58
  59   /**
  60    * Short name for the plugin (i.e. "sqlite").
  61    */
  62   char *short_name;
  63
  64   /**
  65    * Name of the library (i.e. "gnunet_plugin_datastore_sqlite").
  66    */
  67   char *lib_name;
  68
  69   /**
  70    * Environment this transport service is using
  71    * for this plugin.
  72    */
  73   struct GNUNET_DATASTORE_PluginEnvironment env;
  74
  75 };
  76
  77
  78 /**
  79  * Linked list of active reservations.
  80  */
  81 struct ReservationList
  82 {
  83
  84   /**
  85    * This is a linked list.
  86    */
  87   struct ReservationList *next;
  88
  89   /**
  90    * Client that made the reservation.
  91    */
  92   struct GNUNET_SERVER_Client *client;
  93
  94   /**
  95    * Number of bytes (still) reserved.
  96    */
  97   uint64_t amount;
  98
  99   /**
 100    * Number of items (still) reserved.
 101    */
 102   uint64_t entries;
 103
 104   /**
 105    * Reservation identifier.
 106    */
 107   int32_t rid;
 108
 109 };
 110
 111
 112 /**
 113  * Our datastore plugin (NULL if not available).
 114  */
 115 static struct DatastorePlugin *plugin;
 116
 117 /**
 118  * Linked list of space reservations made by clients.
 119  */
 120 static struct ReservationList *reservations;
 121
 122 /**
 123  * Bloomfilter to quickly tell if we don't have the content.
 124  */
 125 static struct GNUNET_CONTAINER_BloomFilter *filter;
 126
 127 /**
 128  * How much space are we allowed to use?
 129  */
 130 static unsigned long long quota;
 131
 132 /**
 133  * How much space are we using for the cache?  (space available for
 134  * insertions that will be instantly reclaimed by discarding less
 135  * important content --- or possibly whatever we just inserted into
 136  * the "cache").
 137  */
 138 static unsigned long long cache_size;
 139
 140 /**
 141  * How much space have we currently reserved?
 142  */
 143 static unsigned long long reserved;
 144
 145 /**
 146  * Identity of the task that is used to delete
 147  * expired content.
 148  */
 149 static GNUNET_SCHEDULER_TaskIdentifier expired_kill_task;
 150
 151 /**
 152  * Our configuration.
 153  */
 154 const struct GNUNET_CONFIGURATION_Handle *cfg;
 155
 156 /**
 157  * Our scheduler.
 158  */
 159 struct GNUNET_SCHEDULER_Handle *sched;
 160
 161 /**
 162  * Handle for reporting statistics.
 163  */
 164 static struct GNUNET_STATISTICS_Handle *stats;
 165
 166
 167 /**
 168  * Function called once the transmit operation has
 169  * either failed or succeeded.
 170  *
 171  * @param cls closure
 172  * @param status GNUNET_OK on success, GNUNET_SYSERR on error
 173  */
 174 typedef void (*TransmitContinuation)(void *cls,
 175                                      int status);
 176
 177
 178 /**
 179  * Context for transmitting replies to clients.
 180  */
 181 struct TransmitCallbackContext
 182 {
 183
 184   /**
 185    * We keep these in a doubly-linked list (for cleanup).
 186    */
 187   struct TransmitCallbackContext *next;
 188
 189   /**
 190    * We keep these in a doubly-linked list (for cleanup).
 191    */
 192   struct TransmitCallbackContext *prev;
 193
 194   /**
 195    * The message that we're asked to transmit.
 196    */
 197   struct GNUNET_MessageHeader *msg;
 198
 199   /**
 200    * Handle for the transmission request.
 201    */
 202   struct GNUNET_CONNECTION_TransmitHandle *th;
 203
 204   /**
 205    * Client that we are transmitting to.
 206    */
 207   struct GNUNET_SERVER_Client *client;
 208
 209   /**
 210    * Function to call once msg has been transmitted
 211    * (or at least added to the buffer).
 212    */
 213   TransmitContinuation tc;
 214
 215   /**
 216    * Closure for tc.
 217    */
 218   void *tc_cls;
 219
 220   /**
 221    * GNUNET_YES if we are supposed to signal the server
 222    * completion of the client's request.
 223    */
 224   int end;
 225 };
 226
 227
 228 /**
 229  * Head of the doubly-linked list (for cleanup).
 230  */
 231 static struct TransmitCallbackContext *tcc_head;
 232
 233 /**
 234  * Tail of the doubly-linked list (for cleanup).
 235  */
 236 static struct TransmitCallbackContext *tcc_tail;
 237
 238 /**
 239  * Have we already cleaned up the TCCs and are hence no longer
 240  * willing (or able) to transmit anything to anyone?
 241  */
 242 static int cleaning_done;
 243
 244 /**
 245  * Task that is used to remove expired entries from
 246  * the datastore.  This task will schedule itself
 247  * again automatically to always delete all expired
 248  * content quickly.
 249  *
 250  * @param cls not used
 251  * @param tc task context
 252  */
 253 static void
 254 delete_expired (void *cls,
 255                 const struct GNUNET_SCHEDULER_TaskContext *tc);
 256
 257
 258 /**
 259  * Iterate over the expired items stored in the datastore.
 260  * Delete all expired items; once we have processed all
 261  * expired items, re-schedule the "delete_expired" task.
 262  *
 263  * @param cls not used
 264  * @param next_cls closure to pass to the "next" function.
 265  * @param key key for the content
 266  * @param size number of bytes in data
 267  * @param data content stored
 268  * @param type type of the content
 269  * @param priority priority of the content
 270  * @param anonymity anonymity-level for the content
 271  * @param expiration expiration time for the content
 272  * @param uid unique identifier for the datum;
 273  *        maybe 0 if no unique identifier is available
 274  *
 275  * @return GNUNET_SYSERR to abort the iteration, GNUNET_OK to continue
 276  *         (continue on call to "next", of course),
 277  *         GNUNET_NO to delete the item and continue (if supported)
 278  */
 279 static int
 280 expired_processor (void *cls,
 281                    void *next_cls,
 282                    const GNUNET_HashCode * key,
 283                    uint32_t size,
 284                    const void *data,
 285                    enum GNUNET_BLOCK_Type type,
 286                    uint32_t priority,
 287                    uint32_t anonymity,
 288                    struct GNUNET_TIME_Absolute
 289                    expiration,
 290                    uint64_t uid)
 291 {
 292   struct GNUNET_TIME_Absolute now;
 293
 294   if (key == NULL)
 295     {
 296       expired_kill_task
 297         = GNUNET_SCHEDULER_add_delayed (sched,
 298                                         MAX_EXPIRE_DELAY,
 299                                         &delete_expired,
 300                                         NULL);
 301       return GNUNET_SYSERR;
 302     }
 303   now = GNUNET_TIME_absolute_get ();
 304   if (expiration.value > now.value)
 305     {
 306       /* finished processing */
 307       plugin->api->next_request (next_cls, GNUNET_YES);
 308       return GNUNET_SYSERR;
 309     }
 310   plugin->api->next_request (next_cls, GNUNET_NO);
 311 #if DEBUG_DATASTORE
 312   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 313               "Deleting content `%s' of type %u that expired %llu ms ago\n",
 314               GNUNET_h2s (key),
 315               type,
 316               (unsigned long long) (now.value - expiration.value));
 317 #endif
 318   GNUNET_STATISTICS_update (stats,
 319                             gettext_noop ("# bytes expired"),
 320                             size,
 321                             GNUNET_YES);
 322   GNUNET_CONTAINER_bloomfilter_remove (filter,
 323                                        key);
 324   return GNUNET_NO; /* delete */
 325 }
 326
 327
 328 /**
 329  * Task that is used to remove expired entries from
 330  * the datastore.  This task will schedule itself
 331  * again automatically to always delete all expired
 332  * content quickly.
 333  *
 334  * @param cls not used
 335  * @param tc task context
 336  */
 337 static void
 338 delete_expired (void *cls,
 339                 const struct GNUNET_SCHEDULER_TaskContext *tc)
 340 {
 341   expired_kill_task = GNUNET_SCHEDULER_NO_TASK;
 342   plugin->api->iter_ascending_expiration (plugin->api->cls,
 343                                           0,
 344                                           &expired_processor,
 345                                           NULL);
 346 }
 347
 348
 349 /**
 350  * An iterator over a set of items stored in the datastore.
 351  *
 352  * @param cls closure
 353  * @param next_cls closure to pass to the "next" function.
 354  * @param key key for the content
 355  * @param size number of bytes in data
 356  * @param data content stored
 357  * @param type type of the content
 358  * @param priority priority of the content
 359  * @param anonymity anonymity-level for the content
 360  * @param expiration expiration time for the content
 361  * @param uid unique identifier for the datum;
 362  *        maybe 0 if no unique identifier is available
 363  *
 364  * @return GNUNET_SYSERR to abort the iteration, GNUNET_OK to continue
 365  *         (continue on call to "next", of course),
 366  *         GNUNET_NO to delete the item and continue (if supported)
 367  */
 368 static int
 369 manage (void *cls,
 370         void *next_cls,
 371         const GNUNET_HashCode * key,
 372         uint32_t size,
 373         const void *data,
 374         enum GNUNET_BLOCK_Type type,
 375         uint32_t priority,
 376         uint32_t anonymity,
 377         struct GNUNET_TIME_Absolute
 378         expiration,
 379         uint64_t uid)
 380 {
 381   unsigned long long *need = cls;
 382
 383   if (NULL == key)
 384     {
 385       GNUNET_free (need);
 386       return GNUNET_SYSERR;
 387     }
 388   if (size + GNUNET_DATASTORE_ENTRY_OVERHEAD > *need)
 389     *need = 0;
 390   else
 391     *need -= size + GNUNET_DATASTORE_ENTRY_OVERHEAD;
 392   plugin->api->next_request (next_cls,
 393                              (0 == *need) ? GNUNET_YES : GNUNET_NO);
 394 #if DEBUG_DATASTORE
 395   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 396               "Deleting %llu bytes of low-priority content `%s' of type %u (still trying to free another %llu bytes)\n",
 397               (unsigned long long) (size + GNUNET_DATASTORE_ENTRY_OVERHEAD),
 398               GNUNET_h2s (key),
 399               type,
 400               *need);
 401 #endif
 402   GNUNET_STATISTICS_update (stats,
 403                             gettext_noop ("# bytes purged (low-priority)"),
 404                             size,
 405                             GNUNET_YES);
 406   GNUNET_CONTAINER_bloomfilter_remove (filter,
 407                                        key);
 408   return GNUNET_NO;
 409 }
 410
 411
 412 /**
 413  * Manage available disk space by running tasks
 414  * that will discard content if necessary.  This
 415  * function will be run whenever a request for
 416  * "need" bytes of storage could only be satisfied
 417  * by eating into the "cache" (and we want our cache
 418  * space back).
 419  *
 420  * @param need number of bytes of content that were
 421  *        placed into the "cache" (and hence the
 422  *        number of bytes that should be removed).
 423  */
 424 static void
 425 manage_space (unsigned long long need)
 426 {
 427   unsigned long long *n;
 428
 429 #if DEBUG_DATASTORE
 430   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 431               "Asked to free up %llu bytes of cache space\n",
 432               need);
 433 #endif
 434   n = GNUNET_malloc (sizeof(unsigned long long));
 435   *n = need;
 436   plugin->api->iter_low_priority (plugin->api->cls,
 437                                   0,
 438                                   &manage,
 439                                   n);
 440 }
 441
 442
 443 /**
 444  * Function called to notify a client about the socket
 445  * begin ready to queue more data.  "buf" will be
 446  * NULL and "size" zero if the socket was closed for
 447  * writing in the meantime.
 448  *
 449  * @param cls closure
 450  * @param size number of bytes available in buf
 451  * @param buf where the callee should write the message
 452  * @return number of bytes written to buf
 453  */
 454 static size_t
 455 transmit_callback (void *cls,
 456                    size_t size, void *buf)
 457 {
 458   struct TransmitCallbackContext *tcc = cls;
 459   size_t msize;
 460
 461   tcc->th = NULL;
 462   GNUNET_CONTAINER_DLL_remove (tcc_head,
 463                                tcc_tail,
 464                                tcc);
 465   msize = ntohs(tcc->msg->size);
 466   if (size == 0)
 467     {
 468 #if DEBUG_DATASTORE
 469       GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 470                   "Transmission failed.\n");
 471 #endif
 472       if (tcc->tc != NULL)
 473         tcc->tc (tcc->tc_cls, GNUNET_SYSERR);
 474       if (GNUNET_YES == tcc->end)
 475         GNUNET_SERVER_receive_done (tcc->client, GNUNET_SYSERR);
 476       GNUNET_SERVER_client_drop (tcc->client);
 477       GNUNET_free (tcc->msg);
 478       GNUNET_free (tcc);
 479       return 0;
 480     }
 481   GNUNET_assert (size >= msize);
 482   memcpy (buf, tcc->msg, msize);
 483   if (tcc->tc != NULL)
 484     tcc->tc (tcc->tc_cls, GNUNET_OK);
 485   if (GNUNET_YES == tcc->end)
 486     {
 487       GNUNET_SERVER_receive_done (tcc->client, GNUNET_OK);
 488     }
 489   else
 490     {
 491 #if DEBUG_DATASTORE
 492       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 493                   "Response transmitted, more pending!\n");
 494 #endif
 495     }
 496   GNUNET_SERVER_client_drop (tcc->client);
 497   GNUNET_free (tcc->msg);
 498   GNUNET_free (tcc);
 499   return msize;
 500 }
 501
 502
 503 /**
 504  * Transmit the given message to the client.
 505  *
 506  * @param client target of the message
 507  * @param msg message to transmit, will be freed!
 508  * @param tc function to call afterwards
 509  * @param tc_cls closure for tc
 510  * @param end is this the last response (and we should
 511  *        signal the server completion accodingly after
 512  *        transmitting this message)?
 513  */
 514 static void
 515 transmit (struct GNUNET_SERVER_Client *client,
 516           struct GNUNET_MessageHeader *msg,
 517           TransmitContinuation tc,
 518           void *tc_cls,
 519           int end)
 520 {
 521   struct TransmitCallbackContext *tcc;
 522
 523   if (GNUNET_YES == cleaning_done)
 524     {
 525 #if DEBUG_DATASTORE
 526       GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 527                   "Shutdown in progress, aborting transmission.\n");
 528 #endif
 529       GNUNET_free (msg);
 530       if (NULL != tc)
 531         tc (tc_cls, GNUNET_SYSERR);
 532       return;
 533     }
 534   tcc = GNUNET_malloc (sizeof(struct TransmitCallbackContext));
 535   tcc->msg = msg;
 536   tcc->client = client;
 537   tcc->tc = tc;
 538   tcc->tc_cls = tc_cls;
 539   tcc->end = end;
 540   if (NULL ==
 541       (tcc->th = GNUNET_SERVER_notify_transmit_ready (client,
 542                                                       ntohs(msg->size),
 543                                                       GNUNET_TIME_UNIT_FOREVER_REL,
 544                                                       &transmit_callback,
 545                                                       tcc)))
 546     {
 547       GNUNET_break (0);
 548       if (GNUNET_YES == end)
 549         {
 550 #if DEBUG_DATASTORE
 551           GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 552                       "Disconnecting client.\n");
 553 #endif
 554           GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
 555         }
 556       if (NULL != tc)
 557         tc (tc_cls, GNUNET_SYSERR);
 558       GNUNET_free (msg);
 559       GNUNET_free (tcc);
 560       return;
 561     }
 562   GNUNET_SERVER_client_keep (client);
 563   GNUNET_CONTAINER_DLL_insert (tcc_head,
 564                                tcc_tail,
 565                                tcc);
 566 }
 567
 568
 569 /**
 570  * Transmit a status code to the client.
 571  *
 572  * @param client receiver of the response
 573  * @param code status code
 574  * @param msg optional error message (can be NULL)
 575  */
 576 static void
 577 transmit_status (struct GNUNET_SERVER_Client *client,
 578                  int code,
 579                  const char *msg)
 580 {
 581   struct StatusMessage *sm;
 582   size_t slen;
 583
 584 #if DEBUG_DATASTORE
 585   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 586               "Transmitting `%s' message with value %d and message `%s'\n",
 587               "STATUS",
 588               code,
 589               msg != NULL ? msg : "(none)");
 590 #endif
 591   slen = (msg == NULL) ? 0 : strlen(msg) + 1;
 592   sm = GNUNET_malloc (sizeof(struct StatusMessage) + slen);
 593   sm->header.size = htons(sizeof(struct StatusMessage) + slen);
 594   sm->header.type = htons(GNUNET_MESSAGE_TYPE_DATASTORE_STATUS);
 595   sm->status = htonl(code);
 596   if (slen > 0)
 597     memcpy (&sm[1], msg, slen);
 598   transmit (client, &sm->header, NULL, NULL, GNUNET_YES);
 599 }
 600
 601
 602 /**
 603  * Function called once the transmit operation has
 604  * either failed or succeeded.
 605  *
 606  * @param next_cls closure for calling "next_request" callback
 607  * @param status GNUNET_OK on success, GNUNET_SYSERR on error
 608  */
 609 static void
 610 get_next(void *next_cls,
 611          int status)
 612 {
 613   if (status != GNUNET_OK)
 614     {
 615       GNUNET_log (GNUNET_ERROR_TYPE_INFO,
 616                   _("Failed to transmit an item to the client; aborting iteration.\n"));
 617       if (plugin != NULL)
 618         plugin->api->next_request (next_cls, GNUNET_YES);
 619       return;
 620     }
 621   plugin->api->next_request (next_cls, GNUNET_NO);
 622 }
 623
 624
 625 /**
 626  * Function that will transmit the given datastore entry
 627  * to the client.
 628  *
 629  * @param cls closure, pointer to the client (of type GNUNET_SERVER_Client).
 630  * @param next_cls closure to use to ask for the next item
 631  * @param key key for the content
 632  * @param size number of bytes in data
 633  * @param data content stored
 634  * @param type type of the content
 635  * @param priority priority of the content
 636  * @param anonymity anonymity-level for the content
 637  * @param expiration expiration time for the content
 638  * @param uid unique identifier for the datum;
 639  *        maybe 0 if no unique identifier is available
 640  *
 641  * @return GNUNET_SYSERR to abort the iteration, GNUNET_OK to continue,
 642  *         GNUNET_NO to delete the item and continue (if supported)
 643  */
 644 static int
 645 transmit_item (void *cls,
 646                void *next_cls,
 647                const GNUNET_HashCode * key,
 648                uint32_t size,
 649                const void *data,
 650                enum GNUNET_BLOCK_Type type,
 651                uint32_t priority,
 652                uint32_t anonymity,
 653                struct GNUNET_TIME_Absolute
 654                expiration, uint64_t uid)
 655 {
 656   struct GNUNET_SERVER_Client *client = cls;
 657   struct GNUNET_MessageHeader *end;
 658   struct DataMessage *dm;
 659
 660   if (key == NULL)
 661     {
 662       /* transmit 'DATA_END' */
 663 #if DEBUG_DATASTORE
 664       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 665                   "Transmitting `%s' message\n",
 666                   "DATA_END");
 667 #endif
 668       end = GNUNET_malloc (sizeof(struct GNUNET_MessageHeader));
 669       end->size = htons(sizeof(struct GNUNET_MessageHeader));
 670       end->type = htons(GNUNET_MESSAGE_TYPE_DATASTORE_DATA_END);
 671       transmit (client, end, NULL, NULL, GNUNET_YES);
 672       GNUNET_SERVER_client_drop (client);
 673       return GNUNET_OK;
 674     }
 675   dm = GNUNET_malloc (sizeof(struct DataMessage) + size);
 676   dm->header.size = htons(sizeof(struct DataMessage) + size);
 677   dm->header.type = htons(GNUNET_MESSAGE_TYPE_DATASTORE_DATA);
 678   dm->rid = htonl(0);
 679   dm->size = htonl(size);
 680   dm->type = htonl(type);
 681   dm->priority = htonl(priority);
 682   dm->anonymity = htonl(anonymity);
 683   dm->expiration = GNUNET_TIME_absolute_hton(expiration);
 684   dm->uid = GNUNET_htonll(uid);
 685   dm->key = *key;
 686   memcpy (&dm[1], data, size);
 687 #if DEBUG_DATASTORE
 688   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 689               "Transmitting `%s' message for `%s' of type %u\n",
 690               "DATA",
 691               GNUNET_h2s (key),
 692               type);
 693 #endif
 694   GNUNET_STATISTICS_update (stats,
 695                             gettext_noop ("# results found"),
 696                             1,
 697                             GNUNET_NO);
 698   transmit (client, &dm->header, &get_next, next_cls, GNUNET_NO);
 699   return GNUNET_OK;
 700 }
 701
 702
 703 /**
 704  * Handle RESERVE-message.
 705  *
 706  * @param cls closure
 707  * @param client identification of the client
 708  * @param message the actual message
 709  */
 710 static void
 711 handle_reserve (void *cls,
 712                 struct GNUNET_SERVER_Client *client,
 713                 const struct GNUNET_MessageHeader *message)
 714 {
 715   /**
 716    * Static counter to produce reservation identifiers.
 717    */
 718   static int reservation_gen;
 719
 720   const struct ReserveMessage *msg = (const struct ReserveMessage*) message;
 721   struct ReservationList *e;
 722   unsigned long long used;
 723   unsigned long long req;
 724   uint64_t amount;
 725   uint32_t entries;
 726
 727 #if DEBUG_DATASTORE
 728   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 729               "Processing `%s' request\n",
 730               "RESERVE");
 731 #endif
 732   amount = GNUNET_ntohll(msg->amount);
 733   entries = ntohl(msg->entries);
 734   used = plugin->api->get_size (plugin->api->cls) + reserved;
 735   req = amount + ((unsigned long long) GNUNET_DATASTORE_ENTRY_OVERHEAD) * entries;
 736   if (used + req > quota)
 737     {
 738       if (quota < used)
 739         used = quota; /* cheat a bit for error message (to avoid negative numbers) */
 740       GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 741                   _("Insufficient space (%llu bytes are available) to satisfy `%s' request for %llu bytes\n"),
 742                   quota - used,
 743                   "RESERVE",
 744                   req);
 745       if (cache_size < req)
 746         {
 747           /* TODO: document this in the FAQ; essentially, if this
 748              message happens, the insertion request could be blocked
 749              by less-important content from migration because it is
 750              larger than 1/8th of the overall available space, and
 751              we only reserve 1/8th for "fresh" insertions */
 752           GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 753                       _("The requested amount (%llu bytes) is larger than the cache size (%llu bytes)\n"),
 754                       req,
 755                       cache_size);
 756           transmit_status (client, 0,
 757                            gettext_noop ("Insufficient space to satisfy request and "
 758                                          "requested amount is larger than cache size"));
 759         }
 760       else
 761         {
 762           transmit_status (client, 0,
 763                            gettext_noop ("Insufficient space to satisfy request"));
 764         }
 765       return;
 766     }
 767   reserved += req;
 768   GNUNET_STATISTICS_set (stats,
 769                          gettext_noop ("# reserved"),
 770                          reserved,
 771                          GNUNET_NO);
 772   e = GNUNET_malloc (sizeof(struct ReservationList));
 773   e->next = reservations;
 774   reservations = e;
 775   e->client = client;
 776   e->amount = amount;
 777   e->entries = entries;
 778   e->rid = ++reservation_gen;
 779   if (reservation_gen < 0)
 780     reservation_gen = 0; /* wrap around */
 781   transmit_status (client, e->rid, NULL);
 782 }
 783
 784
 785 /**
 786  * Handle RELEASE_RESERVE-message.
 787  *
 788  * @param cls closure
 789  * @param client identification of the client
 790  * @param message the actual message
 791  */
 792 static void
 793 handle_release_reserve (void *cls,
 794                         struct GNUNET_SERVER_Client *client,
 795                         const struct GNUNET_MessageHeader *message)
 796 {
 797   const struct ReleaseReserveMessage *msg = (const struct ReleaseReserveMessage*) message;
 798   struct ReservationList *pos;
 799   struct ReservationList *prev;
 800   struct ReservationList *next;
 801   int rid = ntohl(msg->rid);
 802   unsigned long long rem;
 803
 804 #if DEBUG_DATASTORE
 805   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 806               "Processing `%s' request\n",
 807               "RELEASE_RESERVE");
 808 #endif
 809   next = reservations;
 810   prev = NULL;
 811   while (NULL != (pos = next))
 812     {
 813       next = pos->next;
 814       if (rid == pos->rid)
 815         {
 816           if (prev == NULL)
 817             reservations = next;
 818           else
 819             prev->next = next;
 820           rem = pos->amount + ((unsigned long long) GNUNET_DATASTORE_ENTRY_OVERHEAD) * pos->entries;
 821           GNUNET_assert (reserved >= rem);
 822           reserved -= rem;
 823           GNUNET_STATISTICS_set (stats,
 824                          gettext_noop ("# reserved"),
 825                                  reserved,
 826                                  GNUNET_NO);
 827 #if DEBUG_DATASTORE
 828           GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 829                       "Returning %llu remaining reserved bytes to storage pool\n",
 830                       rem);
 831 #endif
 832           GNUNET_free (pos);
 833           transmit_status (client, GNUNET_OK, NULL);
 834           return;
 835         }
 836       prev = pos;
 837     }
 838   GNUNET_break (0);
 839   transmit_status (client, GNUNET_SYSERR, gettext_noop ("Could not find matching reservation"));
 840 }
 841
 842
 843 /**
 844  * Check that the given message is a valid data message.
 845  *
 846  * @return NULL if the message is not well-formed, otherwise the message
 847  */
 848 static const struct DataMessage *
 849 check_data (const struct GNUNET_MessageHeader *message)
 850 {
 851   uint16_t size;
 852   uint32_t dsize;
 853   const struct DataMessage *dm;
 854
 855   size = ntohs(message->size);
 856   if (size < sizeof(struct DataMessage))
 857     {
 858       GNUNET_break (0);
 859       return NULL;
 860     }
 861   dm = (const struct DataMessage *) message;
 862   dsize = ntohl(dm->size);
 863   if (size != dsize + sizeof(struct DataMessage))
 864     {
 865       GNUNET_break (0);
 866       return NULL;
 867     }
 868   return dm;
 869 }
 870
 871
 872 /**
 873  * Handle PUT-message.
 874  *
 875  * @param cls closure
 876  * @param client identification of the client
 877  * @param message the actual message
 878  */
 879 static void
 880 handle_put (void *cls,
 881             struct GNUNET_SERVER_Client *client,
 882             const struct GNUNET_MessageHeader *message)
 883 {
 884   const struct DataMessage *dm = check_data (message);
 885   char *msg;
 886   int ret;
 887   int rid;
 888   struct ReservationList *pos;
 889   uint32_t size;
 890
 891   if ( (dm == NULL) ||
 892        (ntohl(dm->type) == 0) )
 893     {
 894       GNUNET_break (0);
 895       GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
 896       return;
 897     }
 898 #if DEBUG_DATASTORE
 899   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 900               "Processing `%s' request for `%s' of type %u\n",
 901               "PUT",
 902               GNUNET_h2s (&dm->key),
 903               ntohl (dm->type));
 904 #endif
 905   rid = ntohl(dm->rid);
 906   size = ntohl(dm->size);
 907   if (rid > 0)
 908     {
 909       pos = reservations;
 910       while ( (NULL != pos) &&
 911               (rid != pos->rid) )
 912         pos = pos->next;
 913       GNUNET_break (pos != NULL);
 914       if (NULL != pos)
 915         {
 916           GNUNET_break (pos->entries > 0);
 917           GNUNET_break (pos->amount > size);
 918           pos->entries--;
 919           pos->amount -= size;
 920           reserved -= (size + GNUNET_DATASTORE_ENTRY_OVERHEAD);
 921           GNUNET_STATISTICS_set (stats,
 922                                  gettext_noop ("# reserved"),
 923                                  reserved,
 924                                  GNUNET_NO);
 925         }
 926     }
 927   msg = NULL;
 928   ret = plugin->api->put (plugin->api->cls,
 929                           &dm->key,
 930                           size,
 931                           &dm[1],
 932                           ntohl(dm->type),
 933                           ntohl(dm->priority),
 934                           ntohl(dm->anonymity),
 935                           GNUNET_TIME_absolute_ntoh(dm->expiration),
 936                           &msg);
 937   if (GNUNET_OK == ret)
 938     {
 939       GNUNET_STATISTICS_update (stats,
 940                                 gettext_noop ("# bytes stored"),
 941                                 size,
 942                                 GNUNET_YES);
 943       GNUNET_CONTAINER_bloomfilter_add (filter,
 944                                         &dm->key);
 945 #if DEBUG_DATASTORE
 946       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 947                   "Successfully stored %u bytes of type %u under key `%s'\n",
 948                   size,
 949                   ntohl(dm->type),
 950                   GNUNET_h2s (&dm->key));
 951 #endif
 952     }
 953   transmit_status (client,
 954                    (GNUNET_SYSERR == ret) ? GNUNET_SYSERR : GNUNET_OK,
 955                    msg);
 956   GNUNET_free_non_null (msg);
 957   if (quota - reserved - cache_size < plugin->api->get_size (plugin->api->cls))
 958     {
 959       GNUNET_log (GNUNET_ERROR_TYPE_INFO,
 960                   _("Need %llu bytes more space (%llu allowed, using %llu)\n"),
 961                   (unsigned long long) size + GNUNET_DATASTORE_ENTRY_OVERHEAD,
 962                   (unsigned long long) (quota - reserved - cache_size),
 963                   (unsigned long long) plugin->api->get_size (plugin->api->cls));
 964       manage_space (size + GNUNET_DATASTORE_ENTRY_OVERHEAD);
 965     }
 966 }
 967
 968
 969 /**
 970  * Handle GET-message.
 971  *
 972  * @param cls closure
 973  * @param client identification of the client
 974  * @param message the actual message
 975  */
 976 static void
 977 handle_get (void *cls,
 978             struct GNUNET_SERVER_Client *client,
 979             const struct GNUNET_MessageHeader *message)
 980 {
 981   const struct GetMessage *msg;
 982   uint16_t size;
 983
 984   size = ntohs(message->size);
 985   if ( (size != sizeof(struct GetMessage)) &&
 986        (size != sizeof(struct GetMessage) - sizeof(GNUNET_HashCode)) )
 987     {
 988       GNUNET_break (0);
 989       GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
 990       return;
 991     }
 992   msg = (const struct GetMessage*) message;
 993 #if DEBUG_DATASTORE
 994   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 995               "Processing `%s' request for `%s' of type %u\n",
 996               "GET",
 997               GNUNET_h2s (&msg->key),
 998               ntohl (msg->type));
 999 #endif
1000   GNUNET_STATISTICS_update (stats,
1001                             gettext_noop ("# GET requests received"),
1002                             1,
1003                             GNUNET_NO);
1004   GNUNET_SERVER_client_keep (client);
1005   if ( (size == sizeof(struct GetMessage)) &&
1006        (GNUNET_YES != GNUNET_CONTAINER_bloomfilter_test (filter,
1007                                                          &msg->key)) )
1008     {
1009       /* don't bother database... */
1010 #if DEBUG_DATASTORE
1011       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1012                   "Empty result set for `%s' request for `%s' (bloomfilter).\n",
1013                   "GET",
1014                   GNUNET_h2s (&msg->key));
1015 #endif
1016       GNUNET_STATISTICS_update (stats,
1017                                 gettext_noop ("# requests filtered by bloomfilter"),
1018                                 1,
1019                                 GNUNET_NO);
1020       transmit_item (client,
1021                      NULL, NULL, 0, NULL, 0, 0, 0,
1022                      GNUNET_TIME_UNIT_ZERO_ABS, 0);
1023       return;
1024     }
1025   plugin->api->get (plugin->api->cls,
1026                     ((size == sizeof(struct GetMessage)) ? &msg->key : NULL),
1027                     NULL,
1028                     ntohl(msg->type),
1029                     &transmit_item,
1030                     client);
1031 }
1032
1033
1034 /**
1035  * Handle UPDATE-message.
1036  *
1037  * @param cls closure
1038  * @param client identification of the client
1039  * @param message the actual message
1040  */
1041 static void
1042 handle_update (void *cls,
1043                struct GNUNET_SERVER_Client *client,
1044                const struct GNUNET_MessageHeader *message)
1045 {
1046   const struct UpdateMessage *msg;
1047   int ret;
1048   char *emsg;
1049
1050   GNUNET_STATISTICS_update (stats,
1051                             gettext_noop ("# UPDATE requests received"),
1052                             1,
1053                             GNUNET_NO);
1054   msg = (const struct UpdateMessage*) message;
1055   emsg = NULL;
1056 #if DEBUG_DATASTORE
1057   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1058               "Processing `%s' request for %llu\n",
1059               "UPDATE",
1060               (unsigned long long) GNUNET_ntohll (msg->uid));
1061 #endif
1062   ret = plugin->api->update (plugin->api->cls,
1063                              GNUNET_ntohll(msg->uid),
1064                              (int32_t) ntohl(msg->priority),
1065                              GNUNET_TIME_absolute_ntoh(msg->expiration),
1066                              &emsg);
1067   transmit_status (client, ret, emsg);
1068   GNUNET_free_non_null (emsg);
1069 }
1070
1071
1072 /**
1073  * Handle GET_RANDOM-message.
1074  *
1075  * @param cls closure
1076  * @param client identification of the client
1077  * @param message the actual message
1078  */
1079 static void
1080 handle_get_random (void *cls,
1081                    struct GNUNET_SERVER_Client *client,
1082                    const struct GNUNET_MessageHeader *message)
1083 {
1084 #if DEBUG_DATASTORE
1085   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1086               "Processing `%s' request\n",
1087               "GET_RANDOM");
1088 #endif
1089   GNUNET_STATISTICS_update (stats,
1090                             gettext_noop ("# GET RANDOM requests received"),
1091                             1,
1092                             GNUNET_NO);
1093   GNUNET_SERVER_client_keep (client);
1094   plugin->api->iter_migration_order (plugin->api->cls,
1095                                      0,
1096                                      &transmit_item,
1097                                      client);
1098 }
1099
1100
1101 /**
1102  * Context for the 'remove_callback'.
1103  */
1104 struct RemoveContext
1105 {
1106   /**
1107    * Client for whom we're doing the remvoing.
1108    */
1109   struct GNUNET_SERVER_Client *client;
1110
1111   /**
1112    * GNUNET_YES if we managed to remove something.
1113    */
1114   int found;
1115 };
1116
1117
1118 /**
1119  * Callback function that will cause the item that is passed
1120  * in to be deleted (by returning GNUNET_NO).
1121  */
1122 static int
1123 remove_callback (void *cls,
1124                  void *next_cls,
1125                  const GNUNET_HashCode * key,
1126                  uint32_t size,
1127                  const void *data,
1128                  enum GNUNET_BLOCK_Type type,
1129                  uint32_t priority,
1130                  uint32_t anonymity,
1131                  struct GNUNET_TIME_Absolute
1132                  expiration, uint64_t uid)
1133 {
1134   struct RemoveContext *rc = cls;
1135
1136   if (key == NULL)
1137     {
1138 #if DEBUG_DATASTORE
1139       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1140                   "No further matches for `%s' request.\n",
1141                   "REMOVE");
1142 #endif
1143       if (GNUNET_YES == rc->found)
1144         transmit_status (rc->client, GNUNET_OK, NULL);
1145       else
1146         transmit_status (rc->client, GNUNET_NO, _("Content not found"));
1147       GNUNET_SERVER_client_drop (rc->client);
1148       GNUNET_free (rc);
1149       return GNUNET_OK; /* last item */
1150     }
1151   rc->found = GNUNET_YES;
1152 #if DEBUG_DATASTORE
1153   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1154               "Item %llu matches `%s' request for key `%s' and type %u.\n",
1155               (unsigned long long) uid,
1156               "REMOVE",
1157               GNUNET_h2s (key),
1158               type);
1159 #endif
1160   GNUNET_STATISTICS_update (stats,
1161                             gettext_noop ("# bytes removed (explicit request)"),
1162                             size,
1163                             GNUNET_YES);
1164   GNUNET_CONTAINER_bloomfilter_remove (filter,
1165                                        key);
1166   plugin->api->next_request (next_cls, GNUNET_YES);
1167   return GNUNET_NO;
1168 }
1169
1170
1171 /**
1172  * Handle REMOVE-message.
1173  *
1174  * @param cls closure
1175  * @param client identification of the client
1176  * @param message the actual message
1177  */
1178 static void
1179 handle_remove (void *cls,
1180              struct GNUNET_SERVER_Client *client,
1181              const struct GNUNET_MessageHeader *message)
1182 {
1183   const struct DataMessage *dm = check_data (message);
1184   GNUNET_HashCode vhash;
1185   struct RemoveContext *rc;
1186
1187   if (dm == NULL)
1188     {
1189       GNUNET_break (0);
1190       GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
1191       return;
1192     }
1193 #if DEBUG_DATASTORE
1194   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1195               "Processing `%s' request for `%s' of type %u\n",
1196               "REMOVE",
1197               GNUNET_h2s (&dm->key),
1198               ntohl (dm->type));
1199 #endif
1200   GNUNET_STATISTICS_update (stats,
1201                             gettext_noop ("# REMOVE requests received"),
1202                             1,
1203                             GNUNET_NO);
1204   rc = GNUNET_malloc (sizeof(struct RemoveContext));
1205   GNUNET_SERVER_client_keep (client);
1206   rc->client = client;
1207   GNUNET_CRYPTO_hash (&dm[1],
1208                       ntohl(dm->size),
1209                       &vhash);
1210   plugin->api->get (plugin->api->cls,
1211                     &dm->key,
1212                     &vhash,
1213                     ntohl(dm->type),
1214                     &remove_callback,
1215                     rc);
1216 }
1217
1218
1219 /**
1220  * Handle DROP-message.
1221  *
1222  * @param cls closure
1223  * @param client identification of the client
1224  * @param message the actual message
1225  */
1226 static void
1227 handle_drop (void *cls,
1228              struct GNUNET_SERVER_Client *client,
1229              const struct GNUNET_MessageHeader *message)
1230 {
1231 #if DEBUG_DATASTORE
1232   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1233               "Processing `%s' request\n",
1234               "DROP");
1235 #endif
1236   plugin->api->drop (plugin->api->cls);
1237   GNUNET_SERVER_receive_done (client, GNUNET_OK);
1238 }
1239
1240
1241 /**
1242  * Load the datastore plugin.
1243  */
1244 static struct DatastorePlugin *
1245 load_plugin ()
1246 {
1247   struct DatastorePlugin *ret;
1248   char *libname;
1249   char *name;
1250
1251   if (GNUNET_OK !=
1252       GNUNET_CONFIGURATION_get_value_string (cfg,
1253                                              "DATASTORE", "DATABASE", &name))
1254     {
1255       GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
1256                   _("No `%s' specified for `%s' in configuration!\n"),
1257                   "DATABASE",
1258                   "DATASTORE");
1259       return NULL;
1260     }
1261   ret = GNUNET_malloc (sizeof(struct DatastorePlugin));
1262   ret->env.cfg = cfg;
1263   ret->env.sched = sched;
1264   GNUNET_log (GNUNET_ERROR_TYPE_INFO,
1265               _("Loading `%s' datastore plugin\n"), name);
1266   GNUNET_asprintf (&libname, "libgnunet_plugin_datastore_%s", name);
1267   ret->short_name = name;
1268   ret->lib_name = libname;
1269   ret->api = GNUNET_PLUGIN_load (libname, &ret->env);
1270   if (ret->api == NULL)
1271     {
1272       GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
1273                   _("Failed to load datastore plugin for `%s'\n"), name);
1274       GNUNET_free (ret->short_name);
1275       GNUNET_free (libname);
1276       GNUNET_free (ret);
1277       return NULL;
1278     }
1279   return ret;
1280 }
1281
1282
1283 /**
1284  * Function called when the service shuts
1285  * down.  Unloads our datastore plugin.
1286  *
1287  * @param plug plugin to unload
1288  */
1289 static void
1290 unload_plugin (struct DatastorePlugin *plug)
1291 {
1292 #if DEBUG_DATASTORE
1293   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1294               "Datastore service is unloading plugin...\n");
1295 #endif
1296   GNUNET_break (NULL == GNUNET_PLUGIN_unload (plug->lib_name, plug->api));
1297   GNUNET_free (plug->lib_name);
1298   GNUNET_free (plug->short_name);
1299   GNUNET_free (plug);
1300 }
1301
1302
1303 /**
1304  * Final task run after shutdown.  Unloads plugins and disconnects us from
1305  * statistics.
1306  */
1307 static void
1308 unload_task (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
1309 {
1310   unload_plugin (plugin);
1311   plugin = NULL;
1312   if (filter != NULL)
1313     {
1314       GNUNET_CONTAINER_bloomfilter_free (filter);
1315       filter = NULL;
1316     }
1317   if (stats != NULL)
1318     {
1319       GNUNET_STATISTICS_destroy (stats, GNUNET_YES);
1320       stats = NULL;
1321     }
1322 }
1323
1324
1325 /**
1326  * Last task run during shutdown.  Disconnects us from
1327  * the transport and core.
1328  */
1329 static void
1330 cleaning_task (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
1331 {
1332   struct TransmitCallbackContext *tcc;
1333
1334   cleaning_done = GNUNET_YES;
1335   while (NULL != (tcc = tcc_head))
1336     {
1337       GNUNET_CONTAINER_DLL_remove (tcc_head,
1338                                    tcc_tail,
1339                                    tcc);
1340       if (tcc->th != NULL)
1341         {
1342           GNUNET_CONNECTION_notify_transmit_ready_cancel (tcc->th);
1343           GNUNET_SERVER_client_drop (tcc->client);
1344         }
1345       if (NULL != tcc->tc)
1346         tcc->tc (tcc->tc_cls, GNUNET_SYSERR);
1347       GNUNET_free (tcc->msg);
1348       GNUNET_free (tcc);
1349     }
1350   if (expired_kill_task != GNUNET_SCHEDULER_NO_TASK)
1351     {
1352       GNUNET_SCHEDULER_cancel (sched,
1353                                expired_kill_task);
1354       expired_kill_task = GNUNET_SCHEDULER_NO_TASK;
1355     }
1356   GNUNET_SCHEDULER_add_continuation (sched,
1357                                      &unload_task,
1358                                      NULL,
1359                                      GNUNET_SCHEDULER_REASON_PREREQ_DONE);
1360 }
1361
1362
1363 /**
1364  * Function that removes all active reservations made
1365  * by the given client and releases the space for other
1366  * requests.
1367  *
1368  * @param cls closure
1369  * @param client identification of the client
1370  */
1371 static void
1372 cleanup_reservations (void *cls,
1373                       struct GNUNET_SERVER_Client
1374                       * client)
1375 {
1376   struct ReservationList *pos;
1377   struct ReservationList *prev;
1378   struct ReservationList *next;
1379
1380   if (client == NULL)
1381     return;
1382   prev = NULL;
1383   pos = reservations;
1384   while (NULL != pos)
1385     {
1386       next = pos->next;
1387       if (pos->client == client)
1388         {
1389           if (prev == NULL)
1390             reservations = next;
1391           else
1392             prev->next = next;
1393           reserved -= pos->amount + pos->entries * GNUNET_DATASTORE_ENTRY_OVERHEAD;
1394           GNUNET_free (pos);
1395         }
1396       else
1397         {
1398           prev = pos;
1399         }
1400       pos = next;
1401     }
1402   GNUNET_STATISTICS_set (stats,
1403                          gettext_noop ("# reserved"),
1404                          reserved,
1405                          GNUNET_NO);
1406 }
1407
1408
1409 /**
1410  * Process datastore requests.
1411  *
1412  * @param cls closure
1413  * @param s scheduler to use
1414  * @param server the initialized server
1415  * @param c configuration to use
1416  */
1417 static void
1418 run (void *cls,
1419      struct GNUNET_SCHEDULER_Handle *s,
1420      struct GNUNET_SERVER_Handle *server,
1421      const struct GNUNET_CONFIGURATION_Handle *c)
1422 {
1423   static const struct GNUNET_SERVER_MessageHandler handlers[] = {
1424     {&handle_reserve, NULL, GNUNET_MESSAGE_TYPE_DATASTORE_RESERVE,
1425      sizeof(struct ReserveMessage) },
1426     {&handle_release_reserve, NULL, GNUNET_MESSAGE_TYPE_DATASTORE_RELEASE_RESERVE,
1427      sizeof(struct ReleaseReserveMessage) },
1428     {&handle_put, NULL, GNUNET_MESSAGE_TYPE_DATASTORE_PUT, 0 },
1429     {&handle_update, NULL, GNUNET_MESSAGE_TYPE_DATASTORE_UPDATE,
1430      sizeof (struct UpdateMessage) },
1431     {&handle_get, NULL, GNUNET_MESSAGE_TYPE_DATASTORE_GET, 0 },
1432     {&handle_get_random, NULL, GNUNET_MESSAGE_TYPE_DATASTORE_GET_RANDOM,
1433      sizeof(struct GNUNET_MessageHeader) },
1434     {&handle_remove, NULL, GNUNET_MESSAGE_TYPE_DATASTORE_REMOVE, 0 },
1435     {&handle_drop, NULL, GNUNET_MESSAGE_TYPE_DATASTORE_DROP,
1436      sizeof(struct GNUNET_MessageHeader) },
1437     {NULL, NULL, 0, 0}
1438   };
1439   char *fn;
1440   unsigned int bf_size;
1441
1442   sched = s;
1443   cfg = c;
1444   if (GNUNET_OK !=
1445       GNUNET_CONFIGURATION_get_value_number (cfg,
1446                                              "DATASTORE", "QUOTA", &quota))
1447     {
1448       GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
1449                   _("No `%s' specified for `%s' in configuration!\n"),
1450                   "QUOTA",
1451                   "DATASTORE");
1452       return;
1453     }
1454   stats = GNUNET_STATISTICS_create (sched, "datastore", cfg);
1455   GNUNET_STATISTICS_set (stats,
1456                          gettext_noop ("# quota"),
1457                          quota,
1458                          GNUNET_NO);
1459   cache_size = quota / 8; /* Or should we make this an option? */
1460   GNUNET_STATISTICS_set (stats,
1461                          gettext_noop ("# cache size"),
1462                          cache_size,
1463                          GNUNET_NO);
1464   bf_size = quota / 32; /* 8 bit per entry, 1 bit per 32 kb in DB */
1465   fn = NULL;
1466   if ( (GNUNET_OK !=
1467         GNUNET_CONFIGURATION_get_value_filename (cfg,
1468                                                  "DATASTORE",
1469                                                  "BLOOMFILTER",
1470                                                  &fn)) ||
1471        (GNUNET_OK !=
1472         GNUNET_DISK_directory_create_for_file (fn)) )
1473     {
1474       GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
1475                   _("Could not use specified filename `%s' for bloomfilter.\n"),
1476                   fn != NULL ? fn : "");
1477       GNUNET_free_non_null (fn);
1478       fn = NULL;
1479     }
1480   filter = GNUNET_CONTAINER_bloomfilter_load (fn, bf_size, 5);  /* approx. 3% false positives at max use */
1481   GNUNET_free_non_null (fn);
1482   if (filter == NULL)
1483     {
1484       GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
1485                   _("Failed to initialize bloomfilter.\n"));
1486       if (stats != NULL)
1487         {
1488           GNUNET_STATISTICS_destroy (stats, GNUNET_YES);
1489           stats = NULL;
1490         }
1491       return;
1492     }
1493   plugin = load_plugin ();
1494   if (NULL == plugin)
1495     {
1496       GNUNET_CONTAINER_bloomfilter_free (filter);
1497       filter = NULL;
1498       if (stats != NULL)
1499         {
1500           GNUNET_STATISTICS_destroy (stats, GNUNET_YES);
1501           stats = NULL;
1502         }
1503       return;
1504     }
1505   GNUNET_SERVER_disconnect_notify (server, &cleanup_reservations, NULL);
1506   GNUNET_SERVER_add_handlers (server, handlers);
1507   expired_kill_task
1508     = GNUNET_SCHEDULER_add_with_priority (sched,
1509                                           GNUNET_SCHEDULER_PRIORITY_IDLE,
1510                                           &delete_expired, NULL);
1511   GNUNET_SCHEDULER_add_delayed (sched,
1512                                 GNUNET_TIME_UNIT_FOREVER_REL,
1513                                 &cleaning_task, NULL);
1514
1515 }
1516
1517
1518 /**
1519  * The main function for the datastore service.
1520  *
1521  * @param argc number of arguments from the command line
1522  * @param argv command line arguments
1523  * @return 0 ok, 1 on error
1524  */
1525 int
1526 main (int argc, char *const *argv)
1527 {
1528   int ret;
1529
1530   ret = (GNUNET_OK ==
1531          GNUNET_SERVICE_run (argc,
1532                              argv,
1533                              "datastore",
1534                              GNUNET_SERVICE_OPTION_NONE,
1535                              &run, NULL)) ? 0 : 1;
1536   return ret;
1537 }
1538
1539
1540 /* end of gnunet-service-datastore.c */