Simplify waiter implementation, eliminate sync.Cond (#3629)

provider/mqtt.go

@@ -107,7 +107,7 @@ func (m *Mqtt) newReceiver() *msgHandler {
 	h := &msgHandler{
 		topic:    m.topic,
 		scale:    m.scale,
-		mux:      util.NewWaiter(m.timeout, func() { m.log.DEBUG.Printf("%s wait for initial value", m.topic) }),
+		wait:     util.NewWaiter(m.timeout, func() { m.log.DEBUG.Printf("%s wait for initial value", m.topic) }),
 		pipeline: m.pipeline,
 	}
 

provider/mqtt_handler.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"math"
 	"strconv"
+	"sync"
 	"time"
 
 	"github.com/evcc-io/evcc/provider/pipeline"
@@ -11,7 +12,8 @@ import (
 )
 
 type msgHandler struct {
-	mux      *util.Waiter
+	mux      sync.Mutex
+	wait     *util.Waiter
 	scale    float64
 	topic    string
 	pipeline *pipeline.Pipeline
@@ -23,18 +25,18 @@ func (h *msgHandler) receive(payload string) {
 	defer h.mux.Unlock()
 
 	h.payload = payload
-	h.mux.Update()
+	h.wait.Update()
 }
 
 // hasValue returned the received and processed payload as string
 func (h *msgHandler) hasValue() (string, error) {
-	h.mux.Lock()
-	defer h.mux.Unlock()
-
-	if late := h.mux.Overdue(); late > 0 {
+	if late := h.wait.Overdue(); late > 0 {
 		return "", fmt.Errorf("%s outdated: %v", h.topic, late.Truncate(time.Second))
 	}
 
+	h.mux.Lock()
+	defer h.mux.Unlock()
+
 	if h.pipeline != nil {
 		b, err := h.pipeline.Process([]byte(h.payload))
 		return string(b), err

provider/sma/device.go

@@ -15,7 +15,8 @@ type Device struct {
 	*sunny.Device
 
 	log    *util.Logger
-	mux    *util.Waiter
+	mux    sync.Mutex
+	wait   *util.Waiter
 	values map[sunny.ValueID]interface{}
 	once   sync.Once
 }
@@ -40,7 +41,7 @@ func (d *Device) UpdateValues() error {
 	values, err := d.Device.GetValues()
 	if err == nil {
 		err = mergo.Merge(&d.values, values, mergo.WithOverride)
-		d.mux.Update()
+		d.wait.Update()
 	}
 
 	return err
@@ -53,7 +54,7 @@ func (d *Device) Values() (map[sunny.ValueID]interface{}, error) {
 	d.mux.Lock()
 	defer d.mux.Unlock()
 
-	if late := d.mux.Overdue(); late > 0 {
+	if late := d.wait.Overdue(); late > 0 {
 		return nil, fmt.Errorf("update timeout: %v", late.Truncate(time.Second))
 	}
 

provider/sma/discover.go

@@ -65,7 +65,7 @@ func (d *Discoverer) createDevice(device *sunny.Device) *Device {
 	return &Device{
 		Device: device,
 		log:    d.log,
-		mux:    util.NewWaiter(udpTimeout, func() { d.log.DEBUG.Println("wait for initial value") }),
+		wait:   util.NewWaiter(udpTimeout, func() { d.log.DEBUG.Println("wait for initial value") }),
 		values: make(map[sunny.ValueID]interface{}),
 	}
 }

provider/socket.go

@@ -5,6 +5,7 @@ import (
 	"math"
 	"net/http"
 	"strconv"
+	"sync"
 	"time"
 
 	"github.com/evcc-io/evcc/util"
@@ -21,7 +22,8 @@ const retryDelay = 5 * time.Second
 type Socket struct {
 	*request.Helper
 	log     *util.Logger
-	mux     *util.Waiter
+	mux     sync.Mutex
+	wait    *util.Waiter
 	url     string
 	headers map[string]string
 	scale   float64
@@ -62,7 +64,7 @@ func NewSocketProviderFromConfig(other map[string]interface{}) (IntProvider, err
 	p := &Socket{
 		log:     log,
 		Helper:  request.NewHelper(log),
-		mux:     util.NewWaiter(cc.Timeout, func() { log.DEBUG.Println("wait for initial value") }),
+		wait:    util.NewWaiter(cc.Timeout, func() { log.DEBUG.Println("wait for initial value") }),
 		url:     url,
 		headers: cc.Headers,
 		scale:   cc.Scale,
@@ -129,11 +131,11 @@ func (p *Socket) listen() {
 				v, err := jq.Query(p.jq, b)
 				if err == nil {
 					p.val = v
-					p.mux.Update()
+					p.wait.Update()
 				}
 			} else {
 				p.val = string(b)
-				p.mux.Update()
+				p.wait.Update()
 			}
 			p.mux.Unlock()
 		}
@@ -141,13 +143,13 @@ func (p *Socket) listen() {
 }
 
 func (p *Socket) hasValue() (interface{}, error) {
-	p.mux.Lock()
-	defer p.mux.Unlock()
-
-	if late := p.mux.Overdue(); late > 0 {
+	if late := p.wait.Overdue(); late > 0 {
 		return nil, fmt.Errorf("outdated: %v", late.Truncate(time.Second))
 	}
 
+	p.mux.Lock()
+	defer p.mux.Unlock()
+
 	return p.val, nil
 }
 

util/waiter.go

@@ -9,63 +9,54 @@ var waitInitialTimeout = 10 * time.Second
 
 // Waiter provides monitoring of receive timeouts and reception of initial value
 type Waiter struct {
-	sync.Mutex
+	mu      sync.Mutex
 	log     func()
-	cond    *sync.Cond
 	updated time.Time
 	timeout time.Duration
+	initial chan bool
 }
 
 // NewWaiter creates new waiter
 func NewWaiter(timeout time.Duration, logInitialWait func()) *Waiter {
-	p := &Waiter{
+	return &Waiter{
 		log:     logInitialWait,
 		timeout: timeout,
+		initial: make(chan bool),
 	}
-	p.cond = sync.NewCond(p)
-	return p
 }
 
 // Update is called when client has received data. Update resets the timeout counter.
-// Waiter MUST be locked when calling Update.
 func (p *Waiter) Update() {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+
 	p.updated = time.Now()
-	p.cond.Broadcast()
+
+	select {
+	case <-p.initial:
+	default:
+		close(p.initial)
+	}
 }
 
 // Overdue waits for initial update and returns the duration since the last update
-// in excess of timeout. Waiter MUST be locked when calling Overdue.
+// in excess of timeout.
 func (p *Waiter) Overdue() time.Duration {
-	if p.updated.IsZero() {
+	select {
+	case <-p.initial:
+	default:
 		p.log()
 
-		c := make(chan struct{})
-
-		go func() {
-			defer close(c)
-			p.Lock() // establish lock once go routine has started
-			for p.updated.IsZero() {
-				p.cond.Wait()
-			}
-		}()
-
-		// release lock so external updates can occur
-		p.Unlock()
-
 		select {
-		case <-c:
-			// initial value received, lock established
+		case <-p.initial:
 		case <-time.After(waitInitialTimeout):
-			// establish lock per contract of `Update()``
-			p.Lock()
-			p.Update() // unblock the sync.Cond
-			p.Unlock()
-			<-c                     // wait for goroutine, re-establish lock
-			p.updated = time.Time{} // reset to "initial value missing"
 			return waitInitialTimeout
 		}
 	}
 
+	p.mu.Lock()
+	defer p.mu.Unlock()
+
 	if elapsed := time.Since(p.updated); p.timeout != 0 && elapsed > p.timeout {
 		return elapsed
 	}

util/waiter_test.go

@@ -19,16 +19,12 @@ func TestWaiterInitialUpdateInTime(t *testing.T) {
 
 		go func() {
 			time.Sleep(testTimeout / 2)
-			w.Lock()
 			w.Update()
-			w.Unlock()
 		}()
 
-		w.Lock()
 		if elapsed := w.Overdue(); elapsed != 0 {
 			t.Errorf("expected %v, got %v", 0, elapsed)
 		}
-		w.Unlock()
 	}
 }
 
@@ -36,30 +32,21 @@ func TestWaiterInitialUpdateNotReceived(t *testing.T) {
 	for _, timeout := range []time.Duration{0, testTimeout} {
 		w := NewWaiter(timeout, func() {})
 
-		w.Lock()
 		if elapsed := w.Overdue(); elapsed != waitInitialTimeout {
 			t.Errorf("expected %v, got %v", waitInitialTimeout, elapsed)
 		}
-		w.Unlock()
 	}
 }
 
 func TestWaiterUpdateInTime(t *testing.T) {
 	w := NewWaiter(testTimeout, func() {})
-	w.Lock()
 	w.Update()
-	w.Unlock()
 
 	go func() {
 		time.Sleep(testTimeout / 2)
-		w.Lock()
 		w.Update()
-		w.Unlock()
 	}()
 
-	w.Lock()
-	defer w.Unlock()
-
 	if elapsed := w.Overdue(); elapsed != 0 {
 		t.Errorf("expected %v, got %v", 0, elapsed)
 	}
@@ -67,15 +54,10 @@ func TestWaiterUpdateInTime(t *testing.T) {
 
 func TestWaiterUpdateNotReceived(t *testing.T) {
 	w := NewWaiter(testTimeout, func() {})
-	w.Lock()
 	w.Update()
-	w.Unlock()
 
 	time.Sleep(2 * testTimeout)
 
-	w.Lock()
-	defer w.Unlock()
-
 	if elapsed := w.Overdue(); elapsed < 2*testTimeout {
 		t.Errorf("expected >%v, got %v", 2*testTimeout, elapsed)
 	}