diff --git a/provider/mqtt.go b/provider/mqtt.go index 1fdfe801a..b5786c71a 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 4da973893..af44237c2 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 ae4298809..d6644bbcc 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 fa55c35dc..f891bbad8 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 73e776dac..4669d4bd8 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 d842716ab..a6c0efce3 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 } diff --git a/util/waiter_test.go b/util/waiter_test.go index 7ff903b9a..2c6cdf4bc 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) }