@@ -86,18 +86,14 @@ func newWorker(o workerOpt) (*worker, error) {
8686 return w , nil
8787}
8888
89- func drainWorkers () {
90- watcher .DrainWatcher ()
89+ // EXPERIMENTAL: DrainWorkers finishes all worker scripts before a graceful shutdown
90+ func DrainWorkers () {
91+ _ = drainWorkerThreads ()
9192}
9293
93- // RestartWorkers attempts to restart all workers gracefully
94- func RestartWorkers () {
95- // disallow scaling threads while restarting workers
96- scalingMu .Lock ()
97- defer scalingMu .Unlock ()
98-
94+ func drainWorkerThreads () []* phpThread {
9995 ready := sync.WaitGroup {}
100- threadsToRestart := make ([]* phpThread , 0 )
96+ drainedThreads := make ([]* phpThread , 0 )
10197 for _ , worker := range workers {
10298 worker .threadMutex .RLock ()
10399 ready .Add (len (worker .threads ))
@@ -108,17 +104,33 @@ func RestartWorkers() {
108104 continue
109105 }
110106 close (thread .drainChan )
111- threadsToRestart = append (threadsToRestart , thread )
107+ drainedThreads = append (drainedThreads , thread )
112108 go func (thread * phpThread ) {
113109 thread .state .waitFor (stateYielding )
114110 ready .Done ()
115111 }(thread )
116112 }
117113 worker .threadMutex .RUnlock ()
118114 }
119-
120115 ready .Wait ()
121116
117+ return drainedThreads
118+ }
119+
120+ func drainWatcher () {
121+ if watcherIsEnabled {
122+ watcher .DrainWatcher ()
123+ }
124+ }
125+
126+ // RestartWorkers attempts to restart all workers gracefully
127+ func RestartWorkers () {
128+ // disallow scaling threads while restarting workers
129+ scalingMu .Lock ()
130+ defer scalingMu .Unlock ()
131+
132+ threadsToRestart := drainWorkerThreads ()
133+
122134 for _ , thread := range threadsToRestart {
123135 thread .drainChan = make (chan struct {})
124136 thread .state .set (stateReady )
0 commit comments