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