src/datastore/datastore_api.c

   1 /*
   2      This file is part of GNUnet
   3      (C) 2004, 2005, 2006, 2007, 2009, 2010, 2011 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 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., 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.  Implements
  24  *        a priority queue for requests (with timeouts).
  25  * @author Christian Grothoff
  26  */
  27 #include "platform.h"
  28 #include "gnunet_arm_service.h"
  29 #include "gnunet_constants.h"
  30 #include "gnunet_datastore_service.h"
  31 #include "gnunet_statistics_service.h"
  32 #include "datastore.h"
  33
  34 /**
  35  * If a client stopped asking for more results, how many more do
  36  * we receive from the DB before killing the connection?  Trade-off
  37  * between re-doing TCP handshakes and (needlessly) receiving
  38  * useless results.
  39  */
  40 #define MAX_EXCESS_RESULTS 8
  41
  42 /**
  43  * Context for processing status messages.
  44  */
  45 struct StatusContext
  46 {
  47   /**
  48    * Continuation to call with the status.
  49    */
  50   GNUNET_DATASTORE_ContinuationWithStatus cont;
  51
  52   /**
  53    * Closure for cont.
  54    */
  55   void *cont_cls;
  56
  57 };
  58
  59
  60 /**
  61  * Context for processing result messages.
  62  */
  63 struct ResultContext
  64 {
  65   /**
  66    * Iterator to call with the result.
  67    */
  68   GNUNET_DATASTORE_Iterator iter;
  69
  70   /**
  71    * Closure for iter.
  72    */
  73   void *iter_cls;
  74
  75 };
  76
  77
  78 /**
  79  *  Context for a queue operation.
  80  */
  81 union QueueContext
  82 {
  83
  84   struct StatusContext sc;
  85
  86   struct ResultContext rc;
  87
  88 };
  89
  90
  91
  92 /**
  93  * Entry in our priority queue.
  94  */
  95 struct GNUNET_DATASTORE_QueueEntry
  96 {
  97
  98   /**
  99    * This is a linked list.
 100    */
 101   struct GNUNET_DATASTORE_QueueEntry *next;
 102
 103   /**
 104    * This is a linked list.
 105    */
 106   struct GNUNET_DATASTORE_QueueEntry *prev;
 107
 108   /**
 109    * Handle to the master context.
 110    */
 111   struct GNUNET_DATASTORE_Handle *h;
 112
 113   /**
 114    * Response processor (NULL if we are not waiting for a response).
 115    * This struct should be used for the closure, function-specific
 116    * arguments can be passed via 'qc'.
 117    */
 118   GNUNET_CLIENT_MessageHandler response_proc;
 119
 120   /**
 121    * Function to call after transmission of the request.
 122    */
 123   GNUNET_DATASTORE_ContinuationWithStatus cont;
 124
 125   /**
 126    * Closure for 'cont'.
 127    */
 128   void *cont_cls;
 129
 130   /**
 131    * Context for the operation.
 132    */
 133   union QueueContext qc;
 134
 135   /**
 136    * Task for timeout signalling.
 137    */
 138   GNUNET_SCHEDULER_TaskIdentifier task;
 139
 140   /**
 141    * Timeout for the current operation.
 142    */
 143   struct GNUNET_TIME_Absolute timeout;
 144
 145   /**
 146    * Priority in the queue.
 147    */
 148   unsigned int priority;
 149
 150   /**
 151    * Maximum allowed length of queue (otherwise
 152    * this request should be discarded).
 153    */
 154   unsigned int max_queue;
 155
 156   /**
 157    * Number of bytes in the request message following
 158    * this struct.  32-bit value for nicer memory
 159    * access (and overall struct alignment).
 160    */
 161   uint32_t message_size;
 162
 163   /**
 164    * Has this message been transmitted to the service?
 165    * Only ever GNUNET_YES for the head of the queue.
 166    * Note that the overall struct should end at a
 167    * multiple of 64 bits.
 168    */
 169   int was_transmitted;
 170
 171   /**
 172    * Are we expecting a single message in response to this
 173    * request (and, if it is data, no 'END' message)?
 174    */
 175   int one_shot;
 176
 177 };
 178
 179 /**
 180  * Handle to the datastore service.
 181  */
 182 struct GNUNET_DATASTORE_Handle
 183 {
 184
 185   /**
 186    * Our configuration.
 187    */
 188   const struct GNUNET_CONFIGURATION_Handle *cfg;
 189
 190   /**
 191    * Current connection to the datastore service.
 192    */
 193   struct GNUNET_CLIENT_Connection *client;
 194
 195   /**
 196    * Handle for statistics.
 197    */
 198   struct GNUNET_STATISTICS_Handle *stats;
 199
 200   /**
 201    * Current transmit handle.
 202    */
 203   struct GNUNET_CLIENT_TransmitHandle *th;
 204
 205   /**
 206    * Current head of priority queue.
 207    */
 208   struct GNUNET_DATASTORE_QueueEntry *queue_head;
 209
 210   /**
 211    * Current tail of priority queue.
 212    */
 213   struct GNUNET_DATASTORE_QueueEntry *queue_tail;
 214
 215   /**
 216    * Task for trying to reconnect.
 217    */
 218   GNUNET_SCHEDULER_TaskIdentifier reconnect_task;
 219
 220   /**
 221    * How quickly should we retry?  Used for exponential back-off on
 222    * connect-errors.
 223    */
 224   struct GNUNET_TIME_Relative retry_time;
 225
 226   /**
 227    * Number of entries in the queue.
 228    */
 229   unsigned int queue_size;
 230
 231   /**
 232    * Number of results we're receiving for the current query
 233    * after application stopped to care.  Used to determine when
 234    * to reset the connection.
 235    */
 236   unsigned int result_count;
 237
 238   /**
 239    * Are we currently trying to receive from the service?
 240    */
 241   int in_receive;
 242
 243   /**
 244    * We should either receive (and ignore) an 'END' message or force a
 245    * disconnect for the next message from the service.
 246    */
 247   unsigned int expect_end_or_disconnect;
 248
 249 };
 250
 251
 252
 253 /**
 254  * Connect to the datastore service.
 255  *
 256  * @param cfg configuration to use
 257  * @return handle to use to access the service
 258  */
 259 struct GNUNET_DATASTORE_Handle *
 260 GNUNET_DATASTORE_connect (const struct
 261                           GNUNET_CONFIGURATION_Handle
 262                           *cfg)
 263 {
 264   struct GNUNET_CLIENT_Connection *c;
 265   struct GNUNET_DATASTORE_Handle *h;
 266
 267   c = GNUNET_CLIENT_connect ("datastore", cfg);
 268   if (c == NULL)
 269     return NULL; /* oops */
 270   h = GNUNET_malloc (sizeof(struct GNUNET_DATASTORE_Handle) +
 271                      GNUNET_SERVER_MAX_MESSAGE_SIZE - 1);
 272   h->client = c;
 273   h->cfg = cfg;
 274   h->stats = GNUNET_STATISTICS_create ("datastore-api",
 275                                        cfg);
 276   return h;
 277 }
 278
 279
 280 /**
 281  * Transmit DROP message to datastore service.
 282  *
 283  * @param cls the 'struct GNUNET_DATASTORE_Handle'
 284  * @param size number of bytes that can be copied to buf
 285  * @param buf where to copy the drop message
 286  * @return number of bytes written to buf
 287  */
 288 static size_t
 289 transmit_drop (void *cls,
 290                size_t size,
 291                void *buf)
 292 {
 293   struct GNUNET_DATASTORE_Handle *h = cls;
 294   struct GNUNET_MessageHeader *hdr;
 295
 296   if (buf == NULL)
 297     {
 298       GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 299                   _("Failed to transmit request to drop database.\n"));
 300       GNUNET_DATASTORE_disconnect (h, GNUNET_NO);
 301       return 0;
 302     }
 303   GNUNET_assert (size >= sizeof(struct GNUNET_MessageHeader));
 304   hdr = buf;
 305   hdr->size = htons(sizeof(struct GNUNET_MessageHeader));
 306   hdr->type = htons(GNUNET_MESSAGE_TYPE_DATASTORE_DROP);
 307   GNUNET_DATASTORE_disconnect (h, GNUNET_NO);
 308   return sizeof(struct GNUNET_MessageHeader);
 309 }
 310
 311
 312 /**
 313  * Disconnect from the datastore service (and free
 314  * associated resources).
 315  *
 316  * @param h handle to the datastore
 317  * @param drop set to GNUNET_YES to delete all data in datastore (!)
 318  */
 319 void
 320 GNUNET_DATASTORE_disconnect (struct GNUNET_DATASTORE_Handle *h,
 321                              int drop)
 322 {
 323   struct GNUNET_DATASTORE_QueueEntry *qe;
 324
 325   if (h->client != NULL)
 326     {
 327       GNUNET_CLIENT_disconnect (h->client, GNUNET_NO);
 328       h->client = NULL;
 329     }
 330   if (h->reconnect_task != GNUNET_SCHEDULER_NO_TASK)
 331     {
 332       GNUNET_SCHEDULER_cancel (h->reconnect_task);
 333       h->reconnect_task = GNUNET_SCHEDULER_NO_TASK;
 334     }
 335   while (NULL != (qe = h->queue_head))
 336     {
 337       GNUNET_assert (NULL != qe->response_proc);
 338       qe->response_proc (qe, NULL);
 339     }
 340   if (GNUNET_YES == drop)
 341     {
 342       h->client = GNUNET_CLIENT_connect ("datastore", h->cfg);
 343       if (h->client != NULL)
 344         {
 345           if (NULL !=
 346               GNUNET_CLIENT_notify_transmit_ready (h->client,
 347                                                    sizeof(struct GNUNET_MessageHeader),
 348                                                    GNUNET_TIME_UNIT_MINUTES,
 349                                                    GNUNET_YES,
 350                                                    &transmit_drop,
 351                                                    h))
 352             return;
 353           GNUNET_CLIENT_disconnect (h->client, GNUNET_NO);
 354         }
 355       GNUNET_break (0);
 356     }
 357   GNUNET_STATISTICS_destroy (h->stats,
 358                              GNUNET_NO);
 359   GNUNET_free (h);
 360 }
 361
 362
 363 /**
 364  * A request has timed out (before being transmitted to the service).
 365  *
 366  * @param cls the 'struct GNUNET_DATASTORE_QueueEntry'
 367  * @param tc scheduler context
 368  */
 369 static void
 370 timeout_queue_entry (void *cls,
 371                      const struct GNUNET_SCHEDULER_TaskContext *tc)
 372 {
 373   struct GNUNET_DATASTORE_QueueEntry *qe = cls;
 374
 375   GNUNET_STATISTICS_update (qe->h->stats,
 376                             gettext_noop ("# queue entry timeouts"),
 377                             1,
 378                             GNUNET_NO);
 379   qe->task = GNUNET_SCHEDULER_NO_TASK;
 380   GNUNET_assert (qe->was_transmitted == GNUNET_NO);
 381   qe->response_proc (qe, NULL);
 382 }
 383
 384
 385 /**
 386  * Create a new entry for our priority queue (and possibly discard other entires if
 387  * the queue is getting too long).
 388  *
 389  * @param h handle to the datastore
 390  * @param msize size of the message to queue
 391  * @param queue_priority priority of the entry
 392  * @param max_queue_size at what queue size should this request be dropped
 393  *        (if other requests of higher priority are in the queue)
 394  * @param timeout timeout for the operation
 395  * @param response_proc function to call with replies (can be NULL)
 396  * @param qc client context (NOT a closure for response_proc)
 397  * @return NULL if the queue is full (and this entry was dropped)
 398  */
 399 static struct GNUNET_DATASTORE_QueueEntry *
 400 make_queue_entry (struct GNUNET_DATASTORE_Handle *h,
 401                   size_t msize,
 402                   unsigned int queue_priority,
 403                   unsigned int max_queue_size,
 404                   struct GNUNET_TIME_Relative timeout,
 405                   GNUNET_CLIENT_MessageHandler response_proc,
 406                   const union QueueContext *qc)
 407 {
 408   struct GNUNET_DATASTORE_QueueEntry *ret;
 409   struct GNUNET_DATASTORE_QueueEntry *pos;
 410   unsigned int c;
 411
 412   c = 0;
 413   pos = h->queue_head;
 414   while ( (pos != NULL) &&
 415           (c < max_queue_size) &&
 416           (pos->priority >= queue_priority) )
 417     {
 418       c++;
 419       pos = pos->next;
 420     }
 421   ret = GNUNET_malloc (sizeof (struct GNUNET_DATASTORE_QueueEntry) + msize);
 422   ret->h = h;
 423   ret->response_proc = response_proc;
 424   ret->qc = *qc;
 425   ret->timeout = GNUNET_TIME_relative_to_absolute (timeout);
 426   ret->priority = queue_priority;
 427   ret->max_queue = max_queue_size;
 428   ret->message_size = msize;
 429   ret->was_transmitted = GNUNET_NO;
 430   if (pos == NULL)
 431     {
 432       /* append at the tail */
 433       pos = h->queue_tail;
 434     }
 435   else
 436     {
 437       pos = pos->prev;
 438       /* do not insert at HEAD if HEAD query was already
 439          transmitted and we are still receiving replies! */
 440       if ( (pos == NULL) &&
 441            (h->queue_head->was_transmitted) )
 442         pos = h->queue_head;
 443     }
 444   c++;
 445   GNUNET_STATISTICS_update (h->stats,
 446                             gettext_noop ("# queue entries created"),
 447                             1,
 448                             GNUNET_NO);
 449   GNUNET_CONTAINER_DLL_insert_after (h->queue_head,
 450                                      h->queue_tail,
 451                                      pos,
 452                                      ret);
 453   h->queue_size++;
 454   if (c > max_queue_size)
 455     {
 456       GNUNET_STATISTICS_update (h->stats,
 457                                 gettext_noop ("# queue overflows"),
 458                                 1,
 459                                 GNUNET_NO);
 460       response_proc (ret, NULL);
 461       return NULL;
 462     }
 463   ret->task = GNUNET_SCHEDULER_add_delayed (timeout,
 464                                             &timeout_queue_entry,
 465                                             ret);
 466   pos = ret->next;
 467   while (pos != NULL)
 468     {
 469       if (pos->max_queue < h->queue_size)
 470         {
 471           GNUNET_assert (pos->response_proc != NULL);
 472           pos->response_proc (pos, NULL);
 473           break;
 474         }
 475       pos = pos->next;
 476     }
 477   return ret;
 478 }
 479
 480
 481 /**
 482  * Process entries in the queue (or do nothing if we are already
 483  * doing so).
 484  *
 485  * @param h handle to the datastore
 486  */
 487 static void
 488 process_queue (struct GNUNET_DATASTORE_Handle *h);
 489
 490
 491 /**
 492  * Try reconnecting to the datastore service.
 493  *
 494  * @param cls the 'struct GNUNET_DATASTORE_Handle'
 495  * @param tc scheduler context
 496  */
 497 static void
 498 try_reconnect (void *cls,
 499                const struct GNUNET_SCHEDULER_TaskContext *tc)
 500 {
 501   struct GNUNET_DATASTORE_Handle *h = cls;
 502
 503   if (h->retry_time.rel_value < GNUNET_CONSTANTS_SERVICE_RETRY.rel_value)
 504     h->retry_time = GNUNET_CONSTANTS_SERVICE_RETRY;
 505   else
 506     h->retry_time = GNUNET_TIME_relative_multiply (h->retry_time, 2);
 507   if (h->retry_time.rel_value > GNUNET_CONSTANTS_SERVICE_TIMEOUT.rel_value)
 508     h->retry_time = GNUNET_CONSTANTS_SERVICE_TIMEOUT;
 509   h->reconnect_task = GNUNET_SCHEDULER_NO_TASK;
 510   h->client = GNUNET_CLIENT_connect ("datastore", h->cfg);
 511   if (h->client == NULL)
 512     {
 513       GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 514                   "DATASTORE reconnect failed (fatally)\n");
 515       return;
 516     }
 517   GNUNET_STATISTICS_update (h->stats,
 518                             gettext_noop ("# datastore connections (re)created"),
 519                             1,
 520                             GNUNET_NO);
 521 #if DEBUG_DATASTORE
 522   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 523               "Reconnected to DATASTORE\n");
 524 #endif
 525   process_queue (h);
 526 }
 527
 528
 529 /**
 530  * Disconnect from the service and then try reconnecting to the datastore service
 531  * after some delay.
 532  *
 533  * @param h handle to datastore to disconnect and reconnect
 534  */
 535 static void
 536 do_disconnect (struct GNUNET_DATASTORE_Handle *h)
 537 {
 538   if (h->client == NULL)
 539     {
 540 #if DEBUG_DATASTORE
 541       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 542                   "client NULL in disconnect, will not try to reconnect\n");
 543 #endif
 544       return;
 545     }
 546 #if 0
 547   GNUNET_STATISTICS_update (stats,
 548                             gettext_noop ("# reconnected to DATASTORE"),
 549                             1,
 550                             GNUNET_NO);
 551 #endif
 552   GNUNET_CLIENT_disconnect (h->client, GNUNET_NO);
 553   h->client = NULL;
 554   h->reconnect_task = GNUNET_SCHEDULER_add_delayed (h->retry_time,
 555                                                     &try_reconnect,
 556                                                     h);
 557 }
 558
 559
 560 /**
 561  * Transmit request from queue to datastore service.
 562  *
 563  * @param cls the 'struct GNUNET_DATASTORE_Handle'
 564  * @param size number of bytes that can be copied to buf
 565  * @param buf where to copy the drop message
 566  * @return number of bytes written to buf
 567  */
 568 static size_t
 569 transmit_request (void *cls,
 570                   size_t size,
 571                   void *buf)
 572 {
 573   struct GNUNET_DATASTORE_Handle *h = cls;
 574   struct GNUNET_DATASTORE_QueueEntry *qe;
 575   size_t msize;
 576
 577   h->th = NULL;
 578   if (NULL == (qe = h->queue_head))
 579     return 0; /* no entry in queue */
 580   if (buf == NULL)
 581     {
 582       GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 583                   _("Failed to transmit request to DATASTORE.\n"));
 584       GNUNET_STATISTICS_update (h->stats,
 585                                 gettext_noop ("# transmission request failures"),
 586                                 1,
 587                                 GNUNET_NO);
 588       do_disconnect (h);
 589       return 0;
 590     }
 591   if (size < (msize = qe->message_size))
 592     {
 593       process_queue (h);
 594       return 0;
 595     }
 596  #if DEBUG_DATASTORE
 597   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 598               "Transmitting %u byte request to DATASTORE\n",
 599               msize);
 600 #endif
 601   memcpy (buf, &qe[1], msize);
 602   qe->was_transmitted = GNUNET_YES;
 603   GNUNET_SCHEDULER_cancel (qe->task);
 604   qe->task = GNUNET_SCHEDULER_NO_TASK;
 605   h->in_receive = GNUNET_YES;
 606   GNUNET_CLIENT_receive (h->client,
 607                          qe->response_proc,
 608                          h,
 609                          GNUNET_TIME_absolute_get_remaining (qe->timeout));
 610   GNUNET_STATISTICS_update (h->stats,
 611                             gettext_noop ("# bytes sent to datastore"),
 612                             1,
 613                             GNUNET_NO);
 614   return msize;
 615 }
 616
 617
 618 /**
 619  * Process entries in the queue (or do nothing if we are already
 620  * doing so).
 621  *
 622  * @param h handle to the datastore
 623  */
 624 static void
 625 process_queue (struct GNUNET_DATASTORE_Handle *h)
 626 {
 627   struct GNUNET_DATASTORE_QueueEntry *qe;
 628
 629   if (NULL == (qe = h->queue_head))
 630     {
 631 #if DEBUG_DATASTORE > 1
 632       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 633                   "Queue empty\n");
 634 #endif
 635       return; /* no entry in queue */
 636     }
 637   if (qe->was_transmitted == GNUNET_YES)
 638     {
 639 #if DEBUG_DATASTORE > 1
 640       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 641                   "Head request already transmitted\n");
 642 #endif
 643       return; /* waiting for replies */
 644     }
 645   if (h->th != NULL)
 646     {
 647 #if DEBUG_DATASTORE > 1
 648       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 649                   "Pending transmission request\n");
 650 #endif
 651       return; /* request pending */
 652     }
 653   if (h->client == NULL)
 654     {
 655 #if DEBUG_DATASTORE > 1
 656       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 657                   "Not connected\n");
 658 #endif
 659       return; /* waiting for reconnect */
 660     }
 661 #if DEBUG_DATASTORE
 662   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 663               "Queueing %u byte request to DATASTORE\n",
 664               qe->message_size);
 665 #endif
 666   h->th = GNUNET_CLIENT_notify_transmit_ready (h->client,
 667                                                qe->message_size,
 668                                                GNUNET_TIME_absolute_get_remaining (qe->timeout),
 669                                                GNUNET_YES,
 670                                                &transmit_request,
 671                                                h);
 672 }
 673
 674
 675 /**
 676  * Dummy continuation used to do nothing (but be non-zero).
 677  *
 678  * @param cls closure
 679  * @param result result
 680  * @param emsg error message
 681  */
 682 static void
 683 drop_status_cont (void *cls, int32_t result, const char *emsg)
 684 {
 685   /* do nothing */
 686 }
 687
 688
 689 static void
 690 free_queue_entry (struct GNUNET_DATASTORE_QueueEntry *qe)
 691 {
 692   struct GNUNET_DATASTORE_Handle *h = qe->h;
 693
 694   GNUNET_CONTAINER_DLL_remove (h->queue_head,
 695                                h->queue_tail,
 696                                qe);
 697   if (qe->task != GNUNET_SCHEDULER_NO_TASK)
 698     {
 699       GNUNET_SCHEDULER_cancel (qe->task);
 700       qe->task = GNUNET_SCHEDULER_NO_TASK;
 701     }
 702   h->queue_size--;
 703   GNUNET_free (qe);
 704 }
 705
 706 /**
 707  * Type of a function to call when we receive a message
 708  * from the service.
 709  *
 710  * @param cls closure
 711  * @param msg message received, NULL on timeout or fatal error
 712  */
 713 static void
 714 process_status_message (void *cls,
 715                         const struct
 716                         GNUNET_MessageHeader * msg)
 717 {
 718   struct GNUNET_DATASTORE_Handle *h = cls;
 719   struct GNUNET_DATASTORE_QueueEntry *qe;
 720   struct StatusContext rc;
 721   const struct StatusMessage *sm;
 722   const char *emsg;
 723   int32_t status;
 724   int was_transmitted;
 725
 726   h->in_receive = GNUNET_NO;
 727   if (NULL == (qe = h->queue_head))
 728     {
 729       GNUNET_break (0);
 730       do_disconnect (h);
 731       return;
 732     }
 733   was_transmitted = qe->was_transmitted;
 734   rc = qe->qc.sc;
 735   if (msg == NULL)
 736     {
 737       free_queue_entry (qe);
 738       if (NULL == h->client)
 739         return; /* forced disconnect */
 740       if (rc.cont != NULL)
 741         rc.cont (rc.cont_cls,
 742                  GNUNET_SYSERR,
 743                  _("Failed to receive status response from database."));
 744       if (was_transmitted == GNUNET_YES)
 745         do_disconnect (h);
 746       return;
 747     }
 748   GNUNET_assert (GNUNET_YES == qe->was_transmitted);
 749   free_queue_entry (qe);
 750   if ( (ntohs(msg->size) < sizeof(struct StatusMessage)) ||
 751        (ntohs(msg->type) != GNUNET_MESSAGE_TYPE_DATASTORE_STATUS) )
 752     {
 753       GNUNET_break (0);
 754       h->retry_time = GNUNET_TIME_UNIT_ZERO;
 755       do_disconnect (h);
 756       if (rc.cont != NULL)
 757         rc.cont (rc.cont_cls,
 758                  GNUNET_SYSERR,
 759                  _("Error reading response from datastore service"));
 760       return;
 761     }
 762   sm = (const struct StatusMessage*) msg;
 763   status = ntohl(sm->status);
 764   emsg = NULL;
 765   if (ntohs(msg->size) > sizeof(struct StatusMessage))
 766     {
 767       emsg = (const char*) &sm[1];
 768       if (emsg[ntohs(msg->size) - sizeof(struct StatusMessage) - 1] != '\0')
 769         {
 770           GNUNET_break (0);
 771           emsg = _("Invalid error message received from datastore service");
 772         }
 773     }
 774   if ( (status == GNUNET_SYSERR) &&
 775        (emsg == NULL) )
 776     {
 777       GNUNET_break (0);
 778       emsg = _("Invalid error message received from datastore service");
 779     }
 780 #if DEBUG_DATASTORE
 781   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 782               "Received status %d/%s\n",
 783               (int) status,
 784               emsg);
 785 #endif
 786   GNUNET_STATISTICS_update (h->stats,
 787                             gettext_noop ("# status messages received"),
 788                             1,
 789                             GNUNET_NO);
 790   h->retry_time.rel_value = 0;
 791   process_queue (h);
 792   if (rc.cont != NULL)
 793     rc.cont (rc.cont_cls,
 794              status,
 795              emsg);
 796 }
 797
 798
 799 /**
 800  * Store an item in the datastore.  If the item is already present,
 801  * the priorities are summed up and the higher expiration time and
 802  * lower anonymity level is used.
 803  *
 804  * @param h handle to the datastore
 805  * @param rid reservation ID to use (from "reserve"); use 0 if no
 806  *            prior reservation was made
 807  * @param key key for the value
 808  * @param size number of bytes in data
 809  * @param data content stored
 810  * @param type type of the content
 811  * @param priority priority of the content
 812  * @param anonymity anonymity-level for the content
 813  * @param replication how often should the content be replicated to other peers?
 814  * @param expiration expiration time for the content
 815  * @param queue_priority ranking of this request in the priority queue
 816  * @param max_queue_size at what queue size should this request be dropped
 817  *        (if other requests of higher priority are in the queue)
 818  * @param timeout timeout for the operation
 819  * @param cont continuation to call when done
 820  * @param cont_cls closure for cont
 821  * @return NULL if the entry was not queued, otherwise a handle that can be used to
 822  *         cancel; note that even if NULL is returned, the callback will be invoked
 823  *         (or rather, will already have been invoked)
 824  */
 825 struct GNUNET_DATASTORE_QueueEntry *
 826 GNUNET_DATASTORE_put (struct GNUNET_DATASTORE_Handle *h,
 827                       uint32_t rid,
 828                       const GNUNET_HashCode * key,
 829                       size_t size,
 830                       const void *data,
 831                       enum GNUNET_BLOCK_Type type,
 832                       uint32_t priority,
 833                       uint32_t anonymity,
 834                       uint32_t replication,
 835                       struct GNUNET_TIME_Absolute expiration,
 836                       unsigned int queue_priority,
 837                       unsigned int max_queue_size,
 838                       struct GNUNET_TIME_Relative timeout,
 839                       GNUNET_DATASTORE_ContinuationWithStatus cont,
 840                       void *cont_cls)
 841 {
 842   struct GNUNET_DATASTORE_QueueEntry *qe;
 843   struct DataMessage *dm;
 844   size_t msize;
 845   union QueueContext qc;
 846
 847 #if DEBUG_DATASTORE
 848   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 849               "Asked to put %u bytes of data under key `%s'\n",
 850               size,
 851               GNUNET_h2s (key));
 852 #endif
 853   msize = sizeof(struct DataMessage) + size;
 854   GNUNET_assert (msize < GNUNET_SERVER_MAX_MESSAGE_SIZE);
 855   qc.sc.cont = cont;
 856   qc.sc.cont_cls = cont_cls;
 857   qe = make_queue_entry (h, msize,
 858                          queue_priority, max_queue_size, timeout,
 859                          &process_status_message, &qc);
 860   if (qe == NULL)
 861     {
 862 #if DEBUG_DATASTORE
 863       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 864                   "Could not create queue entry for PUT\n");
 865 #endif
 866       return NULL;
 867     }
 868   GNUNET_STATISTICS_update (h->stats,
 869                             gettext_noop ("# PUT requests executed"),
 870                             1,
 871                             GNUNET_NO);
 872   dm = (struct DataMessage* ) &qe[1];
 873   dm->header.type = htons(GNUNET_MESSAGE_TYPE_DATASTORE_PUT);
 874   dm->header.size = htons(msize);
 875   dm->rid = htonl(rid);
 876   dm->size = htonl( (uint32_t) size);
 877   dm->type = htonl(type);
 878   dm->priority = htonl(priority);
 879   dm->anonymity = htonl(anonymity);
 880   dm->uid = GNUNET_htonll(0);
 881   dm->expiration = GNUNET_TIME_absolute_hton(expiration);
 882   dm->key = *key;
 883   memcpy (&dm[1], data, size);
 884   process_queue (h);
 885   return qe;
 886 }
 887
 888
 889 /**
 890  * Reserve space in the datastore.  This function should be used
 891  * to avoid "out of space" failures during a longer sequence of "put"
 892  * operations (for example, when a file is being inserted).
 893  *
 894  * @param h handle to the datastore
 895  * @param amount how much space (in bytes) should be reserved (for content only)
 896  * @param entries how many entries will be created (to calculate per-entry overhead)
 897  * @param queue_priority ranking of this request in the priority queue
 898  * @param max_queue_size at what queue size should this request be dropped
 899  *        (if other requests of higher priority are in the queue)
 900  * @param timeout how long to wait at most for a response (or before dying in queue)
 901  * @param cont continuation to call when done; "success" will be set to
 902  *             a positive reservation value if space could be reserved.
 903  * @param cont_cls closure for cont
 904  * @return NULL if the entry was not queued, otherwise a handle that can be used to
 905  *         cancel; note that even if NULL is returned, the callback will be invoked
 906  *         (or rather, will already have been invoked)
 907  */
 908 struct GNUNET_DATASTORE_QueueEntry *
 909 GNUNET_DATASTORE_reserve (struct GNUNET_DATASTORE_Handle *h,
 910                           uint64_t amount,
 911                           uint32_t entries,
 912                           unsigned int queue_priority,
 913                           unsigned int max_queue_size,
 914                           struct GNUNET_TIME_Relative timeout,
 915                           GNUNET_DATASTORE_ContinuationWithStatus cont,
 916                           void *cont_cls)
 917 {
 918   struct GNUNET_DATASTORE_QueueEntry *qe;
 919   struct ReserveMessage *rm;
 920   union QueueContext qc;
 921
 922   if (cont == NULL)
 923     cont = &drop_status_cont;
 924 #if DEBUG_DATASTORE
 925   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 926               "Asked to reserve %llu bytes of data and %u entries'\n",
 927               (unsigned long long) amount,
 928               (unsigned int) entries);
 929 #endif
 930   qc.sc.cont = cont;
 931   qc.sc.cont_cls = cont_cls;
 932   qe = make_queue_entry (h, sizeof(struct ReserveMessage),
 933                          queue_priority, max_queue_size, timeout,
 934                          &process_status_message, &qc);
 935   if (qe == NULL)
 936     {
 937 #if DEBUG_DATASTORE
 938       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 939                   "Could not create queue entry to reserve\n");
 940 #endif
 941       return NULL;
 942     }
 943   GNUNET_STATISTICS_update (h->stats,
 944                             gettext_noop ("# RESERVE requests executed"),
 945                             1,
 946                             GNUNET_NO);
 947   rm = (struct ReserveMessage*) &qe[1];
 948   rm->header.type = htons(GNUNET_MESSAGE_TYPE_DATASTORE_RESERVE);
 949   rm->header.size = htons(sizeof (struct ReserveMessage));
 950   rm->entries = htonl(entries);
 951   rm->amount = GNUNET_htonll(amount);
 952   process_queue (h);
 953   return qe;
 954 }
 955
 956
 957 /**
 958  * Signal that all of the data for which a reservation was made has
 959  * been stored and that whatever excess space might have been reserved
 960  * can now be released.
 961  *
 962  * @param h handle to the datastore
 963  * @param rid reservation ID (value of "success" in original continuation
 964  *        from the "reserve" function).
 965  * @param queue_priority ranking of this request in the priority queue
 966  * @param max_queue_size at what queue size should this request be dropped
 967  *        (if other requests of higher priority are in the queue)
 968  * @param queue_priority ranking of this request in the priority queue
 969  * @param max_queue_size at what queue size should this request be dropped
 970  *        (if other requests of higher priority are in the queue)
 971  * @param timeout how long to wait at most for a response
 972  * @param cont continuation to call when done
 973  * @param cont_cls closure for cont
 974  * @return NULL if the entry was not queued, otherwise a handle that can be used to
 975  *         cancel; note that even if NULL is returned, the callback will be invoked
 976  *         (or rather, will already have been invoked)
 977  */
 978 struct GNUNET_DATASTORE_QueueEntry *
 979 GNUNET_DATASTORE_release_reserve (struct GNUNET_DATASTORE_Handle *h,
 980                                   uint32_t rid,
 981                                   unsigned int queue_priority,
 982                                   unsigned int max_queue_size,
 983                                   struct GNUNET_TIME_Relative timeout,
 984                                   GNUNET_DATASTORE_ContinuationWithStatus cont,
 985                                   void *cont_cls)
 986 {
 987   struct GNUNET_DATASTORE_QueueEntry *qe;
 988   struct ReleaseReserveMessage *rrm;
 989   union QueueContext qc;
 990
 991   if (cont == NULL)
 992     cont = &drop_status_cont;
 993 #if DEBUG_DATASTORE
 994   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 995               "Asked to release reserve %d\n",
 996               rid);
 997 #endif
 998   qc.sc.cont = cont;
 999   qc.sc.cont_cls = cont_cls;
1000   qe = make_queue_entry (h, sizeof(struct ReleaseReserveMessage),
1001                          queue_priority, max_queue_size, timeout,
1002                          &process_status_message, &qc);
1003   if (qe == NULL)
1004     {
1005 #if DEBUG_DATASTORE
1006       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1007                   "Could not create queue entry to release reserve\n");
1008 #endif
1009       return NULL;
1010     }
1011   GNUNET_STATISTICS_update (h->stats,
1012                             gettext_noop ("# RELEASE RESERVE requests executed"),
1013                             1,
1014                             GNUNET_NO);
1015   rrm = (struct ReleaseReserveMessage*) &qe[1];
1016   rrm->header.type = htons(GNUNET_MESSAGE_TYPE_DATASTORE_RELEASE_RESERVE);
1017   rrm->header.size = htons(sizeof (struct ReleaseReserveMessage));
1018   rrm->rid = htonl(rid);
1019   process_queue (h);
1020   return qe;
1021 }
1022
1023
1024 /**
1025  * Update a value in the datastore.
1026  *
1027  * @param h handle to the datastore
1028  * @param uid identifier for the value
1029  * @param priority how much to increase the priority of the value
1030  * @param expiration new expiration value should be MAX of existing and this argument
1031  * @param queue_priority ranking of this request in the priority queue
1032  * @param max_queue_size at what queue size should this request be dropped
1033  *        (if other requests of higher priority are in the queue)
1034  * @param timeout how long to wait at most for a response
1035  * @param cont continuation to call when done
1036  * @param cont_cls closure for cont
1037  * @return NULL if the entry was not queued, otherwise a handle that can be used to
1038  *         cancel; note that even if NULL is returned, the callback will be invoked
1039  *         (or rather, will already have been invoked)
1040  */
1041 struct GNUNET_DATASTORE_QueueEntry *
1042 GNUNET_DATASTORE_update (struct GNUNET_DATASTORE_Handle *h,
1043                          uint64_t uid,
1044                          uint32_t priority,
1045                          struct GNUNET_TIME_Absolute expiration,
1046                          unsigned int queue_priority,
1047                          unsigned int max_queue_size,
1048                          struct GNUNET_TIME_Relative timeout,
1049                          GNUNET_DATASTORE_ContinuationWithStatus cont,
1050                          void *cont_cls)
1051 {
1052   struct GNUNET_DATASTORE_QueueEntry *qe;
1053   struct UpdateMessage *um;
1054   union QueueContext qc;
1055
1056   if (cont == NULL)
1057     cont = &drop_status_cont;
1058 #if DEBUG_DATASTORE
1059   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1060               "Asked to update entry %llu raising priority by %u and expiration to %llu\n",
1061               uid,
1062               (unsigned int) priority,
1063               (unsigned long long) expiration.abs_value);
1064 #endif
1065   qc.sc.cont = cont;
1066   qc.sc.cont_cls = cont_cls;
1067   qe = make_queue_entry (h, sizeof(struct UpdateMessage),
1068                          queue_priority, max_queue_size, timeout,
1069                          &process_status_message, &qc);
1070   if (qe == NULL)
1071     {
1072 #if DEBUG_DATASTORE
1073       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1074                   "Could not create queue entry for UPDATE\n");
1075 #endif
1076       return NULL;
1077     }
1078   GNUNET_STATISTICS_update (h->stats,
1079                             gettext_noop ("# UPDATE requests executed"),
1080                             1,
1081                             GNUNET_NO);
1082   um = (struct UpdateMessage*) &qe[1];
1083   um->header.type = htons(GNUNET_MESSAGE_TYPE_DATASTORE_UPDATE);
1084   um->header.size = htons(sizeof (struct UpdateMessage));
1085   um->priority = htonl(priority);
1086   um->expiration = GNUNET_TIME_absolute_hton(expiration);
1087   um->uid = GNUNET_htonll(uid);
1088   process_queue (h);
1089   return qe;
1090 }
1091
1092
1093 /**
1094  * Explicitly remove some content from the database.
1095  * The "cont"inuation will be called with status
1096  * "GNUNET_OK" if content was removed, "GNUNET_NO"
1097  * if no matching entry was found and "GNUNET_SYSERR"
1098  * on all other types of errors.
1099  *
1100  * @param h handle to the datastore
1101  * @param key key for the value
1102  * @param size number of bytes in data
1103  * @param data content stored
1104  * @param queue_priority ranking of this request in the priority queue
1105  * @param max_queue_size at what queue size should this request be dropped
1106  *        (if other requests of higher priority are in the queue)
1107  * @param timeout how long to wait at most for a response
1108  * @param cont continuation to call when done
1109  * @param cont_cls closure for cont
1110  * @return NULL if the entry was not queued, otherwise a handle that can be used to
1111  *         cancel; note that even if NULL is returned, the callback will be invoked
1112  *         (or rather, will already have been invoked)
1113  */
1114 struct GNUNET_DATASTORE_QueueEntry *
1115 GNUNET_DATASTORE_remove (struct GNUNET_DATASTORE_Handle *h,
1116                          const GNUNET_HashCode *key,
1117                          size_t size,
1118                          const void *data,
1119                          unsigned int queue_priority,
1120                          unsigned int max_queue_size,
1121                          struct GNUNET_TIME_Relative timeout,
1122                          GNUNET_DATASTORE_ContinuationWithStatus cont,
1123                          void *cont_cls)
1124 {
1125   struct GNUNET_DATASTORE_QueueEntry *qe;
1126   struct DataMessage *dm;
1127   size_t msize;
1128   union QueueContext qc;
1129
1130   if (cont == NULL)
1131     cont = &drop_status_cont;
1132 #if DEBUG_DATASTORE
1133   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1134               "Asked to remove %u bytes under key `%s'\n",
1135               size,
1136               GNUNET_h2s (key));
1137 #endif
1138   qc.sc.cont = cont;
1139   qc.sc.cont_cls = cont_cls;
1140   msize = sizeof(struct DataMessage) + size;
1141   GNUNET_assert (msize < GNUNET_SERVER_MAX_MESSAGE_SIZE);
1142   qe = make_queue_entry (h, msize,
1143                          queue_priority, max_queue_size, timeout,
1144                          &process_status_message, &qc);
1145   if (qe == NULL)
1146     {
1147 #if DEBUG_DATASTORE
1148       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1149                   "Could not create queue entry for REMOVE\n");
1150 #endif
1151       return NULL;
1152     }
1153   GNUNET_STATISTICS_update (h->stats,
1154                             gettext_noop ("# REMOVE requests executed"),
1155                             1,
1156                             GNUNET_NO);
1157   dm = (struct DataMessage*) &qe[1];
1158   dm->header.type = htons(GNUNET_MESSAGE_TYPE_DATASTORE_REMOVE);
1159   dm->header.size = htons(msize);
1160   dm->rid = htonl(0);
1161   dm->size = htonl(size);
1162   dm->type = htonl(0);
1163   dm->priority = htonl(0);
1164   dm->anonymity = htonl(0);
1165   dm->uid = GNUNET_htonll(0);
1166   dm->expiration = GNUNET_TIME_absolute_hton(GNUNET_TIME_UNIT_ZERO_ABS);
1167   dm->key = *key;
1168   memcpy (&dm[1], data, size);
1169   process_queue (h);
1170   return qe;
1171 }
1172
1173
1174 /**
1175  * Type of a function to call when we receive a message
1176  * from the service.
1177  *
1178  * @param cls closure
1179  * @param msg message received, NULL on timeout or fatal error
1180  */
1181 static void
1182 process_result_message (void *cls,
1183                         const struct GNUNET_MessageHeader *msg)
1184 {
1185   struct GNUNET_DATASTORE_Handle *h = cls;
1186   struct GNUNET_DATASTORE_QueueEntry *qe;
1187   struct ResultContext rc;
1188   const struct DataMessage *dm;
1189   int was_transmitted;
1190
1191   h->in_receive = GNUNET_NO;
1192   if (msg == NULL)
1193     {
1194       if (NULL != (qe = h->queue_head))
1195         {
1196           was_transmitted = qe->was_transmitted;
1197           free_queue_entry (qe);
1198           rc = qe->qc.rc;
1199           if (was_transmitted == GNUNET_YES)
1200             {
1201               GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
1202                           _("Failed to receive response from database.\n"));
1203               do_disconnect (h);
1204             }
1205           else
1206             {
1207 #if DEBUG_DATASTORE
1208               GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1209                           "Request dropped due to finite datastore queue length.\n");
1210 #endif
1211             }
1212           if (rc.iter != NULL)
1213             rc.iter (rc.iter_cls,
1214                      NULL, 0, NULL, 0, 0, 0,
1215                      GNUNET_TIME_UNIT_ZERO_ABS, 0);
1216         }
1217       return;
1218     }
1219   if (ntohs(msg->type) == GNUNET_MESSAGE_TYPE_DATASTORE_DATA_END)
1220     {
1221       GNUNET_break (ntohs(msg->size) == sizeof(struct GNUNET_MessageHeader));
1222       if (h->expect_end_or_disconnect > 0)
1223         {
1224           h->expect_end_or_disconnect--;
1225           process_queue (h);
1226           return;
1227         }
1228       qe = h->queue_head;
1229       rc = qe->qc.rc;
1230       GNUNET_assert (GNUNET_YES == qe->was_transmitted);
1231       free_queue_entry (qe);
1232 #if DEBUG_DATASTORE
1233       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1234                   "Received end of result set, new queue size is %u\n",
1235                   h->queue_size);
1236 #endif
1237       if (rc.iter != NULL)
1238         rc.iter (rc.iter_cls,
1239                  NULL, 0, NULL, 0, 0, 0,
1240                  GNUNET_TIME_UNIT_ZERO_ABS, 0);
1241       h->retry_time.rel_value = 0;
1242       h->result_count = 0;
1243       process_queue (h);
1244       return;
1245     }
1246   if (h->expect_end_or_disconnect > 0)
1247     {
1248       /* only 'END' allowed, must reconnect */
1249       h->retry_time = GNUNET_TIME_UNIT_ZERO;
1250       do_disconnect (h);
1251       return;
1252     }
1253   qe = h->queue_head;
1254   rc = qe->qc.rc;
1255   GNUNET_assert (GNUNET_YES == qe->was_transmitted);
1256   if ( (ntohs(msg->size) < sizeof(struct DataMessage)) ||
1257        (ntohs(msg->type) != GNUNET_MESSAGE_TYPE_DATASTORE_DATA) ||
1258        (ntohs(msg->size) != sizeof(struct DataMessage) + ntohl (((const struct DataMessage*)msg)->size)) )
1259     {
1260       GNUNET_break (0);
1261       free_queue_entry (qe);
1262       h->retry_time = GNUNET_TIME_UNIT_ZERO;
1263       do_disconnect (h);
1264       if (rc.iter != NULL)
1265         rc.iter (rc.iter_cls,
1266                  NULL, 0, NULL, 0, 0, 0,
1267                  GNUNET_TIME_UNIT_ZERO_ABS, 0);
1268       return;
1269     }
1270   GNUNET_STATISTICS_update (h->stats,
1271                             gettext_noop ("# Results received"),
1272                             1,
1273                             GNUNET_NO);
1274   if (rc.iter == NULL)
1275     {
1276       h->result_count++;
1277       GNUNET_STATISTICS_update (h->stats,
1278                                 gettext_noop ("# Excess results received"),
1279                                 1,
1280                                 GNUNET_NO);
1281       if (h->result_count > MAX_EXCESS_RESULTS)
1282         {
1283           free_queue_entry (qe);
1284           GNUNET_STATISTICS_update (h->stats,
1285                                     gettext_noop ("# Forced database connection resets"),
1286                                     1,
1287                                     GNUNET_NO);
1288           h->retry_time = GNUNET_TIME_UNIT_ZERO;
1289           do_disconnect (h);
1290           return;
1291         }
1292       if (GNUNET_YES == qe->one_shot)
1293         free_queue_entry (qe);
1294       else
1295         GNUNET_DATASTORE_iterate_get_next (h);
1296       return;
1297     }
1298   dm = (const struct DataMessage*) msg;
1299 #if DEBUG_DATASTORE
1300   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1301               "Received result %llu with type %u and size %u with key %s\n",
1302               (unsigned long long) GNUNET_ntohll(dm->uid),
1303               ntohl(dm->type),
1304               ntohl(dm->size),
1305               GNUNET_h2s(&dm->key));
1306 #endif
1307   if (GNUNET_YES == qe->one_shot)
1308     free_queue_entry (qe);
1309   h->retry_time.rel_value = 0;
1310   rc.iter (rc.iter_cls,
1311            &dm->key,
1312            ntohl(dm->size),
1313            &dm[1],
1314            ntohl(dm->type),
1315            ntohl(dm->priority),
1316            ntohl(dm->anonymity),
1317            GNUNET_TIME_absolute_ntoh(dm->expiration),
1318            GNUNET_ntohll(dm->uid));
1319 }
1320
1321
1322 /**
1323  * Get a random value from the datastore for content replication.
1324  * Returns a single, random value among those with the highest
1325  * replication score, lowering positive replication scores by one for
1326  * the chosen value (if only content with a replication score exists,
1327  * a random value is returned and replication scores are not changed).
1328  *
1329  * @param h handle to the datastore
1330  * @param queue_priority ranking of this request in the priority queue
1331  * @param max_queue_size at what queue size should this request be dropped
1332  *        (if other requests of higher priority are in the queue)
1333  * @param timeout how long to wait at most for a response
1334  * @param iter function to call on a random value; it
1335  *        will be called once with a value (if available)
1336  *        and always once with a value of NULL.
1337  * @param iter_cls closure for iter
1338  * @return NULL if the entry was not queued, otherwise a handle that can be used to
1339  *         cancel; note that even if NULL is returned, the callback will be invoked
1340  *         (or rather, will already have been invoked)
1341  */
1342 struct GNUNET_DATASTORE_QueueEntry *
1343 GNUNET_DATASTORE_get_for_replication (struct GNUNET_DATASTORE_Handle *h,
1344                                       unsigned int queue_priority,
1345                                       unsigned int max_queue_size,
1346                                       struct GNUNET_TIME_Relative timeout,
1347                                       GNUNET_DATASTORE_Iterator iter,
1348                                       void *iter_cls)
1349 {
1350   struct GNUNET_DATASTORE_QueueEntry *qe;
1351   struct GNUNET_MessageHeader *m;
1352   union QueueContext qc;
1353
1354 #if DEBUG_DATASTORE
1355   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1356               "Asked to get replication entry in %llu ms\n",
1357               (unsigned long long) timeout.rel_value);
1358 #endif
1359   qc.rc.iter = iter;
1360   qc.rc.iter_cls = iter_cls;
1361   qe = make_queue_entry (h, sizeof(struct GNUNET_MessageHeader),
1362                          queue_priority, max_queue_size, timeout,
1363                          &process_result_message, &qc);
1364   if (qe == NULL)
1365     {
1366 #if DEBUG_DATASTORE
1367       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1368                   "Could not create queue entry for GET REPLICATION\n");
1369 #endif
1370       return NULL;
1371     }
1372   qe->one_shot = GNUNET_YES;
1373   GNUNET_STATISTICS_update (h->stats,
1374                             gettext_noop ("# GET REPLICATION requests executed"),
1375                             1,
1376                             GNUNET_NO);
1377   m = (struct GNUNET_MessageHeader*) &qe[1];
1378   m->type = htons(GNUNET_MESSAGE_TYPE_DATASTORE_GET_REPLICATION);
1379   m->size = htons(sizeof (struct GNUNET_MessageHeader));
1380   process_queue (h);
1381   return qe;
1382 }
1383
1384
1385 /**
1386  * Get a zero-anonymity value from the datastore.
1387  *
1388  * @param h handle to the datastore
1389  * @param queue_priority ranking of this request in the priority queue
1390  * @param max_queue_size at what queue size should this request be dropped
1391  *        (if other requests of higher priority are in the queue)
1392  * @param timeout how long to wait at most for a response
1393  * @param type allowed type for the operation
1394  * @param iter function to call on a random value; it
1395  *        will be called once with a value (if available)
1396  *        and always once with a value of NULL.
1397  * @param iter_cls closure for iter
1398  * @return NULL if the entry was not queued, otherwise a handle that can be used to
1399  *         cancel; note that even if NULL is returned, the callback will be invoked
1400  *         (or rather, will already have been invoked)
1401  */
1402 struct GNUNET_DATASTORE_QueueEntry *
1403 GNUNET_DATASTORE_iterate_zero_anonymity (struct GNUNET_DATASTORE_Handle *h,
1404                                          unsigned int queue_priority,
1405                                          unsigned int max_queue_size,
1406                                          struct GNUNET_TIME_Relative timeout,
1407                                          enum GNUNET_BLOCK_Type type,
1408                                          GNUNET_DATASTORE_Iterator iter,
1409                                          void *iter_cls)
1410 {
1411   struct GNUNET_DATASTORE_QueueEntry *qe;
1412   struct GetZeroAnonymityMessage *m;
1413   union QueueContext qc;
1414
1415   GNUNET_assert (type != GNUNET_BLOCK_TYPE_ANY);
1416 #if DEBUG_DATASTORE
1417   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1418               "Asked to get zero-anonymity entry in %llu ms\n",
1419               (unsigned long long) timeout.rel_value);
1420 #endif
1421   qc.rc.iter = iter;
1422   qc.rc.iter_cls = iter_cls;
1423   qe = make_queue_entry (h, sizeof(struct GetZeroAnonymityMessage),
1424                          queue_priority, max_queue_size, timeout,
1425                          &process_result_message, &qc);
1426   if (qe == NULL)
1427     {
1428 #if DEBUG_DATASTORE
1429       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1430                   "Could not create queue entry for zero-anonymity iteration\n");
1431 #endif
1432       return NULL;
1433     }
1434   GNUNET_STATISTICS_update (h->stats,
1435                             gettext_noop ("# GET ZERO ANONYMITY requests executed"),
1436                             1,
1437                             GNUNET_NO);
1438   m = (struct GetZeroAnonymityMessage*) &qe[1];
1439   m->header.type = htons(GNUNET_MESSAGE_TYPE_DATASTORE_GET_ZERO_ANONYMITY);
1440   m->header.size = htons(sizeof (struct GetZeroAnonymityMessage));
1441   m->type = htonl ((uint32_t) type);
1442   process_queue (h);
1443   return qe;
1444 }
1445
1446
1447
1448 /**
1449  * Iterate over the results for a particular key
1450  * in the datastore.  The iterator will only be called
1451  * once initially; if the first call did contain a
1452  * result, further results can be obtained by calling
1453  * "GNUNET_DATASTORE_iterate_get_next" with the given argument.
1454  *
1455  * @param h handle to the datastore
1456  * @param key maybe NULL (to match all entries)
1457  * @param type desired type, 0 for any
1458  * @param queue_priority ranking of this request in the priority queue
1459  * @param max_queue_size at what queue size should this request be dropped
1460  *        (if other requests of higher priority are in the queue)
1461  * @param timeout how long to wait at most for a response
1462  * @param iter function to call on each matching value;
1463  *        will be called once with a NULL value at the end
1464  * @param iter_cls closure for iter
1465  * @return NULL if the entry was not queued, otherwise a handle that can be used to
1466  *         cancel; note that even if NULL is returned, the callback will be invoked
1467  *         (or rather, will already have been invoked)
1468  */
1469 struct GNUNET_DATASTORE_QueueEntry *
1470 GNUNET_DATASTORE_iterate_key (struct GNUNET_DATASTORE_Handle *h,
1471                               const GNUNET_HashCode * key,
1472                               enum GNUNET_BLOCK_Type type,
1473                               unsigned int queue_priority,
1474                               unsigned int max_queue_size,
1475                               struct GNUNET_TIME_Relative timeout,
1476                               GNUNET_DATASTORE_Iterator iter,
1477                               void *iter_cls)
1478 {
1479   struct GNUNET_DATASTORE_QueueEntry *qe;
1480   struct GetMessage *gm;
1481   union QueueContext qc;
1482
1483 #if DEBUG_DATASTORE
1484   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1485               "Asked to look for data of type %u under key `%s'\n",
1486               (unsigned int) type,
1487               GNUNET_h2s (key));
1488 #endif
1489   qc.rc.iter = iter;
1490   qc.rc.iter_cls = iter_cls;
1491   qe = make_queue_entry (h, sizeof(struct GetMessage),
1492                          queue_priority, max_queue_size, timeout,
1493                          &process_result_message, &qc);
1494   if (qe == NULL)
1495     {
1496 #if DEBUG_DATASTORE
1497       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1498                   "Could not queue request for `%s'\n",
1499                   GNUNET_h2s (key));
1500 #endif
1501       return NULL;
1502     }
1503   GNUNET_STATISTICS_update (h->stats,
1504                             gettext_noop ("# GET requests executed"),
1505                             1,
1506                             GNUNET_NO);
1507   gm = (struct GetMessage*) &qe[1];
1508   gm->header.type = htons(GNUNET_MESSAGE_TYPE_DATASTORE_GET);
1509   gm->type = htonl(type);
1510   if (key != NULL)
1511     {
1512       gm->header.size = htons(sizeof (struct GetMessage));
1513       gm->key = *key;
1514     }
1515   else
1516     {
1517       gm->header.size = htons(sizeof (struct GetMessage) - sizeof(GNUNET_HashCode));
1518     }
1519   process_queue (h);
1520   return qe;
1521 }
1522
1523
1524 /**
1525  * Function called to trigger obtaining the next result
1526  * from the datastore.
1527  *
1528  * @param h handle to the datastore
1529  */
1530 void
1531 GNUNET_DATASTORE_iterate_get_next (struct GNUNET_DATASTORE_Handle *h)
1532 {
1533   struct GNUNET_DATASTORE_QueueEntry *qe = h->queue_head;
1534
1535   h->in_receive = GNUNET_YES;
1536   GNUNET_CLIENT_receive (h->client,
1537                          &process_result_message,
1538                          h,
1539                          GNUNET_TIME_absolute_get_remaining (qe->timeout));
1540 }
1541
1542
1543 /**
1544  * Cancel a datastore operation.  The final callback from the
1545  * operation must not have been done yet.
1546  *
1547  * @param qe operation to cancel
1548  */
1549 void
1550 GNUNET_DATASTORE_cancel (struct GNUNET_DATASTORE_QueueEntry *qe)
1551 {
1552   struct GNUNET_DATASTORE_Handle *h;
1553
1554   h = qe->h;
1555 #if DEBUG_DATASTORE
1556   GNUNET_log  (GNUNET_ERROR_TYPE_DEBUG,
1557                "Pending DATASTORE request %p cancelled (%d, %d)\n",
1558                qe,
1559                qe->was_transmitted,
1560                h->queue_head == qe);
1561 #endif
1562   if (GNUNET_YES == qe->was_transmitted)
1563     {
1564       free_queue_entry (qe);
1565       h->expect_end_or_disconnect++;
1566       return;
1567     }
1568   free_queue_entry (qe);
1569   process_queue (h);
1570 }
1571
1572
1573 /* end of datastore_api.c */