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