server_workers.mx raw
1 package server
2
3 import (
4 "fmt"
5 "runtime"
6 "time"
7
8 "git.smesh.lol/morly/pkg/pool"
9 "git.smesh.lol/morly/pkg/relay/wire"
10 "git.smesh.lol/morly/pkg/transport"
11 )
12
13 // --- Worker lifecycle ---
14
15 func (s *Server) startProxyWorkers(n int32) {
16 s.proxyIn, s.proxyOut, s.proxyDone = spawnProxyPool(n)
17 s.proxyBusyTime = []int64{:n}
18 s.proxyPool = pool.NewPool(n)
19 }
20
21 // spawnProxyPool forks the proxy domains in a free function: the channels and
22 // the loop scratch belong to this frame, and only the finished slices land in
23 // the Server's sovereign arena. Forking inside a self-mutating method's loop
24 // left the receiver's slices unusable part-way through the pool.
25 //
26 // The slices are grown with push because an inline []chan T{:n} literal is
27 // miscompiled to an empty slice; see spawnIngestPool.
28 func spawnProxyPool(n int32) (ins []chan wire.ProxyRequest, outs []chan wire.ProxyResponse, dones []chan struct{}) {
29 var in chan wire.ProxyRequest
30 var out chan wire.ProxyResponse
31 var done chan struct{}
32 for i := 0; i < n; i++ {
33 in, out, done = newProxyWorker()
34 ins = push(ins, in)
35 outs = push(outs, out)
36 dones = push(dones, done)
37 }
38 return
39 }
40
41 // newProxyWorker creates the channel pair and forks one proxy domain.
42 func newProxyWorker() (in chan wire.ProxyRequest, out chan wire.ProxyResponse, done chan struct{}) {
43 // Root lifetime: the server stores these and polls them on every tick.
44 runtime.SovereignSetArena(runtime.RootArena())
45 in = chan wire.ProxyRequest{}
46 out = chan wire.ProxyResponse{}
47 done = spawn(wire.ProxyWorker, in, out)
48 runtime.SovereignRestoreArena(runtime.RootArena())
49 return
50 }
51
52 func (s *Server) startBlossomWorkers(n int32) {
53 s.blossomIn, s.blossomOut, s.blossomDone = spawnBlossomPool(n)
54 s.blossomPool = pool.NewPool(n)
55 }
56
57 // spawnBlossomPool mirrors spawnProxyPool for the blossom domains.
58 func spawnBlossomPool(n int32) (ins []chan wire.BlossomRequest, outs []chan wire.BlossomResponse, dones []chan struct{}) {
59 var in chan wire.BlossomRequest
60 var out chan wire.BlossomResponse
61 var done chan struct{}
62 for i := 0; i < n; i++ {
63 in, out, done = newBlossomWorker()
64 ins = push(ins, in)
65 outs = push(outs, out)
66 dones = push(dones, done)
67 }
68 return
69 }
70
71 // newBlossomWorker creates the channel pair and forks one blossom domain.
72 func newBlossomWorker() (in chan wire.BlossomRequest, out chan wire.BlossomResponse, done chan struct{}) {
73 // Root lifetime: the server stores these and polls them on every tick.
74 runtime.SovereignSetArena(runtime.RootArena())
75 in = chan wire.BlossomRequest{}
76 out = chan wire.BlossomResponse{}
77 done = spawn(wire.BlossomWorker, in, out)
78 runtime.SovereignRestoreArena(runtime.RootArena())
79 return
80 }
81
82 func (s *Server) respawnProxyWorker(i int32) {
83 in, out, done := newProxyWorker()
84 s.proxyIn[i] = in
85 s.proxyOut[i] = out
86 s.proxyDone[i] = done
87 s.proxyPool.Busy[i] = false
88 s.proxyBusyTime[i] = 0
89 fmt.Println("respawned proxy worker", i)
90 }
91
92 func (s *Server) respawnBlossomWorker(i int32) {
93 in, out, done := newBlossomWorker()
94 s.blossomIn[i] = in
95 s.blossomOut[i] = out
96 s.blossomDone[i] = done
97 s.blossomPool.Busy[i] = false
98 fmt.Println("respawned blossom worker", i)
99 }
100
101 // --- Proxy dispatch ---
102
103 func (s *Server) doDispatchProxy(fd int32, path string, headers map[string]string) (ok bool) {
104 target := path[len("/proxy/"):]
105 if target == "" {
106 s.t.SendHTTP(fd, 400, map[string]string{"Content-Type": "text/plain"}, []byte("missing url\n"))
107 return true
108 }
109 if transport.HasPrefix(target, s.selfHost) {
110 direct := target[len(s.selfHost)-1:]
111 s.t.SendHTTP(fd, 302, map[string]string{
112 "Location": direct,
113 "Access-Control-Allow-Origin": "*",
114 }, nil)
115 return true
116 }
117 targetURL := []byte("https://" | target)
118 s.nextAsyncID++
119 rid := s.nextAsyncID
120 connClose := headers["connection"] == "close"
121 req := wire.ProxyRequest{ReqID: rid, MaxBytes: 32 * 1024 * 1024, URL: targetURL}
122 s.proxyReapStuck()
123 if i := s.proxyPool.IdleIndex(); i >= 0 {
124 s.asyncPending[rid] = asyncHTTPEntry{connFD: fd, connClose: connClose, createdAt: time.Now().UnixNano()}
125 s.proxyPool.Busy[i] = true
126 s.proxyBusyTime[i] = time.Now().UnixNano()
127 if !s.proxyDispatchFrame(i, req) {
128 s.proxyPool.Busy[i] = false
129 s.proxyBusyTime[i] = 0
130 delete(s.asyncPending, rid)
131 } else {
132 return true
133 }
134 }
135 if len(s.proxyQueue) < 64 {
136 s.proxyQueue = push(s.proxyQueue, pendingProxy{connFD: fd, connClose: connClose, req: req})
137 return true
138 }
139 s.t.SendHTTP(fd, 503, map[string]string{
140 "Content-Type": "text/plain",
141 "Retry-After": "1",
142 }, []byte("proxy workers busy\n"))
143 return true
144 }
145
146 func (s *Server) drainProxyQueue() {
147 var i int32
148 var p pendingProxy
149 for len(s.proxyQueue) > 0 {
150 i = s.proxyPool.IdleIndex()
151 if i < 0 {
152 return
153 }
154 p = s.proxyQueue[0]
155 s.proxyQueue = s.proxyQueue[1:]
156 s.asyncPending[p.req.ReqID] = asyncHTTPEntry{connFD: p.connFD, connClose: p.connClose, createdAt: time.Now().UnixNano()}
157 s.proxyPool.Busy[i] = true
158 s.proxyBusyTime[i] = time.Now().UnixNano()
159 if !s.proxyDispatchFrame(i, p.req) {
160 s.proxyPool.Busy[i] = false
161 s.proxyBusyTime[i] = 0
162 delete(s.asyncPending, p.req.ReqID)
163 s.t.SendHTTP(p.connFD, 502, map[string]string{"Content-Type": "text/plain"}, []byte("proxy dispatch failed\n"))
164 }
165 }
166 }
167
168 func (s *Server) proxyReapStuck() {
169 now := time.Now().UnixNano()
170 reaped := false
171 for i := 0; i < s.proxyPool.Len(); i++ {
172 if s.proxyPool.Busy[i] && s.proxyBusyTime[i] > 0 {
173 elapsed := now - s.proxyBusyTime[i]
174 if elapsed > 15_000_000_000 {
175 fmt.Println("proxy worker", i, "stuck", elapsed/1_000_000_000, "s, respawning")
176 s.respawnProxyWorker(i)
177 reaped = true
178 }
179 }
180 }
181 if reaped {
182 s.drainProxyQueue()
183 }
184 }
185
186 func (s *Server) proxyDispatchFrame(i int32, req wire.ProxyRequest) (ok bool) {
187 s.proxyIn[i] <- req
188 return true
189 }
190
191 // --- Blossom dispatch ---
192
193 func (s *Server) doDispatchBlossom(fd int32, method, path string, headers map[string]string, body []byte) (ok bool) {
194 bpath := path[len("/blossom"):]
195 i := s.blossomPool.IdleIndex()
196 if i >= 0 {
197 s.nextAsyncID++
198 rid := s.nextAsyncID
199 req := wire.BlossomRequest{
200 ReqID: rid,
201 Dir: []byte(s.cfg.BlossomDir),
202 Method: []byte(method),
203 Path: []byte(bpath),
204 ContentType: []byte(headers["content-type"]),
205 Body: body,
206 Upstream: []byte(s.cfg.BlossomUpstream),
207 }
208 s.blossomIn[i] <- req
209 connClose := headers["connection"] == "close"
210 s.asyncPending[rid] = asyncHTTPEntry{connFD: fd, connClose: connClose, createdAt: time.Now().UnixNano()}
211 s.blossomPool.Busy[i] = true
212 return true
213 }
214 s.t.SendHTTP(fd, 503, map[string]string{
215 "Content-Type": "text/plain",
216 "Retry-After": "1",
217 }, []byte("blossom worker busy\n"))
218 return true
219 }
220
221 // asyncReapStuck removes asyncPending entries older than 60s.
222 // Does NOT send a response - connFD may have been reused by another
223 // connection by the time 60s have elapsed.
224 func (s *Server) asyncReapStuck() {
225 cutoff := time.Now().UnixNano() - 60_000_000_000
226 for rid, entry := range s.asyncPending {
227 if entry.createdAt > 0 && entry.createdAt < cutoff {
228 fmt.Println("asyncPending: reaping stale entry rid=", rid, "fd=", entry.connFD)
229 delete(s.asyncPending, rid)
230 }
231 }
232 }
233
234 // --- Worker response handlers ---
235
236 func (s *Server) pollProxyWorkers() {
237 var resp wire.ProxyResponse
238 for i := int32(0); i < s.proxyPool.Len(); i++ {
239 if !workerAlive(s.proxyDone[i]) {
240 s.respawnProxyWorker(i)
241 s.drainProxyQueue()
242 continue
243 }
244 select {
245 case resp = <-s.proxyOut[i]:
246 s.proxyPool.Busy[i] = false
247 s.proxyBusyTime[i] = 0
248 s.drainProxyQueue()
249 s.completeProxyResponse(resp)
250 default:
251 }
252 }
253 }
254
255 func (s *Server) completeProxyResponse(resp wire.ProxyResponse) {
256 entry, ok := s.asyncPending[resp.ReqID]
257 if !ok {
258 return
259 }
260 delete(s.asyncPending, resp.ReqID)
261 var status int32
262 var h map[string]string
263 var body []byte
264 switch {
265 case resp.Status < 0:
266 status = 502
267 h = map[string]string{"Content-Type": "text/plain"}
268 body = []byte("proxy: " | string(resp.Err) | "\n")
269 case resp.Status == 415:
270 status = 415
271 h = map[string]string{"Content-Type": "text/plain"}
272 body = []byte("content-type not allowed\n")
273 case resp.Status >= 200 && resp.Status < 300:
274 status = 200
275 h = map[string]string{
276 "Content-Type": string(resp.ContentType),
277 "Cross-Origin-Resource-Policy": "cross-origin",
278 "Cache-Control": "public, max-age=86400",
279 "Access-Control-Allow-Origin": "*",
280 }
281 body = resp.Body
282 default:
283 status = int32(resp.Status)
284 h = map[string]string{
285 "Content-Type": "text/plain",
286 "Cache-Control": "public, max-age=3600",
287 }
288 body = []byte(fmt.Sprintf("upstream %d\n", resp.Status))
289 }
290 s.t.CompleteHTTP(entry.connFD, status, h, body, entry.connClose)
291 resp.Body = nil
292 resp.ContentType = nil
293 resp.Err = nil
294 body = nil
295 }
296
297 var blossomCORSHeaders map[string]string
298
299 func (s *Server) pollBlossomWorkers() {
300 var resp wire.BlossomResponse
301 for i := int32(0); i < s.blossomPool.Len(); i++ {
302 if !workerAlive(s.blossomDone[i]) {
303 s.respawnBlossomWorker(i)
304 continue
305 }
306 select {
307 case resp = <-s.blossomOut[i]:
308 s.blossomPool.Busy[i] = false
309 s.completeBlossomResponse(resp)
310 default:
311 }
312 }
313 }
314
315 func (s *Server) completeBlossomResponse(resp wire.BlossomResponse) {
316 entry, ok := s.asyncPending[resp.ReqID]
317 if !ok {
318 return
319 }
320 delete(s.asyncPending, resp.ReqID)
321 h := map[string]string{}
322 for k, v := range blossomCORSHeaders {
323 h[k] = v
324 }
325 if len(resp.CT) > 0 {
326 h["Content-Type"] = string(resp.CT)
327 }
328 if resp.Size > 0 {
329 h["Content-Length"] = fmt.Sprintf("%d", resp.Size)
330 h["Content-Type"] = "application/octet-stream"
331 }
332 s.t.CompleteHTTP(entry.connFD, int32(resp.Status), h, resp.Body, entry.connClose)
333 resp.Body = nil
334 resp.CT = nil
335 }
336
337 func init() {
338 blossomCORSHeaders = map[string]string{
339 "Access-Control-Allow-Origin": "*",
340 "Access-Control-Allow-Methods": "GET, PUT, DELETE, HEAD, OPTIONS",
341 "Access-Control-Allow-Headers": "Authorization, Content-Type",
342 }
343 }
344