Skip to content
GitLab
Menu
Projects
Groups
Snippets
Loading...
Help
Help
Support
Community forum
Keyboard shortcuts
?
Submit feedback
Contribute to GitLab
Sign in / Register
Toggle navigation
Menu
Open sidebar
ruanhaishen
redis
Commits
bf4def0f
Unverified
Commit
bf4def0f
authored
Jul 12, 2018
by
Salvatore Sanfilippo
Committed by
GitHub
Jul 12, 2018
Browse files
Merge pull request #5112 from 0xtonyxia/fix-xreadgroup-crash-after-setid
Streams: fix xreadgroup crash after xgroup SETID is sent.
parents
16770551
3f8a3efe
Changes
1
Show whitespace changes
Inline
Side-by-side
src/t_stream.c
View file @
bf4def0f
...
...
@@ -905,24 +905,29 @@ size_t streamReplyWithRange(client *c, stream *s, streamID *start, streamID *end
* will not require extra lookups. We'll fix the problem later
* if we find that there is already a entry for this ID. */
streamNACK
*
nack
=
streamCreateNACK
(
consumer
);
int
retval
=
0
;
retval
+
=
raxTryInsert
(
group
->
pel
,
buf
,
sizeof
(
buf
),
nack
,
NULL
);
retval
+
=
raxTryInsert
(
consumer
->
pel
,
buf
,
sizeof
(
buf
),
nack
,
NULL
);
int
retval
_group_tryins
=
0
,
retval_consumer_tryins
=
0
;
retval
_group_tryins
=
raxTryInsert
(
group
->
pel
,
buf
,
sizeof
(
buf
),
nack
,
NULL
);
retval
_consumer_tryins
=
raxTryInsert
(
consumer
->
pel
,
buf
,
sizeof
(
buf
),
nack
,
NULL
);
/* Now we can check if the entry was already busy, and
* in that case reassign the entry to the new consumer. */
if
(
retval
==
0
)
{
if
(
retval_group_tryins
==
0
)
{
streamNACK
*
oldnack
=
raxFind
(
group
->
pel
,
buf
,
sizeof
(
buf
));
serverAssert
(
oldnack
!=
raxNotFound
);
if
(
retval_consumer_tryins
==
0
)
{
/* In this case, only need to update old nack
* delivery_time and delivery_count. */
oldnack
->
delivery_time
=
mstime
();
oldnack
->
delivery_count
++
;
streamFreeNACK
(
nack
);
nack
=
raxFind
(
group
->
pel
,
buf
,
sizeof
(
buf
));
serverAssert
(
nack
!=
raxNotFound
);
raxRemove
(
nack
->
consumer
->
pel
,
buf
,
sizeof
(
buf
),
NULL
);
/* Update the consumer and idle time. */
nack
->
consumer
=
consumer
;
nack
->
delivery_time
=
mstime
();
nack
->
delivery_count
++
;
/* Add the entry in the new consumer local PEL. */
raxInsert
(
consumer
->
pel
,
buf
,
sizeof
(
buf
),
nack
,
NULL
);
}
else
if
(
retval
==
1
)
{
}
else
{
/* remove the old nack from original consumer */
raxRemove
(
oldnack
->
consumer
->
pel
,
buf
,
sizeof
(
buf
),
NULL
);
/* Overwrite nack in group pel. */
raxInsert
(
group
->
pel
,
buf
,
sizeof
(
buf
),
nack
,
NULL
);
streamFreeNACK
(
oldnack
);
}
}
else
if
(
retval_group_tryins
==
1
&&
retval_consumer_tryins
==
0
)
{
serverPanic
(
"NACK half-created. Should not be possible."
);
}
...
...
Write
Preview
Markdown
is supported
0%
Try again
or
attach a new file
.
Attach a file
Cancel
You are about to add
0
people
to the discussion. Proceed with caution.
Finish editing this message first!
Cancel
Please
register
or
sign in
to comment