From ee230ec3836b2b5fafdbb9eceb3a3657c25e8157 Mon Sep 17 00:00:00 2001 From: Markus <974709+Links2004@users.noreply.github.com> Date: Thu, 10 Apr 2025 13:19:53 +0200 Subject: [PATCH] Mqtt: limit the number of max parallel inflight requests (#20494) --- plugin/mqtt.go | 12 ++++--- plugin/mqtt/client.go | 80 +++++++++++++++++++++++-------------------- server/mqtt.go | 11 ++++-- 3 files changed, 60 insertions(+), 43 deletions(-) diff --git a/plugin/mqtt.go b/plugin/mqtt.go index d733417ba..4abb66a2f 100644 --- a/plugin/mqtt.go +++ b/plugin/mqtt.go @@ -130,7 +130,8 @@ func (m *Mqtt) IntSetter(param string) (func(int64) error, error) { return err } - return m.client.Publish(m.topic, m.retained, payload) + m.client.Publish(m.topic, m.retained, payload) + return nil }, nil } @@ -144,7 +145,8 @@ func (m *Mqtt) FloatSetter(param string) (func(float64) error, error) { return err } - return m.client.Publish(m.topic, m.retained, payload) + m.client.Publish(m.topic, m.retained, payload) + return nil }, nil } @@ -158,7 +160,8 @@ func (m *Mqtt) BoolSetter(param string) (func(bool) error, error) { return err } - return m.client.Publish(m.topic, m.retained, payload) + m.client.Publish(m.topic, m.retained, payload) + return nil }, nil } @@ -172,6 +175,7 @@ func (m *Mqtt) StringSetter(param string) (func(string) error, error) { return err } - return m.client.Publish(m.topic, m.retained, payload) + m.client.Publish(m.topic, m.retained, payload) + return nil }, nil } diff --git a/plugin/mqtt/client.go b/plugin/mqtt/client.go index d0b847921..5b43162ef 100644 --- a/plugin/mqtt/client.go +++ b/plugin/mqtt/client.go @@ -1,15 +1,17 @@ package mqtt import ( + "context" "crypto/tls" "crypto/x509" "fmt" "math/rand/v2" "strings" "sync" - "sync/atomic" "time" + "golang.org/x/sync/semaphore" + paho "github.com/eclipse/paho.mqtt.golang" "github.com/evcc-io/evcc/api" "github.com/evcc-io/evcc/util" @@ -19,6 +21,8 @@ import ( // Instance is the paho Mqtt client singleton var Instance *Client +const parallelInflightLimit int64 = 128 + // ClientID created unique mqtt client id func ClientID() string { return fmt.Sprintf("evcc-%d", rand.Int32()) @@ -40,11 +44,11 @@ type Config struct { type Client struct { log *util.Logger mux sync.Mutex - Client paho.Client + client paho.Client broker string Qos byte - inflight uint32 listener map[string][]func(string) + inflight *semaphore.Weighted } type Option func(*paho.ClientOptions) @@ -65,6 +69,7 @@ func NewClient(log *util.Logger, broker, user, password, clientID string, qos by log: log, Qos: qos, listener: make(map[string][]func(string)), + inflight: semaphore.NewWeighted(parallelInflightLimit), } options := paho.NewClientOptions() @@ -116,7 +121,7 @@ func NewClient(log *util.Logger, broker, user, password, clientID string, qos by return nil, fmt.Errorf("error connecting: %w", token.Error()) } - mc.Client = client + mc.client = client return mc, nil } @@ -141,33 +146,55 @@ func (m *Client) ConnectionHandler(client paho.Client) { // Cleanup recursively removes a topic func (m *Client) Cleanup(topic string, retained bool) error { + timer := time.NewTimer(time.Second) + statusTopic := topic + "/status" - if !m.Client.Subscribe(topic+"/#", m.Qos, func(c paho.Client, msg paho.Message) { + if !m.client.Subscribe(topic+"/#", m.Qos, func(c paho.Client, msg paho.Message) { if len(msg.Payload()) == 0 || msg.Topic() == statusTopic { return } m.log.TRACE.Printf("delete: %s", msg.Topic()) - m.Client.Publish(msg.Topic(), m.Qos, true, []byte{}) + m.Publish(msg.Topic(), true, "") + + // reset timeout + timer.Reset(time.Second) }).WaitTimeout(request.Timeout) { return api.ErrTimeout } - time.Sleep(time.Second) + // wait for cleanup to finish + <-timer.C - if !m.Client.Unsubscribe(topic + "/#").WaitTimeout(request.Timeout) { + if !m.client.Unsubscribe(topic + "/#").WaitTimeout(request.Timeout) { return api.ErrTimeout } return nil } -// Publish synchronously publishes payload using client qos -func (m *Client) Publish(topic string, retained bool, payload interface{}) error { - m.log.TRACE.Printf("send %s: '%v'", topic, payload) - token := m.Client.Publish(topic, m.Qos, retained, payload) - go m.WaitForToken("send", topic, token) - return nil +// Publish asynchronously publishes payload using client qos +func (m *Client) Publish(topic string, retained bool, payload interface{}) { + go func() { + ctx, cancel := context.WithTimeout(context.Background(), request.Timeout) + defer cancel() + if err := m.inflight.Acquire(ctx, 1); err != nil { + m.log.ERROR.Printf("send %s: %v", topic, err) + return + } + defer m.inflight.Release(1) + + m.log.TRACE.Printf("send %s: '%v'", topic, payload) + token := m.client.Publish(topic, m.Qos, retained, payload) + + err := api.ErrTimeout + if token.WaitTimeout(request.Timeout) { + err = token.Error() + } + if err != nil { + m.log.ERROR.Printf("send: %s: %v", topic, err) + } + }() } // Listen attaches listener to slice of listeners for given topic @@ -193,16 +220,14 @@ func (m *Client) ListenSetter(topic string, callback func(string) error) error { if err := callback(payload); err != nil { m.log.ERROR.Printf("set %s: %v", topic, err) } - if err := m.Publish(topic, true, ""); err != nil { - m.log.ERROR.Printf("clear: %s: %v", topic, err) - } + m.Publish(topic, true, "") }) return err } // listen attaches listener to topic func (m *Client) listen(topic string) paho.Token { - token := m.Client.Subscribe(topic, m.Qos, func(c paho.Client, msg paho.Message) { + token := m.client.Subscribe(topic, m.Qos, func(c paho.Client, msg paho.Message) { payload := string(msg.Payload()) m.log.TRACE.Printf("recv %s: '%v'", topic, payload) if len(payload) > 0 { @@ -217,22 +242,3 @@ func (m *Client) listen(topic string) paho.Token { }) return token } - -// WaitForToken synchronously waits until token operation completed -func (m *Client) WaitForToken(action, topic string, token paho.Token) { - if inflight := atomic.LoadUint32(&m.inflight); inflight > 64 { - return - } - - // track inflight token waits - atomic.AddUint32(&m.inflight, 1) - defer atomic.AddUint32(&m.inflight, ^uint32(0)) - - err := api.ErrTimeout - if token.WaitTimeout(request.Timeout) { - err = token.Error() - } - if err != nil { - m.log.ERROR.Printf("%s: %s: %v", action, topic, err) - } -} diff --git a/server/mqtt.go b/server/mqtt.go index 9985e5639..231488821 100644 --- a/server/mqtt.go +++ b/server/mqtt.go @@ -9,6 +9,7 @@ import ( "time" "github.com/evcc-io/evcc/api" + "github.com/evcc-io/evcc/cmd/shutdown" "github.com/evcc-io/evcc/core/loadpoint" "github.com/evcc-io/evcc/core/site" "github.com/evcc-io/evcc/core/vehicle" @@ -47,6 +48,13 @@ func NewMQTT(root string, site site.API) (*MQTT, error) { err = fmt.Errorf("mqtt: %w", err) } + shutdown.Register(func() { + m.log.DEBUG.Println("shutdown cleanup started") + if err := m.Handler.Cleanup(m.root, true); err != nil { + m.log.ERROR.Printf("shutdown cleanup failed: %v", err) + } + }) + return m, err } @@ -141,8 +149,7 @@ func (m *MQTT) publishComplex(topic string, retained bool, payload interface{}) } func (m *MQTT) publishString(topic string, retained bool, payload string) { - token := m.Handler.Client.Publish(topic, m.Handler.Qos, retained, m.encode(payload)) - go m.Handler.WaitForToken("send", topic, token) + m.Handler.Publish(topic, retained, m.encode(payload)) } func (m *MQTT) publishSingleValue(topic string, retained bool, payload interface{}) {