src/util/server_mst.c

   1 /*
   2      This file is part of GNUnet.
   3      (C) 2010 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 2, 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 util/server_mst.c
  23  * @brief convenience functions for handling inbound message buffers
  24  * @author Christian Grothoff
  25  */
  26
  27 #include "platform.h"
  28 #include "gnunet_common.h"
  29 #include "gnunet_connection_lib.h"
  30 #include "gnunet_scheduler_lib.h"
  31 #include "gnunet_server_lib.h"
  32 #include "gnunet_time_lib.h"
  33
  34 #if HAVE_UNALIGNED_64_ACCESS
  35 #define ALIGN_FACTOR 4
  36 #else
  37 #define ALIGN_FACTOR 8
  38 #endif
  39
  40
  41 /**
  42  * Handle to a message stream tokenizer.
  43  */
  44 struct GNUNET_SERVER_MessageStreamTokenizer
  45 {
  46
  47   /**
  48    * Function to call on completed messages.
  49    */
  50   GNUNET_SERVER_MessageTokenizerCallback cb;
  51
  52   /**
  53    * Closure for cb.
  54    */
  55   void *cb_cls;
  56
  57   /**
  58    * Client to pass to cb.
  59    */
  60   void *client_identity;
  61
  62   /**
  63    * Size of the buffer (starting at 'hdr').
  64    */
  65   size_t maxbuf;
  66
  67   /**
  68    * How many bytes in buffer have we already processed?
  69    */
  70   size_t off;
  71
  72   /**
  73    * How many bytes in buffer are valid right now?
  74    */
  75   size_t pos;
  76
  77   /**
  78    * Beginning of the buffer.  Typed like this to force alignment.
  79    */
  80   struct GNUNET_MessageHeader hdr;
  81
  82 };
  83
  84
  85
  86 /**
  87  * Create a message stream tokenizer.
  88  *
  89  * @param maxbuf maximum message size to support (typically
  90  *    GNUNET_SERVER_MAX_MESSAGE_SIZE)
  91  * @param client_identity ID of client for which this is a buffer,
  92  *        can be NULL (will be passed back to 'cb')
  93  * @param cb function to call on completed messages
  94  * @param cb_cls closure for cb
  95  * @return handle to tokenizer
  96  */
  97 struct GNUNET_SERVER_MessageStreamTokenizer *
  98 GNUNET_SERVER_mst_create (size_t maxbuf,
  99                           void *client_identity,
 100                           GNUNET_SERVER_MessageTokenizerCallback cb,
 101                           void *cb_cls)
 102 {
 103   struct GNUNET_SERVER_MessageStreamTokenizer *ret;
 104
 105   ret = GNUNET_malloc (maxbuf + sizeof (struct GNUNET_SERVER_MessageStreamTokenizer));
 106   ret->maxbuf = maxbuf;
 107   ret->client_identity = client_identity;
 108   ret->cb = cb;
 109   ret->cb_cls = cb_cls;
 110   return ret;
 111 }
 112
 113
 114 /**
 115  * Add incoming data to the receive buffer and call the
 116  * callback for all complete messages.
 117  *
 118  * @param mst tokenizer to use
 119  * @param buf input data to add
 120  * @param size number of bytes in buf
 121  * @param purge should any excess bytes in the buffer be discarded
 122  *       (i.e. for packet-based services like UDP)
 123  * @param one_shot only call callback once, keep rest of message in buffer
 124  * @return GNUNET_OK if we are done processing (need more data)
 125  *         GNUNET_NO if one_shot was set and we have another message ready
 126  *         GNUNET_SYSERR if the data stream is corrupt
 127  */
 128 int
 129 GNUNET_SERVER_mst_receive (struct GNUNET_SERVER_MessageStreamTokenizer *mst,
 130                            const char *buf,
 131                            size_t size,
 132                            int purge,
 133                            int one_shot)
 134 {
 135   const struct GNUNET_MessageHeader *hdr;
 136   size_t delta;
 137   uint16_t want;
 138   char *ibuf;
 139   int need_align;
 140   unsigned long offset;
 141   int ret;
 142
 143   ret = GNUNET_OK;
 144   ibuf = (char*) &mst->hdr;
 145   if (mst->pos > 0)
 146     {
 147     do_align:
 148       if ( (mst->maxbuf - mst->off < sizeof (struct GNUNET_MessageHeader)) ||
 149            (0 != (mst->off % ALIGN_FACTOR)) )
 150         {
 151           /* need to align or need more space */
 152           mst->pos -= mst->off;
 153           memmove (ibuf,
 154                    &ibuf[mst->off],
 155                    mst->pos);
 156           mst->off = 0;
 157         }
 158       if (mst->pos - mst->off < sizeof (struct GNUNET_MessageHeader))
 159         {
 160           delta = GNUNET_MIN (sizeof (struct GNUNET_MessageHeader) - (mst->pos - mst->off),
 161                               size);
 162           memcpy (&ibuf[mst->pos],
 163                   buf,
 164                   delta);
 165           mst->pos += delta;
 166           buf += delta;
 167           size -= delta;
 168         }
 169       if (mst->pos - mst->off < sizeof (struct GNUNET_MessageHeader))
 170         {
 171           if (purge)
 172             {
 173               mst->off = 0;
 174               mst->pos = 0;
 175             }
 176           return GNUNET_OK;
 177         }
 178       hdr = (const struct GNUNET_MessageHeader*) &ibuf[mst->off];
 179       want = ntohs (hdr->size);
 180       if (want < sizeof (struct GNUNET_MessageHeader))
 181         {
 182           GNUNET_break_op (0);
 183           return GNUNET_SYSERR;
 184         }
 185       if (mst->maxbuf - mst->off < want)
 186         {
 187           /* need more space */
 188           mst->pos -= mst->off;
 189           memmove (ibuf,
 190                    &ibuf[mst->off],
 191                    mst->pos);
 192           mst->off = 0;
 193         }
 194       if (mst->pos - mst->off < want)
 195         {
 196           delta = GNUNET_MIN (want - (mst->pos - mst->off),
 197                               size);
 198           memcpy (&ibuf[mst->pos],
 199                   buf,
 200                   delta);
 201           mst->pos += delta;
 202           buf += delta;
 203           size -= delta;
 204         }
 205       if (mst->pos - mst->off < want)
 206         {
 207           if (purge)
 208             {
 209               mst->off = 0;
 210               mst->pos = 0;
 211             }
 212           return GNUNET_OK;
 213         }
 214       if (one_shot == GNUNET_SYSERR)
 215         {
 216           /* cannot call callback again, but return value saying that
 217              we have another full message in the buffer */
 218           ret = GNUNET_NO;
 219           goto copy;
 220         }
 221       if (one_shot == GNUNET_YES)
 222         one_shot = GNUNET_SYSERR;
 223       mst->cb (mst->cb_cls, mst->client_identity, hdr);
 224       mst->off += want;
 225       if (mst->off == mst->pos)
 226         {
 227           /* reset to beginning of buffer, it's free right now! */
 228           mst->off = 0;
 229           mst->pos = 0;
 230         }
 231     }
 232   while (size > 0)
 233     {
 234       if (size < sizeof (struct GNUNET_MessageHeader))
 235         break;
 236       offset = (unsigned long) buf;
 237       need_align = (0 != offset % ALIGN_FACTOR) ? GNUNET_YES : GNUNET_NO;
 238       if (GNUNET_NO == need_align)
 239         {
 240           /* can try to do zero-copy and process directly from original buffer */
 241           hdr = (const struct GNUNET_MessageHeader *) buf;
 242           want = ntohs (hdr->size);
 243           if (size < want)
 244             break; /* or not, buffer incomplete, so copy to private buffer... */
 245           if (one_shot == GNUNET_SYSERR)
 246             {
 247               /* cannot call callback again, but return value saying that
 248                  we have another full message in the buffer */
 249               ret = GNUNET_NO;
 250               goto copy;
 251             }
 252           if (one_shot == GNUNET_YES)
 253             one_shot = GNUNET_SYSERR;
 254           mst->cb (mst->cb_cls, mst->client_identity, hdr);
 255           buf += want;
 256           size -= want;
 257         }
 258       else
 259         {
 260           /* need to copy to private buffer to align;
 261              yes, we go a bit more spagetti than usual here */
 262           goto do_align;
 263         }
 264     }
 265  copy:
 266   if ( (size > 0) && (! purge) )
 267     {
 268       GNUNET_assert (mst->pos + size <= mst->maxbuf);
 269       memcpy (&ibuf[mst->pos], buf, size);
 270       mst->pos += size;
 271     }
 272   if (purge)
 273     mst->off = 0;
 274   return ret;
 275 }
 276
 277
 278 /**
 279  * Destroys a tokenizer.
 280  *
 281  * @param mst tokenizer to destroy
 282  */
 283 void
 284 GNUNET_SERVER_mst_destroy (struct GNUNET_SERVER_MessageStreamTokenizer *mst)
 285 {
 286   GNUNET_free (mst);
 287 }
 288
 289
 290
 291 /* end of server_mst.c */