source: benchmark/io/http/main.cfa @ 1e8b4b49

ADTast-experimentalenumforall-pointer-decaypthread-emulationqualifiedEnum
Last change on this file since 1e8b4b49 was 1e8b4b49, checked in by Thierry Delisle <tdelisle@…>, 2 years ago

Added work around for CFA bug with implicit conversion to unions.
  • Property mode set to 100644
File size: 7.2 KB
Line 
1#define _GNU_SOURCE
2
3#include <errno.h>
4#include <stdio.h>
5#include <string.h>
6#include <unistd.h>
7extern "C" {
8        #include <signal.h>
9        #include <sys/socket.h>
10        #include <netinet/in.h>
11}
12
13#include <fstream.hfa>
14#include <kernel.hfa>
15#include <iofwd.hfa>
16#include <stats.hfa>
17#include <time.hfa>
18#include <thread.hfa>
19
20#include "filecache.hfa"
21#include "options.hfa"
22#include "worker.hfa"
23
24extern void register_fixed_files( cluster &, int *, unsigned count );
25
26Duration default_preemption() {
27        return 0;
28}
29
30//=============================================================================================
31// Stats Printer
32//============================================================================================='
33
34thread StatsPrinter {};
35
36void ?{}( StatsPrinter & this, cluster & cl ) {
37        ((thread&)this){ "Stats Printer Thread", cl };
38}
39
40void ^?{}( StatsPrinter & mutex this ) {}
41
42void main(StatsPrinter & this) {
43        LOOP: for() {
44                waitfor( ^?{} : this) {
45                        break LOOP;
46                }
47                or else {}
48
49                sleep(10`s);
50
51                print_stats_now( *active_cluster(), CFA_STATS_READY_Q | CFA_STATS_IO );
52        }
53}
54
55//=============================================================================================
56// Globals
57//=============================================================================================
58struct ServerCluster {
59        cluster self;
60        processor    * procs;
61        // io_context   * ctxs;
62        StatsPrinter * prnt;
63
64};
65
66void ?{}( ServerCluster & this ) {
67        (this.self){ "Server Cluster", options.clopts.params };
68
69        this.procs = alloc(options.clopts.nprocs);
70        for(i; options.clopts.nprocs) {
71                (this.procs[i]){ "Benchmark Processor", this.self };
72
73                #if !defined(__CFA_NO_STATISTICS__)
74                        if( options.clopts.procstats ) {
75                                print_stats_at_exit( *this.procs, this.self.print_stats );
76                        }
77                        if( options.clopts.viewhalts ) {
78                                print_halts( *this.procs );
79                        }
80                #endif
81        }
82
83        if(options.stats) {
84                this.prnt = alloc();
85                (*this.prnt){ this.self };
86        } else {
87                this.prnt = 0p;
88        }
89
90        #if !defined(__CFA_NO_STATISTICS__)
91                print_stats_at_exit( this.self, CFA_STATS_READY_Q | CFA_STATS_IO );
92        #endif
93
94        options.clopts.instance[options.clopts.cltr_cnt] = &this.self;
95        options.clopts.cltr_cnt++;
96}
97
98void ^?{}( ServerCluster & this ) {
99        delete(this.prnt);
100
101        for(i; options.clopts.nprocs) {
102                ^(this.procs[i]){};
103        }
104        free(this.procs);
105
106        ^(this.self){};
107}
108
109extern void init_protocol(void);
110extern void deinit_protocol(void);
111
112//=============================================================================================
113// Main
114//============================================================================================='
115int main( int argc, char * argv[] ) {
116        __sighandler_t s = 1p;
117        signal(SIGPIPE, s);
118
119        //===================
120        // Parse args
121        parse_options(argc, argv);
122
123        //===================
124        // Open Files
125        if( options.file_cache.path ) {
126                sout | "Filling cache from" | options.file_cache.path;
127                fill_cache( options.file_cache.path );
128        }
129
130        //===================
131        // Open Socket
132        sout | getpid() | ": Listening on port" | options.socket.port;
133        int server_fd = socket(AF_INET, SOCK_STREAM, 0);
134        if(server_fd < 0) {
135                abort( "socket error: (%d) %s\n", (int)errno, strerror(errno) );
136        }
137
138        int ret = 0;
139        struct sockaddr_in address;
140        int addrlen = sizeof(address);
141        memset( (char *)&address, '\0' );
142        address.sin_family = AF_INET;
143        address.sin_addr.s_addr = htonl(INADDR_ANY);
144        address.sin_port = htons( options.socket.port );
145
146        int waited = 0;
147        for() {
148                int sockfd = server_fd;
149                __CONST_SOCKADDR_ARG addr;
150                addr.__sockaddr__ = (struct sockaddr *)&address;
151                socklen_t addrlen = sizeof(address);
152                ret = bind( sockfd, addr, addrlen );
153                if(ret < 0) {
154                        if(errno == EADDRINUSE) {
155                                if(waited == 0) {
156                                        if(!options.interactive) abort | "Port already in use in non-interactive mode. Aborting";
157                                        sout | "Waiting for port";
158                                } else {
159                                        sout | "\r" | waited | nonl;
160                                        flush( sout );
161                                }
162                                waited ++;
163                                sleep( 1`s );
164                                continue;
165                        }
166                        abort( "bind error: (%d) %s\n", (int)errno, strerror(errno) );
167                }
168                break;
169        }
170
171        ret = listen( server_fd, options.socket.backlog );
172        if(ret < 0) {
173                abort( "listen error: (%d) %s\n", (int)errno, strerror(errno) );
174        }
175
176        //===================
177        // Run Server Cluster
178        {
179                int pipe_cnt = options.clopts.nworkers * 2;
180                int pipe_off;
181                int * fds;
182                [fds, pipe_off] = filefds( pipe_cnt );
183                for(i; 0 ~ pipe_cnt ~ 2) {
184                        int ret = pipe(&fds[pipe_off + i]);
185                        if( ret < 0 ) { abort( "pipe error: (%d) %s\n", (int)errno, strerror(errno) ); }
186                }
187
188                // if(options.file_cache.path && options.file_cache.fixed_fds) {
189                //      register_fixed_files(cl, fds, pipe_off);
190                // }
191
192                {
193                        ServerCluster cl[options.clopts.nclusters];
194
195                        init_protocol();
196                        {
197                                Worker * workers = anew(options.clopts.nworkers);
198                                for(i; options.clopts.nworkers) {
199                                        // if( options.file_cache.fixed_fds ) {
200                                        //      workers[i].pipe[0] = pipe_off + (i * 2) + 0;
201                                        //      workers[i].pipe[1] = pipe_off + (i * 2) + 1;
202                                        // }
203                                        // else
204                                        {
205                                                workers[i].pipe[0] = fds[pipe_off + (i * 2) + 0];
206                                                workers[i].pipe[1] = fds[pipe_off + (i * 2) + 1];
207                                                workers[i].sockfd  = server_fd;
208                                                workers[i].addr    = (struct sockaddr *)&address;
209                                                workers[i].addrlen = (socklen_t*)&addrlen;
210                                                workers[i].flags   = 0;
211                                        }
212                                        unpark( workers[i] );
213                                }
214                                sout | options.clopts.nworkers | "workers started on" | options.clopts.nprocs | "processors /" | options.clopts.nclusters | "clusters";
215                                for(i; options.clopts.nclusters) {
216                                        sout | options.clopts.thrd_cnt[i] | nonl;
217                                }
218                                sout | nl;
219                                if(!options.interactive) park();
220                                {
221                                        char buffer[128];
222                                        for() {
223                                                int ret = cfa_read(0, buffer, 128, 0);
224                                                if(ret == 0) break;
225                                                if(ret < 0) abort( "main read error: (%d) %s\n", (int)errno, strerror(errno) );
226                                                sout | "User wrote '" | "" | nonl;
227                                                write(sout, buffer, ret - 1);
228                                                sout | "'";
229                                        }
230
231                                        sout | "Shutdown received";
232                                }
233
234                                sout | "Notifying connections..." | nonl; flush( sout );
235                                for(i; options.clopts.nworkers) {
236                                        workers[i].done = true;
237                                }
238                                sout | "done";
239
240                                sout | "Shutting down socket..." | nonl; flush( sout );
241                                int ret = shutdown( server_fd, SHUT_RD );
242                                if( ret < 0 ) {
243                                        abort( "shutdown error: (%d) %s\n", (int)errno, strerror(errno) );
244                                }
245                                sout | "done";
246
247                                //===================
248                                // Close Socket
249                                sout | "Closing Socket..." | nonl; flush( sout );
250                                ret = close( server_fd );
251                                if(ret < 0) {
252                                        abort( "close socket error: (%d) %s\n", (int)errno, strerror(errno) );
253                                }
254                                sout | "done";
255
256                                sout | "Stopping connection threads..." | nonl; flush( sout );
257                                adelete(workers);
258                        }
259                        sout | "done";
260
261                        sout | "Stopping protocol threads..." | nonl; flush( sout );
262                        deinit_protocol();
263                        sout | "done";
264
265                        sout | "Stopping processors/clusters..." | nonl; flush( sout );
266                }
267                sout | "done";
268
269                sout | "Closing splice fds..." | nonl; flush( sout );
270                for(i; pipe_cnt) {
271                        ret = close( fds[pipe_off + i] );
272                        if(ret < 0) {
273                                abort( "close pipe error: (%d) %s\n", (int)errno, strerror(errno) );
274                        }
275                }
276                free(fds);
277                sout | "done";
278
279                sout | "Stopping processors..." | nonl; flush( sout );
280        }
281        sout | "done";
282
283        //===================
284        // Close Files
285        if( options.file_cache.path ) {
286                sout | "Closing open files..." | nonl; flush( sout );
287                close_cache();
288                sout | "done";
289        }
290}
Note: See TracBrowser for help on using the repository browser.