Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Improve Consumer with multiple subjects #4188

Merged
merged 3 commits into from Jun 1, 2023
Merged

Improve Consumer with multiple subjects #4188

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
9a79248
Improve consumer with multiple filters
Jarema Mar 31, 2023
a463dfe
Improve consumer error handling
Jarema May 26, 2023
261f39b
Make sure that o.subjf is nil or checked if empty
Jarema May 28, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
57 changes: 32 additions & 25 deletions server/consumer.go
View file
Open in desktop
Expand Up @@ -1735,7 +1735,12 @@ func (o *consumer) updateConfig(cfg *ConsumerConfig) error {
o.mu.Lock()

// When we're done with signaling, we can replace the subjects.
o.subjf = newSubjf
// If filters were removed, set `o.subjf` to nil.
if len(newSubjf) == 0 {
o.subjf = nil
} else {
o.subjf = newSubjf
}
}

// Record new config for others that do not need special handling.
Expand Down Expand Up @@ -3247,7 +3252,7 @@ func (o *consumer) getNextMsg() (*jsPubMsg, uint64, error) {
store := o.mset.store

// If no filters are specified, optimize to fetch just non-filtered messages.
if o.subjf == nil {
if len(o.subjf) == 0 {
// Grab next message applicable to us.
// We will unlock here in case lots of contention, e.g. WQ.
o.mu.Unlock()
Expand All @@ -3267,7 +3272,6 @@ func (o *consumer) getNextMsg() (*jsPubMsg, uint64, error) {
return pmsg, 1, err
}

var lastErr error
// if we have filters, iterate over filters and optimize by buffering found messages.
for _, filter := range o.subjf {
if filter.nextSeq < o.sseq {
Expand All @@ -3290,12 +3294,10 @@ func (o *consumer) getNextMsg() (*jsPubMsg, uint64, error) {
filter.pmsg = pmsg
derekcollison marked this conversation as resolved.
Show resolved Hide resolved
} else {
pmsg.returnToPool()
pmsg = nil
}
if sseq >= filter.nextSeq {
filter.nextSeq = sseq + 1
if err == ErrStoreEOF {
o.updateSkipped(uint64(filter.currentSeq))
}
}

// If we're sure that this filter has continuous sequence of messages, skip looking up other filters.
Expand All @@ -3311,33 +3313,38 @@ func (o *consumer) getNextMsg() (*jsPubMsg, uint64, error) {
// even if len == 0 or 1.
// TODO(tp): we should have sort based off generics for server
// to avoid reflection.
if o.subjf != nil && len(o.subjf) > 1 {
if len(o.subjf) > 1 {
sort.Slice(o.subjf, func(i, j int) bool {
if o.subjf[j].pmsg != nil && o.subjf[i].pmsg == nil {
derekcollison marked this conversation as resolved.
Show resolved Hide resolved
return false
}
if o.subjf[i].pmsg != nil && o.subjf[j].pmsg == nil {
return true
}
return o.subjf[j].nextSeq > o.subjf[i].nextSeq
})
}

// Grab next message applicable to us.
// Sort sequences first, to grab the first message.
for _, filter := range o.subjf {
// set o.sseq to the first subject sequence
if filter.nextSeq > o.sseq {
o.sseq = filter.nextSeq
}
// This means we got a message in this subject fetched.
if filter.pmsg != nil {
filter.currentSeq = filter.nextSeq
o.sseq = filter.currentSeq
returned := filter.pmsg
filter.pmsg = nil
return returned, 1, filter.err
}
if filter.err != nil {
lastErr = filter.err
}
filter := o.subjf[0]
derekcollison marked this conversation as resolved.
Show resolved Hide resolved
err := filter.err
// This means we got a message in this subject fetched.
if filter.pmsg != nil {
filter.currentSeq = filter.nextSeq
o.sseq = filter.currentSeq
returned := filter.pmsg
filter.pmsg = nil
return returned, 1, err
}
if err == ErrStoreEOF {
o.updateSkipped(filter.nextSeq)
}

return nil, 0, lastErr
// set o.sseq to the first subject sequence
if filter.nextSeq > o.sseq {
o.sseq = filter.nextSeq
}
return nil, 0, err
}

// Will check for expiration and lack of interest on waiting requests.
Expand Down