diff --git a/src/output-json-http.c b/src/output-json-http.c index 4eb8ac7d32..0dbbfee7d1 100644 --- a/src/output-json-http.c +++ b/src/output-json-http.c @@ -380,7 +380,7 @@ static void EveHttpLogJSONHeaders(JsonBuilder *js, uint32_t direction, htp_tx_t static void BodyPrintableBuffer(JsonBuilder *js, HtpBody *body, const char *key) { - if (body->sb != NULL && body->sb->buf != NULL) { + if (body->sb != NULL && body->sb->region.buf != NULL) { uint32_t offset = 0; const uint8_t *body_data; uint32_t body_data_len; @@ -418,7 +418,7 @@ void EveHttpLogJSONBodyPrintable(JsonBuilder *js, Flow *f, uint64_t tx_id) static void BodyBase64Buffer(JsonBuilder *js, HtpBody *body, const char *key) { - if (body->sb != NULL && body->sb->buf != NULL) { + if (body->sb != NULL && body->sb->region.buf != NULL) { const uint8_t *body_data; uint32_t body_data_len; uint64_t body_offset; diff --git a/src/stream-tcp-private.h b/src/stream-tcp-private.h index 4f44fdcc8e..1de3cb7f63 100644 --- a/src/stream-tcp-private.h +++ b/src/stream-tcp-private.h @@ -140,7 +140,7 @@ typedef struct TcpStream_ { struct TCPSACK sack_tree; /**< red back tree of TCP SACK records. */ } TcpStream; -#define STREAM_BASE_OFFSET(stream) ((stream)->sb.stream_offset) +#define STREAM_BASE_OFFSET(stream) ((stream)->sb.region.stream_offset) #define STREAM_APP_PROGRESS(stream) (STREAM_BASE_OFFSET((stream)) + (stream)->app_progress_rel) #define STREAM_RAW_PROGRESS(stream) (STREAM_BASE_OFFSET((stream)) + (stream)->raw_progress_rel) #define STREAM_LOG_PROGRESS(stream) (STREAM_BASE_OFFSET((stream)) + (stream)->log_progress_rel) diff --git a/src/stream-tcp-reassemble.c b/src/stream-tcp-reassemble.c index 9db65f483e..fa0636dd4d 100644 --- a/src/stream-tcp-reassemble.c +++ b/src/stream-tcp-reassemble.c @@ -686,10 +686,10 @@ static uint32_t StreamTcpReassembleCheckDepth(TcpSession *ssn, TcpStream *stream uint32_t StreamDataAvailableForProtoDetect(TcpStream *stream) { if (RB_EMPTY(&stream->sb.sbb_tree)) { - if (stream->sb.stream_offset != 0) + if (stream->sb.region.stream_offset != 0) return 0; - return stream->sb.buf_offset; + return stream->sb.region.buf_offset; } else { DEBUG_VALIDATE_BUG_ON(stream->sb.head == NULL); DEBUG_VALIDATE_BUG_ON(stream->sb.sbb_size == 0); @@ -913,7 +913,6 @@ uint8_t StreamNeedsReassembly(const TcpSession *ssn, uint8_t direction) } const uint64_t right_edge = StreamingBufferGetConsecutiveDataRightEdge(&stream->sb); - SCLogDebug("%s: app %"PRIu64" (use: %s), raw %"PRIu64" (use: %s). Stream right edge: %"PRIu64, dirstr, STREAM_APP_PROGRESS(stream), use_app ? "yes" : "no", @@ -945,7 +944,7 @@ static uint64_t GetStreamSize(TcpStream *stream) uint64_t last_ack_abs = GetAbsLastAck(stream); uint64_t last_re = 0; - SCLogDebug("stream_offset %" PRIu64, stream->sb.stream_offset); + SCLogDebug("stream_offset %" PRIu64, stream->sb.region.stream_offset); TcpSegment *seg; RB_FOREACH(seg, TCPSEG, &stream->seg_tree) { @@ -1065,6 +1064,7 @@ static bool GetAppBuffer(const TcpStream *stream, const uint8_t **data, uint32_t SCLogDebug("blk at offset"); StreamingBufferSBBGetData(&stream->sb, blk, data, data_len); + BUG_ON(blk->len != *data_len); gap_ahead = check_for_gap && GapAhead(stream, blk); diff --git a/src/util-streaming-buffer.c b/src/util-streaming-buffer.c index 3357dc8489..faf209f3c3 100644 --- a/src/util-streaming-buffer.c +++ b/src/util-streaming-buffer.c @@ -1,4 +1,4 @@ -/* Copyright (C) 2015-2016 Open Information Security Foundation +/* Copyright (C) 2015-2023 Open Information Security Foundation * * You can copy, redistribute or modify this Program under the terms of * the GNU General Public License version 2 as published by the Free @@ -22,6 +22,11 @@ #include "util-validate.h" #include "util-debug.h" +static void ListRegions(StreamingBuffer *sb); +#define REGION_MAX_GAP 250000 + +#define DUMP_REGIONS 0 // set to 1 to dump a visual representation of the regions list and sbb tree. + /** * \file * @@ -45,8 +50,7 @@ RB_GENERATE(SBB, StreamingBufferBlock, rb, SBBCompare); int SBBCompare(struct StreamingBufferBlock *a, struct StreamingBufferBlock *b) { - SCLogDebug("a %"PRIu64" len %u, b %"PRIu64" len %u", - a->offset, a->len, b->offset, b->len); + SCLogDebug("a %" PRIu64 " len %u, b %" PRIu64 " len %u", a->offset, a->len, b->offset, b->len); if (a->offset > b->offset) SCReturnInt(1); @@ -77,12 +81,12 @@ static inline int InclusiveCompare(StreamingBufferBlock *lookup, StreamingBuffer StreamingBufferBlock *SBB_RB_FIND_INCLUSIVE(struct SBB *head, StreamingBufferBlock *elm) { - SCLogDebug("looking up %"PRIu64, elm->offset); + SCLogDebug("looking up %" PRIu64, elm->offset); struct StreamingBufferBlock *tmp = RB_ROOT(head); struct StreamingBufferBlock *res = NULL; while (tmp) { - SCLogDebug("compare with %"PRIu64"/%u", tmp->offset, tmp->len); + SCLogDebug("compare with %" PRIu64 "/%u", tmp->offset, tmp->len); const int comp = InclusiveCompare(elm, tmp); SCLogDebug("compare result: %d", comp); if (comp < 0) { @@ -97,14 +101,32 @@ StreamingBufferBlock *SBB_RB_FIND_INCLUSIVE(struct SBB *head, StreamingBufferBlo return res; } +static inline StreamingBufferRegion *InitBufferRegion(StreamingBuffer *sb, const uint32_t min_size) +{ + StreamingBufferRegion *aux_r = CALLOC(sb->cfg, 1, sizeof(*aux_r)); + if (aux_r == NULL) + return NULL; + + aux_r->buf = CALLOC(sb->cfg, 1, MAX(sb->cfg->buf_size, min_size)); + if (aux_r->buf == NULL) { + FREE(sb->cfg, aux_r, sizeof(*aux_r)); + return NULL; + } + aux_r->buf_size = MAX(sb->cfg->buf_size, min_size); + sb->regions++; + sb->max_regions = MAX(sb->regions, sb->max_regions); + + BUG_ON(sb->regions > 100); + return aux_r; +} static inline int InitBuffer(StreamingBuffer *sb) { - sb->buf = CALLOC(sb->cfg, 1, sb->cfg->buf_size); - if (sb->buf == NULL) { + sb->region.buf = CALLOC(sb->cfg, 1, sb->cfg->buf_size); + if (sb->region.buf == NULL) { return -1; } - sb->buf_size = sb->cfg->buf_size; + sb->region.buf_size = sb->cfg->buf_size; return 0; } @@ -112,8 +134,9 @@ StreamingBuffer *StreamingBufferInit(const StreamingBufferConfig *cfg) { StreamingBuffer *sb = CALLOC(cfg, 1, sizeof(StreamingBuffer)); if (sb != NULL) { - sb->buf_size = cfg->buf_size; + sb->region.buf_size = cfg->buf_size; sb->cfg = cfg; + sb->regions = sb->max_regions = 1; if (cfg->buf_size > 0) { if (InitBuffer(sb) == 0) { @@ -131,13 +154,23 @@ StreamingBuffer *StreamingBufferInit(const StreamingBufferConfig *cfg) void StreamingBufferClear(StreamingBuffer *sb) { if (sb != NULL) { - SCLogDebug("sb->buf_size %u max %u", sb->buf_size, sb->buf_size_max); + SCLogDebug("sb->region.buf_size %u max %u", sb->region.buf_size, sb->buf_size_max); SBBFree(sb); - if (sb->buf != NULL) { - FREE(sb->cfg, sb->buf, sb->buf_size); - sb->buf = NULL; + ListRegions(sb); + if (sb->region.buf != NULL) { + FREE(sb->cfg, sb->region.buf, sb->region.buf_size); + sb->region.buf = NULL; + } + + for (StreamingBufferRegion *r = sb->region.next; r != NULL;) { + StreamingBufferRegion *next = r->next; + FREE(sb->cfg, r->buf, r->buf_size); + FREE(sb->cfg, r, sizeof(*r)); + r = next; } + sb->region.next = NULL; + sb->regions = sb->max_regions = 1; } } @@ -154,11 +187,11 @@ static void SBBPrintList(StreamingBuffer *sb) { StreamingBufferBlock *sbb = NULL; RB_FOREACH(sbb, SBB, &sb->sbb_tree) { - SCLogDebug("sbb: offset %"PRIu64", len %u", sbb->offset, sbb->len); + SCLogDebug("sbb: offset %" PRIu64 ", len %u", sbb->offset, sbb->len); StreamingBufferBlock *next = SBB_RB_NEXT(sbb); if (next) { if ((sbb->offset + sbb->len) != next->offset) { - SCLogDebug("gap: offset %"PRIu64", len %"PRIu64, (sbb->offset + sbb->len), + SCLogDebug("gap: offset %" PRIu64 ", len %" PRIu64, (sbb->offset + sbb->len), next->offset - (sbb->offset + sbb->len)); } } @@ -170,26 +203,26 @@ static void SBBPrintList(StreamingBuffer *sb) * * [block][gap][block] **/ -static void SBBInit(StreamingBuffer *sb, - uint32_t rel_offset, uint32_t data_len) +static void SBBInit( + StreamingBuffer *sb, StreamingBufferRegion *region, uint32_t rel_offset, uint32_t data_len) { DEBUG_VALIDATE_BUG_ON(!RB_EMPTY(&sb->sbb_tree)); - DEBUG_VALIDATE_BUG_ON(sb->buf_offset > sb->stream_offset + rel_offset); + DEBUG_VALIDATE_BUG_ON(region->buf_offset > region->stream_offset + rel_offset); /* need to set up 2: existing data block and new data block */ StreamingBufferBlock *sbb = CALLOC(sb->cfg, 1, sizeof(*sbb)); if (sbb == NULL) { return; } - sbb->offset = sb->stream_offset; - sbb->len = sb->buf_offset; + sbb->offset = sb->region.stream_offset; + sbb->len = sb->region.buf_offset; StreamingBufferBlock *sbb2 = CALLOC(sb->cfg, 1, sizeof(*sbb2)); if (sbb2 == NULL) { FREE(sb->cfg, sbb, sizeof(*sbb)); return; } - sbb2->offset = sb->stream_offset + rel_offset; + sbb2->offset = region->stream_offset + rel_offset; sbb2->len = data_len; sb->head = sbb; @@ -197,8 +230,8 @@ static void SBBInit(StreamingBuffer *sb, SBB_RB_INSERT(&sb->sbb_tree, sbb); SBB_RB_INSERT(&sb->sbb_tree, sbb2); - SCLogDebug("sbb1 %"PRIu64", len %u, sbb2 %"PRIu64", len %u", - sbb->offset, sbb->len, sbb2->offset, sbb2->len); + SCLogDebug("sbb1 %" PRIu64 ", len %u, sbb2 %" PRIu64 ", len %u", sbb->offset, sbb->len, + sbb2->offset, sbb2->len); #ifdef DEBUG SBBPrintList(sb); #endif @@ -209,8 +242,8 @@ static void SBBInit(StreamingBuffer *sb, * * [gap][block] **/ -static void SBBInitLeadingGap(StreamingBuffer *sb, - uint64_t offset, uint32_t data_len) +static void SBBInitLeadingGap( + StreamingBuffer *sb, StreamingBufferRegion *region, uint64_t offset, uint32_t data_len) { DEBUG_VALIDATE_BUG_ON(!RB_EMPTY(&sb->sbb_tree)); @@ -224,14 +257,13 @@ static void SBBInitLeadingGap(StreamingBuffer *sb, sb->sbb_size = sbb->len; SBB_RB_INSERT(&sb->sbb_tree, sbb); - SCLogDebug("sbb %"PRIu64", len %u", - sbb->offset, sbb->len); + SCLogDebug("sbb %" PRIu64 ", len %u", sbb->offset, sbb->len); #ifdef DEBUG SBBPrintList(sb); #endif } -static inline void ConsolidateFwd(StreamingBuffer *sb, +static inline void ConsolidateFwd(StreamingBuffer *sb, StreamingBufferRegion *region, struct SBB *tree, StreamingBufferBlock *sa) { uint64_t sa_re = sa->offset + sa->len; @@ -241,13 +273,14 @@ static inline void ConsolidateFwd(StreamingBuffer *sb, continue; const uint64_t tr_re = tr->offset + tr->len; - SCLogDebug("-> (fwd) tr %p %"PRIu64"/%u re %"PRIu64, - tr, tr->offset, tr->len, tr_re); + SCLogDebug("-> (fwd) tr %p %" PRIu64 "/%u re %" PRIu64, tr, tr->offset, tr->len, tr_re); - if (sa_re < tr->offset) + if (sa_re < tr->offset) { + SCLogDebug("entirely before: %" PRIu64 " < %" PRIu64, sa_re, tr->offset); break; // entirely before + } - /* + /* new block (sa) entirely eclipsed by in tree block (tr) sa: [ ] tr: [ ] sa: [ ] @@ -260,28 +293,43 @@ static inline void ConsolidateFwd(StreamingBuffer *sb, sa->len = tr->len; sa->offset = tr->offset; sa_re = sa->offset + sa->len; - SCLogDebug("-> (fwd) tr %p %"PRIu64"/%u REMOVED ECLIPSED2", tr, tr->offset, tr->len); + SCLogDebug("-> (fwd) tr %p %" PRIu64 "/%u REMOVED ECLIPSED (sa overlapped by tr)", tr, + tr->offset, tr->len); SBB_RB_REMOVE(tree, tr); FREE(sb->cfg, tr, sizeof(StreamingBufferBlock)); - /* - sa: [ ] - tr: [ ] - sa: [ ] - tr: [ ] - sa: [ ] - tr: [ ] - */ + /* new block (sa) entire eclipses in tree block (tr) + sa: [ ] + tr: [ ] + sa: [ ] + tr: [ ] + sa: [ ] + tr: [ ] + */ } else if (sa->offset <= tr->offset && sa_re >= tr_re) { - SCLogDebug("-> (fwd) tr %p %"PRIu64"/%u REMOVED ECLIPSED", tr, tr->offset, tr->len); + SCLogDebug("-> (fwd) tr %p %" PRIu64 "/%u REMOVED ECLIPSED (tr overlapped by sa)", tr, + tr->offset, tr->len); SBB_RB_REMOVE(tree, tr); sb->sbb_size -= tr->len; FREE(sb->cfg, tr, sizeof(StreamingBufferBlock)); - /* - sa: [ ] - tr: [ ] - sa: [ ] - tr: [ ] - */ + + SCLogDebug("-> (fwd) tr %p %" PRIu64 "/%u region %p so %" PRIu64 " bo %u sz %u", sa, + sa->offset, sa->len, region, region->stream_offset, region->buf_offset, + region->buf_size); + if (sa->offset == region->stream_offset && + sa_re > (region->stream_offset + region->buf_offset)) { + BUG_ON(sa_re < region->stream_offset); + region->buf_offset = sa_re - region->stream_offset; + SCLogDebug("-> (fwd) tr %p %" PRIu64 "/%u region %p so %" PRIu64 + " bo %u sz %u BUF_OFFSET UPDATED", + sa, sa->offset, sa->len, region, region->stream_offset, region->buf_offset, + region->buf_size); + } + /* new block (sa) extended by in tree block (tr) + sa: [ ] + tr: [ ] + sa: [ ] + tr: [ ] + */ } else if (sa->offset < tr->offset && // starts before sa_re >= tr->offset && sa_re < tr_re) // ends inside { @@ -289,15 +337,28 @@ static inline void ConsolidateFwd(StreamingBuffer *sb, uint32_t combined_len = sa->len + tr->len; sa->len = tr_re - sa->offset; sa_re = sa->offset + sa->len; - SCLogDebug("-> (fwd) tr %p %"PRIu64"/%u REMOVED MERGED", tr, tr->offset, tr->len); + SCLogDebug("-> (fwd) tr %p %" PRIu64 "/%u REMOVED MERGED", tr, tr->offset, tr->len); SBB_RB_REMOVE(tree, tr); sb->sbb_size -= (combined_len - sa->len); // remove what we added twice FREE(sb->cfg, tr, sizeof(StreamingBufferBlock)); + SCLogDebug("-> (fwd) tr %p %" PRIu64 "/%u RESULT", sa, sa->offset, sa->len); + SCLogDebug("-> (fwd) tr %p %" PRIu64 "/%u region %p so %" PRIu64 " bo %u sz %u", sa, + sa->offset, sa->len, region, region->stream_offset, region->buf_offset, + region->buf_size); + if (sa->offset == region->stream_offset && + sa_re > (region->stream_offset + region->buf_offset)) { + BUG_ON(sa_re < region->stream_offset); + region->buf_offset = sa_re - region->stream_offset; + SCLogDebug("-> (fwd) tr %p %" PRIu64 "/%u region %p so %" PRIu64 + " bo %u sz %u BUF_OFFSET UPDATED", + sa, sa->offset, sa->len, region, region->stream_offset, region->buf_offset, + region->buf_size); + } } } } -static inline void ConsolidateBackward(StreamingBuffer *sb, +static inline void ConsolidateBackward(StreamingBuffer *sb, StreamingBufferRegion *region, struct SBB *tree, StreamingBufferBlock *sa) { uint64_t sa_re = sa->offset + sa->len; @@ -306,94 +367,138 @@ static inline void ConsolidateBackward(StreamingBuffer *sb, if (sa == tr) continue; const uint64_t tr_re = tr->offset + tr->len; - SCLogDebug("-> (bwd) tr %p %"PRIu64"/%u", tr, tr->offset, tr->len); + SCLogDebug("-> (bwd) tr %p %" PRIu64 "/%u", tr, tr->offset, tr->len); if (sa->offset > tr_re) break; // entirely after + /* new block (sa) entirely eclipsed by in tree block (tr) + sa: [ ] + tr: [ ] + sa: [ ] + tr: [ ] + sa: [ ] + tr: [ ] + */ if (sa->offset >= tr->offset && sa_re <= tr_re) { sb->sbb_size -= sa->len; // sa entirely eclipsed so remove double accounting sa->len = tr->len; sa->offset = tr->offset; sa_re = sa->offset + sa->len; - SCLogDebug("-> (bwd) tr %p %"PRIu64"/%u REMOVED ECLIPSED2", tr, tr->offset, tr->len); + SCLogDebug("-> (bwd) tr %p %" PRIu64 "/%u REMOVED ECLIPSED (sa overlapped by tr)", tr, + tr->offset, tr->len); if (sb->head == tr) sb->head = sa; SBB_RB_REMOVE(tree, tr); FREE(sb->cfg, tr, sizeof(StreamingBufferBlock)); - /* - sa: [ ] - tr: [ ] - sa: [ ] - tr: [ ] - sa: [ ] - tr: [ ] - */ + /* new block (sa) entire eclipses in tree block (tr) + sa: [ ] + tr: [ ] + sa: [ ] + tr: [ ] + sa: [ ] + tr: [ ] + */ } else if (sa->offset <= tr->offset && sa_re >= tr_re) { - SCLogDebug("-> (bwd) tr %p %"PRIu64"/%u REMOVED ECLIPSED", tr, tr->offset, tr->len); + SCLogDebug("-> (bwd) tr %p %" PRIu64 "/%u REMOVED ECLIPSED (tr overlapped by sa)", tr, + tr->offset, tr->len); if (sb->head == tr) sb->head = sa; SBB_RB_REMOVE(tree, tr); sb->sbb_size -= tr->len; // tr entirely eclipsed so remove double accounting FREE(sb->cfg, tr, sizeof(StreamingBufferBlock)); - /* - sa: [ ] - tr: [ ] - sa: [ ] - tr: [ ] - */ + + SCLogDebug("-> (bwd) tr %p %" PRIu64 "/%u region %p so %" PRIu64 " bo %u sz %u", sa, + sa->offset, sa->len, region, region->stream_offset, region->buf_offset, + region->buf_size); + + if (sa->offset == region->stream_offset && + sa_re > (region->stream_offset + region->buf_offset)) { + BUG_ON(sa_re < region->stream_offset); + region->buf_offset = sa_re - region->stream_offset; + SCLogDebug("-> (bwd) tr %p %" PRIu64 "/%u region %p so %" PRIu64 + " bo %u sz %u BUF_OFFSET UPDATED", + sa, sa->offset, sa->len, region, region->stream_offset, region->buf_offset, + region->buf_size); + } + + /* new block (sa) extends in tree block (tr) + sa: [ ] + tr: [ ] + sa: [ ] + tr: [ ] + */ } else if (sa->offset > tr->offset && sa_re > tr_re && sa->offset <= tr_re) { // merge. sb->sbb_size includes both so we need to adjust that too. uint32_t combined_len = sa->len + tr->len; sa->len = sa_re - tr->offset; sa->offset = tr->offset; sa_re = sa->offset + sa->len; - SCLogDebug("-> (bwd) tr %p %"PRIu64"/%u REMOVED MERGED", tr, tr->offset, tr->len); + SCLogDebug("-> (bwd) tr %p %" PRIu64 "/%u REMOVED MERGED", tr, tr->offset, tr->len); if (sb->head == tr) sb->head = sa; SBB_RB_REMOVE(tree, tr); sb->sbb_size -= (combined_len - sa->len); // remove what we added twice FREE(sb->cfg, tr, sizeof(StreamingBufferBlock)); + + SCLogDebug("-> (bwd) tr %p %" PRIu64 "/%u region %p so %" PRIu64 " bo %u sz %u", sa, + sa->offset, sa->len, region, region->stream_offset, region->buf_offset, + region->buf_size); + if (sa->offset == region->stream_offset && + sa_re > (region->stream_offset + region->buf_offset)) { + BUG_ON(sa_re < region->stream_offset); + region->buf_offset = sa_re - region->stream_offset; + SCLogDebug("-> (bwd) tr %p %" PRIu64 "/%u region %p so %" PRIu64 + " bo %u sz %u BUF_OFFSET UPDATED", + sa, sa->offset, sa->len, region, region->stream_offset, region->buf_offset, + region->buf_size); + } } } } -static int Insert(StreamingBuffer *sb, struct SBB *tree, - uint32_t rel_offset, uint32_t len) +/** \internal + * \param region the region that holds the new data + */ +static int SBBUpdate( + StreamingBuffer *sb, StreamingBufferRegion *region, uint32_t rel_offset, uint32_t len) { + struct SBB *tree = &sb->sbb_tree; SCLogDebug("* inserting: %u/%u", rel_offset, len); StreamingBufferBlock *sbb = CALLOC(sb->cfg, 1, sizeof(*sbb)); if (sbb == NULL) return -1; - sbb->offset = sb->stream_offset + rel_offset; + sbb->offset = region->stream_offset + rel_offset; sbb->len = len; + StreamingBufferBlock *res = SBB_RB_INSERT(tree, sbb); if (res) { // exact overlap - SCLogDebug("* insert failed: exact match in tree with %p %"PRIu64"/%u", res, res->offset, res->len); + SCLogDebug("* insert failed: exact match in tree with %p %" PRIu64 "/%u", res, res->offset, + res->len); FREE(sb->cfg, sbb, sizeof(StreamingBufferBlock)); return 0; } sb->sbb_size += len; // may adjust based on consolidation below + + /* handle backwards and forwards overlaps */ if (SBB_RB_PREV(sbb) == NULL) { sb->head = sbb; } else { - ConsolidateBackward(sb, tree, sbb); + ConsolidateBackward(sb, region, tree, sbb); } - ConsolidateFwd(sb, tree, sbb); + ConsolidateFwd(sb, region, tree, sbb); #ifdef DEBUG SBBPrintList(sb); #endif + if (sbb->offset == sb->region.stream_offset) { + SCLogDebug("insert at head"); + sb->region.buf_offset = sbb->len; + } return 0; } -static void SBBUpdate(StreamingBuffer *sb, - uint32_t rel_offset, uint32_t data_len) -{ - Insert(sb, &sb->sbb_tree, rel_offset, data_len); -} - static void SBBFree(StreamingBuffer *sb) { StreamingBufferBlock *sbb = NULL, *safe = NULL; @@ -407,27 +512,41 @@ static void SBBFree(StreamingBuffer *sb) static void SBBPrune(StreamingBuffer *sb) { - SCLogDebug("pruning %p to %"PRIu64, sb, sb->stream_offset); + SCLogDebug("pruning %p to %" PRIu64, sb, sb->region.stream_offset); StreamingBufferBlock *sbb = NULL, *safe = NULL; RB_FOREACH_SAFE(sbb, SBB, &sb->sbb_tree, safe) { /* completely beyond window, we're done */ - if (sbb->offset >= sb->stream_offset) { + if (sbb->offset >= sb->region.stream_offset) { sb->head = sbb; + if (sbb->offset == sb->region.stream_offset) { + SCLogDebug("set buf_offset?"); + if (sbb->offset == sb->region.stream_offset) { + SCLogDebug("set buf_offset to first sbb len %u", sbb->len); + BUG_ON(sbb->len > sb->region.buf_size); + sb->region.buf_offset = sbb->len; + } + } break; } /* partly before, partly beyond. Adjust */ - if (sbb->offset < sb->stream_offset && - sbb->offset + sbb->len > sb->stream_offset) { - uint32_t shrink_by = sb->stream_offset - sbb->offset; + if (sbb->offset < sb->region.stream_offset && + sbb->offset + sbb->len > sb->region.stream_offset) { + uint32_t shrink_by = sb->region.stream_offset - sbb->offset; DEBUG_VALIDATE_BUG_ON(shrink_by > sbb->len); if (sbb->len >= shrink_by) { sbb->len -= shrink_by; sbb->offset += shrink_by; sb->sbb_size -= shrink_by; - DEBUG_VALIDATE_BUG_ON(sbb->offset != sb->stream_offset); + SCLogDebug("shrunk by %u", shrink_by); + DEBUG_VALIDATE_BUG_ON(sbb->offset != sb->region.stream_offset); } sb->head = sbb; + if (sbb->offset == sb->region.stream_offset) { + SCLogDebug("set buf_offset to first sbb len %u", sbb->len); + BUG_ON(sbb->len > sb->region.buf_size); + sb->region.buf_offset = sbb->len; + } break; } @@ -435,21 +554,24 @@ static void SBBPrune(StreamingBuffer *sb) /* either we set it again for the next sbb, or there isn't any */ sb->head = NULL; sb->sbb_size -= sbb->len; - SCLogDebug("sb %p removed %p %"PRIu64", %u", sb, sbb, sbb->offset, sbb->len); + SCLogDebug("sb %p removed %p %" PRIu64 ", %u", sb, sbb, sbb->offset, sbb->len); FREE(sb->cfg, sbb, sizeof(StreamingBufferBlock)); } +#ifdef DEBUG + SBBPrintList(sb); +#endif } static thread_local bool g2s_warn_once = false; -static int WARN_UNUSED -GrowToSize(StreamingBuffer *sb, uint32_t size) +static inline int WARN_UNUSED GrowRegionToSize( + StreamingBuffer *sb, StreamingBufferRegion *region, const uint32_t size) { - DEBUG_VALIDATE_BUG_ON(sb->buf_size > BIT_U32(30)); + DEBUG_VALIDATE_BUG_ON(region->buf_size > BIT_U32(30)); if (size > BIT_U32(30)) { // 1GiB if (!g2s_warn_once) { - SCLogWarning( - "StreamingBuffer::GrowToSize() tried to alloc %u bytes, exceeds limit of %lu", + SCLogWarning("StreamingBuffer::GrowRegionToSize() tried to alloc %u bytes, exceeds " + "limit of %lu", size, BIT_U32(30)); g2s_warn_once = true; } @@ -461,27 +583,32 @@ GrowToSize(StreamingBuffer *sb, uint32_t size) uint32_t base = size - x; uint32_t grow = base + sb->cfg->buf_size; - void *ptr = REALLOC(sb->cfg, sb->buf, sb->buf_size, grow); - if (ptr == NULL) + void *ptr = REALLOC(sb->cfg, region->buf, region->buf_size, grow); + if (ptr == NULL) { return -1; - + } /* for safe printing and general caution, lets memset the * new data to 0 */ - size_t diff = grow - sb->buf_size; - void *new_mem = ((char *)ptr) + sb->buf_size; + size_t diff = grow - region->buf_size; + void *new_mem = ((char *)ptr) + region->buf_size; memset(new_mem, 0, diff); - sb->buf = ptr; - sb->buf_size = grow; + region->buf = ptr; + region->buf_size = grow; SCLogDebug("grown buffer to %u", grow); #ifdef DEBUG - if (sb->buf_size > sb->buf_size_max) { - sb->buf_size_max = sb->buf_size; + if (region->buf_size > sb->buf_size_max) { + sb->buf_size_max = region->buf_size; } #endif return 0; } +static int WARN_UNUSED GrowToSize(StreamingBuffer *sb, uint32_t size) +{ + return GrowRegionToSize(sb, &sb->region, size); +} + static thread_local bool grow_warn_once = false; /** \internal @@ -491,8 +618,8 @@ static thread_local bool grow_warn_once = false; */ static int WARN_UNUSED Grow(StreamingBuffer *sb) { - DEBUG_VALIDATE_BUG_ON(sb->buf_size > BIT_U32(30)); - uint32_t grow = sb->buf_size * 2; + DEBUG_VALIDATE_BUG_ON(sb->region.buf_size > BIT_U32(30)); + uint32_t grow = sb->region.buf_size * 2; if (grow > BIT_U32(30)) { // 1GiB if (!grow_warn_once) { SCLogWarning("StreamingBuffer::Grow() tried to alloc %u bytes, exceeds limit of %lu", @@ -502,68 +629,263 @@ static int WARN_UNUSED Grow(StreamingBuffer *sb) return -1; } - void *ptr = REALLOC(sb->cfg, sb->buf, sb->buf_size, grow); + void *ptr = REALLOC(sb->cfg, sb->region.buf, sb->region.buf_size, grow); if (ptr == NULL) return -1; /* for safe printing and general caution, lets memset the * new data to 0 */ - size_t diff = grow - sb->buf_size; - void *new_mem = ((char *)ptr) + sb->buf_size; + size_t diff = grow - sb->region.buf_size; + void *new_mem = ((char *)ptr) + sb->region.buf_size; memset(new_mem, 0, diff); - sb->buf = ptr; - sb->buf_size = grow; + sb->region.buf = ptr; + sb->region.buf_size = grow; SCLogDebug("grown buffer to %u", grow); #ifdef DEBUG - if (sb->buf_size > sb->buf_size_max) { - sb->buf_size_max = sb->buf_size; + if (sb->region.buf_size > sb->buf_size_max) { + sb->buf_size_max = sb->region.buf_size; } #endif return 0; } +static inline bool RegionBeforeOffset(const StreamingBufferRegion *r, const uint64_t o) +{ + return (r->stream_offset + r->buf_size <= o); +} + +static inline bool RegionContainsOffset(const StreamingBufferRegion *r, const uint64_t o) +{ + return (o >= r->stream_offset && (r->stream_offset + r->buf_size) >= o); +} + +/** \internal + * \brief slide to offset for regions + * + * + * [ main ] [ gap ] [ aux ] [ gap ] [ aux ] + * ^ + * - main reset to 0 + * + * + * [ main ] [ gap ] [ aux ] [ gap ] [ aux ] + * ^ + * - main shift + * + * [ main ] [ gap ] [ aux ] [ gap ] [ aux ] + * ^ + * - main reset + * - move aux into main + * - free aux + * - shift + * + * [ main ] [ gap ] [ aux ] [ gap ] [ aux ] + * ^ + * - main reset + * - move aux into main + * - free aux + * - no shift + */ +static inline void StreamingBufferSlideToOffsetWithRegions( + StreamingBuffer *sb, const uint64_t slide_offset) +{ + ListRegions(sb); + BUG_ON(slide_offset == sb->region.stream_offset); + + SCLogDebug("slide_offset %" PRIu64, slide_offset); + SCLogDebug("main: offset %" PRIu64 " buf %p size %u offset %u", sb->region.stream_offset, + sb->region.buf, sb->region.buf_size, sb->region.buf_offset); + + StreamingBufferRegion *to_shift = NULL; + const bool main_is_oow = RegionBeforeOffset(&sb->region, slide_offset); + if (main_is_oow) { + SCLogDebug("main_is_oow"); + if (sb->region.buf != NULL) { + SCLogDebug("clearing main"); + FREE(sb->cfg, sb->region.buf, sb->region.buf_size); + sb->region.buf = NULL; + sb->region.buf_size = 0; + sb->region.buf_offset = 0; + sb->region.stream_offset = slide_offset; + } else { + sb->region.stream_offset = slide_offset; + } + + /* remove regions that are out of window & select the region to + * become the new main */ + StreamingBufferRegion *prev = &sb->region; + for (StreamingBufferRegion *r = sb->region.next; r != NULL;) { + SCLogDebug("r %p so %" PRIu64 ", re %" PRIu64, r, r->stream_offset, + r->stream_offset + r->buf_offset); + StreamingBufferRegion *next = r->next; + if (RegionBeforeOffset(r, slide_offset)) { + SCLogDebug("r %p so %" PRIu64 ", re %" PRIu64 " -> before", r, r->stream_offset, + r->stream_offset + r->buf_offset); + BUG_ON(r == &sb->region); + prev->next = next; + + FREE(sb->cfg, r->buf, r->buf_size); + FREE(sb->cfg, r, sizeof(*r)); + sb->regions--; + BUG_ON(sb->regions == 0); + } else if (RegionContainsOffset(r, slide_offset)) { + SCLogDebug("r %p so %" PRIu64 ", re %" PRIu64 " -> within", r, r->stream_offset, + r->stream_offset + r->buf_offset); + /* remove from list, we will xfer contents to main below */ + prev->next = next; + to_shift = r; + break; + } else { + SCLogDebug("r %p so %" PRIu64 ", re %" PRIu64 " -> post", r, r->stream_offset, + r->stream_offset + r->buf_offset); + /* implied beyond slide offset */ + BUG_ON(r->stream_offset < slide_offset); + break; + } + r = next; + } + SCLogDebug("to_shift %p", to_shift); + } else { + to_shift = &sb->region; + SCLogDebug("shift start region %p", to_shift); + } + + // this region is main, or will xfer its buffer to main + if (to_shift) { + SCLogDebug("main: offset %" PRIu64 " buf %p size %u offset %u", to_shift->stream_offset, + to_shift->buf, to_shift->buf_size, to_shift->buf_offset); + if (to_shift != &sb->region) { + BUG_ON(sb->region.buf != NULL); + + sb->region.buf = to_shift->buf; + sb->region.stream_offset = to_shift->stream_offset; + sb->region.buf_offset = to_shift->buf_offset; + sb->region.buf_size = to_shift->buf_size; + sb->region.next = to_shift->next; + + FREE(sb->cfg, to_shift, sizeof(*to_shift)); + to_shift = &sb->region; + sb->regions--; + BUG_ON(sb->regions == 0); + } + + // Do the shift. If new region is exactly at the slide offset we can skip this. + BUG_ON(to_shift->stream_offset > slide_offset); + const uint32_t s = slide_offset - to_shift->stream_offset; + if (s > 0) { + const uint32_t new_size = to_shift->buf_size - s; + SCLogDebug("s %u new_size %u", s, new_size); + memmove(to_shift->buf, to_shift->buf + s, new_size); + void *ptr = REALLOC(sb->cfg, to_shift->buf, to_shift->buf_size, new_size); + BUG_ON(ptr == NULL); // TODO + to_shift->buf = ptr; + to_shift->buf_size = new_size; + if (s < to_shift->buf_offset) + to_shift->buf_offset -= s; + else + to_shift->buf_offset = 0; + to_shift->stream_offset = slide_offset; + } + } + + SCLogDebug("main: offset %" PRIu64 " buf %p size %u offset %u", sb->region.stream_offset, + sb->region.buf, sb->region.buf_size, sb->region.buf_offset); + SCLogDebug("end of slide"); +} + /** * \brief slide to absolute offset * \todo if sliding beyond window, we could perhaps reset? */ void StreamingBufferSlideToOffset(StreamingBuffer *sb, uint64_t offset) { - if (offset > sb->stream_offset && - offset <= sb->stream_offset + sb->buf_offset) - { - uint32_t slide = offset - sb->stream_offset; - uint32_t size = sb->buf_offset - slide; - SCLogDebug("sliding %u forward, size of original buffer left after slide %u", slide, size); - memmove(sb->buf, sb->buf+slide, size); - sb->stream_offset += slide; - sb->buf_offset = size; + SCLogDebug("sliding to offset %" PRIu64, offset); + ListRegions(sb); +#ifdef DEBUG + SBBPrintList(sb); +#endif + + if (sb->region.next) { + StreamingBufferSlideToOffsetWithRegions(sb, offset); + SBBPrune(sb); + SCLogDebug("post SBBPrune"); + ListRegions(sb); +#ifdef DEBUG + SBBPrintList(sb); +#endif + BUG_ON(sb->region.buf != NULL && sb->region.buf_size == 0); + BUG_ON(sb->region.buf_offset > sb->region.buf_size); + BUG_ON(offset > sb->region.stream_offset); + BUG_ON(sb->head && sb->head->offset == sb->region.stream_offset && + sb->head->len > sb->region.buf_offset); + BUG_ON(sb->region.stream_offset < offset); + return; + } + + if (offset > sb->region.stream_offset) { + const uint32_t slide = offset - sb->region.stream_offset; + if (sb->head != NULL) { + /* have sbb's, so can't rely on buf_offset for the slide */ + if (slide < sb->region.buf_size) { + const uint32_t size = sb->region.buf_size - slide; + SCLogDebug("sliding %u forward, size of original buffer left after slide %u", slide, + size); + memmove(sb->region.buf, sb->region.buf + slide, size); + if (sb->region.buf_offset > slide) { + sb->region.buf_offset -= slide; + } else { + sb->region.buf_offset = 0; + } + } else { + sb->region.buf_offset = 0; + } + sb->region.stream_offset = offset; + } else { + /* no sbb's, so we can use buf_offset */ + if (offset <= sb->region.stream_offset + sb->region.buf_offset) { + const uint32_t size = sb->region.buf_offset - slide; + SCLogDebug("sliding %u forward, size of original buffer left after slide %u", slide, + size); + memmove(sb->region.buf, sb->region.buf + slide, size); + sb->region.stream_offset = offset; + sb->region.buf_offset = size; + } else { + /* moved past all data */ + sb->region.stream_offset = offset; + sb->region.buf_offset = 0; + } + } SBBPrune(sb); } +#ifdef DEBUG + SBBPrintList(sb); +#endif + BUG_ON(sb->region.stream_offset < offset); } void StreamingBufferSlide(StreamingBuffer *sb, uint32_t slide) { - uint32_t size = sb->buf_offset - slide; + SCLogDebug("slide with %" PRIu32, slide); + uint32_t size = sb->region.buf_offset - slide; SCLogDebug("sliding %u forward, size of original buffer left after slide %u", slide, size); - memmove(sb->buf, sb->buf+slide, size); - sb->stream_offset += slide; - sb->buf_offset = size; + memmove(sb->region.buf, sb->region.buf + slide, size); + sb->region.stream_offset += slide; + sb->region.buf_offset = size; SBBPrune(sb); } -#define DATA_FITS(sb, len) \ - ((sb)->buf_offset + (len) <= (sb)->buf_size) +#define DATA_FITS(sb, len) ((sb)->region.buf_offset + (len) <= (sb)->region.buf_size) StreamingBufferSegment *StreamingBufferAppendRaw(StreamingBuffer *sb, const uint8_t *data, uint32_t data_len) { - if (sb->buf == NULL) { + if (sb->region.buf == NULL) { if (InitBuffer(sb) == -1) return NULL; } if (!DATA_FITS(sb, data_len)) { - if (sb->buf_size == 0) { + if (sb->region.buf_size == 0) { if (GrowToSize(sb, data_len) != 0) return NULL; } else { @@ -578,14 +900,14 @@ StreamingBufferSegment *StreamingBufferAppendRaw(StreamingBuffer *sb, const uint StreamingBufferSegment *seg = CALLOC(sb->cfg, 1, sizeof(StreamingBufferSegment)); if (seg != NULL) { - memcpy(sb->buf + sb->buf_offset, data, data_len); - seg->stream_offset = sb->stream_offset + sb->buf_offset; + memcpy(sb->region.buf + sb->region.buf_offset, data, data_len); + seg->stream_offset = sb->region.stream_offset + sb->region.buf_offset; seg->segment_len = data_len; - uint32_t rel_offset = sb->buf_offset; - sb->buf_offset += data_len; + uint32_t rel_offset = sb->region.buf_offset; + sb->region.buf_offset += data_len; if (!RB_EMPTY(&sb->sbb_tree)) { - SBBUpdate(sb, rel_offset, data_len); + SBBUpdate(sb, &sb->region, rel_offset, data_len); } return seg; } @@ -597,13 +919,13 @@ int StreamingBufferAppend(StreamingBuffer *sb, StreamingBufferSegment *seg, { BUG_ON(seg == NULL); - if (sb->buf == NULL) { + if (sb->region.buf == NULL) { if (InitBuffer(sb) == -1) return -1; } if (!DATA_FITS(sb, data_len)) { - if (sb->buf_size == 0) { + if (sb->region.buf_size == 0) { if (GrowToSize(sb, data_len) != 0) return -1; } else { @@ -616,14 +938,14 @@ int StreamingBufferAppend(StreamingBuffer *sb, StreamingBufferSegment *seg, } DEBUG_VALIDATE_BUG_ON(!DATA_FITS(sb, data_len)); - memcpy(sb->buf + sb->buf_offset, data, data_len); - seg->stream_offset = sb->stream_offset + sb->buf_offset; + memcpy(sb->region.buf + sb->region.buf_offset, data, data_len); + seg->stream_offset = sb->region.stream_offset + sb->region.buf_offset; seg->segment_len = data_len; - uint32_t rel_offset = sb->buf_offset; - sb->buf_offset += data_len; + uint32_t rel_offset = sb->region.buf_offset; + sb->region.buf_offset += data_len; if (!RB_EMPTY(&sb->sbb_tree)) { - SBBUpdate(sb, rel_offset, data_len); + SBBUpdate(sb, &sb->region, rel_offset, data_len); } return 0; } @@ -634,13 +956,13 @@ int StreamingBufferAppend(StreamingBuffer *sb, StreamingBufferSegment *seg, int StreamingBufferAppendNoTrack(StreamingBuffer *sb, const uint8_t *data, uint32_t data_len) { - if (sb->buf == NULL) { + if (sb->region.buf == NULL) { if (InitBuffer(sb) == -1) return -1; } if (!DATA_FITS(sb, data_len)) { - if (sb->buf_size == 0) { + if (sb->region.buf_size == 0) { if (GrowToSize(sb, data_len) != 0) return -1; } else { @@ -653,18 +975,380 @@ int StreamingBufferAppendNoTrack(StreamingBuffer *sb, } DEBUG_VALIDATE_BUG_ON(!DATA_FITS(sb, data_len)); - memcpy(sb->buf + sb->buf_offset, data, data_len); - uint32_t rel_offset = sb->buf_offset; - sb->buf_offset += data_len; + memcpy(sb->region.buf + sb->region.buf_offset, data, data_len); + uint32_t rel_offset = sb->region.buf_offset; + sb->region.buf_offset += data_len; if (!RB_EMPTY(&sb->sbb_tree)) { - SBBUpdate(sb, rel_offset, data_len); + SBBUpdate(sb, &sb->region, rel_offset, data_len); } return 0; } -#define DATA_FITS_AT_OFFSET(sb, len, offset) \ - ((offset) + (len) <= (sb)->buf_size) +#define DATA_FITS_AT_OFFSET(region, len, offset) ((offset) + (len) <= (region)->buf_size) + +#if defined(DEBUG) || defined(DEBUG_VALIDATION) +static void Validate(const StreamingBuffer *sb) +{ + bool bail = false; + uint32_t cnt = 0; + for (const StreamingBufferRegion *r = &sb->region; r != NULL; r = r->next) { + cnt++; + if (r->next) { + bail |= ((r->stream_offset + r->buf_size) > r->next->stream_offset); + } + + bail |= (r->buf != NULL && r->buf_size == 0); + bail |= (r->buf_offset > r->buf_size); + } + // leading gap, so buf_offset should be 0? + if (sb->head && sb->head->offset > sb->region.stream_offset) { + SCLogDebug("leading gap of size %" PRIu64, sb->head->offset - sb->region.stream_offset); + BUG_ON(sb->region.buf_offset != 0); + } + + if (sb->head && sb->head->offset == sb->region.stream_offset) { + BUG_ON(sb->head->len > sb->region.buf_offset); + BUG_ON(sb->head->len < sb->region.buf_offset); + } + BUG_ON(sb->regions != cnt); + BUG_ON(bail); +} +#endif + +static void ListRegions(StreamingBuffer *sb) +{ + if (sb->cfg == NULL) + return; +#ifdef DEBUG +#if DUMP_REGIONS == 1 + uint32_t cnt = 0; + for (StreamingBufferRegion *r = &sb->region; r != NULL; r = r->next) { + cnt++; + char gap[64] = ""; + if (r->next) { + snprintf(gap, sizeof(gap), "[ gap:%" PRIu64 " ]", + r->next->stream_offset - (r->stream_offset + r->buf_size)); + } + + printf("[ %s offset:%" PRIu64 " size:%u offset:%u ]%s", r == &sb->region ? "main" : "aux", + r->stream_offset, r->buf_size, r->buf_offset, gap); + } + printf("(max %u, cnt %u, sb->regions %u)\n", sb->max_regions, cnt, sb->regions); + bool at_least_one = false; + uint64_t last_re = sb->region.stream_offset; + StreamingBufferBlock *sbb = NULL; + RB_FOREACH(sbb, SBB, &sb->sbb_tree) + { + if (last_re != sbb->offset) { + printf("[ gap:%" PRIu64 " ]", sbb->offset - last_re); + } + printf("[ sbb offset:%" PRIu64 " len:%u ]", sbb->offset, sbb->len); + at_least_one = true; + last_re = sbb->offset + sbb->len; + } + if (at_least_one) + printf("\n"); +#endif +#endif +#if defined(DEBUG) || defined(DEBUG_VALIDATION) + Validate(sb); +#endif +} + +/** \interal + * \brief does data region intersect with list region 'r' + * Takes the max gap into account. + */ +static inline bool RegionsIntersect(const StreamingBuffer *sb, const StreamingBufferRegion *r, + const uint64_t offset, const uint32_t len) +{ + const uint64_t re = offset + len; + + /* create the data range for the region, adding the max gap */ + const uint64_t reg_o = + r->stream_offset > REGION_MAX_GAP ? (r->stream_offset - REGION_MAX_GAP) : 0; + const uint64_t reg_re = r->stream_offset + r->buf_size + REGION_MAX_GAP; + SCLogDebug("r %p: %" PRIu64 "/%" PRIu64 " - adjusted %" PRIu64 "/%" PRIu64, r, r->stream_offset, + r->stream_offset + r->buf_size, reg_o, reg_re); + /* check if data range intersects with region range */ + if (offset >= reg_o && offset <= reg_re) { + SCLogDebug("r %p is in-scope", r); + return true; + } + if (re >= reg_o && re <= reg_re) { + SCLogDebug("r %p is in-scope: %" PRIu64 " >= %" PRIu64 " && %" PRIu64 " <= %" PRIu64, r, re, + reg_o, re, reg_re); + return true; + } + SCLogDebug("r %p is out of scope: %" PRIu64 "/%" PRIu64, r, offset, re); + return false; +} + +/** \internal + * \brief find the first region for merging. + */ +static StreamingBufferRegion *FindFirstRegionForOffset(const StreamingBuffer *sb, + StreamingBufferRegion *r, const uint64_t offset, const uint32_t len) +{ + const uint64_t data_re = offset + len; + SCLogDebug("looking for first region matching %" PRIu64 "/%" PRIu64, offset, data_re); + + for (; r != NULL; r = r->next) { + if (RegionsIntersect(sb, r, offset, data_re) == true) + return r; + } + return NULL; +} + +static StreamingBufferRegion *FindLargestRegionForOffset(const StreamingBuffer *sb, + StreamingBufferRegion *r, const uint64_t offset, const uint32_t len) +{ + const uint64_t data_re = offset + len; + SCLogDebug("starting at %p/%" PRIu64 ", offset %" PRIu64 ", data_re %" PRIu64, r, + r->stream_offset, offset, data_re); + StreamingBufferRegion *candidate = r; + for (; r != NULL; r = r->next) { +#ifdef DEBUG + const uint64_t reg_re = r->stream_offset + r->buf_size; + SCLogDebug("checking: %p/%" PRIu64 "/%" PRIu64 ", offset %" PRIu64 "/%" PRIu64, r, + r->stream_offset, reg_re, offset, data_re); +#endif + if (!RegionsIntersect(sb, r, offset, data_re)) + return candidate; + + if (candidate == NULL) { + candidate = r; + SCLogDebug("candidate %p", candidate); + } else if (r->buf_size > candidate->buf_size) { + SCLogDebug("candidate %p as size %u > %u", candidate, r->buf_size, candidate->buf_size); + candidate = r; + } + } + return candidate; +} + +static StreamingBufferRegion *FindRightEdge(const StreamingBuffer *sb, StreamingBufferRegion *r, + const uint64_t offset, const uint32_t len) +{ + const uint64_t data_re = offset + len; + StreamingBufferRegion *candidate = r; + for (; r != NULL; r = r->next) { + if (!RegionsIntersect(sb, r, offset, data_re)) { + SCLogDebug("r %p is out of scope: %" PRIu64 "/%u", r, offset, len); + return candidate; + } + candidate = r; + } + return candidate; +} + +/** \internal + * \brief process insert by consolidating the affected regions into one + */ +static StreamingBufferRegion *BufferInsertAtRegionConsolidate(StreamingBuffer *sb, + StreamingBufferRegion *dst, StreamingBufferRegion *src_start, + StreamingBufferRegion *src_end, const uint64_t data_offset, const uint32_t data_len) +{ + const uint64_t data_re = data_offset + data_len; + SCLogDebug("sb %p dst %p src_start %p src_end %p data_offset %" PRIu64 + "/data_len %u/data_re %" PRIu64, + sb, dst, src_start, src_end, data_offset, data_len, data_re); + + // 1. determine size for dst. + const uint64_t dst_offset = MIN(src_start->stream_offset, data_offset); + DEBUG_VALIDATE_BUG_ON(dst_offset < sb->region.stream_offset); + const uint64_t dst_re = MAX((src_end->stream_offset + src_end->buf_size), data_re); + const uint32_t dst_size = dst_re - dst_offset; + SCLogDebug("dst_offset %" PRIu64 ", dst_re %" PRIu64 ", dst_size %u", dst_offset, dst_re, + dst_size); + + // 2. resize dst + const uint32_t old_size = dst->buf_size; + const uint32_t dst_copy_offset = dst->stream_offset - dst_offset; +#ifdef DEBUG + const uint32_t old_offset = dst->buf_offset; + SCLogDebug("old_size %u, old_offset %u, dst_copy_offset %u", old_size, old_offset, + dst_copy_offset); +#endif + if (GrowRegionToSize(sb, dst, dst_size) != 0) + return NULL; + SCLogDebug("resized to %u", dst_size); + if (dst_copy_offset != 0) + memmove(dst->buf + dst_copy_offset, dst->buf, old_size); + dst->stream_offset = dst_offset; + + uint32_t new_offset = src_start->buf_offset; + if (data_offset == src_start->stream_offset + src_start->buf_offset) { + new_offset += data_len; + } + + bool start_is_main = false; + StreamingBufferRegion *prev = NULL; + if (src_start == &sb->region) { + DEBUG_VALIDATE_BUG_ON(src_start->stream_offset != dst_offset); + + start_is_main = true; + SCLogDebug("src_start is main region"); + if (src_start != dst) + memcpy(dst->buf, src_start->buf, src_start->buf_offset); + if (src_start == src_end) { + SCLogDebug("src_start == src_end == main, we're done"); + BUG_ON(src_start != dst); + return src_start; + } + prev = src_start; + src_start = src_start->next; // skip in the loop below + } + + // 3. copy all regions from src_start to dst_start into the new region + for (StreamingBufferRegion *r = src_start; r != NULL;) { + SCLogDebug("r %p %" PRIu64 ", offset %u, len %u, %s, last %s", r, r->stream_offset, + r->buf_offset, r->buf_size, r == &sb->region ? "main" : "aux", + BOOL2STR(r == src_end)); + // skip dst + if (r == dst) { + SCLogDebug("skipping r %p as it is 'dst'", r); + if (r == src_end) + break; + prev = r; + r = r->next; + continue; + } + const uint32_t target_offset = r->stream_offset - dst_offset; + SCLogDebug("r %p: target_offset %u", r, target_offset); + memcpy(dst->buf + target_offset, r->buf, r->buf_size); + + StreamingBufferRegion *next = r->next; + FREE(sb->cfg, r->buf, r->buf_size); + FREE(sb->cfg, r, sizeof(*r)); + sb->regions--; + BUG_ON(sb->regions == 0); + if (prev != NULL) { + SCLogDebug("setting prev %p next to %p (was %p)", prev, next, prev->next); + prev->next = next; + } else { + SCLogDebug("no prev yet"); + } + + if (r == src_end) + break; + + prev = r; + r = next; + } + + /* special handling of main region being the start, but not the + * region we expand. In this case we'll have main and dst. We will + * move the buffer from dst into main and free dst. */ + if (start_is_main && dst != &sb->region) { + BUG_ON(sb->region.next != dst); + SCLogDebug("start_is_main && dst != main region"); + FREE(sb->cfg, sb->region.buf, sb->region.buf_size); + sb->region.buf = dst->buf; + sb->region.buf_size = dst->buf_size; + sb->region.buf_offset = new_offset; + SCLogDebug("sb->region.buf_offset set to %u", sb->region.buf_offset); + sb->region.next = dst->next; + FREE(sb->cfg, dst, sizeof(*dst)); + dst = &sb->region; + sb->regions--; + BUG_ON(sb->regions == 0); + } else { + SCLogDebug("dst: %p next %p", dst, dst->next); + } + + SCLogDebug("returning dst %p stream_offset %" PRIu64 " buf_offset %u buf_size %u", dst, + dst->stream_offset, dst->buf_offset, dst->buf_size); + return dst; +} + +static StreamingBufferRegion *BufferInsertAtRegionDo( + StreamingBuffer *sb, const uint64_t offset, const uint32_t len) +{ + SCLogDebug("offset %" PRIu64 ", len %u", offset, len); + StreamingBufferRegion *start = FindFirstRegionForOffset(sb, &sb->region, offset, len); + if (start) { + SCLogDebug("start region %p/%" PRIu64 "/%u", start, start->stream_offset, start->buf_size); + StreamingBufferRegion *big = FindLargestRegionForOffset(sb, start, offset, len); + DEBUG_VALIDATE_BUG_ON(big == NULL); + if (big == NULL) + return NULL; + SCLogDebug("big region %p/%" PRIu64 "/%u", big, big->stream_offset, big->buf_size); + StreamingBufferRegion *end = FindRightEdge(sb, big, offset, len); + DEBUG_VALIDATE_BUG_ON(end == NULL); + if (end == NULL) + return NULL; + SCLogDebug("end region %p/%" PRIu64 "/%u", end, end->stream_offset, end->buf_size); + StreamingBufferRegion *ret = + BufferInsertAtRegionConsolidate(sb, big, start, end, offset, len); + return ret; + } else { + /* if there was no region we can use we add a new region and insert it */ + StreamingBufferRegion *append = &sb->region; + for (StreamingBufferRegion *r = append; r != NULL; r = r->next) { + if (r->stream_offset > offset) { + break; + } else { + append = r; + } + } + + SCLogDebug("no matching region found, append to %p (%s)", append, + append == &sb->region ? "main" : "aux"); + StreamingBufferRegion *add = InitBufferRegion(sb, len); + if (add == NULL) + return NULL; + add->stream_offset = offset; + add->next = append->next; + append->next = add; + SCLogDebug("new region %p offset %" PRIu64, add, add->stream_offset); + return add; + } +} + +/** \internal + * \brief return the region to put the new data in + * + * Will find an existing region, expand it if needed. If no existing region exists or is + * a good fit, it will try to set up a new region. If the region then overlaps or gets + * too close to the next, merge them. + */ +static StreamingBufferRegion *BufferInsertAtRegion(StreamingBuffer *sb, const uint8_t *data, + const uint32_t data_len, const uint64_t data_offset) +{ + SCLogDebug("data_offset %" PRIu64 ", data_len %u, re %" PRIu64, data_offset, data_len, + data_offset + data_len); + ListRegions(sb); + + if (RegionsIntersect(sb, &sb->region, data_offset, data_len)) { + SCLogDebug("data_offset %" PRIu64 ", data_len %u intersects with main region (next %p)", + data_offset, data_len, sb->region.next); + if (sb->region.next == NULL || + !RegionsIntersect(sb, sb->region.next, data_offset, data_len)) { + SCLogDebug( + "data_offset %" PRIu64 + ", data_len %u intersects with main region, no next or way before next region", + data_offset, data_len); + if (sb->region.buf == NULL) + if (InitBuffer(sb) == -1) // TODO init with size + return NULL; + return &sb->region; + } + } else if (sb->region.next == NULL) { + StreamingBufferRegion *aux_r = sb->region.next = InitBufferRegion(sb, data_len); + if (aux_r == NULL) + return NULL; + aux_r->stream_offset = data_offset; + DEBUG_VALIDATE_BUG_ON(data_len > aux_r->buf_size); + SCLogDebug("created new region %p with offset %" PRIu64 ", size %u", aux_r, + aux_r->stream_offset, aux_r->buf_size); + return aux_r; + } + StreamingBufferRegion *blob = BufferInsertAtRegionDo(sb, data_offset, data_len); + SCLogDebug("blob %p (%s)", blob, blob == &sb->region ? "main" : "aux"); + return blob; +} /** * \param offset offset relative to StreamingBuffer::stream_offset @@ -678,59 +1362,106 @@ int StreamingBufferInsertAt(StreamingBuffer *sb, StreamingBufferSegment *seg, uint64_t offset) { BUG_ON(seg == NULL); - DEBUG_VALIDATE_BUG_ON(offset < sb->stream_offset); - if (offset < sb->stream_offset) + DEBUG_VALIDATE_BUG_ON(offset < sb->region.stream_offset); + if (offset < sb->region.stream_offset) return -2; - if (sb->buf == NULL) { - if (InitBuffer(sb) == -1) - return -1; + StreamingBufferRegion *region = BufferInsertAtRegion(sb, data, data_len, offset); + if (region == NULL) { + return -1; } - uint32_t rel_offset = offset - sb->stream_offset; - if (!DATA_FITS_AT_OFFSET(sb, data_len, rel_offset)) { + const bool region_is_main = region == &sb->region; + + SCLogDebug("inserting %" PRIu64 "/%u using %s region %p", offset, data_len, + region == &sb->region ? "main" : "aux", region); + + uint32_t rel_offset = offset - region->stream_offset; + if (!DATA_FITS_AT_OFFSET(region, data_len, rel_offset)) { if (GrowToSize(sb, (rel_offset + data_len)) != 0) return -1; } - DEBUG_VALIDATE_BUG_ON(!DATA_FITS_AT_OFFSET(sb, data_len, rel_offset)); + DEBUG_VALIDATE_BUG_ON(!DATA_FITS_AT_OFFSET(region, data_len, rel_offset)); - memcpy(sb->buf + rel_offset, data, data_len); + SCLogDebug("offset %" PRIu64 " data_len %u, rel_offset %u into region offset %" PRIu64 + ", buf_offset %u, buf_size %u", + offset, data_len, rel_offset, region->stream_offset, region->buf_offset, + region->buf_size); + memcpy(region->buf + rel_offset, data, data_len); seg->stream_offset = offset; seg->segment_len = data_len; - SCLogDebug("rel_offset %u sb->stream_offset %"PRIu64", buf_offset %u", - rel_offset, sb->stream_offset, sb->buf_offset); + SCLogDebug("rel_offset %u region->stream_offset %" PRIu64 ", buf_offset %u", rel_offset, + region->stream_offset, sb->region.buf_offset); if (RB_EMPTY(&sb->sbb_tree)) { SCLogDebug("empty sbb list"); - if (sb->stream_offset == offset) { - SCLogDebug("empty sbb list: block exactly what was expected, fall through"); - /* empty list, data is exactly what is expected (append), - * so do nothing */ - } else if ((rel_offset + data_len) <= sb->buf_offset) { - SCLogDebug("empty sbb list: block is within existing region"); + if (region_is_main) { + if (sb->region.stream_offset == offset) { + SCLogDebug("empty sbb list: block exactly what was expected, fall through"); + /* empty list, data is exactly what is expected (append), + * so do nothing. + * Update buf_offset if needed, but in case of overlaps it might be beyond us. */ + sb->region.buf_offset = MAX(sb->region.buf_offset, rel_offset + data_len); + } else if ((rel_offset + data_len) <= sb->region.buf_offset) { + SCLogDebug("empty sbb list: block is within existing main data region"); + } else { + if (sb->region.buf_offset && rel_offset == sb->region.buf_offset) { + SCLogDebug("exactly at expected offset"); + // nothing to do + sb->region.buf_offset = rel_offset + data_len; + + } else if (rel_offset < sb->region.buf_offset) { + // nothing to do + + SCLogDebug("before expected offset: %u < sb->region.buf_offset %u", rel_offset, + sb->region.buf_offset); + if (rel_offset + data_len > sb->region.buf_offset) { + SCLogDebug("before expected offset, ends after: %u < sb->region.buf_offset " + "%u, %u > %u", + rel_offset, sb->region.buf_offset, rel_offset + data_len, + sb->region.buf_offset); + sb->region.buf_offset = rel_offset + data_len; + } + + } else if (sb->region.buf_offset) { + SCLogDebug("beyond expected offset: SBBInit"); + /* existing data, but there is a gap between us */ + SBBInit(sb, region, rel_offset, data_len); + } else { + /* gap before data in empty list */ + SCLogDebug("empty sbb list: invoking SBBInitLeadingGap"); + SBBInitLeadingGap(sb, region, offset, data_len); + } + } } else { - if (sb->buf_offset && rel_offset == sb->buf_offset) { - // nothing to do - } else if (rel_offset < sb->buf_offset) { - // nothing to do - } else if (sb->buf_offset) { + if (sb->region.buf_offset) { /* existing data, but there is a gap between us */ - SBBInit(sb, rel_offset, data_len); + SCLogDebug("empty sbb list, no data in main: use SBBInit"); + SBBInit(sb, region, rel_offset, data_len); } else { /* gap before data in empty list */ SCLogDebug("empty sbb list: invoking SBBInitLeadingGap"); - SBBInitLeadingGap(sb, offset, data_len); + SBBInitLeadingGap(sb, region, offset, data_len); + } + if (rel_offset == region->buf_offset) { + SCLogDebug("pre region->buf_offset %u", region->buf_offset); + region->buf_offset = rel_offset + data_len; + SCLogDebug("post region->buf_offset %u", region->buf_offset); } } } else { + SCLogDebug("updating sbb tree"); /* already have blocks, so append new block based on new data */ - SBBUpdate(sb, rel_offset, data_len); + SBBUpdate(sb, region, rel_offset, data_len); } + BUG_ON(!region_is_main && sb->head == NULL); - if (rel_offset + data_len > sb->buf_offset) - sb->buf_offset = rel_offset + data_len; + ListRegions(sb); + if (RB_EMPTY(&sb->sbb_tree)) { + BUG_ON(offset + data_len > sb->region.stream_offset + sb->region.buf_offset); + } return 0; } @@ -738,33 +1469,61 @@ int StreamingBufferInsertAt(StreamingBuffer *sb, StreamingBufferSegment *seg, int StreamingBufferSegmentIsBeforeWindow(const StreamingBuffer *sb, const StreamingBufferSegment *seg) { - if (seg->stream_offset < sb->stream_offset) { - if (seg->stream_offset + seg->segment_len <= sb->stream_offset) { + if (seg->stream_offset < sb->region.stream_offset) { + if (seg->stream_offset + seg->segment_len <= sb->region.stream_offset) { return 1; } } return 0; } +static inline const StreamingBufferRegion *GetRegionForOffset( + const StreamingBuffer *sb, const uint64_t offset) +{ + if (sb == NULL) + return NULL; + if (sb->region.next == NULL) { + return &sb->region; + } + if (offset >= sb->region.stream_offset && + offset < (sb->region.stream_offset + sb->region.buf_size)) { + return &sb->region; + } + for (const StreamingBufferRegion *r = sb->region.next; r != NULL; r = r->next) { + if (offset >= r->stream_offset && offset < (r->stream_offset + r->buf_size)) { + return r; + } + } + return NULL; +} + /** \brief get the data for one SBB */ void StreamingBufferSBBGetData(const StreamingBuffer *sb, const StreamingBufferBlock *sbb, const uint8_t **data, uint32_t *data_len) { - if (sbb->offset >= sb->stream_offset) { - uint64_t offset = sbb->offset - sb->stream_offset; - *data = sb->buf + offset; - if (offset + sbb->len > sb->buf_offset) - *data_len = sb->buf_offset - offset; - else + ListRegions((StreamingBuffer *)sb); + const StreamingBufferRegion *region = GetRegionForOffset(sb, sbb->offset); + SCLogDebug("first find our region (offset %" PRIu64 ") -> %p", sbb->offset, region); + if (region) { + SCLogDebug("region %p found %" PRIu64 "/%u/%u", region, region->stream_offset, + region->buf_size, region->buf_offset); + if (sbb->offset >= region->stream_offset) { + SCLogDebug("1"); + uint64_t offset = sbb->offset - region->stream_offset; + *data = region->buf + offset; + BUG_ON(offset + sbb->len > region->buf_size); *data_len = sbb->len; - return; - } else { - uint64_t offset = sb->stream_offset - sbb->offset; - if (offset < sbb->len) { - *data = sb->buf; - *data_len = sbb->len - offset; return; + } else { + SCLogDebug("2"); + uint64_t offset = region->stream_offset - sbb->offset; + if (offset < sbb->len) { + *data = region->buf; + *data_len = sbb->len - offset; + return; + } + SCLogDebug("3"); } } *data = NULL; @@ -778,22 +1537,31 @@ void StreamingBufferSBBGetDataAtOffset(const StreamingBuffer *sb, const uint8_t **data, uint32_t *data_len, uint64_t offset) { - if (offset >= sbb->offset && offset < (sbb->offset + sbb->len)) { + /* validate that we are looking for a offset within the sbb */ + DEBUG_VALIDATE_BUG_ON(!(offset >= sbb->offset && offset < (sbb->offset + sbb->len))); + if (!(offset >= sbb->offset && offset < (sbb->offset + sbb->len))) { + *data = NULL; + *data_len = 0; + return; + } + + const StreamingBufferRegion *region = GetRegionForOffset(sb, offset); + if (region) { uint32_t sbblen = sbb->len - (offset - sbb->offset); - if (offset >= sb->stream_offset) { - uint64_t data_offset = offset - sb->stream_offset; - *data = sb->buf + data_offset; - if (data_offset + sbblen > sb->buf_size) - *data_len = sb->buf_size - data_offset; + if (offset >= region->stream_offset) { + uint64_t data_offset = offset - region->stream_offset; + *data = region->buf + data_offset; + if (data_offset + sbblen > region->buf_size) + *data_len = region->buf_size - data_offset; else *data_len = sbblen; BUG_ON(*data_len > sbblen); return; } else { - uint64_t data_offset = sb->stream_offset - sbb->offset; + uint64_t data_offset = region->stream_offset - sbb->offset; if (data_offset < sbblen) { - *data = sb->buf; + *data = region->buf; *data_len = sbblen - data_offset; BUG_ON(*data_len > sbblen); return; @@ -810,20 +1578,23 @@ void StreamingBufferSegmentGetData(const StreamingBuffer *sb, const StreamingBufferSegment *seg, const uint8_t **data, uint32_t *data_len) { - if (likely(sb->buf)) { - if (seg->stream_offset >= sb->stream_offset) { - uint64_t offset = seg->stream_offset - sb->stream_offset; - *data = sb->buf + offset; - if (offset + seg->segment_len > sb->buf_size) - *data_len = sb->buf_size - offset; + const StreamingBufferRegion *region = GetRegionForOffset(sb, seg->stream_offset); + if (region) { + if (seg->stream_offset >= region->stream_offset) { + uint64_t offset = seg->stream_offset - region->stream_offset; + *data = region->buf + offset; + if (offset + seg->segment_len > region->buf_size) + *data_len = region->buf_size - offset; else *data_len = seg->segment_len; + SCLogDebug("*data_len %u", *data_len); return; } else { - uint64_t offset = sb->stream_offset - seg->stream_offset; + uint64_t offset = region->stream_offset - seg->stream_offset; if (offset < seg->segment_len) { - *data = sb->buf; + *data = region->buf; *data_len = seg->segment_len - offset; + SCLogDebug("*data_len %u", *data_len); return; } } @@ -857,10 +1628,10 @@ int StreamingBufferGetData(const StreamingBuffer *sb, const uint8_t **data, uint32_t *data_len, uint64_t *stream_offset) { - if (sb != NULL && sb->buf != NULL) { - *data = sb->buf; - *data_len = sb->buf_offset; - *stream_offset = sb->stream_offset; + if (sb != NULL && sb->region.buf != NULL) { + *data = sb->region.buf; + *data_len = sb->region.buf_offset; + *stream_offset = sb->region.stream_offset; return 1; } else { *data = NULL; @@ -874,13 +1645,12 @@ int StreamingBufferGetDataAtOffset (const StreamingBuffer *sb, const uint8_t **data, uint32_t *data_len, uint64_t offset) { - if (sb != NULL && sb->buf != NULL && - offset >= sb->stream_offset && - offset < (sb->stream_offset + sb->buf_offset)) - { - uint32_t skip = offset - sb->stream_offset; - *data = sb->buf + skip; - *data_len = sb->buf_offset - skip; + const StreamingBufferRegion *region = GetRegionForOffset(sb, offset); + if (region != NULL && region->buf != NULL && offset >= region->stream_offset && + offset < (region->stream_offset + region->buf_offset)) { + uint32_t skip = offset - region->stream_offset; + *data = region->buf + skip; + *data_len = region->buf_offset - skip; return 1; } else { *data = NULL; @@ -907,7 +1677,7 @@ int StreamingBufferCompareRawData(const StreamingBuffer *sb, { return 1; } - SCLogDebug("sbdata_len %u, offset %"PRIu64, sbdata_len, offset); + SCLogDebug("sbdata_len %u, offset %" PRIu64, sbdata_len, offset); printf("got:\n"); PrintRawDataFp(stdout, sbdata,sbdata_len); printf("wanted:\n"); @@ -918,7 +1688,7 @@ int StreamingBufferCompareRawData(const StreamingBuffer *sb, #ifdef UNITTESTS static void Dump(StreamingBuffer *sb) { - PrintRawDataFp(stdout, sb->buf, sb->buf_offset); + PrintRawDataFp(stdout, sb->region.buf, sb->region.buf_offset); } static void DumpSegment(StreamingBuffer *sb, StreamingBufferSegment *seg) @@ -941,8 +1711,8 @@ static int StreamingBufferTest02(void) FAIL_IF(StreamingBufferAppend(sb, &seg1, (const uint8_t *)"ABCDEFGH", 8) != 0); StreamingBufferSegment seg2; FAIL_IF(StreamingBufferAppend(sb, &seg2, (const uint8_t *)"01234567", 8) != 0); - FAIL_IF(sb->stream_offset != 0); - FAIL_IF(sb->buf_offset != 16); + FAIL_IF(sb->region.stream_offset != 0); + FAIL_IF(sb->region.buf_offset != 16); FAIL_IF(seg1.stream_offset != 0); FAIL_IF(seg2.stream_offset != 8); FAIL_IF(StreamingBufferSegmentIsBeforeWindow(sb,&seg1)); @@ -959,8 +1729,8 @@ static int StreamingBufferTest02(void) StreamingBufferSegment seg3; FAIL_IF(StreamingBufferAppend(sb, &seg3, (const uint8_t *)"QWERTY", 6) != 0); - FAIL_IF(sb->stream_offset != 6); - FAIL_IF(sb->buf_offset != 16); + FAIL_IF(sb->region.stream_offset != 6); + FAIL_IF(sb->region.buf_offset != 16); FAIL_IF(seg3.stream_offset != 16); FAIL_IF(StreamingBufferSegmentIsBeforeWindow(sb,&seg1)); FAIL_IF(StreamingBufferSegmentIsBeforeWindow(sb,&seg2)); @@ -997,8 +1767,8 @@ static int StreamingBufferTest03(void) FAIL_IF(StreamingBufferAppend(sb, &seg1, (const uint8_t *)"ABCDEFGH", 8) != 0); StreamingBufferSegment seg2; FAIL_IF(StreamingBufferInsertAt(sb, &seg2, (const uint8_t *)"01234567", 8, 14) != 0); - FAIL_IF(sb->stream_offset != 0); - FAIL_IF(sb->buf_offset != 22); + FAIL_IF(sb->region.stream_offset != 0); + FAIL_IF(sb->region.buf_offset != 8); FAIL_IF(seg1.stream_offset != 0); FAIL_IF(seg2.stream_offset != 14); FAIL_IF(StreamingBufferSegmentIsBeforeWindow(sb,&seg1)); @@ -1012,8 +1782,8 @@ static int StreamingBufferTest03(void) StreamingBufferSegment seg3; FAIL_IF(StreamingBufferInsertAt(sb, &seg3, (const uint8_t *)"QWERTY", 6, 8) != 0); - FAIL_IF(sb->stream_offset != 0); - FAIL_IF(sb->buf_offset != 22); + FAIL_IF(sb->region.stream_offset != 0); + FAIL_IF(sb->region.buf_offset != 22); FAIL_IF(seg3.stream_offset != 8); FAIL_IF(StreamingBufferSegmentIsBeforeWindow(sb,&seg1)); FAIL_IF(StreamingBufferSegmentIsBeforeWindow(sb,&seg2)); @@ -1053,8 +1823,8 @@ static int StreamingBufferTest04(void) FAIL_IF(!RB_EMPTY(&sb->sbb_tree)); StreamingBufferSegment seg2; FAIL_IF(StreamingBufferInsertAt(sb, &seg2, (const uint8_t *)"01234567", 8, 14) != 0); - FAIL_IF(sb->stream_offset != 0); - FAIL_IF(sb->buf_offset != 22); + FAIL_IF(sb->region.stream_offset != 0); + FAIL_IF(sb->region.buf_offset != 8); FAIL_IF(seg1.stream_offset != 0); FAIL_IF(seg2.stream_offset != 14); FAIL_IF(StreamingBufferSegmentIsBeforeWindow(sb,&seg1)); @@ -1077,8 +1847,8 @@ static int StreamingBufferTest04(void) StreamingBufferSegment seg3; FAIL_IF(StreamingBufferInsertAt(sb, &seg3, (const uint8_t *)"QWERTY", 6, 8) != 0); - FAIL_IF(sb->stream_offset != 0); - FAIL_IF(sb->buf_offset != 22); + FAIL_IF(sb->region.stream_offset != 0); + FAIL_IF(sb->region.buf_offset != 22); FAIL_IF(seg3.stream_offset != 8); FAIL_IF(StreamingBufferSegmentIsBeforeWindow(sb,&seg1)); FAIL_IF(StreamingBufferSegmentIsBeforeWindow(sb,&seg2)); @@ -1100,9 +1870,9 @@ static int StreamingBufferTest04(void) /* far ahead of curve: */ StreamingBufferSegment seg4; FAIL_IF(StreamingBufferInsertAt(sb, &seg4, (const uint8_t *)"XYZ", 3, 124) != 0); - FAIL_IF(sb->stream_offset != 0); - FAIL_IF(sb->buf_offset != 127); - FAIL_IF(sb->buf_size != 128); + FAIL_IF(sb->region.stream_offset != 0); + FAIL_IF(sb->region.buf_offset != 22); + FAIL_IF(sb->region.buf_size != 128); FAIL_IF(seg4.stream_offset != 124); FAIL_IF(StreamingBufferSegmentIsBeforeWindow(sb,&seg1)); FAIL_IF(StreamingBufferSegmentIsBeforeWindow(sb,&seg2)); diff --git a/src/util-streaming-buffer.h b/src/util-streaming-buffer.h index 17844cc243..3bbfcf438d 100644 --- a/src/util-streaming-buffer.h +++ b/src/util-streaming-buffer.h @@ -74,6 +74,19 @@ typedef struct StreamingBufferConfig_ { 0, 0, NULL, NULL, NULL, \ } +#define STREAMING_BUFFER_REGION_INIT \ + { \ + NULL, 0, 0, 0ULL, NULL, \ + } + +typedef struct StreamingBufferRegion_ { + uint8_t *buf; /**< memory block for reassembly */ + uint32_t buf_size; /**< size of memory block */ + uint32_t buf_offset; /**< how far we are in buf_size */ + uint64_t stream_offset; /**< stream offset of this region */ + struct StreamingBufferRegion_ *next; +} StreamingBufferRegion; + /** * \brief block of continues data */ @@ -92,15 +105,12 @@ StreamingBufferBlock *SBB_RB_FIND_INCLUSIVE(struct SBB *head, StreamingBufferBlo typedef struct StreamingBuffer_ { const StreamingBufferConfig *cfg; - uint64_t stream_offset; /**< offset of the start of the memory block */ - - uint8_t *buf; /**< memory block for reassembly */ - uint32_t buf_size; /**< size of memory block */ - uint32_t buf_offset; /**< how far we are in buf_size */ - + StreamingBufferRegion region; struct SBB sbb_tree; /**< red black tree of Stream Buffer Blocks */ StreamingBufferBlock *head; /**< head, should always be the same as RB_MIN */ uint32_t sbb_size; /**< data size covered by sbbs */ + uint16_t regions; + uint16_t max_regions; #ifdef DEBUG uint32_t buf_size_max; #endif @@ -108,33 +118,34 @@ typedef struct StreamingBuffer_ { static inline bool StreamingBufferHasData(const StreamingBuffer *sb) { - return (sb->stream_offset || sb->buf_offset || !RB_EMPTY(&sb->sbb_tree)); + return (sb->region.stream_offset || sb->region.buf_offset || sb->region.next != NULL || + !RB_EMPTY(&sb->sbb_tree)); } static inline uint64_t StreamingBufferGetConsecutiveDataRightEdge(const StreamingBuffer *sb) { - return sb->stream_offset + sb->buf_offset; + return sb->region.stream_offset + sb->region.buf_offset; } static inline uint64_t StreamingBufferGetOffset(const StreamingBuffer *sb) { - return sb->stream_offset; + return sb->region.stream_offset; } #ifndef DEBUG #define STREAMING_BUFFER_INITIALIZER(cfg) \ { \ (cfg), \ - 0, \ - NULL, \ - 0, \ - 0, \ + STREAMING_BUFFER_REGION_INIT, \ { NULL }, \ NULL, \ 0, \ + 1, \ + 1, \ }; #else -#define STREAMING_BUFFER_INITIALIZER(cfg) { (cfg), 0, NULL, 0, 0, { NULL }, NULL, 0, 0 }; +#define STREAMING_BUFFER_INITIALIZER(cfg) \ + { (cfg), STREAMING_BUFFER_REGION_INIT, { NULL }, NULL, 0, 1, 1, 0 }; #endif typedef struct StreamingBufferSegment_ {