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