Move notifyQueueProcessor to acquireShard
@@ -1615,14 +1615,6 @@ func (s *ContextImpl) transition(request contextRequest) error {
return errInvalidTransition
}
- engine, err := s.engineFuture.Get(s.lifecycleCtx)
- if err != nil {
- // this should not happen, already checked engineFuture is ready above
- s.contextTaggedLogger.Warn("transition to acquired but unable to get engine", tag.Error(err))
- return errInvalidTransition
- }
- s.notifyQueueProcessor(engine)
-
return nil
case contextRequestLost:
return nil // nothing to do, already acquiring
@@ -1664,7 +1656,20 @@ func (s *ContextImpl) transition(request contextRequest) error {
return errInvalidTransition
}
-func (s *ContextImpl) notifyQueueProcessor(engine Engine) {
+// notifyQueueProcessor sends notification to all queue processors for triggering a load
+// NOTE: this method assumes engineFuture is already in a ready state.
+func (s *ContextImpl) notifyQueueProcessor() {
+ // use a cancelled ctx so the method won't be blocked if engineFuture is not ready
+ cancelledCtx, cancel := context.WithCancel(context.Background())
+ cancel()
+
+ // we will get the engine when the Future is ready
+ engine, err := s.engineFuture.Get(cancelledCtx)
+ if err != nil {
+ s.contextTaggedLogger.Warn("tried to notify queue processor when engine is not ready")
+ return
+ }
+
now := s.timeSource.Now()
fakeTasks := make(map[tasks.Category][]tasks.Task)
for _, category := range tasks.GetCategories() {
@@ -1868,12 +1873,18 @@ func (s *ContextImpl) acquireShard() {
err = s.transition(contextRequestAcquired{engine: engine})
- if err != nil && engine != nil {
+ if err != nil {
+ if engine != nil {
// We tried to set the engine but the context was already stopped
engine.Stop()
+ }
return err
}
+ // we know engineFuture must be ready here, and we can notify queue processor
+ // to trigger a load as queue max level can be updated to a newer value
+ s.notifyQueueProcessor()
+
return nil
}
