Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

OCTRL-920 Fixes for stuck calibration runs #609

Merged
merged 3 commits into from
Sep 18, 2024
Merged

OCTRL-920 Fixes for stuck calibration runs #609

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
4fdeae0
[core] improve a few logs
knopers8 Sep 12, 2024
bda8717
OCTRL-920 [core] invalidating auto-stop lets the goroutine exit
knopers8 Sep 12, 2024
2c0940b
[core] OCTRL-920 safer concurrency in KillTasks
knopers8 Sep 13, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 18 additions & 4 deletions core/environment/environment.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -88,7 +88,8 @@ type Environment struct {
callsPendingAwait map[string] /*await expression, trigger only*/ callable.CallsMap
currentTransition string

autoStopTimer *time.Timer
autoStopTimer *time.Timer
autoStopCancelFcn context.CancelFunc
}

func (env *Environment) NotifyEvent(e event.DeviceEvent) {
Expand Down Expand Up @@ -593,6 +594,7 @@ func newEnvironment(userVars map[string]string, newId uid.ID) (env *Environment,
log.WithField("partition", envId.String()).
Debug("O2 End Completion time already set before after_GO_ERROR")
}
env.invalidateAutoStopTransition()
}

errHooks = errors.Join(errHooks, env.handleHooksWithPositiveWeights(env.Workflow(), trigger))
Expand Down Expand Up @@ -959,8 +961,11 @@ func (env *Environment) runTasksAsHooks(hooksToTrigger task.Tasks) (errorMap map
func (env *Environment) TryTransition(t Transition) (err error) {
if !env.transitionMutex.TryLock() {
log.WithField("partition", env.id.String()).
Warnf("environment transition attempt delayed: transition '%s' in progress. waiting for completion or failure", env.currentTransition)
Warnf("environment transition '%s' attempt delayed: transition '%s' in progress. waiting for completion or failure", t.eventName(), env.currentTransition)
env.transitionMutex.Lock()
log.WithField("level", infologger.IL_Support).
WithField("partition", env.id.String()).
Infof("environment transition '%s' attempt resumed", t.eventName())
}
defer env.transitionMutex.Unlock()

Expand Down Expand Up @@ -1400,6 +1405,8 @@ func (env *Environment) scheduleAutoStopTransition() (scheduled bool, expected t
}

env.autoStopTimer = time.NewTimer(autoStopDuration)
ctx, cancel := context.WithCancel(context.Background())
env.autoStopCancelFcn = cancel
go func() {
select {
case <-env.autoStopTimer.C:
Expand All @@ -1420,6 +1427,10 @@ func (env *Environment) scheduleAutoStopTransition() (scheduled bool, expected t
}
return
}
case <-ctx.Done():
log.WithField("partition", env.id).
WithField("run", env.currentRunNumber).
Debugf("Scheduled auto stop transition was cancelled")
}
}()

Expand All @@ -1434,7 +1445,10 @@ func (env *Environment) scheduleAutoStopTransition() (scheduled bool, expected t

func (env *Environment) invalidateAutoStopTransition() {
// Only try to stop an initialized timer
if env.autoStopTimer != nil {
env.autoStopTimer.Stop()
if env.autoStopTimer == nil {
return
}
if env.autoStopTimer.Stop() && env.autoStopCancelFcn != nil {
env.autoStopCancelFcn()
}
}
9 changes: 6 additions & 3 deletions core/environment/manager.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -431,7 +431,7 @@ func (envs *Manager) CreateEnvironment(workflowPath string, userVars map[string]
goErrorKillDestroy := func(op string) {
envState := env.CurrentState()
log.WithField("state", envState).
WithField("environment", env.Id().String()).
WithField("partition", env.Id().String()).
WithError(err).
Warnf("auto-transitioning environment failed %s, cleanup in progress", op)

Expand Down Expand Up @@ -586,6 +586,9 @@ func (envs *Manager) TeardownEnvironment(environmentId uid.ID, force bool) error
log.WithField("partition", environmentId.String()).
Warnf("environment teardown attempt delayed: transition '%s' in progress. waiting for completion or failure", env.currentTransition)
env.transitionMutex.Lock()
log.WithField("level", infologger.IL_Support).
WithField("partition", environmentId.String()).
Infof("environment teardown attempt resumed")
}
defer env.transitionMutex.Unlock()

Expand Down Expand Up @@ -1212,7 +1215,7 @@ func (envs *Manager) CreateAutoEnvironment(workflowPath string, userVars map[str
env.Public, env.Description, err = parseWorkflowPublicInfo(workflowPath)
if err != nil {
log.WithField("public info", env.Public).
WithField("environment", env.Id().String()).
WithField("partition", env.Id().String()).
WithError(err).
Warn("parse workflow public info failed.")
}
Expand Down Expand Up @@ -1272,7 +1275,7 @@ func (envs *Manager) CreateAutoEnvironment(workflowPath string, userVars map[str
envState := env.CurrentState()
env.sendEnvironmentEvent(&event.EnvironmentEvent{Message: fmt.Sprintf("environment is in %s, handling ERROR", envState), EnvironmentID: env.Id().String(), Error: err})
log.WithField("state", envState).
WithField("environment", env.Id().String()).
WithField("partition", env.Id().String()).
WithError(err).
Warnf("auto-transitioning environment failed %s, cleanup in progress", op)

Expand Down
16 changes: 13 additions & 3 deletions core/task/manager.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -100,6 +100,7 @@ type Manager struct {
schedulerState *schedulerState
internalEventCh chan<- event.Event
ackKilledTasks *safeAcks
killTasksMu sync.Mutex // to avoid races when attempting to kill the same tasks in different goroutines
}

func NewManager(shutdown func(), internalEventCh chan<- event.Event) (taskman *Manager, err error) {
Expand Down Expand Up @@ -938,6 +939,7 @@ func (m *Manager) updateTaskState(taskId string, state string) {
taskPtr := m.roster.getByTaskId(taskId)
if taskPtr == nil {
log.WithField("taskId", taskId).
WithField("state", state).
Warn("attempted state update of task not in roster")
return
}
Expand Down Expand Up @@ -1041,7 +1043,7 @@ func (m *Manager) Cleanup() (killed Tasks, running Tasks, err error) {
// If the task list includes locked tasks, TaskNotFoundError is returned.
func (m *Manager) KillTasks(taskIds []string) (killed Tasks, running Tasks, err error) {
taskCanBeKilledFilter := func(t *Task) bool {
if t.IsLocked() {
if t.IsLocked() || m.ackKilledTasks.contains(t.taskId) {
return false
}
for _, id := range taskIds {
Expand All @@ -1052,20 +1054,27 @@ func (m *Manager) KillTasks(taskIds []string) (killed Tasks, running Tasks, err
return false
}

if !m.killTasksMu.TryLock() {
log.WithField("level", infologger.IL_Support).Warnf("Scheduling killing tasks was delayed until another goroutine finishes doing so")
m.killTasksMu.Lock()
log.WithField("level", infologger.IL_Support).Infof("Scheduling killing tasks is resumed")
}
// TODO: use grouping instead of 2 passes of filtering for performance
toKill := m.roster.filtered(taskCanBeKilledFilter)
unkillable := m.roster.filtered(func(t *Task) bool { return !taskCanBeKilledFilter(t) })

if len(toKill) < len(taskIds) {
unkillable := m.roster.filtered(func(t *Task) bool { return !taskCanBeKilledFilter(t) })
log.WithField("taskIds", strings.Join(unkillable.GetTaskIds(), ", ")).
Debugf("some tasks cannot be physically killed (already dead?), will instead only be removed from roster")
Debugf("some tasks cannot be physically killed (already dead or being killed in another goroutine?), will instead only be removed from roster")
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

does this happen a lot? Shouldn't this be a Warning?

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

i think this can be a somewhat valid situation if two kill requests happen subsequently, so I'm not sure if a warning is needed. But I don't know how often this may happen.

}

for _, id := range toKill.GetTaskIds() {
m.ackKilledTasks.addAckChannel(id)
}

killed, running, err = m.doKillTasks(toKill)
m.killTasksMu.Unlock()

for _, id := range killed.GetTaskIds() {
ack, ok := m.ackKilledTasks.getValue(id)
if ok {
Expand All @@ -1087,6 +1096,7 @@ func (m *Manager) doKillTasks(tasks Tasks) (killed Tasks, running Tasks, err err
inactiveTasks := tasks.Filtered(func(task *Task) bool {
return task.status != ACTIVE
})

// Remove from the roster the tasks which are also in the inactiveTasks list to delete
m.roster.updateTasks(m.roster.filtered(func(task *Task) bool {
return !inactiveTasks.Contains(func(t *Task) bool {
Expand Down
Loading