Try and remove blocking operations that can deadlock and data races

eclipse-paho#122

Author: alsm

client.go

@@ -20,15 +20,14 @@ import (
 	"fmt"
 	"net"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/eclipse/paho.mqtt.golang/packets"
 )
 
-type connStatus uint
-
 const (
-	disconnected connStatus = iota
+	disconnected uint32 = iota
 	connecting
 	reconnecting
 	connected
@@ -78,10 +77,10 @@ type client struct {
 	stop            chan struct{}
 	persist         Store
 	options         ClientOptions
-	pingResp        chan struct{}
-	packetResp      chan struct{}
-	keepaliveReset  chan struct{}
-	status          connStatus
+	pingResp        *sync.Cond
+	packetResp      *sync.Cond
+	keepaliveReset  *sync.Cond
+	status          uint32
 	workers         sync.WaitGroup
 }
 
@@ -125,26 +124,28 @@ func (c *client) AddRoute(topic string, callback MessageHandler) {
 func (c *client) IsConnected() bool {
 	c.RLock()
 	defer c.RUnlock()
+	status := atomic.LoadUint32(&c.status)
 	switch {
-	case c.status == connected:
+	case status == connected:
 		return true
-	case c.options.AutoReconnect && c.status > disconnected:
+	case c.options.AutoReconnect && status > disconnected:
 		return true
 	default:
 		return false
 	}
 }
 
-func (c *client) connectionStatus() connStatus {
+func (c *client) connectionStatus() uint32 {
 	c.RLock()
 	defer c.RUnlock()
-	return c.status
+	status := atomic.LoadUint32(&c.status)
+	return status
 }
 
-func (c *client) setConnected(status connStatus) {
+func (c *client) setConnected(status uint32) {
 	c.Lock()
 	defer c.Unlock()
-	c.status = status
+	atomic.StoreUint32(&c.status, uint32(status))
 }
 
 //ErrNotConnected is the error returned from function calls that are
@@ -236,9 +237,9 @@ func (c *client) Connect() Token {
 		c.ibound = make(chan packets.ControlPacket)
 		c.errors = make(chan error, 1)
 		c.stop = make(chan struct{})
-		c.pingResp = make(chan struct{}, 1)
-		c.packetResp = make(chan struct{}, 1)
-		c.keepaliveReset = make(chan struct{}, 1)
+		c.pingResp = sync.NewCond(&sync.Mutex{})
+		c.packetResp = sync.NewCond(&sync.Mutex{})
+		c.keepaliveReset = sync.NewCond(&sync.Mutex{})
 
 		if c.options.KeepAlive != 0 {
 			c.workers.Add(1)
@@ -336,7 +337,7 @@ func (c *client) reconnect() {
 		}
 	}
 	// Disconnect() must have been called while we were trying to reconnect.
-	if c.status == disconnected {
+	if c.connectionStatus() == disconnected {
 		DEBUG.Println(CLI, "Client moved to disconnected state while reconnecting, abandoning reconnect")
 		return
 	}
@@ -393,7 +394,8 @@ func (c *client) connect() byte {
 // the specified number of milliseconds to wait for existing work to be
 // completed.
 func (c *client) Disconnect(quiesce uint) {
-	if c.status == connected {
+	status := atomic.LoadUint32(&c.status)
+	if status == connected {
 		DEBUG.Println(CLI, "disconnecting")
 		c.setConnected(disconnected)
 

fvt_client_test.go

@@ -866,6 +866,7 @@ func Test_PublishEmptyMessage(t *testing.T) {
 	wait(choke)
 
 	p.Disconnect(250)
+	s.Disconnect(250)
 }
 
 // func Test_Cleanstore(t *testing.T) {
@@ -962,14 +963,14 @@ func Test_ping1_idle5(t *testing.T) {
 	ops.SetConnectionLostHandler(func(c Client, err error) {
 		t.Fatalf("Connection-lost handler was called: %s", err)
 	})
-	ops.SetKeepAlive(2 * time.Second)
+	ops.SetKeepAlive(3 * time.Second)
 
 	c := NewClient(ops)
 
 	if token := c.Connect(); token.Wait() && token.Error() != nil {
 		t.Fatalf("Error on Client.Connect(): %v", token.Error())
 	}
-	time.Sleep(5 * time.Second)
+	time.Sleep(8 * time.Second)
 	c.Disconnect(250)
 }
 

net.go

@@ -125,7 +125,7 @@ func incoming(c *client) {
 		case c.ibound <- cp:
 			// Notify keepalive logic that we recently received a packet
 			if c.options.KeepAlive != 0 {
-				c.packetResp <- struct{}{}
+				c.packetResp.Broadcast()
 			}
 		case <-c.stop:
 			// This avoids a deadlock should a message arrive while shutting down.
@@ -205,11 +205,7 @@ func outgoing(c *client) {
 		}
 		// Reset ping timer after sending control packet.
 		if c.options.KeepAlive != 0 {
-			select {
-			case c.keepaliveReset <- struct{}{}:
-			default:
-				DEBUG.Println(NET, "couldn't send keepalive signal in outbound as channel full")
-			}
+			c.keepaliveReset.Broadcast()
 		}
 	}
 }
@@ -232,7 +228,7 @@ func alllogic(c *client) {
 			switch m := msg.(type) {
 			case *packets.PingrespPacket:
 				DEBUG.Println(NET, "received pingresp")
-				c.pingResp <- struct{}{}
+				c.pingResp.Broadcast()
 			case *packets.SubackPacket:
 				DEBUG.Println(NET, "received suback, id:", m.MessageID)
 				token := c.getToken(m.MessageID).(*SubscribeToken)
@@ -257,18 +253,27 @@ func alllogic(c *client) {
 					pr := packets.NewControlPacket(packets.Pubrec).(*packets.PubrecPacket)
 					pr.MessageID = m.MessageID
 					DEBUG.Println(NET, "putting pubrec msg on obound")
-					c.oboundP <- &PacketAndToken{p: pr, t: nil}
+					select {
+					case c.oboundP <- &PacketAndToken{p: pr, t: nil}:
+					case <-c.stop:
+					}
 					DEBUG.Println(NET, "done putting pubrec msg on obound")
 				case 1:
 					c.incomingPubChan <- m
 					DEBUG.Println(NET, "done putting msg on incomingPubChan")
 					pa := packets.NewControlPacket(packets.Puback).(*packets.PubackPacket)
 					pa.MessageID = m.MessageID
 					DEBUG.Println(NET, "putting puback msg on obound")
-					c.oboundP <- &PacketAndToken{p: pa, t: nil}
+					select {
+					case c.oboundP <- &PacketAndToken{p: pa, t: nil}:
+					case <-c.stop:
+					}
 					DEBUG.Println(NET, "done putting puback msg on obound")
 				case 0:
-					c.incomingPubChan <- m
+					select {
+					case c.incomingPubChan <- m:
+					case <-c.stop:
+					}
 					DEBUG.Println(NET, "done putting msg on incomingPubChan")
 				}
 			case *packets.PubackPacket:
@@ -283,15 +288,15 @@ func alllogic(c *client) {
 				prel.MessageID = m.MessageID
 				select {
 				case c.oboundP <- &PacketAndToken{p: prel, t: nil}:
-				case <-time.After(time.Second):
+				case <-c.stop:
 				}
 			case *packets.PubrelPacket:
 				DEBUG.Println(NET, "received pubrel, id:", m.MessageID)
 				pc := packets.NewControlPacket(packets.Pubcomp).(*packets.PubcompPacket)
 				pc.MessageID = m.MessageID
 				select {
 				case c.oboundP <- &PacketAndToken{p: pc, t: nil}:
-				case <-time.After(time.Second):
+				case <-c.stop:
 				}
 			case *packets.PubcompPacket:
 				DEBUG.Println(NET, "received pubcomp, id:", m.MessageID)

options_reader.go

@@ -25,6 +25,7 @@ type ClientOptionsReader struct {
 	options *ClientOptions
 }
 
+//Servers returns a slice of the servers defined in the clientoptions
 func (r *ClientOptionsReader) Servers() []*url.URL {
 	s := make([]*url.URL, len(r.options.Servers))
 
@@ -36,21 +37,25 @@ func (r *ClientOptionsReader) Servers() []*url.URL {
 	return s
 }
 
+//ClientID returns the set client id
 func (r *ClientOptionsReader) ClientID() string {
 	s := r.options.ClientID
 	return s
 }
 
+//Username returns the set username
 func (r *ClientOptionsReader) Username() string {
 	s := r.options.Username
 	return s
 }
 
+//Password returns the set password
 func (r *ClientOptionsReader) Password() string {
 	s := r.options.Password
 	return s
 }
 
+//CleanSession returns whether Cleansession is set
 func (r *ClientOptionsReader) CleanSession() bool {
 	s := r.options.CleanSession
 	return s

ping.go

@@ -16,6 +16,7 @@ package mqtt
 
 import (
 	"errors"
+	"sync"
 	"time"
 
 	"github.com/eclipse/paho.mqtt.golang/packets"
@@ -24,32 +25,81 @@ import (
 func keepalive(c *client) {
 	DEBUG.Println(PNG, "keepalive starting")
 
+	var condWG sync.WaitGroup
+
+	defer func() {
+		c.keepaliveReset.Broadcast()
+		c.pingResp.Broadcast()
+		c.packetResp.Broadcast()
+		condWG.Wait()
+		c.workers.Done()
+	}()
+
 	receiveInterval := c.options.KeepAlive + (1 * time.Second)
 	pingTimer := timer{Timer: time.NewTimer(c.options.KeepAlive)}
 	receiveTimer := timer{Timer: time.NewTimer(receiveInterval)}
 	pingRespTimer := timer{Timer: time.NewTimer(c.options.PingTimeout)}
 
 	pingRespTimer.Stop()
 
+	condWG.Add(3)
+	go func() {
+		defer condWG.Done()
+		for {
+			c.pingResp.L.Lock()
+			c.pingResp.Wait()
+			c.pingResp.L.Unlock()
+			select {
+			case <-c.stop:
+				return
+			default:
+			}
+			DEBUG.Println(NET, "resetting ping timeout timer")
+			pingRespTimer.Stop()
+			pingTimer.Reset(c.options.KeepAlive)
+			receiveTimer.Reset(receiveInterval)
+		}
+	}()
+
+	go func() {
+		defer condWG.Done()
+		for {
+			c.packetResp.L.Lock()
+			c.packetResp.Wait()
+			c.packetResp.L.Unlock()
+			select {
+			case <-c.stop:
+				return
+			default:
+			}
+			DEBUG.Println(NET, "resetting receive timer")
+			receiveTimer.Reset(receiveInterval)
+		}
+	}()
+
+	go func() {
+		defer condWG.Done()
+		for {
+			c.keepaliveReset.L.Lock()
+			c.keepaliveReset.Wait()
+			c.keepaliveReset.L.Unlock()
+			select {
+			case <-c.stop:
+				return
+			default:
+			}
+			DEBUG.Println(NET, "resetting ping timer")
+			pingTimer.Reset(c.options.KeepAlive)
+		}
+	}()
+
 	for {
 		select {
 		case <-c.stop:
 			DEBUG.Println(PNG, "keepalive stopped")
-			c.workers.Done()
 			return
 		case <-pingTimer.C:
 			sendPing(&pingTimer, &pingRespTimer, c)
-		case <-c.keepaliveReset:
-			DEBUG.Println(NET, "resetting ping timer")
-			pingTimer.Reset(c.options.KeepAlive)
-		case <-c.pingResp:
-			DEBUG.Println(NET, "resetting ping timeout timer")
-			pingRespTimer.Stop()
-			pingTimer.Reset(c.options.KeepAlive)
-			receiveTimer.Reset(receiveInterval)
-		case <-c.packetResp:
-			DEBUG.Println(NET, "resetting receive timer")
-			receiveTimer.Reset(receiveInterval)
 		case <-receiveTimer.C:
 			receiveTimer.SetRead(true)
 			receiveTimer.Reset(receiveInterval)
@@ -66,16 +116,21 @@ func keepalive(c *client) {
 }
 
 type timer struct {
+	sync.Mutex
 	*time.Timer
 	readFrom bool
 }
 
 func (t *timer) SetRead(v bool) {
+	t.Lock()
 	t.readFrom = v
+	t.Unlock()
 }
 
 func (t *timer) Stop() bool {
+	t.Lock()
 	defer t.SetRead(true)
+	defer t.Unlock()
 
 	if !t.Timer.Stop() && !t.readFrom {
 		<-t.C
@@ -85,8 +140,13 @@ func (t *timer) Stop() bool {
 }
 
 func (t *timer) Reset(d time.Duration) bool {
+	t.Lock()
 	defer t.SetRead(false)
-	t.Stop()
+	defer t.Unlock()
+	if !t.Timer.Stop() && !t.readFrom {
+		<-t.C
+	}
+
 	return t.Timer.Reset(d)
 }