src/datastore/gnunet-service-datastore.c

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