MCPcopy Create free account
hub / github.com/Snapchat/KeyDB / streamTrim

Function streamTrim

src/t_stream.cpp:715–866  ·  view source on GitHub ↗

Trim the stream 's' according to args->trim_strategy, and return the * number of elements removed from the stream. The 'approx' option, if non-zero, * specifies that the trimming must be performed in a approximated way in * order to maximize performances. This means that the stream may contain * entries with IDs < 'id' in case of MINID (or more elements than 'maxlen' * in case of MAXLEN), and

Source from the content-addressed store, hash-verified

713 * IDs < 'id' (or number of elements >= maxlen in case of MAXLEN).
714 */
715int64_t streamTrim(stream *s, streamAddTrimArgs *args) {
716 size_t maxlen = args->maxlen;
717 streamID *id = &args->minid;
718 int approx = args->approx_trim;
719 int64_t limit = args->limit;
720 int trim_strategy = args->trim_strategy;
721
722 if (trim_strategy == TRIM_STRATEGY_NONE)
723 return 0;
724
725 raxIterator ri;
726 raxStart(&ri,s->rax);
727 raxSeek(&ri,"^",NULL,0);
728
729 int64_t deleted = 0;
730 while (raxNext(&ri)) {
731 if (trim_strategy == TRIM_STRATEGY_MAXLEN && s->length <= maxlen)
732 break;
733
734 unsigned char *lp = (unsigned char*)ri.data, *p = lpFirst(lp);
735 int64_t entries = lpGetInteger(p);
736
737 /* Check if we exceeded the amount of work we could do */
738 if (limit && (deleted + entries) > limit)
739 break;
740
741 /* Check if we can remove the whole node. */
742 int remove_node;
743 streamID master_id = {0}; /* For MINID */
744 if (trim_strategy == TRIM_STRATEGY_MAXLEN) {
745 remove_node = s->length - entries >= maxlen;
746 } else {
747 /* Read the master ID from the radix tree key. */
748 streamDecodeID(ri.key, &master_id);
749
750 /* Read last ID. */
751 streamID last_id;
752 lpGetEdgeStreamID(lp, 0, &master_id, &last_id);
753
754 /* We can remove the entire node id its last ID < 'id' */
755 remove_node = streamCompareID(&last_id, id) < 0;
756 }
757
758 if (remove_node) {
759 lpFree(lp);
760 raxRemove(s->rax,ri.key,ri.key_len,NULL);
761 raxSeek(&ri,">=",ri.key,ri.key_len);
762 s->length -= entries;
763 deleted += entries;
764 continue;
765 }
766
767 /* If we cannot remove a whole element, and approx is true,
768 * stop here. */
769 if (approx) break;
770
771 /* Now we have to trim entries from within 'lp' */
772 int64_t deleted_from_lp = 0;

Callers 4

streamTrimByLengthFunction · 0.85
streamTrimByIDFunction · 0.85
xaddCommandFunction · 0.85
xtrimCommandFunction · 0.85

Calls 13

raxStartFunction · 0.85
raxSeekFunction · 0.85
raxNextFunction · 0.85
lpFirstFunction · 0.85
streamDecodeIDFunction · 0.85
lpGetEdgeStreamIDFunction · 0.85
streamCompareIDFunction · 0.85
lpFreeFunction · 0.85
raxRemoveFunction · 0.85
lpNextFunction · 0.85
lpReplaceIntegerFunction · 0.85
raxInsertFunction · 0.85

Tested by

no test coverage detected