vikarti.anatra/safing-portbase
1
0
Fork 0
mirror of https://github.com/safing/portbase synced 2025-04-12 13:39:09 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Improve stopping of modules

Browse source
This commit is contained in:
Daniel 2023-10-13 11:46:17 +02:00
parent 3dbffd9c1a
commit 3b22f8497d
2 changed files with 78 additions and 67 deletions

90
modules/modules.go
View file

@ -57,10 +57,11 @@ type Module struct { //nolint:maligned
// start
startComplete chan struct{}
// stop
Ctx context.Context
cancelCtx func()
stopFlag *abool.AtomicBool
stopComplete chan struct{}
Ctx context.Context
cancelCtx func()
stopFlag *abool.AtomicBool
stopCompleted *abool.AtomicBool
stopComplete chan struct{}
// workers/tasks
ctrlFuncRunning *abool.AtomicBool
@ -255,12 +256,10 @@ func (m *Module) checkIfStopComplete() {
atomic.LoadInt32(m.taskCnt) == 0 &&
atomic.LoadInt32(m.microTaskCnt) == 0 {
m.Lock()
defer m.Unlock()
if m.stopComplete != nil {
if m.stopCompleted.SetToIf(false, true) {
m.Lock()
defer m.Unlock()
close(m.stopComplete)
m.stopComplete = nil
}
}
}
@ -283,60 +282,56 @@ func (m *Module) stop(reports chan *report) {
// Reset start/stop signal channels.
m.startComplete = make(chan struct{})
m.stopComplete = make(chan struct{})
m.stopCompleted.SetTo(false)
// Make a copy of the stop channel.
stopComplete := m.stopComplete
// Set status and cancel context.
// Set status.
m.status = StatusStopping
m.stopFlag.Set()
m.cancelCtx()
go m.stopAllTasks(reports, stopComplete)
go m.stopAllTasks(reports)
}
func (m *Module) stopAllTasks(reports chan *report, stopComplete chan struct{}) {
// start shutdown function
var stopFnError error
stopFuncRunning := abool.New()
if m.stopFn != nil {
stopFuncRunning.Set()
go func() {
stopFnError = m.runCtrlFn("stop module", m.stopFn)
stopFuncRunning.UnSet()
m.checkIfStopComplete()
}()
} else {
m.checkIfStopComplete()
}
func (m *Module) stopAllTasks(reports chan *report) {
// Manually set the control function flag in order to stop completion by race
// condition before stop function has even started.
m.ctrlFuncRunning.Set()
// Set stop flag for everyone checking this flag before we activate any stop trigger.
m.stopFlag.Set()
// Cancel the context to notify all workers and tasks.
m.cancelCtx()
// Start stop function.
stopFnError := m.startCtrlFn("stop module", m.stopFn)
// wait for results
select {
case <-stopComplete:
// case <-time.After(moduleStopTimeout):
case <-m.stopComplete:
// Complete!
case <-time.After(moduleStopTimeout):
log.Warningf(
"%s: timed out while waiting for stopfn/workers/tasks to finish: stopFn=%v/%v workers=%d tasks=%d microtasks=%d, continuing shutdown...",
"%s: timed out while waiting for stopfn/workers/tasks to finish: stopFn=%v workers=%d tasks=%d microtasks=%d, continuing shutdown...",
m.Name,
stopFuncRunning.IsSet(), m.ctrlFuncRunning.IsSet(),
m.ctrlFuncRunning.IsSet(),
atomic.LoadInt32(m.workerCnt),
atomic.LoadInt32(m.taskCnt),
atomic.LoadInt32(m.microTaskCnt),
)
}
// collect error
// Check for stop fn status.
var err error
if stopFuncRunning.IsNotSet() && stopFnError != nil {
err = stopFnError
}
// set status
if err != nil {
m.Error(
fmt.Sprintf("%s:stop-failed", m.Name),
fmt.Sprintf("Stopping module %s failed", m.Name),
fmt.Sprintf("Failed to stop module: %s", err.Error()),
)
select {
case err = <-stopFnError:
if err != nil {
// Set error as module error.
m.Error(
fmt.Sprintf("%s:stop-failed", m.Name),
fmt.Sprintf("Stopping module %s failed", m.Name),
fmt.Sprintf("Failed to stop module: %s", err.Error()),
)
}
default:
}
// Always set to offline in order to let other modules shutdown in order.
@ -384,7 +379,7 @@ func initNewModule(name string, prep, start, stop func() error, dependencies ...
Name: name,
enabled: abool.NewBool(false),
enabledAsDependency: abool.NewBool(false),
sleepMode: abool.NewBool(true),
sleepMode: abool.NewBool(true), // Change (for init) is triggered below.
sleepWaitingChannel: make(chan time.Time),
prepFn: prep,
startFn: start,
@ -393,6 +388,7 @@ func initNewModule(name string, prep, start, stop func() error, dependencies ...
Ctx: ctx,
cancelCtx: cancelCtx,
stopFlag: abool.NewBool(false),
stopCompleted: abool.NewBool(true),
ctrlFuncRunning: abool.NewBool(false),
workerCnt: &workerCnt,
taskCnt: &taskCnt,
@ -401,7 +397,7 @@ func initNewModule(name string, prep, start, stop func() error, dependencies ...
depNames: dependencies,
}
// Sleep mode is disabled by default
// Sleep mode is disabled by default.
newModule.Sleep(false)
return newModule

55
modules/worker.go
View file

@ -135,10 +135,7 @@ func (m *Module) runWorker(name string, fn func(context.Context) error) (err err
}
func (m *Module) runCtrlFnWithTimeout(name string, timeout time.Duration, fn func() error) error {
stopFnError := make(chan error)
go func() {
stopFnError <- m.runCtrlFn(name, fn)
}()
stopFnError := m.startCtrlFn(name, fn)
// wait for results
select {
@ -149,26 +146,44 @@ func (m *Module) runCtrlFnWithTimeout(name string, timeout time.Duration, fn fun
}
}
func (m *Module) runCtrlFn(name string, fn func() error) (err error) {
func (m *Module) startCtrlFn(name string, fn func() error) chan error {
ctrlFnError := make(chan error, 1)
// If no function is given, still act as if it was run.
if fn == nil {
return
// Signal finish.
m.ctrlFuncRunning.UnSet()
m.checkIfStopComplete()
// Report nil error and return.
ctrlFnError <- nil
return ctrlFnError
}
if m.ctrlFuncRunning.SetToIf(false, true) {
defer m.ctrlFuncRunning.SetToIf(true, false)
}
// Signal that a control function is running.
m.ctrlFuncRunning.Set()
defer func() {
// recover from panic
panicVal := recover()
if panicVal != nil {
me := m.NewPanicError(name, "module-control", panicVal)
me.Report()
err = me
}
// Start control function in goroutine.
go func() {
// Recover from panic and reset control function signal.
defer func() {
// recover from panic
panicVal := recover()
if panicVal != nil {
me := m.NewPanicError(name, "module-control", panicVal)
me.Report()
ctrlFnError <- fmt.Errorf("panic: %s", panicVal)
}
// Signal finish.
m.ctrlFuncRunning.UnSet()
m.checkIfStopComplete()
}()
// Run control function and report error.
err := fn()
ctrlFnError <- err
}()
// run
err = fn()
return
return ctrlFnError
}