streaming: add blocklist

Add list of 'blocks'. This list contains offsets and lengths to
continuous data blocks. This is useful for TCP tracking where we
can have data gaps.

The blocks don't contain any data themselves, instead they contain
lenght and offsets. This way no extra copying is needed.

On inserting new data, existing blocks are expanded instead of
having multiple neighbouring blocks.

diff --git a/src/util-streaming-buffer.c b/src/util-streaming-buffer.c
index 9305a41ef1269cf3092004489bdd63e3fb75c210..fdfcd42e34e2978d7acf3d08b4f639b3587d7f8d 100644 (file)
--- a/src/util-streaming-buffer.c
+++ b/src/util-streaming-buffer.c
@@ -39,6 +39,8 @@
 #define FREE(cfg, ptr, s) \
     (cfg)->Free ? (cfg)->Free((ptr), (s)) : SCFree((ptr))
 
+static void SBBFree(StreamingBuffer *sb);
+
 static inline int InitBuffer(StreamingBuffer *sb)
 {
     sb->buf = CALLOC(sb->cfg, 1, sb->cfg->buf_size);
@@ -74,6 +76,7 @@ void StreamingBufferClear(StreamingBuffer *sb)
     if (sb != NULL) {
         SCLogDebug("sb->buf_size %u max %u", sb->buf_size, sb->buf_size_max);
 
+        SBBFree(sb);
         if (sb->buf != NULL) {
             FREE(sb->cfg, sb->buf, sb->buf_size);
             sb->buf = NULL;
@@ -89,6 +92,386 @@ void StreamingBufferFree(StreamingBuffer *sb)
     }
 }
 
+#ifdef DEBUG
+static void SBBPrintList(const StreamingBuffer *sb)
+{
+    const StreamingBufferBlock *sbb = sb->block_list;
+    while (sbb) {
+        SCLogDebug("sbb: offset %"PRIu64", len %u", sbb->offset, sbb->len);
+        if (sbb->next) {
+            if ((sbb->offset + sbb->len) != sbb->next->offset) {
+                SCLogDebug("gap: offset %"PRIu64", len %"PRIu64, (sbb->offset + sbb->len),
+                        sbb->next->offset - (sbb->offset + sbb->len));
+            }
+        }
+
+        sbb = sbb->next;
+    }
+}
+#endif
+
+/* setup with gap between 2 blocks
+ *
+ * [block][gap][block]
+ **/
+static void SBBInit(StreamingBuffer *sb,
+                    uint32_t rel_offset, uint32_t data_len)
+{
+    BUG_ON(sb->block_list != NULL);
+    BUG_ON(sb->buf_offset > sb->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;
+
+    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->len = data_len;
+
+    sb->block_list = sbb;
+    sbb->next = sbb2;
+    sb->block_list_tail = sbb2;
+
+    SCLogDebug("sbb1 %"PRIu64", len %u, sbb2 %"PRIu64", len %u",
+            sbb->offset, sbb->len, sbb2->offset, sbb2->len);
+#ifdef DEBUG
+    SBBPrintList(sb);
+#endif
+    BUG_ON(sbb2->offset < sbb->len);
+}
+
+/* setup with leading gap
+ *
+ * [gap][block]
+ **/
+static void SBBInitLeadingGap(StreamingBuffer *sb,
+                              uint64_t offset, uint32_t data_len)
+{
+    BUG_ON(sb->block_list != NULL);
+
+    StreamingBufferBlock *sbb = CALLOC(sb->cfg, 1, sizeof(*sbb));
+    if (sbb == NULL)
+        return;
+    sbb->offset = offset;
+    sbb->len = data_len;
+
+    sb->block_list = sbb;
+    sb->block_list_tail = sbb;
+
+    SCLogDebug("sbb %"PRIu64", len %u",
+            sbb->offset, sbb->len);
+#ifdef DEBUG
+    SBBPrintList(sb);
+#endif
+}
+
+static int IsBefore(StreamingBufferBlock *me, StreamingBufferBlock *you)
+{
+    if ((me->offset + me->len) < you->offset) {
+        return 1;
+    }
+    return 0;
+}
+
+static int StartsBefore(StreamingBufferBlock *me, StreamingBufferBlock *you)
+{
+    if (me->offset < you->offset)
+        return 1;
+    return 0;
+}
+
+static int IsAfter(StreamingBufferBlock *me, StreamingBufferBlock *you)
+{
+    if (you->offset + you->len < me->offset)
+        return 1;
+    return 0;
+}
+
+static int IsOverlappedBy(StreamingBufferBlock *me, StreamingBufferBlock *you)
+{
+    if (you->offset <= me->offset && (you->offset + you->len) >= (me->offset + me->len))
+        return 1;
+    return 0;
+}
+
+static int EndsAfter(StreamingBufferBlock *me, StreamingBufferBlock *you)
+{
+    if ((me->offset + me->len) > (you->offset + you->len))
+        return 1;
+    return 0;
+}
+
+static StreamingBufferBlock *GetNew(StreamingBuffer *sb,
+                                    uint64_t offset, uint32_t len,
+                                    StreamingBufferBlock *next)
+{
+    StreamingBufferBlock *new_sbb = CALLOC(sb->cfg, 1, sizeof(*new_sbb));
+    if (new_sbb == NULL)
+        return NULL;
+    new_sbb->offset = offset;
+    new_sbb->len = len;
+    new_sbb->next = next;
+    return new_sbb;
+}
+
+/* expand our sbb forward if possible */
+static int SBBUpdateLookForward(StreamingBuffer *sb,
+                                StreamingBufferBlock *sbb,
+                                StreamingBufferBlock *my_block)
+{
+    SCLogDebug("EndsAfter: consider next");
+
+    while (sbb->offset + sbb->len == sbb->next->offset)
+    {
+        SCLogDebug("EndsAfter: gobble up next: %u/%u", (uint)sbb->next->offset, sbb->next->len);
+        uint64_t right_edge = sbb->next->offset + sbb->next->len;
+        uint32_t expand_by = right_edge - (sbb->offset + sbb->len);
+        sbb->len += expand_by;
+        SCLogDebug("EndsAfter: expand_by %u (part 2)", expand_by);
+        SCLogDebug("EndsAfter: (loop) sbb now %u/%u", (uint)sbb->offset, sbb->len);
+
+        /* we can gobble up next */
+        StreamingBufferBlock *to_free = sbb->next;
+        sbb->next = sbb->next->next;
+        FREE(sb->cfg, to_free, sizeof(StreamingBufferBlock));
+        if (sbb->next == NULL)
+            sb->block_list_tail = sbb;
+
+        /* update my block */
+        if (expand_by >= my_block->len) {
+            return 1;
+        }
+
+        my_block->len -= expand_by;
+        my_block->offset += expand_by;
+
+        if (sbb->next == NULL) {
+            /* if we have nothing left in the list we're almost done,
+             * except we need to check if we have some of our block
+             * left */
+            sbb->len += my_block->len;
+            my_block->offset += my_block->len;
+            my_block->len = 0;
+            return 1;
+        } else {
+            /* if next is not directly connected and we have some
+             * block len left, expand sbb further */
+            uint32_t gap = sbb->next->offset - (sbb->offset + sbb->len);
+            SCLogDebug("EndsAfter: we now have a gap of %u and a block of %u/%u", gap, (uint)my_block->offset, my_block->len);
+
+            if (my_block->len < gap) {
+                sbb->len += my_block->len;
+                my_block->offset += my_block->len;
+                my_block->len = 0;
+                return 1;
+            } else {
+                sbb->len += gap;
+                my_block->offset += gap;
+                my_block->len -= gap;
+                SCLogDebug("EndsAfter: (loop) block at %u/%u, sbb %u/%u", (uint)my_block->offset, my_block->len, (uint)sbb->offset, sbb->len);
+                SCLogDebug("EndsAfter: (loop) sbb->next %u/%u", (uint)sbb->next->offset, sbb->next->len);
+            }
+        }
+    }
+    return 0;
+}
+
+static void SBBUpdate(StreamingBuffer *sb,
+                      uint32_t rel_offset, uint32_t data_len)
+{
+    StreamingBufferBlock my_block = { .offset = sb->stream_offset + rel_offset,
+                                      .len = data_len,
+                                      .next = NULL  };
+    const uint64_t my_block_right_edge = my_block.offset + my_block.len;
+
+    StreamingBufferBlock *tail = sb->block_list_tail;
+
+    /* fast path 1: expands tail */
+    if (tail && ((tail->offset + tail->len) == my_block.offset))
+    {
+        tail->len = my_block_right_edge - tail->offset;
+        goto done;
+    }
+    /* fast path 2: new isolated block after tail */
+    else if (tail && IsAfter(&my_block, tail)) {
+        StreamingBufferBlock *new_sbb = GetNew(sb, my_block.offset, my_block.len, NULL);
+        sb->block_list_tail = tail->next = new_sbb;
+        SCLogDebug("tail: new block at %u/%u", (uint)my_block.offset, my_block.len);
+        goto done;
+    }
+
+    BUG_ON(sb->block_list == NULL);
+#ifdef DEBUG
+    SBBPrintList(sb);
+#endif
+    SCLogDebug("PreInsert: block at %u/%u", (uint)my_block.offset, my_block.len);
+    StreamingBufferBlock *sbb = sb->block_list, *prev = NULL;
+    while (sbb) {
+        SCLogDebug("sbb %"PRIu64"/%u data %"PRIu64"/%u. Next %s", sbb->offset, sbb->len,
+                my_block.offset, my_block.len, sbb->next ? "true" : "false");
+
+        if (IsBefore(&my_block, sbb)) {
+            StreamingBufferBlock *new_sbb = GetNew(sb, my_block.offset, my_block.len, sbb);
+
+            /* place before, maybe replace list head */
+            if (sbb == sb->block_list) {
+                sb->block_list = new_sbb;
+            } else {
+                prev->next = new_sbb;
+            }
+            SCLogDebug("IsBefore: new block at %u/%u", (uint)my_block.offset, my_block.len);
+            break;
+
+        } else if (IsOverlappedBy(&my_block, sbb)) {
+            /* nothing to do */
+            SCLogDebug("IsOverlappedBy: overlapped block at %u/%u", (uint)my_block.offset, my_block.len);
+            break;
+
+        } else if (IsAfter(&my_block, sbb)) {
+
+            /* if no next, place after, otherwise, iterate */
+            if (sbb->next == NULL) {
+                StreamingBufferBlock *new_sbb = GetNew(sb, my_block.offset, my_block.len, NULL);
+                sbb->next = new_sbb;
+                sb->block_list_tail = new_sbb;
+                SCLogDebug("new block at %u/%u", (uint)my_block.offset, my_block.len);
+                break;
+            }
+            SCLogDebug("IsAfter: block at %u/%u, is after sbb", (uint)my_block.offset, my_block.len);
+
+        } else {
+
+            /* those were the simple cases */
+
+            if (StartsBefore(&my_block, sbb)) {
+                /* expand sbb */
+                uint32_t expand_by = sbb->offset - my_block.offset;
+                SCLogDebug("StartsBefore: expand_by %u", expand_by);
+                sbb->offset = my_block.offset;
+                sbb->len += expand_by;
+
+                /* if my_block ends before sbb right edge, we are done */
+                if (my_block_right_edge <= (sbb->offset + sbb->len))
+                    break;
+
+                my_block.offset = sbb->offset + sbb->len;
+                my_block.len = my_block_right_edge - my_block.offset;
+                SCLogDebug("StartsBefore: block now %u/%u", (uint)my_block.offset, my_block.len);
+
+                if (sbb->next == NULL) {
+                    sbb->len += my_block.len;
+                    break;
+                }
+                /* expand, but consider next */
+                uint64_t right_edge = my_block_right_edge;
+                if (right_edge > sbb->next->offset) {
+                    right_edge = sbb->next->offset;
+                }
+
+                expand_by = right_edge - (sbb->offset + sbb->len);
+                SCLogDebug("EndsAfter: expand_by %u", expand_by);
+                sbb->len += expand_by;
+                SCLogDebug("EndsAfter: sbb now %u/%u", (uint)sbb->offset, sbb->len);
+
+                my_block.offset = sbb->offset + sbb->len;
+                my_block.len = my_block_right_edge - my_block.offset;
+                SCLogDebug("StartsBefore: sbb now %u/%u", (uint)sbb->offset, sbb->len);
+
+            } else if (EndsAfter(&my_block, sbb)) {
+                /* expand sbb, but we need to mind "next" */
+
+                if (sbb->next == NULL) {
+                    /* last, so just expand sbb */
+                    sbb->len = my_block_right_edge - sbb->offset;
+                    break;
+                }
+
+                /* expand, but consider next */
+                uint64_t right_edge = my_block_right_edge;
+                if (right_edge > sbb->next->offset) {
+                    right_edge = sbb->next->offset;
+                }
+
+                uint32_t expand_by = right_edge - (sbb->offset + sbb->len);
+                SCLogDebug("EndsAfter: expand_by %u", expand_by);
+                sbb->len += expand_by;
+                SCLogDebug("EndsAfter: sbb now %u/%u", (uint)sbb->offset, sbb->len);
+
+                my_block.offset = sbb->offset + sbb->len;
+                my_block.len = my_block_right_edge - my_block.offset;
+            }
+
+            if (sbb->next != NULL) {
+                SCLogDebug("EndsAfter: consider next");
+
+                if (SBBUpdateLookForward(sb, sbb, &my_block) == 1)
+                    goto done;
+            }
+
+            SCLogDebug("EndsAfter: block at %u/%u, is after sbb", (uint)my_block.offset, my_block.len);
+
+            if (my_block.len == 0)
+                break;
+        }
+        prev = sbb;
+        sbb = sbb->next;
+    }
+done:
+    SCLogDebug("PostInsert: block at %u/%u", (uint)my_block.offset, my_block.len);
+    SCLogDebug("PostInsert");
+#ifdef DEBUG
+    SBBPrintList(sb);
+#endif
+}
+
+static void SBBFree(StreamingBuffer *sb)
+{
+    StreamingBufferBlock *sbb = sb->block_list;
+    while (sbb) {
+        StreamingBufferBlock *next = sbb->next;
+        FREE(sb->cfg, sbb, sizeof(StreamingBufferBlock));
+        sbb = next;
+    }
+    sb->block_list = NULL;
+}
+
+static void SBBPrune(StreamingBuffer *sb)
+{
+    StreamingBufferBlock *sbb = sb->block_list;
+    while (sbb) {
+        /* completely beyond window, we're done */
+        if (sbb->offset > sb->stream_offset)
+            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;
+            BUG_ON(shrink_by > sbb->len);
+            sbb->len -=  shrink_by;
+            sbb->offset += shrink_by;
+            BUG_ON(sbb->offset != sb->stream_offset);
+            break;
+        }
+
+        StreamingBufferBlock *next = sbb->next;
+        FREE(sb->cfg, sbb, sizeof(StreamingBufferBlock));
+
+        sbb = next;
+        sb->block_list = next;
+        if (sbb && sbb->next == NULL)
+            sb->block_list_tail = NULL;
+    }
+}
+
 /**
  * \internal
  * \brief move buffer forward by 'slide'
@@ -101,6 +484,7 @@ static void AutoSlide(StreamingBuffer *sb)
     memmove(sb->buf, sb->buf+slide, size);
     sb->stream_offset += slide;
     sb->buf_offset = size;
+    SBBPrune(sb);
 }
 
 static int __attribute__((warn_unused_result))
@@ -176,6 +560,7 @@ void StreamingBufferSlideToOffset(StreamingBuffer *sb, uint64_t offset)
         memmove(sb->buf, sb->buf+slide, size);
         sb->stream_offset += slide;
         sb->buf_offset = size;
+        SBBPrune(sb);
     }
 }
 
@@ -186,6 +571,7 @@ void StreamingBufferSlide(StreamingBuffer *sb, uint32_t slide)
     memmove(sb->buf, sb->buf+slide, size);
     sb->stream_offset += slide;
     sb->buf_offset = size;
+    SBBPrune(sb);
 }
 
 #define DATA_FITS(sb, len) \
@@ -221,7 +607,12 @@ StreamingBufferSegment *StreamingBufferAppendRaw(StreamingBuffer *sb, const uint
         memcpy(sb->buf + sb->buf_offset, data, data_len);
         seg->stream_offset = sb->stream_offset + sb->buf_offset;
         seg->segment_len = data_len;
+        uint32_t rel_offset = sb->buf_offset;
         sb->buf_offset += data_len;
+
+        if (sb->block_list) {
+            SBBUpdate(sb, rel_offset, data_len);
+        }
         return seg;
     }
     return NULL;
@@ -258,7 +649,12 @@ int StreamingBufferAppend(StreamingBuffer *sb, StreamingBufferSegment *seg,
     memcpy(sb->buf + sb->buf_offset, data, data_len);
     seg->stream_offset = sb->stream_offset + sb->buf_offset;
     seg->segment_len = data_len;
+    uint32_t rel_offset = sb->buf_offset;
     sb->buf_offset += data_len;
+
+    if (sb->block_list) {
+        SBBUpdate(sb, rel_offset, data_len);
+    }
     return 0;
 }
 
@@ -292,7 +688,12 @@ int StreamingBufferAppendNoTrack(StreamingBuffer *sb,
     }
 
     memcpy(sb->buf + sb->buf_offset, data, data_len);
+    uint32_t rel_offset = sb->buf_offset;
     sb->buf_offset += data_len;
+
+    if (sb->block_list) {
+        SBBUpdate(sb, rel_offset, data_len);
+    }
     return 0;
 }
 
@@ -332,8 +733,41 @@ int StreamingBufferInsertAt(StreamingBuffer *sb, StreamingBufferSegment *seg,
     memcpy(sb->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);
+
+    if (sb->block_list == NULL) {
+        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");
+        } 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) {
+                /* existing data, but there is a gap between us */
+                SBBInit(sb, rel_offset, data_len);
+            } else {
+                /* gap before data in empty list */
+                SCLogDebug("empty sbb list: invoking SBBInitLeadingGap");
+                SBBInitLeadingGap(sb, offset, data_len);
+            }
+        }
+    } else {
+        /* already have blocks, so append new block based on new data */
+        SBBUpdate(sb, rel_offset, data_len);
+    }
+
     if (rel_offset + data_len > sb->buf_offset)
         sb->buf_offset = rel_offset + data_len;
+
     return 0;
 }
 
@@ -348,6 +782,66 @@ int StreamingBufferSegmentIsBeforeWindow(const StreamingBuffer *sb,
     return 0;
 }
 
+/** \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_size)
+            *data_len = sb->buf_size - offset;
+        else
+            *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;
+        }
+    }
+    *data = NULL;
+    *data_len = 0;
+    return;
+}
+
+/** \brief get the data for one SBB */
+void StreamingBufferSBBGetDataAtOffset(const StreamingBuffer *sb,
+                                       const StreamingBufferBlock *sbb,
+                                       const uint8_t **data, uint32_t *data_len,
+                                       uint64_t offset)
+{
+    if (offset >= sbb->offset && offset < (sbb->offset + sbb->len)) {
+        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;
+            else
+                *data_len = sbblen;
+            BUG_ON(*data_len > sbblen);
+            return;
+        } else {
+            uint64_t data_offset = sb->stream_offset - sbb->offset;
+            if (data_offset < sbblen) {
+                *data = sb->buf;
+                *data_len = sbblen - data_offset;
+                BUG_ON(*data_len > sbblen);
+                return;
+            }
+        }
+    }
+
+    *data = NULL;
+    *data_len = 0;
+    return;
+}
+
 void StreamingBufferSegmentGetData(const StreamingBuffer *sb,
                                    const StreamingBufferSegment *seg,
                                    const uint8_t **data, uint32_t *data_len)
@@ -447,8 +941,8 @@ int StreamingBufferCompareRawData(const StreamingBuffer *sb,
     {
         return 1;
     }
-    SCLogInfo("sbdata_len %u, offset %u", sbdata_len, (uint)offset);
-    PrintRawDataFp(stdout, sbdata,sbdata_len);
+    SCLogDebug("sbdata_len %u, offset %u", sbdata_len, (uint)offset);
+    //PrintRawDataFp(stdout, sbdata,sbdata_len);
     return 0;
 }
 
@@ -644,6 +1138,7 @@ static int StreamingBufferTest04(void)
 
     StreamingBufferSegment seg1;
     FAIL_IF(StreamingBufferAppend(sb, &seg1, (const uint8_t *)"ABCDEFGH", 8) != 0);
+    FAIL_IF(sb->block_list != NULL);
     StreamingBufferSegment seg2;
     FAIL_IF(StreamingBufferInsertAt(sb, &seg2, (const uint8_t *)"01234567", 8, 14) != 0);
     FAIL_IF(sb->stream_offset != 0);
@@ -652,6 +1147,12 @@ static int StreamingBufferTest04(void)
     FAIL_IF(seg2.stream_offset != 14);
     FAIL_IF(StreamingBufferSegmentIsBeforeWindow(sb,&seg1));
     FAIL_IF(StreamingBufferSegmentIsBeforeWindow(sb,&seg2));
+    FAIL_IF(sb->block_list == NULL);
+    FAIL_IF(sb->block_list->offset != 0);
+    FAIL_IF(sb->block_list->len != 8);
+    FAIL_IF(sb->block_list->next == NULL);
+    FAIL_IF(sb->block_list->next->offset != 14);
+    FAIL_IF(sb->block_list->next->len != 8);
     Dump(sb);
     DumpSegment(sb, &seg1);
     DumpSegment(sb, &seg2);
@@ -664,6 +1165,10 @@ static int StreamingBufferTest04(void)
     FAIL_IF(StreamingBufferSegmentIsBeforeWindow(sb,&seg1));
     FAIL_IF(StreamingBufferSegmentIsBeforeWindow(sb,&seg2));
     FAIL_IF(StreamingBufferSegmentIsBeforeWindow(sb,&seg3));
+    FAIL_IF(sb->block_list == NULL);
+    FAIL_IF(sb->block_list->offset != 0);
+    FAIL_IF(sb->block_list->len != 22);
+    FAIL_IF(sb->block_list->next != NULL);
     Dump(sb);
     DumpSegment(sb, &seg1);
     DumpSegment(sb, &seg2);
@@ -680,6 +1185,10 @@ static int StreamingBufferTest04(void)
     FAIL_IF(StreamingBufferSegmentIsBeforeWindow(sb,&seg2));
     FAIL_IF(StreamingBufferSegmentIsBeforeWindow(sb,&seg3));
     FAIL_IF(StreamingBufferSegmentIsBeforeWindow(sb,&seg4));
+    FAIL_IF(sb->block_list == NULL);
+    FAIL_IF(sb->block_list->offset != 0);
+    FAIL_IF(sb->block_list->len != 22);
+    FAIL_IF(sb->block_list->next == NULL);
     Dump(sb);
     DumpSegment(sb, &seg1);
     DumpSegment(sb, &seg2);
@@ -731,6 +1240,218 @@ static int StreamingBufferTest05(void)
     StreamingBufferClear(&sb);
     PASS;
 }
+
+/** \test lots of gaps in block list */
+static int StreamingBufferTest06(void)
+{
+    StreamingBufferConfig cfg = { 0, 8, 16, NULL, NULL, NULL, NULL };
+    StreamingBuffer *sb = StreamingBufferInit(&cfg);
+    FAIL_IF(sb == NULL);
+
+    StreamingBufferSegment seg1;
+    FAIL_IF(StreamingBufferAppend(sb, &seg1, (const uint8_t *)"A", 1) != 0);
+    StreamingBufferSegment seg2;
+    FAIL_IF(StreamingBufferInsertAt(sb, &seg2, (const uint8_t *)"C", 1, 2) != 0);
+    Dump(sb);
+
+    StreamingBufferSegment seg3;
+    FAIL_IF(StreamingBufferInsertAt(sb, &seg3, (const uint8_t *)"F", 1, 5) != 0);
+    Dump(sb);
+
+    StreamingBufferSegment seg4;
+    FAIL_IF(StreamingBufferInsertAt(sb, &seg4, (const uint8_t *)"H", 1, 7) != 0);
+    Dump(sb);
+
+    StreamingBufferSegment seg5;
+    FAIL_IF(StreamingBufferInsertAt(sb, &seg5, (const uint8_t *)"ABCDEFGHIJ", 10, 0) != 0);
+    Dump(sb);
+    FAIL_IF(sb->block_list == NULL);
+    FAIL_IF(sb->block_list->offset != 0);
+    FAIL_IF(sb->block_list->len != 10);
+    FAIL_IF(sb->block_list->next != NULL);
+
+    StreamingBufferSegment seg6;
+    FAIL_IF(StreamingBufferInsertAt(sb, &seg6, (const uint8_t *)"abcdefghij", 10, 0) != 0);
+    Dump(sb);
+    FAIL_IF(sb->block_list == NULL);
+    FAIL_IF(sb->block_list->offset != 0);
+    FAIL_IF(sb->block_list->len != 10);
+    FAIL_IF(sb->block_list->next != NULL);
+
+    StreamingBufferFree(sb);
+    PASS;
+}
+
+/** \test lots of gaps in block list */
+static int StreamingBufferTest07(void)
+{
+    StreamingBufferConfig cfg = { 0, 8, 16, NULL, NULL, NULL, NULL };
+    StreamingBuffer *sb = StreamingBufferInit(&cfg);
+    FAIL_IF(sb == NULL);
+
+    StreamingBufferSegment seg1;
+    FAIL_IF(StreamingBufferInsertAt(sb, &seg1, (const uint8_t *)"B", 1, 1) != 0);
+    StreamingBufferSegment seg2;
+    FAIL_IF(StreamingBufferInsertAt(sb, &seg2, (const uint8_t *)"D", 1, 3) != 0);
+    Dump(sb);
+
+    StreamingBufferSegment seg3;
+    FAIL_IF(StreamingBufferInsertAt(sb, &seg3, (const uint8_t *)"F", 1, 5) != 0);
+    Dump(sb);
+
+    StreamingBufferSegment seg4;
+    FAIL_IF(StreamingBufferInsertAt(sb, &seg4, (const uint8_t *)"H", 1, 7) != 0);
+    Dump(sb);
+
+    StreamingBufferSegment seg5;
+    FAIL_IF(StreamingBufferInsertAt(sb, &seg5, (const uint8_t *)"ABCDEFGHIJ", 10, 0) != 0);
+    Dump(sb);
+    FAIL_IF(sb->block_list == NULL);
+    FAIL_IF(sb->block_list->offset != 0);
+    FAIL_IF(sb->block_list->len != 10);
+    FAIL_IF(sb->block_list->next != NULL);
+
+    StreamingBufferSegment seg6;
+    FAIL_IF(StreamingBufferInsertAt(sb, &seg6, (const uint8_t *)"abcdefghij", 10, 0) != 0);
+    Dump(sb);
+    FAIL_IF(sb->block_list == NULL);
+    FAIL_IF(sb->block_list->offset != 0);
+    FAIL_IF(sb->block_list->len != 10);
+    FAIL_IF(sb->block_list->next != NULL);
+
+    StreamingBufferFree(sb);
+    PASS;
+}
+
+/** \test lots of gaps in block list */
+static int StreamingBufferTest08(void)
+{
+    StreamingBufferConfig cfg = { 0, 8, 16, NULL, NULL, NULL, NULL };
+    StreamingBuffer *sb = StreamingBufferInit(&cfg);
+    FAIL_IF(sb == NULL);
+
+    StreamingBufferSegment seg1;
+    FAIL_IF(StreamingBufferInsertAt(sb, &seg1, (const uint8_t *)"B", 1, 1) != 0);
+    StreamingBufferSegment seg2;
+    FAIL_IF(StreamingBufferInsertAt(sb, &seg2, (const uint8_t *)"D", 1, 3) != 0);
+    Dump(sb);
+
+    StreamingBufferSegment seg3;
+    FAIL_IF(StreamingBufferInsertAt(sb, &seg3, (const uint8_t *)"F", 1, 5) != 0);
+    Dump(sb);
+
+    StreamingBufferSegment seg4;
+    FAIL_IF(StreamingBufferInsertAt(sb, &seg4, (const uint8_t *)"H", 1, 7) != 0);
+    Dump(sb);
+
+    StreamingBufferSegment seg5;
+    FAIL_IF(StreamingBufferInsertAt(sb, &seg5, (const uint8_t *)"ABCDEFGHIJ", 10, 0) != 0);
+    Dump(sb);
+    FAIL_IF(sb->block_list == NULL);
+    FAIL_IF(sb->block_list->offset != 0);
+    FAIL_IF(sb->block_list->len != 10);
+    FAIL_IF(sb->block_list->next != NULL);
+
+    StreamingBufferSegment seg6;
+    FAIL_IF(StreamingBufferAppend(sb, &seg6, (const uint8_t *)"abcdefghij", 10) != 0);
+    Dump(sb);
+    FAIL_IF(sb->block_list == NULL);
+    FAIL_IF(sb->block_list->offset != 0);
+    FAIL_IF(sb->block_list->len != 20);
+    FAIL_IF(sb->block_list->next != NULL);
+
+    StreamingBufferFree(sb);
+    PASS;
+}
+
+/** \test lots of gaps in block list */
+static int StreamingBufferTest09(void)
+{
+    StreamingBufferConfig cfg = { 0, 8, 16, NULL, NULL, NULL, NULL };
+    StreamingBuffer *sb = StreamingBufferInit(&cfg);
+    FAIL_IF(sb == NULL);
+
+    StreamingBufferSegment seg1;
+    FAIL_IF(StreamingBufferInsertAt(sb, &seg1, (const uint8_t *)"B", 1, 1) != 0);
+    StreamingBufferSegment seg2;
+    FAIL_IF(StreamingBufferInsertAt(sb, &seg2, (const uint8_t *)"D", 1, 3) != 0);
+    Dump(sb);
+
+    StreamingBufferSegment seg3;
+    FAIL_IF(StreamingBufferInsertAt(sb, &seg3, (const uint8_t *)"H", 1, 7) != 0);
+    Dump(sb);
+
+    StreamingBufferSegment seg4;
+    FAIL_IF(StreamingBufferInsertAt(sb, &seg4, (const uint8_t *)"F", 1, 5) != 0);
+    Dump(sb);
+
+    StreamingBufferSegment seg5;
+    FAIL_IF(StreamingBufferInsertAt(sb, &seg5, (const uint8_t *)"ABCDEFGHIJ", 10, 0) != 0);
+    Dump(sb);
+    FAIL_IF(sb->block_list == NULL);
+    FAIL_IF(sb->block_list->offset != 0);
+    FAIL_IF(sb->block_list->len != 10);
+    FAIL_IF(sb->block_list->next != NULL);
+
+    StreamingBufferSegment seg6;
+    FAIL_IF(StreamingBufferInsertAt(sb, &seg6, (const uint8_t *)"abcdefghij", 10, 0) != 0);
+    Dump(sb);
+    FAIL_IF(sb->block_list == NULL);
+    FAIL_IF(sb->block_list->offset != 0);
+    FAIL_IF(sb->block_list->len != 10);
+    FAIL_IF(sb->block_list->next != NULL);
+
+    StreamingBufferFree(sb);
+    PASS;
+}
+
+/** \test lots of gaps in block list */
+static int StreamingBufferTest10(void)
+{
+    StreamingBufferConfig cfg = { 0, 8, 16, NULL, NULL, NULL, NULL };
+    StreamingBuffer *sb = StreamingBufferInit(&cfg);
+    FAIL_IF(sb == NULL);
+
+    StreamingBufferSegment seg1;
+    FAIL_IF(StreamingBufferInsertAt(sb, &seg1, (const uint8_t *)"A", 1, 0) != 0);
+    StreamingBufferSegment seg2;
+    FAIL_IF(StreamingBufferInsertAt(sb, &seg2, (const uint8_t *)"D", 1, 3) != 0);
+    Dump(sb);
+
+    StreamingBufferSegment seg3;
+    FAIL_IF(StreamingBufferInsertAt(sb, &seg3, (const uint8_t *)"H", 1, 7) != 0);
+    Dump(sb);
+
+    StreamingBufferSegment seg4;
+    FAIL_IF(StreamingBufferInsertAt(sb, &seg4, (const uint8_t *)"B", 1, 1) != 0);
+    Dump(sb);
+    StreamingBufferSegment seg5;
+    FAIL_IF(StreamingBufferInsertAt(sb, &seg5, (const uint8_t *)"C", 1, 2) != 0);
+    Dump(sb);
+    StreamingBufferSegment seg6;
+    FAIL_IF(StreamingBufferInsertAt(sb, &seg6, (const uint8_t *)"G", 1, 6) != 0);
+    Dump(sb);
+
+    StreamingBufferSegment seg7;
+    FAIL_IF(StreamingBufferInsertAt(sb, &seg7, (const uint8_t *)"ABCDEFGHIJ", 10, 0) != 0);
+    Dump(sb);
+    FAIL_IF(sb->block_list == NULL);
+    FAIL_IF(sb->block_list->offset != 0);
+    FAIL_IF(sb->block_list->len != 10);
+    FAIL_IF(sb->block_list->next != NULL);
+
+    StreamingBufferSegment seg8;
+    FAIL_IF(StreamingBufferInsertAt(sb, &seg8, (const uint8_t *)"abcdefghij", 10, 0) != 0);
+    Dump(sb);
+    FAIL_IF(sb->block_list == NULL);
+    FAIL_IF(sb->block_list->offset != 0);
+    FAIL_IF(sb->block_list->len != 10);
+    FAIL_IF(sb->block_list->next != NULL);
+
+    StreamingBufferFree(sb);
+    PASS;
+}
+
 #endif
 
 void StreamingBufferRegisterTests(void)
@@ -741,5 +1462,10 @@ void StreamingBufferRegisterTests(void)
     UtRegisterTest("StreamingBufferTest03", StreamingBufferTest03);
     UtRegisterTest("StreamingBufferTest04", StreamingBufferTest04);
     UtRegisterTest("StreamingBufferTest05", StreamingBufferTest05);
+    UtRegisterTest("StreamingBufferTest06", StreamingBufferTest06);
+    UtRegisterTest("StreamingBufferTest07", StreamingBufferTest07);
+    UtRegisterTest("StreamingBufferTest08", StreamingBufferTest08);
+    UtRegisterTest("StreamingBufferTest09", StreamingBufferTest09);
+    UtRegisterTest("StreamingBufferTest10", StreamingBufferTest10);
 #endif
 }

diff --git a/src/util-streaming-buffer.h b/src/util-streaming-buffer.h
index 20c0a8904b7985709afa7e161fa0bf4323842813..91dd011514029993117aa39aaabe41124e6dc8b6 100644 (file)
--- a/src/util-streaming-buffer.h
+++ b/src/util-streaming-buffer.h
@@ -74,6 +74,15 @@ typedef struct StreamingBufferConfig_ {
 
 #define STREAMING_BUFFER_CONFIG_INITIALIZER { 0, 0, 0, NULL, NULL, NULL, NULL, }
 
+/**
+ *  \brief block of continues data
+ */
+typedef struct StreamingBufferBlock_ {
+    uint64_t offset;
+    uint32_t len;
+    struct StreamingBufferBlock_ *next;
+} StreamingBufferBlock;
+
 typedef struct StreamingBuffer_ {
     const StreamingBufferConfig *cfg;
     uint64_t stream_offset; /**< offset of the start of the memory block */
@@ -81,15 +90,18 @@ typedef struct StreamingBuffer_ {
     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 */
+
+    StreamingBufferBlock *block_list;
+    StreamingBufferBlock *block_list_tail;
 #ifdef DEBUG
     uint32_t buf_size_max;
 #endif
 } StreamingBuffer;
 
 #ifndef DEBUG
-#define STREAMING_BUFFER_INITIALIZER(cfg) { (cfg), 0, NULL, 0, 0, };
+#define STREAMING_BUFFER_INITIALIZER(cfg) { (cfg), 0, NULL, 0, 0, NULL, NULL};
 #else
-#define STREAMING_BUFFER_INITIALIZER(cfg) { (cfg), 0, NULL, 0, 0, 0 };
+#define STREAMING_BUFFER_INITIALIZER(cfg) { (cfg), 0, NULL, 0, 0, NULL, NULL, 0 };
 #endif
 
 typedef struct StreamingBufferSegment_ {
@@ -118,6 +130,15 @@ void StreamingBufferSegmentGetData(const StreamingBuffer *sb,
                                    const StreamingBufferSegment *seg,
                                    const uint8_t **data, uint32_t *data_len);
 
+void StreamingBufferSBBGetData(const StreamingBuffer *sb,
+                               const StreamingBufferBlock *sbb,
+                               const uint8_t **data, uint32_t *data_len);
+
+void StreamingBufferSBBGetDataAtOffset(const StreamingBuffer *sb,
+                                       const StreamingBufferBlock *sbb,
+                                       const uint8_t **data, uint32_t *data_len,
+                                       uint64_t offset);
+
 int StreamingBufferSegmentCompareRawData(const StreamingBuffer *sb,
                                          const StreamingBufferSegment *seg,
                                          const uint8_t *rawdata, uint32_t rawdata_len);