source: benchmark/io/http/main.cfa @ b57db73

ADTarm-ehast-experimentalenumforall-pointer-decayjacob/cs343-translationnew-ast-unique-exprpthread-emulationqualifiedEnum
Last change on this file since b57db73 was b57db73, checked in by Thierry Delisle <tdelisle@…>, 3 years ago

httpforall now only loads file if explicit path is given.

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