Disconnect() can hang waiting on c.workers.Wait() #554
Comments
|
It looks like your issue is here: router.go:210 is: So one of your handlers is not returning. As per common problems in the readme:
I'm really not sure that the library can handle this any differently - |
|
I see, thanks for the pointer! It turns out that the handler is signaling on a channel that is insufficiently deep and blocking as a result. To elaborate further on the scenario, I have an adapter around MQTT that may be passed arbitrary handler functions and I would like to be able to recover the client by stopping and bringing up new instances of it. Let's say I wrap the handler to include a cancellation context controlled by the adapter. When I cancel the context and call Disconnect, what resources, if any, would be leaked in the mqtt lib by this approach? On a related note, while the handler seen by the mqtt lib does complete in this case (cancellation forces progress) it looks like router.go:210 always calls |
None if it shuts down properly (i.e. cancelling your context also terminates any callbacks). If the callback is blocked then go-routines are leaked (along with any resources they hold). Due to the way this library was written (over a long period by quite a few people) its not particularly easy to modify and keeping track of the various go-routines and what blocks what is difficult. This is one of the reasons that the mqtt v5 client is a total (incompatible) rewrite. I rewrote a large portion of this library a couple of years ago (simplifying, documenting, and removing a range of potential deadlocks) but was constrained by the need to avoid breaking existing clients.
See issue #459 for a discussion on this point (there was also a discussion on the Paho mailing list). This package was originally designed on the basis that the ACK is a protocol level thing (so once the message has been received the ACK should be sent regardless of any issues the user may have processing the message). As per the comments on that issue I'm open to seeing this change implemented but it's not something I'm really interested in doing myself (have no use-case). |
|
Thanks @MattBrittan, this was incredibly helpful. I'll keep an eye on eclipse-paho/paho.golang#53 and the v5 betas. |
Related to #550, but the hang is not at the previously noted race on
<-c.commsStoppedbut prior to that on line 672 atc.workers.Wait(): https://github.com/eclipse/paho.mqtt.golang/blob/88d53342c440d7e9fb3b2927188e6bd8645dce8c/client.go#L670-L680The repro also does not involve attempting to reuse a client via
Disconnect/Reconnect:SetCleanSession(False)(not salient, hang reproes either way)Connectthe client and callSubscribewith a QoS level 1Disconnecton the client as part of teardownIs there an idiomatic way of releasing the client and any resources it holds other than calling
Disconnecton it?It looks like sometimes after
c.stopis signaled as part ofDisconnect, the goroutine responsible for flushing any acks doesn't end up getting toc.workers.Done():https://github.com/eclipse/paho.mqtt.golang/blob/88d53342c440d7e9fb3b2927188e6bd8645dce8c/client.go#L579-L586
Relevant stacks I'm basing that hypothesis on highlighted below:
goroutine 99 [running]: testing.(*M).startAlarm.func1() /usr/local/go/src/testing/testing.go:1788 +0xbb created by time.goFunc /usr/local/go/src/time/sleep.go:180 +0x4a goroutine 1 [chan receive, 2 minutes]: testing.(*T).Run(0xc000501380, {0x3459033, 0x17}, 0x3531160) /usr/local/go/src/testing/testing.go:1307 +0x752 testing.runTests.func1(0x0) /usr/local/go/src/testing/testing.go:1598 +0x9a testing.tRunner(0xc000501380, 0xc00065fbf8) /usr/local/go/src/testing/testing.go:1259 +0x230 testing.runTests(0xc000126800, {0x5654000, 0x11, 0x11}, {0x0, 0xc00032f340, 0x5781700}) /usr/local/go/src/testing/testing.go:1596 +0x7cb testing.(*M).Run(0xc000126800) /usr/local/go/src/testing/testing.go:1504 +0x9d2 main.main() _testmain.go:75 +0x22c goroutine 34 [chan receive]: k8s.io/klog/v2.(*loggingT).flushDaemon(0x0) /go/pkg/mod/k8s.io/klog/v2@v2.4.0/klog.go:1169 +0x8b created by k8s.io/klog/v2.init.0 /go/pkg/mod/k8s.io/klog/v2@v2.4.0/klog.go:417 +0x1c5 goroutine 52 [chan receive, 2 minutes]: testing.(*T).Run(0xc000501520, {0xc000058d20, 0x13}, 0xc0001b9410) /usr/local/go/src/testing/testing.go:1307 +0x752 github.com/dapr/components-contrib/tests/conformance.(*TestConfiguration).Run(0xc000173fb0, 0xc000501520) /go/src/github.com/dapr/components-contrib/tests/conformance/common.go:266 +0x2b7 github.com/dapr/components-contrib/tests/conformance.TestBindingsConformance(0x0) /go/src/github.com/dapr/components-contrib/tests/conformance/bindings_test.go:20 +0x94 testing.tRunner(0xc000501520, 0x3531160) /usr/local/go/src/testing/testing.go:1259 +0x230 created by testing.(*T).Run /usr/local/go/src/testing/testing.go:1306 +0x727 goroutine 53 [chan receive]: testing.(*T).Run(0xc000501a00, {0x343af9e, 0x5}, 0xc000706c00) /usr/local/go/src/testing/testing.go:1307 +0x752 github.com/dapr/components-contrib/tests/conformance/bindings.ConformanceTests(0xc0000522bc, 0xc000039080, {0x423a050, 0xc00066b980}, {0x4240c40, 0xc00066b9e0}, {{{0x343e22c, 0x8}, {0xc0000522bc, 0x4}, ...}, ...}) /go/src/github.com/dapr/components-contrib/tests/conformance/bindings/bindings.go:260 +0xef2 github.com/dapr/components-contrib/tests/conformance.(*TestConfiguration).Run.func1(0xc000501a00) /go/src/github.com/dapr/components-contrib/tests/conformance/common.go:332 +0x10e8 testing.tRunner(0xc000501a00, 0xc0001b9410) /usr/local/go/src/testing/testing.go:1259 +0x230 created by testing.(*T).Run /usr/local/go/src/testing/testing.go:1306 +0x727 goroutine 97 [chan receive]: github.com/eclipse/paho%2emqtt%2egolang.(*client).disconnect(0xc000032000) /go/pkg/mod/github.com/eclipse/paho.mqtt.golang@v1.3.5/client.go:481 +0x45 github.com/eclipse/paho%2emqtt%2egolang.(*client).Disconnect(0xc000032000, 0x1) /go/pkg/mod/github.com/eclipse/paho.mqtt.golang@v1.3.5/client.go:463 +0x792 github.com/dapr/components-contrib/bindings/mqtt.(*MQTT).Close(0xc00066b980) /go/src/github.com/dapr/components-contrib/bindings/mqtt/mqtt.go:328 +0x68 github.com/dapr/components-contrib/tests/conformance/bindings.ConformanceTests.func9(0x0) /go/src/github.com/dapr/components-contrib/tests/conformance/bindings/bindings.go:266 +0x207 testing.tRunner(0xc00070b380, 0xc000706c00) /usr/local/go/src/testing/testing.go:1259 +0x230 created by testing.(*T).Run /usr/local/go/src/testing/testing.go:1306 +0x727 goroutine 66 [syscall]: os/signal.signal_recv() /usr/local/go/src/runtime/sigqueue.go:169 +0x98 os/signal.loop() /usr/local/go/src/os/signal/signal_unix.go:24 +0x25 created by os/signal.Notify.func1.1 /usr/local/go/src/os/signal/signal.go:151 +0x53 goroutine 98 [semacquire]: sync.runtime_Semacquire(0xc0000321f8) /usr/local/go/src/runtime/sema.go:56 +0x25 sync.(*WaitGroup).Wait(0xc0000321f0) /usr/local/go/src/sync/waitgroup.go:130 +0xea github.com/eclipse/paho%2emqtt%2egolang.(*client).stopCommsWorkers.func1() /go/pkg/mod/github.com/eclipse/paho.mqtt.golang@v1.3.5/client.go:672 +0x108 created by github.com/eclipse/paho%2emqtt%2egolang.(*client).stopCommsWorkers /go/pkg/mod/github.com/eclipse/paho.mqtt.golang@v1.3.5/client.go:670 +0x265 goroutine 87 [chan send]: github.com/dapr/components-contrib/tests/conformance/bindings.ConformanceTests.func3.1.1(0xc0001b8588) /go/src/github.com/dapr/components-contrib/tests/conformance/bindings/bindings.go:167 +0x86 github.com/dapr/components-contrib/bindings/mqtt.(*MQTT).Read.func1({0xc0007060f0, 0x0}, {0x4285e00, 0xc00012e320}) /go/src/github.com/dapr/components-contrib/bindings/mqtt/mqtt.go:237 +0x254 github.com/eclipse/paho%2emqtt%2egolang.(*router).matchAndDispatch.func2() /go/pkg/mod/github.com/eclipse/paho.mqtt.golang@v1.3.5/router.go:210 +0xf03 created by github.com/eclipse/paho%2emqtt%2egolang.(*router).matchAndDispatch /go/pkg/mod/github.com/eclipse/paho.mqtt.golang@v1.3.5/router.go:169 +0x558 goroutine 88 [chan receive]: github.com/eclipse/paho%2emqtt%2egolang.(*client).startCommsWorkers.func1() /go/pkg/mod/github.com/eclipse/paho.mqtt.golang@v1.3.5/client.go:582 +0x50d created by github.com/eclipse/paho%2emqtt%2egolang.(*client).startCommsWorkers /go/pkg/mod/github.com/eclipse/paho.mqtt.golang@v1.3.5/client.go:564 +0xa14 goroutine 89 [chan send]: github.com/eclipse/paho%2emqtt%2egolang.startIncoming.func1() /go/pkg/mod/github.com/eclipse/paho.mqtt.golang@v1.3.5/net.go:131 +0xdb created by github.com/eclipse/paho%2emqtt%2egolang.startIncoming /go/pkg/mod/github.com/eclipse/paho.mqtt.golang@v1.3.5/net.go:117 +0x2b9 goroutine 90 [chan send]: github.com/eclipse/paho%2emqtt%2egolang.startIncomingComms.func1() /go/pkg/mod/github.com/eclipse/paho.mqtt.golang@v1.3.5/net.go:219 +0xdd9 created by github.com/eclipse/paho%2emqtt%2egolang.startIncomingComms /go/pkg/mod/github.com/eclipse/paho.mqtt.golang@v1.3.5/net.go:159 +0x2fa goroutine 91 [select]: github.com/eclipse/paho%2emqtt%2egolang.startOutgoingComms.func1() /go/pkg/mod/github.com/eclipse/paho.mqtt.golang@v1.3.5/net.go:272 +0x326 created by github.com/eclipse/paho%2emqtt%2egolang.startOutgoingComms /go/pkg/mod/github.com/eclipse/paho.mqtt.golang@v1.3.5/net.go:259 +0x3da goroutine 92 [chan send]: github.com/eclipse/paho%2emqtt%2egolang.startComms.func1() /go/pkg/mod/github.com/eclipse/paho.mqtt.golang@v1.3.5/net.go:411 +0xe5 created by github.com/eclipse/paho%2emqtt%2egolang.startComms /go/pkg/mod/github.com/eclipse/paho.mqtt.golang@v1.3.5/net.go:400 +0x3c8 goroutine 93 [chan receive]: github.com/eclipse/paho%2emqtt%2egolang.startComms.func2() /go/pkg/mod/github.com/eclipse/paho.mqtt.golang@v1.3.5/net.go:424 +0x7d created by github.com/eclipse/paho%2emqtt%2egolang.startComms /go/pkg/mod/github.com/eclipse/paho.mqtt.golang@v1.3.5/net.go:423 +0x4d4 goroutine 94 [semacquire]: sync.runtime_Semacquire(0xc0000c0554) /usr/local/go/src/runtime/sema.go:56 +0x25 sync.(*WaitGroup).Wait(0xc0000c0554) /usr/local/go/src/sync/waitgroup.go:130 +0xea github.com/eclipse/paho%2emqtt%2egolang.startComms.func3() /go/pkg/mod/github.com/eclipse/paho.mqtt.golang@v1.3.5/net.go:432 +0x3e created by github.com/eclipse/paho%2emqtt%2egolang.startComms /go/pkg/mod/github.com/eclipse/paho.mqtt.golang@v1.3.5/net.go:431 +0x585 goroutine 95 [select]: github.com/eclipse/paho%2emqtt%2egolang.(*client).startCommsWorkers.func2() /go/pkg/mod/github.com/eclipse/paho.mqtt.golang@v1.3.5/client.go:613 +0x5a5 created by github.com/eclipse/paho%2emqtt%2egolang.(*client).startCommsWorkers /go/pkg/mod/github.com/eclipse/paho.mqtt.golang@v1.3.5/client.go:597 +0xcf4The text was updated successfully, but these errors were encountered: