src/statistics/statistics_api.c

   1 /*
   2      This file is part of GNUnet.
   3      (C) 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 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 statistics/statistics_api.c
  23  * @brief API of the statistics service
  24  * @author Christian Grothoff
  25  */
  26 #include "platform.h"
  27 #include "gnunet_client_lib.h"
  28 #include "gnunet_constants.h"
  29 #include "gnunet_container_lib.h"
  30 #include "gnunet_protocols.h"
  31 #include "gnunet_server_lib.h"
  32 #include "gnunet_statistics_service.h"
  33 #include "gnunet_strings_lib.h"
  34 #include "statistics.h"
  35
  36 /**
  37  * How long do we wait until a statistics request for setting
  38  * a value times out?  (The update will be lost if the
  39  * service does not react within this timeframe).
  40  */
  41 #define SET_TRANSMIT_TIMEOUT GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 2)
  42
  43 #define LOG(kind,...) GNUNET_log_from (kind, "statistics-api",__VA_ARGS__)
  44
  45 /**
  46  * Types of actions.
  47  */
  48 enum ActionType
  49 {
  50   /**
  51    * Get a value.
  52    */
  53   ACTION_GET,
  54
  55   /**
  56    * Set a value.
  57    */
  58   ACTION_SET,
  59
  60   /**
  61    * Update a value.
  62    */
  63   ACTION_UPDATE,
  64
  65   /**
  66    * Watch a value.
  67    */
  68   ACTION_WATCH
  69 };
  70
  71
  72 /**
  73  * Entry kept for each value we are watching.
  74  */
  75 struct GNUNET_STATISTICS_WatchEntry
  76 {
  77
  78   /**
  79    * What subsystem is this action about? (never NULL)
  80    */
  81   char *subsystem;
  82
  83   /**
  84    * What value is this action about? (never NULL)
  85    */
  86   char *name;
  87
  88   /**
  89    * Function to call
  90    */
  91   GNUNET_STATISTICS_Iterator proc;
  92
  93   /**
  94    * Closure for proc
  95    */
  96   void *proc_cls;
  97
  98 };
  99
 100
 101 /**
 102  * Linked list of things we still need to do.
 103  */
 104 struct GNUNET_STATISTICS_GetHandle
 105 {
 106
 107   /**
 108    * This is a doubly linked list.
 109    */
 110   struct GNUNET_STATISTICS_GetHandle *next;
 111
 112   /**
 113    * This is a doubly linked list.
 114    */
 115   struct GNUNET_STATISTICS_GetHandle *prev;
 116
 117   /**
 118    * Main statistics handle.
 119    */
 120   struct GNUNET_STATISTICS_Handle *sh;
 121
 122   /**
 123    * What subsystem is this action about? (can be NULL)
 124    */
 125   char *subsystem;
 126
 127   /**
 128    * What value is this action about? (can be NULL)
 129    */
 130   char *name;
 131
 132   /**
 133    * Continuation to call once action is complete.
 134    */
 135   GNUNET_STATISTICS_Callback cont;
 136
 137   /**
 138    * Function to call (for GET actions only).
 139    */
 140   GNUNET_STATISTICS_Iterator proc;
 141
 142   /**
 143    * Closure for proc and cont.
 144    */
 145   void *cls;
 146
 147   /**
 148    * Timeout for this action.
 149    */
 150   struct GNUNET_TIME_Absolute timeout;
 151
 152   /**
 153    * Associated value.
 154    */
 155   uint64_t value;
 156
 157   /**
 158    * Flag for SET/UPDATE actions.
 159    */
 160   int make_persistent;
 161
 162   /**
 163    * Has the current iteration been aborted; for GET actions.
 164    */
 165   int aborted;
 166
 167   /**
 168    * Is this a GET, SET, UPDATE or WATCH?
 169    */
 170   enum ActionType type;
 171
 172   /**
 173    * Size of the message that we will be transmitting.
 174    */
 175   uint16_t msize;
 176
 177 };
 178
 179
 180 /**
 181  * Handle for the service.
 182  */
 183 struct GNUNET_STATISTICS_Handle
 184 {
 185   /**
 186    * Name of our subsystem.
 187    */
 188   char *subsystem;
 189
 190   /**
 191    * Configuration to use.
 192    */
 193   const struct GNUNET_CONFIGURATION_Handle *cfg;
 194
 195   /**
 196    * Socket (if available).
 197    */
 198   struct GNUNET_CLIENT_Connection *client;
 199
 200   /**
 201    * Currently pending transmission request.
 202    */
 203   struct GNUNET_CLIENT_TransmitHandle *th;
 204
 205   /**
 206    * Head of the linked list of pending actions (first action
 207    * to be performed).
 208    */
 209   struct GNUNET_STATISTICS_GetHandle *action_head;
 210
 211   /**
 212    * Tail of the linked list of actions (for fast append).
 213    */
 214   struct GNUNET_STATISTICS_GetHandle *action_tail;
 215
 216   /**
 217    * Action we are currently busy with (action request has been
 218    * transmitted, we're now receiving the response from the
 219    * service).
 220    */
 221   struct GNUNET_STATISTICS_GetHandle *current;
 222
 223   /**
 224    * Array of watch entries.
 225    */
 226   struct GNUNET_STATISTICS_WatchEntry **watches;
 227
 228   /**
 229    * Task doing exponential back-off trying to reconnect.
 230    */
 231   GNUNET_SCHEDULER_TaskIdentifier backoff_task;
 232
 233   /**
 234    * Time for next connect retry.
 235    */
 236   struct GNUNET_TIME_Relative backoff;
 237
 238   /**
 239    * Size of the 'watches' array.
 240    */
 241   unsigned int watches_size;
 242
 243   /**
 244    * Should this handle auto-destruct once all actions have
 245    * been processed?
 246    */
 247   int do_destroy;
 248
 249   /**
 250    * Are we currently receiving from the service?
 251    */
 252   int receiving;
 253
 254 };
 255
 256
 257 /**
 258  * Schedule the next action to be performed.
 259  *
 260  * @param h statistics handle to reconnect
 261  */
 262 static void
 263 schedule_action (struct GNUNET_STATISTICS_Handle *h);
 264
 265
 266 /**
 267  * Transmit request to service that we want to watch
 268  * the development of a particular value.
 269  *
 270  * @param h statistics handle
 271  * @param watch watch entry of the value to watch
 272  */
 273 static void
 274 schedule_watch_request (struct GNUNET_STATISTICS_Handle *h,
 275                         struct GNUNET_STATISTICS_WatchEntry *watch)
 276 {
 277
 278   struct GNUNET_STATISTICS_GetHandle *ai;
 279   size_t slen;
 280   size_t nlen;
 281   size_t nsize;
 282
 283   GNUNET_assert (h != NULL);
 284   GNUNET_assert (watch != NULL);
 285
 286   slen = strlen (watch->subsystem) + 1;
 287   nlen = strlen (watch->name) + 1;
 288   nsize = sizeof (struct GNUNET_MessageHeader) + slen + nlen;
 289   if (nsize >= GNUNET_SERVER_MAX_MESSAGE_SIZE)
 290   {
 291     GNUNET_break (0);
 292     return;
 293   }
 294   ai = GNUNET_malloc (sizeof (struct GNUNET_STATISTICS_GetHandle));
 295   ai->sh = h;
 296   ai->subsystem = GNUNET_strdup (watch->subsystem);
 297   ai->name = GNUNET_strdup (watch->name);
 298   ai->timeout = GNUNET_TIME_UNIT_FOREVER_ABS;
 299   ai->msize = nsize;
 300   ai->type = ACTION_WATCH;
 301   ai->proc = watch->proc;
 302   ai->cls = watch->proc_cls;
 303   GNUNET_CONTAINER_DLL_insert_tail (h->action_head, h->action_tail,
 304                                     ai);
 305   schedule_action (h);
 306 }
 307
 308
 309 /**
 310  * Free memory associated with the given action item.
 311  *
 312  * @param gh action item to free
 313  */
 314 static void
 315 free_action_item (struct GNUNET_STATISTICS_GetHandle *gh)
 316 {
 317   GNUNET_free_non_null (gh->subsystem);
 318   GNUNET_free_non_null (gh->name);
 319   GNUNET_free (gh);
 320 }
 321
 322
 323 /**
 324  * Disconnect from the statistics service.
 325  *
 326  * @param h statistics handle to disconnect from
 327  */
 328 static void
 329 do_disconnect (struct GNUNET_STATISTICS_Handle *h)
 330 {
 331   struct GNUNET_STATISTICS_GetHandle *c;
 332
 333   if (NULL != h->th)
 334   {
 335     GNUNET_CLIENT_notify_transmit_ready_cancel (h->th);
 336     h->th = NULL;
 337   }
 338   if (NULL != h->client)
 339   {
 340     GNUNET_CLIENT_disconnect (h->client);
 341     h->client = NULL;
 342   }
 343   h->receiving = GNUNET_NO;
 344   if (NULL != (c = h->current))
 345   {
 346     h->current = NULL;
 347     if (c->cont != NULL)
 348       c->cont (c->cls, GNUNET_SYSERR);
 349     free_action_item (c);
 350   }
 351 }
 352
 353
 354 /**
 355  * Try to (re)connect to the statistics service.
 356  *
 357  * @param h statistics handle to reconnect
 358  * @return GNUNET_YES on success, GNUNET_NO on failure.
 359  */
 360 static int
 361 try_connect (struct GNUNET_STATISTICS_Handle *h)
 362 {
 363   struct GNUNET_STATISTICS_GetHandle *gh;
 364   struct GNUNET_STATISTICS_GetHandle *gn;
 365   unsigned int i;
 366
 367   if (h->backoff_task != GNUNET_SCHEDULER_NO_TASK)
 368     return GNUNET_NO;
 369   if (h->client != NULL)
 370     return GNUNET_YES;
 371   h->client = GNUNET_CLIENT_connect ("statistics", h->cfg);
 372   if (h->client != NULL)
 373   {
 374     gn = h->action_head;
 375     while (NULL != (gh = gn))
 376     {
 377       gn = gh->next;
 378       if (gh->type == ACTION_WATCH)
 379       {
 380         GNUNET_CONTAINER_DLL_remove (h->action_head,
 381                                      h->action_tail,
 382                                      gh);
 383         free_action_item (gh);
 384       }
 385     }
 386     for (i = 0; i < h->watches_size; i++)
 387     {
 388       if (NULL != h->watches[i])
 389         schedule_watch_request (h, h->watches[i]);
 390     }
 391     return GNUNET_YES;
 392   }
 393   LOG (GNUNET_ERROR_TYPE_DEBUG,
 394        "Failed to connect to statistics service!\n");
 395   return GNUNET_NO;
 396 }
 397
 398
 399 /**
 400  * We've waited long enough, reconnect now.
 401  *
 402  * @param cls the 'struct GNUNET_STATISTICS_Handle' to reconnect
 403  * @param tc scheduler context (unused)
 404  */
 405 static void
 406 reconnect_task (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 407 {
 408   struct GNUNET_STATISTICS_Handle *h = cls;
 409
 410   h->backoff_task = GNUNET_SCHEDULER_NO_TASK;
 411   schedule_action (h);
 412 }
 413
 414
 415
 416 /**
 417  * Task used by 'reconnect_later' to shutdown the handle
 418  *
 419  * @param cls the statistics handle
 420  * @param tc scheduler context
 421  */
 422 static void
 423 do_destroy (void *cls,
 424                const struct GNUNET_SCHEDULER_TaskContext *tc)
 425 {
 426   struct GNUNET_STATISTICS_Handle *h = cls;
 427
 428   GNUNET_STATISTICS_destroy (h, GNUNET_NO);
 429 }
 430
 431 /**
 432  * Reconnect at a later time, respecting back-off.
 433  *
 434  * @param h statistics handle
 435  */
 436 static void
 437 reconnect_later (struct GNUNET_STATISTICS_Handle *h)
 438 {
 439   GNUNET_assert (GNUNET_SCHEDULER_NO_TASK == h->backoff_task);
 440   if (h->do_destroy)
 441   {
 442     /* So we are shutting down and the service is not reachable.
 443      * Chances are that it's down for good and we are not going to connect to
 444      * it anymore.
 445      * Give up and don't sync the rest of the data.
 446      */
 447     GNUNET_break (0);
 448     h->do_destroy = GNUNET_NO;
 449     GNUNET_SCHEDULER_add_continuation (&do_destroy, h,
 450                                        GNUNET_SCHEDULER_REASON_PREREQ_DONE);
 451     return;
 452   }
 453   h->backoff_task =
 454     GNUNET_SCHEDULER_add_delayed (h->backoff, &reconnect_task, h);
 455   h->backoff = GNUNET_TIME_relative_multiply (h->backoff, 2);
 456   h->backoff =
 457     GNUNET_TIME_relative_min (h->backoff, GNUNET_CONSTANTS_SERVICE_TIMEOUT);
 458 }
 459
 460
 461 /**
 462  * Process a 'GNUNET_MESSAGE_TYPE_STATISTICS_VALUE' message.
 463  *
 464  * @param h statistics handle
 465  * @param msg message received from the service, never NULL
 466  * @return GNUNET_OK if the message was well-formed
 467  */
 468 static int
 469 process_statistics_value_message (struct GNUNET_STATISTICS_Handle *h,
 470                                   const struct GNUNET_MessageHeader *msg)
 471 {
 472   char *service;
 473   char *name;
 474   const struct GNUNET_STATISTICS_ReplyMessage *smsg;
 475   uint16_t size;
 476
 477   if (h->current->aborted)
 478   {
 479     LOG (GNUNET_ERROR_TYPE_DEBUG, "Iteration was aborted, ignoring VALUE\n");
 480     return GNUNET_OK;           /* don't bother */
 481   }
 482   size = ntohs (msg->size);
 483   if (size < sizeof (struct GNUNET_STATISTICS_ReplyMessage))
 484   {
 485     GNUNET_break (0);
 486     return GNUNET_SYSERR;
 487   }
 488   smsg = (const struct GNUNET_STATISTICS_ReplyMessage *) msg;
 489   size -= sizeof (struct GNUNET_STATISTICS_ReplyMessage);
 490   if (size !=
 491       GNUNET_STRINGS_buffer_tokenize ((const char *) &smsg[1], size, 2,
 492                                       &service, &name))
 493   {
 494     GNUNET_break (0);
 495     return GNUNET_SYSERR;
 496   }
 497   LOG (GNUNET_ERROR_TYPE_DEBUG, "Received valid statistic on `%s:%s': %llu\n",
 498        service, name, GNUNET_ntohll (smsg->value));
 499   if (GNUNET_OK !=
 500       h->current->proc (h->current->cls, service, name,
 501                         GNUNET_ntohll (smsg->value),
 502                         0 !=
 503                         (ntohl (smsg->uid) & GNUNET_STATISTICS_PERSIST_BIT)))
 504   {
 505     LOG (GNUNET_ERROR_TYPE_DEBUG,
 506          "Processing of remaining statistics aborted by client.\n");
 507     h->current->aborted = GNUNET_YES;
 508   }
 509   LOG (GNUNET_ERROR_TYPE_DEBUG, "VALUE processed successfully\n");
 510   return GNUNET_OK;
 511 }
 512
 513
 514 /**
 515  * We have received a watch value from the service.  Process it.
 516  *
 517  * @param h statistics handle
 518  * @param msg the watch value message
 519  * @return GNUNET_OK if the message was well-formed, GNUNET_SYSERR if not,
 520  *         GNUNET_NO if this watch has been cancelled
 521  */
 522 static int
 523 process_watch_value (struct GNUNET_STATISTICS_Handle *h,
 524                      const struct GNUNET_MessageHeader *msg)
 525 {
 526   const struct GNUNET_STATISTICS_WatchValueMessage *wvm;
 527   struct GNUNET_STATISTICS_WatchEntry *w;
 528   uint32_t wid;
 529
 530   if (sizeof (struct GNUNET_STATISTICS_WatchValueMessage) != ntohs (msg->size))
 531   {
 532     GNUNET_break (0);
 533     return GNUNET_SYSERR;
 534   }
 535   wvm = (const struct GNUNET_STATISTICS_WatchValueMessage *) msg;
 536   GNUNET_break (0 == ntohl (wvm->reserved));
 537   wid = ntohl (wvm->wid);
 538   if (wid >= h->watches_size)
 539   {
 540     GNUNET_break (0);
 541     return GNUNET_SYSERR;
 542   }
 543   w = h->watches[wid];
 544   if (NULL == w)
 545     return GNUNET_NO;
 546   (void) w->proc (w->proc_cls, w->subsystem, w->name,
 547                   GNUNET_ntohll (wvm->value),
 548                   0 != (ntohl (wvm->flags) & GNUNET_STATISTICS_PERSIST_BIT));
 549   return GNUNET_OK;
 550 }
 551
 552
 553 /**
 554  * Function called with messages from stats service.
 555  *
 556  * @param cls closure
 557  * @param msg message received, NULL on timeout or fatal error
 558  */
 559 static void
 560 receive_stats (void *cls, const struct GNUNET_MessageHeader *msg)
 561 {
 562   struct GNUNET_STATISTICS_Handle *h = cls;
 563   struct GNUNET_STATISTICS_GetHandle *c;
 564   int ret;
 565
 566   if (msg == NULL)
 567   {
 568     LOG (GNUNET_ERROR_TYPE_DEBUG | GNUNET_ERROR_TYPE_BULK,
 569          "Error receiving statistics from service, is the service running?\n");
 570     do_disconnect (h);
 571     reconnect_later (h);
 572     return;
 573   }
 574   switch (ntohs (msg->type))
 575   {
 576   case GNUNET_MESSAGE_TYPE_STATISTICS_END:
 577     LOG (GNUNET_ERROR_TYPE_DEBUG, "Received end of statistics marker\n");
 578     if (NULL == (c = h->current))
 579     {
 580       GNUNET_break (0);
 581       do_disconnect (h);
 582       reconnect_later (h);
 583       return;
 584     }
 585     h->backoff = GNUNET_TIME_UNIT_MILLISECONDS;
 586     if (h->watches_size > 0)
 587     {
 588       GNUNET_CLIENT_receive (h->client, &receive_stats, h,
 589                              GNUNET_TIME_UNIT_FOREVER_REL);
 590     }
 591     else
 592     {
 593       h->receiving = GNUNET_NO;
 594     }
 595     h->current = NULL;
 596     schedule_action (h);
 597     if (c->cont != NULL)
 598       c->cont (c->cls, GNUNET_OK);
 599     free_action_item (c);
 600     return;
 601   case GNUNET_MESSAGE_TYPE_STATISTICS_VALUE:
 602     if (GNUNET_OK != process_statistics_value_message (h, msg))
 603     {
 604       do_disconnect (h);
 605       reconnect_later (h);
 606       return;
 607     }
 608     /* finally, look for more! */
 609     LOG (GNUNET_ERROR_TYPE_DEBUG,
 610          "Processing VALUE done, now reading more\n");
 611     GNUNET_CLIENT_receive (h->client, &receive_stats, h,
 612                            GNUNET_TIME_absolute_get_remaining (h->
 613                                                                current->timeout));
 614     h->backoff = GNUNET_TIME_UNIT_MILLISECONDS;
 615     return;
 616   case GNUNET_MESSAGE_TYPE_STATISTICS_WATCH_VALUE:
 617     if (GNUNET_OK !=
 618         (ret = process_watch_value (h, msg)))
 619     {
 620       do_disconnect (h);
 621       if (GNUNET_NO == ret)
 622         h->backoff = GNUNET_TIME_UNIT_MILLISECONDS;
 623       reconnect_later (h);
 624       return;
 625     }
 626     h->backoff = GNUNET_TIME_UNIT_MILLISECONDS;
 627     GNUNET_assert (h->watches_size > 0);
 628     GNUNET_CLIENT_receive (h->client, &receive_stats, h,
 629                            GNUNET_TIME_UNIT_FOREVER_REL);
 630     return;
 631   default:
 632     GNUNET_break (0);
 633     do_disconnect (h);
 634     reconnect_later (h);
 635     return;
 636   }
 637 }
 638
 639
 640 /**
 641  * Transmit a GET request (and if successful, start to receive
 642  * the response).
 643  *
 644  * @param handle statistics handle
 645  * @param size how many bytes can we write to buf
 646  * @param buf where to write requests to the service
 647  * @return number of bytes written to buf
 648  */
 649 static size_t
 650 transmit_get (struct GNUNET_STATISTICS_Handle *handle, size_t size, void *buf)
 651 {
 652   struct GNUNET_STATISTICS_GetHandle *c;
 653   struct GNUNET_MessageHeader *hdr;
 654   size_t slen1;
 655   size_t slen2;
 656   uint16_t msize;
 657
 658   GNUNET_assert (NULL != (c = handle->current));
 659   if (buf == NULL)
 660   {
 661     /* timeout / error */
 662     LOG (GNUNET_ERROR_TYPE_DEBUG,
 663          "Transmission of request for statistics failed!\n");
 664     do_disconnect (handle);
 665     reconnect_later (handle);
 666     return 0;
 667   }
 668   slen1 = strlen (c->subsystem) + 1;
 669   slen2 = strlen (c->name) + 1;
 670   msize = slen1 + slen2 + sizeof (struct GNUNET_MessageHeader);
 671   GNUNET_assert (msize <= size);
 672   hdr = (struct GNUNET_MessageHeader *) buf;
 673   hdr->size = htons (msize);
 674   hdr->type = htons (GNUNET_MESSAGE_TYPE_STATISTICS_GET);
 675   GNUNET_assert (slen1 + slen2 ==
 676                  GNUNET_STRINGS_buffer_fill ((char *) &hdr[1], slen1 + slen2, 2,
 677                                              c->subsystem,
 678                                              c->name));
 679   if (GNUNET_YES != handle->receiving)
 680   {
 681     LOG (GNUNET_ERROR_TYPE_DEBUG,
 682          "Transmission of GET done, now reading response\n");
 683     handle->receiving = GNUNET_YES;
 684     GNUNET_CLIENT_receive (handle->client, &receive_stats, handle,
 685                            GNUNET_TIME_absolute_get_remaining (c->timeout));
 686   }
 687   return msize;
 688 }
 689
 690
 691 /**
 692  * Transmit a WATCH request (and if successful, start to receive
 693  * the response).
 694  *
 695  * @param handle statistics handle
 696  * @param size how many bytes can we write to buf
 697  * @param buf where to write requests to the service
 698  * @return number of bytes written to buf
 699  */
 700 static size_t
 701 transmit_watch (struct GNUNET_STATISTICS_Handle *handle, size_t size, void *buf)
 702 {
 703   struct GNUNET_MessageHeader *hdr;
 704   size_t slen1;
 705   size_t slen2;
 706   uint16_t msize;
 707
 708   if (buf == NULL)
 709   {
 710     /* timeout / error */
 711     LOG (GNUNET_ERROR_TYPE_DEBUG,
 712          "Transmission of request for statistics failed!\n");
 713     do_disconnect (handle);
 714     reconnect_later (handle);
 715     return 0;
 716   }
 717   LOG (GNUNET_ERROR_TYPE_DEBUG, "Transmitting watch request for `%s'\n",
 718        handle->current->name);
 719   slen1 = strlen (handle->current->subsystem) + 1;
 720   slen2 = strlen (handle->current->name) + 1;
 721   msize = slen1 + slen2 + sizeof (struct GNUNET_MessageHeader);
 722   GNUNET_assert (msize <= size);
 723   hdr = (struct GNUNET_MessageHeader *) buf;
 724   hdr->size = htons (msize);
 725   hdr->type = htons (GNUNET_MESSAGE_TYPE_STATISTICS_WATCH);
 726   GNUNET_assert (slen1 + slen2 ==
 727                  GNUNET_STRINGS_buffer_fill ((char *) &hdr[1], slen1 + slen2, 2,
 728                                              handle->current->subsystem,
 729                                              handle->current->name));
 730   if (GNUNET_YES != handle->receiving)
 731   {
 732     handle->receiving = GNUNET_YES;
 733     GNUNET_CLIENT_receive (handle->client, &receive_stats, handle,
 734                            GNUNET_TIME_UNIT_FOREVER_REL);
 735   }
 736   GNUNET_assert (NULL == handle->current->cont);
 737   free_action_item (handle->current);
 738   handle->current = NULL;
 739   return msize;
 740 }
 741
 742
 743 /**
 744  * Transmit a SET/UPDATE request.
 745  *
 746  * @param handle statistics handle
 747  * @param size how many bytes can we write to buf
 748  * @param buf where to write requests to the service
 749  * @return number of bytes written to buf
 750  */
 751 static size_t
 752 transmit_set (struct GNUNET_STATISTICS_Handle *handle, size_t size, void *buf)
 753 {
 754   struct GNUNET_STATISTICS_SetMessage *r;
 755   size_t slen;
 756   size_t nlen;
 757   size_t nsize;
 758
 759   if (NULL == buf)
 760   {
 761     do_disconnect (handle);
 762     reconnect_later (handle);
 763     return 0;
 764   }
 765   slen = strlen (handle->current->subsystem) + 1;
 766   nlen = strlen (handle->current->name) + 1;
 767   nsize = sizeof (struct GNUNET_STATISTICS_SetMessage) + slen + nlen;
 768   if (size < nsize)
 769   {
 770     GNUNET_break (0);
 771     do_disconnect (handle);
 772     reconnect_later (handle);
 773     return 0;
 774   }
 775   r = buf;
 776   r->header.size = htons (nsize);
 777   r->header.type = htons (GNUNET_MESSAGE_TYPE_STATISTICS_SET);
 778   r->flags = 0;
 779   r->value = GNUNET_htonll (handle->current->value);
 780   if (handle->current->make_persistent)
 781     r->flags |= htonl (GNUNET_STATISTICS_SETFLAG_PERSISTENT);
 782   if (handle->current->type == ACTION_UPDATE)
 783     r->flags |= htonl (GNUNET_STATISTICS_SETFLAG_RELATIVE);
 784   GNUNET_assert (slen + nlen ==
 785                  GNUNET_STRINGS_buffer_fill ((char *) &r[1], slen + nlen, 2,
 786                                              handle->current->subsystem,
 787                                              handle->current->name));
 788   GNUNET_assert (NULL == handle->current->cont);
 789   free_action_item (handle->current);
 790   handle->current = NULL;
 791   return nsize;
 792 }
 793
 794
 795 /**
 796  * Function called when we are ready to transmit a request to the service.
 797  *
 798  * @param cls the 'struct GNUNET_STATISTICS_Handle'
 799  * @param size how many bytes can we write to buf
 800  * @param buf where to write requests to the service
 801  * @return number of bytes written to buf
 802  */
 803 static size_t
 804 transmit_action (void *cls, size_t size, void *buf)
 805 {
 806   struct GNUNET_STATISTICS_Handle *h = cls;
 807   size_t ret;
 808
 809   h->th = NULL;
 810   ret = 0;
 811   if (NULL != h->current)
 812     switch (h->current->type)
 813     {
 814     case ACTION_GET:
 815       ret = transmit_get (h, size, buf);
 816       break;
 817     case ACTION_SET:
 818     case ACTION_UPDATE:
 819       ret = transmit_set (h, size, buf);
 820       break;
 821     case ACTION_WATCH:
 822       ret = transmit_watch (h, size, buf);
 823       break;
 824     default:
 825       GNUNET_assert (0);
 826       break;
 827     }
 828   schedule_action (h);
 829   return ret;
 830 }
 831
 832
 833 /**
 834  * Get handle for the statistics service.
 835  *
 836  * @param subsystem name of subsystem using the service
 837  * @param cfg services configuration in use
 838  * @return handle to use
 839  */
 840 struct GNUNET_STATISTICS_Handle *
 841 GNUNET_STATISTICS_create (const char *subsystem,
 842                           const struct GNUNET_CONFIGURATION_Handle *cfg)
 843 {
 844   struct GNUNET_STATISTICS_Handle *ret;
 845
 846   GNUNET_assert (subsystem != NULL);
 847   GNUNET_assert (cfg != NULL);
 848   ret = GNUNET_malloc (sizeof (struct GNUNET_STATISTICS_Handle));
 849   ret->cfg = cfg;
 850   ret->subsystem = GNUNET_strdup (subsystem);
 851   ret->backoff = GNUNET_TIME_UNIT_MILLISECONDS;
 852   return ret;
 853 }
 854
 855
 856 /**
 857  * Destroy a handle (free all state associated with
 858  * it).
 859  *
 860  * @param h statistics handle to destroy
 861  * @param sync_first set to GNUNET_YES if pending SET requests should
 862  *        be completed
 863  */
 864 void
 865 GNUNET_STATISTICS_destroy (struct GNUNET_STATISTICS_Handle *h, int sync_first)
 866 {
 867   struct GNUNET_STATISTICS_GetHandle *pos;
 868   struct GNUNET_STATISTICS_GetHandle *next;
 869   struct GNUNET_TIME_Relative timeout;
 870   int i;
 871
 872   if (h == NULL)
 873     return;
 874   GNUNET_assert (GNUNET_NO == h->do_destroy); // Don't call twice.
 875   if (GNUNET_SCHEDULER_NO_TASK != h->backoff_task)
 876   {
 877     GNUNET_SCHEDULER_cancel (h->backoff_task);
 878     h->backoff_task = GNUNET_SCHEDULER_NO_TASK;
 879   }
 880   if (sync_first)
 881   {
 882     if (h->current != NULL)
 883     {
 884       if (h->current->type == ACTION_GET)
 885       {
 886         GNUNET_CLIENT_notify_transmit_ready_cancel (h->th);
 887         h->th = NULL;
 888         free_action_item (h->current);
 889         h->current = NULL;
 890       }
 891     }
 892     next = h->action_head;
 893     while (NULL != (pos = next))
 894     {
 895       next = pos->next;
 896       if (pos->type == ACTION_GET)
 897       {
 898         GNUNET_CONTAINER_DLL_remove (h->action_head,
 899                                      h->action_tail,
 900                                      pos);
 901         free_action_item (pos);
 902       }
 903     }
 904     if ( (NULL == h->current) &&
 905          (NULL != (h->current = h->action_head)) )
 906       GNUNET_CONTAINER_DLL_remove (h->action_head,
 907                                    h->action_tail,
 908                                    h->current);
 909     h->do_destroy = GNUNET_YES;
 910     if ((h->current != NULL) && (h->th == NULL)&&
 911         (NULL != h->client))
 912     {
 913       timeout = GNUNET_TIME_absolute_get_remaining (h->current->timeout);
 914       h->th =
 915         GNUNET_CLIENT_notify_transmit_ready (h->client, h->current->msize,
 916                                              timeout, GNUNET_YES,
 917                                              &transmit_action, h);
 918       GNUNET_assert (NULL != h->th);
 919     }
 920     if (h->th != NULL)
 921       return; /* do not finish destruction just yet */
 922   }
 923   while (NULL != (pos = h->action_head))
 924   {
 925     GNUNET_CONTAINER_DLL_remove (h->action_head,
 926                                  h->action_tail,
 927                                  pos);
 928     free_action_item (pos);
 929   }
 930   do_disconnect (h);
 931   for (i = 0; i < h->watches_size; i++)
 932   {
 933     if (NULL == h->watches[i])
 934       continue;
 935     GNUNET_free (h->watches[i]->subsystem);
 936     GNUNET_free (h->watches[i]->name);
 937     GNUNET_free (h->watches[i]);
 938   }
 939   GNUNET_array_grow (h->watches, h->watches_size, 0);
 940   GNUNET_free (h->subsystem);
 941   GNUNET_free (h);
 942 }
 943
 944
 945 static void
 946 destroy_task (void *cls,
 947               const struct GNUNET_SCHEDULER_TaskContext *tc)
 948 {
 949   struct GNUNET_STATISTICS_Handle *h = cls;
 950
 951   GNUNET_STATISTICS_destroy (h, GNUNET_YES);
 952 }
 953
 954
 955 /**
 956  * Schedule the next action to be performed.
 957  *
 958  * @param h statistics handle
 959  */
 960 static void
 961 schedule_action (struct GNUNET_STATISTICS_Handle *h)
 962 {
 963   struct GNUNET_TIME_Relative timeout;
 964
 965   if ( (h->th != NULL) ||
 966        (h->backoff_task != GNUNET_SCHEDULER_NO_TASK) )
 967     return;                     /* action already pending */
 968   if (GNUNET_YES != try_connect (h))
 969   {
 970     reconnect_later (h);
 971     return;
 972   }
 973   if (NULL != h->current)
 974     return; /* action already pending */
 975   /* schedule next action */
 976   h->current = h->action_head;
 977   if (NULL == h->current)
 978   {
 979     if (h->do_destroy)
 980     {
 981       h->do_destroy = GNUNET_NO;
 982       GNUNET_SCHEDULER_add_continuation (&destroy_task, h,
 983                                          GNUNET_SCHEDULER_REASON_PREREQ_DONE);
 984     }
 985     return;
 986   }
 987   GNUNET_CONTAINER_DLL_remove (h->action_head, h->action_tail, h->current);
 988   timeout = GNUNET_TIME_absolute_get_remaining (h->current->timeout);
 989   if (NULL ==
 990       (h->th =
 991        GNUNET_CLIENT_notify_transmit_ready (h->client, h->current->msize,
 992                                             timeout, GNUNET_YES,
 993                                             &transmit_action, h)))
 994   {
 995     LOG (GNUNET_ERROR_TYPE_DEBUG,
 996          "Failed to transmit request to statistics service.\n");
 997     do_disconnect (h);
 998     reconnect_later (h);
 999   }
1000 }
1001
1002
1003 /**
1004  * Get statistic from the peer.
1005  *
1006  * @param handle identification of the statistics service
1007  * @param subsystem limit to the specified subsystem, NULL for our subsystem
1008  * @param name name of the statistic value, NULL for all values
1009  * @param timeout after how long should we give up (and call
1010  *        cont with an error code)?
1011  * @param cont continuation to call when done (can be NULL)
1012  * @param proc function to call on each value
1013  * @param cls closure for cont and proc
1014  * @return NULL on error
1015  */
1016 struct GNUNET_STATISTICS_GetHandle *
1017 GNUNET_STATISTICS_get (struct GNUNET_STATISTICS_Handle *handle,
1018                        const char *subsystem, const char *name,
1019                        struct GNUNET_TIME_Relative timeout,
1020                        GNUNET_STATISTICS_Callback cont,
1021                        GNUNET_STATISTICS_Iterator proc, void *cls)
1022 {
1023   size_t slen1;
1024   size_t slen2;
1025   struct GNUNET_STATISTICS_GetHandle *ai;
1026
1027   if (NULL == handle)
1028     return NULL;
1029   GNUNET_assert (proc != NULL);
1030   GNUNET_assert (GNUNET_NO == handle->do_destroy);
1031   if (subsystem == NULL)
1032     subsystem = "";
1033   if (name == NULL)
1034     name = "";
1035   slen1 = strlen (subsystem) + 1;
1036   slen2 = strlen (name) + 1;
1037   GNUNET_assert (slen1 + slen2 + sizeof (struct GNUNET_MessageHeader) <
1038                  GNUNET_SERVER_MAX_MESSAGE_SIZE);
1039   ai = GNUNET_malloc (sizeof (struct GNUNET_STATISTICS_GetHandle));
1040   ai->sh = handle;
1041   ai->subsystem = GNUNET_strdup (subsystem);
1042   ai->name = GNUNET_strdup (name);
1043   ai->cont = cont;
1044   ai->proc = proc;
1045   ai->cls = cls;
1046   ai->timeout = GNUNET_TIME_relative_to_absolute (timeout);
1047   ai->type = ACTION_GET;
1048   ai->msize = slen1 + slen2 + sizeof (struct GNUNET_MessageHeader);
1049   GNUNET_CONTAINER_DLL_insert_tail (handle->action_head, handle->action_tail,
1050                                     ai);
1051   schedule_action (handle);
1052   return ai;
1053 }
1054
1055
1056 /**
1057  * Cancel a 'get' request.  Must be called before the 'cont'
1058  * function is called.
1059  *
1060  * @param gh handle of the request to cancel
1061  */
1062 void
1063 GNUNET_STATISTICS_get_cancel (struct GNUNET_STATISTICS_GetHandle *gh)
1064 {
1065   if (NULL == gh)
1066     return;
1067   if (gh->sh->current == gh)
1068   {
1069     gh->aborted = GNUNET_YES;
1070   }
1071   else
1072   {
1073     GNUNET_CONTAINER_DLL_remove (gh->sh->action_head, gh->sh->action_tail, gh);
1074     GNUNET_free (gh->name);
1075     GNUNET_free (gh->subsystem);
1076     GNUNET_free (gh);
1077   }
1078 }
1079
1080
1081 /**
1082  * Watch statistics from the peer (be notified whenever they change).
1083  *
1084  * @param handle identification of the statistics service
1085  * @param subsystem limit to the specified subsystem, never NULL
1086  * @param name name of the statistic value, never NULL
1087  * @param proc function to call on each value
1088  * @param proc_cls closure for proc
1089  * @return GNUNET_OK on success, GNUNET_SYSERR on error
1090  */
1091 int
1092 GNUNET_STATISTICS_watch (struct GNUNET_STATISTICS_Handle *handle,
1093                          const char *subsystem, const char *name,
1094                          GNUNET_STATISTICS_Iterator proc, void *proc_cls)
1095 {
1096   struct GNUNET_STATISTICS_WatchEntry *w;
1097
1098   if (handle == NULL)
1099     return GNUNET_SYSERR;
1100   w = GNUNET_malloc (sizeof (struct GNUNET_STATISTICS_WatchEntry));
1101   w->subsystem = GNUNET_strdup (subsystem);
1102   w->name = GNUNET_strdup (name);
1103   w->proc = proc;
1104   w->proc_cls = proc_cls;
1105   GNUNET_array_append (handle->watches, handle->watches_size, w);
1106   schedule_watch_request (handle, w);
1107   return GNUNET_OK;
1108 }
1109
1110
1111 /**
1112  * Stop watching statistics from the peer.
1113  *
1114  * @param handle identification of the statistics service
1115  * @param subsystem limit to the specified subsystem, never NULL
1116  * @param name name of the statistic value, never NULL
1117  * @param proc function to call on each value
1118  * @param proc_cls closure for proc
1119  * @return GNUNET_OK on success, GNUNET_SYSERR on error (no such watch)
1120  */
1121 int
1122 GNUNET_STATISTICS_watch_cancel (struct GNUNET_STATISTICS_Handle *handle,
1123                                 const char *subsystem, const char *name,
1124                                 GNUNET_STATISTICS_Iterator proc, void *proc_cls)
1125 {
1126   struct GNUNET_STATISTICS_WatchEntry *w;
1127   unsigned int i;
1128
1129   if (handle == NULL)
1130     return GNUNET_SYSERR;
1131   for (i=0;i<handle->watches_size;i++)
1132   {
1133     w = handle->watches[i];
1134     if ( (w->proc == proc) &&
1135          (w->proc_cls == proc_cls) &&
1136          (0 == strcmp (w->name, name)) &&
1137          (0 == strcmp (w->subsystem, subsystem)) )
1138     {
1139       GNUNET_free (w->name);
1140       GNUNET_free (w->subsystem);
1141       GNUNET_free (w);
1142       handle->watches[i] = NULL;
1143       return GNUNET_OK;
1144     }
1145   }
1146   return GNUNET_SYSERR;
1147 }
1148
1149
1150
1151 /**
1152  * Queue a request to change a statistic.
1153  *
1154  * @param h statistics handle
1155  * @param name name of the value
1156  * @param make_persistent  should the value be kept across restarts?
1157  * @param value new value or change
1158  * @param type type of the action (ACTION_SET or ACTION_UPDATE)
1159  */
1160 static void
1161 add_setter_action (struct GNUNET_STATISTICS_Handle *h, const char *name,
1162                    int make_persistent, uint64_t value, enum ActionType type)
1163 {
1164   struct GNUNET_STATISTICS_GetHandle *ai;
1165   size_t slen;
1166   size_t nlen;
1167   size_t nsize;
1168   int64_t delta;
1169
1170   GNUNET_assert (h != NULL);
1171   GNUNET_assert (name != NULL);
1172   slen = strlen (h->subsystem) + 1;
1173   nlen = strlen (name) + 1;
1174   nsize = sizeof (struct GNUNET_STATISTICS_SetMessage) + slen + nlen;
1175   if (nsize >= GNUNET_SERVER_MAX_MESSAGE_SIZE)
1176   {
1177     GNUNET_break (0);
1178     return;
1179   }
1180   for (ai = h->action_head; ai != NULL; ai = ai->next)
1181   {
1182     if (! ( (0 == strcmp (ai->subsystem, h->subsystem)) &&
1183             (0 == strcmp (ai->name, name)) &&
1184             ( (ai->type == ACTION_UPDATE) ||
1185               (ai->type == ACTION_SET) ) ) )
1186       continue;
1187     if (ai->type == ACTION_SET)
1188     {
1189       if (type == ACTION_UPDATE)
1190       {
1191         delta = (int64_t) value;
1192         if (delta > 0)
1193         {
1194           /* update old set by new delta */
1195           ai->value += delta;
1196         }
1197         else
1198         {
1199           /* update old set by new delta, but never go negative */
1200           if (ai->value < -delta)
1201             ai->value = 0;
1202           else
1203             ai->value += delta;
1204         }
1205       }
1206       else
1207       {
1208         /* new set overrides old set */
1209         ai->value = value;
1210       }
1211     }
1212     else
1213     {
1214       if (type == ACTION_UPDATE)
1215       {
1216         /* make delta cummulative */
1217         delta = (int64_t) value;
1218         ai->value += delta;
1219       }
1220       else
1221       {
1222         /* drop old 'update', use new 'set' instead */
1223         ai->value = value;
1224         ai->type = type;
1225       }
1226     }
1227     ai->timeout = GNUNET_TIME_relative_to_absolute (SET_TRANSMIT_TIMEOUT);
1228     ai->make_persistent = make_persistent;
1229     return;
1230   }
1231   /* no existing entry matches, create a fresh one */
1232   ai = GNUNET_malloc (sizeof (struct GNUNET_STATISTICS_GetHandle));
1233   ai->sh = h;
1234   ai->subsystem = GNUNET_strdup (h->subsystem);
1235   ai->name = GNUNET_strdup (name);
1236   ai->timeout = GNUNET_TIME_relative_to_absolute (SET_TRANSMIT_TIMEOUT);
1237   ai->make_persistent = make_persistent;
1238   ai->msize = nsize;
1239   ai->value = value;
1240   ai->type = type;
1241   GNUNET_CONTAINER_DLL_insert_tail (h->action_head, h->action_tail,
1242                                     ai);
1243   schedule_action (h);
1244 }
1245
1246
1247 /**
1248  * Set statistic value for the peer.  Will always use our
1249  * subsystem (the argument used when "handle" was created).
1250  *
1251  * @param handle identification of the statistics service
1252  * @param name name of the statistic value
1253  * @param value new value to set
1254  * @param make_persistent should the value be kept across restarts?
1255  */
1256 void
1257 GNUNET_STATISTICS_set (struct GNUNET_STATISTICS_Handle *handle,
1258                        const char *name, uint64_t value, int make_persistent)
1259 {
1260   if (handle == NULL)
1261     return;
1262   GNUNET_assert (GNUNET_NO == handle->do_destroy);
1263   add_setter_action (handle, name, make_persistent, value, ACTION_SET);
1264 }
1265
1266
1267 /**
1268  * Set statistic value for the peer.  Will always use our
1269  * subsystem (the argument used when "handle" was created).
1270  *
1271  * @param handle identification of the statistics service
1272  * @param name name of the statistic value
1273  * @param delta change in value (added to existing value)
1274  * @param make_persistent should the value be kept across restarts?
1275  */
1276 void
1277 GNUNET_STATISTICS_update (struct GNUNET_STATISTICS_Handle *handle,
1278                           const char *name, int64_t delta, int make_persistent)
1279 {
1280   if (handle == NULL)
1281     return;
1282   if (delta == 0)
1283     return;
1284   GNUNET_assert (GNUNET_NO == handle->do_destroy);
1285   add_setter_action (handle, name, make_persistent, (uint64_t) delta,
1286                      ACTION_UPDATE);
1287 }
1288
1289
1290 /* end of statistics_api.c */