From 4c803d950f44f169cbe393d971672c4a27d6d0cc Mon Sep 17 00:00:00 2001 From: Paul Lorenz Date: Thu, 5 Oct 2023 10:03:21 -0400 Subject: [PATCH] stop entity change event loop when controller shuts down. Fixes #1406 --- controller/events/dispatcher.go | 1 + controller/events/dispatcher_entity_change.go | 9 ++++++++- 2 files changed, 9 insertions(+), 1 deletion(-) diff --git a/controller/events/dispatcher.go b/controller/events/dispatcher.go index 748f175ff..c0c0905c2 100644 --- a/controller/events/dispatcher.go +++ b/controller/events/dispatcher.go @@ -47,6 +47,7 @@ func NewDispatcher(closeNotify <-chan struct{}) *Dispatcher { result := &Dispatcher{ closeNotify: closeNotify, entityChangeEventsDispatcher: entityChangeEventDispatcher{ + closeNotify: closeNotify, notifyCh: make(chan struct{}, 1), globalMetadata: map[string]any{}, }, diff --git a/controller/events/dispatcher_entity_change.go b/controller/events/dispatcher_entity_change.go index 5ef20f978..54e35874b 100644 --- a/controller/events/dispatcher_entity_change.go +++ b/controller/events/dispatcher_entity_change.go @@ -163,6 +163,7 @@ func bytesToTxId(b []byte) uint64 { type entityChangeEventDispatcher struct { network *network.Network dispatcher *Dispatcher + closeNotify <-chan struct{} notifyCh chan struct{} globalMetadata map[string]any } @@ -298,12 +299,18 @@ func (self *entityChangeEventDispatcher) notifyFlush() { func (self *entityChangeEventDispatcher) flushLoop() { for { // wait to be notified of an event - <-self.notifyCh + select { + case <-self.closeNotify: + return + case <-self.notifyCh: + } // wait until we've not gotten an event for 5 seconds before cleaning up flushed := false for !flushed { select { + case <-self.closeNotify: + return case <-self.notifyCh: case <-time.After(5 * time.Second): pfxlog.Logger().Debug("cleaning up entity change events")