Skip to content

Commit 4ee7db8

Browse files
sggeorgievsggeorgiev
committed
fixed: issue from review
1 parent 9899403 commit 4ee7db8

2 files changed

Lines changed: 21 additions & 19 deletions

File tree

‎src/stream.h‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -172,9 +172,11 @@ typedef struct streamReplyRangeArgs {
172172
int flags; /* STREAM_RWR_* flags. */
173173
streamPropInfo *spi; /* Propagation info, or NULL for no propagation. */
174174
unsigned long *propCount; /* Out: number of propagated commands, or NULL. */
175-
long long maxsize; /* Byte budget for the reply (0 means unlimited). */
175+
long long maxsize; /* Byte budget for the reply (0 means unlimited).
176+
Already includes the output-bytes baseline, so
177+
MAXSIZE is checked against the absolute
178+
c->net_output_bytes_curr_cmd. */
176179
size_t emitted_before; /* Entries already emitted before this call. */
177-
size_t maxsize_base; /* Output-bytes baseline; MAXSIZE checks the delta. */
178180
} streamReplyRangeArgs;
179181

180182
/* Prototypes of exported APIs. */

‎src/t_stream.c‎

Lines changed: 17 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -37,7 +37,7 @@
3737

3838
void streamFreeCGGeneric(void *cg, void *s);
3939
void streamFreeNACK(stream *s, streamNACK *na);
40-
size_t streamReplyWithRangeFromConsumerPEL(client *c, stream *s, streamID *start, streamID *end, size_t count, streamCG *group, streamConsumer *consumer, long long maxsize, size_t emitted_before, size_t maxsize_base);
40+
size_t streamReplyWithRangeFromConsumerPEL(client *c, stream *s, streamID *start, streamID *end, size_t count, streamCG *group, streamConsumer *consumer, long long maxsize, size_t emitted_before);
4141
int streamParseStrictIDOrReply(client *c, robj *o, streamID *id, uint64_t missing_seq, int *seq_given);
4242
int streamParseIDOrReply(client *c, robj *o, streamID *id, uint64_t missing_seq);
4343

@@ -1967,12 +1967,12 @@ void streamPropagateConsumerCreation(client *c, robj *key, robj *groupname, sds
19671967
/* Returns non-zero if the MAXSIZE byte budget has been reached. 'maxsize' is the
19681968
* budget (0 = no limit) and 'emitted' the entries emitted so far; the budget is
19691969
* only enforced after at least one entry, so a single oversized message can still
1970-
* exceed it. The budget applies to the delta from 'maxsize_base' (output bytes at
1971-
* serve-start) so earlier commands in the same MULTI/EXEC don't count; outside a
1972-
* transaction the base is 0. */
1973-
static inline int streamReplyMaxsizeReached(client *c, long long maxsize, size_t emitted, size_t maxsize_base) {
1970+
* exceed it. 'maxsize' already includes the output-bytes baseline (output bytes at
1971+
* serve-start), so earlier commands in the same MULTI/EXEC don't count; outside a
1972+
* transaction the baseline is 0. */
1973+
static inline int streamReplyMaxsizeReached(client *c, long long maxsize, size_t emitted) {
19741974
return maxsize && emitted > 0 &&
1975-
(c->net_output_bytes_curr_cmd - maxsize_base) >= (size_t)maxsize;
1975+
c->net_output_bytes_curr_cmd >= (size_t)maxsize;
19761976
}
19771977

19781978
/* Send the stream items in the specified range to the client 'c'. The range
@@ -2059,7 +2059,6 @@ size_t streamReplyWithRange(client *c, stream *s, streamReplyRangeArgs *args) {
20592059
unsigned long *propCount = args->propCount;
20602060
long long maxsize = args->maxsize;
20612061
size_t emitted_before = args->emitted_before;
2062-
size_t maxsize_base = args->maxsize_base;
20632062
void *arraylen_ptr = NULL;
20642063
size_t arraylen = 0;
20652064
streamIterator si;
@@ -2105,7 +2104,7 @@ size_t streamReplyWithRange(client *c, stream *s, streamReplyRangeArgs *args) {
21052104
uint64_t idle = cmd_time_snapshot - nack->delivery_time;
21062105
if (idle < (uint64_t)min_idle_time) break;
21072106

2108-
if (streamReplyMaxsizeReached(c, maxsize, emitted_before + arraylen, maxsize_base))
2107+
if (streamReplyMaxsizeReached(c, maxsize, emitted_before + arraylen))
21092108
break;
21102109

21112110
/* Process and claim this entry */
@@ -2180,7 +2179,7 @@ size_t streamReplyWithRange(client *c, stream *s, streamReplyRangeArgs *args) {
21802179
}
21812180
return streamReplyWithRangeFromConsumerPEL(c,s,start,end,count,
21822181
group, consumer,
2183-
maxsize, emitted_before, maxsize_base);
2182+
maxsize, emitted_before);
21842183
}
21852184

21862185
/* Stop here if client only wants claimed entries or count is satisfied. */
@@ -2200,7 +2199,7 @@ size_t streamReplyWithRange(client *c, stream *s, streamReplyRangeArgs *args) {
22002199
while (streamIteratorGetID(&si,&id,&numfields)) {
22012200
/* Break before delivering the entry so it is neither sent nor added to
22022201
* the consumer's PEL. */
2203-
if (streamReplyMaxsizeReached(c, maxsize, emitted_before + arraylen, maxsize_base))
2202+
if (streamReplyMaxsizeReached(c, maxsize, emitted_before + arraylen))
22042203
break;
22052204

22062205
/* Update the group last_id if needed. */
@@ -2343,7 +2342,7 @@ size_t streamReplyWithRange(client *c, stream *s, streamReplyRangeArgs *args) {
23432342
* seek into the radix tree of the messages in order to emit the full message
23442343
* to the client. However clients only reach this code path when they are
23452344
* fetching the history of already retrieved messages, which is rare. */
2346-
size_t streamReplyWithRangeFromConsumerPEL(client *c, stream *s, streamID *start, streamID *end, size_t count, streamCG *group, streamConsumer *consumer, long long maxsize, size_t emitted_before, size_t maxsize_base) {
2345+
size_t streamReplyWithRangeFromConsumerPEL(client *c, stream *s, streamID *start, streamID *end, size_t count, streamCG *group, streamConsumer *consumer, long long maxsize, size_t emitted_before) {
23472346
raxIterator ri;
23482347
unsigned char startkey[sizeof(streamID)];
23492348
unsigned char endkey[sizeof(streamID)];
@@ -2356,7 +2355,7 @@ size_t streamReplyWithRangeFromConsumerPEL(client *c, stream *s, streamID *start
23562355
raxSeek(&ri,">=",startkey,sizeof(startkey));
23572356
while(raxNext(&ri) && (!count || arraylen < count)) {
23582357
if (end && memcmp(ri.key,endkey,ri.key_len) > 0) break;
2359-
if (streamReplyMaxsizeReached(c, maxsize, emitted_before + arraylen, maxsize_base))
2358+
if (streamReplyMaxsizeReached(c, maxsize, emitted_before + arraylen))
23602359
break;
23612360
streamID thisid;
23622361
streamDecodeID(ri.key,&thisid);
@@ -2983,15 +2982,17 @@ void xreadCommand(client *c) {
29832982
mstime_t min_pel_delivery_time = LLONG_MAX;
29842983
size_t total_entries = 0; /* Entries emitted so far across all streams,
29852984
used to enforce MAXCOUNT/MAXSIZE. */
2986-
/* MAXSIZE budget baseline: bytes accumulated before serving (non-zero inside
2987-
* MULTI/EXEC). The budget is enforced on the delta from this value. */
2988-
size_t maxsize_base = c->net_output_bytes_curr_cmd;
2985+
/* MAXSIZE budget baseline: fold the bytes accumulated before serving
2986+
* (non-zero inside MULTI/EXEC) into the budget so MAXSIZE is enforced on the
2987+
* absolute c->net_output_bytes_curr_cmd. Only adjust a real (non-zero) budget,
2988+
* since maxsize == 0 means "no limit". */
2989+
if (maxsize) maxsize += c->net_output_bytes_curr_cmd;
29892990
for (int i = 0; i < streams_count; i++) {
29902991
/* Stop scanning further streams once a cumulative limit is reached.
29912992
* At least the first stream is always served, so a single message
29922993
* larger than MAXSIZE is still returned. */
29932994
if (maxcount && total_entries >= (size_t)maxcount) break;
2994-
if (streamReplyMaxsizeReached(c, maxsize, total_entries, maxsize_base)) break;
2995+
if (streamReplyMaxsizeReached(c, maxsize, total_entries)) break;
29952996

29962997
kvobj *o = lookupKeyRead(c->db, c->argv[streams_arg + i]);
29972998
if (o == NULL) continue;
@@ -3123,7 +3124,6 @@ void xreadCommand(client *c) {
31233124
.group = groups ? groups[i] : NULL, .consumer = consumer,
31243125
.flags = flags, .spi = &spi, .propCount = &propCount,
31253126
.maxsize = maxsize, .emitted_before = total_entries,
3126-
.maxsize_base = maxsize_base,
31273127
};
31283128
total_entries += streamReplyWithRange(c,s,&args);
31293129
if (server.memory_tracking_enabled)

0 commit comments

Comments
 (0)