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