goroutine leak when connectionUp(true) return error

[kiqi007]

Issue Description: Encountered a goroutine leak scenario when using the MQTT package.

    for {
        select {
        case <-m.checkDone:
            hlog.Info("mqtt check done")
            return
        default:
            if m.client == nil || !m.client.IsConnectionOpen() {
                if nil != m.client {
                    m.client.Disconnect(1000)
                }
                err := m.Connect() // create a new client, and do connect
                if nil != err {
                    hlog.Error("mqtt Connect error", elog.String("host", m.host), elog.Int("port", m.port), elog.FieldErr(err))
                } else {
                    hlog.Info("mqtt Connect success", elog.String("host", m.host), elog.Int("port", m.port))
                }
            }
            <-m.connCheckTicker.C
        }
    }

gorouine pprof:
goroutine profile: total 2787
1339 @ 0xe2e556 0xe3edac 0xe3ed86 0xe57ae5 0xe77032 0x16e6c25 0x16e43dc 0xe5ca61
#   0xe57ae4    sync.runtime_Semacquire+0x24                            /usr/local/go/src/runtime/sema.go:56
#   0xe77031    sync.(*WaitGroup).Wait+0x51                         /usr/local/go/src/sync/waitgroup.go:136
#   0x16e6c24   github.com/eclipse/paho%2emqtt%2egolang.(*client).startCommsWorkers+0x4a4   /home/jenkins/go/pkg/mod/github.com/eclipse/paho.mqtt.golang@v1.4.2/client.go:608
#   0x16e43db   github.com/eclipse/paho%2emqtt%2egolang.(*client).Connect.func1+0x49b       /home/jenkins/go/pkg/mod/github.com/eclipse/paho.mqtt.golang@v1.4.2/client.go:282

1339 @ 0xe2e556 0xe3edac 0xe3ed86 0xe57c05 0xe755a5 0x16e7cc9 0x16e7c9e 0x16e5cc5 0x16e5c06 0x16e5b87 0xe5ca61
#   0xe57c04    sync.runtime_SemacquireMutex+0x24                       /usr/local/go/src/runtime/sema.go:71
#   0xe755a4    sync.(*Mutex).lockSlow+0x164                            /usr/local/go/src/sync/mutex.go:162
#   0x16e7cc8   sync.(*Mutex).Lock+0xa8                             /usr/local/go/src/sync/mutex.go:81
#   0x16e7c9d   github.com/eclipse/paho%2emqtt%2egolang.(*client).stopCommsWorkers+0x7d     /home/jenkins/go/pkg/mod/github.com/eclipse/paho.mqtt.golang@v1.4.2/client.go:711
#   0x16e5cc4   github.com/eclipse/paho%2emqtt%2egolang.(*client).disconnect+0x24       /home/jenkins/go/pkg/mod/github.com/eclipse/paho.mqtt.golang@v1.4.2/client.go:498
#   0x16e5c05   github.com/eclipse/paho%2emqtt%2egolang.(*client).Disconnect.func1.1+0x25   /home/jenkins/go/pkg/mod/github.com/eclipse/paho.mqtt.golang@v1.4.2/client.go:451
#   0x16e5b86   github.com/eclipse/paho%2emqtt%2egolang.(*client).Disconnect.func1+0x4a6    /home/jenkins/go/pkg/mod/github.com/eclipse/paho.mqtt.golang@v1.4.2/client.go:470

Local Reproduction Steps: Perform the following code changes and simulate network fluctuations.

• Several key nodes are marked in the code through issue comments

// github.com/eclipse/paho.mqtt.golang/client.go:574 startCommsWorkers
        c.workers.Add(1)
        go keepalive(c, conn)  // issue:  This goroutine will change the connect status to disconnecting
    }

    // matchAndDispatch will process messages received from the network. It may generate acknowledgements
    // It will complete when incomingPubChan is closed and will close ackOut prior to exiting
    incomingPubChan := make(chan *packets.PublishPacket)
    c.workers.Add(1) // Done will be called when ackOut is closed
    ackOut := c.msgRouter.matchAndDispatch(incomingPubChan, c.options.Order, c)

    time.Sleep(10 * time.Second) // issue:  Simulate the situation where keepalive has already converted status to disconnecting, but connectionUp has not yet been executed

    // The connection is now ready for use (we spin up a few go routines below). It is possible that
    // Disconnect has been called in the interim...
    if err := connectionUp(true); err != nil {
        DEBUG.Println(CLI, err)
        close(c.stop) // Tidy up anything we have already started
        close(incomingPubChan)
        c.workers.Wait()       // issue: Goroutine will block at this location
        c.conn.Close()

Issue Cause: Within the code mentioned above, there is no call to Done() between Add() and Wait(), which causes the Wait() method to block indefinitely. To resolve this issue, a call to Done() needs to be properly placed.

[MattBrittan]

Hmmm, so we probably need to drain ackOut and then call c.workers.Done() before calling c.workers.Wait().

[kiqi007]

Hey there! Any idea when we might tackle this issue? Also, could anyone lend a hand with fixing it? I stumbled upon this while assisting my colleagues in debugging a memory leak, but I'm not very familiar with the MQTT library itself. So, if someone could help fix this problem, that would be great!

[MattBrittan]

@kiqi007 I flagged it as "Help Wanted" because I doubt it's something I will get around to in the foreseeable future (don't ever call Disconnect personally and am investing more time in the v5 client).