src/conversation/mst.c

   1 /*
   2      This file is part of GNUnet.
   3      (C) 2008, 2011 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 3, 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 conversation/mst.c
  23  * @brief Message tokenizer
  24  * @author Christian Grothoff
  25  */
  26
  27 #include <gnunet/platform.h>
  28 #include <gnunet/gnunet_constants.h>
  29
  30 /**
  31  * To what multiple do we align messages?  8 byte should suffice for everyone
  32  * for now.
  33  */
  34 #define ALIGN_FACTOR 8
  35
  36 /**
  37  * Smallest supported message.
  38  */
  39 #define MIN_BUFFER_SIZE sizeof (struct GNUNET_MessageHeader)
  40
  41
  42 /**
  43  * Functions with this signature are called whenever a
  44  * complete message is received by the tokenizer.
  45  *
  46  * @param cls closure
  47  * @param message the actual message
  48  */
  49 typedef void (*MessageTokenizerCallback) (void *cls,
  50                                           const struct
  51                                           GNUNET_MessageHeader *
  52                                           message);
  53
  54 /**
  55  * Handle to a message stream tokenizer.
  56  */
  57 struct MessageStreamTokenizer
  58 {
  59
  60   /**
  61    * Function to call on completed messages.
  62    */
  63   MessageTokenizerCallback cb;
  64
  65   /**
  66    * Closure for cb.
  67    */
  68   void *cb_cls;
  69
  70   /**
  71    * Size of the buffer (starting at 'hdr').
  72    */
  73   size_t curr_buf;
  74
  75   /**
  76    * How many bytes in buffer have we already processed?
  77    */
  78   size_t off;
  79
  80   /**
  81    * How many bytes in buffer are valid right now?
  82    */
  83   size_t pos;
  84
  85   /**
  86    * Beginning of the buffer.  Typed like this to force alignment.
  87    */
  88   struct GNUNET_MessageHeader *hdr;
  89
  90 };
  91
  92
  93 /**
  94  * Create a message stream tokenizer.
  95  *
  96  * @param cb function to call on completed messages
  97  * @param cb_cls closure for cb
  98  * @return handle to tokenizer
  99  */
 100 static struct MessageStreamTokenizer *
 101 mst_create (MessageTokenizerCallback cb,
 102                 void *cb_cls)
 103 {
 104   struct MessageStreamTokenizer *ret;
 105
 106   ret = malloc (sizeof (struct MessageStreamTokenizer));
 107   if (NULL == ret)
 108   {
 109         fprintf (stderr, "Failed to allocate buffer for tokenizer\n");
 110         exit (1);
 111   }
 112   ret->hdr = malloc (MIN_BUFFER_SIZE);
 113   if (NULL == ret->hdr)
 114   {
 115         fprintf (stderr, "Failed to allocate buffer for alignment\n");
 116         exit (1);
 117   }
 118   ret->curr_buf = MIN_BUFFER_SIZE;
 119   ret->cb = cb;
 120   ret->cb_cls = cb_cls;
 121   return ret;
 122 }
 123
 124
 125 /**
 126  * Add incoming data to the receive buffer and call the
 127  * callback for all complete messages.
 128  *
 129  * @param mst tokenizer to use
 130  * @param buf input data to add
 131  * @param size number of bytes in buf
 132  * @return GNUNET_OK if we are done processing (need more data)
 133  *         GNUNET_SYSERR if the data stream is corrupt
 134  */
 135 static int
 136 mst_receive (struct MessageStreamTokenizer *mst,
 137                  const char *buf, size_t size)
 138 {
 139   const struct GNUNET_MessageHeader *hdr;
 140   size_t delta;
 141   uint16_t want;
 142   char *ibuf;
 143   int need_align;
 144   unsigned long offset;
 145   int ret;
 146
 147   ret = GNUNET_OK;
 148   ibuf = (char *) mst->hdr;
 149   while (mst->pos > 0)
 150   {
 151 do_align:
 152         if ((mst->curr_buf - mst->off < sizeof (struct GNUNET_MessageHeader)) ||
 153                 (0 != (mst->off % ALIGN_FACTOR)))
 154         {
 155           /* need to align or need more space */
 156           mst->pos -= mst->off;
 157           memmove (ibuf, &ibuf[mst->off], mst->pos);
 158           mst->off = 0;
 159         }
 160         if (mst->pos - mst->off < sizeof (struct GNUNET_MessageHeader))
 161         {
 162           delta =
 163                   GNUNET_MIN (sizeof (struct GNUNET_MessageHeader) -
 164                                           (mst->pos - mst->off), size);
 165           memcpy (&ibuf[mst->pos], buf, delta);
 166           mst->pos += delta;
 167           buf += delta;
 168           size -= delta;
 169         }
 170         if (mst->pos - mst->off < sizeof (struct GNUNET_MessageHeader))
 171         {
 172           return GNUNET_OK;
 173         }
 174         hdr = (const struct GNUNET_MessageHeader *) &ibuf[mst->off];
 175         want = ntohs (hdr->size);
 176         if (want < sizeof (struct GNUNET_MessageHeader))
 177         {
 178           fprintf (stderr,
 179                    "Received invalid message from stdin\n");
 180           exit (1);
 181         }
 182         if (mst->curr_buf - mst->off < want)
 183         {
 184           /* need more space */
 185           mst->pos -= mst->off;
 186           memmove (ibuf, &ibuf[mst->off], mst->pos);
 187           mst->off = 0;
 188         }
 189         if (want > mst->curr_buf)
 190         {
 191           mst->hdr = realloc (mst->hdr, want);
 192           if (NULL == mst->hdr)
 193           {
 194         fprintf (stderr, "Failed to allocate buffer for alignment\n");
 195         exit (1);
 196           }
 197           ibuf = (char *) mst->hdr;
 198           mst->curr_buf = want;
 199         }
 200         hdr = (const struct GNUNET_MessageHeader *) &ibuf[mst->off];
 201         if (mst->pos - mst->off < want)
 202         {
 203           delta = GNUNET_MIN (want - (mst->pos - mst->off), size);
 204           memcpy (&ibuf[mst->pos], buf, delta);
 205           mst->pos += delta;
 206           buf += delta;
 207           size -= delta;
 208         }
 209         if (mst->pos - mst->off < want)
 210         {
 211           return GNUNET_OK;
 212         }
 213         mst->cb (mst->cb_cls, hdr);
 214         mst->off += want;
 215         if (mst->off == mst->pos)
 216         {
 217           /* reset to beginning of buffer, it's free right now! */
 218           mst->off = 0;
 219           mst->pos = 0;
 220         }
 221   }
 222   while (size > 0)
 223   {
 224         if (size < sizeof (struct GNUNET_MessageHeader))
 225           break;
 226         offset = (unsigned long) buf;
 227         need_align = (0 != offset % ALIGN_FACTOR) ? GNUNET_YES : GNUNET_NO;
 228         if (GNUNET_NO == need_align)
 229         {
 230           /* can try to do zero-copy and process directly from original buffer */
 231           hdr = (const struct GNUNET_MessageHeader *) buf;
 232           want = ntohs (hdr->size);
 233           if (want < sizeof (struct GNUNET_MessageHeader))
 234           {
 235         fprintf (stderr,
 236                  "Received invalid message from stdin\n");
 237         exit (1);
 238           }
 239           if (size < want)
 240                 break;                  /* or not, buffer incomplete, so copy to private buffer... */
 241           mst->cb (mst->cb_cls, hdr);
 242           buf += want;
 243           size -= want;
 244         }
 245         else
 246         {
 247           /* need to copy to private buffer to align;
 248            * yes, we go a bit more spagetti than usual here */
 249           goto do_align;
 250         }
 251   }
 252   if (size > 0)
 253   {
 254         if (size + mst->pos > mst->curr_buf)
 255         {
 256           mst->hdr = realloc (mst->hdr, size + mst->pos);
 257           if (NULL == mst->hdr)
 258           {
 259         fprintf (stderr, "Failed to allocate buffer for alignment\n");
 260         exit (1);
 261           }
 262           ibuf = (char *) mst->hdr;
 263           mst->curr_buf = size + mst->pos;
 264         }
 265         if (mst->pos + size > mst->curr_buf)
 266         {
 267           fprintf (stderr,
 268                    "Assertion failed\n");
 269           exit (1);
 270         }
 271         memcpy (&ibuf[mst->pos], buf, size);
 272         mst->pos += size;
 273   }
 274   return ret;
 275 }
 276
 277
 278 /**
 279  * Destroys a tokenizer.
 280  *
 281  * @param mst tokenizer to destroy
 282  */
 283 static void
 284 mst_destroy (struct MessageStreamTokenizer *mst)
 285 {
 286   free (mst->hdr);
 287   free (mst);
 288 }