From 39e7d6d264cc71fd2cca8798efed67c7f059fadb Mon Sep 17 00:00:00 2001 From: mfuchs1984 <57141790+mfuchs1984@users.noreply.github.com> Date: Fri, 7 Feb 2025 15:23:12 +0100 Subject: [PATCH] Ocpp: cache and re-use initial status (4th attempt) (#18597) --- charger/ocpp.go | 4 ++ charger/ocpp/connector.go | 27 +++++-- charger/ocpp/connector_test.go | 2 + charger/ocpp/cp_setup.go | 7 -- charger/ocpp/cs.go | 124 ++++++++++++++++++--------------- charger/ocpp/cs_core.go | 9 +++ charger/ocpp/instance.go | 4 +- 7 files changed, 108 insertions(+), 69 deletions(-) diff --git a/charger/ocpp.go b/charger/ocpp.go index fef6df3a8..d37d194fc 100644 --- a/charger/ocpp.go +++ b/charger/ocpp.go @@ -144,6 +144,8 @@ func NewOCPP(id string, connector int, idTag string, ) (*OCPP, error) { log := util.NewLogger(fmt.Sprintf("%s-%d", lo.CoalesceOrEmpty(id, "ocpp"), connector)) + log.DEBUG.Printf("!! registering %s:%d", id, connector) + cp, err := ocpp.Instance().RegisterChargepoint(id, func() *ocpp.CP { return ocpp.NewChargePoint(log, id) @@ -164,6 +166,8 @@ func NewOCPP(id string, connector int, idTag string, return nil, err } + log.DEBUG.Printf("!! connected %s:%d", id, connector) + if cp.NumberOfConnectors > 0 && connector > cp.NumberOfConnectors { return nil, fmt.Errorf("invalid connector: %d", connector) } diff --git a/charger/ocpp/connector.go b/charger/ocpp/connector.go index 6dea4275c..43cbf7a8b 100644 --- a/charger/ocpp/connector.go +++ b/charger/ocpp/connector.go @@ -45,9 +45,28 @@ func NewConnector(log *util.Logger, id int, cp *CP, idTag string) (*Connector, e remoteIdTag: idTag, } - err := cp.registerConnector(id, conn) + if err := cp.registerConnector(id, conn); err != nil { + return nil, err + } - return conn, err + // trigger status for all connectors + + var ok bool + // apply cached status if available + instance.WithConnectorStatus(cp.ID(), id, func(status *core.StatusNotificationRequest) { + if _, err := cp.OnStatusNotification(status); err == nil { + ok = true + } + }) + + // only trigger if we don't already have a status + if !ok && cp.HasRemoteTriggerFeature { + if err := cp.TriggerMessageRequest(0, core.StatusNotificationFeatureName); err != nil { + cp.log.WARN.Printf("failed triggering StatusNotification: %v", err) + } + } + + return conn, nil } func (conn *Connector) TestClock(clock clock.Clock) { @@ -88,7 +107,7 @@ func (conn *Connector) WatchDog(timeout time.Duration) { update := conn.clock.Since(conn.meterUpdated) > timeout conn.mu.Unlock() - if update { + if update && conn.cp.HasRemoteTriggerFeature { conn.TriggerMessageRequest(core.MeterValuesFeatureName) } } @@ -103,7 +122,7 @@ func (conn *Connector) Initialized() error { case <-conn.statusC: return nil - case <-trigger: // try to trigger StatusNotification again as last resort + case <-trigger: // try to trigger StatusNotification again as last resort even when the charger does not report RemoteTrigger support conn.TriggerMessageRequest(core.StatusNotificationFeatureName) case <-timeout: diff --git a/charger/ocpp/connector_test.go b/charger/ocpp/connector_test.go index ba0d80918..95e7b9f27 100644 --- a/charger/ocpp/connector_test.go +++ b/charger/ocpp/connector_test.go @@ -23,6 +23,8 @@ type connTestSuite struct { } func (suite *connTestSuite) SetupTest() { + // setup instance + Instance() suite.cp = NewChargePoint(util.NewLogger("foo"), "abc") suite.conn, _ = NewConnector(util.NewLogger("foo"), 1, suite.cp, "") diff --git a/charger/ocpp/cp_setup.go b/charger/ocpp/cp_setup.go index 44488d4aa..1b9999947 100644 --- a/charger/ocpp/cp_setup.go +++ b/charger/ocpp/cp_setup.go @@ -155,13 +155,6 @@ func (cp *CP) Setup(meterValues string, meterInterval time.Duration) error { cp.log.DEBUG.Printf("failed configuring %s: %v", KeyWebSocketPingInterval, err) } - // trigger status for all connectors - if cp.HasRemoteTriggerFeature { - if err := cp.TriggerMessageRequest(0, core.StatusNotificationFeatureName); err != nil { - cp.log.WARN.Printf("failed triggering StatusNotification: %v", err) - } - } - return nil } diff --git a/charger/ocpp/cs.go b/charger/ocpp/cs.go index 376fc22e2..677e14ddf 100644 --- a/charger/ocpp/cs.go +++ b/charger/ocpp/cs.go @@ -8,43 +8,26 @@ import ( "github.com/evcc-io/evcc/util" ocpp16 "github.com/lorenzodonini/ocpp-go/ocpp1.6" + "github.com/lorenzodonini/ocpp-go/ocpp1.6/core" ) -type CS struct { - mu sync.Mutex - log *util.Logger - ocpp16.CentralSystem - cps map[string]*CP - init map[string]*sync.Mutex - txnId atomic.Int64 +type registration struct { + mu sync.RWMutex + setup sync.RWMutex // serialises chargepoint setup + cp *CP // guarded by setup and CS mutexes + status map[int]*core.StatusNotificationRequest // guarded by mu mutex } -// Register registers a charge point with the central system. -// The charge point identified by id may already be connected in which case initial connection is triggered. -func (cs *CS) register(id string, new *CP) error { - cs.mu.Lock() - defer cs.mu.Unlock() +func newRegistration() *registration { + return ®istration{status: make(map[int]*core.StatusNotificationRequest)} +} - cp, ok := cs.cps[id] - - // case 1: charge point neither registered nor physically connected - if !ok { - cs.cps[id] = new - return nil - } - - // case 2: duplicate registration of id empty - if id == "" { - return errors.New("cannot have >1 charge point with empty station id") - } - - // case 3: charge point not registered but physically already connected - if cp == nil { - cs.cps[id] = new - new.connect(true) - } - - return nil +type CS struct { + ocpp16.CentralSystem + mu sync.Mutex + log *util.Logger + regs map[string]*registration // guarded by mu mutex + txnId atomic.Int64 } // errorHandler logs error channel @@ -58,38 +41,67 @@ func (cs *CS) ChargepointByID(id string) (*CP, error) { cs.mu.Lock() defer cs.mu.Unlock() - cp, ok := cs.cps[id] + reg, ok := cs.regs[id] if !ok { return nil, fmt.Errorf("unknown charge point: %s", id) } - if cp == nil { + if reg.cp == nil { return nil, fmt.Errorf("charge point not configured: %s", id) } - return cp, nil + return reg.cp, nil } +func (cs *CS) WithConnectorStatus(id string, connector int, fun func(status *core.StatusNotificationRequest)) { + cs.mu.Lock() + defer cs.mu.Unlock() + + if reg, ok := cs.regs[id]; ok { + reg.mu.RLock() + if status, ok := reg.status[connector]; ok { + fun(status) + } + reg.mu.RUnlock() + } +} + +// RegisterChargepoint registers a charge point with the central system of returns an already registered charge point func (cs *CS) RegisterChargepoint(id string, newfun func() *CP, init func(*CP) error) (*CP, error) { cs.mu.Lock() - cpmu, ok := cs.init[id] - if !ok { - cpmu = new(sync.Mutex) - cs.init[id] = cpmu + + // prepare shadow state + reg, registered := cs.regs[id] + if !registered { + reg = newRegistration() + cs.regs[id] = reg } - cs.mu.Unlock() // serialise on chargepoint id - cpmu.Lock() - defer cpmu.Unlock() + reg.setup.Lock() + defer reg.setup.Unlock() + + cp := reg.cp + + cs.mu.Unlock() + + // setup already completed? + if cp != nil { + // duplicate registration of id empty + if id == "" { + return nil, errors.New("cannot have >1 charge point with empty station id") + } - // already registered? - if cp, err := cs.ChargepointByID(id); err == nil { return cp, nil } - // first time- registration should not error - cp := newfun() - if err := cs.register(id, cp); err != nil { - return nil, err + // first time- create the charge point + cp = newfun() + + cs.mu.Lock() + reg.cp = cp + cs.mu.Unlock() + + if registered { + cp.connect(true) } return cp, init(cp) @@ -101,12 +113,12 @@ func (cs *CS) NewChargePoint(chargePoint ocpp16.ChargePointConnection) { defer cs.mu.Unlock() // check for configured charge point - cp, ok := cs.cps[chargePoint.ID()] + reg, ok := cs.regs[chargePoint.ID()] if ok { cs.log.DEBUG.Printf("charge point connected: %s", chargePoint.ID()) // trigger initial connection if charge point is already setup - if cp != nil { + if cp := reg.cp; cp != nil { cp.connect(true) } @@ -114,15 +126,15 @@ func (cs *CS) NewChargePoint(chargePoint ocpp16.ChargePointConnection) { } // check for configured anonymous charge point - cp, ok = cs.cps[""] - if ok && cp != nil { + reg, ok = cs.regs[""] + if ok && reg.cp != nil { + cp := reg.cp cs.log.INFO.Printf("charge point connected, registering: %s", chargePoint.ID()) // update id cp.RegisterID(chargePoint.ID()) - - cs.cps[chargePoint.ID()] = cp - delete(cs.cps, "") + cs.regs[chargePoint.ID()] = reg + delete(cs.regs, "") cp.connect(true) @@ -133,7 +145,7 @@ func (cs *CS) NewChargePoint(chargePoint ocpp16.ChargePointConnection) { // register unknown charge point // when charge point setup is complete, it will eventually be associated with the connected id - cs.cps[chargePoint.ID()] = nil + cs.regs[chargePoint.ID()] = newRegistration() } // ChargePointDisconnected implements ocpp16.ChargePointConnectionHandler diff --git a/charger/ocpp/cs_core.go b/charger/ocpp/cs_core.go index f6b1366a5..d56e7e325 100644 --- a/charger/ocpp/cs_core.go +++ b/charger/ocpp/cs_core.go @@ -62,6 +62,15 @@ func (cs *CS) OnMeterValues(id string, request *core.MeterValuesRequest) (*core. } func (cs *CS) OnStatusNotification(id string, request *core.StatusNotificationRequest) (*core.StatusNotificationConfirmation, error) { + cs.mu.Lock() + // cache status for future cp connection + if reg, ok := cs.regs[id]; ok && request != nil { + reg.mu.Lock() + reg.status[request.ConnectorId] = request + reg.mu.Unlock() + } + cs.mu.Unlock() + if cp, err := cs.ChargepointByID(id); err == nil { return cp.OnStatusNotification(request) } diff --git a/charger/ocpp/instance.go b/charger/ocpp/instance.go index 33ba56f26..9fec88103 100644 --- a/charger/ocpp/instance.go +++ b/charger/ocpp/instance.go @@ -40,10 +40,10 @@ func Instance() *CS { instance = &CS{ log: log, - cps: make(map[string]*CP), - init: make(map[string]*sync.Mutex), + regs: make(map[string]*registration), CentralSystem: cs, } + instance.txnId.Store(time.Now().UTC().Unix()) ocppj.SetLogger(instance)