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