src/datastore/gnunet-service-datastore.c

   1 /*
   2      This file is part of GNUnet
   3      Copyright (C) 2004-2014 GNUnet e.V.
   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 3, 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., 51 Franklin Street, Fifth Floor,
  18      Boston, MA 02110-1301, 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 "gnunet_datastore_plugin.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  * How fast are we allowed to query the database for deleting
  47  * expired content? (1 item per second).
  48  */
  49 #define MIN_EXPIRE_DELAY GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 1)
  50
  51 /**
  52  * Name under which we store current space consumption.
  53  */
  54 static char *quota_stat_name;
  55
  56 /**
  57  * Task to timeout stat GET.
  58  */
  59 static struct GNUNET_SCHEDULER_Task *stat_timeout_task;
  60
  61 /**
  62  * After how many payload-changing operations
  63  * do we sync our statistics?
  64  */
  65 #define MAX_STAT_SYNC_LAG 50
  66
  67
  68 /**
  69  * Our datastore plugin.
  70  */
  71 struct DatastorePlugin
  72 {
  73
  74   /**
  75    * API of the transport as returned by the plugin's
  76    * initialization function.
  77    */
  78   struct GNUNET_DATASTORE_PluginFunctions *api;
  79
  80   /**
  81    * Short name for the plugin (i.e. "sqlite").
  82    */
  83   char *short_name;
  84
  85   /**
  86    * Name of the library (i.e. "gnunet_plugin_datastore_sqlite").
  87    */
  88   char *lib_name;
  89
  90   /**
  91    * Environment this transport service is using
  92    * for this plugin.
  93    */
  94   struct GNUNET_DATASTORE_PluginEnvironment env;
  95
  96 };
  97
  98
  99 /**
 100  * Linked list of active reservations.
 101  */
 102 struct ReservationList
 103 {
 104
 105   /**
 106    * This is a linked list.
 107    */
 108   struct ReservationList *next;
 109
 110   /**
 111    * Client that made the reservation.
 112    */
 113   struct GNUNET_SERVER_Client *client;
 114
 115   /**
 116    * Number of bytes (still) reserved.
 117    */
 118   uint64_t amount;
 119
 120   /**
 121    * Number of items (still) reserved.
 122    */
 123   uint64_t entries;
 124
 125   /**
 126    * Reservation identifier.
 127    */
 128   int32_t rid;
 129
 130 };
 131
 132
 133
 134 /**
 135  * Our datastore plugin (NULL if not available).
 136  */
 137 static struct DatastorePlugin *plugin;
 138
 139 /**
 140  * Linked list of space reservations made by clients.
 141  */
 142 static struct ReservationList *reservations;
 143
 144 /**
 145  * Bloomfilter to quickly tell if we don't have the content.
 146  */
 147 static struct GNUNET_CONTAINER_BloomFilter *filter;
 148
 149 /**
 150  * Name of our plugin.
 151  */
 152 static char *plugin_name;
 153
 154 /**
 155  * Our configuration.
 156  */
 157 static const struct GNUNET_CONFIGURATION_Handle *cfg;
 158
 159 /**
 160  * Handle for reporting statistics.
 161  */
 162 static struct GNUNET_STATISTICS_Handle *stats;
 163
 164 /**
 165  * How much space are we using for the cache?  (space available for
 166  * insertions that will be instantly reclaimed by discarding less
 167  * important content --- or possibly whatever we just inserted into
 168  * the "cache").
 169  */
 170 static unsigned long long cache_size;
 171
 172 /**
 173  * How much space have we currently reserved?
 174  */
 175 static unsigned long long reserved;
 176
 177 /**
 178  * How much data are we currently storing
 179  * in the database?
 180  */
 181 static unsigned long long payload;
 182
 183 /**
 184  * Identity of the task that is used to delete
 185  * expired content.
 186  */
 187 static struct GNUNET_SCHEDULER_Task * expired_kill_task;
 188
 189 /**
 190  * Minimum time that content should have to not be discarded instantly
 191  * (time stamp of any content that we've been discarding recently to
 192  * stay below the quota).  FOREVER if we had to expire content with
 193  * non-zero priority.
 194  */
 195 static struct GNUNET_TIME_Absolute min_expiration;
 196
 197 /**
 198  * How much space are we allowed to use?
 199  */
 200 static unsigned long long quota;
 201
 202 /**
 203  * Should the database be dropped on exit?
 204  */
 205 static int do_drop;
 206
 207 /**
 208  * Should we refresh the BF when the DB is loaded?
 209  */
 210 static int refresh_bf;
 211
 212 /**
 213  * Number of updates that were made to the
 214  * payload value since we last synchronized
 215  * it with the statistics service.
 216  */
 217 static unsigned int last_sync;
 218
 219 /**
 220  * Did we get an answer from statistics?
 221  */
 222 static int stats_worked;
 223
 224
 225 /**
 226  * Synchronize our utilization statistics with the
 227  * statistics service.
 228  */
 229 static void
 230 sync_stats ()
 231 {
 232   GNUNET_STATISTICS_set (stats,
 233                          quota_stat_name,
 234                          payload,
 235                          GNUNET_YES);
 236   GNUNET_STATISTICS_set (stats,
 237                          "# utilization by current datastore",
 238                          payload,
 239                          GNUNET_NO);
 240   last_sync = 0;
 241 }
 242
 243
 244 /**
 245  * Context for transmitting replies to clients.
 246  */
 247 struct TransmitCallbackContext
 248 {
 249
 250   /**
 251    * We keep these in a doubly-linked list (for cleanup).
 252    */
 253   struct TransmitCallbackContext *next;
 254
 255   /**
 256    * We keep these in a doubly-linked list (for cleanup).
 257    */
 258   struct TransmitCallbackContext *prev;
 259
 260   /**
 261    * The message that we're asked to transmit.
 262    */
 263   struct GNUNET_MessageHeader *msg;
 264
 265   /**
 266    * Handle for the transmission request.
 267    */
 268   struct GNUNET_SERVER_TransmitHandle *th;
 269
 270   /**
 271    * Client that we are transmitting to.
 272    */
 273   struct GNUNET_SERVER_Client *client;
 274
 275 };
 276
 277
 278 /**
 279  * Head of the doubly-linked list (for cleanup).
 280  */
 281 static struct TransmitCallbackContext *tcc_head;
 282
 283 /**
 284  * Tail of the doubly-linked list (for cleanup).
 285  */
 286 static struct TransmitCallbackContext *tcc_tail;
 287
 288 /**
 289  * Have we already cleaned up the TCCs and are hence no longer
 290  * willing (or able) to transmit anything to anyone?
 291  */
 292 static int cleaning_done;
 293
 294 /**
 295  * Handle for pending get request.
 296  */
 297 static struct GNUNET_STATISTICS_GetHandle *stat_get;
 298
 299 /**
 300  * Handle to our server.
 301  */
 302 static struct GNUNET_SERVER_Handle *server;
 303
 304 /**
 305  * Task that is used to remove expired entries from
 306  * the datastore.  This task will schedule itself
 307  * again automatically to always delete all expired
 308  * content quickly.
 309  *
 310  * @param cls not used
 311  */
 312 static void
 313 delete_expired (void *cls);
 314
 315
 316 /**
 317  * Iterate over the expired items stored in the datastore.
 318  * Delete all expired items; once we have processed all
 319  * expired items, re-schedule the "delete_expired" task.
 320  *
 321  * @param cls not used
 322  * @param key key for the content
 323  * @param size number of bytes in data
 324  * @param data content stored
 325  * @param type type of the content
 326  * @param priority priority of the content
 327  * @param anonymity anonymity-level for the content
 328  * @param expiration expiration time for the content
 329  * @param uid unique identifier for the datum;
 330  *        maybe 0 if no unique identifier is available
 331  *
 332  * @return #GNUNET_SYSERR to abort the iteration, #GNUNET_OK to continue
 333  *         (continue on call to "next", of course),
 334  *         #GNUNET_NO to delete the item and continue (if supported)
 335  */
 336 static int
 337 expired_processor (void *cls,
 338                    const struct GNUNET_HashCode *key,
 339                    uint32_t size,
 340                    const void *data,
 341                    enum GNUNET_BLOCK_Type type,
 342                    uint32_t priority,
 343                    uint32_t anonymity,
 344                    struct GNUNET_TIME_Absolute expiration,
 345                    uint64_t uid)
 346 {
 347   struct GNUNET_TIME_Absolute now;
 348
 349   if (key == NULL)
 350   {
 351     expired_kill_task =
 352         GNUNET_SCHEDULER_add_delayed_with_priority (MAX_EXPIRE_DELAY,
 353                                                     GNUNET_SCHEDULER_PRIORITY_IDLE,
 354                                                     &delete_expired, NULL);
 355     return GNUNET_SYSERR;
 356   }
 357   now = GNUNET_TIME_absolute_get ();
 358   if (expiration.abs_value_us > now.abs_value_us)
 359   {
 360     /* finished processing */
 361     expired_kill_task =
 362         GNUNET_SCHEDULER_add_delayed_with_priority (MAX_EXPIRE_DELAY,
 363                                                     GNUNET_SCHEDULER_PRIORITY_IDLE,
 364                                                     &delete_expired, NULL);
 365     return GNUNET_SYSERR;
 366   }
 367   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 368               "Deleting content `%s' of type %u that expired %s ago\n",
 369               GNUNET_h2s (key), type,
 370               GNUNET_STRINGS_relative_time_to_string (GNUNET_TIME_absolute_get_difference (expiration,
 371                                                                                            now),
 372                                                       GNUNET_YES));
 373   min_expiration = now;
 374   GNUNET_STATISTICS_update (stats,
 375                             gettext_noop ("# bytes expired"),
 376                             size,
 377                             GNUNET_YES);
 378   GNUNET_CONTAINER_bloomfilter_remove (filter, key);
 379   expired_kill_task =
 380       GNUNET_SCHEDULER_add_delayed_with_priority (MIN_EXPIRE_DELAY,
 381                                                   GNUNET_SCHEDULER_PRIORITY_IDLE,
 382                                                   &delete_expired, NULL);
 383   return GNUNET_NO;
 384 }
 385
 386
 387 /**
 388  * Task that is used to remove expired entries from
 389  * the datastore.  This task will schedule itself
 390  * again automatically to always delete all expired
 391  * content quickly.
 392  *
 393  * @param cls not used
 394  */
 395 static void
 396 delete_expired (void *cls)
 397 {
 398   expired_kill_task = NULL;
 399   plugin->api->get_expiration (plugin->api->cls,
 400                                &expired_processor,
 401                                NULL);
 402 }
 403
 404
 405 /**
 406  * An iterator over a set of items stored in the datastore
 407  * that deletes until we're happy with respect to our quota.
 408  *
 409  * @param cls closure
 410  * @param key key for the content
 411  * @param size number of bytes in data
 412  * @param data content stored
 413  * @param type type of the content
 414  * @param priority priority of the content
 415  * @param anonymity anonymity-level for the content
 416  * @param expiration expiration time for the content
 417  * @param uid unique identifier for the datum;
 418  *        maybe 0 if no unique identifier is available
 419  * @return #GNUNET_SYSERR to abort the iteration, #GNUNET_OK to continue
 420  *         (continue on call to "next", of course),
 421  *         #GNUNET_NO to delete the item and continue (if supported)
 422  */
 423 static int
 424 quota_processor (void *cls,
 425                  const struct GNUNET_HashCode *key,
 426                  uint32_t size,
 427                  const void *data,
 428                  enum GNUNET_BLOCK_Type type,
 429                  uint32_t priority,
 430                  uint32_t anonymity,
 431                  struct GNUNET_TIME_Absolute expiration,
 432                  uint64_t uid)
 433 {
 434   unsigned long long *need = cls;
 435
 436   if (NULL == key)
 437     return GNUNET_SYSERR;
 438   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 439               "Deleting %llu bytes of low-priority (%u) content `%s' of type %u at %s prior to expiration (still trying to free another %llu bytes)\n",
 440               (unsigned long long) (size + GNUNET_DATASTORE_ENTRY_OVERHEAD),
 441               (unsigned int) priority,
 442               GNUNET_h2s (key), type,
 443               GNUNET_STRINGS_relative_time_to_string (GNUNET_TIME_absolute_get_remaining (expiration),
 444                                                       GNUNET_YES),
 445               *need);
 446   if (size + GNUNET_DATASTORE_ENTRY_OVERHEAD > *need)
 447     *need = 0;
 448   else
 449     *need -= size + GNUNET_DATASTORE_ENTRY_OVERHEAD;
 450   if (priority > 0)
 451     min_expiration = GNUNET_TIME_UNIT_FOREVER_ABS;
 452   else
 453     min_expiration = expiration;
 454   GNUNET_STATISTICS_update (stats,
 455                             gettext_noop ("# bytes purged (low-priority)"),
 456                             size, GNUNET_YES);
 457   GNUNET_CONTAINER_bloomfilter_remove (filter, key);
 458   return GNUNET_NO;
 459 }
 460
 461
 462 /**
 463  * Manage available disk space by running tasks
 464  * that will discard content if necessary.  This
 465  * function will be run whenever a request for
 466  * "need" bytes of storage could only be satisfied
 467  * by eating into the "cache" (and we want our cache
 468  * space back).
 469  *
 470  * @param need number of bytes of content that were
 471  *        placed into the "cache" (and hence the
 472  *        number of bytes that should be removed).
 473  */
 474 static void
 475 manage_space (unsigned long long need)
 476 {
 477   unsigned long long last;
 478
 479   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 480               "Asked to free up %llu bytes of cache space\n",
 481               need);
 482   last = 0;
 483   while ((need > 0) && (last != need))
 484   {
 485     last = need;
 486     plugin->api->get_expiration (plugin->api->cls,
 487                                  &quota_processor,
 488                                  &need);
 489   }
 490 }
 491
 492
 493 /**
 494  * Function called to notify a client about the socket
 495  * begin ready to queue more data.  "buf" will be
 496  * NULL and "size" zero if the socket was closed for
 497  * writing in the meantime.
 498  *
 499  * @param cls closure
 500  * @param size number of bytes available in buf
 501  * @param buf where the callee should write the message
 502  * @return number of bytes written to buf
 503  */
 504 static size_t
 505 transmit_callback (void *cls, size_t size, void *buf)
 506 {
 507   struct TransmitCallbackContext *tcc = cls;
 508   size_t msize;
 509
 510   tcc->th = NULL;
 511   GNUNET_CONTAINER_DLL_remove (tcc_head, tcc_tail, tcc);
 512   msize = ntohs (tcc->msg->size);
 513   if (size == 0)
 514   {
 515     GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 516                 _("Transmission to client failed!\n"));
 517     GNUNET_SERVER_receive_done (tcc->client, GNUNET_SYSERR);
 518     GNUNET_SERVER_client_drop (tcc->client);
 519     GNUNET_free (tcc->msg);
 520     GNUNET_free (tcc);
 521     return 0;
 522   }
 523   GNUNET_assert (size >= msize);
 524   memcpy (buf, tcc->msg, msize);
 525   GNUNET_SERVER_receive_done (tcc->client, GNUNET_OK);
 526   GNUNET_SERVER_client_drop (tcc->client);
 527   GNUNET_free (tcc->msg);
 528   GNUNET_free (tcc);
 529   return msize;
 530 }
 531
 532
 533 /**
 534  * Transmit the given message to the client.
 535  *
 536  * @param client target of the message
 537  * @param msg message to transmit, will be freed!
 538  */
 539 static void
 540 transmit (struct GNUNET_SERVER_Client *client, struct GNUNET_MessageHeader *msg)
 541 {
 542   struct TransmitCallbackContext *tcc;
 543
 544   if (GNUNET_YES == cleaning_done)
 545   {
 546     GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 547                 _("Shutdown in progress, aborting transmission.\n"));
 548     GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
 549     GNUNET_free (msg);
 550     return;
 551   }
 552   tcc = GNUNET_new (struct TransmitCallbackContext);
 553   tcc->msg = msg;
 554   tcc->client = client;
 555   if (NULL ==
 556       (tcc->th =
 557        GNUNET_SERVER_notify_transmit_ready (client, ntohs (msg->size),
 558                                             GNUNET_TIME_UNIT_FOREVER_REL,
 559                                             &transmit_callback, tcc)))
 560   {
 561     GNUNET_break (0);
 562     GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
 563     GNUNET_free (msg);
 564     GNUNET_free (tcc);
 565     return;
 566   }
 567   GNUNET_SERVER_client_keep (client);
 568   GNUNET_CONTAINER_DLL_insert (tcc_head, tcc_tail, tcc);
 569 }
 570
 571
 572 /**
 573  * Transmit a status code to the client.
 574  *
 575  * @param client receiver of the response
 576  * @param code status code
 577  * @param msg optional error message (can be NULL)
 578  */
 579 static void
 580 transmit_status (struct GNUNET_SERVER_Client *client, int code, const char *msg)
 581 {
 582   struct StatusMessage *sm;
 583   size_t slen;
 584
 585   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 586               "Transmitting `%s' message with value %d and message `%s'\n",
 587               "STATUS", code, msg != NULL ? msg : "(none)");
 588   slen = (msg == NULL) ? 0 : strlen (msg) + 1;
 589   sm = GNUNET_malloc (sizeof (struct StatusMessage) + slen);
 590   sm->header.size = htons (sizeof (struct StatusMessage) + slen);
 591   sm->header.type = htons (GNUNET_MESSAGE_TYPE_DATASTORE_STATUS);
 592   sm->status = htonl (code);
 593   sm->min_expiration = GNUNET_TIME_absolute_hton (min_expiration);
 594   if (slen > 0)
 595     memcpy (&sm[1], msg, slen);
 596   transmit (client, &sm->header);
 597 }
 598
 599
 600 /**
 601  * Function that will transmit the given datastore entry
 602  * to the client.
 603  *
 604  * @param cls closure, pointer to the client (of type GNUNET_SERVER_Client).
 605  * @param key key for the content
 606  * @param size number of bytes in data
 607  * @param data content stored
 608  * @param type type of the content
 609  * @param priority priority of the content
 610  * @param anonymity anonymity-level for the content
 611  * @param expiration expiration time for the content
 612  * @param uid unique identifier for the datum;
 613  *        maybe 0 if no unique identifier is available
 614  *
 615  * @return GNUNET_SYSERR to abort the iteration, GNUNET_OK to continue,
 616  *         GNUNET_NO to delete the item and continue (if supported)
 617  */
 618 static int
 619 transmit_item (void *cls, const struct GNUNET_HashCode * key, uint32_t size,
 620                const void *data, enum GNUNET_BLOCK_Type type, uint32_t priority,
 621                uint32_t anonymity, struct GNUNET_TIME_Absolute expiration,
 622                uint64_t uid)
 623 {
 624   struct GNUNET_SERVER_Client *client = cls;
 625   struct GNUNET_MessageHeader *end;
 626   struct DataMessage *dm;
 627
 628   if (key == NULL)
 629   {
 630     /* transmit 'DATA_END' */
 631     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Transmitting `%s' message\n",
 632                 "DATA_END");
 633     end = GNUNET_new (struct GNUNET_MessageHeader);
 634     end->size = htons (sizeof (struct GNUNET_MessageHeader));
 635     end->type = htons (GNUNET_MESSAGE_TYPE_DATASTORE_DATA_END);
 636     transmit (client, end);
 637     GNUNET_SERVER_client_drop (client);
 638     return GNUNET_OK;
 639   }
 640   GNUNET_assert (sizeof (struct DataMessage) + size <
 641                  GNUNET_SERVER_MAX_MESSAGE_SIZE);
 642   dm = GNUNET_malloc (sizeof (struct DataMessage) + size);
 643   dm->header.size = htons (sizeof (struct DataMessage) + size);
 644   dm->header.type = htons (GNUNET_MESSAGE_TYPE_DATASTORE_DATA);
 645   dm->rid = htonl (0);
 646   dm->size = htonl (size);
 647   dm->type = htonl (type);
 648   dm->priority = htonl (priority);
 649   dm->anonymity = htonl (anonymity);
 650   dm->replication = htonl (0);
 651   dm->reserved = htonl (0);
 652   dm->expiration = GNUNET_TIME_absolute_hton (expiration);
 653   dm->uid = GNUNET_htonll (uid);
 654   dm->key = *key;
 655   memcpy (&dm[1], data, size);
 656   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 657               "Transmitting `%s' message for `%s' of type %u with expiration %s (in: %s)\n",
 658               "DATA", GNUNET_h2s (key), type,
 659               GNUNET_STRINGS_absolute_time_to_string (expiration),
 660               GNUNET_STRINGS_relative_time_to_string (GNUNET_TIME_absolute_get_remaining (expiration),
 661                                                       GNUNET_YES));
 662   GNUNET_STATISTICS_update (stats,
 663                             gettext_noop ("# results found"),
 664                             1,
 665                             GNUNET_NO);
 666   transmit (client, &dm->header);
 667   GNUNET_SERVER_client_drop (client);
 668   return GNUNET_OK;
 669 }
 670
 671
 672 /**
 673  * Handle RESERVE-message.
 674  *
 675  * @param cls closure
 676  * @param client identification of the client
 677  * @param message the actual message
 678  */
 679 static void
 680 handle_reserve (void *cls, struct GNUNET_SERVER_Client *client,
 681                 const struct GNUNET_MessageHeader *message)
 682 {
 683   /**
 684    * Static counter to produce reservation identifiers.
 685    */
 686   static int reservation_gen;
 687
 688   const struct ReserveMessage *msg = (const struct ReserveMessage *) message;
 689   struct ReservationList *e;
 690   unsigned long long used;
 691   unsigned long long req;
 692   uint64_t amount;
 693   uint32_t entries;
 694
 695   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 696               "Processing RESERVE request\n");
 697   amount = GNUNET_ntohll (msg->amount);
 698   entries = ntohl (msg->entries);
 699   used = payload + reserved;
 700   req =
 701       amount + ((unsigned long long) GNUNET_DATASTORE_ENTRY_OVERHEAD) * entries;
 702   if (used + req > quota)
 703   {
 704     if (quota < used)
 705       used = quota;             /* cheat a bit for error message (to avoid negative numbers) */
 706     GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 707                 _
 708                 ("Insufficient space (%llu bytes are available) to satisfy `%s' request for %llu bytes\n"),
 709                 quota - used, "RESERVE", req);
 710     if (cache_size < req)
 711     {
 712       /* TODO: document this in the FAQ; essentially, if this
 713        * message happens, the insertion request could be blocked
 714        * by less-important content from migration because it is
 715        * larger than 1/8th of the overall available space, and
 716        * we only reserve 1/8th for "fresh" insertions */
 717       GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 718                   _
 719                   ("The requested amount (%llu bytes) is larger than the cache size (%llu bytes)\n"),
 720                   req, cache_size);
 721       transmit_status (client, 0,
 722                        gettext_noop
 723                        ("Insufficient space to satisfy request and "
 724                         "requested amount is larger than cache size"));
 725     }
 726     else
 727     {
 728       transmit_status (client, 0,
 729                        gettext_noop ("Insufficient space to satisfy request"));
 730     }
 731     return;
 732   }
 733   reserved += req;
 734   GNUNET_STATISTICS_set (stats,
 735                          gettext_noop ("# reserved"),
 736                          reserved,
 737                          GNUNET_NO);
 738   e = GNUNET_new (struct ReservationList);
 739   e->next = reservations;
 740   reservations = e;
 741   e->client = client;
 742   e->amount = amount;
 743   e->entries = entries;
 744   e->rid = ++reservation_gen;
 745   if (reservation_gen < 0)
 746     reservation_gen = 0;        /* wrap around */
 747   transmit_status (client, e->rid, NULL);
 748 }
 749
 750
 751 /**
 752  * Handle RELEASE_RESERVE-message.
 753  *
 754  * @param cls closure
 755  * @param client identification of the client
 756  * @param message the actual message
 757  */
 758 static void
 759 handle_release_reserve (void *cls,
 760                         struct GNUNET_SERVER_Client *client,
 761                         const struct GNUNET_MessageHeader *message)
 762 {
 763   const struct ReleaseReserveMessage *msg =
 764       (const struct ReleaseReserveMessage *) message;
 765   struct ReservationList *pos;
 766   struct ReservationList *prev;
 767   struct ReservationList *next;
 768   int rid = ntohl (msg->rid);
 769   unsigned long long rem;
 770
 771   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 772               "Processing RELEASE_RESERVE request\n");
 773   next = reservations;
 774   prev = NULL;
 775   while (NULL != (pos = next))
 776   {
 777     next = pos->next;
 778     if (rid == pos->rid)
 779     {
 780       if (prev == NULL)
 781         reservations = next;
 782       else
 783         prev->next = next;
 784       rem =
 785           pos->amount +
 786           ((unsigned long long) GNUNET_DATASTORE_ENTRY_OVERHEAD) * pos->entries;
 787       GNUNET_assert (reserved >= rem);
 788       reserved -= rem;
 789       GNUNET_STATISTICS_set (stats,
 790                              gettext_noop ("# reserved"),
 791                              reserved,
 792                              GNUNET_NO);
 793       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 794                   "Returning %llu remaining reserved bytes to storage pool\n",
 795                   rem);
 796       GNUNET_free (pos);
 797       transmit_status (client, GNUNET_OK, NULL);
 798       return;
 799     }
 800     prev = pos;
 801   }
 802   GNUNET_break (0);
 803   transmit_status (client, GNUNET_SYSERR,
 804                    gettext_noop ("Could not find matching reservation"));
 805 }
 806
 807
 808 /**
 809  * Check that the given message is a valid data message.
 810  *
 811  * @return NULL if the message is not well-formed, otherwise the message
 812  */
 813 static const struct DataMessage *
 814 check_data (const struct GNUNET_MessageHeader *message)
 815 {
 816   uint16_t size;
 817   uint32_t dsize;
 818   const struct DataMessage *dm;
 819
 820   size = ntohs (message->size);
 821   if (size < sizeof (struct DataMessage))
 822   {
 823     GNUNET_break (0);
 824     return NULL;
 825   }
 826   dm = (const struct DataMessage *) message;
 827   dsize = ntohl (dm->size);
 828   if (size != dsize + sizeof (struct DataMessage))
 829   {
 830     GNUNET_break (0);
 831     return NULL;
 832   }
 833   return dm;
 834 }
 835
 836
 837 /**
 838  * Context for a PUT request used to see if the content is
 839  * already present.
 840  */
 841 struct PutContext
 842 {
 843   /**
 844    * Client to notify on completion.
 845    */
 846   struct GNUNET_SERVER_Client *client;
 847
 848 #if ! HAVE_UNALIGNED_64_ACCESS
 849   void *reserved;
 850 #endif
 851
 852   /* followed by the 'struct DataMessage' */
 853 };
 854
 855
 856 /**
 857  * Put continuation.
 858  *
 859  * @param cls closure
 860  * @param key key for the item stored
 861  * @param size size of the item stored
 862  * @param status #GNUNET_OK or #GNUNET_SYSERROR
 863  * @param msg error message on error
 864  */
 865 static void
 866 put_continuation (void *cls,
 867                   const struct GNUNET_HashCode *key,
 868                   uint32_t size,
 869                   int status,
 870                   const char *msg)
 871 {
 872   struct PutContext *pc = cls;
 873
 874   if (GNUNET_OK == status)
 875   {
 876     GNUNET_STATISTICS_update (stats,
 877                               gettext_noop ("# bytes stored"),
 878                               size,
 879                               GNUNET_YES);
 880     GNUNET_CONTAINER_bloomfilter_add (filter, key);
 881     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 882                 "Successfully stored %u bytes under key `%s'\n",
 883                 size, GNUNET_h2s (key));
 884   }
 885   transmit_status (pc->client, status, msg);
 886   GNUNET_SERVER_client_drop (pc->client);
 887   GNUNET_free (pc);
 888   if (quota - reserved - cache_size < payload)
 889   {
 890     GNUNET_log (GNUNET_ERROR_TYPE_INFO,
 891                 _("Need %llu bytes more space (%llu allowed, using %llu)\n"),
 892                 (unsigned long long) size + GNUNET_DATASTORE_ENTRY_OVERHEAD,
 893                 (unsigned long long) (quota - reserved - cache_size),
 894                 (unsigned long long) payload);
 895     manage_space (size + GNUNET_DATASTORE_ENTRY_OVERHEAD);
 896   }
 897 }
 898
 899
 900 /**
 901  * Actually put the data message.
 902  *
 903  * @param pc put context
 904  */
 905 static void
 906 execute_put (struct PutContext *pc)
 907 {
 908   const struct DataMessage *dm;
 909
 910   dm = (const struct DataMessage *) &pc[1];
 911   plugin->api->put (plugin->api->cls, &dm->key, ntohl (dm->size), &dm[1],
 912                     ntohl (dm->type), ntohl (dm->priority),
 913                     ntohl (dm->anonymity), ntohl (dm->replication),
 914                     GNUNET_TIME_absolute_ntoh (dm->expiration),
 915                     &put_continuation, pc);
 916 }
 917
 918
 919 /**
 920  *
 921  * @param cls closure
 922  * @param status #GNUNET_OK or #GNUNET_SYSERR
 923  * @param msg error message on error
 924  */
 925 static void
 926 check_present_continuation (void *cls,
 927                             int status,
 928                             const char *msg)
 929 {
 930   struct GNUNET_SERVER_Client *client = cls;
 931
 932   transmit_status (client, GNUNET_NO, NULL);
 933   GNUNET_SERVER_client_drop (client);
 934 }
 935
 936
 937 /**
 938  * Function that will check if the given datastore entry
 939  * matches the put and if none match executes the put.
 940  *
 941  * @param cls closure, pointer to the client (of type `struct PutContext`).
 942  * @param key key for the content
 943  * @param size number of bytes in data
 944  * @param data content stored
 945  * @param type type of the content
 946  * @param priority priority of the content
 947  * @param anonymity anonymity-level for the content
 948  * @param expiration expiration time for the content
 949  * @param uid unique identifier for the datum;
 950  *        maybe 0 if no unique identifier is available
 951  * @return #GNUNET_OK usually
 952  *         #GNUNET_NO to delete the item
 953  */
 954 static int
 955 check_present (void *cls,
 956                const struct GNUNET_HashCode *key,
 957                uint32_t size,
 958                const void *data,
 959                enum GNUNET_BLOCK_Type type,
 960                uint32_t priority,
 961                uint32_t anonymity,
 962                struct GNUNET_TIME_Absolute expiration,
 963                uint64_t uid)
 964 {
 965   struct PutContext *pc = cls;
 966   const struct DataMessage *dm;
 967
 968   dm = (const struct DataMessage *) &pc[1];
 969   if (key == NULL)
 970   {
 971     execute_put (pc);
 972     return GNUNET_OK;
 973   }
 974   if ( (GNUNET_BLOCK_TYPE_FS_DBLOCK == type) ||
 975        (GNUNET_BLOCK_TYPE_FS_IBLOCK == type) ||
 976        ( (size == ntohl (dm->size)) &&
 977          (0 == memcmp (&dm[1], data, size)) ) )
 978   {
 979     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 980                 "Result already present in datastore\n");
 981     /* FIXME: change API to allow increasing 'replication' counter */
 982     if ((ntohl (dm->priority) > 0) ||
 983         (GNUNET_TIME_absolute_ntoh (dm->expiration).abs_value_us >
 984          expiration.abs_value_us))
 985       plugin->api->update (plugin->api->cls,
 986                            uid,
 987                            (int32_t) ntohl (dm->priority),
 988                            GNUNET_TIME_absolute_ntoh (dm->expiration),
 989                            &check_present_continuation,
 990                            pc->client);
 991     else
 992     {
 993       transmit_status (pc->client, GNUNET_NO, NULL);
 994       GNUNET_SERVER_client_drop (pc->client);
 995     }
 996     GNUNET_free (pc);
 997   }
 998   else
 999   {
1000     execute_put (pc);
1001   }
1002   return GNUNET_OK;
1003 }
1004
1005
1006 /**
1007  * Handle PUT-message.
1008  *
1009  * @param cls closure
1010  * @param client identification of the client
1011  * @param message the actual message
1012  */
1013 static void
1014 handle_put (void *cls,
1015             struct GNUNET_SERVER_Client *client,
1016             const struct GNUNET_MessageHeader *message)
1017 {
1018   const struct DataMessage *dm = check_data (message);
1019   int rid;
1020   struct ReservationList *pos;
1021   struct PutContext *pc;
1022   struct GNUNET_HashCode vhash;
1023   uint32_t size;
1024
1025   if ((dm == NULL) || (ntohl (dm->type) == 0))
1026   {
1027     GNUNET_break (0);
1028     GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
1029     return;
1030   }
1031   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1032               "Processing PUT request for `%s' of type %u\n",
1033               GNUNET_h2s (&dm->key),
1034               ntohl (dm->type));
1035   rid = ntohl (dm->rid);
1036   size = ntohl (dm->size);
1037   if (rid > 0)
1038   {
1039     pos = reservations;
1040     while ((NULL != pos) && (rid != pos->rid))
1041       pos = pos->next;
1042     GNUNET_break (pos != NULL);
1043     if (NULL != pos)
1044     {
1045       GNUNET_break (pos->entries > 0);
1046       GNUNET_break (pos->amount >= size);
1047       pos->entries--;
1048       pos->amount -= size;
1049       reserved -= (size + GNUNET_DATASTORE_ENTRY_OVERHEAD);
1050       GNUNET_STATISTICS_set (stats,
1051                              gettext_noop ("# reserved"),
1052                              reserved,
1053                              GNUNET_NO);
1054     }
1055   }
1056   pc = GNUNET_malloc (sizeof (struct PutContext) + size +
1057                       sizeof (struct DataMessage));
1058   pc->client = client;
1059   GNUNET_SERVER_client_keep (client);
1060   memcpy (&pc[1], dm, size + sizeof (struct DataMessage));
1061   if (GNUNET_YES == GNUNET_CONTAINER_bloomfilter_test (filter, &dm->key))
1062   {
1063     GNUNET_CRYPTO_hash (&dm[1], size, &vhash);
1064     plugin->api->get_key (plugin->api->cls,
1065                           0,
1066                           &dm->key,
1067                           &vhash,
1068                           ntohl (dm->type),
1069                           &check_present,
1070                           pc);
1071     return;
1072   }
1073   execute_put (pc);
1074 }
1075
1076
1077 /**
1078  * Handle #GNUNET_MESSAGE_TYPE_DATASTORE_GET-message.
1079  *
1080  * @param cls closure
1081  * @param client identification of the client
1082  * @param message the actual message
1083  */
1084 static void
1085 handle_get (void *cls,
1086             struct GNUNET_SERVER_Client *client,
1087             const struct GNUNET_MessageHeader *message)
1088 {
1089   const struct GetMessage *msg;
1090
1091   msg = (const struct GetMessage *) message;
1092   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1093               "Processing GET request of type %u\n",
1094               ntohl (msg->type));
1095   GNUNET_STATISTICS_update (stats,
1096                             gettext_noop ("# GET requests received"),
1097                             1,
1098                             GNUNET_NO);
1099   GNUNET_SERVER_client_keep (client);
1100   plugin->api->get_key (plugin->api->cls,
1101                         GNUNET_ntohll (msg->offset),
1102                         NULL,
1103                         NULL,
1104                         ntohl (msg->type),
1105                         &transmit_item,
1106                         client);
1107 }
1108
1109 /**
1110  * Handle #GNUNET_MESSAGE_TYPE_DATASTORE_GET_KEY-message.
1111  *
1112  * @param cls closure
1113  * @param client identification of the client
1114  * @param message the actual message
1115  */
1116 static void
1117 handle_get_key (void *cls,
1118                 struct GNUNET_SERVER_Client *client,
1119                 const struct GNUNET_MessageHeader *message)
1120 {
1121   const struct GetKeyMessage *msg;
1122
1123   msg = (const struct GetKeyMessage *) message;
1124   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1125               "Processing GET request for `%s' of type %u\n",
1126               GNUNET_h2s (&msg->key),
1127               ntohl (msg->type));
1128   GNUNET_STATISTICS_update (stats,
1129                             gettext_noop ("# GET KEY requests received"),
1130                             1,
1131                             GNUNET_NO);
1132   GNUNET_SERVER_client_keep (client);
1133   if (GNUNET_YES !=
1134       GNUNET_CONTAINER_bloomfilter_test (filter,
1135                                          &msg->key))
1136   {
1137     /* don't bother database... */
1138     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1139                 "Empty result set for GET request for `%s' (bloomfilter).\n",
1140                 GNUNET_h2s (&msg->key));
1141     GNUNET_STATISTICS_update (stats,
1142                               gettext_noop
1143                               ("# requests filtered by bloomfilter"),
1144                               1,
1145                               GNUNET_NO);
1146     transmit_item (client,
1147                    NULL, 0, NULL, 0, 0, 0,
1148                    GNUNET_TIME_UNIT_ZERO_ABS,
1149                    0);
1150     return;
1151   }
1152   plugin->api->get_key (plugin->api->cls,
1153                         GNUNET_ntohll (msg->offset),
1154                         &msg->key,
1155                         NULL,
1156                         ntohl (msg->type),
1157                         &transmit_item,
1158                         client);
1159 }
1160
1161
1162 /**
1163  * Function called with the result of an update operation.
1164  *
1165  * @param cls closure
1166  * @param status #GNUNET_OK or #GNUNET_SYSERR
1167  * @param msg error message on error
1168  */
1169 static void
1170 update_continuation (void *cls,
1171                      int status,
1172                      const char *msg)
1173 {
1174   struct GNUNET_SERVER_Client *client = cls;
1175
1176   transmit_status (client, status, msg);
1177   GNUNET_SERVER_client_drop (client);
1178 }
1179
1180
1181 /**
1182  * Handle UPDATE-message.
1183  *
1184  * @param cls closure
1185  * @param client identification of the client
1186  * @param message the actual message
1187  */
1188 static void
1189 handle_update (void *cls,
1190                struct GNUNET_SERVER_Client *client,
1191                const struct GNUNET_MessageHeader *message)
1192 {
1193   const struct UpdateMessage *msg;
1194
1195   GNUNET_STATISTICS_update (stats,
1196                             gettext_noop ("# UPDATE requests received"),
1197                             1,
1198                             GNUNET_NO);
1199   msg = (const struct UpdateMessage *) message;
1200   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1201               "Processing UPDATE request for %llu\n",
1202               (unsigned long long) GNUNET_ntohll (msg->uid));
1203   GNUNET_SERVER_client_keep (client);
1204   plugin->api->update (plugin->api->cls,
1205                        GNUNET_ntohll (msg->uid),
1206                        (int32_t) ntohl (msg->priority),
1207                        GNUNET_TIME_absolute_ntoh (msg->expiration),
1208                        &update_continuation, client);
1209 }
1210
1211
1212 /**
1213  * Handle GET_REPLICATION-message.
1214  *
1215  * @param cls closure
1216  * @param client identification of the client
1217  * @param message the actual message
1218  */
1219 static void
1220 handle_get_replication (void *cls,
1221                         struct GNUNET_SERVER_Client *client,
1222                         const struct GNUNET_MessageHeader *message)
1223 {
1224   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1225               "Processing `%s' request\n",
1226               "GET_REPLICATION");
1227   GNUNET_STATISTICS_update (stats,
1228                             gettext_noop ("# GET REPLICATION requests received"),
1229                             1,
1230                             GNUNET_NO);
1231   GNUNET_SERVER_client_keep (client);
1232   plugin->api->get_replication (plugin->api->cls,
1233                                 &transmit_item, client);
1234 }
1235
1236
1237 /**
1238  * Handle GET_ZERO_ANONYMITY-message.
1239  *
1240  * @param cls closure
1241  * @param client identification of the client
1242  * @param message the actual message
1243  */
1244 static void
1245 handle_get_zero_anonymity (void *cls,
1246                            struct GNUNET_SERVER_Client *client,
1247                            const struct GNUNET_MessageHeader *message)
1248 {
1249   const struct GetZeroAnonymityMessage *msg =
1250       (const struct GetZeroAnonymityMessage *) message;
1251   enum GNUNET_BLOCK_Type type;
1252
1253   type = (enum GNUNET_BLOCK_Type) ntohl (msg->type);
1254   if (type == GNUNET_BLOCK_TYPE_ANY)
1255   {
1256     GNUNET_break (0);
1257     GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
1258     return;
1259   }
1260   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1261               "Processing `%s' request\n",
1262               "GET_ZERO_ANONYMITY");
1263   GNUNET_STATISTICS_update (stats,
1264                             gettext_noop ("# GET ZERO ANONYMITY requests received"),
1265                             1,
1266                             GNUNET_NO);
1267   GNUNET_SERVER_client_keep (client);
1268   plugin->api->get_zero_anonymity (plugin->api->cls,
1269                                    GNUNET_ntohll (msg->offset),
1270                                    type,
1271                                    &transmit_item, client);
1272 }
1273
1274
1275 /**
1276  * Callback function that will cause the item that is passed
1277  * in to be deleted (by returning #GNUNET_NO).
1278  *
1279  * @param cls closure
1280  * @param key key for the content
1281  * @param size number of bytes in data
1282  * @param data content stored
1283  * @param type type of the content
1284  * @param priority priority of the content
1285  * @param anonymity anonymity-level for the content
1286  * @param expiration expiration time for the content
1287  * @param uid unique identifier for the datum
1288  * @return #GNUNET_OK to keep the item
1289  *         #GNUNET_NO to delete the item
1290  */
1291 static int
1292 remove_callback (void *cls,
1293                  const struct GNUNET_HashCode *key,
1294                  uint32_t size,
1295                  const void *data,
1296                  enum GNUNET_BLOCK_Type type,
1297                  uint32_t priority,
1298                  uint32_t anonymity,
1299                  struct GNUNET_TIME_Absolute expiration,
1300                  uint64_t uid)
1301 {
1302   struct GNUNET_SERVER_Client *client = cls;
1303
1304   if (NULL == key)
1305   {
1306     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1307                 "No further matches for REMOVE request.\n");
1308     transmit_status (client,
1309                      GNUNET_NO,
1310                      _("Content not found"));
1311     GNUNET_SERVER_client_drop (client);
1312     return GNUNET_OK;           /* last item */
1313   }
1314   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1315               "Item %llu matches REMOVE request for key `%s' and type %u.\n",
1316               (unsigned long long) uid,
1317               GNUNET_h2s (key),
1318               type);
1319   GNUNET_STATISTICS_update (stats,
1320                             gettext_noop ("# bytes removed (explicit request)"),
1321                             size,
1322                             GNUNET_YES);
1323   GNUNET_CONTAINER_bloomfilter_remove (filter, key);
1324   transmit_status (client, GNUNET_OK, NULL);
1325   GNUNET_SERVER_client_drop (client);
1326   return GNUNET_NO;
1327 }
1328
1329
1330 /**
1331  * Handle REMOVE-message.
1332  *
1333  * @param cls closure
1334  * @param client identification of the client
1335  * @param message the actual message
1336  */
1337 static void
1338 handle_remove (void *cls,
1339                struct GNUNET_SERVER_Client *client,
1340                const struct GNUNET_MessageHeader *message)
1341 {
1342   const struct DataMessage *dm = check_data (message);
1343   struct GNUNET_HashCode vhash;
1344
1345   if (NULL == dm)
1346   {
1347     GNUNET_break (0);
1348     GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
1349     return;
1350   }
1351   GNUNET_STATISTICS_update (stats,
1352                             gettext_noop ("# REMOVE requests received"),
1353                             1, GNUNET_NO);
1354   GNUNET_SERVER_client_keep (client);
1355   GNUNET_CRYPTO_hash (&dm[1],
1356                       ntohl (dm->size),
1357                       &vhash);
1358   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1359               "Processing REMOVE request for `%s' of type %u\n",
1360               GNUNET_h2s (&dm->key),
1361               ntohl (dm->type));
1362   plugin->api->get_key (plugin->api->cls,
1363                         0,
1364                         &dm->key,
1365                         &vhash,
1366                         (enum GNUNET_BLOCK_Type) ntohl (dm->type),
1367                         &remove_callback, client);
1368 }
1369
1370
1371 /**
1372  * Handle DROP-message.
1373  *
1374  * @param cls closure
1375  * @param client identification of the client
1376  * @param message the actual message
1377  */
1378 static void
1379 handle_drop (void *cls,
1380              struct GNUNET_SERVER_Client *client,
1381              const struct GNUNET_MessageHeader *message)
1382 {
1383   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1384               "Processing DROP request\n");
1385   do_drop = GNUNET_YES;
1386   GNUNET_SERVER_receive_done (client, GNUNET_OK);
1387 }
1388
1389
1390 /**
1391  * Function called by plugins to notify us about a
1392  * change in their disk utilization.
1393  *
1394  * @param cls closure (NULL)
1395  * @param delta change in disk utilization,
1396  *        0 for "reset to empty"
1397  */
1398 static void
1399 disk_utilization_change_cb (void *cls,
1400                             int delta)
1401 {
1402   if ((delta < 0) && (payload < -delta))
1403   {
1404     GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
1405                 _("Datastore payload must have been inaccurate (%lld < %lld). Recomputing it.\n"),
1406                 (long long) payload,
1407                 (long long) -delta);
1408     plugin->api->estimate_size (plugin->api->cls,
1409                                 &payload);
1410     GNUNET_log (GNUNET_ERROR_TYPE_INFO,
1411                 _("New payload: %lld\n"),
1412                 (long long) payload);
1413      sync_stats ();
1414     return;
1415   }
1416   payload += delta;
1417   last_sync++;
1418   if (last_sync >= MAX_STAT_SYNC_LAG)
1419     sync_stats ();
1420 }
1421
1422
1423 /**
1424  * Callback function to process statistic values.
1425  *
1426  * @param cls closure (struct Plugin*)
1427  * @param subsystem name of subsystem that created the statistic
1428  * @param name the name of the datum
1429  * @param value the current value
1430  * @param is_persistent #GNUNET_YES if the value is persistent, #GNUNET_NO if not
1431  * @return #GNUNET_OK to continue, #GNUNET_SYSERR to abort iteration
1432  */
1433 static int
1434 process_stat_in (void *cls,
1435                  const char *subsystem,
1436                  const char *name,
1437                  uint64_t value,
1438                  int is_persistent)
1439 {
1440   GNUNET_assert (GNUNET_NO == stats_worked);
1441   stats_worked = GNUNET_YES;
1442   payload += value;
1443   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1444               "Notification from statistics about existing payload (%llu), new payload is %llu\n",
1445               (unsigned long long) value,
1446               (unsigned long long) payload);
1447   return GNUNET_OK;
1448 }
1449
1450
1451 /**
1452  * Load the datastore plugin.
1453  */
1454 static struct DatastorePlugin *
1455 load_plugin ()
1456 {
1457   struct DatastorePlugin *ret;
1458   char *libname;
1459
1460   ret = GNUNET_new (struct DatastorePlugin);
1461   ret->env.cfg = cfg;
1462   ret->env.duc = &disk_utilization_change_cb;
1463   ret->env.cls = NULL;
1464   GNUNET_log (GNUNET_ERROR_TYPE_INFO,
1465               _("Loading `%s' datastore plugin\n"),
1466               plugin_name);
1467   GNUNET_asprintf (&libname,
1468                    "libgnunet_plugin_datastore_%s",
1469                    plugin_name);
1470   ret->short_name = GNUNET_strdup (plugin_name);
1471   ret->lib_name = libname;
1472   ret->api = GNUNET_PLUGIN_load (libname, &ret->env);
1473   if (NULL == ret->api)
1474   {
1475     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
1476                 _("Failed to load datastore plugin for `%s'\n"),
1477                 plugin_name);
1478     GNUNET_free (ret->short_name);
1479     GNUNET_free (libname);
1480     GNUNET_free (ret);
1481     return NULL;
1482   }
1483   return ret;
1484 }
1485
1486
1487 /**
1488  * Function called when the service shuts
1489  * down.  Unloads our datastore plugin.
1490  *
1491  * @param plug plugin to unload
1492  */
1493 static void
1494 unload_plugin (struct DatastorePlugin *plug)
1495 {
1496   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1497               "Datastore service is unloading plugin...\n");
1498   GNUNET_break (NULL == GNUNET_PLUGIN_unload (plug->lib_name, plug->api));
1499   GNUNET_free (plug->lib_name);
1500   GNUNET_free (plug->short_name);
1501   GNUNET_free (plug);
1502 }
1503
1504
1505 static const struct GNUNET_SERVER_MessageHandler handlers[] = {
1506   {&handle_reserve, NULL, GNUNET_MESSAGE_TYPE_DATASTORE_RESERVE,
1507    sizeof (struct ReserveMessage)},
1508   {&handle_release_reserve, NULL,
1509    GNUNET_MESSAGE_TYPE_DATASTORE_RELEASE_RESERVE,
1510    sizeof (struct ReleaseReserveMessage)},
1511   {&handle_put, NULL, GNUNET_MESSAGE_TYPE_DATASTORE_PUT, 0},
1512   {&handle_update, NULL, GNUNET_MESSAGE_TYPE_DATASTORE_UPDATE,
1513    sizeof (struct UpdateMessage)},
1514   {&handle_get, NULL, GNUNET_MESSAGE_TYPE_DATASTORE_GET,
1515    sizeof (struct GetMessage) },
1516   {&handle_get_key, NULL, GNUNET_MESSAGE_TYPE_DATASTORE_GET_KEY,
1517    sizeof (struct GetKeyMessage) },
1518   {&handle_get_replication, NULL,
1519    GNUNET_MESSAGE_TYPE_DATASTORE_GET_REPLICATION,
1520    sizeof (struct GNUNET_MessageHeader)},
1521   {&handle_get_zero_anonymity, NULL,
1522    GNUNET_MESSAGE_TYPE_DATASTORE_GET_ZERO_ANONYMITY,
1523    sizeof (struct GetZeroAnonymityMessage)},
1524   {&handle_remove, NULL, GNUNET_MESSAGE_TYPE_DATASTORE_REMOVE, 0},
1525   {&handle_drop, NULL, GNUNET_MESSAGE_TYPE_DATASTORE_DROP,
1526    sizeof (struct GNUNET_MessageHeader)},
1527   {NULL, NULL, 0, 0}
1528 };
1529
1530
1531 /**
1532  * Adds a given @a key to the bloomfilter in @a cls @a count times.
1533  *
1534  * @param cls the bloomfilter
1535  * @param key key to add
1536  * @param count number of times to add key
1537  */
1538 static void
1539 add_key_to_bloomfilter (void *cls,
1540                         const struct GNUNET_HashCode *key,
1541                         unsigned int count)
1542 {
1543   struct GNUNET_CONTAINER_BloomFilter *bf = cls;
1544
1545   if (NULL == key)
1546   {
1547     GNUNET_log (GNUNET_ERROR_TYPE_INFO,
1548                 _("Bloomfilter construction complete.\n"));
1549     GNUNET_SERVER_add_handlers (server, handlers);
1550     GNUNET_SERVER_resume (server);
1551     expired_kill_task
1552       = GNUNET_SCHEDULER_add_with_priority (GNUNET_SCHEDULER_PRIORITY_IDLE,
1553                                             &delete_expired,
1554                                             NULL);
1555     return;
1556   }
1557
1558   while (0 < count--)
1559     GNUNET_CONTAINER_bloomfilter_add (bf, key);
1560 }
1561
1562
1563 /**
1564  * We finished receiving the statistic.  Initialize the plugin; if
1565  * loading the statistic failed, run the estimator.
1566  *
1567  * @param cls NULL
1568  * @param success #GNUNET_NO if we failed to read the stat
1569  */
1570 static void
1571 process_stat_done (void *cls,
1572                    int success)
1573 {
1574   stat_get = NULL;
1575   if (NULL != stat_timeout_task)
1576   {
1577     GNUNET_SCHEDULER_cancel (stat_timeout_task);
1578     stat_timeout_task = NULL;
1579   }
1580   plugin = load_plugin ();
1581   if (NULL == plugin)
1582   {
1583     GNUNET_CONTAINER_bloomfilter_free (filter);
1584     filter = NULL;
1585     if (NULL != stats)
1586     {
1587       GNUNET_STATISTICS_destroy (stats, GNUNET_YES);
1588       stats = NULL;
1589     }
1590     return;
1591   }
1592   if (GNUNET_NO == stats_worked)
1593   {
1594     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1595                 "Failed to obtain value from statistics service, recomputing it\n");
1596     plugin->api->estimate_size (plugin->api->cls,
1597                                 &payload);
1598     GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
1599                 _("New payload: %lld\n"),
1600                 (long long) payload);
1601
1602   }
1603   if (GNUNET_YES == refresh_bf)
1604   {
1605     GNUNET_log (GNUNET_ERROR_TYPE_INFO,
1606                 _("Rebuilding bloomfilter.  Please be patient.\n"));
1607     if (NULL != plugin->api->get_keys)
1608     {
1609       plugin->api->get_keys (plugin->api->cls,
1610                              &add_key_to_bloomfilter,
1611                              filter);
1612       return;
1613     }
1614     else
1615       GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
1616                   _("Plugin does not support get_keys function. Please fix!\n"));
1617
1618     GNUNET_log (GNUNET_ERROR_TYPE_INFO,
1619                 _("Bloomfilter construction complete.\n"));
1620   }
1621
1622   GNUNET_SERVER_add_handlers (server, handlers);
1623   GNUNET_SERVER_resume (server);
1624   expired_kill_task
1625     = GNUNET_SCHEDULER_add_with_priority (GNUNET_SCHEDULER_PRIORITY_IDLE,
1626                                           &delete_expired,
1627                                           NULL);
1628 }
1629
1630
1631 /**
1632  * Fetching stats took to long, run without.
1633  *
1634  * @param cls NULL
1635  */
1636 static void
1637 stat_timeout (void *cls)
1638 {
1639   stat_timeout_task = NULL;
1640   GNUNET_STATISTICS_get_cancel (stat_get);
1641   process_stat_done (NULL, GNUNET_NO);
1642 }
1643
1644
1645 /**
1646  * Task run during shutdown.
1647  */
1648 static void
1649 cleaning_task (void *cls)
1650 {
1651   struct TransmitCallbackContext *tcc;
1652
1653   cleaning_done = GNUNET_YES;
1654   while (NULL != (tcc = tcc_head))
1655   {
1656     GNUNET_CONTAINER_DLL_remove (tcc_head, tcc_tail, tcc);
1657     if (tcc->th != NULL)
1658     {
1659       GNUNET_SERVER_notify_transmit_ready_cancel (tcc->th);
1660       GNUNET_SERVER_client_drop (tcc->client);
1661     }
1662     GNUNET_free (tcc->msg);
1663     GNUNET_free (tcc);
1664   }
1665   if (NULL != expired_kill_task)
1666   {
1667     GNUNET_SCHEDULER_cancel (expired_kill_task);
1668     expired_kill_task = NULL;
1669   }
1670   if (GNUNET_YES == do_drop)
1671   {
1672     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1673                 "Dropping database!\n");
1674     plugin->api->drop (plugin->api->cls);
1675     payload = 0;
1676     last_sync++;
1677   }
1678   if (NULL != plugin)
1679   {
1680     unload_plugin (plugin);
1681     plugin = NULL;
1682   }
1683   if (NULL != filter)
1684   {
1685     GNUNET_CONTAINER_bloomfilter_free (filter);
1686     filter = NULL;
1687   }
1688   if (NULL != stat_get)
1689   {
1690     GNUNET_STATISTICS_get_cancel (stat_get);
1691     stat_get = NULL;
1692   }
1693   if (NULL != stat_timeout_task)
1694   {
1695     GNUNET_SCHEDULER_cancel (stat_timeout_task);
1696     stat_timeout_task = NULL;
1697   }
1698   GNUNET_free_non_null (plugin_name);
1699   plugin_name = NULL;
1700   if (last_sync > 0)
1701     sync_stats ();
1702   if (NULL != stats)
1703   {
1704     GNUNET_STATISTICS_destroy (stats,
1705                                GNUNET_YES);
1706     stats = NULL;
1707   }
1708   GNUNET_free (quota_stat_name);
1709   quota_stat_name = NULL;
1710 }
1711
1712
1713 /**
1714  * Function that removes all active reservations made
1715  * by the given client and releases the space for other
1716  * requests.
1717  *
1718  * @param cls closure
1719  * @param client identification of the client
1720  */
1721 static void
1722 cleanup_reservations (void *cls,
1723                       struct GNUNET_SERVER_Client *client)
1724 {
1725   struct ReservationList *pos;
1726   struct ReservationList *prev;
1727   struct ReservationList *next;
1728
1729   if (NULL == client)
1730     return;
1731   prev = NULL;
1732   pos = reservations;
1733   while (NULL != pos)
1734   {
1735     next = pos->next;
1736     if (pos->client == client)
1737     {
1738       if (NULL == prev)
1739         reservations = next;
1740       else
1741         prev->next = next;
1742       reserved -= pos->amount + pos->entries * GNUNET_DATASTORE_ENTRY_OVERHEAD;
1743       GNUNET_free (pos);
1744     }
1745     else
1746     {
1747       prev = pos;
1748     }
1749     pos = next;
1750   }
1751   GNUNET_STATISTICS_set (stats,
1752                          gettext_noop ("# reserved"),
1753                          reserved,
1754                          GNUNET_NO);
1755 }
1756
1757
1758 /**
1759  * Process datastore requests.
1760  *
1761  * @param cls closure
1762  * @param serv the initialized server
1763  * @param c configuration to use
1764  */
1765 static void
1766 run (void *cls,
1767      struct GNUNET_SERVER_Handle *serv,
1768      const struct GNUNET_CONFIGURATION_Handle *c)
1769 {
1770   char *fn;
1771   char *pfn;
1772   unsigned int bf_size;
1773
1774   server = serv;
1775   cfg = c;
1776   if (GNUNET_OK !=
1777       GNUNET_CONFIGURATION_get_value_string (cfg,
1778                                              "DATASTORE",
1779                                              "DATABASE",
1780                                              &plugin_name))
1781   {
1782     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
1783                 _("No `%s' specified for `%s' in configuration!\n"),
1784                 "DATABASE",
1785                 "DATASTORE");
1786     return;
1787   }
1788   GNUNET_asprintf (&quota_stat_name,
1789                    _("# bytes used in file-sharing datastore `%s'"),
1790                    plugin_name);
1791   if (GNUNET_OK !=
1792       GNUNET_CONFIGURATION_get_value_size (cfg, "DATASTORE", "QUOTA", &quota))
1793   {
1794     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
1795                 _("No `%s' specified for `%s' in configuration!\n"),
1796                 "QUOTA",
1797                 "DATASTORE");
1798     return;
1799   }
1800   stats = GNUNET_STATISTICS_create ("datastore", cfg);
1801   GNUNET_STATISTICS_set (stats, gettext_noop ("# quota"), quota, GNUNET_NO);
1802   cache_size = quota / 8;       /* Or should we make this an option? */
1803   GNUNET_STATISTICS_set (stats, gettext_noop ("# cache size"), cache_size,
1804                          GNUNET_NO);
1805   if (quota / (32 * 1024LL) > (1 << 31))
1806     bf_size = (1 << 31);          /* absolute limit: ~2 GB, beyond that BF just won't help anyway */
1807   else
1808     bf_size = quota / (32 * 1024LL);         /* 8 bit per entry, 1 bit per 32 kb in DB */
1809   fn = NULL;
1810   if ((GNUNET_OK !=
1811        GNUNET_CONFIGURATION_get_value_filename (cfg,
1812                                                 "DATASTORE",
1813                                                 "BLOOMFILTER",
1814                                                 &fn)) ||
1815       (GNUNET_OK != GNUNET_DISK_directory_create_for_file (fn)))
1816   {
1817     GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
1818                 _("Could not use specified filename `%s' for bloomfilter.\n"),
1819                 NULL != fn ? fn : "");
1820     GNUNET_free_non_null (fn);
1821     fn = NULL;
1822   }
1823   if (NULL != fn)
1824   {
1825     GNUNET_asprintf (&pfn, "%s.%s", fn, plugin_name);
1826     if (GNUNET_YES == GNUNET_DISK_file_test (pfn))
1827     {
1828       filter = GNUNET_CONTAINER_bloomfilter_load (pfn, bf_size, 5);        /* approx. 3% false positives at max use */
1829       if (NULL == filter)
1830       {
1831         /* file exists but not valid, remove and try again, but refresh */
1832         if (0 != UNLINK (pfn))
1833         {
1834           /* failed to remove, run without file */
1835           GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
1836                       _("Failed to remove bogus bloomfilter file `%s'\n"),
1837                       pfn);
1838           GNUNET_free (pfn);
1839           pfn = NULL;
1840           filter = GNUNET_CONTAINER_bloomfilter_load (NULL, bf_size, 5);        /* approx. 3% false positives at max use */
1841           refresh_bf = GNUNET_YES;
1842         }
1843         else
1844         {
1845           /* try again after remove */
1846           filter = GNUNET_CONTAINER_bloomfilter_load (pfn, bf_size, 5);        /* approx. 3% false positives at max use */
1847           refresh_bf = GNUNET_YES;
1848           if (NULL == filter)
1849           {
1850             /* failed yet again, give up on using file */
1851             GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
1852                         _("Failed to remove bogus bloomfilter file `%s'\n"),
1853                         pfn);
1854             GNUNET_free (pfn);
1855             pfn = NULL;
1856             filter = GNUNET_CONTAINER_bloomfilter_init (NULL, bf_size, 5);        /* approx. 3% false positives at max use */
1857           }
1858         }
1859       }
1860       else
1861       {
1862         /* normal case: have an existing valid bf file, no need to refresh */
1863         refresh_bf = GNUNET_NO;
1864       }
1865     }
1866     else
1867     {
1868       filter = GNUNET_CONTAINER_bloomfilter_load (pfn, bf_size, 5);        /* approx. 3% false positives at max use */
1869       refresh_bf = GNUNET_YES;
1870     }
1871     GNUNET_free (pfn);
1872   }
1873   else
1874   {
1875     filter = GNUNET_CONTAINER_bloomfilter_init (NULL, bf_size, 5);      /* approx. 3% false positives at max use */
1876     refresh_bf = GNUNET_YES;
1877   }
1878   GNUNET_free_non_null (fn);
1879   if (NULL == filter)
1880   {
1881     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
1882                 _("Failed to initialize bloomfilter.\n"));
1883     if (NULL != stats)
1884     {
1885       GNUNET_STATISTICS_destroy (stats, GNUNET_YES);
1886       stats = NULL;
1887     }
1888     return;
1889   }
1890   GNUNET_SERVER_suspend (server);
1891   stat_get =
1892       GNUNET_STATISTICS_get (stats,
1893                              "datastore",
1894                              quota_stat_name,
1895                              &process_stat_done,
1896                              &process_stat_in,
1897                              NULL);
1898   if (NULL == stat_get)
1899     process_stat_done (NULL, GNUNET_SYSERR);
1900   else
1901     stat_timeout_task = GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_SECONDS,
1902                                                       &stat_timeout,
1903                                                       NULL);
1904   GNUNET_SERVER_disconnect_notify (server,
1905                                    &cleanup_reservations,
1906                                    NULL);
1907   GNUNET_SCHEDULER_add_shutdown (&cleaning_task,
1908                                  NULL);
1909 }
1910
1911
1912 /**
1913  * The main function for the datastore service.
1914  *
1915  * @param argc number of arguments from the command line
1916  * @param argv command line arguments
1917  * @return 0 ok, 1 on error
1918  */
1919 int
1920 main (int argc,
1921       char *const *argv)
1922 {
1923   int ret;
1924
1925   ret =
1926       (GNUNET_OK ==
1927        GNUNET_SERVICE_run (argc, argv, "datastore",
1928                            GNUNET_SERVICE_OPTION_NONE,
1929                            &run, NULL)) ? 0 : 1;
1930   return ret;
1931 }
1932
1933
1934 /* end of gnunet-service-datastore.c */