source: benchmark/io/http/main.cfa @ 86c12d65

ADTast-experimentalpthread-emulationqualifiedEnum
Last change on this file since 86c12d65 was 86c12d65, checked in by Thierry Delisle <tdelisle@…>, 2 years ago

Checkpoint of the broken version of reuseport
  • Property mode set to 100644
File size: 11.0 KB
Line 
1#define _GNU_SOURCE
2
3#include <errno.h>
4#include <signal.h>
5#include <stdio.h>
6#include <string.h>
7#include <unistd.h>
8extern "C" {
9        #include <sched.h>
10        #include <signal.h>
11        #include <sys/eventfd.h>
12        #include <sys/socket.h>
13        #include <netinet/in.h>
14}
15
16#include <fstream.hfa>
17#include <kernel.hfa>
18#include <locks.hfa>
19#include <iofwd.hfa>
20#include <stats.hfa>
21#include <time.hfa>
22#include <thread.hfa>
23
24#include "filecache.hfa"
25#include "options.hfa"
26#include "socket.hfa"
27#include "worker.hfa"
28
29extern void register_fixed_files( cluster &, int *, unsigned count );
30
31Duration default_preemption() {
32        return 0;
33}
34
35//=============================================================================================
36// Stats Printer
37//============================================================================================='
38
39thread StatsPrinter {
40        Worker * workers;
41        int worker_cnt;
42        condition_variable(fast_block_lock) var;
43};
44
45void ?{}( StatsPrinter & this, cluster & cl ) {
46        ((thread&)this){ "Stats Printer Thread", cl };
47        this.worker_cnt = 0;
48}
49
50void ^?{}( StatsPrinter & mutex this ) {}
51
52#define eng3(X) (ws(3, 3, unit(eng( X ))))
53
54void main(StatsPrinter & this) {
55        LOOP: for() {
56                waitfor( ^?{} : this) {
57                        break LOOP;
58                }
59                or else {}
60
61                wait(this.var, 10`s);
62
63                print_stats_now( *active_cluster(), CFA_STATS_READY_Q | CFA_STATS_IO );
64                if(this.worker_cnt != 0) {
65                        uint64_t tries = 0;
66                        uint64_t calls = 0;
67                        uint64_t header = 0;
68                        uint64_t splcin = 0;
69                        uint64_t splcot = 0;
70                        struct {
71                                volatile uint64_t calls;
72                                volatile uint64_t bytes;
73                        } avgrd[zipf_cnts];
74                        memset(avgrd, 0, sizeof(avgrd));
75
76                        for(i; this.worker_cnt) {
77                                tries += this.workers[i].stats.sendfile.tries;
78                                calls += this.workers[i].stats.sendfile.calls;
79                                header += this.workers[i].stats.sendfile.header;
80                                splcin += this.workers[i].stats.sendfile.splcin;
81                                splcot += this.workers[i].stats.sendfile.splcot;
82                                for(j; zipf_cnts) {
83                                        avgrd[j].calls += this.workers[i].stats.sendfile.avgrd[j].calls;
84                                        avgrd[j].bytes += this.workers[i].stats.sendfile.avgrd[j].bytes;
85                                }
86                        }
87
88                        double ratio = ((double)tries) / calls;
89
90                        sout | "----- Worker Stats -----";
91                        sout | "sendfile  : " | calls | "calls," | tries | "tries (" | ratio | " try/call)";
92                        sout | "            " | header | "header," | splcin | "splice in," | splcot | "splice out";
93                        sout | " - zipf sizes:";
94                        for(i; zipf_cnts) {
95                                double written = avgrd[i].calls > 0 ? ((double)avgrd[i].bytes) / avgrd[i].calls : 0;
96                                sout | "        " | zipf_sizes[i] | "bytes," | avgrd[i].calls | "shorts," | written | "written";
97                        }
98                }
99                else {
100                        sout | "No Workers!";
101                }
102        }
103}
104
105//=============================================================================================
106// Globals
107//=============================================================================================
108struct ServerCluster {
109        cluster self;
110        processor    * procs;
111        // io_context   * ctxs;
112        StatsPrinter * prnt;
113
114};
115
116void ?{}( ServerCluster & this ) {
117        (this.self){ "Server Cluster", options.clopts.params };
118
119        cpu_set_t fullset;
120        CPU_ZERO(&fullset);
121        int ret = sched_getaffinity(getpid(), sizeof(fullset), &fullset);
122        if( ret != 0 ) abort | "sched_getaffinity failed with" | errno | strerror( errno );
123        int cnt = CPU_COUNT(&fullset);
124
125        this.procs = alloc(options.clopts.nprocs);
126        for(i; options.clopts.nprocs) {
127                (this.procs[i]){ "Benchmark Processor", this.self };
128
129                int c = 0;
130                int n = 1 + (i % cnt);
131                for(int j = 0; j < CPU_SETSIZE; j++) {
132                        if(CPU_ISSET(j, &fullset)) n--;
133                        if(n == 0) {
134                                c = j;
135                                break;
136                        }
137                }
138                cpu_set_t localset;
139                CPU_ZERO(&localset);
140                CPU_SET(c, &localset);
141                ret = pthread_setaffinity_np(this.procs[i].kernel_thread, sizeof(localset), &localset);
142                if( ret != 0 ) abort | "sched_getaffinity failed with" | ret | strerror( ret );
143
144                #if !defined(__CFA_NO_STATISTICS__)
145                        if( options.clopts.procstats ) {
146                                print_stats_at_exit( *this.procs, this.self.print_stats );
147                        }
148                        if( options.clopts.viewhalts ) {
149                                print_halts( *this.procs );
150                        }
151                #endif
152        }
153
154        if(options.stats) {
155                this.prnt = alloc();
156                (*this.prnt){ this.self };
157        } else {
158                this.prnt = 0p;
159        }
160
161        #if !defined(__CFA_NO_STATISTICS__)
162                print_stats_at_exit( this.self, CFA_STATS_READY_Q | CFA_STATS_IO );
163        #endif
164
165        options.clopts.instance[options.clopts.cltr_cnt] = &this.self;
166        options.clopts.cltr_cnt++;
167}
168
169void ^?{}( ServerCluster & this ) {
170        delete(this.prnt);
171
172        for(i; options.clopts.nprocs) {
173                ^(this.procs[i]){};
174        }
175        free(this.procs);
176
177        ^(this.self){};
178}
179
180extern void init_protocol(void);
181extern void deinit_protocol(void);
182
183//=============================================================================================
184// Termination
185//=============================================================================================
186
187int closefd;
188void cleanstop(int) {
189        eventfd_t buffer = 1;
190        char * buffer_s = (char*)&buffer;
191        int ret = write(closefd, buffer_s, sizeof(buffer));
192        if(ret < 0) abort( "eventfd write error: (%d) %s\n", (int)errno, strerror(errno) );
193        return;
194}
195
196//=============================================================================================
197// Main
198//============================================================================================='
199int main( int argc, char * argv[] ) {
200        int ret;
201        __sighandler_t s = 1p;
202        signal(SIGPIPE, s);
203
204        //===================
205        // Parse args
206        parse_options(argc, argv);
207
208        //===================
209        // Setup non-interactive termination
210        if(!options.interactive) {
211                closefd = eventfd(0, 0);
212                if(closefd < 0) abort( "eventfd error: (%d) %s\n", (int)errno, strerror(errno) );
213
214                sighandler_t prev = signal(SIGTERM, cleanstop);
215                intptr_t prev_workaround = (intptr_t) prev;
216                // can't use SIG_ERR it crashes the compiler
217                if(prev_workaround == -1) abort( "signal setup error: (%d) %s\n", (int)errno, strerror(errno) );
218
219                sout | "Signal termination ready";
220        }
221
222        //===================
223        // Open Files
224        if( options.file_cache.path ) {
225                sout | "Filling cache from" | options.file_cache.path;
226                fill_cache( options.file_cache.path );
227        }
228
229        //===================
230        // Open Socket
231        sout | getpid() | ": Listening on port" | options.socket.port;
232
233        struct sockaddr_in address;
234        int addrlen = prepaddr(address);
235
236        int server_fd;
237        if(!options.socket.manyreuse) {
238                server_fd = listener(address, addrlen);
239        }
240
241        //===================
242        // Run Server Cluster
243        {
244                int pipe_cnt = options.clopts.nworkers * 2;
245                int pipe_off;
246                int * fds;
247                [fds, pipe_off] = filefds( pipe_cnt );
248                for(i; 0 ~ pipe_cnt ~ 2) {
249                        int ret = pipe(&fds[pipe_off + i]);
250                        if( ret < 0 ) { abort( "pipe error: (%d) %s\n", (int)errno, strerror(errno) ); }
251                }
252
253                // if(options.file_cache.path && options.file_cache.fixed_fds) {
254                //      register_fixed_files(cl, fds, pipe_off);
255                // }
256
257                {
258                        // Stats printer makes a copy so this needs to persist longer than normal
259                        Worker * workers;
260                        ServerCluster cl[options.clopts.nclusters];
261
262                        init_protocol();
263                        {
264                                workers = anew(options.clopts.nworkers);
265                                cl[0].prnt->workers = workers;
266                                cl[0].prnt->worker_cnt = options.clopts.nworkers;
267                                for(i; options.clopts.nworkers) {
268                                        // if( options.file_cache.fixed_fds ) {
269                                        //      workers[i].pipe[0] = pipe_off + (i * 2) + 0;
270                                        //      workers[i].pipe[1] = pipe_off + (i * 2) + 1;
271                                        // }
272                                        // else
273                                        {
274                                                workers[i].pipe[0] = fds[pipe_off + (i * 2) + 0];
275                                                workers[i].pipe[1] = fds[pipe_off + (i * 2) + 1];
276                                                workers[i].sockfd  = options.socket.manyreuse ?  listener(address, addrlen) : server_fd;
277                                                workers[i].addr    = (struct sockaddr *)&address;
278                                                workers[i].addrlen = (socklen_t*)&addrlen;
279                                                workers[i].flags   = 0;
280                                        }
281                                        unpark( workers[i] );
282                                }
283                                sout | options.clopts.nworkers | "workers started on" | options.clopts.nprocs | "processors /" | options.clopts.nclusters | "clusters";
284                                for(i; options.clopts.nclusters) {
285                                        sout | options.clopts.thrd_cnt[i] | nonl;
286                                }
287                                sout | nl;
288                                {
289                                        if(options.interactive) {
290                                                char buffer[128];
291                                                for() {
292                                                        int ret = cfa_read(0, buffer, 128, 0);
293                                                        if(ret == 0) break;
294                                                        if(ret < 0) abort( "main read error: (%d) %s\n", (int)errno, strerror(errno) );
295                                                        sout | "User wrote '" | "" | nonl;
296                                                        write(sout, buffer, ret - 1);
297                                                        sout | "'";
298                                                }
299                                        }
300                                        else {
301                                                char buffer[sizeof(eventfd_t)];
302                                                int ret = cfa_read(closefd, buffer, sizeof(eventfd_t), 0);
303                                                if(ret < 0) abort( "main read error: (%d) %s\n", (int)errno, strerror(errno) );
304                                        }
305
306                                        sout | "Shutdown received";
307                                }
308
309                                sout | "Notifying connections..." | nonl; flush( sout );
310                                for(i; options.clopts.nworkers) {
311                                        workers[i].done = true;
312                                }
313                                sout | "done";
314
315                                sout | "Shutting down socket..." | nonl; flush( sout );
316                                if(options.socket.manyreuse) {
317                                        for(i; options.clopts.nworkers) {
318                                                ret = shutdown( workers[i].sockfd, SHUT_RD );
319                                                if(ret < 0) abort( "close socket %d error: (%d) %s\n", i, (int)errno, strerror(errno) );
320                                        }
321                                }
322                                else {
323                                        ret = shutdown( server_fd, SHUT_RD );
324                                        if( ret < 0 ) {
325                                                abort( "shutdown error: (%d) %s\n", (int)errno, strerror(errno) );
326                                        }
327                                }
328                                sout | "done";
329
330                                //===================
331                                // Close Socket
332                                sout | "Closing Socket..." | nonl; flush( sout );
333                                if(options.socket.manyreuse) {
334                                        for(i; options.clopts.nworkers) {
335                                                ret = close(workers[i].sockfd);
336                                                if(ret < 0) abort( "close socket %d error: (%d) %s\n", i, (int)errno, strerror(errno) );
337                                        }
338                                }
339                                else {
340                                        ret = close( server_fd );
341                                        if(ret < 0) {
342                                                abort( "close socket error: (%d) %s\n", (int)errno, strerror(errno) );
343                                        }
344                                }
345                                sout | "done";
346
347                                sout | "Stopping connection threads..." | nonl; flush( sout );
348                                for(i; options.clopts.nworkers) {
349                                        for(j; 2) {
350                                                ret = close(workers[i].pipe[j]);
351                                                if(ret < 0) abort( "close pipe %d error: (%d) %s\n", j, (int)errno, strerror(errno) );
352                                        }
353                                        join(workers[i]);
354                                }
355                        }
356                        sout | "done";
357
358                        sout | "Stopping protocol threads..." | nonl; flush( sout );
359                        deinit_protocol();
360                        sout | "done";
361
362                        sout | "Stopping printer threads..." | nonl; flush( sout );
363                        for(i; options.clopts.nclusters) {
364                                StatsPrinter * p = cl[i].prnt;
365                                if(p) {
366                                        notify_one(p->var);
367                                        join(*p);
368                                }
369                        }
370                        sout | "done";
371
372                        // Now that the stats printer is stopped, we can reclaim this
373                        adelete(workers);
374
375                        sout | "Stopping processors/clusters..." | nonl; flush( sout );
376                }
377                sout | "done";
378
379                // sout | "Closing splice fds..." | nonl; flush( sout );
380                // for(i; pipe_cnt) {
381                //      ret = close( fds[pipe_off + i] );
382                //      if(ret < 0) {
383                //              abort( "close pipe error: (%d) %s\n", (int)errno, strerror(errno) );
384                //      }
385                // }
386                free(fds);
387                sout | "done";
388
389                sout | "Stopping processors..." | nonl; flush( sout );
390        }
391        sout | "done";
392
393        //===================
394        // Close Files
395        if( options.file_cache.path ) {
396                sout | "Closing open files..." | nonl; flush( sout );
397                close_cache();
398                sout | "done";
399        }
400}
401
402const size_t zipf_sizes[] = { 102, 204, 307, 409, 512, 614, 716, 819, 921, 1024, 2048, 3072, 4096, 5120, 6144, 7168, 8192, 9216, 10240, 20480, 30720, 40960, 51200, 61440, 71680, 81920, 92160, 102400, 204800, 307200, 409600, 512000, 614400, 716800, 819200, 921600 };
403static_assert(zipf_cnts == sizeof(zipf_sizes) / sizeof(zipf_sizes[0]));
Note: See TracBrowser for help on using the repository browser.