Changes to default pending and default error handler. #607
Conversation
Slow consumer state still seems off, this is a first step trying to improve. Increased the default number of pending messages limit but kept bytes the same. Also introduced a default ErrHandler that will print out to stderr in case none has been set. Signed-off-by: Derek Collison <derek@nats.io>
nats.go
Outdated
| if nc != nil { | ||
| nc.mu.RLock() | ||
| cid = nc.info.CID | ||
| defer nc.mu.RUnlock() |
There was a problem hiding this comment.
Is it intentional that you keep the connection's lock until the end of this function? Error handlers are already serialized, if that was the concern.
| sub := nc.subs[nc.ps.ma.sid] | ||
| nc.subsMu.RUnlock() |
There was a problem hiding this comment.
This function was protected by the subsMu so that remove could not happen while processing a message. I believe this is the reason we now can get the panic:
=== RUN TestCloseChanOnSubscriber
panic: runtime error: invalid memory address or nil pointer dereference
[signal SIGSEGV: segmentation violation code=0x1 addr=0x0 pc=0x4bcb30]
goroutine 6453 [running]:
sync.(*Cond).Signal(0x0)
/home/travis/.gimme/versions/go1.15.5.linux.amd64/src/sync/cond.go:65 +0x30
github.com/nats-io/nats%2ego.(*Conn).processMsg(0xc000702600, 0xc0005a000d, 0x5, 0x7ff3)
/home/travis/gopath/src/github.com/nats-io/nats.go/nats.go:2498 +0xc7a
github.com/nats-io/nats%2ego.(*Conn).parse(0xc000702600, 0xc0005a0000, 0x398, 0x8000, 0x398, 0x0)
/home/travis/gopath/src/github.com/nats-io/nats.go/parser.go:192 +0x26e5
github.com/nats-io/nats%2ego.(*Conn).readLoop(0xc000702600)
/home/travis/gopath/src/github.com/nats-io/nats.go/nats.go:2323 +0x279
created by github.com/nats-io/nats%2ego.(*Conn).processConnectInit
/home/travis/gopath/src/github.com/nats-io/nats.go/nats.go:1639 +0x32b
FAIL github.com/nats-io/nats.go/test 42.511s
You need to check for sub.closed line 2466, after lock the sub. If closed, return.
There was a problem hiding this comment.
By the way, for this test: TestUnsubscribeChanOnSubscriber, I also see the slow consumer error reported in some cases, so may have to set dummy error handler there too.
There was a problem hiding this comment.
ok will take a deeper look.
| @@ -908,10 +918,19 @@ func TestChanSubscriberPendingLimits(t *testing.T) { | |||
| switch typeSubs { | |||
| case 0: | |||
| sub, err = nc.ChanSubscribe("foo", ch) | |||
| if err := sub.SetPendingLimits(pending, -1); err == nil { | |||
There was a problem hiding this comment.
I thought limits did not apply to Channel subscribe, because the limit is really whatever the user sets for the provided channel?
There was a problem hiding this comment.
The test expects an error since its checking for err == nil but the t.Fatalf descriptions are wrong, will fix.
Signed-off-by: Derek Collison <derek@nats.io>
Signed-off-by: Derek Collison <derek@nats.io>
Slow consumer state still seems off, this is a first step trying to improve.
Increased the default number of pending messages limit but kept bytes the same.
Also introduced a default ErrHandler that will print out to stderr in case none has been set.
Someone needs to look deeper into why we are so imbalanced when receiving messages like queued up JetStream messages for a consumer.
Signed-off-by: Derek Collison derek@nats.io