suricata
util-streaming-buffer.c
Go to the documentation of this file.
1 /* Copyright (C) 2015-2023 Open Information Security Foundation
2  *
3  * You can copy, redistribute or modify this Program under the terms of
4  * the GNU General Public License version 2 as published by the Free
5  * Software Foundation.
6  *
7  * This program is distributed in the hope that it will be useful,
8  * but WITHOUT ANY WARRANTY; without even the implied warranty of
9  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
10  * GNU General Public License for more details.
11  *
12  * You should have received a copy of the GNU General Public License
13  * version 2 along with this program; if not, write to the Free Software
14  * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
15  * 02110-1301, USA.
16  */
17 
18 #include "suricata-common.h"
19 #include "util-streaming-buffer.h"
20 #include "util-unittest.h"
21 #include "util-print.h"
22 #include "util-validate.h"
23 #include "util-debug.h"
24 #include "util-error.h"
25 
26 static void ListRegions(StreamingBuffer *sb);
27 
28 #define DUMP_REGIONS 0 // set to 1 to dump a visual representation of the regions list and sbb tree.
29 
30 /**
31  * \file
32  *
33  * \author Victor Julien <victor@inliniac.net>
34  *
35  * \brief Streaming Buffer API
36  */
37 
38 static void *ReallocFunc(void *ptr, const size_t size)
39 {
40  void *ptrmem = SCRealloc(ptr, size);
41  if (unlikely(ptrmem == NULL)) {
42  sc_errno = SC_ENOMEM;
43  }
44  return ptrmem;
45 }
46 
47 static void *CallocFunc(const size_t nm, const size_t sz)
48 {
49  void *ptrmem = SCCalloc(nm, sz);
50  if (unlikely(ptrmem == NULL)) {
51  sc_errno = SC_ENOMEM;
52  }
53  return ptrmem;
54 }
55 
56 /* memory handling wrappers. If config doesn't define it's own set of
57  * functions, use the defaults */
58 // TODO the default allocators don't set `sc_errno` yet.
59 #define CALLOC(cfg, n, s) (cfg)->Calloc ? (cfg)->Calloc((n), (s)) : CallocFunc((n), (s))
60 #define REALLOC(cfg, ptr, orig_s, s) \
61  (cfg)->Realloc ? (cfg)->Realloc((ptr), (orig_s), (s)) : ReallocFunc((ptr), (s))
62 #define FREE(cfg, ptr, s) \
63  (cfg)->Free ? (cfg)->Free((ptr), (s)) : SCFree((ptr))
64 
65 static void SBBFree(StreamingBuffer *sb, const StreamingBufferConfig *cfg);
66 
67 RB_GENERATE(SBB, StreamingBufferBlock, rb, SBBCompare);
68 
69 int SBBCompare(struct StreamingBufferBlock *a, struct StreamingBufferBlock *b)
70 {
71  SCLogDebug("a %" PRIu64 " len %u, b %" PRIu64 " len %u", a->offset, a->len, b->offset, b->len);
72 
73  if (a->offset > b->offset)
74  SCReturnInt(1);
75  else if (a->offset < b->offset)
76  SCReturnInt(-1);
77  else {
78  if (a->len == 0 || b->len == 0 || a->len == b->len)
79  SCReturnInt(0);
80  else if (a->len > b->len)
81  SCReturnInt(1);
82  else
83  SCReturnInt(-1);
84  }
85 }
86 
87 /* inclusive compare function that also considers the right edge,
88  * not just the offset. */
89 static inline int InclusiveCompare(StreamingBufferBlock *lookup, StreamingBufferBlock *intree) {
90  const uint64_t lre = lookup->offset + lookup->len;
91  const uint64_t tre = intree->offset + intree->len;
92  if (lre <= intree->offset) // entirely before
93  return -1;
94  else if (lookup->offset < tre && lre <= tre) // (some) overlap
95  return 0;
96  else
97  return 1; // entirely after
98 }
99 
100 StreamingBufferBlock *SBB_RB_FIND_INCLUSIVE(struct SBB *head, StreamingBufferBlock *elm)
101 {
102  SCLogDebug("looking up %" PRIu64, elm->offset);
103 
104  struct StreamingBufferBlock *tmp = RB_ROOT(head);
105  struct StreamingBufferBlock *res = NULL;
106  while (tmp) {
107  SCLogDebug("compare with %" PRIu64 "/%u", tmp->offset, tmp->len);
108  const int comp = InclusiveCompare(elm, tmp);
109  SCLogDebug("compare result: %d", comp);
110  if (comp < 0) {
111  res = tmp;
112  tmp = RB_LEFT(tmp, rb);
113  } else if (comp > 0) {
114  tmp = RB_RIGHT(tmp, rb);
115  } else {
116  return tmp;
117  }
118  }
119  return res;
120 }
121 
122 /** \interal
123  * \brief does data region intersect with list region 'r'
124  * Takes the max gap into account.
125  */
126 static inline bool RegionsIntersect(const StreamingBuffer *sb, const StreamingBufferConfig *cfg,
127  const StreamingBufferRegion *r, const uint64_t offset, const uint64_t re)
128 {
129  /* create the data range for the region, adding the max gap */
130  const uint64_t reg_o =
131  r->stream_offset > cfg->region_gap ? (r->stream_offset - cfg->region_gap) : 0;
132  const uint64_t reg_re = r->stream_offset + r->buf_size + cfg->region_gap;
133  SCLogDebug("r %p: %" PRIu64 "/%" PRIu64 " - adjusted %" PRIu64 "/%" PRIu64, r, r->stream_offset,
134  r->stream_offset + r->buf_size, reg_o, reg_re);
135  /* check if data range intersects with region range */
136  if (offset >= reg_o && offset <= reg_re) {
137  SCLogDebug("r %p is in-scope", r);
138  return true;
139  }
140  if (re >= reg_o && re <= reg_re) {
141  SCLogDebug("r %p is in-scope: %" PRIu64 " >= %" PRIu64 " && %" PRIu64 " <= %" PRIu64, r, re,
142  reg_o, re, reg_re);
143  return true;
144  }
145  SCLogDebug("r %p is out of scope: %" PRIu64 "/%" PRIu64, r, offset, re);
146  return false;
147 }
148 
149 /** \internal
150  * \brief find the first region for merging.
151  */
152 static StreamingBufferRegion *FindFirstRegionForOffset(const StreamingBuffer *sb,
153  const StreamingBufferConfig *cfg, StreamingBufferRegion *r, const uint64_t offset,
154  const uint32_t len, StreamingBufferRegion **prev)
155 {
156  const uint64_t data_re = offset + len;
157  SCLogDebug("looking for first region matching %" PRIu64 "/%" PRIu64, offset, data_re);
158 
159  StreamingBufferRegion *p = NULL;
160  for (; r != NULL; r = r->next) {
161  if (RegionsIntersect(sb, cfg, r, offset, data_re) == true) {
162  *prev = p;
163  return r;
164  }
165  p = r;
166  }
167  *prev = NULL;
168  return NULL;
169 }
170 
171 static StreamingBufferRegion *FindLargestRegionForOffset(const StreamingBuffer *sb,
172  const StreamingBufferConfig *cfg, StreamingBufferRegion *r, const uint64_t offset,
173  const uint32_t len)
174 {
175  const uint64_t data_re = offset + len;
176  SCLogDebug("starting at %p/%" PRIu64 ", offset %" PRIu64 ", data_re %" PRIu64, r,
177  r->stream_offset, offset, data_re);
178  StreamingBufferRegion *candidate = r;
179  for (; r != NULL; r = r->next) {
180 #ifdef DEBUG
181  const uint64_t reg_re = r->stream_offset + r->buf_size;
182  SCLogDebug("checking: %p/%" PRIu64 "/%" PRIu64 ", offset %" PRIu64 "/%" PRIu64, r,
183  r->stream_offset, reg_re, offset, data_re);
184 #endif
185  if (!RegionsIntersect(sb, cfg, r, offset, data_re))
186  return candidate;
187 
188  if (r->buf_size > candidate->buf_size) {
189  SCLogDebug("candidate %p as size %u > %u", candidate, r->buf_size, candidate->buf_size);
190  candidate = r;
191  }
192  }
193  return candidate;
194 }
195 
196 static StreamingBufferRegion *FindRightEdge(const StreamingBuffer *sb,
197  const StreamingBufferConfig *cfg, StreamingBufferRegion *r, const uint64_t offset,
198  const uint32_t len)
199 {
200  const uint64_t data_re = offset + len;
201  StreamingBufferRegion *candidate = r;
202  for (; r != NULL; r = r->next) {
203  if (!RegionsIntersect(sb, cfg, r, offset, data_re)) {
204  SCLogDebug(
205  "r %p is out of scope: %" PRIu64 "/%u/%" PRIu64, r, offset, len, offset + len);
206  return candidate;
207  }
208  candidate = r;
209  }
210  return candidate;
211 }
212 
213 /** \note uses sc_errno to give error details */
214 static inline StreamingBufferRegion *InitBufferRegion(
215  StreamingBuffer *sb, const StreamingBufferConfig *cfg, const uint32_t min_size)
216 {
217  if (sb->regions == USHRT_MAX || (cfg->max_regions != 0 && sb->regions >= cfg->max_regions)) {
218  SCLogDebug("max regions reached");
219  sc_errno = SC_ELIMIT;
220  return NULL;
221  }
222 
223  StreamingBufferRegion *aux_r = CALLOC(cfg, 1, sizeof(*aux_r));
224  if (aux_r == NULL) {
225  return NULL;
226  }
227 
228  aux_r->buf = CALLOC(cfg, 1, MAX(cfg->buf_size, min_size));
229  if (aux_r->buf == NULL) {
230  FREE(cfg, aux_r, sizeof(*aux_r));
231  return NULL;
232  }
233  aux_r->buf_size = MAX(cfg->buf_size, min_size);
234  sb->regions++;
235  sb->max_regions = MAX(sb->regions, sb->max_regions);
236  return aux_r;
237 }
238 
239 static inline int InitBuffer(StreamingBuffer *sb, const StreamingBufferConfig *cfg)
240 {
241  sb->region.buf = CALLOC(cfg, 1, cfg->buf_size);
242  if (sb->region.buf == NULL) {
243  return sc_errno;
244  }
245  sb->region.buf_size = cfg->buf_size;
246  return SC_OK;
247 }
248 
249 StreamingBuffer *StreamingBufferInit(const StreamingBufferConfig *cfg)
250 {
251  StreamingBuffer *sb = CALLOC(cfg, 1, sizeof(StreamingBuffer));
252  if (sb == NULL) {
253  return NULL;
254  }
255 
256  sb->region.buf_size = cfg->buf_size;
257  sb->regions = sb->max_regions = 1;
258 
259  if (cfg->buf_size == 0) {
260  return sb;
261  }
262 
263  int r = InitBuffer(sb, cfg);
264  if (r != SC_OK) {
265  FREE(cfg, sb, sizeof(StreamingBuffer));
266  sc_errno = r;
267  return NULL;
268  }
269  return sb;
270 }
271 
272 void StreamingBufferClear(StreamingBuffer *sb, const StreamingBufferConfig *cfg)
273 {
274  if (sb != NULL) {
275  SCLogDebug("sb->region.buf_size %u max %u", sb->region.buf_size, sb->buf_size_max);
276 
277  SBBFree(sb, cfg);
278  ListRegions(sb);
279  if (sb->region.buf != NULL) {
280  FREE(cfg, sb->region.buf, sb->region.buf_size);
281  sb->region.buf = NULL;
282  }
283 
284  for (StreamingBufferRegion *r = sb->region.next; r != NULL;) {
285  StreamingBufferRegion *next = r->next;
286  FREE(cfg, r->buf, r->buf_size);
287  FREE(cfg, r, sizeof(*r));
288  r = next;
289  }
290  sb->region.next = NULL;
291  sb->regions = sb->max_regions = 1;
292  }
293 }
294 
295 void StreamingBufferFree(StreamingBuffer *sb, const StreamingBufferConfig *cfg)
296 {
297  if (sb != NULL) {
298  StreamingBufferClear(sb, cfg);
299  FREE(cfg, sb, sizeof(StreamingBuffer));
300  }
301 }
302 
303 #ifdef DEBUG
304 static void SBBPrintList(StreamingBuffer *sb)
305 {
306  StreamingBufferBlock *sbb = NULL;
307  RB_FOREACH(sbb, SBB, &sb->sbb_tree) {
308  SCLogDebug("sbb: offset %" PRIu64 ", len %u", sbb->offset, sbb->len);
309  StreamingBufferBlock *next = SBB_RB_NEXT(sbb);
310  if (next) {
311  if ((sbb->offset + sbb->len) != next->offset) {
312  SCLogDebug("gap: offset %" PRIu64 ", len %" PRIu64, (sbb->offset + sbb->len),
313  next->offset - (sbb->offset + sbb->len));
314  }
315  }
316  }
317 }
318 #endif
319 
320 /* setup with gap between 2 blocks
321  *
322  * [block][gap][block]
323  **/
324 static int WARN_UNUSED SBBInit(StreamingBuffer *sb, const StreamingBufferConfig *cfg,
325  StreamingBufferRegion *region, uint32_t rel_offset, uint32_t data_len)
326 {
327  DEBUG_VALIDATE_BUG_ON(!RB_EMPTY(&sb->sbb_tree));
328  DEBUG_VALIDATE_BUG_ON(region->buf_offset > region->stream_offset + rel_offset);
329 
330  /* need to set up 2: existing data block and new data block */
331  StreamingBufferBlock *sbb = CALLOC(cfg, 1, sizeof(*sbb));
332  if (sbb == NULL) {
333  return sc_errno;
334  }
335  sbb->offset = sb->region.stream_offset;
336  sbb->len = sb->region.buf_offset;
337  (void)SBB_RB_INSERT(&sb->sbb_tree, sbb);
338  sb->head = sbb;
339  sb->sbb_size = sbb->len;
340 
341  StreamingBufferBlock *sbb2 = CALLOC(cfg, 1, sizeof(*sbb2));
342  if (sbb2 == NULL) {
343  return sc_errno;
344  }
345  sbb2->offset = region->stream_offset + rel_offset;
346  sbb2->len = data_len;
347  DEBUG_VALIDATE_BUG_ON(sbb2->offset < sbb->len);
348  SCLogDebug("sbb1 %" PRIu64 ", len %u, sbb2 %" PRIu64 ", len %u", sbb->offset, sbb->len,
349  sbb2->offset, sbb2->len);
350 
351  sb->sbb_size += sbb2->len;
352  if (SBB_RB_INSERT(&sb->sbb_tree, sbb2) != NULL) {
353  FREE(cfg, sbb2, sizeof(*sbb2));
354  return SC_EINVAL;
355  }
356 
357 #ifdef DEBUG
358  SBBPrintList(sb);
359 #endif
360  return SC_OK;
361 }
362 
363 /* setup with leading gap
364  *
365  * [gap][block]
366  **/
367 static int WARN_UNUSED SBBInitLeadingGap(StreamingBuffer *sb, const StreamingBufferConfig *cfg,
368  StreamingBufferRegion *region, uint64_t offset, uint32_t data_len)
369 {
370  DEBUG_VALIDATE_BUG_ON(!RB_EMPTY(&sb->sbb_tree));
371 
372  StreamingBufferBlock *sbb = CALLOC(cfg, 1, sizeof(*sbb));
373  if (sbb == NULL) {
374  return sc_errno;
375  }
376  sbb->offset = offset;
377  sbb->len = data_len;
378 
379  sb->head = sbb;
380  sb->sbb_size = sbb->len;
381  (void)SBB_RB_INSERT(&sb->sbb_tree, sbb);
382 
383  SCLogDebug("sbb %" PRIu64 ", len %u", sbb->offset, sbb->len);
384 #ifdef DEBUG
385  SBBPrintList(sb);
386 #endif
387  return SC_OK;
388 }
389 
390 static inline void ConsolidateFwd(StreamingBuffer *sb, const StreamingBufferConfig *cfg,
391  StreamingBufferRegion *region, struct SBB *tree, StreamingBufferBlock *sa)
392 {
393  uint64_t sa_re = sa->offset + sa->len;
394  StreamingBufferBlock *tr, *s = sa;
395  RB_FOREACH_FROM(tr, SBB, s) {
396  if (sa == tr)
397  continue;
398 
399  const uint64_t tr_re = tr->offset + tr->len;
400  SCLogDebug("-> (fwd) tr %p %" PRIu64 "/%u re %" PRIu64, tr, tr->offset, tr->len, tr_re);
401 
402  if (sa_re < tr->offset) {
403  SCLogDebug("entirely before: %" PRIu64 " < %" PRIu64, sa_re, tr->offset);
404  break; // entirely before
405  }
406 
407  /* new block (sa) entirely eclipsed by in tree block (tr)
408  sa: [ ]
409  tr: [ ]
410  sa: [ ]
411  tr: [ ]
412  sa: [ ]
413  tr: [ ]
414  */
415  if (sa->offset >= tr->offset && sa_re <= tr_re) {
416  sb->sbb_size -= sa->len;
417  sa->len = tr->len;
418  sa->offset = tr->offset;
419  sa_re = sa->offset + sa->len;
420  SCLogDebug("-> (fwd) tr %p %" PRIu64 "/%u REMOVED ECLIPSED (sa overlapped by tr)", tr,
421  tr->offset, tr->len);
422  SBB_RB_REMOVE(tree, tr);
423  FREE(cfg, tr, sizeof(StreamingBufferBlock));
424  /* new block (sa) entire eclipses in tree block (tr)
425  sa: [ ]
426  tr: [ ]
427  sa: [ ]
428  tr: [ ]
429  sa: [ ]
430  tr: [ ]
431  */
432  } else if (sa->offset <= tr->offset && sa_re >= tr_re) {
433  SCLogDebug("-> (fwd) tr %p %" PRIu64 "/%u REMOVED ECLIPSED (tr overlapped by sa)", tr,
434  tr->offset, tr->len);
435  SBB_RB_REMOVE(tree, tr);
436  sb->sbb_size -= tr->len;
437  FREE(cfg, tr, sizeof(StreamingBufferBlock));
438 
439  SCLogDebug("-> (fwd) tr %p %" PRIu64 "/%u region %p so %" PRIu64 " bo %u sz %u", sa,
440  sa->offset, sa->len, region, region->stream_offset, region->buf_offset,
441  region->buf_size);
442  if (sa->offset == region->stream_offset &&
443  sa_re > (region->stream_offset + region->buf_offset)) {
444  DEBUG_VALIDATE_BUG_ON(sa_re < region->stream_offset);
445  region->buf_offset = sa_re - region->stream_offset;
446  SCLogDebug("-> (fwd) tr %p %" PRIu64 "/%u region %p so %" PRIu64
447  " bo %u sz %u BUF_OFFSET UPDATED",
448  sa, sa->offset, sa->len, region, region->stream_offset, region->buf_offset,
449  region->buf_size);
450  }
451  /* new block (sa) extended by in tree block (tr)
452  sa: [ ]
453  tr: [ ]
454  sa: [ ]
455  tr: [ ]
456  */
457  } else if (sa->offset < tr->offset && // starts before
458  sa_re >= tr->offset && sa_re < tr_re) // ends inside
459  {
460  // merge. sb->sbb_size includes both so we need to adjust that too.
461  uint32_t combined_len = sa->len + tr->len;
462  sa->len = tr_re - sa->offset;
463  sa_re = sa->offset + sa->len;
464  SCLogDebug("-> (fwd) tr %p %" PRIu64 "/%u REMOVED MERGED", tr, tr->offset, tr->len);
465  SBB_RB_REMOVE(tree, tr);
466  sb->sbb_size -= (combined_len - sa->len); // remove what we added twice
467  FREE(cfg, tr, sizeof(StreamingBufferBlock));
468  SCLogDebug("-> (fwd) tr %p %" PRIu64 "/%u RESULT", sa, sa->offset, sa->len);
469  SCLogDebug("-> (fwd) tr %p %" PRIu64 "/%u region %p so %" PRIu64 " bo %u sz %u", sa,
470  sa->offset, sa->len, region, region->stream_offset, region->buf_offset,
471  region->buf_size);
472  if (sa->offset == region->stream_offset &&
473  sa_re > (region->stream_offset + region->buf_offset)) {
474  DEBUG_VALIDATE_BUG_ON(sa_re < region->stream_offset);
475  region->buf_offset = sa_re - region->stream_offset;
476  SCLogDebug("-> (fwd) tr %p %" PRIu64 "/%u region %p so %" PRIu64
477  " bo %u sz %u BUF_OFFSET UPDATED",
478  sa, sa->offset, sa->len, region, region->stream_offset, region->buf_offset,
479  region->buf_size);
480  }
481  }
482  }
483 }
484 
485 static inline void ConsolidateBackward(StreamingBuffer *sb, const StreamingBufferConfig *cfg,
486  StreamingBufferRegion *region, struct SBB *tree, StreamingBufferBlock *sa)
487 {
488  uint64_t sa_re = sa->offset + sa->len;
489  StreamingBufferBlock *tr, *s = sa;
490  RB_FOREACH_REVERSE_FROM(tr, SBB, s) {
491  if (sa == tr)
492  continue;
493  const uint64_t tr_re = tr->offset + tr->len;
494  SCLogDebug("-> (bwd) tr %p %" PRIu64 "/%u", tr, tr->offset, tr->len);
495 
496  if (sa->offset > tr_re)
497  break; // entirely after
498 
499  /* new block (sa) entirely eclipsed by in tree block (tr)
500  sa: [ ]
501  tr: [ ]
502  sa: [ ]
503  tr: [ ]
504  sa: [ ]
505  tr: [ ]
506  */
507  if (sa->offset >= tr->offset && sa_re <= tr_re) {
508  sb->sbb_size -= sa->len; // sa entirely eclipsed so remove double accounting
509  sa->len = tr->len;
510  sa->offset = tr->offset;
511  sa_re = sa->offset + sa->len;
512  SCLogDebug("-> (bwd) tr %p %" PRIu64 "/%u REMOVED ECLIPSED (sa overlapped by tr)", tr,
513  tr->offset, tr->len);
514  if (sb->head == tr)
515  sb->head = sa;
516  SBB_RB_REMOVE(tree, tr);
517  FREE(cfg, tr, sizeof(StreamingBufferBlock));
518  /* new block (sa) entire eclipses in tree block (tr)
519  sa: [ ]
520  tr: [ ]
521  sa: [ ]
522  tr: [ ]
523  sa: [ ]
524  tr: [ ]
525  */
526  } else if (sa->offset <= tr->offset && sa_re >= tr_re) {
527  SCLogDebug("-> (bwd) tr %p %" PRIu64 "/%u REMOVED ECLIPSED (tr overlapped by sa)", tr,
528  tr->offset, tr->len);
529  if (sb->head == tr)
530  sb->head = sa;
531  SBB_RB_REMOVE(tree, tr);
532  sb->sbb_size -= tr->len; // tr entirely eclipsed so remove double accounting
533  FREE(cfg, tr, sizeof(StreamingBufferBlock));
534 
535  SCLogDebug("-> (bwd) tr %p %" PRIu64 "/%u region %p so %" PRIu64 " bo %u sz %u", sa,
536  sa->offset, sa->len, region, region->stream_offset, region->buf_offset,
537  region->buf_size);
538 
539  if (sa->offset == region->stream_offset &&
540  sa_re > (region->stream_offset + region->buf_offset)) {
541  DEBUG_VALIDATE_BUG_ON(sa_re < region->stream_offset);
542  region->buf_offset = sa_re - region->stream_offset;
543  SCLogDebug("-> (bwd) tr %p %" PRIu64 "/%u region %p so %" PRIu64
544  " bo %u sz %u BUF_OFFSET UPDATED",
545  sa, sa->offset, sa->len, region, region->stream_offset, region->buf_offset,
546  region->buf_size);
547  }
548 
549  /* new block (sa) extends in tree block (tr)
550  sa: [ ]
551  tr: [ ]
552  sa: [ ]
553  tr: [ ]
554  */
555  } else if (sa->offset > tr->offset && sa_re > tr_re && sa->offset <= tr_re) {
556  // merge. sb->sbb_size includes both so we need to adjust that too.
557  uint32_t combined_len = sa->len + tr->len;
558  sa->len = sa_re - tr->offset;
559  sa->offset = tr->offset;
560  sa_re = sa->offset + sa->len;
561  SCLogDebug("-> (bwd) tr %p %" PRIu64 "/%u REMOVED MERGED", tr, tr->offset, tr->len);
562  if (sb->head == tr)
563  sb->head = sa;
564  SBB_RB_REMOVE(tree, tr);
565  sb->sbb_size -= (combined_len - sa->len); // remove what we added twice
566  FREE(cfg, tr, sizeof(StreamingBufferBlock));
567 
568  SCLogDebug("-> (bwd) tr %p %" PRIu64 "/%u region %p so %" PRIu64 " bo %u sz %u", sa,
569  sa->offset, sa->len, region, region->stream_offset, region->buf_offset,
570  region->buf_size);
571  if (sa->offset == region->stream_offset &&
572  sa_re > (region->stream_offset + region->buf_offset)) {
573  DEBUG_VALIDATE_BUG_ON(sa_re < region->stream_offset);
574  region->buf_offset = sa_re - region->stream_offset;
575  SCLogDebug("-> (bwd) tr %p %" PRIu64 "/%u region %p so %" PRIu64
576  " bo %u sz %u BUF_OFFSET UPDATED",
577  sa, sa->offset, sa->len, region, region->stream_offset, region->buf_offset,
578  region->buf_size);
579  }
580  }
581  }
582 }
583 
584 /** \internal
585  * \param region the region that holds the new data
586  */
587 static int SBBUpdate(StreamingBuffer *sb, const StreamingBufferConfig *cfg,
588  StreamingBufferRegion *region, uint32_t rel_offset, uint32_t len)
589 {
590  struct SBB *tree = &sb->sbb_tree;
591  SCLogDebug("* inserting: %u/%u", rel_offset, len);
592 
593  StreamingBufferBlock *sbb = CALLOC(cfg, 1, sizeof(*sbb));
594  if (sbb == NULL) {
595  return sc_errno;
596  }
597  sbb->offset = region->stream_offset + rel_offset;
598  sbb->len = len;
599 
600  StreamingBufferBlock *res = SBB_RB_INSERT(tree, sbb);
601  if (res) {
602  // exact overlap
603  SCLogDebug("* insert failed: exact match in tree with %p %" PRIu64 "/%u", res, res->offset,
604  res->len);
605  FREE(cfg, sbb, sizeof(StreamingBufferBlock));
606  return SC_OK;
607  }
608  sb->sbb_size += len; // may adjust based on consolidation below
609 
610  /* handle backwards and forwards overlaps */
611  if (SBB_RB_PREV(sbb) == NULL) {
612  sb->head = sbb;
613  } else {
614  ConsolidateBackward(sb, cfg, region, tree, sbb);
615  }
616  ConsolidateFwd(sb, cfg, region, tree, sbb);
617 #ifdef DEBUG
618  SBBPrintList(sb);
619 #endif
620  if (sbb->offset == region->stream_offset) {
621  SCLogDebug("insert at region head");
622  region->buf_offset = sbb->len;
623  }
624  return SC_OK;
625 }
626 
627 static void SBBFree(StreamingBuffer *sb, const StreamingBufferConfig *cfg)
628 {
629  StreamingBufferBlock *sbb = NULL, *safe = NULL;
630  RB_FOREACH_SAFE(sbb, SBB, &sb->sbb_tree, safe) {
631  SBB_RB_REMOVE(&sb->sbb_tree, sbb);
632  sb->sbb_size -= sbb->len;
633  FREE(cfg, sbb, sizeof(StreamingBufferBlock));
634  }
635  sb->head = NULL;
636 }
637 
638 static void SBBPrune(StreamingBuffer *sb, const StreamingBufferConfig *cfg)
639 {
640  SCLogDebug("pruning %p to %" PRIu64, sb, sb->region.stream_offset);
641  StreamingBufferBlock *sbb = NULL, *safe = NULL;
642  RB_FOREACH_SAFE(sbb, SBB, &sb->sbb_tree, safe) {
643  /* completely beyond window, we're done */
644  if (sbb->offset >= sb->region.stream_offset) {
645  sb->head = sbb;
646  if (sbb->offset == sb->region.stream_offset) {
647  SCLogDebug("set buf_offset?");
648  if (sbb->offset == sb->region.stream_offset) {
649  SCLogDebug("set buf_offset to first sbb len %u", sbb->len);
650  DEBUG_VALIDATE_BUG_ON(sbb->len > sb->region.buf_size);
651  sb->region.buf_offset = sbb->len;
652  }
653  }
654  break;
655  }
656 
657  /* partly before, partly beyond. Adjust */
658  if (sbb->offset < sb->region.stream_offset &&
659  sbb->offset + sbb->len > sb->region.stream_offset) {
660  uint32_t shrink_by = sb->region.stream_offset - sbb->offset;
661  DEBUG_VALIDATE_BUG_ON(shrink_by > sbb->len);
662  if (sbb->len >= shrink_by) {
663  sbb->len -= shrink_by;
664  sbb->offset += shrink_by;
665  sb->sbb_size -= shrink_by;
666  SCLogDebug("shrunk by %u", shrink_by);
667  DEBUG_VALIDATE_BUG_ON(sbb->offset != sb->region.stream_offset);
668  }
669  sb->head = sbb;
670  if (sbb->offset == sb->region.stream_offset) {
671  SCLogDebug("set buf_offset to first sbb len %u", sbb->len);
672  DEBUG_VALIDATE_BUG_ON(sbb->len > sb->region.buf_size);
673  sb->region.buf_offset = sbb->len;
674  }
675  break;
676  }
677 
678  SBB_RB_REMOVE(&sb->sbb_tree, sbb);
679  /* either we set it again for the next sbb, or there isn't any */
680  sb->head = NULL;
681  sb->sbb_size -= sbb->len;
682  SCLogDebug("sb %p removed %p %" PRIu64 ", %u", sb, sbb, sbb->offset, sbb->len);
683  FREE(cfg, sbb, sizeof(StreamingBufferBlock));
684  }
685 #ifdef DEBUG
686  SBBPrintList(sb);
687 #endif
688 }
689 
690 static inline uint32_t ToNextMultipleOf(const uint32_t in, const uint32_t m)
691 {
692  uint32_t r = in;
693  if (m > 0) {
694  const uint32_t x = in % m;
695  if (x != 0) {
696  r = (in - x) + m;
697  }
698  }
699  return r;
700 }
701 
702 static thread_local bool g2s_warn_once = false;
703 
704 static inline int WARN_UNUSED GrowRegionToSize(StreamingBuffer *sb,
705  const StreamingBufferConfig *cfg, StreamingBufferRegion *region, const uint32_t size)
706 {
707  DEBUG_VALIDATE_BUG_ON(region->buf_size > BIT_U32(30));
708  if (size > BIT_U32(30)) { // 1GiB
709  if (!g2s_warn_once) {
710  SCLogWarning("StreamingBuffer::GrowRegionToSize() tried to alloc %u bytes, exceeds "
711  "limit of %lu",
712  size, BIT_U32(30));
713  g2s_warn_once = true;
714  }
715  return SC_ELIMIT;
716  }
717 
718  /* try to grow in multiples of cfg->buf_size */
719  const uint32_t grow = ToNextMultipleOf(size, cfg->buf_size);
720  SCLogDebug("grow %u", grow);
721 
722  void *ptr = REALLOC(cfg, region->buf, region->buf_size, grow);
723  if (ptr == NULL) {
724  return sc_errno;
725  }
726  /* for safe printing and general caution, lets memset the
727  * new data to 0 */
728  size_t diff = grow - region->buf_size;
729  void *new_mem = ((char *)ptr) + region->buf_size;
730  memset(new_mem, 0, diff);
731 
732  region->buf = ptr;
733  region->buf_size = grow;
734  SCLogDebug("grown buffer to %u", grow);
735 #ifdef DEBUG
736  if (region->buf_size > sb->buf_size_max) {
737  sb->buf_size_max = region->buf_size;
738  }
739 #endif
740  return SC_OK;
741 }
742 
743 static int WARN_UNUSED GrowToSize(
744  StreamingBuffer *sb, const StreamingBufferConfig *cfg, uint32_t size)
745 {
746  return GrowRegionToSize(sb, cfg, &sb->region, size);
747 }
748 
749 static inline bool RegionBeforeOffset(const StreamingBufferRegion *r, const uint64_t o)
750 {
751  return (r->stream_offset + r->buf_size <= o);
752 }
753 
754 static inline bool RegionContainsOffset(const StreamingBufferRegion *r, const uint64_t o)
755 {
756  return (o >= r->stream_offset && (r->stream_offset + r->buf_size) >= o);
757 }
758 
759 /** \internal
760  * \brief slide to offset for regions
761  *
762  *
763  * [ main ] [ gap ] [ aux ] [ gap ] [ aux ]
764  * ^
765  * - main reset to 0
766  *
767  *
768  * [ main ] [ gap ] [ aux ] [ gap ] [ aux ]
769  * ^
770  * - main shift
771  *
772  * [ main ] [ gap ] [ aux ] [ gap ] [ aux ]
773  * ^
774  * - main reset
775  * - move aux into main
776  * - free aux
777  * - shift
778  *
779  * [ main ] [ gap ] [ aux ] [ gap ] [ aux ]
780  * ^
781  * - main reset
782  * - move aux into main
783  * - free aux
784  * - no shift
785  */
786 static inline void StreamingBufferSlideToOffsetWithRegions(
787  StreamingBuffer *sb, const StreamingBufferConfig *cfg, const uint64_t slide_offset)
788 {
789  ListRegions(sb);
790  DEBUG_VALIDATE_BUG_ON(slide_offset == sb->region.stream_offset);
791 
792  SCLogDebug("slide_offset %" PRIu64, slide_offset);
793  SCLogDebug("main: offset %" PRIu64 " buf %p size %u offset %u", sb->region.stream_offset,
794  sb->region.buf, sb->region.buf_size, sb->region.buf_offset);
795 
796  StreamingBufferRegion *to_shift = NULL;
797  const bool main_is_oow = RegionBeforeOffset(&sb->region, slide_offset);
798  if (main_is_oow) {
799  SCLogDebug("main_is_oow");
800  if (sb->region.buf != NULL) {
801  SCLogDebug("clearing main");
802  FREE(cfg, sb->region.buf, sb->region.buf_size);
803  sb->region.buf = NULL;
804  sb->region.buf_size = 0;
805  sb->region.buf_offset = 0;
806  sb->region.stream_offset = slide_offset;
807  } else {
808  sb->region.stream_offset = slide_offset;
809  }
810 
811  /* remove regions that are out of window & select the region to
812  * become the new main */
813  StreamingBufferRegion *prev = &sb->region;
814  for (StreamingBufferRegion *r = sb->region.next; r != NULL;) {
815  SCLogDebug("r %p so %" PRIu64 ", re %" PRIu64, r, r->stream_offset,
816  r->stream_offset + r->buf_offset);
817  StreamingBufferRegion *next = r->next;
818  if (RegionBeforeOffset(r, slide_offset)) {
819  SCLogDebug("r %p so %" PRIu64 ", re %" PRIu64 " -> before", r, r->stream_offset,
820  r->stream_offset + r->buf_offset);
821  DEBUG_VALIDATE_BUG_ON(r == &sb->region);
822  prev->next = next;
823 
824  FREE(cfg, r->buf, r->buf_size);
825  FREE(cfg, r, sizeof(*r));
826  sb->regions--;
827  DEBUG_VALIDATE_BUG_ON(sb->regions == 0);
828  } else if (RegionContainsOffset(r, slide_offset)) {
829  SCLogDebug("r %p so %" PRIu64 ", re %" PRIu64 " -> within", r, r->stream_offset,
830  r->stream_offset + r->buf_offset);
831  /* remove from list, we will xfer contents to main below */
832  prev->next = next;
833  to_shift = r;
834  break;
835  } else {
836  SCLogDebug("r %p so %" PRIu64 ", re %" PRIu64 " -> post", r, r->stream_offset,
837  r->stream_offset + r->buf_offset);
838  /* implied beyond slide offset */
839  DEBUG_VALIDATE_BUG_ON(r->stream_offset < slide_offset);
840  break;
841  }
842  r = next;
843  }
844  SCLogDebug("to_shift %p", to_shift);
845  } else {
846  to_shift = &sb->region;
847  SCLogDebug("shift start region %p", to_shift);
848  }
849 
850  // this region is main, or will xfer its buffer to main
851  if (to_shift) {
852  SCLogDebug("main: offset %" PRIu64 " buf %p size %u offset %u", to_shift->stream_offset,
853  to_shift->buf, to_shift->buf_size, to_shift->buf_offset);
854  if (to_shift != &sb->region) {
855  DEBUG_VALIDATE_BUG_ON(sb->region.buf != NULL);
856 
857  sb->region.buf = to_shift->buf;
858  sb->region.stream_offset = to_shift->stream_offset;
859  sb->region.buf_offset = to_shift->buf_offset;
860  sb->region.buf_size = to_shift->buf_size;
861  sb->region.next = to_shift->next;
862 
863  FREE(cfg, to_shift, sizeof(*to_shift));
864  to_shift = &sb->region;
865  sb->regions--;
866  DEBUG_VALIDATE_BUG_ON(sb->regions == 0);
867  }
868 
869  // Do the shift. If new region is exactly at the slide offset we can skip this.
870  DEBUG_VALIDATE_BUG_ON(to_shift->stream_offset > slide_offset);
871  const uint32_t s = slide_offset - to_shift->stream_offset;
872  if (s > 0) {
873  const uint32_t new_data_size = to_shift->buf_size - s;
874  uint32_t new_mem_size = ToNextMultipleOf(new_data_size, cfg->buf_size);
875 
876  /* see if after the slide we'd overlap with the next region. If so, we need
877  * to consolidate them into one. Error handling is a bit peculiar. We need
878  * to grow a region, move data from another region into it, then free the
879  * other region. This could fail if we're close to the memcap. In this case
880  * we relax our rounding up logic so we only shrink and don't merge the 2
881  * regions after all. */
882  if (to_shift->next && slide_offset + new_mem_size >= to_shift->next->stream_offset) {
883  StreamingBufferRegion *start = to_shift;
884  StreamingBufferRegion *next = start->next;
885  const uint64_t next_re = next->stream_offset + next->buf_size;
886  const uint32_t mem_size = ToNextMultipleOf(next_re - slide_offset, cfg->buf_size);
887 
888  /* using next as the new main */
889  if (start->buf_size < next->buf_size) {
890  SCLogDebug("replace main with the next bigger region");
891 
892  const uint32_t next_data_offset = next->stream_offset - slide_offset;
893  const uint32_t prev_buf_size = next->buf_size;
894  const uint32_t start_data_offset = slide_offset - start->stream_offset;
895  DEBUG_VALIDATE_BUG_ON(start_data_offset > start->buf_size);
896  if (start_data_offset > start->buf_size) {
897  new_mem_size = new_data_size;
898  goto just_main;
899  }
900  /* expand "next" to include relevant part of "start" */
901  if (GrowRegionToSize(sb, cfg, next, mem_size) != 0) {
902  new_mem_size = new_data_size;
903  goto just_main;
904  }
905  SCLogDebug("region now sized %u", mem_size);
906 
907  // slide "next":
908  // pre-grow: [nextnextnext]
909  // post-grow [nextnextnextXXX]
910  // post-move [XXXnextnextnext]
911  memmove(next->buf + next_data_offset, next->buf, prev_buf_size);
912 
913  // move portion of "start" into "next"
914  //
915  // start: [ooooookkkkk] (o: old, k: keep)
916  // pre-next [xxxxxnextnextnext]
917  // post-next [kkkkknextnextnext]
918  const uint32_t start_data_size = start->buf_size - start_data_offset;
919  memcpy(next->buf, start->buf + start_data_offset, start_data_size);
920 
921  // free "start"s buffer, we will use the one from "next"
922  FREE(cfg, start->buf, start->buf_size);
923 
924  // update "main" to use "next"
925  start->stream_offset = slide_offset;
926  start->buf = next->buf;
927  start->buf_size = next->buf_size;
928  start->next = next->next;
929 
930  // free "next"
931  FREE(cfg, next, sizeof(*next));
932  sb->regions--;
933  DEBUG_VALIDATE_BUG_ON(sb->regions == 0);
934  goto done;
935  } else {
936  /* using "main", expand to include "next" */
937  if (GrowRegionToSize(sb, cfg, start, mem_size) != 0) {
938  new_mem_size = new_data_size;
939  goto just_main;
940  }
941  SCLogDebug("start->buf now size %u", mem_size);
942 
943  // slide "start"
944  // pre: [xxxxxxxAAA]
945  // post: [AAAxxxxxxx]
946  SCLogDebug("s %u new_data_size %u", s, new_data_size);
947  memmove(start->buf, start->buf + s, new_data_size);
948 
949  // copy in "next"
950  // pre: [AAAxxxxxxx]
951  // [BBBBBBB]
952  // post: [AAABBBBBBB]
953  SCLogDebug("copy next->buf %p/%u to start->buf offset %u", next->buf,
954  next->buf_size, new_data_size);
955  memcpy(start->buf + new_data_size, next->buf, next->buf_size);
956 
957  start->stream_offset = slide_offset;
958  start->next = next->next;
959 
960  // free "next"
961  FREE(cfg, next->buf, next->buf_size);
962  FREE(cfg, next, sizeof(*next));
963  sb->regions--;
964  DEBUG_VALIDATE_BUG_ON(sb->regions == 0);
965  goto done;
966  }
967  }
968 
969  just_main:
970  SCLogDebug("s %u new_data_size %u", s, new_data_size);
971  memmove(to_shift->buf, to_shift->buf + s, new_data_size);
972  /* shrink memory region. If this fails we keep the old */
973  void *ptr = REALLOC(cfg, to_shift->buf, to_shift->buf_size, new_mem_size);
974  if (ptr != NULL) {
975  to_shift->buf = ptr;
976  to_shift->buf_size = new_mem_size;
977  SCLogDebug("new buf_size %u", to_shift->buf_size);
978  }
979  if (s < to_shift->buf_offset)
980  to_shift->buf_offset -= s;
981  else
982  to_shift->buf_offset = 0;
983  to_shift->stream_offset = slide_offset;
984  }
985  }
986 
987 done:
988  SCLogDebug("main: offset %" PRIu64 " buf %p size %u offset %u", sb->region.stream_offset,
989  sb->region.buf, sb->region.buf_size, sb->region.buf_offset);
990  SCLogDebug("end of slide");
991 }
992 
993 /**
994  * \brief slide to absolute offset
995  * \todo if sliding beyond window, we could perhaps reset?
996  */
997 void StreamingBufferSlideToOffset(
998  StreamingBuffer *sb, const StreamingBufferConfig *cfg, uint64_t offset)
999 {
1000  SCLogDebug("sliding to offset %" PRIu64, offset);
1001  ListRegions(sb);
1002 #ifdef DEBUG
1003  SBBPrintList(sb);
1004 #endif
1005 
1006  if (sb->region.next) {
1007  StreamingBufferSlideToOffsetWithRegions(sb, cfg, offset);
1008  SBBPrune(sb, cfg);
1009  SCLogDebug("post SBBPrune");
1010  ListRegions(sb);
1011 #ifdef DEBUG
1012  SBBPrintList(sb);
1013 #endif
1014  DEBUG_VALIDATE_BUG_ON(sb->region.buf != NULL && sb->region.buf_size == 0);
1015  DEBUG_VALIDATE_BUG_ON(sb->region.buf_offset > sb->region.buf_size);
1016  DEBUG_VALIDATE_BUG_ON(offset > sb->region.stream_offset);
1017  DEBUG_VALIDATE_BUG_ON(sb->head && sb->head->offset == sb->region.stream_offset &&
1018  sb->head->len > sb->region.buf_offset);
1019  DEBUG_VALIDATE_BUG_ON(sb->region.stream_offset < offset);
1020  return;
1021  }
1022 
1023  if (offset > sb->region.stream_offset) {
1024  const uint32_t slide = offset - sb->region.stream_offset;
1025  if (sb->head != NULL) {
1026  /* have sbb's, so can't rely on buf_offset for the slide */
1027  if (slide < sb->region.buf_size) {
1028  const uint32_t size = sb->region.buf_size - slide;
1029  SCLogDebug("sliding %u forward, size of original buffer left after slide %u", slide,
1030  size);
1031  memmove(sb->region.buf, sb->region.buf + slide, size);
1032  if (sb->region.buf_offset > slide) {
1033  sb->region.buf_offset -= slide;
1034  } else {
1035  sb->region.buf_offset = 0;
1036  }
1037  } else {
1038  sb->region.buf_offset = 0;
1039  }
1040  sb->region.stream_offset = offset;
1041  } else {
1042  /* no sbb's, so we can use buf_offset */
1043  if (offset <= sb->region.stream_offset + sb->region.buf_offset) {
1044  const uint32_t size = sb->region.buf_offset - slide;
1045  SCLogDebug("sliding %u forward, size of original buffer left after slide %u", slide,
1046  size);
1047  memmove(sb->region.buf, sb->region.buf + slide, size);
1048  sb->region.stream_offset = offset;
1049  sb->region.buf_offset = size;
1050  } else {
1051  /* moved past all data */
1052  sb->region.stream_offset = offset;
1053  sb->region.buf_offset = 0;
1054  }
1055  }
1056  SBBPrune(sb, cfg);
1057  }
1058 #ifdef DEBUG
1059  SBBPrintList(sb);
1060 #endif
1061  DEBUG_VALIDATE_BUG_ON(sb->region.stream_offset < offset);
1062 }
1063 
1064 #define DATA_FITS(sb, len) ((sb)->region.buf_offset + (len) <= (sb)->region.buf_size)
1065 
1066 int StreamingBufferAppend(StreamingBuffer *sb, const StreamingBufferConfig *cfg,
1067  StreamingBufferSegment *seg, const uint8_t *data, uint32_t data_len)
1068 {
1069  DEBUG_VALIDATE_BUG_ON(seg == NULL);
1070 
1071  if (sb->region.buf == NULL) {
1072  if (InitBuffer(sb, cfg) == -1)
1073  return -1;
1074  }
1075 
1076  if (!DATA_FITS(sb, data_len)) {
1077  if (sb->region.buf_size == 0) {
1078  if (GrowToSize(sb, cfg, data_len) != SC_OK)
1079  return -1;
1080  } else {
1081  if (GrowToSize(sb, cfg, sb->region.buf_offset + data_len) != SC_OK)
1082  return -1;
1083  }
1084  }
1085  DEBUG_VALIDATE_BUG_ON(!DATA_FITS(sb, data_len));
1086 
1087  memcpy(sb->region.buf + sb->region.buf_offset, data, data_len);
1088  seg->stream_offset = sb->region.stream_offset + sb->region.buf_offset;
1089  seg->segment_len = data_len;
1090  uint32_t rel_offset = sb->region.buf_offset;
1091  sb->region.buf_offset += data_len;
1092 
1093  if (!RB_EMPTY(&sb->sbb_tree)) {
1094  return SBBUpdate(sb, cfg, &sb->region, rel_offset, data_len);
1095  } else {
1096  return 0;
1097  }
1098 }
1099 
1100 /**
1101  * \brief add data w/o tracking a segment
1102  */
1103 int StreamingBufferAppendNoTrack(StreamingBuffer *sb, const StreamingBufferConfig *cfg,
1104  const uint8_t *data, uint32_t data_len)
1105 {
1106  if (sb->region.buf == NULL) {
1107  if (InitBuffer(sb, cfg) == -1)
1108  return -1;
1109  }
1110 
1111  if (!DATA_FITS(sb, data_len)) {
1112  if (sb->region.buf_size == 0) {
1113  if (GrowToSize(sb, cfg, data_len) != SC_OK)
1114  return -1;
1115  } else {
1116  if (GrowToSize(sb, cfg, sb->region.buf_offset + data_len) != SC_OK)
1117  return -1;
1118  }
1119  }
1120  DEBUG_VALIDATE_BUG_ON(!DATA_FITS(sb, data_len));
1121 
1122  memcpy(sb->region.buf + sb->region.buf_offset, data, data_len);
1123  uint32_t rel_offset = sb->region.buf_offset;
1124  sb->region.buf_offset += data_len;
1125 
1126  if (!RB_EMPTY(&sb->sbb_tree)) {
1127  return SBBUpdate(sb, cfg, &sb->region, rel_offset, data_len);
1128  } else {
1129  return 0;
1130  }
1131 }
1132 
1133 #define DATA_FITS_AT_OFFSET(region, len, offset) ((offset) + (len) <= (region)->buf_size)
1134 
1135 #if defined(DEBUG) || defined(DEBUG_VALIDATION)
1136 static void Validate(const StreamingBuffer *sb)
1137 {
1138  bool bail = false;
1139  uint32_t cnt = 0;
1140  for (const StreamingBufferRegion *r = &sb->region; r != NULL; r = r->next) {
1141  cnt++;
1142  if (r->next) {
1143  bail |= ((r->stream_offset + r->buf_size) > r->next->stream_offset);
1144  }
1145 
1146  bail |= (r->buf != NULL && r->buf_size == 0);
1147  bail |= (r->buf_offset > r->buf_size);
1148  }
1149  // leading gap, so buf_offset should be 0?
1150  if (sb->head && sb->head->offset > sb->region.stream_offset) {
1151  SCLogDebug("leading gap of size %" PRIu64, sb->head->offset - sb->region.stream_offset);
1152  BUG_ON(sb->region.buf_offset != 0);
1153  }
1154 
1155  if (sb->head && sb->head->offset == sb->region.stream_offset) {
1156  BUG_ON(sb->head->len > sb->region.buf_offset);
1157  BUG_ON(sb->head->len < sb->region.buf_offset);
1158  }
1159  BUG_ON(sb->regions != cnt);
1160  BUG_ON(bail);
1161 }
1162 #endif
1163 
1164 static void ListRegions(StreamingBuffer *sb)
1165 {
1166  if (sb->region.buf == NULL && sb->region.buf_offset == 0 && sb->region.next == NULL)
1167  return;
1168 #if defined(DEBUG) && DUMP_REGIONS == 1
1169  uint32_t cnt = 0;
1170  for (StreamingBufferRegion *r = &sb->region; r != NULL; r = r->next) {
1171  cnt++;
1172  char gap[64] = "";
1173  if (r->next) {
1174  snprintf(gap, sizeof(gap), "[ gap:%" PRIu64 " ]",
1175  r->next->stream_offset - (r->stream_offset + r->buf_size));
1176  }
1177 
1178  printf("[ %s offset:%" PRIu64 " size:%u offset:%u ]%s", r == &sb->region ? "main" : "aux",
1179  r->stream_offset, r->buf_size, r->buf_offset, gap);
1180  }
1181  printf("(max %u, cnt %u, sb->regions %u)\n", sb->max_regions, cnt, sb->regions);
1182  bool at_least_one = false;
1183  uint64_t last_re = sb->region.stream_offset;
1184  StreamingBufferBlock *sbb = NULL;
1185  RB_FOREACH(sbb, SBB, &sb->sbb_tree)
1186  {
1187  if (last_re != sbb->offset) {
1188  printf("[ gap:%" PRIu64 " ]", sbb->offset - last_re);
1189  }
1190  printf("[ sbb offset:%" PRIu64 " len:%u ]", sbb->offset, sbb->len);
1191  at_least_one = true;
1192  last_re = sbb->offset + sbb->len;
1193  }
1194  if (at_least_one)
1195  printf("\n");
1196 #endif
1197 #if defined(DEBUG) || defined(DEBUG_VALIDATION)
1198  StreamingBufferBlock *sbb2 = NULL;
1199  RB_FOREACH(sbb2, SBB, &sb->sbb_tree)
1200  {
1201  const uint8_t *_data = NULL;
1202  uint32_t _data_len = 0;
1203  (void)StreamingBufferSBBGetData(sb, sbb2, &_data, &_data_len);
1204  }
1205  Validate(sb);
1206 #endif
1207 }
1208 
1209 /** \internal
1210  * \brief process insert by consolidating the affected regions into one
1211  * \note sets sc_errno
1212  */
1213 static StreamingBufferRegion *BufferInsertAtRegionConsolidate(StreamingBuffer *sb,
1214  const StreamingBufferConfig *cfg, StreamingBufferRegion *dst,
1215  StreamingBufferRegion *src_start, StreamingBufferRegion *src_end,
1216  const uint64_t data_offset, const uint32_t data_len, StreamingBufferRegion *prev,
1217  uint32_t dst_buf_size)
1218 {
1219  int retval;
1220 #ifdef DEBUG
1221  const uint64_t data_re = data_offset + data_len;
1222  SCLogDebug("sb %p dst %p src_start %p src_end %p data_offset %" PRIu64
1223  "/data_len %u/data_re %" PRIu64,
1224  sb, dst, src_start, src_end, data_offset, data_len, data_re);
1225 #endif
1226 
1227  // 1. determine size and offset for dst.
1228  const uint64_t dst_offset = MIN(src_start->stream_offset, data_offset);
1229  DEBUG_VALIDATE_BUG_ON(dst_offset < sb->region.stream_offset);
1230  const uint32_t dst_size = dst_buf_size;
1231  SCLogDebug("dst_size %u", dst_size);
1232 
1233  // 2. resize dst
1234  const uint32_t old_size = dst->buf_size;
1235  const uint32_t dst_copy_offset = dst->stream_offset - dst_offset;
1236 #ifdef DEBUG
1237  const uint32_t old_offset = dst->buf_offset;
1238  SCLogDebug("old_size %u, old_offset %u, dst_copy_offset %u", old_size, old_offset,
1239  dst_copy_offset);
1240 #endif
1241  if ((retval = GrowRegionToSize(sb, cfg, dst, dst_size)) != SC_OK) {
1242  sc_errno = retval;
1243  return NULL;
1244  }
1245  SCLogDebug("resized to %u -> %u", dst_size, dst->buf_size);
1246  /* validate that the size is exactly what we asked for */
1247  DEBUG_VALIDATE_BUG_ON(dst_size != dst->buf_size);
1248  if (dst_copy_offset != 0)
1249  memmove(dst->buf + dst_copy_offset, dst->buf, old_size);
1250  if (dst_offset != dst->stream_offset) {
1251  dst->stream_offset = dst_offset;
1252  // buf_offset no longer valid, reset.
1253  dst->buf_offset = 0;
1254  }
1255 
1256  uint32_t new_offset = src_start->buf_offset;
1257  if (data_offset == src_start->stream_offset + src_start->buf_offset) {
1258  new_offset += data_len;
1259  }
1260 
1261  bool start_is_main = false;
1262  if (src_start == &sb->region) {
1263  DEBUG_VALIDATE_BUG_ON(src_start->stream_offset != dst_offset);
1264 
1265  start_is_main = true;
1266  SCLogDebug("src_start is main region");
1267  if (src_start != dst)
1268  memcpy(dst->buf, src_start->buf, src_start->buf_offset);
1269  if (src_start == src_end) {
1270  SCLogDebug("src_start == src_end == main, we're done");
1271  DEBUG_VALIDATE_BUG_ON(src_start != dst);
1272  return src_start;
1273  }
1274  prev = src_start;
1275  src_start = src_start->next; // skip in the loop below
1276  }
1277 
1278  // 3. copy all regions from src_start to dst_start into the new region
1279  for (StreamingBufferRegion *r = src_start; r != NULL;) {
1280  SCLogDebug("r %p %" PRIu64 ", offset %u, len %u, %s, last %s", r, r->stream_offset,
1281  r->buf_offset, r->buf_size, r == &sb->region ? "main" : "aux",
1282  BOOL2STR(r == src_end));
1283  // skip dst
1284  if (r == dst) {
1285  SCLogDebug("skipping r %p as it is 'dst'", r);
1286  if (r == src_end)
1287  break;
1288  prev = r;
1289  r = r->next;
1290  continue;
1291  }
1292  const uint32_t target_offset = r->stream_offset - dst_offset;
1293  SCLogDebug("r %p: target_offset %u", r, target_offset);
1294  DEBUG_VALIDATE_BUG_ON(target_offset > dst->buf_size);
1295  DEBUG_VALIDATE_BUG_ON(target_offset + r->buf_size > dst->buf_size);
1296  memcpy(dst->buf + target_offset, r->buf, r->buf_size);
1297 
1298  StreamingBufferRegion *next = r->next;
1299  FREE(cfg, r->buf, r->buf_size);
1300  FREE(cfg, r, sizeof(*r));
1301  sb->regions--;
1302  DEBUG_VALIDATE_BUG_ON(sb->regions == 0);
1303 
1304  DEBUG_VALIDATE_BUG_ON(prev == NULL && src_start != &sb->region);
1305  if (prev != NULL) {
1306  SCLogDebug("setting prev %p next to %p (was %p)", prev, next, prev->next);
1307  prev->next = next;
1308  } else {
1309  SCLogDebug("no prev yet");
1310  }
1311 
1312  if (r == src_end)
1313  break;
1314  r = next;
1315  }
1316 
1317  /* special handling of main region being the start, but not the
1318  * region we expand. In this case we'll have main and dst. We will
1319  * move the buffer from dst into main and free dst. */
1320  if (start_is_main && dst != &sb->region) {
1321  DEBUG_VALIDATE_BUG_ON(sb->region.next != dst);
1322  SCLogDebug("start_is_main && dst != main region");
1323  FREE(cfg, sb->region.buf, sb->region.buf_size);
1324  sb->region.buf = dst->buf;
1325  sb->region.buf_size = dst->buf_size;
1326  sb->region.buf_offset = new_offset;
1327  SCLogDebug("sb->region.buf_offset set to %u", sb->region.buf_offset);
1328  sb->region.next = dst->next;
1329  FREE(cfg, dst, sizeof(*dst));
1330  dst = &sb->region;
1331  sb->regions--;
1332  DEBUG_VALIDATE_BUG_ON(sb->regions == 0);
1333  } else {
1334  SCLogDebug("dst: %p next %p", dst, dst->next);
1335  }
1336 
1337  SCLogDebug("returning dst %p stream_offset %" PRIu64 " buf_offset %u buf_size %u", dst,
1338  dst->stream_offset, dst->buf_offset, dst->buf_size);
1339  return dst;
1340 }
1341 
1342 /** \note sets sc_errno */
1343 static StreamingBufferRegion *BufferInsertAtRegionDo(StreamingBuffer *sb,
1344  const StreamingBufferConfig *cfg, const uint64_t offset, const uint32_t len)
1345 {
1346  SCLogDebug("offset %" PRIu64 ", len %u", offset, len);
1347  StreamingBufferRegion *start_prev = NULL;
1348  StreamingBufferRegion *start =
1349  FindFirstRegionForOffset(sb, cfg, &sb->region, offset, len, &start_prev);
1350  if (start) {
1351  const uint64_t insert_re = offset + len;
1352  const uint64_t insert_start_offset = MIN(start->stream_offset, offset);
1353  uint64_t insert_adjusted_re = insert_re;
1354 
1355  SCLogDebug("start region %p/%" PRIu64 "/%u", start, start->stream_offset, start->buf_size);
1356  StreamingBufferRegion *big = FindLargestRegionForOffset(sb, cfg, start, offset, len);
1357  DEBUG_VALIDATE_BUG_ON(big == NULL);
1358  if (big == NULL) {
1359  sc_errno = SC_EINVAL;
1360  return NULL;
1361  }
1362  SCLogDebug("big region %p/%" PRIu64 "/%u", big, big->stream_offset, big->buf_size);
1363  StreamingBufferRegion *end = FindRightEdge(sb, cfg, big, offset, len);
1364  DEBUG_VALIDATE_BUG_ON(end == NULL);
1365  if (end == NULL) {
1366  sc_errno = SC_EINVAL;
1367  return NULL;
1368  }
1369  insert_adjusted_re = MAX(insert_adjusted_re, (end->stream_offset + end->buf_size));
1370 
1371  uint32_t new_buf_size =
1372  ToNextMultipleOf(insert_adjusted_re - insert_start_offset, cfg->buf_size);
1373  SCLogDebug("new_buf_size %u", new_buf_size);
1374 
1375  /* see if our new buf size + cfg->buf_size margin leads to an overlap with the next region.
1376  * If so, include that in the consolidation. */
1377  if (end->next != NULL && new_buf_size + insert_start_offset > end->next->stream_offset) {
1378  SCLogDebug("adjusted end from %p to %p", end, end->next);
1379  end = end->next;
1380  insert_adjusted_re = MAX(insert_adjusted_re, (end->stream_offset + end->buf_size));
1381  new_buf_size =
1382  ToNextMultipleOf(insert_adjusted_re - insert_start_offset, cfg->buf_size);
1383  SCLogDebug("new_buf_size %u", new_buf_size);
1384  }
1385 
1386  SCLogDebug("end region %p/%" PRIu64 "/%u", end, end->stream_offset, end->buf_size);
1387  /* sets sc_errno */
1388  StreamingBufferRegion *ret = BufferInsertAtRegionConsolidate(
1389  sb, cfg, big, start, end, offset, len, start_prev, new_buf_size);
1390  return ret;
1391  } else {
1392  /* if there was no region we can use we add a new region and insert it */
1393  StreamingBufferRegion *append = &sb->region;
1394  for (StreamingBufferRegion *r = append; r != NULL; r = r->next) {
1395  if (r->stream_offset > offset) {
1396  break;
1397  } else {
1398  append = r;
1399  }
1400  }
1401 
1402  SCLogDebug("no matching region found, append to %p (%s)", append,
1403  append == &sb->region ? "main" : "aux");
1404  /* sets sc_errno */
1405  StreamingBufferRegion *add = InitBufferRegion(sb, cfg, len);
1406  if (add == NULL)
1407  return NULL;
1408  add->stream_offset = offset;
1409  add->next = append->next;
1410  append->next = add;
1411  SCLogDebug("new region %p offset %" PRIu64, add, add->stream_offset);
1412  return add;
1413  }
1414 }
1415 
1416 /** \internal
1417  * \brief return the region to put the new data in
1418  *
1419  * Will find an existing region, expand it if needed. If no existing region exists or is
1420  * a good fit, it will try to set up a new region. If the region then overlaps or gets
1421  * too close to the next, merge them.
1422  *
1423  * \note sets sc_errno
1424  */
1425 static StreamingBufferRegion *BufferInsertAtRegion(StreamingBuffer *sb,
1426  const StreamingBufferConfig *cfg, const uint8_t *data, const uint32_t data_len,
1427  const uint64_t data_offset)
1428 {
1429  SCLogDebug("data_offset %" PRIu64 ", data_len %u, re %" PRIu64, data_offset, data_len,
1430  data_offset + data_len);
1431  ListRegions(sb);
1432 
1433  if (RegionsIntersect(sb, cfg, &sb->region, data_offset, data_offset + data_len)) {
1434  SCLogDebug("data_offset %" PRIu64 ", data_len %u intersects with main region (next %p)",
1435  data_offset, data_len, sb->region.next);
1436  if (sb->region.next == NULL ||
1437  !RegionsIntersect(sb, cfg, sb->region.next, data_offset, data_offset + data_len)) {
1438  SCLogDebug(
1439  "data_offset %" PRIu64
1440  ", data_len %u intersects with main region, no next or way before next region",
1441  data_offset, data_len);
1442  if (sb->region.buf == NULL) {
1443  int r;
1444  if ((r = InitBuffer(sb, cfg)) != SC_OK) { // TODO init with size
1445  sc_errno = r;
1446  return NULL;
1447  }
1448  }
1449  return &sb->region;
1450  }
1451  } else if (sb->region.next == NULL) {
1452  /* InitBufferRegion sets sc_errno */
1453  StreamingBufferRegion *aux_r = sb->region.next = InitBufferRegion(sb, cfg, data_len);
1454  if (aux_r == NULL)
1455  return NULL;
1456  aux_r->stream_offset = data_offset;
1457  DEBUG_VALIDATE_BUG_ON(data_len > aux_r->buf_size);
1458  SCLogDebug("created new region %p with offset %" PRIu64 ", size %u", aux_r,
1459  aux_r->stream_offset, aux_r->buf_size);
1460  return aux_r;
1461  }
1462  /* BufferInsertAtRegionDo sets sc_errno */
1463  StreamingBufferRegion *blob = BufferInsertAtRegionDo(sb, cfg, data_offset, data_len);
1464  SCLogDebug("blob %p (%s)", blob, blob == &sb->region ? "main" : "aux");
1465  return blob;
1466 }
1467 
1468 /**
1469  * \param offset offset relative to StreamingBuffer::stream_offset
1470  *
1471  * \return SC_OK in case of success
1472  * \return SC_E* errors otherwise
1473  */
1474 int StreamingBufferInsertAt(StreamingBuffer *sb, const StreamingBufferConfig *cfg,
1475  StreamingBufferSegment *seg, const uint8_t *data, uint32_t data_len, uint64_t offset)
1476 {
1477  int r;
1478 
1479  DEBUG_VALIDATE_BUG_ON(seg == NULL);
1480  DEBUG_VALIDATE_BUG_ON(offset < sb->region.stream_offset);
1481  if (offset < sb->region.stream_offset) {
1482  return SC_EINVAL;
1483  }
1484 
1485  StreamingBufferRegion *region = BufferInsertAtRegion(sb, cfg, data, data_len, offset);
1486  if (region == NULL) {
1487  return sc_errno;
1488  }
1489 
1490  const bool region_is_main = region == &sb->region;
1491 
1492  SCLogDebug("inserting %" PRIu64 "/%u using %s region %p", offset, data_len,
1493  region == &sb->region ? "main" : "aux", region);
1494 
1495  uint32_t rel_offset = offset - region->stream_offset;
1496  if (!DATA_FITS_AT_OFFSET(region, data_len, rel_offset)) {
1497  if ((r = GrowToSize(sb, cfg, (rel_offset + data_len))) != SC_OK)
1498  return r;
1499  }
1500  DEBUG_VALIDATE_BUG_ON(!DATA_FITS_AT_OFFSET(region, data_len, rel_offset));
1501 
1502  SCLogDebug("offset %" PRIu64 " data_len %u, rel_offset %u into region offset %" PRIu64
1503  ", buf_offset %u, buf_size %u",
1504  offset, data_len, rel_offset, region->stream_offset, region->buf_offset,
1505  region->buf_size);
1506  memcpy(region->buf + rel_offset, data, data_len);
1507  seg->stream_offset = offset;
1508  seg->segment_len = data_len;
1509 
1510  SCLogDebug("rel_offset %u region->stream_offset %" PRIu64 ", buf_offset %u", rel_offset,
1511  region->stream_offset, region->buf_offset);
1512 
1513  if (RB_EMPTY(&sb->sbb_tree)) {
1514  SCLogDebug("empty sbb list");
1515 
1516  if (region_is_main) {
1517  if (sb->region.stream_offset == offset) {
1518  SCLogDebug("empty sbb list: block exactly what was expected, fall through");
1519  /* empty list, data is exactly what is expected (append),
1520  * so do nothing.
1521  * Update buf_offset if needed, but in case of overlaps it might be beyond us. */
1522  sb->region.buf_offset = MAX(sb->region.buf_offset, rel_offset + data_len);
1523  } else if ((rel_offset + data_len) <= sb->region.buf_offset) {
1524  SCLogDebug("empty sbb list: block is within existing main data region");
1525  } else {
1526  if (sb->region.buf_offset && rel_offset == sb->region.buf_offset) {
1527  SCLogDebug("exactly at expected offset");
1528  // nothing to do
1529  sb->region.buf_offset = rel_offset + data_len;
1530 
1531  } else if (rel_offset < sb->region.buf_offset) {
1532  // nothing to do
1533 
1534  SCLogDebug("before expected offset: %u < sb->region.buf_offset %u", rel_offset,
1535  sb->region.buf_offset);
1536  if (rel_offset + data_len > sb->region.buf_offset) {
1537  SCLogDebug("before expected offset, ends after: %u < sb->region.buf_offset "
1538  "%u, %u > %u",
1539  rel_offset, sb->region.buf_offset, rel_offset + data_len,
1540  sb->region.buf_offset);
1541  sb->region.buf_offset = rel_offset + data_len;
1542  }
1543 
1544  } else if (sb->region.buf_offset) {
1545  SCLogDebug("beyond expected offset: SBBInit");
1546  /* existing data, but there is a gap between us */
1547  if ((r = SBBInit(sb, cfg, region, rel_offset, data_len)) != SC_OK)
1548  return r;
1549  } else {
1550  /* gap before data in empty list */
1551  SCLogDebug("empty sbb list: invoking SBBInitLeadingGap");
1552  if ((r = SBBInitLeadingGap(sb, cfg, region, offset, data_len)) != SC_OK)
1553  return r;
1554  }
1555  }
1556  } else {
1557  if (sb->region.buf_offset) {
1558  /* existing data, but there is a gap between us */
1559  SCLogDebug("empty sbb list, no data in main: use SBBInit");
1560  if ((r = SBBInit(sb, cfg, region, rel_offset, data_len)) != SC_OK)
1561  return r;
1562  } else {
1563  /* gap before data in empty list */
1564  SCLogDebug("empty sbb list: invoking SBBInitLeadingGap");
1565  if ((r = SBBInitLeadingGap(sb, cfg, region, offset, data_len)) != SC_OK)
1566  return r;
1567  }
1568  if (rel_offset == region->buf_offset) {
1569  SCLogDebug("pre region->buf_offset %u", region->buf_offset);
1570  region->buf_offset = rel_offset + data_len;
1571  SCLogDebug("post region->buf_offset %u", region->buf_offset);
1572  }
1573  }
1574  } else {
1575  SCLogDebug("updating sbb tree");
1576  /* already have blocks, so append new block based on new data */
1577  if ((r = SBBUpdate(sb, cfg, region, rel_offset, data_len)) != SC_OK)
1578  return r;
1579  }
1580  DEBUG_VALIDATE_BUG_ON(!region_is_main && sb->head == NULL);
1581 
1582  ListRegions(sb);
1583  if (RB_EMPTY(&sb->sbb_tree)) {
1584  DEBUG_VALIDATE_BUG_ON(offset + data_len > sb->region.stream_offset + sb->region.buf_offset);
1585  }
1586 
1587  return SC_OK;
1588 }
1589 
1590 int StreamingBufferSegmentIsBeforeWindow(const StreamingBuffer *sb,
1591  const StreamingBufferSegment *seg)
1592 {
1593  if (seg->stream_offset < sb->region.stream_offset) {
1594  if (seg->stream_offset + seg->segment_len <= sb->region.stream_offset) {
1595  return 1;
1596  }
1597  }
1598  return 0;
1599 }
1600 
1601 static inline const StreamingBufferRegion *GetRegionForOffset(
1602  const StreamingBuffer *sb, const uint64_t offset)
1603 {
1604  if (sb == NULL)
1605  return NULL;
1606  if (sb->region.next == NULL) {
1607  return &sb->region;
1608  }
1609  if (offset >= sb->region.stream_offset &&
1610  offset < (sb->region.stream_offset + sb->region.buf_size)) {
1611  return &sb->region;
1612  }
1613  for (const StreamingBufferRegion *r = sb->region.next; r != NULL; r = r->next) {
1614  if (offset >= r->stream_offset && offset < (r->stream_offset + r->buf_size)) {
1615  return r;
1616  }
1617  }
1618  return NULL;
1619 }
1620 
1621 /** \brief get the data for one SBB */
1622 void StreamingBufferSBBGetData(const StreamingBuffer *sb,
1623  const StreamingBufferBlock *sbb,
1624  const uint8_t **data, uint32_t *data_len)
1625 {
1626  const StreamingBufferRegion *region = GetRegionForOffset(sb, sbb->offset);
1627  SCLogDebug("first find our region (offset %" PRIu64 ") -> %p", sbb->offset, region);
1628  if (region) {
1629  SCLogDebug("region %p found %" PRIu64 "/%u/%u", region, region->stream_offset,
1630  region->buf_size, region->buf_offset);
1631  DEBUG_VALIDATE_BUG_ON(
1632  region->stream_offset == sbb->offset && region->buf_offset > sbb->len);
1633  // buf_offset should match first sbb len if it has the same offset
1634 
1635  if (sbb->offset >= region->stream_offset) {
1636  SCLogDebug("1");
1637  uint64_t offset = sbb->offset - region->stream_offset;
1638  *data = region->buf + offset;
1639  DEBUG_VALIDATE_BUG_ON(offset + sbb->len > region->buf_size);
1640  *data_len = sbb->len;
1641  return;
1642  } else {
1643  SCLogDebug("2");
1644  uint64_t offset = region->stream_offset - sbb->offset;
1645  if (offset < sbb->len) {
1646  *data = region->buf;
1647  *data_len = sbb->len - offset;
1648  return;
1649  }
1650  SCLogDebug("3");
1651  }
1652  }
1653  *data = NULL;
1654  *data_len = 0;
1655  return;
1656 }
1657 
1658 /** \brief get the data for one SBB */
1659 void StreamingBufferSBBGetDataAtOffset(const StreamingBuffer *sb,
1660  const StreamingBufferBlock *sbb,
1661  const uint8_t **data, uint32_t *data_len,
1662  uint64_t offset)
1663 {
1664  /* validate that we are looking for a offset within the sbb */
1665  DEBUG_VALIDATE_BUG_ON(!(offset >= sbb->offset && offset < (sbb->offset + sbb->len)));
1666  if (!(offset >= sbb->offset && offset < (sbb->offset + sbb->len))) {
1667  *data = NULL;
1668  *data_len = 0;
1669  return;
1670  }
1671 
1672  const StreamingBufferRegion *region = GetRegionForOffset(sb, offset);
1673  if (region) {
1674  uint32_t sbblen = sbb->len - (offset - sbb->offset);
1675 
1676  if (offset >= region->stream_offset) {
1677  uint64_t data_offset = offset - region->stream_offset;
1678  *data = region->buf + data_offset;
1679  if (data_offset + sbblen > region->buf_size)
1680  *data_len = region->buf_size - data_offset;
1681  else
1682  *data_len = sbblen;
1683  DEBUG_VALIDATE_BUG_ON(*data_len > sbblen);
1684  return;
1685  } else {
1686  uint64_t data_offset = region->stream_offset - sbb->offset;
1687  if (data_offset < sbblen) {
1688  *data = region->buf;
1689  *data_len = sbblen - data_offset;
1690  DEBUG_VALIDATE_BUG_ON(*data_len > sbblen);
1691  return;
1692  }
1693  }
1694  }
1695 
1696  *data = NULL;
1697  *data_len = 0;
1698  return;
1699 }
1700 
1701 void StreamingBufferSegmentGetData(const StreamingBuffer *sb,
1702  const StreamingBufferSegment *seg,
1703  const uint8_t **data, uint32_t *data_len)
1704 {
1705  const StreamingBufferRegion *region = GetRegionForOffset(sb, seg->stream_offset);
1706  if (region) {
1707  if (seg->stream_offset >= region->stream_offset) {
1708  uint64_t offset = seg->stream_offset - region->stream_offset;
1709  *data = region->buf + offset;
1710  if (offset + seg->segment_len > region->buf_size)
1711  *data_len = region->buf_size - offset;
1712  else
1713  *data_len = seg->segment_len;
1714  SCLogDebug("*data_len %u", *data_len);
1715  return;
1716  } else {
1717  uint64_t offset = region->stream_offset - seg->stream_offset;
1718  if (offset < seg->segment_len) {
1719  *data = region->buf;
1720  *data_len = seg->segment_len - offset;
1721  SCLogDebug("*data_len %u", *data_len);
1722  return;
1723  }
1724  }
1725  }
1726  *data = NULL;
1727  *data_len = 0;
1728  return;
1729 }
1730 
1731 /**
1732  * \retval 1 data is the same
1733  * \retval 0 data is different
1734  */
1735 int StreamingBufferSegmentCompareRawData(const StreamingBuffer *sb,
1736  const StreamingBufferSegment *seg,
1737  const uint8_t *rawdata, uint32_t rawdata_len)
1738 {
1739  const uint8_t *segdata = NULL;
1740  uint32_t segdata_len = 0;
1741  StreamingBufferSegmentGetData(sb, seg, &segdata, &segdata_len);
1742  if (segdata && segdata_len &&
1743  segdata_len == rawdata_len &&
1744  memcmp(segdata, rawdata, segdata_len) == 0)
1745  {
1746  return 1;
1747  }
1748  return 0;
1749 }
1750 
1751 int StreamingBufferGetData(const StreamingBuffer *sb,
1752  const uint8_t **data, uint32_t *data_len,
1753  uint64_t *stream_offset)
1754 {
1755  if (sb != NULL && sb->region.buf != NULL) {
1756  *data = sb->region.buf;
1757  *data_len = sb->region.buf_offset;
1758  *stream_offset = sb->region.stream_offset;
1759  return 1;
1760  } else {
1761  *data = NULL;
1762  *data_len = 0;
1763  *stream_offset = 0;
1764  return 0;
1765  }
1766 }
1767 
1768 int StreamingBufferGetDataAtOffset (const StreamingBuffer *sb,
1769  const uint8_t **data, uint32_t *data_len,
1770  uint64_t offset)
1771 {
1772  const StreamingBufferRegion *region = GetRegionForOffset(sb, offset);
1773  if (region != NULL && region->buf != NULL && offset >= region->stream_offset &&
1774  offset < (region->stream_offset + region->buf_offset)) {
1775  uint32_t skip = offset - region->stream_offset;
1776  *data = region->buf + skip;
1777  *data_len = region->buf_offset - skip;
1778  return 1;
1779  } else {
1780  *data = NULL;
1781  *data_len = 0;
1782  return 0;
1783  }
1784 }
1785 
1786 /**
1787  * \retval 1 data is the same
1788  * \retval 0 data is different
1789  */
1790 int StreamingBufferCompareRawData(const StreamingBuffer *sb,
1791  const uint8_t *rawdata, uint32_t rawdata_len)
1792 {
1793  const uint8_t *sbdata = NULL;
1794  uint32_t sbdata_len = 0;
1795  uint64_t offset = 0;
1796  StreamingBufferGetData(sb, &sbdata, &sbdata_len, &offset);
1797  if (offset == 0 &&
1798  sbdata && sbdata_len &&
1799  sbdata_len == rawdata_len &&
1800  memcmp(sbdata, rawdata, sbdata_len) == 0)
1801  {
1802  return 1;
1803  }
1804  SCLogDebug("sbdata_len %u, offset %" PRIu64, sbdata_len, offset);
1805  printf("got:\n");
1806  PrintRawDataFp(stdout, sbdata,sbdata_len);
1807  printf("wanted:\n");
1808  PrintRawDataFp(stdout, rawdata,rawdata_len);
1809  return 0;
1810 }
1811 
1812 #ifdef UNITTESTS
1813 static void Dump(StreamingBuffer *sb)
1814 {
1815  PrintRawDataFp(stdout, sb->region.buf, sb->region.buf_offset);
1816 }
1817 
1818 static void DumpSegment(StreamingBuffer *sb, StreamingBufferSegment *seg)
1819 {
1820  const uint8_t *data = NULL;
1821  uint32_t data_len = 0;
1822  StreamingBufferSegmentGetData(sb, seg, &data, &data_len);
1823  if (data && data_len) {
1824  PrintRawDataFp(stdout, data, data_len);
1825  }
1826 }
1827 
1828 static int StreamingBufferTest02(void)
1829 {
1830  StreamingBufferConfig cfg = { 24, 1, STREAMING_BUFFER_REGION_GAP_DEFAULT, NULL, NULL, NULL };
1831  StreamingBuffer *sb = StreamingBufferInit(&cfg);
1832  FAIL_IF(sb == NULL);
1833 
1834  StreamingBufferSegment seg1;
1835  FAIL_IF(StreamingBufferAppend(sb, &cfg, &seg1, (const uint8_t *)"ABCDEFGH", 8) != 0);
1836  StreamingBufferSegment seg2;
1837  FAIL_IF(StreamingBufferAppend(sb, &cfg, &seg2, (const uint8_t *)"01234567", 8) != 0);
1838  FAIL_IF(sb->region.stream_offset != 0);
1839  FAIL_IF(sb->region.buf_offset != 16);
1840  FAIL_IF(seg1.stream_offset != 0);
1841  FAIL_IF(seg2.stream_offset != 8);
1842  FAIL_IF(StreamingBufferSegmentIsBeforeWindow(sb,&seg1));
1843  FAIL_IF(StreamingBufferSegmentIsBeforeWindow(sb,&seg2));
1844  Dump(sb);
1845  DumpSegment(sb, &seg1);
1846  DumpSegment(sb, &seg2);
1847  FAIL_IF_NOT_NULL(sb->head);
1848  FAIL_IF_NOT(sb->head == RB_MIN(SBB, &sb->sbb_tree));
1849 
1850  StreamingBufferSlideToOffset(sb, &cfg, 6);
1851  FAIL_IF_NOT_NULL(sb->head);
1852  FAIL_IF_NOT(sb->head == RB_MIN(SBB, &sb->sbb_tree));
1853 
1854  StreamingBufferSegment seg3;
1855  FAIL_IF(StreamingBufferAppend(sb, &cfg, &seg3, (const uint8_t *)"QWERTY", 6) != 0);
1856  FAIL_IF(sb->region.stream_offset != 6);
1857  FAIL_IF(sb->region.buf_offset != 16);
1858  FAIL_IF(seg3.stream_offset != 16);
1859  FAIL_IF(StreamingBufferSegmentIsBeforeWindow(sb,&seg1));
1860  FAIL_IF(StreamingBufferSegmentIsBeforeWindow(sb,&seg2));
1861  FAIL_IF(StreamingBufferSegmentIsBeforeWindow(sb,&seg3));
1862  Dump(sb);
1863  DumpSegment(sb, &seg1);
1864  DumpSegment(sb, &seg2);
1865  DumpSegment(sb, &seg3);
1866  FAIL_IF_NOT_NULL(sb->head);
1867  FAIL_IF_NOT(sb->head == RB_MIN(SBB, &sb->sbb_tree));
1868 
1869  StreamingBufferSlideToOffset(sb, &cfg, 12);
1870  FAIL_IF(!StreamingBufferSegmentIsBeforeWindow(sb,&seg1));
1871  FAIL_IF(StreamingBufferSegmentIsBeforeWindow(sb,&seg2));
1872  FAIL_IF(StreamingBufferSegmentIsBeforeWindow(sb,&seg3));
1873  Dump(sb);
1874  DumpSegment(sb, &seg1);
1875  DumpSegment(sb, &seg2);
1876  DumpSegment(sb, &seg3);
1877  FAIL_IF_NOT_NULL(sb->head);
1878  FAIL_IF_NOT(sb->head == RB_MIN(SBB, &sb->sbb_tree));
1879 
1880  StreamingBufferFree(sb, &cfg);
1881  PASS;
1882 }
1883 
1884 static int StreamingBufferTest03(void)
1885 {
1886  StreamingBufferConfig cfg = { 24, 1, STREAMING_BUFFER_REGION_GAP_DEFAULT, NULL, NULL, NULL };
1887  StreamingBuffer *sb = StreamingBufferInit(&cfg);
1888  FAIL_IF(sb == NULL);
1889 
1890  StreamingBufferSegment seg1;
1891  FAIL_IF(StreamingBufferAppend(sb, &cfg, &seg1, (const uint8_t *)"ABCDEFGH", 8) != 0);
1892  StreamingBufferSegment seg2;
1893  FAIL_IF(StreamingBufferInsertAt(sb, &cfg, &seg2, (const uint8_t *)"01234567", 8, 14) != 0);
1894  FAIL_IF(sb->region.stream_offset != 0);
1895  FAIL_IF(sb->region.buf_offset != 8);
1896  FAIL_IF(seg1.stream_offset != 0);
1897  FAIL_IF(seg2.stream_offset != 14);
1898  FAIL_IF(StreamingBufferSegmentIsBeforeWindow(sb,&seg1));
1899  FAIL_IF(StreamingBufferSegmentIsBeforeWindow(sb,&seg2));
1900  Dump(sb);
1901  DumpSegment(sb, &seg1);
1902  DumpSegment(sb, &seg2);
1903  FAIL_IF_NULL(sb->head);
1904  FAIL_IF_NOT(sb->sbb_size == 16);
1905  FAIL_IF_NOT(sb->head == RB_MIN(SBB, &sb->sbb_tree));
1906 
1907  StreamingBufferSegment seg3;
1908  FAIL_IF(StreamingBufferInsertAt(sb, &cfg, &seg3, (const uint8_t *)"QWERTY", 6, 8) != 0);
1909  FAIL_IF(sb->region.stream_offset != 0);
1910  FAIL_IF(sb->region.buf_offset != 22);
1911  FAIL_IF(seg3.stream_offset != 8);
1912  FAIL_IF(StreamingBufferSegmentIsBeforeWindow(sb,&seg1));
1913  FAIL_IF(StreamingBufferSegmentIsBeforeWindow(sb,&seg2));
1914  FAIL_IF(StreamingBufferSegmentIsBeforeWindow(sb,&seg3));
1915  Dump(sb);
1916  DumpSegment(sb, &seg1);
1917  DumpSegment(sb, &seg2);
1918  DumpSegment(sb, &seg3);
1919  FAIL_IF_NULL(sb->head);
1920  FAIL_IF_NOT(sb->sbb_size == 22);
1921  FAIL_IF_NOT(sb->head == RB_MIN(SBB, &sb->sbb_tree));
1922 
1923  StreamingBufferSlideToOffset(sb, &cfg, 10);
1924  FAIL_IF(!StreamingBufferSegmentIsBeforeWindow(sb,&seg1));
1925  FAIL_IF(StreamingBufferSegmentIsBeforeWindow(sb,&seg2));
1926  FAIL_IF(StreamingBufferSegmentIsBeforeWindow(sb,&seg3));
1927  Dump(sb);
1928  DumpSegment(sb, &seg1);
1929  DumpSegment(sb, &seg2);
1930  DumpSegment(sb, &seg3);
1931  FAIL_IF_NULL(sb->head);
1932  FAIL_IF_NOT(sb->sbb_size == 12);
1933  FAIL_IF_NOT(sb->head == RB_MIN(SBB, &sb->sbb_tree));
1934 
1935  StreamingBufferFree(sb, &cfg);
1936  PASS;
1937 }
1938 
1939 static int StreamingBufferTest04(void)
1940 {
1941  StreamingBufferConfig cfg = { 16, 1, STREAMING_BUFFER_REGION_GAP_DEFAULT, NULL, NULL, NULL };
1942  StreamingBuffer *sb = StreamingBufferInit(&cfg);
1943  FAIL_IF(sb == NULL);
1944 
1945  StreamingBufferSegment seg1;
1946  FAIL_IF(StreamingBufferAppend(sb, &cfg, &seg1, (const uint8_t *)"ABCDEFGH", 8) != 0);
1947  FAIL_IF(!RB_EMPTY(&sb->sbb_tree));
1948  StreamingBufferSegment seg2;
1949  FAIL_IF(StreamingBufferInsertAt(sb, &cfg, &seg2, (const uint8_t *)"01234567", 8, 14) != 0);
1950  FAIL_IF(sb->region.stream_offset != 0);
1951  FAIL_IF(sb->region.buf_offset != 8);
1952  FAIL_IF(seg1.stream_offset != 0);
1953  FAIL_IF(seg2.stream_offset != 14);
1954  FAIL_IF(StreamingBufferSegmentIsBeforeWindow(sb,&seg1));
1955  FAIL_IF(StreamingBufferSegmentIsBeforeWindow(sb,&seg2));
1956  FAIL_IF(RB_EMPTY(&sb->sbb_tree));
1957  StreamingBufferBlock *sbb1 = RB_MIN(SBB, &sb->sbb_tree);
1958  FAIL_IF(sbb1 != sb->head);
1959  FAIL_IF_NULL(sbb1);
1960  FAIL_IF(sbb1->offset != 0);
1961  FAIL_IF(sbb1->len != 8);
1962  StreamingBufferBlock *sbb2 = SBB_RB_NEXT(sbb1);
1963  FAIL_IF_NULL(sbb2);
1964  FAIL_IF(sbb2 == sb->head);
1965  FAIL_IF(sbb2->offset != 14);
1966  FAIL_IF(sbb2->len != 8);
1967  Dump(sb);
1968  DumpSegment(sb, &seg1);
1969  DumpSegment(sb, &seg2);
1970  FAIL_IF_NOT(sb->head == RB_MIN(SBB, &sb->sbb_tree));
1971 
1972  StreamingBufferSegment seg3;
1973  FAIL_IF(StreamingBufferInsertAt(sb, &cfg, &seg3, (const uint8_t *)"QWERTY", 6, 8) != 0);
1974  FAIL_IF(sb->region.stream_offset != 0);
1975  FAIL_IF(sb->region.buf_offset != 22);
1976  FAIL_IF(seg3.stream_offset != 8);
1977  FAIL_IF(StreamingBufferSegmentIsBeforeWindow(sb,&seg1));
1978  FAIL_IF(StreamingBufferSegmentIsBeforeWindow(sb,&seg2));
1979  FAIL_IF(StreamingBufferSegmentIsBeforeWindow(sb,&seg3));
1980  sbb1 = RB_MIN(SBB, &sb->sbb_tree);
1981  FAIL_IF_NULL(sbb1);
1982  FAIL_IF(sbb1 != sb->head);
1983  FAIL_IF(sbb1->offset != 0);
1984  FAIL_IF(sbb1->len != 22);
1985  FAIL_IF(SBB_RB_NEXT(sbb1));
1986  Dump(sb);
1987  DumpSegment(sb, &seg1);
1988  DumpSegment(sb, &seg2);
1989  DumpSegment(sb, &seg3);
1990  FAIL_IF_NULL(sb->head);
1991  FAIL_IF_NOT(sb->sbb_size == 22);
1992  FAIL_IF_NOT(sb->head == RB_MIN(SBB, &sb->sbb_tree));
1993 
1994  /* far ahead of curve: */
1995  StreamingBufferSegment seg4;
1996  FAIL_IF(StreamingBufferInsertAt(sb, &cfg, &seg4, (const uint8_t *)"XYZ", 3, 124) != 0);
1997  FAIL_IF(sb->region.stream_offset != 0);
1998  FAIL_IF(sb->region.buf_offset != 22);
1999  FAIL_IF(sb->region.buf_size != 128);
2000  FAIL_IF(seg4.stream_offset != 124);
2001  FAIL_IF(StreamingBufferSegmentIsBeforeWindow(sb,&seg1));
2002  FAIL_IF(StreamingBufferSegmentIsBeforeWindow(sb,&seg2));
2003  FAIL_IF(StreamingBufferSegmentIsBeforeWindow(sb,&seg3));
2004  FAIL_IF(StreamingBufferSegmentIsBeforeWindow(sb,&seg4));
2005  sbb1 = RB_MIN(SBB, &sb->sbb_tree);
2006  FAIL_IF_NULL(sbb1);
2007  FAIL_IF(sbb1 != sb->head);
2008  FAIL_IF(sbb1->offset != 0);
2009  FAIL_IF(sbb1->len != 22);
2010  FAIL_IF(!SBB_RB_NEXT(sbb1));
2011  Dump(sb);
2012  DumpSegment(sb, &seg1);
2013  DumpSegment(sb, &seg2);
2014  DumpSegment(sb, &seg3);
2015  DumpSegment(sb, &seg4);
2016  FAIL_IF_NULL(sb->head);
2017  FAIL_IF_NOT(sb->sbb_size == 25);
2018  FAIL_IF_NOT(sb->head == RB_MIN(SBB, &sb->sbb_tree));
2019 
2020  FAIL_IF(!StreamingBufferSegmentCompareRawData(sb,&seg1,(const uint8_t *)"ABCDEFGH", 8));
2021  FAIL_IF(!StreamingBufferSegmentCompareRawData(sb,&seg2,(const uint8_t *)"01234567", 8));
2022  FAIL_IF(!StreamingBufferSegmentCompareRawData(sb,&seg3,(const uint8_t *)"QWERTY", 6));
2023  FAIL_IF(!StreamingBufferSegmentCompareRawData(sb,&seg4,(const uint8_t *)"XYZ", 3));
2024 
2025  StreamingBufferFree(sb, &cfg);
2026  PASS;
2027 }
2028 
2029 /** \test lots of gaps in block list */
2030 static int StreamingBufferTest06(void)
2031 {
2032  StreamingBufferConfig cfg = { 16, 1, STREAMING_BUFFER_REGION_GAP_DEFAULT, NULL, NULL, NULL };
2033  StreamingBuffer *sb = StreamingBufferInit(&cfg);
2034  FAIL_IF(sb == NULL);
2035 
2036  StreamingBufferSegment seg1;
2037  FAIL_IF(StreamingBufferAppend(sb, &cfg, &seg1, (const uint8_t *)"A", 1) != 0);
2038  StreamingBufferSegment seg2;
2039  FAIL_IF(StreamingBufferInsertAt(sb, &cfg, &seg2, (const uint8_t *)"C", 1, 2) != 0);
2040  Dump(sb);
2041  FAIL_IF_NULL(sb->head);
2042  FAIL_IF_NOT(sb->sbb_size == 2);
2043  FAIL_IF_NOT(sb->head == RB_MIN(SBB, &sb->sbb_tree));
2044 
2045  StreamingBufferSegment seg3;
2046  FAIL_IF(StreamingBufferInsertAt(sb, &cfg, &seg3, (const uint8_t *)"F", 1, 5) != 0);
2047  Dump(sb);
2048  FAIL_IF_NULL(sb->head);
2049  FAIL_IF_NOT(sb->sbb_size == 3);
2050  FAIL_IF_NOT(sb->head == RB_MIN(SBB, &sb->sbb_tree));
2051 
2052  StreamingBufferSegment seg4;
2053  FAIL_IF(StreamingBufferInsertAt(sb, &cfg, &seg4, (const uint8_t *)"H", 1, 7) != 0);
2054  Dump(sb);
2055  FAIL_IF_NULL(sb->head);
2056  FAIL_IF_NOT(sb->sbb_size == 4);
2057  FAIL_IF_NOT(sb->head == RB_MIN(SBB, &sb->sbb_tree));
2058 
2059  StreamingBufferSegment seg5;
2060  FAIL_IF(StreamingBufferInsertAt(sb, &cfg, &seg5, (const uint8_t *)"ABCDEFGHIJ", 10, 0) != 0);
2061  Dump(sb);
2062  StreamingBufferBlock *sbb1 = RB_MIN(SBB, &sb->sbb_tree);
2063  FAIL_IF_NULL(sbb1);
2064  FAIL_IF(sbb1->offset != 0);
2065  FAIL_IF(sbb1->len != 10);
2066  FAIL_IF(SBB_RB_NEXT(sbb1));
2067  FAIL_IF_NULL(sb->head);
2068  FAIL_IF_NOT(sb->sbb_size == 10);
2069  FAIL_IF_NOT(sb->head == RB_MIN(SBB, &sb->sbb_tree));
2070 
2071  StreamingBufferSegment seg6;
2072  FAIL_IF(StreamingBufferInsertAt(sb, &cfg, &seg6, (const uint8_t *)"abcdefghij", 10, 0) != 0);
2073  Dump(sb);
2074  sbb1 = RB_MIN(SBB, &sb->sbb_tree);
2075  FAIL_IF_NULL(sbb1);
2076  FAIL_IF(sbb1->offset != 0);
2077  FAIL_IF(sbb1->len != 10);
2078  FAIL_IF(SBB_RB_NEXT(sbb1));
2079  FAIL_IF_NULL(sb->head);
2080  FAIL_IF_NOT(sb->sbb_size == 10);
2081  FAIL_IF_NOT(sb->head == RB_MIN(SBB, &sb->sbb_tree));
2082 
2083  StreamingBufferFree(sb, &cfg);
2084  PASS;
2085 }
2086 
2087 /** \test lots of gaps in block list */
2088 static int StreamingBufferTest07(void)
2089 {
2090  StreamingBufferConfig cfg = { 16, 1, STREAMING_BUFFER_REGION_GAP_DEFAULT, NULL, NULL, NULL };
2091  StreamingBuffer *sb = StreamingBufferInit(&cfg);
2092  FAIL_IF(sb == NULL);
2093 
2094  StreamingBufferSegment seg1;
2095  FAIL_IF(StreamingBufferInsertAt(sb, &cfg, &seg1, (const uint8_t *)"B", 1, 1) != 0);
2096  StreamingBufferSegment seg2;
2097  FAIL_IF(StreamingBufferInsertAt(sb, &cfg, &seg2, (const uint8_t *)"D", 1, 3) != 0);
2098  Dump(sb);
2099  FAIL_IF_NULL(sb->head);
2100  FAIL_IF_NOT(sb->sbb_size == 2);
2101  FAIL_IF_NOT(sb->head == RB_MIN(SBB, &sb->sbb_tree));
2102 
2103  StreamingBufferSegment seg3;
2104  FAIL_IF(StreamingBufferInsertAt(sb, &cfg, &seg3, (const uint8_t *)"F", 1, 5) != 0);
2105  Dump(sb);
2106  FAIL_IF_NULL(sb->head);
2107  FAIL_IF_NOT(sb->sbb_size == 3);
2108  FAIL_IF_NOT(sb->head == RB_MIN(SBB, &sb->sbb_tree));
2109 
2110  StreamingBufferSegment seg4;
2111  FAIL_IF(StreamingBufferInsertAt(sb, &cfg, &seg4, (const uint8_t *)"H", 1, 7) != 0);
2112  Dump(sb);
2113  FAIL_IF_NULL(sb->head);
2114  FAIL_IF_NOT(sb->sbb_size == 4);
2115  FAIL_IF_NOT(sb->head == RB_MIN(SBB, &sb->sbb_tree));
2116 
2117  StreamingBufferSegment seg5;
2118  FAIL_IF(StreamingBufferInsertAt(sb, &cfg, &seg5, (const uint8_t *)"ABCDEFGHIJ", 10, 0) != 0);
2119  Dump(sb);
2120  StreamingBufferBlock *sbb1 = RB_MIN(SBB, &sb->sbb_tree);
2121  FAIL_IF_NULL(sbb1);
2122  FAIL_IF(sbb1->offset != 0);
2123  FAIL_IF(sbb1->len != 10);
2124  FAIL_IF(SBB_RB_NEXT(sbb1));
2125  FAIL_IF_NULL(sb->head);
2126  FAIL_IF_NOT(sb->sbb_size == 10);
2127  FAIL_IF_NOT(sb->head == RB_MIN(SBB, &sb->sbb_tree));
2128 
2129  StreamingBufferSegment seg6;
2130  FAIL_IF(StreamingBufferInsertAt(sb, &cfg, &seg6, (const uint8_t *)"abcdefghij", 10, 0) != 0);
2131  Dump(sb);
2132  sbb1 = RB_MIN(SBB, &sb->sbb_tree);
2133  FAIL_IF_NULL(sbb1);
2134  FAIL_IF(sbb1->offset != 0);
2135  FAIL_IF(sbb1->len != 10);
2136  FAIL_IF(SBB_RB_NEXT(sbb1));
2137  FAIL_IF_NULL(sb->head);
2138  FAIL_IF_NOT(sb->sbb_size == 10);
2139  FAIL_IF_NOT(sb->head == RB_MIN(SBB, &sb->sbb_tree));
2140 
2141  StreamingBufferFree(sb, &cfg);
2142  PASS;
2143 }
2144 
2145 /** \test lots of gaps in block list */
2146 static int StreamingBufferTest08(void)
2147 {
2148  StreamingBufferConfig cfg = { 16, 1, STREAMING_BUFFER_REGION_GAP_DEFAULT, NULL, NULL, NULL };
2149  StreamingBuffer *sb = StreamingBufferInit(&cfg);
2150  FAIL_IF(sb == NULL);
2151 
2152  StreamingBufferSegment seg1;
2153  FAIL_IF(StreamingBufferInsertAt(sb, &cfg, &seg1, (const uint8_t *)"B", 1, 1) != 0);
2154  StreamingBufferSegment seg2;
2155  FAIL_IF(StreamingBufferInsertAt(sb, &cfg, &seg2, (const uint8_t *)"D", 1, 3) != 0);
2156  Dump(sb);
2157  FAIL_IF_NULL(sb->head);
2158  FAIL_IF_NOT(sb->sbb_size == 2);
2159  FAIL_IF_NOT(sb->head == RB_MIN(SBB, &sb->sbb_tree));
2160 
2161  StreamingBufferSegment seg3;
2162  FAIL_IF(StreamingBufferInsertAt(sb, &cfg, &seg3, (const uint8_t *)"F", 1, 5) != 0);
2163  Dump(sb);
2164  FAIL_IF_NULL(sb->head);
2165  FAIL_IF_NOT(sb->sbb_size == 3);
2166  FAIL_IF_NOT(sb->head == RB_MIN(SBB, &sb->sbb_tree));
2167 
2168  StreamingBufferSegment seg4;
2169  FAIL_IF(StreamingBufferInsertAt(sb, &cfg, &seg4, (const uint8_t *)"H", 1, 7) != 0);
2170  Dump(sb);
2171  FAIL_IF_NULL(sb->head);
2172  FAIL_IF_NOT(sb->sbb_size == 4);
2173  FAIL_IF_NOT(sb->head == RB_MIN(SBB, &sb->sbb_tree));
2174 
2175  StreamingBufferSegment seg5;
2176  FAIL_IF(StreamingBufferInsertAt(sb, &cfg, &seg5, (const uint8_t *)"ABCDEFGHIJ", 10, 0) != 0);
2177  Dump(sb);
2178  StreamingBufferBlock *sbb1 = RB_MIN(SBB, &sb->sbb_tree);
2179  FAIL_IF_NULL(sbb1);
2180  FAIL_IF(sbb1->offset != 0);
2181  FAIL_IF(sbb1->len != 10);
2182  FAIL_IF(SBB_RB_NEXT(sbb1));
2183  FAIL_IF_NULL(sb->head);
2184  FAIL_IF_NOT(sb->sbb_size == 10);
2185  FAIL_IF_NOT(sb->head == RB_MIN(SBB, &sb->sbb_tree));
2186 
2187  StreamingBufferSegment seg6;
2188  FAIL_IF(StreamingBufferAppend(sb, &cfg, &seg6, (const uint8_t *)"abcdefghij", 10) != 0);
2189  Dump(sb);
2190  sbb1 = RB_MIN(SBB, &sb->sbb_tree);
2191  FAIL_IF_NULL(sbb1);
2192  FAIL_IF(sbb1->offset != 0);
2193  FAIL_IF(sbb1->len != 20);
2194  FAIL_IF(SBB_RB_NEXT(sbb1));
2195  FAIL_IF_NULL(sb->head);
2196  FAIL_IF_NOT(sb->sbb_size == 20);
2197  FAIL_IF_NOT(sb->head == RB_MIN(SBB, &sb->sbb_tree));
2198 
2199  StreamingBufferFree(sb, &cfg);
2200  PASS;
2201 }
2202 
2203 /** \test lots of gaps in block list */
2204 static int StreamingBufferTest09(void)
2205 {
2206  StreamingBufferConfig cfg = { 16, 1, STREAMING_BUFFER_REGION_GAP_DEFAULT, NULL, NULL, NULL };
2207  StreamingBuffer *sb = StreamingBufferInit(&cfg);
2208  FAIL_IF(sb == NULL);
2209 
2210  StreamingBufferSegment seg1;
2211  FAIL_IF(StreamingBufferInsertAt(sb, &cfg, &seg1, (const uint8_t *)"B", 1, 1) != 0);
2212  StreamingBufferSegment seg2;
2213  FAIL_IF(StreamingBufferInsertAt(sb, &cfg, &seg2, (const uint8_t *)"D", 1, 3) != 0);
2214  Dump(sb);
2215  FAIL_IF_NULL(sb->head);
2216  FAIL_IF_NOT(sb->sbb_size == 2);
2217  FAIL_IF_NOT(sb->head == RB_MIN(SBB, &sb->sbb_tree));
2218 
2219  StreamingBufferSegment seg3;
2220  FAIL_IF(StreamingBufferInsertAt(sb, &cfg, &seg3, (const uint8_t *)"H", 1, 7) != 0);
2221  Dump(sb);
2222  FAIL_IF_NULL(sb->head);
2223  FAIL_IF_NOT(sb->sbb_size == 3);
2224  FAIL_IF_NOT(sb->head == RB_MIN(SBB, &sb->sbb_tree));
2225 
2226  StreamingBufferSegment seg4;
2227  FAIL_IF(StreamingBufferInsertAt(sb, &cfg, &seg4, (const uint8_t *)"F", 1, 5) != 0);
2228  Dump(sb);
2229  FAIL_IF_NULL(sb->head);
2230  FAIL_IF_NOT(sb->sbb_size == 4);
2231  FAIL_IF_NOT(sb->head == RB_MIN(SBB, &sb->sbb_tree));
2232 
2233  StreamingBufferSegment seg5;
2234  FAIL_IF(StreamingBufferInsertAt(sb, &cfg, &seg5, (const uint8_t *)"ABCDEFGHIJ", 10, 0) != 0);
2235  Dump(sb);
2236  StreamingBufferBlock *sbb1 = RB_MIN(SBB, &sb->sbb_tree);
2237  FAIL_IF_NULL(sbb1);
2238  FAIL_IF(sbb1->offset != 0);
2239  FAIL_IF(sbb1->len != 10);
2240  FAIL_IF(SBB_RB_NEXT(sbb1));
2241  FAIL_IF_NULL(sb->head);
2242  FAIL_IF_NOT(sb->sbb_size == 10);
2243  FAIL_IF_NOT(sb->head == RB_MIN(SBB, &sb->sbb_tree));
2244 
2245  StreamingBufferSegment seg6;
2246  FAIL_IF(StreamingBufferInsertAt(sb, &cfg, &seg6, (const uint8_t *)"abcdefghij", 10, 0) != 0);
2247  Dump(sb);
2248  sbb1 = RB_MIN(SBB, &sb->sbb_tree);
2249  FAIL_IF_NULL(sbb1);
2250  FAIL_IF(sbb1->offset != 0);
2251  FAIL_IF(sbb1->len != 10);
2252  FAIL_IF(SBB_RB_NEXT(sbb1));
2253  FAIL_IF_NULL(sb->head);
2254  FAIL_IF_NOT(sb->sbb_size == 10);
2255  FAIL_IF_NOT(sb->head == RB_MIN(SBB, &sb->sbb_tree));
2256 
2257  StreamingBufferFree(sb, &cfg);
2258  PASS;
2259 }
2260 
2261 /** \test lots of gaps in block list */
2262 static int StreamingBufferTest10(void)
2263 {
2264  StreamingBufferConfig cfg = { 16, 1, STREAMING_BUFFER_REGION_GAP_DEFAULT, NULL, NULL, NULL };
2265  StreamingBuffer *sb = StreamingBufferInit(&cfg);
2266  FAIL_IF(sb == NULL);
2267 
2268  StreamingBufferSegment seg1;
2269  FAIL_IF(StreamingBufferInsertAt(sb, &cfg, &seg1, (const uint8_t *)"A", 1, 0) != 0);
2270  Dump(sb);
2271  StreamingBufferSegment seg2;
2272  FAIL_IF(StreamingBufferInsertAt(sb, &cfg, &seg2, (const uint8_t *)"D", 1, 3) != 0);
2273  Dump(sb);
2274  StreamingBufferSegment seg3;
2275  FAIL_IF(StreamingBufferInsertAt(sb, &cfg, &seg3, (const uint8_t *)"H", 1, 7) != 0);
2276  Dump(sb);
2277  FAIL_IF_NULL(sb->head);
2278  FAIL_IF_NOT(sb->sbb_size == 3);
2279 
2280  StreamingBufferSegment seg4;
2281  FAIL_IF(StreamingBufferInsertAt(sb, &cfg, &seg4, (const uint8_t *)"B", 1, 1) != 0);
2282  Dump(sb);
2283  StreamingBufferSegment seg5;
2284  FAIL_IF(StreamingBufferInsertAt(sb, &cfg, &seg5, (const uint8_t *)"C", 1, 2) != 0);
2285  Dump(sb);
2286  StreamingBufferSegment seg6;
2287  FAIL_IF(StreamingBufferInsertAt(sb, &cfg, &seg6, (const uint8_t *)"G", 1, 6) != 0);
2288  Dump(sb);
2289  FAIL_IF_NOT(sb->head == RB_MIN(SBB, &sb->sbb_tree));
2290  FAIL_IF_NULL(sb->head);
2291  FAIL_IF_NOT(sb->sbb_size == 6);
2292 
2293  StreamingBufferSegment seg7;
2294  FAIL_IF(StreamingBufferInsertAt(sb, &cfg, &seg7, (const uint8_t *)"ABCDEFGHIJ", 10, 0) != 0);
2295  Dump(sb);
2296  StreamingBufferBlock *sbb1 = RB_MIN(SBB, &sb->sbb_tree);
2297  FAIL_IF_NULL(sbb1);
2298  FAIL_IF(sbb1->offset != 0);
2299  FAIL_IF(sbb1->len != 10);
2300  FAIL_IF(SBB_RB_NEXT(sbb1));
2301  FAIL_IF_NULL(sb->head);
2302  FAIL_IF_NOT(sb->sbb_size == 10);
2303 
2304  StreamingBufferSegment seg8;
2305  FAIL_IF(StreamingBufferInsertAt(sb, &cfg, &seg8, (const uint8_t *)"abcdefghij", 10, 0) != 0);
2306  Dump(sb);
2307  sbb1 = RB_MIN(SBB, &sb->sbb_tree);
2308  FAIL_IF_NOT(sb->head == RB_MIN(SBB, &sb->sbb_tree));
2309  FAIL_IF_NULL(sbb1);
2310  FAIL_IF(sbb1->offset != 0);
2311  FAIL_IF(sbb1->len != 10);
2312  FAIL_IF(SBB_RB_NEXT(sbb1));
2313  FAIL_IF_NULL(sb->head);
2314  FAIL_IF_NOT(sb->sbb_size == 10);
2315 
2316  StreamingBufferFree(sb, &cfg);
2317  PASS;
2318 }
2319 
2320 #endif
2321 
2322 void StreamingBufferRegisterTests(void)
2323 {
2324 #ifdef UNITTESTS
2325  UtRegisterTest("StreamingBufferTest02", StreamingBufferTest02);
2326  UtRegisterTest("StreamingBufferTest03", StreamingBufferTest03);
2327  UtRegisterTest("StreamingBufferTest04", StreamingBufferTest04);
2328  UtRegisterTest("StreamingBufferTest06", StreamingBufferTest06);
2329  UtRegisterTest("StreamingBufferTest07", StreamingBufferTest07);
2330  UtRegisterTest("StreamingBufferTest08", StreamingBufferTest08);
2331  UtRegisterTest("StreamingBufferTest09", StreamingBufferTest09);
2332  UtRegisterTest("StreamingBufferTest10", StreamingBufferTest10);
2333 #endif
2334 }
FREE
#define FREE(cfg, ptr, s)
Definition: util-streaming-buffer.c:62
StreamingBufferSlideToOffset
void StreamingBufferSlideToOffset(StreamingBuffer *sb, const StreamingBufferConfig *cfg, uint64_t offset)
slide to absolute offset
Definition: util-streaming-buffer.c:997
len
uint8_t len
Definition: app-layer-dnp3.h:2
FAIL_IF_NULL
#define FAIL_IF_NULL(expr)
Fail a test if expression evaluates to NULL.
Definition: util-unittest.h:89
StreamingBufferConfig_::buf_size
uint32_t buf_size
Definition: util-streaming-buffer.h:67
offset
uint64_t offset
Definition: util-streaming-buffer.h:0
StreamingBuffer_::head
StreamingBufferBlock * head
Definition: util-streaming-buffer.h:112
unlikely
#define unlikely(expr)
Definition: util-optimize.h:35
StreamingBufferRegion_::next
struct StreamingBufferRegion_ * next
Definition: util-streaming-buffer.h:90
UtRegisterTest
void UtRegisterTest(const char *name, int(*TestFn)(void))
Register unit test.
Definition: util-unittest.c:103
SCLogDebug
#define SCLogDebug(...)
Definition: util-debug.h:269
StreamingBufferFree
void StreamingBufferFree(StreamingBuffer *sb, const StreamingBufferConfig *cfg)
Definition: util-streaming-buffer.c:295
next
struct HtpBodyChunk_ * next
Definition: app-layer-htp.h:0
RB_LEFT
#define RB_LEFT(elm, field)
Definition: tree.h:322
SC_EINVAL
@ SC_EINVAL
Definition: util-error.h:30
StreamingBuffer_::max_regions
uint16_t max_regions
Definition: util-streaming-buffer.h:115
StreamingBuffer_::regions
uint16_t regions
Definition: util-streaming-buffer.h:114
StreamingBufferGetData
int StreamingBufferGetData(const StreamingBuffer *sb, const uint8_t **data, uint32_t *data_len, uint64_t *stream_offset)
Definition: util-streaming-buffer.c:1751
RB_MIN
#define RB_MIN(name, x)
Definition: tree.h:778
MIN
#define MIN(x, y)
Definition: suricata-common.h:386
StreamingBufferAppend
int StreamingBufferAppend(StreamingBuffer *sb, const StreamingBufferConfig *cfg, StreamingBufferSegment *seg, const uint8_t *data, uint32_t data_len)
Definition: util-streaming-buffer.c:1066
segment_len
uint32_t segment_len
Definition: util-streaming-buffer.h:0
m
SCMutex m
Definition: flow-hash.h:6
StreamingBufferGetDataAtOffset
int StreamingBufferGetDataAtOffset(const StreamingBuffer *sb, const uint8_t **data, uint32_t *data_len, uint64_t offset)
Definition: util-streaming-buffer.c:1768
MAX
#define MAX(x, y)
Definition: suricata-common.h:390
StreamingBufferSegmentCompareRawData
int StreamingBufferSegmentCompareRawData(const StreamingBuffer *sb, const StreamingBufferSegment *seg, const uint8_t *rawdata, uint32_t rawdata_len)
Definition: util-streaming-buffer.c:1735
stream_offset
uint64_t stream_offset
Definition: util-streaming-buffer.h:1
StreamingBufferConfig_::max_regions
uint16_t max_regions
Definition: util-streaming-buffer.h:68
SC_ELIMIT
@ SC_ELIMIT
Definition: util-error.h:31
util-unittest.h
FAIL_IF_NOT
#define FAIL_IF_NOT(expr)
Fail a test if expression evaluates to false.
Definition: util-unittest.h:82
SC_ENOMEM
@ SC_ENOMEM
Definition: util-error.h:29
RB_EMPTY
#define RB_EMPTY(head)
Definition: tree.h:327
FAIL_IF_NOT_NULL
#define FAIL_IF_NOT_NULL(expr)
Fail a test if expression evaluates to non-NULL.
Definition: util-unittest.h:96
util-debug.h
StreamingBufferRegion_::buf_size
uint32_t buf_size
Definition: util-streaming-buffer.h:87
PASS
#define PASS
Pass the test.
Definition: util-unittest.h:105
util-error.h
RB_ROOT
#define RB_ROOT(head)
Definition: tree.h:326
StreamingBufferInit
StreamingBuffer * StreamingBufferInit(const StreamingBufferConfig *cfg)
Definition: util-streaming-buffer.c:249
StreamingBufferSegmentIsBeforeWindow
int StreamingBufferSegmentIsBeforeWindow(const StreamingBuffer *sb, const StreamingBufferSegment *seg)
Definition: util-streaming-buffer.c:1590
BIT_U32
#define BIT_U32(n)
Definition: suricata-common.h:395
REALLOC
#define REALLOC(cfg, ptr, orig_s, s)
Definition: util-streaming-buffer.c:60
CALLOC
#define CALLOC(cfg, n, s)
Definition: util-streaming-buffer.c:59
BOOL2STR
#define BOOL2STR(b)
Definition: util-debug.h:527
util-print.h
RB_FOREACH_SAFE
#define RB_FOREACH_SAFE(x, name, head, y)
Definition: tree.h:791
StreamingBufferClear
void StreamingBufferClear(StreamingBuffer *sb, const StreamingBufferConfig *cfg)
Definition: util-streaming-buffer.c:272
SCLogWarning
#define SCLogWarning(...)
Macro used to log WARNING messages.
Definition: util-debug.h:249
PrintRawDataFp
void PrintRawDataFp(FILE *fp, const uint8_t *buf, uint32_t buflen)
Definition: util-print.c:143
BUG_ON
#define BUG_ON(x)
Definition: suricata-common.h:295
RB_FOREACH
#define RB_FOREACH(x, name, head)
Definition: tree.h:781
StreamingBuffer_::sbb_size
uint32_t sbb_size
Definition: util-streaming-buffer.h:113
STREAMING_BUFFER_REGION_GAP_DEFAULT
#define STREAMING_BUFFER_REGION_GAP_DEFAULT
Definition: util-streaming-buffer.h:64
StreamingBufferRegion_::buf
uint8_t * buf
Definition: util-streaming-buffer.h:86
StreamingBufferCompareRawData
int StreamingBufferCompareRawData(const StreamingBuffer *sb, const uint8_t *rawdata, uint32_t rawdata_len)
Definition: util-streaming-buffer.c:1790
SC_OK
@ SC_OK
Definition: util-error.h:27
RB_GENERATE
RB_GENERATE(SBB, StreamingBufferBlock, rb, SBBCompare)
StreamingBufferRegion_::buf_offset
uint32_t buf_offset
Definition: util-streaming-buffer.h:88
SBB_RB_FIND_INCLUSIVE
StreamingBufferBlock * SBB_RB_FIND_INCLUSIVE(struct SBB *head, StreamingBufferBlock *elm)
Definition: util-streaming-buffer.c:100
SCRealloc
#define SCRealloc(ptr, sz)
Definition: util-mem.h:50
StreamingBuffer_::sbb_tree
struct SBB sbb_tree
Definition: util-streaming-buffer.h:111
StreamingBuffer_
Definition: util-streaming-buffer.h:109
StreamingBufferSBBGetDataAtOffset
void StreamingBufferSBBGetDataAtOffset(const StreamingBuffer *sb, const StreamingBufferBlock *sbb, const uint8_t **data, uint32_t *data_len, uint64_t offset)
get the data for one SBB
Definition: util-streaming-buffer.c:1659
FAIL_IF
#define FAIL_IF(expr)
Fail a test if expression evaluates to true.
Definition: util-unittest.h:71
StreamingBufferAppendNoTrack
int StreamingBufferAppendNoTrack(StreamingBuffer *sb, const StreamingBufferConfig *cfg, const uint8_t *data, uint32_t data_len)
add data w/o tracking a segment
Definition: util-streaming-buffer.c:1103
suricata-common.h
util-streaming-buffer.h
util-validate.h
StreamingBufferConfig_
Definition: util-streaming-buffer.h:66
HtpBodyChunk_::next
struct HtpBodyChunk_ * next
Definition: app-layer-htp.h:178
RB_RIGHT
#define RB_RIGHT(elm, field)
Definition: tree.h:323
StreamingBufferInsertAt
int StreamingBufferInsertAt(StreamingBuffer *sb, const StreamingBufferConfig *cfg, StreamingBufferSegment *seg, const uint8_t *data, uint32_t data_len, uint64_t offset)
Definition: util-streaming-buffer.c:1474
head
Flow * head
Definition: flow-hash.h:1
DATA_FITS_AT_OFFSET
#define DATA_FITS_AT_OFFSET(region, len, offset)
Definition: util-streaming-buffer.c:1133
sc_errno
thread_local SCError sc_errno
Definition: util-error.c:31
StreamingBufferRegion_
Definition: util-streaming-buffer.h:85
RB_FOREACH_REVERSE_FROM
#define RB_FOREACH_REVERSE_FROM(x, name, y)
Definition: tree.h:801
SBBCompare
int SBBCompare(struct StreamingBufferBlock *a, struct StreamingBufferBlock *b)
Definition: util-streaming-buffer.c:69
dst
uint16_t dst
Definition: app-layer-dnp3.h:4
RB_FOREACH_FROM
#define RB_FOREACH_FROM(x, name, y)
Definition: tree.h:786
StreamingBuffer_::region
StreamingBufferRegion region
Definition: util-streaming-buffer.h:110
StreamingBufferSegmentGetData
void StreamingBufferSegmentGetData(const StreamingBuffer *sb, const StreamingBufferSegment *seg, const uint8_t **data, uint32_t *data_len)
Definition: util-streaming-buffer.c:1701
SCCalloc
#define SCCalloc(nm, sz)
Definition: util-mem.h:53
StreamingBufferConfig_::region_gap
uint32_t region_gap
Definition: util-streaming-buffer.h:69
SCReturnInt
#define SCReturnInt(x)
Definition: util-debug.h:275
StreamingBufferBlock::len
uint32_t len
Definition: util-streaming-buffer.h:99
StreamingBufferBlock::offset
uint64_t offset
Definition: util-streaming-buffer.h:97
StreamingBufferSBBGetData
void StreamingBufferSBBGetData(const StreamingBuffer *sb, const StreamingBufferBlock *sbb, const uint8_t **data, uint32_t *data_len)
get the data for one SBB
Definition: util-streaming-buffer.c:1622
DATA_FITS
#define DATA_FITS(sb, len)
Definition: util-streaming-buffer.c:1064
DEBUG_VALIDATE_BUG_ON
#define DEBUG_VALIDATE_BUG_ON(exp)
Definition: util-validate.h:104
StreamingBufferBlock
block of continues data
Definition: util-streaming-buffer.h:96
WARN_UNUSED
#define WARN_UNUSED
Definition: suricata-common.h:398
StreamingBufferRegion_::stream_offset
uint64_t stream_offset
Definition: util-streaming-buffer.h:89
StreamingBufferRegisterTests
void StreamingBufferRegisterTests(void)
Definition: util-streaming-buffer.c:2322