source: benchmark/io/http/worker.cfa

Last change on this file was 0f1336c, checked in by Thierry Delisle <tdelisle@…>, 3 months ago

Minor error handling improvement.

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