source: benchmark/io/http/worker.cfa@ 80d16f8

ADT ast-experimental pthread-emulation
Last change on this file since 80d16f8 was 3f95dab, checked in by Thierry Delisle <tdelisle@…>, 3 years ago

Committing hopefully last version of the webserver
  • Property mode set to 100644
File size: 10.2 KB
RevLine 
[0aec496]1#include "worker.hfa"
2
3#include <errno.h>
4#include <stdio.h>
5#include <string.h>
[c82af9f]6#include <unistd.h>
[0aec496]7
[8c43d05]8#include <fstream.hfa>
[0aec496]9#include <iofwd.hfa>
[7f0ac12]10#include <mutex_stmt.hfa>
[0aec496]11
12#include "options.hfa"
13#include "protocol.hfa"
14#include "filecache.hfa"
15
[3f95dab]16static const unsigned long long period = 50_000_000;
17
[0aec496]18//=============================================================================================
[7f0ac12]19// Generic connection handling
[0aec496]20//=============================================================================================
[137974ae]21static void handle_connection( connection & this, volatile int & fd, char * buffer, size_t len, io_future_t * f, unsigned long long & last ) {
[7f0ac12]22 REQUEST:
23 for() {
24 bool closed;
25 HttpCode code;
26 const char * file;
27 size_t name_size;
28
29 // Read the http request
[10ba012]30 if( options.log ) mutex(sout) sout | "=== Reading request ===";
[7f0ac12]31 [code, closed, file, name_size] = http_read(fd, buffer, len, f);
32 f = 0p;
33
34 // if we are done, break out of the loop
35 if( closed ) break REQUEST;
36
37 // If this wasn't a request retrun 400
38 if( code != OK200 ) {
39 sout | "=== Invalid Request :" | code_val(code) | "===";
40 answer_error(fd, code);
41 continue REQUEST;
42 }
[ef3c383]43
[7f0ac12]44 if(0 == strncmp(file, "plaintext", min(name_size, sizeof("plaintext") ))) {
[10ba012]45 if( options.log ) mutex(sout) sout | "=== Request for /plaintext ===";
[7f0ac12]46
47 int ret = answer_plaintext(fd);
48 if( ret == -ECONNRESET ) break REQUEST;
49
[10ba012]50 if( options.log ) mutex(sout) sout | "=== Answer sent ===";
[7f0ac12]51 continue REQUEST;
52 }
53
54 if(0 == strncmp(file, "ping", min(name_size, sizeof("ping") ))) {
[10ba012]55 if( options.log ) mutex(sout) sout | "=== Request for /ping ===";
[7f0ac12]56
57 // Send the header
58 int ret = answer_empty(fd);
59 if( ret == -ECONNRESET ) break REQUEST;
60
[10ba012]61 if( options.log ) mutex(sout) sout | "=== Answer sent ===";
[7f0ac12]62 continue REQUEST;
63 }
64
65 if( options.log ) {
66 sout | "=== Request for file " | nonl;
67 write(sout, file, name_size);
68 sout | " ===";
69 }
70
71 if( !options.file_cache.path ) {
72 if( options.log ) {
73 sout | "=== File Not Found (" | nonl;
74 write(sout, file, name_size);
75 sout | ") ===";
76 }
77 answer_error(fd, E405);
78 continue REQUEST;
79 }
80
81 // Get the fd from the file cache
82 int ans_fd;
83 size_t count;
84 [ans_fd, count] = get_file( file, name_size );
85
86 // If we can't find the file, return 404
87 if( ans_fd < 0 ) {
88 if( options.log ) {
89 sout | "=== File Not Found (" | nonl;
90 write(sout, file, name_size);
91 sout | ") ===";
92 }
93 answer_error(fd, E404);
94 continue REQUEST;
95 }
96
97 // Send the desired file
98 int ret = answer_sendfile( this.pipe, fd, ans_fd, count, this.stats.sendfile );
[3f95dab]99 if(ret < 0) {
100 if( ret == -ECONNABORTED ) break REQUEST;
101 if( ret == -ECONNRESET ) break REQUEST;
102 if( ret == -EPIPE ) break REQUEST;
103 abort( "sendfile error: %d (%d) %s\n", ret, (int)errno, strerror(errno) );
104 }
[7f0ac12]105
[10ba012]106 if( options.log ) mutex(sout) sout | "=== Answer sent ===";
[ef3c383]107 }
[137974ae]108
109 if (stats_thrd) {
110 unsigned long long next = rdtscl();
[3f95dab]111 if(next > (last + period)) {
[10ba012]112 if(try_lock(stats_thrd->stats.lock __cfaabi_dbg_ctx2)) {
[137974ae]113 push(this.stats.sendfile, stats_thrd->stats.send);
114 unlock(stats_thrd->stats.lock);
115 last = next;
116 }
117 }
118 }
[481ee28]119}
120
[7f0ac12]121//=============================================================================================
122// Self Accepting Worker Thread
123//=============================================================================================
124void ?{}( AcceptWorker & this ) {
[329e26a]125 size_t cli = rand() % options.clopts.cltr_cnt;
126 ((thread&)this){ "Server Worker Thread", *options.clopts.instance[cli], 64000 };
127 options.clopts.thrd_cnt[cli]++;
[7f0ac12]128 this.done = false;
[0aec496]129}
130
[7f0ac12]131void main( AcceptWorker & this ) {
[e235429]132 park();
[137974ae]133 unsigned long long last = rdtscl();
[7f0ac12]134 /* paranoid */ assert( this.conn.pipe[0] != -1 );
135 /* paranoid */ assert( this.conn.pipe[1] != -1 );
[e95a117]136 for() {
[10ba012]137 if( options.log ) mutex(sout) sout | "=== Accepting connection ===";
[7f0ac12]138 int fd = cfa_accept4( this.sockfd, this.[addr, addrlen, flags], CFA_IO_LAZY );
[8e3034d]139 if(fd < 0) {
140 if( errno == ECONNABORTED ) break;
[ee59ede]141 if( this.done && (errno == EINVAL || errno == EBADF) ) break;
[8e3034d]142 abort( "accept error: (%d) %s\n", (int)errno, strerror(errno) );
143 }
[4f762d3]144 if(this.done) break;
[e95a117]145
[10ba012]146 if( options.log ) mutex(sout) sout | "=== New connection" | fd | "" | ", waiting for requests ===";
[7f0ac12]147 size_t len = options.socket.buflen;
148 char buffer[len];
[137974ae]149 handle_connection( this.conn, fd, buffer, len, 0p, last );
[3f95dab]150 this.conn.stats.sendfile.maxfd = max(this.conn.stats.sendfile.maxfd, fd);
151 this.conn.stats.sendfile.close++;
[0aec496]152
[10ba012]153 if( options.log ) mutex(sout) sout | "=== Connection closed ===";
[7f0ac12]154 }
155}
[0aec496]156
[561dd26]157
[7f0ac12]158//=============================================================================================
159// Channel Worker Thread
160//=============================================================================================
161void ?{}( ChannelWorker & this ) {
[329e26a]162 size_t cli = rand() % options.clopts.cltr_cnt;
163 ((thread&)this){ "Server Worker Thread", *options.clopts.instance[cli], 64000 };
164 options.clopts.thrd_cnt[cli]++;
[7f0ac12]165 this.done = false;
166}
[561dd26]167
[7f0ac12]168void main( ChannelWorker & this ) {
169 park();
[137974ae]170 unsigned long long last = rdtscl();
[7f0ac12]171 /* paranoid */ assert( this.conn.pipe[0] != -1 );
172 /* paranoid */ assert( this.conn.pipe[1] != -1 );
[3f95dab]173 this.conn.stats.sendfile.maxfd = max(this.conn.pipe[0], this.conn.pipe[1]);
[7f0ac12]174 for() {
175 size_t len = options.socket.buflen;
176 char buffer[len];
177 PendingRead p;
[10ba012]178 p.next = 0p;
[7f0ac12]179 p.in.buf = (void*)buffer;
180 p.in.len = len;
181 push(*this.queue, &p);
[561dd26]182
[10ba012]183 if( options.log ) mutex(sout) sout | "=== Waiting new connection ===";
[137974ae]184 handle_connection( this.conn, p.out.fd, buffer, len, &p.f, last );
[3f95dab]185 if(this.done) break;
186 this.conn.stats.sendfile.maxfd = max(this.conn.stats.sendfile.maxfd, p.out.fd);
187 this.conn.stats.sendfile.close++;
[561dd26]188
[10ba012]189 if( options.log ) mutex(sout) sout | "=== Connection closed ===";
[7f0ac12]190 }
[3f95dab]191
192 lock(stats_thrd->stats.lock __cfaabi_dbg_ctx2);
193 push(this.conn.stats.sendfile, stats_thrd->stats.send);
194 unlock(stats_thrd->stats.lock);
[7f0ac12]195}
[0aec496]196
[7f0ac12]197extern "C" {
198extern int accept4(int sockfd, struct sockaddr *addr, socklen_t *addrlen, int flags);
199}
[97748ee]200
[329e26a]201void ?{}( Acceptor & this, int cli ) {
202 ((thread&)this){ "Server Acceptor Thread", *options.clopts.instance[cli], 64000 };
203 options.clopts.thrd_cnt[cli]++;
[7f0ac12]204 this.done = false;
205}
[b57db73]206
[07997cd]207static inline __s32 get_res( io_future_t & this ) {
208 if( this.result < 0 ) {{
209 errno = -this.result;
210 return -1;
211 }}
212 return this.result;
213}
214
215static inline void push_connection( Acceptor & this, int fd ) {
[3f95dab]216 this.stats.accepts++;
[07997cd]217 PendingRead * p = 0p;
218 for() {
219 if(this.done) return;
220 p = pop(*this.queue);
221 if(p) break;
222 yield();
223 this.stats.creates++;
224 };
225
226 p->out.fd = fd;
227 async_recv(p->f, p->out.fd, p->in.buf, p->in.len, 0, CFA_IO_LAZY);
228}
229
230// #define ACCEPT_SPIN
[3f95dab]231#define ACCEPT_ONE
232// #define ACCEPT_MANY
[c25338d]233
[7f0ac12]234void main( Acceptor & this ) {
235 park();
[137974ae]236 unsigned long long last = rdtscl();
[c25338d]237
238#if defined(ACCEPT_SPIN)
[7f0ac12]239 if( options.log ) sout | "=== Accepting connection ===";
240 for() {
241 int fd = accept4(this.sockfd, this.[addr, addrlen, flags]);
242 if(fd < 0) {
243 if( errno == EWOULDBLOCK) {
[137974ae]244 this.stats.eagains++;
[7f0ac12]245 yield();
246 continue;
[97748ee]247 }
[7f0ac12]248 if( errno == ECONNABORTED ) break;
249 if( this.done && (errno == EINVAL || errno == EBADF) ) break;
250 abort( "accept error: (%d) %s\n", (int)errno, strerror(errno) );
251 }
[137974ae]252
[7f0ac12]253 if(this.done) return;
[97748ee]254
[7f0ac12]255 if( options.log ) sout | "=== New connection" | fd | "" | ", waiting for requests ===";
[97748ee]256
[07997cd]257 if(fd) push_connection(this, fd);
[ee59ede]258
[137974ae]259 if (stats_thrd) {
260 unsigned long long next = rdtscl();
[3f95dab]261 if(next > (last + period)) {
262 if(try_lock(stats_thrd->stats.lock)) {
263 push(this.stats, stats_thrd->stats.accpt);
264 unlock(stats_thrd->stats.lock);
265 last = next;
266 }
267 }
268 }
269
270 if( options.log ) sout | "=== Accepting connection ===";
271 }
272
273#elif defined(ACCEPT_ONE)
274 if( options.log ) sout | "=== Accepting connection ===";
275 for() {
276 int fd = cfa_accept4(this.sockfd, this.[addr, addrlen, flags], 0);
277 if(fd < 0) {
278 if( errno == ECONNABORTED ) break;
279 if( this.done && (errno == EINVAL || errno == EBADF) ) break;
280 abort( "accept error: (%d) %s\n", (int)errno, strerror(errno) );
281 }
282
283 if(this.done) return;
284
285 if( options.log ) sout | "=== New connection" | fd | "" | ", waiting for requests ===";
286
287 if(fd) push_connection(this, fd);
288
289 if (stats_thrd) {
290 unsigned long long next = rdtscl();
291 if(next > (last + period)) {
[137974ae]292 if(try_lock(stats_thrd->stats.lock)) {
293 push(this.stats, stats_thrd->stats.accpt);
294 unlock(stats_thrd->stats.lock);
295 last = next;
296 }
297 }
298 }
299
[7f0ac12]300 if( options.log ) sout | "=== Accepting connection ===";
[0aec496]301 }
[c25338d]302
[07997cd]303#elif defined(ACCEPT_MANY)
[c25338d]304 const int nacc = 10;
305 io_future_t results[nacc];
306
307 for(i; nacc) {
308 io_future_t & res = results[i];
309 reset(res);
310 /* paranoid */ assert(!available(res));
311 if( options.log ) mutex(sout) sout | "=== Re-arming accept no" | i | " ===";
312 async_accept4(res, this.sockfd, this.[addr, addrlen, flags], CFA_IO_LAZY);
313 }
314
315 for() {
316 if (stats_thrd) {
317 unsigned long long next = rdtscl();
[3f95dab]318 if(next > (last + period)) {
[c25338d]319 if(try_lock(stats_thrd->stats.lock __cfaabi_dbg_ctx2)) {
320 push(this.stats, stats_thrd->stats.accpt);
321 unlock(stats_thrd->stats.lock);
322 last = next;
323 }
324 }
325 }
326
327 for(i; nacc) {
328 io_future_t & res = results[i];
329 if(available(res)) {
330 if( options.log ) mutex(sout) sout | "=== Accept no " | i | "completed with result" | res.result | "===";
331 int fd = get_res(res);
332 reset(res);
333 if(fd < 0) {
334 if( errno == ECONNABORTED ) continue;
335 if( this.done && (errno == EINVAL || errno == EBADF) ) continue;
336 abort( "accept error: (%d) %s\n", (int)errno, strerror(errno) );
337 }
338 push_connection( this, fd );
339
340 /* paranoid */ assert(!available(res));
341 if( options.log ) mutex(sout) sout | "=== Re-arming accept no" | i | " ===";
342 async_accept4(res, this.sockfd, this.[addr, addrlen, flags], CFA_IO_LAZY);
343 }
344 }
345 if(this.done) return;
346
347 if( options.log ) mutex(sout) sout | "=== Waiting for any accept ===";
348 this.stats.eagains++;
349 wait_any(results, nacc);
350
351 if( options.log ) mutex(sout) {
352 sout | "=== Acceptor wake-up ===";
353 for(i; nacc) {
354 io_future_t & res = results[i];
355 sout | i | "available:" | available(res);
356 }
357 }
358
359 }
360
361 for(i; nacc) {
362 wait(results[i]);
363 }
364#else
365#error no accept algorithm specified
366#endif
[3f95dab]367 lock(stats_thrd->stats.lock);
368 push(this.stats, stats_thrd->stats.accpt);
369 unlock(stats_thrd->stats.lock);
[7f0ac12]370}
Note: See TracBrowser for help on using the repository browser.