Skip to content
Merged
Show file tree
Hide file tree
Changes from 2 commits
Commits
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
2 changes: 1 addition & 1 deletion caddy/caddy.go
Original file line number Diff line number Diff line change
Expand Up @@ -113,7 +113,7 @@ func (f *FrankenPHPApp) Stop() error {
// note: Exiting() is currently marked as 'experimental'
// https://github.com/caddyserver/caddy/blob/e76405d55058b0a3e5ba222b44b5ef00516116aa/caddy.go#L810
if caddy.Exiting() {
frankenphp.Shutdown()
frankenphp.DrainWorkers()
}

// reset configuration so it doesn't bleed into later tests
Expand Down
2 changes: 1 addition & 1 deletion frankenphp.go
Original file line number Diff line number Diff line change
Expand Up @@ -372,7 +372,7 @@ func Shutdown() {
return
}

drainWorkers()
drainWatcher()
drainAutoScaling()
drainPHPThreads()

Expand Down
12 changes: 6 additions & 6 deletions metrics.go
Original file line number Diff line number Diff line change
Expand Up @@ -380,9 +380,9 @@ func (m *PrometheusMetrics) Shutdown() {
}

if err := m.registry.Register(m.queueDepth); err != nil &&
!errors.As(err, &prometheus.AlreadyRegisteredError{}) {
panic(err)
}
!errors.As(err, &prometheus.AlreadyRegisteredError{}) {
panic(err)
}
}

func getWorkerNameForMetrics(name string) string {
Expand Down Expand Up @@ -432,9 +432,9 @@ func NewPrometheusMetrics(registry prometheus.Registerer) *PrometheusMetrics {
}

if err := m.registry.Register(m.queueDepth); err != nil &&
!errors.As(err, &prometheus.AlreadyRegisteredError{}) {
panic(err)
}
!errors.As(err, &prometheus.AlreadyRegisteredError{}) {
panic(err)
}

return m
}
34 changes: 23 additions & 11 deletions worker.go
Original file line number Diff line number Diff line change
Expand Up @@ -86,18 +86,14 @@ func newWorker(o workerOpt) (*worker, error) {
return w, nil
}

func drainWorkers() {
watcher.DrainWatcher()
// DrainWorkers finishes all worker scripts before a graceful shutdown
func DrainWorkers() {
Comment thread
dunglas marked this conversation as resolved.
_ = drainWorkerThreads()
}

// RestartWorkers attempts to restart all workers gracefully
func RestartWorkers() {
// disallow scaling threads while restarting workers
scalingMu.Lock()
defer scalingMu.Unlock()

func drainWorkerThreads() []*phpThread {
ready := sync.WaitGroup{}
threadsToRestart := make([]*phpThread, 0)
drainedThreads := make([]*phpThread, 0)
for _, worker := range workers {
worker.threadMutex.RLock()
ready.Add(len(worker.threads))
Expand All @@ -108,17 +104,33 @@ func RestartWorkers() {
continue
}
close(thread.drainChan)
threadsToRestart = append(threadsToRestart, thread)
drainedThreads = append(drainedThreads, thread)
go func(thread *phpThread) {
thread.state.waitFor(stateYielding)
ready.Done()
}(thread)
}
worker.threadMutex.RUnlock()
}

ready.Wait()

return drainedThreads
}

func drainWatcher() {
if watcherIsEnabled {
watcher.DrainWatcher()
}
}

// RestartWorkers attempts to restart all workers gracefully
func RestartWorkers() {
// disallow scaling threads while restarting workers
scalingMu.Lock()
defer scalingMu.Unlock()

threadsToRestart := drainWorkerThreads()

for _, thread := range threadsToRestart {
thread.drainChan = make(chan struct{})
thread.state.set(stateReady)
Expand Down