Repository navigation
Expand file tree
/
Copy paththreadworker.go
More file actions
320 lines (264 loc) · 10.9 KB
/
Copy paththreadworker.go
File metadata and controls
320 lines (264 loc) · 10.9 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
package frankenphp
// #include "frankenphp.h"
import "C"
import (
"fmt"
"log/slog"
"time"
"unsafe"
"github.com/dunglas/frankenphp/internal/state"
)
// representation of a thread assigned to a worker script
// executes the PHP worker script in a loop
// implements the threadHandler interface
type workerThread struct {
state *state.ThreadState
thread *phpThread
worker *worker
dummyFrankenPHPContext *frankenPHPContext
workerFrankenPHPContext *frankenPHPContext
isBootingScript bool // true if the worker has not reached frankenphp_handle_request yet
failureCount int // number of consecutive startup failures
requestCount int // number of requests handled since last restart
}
func convertToWorkerThread(thread *phpThread, worker *worker) {
thread.setHandler(&workerThread{
state: thread.state,
thread: thread,
worker: worker,
})
worker.attachThread(thread)
}
// beforeScriptExecution returns the name of the script or an empty string on shutdown
func (handler *workerThread) beforeScriptExecution() string {
switch handler.state.Get() {
case state.TransitionRequested:
if handler.worker.onThreadShutdown != nil {
handler.worker.onThreadShutdown(handler.thread.threadIndex)
}
handler.worker.detachThread(handler.thread)
return handler.thread.transitionToNewHandler()
case state.Ready, state.TransitionComplete:
handler.thread.updateContext(true)
if handler.worker.onThreadReady != nil {
handler.worker.onThreadReady(handler.thread.threadIndex)
}
setupWorkerScript(handler, handler.worker)
return handler.worker.fileName
case state.Rebooting, state.ForceRebooting:
return ""
case state.RebootReady:
handler.requestCount = 0
handler.state.Set(state.Ready)
return handler.beforeScriptExecution()
case state.ShuttingDown:
if handler.worker.onThreadShutdown != nil {
handler.worker.onThreadShutdown(handler.thread.threadIndex)
}
handler.worker.detachThread(handler.thread)
// signal to stop
return ""
default:
panic("unexpected state: " + handler.state.Name())
}
}
func (handler *workerThread) afterScriptExecution(exitStatus int) {
tearDownWorkerScript(handler, exitStatus)
}
func (handler *workerThread) frankenPHPContext() *frankenPHPContext {
if handler.workerFrankenPHPContext != nil {
return handler.workerFrankenPHPContext
}
return handler.dummyFrankenPHPContext
}
func (handler *workerThread) name() string {
return "Worker PHP Thread - " + handler.worker.fileName
}
func (handler *workerThread) drain() {}
func setupWorkerScript(handler *workerThread, worker *worker) {
metrics.StartWorker(worker.name)
// Create a dummy request to set up the worker
fc, err := newWorkerDummyContext(worker)
if err != nil {
panic(err)
}
handler.dummyFrankenPHPContext = fc
handler.isBootingScript = true
handler.requestCount = 0
if fc.logger.Enabled(fc.ctx, slog.LevelDebug) {
fc.logger.LogAttrs(fc.ctx, slog.LevelDebug, "starting", slog.String("worker", worker.name), slog.Int("thread", handler.thread.threadIndex))
}
}
func tearDownWorkerScript(handler *workerThread, exitStatus int) {
worker := handler.worker
handler.dummyFrankenPHPContext = nil
// if the worker request is not nil, the script might have crashed
// make sure to close the worker request context
if handler.workerFrankenPHPContext != nil {
handler.workerFrankenPHPContext.closeContext()
handler.thread.contextMu.Lock()
handler.workerFrankenPHPContext = nil
handler.thread.contextMu.Unlock()
}
// on exit status 0 we just run the worker script again
if exitStatus == 0 && !handler.isBootingScript {
metrics.StopWorker(worker.name, StopReasonRestart)
if globalLogger.Enabled(globalCtx, slog.LevelDebug) {
globalLogger.LogAttrs(globalCtx, slog.LevelDebug, "restarting", slog.String("worker", worker.name), slog.Int("thread", handler.thread.threadIndex), slog.Int("exit_status", exitStatus))
}
return
}
// worker has thrown a fatal error or has not reached frankenphp_handle_request
if handler.isBootingScript {
metrics.StopWorker(worker.name, StopReasonBootFailure)
} else {
metrics.StopWorker(worker.name, StopReasonCrash)
}
if !handler.isBootingScript {
// fatal error (could be due to exit(1), timeouts, etc.)
// unlike a clean restart, this took down any in-flight request, so
// surface it above debug level, with the exit status needed to triage it
if globalLogger.Enabled(globalCtx, slog.LevelWarn) {
globalLogger.LogAttrs(globalCtx, slog.LevelWarn, "unexpected termination, restarting", slog.String("worker", worker.name), slog.Int("thread", handler.thread.threadIndex), slog.Int("exit_status", exitStatus))
}
return
}
if worker.maxConsecutiveFailures >= 0 && startupFailChan != nil && !watcherIsEnabled && handler.failureCount >= worker.maxConsecutiveFailures {
startupFailChan <- fmt.Errorf("too many consecutive failures: worker %s has not reached frankenphp_handle_request()", worker.fileName)
handler.thread.state.Set(state.ShuttingDown)
return
}
if watcherIsEnabled {
// worker script has probably failed due to script changes while watcher is enabled
if globalLogger.Enabled(globalCtx, slog.LevelError) {
globalLogger.LogAttrs(globalCtx, slog.LevelError, "(watcher enabled) worker script has not reached frankenphp_handle_request()", slog.String("worker", worker.name), slog.Int("thread", handler.thread.threadIndex))
}
} else {
// rare case where worker script has failed on a restart during normal operation
// this can happen if startup success depends on external resources
if globalLogger.Enabled(globalCtx, slog.LevelWarn) {
globalLogger.LogAttrs(globalCtx, slog.LevelWarn, "worker script has failed on restart", slog.String("worker", worker.name), slog.Int("thread", handler.thread.threadIndex), slog.Int("failures", handler.failureCount))
}
}
// wait a bit and try again (exponential backoff)
backoffDuration := time.Duration(handler.failureCount*handler.failureCount*100) * time.Millisecond
if backoffDuration > time.Second {
backoffDuration = time.Second
}
handler.failureCount++
time.Sleep(backoffDuration)
}
// waitForWorkerRequest is called during frankenphp_handle_request in the php worker script.
func (handler *workerThread) waitForWorkerRequest() (bool, any) {
// unpin any memory left over from previous requests
handler.thread.Unpin()
if globalLogger.Enabled(globalCtx, slog.LevelDebug) {
globalLogger.LogAttrs(globalCtx, slog.LevelDebug, "waiting for request", slog.String("worker", handler.worker.name), slog.Int("thread", handler.thread.threadIndex))
}
// Clear the first dummy request created to initialize the worker
if handler.isBootingScript {
handler.isBootingScript = false
handler.failureCount = 0
if !C.frankenphp_shutdown_dummy_request() {
panic("Not in CGI context")
}
// worker is truly ready only after reaching frankenphp_handle_request()
metrics.ReadyWorker(handler.worker.name)
}
// max_requests reached: signal reboot for full ZTS cleanup
if maxRequestsPerThread > 0 && handler.requestCount >= maxRequestsPerThread {
if globalLogger.Enabled(globalCtx, slog.LevelDebug) {
globalLogger.LogAttrs(globalCtx, slog.LevelDebug, "max requests reached, restarting",
slog.String("worker", handler.worker.name),
slog.Int("thread", handler.thread.threadIndex),
slog.Int("max_requests", maxRequestsPerThread),
)
}
if handler.thread.reboot() {
return false, nil
}
}
if handler.state.Is(state.TransitionComplete) {
handler.state.Set(state.Ready)
}
handler.state.MarkAsWaiting(true)
var fc *frankenPHPContext
select {
case <-handler.thread.drainChan:
if globalLogger.Enabled(globalCtx, slog.LevelDebug) {
globalLogger.LogAttrs(globalCtx, slog.LevelDebug, "shutting down", slog.String("worker", handler.worker.name), slog.Int("thread", handler.thread.threadIndex))
}
return false, nil
case fc = <-handler.thread.requestChan:
case fc = <-handler.worker.requestChan:
}
handler.requestCount++
handler.thread.contextMu.Lock()
handler.workerFrankenPHPContext = fc
handler.thread.contextMu.Unlock()
handler.state.MarkAsWaiting(false)
if fc.logger.Enabled(fc.ctx, slog.LevelDebug) {
if handler.workerFrankenPHPContext.request == nil {
fc.logger.LogAttrs(fc.ctx, slog.LevelDebug, "request handling started", slog.String("worker", handler.worker.name), slog.Int("thread", handler.thread.threadIndex))
} else {
fc.logger.LogAttrs(fc.ctx, slog.LevelDebug, "request handling started", slog.String("worker", handler.worker.name), slog.Int("thread", handler.thread.threadIndex), slog.String("url", handler.workerFrankenPHPContext.request.RequestURI))
}
}
return true, handler.workerFrankenPHPContext.handlerParameters
}
// go_frankenphp_worker_handle_request_start is called at the start of every php request served.
//
//export go_frankenphp_worker_handle_request_start
func go_frankenphp_worker_handle_request_start(threadIndex C.uintptr_t) (C.bool, unsafe.Pointer) {
handler := phpThreads[threadIndex].handler.(*workerThread)
hasRequest, parameters := handler.waitForWorkerRequest()
if parameters != nil {
var ptr unsafe.Pointer
switch p := parameters.(type) {
case unsafe.Pointer:
ptr = p
default:
ptr = PHPValue(p)
}
handler.thread.Pin(ptr)
return C.bool(hasRequest), ptr
}
return C.bool(hasRequest), nil
}
// go_frankenphp_finish_worker_request is called at the end of every php request served.
//
//export go_frankenphp_finish_worker_request
func go_frankenphp_finish_worker_request(threadIndex C.uintptr_t, retval *C.zval) {
thread := phpThreads[threadIndex]
fc := thread.handler.frankenPHPContext()
if retval != nil {
r, err := GoValue[any](unsafe.Pointer(retval))
if err != nil && fc.logger.Enabled(fc.ctx, slog.LevelError) {
fc.logger.LogAttrs(fc.ctx, slog.LevelError, "cannot convert return value", slog.Any("error", err), slog.Int("thread", thread.threadIndex))
}
fc.handlerReturn = r
}
thread.requestCount.Add(1)
fc.closeContext()
thread.contextMu.Lock()
thread.handler.(*workerThread).workerFrankenPHPContext = nil
thread.contextMu.Unlock()
if fc.logger.Enabled(fc.ctx, slog.LevelDebug) {
if fc.request == nil {
fc.logger.LogAttrs(fc.ctx, slog.LevelDebug, "request handling finished", slog.String("worker", fc.worker.name), slog.Int("thread", thread.threadIndex))
} else {
fc.logger.LogAttrs(fc.ctx, slog.LevelDebug, "request handling finished", slog.String("worker", fc.worker.name), slog.Int("thread", thread.threadIndex), slog.String("url", fc.request.RequestURI))
}
}
}
// when frankenphp_finish_request() is directly called from PHP
//
//export go_frankenphp_finish_php_request
func go_frankenphp_finish_php_request(threadIndex C.uintptr_t) {
thread := phpThreads[threadIndex]
fc := thread.handler.frankenPHPContext()
fc.closeContext()
if fc.logger.Enabled(fc.ctx, slog.LevelDebug) {
fc.logger.LogAttrs(fc.ctx, slog.LevelDebug, "request handling finished", slog.Int("thread", thread.threadIndex), slog.String("url", fc.request.RequestURI))
}
}