src/util/mst.c

   1 /*
   2      This file is part of GNUnet.
   3      Copyright (C) 2010, 2016, 2017 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 util/mst.c
  18  * @brief convenience functions for handling inbound message buffers
  19  * @author Christian Grothoff
  20  */
  21
  22 #include "platform.h"
  23 #include "gnunet_util_lib.h"
  24
  25
  26 #if HAVE_UNALIGNED_64_ACCESS
  27 #define ALIGN_FACTOR 4
  28 #else
  29 #define ALIGN_FACTOR 8
  30 #endif
  31
  32 #define LOG(kind,...) GNUNET_log_from (kind, "util-mst", __VA_ARGS__)
  33
  34
  35 /**
  36  * Handle to a message stream tokenizer.
  37  */
  38 struct GNUNET_MessageStreamTokenizer
  39 {
  40
  41   /**
  42    * Function to call on completed messages.
  43    */
  44   GNUNET_MessageTokenizerCallback cb;
  45
  46   /**
  47    * Closure for @e cb.
  48    */
  49   void *cb_cls;
  50
  51   /**
  52    * Size of the buffer (starting at @e hdr).
  53    */
  54   size_t curr_buf;
  55
  56   /**
  57    * How many bytes in buffer have we already processed?
  58    */
  59   size_t off;
  60
  61   /**
  62    * How many bytes in buffer are valid right now?
  63    */
  64   size_t pos;
  65
  66   /**
  67    * Beginning of the buffer.  Typed like this to force alignment.
  68    */
  69   struct GNUNET_MessageHeader *hdr;
  70
  71 };
  72
  73
  74 /**
  75  * Create a message stream tokenizer.
  76  *
  77  * @param cb function to call on completed messages
  78  * @param cb_cls closure for @a cb
  79  * @return handle to tokenizer
  80  */
  81 struct GNUNET_MessageStreamTokenizer *
  82 GNUNET_MST_create (GNUNET_MessageTokenizerCallback cb,
  83                    void *cb_cls)
  84 {
  85   struct GNUNET_MessageStreamTokenizer *ret;
  86
  87   ret = GNUNET_new (struct GNUNET_MessageStreamTokenizer);
  88   ret->hdr = GNUNET_malloc (GNUNET_MIN_MESSAGE_SIZE);
  89   ret->curr_buf = GNUNET_MIN_MESSAGE_SIZE;
  90   ret->cb = cb;
  91   ret->cb_cls = cb_cls;
  92   return ret;
  93 }
  94
  95
  96 /**
  97  * Add incoming data to the receive buffer and call the
  98  * callback for all complete messages.
  99  *
 100  * @param mst tokenizer to use
 101  * @param buf input data to add
 102  * @param size number of bytes in @a buf
 103  * @param purge should any excess bytes in the buffer be discarded
 104  *       (i.e. for packet-based services like UDP)
 105  * @param one_shot only call callback once, keep rest of message in buffer
 106  * @return #GNUNET_OK if we are done processing (need more data)
 107  *         #GNUNET_NO if @a one_shot was set and we have another message ready
 108  *         #GNUNET_SYSERR if the data stream is corrupt
 109  */
 110 int
 111 GNUNET_MST_from_buffer (struct GNUNET_MessageStreamTokenizer *mst,
 112                         const char *buf,
 113                         size_t size,
 114                         int purge,
 115                         int one_shot)
 116 {
 117   const struct GNUNET_MessageHeader *hdr;
 118   size_t delta;
 119   uint16_t want;
 120   char *ibuf;
 121   int need_align;
 122   unsigned long offset;
 123   int ret;
 124   int cbret;
 125
 126   GNUNET_assert (mst->off <= mst->pos);
 127   GNUNET_assert (mst->pos <= mst->curr_buf);
 128   LOG (GNUNET_ERROR_TYPE_DEBUG,
 129        "MST receives %u bytes with %u bytes already in private buffer\n",
 130        (unsigned int) size,
 131        (unsigned int) (mst->pos - mst->off));
 132   ret = GNUNET_OK;
 133   ibuf = (char *) mst->hdr;
 134   while (mst->pos > 0)
 135   {
 136 do_align:
 137     GNUNET_assert (mst->pos >= mst->off);
 138     if ((mst->curr_buf - mst->off < sizeof (struct GNUNET_MessageHeader)) ||
 139         (0 != (mst->off % ALIGN_FACTOR)))
 140     {
 141       /* need to align or need more space */
 142       mst->pos -= mst->off;
 143       memmove (ibuf,
 144                &ibuf[mst->off],
 145                mst->pos);
 146       mst->off = 0;
 147     }
 148     if (mst->pos - mst->off < sizeof (struct GNUNET_MessageHeader))
 149     {
 150       delta
 151         = GNUNET_MIN (sizeof (struct GNUNET_MessageHeader)
 152                       - (mst->pos - mst->off),
 153                       size);
 154       GNUNET_memcpy (&ibuf[mst->pos],
 155                      buf,
 156                      delta);
 157       mst->pos += delta;
 158       buf += delta;
 159       size -= delta;
 160     }
 161     if (mst->pos - mst->off < sizeof (struct GNUNET_MessageHeader))
 162     {
 163       if (purge)
 164       {
 165         mst->off = 0;
 166         mst->pos = 0;
 167       }
 168       return GNUNET_OK;
 169     }
 170     hdr = (const struct GNUNET_MessageHeader *) &ibuf[mst->off];
 171     want = ntohs (hdr->size);
 172     if (want < sizeof (struct GNUNET_MessageHeader))
 173     {
 174       GNUNET_break_op (0);
 175       return GNUNET_SYSERR;
 176     }
 177     if ( (mst->curr_buf - mst->off < want) &&
 178          (mst->off > 0) )
 179     {
 180       /* can get more space by moving */
 181       mst->pos -= mst->off;
 182       memmove (ibuf,
 183                &ibuf[mst->off],
 184                mst->pos);
 185       mst->off = 0;
 186     }
 187     if (mst->curr_buf < want)
 188     {
 189       /* need to get more space by growing buffer */
 190       GNUNET_assert (0 == mst->off);
 191       mst->hdr = GNUNET_realloc (mst->hdr,
 192                                  want);
 193       ibuf = (char *) mst->hdr;
 194       mst->curr_buf = want;
 195     }
 196     hdr = (const struct GNUNET_MessageHeader *) &ibuf[mst->off];
 197     if (mst->pos - mst->off < want)
 198     {
 199       delta = GNUNET_MIN (want - (mst->pos - mst->off),
 200                           size);
 201       GNUNET_assert (mst->pos + delta <= mst->curr_buf);
 202       GNUNET_memcpy (&ibuf[mst->pos],
 203                      buf,
 204                      delta);
 205       mst->pos += delta;
 206       buf += delta;
 207       size -= delta;
 208     }
 209     if (mst->pos - mst->off < want)
 210     {
 211       if (purge)
 212       {
 213         mst->off = 0;
 214         mst->pos = 0;
 215       }
 216       return GNUNET_OK;
 217     }
 218     if (one_shot == GNUNET_SYSERR)
 219     {
 220       /* cannot call callback again, but return value saying that
 221        * we have another full message in the buffer */
 222       ret = GNUNET_NO;
 223       goto copy;
 224     }
 225     if (one_shot == GNUNET_YES)
 226       one_shot = GNUNET_SYSERR;
 227     mst->off += want;
 228     if (GNUNET_OK !=
 229         (cbret = mst->cb (mst->cb_cls,
 230                            hdr)))
 231     {
 232       if (GNUNET_SYSERR == cbret)
 233         GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 234                     "Failure processing message of type %u and size %u\n",
 235                     ntohs (hdr->type),
 236                     ntohs (hdr->size));
 237       return GNUNET_SYSERR;
 238     }
 239     if (mst->off == mst->pos)
 240     {
 241       /* reset to beginning of buffer, it's free right now! */
 242       mst->off = 0;
 243       mst->pos = 0;
 244     }
 245   }
 246   GNUNET_assert (0 == mst->pos);
 247   while (size > 0)
 248   {
 249     LOG (GNUNET_ERROR_TYPE_DEBUG,
 250          "Server-mst has %u bytes left in inbound buffer\n",
 251          (unsigned int) size);
 252     if (size < sizeof (struct GNUNET_MessageHeader))
 253       break;
 254     offset = (unsigned long) buf;
 255     need_align = (0 != (offset % ALIGN_FACTOR)) ? GNUNET_YES : GNUNET_NO;
 256     if (GNUNET_NO == need_align)
 257     {
 258       /* can try to do zero-copy and process directly from original buffer */
 259       hdr = (const struct GNUNET_MessageHeader *) buf;
 260       want = ntohs (hdr->size);
 261       if (want < sizeof (struct GNUNET_MessageHeader))
 262       {
 263         GNUNET_break_op (0);
 264         mst->off = 0;
 265         return GNUNET_SYSERR;
 266       }
 267       if (size < want)
 268         break;                  /* or not: buffer incomplete, so copy to private buffer... */
 269       if (one_shot == GNUNET_SYSERR)
 270       {
 271         /* cannot call callback again, but return value saying that
 272          * we have another full message in the buffer */
 273         ret = GNUNET_NO;
 274         goto copy;
 275       }
 276       if (one_shot == GNUNET_YES)
 277         one_shot = GNUNET_SYSERR;
 278       if (GNUNET_OK !=
 279           (cbret = mst->cb (mst->cb_cls,
 280                             hdr)))
 281       {
 282         if (GNUNET_SYSERR == cbret)
 283           GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 284                       "Failure processing message of type %u and size %u\n",
 285                       ntohs (hdr->type),
 286                       ntohs (hdr->size));
 287         return GNUNET_SYSERR;
 288       }
 289       buf += want;
 290       size -= want;
 291     }
 292     else
 293     {
 294       /* need to copy to private buffer to align;
 295        * yes, we go a bit more spagetti than usual here */
 296       goto do_align;
 297     }
 298   }
 299 copy:
 300   if ((size > 0) && (!purge))
 301   {
 302     if (size + mst->pos > mst->curr_buf)
 303     {
 304       mst->hdr = GNUNET_realloc (mst->hdr,
 305                                  size + mst->pos);
 306       ibuf = (char *) mst->hdr;
 307       mst->curr_buf = size + mst->pos;
 308     }
 309     GNUNET_assert (size + mst->pos <= mst->curr_buf);
 310     GNUNET_memcpy (&ibuf[mst->pos],
 311                    buf,
 312                    size);
 313     mst->pos += size;
 314   }
 315   if (purge)
 316   {
 317     mst->off = 0;
 318     mst->pos = 0;
 319   }
 320   LOG (GNUNET_ERROR_TYPE_DEBUG,
 321        "Server-mst leaves %u bytes in private buffer\n",
 322        (unsigned int) (mst->pos - mst->off));
 323   return ret;
 324 }
 325
 326
 327 /**
 328  * Add incoming data to the receive buffer and call the
 329  * callback for all complete messages.
 330  *
 331  * @param mst tokenizer to use
 332  * @param buf input data to add
 333  * @param size number of bytes in @a buf
 334  * @param purge should any excess bytes in the buffer be discarded
 335  *       (i.e. for packet-based services like UDP)
 336  * @param one_shot only call callback once, keep rest of message in buffer
 337  * @return #GNUNET_OK if we are done processing (need more data)
 338  *         #GNUNET_NO if one_shot was set and we have another message ready
 339  *         #GNUNET_SYSERR if the data stream is corrupt
 340  */
 341 int
 342 GNUNET_MST_read (struct GNUNET_MessageStreamTokenizer *mst,
 343                  struct GNUNET_NETWORK_Handle *sock,
 344                  int purge,
 345                  int one_shot)
 346 {
 347   ssize_t ret;
 348   size_t left;
 349   char *buf;
 350
 351   left = mst->curr_buf - mst->pos;
 352   buf = (char *) mst->hdr;
 353   ret = GNUNET_NETWORK_socket_recv (sock,
 354                                     &buf[mst->pos],
 355                                     left);
 356   if (-1 == ret)
 357   {
 358     if ( (EAGAIN == errno) ||
 359          (EINTR == errno) )
 360       return GNUNET_OK;
 361     GNUNET_log_strerror (GNUNET_ERROR_TYPE_INFO,
 362                          "recv");
 363     return GNUNET_SYSERR;
 364   }
 365   if (0 == ret)
 366   {
 367     /* other side closed connection, treat as error */
 368     return GNUNET_SYSERR;
 369   }
 370   mst->pos += ret;
 371   return GNUNET_MST_from_buffer (mst,
 372                                  NULL,
 373                                  0,
 374                                  purge,
 375                                  one_shot);
 376 }
 377
 378
 379 /**
 380  * Obtain the next message from the @a mst, assuming that
 381  * there are more unprocessed messages in the internal buffer
 382  * of the @a mst.
 383  *
 384  * @param mst tokenizer to use
 385  * @param one_shot only call callback once, keep rest of message in buffer
 386  * @return #GNUNET_OK if we are done processing (need more data)
 387  *         #GNUNET_NO if one_shot was set and we have another message ready
 388  *         #GNUNET_SYSERR if the data stream is corrupt
 389  */
 390 int
 391 GNUNET_MST_next (struct GNUNET_MessageStreamTokenizer *mst,
 392                  int one_shot)
 393 {
 394   return GNUNET_MST_from_buffer (mst,
 395                                  NULL,
 396                                  0,
 397                                  GNUNET_NO,
 398                                  one_shot);
 399 }
 400
 401
 402 /**
 403  * Destroys a tokenizer.
 404  *
 405  * @param mst tokenizer to destroy
 406  */
 407 void
 408 GNUNET_MST_destroy (struct GNUNET_MessageStreamTokenizer *mst)
 409 {
 410   GNUNET_free (mst->hdr);
 411   GNUNET_free (mst);
 412 }
 413
 414
 415
 416 /* end of server_mst.c */