hacks from trip

5 files changed, 719 insertions, 57 deletions

diff --git a/src/statistics/gnunet-service-statistics.c b/src/statistics/gnunet-service-statistics.c
index 3ff751efe..251302a66 100644
--- a/src/statistics/gnunet-service-statistics.c
+++ b/src/statistics/gnunet-service-statistics.c
@@ -1,6 +1,6 @@
 /*
      This file is part of GNUnet.
-     (C) 2009 Christian Grothoff (and other contributing authors)
+     (C) 2009, 2010 Christian Grothoff (and other contributing authors)
 
      GNUnet is free software; you can redistribute it and/or modify
      it under the terms of the GNU General Public License as published
@@ -27,6 +27,7 @@
  * - use BIO for IO operations
  */
 #include "platform.h"
+#include "gnunet_container_lib.h"
 #include "gnunet_disk_lib.h"
 #include "gnunet_getopt_lib.h"
 #include "gnunet_protocols.h"
@@ -37,6 +38,41 @@
 #include "statistics.h"
 
 /**
+ * Watch entry.
+ */
+struct WatchEntry
+{
+
+  struct WatchEntry *next;
+
+  struct WatchEntry *prev;
+
+  struct GNUNET_SERVER_Client *client;
+
+  uint64_t last_value;
+  
+  uint32_t wid;
+
+};
+
+
+/**
+ * Client entry.
+ */
+struct ClientEntry
+{
+
+  struct ClientEntry *next;
+
+  struct ClientEntry *prev;
+
+  struct GNUNET_SERVER_Client *client;
+  
+  uint32_t max_wid;
+
+};
+
+/**
  * Entry in the statistics list.
  */
 struct StatsEntry
@@ -66,6 +102,18 @@ struct StatsEntry
   struct GNUNET_STATISTICS_SetMessage *msg;
 
   /**
+   * Watch context for changes to this
+   * value, or NULL for none.
+   */
+  struct WatchEntry *we_head;
+
+  /**
+   * Watch context for changes to this
+   * value, or NULL for none.
+   */
+  struct WatchEntry *we_tail;
+
+  /**
    * Our value.
    */
   uint64_t value;
@@ -92,6 +140,21 @@ static const struct GNUNET_CONFIGURATION_Handle *cfg;
  */
 static struct StatsEntry *start;
 
+static struct ClientEntry *client_head;
+
+static struct ClientEntry *client_tail;
+
+/**
+ * Our notification context.
+ */
+static struct GNUNET_SERVER_NotificationContext *nc;
+
+/**
+ * Counter used to generate unique values.
+ */
+static uint32_t uidgen;
+
+
 /**
  * Load persistent values from disk.  Disk format is
  * exactly the same format that we also use for
@@ -209,7 +272,7 @@ save ()
  * Transmit the given stats value.
  */
 static void
-transmit (struct GNUNET_SERVER_TransmitContext *tc,
+transmit (struct GNUNET_SERVER_Client *client,
           const struct StatsEntry *e)
 {
   struct GNUNET_STATISTICS_ReplyMessage *m;
@@ -232,10 +295,11 @@ transmit (struct GNUNET_SERVER_TransmitContext *tc,
                                                      2, e->service, e->name));
 #if DEBUG_STATISTICS
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-              "Transmitting value for `%s:%s': %llu\n",
-              e->service, e->name, e->value);
+              "Transmitting value for `%s:%s' (%d): %llu\n",
+              e->service, e->name,
+              e->persistent, e->value);
 #endif
-  GNUNET_SERVER_transmit_context_append_message (tc, &m->header);
+  GNUNET_SERVER_notification_context_unicast (nc, client, &m->header, GNUNET_NO);
   GNUNET_free (m);
 }
 
@@ -252,6 +316,32 @@ matches (const struct StatsEntry *e, const char *service, const char *name)
 }
 
 
+static struct ClientEntry *
+make_client_entry (struct GNUNET_SERVER_Client *client)
+{
+  struct ClientEntry *ce;
+
+  if (client == NULL)
+    return NULL;
+  ce = client_head;
+  while (ce != NULL)
+    {
+      if (ce->client == client)
+        return ce;
+      ce = ce->next;
+    }
+  ce = GNUNET_malloc (sizeof (struct ClientEntry));
+  ce->client = client;
+  GNUNET_SERVER_client_keep (client);
+  GNUNET_CONTAINER_DLL_insert (client_head,
+                               client_tail,
+                               ce);
+  GNUNET_SERVER_notification_context_add (nc,
+                                          client);
+  return ce;
+}
+
+
 /**
  * Handle GET-message.
  *
@@ -266,12 +356,13 @@ handle_get (void *cls,
             struct GNUNET_SERVER_Client *client,
             const struct GNUNET_MessageHeader *message)
 {
+  struct GNUNET_MessageHeader end;
   char *service;
   char *name;
   struct StatsEntry *pos;
-  struct GNUNET_SERVER_TransmitContext *tc;
   size_t size;
 
+  make_client_entry (client);
   size = ntohs (message->size) - sizeof (struct GNUNET_MessageHeader);
   if (size != GNUNET_STRINGS_buffer_tokenize ((const char *) &message[1],
                                               size, 2, &service, &name))
@@ -285,20 +376,51 @@ handle_get (void *cls,
               "Received request for statistics on `%s:%s'\n",
               strlen (service) ? service : "*", strlen (name) ? name : "*");
 #endif
-  tc = GNUNET_SERVER_transmit_context_create (client);
   pos = start;
   while (pos != NULL)
     {
       if (matches (pos, service, name))
-        transmit (tc, pos);
+        transmit (client, pos);
       pos = pos->next;
     }
-  GNUNET_SERVER_transmit_context_append_data (tc, NULL, 0,
-                                              GNUNET_MESSAGE_TYPE_STATISTICS_END);
-  GNUNET_SERVER_transmit_context_run (tc, GNUNET_TIME_UNIT_FOREVER_REL);
+  end.size = htons (sizeof (struct GNUNET_MessageHeader));
+  end.type = htons (GNUNET_MESSAGE_TYPE_STATISTICS_END);
+  GNUNET_SERVER_notification_context_unicast (nc,
+                                              client,
+                                              &end,
+                                              GNUNET_NO);
+  GNUNET_SERVER_receive_done (client,
+                              GNUNET_OK);
 }
 
 
+static void
+notify_change (struct StatsEntry *se)
+{
+  struct GNUNET_STATISTICS_WatchValueMessage wvm;
+  struct WatchEntry *pos;
+
+  pos = se->we_head;
+  while (pos != NULL)
+    {
+      if (pos->last_value != se->value)
+        {
+          wvm.header.type = htons (GNUNET_MESSAGE_TYPE_STATISTICS_WATCH_VALUE);
+          wvm.header.size = htons (sizeof (struct GNUNET_STATISTICS_WatchValueMessage));
+          wvm.flags = htonl (se->persistent ? GNUNET_STATISTICS_PERSIST_BIT : 0);
+          wvm.wid = htonl (pos->wid);
+          wvm.reserved = htonl (0);
+          wvm.value = GNUNET_htonll (se->value);
+          GNUNET_SERVER_notification_context_unicast (nc,
+                                                      pos->client,
+                                                      &wvm.header,
+                                                      GNUNET_NO);
+          pos->last_value = se->value;
+        }
+      pos = pos->next;
+    }
+}
+
 /**
  * Handle SET-message.
  *
@@ -311,11 +433,6 @@ handle_set (void *cls,
             struct GNUNET_SERVER_Client *client,
             const struct GNUNET_MessageHeader *message)
 {
-  /**
-   * Counter used to generate unique values.
-   */
-  static uint32_t uidgen;
-
   char *service;
   char *name;
   uint16_t msize;
@@ -326,7 +443,9 @@ handle_set (void *cls,
   uint32_t flags;
   uint64_t value;
   int64_t delta;
+  int changed;
 
+  make_client_entry (client);
   msize = ntohs (message->size);
   if (msize < sizeof (struct GNUNET_STATISTICS_SetMessage))
     {
@@ -344,13 +463,15 @@ handle_set (void *cls,
       GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
       return;
     }
+  flags = ntohl (msg->flags);
+  value = GNUNET_ntohll (msg->value);
 #if DEBUG_STATISTICS
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-              "Received request to update statistic on `%s:%s'\n",
-              service, name);
+              "Received request to update statistic on `%s:%s' (%u) to/by %llu\n",
+              service, name,
+              (unsigned int) flags,
+              (unsigned long long) value);
 #endif
-  flags = ntohl (msg->flags);
-  value = GNUNET_ntohll (msg->value);
   pos = start;
   prev = NULL;
   while (pos != NULL)
@@ -359,17 +480,20 @@ handle_set (void *cls,
         {
           if ((flags & GNUNET_STATISTICS_SETFLAG_RELATIVE) == 0)
             {
+              changed = (pos->value != value);
               pos->value = value;
             }
           else
             {
               delta = (int64_t) value;
               if ((delta < 0) && (pos->value < -delta))
-                {
-                  pos->value = 0;
+                {
+                  changed = (pos->value != 0);
+                  pos->value = 0;
                 }
               else
                 {
+                  changed = (delta != 0);
                   GNUNET_break ((delta <= 0) ||
                                 (pos->value + delta > pos->value));
                   pos->value += delta;
@@ -391,6 +515,8 @@ handle_set (void *cls,
                       "Statistic `%s:%s' updated to value %llu.\n",
                       service, name, pos->value);
 #endif
+          if (changed) 
+            notify_change (pos);
           GNUNET_SERVER_receive_done (client, GNUNET_OK);
           return;
         }
@@ -420,6 +546,86 @@ handle_set (void *cls,
 
 
 /**
+ * Handle WATCH-message.
+ *
+ * @param cls closure
+ * @param client identification of the client
+ * @param message the actual message
+ */
+static void
+handle_watch (void *cls,
+              struct GNUNET_SERVER_Client *client,
+              const struct GNUNET_MessageHeader *message)
+{
+  char *service;
+  char *name;
+  uint16_t msize;
+  uint16_t size;
+  struct StatsEntry *pos;
+  struct ClientEntry *ce;
+  struct WatchEntry *we;
+
+  ce = make_client_entry (client);
+  msize = ntohs (message->size);
+  if (msize < sizeof (struct GNUNET_MessageHeader))
+    {
+      GNUNET_break (0);
+      GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
+      return;
+    }
+  size = msize - sizeof (struct GNUNET_MessageHeader);
+  if (size != GNUNET_STRINGS_buffer_tokenize ((const char *) &message[1],
+                                              size, 2, &service, &name))
+    {
+      GNUNET_break (0);
+      GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
+      return;
+    }
+#if DEBUG_STATISTICS
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "Received request to watch statistic on `%s:%s'\n",
+              service, name);
+#endif
+  pos = start;
+  while (pos != NULL)
+    {
+      if (matches (pos, service, name))
+        break;
+      pos = pos->next;
+    }
+  if (pos == NULL)
+    {
+      pos = GNUNET_malloc (sizeof (struct StatsEntry) + 
+                           sizeof (struct GNUNET_STATISTICS_SetMessage) + 
+                           size);
+      pos->next = start;
+      pos->uid = uidgen++;
+      pos->msg = (void *) &pos[1];
+      pos->msg->header.size = htons (sizeof (struct GNUNET_STATISTICS_SetMessage) + 
+                                     size);
+      pos->msg->header.type = htons (GNUNET_MESSAGE_TYPE_STATISTICS_SET);
+      memcpy (pos->msg, message, ntohs (message->size));
+      pos->service = (const char *) &pos->msg[1];
+      memcpy (&pos->msg[1], service, strlen (service)+1);
+      pos->name = &pos->service[strlen (pos->service) + 1];
+      memcpy ((void*) pos->name, name, strlen (name)+1);
+      start = pos;
+    }
+  we = GNUNET_malloc (sizeof (struct WatchEntry));
+  we->client = client;
+  GNUNET_SERVER_client_keep (client);
+  we->wid = ce->max_wid++;
+  GNUNET_CONTAINER_DLL_insert (pos->we_head,
+                               pos->we_tail,
+                               we);
+  if (pos->value != 0)
+    notify_change (pos);
+  GNUNET_SERVER_receive_done (client,
+                              GNUNET_OK);
+}
+
+
+/**
  * Task run during shutdown.
  *
  * @param cls unused
@@ -429,7 +635,84 @@ static void
 shutdown_task (void *cls,
                const struct GNUNET_SCHEDULER_TaskContext *tc)
 {
+  struct ClientEntry *ce;
+  struct WatchEntry *we;
+  struct StatsEntry *se;
+
   save ();
+  GNUNET_SERVER_notification_context_destroy (nc);
+  nc = NULL;
+  while (NULL != (ce = client_head))
+    {
+      GNUNET_SERVER_client_drop (ce->client);
+      GNUNET_CONTAINER_DLL_remove (client_head,
+                                   client_tail,
+                                   ce);
+      GNUNET_free (ce);
+    }
+  while (NULL != (se = start))
+    {
+      start = se->next;
+      while (NULL != (we = se->we_head))
+        {
+          GNUNET_SERVER_client_drop (we->client);
+          GNUNET_CONTAINER_DLL_remove (se->we_head,
+                                       se->we_tail,
+                                       we);
+          GNUNET_free (we);
+        }
+      GNUNET_free (se);
+    }
+}
+
+
+/**
+ * A client disconnected.  Remove all of its data structure entries.
+ *
+ * @param cls closure, NULL
+ * @param client identification of the client
+ */
+static void
+handle_client_disconnect (void *cls,
+                          struct GNUNET_SERVER_Client
+                          * client)
+{
+  struct ClientEntry *ce;
+  struct WatchEntry *we;
+  struct WatchEntry *wen;
+  struct StatsEntry *se;
+  
+  ce = client_head;
+  while (NULL != ce)
+    {
+      if (ce->client == client)
+        {
+          GNUNET_SERVER_client_drop (ce->client);
+          GNUNET_CONTAINER_DLL_remove (client_head,
+                                       client_tail,
+                                       ce);
+          GNUNET_free (ce);
+          break;
+        }
+      ce = ce->next;
+    }
+  se = start;
+  while (NULL != se)
+    {
+      wen = se->we_head;
+      while (NULL != (we = wen))
+        {
+          wen = we->next;
+          if (we->client != client)
+            continue;
+          GNUNET_SERVER_client_drop (we->client);
+          GNUNET_CONTAINER_DLL_remove (se->we_head,
+                                       se->we_tail,
+                                       we);
+          GNUNET_free (we);
+        }
+      se = se->next;
+    }
 }
 
 
@@ -450,10 +733,15 @@ run (void *cls,
   static const struct GNUNET_SERVER_MessageHandler handlers[] = {
     {&handle_set, NULL, GNUNET_MESSAGE_TYPE_STATISTICS_SET, 0},
     {&handle_get, NULL, GNUNET_MESSAGE_TYPE_STATISTICS_GET, 0},
+    {&handle_watch, NULL, GNUNET_MESSAGE_TYPE_STATISTICS_WATCH, 0},
     {NULL, NULL, 0, 0}
   };
   cfg = c;
   GNUNET_SERVER_add_handlers (server, handlers);
+  nc = GNUNET_SERVER_notification_context_create (server, 16);
+  GNUNET_SERVER_disconnect_notify (server, 
+                                   &handle_client_disconnect,
+                                   NULL);
   load (server);
   GNUNET_SCHEDULER_add_delayed (sched,
                                 GNUNET_TIME_UNIT_FOREVER_REL,
diff --git a/src/statistics/statistics.h b/src/statistics/statistics.h
index c4a79765a..070a0aba5 100644
--- a/src/statistics/statistics.h
+++ b/src/statistics/statistics.h
@@ -91,4 +91,43 @@ struct GNUNET_STATISTICS_SetMessage
 
 };
 
+
+/**
+ * Message transmitted if a watched value changes.
+ */
+struct GNUNET_STATISTICS_WatchValueMessage
+{
+  /**
+   * Type: GNUNET_MESSAGE_TYPE_STATISTICS_WATCH_VALUE
+   */
+  struct GNUNET_MessageHeader header;
+
+  /**
+   * 0 for absolute value, 1 for relative value; 2 to make persistent
+   * (see GNUNET_STATISTICS_SETFLAG_*).
+   */
+  uint32_t flags GNUNET_PACKED;
+
+  /**
+   * Unique watch identification number (watch
+   * requests are enumerated in the order they 
+   * are received, the first request having 
+   * a wid of zero). 
+   */
+  uint32_t wid GNUNET_PACKED;
+
+  /**
+   * Reserved (always 0).
+   */
+  uint32_t reserved GNUNET_PACKED;
+
+  /**
+   * Value. Note that if this is a relative value, it will
+   * be signed even though the type given here is unsigned.
+   */
+  uint64_t value GNUNET_PACKED;
+  
+};
+
+
 #endif
diff --git a/src/statistics/statistics_api.c b/src/statistics/statistics_api.c
index 9de9f78fd..a5dde0e55 100644
--- a/src/statistics/statistics_api.c
+++ b/src/statistics/statistics_api.c
@@ -1,6 +1,6 @@
 /*
      This file is part of GNUnet.
-     (C) 2009 Christian Grothoff (and other contributing authors)
+     (C) 2009, 2010 Christian Grothoff (and other contributing authors)
 
      GNUnet is free software; you can redistribute it and/or modify
      it under the terms of the GNU General Public License as published
@@ -25,6 +25,7 @@
  */
 #include "platform.h"
 #include "gnunet_client_lib.h"
+#include "gnunet_constants.h"
 #include "gnunet_container_lib.h"
 #include "gnunet_protocols.h"
 #include "gnunet_server_lib.h"
@@ -47,7 +48,37 @@ enum ActionType
 {
   ACTION_GET,
   ACTION_SET,
-  ACTION_UPDATE
+  ACTION_UPDATE,
+  ACTION_WATCH
+};
+
+
+/**
+ * Entry kept for each value we are watching.
+ */
+struct GNUNET_STATISTICS_WatchEntry
+{
+ 
+  /**
+   * What subsystem is this action about? (never NULL)
+   */
+  char *subsystem;
+
+  /**
+   * What value is this action about? (never NULL)
+   */
+  char *name;
+
+  /**
+   * Function to call
+   */
+  GNUNET_STATISTICS_Iterator proc;
+
+  /**
+   * Closure for proc
+   */
+  void *proc_cls;
+
 };
 
 
@@ -118,7 +149,7 @@ struct GNUNET_STATISTICS_GetHandle
   int aborted;
 
   /**
-   * Is this a GET, SET or UPDATE?
+   * Is this a GET, SET, UPDATE or WATCH?
    */
   enum ActionType type;
 
@@ -179,14 +210,103 @@ struct GNUNET_STATISTICS_Handle
   struct GNUNET_STATISTICS_GetHandle *current;
 
   /**
+   * Array of watch entries.
+   */
+  struct GNUNET_STATISTICS_WatchEntry **watches;
+
+  /**
+   * Task doing exponential back-off trying to reconnect.
+   */
+  GNUNET_SCHEDULER_TaskIdentifier backoff_task;
+
+  /**
+   * Time for next connect retry.
+   */
+  struct GNUNET_TIME_Relative backoff;
+
+  /**
+   * Size of the 'watches' array.
+   */
+  unsigned int watches_size;
+
+  /**
    * Should this handle auto-destruct once all actions have
    * been processed?
    */
   int do_destroy;
 
+  /**
+   * Are we currently receiving from the service?
+   */
+  int receiving;
+
 };
 
 
+
+/**
+ * Schedule the next action to be performed.
+ */
+static void schedule_action (struct GNUNET_STATISTICS_Handle *h);
+
+/**
+ * Try to (re)connect to the statistics service.
+ *
+ * @return GNUNET_YES on success, GNUNET_NO on failure.
+ */
+static int
+try_connect (struct GNUNET_STATISTICS_Handle *ret);
+
+
+static void
+insert_ai (struct GNUNET_STATISTICS_Handle *h, struct GNUNET_STATISTICS_GetHandle *ai)
+{
+  GNUNET_CONTAINER_DLL_insert_after (h->action_head,
+                                     h->action_tail,
+                                     h->action_tail,
+                                     ai);                                    
+  if (h->action_head == ai)
+    schedule_action (h);
+}
+
+
+static void
+schedule_watch_request (struct GNUNET_STATISTICS_Handle *h,
+                        struct GNUNET_STATISTICS_WatchEntry *watch)
+{
+
+  struct GNUNET_STATISTICS_GetHandle *ai;
+  size_t slen;
+  size_t nlen;
+  size_t nsize;
+  
+  GNUNET_assert (h != NULL);
+  if (GNUNET_YES != try_connect (h))
+    {
+      schedule_action (h);
+      return;
+    }
+  slen = strlen (watch->subsystem) + 1;
+  nlen = strlen (watch->name) + 1;
+  nsize = sizeof (struct GNUNET_MessageHeader) + slen + nlen;
+  if (nsize >= GNUNET_SERVER_MAX_MESSAGE_SIZE)
+    {
+      GNUNET_break (0);
+      return;
+    }
+  ai = GNUNET_malloc (sizeof (struct GNUNET_STATISTICS_GetHandle));
+  ai->sh = h;
+  ai->subsystem = GNUNET_strdup (watch->subsystem);
+  ai->name = GNUNET_strdup (watch->name);
+  ai->timeout = GNUNET_TIME_UNIT_FOREVER_ABS;
+  ai->msize = nsize;
+  ai->type = ACTION_WATCH;
+  ai->proc = watch->proc;
+  ai->cls = watch->proc_cls;
+  insert_ai (h, ai);
+}
+
+
 /**
  * Try to (re)connect to the statistics service.
  *
@@ -195,11 +315,16 @@ struct GNUNET_STATISTICS_Handle
 static int
 try_connect (struct GNUNET_STATISTICS_Handle *ret)
 {
+  unsigned int i;
   if (ret->client != NULL)
     return GNUNET_YES;
   ret->client = GNUNET_CLIENT_connect (ret->sched, "statistics", ret->cfg);
   if (ret->client != NULL)
-    return GNUNET_YES;
+    {
+      for (i=0;i<ret->watches_size;i++)
+        schedule_watch_request (ret, ret->watches[i]);
+      return GNUNET_YES;
+    }
 #if DEBUG_STATISTICS
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
               _("Failed to connect to statistics service!\n"));
@@ -221,12 +346,6 @@ free_action_item (struct GNUNET_STATISTICS_GetHandle *ai)
 
 
 /**
- * Schedule the next action to be performed.
- */
-static void schedule_action (struct GNUNET_STATISTICS_Handle *h);
-
-
-/**
  * GET processing is complete, tell client about it.
  */
 static void
@@ -259,7 +378,13 @@ process_message (struct GNUNET_STATISTICS_Handle *h,
   uint16_t size;
 
   if (h->current->aborted)
-    return GNUNET_OK;           /* don't bother */
+    {
+#if DEBUG_STATISTICS
+      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                  "Iteration was aborted, ignoring VALUE\n");
+#endif      
+      return GNUNET_OK;           /* don't bother */
+    }
   size = ntohs (msg->size);
   if (size < sizeof (struct GNUNET_STATISTICS_ReplyMessage))
     {
@@ -293,6 +418,42 @@ process_message (struct GNUNET_STATISTICS_Handle *h,
 #endif
       h->current->aborted = GNUNET_YES;    
     }
+#if DEBUG_STATISTICS
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "VALUE processed successfully\n");
+#endif      
+  return GNUNET_OK;
+}
+
+
+static int
+process_watch_value (struct GNUNET_STATISTICS_Handle *h,
+                     const struct GNUNET_MessageHeader *msg)
+{
+  const struct GNUNET_STATISTICS_WatchValueMessage *wvm;
+  struct GNUNET_STATISTICS_WatchEntry *w;
+  uint32_t wid;
+
+  if (sizeof(struct GNUNET_STATISTICS_WatchValueMessage) !=
+      ntohs (msg->size))
+    {
+      GNUNET_break (0);
+      return GNUNET_SYSERR;
+    }
+  wvm = (const struct GNUNET_STATISTICS_WatchValueMessage *)msg;
+  wid = ntohl (wvm->wid);
+  if (wid >= h->watches_size)
+    {
+      GNUNET_break (0);
+      return GNUNET_SYSERR;
+    }
+  w = h->watches[wid];
+  (void) w->proc (w->proc_cls,
+                  w->subsystem,
+                  w->name,
+                  GNUNET_ntohll (wvm->value),
+                  0 !=
+                  (ntohl (wvm->flags) & GNUNET_STATISTICS_PERSIST_BIT));
   return GNUNET_OK;
 }
 
@@ -329,21 +490,53 @@ receive_stats (void *cls, const struct GNUNET_MessageHeader *msg)
       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                   "Received end of statistics marker\n");
 #endif
+      h->backoff = GNUNET_TIME_UNIT_MILLISECONDS;
       finish (h, GNUNET_OK);
+      if (h->watches_size > 0)
+        {
+          GNUNET_CLIENT_receive (h->client,
+                                 &receive_stats,
+                                 h,
+                                 GNUNET_TIME_UNIT_FOREVER_REL);
+        }
+      else
+        {
+          h->receiving = GNUNET_NO;
+        }
       return;
     case GNUNET_MESSAGE_TYPE_STATISTICS_VALUE:
       if (GNUNET_OK == process_message (h, msg))
         {
           /* finally, look for more! */
+#if DEBUG_STATISTICS
+          GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                      "Processing VALUE done, now reading more\n");
+#endif      
           GNUNET_CLIENT_receive (h->client,
                                  &receive_stats,
                                  h,
                                  GNUNET_TIME_absolute_get_remaining
                                  (h->current->timeout));
+          h->backoff = GNUNET_TIME_UNIT_MILLISECONDS;
           return;
         }
       GNUNET_break (0);
       break;
+    case GNUNET_MESSAGE_TYPE_STATISTICS_WATCH_VALUE:
+      if (GNUNET_OK ==
+          process_watch_value (h, 
+                               msg))
+        {
+          h->backoff = GNUNET_TIME_UNIT_MILLISECONDS;
+          GNUNET_assert (h->watches_size > 0);
+          GNUNET_CLIENT_receive (h->client,
+                                 &receive_stats,
+                                 h,
+                                 GNUNET_TIME_UNIT_FOREVER_REL);
+          return;
+        }
+      GNUNET_break (0);
+      break;
     default:
       GNUNET_break (0);
       break;
@@ -392,15 +585,69 @@ transmit_get (struct GNUNET_STATISTICS_Handle *handle, size_t size, void *buf)
                                              2,
                                              handle->current->subsystem,
                                              handle->current->name));
-  GNUNET_CLIENT_receive (handle->client,
-                         &receive_stats,
-                         handle,
-                         GNUNET_TIME_absolute_get_remaining (handle->
-                                                             current->timeout));
+  if (! handle->receiving)
+    {
+#if DEBUG_STATISTICS
+      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                  "Transmission of GET done, now reading response\n");
+#endif      
+      handle->receiving = GNUNET_YES;
+      GNUNET_CLIENT_receive (handle->client,
+                             &receive_stats,
+                             handle,
+                             GNUNET_TIME_absolute_get_remaining (handle->
+                                                                 current->timeout));
+    }
   return msize;
 }
 
 
+/**
+ * Transmit a WATCH request (and if successful, start to receive
+ * the response).
+ */
+static size_t
+transmit_watch (struct GNUNET_STATISTICS_Handle *handle, size_t size, void *buf)
+{
+  struct GNUNET_MessageHeader *hdr;
+  size_t slen1;
+  size_t slen2;
+  uint16_t msize;
+
+  if (buf == NULL)
+    {
+      /* timeout / error */
+#if DEBUG_STATISTICS
+      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                  "Transmission of request for statistics failed!\n");
+#endif
+      finish (handle, GNUNET_SYSERR);
+      return 0;
+    }
+  slen1 = strlen (handle->current->subsystem) + 1;
+  slen2 = strlen (handle->current->name) + 1;
+  msize = slen1 + slen2 + sizeof (struct GNUNET_MessageHeader);
+  GNUNET_assert (msize <= size);
+  hdr = (struct GNUNET_MessageHeader *) buf;
+  hdr->size = htons (msize);
+  hdr->type = htons (GNUNET_MESSAGE_TYPE_STATISTICS_WATCH);
+  GNUNET_assert (slen1 + slen2 ==
+                 GNUNET_STRINGS_buffer_fill ((char *) &hdr[1],
+                                             slen1 + slen2,
+                                             2,
+                                             handle->current->subsystem,
+                                             handle->current->name));
+  if (! handle->receiving)
+    {
+      handle->receiving = GNUNET_YES;
+      GNUNET_CLIENT_receive (handle->client,
+                             &receive_stats,
+                             handle,
+                             GNUNET_TIME_UNIT_FOREVER_REL);
+    }
+  return msize;
+}
+
 
 /**
  * Transmit a SET/UPDATE request.
@@ -464,6 +711,9 @@ transmit_action (void *cls, size_t size, void *buf)
     case ACTION_UPDATE:
       ret = transmit_set (handle, size, buf);
       break;
+    case ACTION_WATCH:
+      ret = transmit_watch (handle, size, buf);
+      break;
     default:
       ret = 0;
       GNUNET_break (0);
@@ -495,6 +745,7 @@ GNUNET_STATISTICS_create (struct GNUNET_SCHEDULER_Handle *sched,
   ret->sched = sched;
   ret->cfg = cfg;
   ret->subsystem = GNUNET_strdup (subsystem);
+  ret->backoff = GNUNET_TIME_UNIT_MILLISECONDS;
   try_connect (ret);
   return ret;
 }
@@ -516,7 +767,11 @@ GNUNET_STATISTICS_destroy (struct GNUNET_STATISTICS_Handle *h,
   struct GNUNET_STATISTICS_GetHandle *next;
   struct GNUNET_STATISTICS_GetHandle *prev;
   struct GNUNET_TIME_Relative timeout;
+  int i;
 
+  if (GNUNET_SCHEDULER_NO_TASK != h->backoff_task)
+    GNUNET_SCHEDULER_cancel (h->sched,
+                             h->backoff_task);
   if (sync_first)
     {
       if (h->current != NULL)
@@ -591,11 +846,30 @@ GNUNET_STATISTICS_destroy (struct GNUNET_STATISTICS_Handle *h,
       GNUNET_CLIENT_disconnect (h->client, GNUNET_YES);
       h->client = NULL;
     }
+  for (i=0;i<h->watches_size;i++)
+    {
+      GNUNET_free (h->watches[i]->subsystem);
+      GNUNET_free (h->watches[i]->name);
+      GNUNET_free (h->watches[i]);
+    }
+  GNUNET_array_grow (h->watches,
+                     h->watches_size,
+                     0);
   GNUNET_free (h->subsystem);
   GNUNET_free (h);
 }
 
 
+static void
+finish_task (void *cls,
+             const struct GNUNET_SCHEDULER_TaskContext *tc)
+{
+  struct GNUNET_STATISTICS_Handle *h = cls;
+
+  h->backoff_task = GNUNET_SCHEDULER_NO_TASK;
+  finish (h, GNUNET_SYSERR);
+}
+
 
 /**
  * Schedule the next action to be performed.
@@ -609,7 +883,13 @@ schedule_action (struct GNUNET_STATISTICS_Handle *h)
     return;                     /* action already pending */
   if (GNUNET_YES != try_connect (h))
     {
-      finish (h, GNUNET_SYSERR);
+      h->backoff_task = GNUNET_SCHEDULER_add_delayed (h->sched,
+                                                      h->backoff,
+                                                      &finish_task,
+                                                      h);
+      h->backoff = GNUNET_TIME_relative_multiply (h->backoff, 2);
+      h->backoff = GNUNET_TIME_relative_min (h->backoff,
+                                             GNUNET_CONSTANTS_SERVICE_TIMEOUT);
       return;
     }
 
@@ -643,19 +923,6 @@ schedule_action (struct GNUNET_STATISTICS_Handle *h)
     }
 }
 
-
-static void
-insert_ai (struct GNUNET_STATISTICS_Handle *h, struct GNUNET_STATISTICS_GetHandle *ai)
-{
-  GNUNET_CONTAINER_DLL_insert_after (h->action_head,
-                                     h->action_tail,
-                                     h->action_tail,
-                                     ai);                                    
-  if (h->action_head == ai)
-    schedule_action (h);
-}
-
-
 /**
  * Get statistic from the peer.
  *
@@ -742,6 +1009,40 @@ GNUNET_STATISTICS_get_cancel (struct GNUNET_STATISTICS_GetHandle *gh)
 }
 
 
+/**
+ * Watch statistics from the peer (be notified whenever they change).
+ * Note that the only way to cancel a "watch" request is to destroy
+ * the statistics handle given as the first argument to this call.
+ *
+ * @param handle identification of the statistics service
+ * @param subsystem limit to the specified subsystem, never NULL
+ * @param name name of the statistic value, never NULL
+ * @param proc function to call on each value
+ * @param proc_cls closure for proc
+ * @return GNUNET_OK on success, GNUNET_SYSERR on error
+ */
+int
+GNUNET_STATISTICS_watch (struct GNUNET_STATISTICS_Handle *handle,
+                         const char *subsystem,
+                         const char *name,
+                         GNUNET_STATISTICS_Iterator proc, 
+                         void *proc_cls)
+{
+  struct GNUNET_STATISTICS_WatchEntry *w;
+
+  w = GNUNET_malloc (sizeof (struct GNUNET_STATISTICS_WatchEntry));
+  w->subsystem = GNUNET_strdup (subsystem);
+  w->name = GNUNET_strdup (name);
+  w->proc = proc;
+  w->proc_cls = proc_cls;
+  GNUNET_array_append (handle->watches,
+                       handle->watches_size,
+                       w);
+  schedule_watch_request (handle, w);
+  return GNUNET_OK;
+}
+
+
 static void
 add_setter_action (struct GNUNET_STATISTICS_Handle *h,
                    const char *name,
@@ -809,7 +1110,8 @@ add_setter_action (struct GNUNET_STATISTICS_Handle *h,
                   ai->type = type;
                 }
             }
-          ai->timeout = GNUNET_TIME_relative_to_absolute (SET_TRANSMIT_TIMEOUT);                  
+          ai->timeout = GNUNET_TIME_relative_to_absolute (SET_TRANSMIT_TIMEOUT);
+          ai->make_persistent = make_persistent;
           return;
         }
       ai = ai->next;
@@ -824,7 +1126,6 @@ add_setter_action (struct GNUNET_STATISTICS_Handle *h,
   ai->value = value;
   ai->type = type;
   insert_ai (h, ai);
-  schedule_action (h);
 }
 
 
diff --git a/src/statistics/test_statistics_api.c b/src/statistics/test_statistics_api.c
index 378e17d0c..5c55a11a6 100644
--- a/src/statistics/test_statistics_api.c
+++ b/src/statistics/test_statistics_api.c
@@ -29,7 +29,7 @@
 #include "gnunet_scheduler_lib.h"
 #include "gnunet_statistics_service.h"
 
-#define VERBOSE GNUNET_NO
+#define DEBUG_STATISTICS GNUNET_NO
 
 #define START_SERVICE GNUNET_YES
 
@@ -38,6 +38,11 @@ check_1 (void *cls,
          const char *subsystem,
          const char *name, uint64_t value, int is_persistent)
 {
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "Received value %llu for `%s:%s\n",
+              (unsigned long long) value,
+              subsystem,
+              name);
   GNUNET_assert (0 == strcmp (name, "test-1"));
   GNUNET_assert (0 == strcmp (subsystem, "test-statistics-api"));
   GNUNET_assert (value == 1);
@@ -50,6 +55,11 @@ check_2 (void *cls,
          const char *subsystem,
          const char *name, uint64_t value, int is_persistent)
 {
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "Received value %llu for `%s:%s\n",
+              (unsigned long long) value,
+              subsystem,
+              name);
   GNUNET_assert (0 == strcmp (name, "test-2"));
   GNUNET_assert (0 == strcmp (subsystem, "test-statistics-api"));
   GNUNET_assert (value == 2);
@@ -62,6 +72,11 @@ check_3 (void *cls,
          const char *subsystem,
          const char *name, uint64_t value, int is_persistent)
 {
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "Received value %llu for `%s:%s\n",
+              (unsigned long long) value,
+              subsystem,
+              name);
   GNUNET_assert (0 == strcmp (name, "test-3"));
   GNUNET_assert (0 == strcmp (subsystem, "test-statistics-api"));
   GNUNET_assert (value == 3);
@@ -85,6 +100,8 @@ static void
 next (void *cls, int success)
 {
   GNUNET_assert (success == GNUNET_OK);
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "Issuing GET request\n");       
   GNUNET_break (NULL !=
                 GNUNET_STATISTICS_get (h, NULL, "test-2",
                                        GNUNET_TIME_UNIT_SECONDS, &next_fin, &check_2, cls));
@@ -102,6 +119,8 @@ run (void *cls,
   GNUNET_STATISTICS_set (h, "test-2", 2, GNUNET_NO);
   GNUNET_STATISTICS_set (h, "test-3", 2, GNUNET_NO);
   GNUNET_STATISTICS_update (h, "test-3", 1, GNUNET_YES);
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "Issuing GET request\n");       
   GNUNET_break (NULL != 
                 GNUNET_STATISTICS_get (h, NULL, "test-1",
                                        GNUNET_TIME_UNIT_SECONDS, &next, &check_1, cls));
@@ -127,6 +146,11 @@ check ()
   char *const argv[] = { "test-statistics-api",
     "-c",
     "test_statistics_api_data.conf",
+#if DEBUG_STATISTICS
+                         "-L", "DEBUG",
+#else
+                         "-L", "WARNING",
+#endif
     NULL
   };
   struct GNUNET_GETOPT_CommandLineOption options[] = {
@@ -141,7 +165,7 @@ check ()
 #endif
                                  "-c", "test_statistics_api_data.conf", NULL);
 #endif
-  GNUNET_PROGRAM_run (3, argv, "test-statistics-api", "nohelp",
+  GNUNET_PROGRAM_run (5, argv, "test-statistics-api", "nohelp",
                       options, &run, &ok);
 #if START_SERVICE
   if (0 != PLIBC_KILL (pid, SIGTERM))
@@ -163,7 +187,7 @@ check ()
 #endif
                                  "-c", "test_statistics_api_data.conf", NULL);
 #endif
-  GNUNET_PROGRAM_run (3, argv, "test-statistics-api", "nohelp",
+  GNUNET_PROGRAM_run (5, argv, "test-statistics-api", "nohelp",
                       options, &run_more, &ok);
 #if START_SERVICE
   if (0 != PLIBC_KILL (pid, SIGTERM))
@@ -181,6 +205,13 @@ main (int argc, char *argv[])
 {
   int ret;
 
+  GNUNET_log_setup ("test_statistics_api", 
+#if DEBUG_STATISTICS
+                    "DEBUG",
+#else
+                    "WARNING",
+#endif
+                    NULL);
   ret = check ();
 
   return ret;
diff --git a/src/statistics/test_statistics_api_data.conf b/src/statistics/test_statistics_api_data.conf
index 41cfd9596..ba8d4d68e 100644
--- a/src/statistics/test_statistics_api_data.conf
+++ b/src/statistics/test_statistics_api_data.conf
@@ -4,9 +4,12 @@ DEFAULTCONFIG = test_statistics_api_data.conf
 
 [statistics]
 PORT = 22353
+UNIXPATH = /tmp/test-statistics-service-statistics.unix
+DEBUG = YES
 
 [arm]
 PORT = 22354
 DEFAULTSERVICES = 
+UNIXPATH = /tmp/test-statistics-service-arm.unix
 # DEBUG = YES