source: benchmark/io/http/main.cfa @ 8c43d05

arm-ehjacob/cs343-translationnew-ast-unique-expr
Last change on this file since 8c43d05 was 8c43d05, checked in by Thierry Delisle <tdelisle@…>, 11 months ago

httpforall now uses sout rather than printf

  • Property mode set to 100644
File size: 5.8 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        const char * path = parse_options(argc, argv);
83
84        //===================
85        // Open Files
86        sout | "Filling cache from" | path;
87        fill_cache( path );
88
89        //===================
90        // Open Socket
91        sout | getpid() | ": Listening on port" | options.socket.port;
92        int server_fd = socket(AF_INET, SOCK_STREAM, 0);
93        if(server_fd < 0) {
94                abort( "socket error: (%d) %s\n", (int)errno, strerror(errno) );
95        }
96
97        int ret = 0;
98        struct sockaddr_in address;
99        int addrlen = sizeof(address);
100        memset( (char *)&address, '\0' );
101        address.sin_family = AF_INET;
102        address.sin_addr.s_addr = htonl(INADDR_ANY);
103        address.sin_port = htons( options.socket.port );
104
105        int waited = 0;
106        for() {
107                ret = bind( server_fd, (struct sockaddr *)&address, sizeof(address) );
108                if(ret < 0) {
109                        if(errno == EADDRINUSE) {
110                                if(waited == 0) {
111                                        sout | "Waiting for port";
112                                } else {
113                                        sout | "\r" | waited | nonl;
114                                        flush( sout );
115                                }
116                                waited ++;
117                                sleep( 1`s );
118                                continue;
119                        }
120                        abort( "bind error: (%d) %s\n", (int)errno, strerror(errno) );
121                }
122                break;
123        }
124
125        ret = listen( server_fd, options.socket.backlog );
126        if(ret < 0) {
127                abort( "listen error: (%d) %s\n", (int)errno, strerror(errno) );
128        }
129
130        //===================
131        // Run Server Cluster
132        {
133                cluster cl = { "Server Cluster", options.clopts.params };
134                #if !defined(__CFA_NO_STATISTICS__)
135                        print_stats_at_exit( cl, CFA_STATS_READY_Q | CFA_STATS_IO );
136                #endif
137                options.clopts.instance = &cl;
138
139
140                int pipe_cnt = options.clopts.nworkers * 2;
141                int pipe_off;
142                int * fds;
143                [fds, pipe_off] = filefds( pipe_cnt );
144                for(i; 0 ~ pipe_cnt ~ 2) {
145                        int ret = pipe(&fds[pipe_off + i]);
146                        if( ret < 0 ) { abort( "pipe error: (%d) %s\n", (int)errno, strerror(errno) ); }
147                }
148
149                if(options.file_cache.fixed_fds) {
150                        register_fixed_files(cl, fds, pipe_off);
151                }
152
153                {
154                        ServerProc procs[options.clopts.nprocs];
155                        StatsPrinter printer;
156
157                        init_protocol();
158                        {
159                                Worker workers[options.clopts.nworkers];
160                                for(i; options.clopts.nworkers) {
161                                        // if( options.file_cache.fixed_fds ) {
162                                        //      workers[i].pipe[0] = pipe_off + (i * 2) + 0;
163                                        //      workers[i].pipe[1] = pipe_off + (i * 2) + 1;
164                                        // }
165                                        // else
166                                        {
167                                                workers[i].pipe[0] = fds[pipe_off + (i * 2) + 0];
168                                                workers[i].pipe[1] = fds[pipe_off + (i * 2) + 1];
169                                                workers[i].sockfd  = server_fd;
170                                                workers[i].addr    = (struct sockaddr *)&address;
171                                                workers[i].addrlen = (socklen_t*)&addrlen;
172                                                workers[i].flags   = 0;
173                                        }
174                                        unpark( workers[i] );
175                                }
176                                sout | options.clopts.nworkers | "workers started on" | options.clopts.nprocs | "processors";
177                                {
178                                        char buffer[128];
179                                        while(int ret = cfa_read(0, buffer, 128, 0, -1`s, 0p, 0p); ret != 0) {
180                                                if(ret < 0) abort( "main read error: (%d) %s\n", (int)errno, strerror(errno) );
181                                        }
182
183                                        sout | "Shutdown received";
184                                }
185
186                                sout | "Notifying connections";
187                                for(i; options.clopts.nworkers) {
188                                        workers[i].done = true;
189                                        cancel(workers[i].cancel);
190                                }
191
192                                sout | "Shutting down socket";
193                                int ret = shutdown( server_fd, SHUT_RD );
194                                if( ret < 0 ) { abort( "shutdown error: (%d) %s\n", (int)errno, strerror(errno) ); }
195
196                                //===================
197                                // Close Socket
198                                sout | "Closing Socket";
199                                ret = close( server_fd );
200                                if(ret < 0) {
201                                        abort( "close socket error: (%d) %s\n", (int)errno, strerror(errno) );
202                                }
203                                sout | "Stopping connection threads..." | nonl;
204                        }
205                        sout | "done";
206
207                        sout | "Stopping protocol threads..." | nonl;
208                        deinit_protocol();
209                        sout | "done";
210
211                        sout | "Stopping processors..." | nonl;
212                }
213                sout | "done";
214
215                sout | "Closing splice fds..." | nonl;
216                for(i; pipe_cnt) {
217                        ret = close( fds[pipe_off + i] );
218                        if(ret < 0) {
219                                abort( "close pipe error: (%d) %s\n", (int)errno, strerror(errno) );
220                        }
221                }
222                free(fds);
223                sout | "done";
224
225                sout | "Stopping processors..." | nonl;
226        }
227        sout | "done";
228
229        //===================
230        // Close Files
231        sout | "Closing open files..." | nonl;
232        close_cache();
233        sout | "done";
234}
Note: See TracBrowser for help on using the repository browser.