==> src/http_relay.c <==1/* The HTTP port: read a head (2 s, at most the http program's request window), route, answer, close. */
2#include "mesh/http_relay.h"
3#include "mesh/abi.h"
4#include "mesh/buf.h"
5#include "mesh/control.h"
6#include "mesh/infer_request.h"
7#include "mesh/inferd.h"
8#include "mesh/log.h"
9#include "mesh/rsfmt.h"
10#include "mesh/views.h"
11
12#include <arpa/inet.h>
13#include <errno.h>
14#include <netinet/in.h>
15#include <poll.h>
16#include <pthread.h>
17#include <stdatomic.h>
18#include <stdio.h>
19#include <stdlib.h>
20#include <string.h>
21#include <sys/socket.h>
22#include <sys/time.h>
23#include <unistd.h>
24
25#define MAX_CLIENTS 64
26#define HEAD_TIME_MS 2000
27#define WRITE_TIME_S 5
28
29static bool has_blank(const mesh_buf *b) {
30 for (size_t i = 0; i + 4 <= b->len; i++)
31 if (b->data[i] == '\r' && b->data[i + 1] == '\n' && b->data[i + 2] == '\r' && b->data[i + 3] == '\n') return true;
32 return false;
33}
34
35/* The request head: up to the http program's request window, until a blank line, EOF or HEAD_TIME in all. */
36static void read_head(int fd, mesh_buf *head) {
37 size_t max = MESH_HTTP_REQ_MAX;
38 int64_t deadline = mesh_mono_ms() + HEAD_TIME_MS;
39 char buf[512];
40 while (head->len < max && !has_blank(head)) {
41 int64_t left = deadline - mesh_mono_ms();
42 if (left <= 0) break;
43 struct pollfd p = {fd, POLLIN, 0};
44 int r = poll(&p, 1, (int)left);
45 if (r <= 0) break;
46 ssize_t k = read(fd, buf, sizeof buf);
47 if (k <= 0) break;
48 mesh_buf_put(head, buf, (size_t)k);
49 }
50 mesh_buf_truncate(head, max);
51}
52
53bool mesh_http_view_route(const char *first, const char **view, const char **query, size_t *qlen) {
54 if (strncmp(first, "GET ", 4) != 0) return false;
55 const char *t = first + 4, *sp = strchr(t, ' ');
56 size_t tn = sp ? (size_t)(sp - t) : strlen(t);
57 const char *q = memchr(t, '?', tn);
58 size_t pn = q ? (size_t)(q - t) : tn;
59 static const char *const views[3] = {"fleet", "jobs", "pipes"};
60 for (int i = 0; i < 3; i++) {
61 size_t k = strlen(views[i]);
62 if (pn == k + 1 && t[0] == '/' && strncmp(t + 1, views[i], k) == 0) {
63 *view = views[i];
64 *query = q ? q + 1 : t + tn;
65 *qlen = q ? tn - pn - 1 : 0;
66 return true;
67 }
68 }
69 return false;
70}
71
72bool mesh_http_render_route(const char *first, const char **what, char **job) {
73 if (strncmp(first, "GET ", 4) != 0) return false;
74 const char *t = first + 4, *sp = strchr(t, ' ');
75 size_t tn = sp ? (size_t)(sp - t) : strlen(t);
76 const char *q = memchr(t, '?', tn);
77 if (!q) return false;
78 size_t pn = (size_t)(q - t);
79 if (pn == 7 && strncmp(t, "/render", 7) == 0) *what = "status";
80 else if (pn == 11 && strncmp(t, "/render.bmp", 11) == 0) *what = "bmp";
81 else if (pn == 11 && strncmp(t, "/render.ppm", 11) == 0) *what = "ppm";
82 else return false;
83 const char *p = q + 1, *end = t + tn;
84 while (p <= end) {
85 const char *amp = memchr(p, '&', (size_t)(end - p));
86 const char *e = amp ? amp : end;
87 if ((size_t)(e - p) >= 4 && strncmp(p, "job=", 4) == 0) {
88 *job = e - p == 4 ? mesh_xstrdup("latest") : mesh_xstrndup(p + 4, (size_t)(e - p - 4));
89 return true;
90 }
91 if (!amp) break;
92 p = amp + 1;
93 }
94 return false;
95}
96
97/* The head's first line, lossily as text. */
98static char *first_line(const uint8_t *b, size_t n) {
99 size_t k = 0;
100 while (k < n && b[k] != '\r' && b[k] != '\n') k++;
101 mesh_buf t = MESH_BUF_INIT;
102 mesh_rs_lossy(&t, b, k);
103 return mesh_buf_take(&t);
104}
105
106static size_t write_all(int fd, const void *p, size_t n) {
107 const char *c = p;
108 size_t left = n;
109 while (left) {
110 ssize_t w = write(fd, c, left);
111 if (w < 0 && errno == EINTR) continue;
112 if (w <= 0) return 0;
113 c += w;
114 left -= (size_t)w;
115 }
116 return n;
117}
118
119static void timing(mesh_buf *l, int64_t t0) {
120 mesh_rs_fixed(l, (double)(mesh_mono_ns() - t0) / 1e6, 2);
121}
122
123void mesh_http_client(mesh_shared *sh, int fd) {
124 struct sockaddr_storage peer;
125 socklen_t pl = sizeof peer;
126 char who[80] = "";
127 if (getpeername(fd, (struct sockaddr *)&peer, &pl) == 0 && peer.ss_family == AF_INET) {
128 struct sockaddr_in *a = (struct sockaddr_in *)&peer;
129 char ip[32];
130 inet_ntop(AF_INET, &a->sin_addr, ip, sizeof ip);
131 snprintf(who, sizeof who, "%s:%u", ip, ntohs(a->sin_port));
132 }
133 struct timeval wt = {WRITE_TIME_S, 0};
134 setsockopt(fd, SOL_SOCKET, SO_SNDTIMEO, &wt, sizeof wt);
135 mesh_buf head = MESH_BUF_INIT;
136 read_head(fd, &head);
137 if (!head.len) { mesh_net_counters_http(&sh->host.net, 0, 0); close(fd); return; }
138 int64_t t0 = mesh_mono_ns();
139 char *first = first_line((uint8_t *)head.data, head.len);
140 const char *view, *query;
141 size_t qlen;
142 const char *what;
143 char *job = NULL;
144 mesh_buf resp = MESH_BUF_INIT, log = MESH_BUF_INIT;
145 if (mesh_http_view_route(first, &view, &query, &qlen)) {
146 /* GET /fleet, /jobs, /pipes [?format=json|text]: the fleet's views, from meshd. */
147 char *q = mesh_xstrndup(query, qlen);
148 bool json = false;
149 char *err = NULL;
150 mesh_buf body = MESH_BUF_INIT;
151 const char *code, *ctype;
152 if (mesh_views_format_of(q, &json, &err) == 0 && mesh_control_view(sh, view, json, &body, &err) == 0) {
153 code = "200 OK";
154 ctype = json ? "application/json" : "text/plain; charset=utf-8";
155 } else {
156 code = "400 Bad Request";
157 ctype = "text/plain; charset=utf-8";
158 mesh_buf_clear(&body);
159 mesh_buf_printf(&body, "meshd: %s\n", err);
160 free(err);
161 }
162 mesh_buf_printf(&resp, "HTTP/1.0 %s\r\nContent-Type: %s\r\nContent-Length: %zu\r\nCache-Control: no-store\r\n\r\n", code, ctype, body.len);
163 mesh_buf_put(&resp, body.data ? body.data : "", body.len);
164 size_t sent = write_all(fd, resp.data, resp.len);
165 mesh_net_counters_http(&sh->host.net, head.len, sent);
166 mesh_buf_printf(&log, "http %s \"%s\" -> %.3s %zu bytes ", who, first, code, resp.len);
167 timing(&log, t0);
168 mesh_buf_printf(&log, " ms (meshd: the %s view)", view);
169 mesh_info("%s", mesh_buf_str(&log));
170 mesh_buf_free(&body);
171 free(q);
172 } else if (strncmp(first, "GET /infer?", 11) == 0) {
173 /* GET /infer?prompt=..&max=..&mode=..&prefill=..&decode=..&mono=..: GPT-2 across the fleet. */
174 const char *qs = first + 11, *sp = strchr(qs, ' ');
175 char *q = sp ? mesh_xstrndup(qs, (size_t)(sp - qs)) : mesh_xstrdup(qs);
176 mesh_infer_request r;
177 mesh_infer_msg res;
178 bool full = false;
179 char *err = NULL;
180 mesh_buf body = MESH_BUF_INIT;
181 const char *code;
182 if (mesh_infer_from_query(q, &r, &full, &err) == 0 && mesh_inferd_run(sh, &r, &res, &err) == 0) {
183 code = "200 OK";
184 if (full) mesh_infer_text(&body, &res);
185 else mesh_buf_printf(&body, "%s\n", res.body);
186 mesh_infer_msg_free(&res);
187 } else {
188 code = "400 Bad Request";
189 mesh_buf_printf(&body, "meshd: infer: %s\n", err);
190 free(err);
191 }
192 mesh_infer_request_free(&r);
193 mesh_buf_printf(&resp, "HTTP/1.0 %s\r\nContent-Type: text/plain; charset=utf-8\r\nContent-Length: %zu\r\nCache-Control: no-store\r\n\r\n", code,
194 body.len);
195 mesh_buf_put(&resp, body.data ? body.data : "", body.len);
196 size_t sent = write_all(fd, resp.data, resp.len);
197 mesh_net_counters_http(&sh->host.net, head.len, sent);
198 mesh_buf_printf(&log, "http %s \"%s\" -> %.3s %zu bytes ", who, first, code, resp.len);
199 timing(&log, t0);
200 mesh_buf_puts(&log, " ms (meshd: infer)");
201 mesh_info("%s", mesh_buf_str(&log));
202 mesh_buf_free(&body);
203 free(q);
204 } else if (mesh_http_render_route(first, &what, &job)) {
205 char *words[2] = {job, (char *)what};
206 uint8_t *b = NULL;
207 size_t n = 0;
208 char *err = NULL;
209 bool ok = mesh_control_render ? mesh_control_render(sh, words, 2, &b, &n, &err) : (err = mesh_xstrdup("the blob thread is gone"), false);
210 const char *code = ok ? "200 OK" : "404 Not Found";
211 const char *ctype = !ok ? "text/plain; charset=utf-8" : strcmp(what, "bmp") == 0 ? "image/bmp" : strcmp(what, "ppm") == 0 ? "image/x-portable-pixmap"
212 : "text/plain; charset=utf-8";
213 mesh_buf body = MESH_BUF_INIT;
214 if (ok) mesh_buf_put(&body, b ? b : (uint8_t *)"", n);
215 else mesh_buf_printf(&body, "meshd: %s\n", err);
216 mesh_buf_printf(&resp, "HTTP/1.0 %s\r\nContent-Type: %s\r\nContent-Length: %zu\r\nCache-Control: no-store\r\n\r\n", code, ctype, body.len);
217 mesh_buf_put(&resp, body.data ? body.data : "", body.len);
218 size_t sent = write_all(fd, resp.data, resp.len);
219 mesh_net_counters_http(&sh->host.net, head.len, sent);
220 mesh_buf_printf(&log, "http %s \"%s\" -> %.3s %zu bytes ", who, first, code, resp.len);
221 timing(&log, t0);
222 mesh_buf_puts(&log, " ms (meshd: the assembled image)");
223 mesh_info("%s", mesh_buf_str(&log));
224 mesh_buf_free(&body);
225 free(b);
226 free(err);
227 free(job);
228 } else {
229 uint8_t *r = NULL;
230 size_t rn = 0;
231 char *err = NULL;
232 if (mesh_shared_http(sh, (uint8_t *)head.data, head.len, &r, &rn, &err)) {
233 size_t sent = write_all(fd, r, rn);
234 mesh_net_counters_http(&sh->host.net, head.len, sent);
235 char *rl = first_line(r, rn);
236 mesh_buf_printf(&log, "http %s \"%s\" -> \"%s\" %zu bytes ", who, first, rl, rn);
237 timing(&log, t0);
238 mesh_buf_puts(&log, " ms");
239 mesh_info("%s", mesh_buf_str(&log));
240 free(rl);
241 free(r);
242 } else {
243 /* No GPU response (no http program, or it failed): say so, never hang the client. */
244 mesh_error("http %s \"%s\": %s", who, first, err);
245 char *body = mesh_buf_fmt("meshd: %s\n", err);
246 mesh_buf_printf(&resp, "HTTP/1.0 503 Service Unavailable\r\nContent-Type: text/plain\r\nContent-Length: %zu\r\n\r\n%s", strlen(body), body);
247 size_t sent = write_all(fd, resp.data, resp.len);
248 mesh_net_counters_http(&sh->host.net, head.len, sent);
249 free(body);
250 free(err);
251 }
252 }
253 shutdown(fd, SHUT_RDWR);
254 close(fd);
255 mesh_buf_free(&resp);
256 mesh_buf_free(&log);
257 mesh_buf_free(&head);
258 free(first);
259}
260
261typedef struct {
262 mesh_shared *sh;
263 int fd;
264} client_arg;
265
266static _Atomic int clients;
267
268static void *client_thread(void *p) {
269 client_arg *a = p;
270 mesh_http_client(a->sh, a->fd);
271 atomic_fetch_sub(&clients, 1);
272 free(a);
273 return NULL;
274}
275
276void mesh_http_serve(mesh_shared *sh, int listen_fd) {
277 for (;;) {
278 struct sockaddr_storage peer;
279 socklen_t pl = sizeof peer;
280 int c = accept(listen_fd, (struct sockaddr *)&peer, &pl);
281 if (c < 0) {
282 if (errno == EINTR) continue;
283 mesh_error("http accept: %s (os error %d)", strerror(errno), errno);
284 if (errno == EBADF || errno == EINVAL) return;
285 continue;
286 }
287 if (atomic_load(&clients) >= MAX_CLIENTS) {
288 char who[80] = "";
289 if (peer.ss_family == AF_INET) {
290 struct sockaddr_in *a = (struct sockaddr_in *)&peer;
291 char ip[32];
292 inet_ntop(AF_INET, &a->sin_addr, ip, sizeof ip);
293 snprintf(who, sizeof who, "%s:%u", ip, ntohs(a->sin_port));
294 }
295 mesh_error("http: %d clients already; closing %s", MAX_CLIENTS, who);
296 close(c);
297 continue;
298 }
299 atomic_fetch_add(&clients, 1);
300 client_arg *a = mesh_xmalloc(sizeof *a);
301 a->sh = sh;
302 a->fd = c;
303 pthread_t th;
304 pthread_attr_t at;
305 pthread_attr_init(&at);
306 pthread_attr_setdetachstate(&at, PTHREAD_CREATE_DETACHED);
307 if (pthread_create(&th, &at, client_thread, a) != 0) {
308 atomic_fetch_sub(&clients, 1);
309 mesh_error("http: cannot spawn a client thread: %s", strerror(errno));
310 close(c);
311 free(a);
312 }
313 pthread_attr_destroy(&at);
314 }
315}