src/testbed-logger/testbed_logger_api.c

   1 /*
   2       This file is part of GNUnet
   3       Copyright (C) 2008--2013, 2016 GNUnet e.V.
   4
   5       GNUnet is free software: you can redistribute it and/or modify it
   6       under the terms of the GNU General Public License as published
   7       by the Free Software Foundation, either version 3 of the License,
   8       or (at your 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       Affero General Public License for more details.
  14  */
  15
  16 /**
  17  * @file testbed-logger/testbed_logger_api.c
  18  * @brief Client-side routines for communicating with the tesbted logger service
  19  * @author Sree Harsha Totakura <sreeharsha@totakura.in>
  20  * @author Christian Grothoff
  21  */
  22
  23 #include "platform.h"
  24 #include "gnunet_util_lib.h"
  25 #include "gnunet_testbed_logger_service.h"
  26
  27 /**
  28  * Generic logging shorthand
  29  */
  30 #define LOG(kind, ...)                          \
  31   GNUNET_log_from (kind, "testbed-logger-api", __VA_ARGS__)
  32
  33
  34 /**
  35  * The size of the buffer we fill before sending out the message
  36  */
  37 #define BUFFER_SIZE (GNUNET_MAX_MESSAGE_SIZE - sizeof (struct GNUNET_MessageHeader))
  38
  39 /**
  40  * Connection handle for the logger service
  41  */
  42 struct GNUNET_TESTBED_LOGGER_Handle
  43 {
  44   /**
  45    * Client connection
  46    */
  47   struct GNUNET_MQ_Handle *mq;
  48
  49   /**
  50    * Flush completion callback
  51    */
  52   GNUNET_TESTBED_LOGGER_FlushCompletion cb;
  53
  54   /**
  55    * Closure for @e cb
  56    */
  57   void *cb_cls;
  58
  59   /**
  60    * Local buffer for data to be transmitted
  61    */
  62   char buf[BUFFER_SIZE];
  63
  64   /**
  65    * How many bytes in @a buf are in use?
  66    */
  67   size_t buse;
  68
  69   /**
  70    * Number of bytes wrote since last flush
  71    */
  72   size_t bwrote;
  73
  74   /**
  75    * How long after should we retry sending a message to the service?
  76    */
  77   struct GNUNET_TIME_Relative retry_backoff;
  78
  79   /**
  80    * Task to call the flush completion callback
  81    */
  82   struct GNUNET_SCHEDULER_Task *flush_completion_task;
  83
  84   /**
  85    * Number of entries in the MQ.
  86    */
  87   unsigned int mq_len;
  88 };
  89
  90
  91 /**
  92  * Task to call the flush completion notification
  93  *
  94  * @param cls the logger handle
  95  */
  96 static void
  97 call_flush_completion (void *cls)
  98 {
  99   struct GNUNET_TESTBED_LOGGER_Handle *h = cls;
 100   GNUNET_TESTBED_LOGGER_FlushCompletion cb;
 101   void *cb_cls;
 102   size_t bw;
 103
 104   h->flush_completion_task = NULL;
 105   bw = h->bwrote;
 106   h->bwrote = 0;
 107   cb = h->cb;
 108   h->cb = NULL;
 109   cb_cls = h->cb_cls;
 110   h->cb_cls = NULL;
 111   if (NULL != cb)
 112     cb (cb_cls, bw);
 113 }
 114
 115
 116 /**
 117  * Schedule the flush completion notification task
 118  *
 119  * @param h logger handle
 120  */
 121 static void
 122 trigger_flush_notification (struct GNUNET_TESTBED_LOGGER_Handle *h)
 123 {
 124   if (NULL != h->flush_completion_task)
 125     GNUNET_SCHEDULER_cancel (h->flush_completion_task);
 126   h->flush_completion_task
 127     = GNUNET_SCHEDULER_add_now (&call_flush_completion,
 128                                 h);
 129 }
 130
 131
 132 /**
 133  * Send the buffered data to the service
 134  *
 135  * @param h the logger handle
 136  */
 137 static void
 138 dispatch_buffer (struct GNUNET_TESTBED_LOGGER_Handle *h);
 139
 140
 141 /**
 142  * MQ successfully sent a message.
 143  *
 144  * @param cls our handle
 145  */
 146 static void
 147 notify_sent (void *cls)
 148 {
 149   struct GNUNET_TESTBED_LOGGER_Handle *h = cls;
 150
 151   h->mq_len--;
 152   if ( (0 == h->mq_len) &&
 153        (NULL != h->cb) )
 154   {
 155     if (0 == h->buse)
 156       trigger_flush_notification (h);
 157     else
 158       dispatch_buffer (h);
 159   }
 160 }
 161
 162
 163 /**
 164  * Send the buffered data to the service
 165  *
 166  * @param h the logger handle
 167  */
 168 static void
 169 dispatch_buffer (struct GNUNET_TESTBED_LOGGER_Handle *h)
 170 {
 171   struct GNUNET_MessageHeader *msg;
 172   struct GNUNET_MQ_Envelope *env;
 173
 174   env = GNUNET_MQ_msg_extra (msg,
 175                              h->buse,
 176                              GNUNET_MESSAGE_TYPE_TESTBED_LOGGER_MSG);
 177   GNUNET_memcpy (&msg[1],
 178           h->buf,
 179           h->buse);
 180   h->bwrote += h->buse;
 181   h->buse = 0;
 182   h->mq_len++;
 183   GNUNET_MQ_notify_sent (env,
 184                          &notify_sent,
 185                          h);
 186   GNUNET_MQ_send (h->mq,
 187                   env);
 188 }
 189
 190
 191 /**
 192  * We got disconnected from the logger.  Stop logging.
 193  *
 194  * @param cls the `struct GNUNET_TESTBED_LOGGER_Handle`
 195  * @param error error code
 196  */
 197 static void
 198 mq_error_handler (void *cls,
 199                   enum GNUNET_MQ_Error error)
 200 {
 201   struct GNUNET_TESTBED_LOGGER_Handle *h = cls;
 202
 203   GNUNET_break (0);
 204   GNUNET_MQ_destroy (h->mq);
 205   h->mq = NULL;
 206 }
 207
 208
 209 /**
 210  * Connect to the testbed logger service
 211  *
 212  * @param cfg configuration to use
 213  * @return the handle which can be used for sending data to the service; NULL
 214  *           upon any error
 215  */
 216 struct GNUNET_TESTBED_LOGGER_Handle *
 217 GNUNET_TESTBED_LOGGER_connect (const struct GNUNET_CONFIGURATION_Handle *cfg)
 218 {
 219   struct GNUNET_TESTBED_LOGGER_Handle *h;
 220
 221   h = GNUNET_new (struct GNUNET_TESTBED_LOGGER_Handle);
 222   h->mq = GNUNET_CLIENT_connect (cfg,
 223                                  "testbed-logger",
 224                                  NULL,
 225                                  &mq_error_handler,
 226                                  h);
 227   if (NULL == h->mq)
 228   {
 229     GNUNET_free (h);
 230     return NULL;
 231   }
 232   return h;
 233 }
 234
 235
 236 /**
 237  * Disconnect from the logger service.
 238  *
 239  * @param h the logger handle
 240  */
 241 void
 242 GNUNET_TESTBED_LOGGER_disconnect (struct GNUNET_TESTBED_LOGGER_Handle *h)
 243 {
 244   if (NULL != h->flush_completion_task)
 245   {
 246     GNUNET_SCHEDULER_cancel (h->flush_completion_task);
 247     h->flush_completion_task = NULL;
 248   }
 249   if (0 != h->mq_len)
 250     LOG (GNUNET_ERROR_TYPE_WARNING,
 251          "Disconnect lost %u logger message[s]\n",
 252          h->mq_len);
 253   if (NULL != h->mq)
 254   {
 255     GNUNET_MQ_destroy (h->mq);
 256     h->mq = NULL;
 257   }
 258   GNUNET_free (h);
 259 }
 260
 261
 262 /**
 263  * Send data to be logged to the logger service.  The data will be buffered and
 264  * will be sent upon an explicit call to GNUNET_TESTBED_LOGGER_flush() or upon
 265  * exceeding a threshold size.
 266  *
 267  * @param h the logger handle
 268  * @param data the data to send;
 269  * @param size how many bytes of @a data to send
 270  */
 271 void
 272 GNUNET_TESTBED_LOGGER_write (struct GNUNET_TESTBED_LOGGER_Handle *h,
 273                              const void *data,
 274                              size_t size)
 275 {
 276   if (NULL == h->mq)
 277     return;
 278   while (0 != size)
 279   {
 280     size_t fit_size = GNUNET_MIN (size,
 281                                   BUFFER_SIZE - h->buse);
 282     GNUNET_memcpy (&h->buf[h->buse],
 283             data,
 284             fit_size);
 285     h->buse += fit_size;
 286     data += fit_size;
 287     size -= fit_size;
 288     if (0 != size)
 289       dispatch_buffer (h);
 290   }
 291 }
 292
 293
 294 /**
 295  * Flush the buffered data to the logger service
 296  *
 297  * @param h the logger handle
 298  * @param cb the callback to call after the data is flushed
 299  * @param cb_cls the closure for the above callback
 300  */
 301 void
 302 GNUNET_TESTBED_LOGGER_flush (struct GNUNET_TESTBED_LOGGER_Handle *h,
 303                              GNUNET_TESTBED_LOGGER_FlushCompletion cb,
 304                              void *cb_cls)
 305 {
 306   GNUNET_assert (NULL == h->cb);
 307   h->cb = cb;
 308   h->cb_cls = cb_cls;
 309   if ( (NULL == h->mq) ||
 310        (0 == h->buse) )
 311   {
 312     trigger_flush_notification (h);
 313     return;
 314   }
 315   dispatch_buffer (h);
 316 }
 317
 318
 319 /**
 320  * Cancel notification upon flush.  Should only be used when the flush
 321  * completion callback given to GNUNET_TESTBED_LOGGER_flush() is not already
 322  * called.
 323  *
 324  * @param h the logger handle
 325  */
 326 void
 327 GNUNET_TESTBED_LOGGER_flush_cancel (struct GNUNET_TESTBED_LOGGER_Handle *h)
 328 {
 329   if (NULL != h->flush_completion_task)
 330   {
 331     GNUNET_SCHEDULER_cancel (h->flush_completion_task);
 332     h->flush_completion_task = NULL;
 333   }
 334   h->cb = NULL;
 335   h->cb_cls = NULL;
 336 }
 337
 338 /* End of testbed_logger_api.c */