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