Repository: evcc-io/evcc
Pull request #3629 (2022-06-12 11:07:07)

Simplify waiter implementation, eliminate sync.Cond
/cc @bboehmke könntest Du evtl. mit kurzem Test bestätigen, dass ich hier bei SMA nix kaputt gemacht habe?

bboehmke: Ich glaube dein Umbau von `StartUpdateLoop` nach `Run` hat den SMA provider kaputt gemacht. Ich habe es zwar noch nicht getestet meine aber da fehlt das starten von `Run` jetzt. (Das wurde vorher ja im zweifel beim ersten Aufruf `Values()` gemacht)

Die normale Nutzung als Meter scheint auf den ersten blick zu funktionieren
andig: Danke, hab den Teil zurück gedreht, hat egtl. mit dem PR auch nix zu tun...
bboehmke: Ich hab nochmal drüber geschaut und jetzt scheint es soweit zu passen

Patch:
diff --git a/provider/mqtt.go b/provider/mqtt.go
index 1fdfe801..b5786c71 100644
--- a/provider/mqtt.go
+++ b/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,
 	}
 
diff --git a/provider/mqtt_handler.go b/provider/mqtt_handler.go
index 4da97389..af44237c 100644
--- a/provider/mqtt_handler.go
+++ b/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
diff --git a/provider/sma/device.go b/provider/sma/device.go
index ae429880..d6644bbc 100644
--- a/provider/sma/device.go
+++ b/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))
 	}
 
diff --git a/provider/sma/discover.go b/provider/sma/discover.go
index fa55c35d..f891bbad 100644
--- a/provider/sma/discover.go
+++ b/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{}),
 	}
 }
diff --git a/provider/socket.go b/provider/socket.go
index 73e776da..4669d4bd 100644
--- a/provider/socket.go
+++ b/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
 }
 
diff --git a/util/waiter.go b/util/waiter.go
index d842716a..a6c0efce 100644
--- a/util/waiter.go
+++ b/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
 	}


Test patch:
diff --git a/util/waiter_test.go b/util/waiter_test.go
index 7ff903b9..2c6cdf4b 100644
--- a/util/waiter_test.go
+++ b/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)
 	}