src/datastore/datastore_api.c

   1 /*
   2      This file is part of GNUnet
   3      (C) 2004, 2005, 2006, 2007, 2009, 2010 Christian Grothoff (and other contributing authors)
   4
   5      GNUnet is free software; you can redistribute it and/or modify
   6      it under the terms of the GNU General Public License as published
   7      by the Free Software Foundation; either version 2, or (at your
   8      option) any later version.
   9
  10      GNUnet is distributed in the hope that it will be useful, but
  11      WITHOUT ANY WARRANTY; without even the implied warranty of
  12      MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13      General Public License for more details.
  14
  15      You should have received a copy of the GNU General Public License
  16      along with GNUnet; see the file COPYING.  If not, write to the
  17      Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  18      Boston, MA 02111-1307, USA.
  19 */
  20
  21 /**
  22  * @file datastore/datastore_api.c
  23  * @brief Management for the datastore for files stored on a GNUnet node
  24  * @author Christian Grothoff
  25  */
  26 #include "platform.h"
  27 #include "gnunet_arm_service.h"
  28 #include "gnunet_datastore_service.h"
  29 #include "datastore.h"
  30
  31 /**
  32  * Entry in our priority queue.
  33  */
  34 struct QueueEntry
  35 {
  36
  37   /**
  38    * This is a linked list.
  39    */
  40   struct QueueEntry *next;
  41
  42   /**
  43    * This is a linked list.
  44    */
  45   struct QueueEntry *prev;
  46
  47   /**
  48    * Handle to the master context.
  49    */
  50   struct GNUNET_DATASTORE_Handle *h;
  51
  52   /**
  53    * Task for timeout signalling.
  54    */
  55   GNUNET_SCHEDULER_TaskIdentifier task;
  56
  57   /**
  58    * Timeout for the current operation.
  59    */
  60   struct GNUNET_TIME_Absolute timeout;
  61
  62   /**
  63    * Priority in the queue.
  64    */
  65   unsigned int priority;
  66
  67   /**
  68    * Maximum allowed length of queue (otherwise
  69    * this request should be discarded).
  70    */
  71   unsigned int max_queue;
  72
  73   /**
  74    * Number of bytes in the request message following
  75    * this struct.
  76    */
  77   uint16_t message_size;
  78
  79   /**
  80    * Has this message been transmitted to the service?
  81    * Only ever GNUNET_YES for the head of the queue.
  82    */
  83   int16_t was_transmitted;
  84
  85   /**
  86    * Response processor (NULL if we are not waiting for a response).
  87    * This struct should be used for the closure, function-specific
  88    * arguments can be passed via 'client_ctx'.
  89    */
  90   GNUNET_CLIENT_MessageHandler response_proc;
  91
  92   /**
  93    * Specific context (variable argument that
  94    * can be used by the response processor).
  95    */
  96   void *client_ctx;
  97
  98 };
  99
 100 /**
 101  * Handle to the datastore service.  Followed
 102  * by 65536 bytes used for storing messages.
 103  */
 104 struct GNUNET_DATASTORE_Handle
 105 {
 106
 107   /**
 108    * Our configuration.
 109    */
 110   const struct GNUNET_CONFIGURATION_Handle *cfg;
 111
 112   /**
 113    * Our scheduler.
 114    */
 115   struct GNUNET_SCHEDULER_Handle *sched;
 116
 117   /**
 118    * Current connection to the datastore service.
 119    */
 120   struct GNUNET_CLIENT_Connection *client;
 121
 122   /**
 123    * Current head of priority queue.
 124    */
 125   struct QueueEntry *queue_head;
 126
 127   /**
 128    * Current tail of priority queue.
 129    */
 130   struct QueueEntry *queue_tail;
 131
 132   /**
 133    * Number of entries in the queue.
 134    */
 135   unsigned int queue_size;
 136
 137 };
 138
 139
 140
 141 /**
 142  * Connect to the datastore service.
 143  *
 144  * @param cfg configuration to use
 145  * @param sched scheduler to use
 146  * @return handle to use to access the service
 147  */
 148 struct GNUNET_DATASTORE_Handle *
 149 GNUNET_DATASTORE_connect (const struct
 150                           GNUNET_CONFIGURATION_Handle
 151                           *cfg,
 152                           struct
 153                           GNUNET_SCHEDULER_Handle
 154                           *sched)
 155 {
 156   struct GNUNET_CLIENT_Connection *c;
 157   struct GNUNET_DATASTORE_Handle *h;
 158
 159   c = GNUNET_CLIENT_connect (sched, "datastore", cfg);
 160   if (c == NULL)
 161     return NULL; /* oops */
 162   h = GNUNET_malloc (sizeof(struct GNUNET_DATASTORE_Handle) +
 163                      GNUNET_SERVER_MAX_MESSAGE_SIZE);
 164   h->client = c;
 165   h->cfg = cfg;
 166   h->sched = sched;
 167   return h;
 168 }
 169
 170
 171 /**
 172  * Transmit DROP message to datastore service.
 173  *
 174  * @param cls the 'struct GNUNET_DATASTORE_Handle'
 175  * @param size number of bytes that can be copied to buf
 176  * @param buf where to copy the drop message
 177  * @return number of bytes written to buf
 178  */
 179 static size_t
 180 transmit_drop (void *cls,
 181                size_t size,
 182                void *buf)
 183 {
 184   struct GNUNET_DATASTORE_Handle *h = cls;
 185   struct GNUNET_MessageHeader *hdr;
 186
 187   if (buf == NULL)
 188     {
 189       GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 190                   _("Failed to transmit request to drop database.\n"));
 191       GNUNET_DATASTORE_disconnect (h, GNUNET_NO);
 192       return 0;
 193     }
 194   GNUNET_assert (size >= sizeof(struct GNUNET_MessageHeader));
 195   hdr = buf;
 196   hdr->size = htons(sizeof(struct GNUNET_MessageHeader));
 197   hdr->type = htons(GNUNET_MESSAGE_TYPE_DATASTORE_DROP);
 198   GNUNET_DATASTORE_disconnect (h, GNUNET_NO);
 199   return sizeof(struct GNUNET_MessageHeader);
 200 }
 201
 202
 203 /**
 204  * Disconnect from the datastore service (and free
 205  * associated resources).
 206  *
 207  * @param h handle to the datastore
 208  * @param drop set to GNUNET_YES to delete all data in datastore (!)
 209  */
 210 void GNUNET_DATASTORE_disconnect (struct GNUNET_DATASTORE_Handle *h,
 211                                   int drop)
 212 {
 213   struct QueueEntry *qe;
 214
 215   if (h->client != NULL)
 216     GNUNET_CLIENT_disconnect (h->client, GNUNET_NO);
 217   h->client = NULL;
 218   while (NULL != (qe = h->queue_head))
 219     {
 220       GNUNET_CONTAINER_DLL_remove (h->queue_head,
 221                                    h->queue_tail,
 222                                    qe);
 223       if (NULL != qe->response_proc)
 224         qe->response_proc (qe, NULL);
 225       GNUNET_free (qe);
 226     }
 227   if (GNUNET_YES == drop)
 228     {
 229       h->client = GNUNET_CLIENT_connect (h->sched, "datastore", h->cfg);
 230       if (h->client != NULL)
 231         {
 232           if (NULL !=
 233               GNUNET_CLIENT_notify_transmit_ready (h->client,
 234                                                    sizeof(struct GNUNET_MessageHeader),
 235                                                    GNUNET_TIME_UNIT_MINUTES,
 236                                                    GNUNET_YES,
 237                                                    &transmit_drop,
 238                                                    h))
 239             return;
 240           GNUNET_CLIENT_disconnect (h->client, GNUNET_NO);
 241         }
 242       GNUNET_break (0);
 243     }
 244   GNUNET_free (h);
 245 }
 246
 247
 248 #if 0
 249 /**
 250  * Type of a function to call when we receive a message
 251  * from the service.  This specific function is used
 252  * to handle messages of type "struct StatusMessage".
 253  *
 254  * @param cls closure
 255  * @param msg message received, NULL on timeout or fatal error
 256  */
 257 static void
 258 with_status_response_handler (void *cls,
 259                               const struct
 260                               GNUNET_MessageHeader * msg)
 261 {
 262   struct GNUNET_DATASTORE_Handle *h = cls;
 263   GNUNET_DATASTORE_ContinuationWithStatus cont = h->response_proc;
 264   const struct StatusMessage *sm;
 265   const char *emsg;
 266   int status;
 267
 268   h->message_size = 0;
 269   if (msg == NULL)
 270     {
 271       h->response_proc = NULL;
 272       GNUNET_CLIENT_disconnect (h->client, GNUNET_NO);
 273       h->client = GNUNET_CLIENT_connect (h->sched, "datastore", h->cfg);
 274       cont (h->response_proc_cls,
 275             GNUNET_SYSERR,
 276             _("Timeout trying to read response from datastore service"));
 277       return;
 278     }
 279   if ( (ntohs(msg->size) < sizeof(struct StatusMessage)) ||
 280        (ntohs(msg->type) != GNUNET_MESSAGE_TYPE_DATASTORE_STATUS) )
 281     {
 282       GNUNET_break (0);
 283       h->response_proc = NULL;
 284       GNUNET_CLIENT_disconnect (h->client, GNUNET_NO);
 285       h->client = GNUNET_CLIENT_connect (h->sched, "datastore", h->cfg);
 286       cont (h->response_proc_cls,
 287             GNUNET_SYSERR,
 288             _("Error reading response from datastore service"));
 289       return;
 290     }
 291   sm = (const struct StatusMessage*) msg;
 292   status = ntohl(sm->status);
 293   emsg = NULL;
 294   if (ntohs(msg->size) > sizeof(struct StatusMessage))
 295     {
 296       emsg = (const char*) &sm[1];
 297       if (emsg[ntohs(msg->size) - sizeof(struct StatusMessage) - 1] != '\0')
 298         {
 299           GNUNET_break (0);
 300           emsg = _("Invalid error message received from datastore service");
 301         }
 302     }
 303   if ( (status == GNUNET_SYSERR) &&
 304        (emsg == NULL) )
 305     {
 306       GNUNET_break (0);
 307       emsg = _("Invalid error message received from datastore service");
 308     }
 309   h->response_proc = NULL;
 310 #if DEBUG_DATASTORE
 311   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 312               "Received status %d/%s\n",
 313               status,
 314               emsg);
 315 #endif
 316   cont (h->response_proc_cls,
 317         status,
 318         emsg);
 319 }
 320
 321
 322 /**
 323  * Helper function that will initiate the
 324  * transmission of a message to the datastore
 325  * service.  The message must already be prepared
 326  * and stored in the buffer at the end of the
 327  * handle.  The message must be of a type that
 328  * expects a "StatusMessage" in response.
 329  *
 330  * @param h handle to the service with prepared message
 331  * @param cont function to call with result
 332  * @param cont_cls closure
 333  * @param timeout timeout for the operation
 334  */
 335 static void
 336 transmit_for_status (struct GNUNET_DATASTORE_Handle *h,
 337                      GNUNET_DATASTORE_ContinuationWithStatus cont,
 338                      void *cont_cls,
 339                      struct GNUNET_TIME_Relative timeout)
 340 {
 341   const struct GNUNET_MessageHeader *hdr;
 342   uint16_t msize;
 343
 344   GNUNET_assert (cont != NULL);
 345   hdr = (const struct GNUNET_MessageHeader*) &h[1];
 346   msize = ntohs(hdr->size);
 347 #if DEBUG_DATASTORE
 348   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 349               "Transmitting %u byte message of type %u to datastore service\n",
 350               msize,
 351               ntohs(hdr->type));
 352 #endif
 353   GNUNET_assert (h->response_proc == NULL);
 354   h->response_proc = cont;
 355   h->response_proc_cls = cont_cls;
 356   h->timeout = GNUNET_TIME_relative_to_absolute (timeout);
 357   h->message_size = msize;
 358   if (GNUNET_OK !=
 359       GNUNET_CLIENT_transmit_and_get_response (h->client,
 360                                                hdr,
 361                                                timeout,
 362                                                GNUNET_YES,
 363                                                &with_status_response_handler,
 364                                                h))
 365     {
 366       GNUNET_break (0);
 367       h->response_proc = NULL;
 368       h->message_size = 0;
 369       cont (cont_cls,
 370             GNUNET_SYSERR,
 371             _("Not ready to transmit request to datastore service"));
 372     }
 373 }
 374
 375
 376 /**
 377  * Store an item in the datastore.  If the item is already present,
 378  * the priorities are summed up and the higher expiration time and
 379  * lower anonymity level is used.
 380  *
 381  * @param h handle to the datastore
 382  * @param rid reservation ID to use (from "reserve"); use 0 if no
 383  *            prior reservation was made
 384  * @param key key for the value
 385  * @param size number of bytes in data
 386  * @param data content stored
 387  * @param type type of the content
 388  * @param priority priority of the content
 389  * @param anonymity anonymity-level for the content
 390  * @param expiration expiration time for the content
 391  * @param timeout timeout for the operation
 392  * @param cont continuation to call when done
 393  * @param cont_cls closure for cont
 394  */
 395 void
 396 GNUNET_DATASTORE_put (struct GNUNET_DATASTORE_Handle *h,
 397                       int rid,
 398                       const GNUNET_HashCode * key,
 399                       uint32_t size,
 400                       const void *data,
 401                       enum GNUNET_BLOCK_Type type,
 402                       uint32_t priority,
 403                       uint32_t anonymity,
 404                       struct GNUNET_TIME_Absolute expiration,
 405                       struct GNUNET_TIME_Relative timeout,
 406                       GNUNET_DATASTORE_ContinuationWithStatus cont,
 407                       void *cont_cls)
 408 {
 409   struct DataMessage *dm;
 410   size_t msize;
 411
 412 #if DEBUG_DATASTORE
 413   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 414               "Asked to put %u bytes of data under key `%s'\n",
 415               size,
 416               GNUNET_h2s (key));
 417 #endif
 418   msize = sizeof(struct DataMessage) + size;
 419   GNUNET_assert (msize <= GNUNET_SERVER_MAX_MESSAGE_SIZE);
 420   dm = (struct DataMessage*) &h[1];
 421   dm->header.type = htons(GNUNET_MESSAGE_TYPE_DATASTORE_PUT);
 422   dm->header.size = htons(msize);
 423   dm->rid = htonl(rid);
 424   dm->size = htonl(size);
 425   dm->type = htonl(type);
 426   dm->priority = htonl(priority);
 427   dm->anonymity = htonl(anonymity);
 428   dm->uid = GNUNET_htonll(0);
 429   dm->expiration = GNUNET_TIME_absolute_hton(expiration);
 430   dm->key = *key;
 431   memcpy (&dm[1], data, size);
 432   transmit_for_status (h, cont, cont_cls, timeout);
 433 }
 434
 435
 436 /**
 437  * Reserve space in the datastore.  This function should be used
 438  * to avoid "out of space" failures during a longer sequence of "put"
 439  * operations (for example, when a file is being inserted).
 440  *
 441  * @param h handle to the datastore
 442  * @param amount how much space (in bytes) should be reserved (for content only)
 443  * @param entries how many entries will be created (to calculate per-entry overhead)
 444  * @param cont continuation to call when done; "success" will be set to
 445  *             a positive reservation value if space could be reserved.
 446  * @param cont_cls closure for cont
 447  * @param timeout how long to wait at most for a response
 448  */
 449 void
 450 GNUNET_DATASTORE_reserve (struct GNUNET_DATASTORE_Handle *h,
 451                           uint64_t amount,
 452                           uint32_t entries,
 453                           GNUNET_DATASTORE_ContinuationWithStatus cont,
 454                           void *cont_cls,
 455                           struct GNUNET_TIME_Relative timeout)
 456 {
 457   struct ReserveMessage *rm;
 458
 459   rm = (struct ReserveMessage*) &h[1];
 460   rm->header.type = htons(GNUNET_MESSAGE_TYPE_DATASTORE_RESERVE);
 461   rm->header.size = htons(sizeof (struct ReserveMessage));
 462   rm->entries = htonl(entries);
 463   rm->amount = GNUNET_htonll(amount);
 464   transmit_for_status (h, cont, cont_cls, timeout);
 465 }
 466
 467
 468 /**
 469  * Signal that all of the data for which a reservation was made has
 470  * been stored and that whatever excess space might have been reserved
 471  * can now be released.
 472  *
 473  * @param h handle to the datastore
 474  * @param rid reservation ID (value of "success" in original continuation
 475  *        from the "reserve" function).
 476  * @param cont continuation to call when done
 477  * @param cont_cls closure for cont
 478  * @param timeout how long to wait at most for a response
 479  */
 480 void
 481 GNUNET_DATASTORE_release_reserve (struct GNUNET_DATASTORE_Handle *h,
 482                                   int rid,
 483                                   GNUNET_DATASTORE_ContinuationWithStatus cont,
 484                                   void *cont_cls,
 485                                   struct GNUNET_TIME_Relative timeout)
 486 {
 487   struct ReleaseReserveMessage *rrm;
 488
 489   rrm = (struct ReleaseReserveMessage*) &h[1];
 490   rrm->header.type = htons(GNUNET_MESSAGE_TYPE_DATASTORE_RELEASE_RESERVE);
 491   rrm->header.size = htons(sizeof (struct ReleaseReserveMessage));
 492   rrm->rid = htonl(rid);
 493   transmit_for_status (h, cont, cont_cls, timeout);
 494 }
 495
 496
 497 /**
 498  * Update a value in the datastore.
 499  *
 500  * @param h handle to the datastore
 501  * @param uid identifier for the value
 502  * @param priority how much to increase the priority of the value
 503  * @param expiration new expiration value should be MAX of existing and this argument
 504  * @param cont continuation to call when done
 505  * @param cont_cls closure for cont
 506  * @param timeout how long to wait at most for a response
 507  */
 508 void
 509 GNUNET_DATASTORE_update (struct GNUNET_DATASTORE_Handle *h,
 510                          unsigned long long uid,
 511                          uint32_t priority,
 512                          struct GNUNET_TIME_Absolute expiration,
 513                          GNUNET_DATASTORE_ContinuationWithStatus cont,
 514                          void *cont_cls,
 515                          struct GNUNET_TIME_Relative timeout)
 516 {
 517   struct UpdateMessage *um;
 518
 519   um = (struct UpdateMessage*) &h[1];
 520   um->header.type = htons(GNUNET_MESSAGE_TYPE_DATASTORE_UPDATE);
 521   um->header.size = htons(sizeof (struct UpdateMessage));
 522   um->priority = htonl(priority);
 523   um->expiration = GNUNET_TIME_absolute_hton(expiration);
 524   um->uid = GNUNET_htonll(uid);
 525   transmit_for_status (h, cont, cont_cls, timeout);
 526 }
 527
 528
 529 /**
 530  * Helper function that will initiate the transmission of a message to
 531  * the datastore service.  The message must already be prepared and
 532  * stored in the buffer at the end of the handle.  The message must be
 533  * of a type that expects a "DataMessage" in response.
 534  *
 535  * @param h handle to the service with prepared message
 536  * @param cont function to call with result
 537  * @param cont_cls closure
 538  * @param timeout timeout for the operation
 539  */
 540 static void
 541 transmit_for_result (struct GNUNET_DATASTORE_Handle *h,
 542                      GNUNET_DATASTORE_Iterator cont,
 543                      void *cont_cls,
 544                      struct GNUNET_TIME_Relative timeout);
 545
 546
 547 /**
 548  * Type of a function to call when we receive a message
 549  * from the service.  This specific function is used
 550  * to handle messages of type "struct DataMessage".
 551  *
 552  * @param cls closure
 553  * @param msg message received, NULL on timeout or fatal error
 554  */
 555 static void
 556 with_result_response_handler (void *cls,
 557                               const struct
 558                               GNUNET_MessageHeader * msg)
 559 {
 560   struct GNUNET_DATASTORE_Handle *h = cls;
 561   GNUNET_DATASTORE_Iterator cont = h->response_proc;
 562   const struct DataMessage *dm;
 563   size_t msize;
 564   struct GNUNET_TIME_Relative remaining;
 565
 566   if (msg == NULL)
 567     {
 568 #if DEBUG_DATASTORE
 569       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 570                   "Got disconnected from datastore\n");
 571 #endif
 572       h->response_proc = NULL;
 573       GNUNET_CLIENT_disconnect (h->client, GNUNET_NO);
 574       h->client = GNUNET_CLIENT_connect (h->sched, "datastore", h->cfg);
 575       remaining = GNUNET_TIME_absolute_get_remaining (h->timeout);
 576       if (remaining.value > 0)
 577         {
 578           transmit_for_result (h,
 579                                cont,
 580                                h->response_proc_cls,
 581                                remaining);
 582         }
 583       else
 584         {
 585           h->message_size = 0;
 586           cont (h->response_proc_cls,
 587                 NULL, 0, NULL, 0, 0, 0,
 588                 GNUNET_TIME_UNIT_ZERO_ABS, 0);
 589         }
 590       return;
 591     }
 592   h->message_size = 0;
 593   if (ntohs(msg->type) == GNUNET_MESSAGE_TYPE_DATASTORE_DATA_END)
 594     {
 595       GNUNET_break (ntohs(msg->size) == sizeof(struct GNUNET_MessageHeader));
 596       h->response_proc = NULL;
 597 #if DEBUG_DATASTORE
 598       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 599                   "Received end of result set\n");
 600 #endif
 601       cont (h->response_proc_cls,
 602             NULL, 0, NULL, 0, 0, 0,
 603             GNUNET_TIME_UNIT_ZERO_ABS, 0);
 604       return;
 605     }
 606   if ( (ntohs(msg->size) < sizeof(struct DataMessage)) ||
 607        (ntohs(msg->type) != GNUNET_MESSAGE_TYPE_DATASTORE_DATA) )
 608     {
 609       GNUNET_break (0);
 610       GNUNET_CLIENT_disconnect (h->client, GNUNET_NO);
 611       h->client = GNUNET_CLIENT_connect (h->sched, "datastore", h->cfg);
 612       h->response_proc = NULL;
 613       cont (h->response_proc_cls,
 614             NULL, 0, NULL, 0, 0, 0,
 615             GNUNET_TIME_UNIT_ZERO_ABS, 0);
 616       return;
 617     }
 618   dm = (const struct DataMessage*) msg;
 619   msize = ntohl(dm->size);
 620   if (ntohs(msg->size) != msize + sizeof(struct DataMessage))
 621     {
 622       GNUNET_break (0);
 623       GNUNET_CLIENT_disconnect (h->client, GNUNET_NO);
 624       h->client = GNUNET_CLIENT_connect (h->sched, "datastore", h->cfg);
 625       h->response_proc = NULL;
 626       cont (h->response_proc_cls,
 627             NULL, 0, NULL, 0, 0, 0,
 628             GNUNET_TIME_UNIT_ZERO_ABS, 0);
 629       return;
 630     }
 631 #if DEBUG_DATASTORE
 632   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 633               "Received result %llu with type %u and size %u with key %s\n",
 634               (unsigned long long) GNUNET_ntohll(dm->uid),
 635               ntohl(dm->type),
 636               msize,
 637               GNUNET_h2s(&dm->key));
 638 #endif
 639   cont (h->response_proc_cls,
 640         &dm->key,
 641         msize,
 642         &dm[1],
 643         ntohl(dm->type),
 644         ntohl(dm->priority),
 645         ntohl(dm->anonymity),
 646         GNUNET_TIME_absolute_ntoh(dm->expiration),
 647         GNUNET_ntohll(dm->uid));
 648 }
 649
 650
 651 /**
 652  * Function called to trigger obtaining the next result
 653  * from the datastore.
 654  *
 655  * @param h handle to the datastore
 656  * @param more GNUNET_YES to get moxre results, GNUNET_NO to abort
 657  *        iteration (with a final call to "iter" with key/data == NULL).
 658  */
 659 void
 660 GNUNET_DATASTORE_get_next (struct GNUNET_DATASTORE_Handle *h,
 661                            int more)
 662 {
 663   GNUNET_DATASTORE_Iterator cont;
 664
 665   if (GNUNET_YES == more)
 666     {
 667       GNUNET_CLIENT_receive (h->client,
 668                              &with_result_response_handler,
 669                              h,
 670                              GNUNET_TIME_absolute_get_remaining (h->timeout));
 671       return;
 672     }
 673   cont = h->response_proc;
 674   h->response_proc = NULL;
 675   GNUNET_CLIENT_disconnect (h->client, GNUNET_NO);
 676   h->client = GNUNET_CLIENT_connect (h->sched, "datastore", h->cfg);
 677   cont (h->response_proc_cls,
 678         NULL, 0, NULL, 0, 0, 0,
 679         GNUNET_TIME_UNIT_ZERO_ABS, 0);
 680 }
 681
 682
 683 /**
 684  * Helper function that will initiate the transmission of a message to
 685  * the datastore service.  The message must already be prepared and
 686  * stored in the buffer at the end of the handle.  The message must be
 687  * of a type that expects a "DataMessage" in response.
 688  *
 689  * @param h handle to the service with prepared message
 690  * @param cont function to call with result
 691  * @param cont_cls closure
 692  * @param timeout timeout for the operation
 693  */
 694 static void
 695 transmit_for_result (struct GNUNET_DATASTORE_Handle *h,
 696                      GNUNET_DATASTORE_Iterator cont,
 697                      void *cont_cls,
 698                      struct GNUNET_TIME_Relative timeout)
 699 {
 700   const struct GNUNET_MessageHeader *hdr;
 701   uint16_t msize;
 702
 703   GNUNET_assert (cont != NULL);
 704   hdr = (const struct GNUNET_MessageHeader*) &h[1];
 705   msize = ntohs(hdr->size);
 706 #if DEBUG_DATASTORE
 707   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 708               "Transmitting %u byte message of type %u to datastore service\n",
 709               msize,
 710               ntohs(hdr->type));
 711 #endif
 712   GNUNET_assert (h->response_proc == NULL);
 713   h->response_proc = cont;
 714   h->response_proc_cls = cont_cls;
 715   h->timeout = GNUNET_TIME_relative_to_absolute (timeout);
 716   h->message_size = msize;
 717   if (GNUNET_OK !=
 718       GNUNET_CLIENT_transmit_and_get_response (h->client,
 719                                                hdr,
 720                                                timeout,
 721                                                GNUNET_YES,
 722                                                &with_result_response_handler,
 723                                                h))
 724     {
 725       GNUNET_break (0);
 726       h->response_proc = NULL;
 727       h->message_size = 0;
 728       cont (h->response_proc_cls,
 729             NULL, 0, NULL, 0, 0, 0,
 730             GNUNET_TIME_UNIT_ZERO_ABS, 0);
 731     }
 732 }
 733
 734
 735 /**
 736  * Iterate over the results for a particular key
 737  * in the datastore.
 738  *
 739  * @param h handle to the datastore
 740  * @param key maybe NULL (to match all entries)
 741  * @param type desired type, 0 for any
 742  * @param iter function to call on each matching value;
 743  *        will be called once with a NULL value at the end
 744  * @param iter_cls closure for iter
 745  * @param timeout how long to wait at most for a response
 746  */
 747 void
 748 GNUNET_DATASTORE_get (struct GNUNET_DATASTORE_Handle *h,
 749                       const GNUNET_HashCode * key,
 750                       enum GNUNET_BLOCK_Type type,
 751                       GNUNET_DATASTORE_Iterator iter, void *iter_cls,
 752                       struct GNUNET_TIME_Relative timeout)
 753 {
 754   struct GetMessage *gm;
 755
 756 #if DEBUG_DATASTORE
 757   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 758               "Asked to look for data under key `%s'\n",
 759               GNUNET_h2s (key));
 760 #endif
 761   gm = (struct GetMessage*) &h[1];
 762   gm->header.type = htons(GNUNET_MESSAGE_TYPE_DATASTORE_GET);
 763   gm->type = htonl(type);
 764   if (key != NULL)
 765     {
 766       gm->header.size = htons(sizeof (struct GetMessage));
 767       gm->key = *key;
 768     }
 769   else
 770     {
 771       gm->header.size = htons(sizeof (struct GetMessage) - sizeof(GNUNET_HashCode));
 772     }
 773   GNUNET_assert (h->response_proc == NULL);
 774   transmit_for_result (h, iter, iter_cls, timeout);
 775 }
 776
 777
 778 /**
 779  * Get a random value from the datastore.
 780  *
 781  * @param h handle to the datastore
 782  * @param iter function to call on a random value; it
 783  *        will be called exactly once; if no values
 784  *        are available, the value will be NULL.
 785  * @param iter_cls closure for iter
 786  * @param timeout how long to wait at most for a response
 787  */
 788 void
 789 GNUNET_DATASTORE_get_random (struct GNUNET_DATASTORE_Handle *h,
 790                              GNUNET_DATASTORE_Iterator iter, void *iter_cls,
 791                              struct GNUNET_TIME_Relative timeout)
 792 {
 793   struct GNUNET_MessageHeader *m;
 794
 795   m = (struct GNUNET_MessageHeader*) &h[1];
 796   m->type = htons(GNUNET_MESSAGE_TYPE_DATASTORE_GET_RANDOM);
 797   m->size = htons(sizeof (struct GNUNET_MessageHeader));
 798   GNUNET_assert (h->response_proc == NULL);
 799   transmit_for_result (h, iter, iter_cls, timeout);
 800 }
 801
 802
 803 /**
 804  * Explicitly remove some content from the database.
 805  *
 806  * @param h handle to the datastore
 807  * @param key key for the value
 808  * @param size number of bytes in data
 809  * @param data content stored
 810  * @param cont continuation to call when done
 811  * @param cont_cls closure for cont
 812  * @param timeout how long to wait at most for a response
 813  */
 814 void
 815 GNUNET_DATASTORE_remove (struct GNUNET_DATASTORE_Handle *h,
 816                          const GNUNET_HashCode * key,
 817                          uint32_t size, const void *data,
 818                          GNUNET_DATASTORE_ContinuationWithStatus cont,
 819                          void *cont_cls,
 820                          struct GNUNET_TIME_Relative timeout)
 821 {
 822   struct DataMessage *dm;
 823   size_t msize;
 824
 825 #if DEBUG_DATASTORE
 826   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 827               "Asked to remove %u bytes of data under key `%s'\n",
 828               size,
 829               GNUNET_h2s (key));
 830 #endif
 831   msize = sizeof(struct DataMessage) + size;
 832   GNUNET_assert (msize <= GNUNET_SERVER_MAX_MESSAGE_SIZE);
 833   dm = (struct DataMessage*) &h[1];
 834   dm->header.type = htons(GNUNET_MESSAGE_TYPE_DATASTORE_REMOVE);
 835   dm->header.size = htons(msize);
 836   dm->rid = htonl(0);
 837   dm->size = htonl(size);
 838   dm->type = htonl(0);
 839   dm->priority = htonl(0);
 840   dm->anonymity = htonl(0);
 841   dm->uid = GNUNET_htonll(0);
 842   dm->expiration = GNUNET_TIME_absolute_hton(GNUNET_TIME_UNIT_ZERO_ABS);
 843   dm->key = *key;
 844   memcpy (&dm[1], data, size);
 845   transmit_for_status (h, cont, cont_cls, timeout);
 846 }
 847 #endif
 848
 849 /* end of datastore_api.c */