From cd43f09263bbaeb8f9c3c598a101fb896940a171 Mon Sep 17 00:00:00 2001 From: andig Date: Sun, 12 Dec 2021 20:06:41 +0100 Subject: [PATCH] Make waiter always expect initial value even if timeout is zero (#2031) --- provider/mqtt.go | 8 +---- provider/mqtt_handler.go | 6 ++-- provider/sma/device.go | 6 ++-- provider/socket.go | 6 ++-- util/waiter.go | 48 ++++++++++++------------- util/waiter_test.go | 76 ++++++++++++++++++++++++++++++++++++++++ 6 files changed, 110 insertions(+), 40 deletions(-) create mode 100644 util/waiter_test.go diff --git a/provider/mqtt.go b/provider/mqtt.go index a92a88b7c..291b1d6f0 100644 --- a/provider/mqtt.go +++ b/provider/mqtt.go @@ -7,7 +7,6 @@ import ( "github.com/evcc-io/evcc/provider/mqtt" "github.com/evcc-io/evcc/util" - "github.com/evcc-io/evcc/util/request" "github.com/itchyny/gojq" ) @@ -117,15 +116,10 @@ var _ FloatProvider = (*Mqtt)(nil) // receiver will ensure actual data guarded by `timeout` and return error // if initial value is not received within `timeout` or max. 10s if timeout is not given. func (m *Mqtt) newReceiver() *msgHandler { - wait := m.timeout - if wait == 0 { - wait = request.Timeout - } - h := &msgHandler{ topic: m.topic, scale: m.scale, - mux: util.NewWaiter(wait, func() { m.log.DEBUG.Printf("%s wait for initial value", m.topic) }), + mux: util.NewWaiter(m.timeout, func() { m.log.DEBUG.Printf("%s wait for initial value", m.topic) }), re: m.re, jq: m.jq, } diff --git a/provider/mqtt_handler.go b/provider/mqtt_handler.go index 70144f5ca..b5f3bb120 100644 --- a/provider/mqtt_handler.go +++ b/provider/mqtt_handler.go @@ -31,11 +31,11 @@ func (h *msgHandler) receive(payload string) { // hasValue returned the received and processed payload as string func (h *msgHandler) hasValue() (string, error) { - elapsed := h.mux.LockWithTimeout() + h.mux.Lock() defer h.mux.Unlock() - if elapsed > 0 { - return "", fmt.Errorf("%s outdated: %v", h.topic, elapsed.Truncate(time.Second)) + if late := h.mux.Overdue(); late > 0 { + return "", fmt.Errorf("%s outdated: %v", h.topic, late.Truncate(time.Second)) } var err error diff --git a/provider/sma/device.go b/provider/sma/device.go index d1a658f90..ae4298809 100644 --- a/provider/sma/device.go +++ b/provider/sma/device.go @@ -50,11 +50,11 @@ func (d *Device) Values() (map[sunny.ValueID]interface{}, error) { // ensure update loop was started d.StartUpdateLoop() - elapsed := d.mux.LockWithTimeout() + d.mux.Lock() defer d.mux.Unlock() - if elapsed > 0 { - return nil, fmt.Errorf("update timeout: %v", elapsed.Truncate(time.Second)) + if late := d.mux.Overdue(); late > 0 { + return nil, fmt.Errorf("update timeout: %v", late.Truncate(time.Second)) } // return a copy of the map to avoid race conditions diff --git a/provider/socket.go b/provider/socket.go index bc434e113..73e776dac 100644 --- a/provider/socket.go +++ b/provider/socket.go @@ -141,11 +141,11 @@ func (p *Socket) listen() { } func (p *Socket) hasValue() (interface{}, error) { - elapsed := p.mux.LockWithTimeout() + p.mux.Lock() defer p.mux.Unlock() - if elapsed > 0 { - return nil, fmt.Errorf("outdated: %v", elapsed.Truncate(time.Second)) + if late := p.mux.Overdue(); late > 0 { + return nil, fmt.Errorf("outdated: %v", late.Truncate(time.Second)) } return p.val, nil diff --git a/util/waiter.go b/util/waiter.go index e8027fa9a..61047dd05 100644 --- a/util/waiter.go +++ b/util/waiter.go @@ -5,59 +5,59 @@ import ( "time" ) -const waitTimeout = 50 * time.Millisecond // polling interval when waiting for initial value +var waitInitialTimeout = 10 * time.Second // Waiter provides monitoring of receive timeouts and reception of initial value type Waiter struct { sync.Mutex log func() - once sync.Once + cond *sync.Cond updated time.Time timeout time.Duration } // NewWaiter creates new waiter func NewWaiter(timeout time.Duration, logInitialWait func()) *Waiter { - return &Waiter{ + p := &Waiter{ log: logInitialWait, timeout: timeout, } + p.cond = sync.NewCond(p) + return p } // Update is called when client has received data. Update resets the timeout counter. // It is client responsibility to ensure that the waiter is not locked when Update is called. func (p *Waiter) Update() { p.updated = time.Now() + p.cond.Broadcast() } -// waitForInitialValue blocks until Update has been called at least once. -// It assumes lock has been obtained before and returns with lock active. -func (p *Waiter) waitForInitialValue() { +// Overdue waits for initial update and returns the duration since the last update +// in excess of timeout. Waiter MUST be locked when calling Overdue. +func (p *Waiter) Overdue() time.Duration { if p.updated.IsZero() { p.log() - // wait for initial update - waitStarted := time.Now() - for p.updated.IsZero() { - p.Unlock() - time.Sleep(waitTimeout) - p.Lock() + c := make(chan struct{}) - // abort initial wait with error - if p.timeout != 0 && time.Since(waitStarted) > p.timeout { - p.updated = waitStarted - return + go func() { + defer close(c) + for p.updated.IsZero() { + p.cond.Wait() } + }() + + select { + case <-c: + // initial value received, lock established + case <-time.After(waitInitialTimeout): + p.Update() // unblock the sync.Cond + <-c // wait for goroutine, re-establish lock + p.updated = time.Time{} // reset updated to initial value missing + return waitInitialTimeout } } -} - -// LockWithTimeout waits for initial value and checks if update timeout has elapsed -func (p *Waiter) LockWithTimeout() time.Duration { - p.Lock() - - // waiting assumes lock acquired and returns with lock - p.once.Do(p.waitForInitialValue) if elapsed := time.Since(p.updated); p.timeout != 0 && elapsed > p.timeout { return elapsed diff --git a/util/waiter_test.go b/util/waiter_test.go new file mode 100644 index 000000000..182b54c59 --- /dev/null +++ b/util/waiter_test.go @@ -0,0 +1,76 @@ +package util + +import ( + "os" + "testing" + "time" +) + +const testTimeout = 100 * time.Millisecond + +func TestMain(t *testing.M) { + waitInitialTimeout = 2 * testTimeout + os.Exit(t.Run()) +} + +func TestWaiterInitialUpdateInTime(t *testing.T) { + for _, timeout := range []time.Duration{0, testTimeout} { + w := NewWaiter(timeout, func() {}) + + go func() { + time.Sleep(testTimeout / 2) + w.Update() + }() + + w.Lock() + defer w.Unlock() + + if elapsed := w.Overdue(); elapsed != 0 { + t.Errorf("expected %v, got %v", 0, elapsed) + } + } +} + +func TestWaiterInitialUpdateNotReceived(t *testing.T) { + for _, timeout := range []time.Duration{0, testTimeout} { + w := NewWaiter(timeout, func() {}) + + w.Lock() + defer w.Unlock() + + if elapsed := w.Overdue(); elapsed != waitInitialTimeout { + t.Errorf("expected %v, got %v", waitInitialTimeout, elapsed) + } + } +} + +func TestWaiterUpdateInTime(t *testing.T) { + w := NewWaiter(testTimeout, func() {}) + w.Update() + + go func() { + time.Sleep(testTimeout / 2) + w.Update() + }() + + w.Lock() + defer w.Unlock() + + if elapsed := w.Overdue(); elapsed != 0 { + t.Errorf("expected %v, got %v", 0, elapsed) + } +} + +func TestWaiterUpdateNotReceived(t *testing.T) { + w := NewWaiter(testTimeout, func() {}) + w.Update() + + 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) + } +}