Re The slow subscriber problem

This is an exciting topic for discussion.

Option 1

On one hand, we have blocking send current implementation:

func (state *state) send(msg interface{}, tags TagMap) {
    for q, clientToChannelMap := range state.queries {
        if q.Matches(tags) {
            for _, ch := range clientToChannelMap {
                ch <- msg
            }
        }
    }
}

If there's slow subscriber, all other subscribers must wait.

Jae's thoughts: #951 (comment)

Option 2

Non-blocking send

for _, ch := range clientToChannelMap {
        select {
        case ch <- msg:
        default:
            // implement retry logic, or probably just do nothing,
            // drop the message for this subscriber on this query
            // it seems fair if it couldn't keep up with its buffer
        }
    }

If there's a slow client, others do not wait because we skip it. The problem here is that slow client does not know it had missed the message and this can be a source of potentially hidden&hardtofind bugs.

libs/pubsub: use non-blocking send when dispatching messages #1880

Description

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions