==> src/server.c <==1/* The mock's control port: an accepting thread polls a non-blocking listener, each connection
2 * gets a thread that reads request lines and answers framed bytes (u32 LE length + bytes) or one
3 * text line, exactly as mock.rs's Mock::start_with / serve do. */
4#include "mesh/top_mock.h"
5
6#include <arpa/inet.h>
7#include <errno.h>
8#include <fcntl.h>
9#include <netinet/in.h>
10#include <poll.h>
11#include <pthread.h>
12#include <stdatomic.h>
13#include <stdio.h>
14#include <stdlib.h>
15#include <string.h>
16#include <sys/socket.h>
17#include <unistd.h>
18
19typedef struct {
20 pthread_t th;
21 int fd;
22 struct mt_mock *m;
23} conn;
24
25struct mt_mock {
26 char addr[64];
27 mt_mock_world *world;
28 bool owned;
29 pthread_mutex_t world_lock;
30 pthread_mutex_t log_lock;
31 char **log;
32 size_t nlog, caplog;
33 int lfd;
34 atomic_bool stop;
35 pthread_t acceptor;
36 pthread_mutex_t conns_lock;
37 conn **conns;
38 size_t nconns, capconns;
39};
40
41static bool split_addr(const char *addr, struct sockaddr_in *sa) {
42 const char *colon = strrchr(addr, ':');
43 if (!colon) return false;
44 char host[64];
45 size_t hl = (size_t)(colon - addr);
46 if (hl >= sizeof host) return false;
47 memcpy(host, addr, hl);
48 host[hl] = 0;
49 if (strcmp(host, "localhost") == 0) strcpy(host, "127.0.0.1");
50 char *end;
51 unsigned long port = strtoul(colon + 1, &end, 10);
52 if (*end || port > 65535) return false;
53 memset(sa, 0, sizeof *sa);
54 sa->sin_family = AF_INET;
55 sa->sin_port = htons((uint16_t)port);
56 return inet_pton(AF_INET, host, &sa->sin_addr) == 1;
57}
58
59static bool send_all(int fd, const void *p, size_t n) {
60 const uint8_t *b = p;
61 while (n) {
62 ssize_t k = send(fd, b, n, MSG_NOSIGNAL);
63 if (k < 0 && errno == EINTR) continue;
64 if (k <= 0) return false;
65 b += k;
66 n -= (size_t)k;
67 }
68 return true;
69}
70
71static void log_req(struct mt_mock *m, const char *req) {
72 pthread_mutex_lock(&m->log_lock);
73 if (m->nlog == m->caplog) {
74 m->caplog = m->caplog ? m->caplog * 2 : 64;
75 m->log = realloc(m->log, m->caplog * sizeof *m->log);
76 if (!m->log) abort();
77 }
78 m->log[m->nlog] = malloc(strlen(req) + 1);
79 if (!m->log[m->nlog]) abort();
80 strcpy(m->log[m->nlog++], req);
81 pthread_mutex_unlock(&m->log_lock);
82}
83
84/* The trimmed request, as the log keeps it. */
85static char *trimmed(const char *s, size_t l) {
86 while (l && (*s == ' ' || (*s >= '\t' && *s <= '\r'))) s++, l--;
87 while (l && (s[l - 1] == ' ' || (s[l - 1] >= '\t' && s[l - 1] <= '\r'))) l--;
88 char *r = malloc(l + 1);
89 if (!r) abort();
90 memcpy(r, s, l);
91 r[l] = 0;
92 return r;
93}
94
95static void *serve(void *arg) {
96 conn *c = arg;
97 struct mt_mock *m = c->m;
98 size_t cap = 4096, have = 0;
99 char *buf = malloc(cap);
100 if (!buf) abort();
101 for (;;) {
102 char *nl = have ? memchr(buf, '\n', have) : NULL;
103 if (!nl) {
104 if (have == cap) {
105 cap *= 2;
106 buf = realloc(buf, cap);
107 if (!buf) abort();
108 }
109 ssize_t k = recv(c->fd, buf + have, cap - have, 0);
110 if (k < 0 && errno == EINTR) continue;
111 if (k <= 0) break; /* EOF (a partial last line is dropped, as read_line's caller does) */
112 have += (size_t)k;
113 continue;
114 }
115 size_t ll = (size_t)(nl - buf) + 1;
116 char *req = trimmed(buf, ll);
117 memmove(buf, buf + ll, have - ll);
118 have -= ll;
119 uint8_t *bytes;
120 size_t len;
121 char *text;
122 pthread_mutex_lock(&m->world_lock);
123 mt_mock_request(m->world, req, &bytes, &len, &text);
124 pthread_mutex_unlock(&m->world_lock);
125 log_req(m, req);
126 free(req);
127 bool ok;
128 if (text) {
129 size_t tl = strlen(text);
130 char *t = malloc(tl + 2);
131 if (!t) abort();
132 memcpy(t, text, tl);
133 t[tl] = '\n';
134 ok = send_all(c->fd, t, tl + 1);
135 free(t);
136 free(text);
137 } else {
138 uint8_t hdr[4] = {(uint8_t)len, (uint8_t)(len >> 8), (uint8_t)(len >> 16), (uint8_t)(len >> 24)};
139 ok = send_all(c->fd, hdr, 4) && send_all(c->fd, bytes, len);
140 free(bytes);
141 }
142 if (!ok) break;
143 }
144 free(buf);
145 return NULL;
146}
147
148static void *accept_loop(void *arg) {
149 struct mt_mock *m = arg;
150 while (!atomic_load(&m->stop)) {
151 struct pollfd p = {.fd = m->lfd, .events = POLLIN};
152 if (poll(&p, 1, 10) <= 0) continue;
153 int fd = accept(m->lfd, NULL, NULL);
154 if (fd < 0) continue;
155 int fl = fcntl(fd, F_GETFL);
156 if (fl >= 0) fcntl(fd, F_SETFL, fl & ~O_NONBLOCK);
157 conn *c = malloc(sizeof *c);
158 if (!c) abort();
159 c->fd = fd;
160 c->m = m;
161 pthread_mutex_lock(&m->conns_lock);
162 if (atomic_load(&m->stop)) {
163 pthread_mutex_unlock(&m->conns_lock);
164 close(fd);
165 free(c);
166 break;
167 }
168 if (m->nconns == m->capconns) {
169 m->capconns = m->capconns ? m->capconns * 2 : 8;
170 m->conns = realloc(m->conns, m->capconns * sizeof *m->conns);
171 if (!m->conns) abort();
172 }
173 if (pthread_create(&c->th, NULL, serve, c) != 0) {
174 pthread_mutex_unlock(&m->conns_lock);
175 close(fd);
176 free(c);
177 continue;
178 }
179 m->conns[m->nconns++] = c;
180 pthread_mutex_unlock(&m->conns_lock);
181 }
182 return NULL;
183}
184
185static mt_mock *start(const char *addr, mt_mock_world *w, bool owned) {
186 struct sockaddr_in sa;
187 if (!split_addr(addr, &sa)) return NULL;
188 int fd = socket(AF_INET, SOCK_STREAM, 0);
189 if (fd < 0) return NULL;
190 int one = 1;
191 setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &one, sizeof one);
192 socklen_t sl = sizeof sa;
193 if (bind(fd, (struct sockaddr *)&sa, sizeof sa) != 0 || listen(fd, 64) != 0 ||
194 getsockname(fd, (struct sockaddr *)&sa, &sl) != 0) {
195 close(fd);
196 return NULL;
197 }
198 int fl = fcntl(fd, F_GETFL);
199 if (fl >= 0) fcntl(fd, F_SETFL, fl | O_NONBLOCK);
200 mt_mock *m = calloc(1, sizeof *m);
201 if (!m) abort();
202 char ip[INET_ADDRSTRLEN];
203 inet_ntop(AF_INET, &sa.sin_addr, ip, sizeof ip);
204 snprintf(m->addr, sizeof m->addr, "%s:%u", ip, (unsigned)ntohs(sa.sin_port));
205 m->world = w;
206 m->owned = owned;
207 m->lfd = fd;
208 atomic_init(&m->stop, false);
209 pthread_mutex_init(&m->world_lock, NULL);
210 pthread_mutex_init(&m->log_lock, NULL);
211 pthread_mutex_init(&m->conns_lock, NULL);
212 if (pthread_create(&m->acceptor, NULL, accept_loop, m) != 0) {
213 close(fd);
214 free(m);
215 return NULL;
216 }
217 return m;
218}
219
220mt_mock *mt_mock_start(const char *addr, const char *host, uint32_t system) {
221 mt_mock_world *w = malloc(sizeof *w);
222 if (!w) abort();
223 mt_mock_world_init(w, host, system);
224 mt_mock *m = start(addr, w, true);
225 if (!m) {
226 mt_mock_world_free(w);
227 free(w);
228 }
229 return m;
230}
231
232mt_mock *mt_mock_start_with(const char *addr, mt_mock_world *w) { return start(addr, w, false); }
233
234const char *mt_mock_addr(const mt_mock *m) { return m->addr; }
235
236mt_mock_world *mt_mock_world_of(mt_mock *m) { return m->world; }
237
238void mt_mock_lock(mt_mock *m) { pthread_mutex_lock(&m->world_lock); }
239
240void mt_mock_unlock(mt_mock *m) { pthread_mutex_unlock(&m->world_lock); }
241
242bool mt_mock_saw(mt_mock *m, const char *req) {
243 pthread_mutex_lock(&m->log_lock);
244 bool saw = false;
245 for (size_t i = 0; i < m->nlog && !saw; i++) saw = strcmp(m->log[i], req) == 0;
246 pthread_mutex_unlock(&m->log_lock);
247 return saw;
248}
249
250void mt_mock_stop(mt_mock *m) {
251 if (!m) return;
252 atomic_store(&m->stop, true);
253 pthread_join(m->acceptor, NULL);
254 close(m->lfd);
255 pthread_mutex_lock(&m->conns_lock);
256 for (size_t i = 0; i < m->nconns; i++) shutdown(m->conns[i]->fd, SHUT_RDWR);
257 pthread_mutex_unlock(&m->conns_lock);
258 for (size_t i = 0; i < m->nconns; i++) {
259 pthread_join(m->conns[i]->th, NULL);
260 close(m->conns[i]->fd);
261 free(m->conns[i]);
262 }
263 free(m->conns);
264 for (size_t i = 0; i < m->nlog; i++) free(m->log[i]);
265 free(m->log);
266 if (m->owned) {
267 mt_mock_world_free(m->world);
268 free(m->world);
269 }
270 pthread_mutex_destroy(&m->world_lock);
271 pthread_mutex_destroy(&m->log_lock);
272 pthread_mutex_destroy(&m->conns_lock);
273 free(m);
274}