-
Notifications
You must be signed in to change notification settings - Fork 835
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add event bus for scheduler-internal events (#99)
* Add bus-based event hub Use mustafaturan/bus as a seemingly well-featured bus implementation. Use an atomic counter as its sequence generator, to avoid adding further dependencies for a simple construct. The sequence generator is per hub to minimise the risk of shared state during testing. * Embed event bus in hub for method use * Register topics for hub on creation * Add source field to event hub logger * Add docstring for NewEventHub * Add sync -> async adapter func for model event message handling This function attempts to solve two issues: * Limit leaking implementation details about the use of mustafaturan/bus to the wider application * Provide a simple interface for users with an asynchronous approach * Refactor model event message handler creation This should be slightly more legible, but more importantly gives a name to the actual handle function, to improve logging. * Use bus-based event in scheduler's main package * Use bus-based event hub in in-memory store * Use bus-based event hub in XDS incremental processor * Make XDS processor's sync-handling internal * Use bus-based event hub in agent server + make handling of syncs internal * Use bus-based event hub in version cleaner + make handling of syncs internal * Use bus-based event hub in scheduler server + make handling of events/listeners internal * Add comment re use of Goroutines in agent server * Add comment re (graceful) shutdown of scheduler * Fix typos in event hub implementation * Format of long lines * Add model event publish method for event hub Also add comment on use of goroutines. The only place writing into the event hub previously was memory_status.go. That chose to publish events from goroutines, but that may not be the best policy - we should consider this. * Replace event hub publish calls when updating model state These used to be 'trigger' calls to the old event hub, but are now 'publish' calls to the bus-based event hub. * Remove unused methods for ModelEventHub * Fix typos & build issues * Add comment re closing channels for event hub * Fix store package tests re use of event hub * Fix scheduler server package tests re use of event hub * Fix agent server package tests re use of event hub * Remove ModelEventHub as fully replaced * Refactor event hub types to separate file within package * Use consistent handler keys for event hub handlers Rather than using free-form text as keys, this change uses dot-separated fields. The fields identify the handling type's purpose first, followed by the event's semantics. For example, the scheduler server listens to model events but treats them in two different ways: as model events, but also as events that can affect servers. * Refactor event hub handler names to package-level constants * Add logging for event publishing failures * Refactor event source names to package-level constants * Add mechanism for closing event hub & its handler channels * Stop processing messages of invalid type * Remove exposure of 'bus' library from event hub interface * Fix typos in channel consumption * Refactor to abstract creation of event hub handlers from 'bus' implementation * Make event publishing synchronous This avoids potential reordering of events caused by the use of goroutines. Instead, callers have explicit control over sync/async operation. * Make event hub topic name private * Add test for event hub creation * Use 'require' for assertion on error in test * Add test cases for registering handlers * Simplify event handler test to use single counter with atomic increments * Add test cases for publishing events * Add test cases for closing event hub
- Loading branch information
Showing
17 changed files
with
713 additions
and
174 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,47 +1,143 @@ | ||
| package coordinator | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
| "reflect" | ||
| "sync" | ||
| "sync/atomic" | ||
|
|
||
| busV3 "github.com/mustafaturan/bus/v3" | ||
| log "github.com/sirupsen/logrus" | ||
| ) | ||
|
|
||
| const ( | ||
| topicModelEvents = "model.event" | ||
| ) | ||
|
|
||
| type ModelEventMsg struct { | ||
| ModelName string | ||
| ModelVersion uint32 | ||
| type SequenceGenerator struct { | ||
| counter int64 | ||
| } | ||
|
|
||
| func (m ModelEventMsg) String() string { | ||
| return fmt.Sprintf("%s:%d", m.ModelName, m.ModelVersion) | ||
| func (g *SequenceGenerator) Generate() string { | ||
| next := atomic.AddInt64(&g.counter, 1) | ||
| return fmt.Sprintf("%d", next) | ||
| } | ||
|
|
||
| var _ busV3.IDGenerator = (*SequenceGenerator)(nil) | ||
|
|
||
| type EventHub struct { | ||
| bus *busV3.Bus | ||
| logger log.FieldLogger | ||
| modelEventHandlerChannels []chan ModelEventMsg | ||
| lock sync.RWMutex | ||
| closed bool | ||
| } | ||
|
|
||
| // NewEventHub creates a new EventHub with topics pre-registered. | ||
| // The logger l does not need fields preset. | ||
| func NewEventHub(l log.FieldLogger) (*EventHub, error) { | ||
| generator := &SequenceGenerator{} | ||
| bus, err := busV3.NewBus(generator) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| hub := EventHub{ | ||
| logger: l.WithField("source", "EventHub"), | ||
| bus: bus, | ||
| } | ||
|
|
||
| hub.bus.RegisterTopics(topicModelEvents) | ||
|
|
||
| return &hub, nil | ||
| } | ||
|
|
||
| type ModelEventHub struct { | ||
| mu sync.RWMutex | ||
| closed bool | ||
| listeners []chan<- ModelEventMsg | ||
| func (h *EventHub) Close() { | ||
| h.lock.Lock() | ||
| defer h.lock.Unlock() | ||
|
|
||
| h.closed = true | ||
|
|
||
| for _, c := range h.modelEventHandlerChannels { | ||
| close(c) | ||
| } | ||
| } | ||
|
|
||
| func (h *ModelEventHub) AddListener(c chan<- ModelEventMsg) { | ||
| h.mu.Lock() | ||
| defer h.mu.Unlock() | ||
| h.listeners = append(h.listeners, c) | ||
| func (h *EventHub) RegisterHandler( | ||
| name string, | ||
| queueSize int, | ||
| logger log.FieldLogger, | ||
| handle func(event ModelEventMsg), | ||
| ) { | ||
| events := make(chan ModelEventMsg, queueSize) | ||
| h.addModelEventHandlerChannel(events) | ||
|
|
||
| go func() { | ||
| for e := range events { | ||
| handle(e) | ||
| } | ||
| }() | ||
|
|
||
| handler := h.newModelEventHandler(logger, events, handle) | ||
| h.bus.RegisterHandler(name, handler) | ||
| } | ||
|
|
||
| func (h *ModelEventHub) TriggerModelEvent(event ModelEventMsg) { | ||
| h.mu.RLock() | ||
| defer h.mu.RUnlock() | ||
| if h.closed { | ||
| return | ||
| func (h *EventHub) newModelEventHandler( | ||
| logger log.FieldLogger, | ||
| events chan ModelEventMsg, | ||
| handle func(event ModelEventMsg), | ||
| ) busV3.Handler { | ||
| handleModelEventMessage := func(_ context.Context, e busV3.Event) { | ||
| l := logger.WithField("func", "handleModelEventMessage") | ||
| l.Debugf("Received event on %s from %s (ID: %s, TxID: %s)", e.Topic, e.Source, e.ID, e.TxID) | ||
|
|
||
| me, ok := e.Data.(ModelEventMsg) | ||
| if !ok { | ||
| l.Warnf( | ||
| "Event (ID %s, TxID %s) on topic %s from %s is not a ModelEventMsg: %s", | ||
| e.ID, | ||
| e.TxID, | ||
| e.Topic, | ||
| e.Source, | ||
| reflect.TypeOf(e.Data).String(), | ||
| ) | ||
| return | ||
| } | ||
|
|
||
| h.lock.RLock() | ||
| if h.closed { | ||
| return | ||
| } | ||
| events <- me | ||
| h.lock.RUnlock() | ||
| } | ||
| for _, listener := range h.listeners { | ||
| listener <- event | ||
|
|
||
| return busV3.Handler{ | ||
| Matcher: topicModelEvents, | ||
| Handle: handleModelEventMessage, | ||
| } | ||
| } | ||
|
|
||
| func (h *ModelEventHub) Close() { | ||
| h.mu.Lock() | ||
| defer h.mu.Unlock() | ||
| for _, listener := range h.listeners { | ||
| close(listener) | ||
| func (h *EventHub) addModelEventHandlerChannel(c chan ModelEventMsg) { | ||
| h.lock.Lock() | ||
| defer h.lock.Unlock() | ||
|
|
||
| h.modelEventHandlerChannels = append(h.modelEventHandlerChannels, c) | ||
| } | ||
|
|
||
| func (h *EventHub) PublishModelEvent(source string, event ModelEventMsg) { | ||
| err := h.bus.EmitWithOpts( | ||
| context.Background(), | ||
| topicModelEvents, | ||
| event, | ||
| busV3.WithSource(source), | ||
| ) | ||
| if err != nil { | ||
| h.logger.WithError(err).Errorf( | ||
| "unable to publish model event message from %s to %s", | ||
| source, | ||
| topicModelEvents, | ||
| ) | ||
| } | ||
| h.closed = true | ||
| } |
Oops, something went wrong.