From 739ac84730e1d5f1b5affdeaedec11a986d2faa6 Mon Sep 17 00:00:00 2001 From: Christian Grothoff Date: Tue, 22 Jun 2010 14:13:02 +0000 Subject: [PATCH] mst fixes --- src/util/server_mst.c | 131 +++++++++++++++++++++++++++++++----------- 1 file changed, 96 insertions(+), 35 deletions(-) diff --git a/src/util/server_mst.c b/src/util/server_mst.c index 8df4e4b7d..27c95815a 100644 --- a/src/util/server_mst.c +++ b/src/util/server_mst.c @@ -1,6 +1,6 @@ /* This file is part of GNUnet. - (C) 2009 Christian Grothoff (and other contributing authors) + (C) 2010 Christian Grothoff (and other contributing authors) GNUnet is free software; you can redistribute it and/or modify it under the terms of the GNU General Public License as published @@ -31,6 +31,12 @@ #include "gnunet_server_lib.h" #include "gnunet_time_lib.h" +#if HAVE_UNALIGNED_64_ACCESS +#define ALIGN_FACTOR 4 +#else +#define ALIGN_FACTOR 8 +#endif + /** * Handle to a message stream tokenizer. @@ -38,18 +44,38 @@ struct GNUNET_SERVER_MessageStreamTokenizer { - size_t maxbuf; - - size_t off; + /** + * Function to call on completed messages. + */ + GNUNET_SERVER_MessageTokenizerCallback cb; + + /** + * Closure for cb. + */ + void *cb_cls; + /** + * Client to pass to cb. + */ void *client_identity; - GNUNET_SERVER_MessageTokenizerCallback cb; + /** + * Size of the buffer (starting at 'hdr'). + */ + size_t maxbuf; - void *cb_cls; + /** + * How many bytes in buffer have we already processed? + */ + size_t off; + + /** + * How many bytes in buffer are valid right now? + */ + size_t pos; /** - * Beginning of the buffer. + * Beginning of the buffer. Typed like this to force alignment. */ struct GNUNET_MessageHeader hdr; @@ -64,6 +90,8 @@ struct GNUNET_SERVER_MessageStreamTokenizer * GNUNET_SERVER_MAX_MESSAGE_SIZE) * @param client_identity ID of client for which this is a buffer, * can be NULL (will be passed back to 'cb') + * @param cb function to call on completed messages + * @param cb_cls closure for cb * @return handle to tokenizer */ struct GNUNET_SERVER_MessageStreamTokenizer * @@ -106,7 +134,7 @@ GNUNET_SERVER_mst_receive (struct GNUNET_SERVER_MessageStreamTokenizer *mst, { const struct GNUNET_MessageHeader *hdr; size_t delta; - size_t want; + uint16_t want; char *ibuf; int need_align; unsigned long offset; @@ -114,78 +142,110 @@ GNUNET_SERVER_mst_receive (struct GNUNET_SERVER_MessageStreamTokenizer *mst, ret = GNUNET_OK; ibuf = (char*) &mst->hdr; - if (mst->off > 0) + if (mst->pos > 0) { do_align: - if (mst->off < sizeof (struct GNUNET_MessageHeader)) + if ( (mst->maxbuf - mst->off < sizeof (struct GNUNET_MessageHeader)) || + (0 != (mst->off % ALIGN_FACTOR)) ) + { + /* need to align or need more space */ + mst->pos -= mst->off; + memmove (ibuf, + &ibuf[mst->off], + mst->pos); + mst->off = 0; + } + if (mst->pos - mst->off < sizeof (struct GNUNET_MessageHeader)) { - delta = GNUNET_MIN (sizeof (struct GNUNET_MessageHeader) - mst->off, + delta = GNUNET_MIN (sizeof (struct GNUNET_MessageHeader) - (mst->pos - mst->off), size); - memcpy (&ibuf[mst->off], + memcpy (&ibuf[mst->pos], buf, delta); - mst->off += delta; + mst->pos += delta; buf += delta; size -= delta; } - if (mst->off < sizeof (struct GNUNET_MessageHeader)) + if (mst->pos - mst->off < sizeof (struct GNUNET_MessageHeader)) { if (purge) - mst->off = 0; + { + mst->off = 0; + mst->pos = 0; + } return GNUNET_OK; } - want = ntohs (mst->hdr.size); + hdr = (const struct GNUNET_MessageHeader*) &ibuf[mst->off]; + want = ntohs (hdr->size); if (want < sizeof (struct GNUNET_MessageHeader)) { GNUNET_break_op (0); return GNUNET_SYSERR; } - if (want < mst->off) + if (mst->maxbuf - mst->off < want) { - delta = GNUNET_MIN (want - mst->off, + /* need more space */ + mst->pos -= mst->off; + memmove (ibuf, + &ibuf[mst->off], + mst->pos); + mst->off = 0; + } + if (mst->pos - mst->off < want) + { + delta = GNUNET_MIN (want - (mst->pos - mst->off), size); - memcpy (&ibuf[mst->off], + memcpy (&ibuf[mst->pos], buf, delta); - mst->off += delta; + mst->pos += delta; buf += delta; size -= delta; } - if (want < mst->off) + if (mst->pos - mst->off < want) { if (purge) - mst->off = 0; + { + mst->off = 0; + mst->pos = 0; + } return GNUNET_OK; } if (one_shot == GNUNET_SYSERR) { + /* cannot call callback again, but return value saying that + we have another full message in the buffer */ ret = GNUNET_NO; goto copy; } if (one_shot == GNUNET_YES) one_shot = GNUNET_SYSERR; - mst->cb (mst->cb_cls, mst->client_identity, &mst->hdr); - mst->off = 0; + mst->cb (mst->cb_cls, mst->client_identity, hdr); + mst->off += want; + if (mst->off == mst->pos) + { + /* reset to beginning of buffer, it's free right now! */ + mst->off = 0; + mst->pos = 0; + } } while (size > 0) { if (size < sizeof (struct GNUNET_MessageHeader)) break; offset = (unsigned long) buf; -#if HAVE_UNALIGNED_64_ACCESS - need_align = (0 != offset % 4) ? GNUNET_YES : GNUNET_NO; -#else - need_align = (0 != offset % 8) ? GNUNET_YES : GNUNET_NO; -#endif + need_align = (0 != offset % ALIGN_FACTOR) ? GNUNET_YES : GNUNET_NO; if (GNUNET_NO == need_align) { - /* can try to do zero-copy */ + /* can try to do zero-copy and process directly from original buffer */ hdr = (const struct GNUNET_MessageHeader *) buf; want = ntohs (hdr->size); if (size < want) - break; /* or not, buffer incomplete... */ + break; /* or not, buffer incomplete, so copy to private buffer... */ if (one_shot == GNUNET_SYSERR) { + /* cannot call callback again, but return value saying that + we have another full message in the buffer */ ret = GNUNET_NO; goto copy; } @@ -197,16 +257,17 @@ GNUNET_SERVER_mst_receive (struct GNUNET_SERVER_MessageStreamTokenizer *mst, } else { - /* yes, we go a bit more spagetti than usual here */ + /* need to copy to private buffer to align; + yes, we go a bit more spagetti than usual here */ goto do_align; } } copy: if ( (size > 0) && (! purge) ) { - memcpy (&ibuf[mst->off], buf, size); - mst->off += size; - size = 0; + GNUNET_assert (mst->pos + size <= mst->maxbuf); + memcpy (&ibuf[mst->pos], buf, size); + mst->pos += size; } if (purge) mst->off = 0; -- 2.25.1