source: benchmark/io/http/main.cfa @ 153dc387

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

Main now print stats every 10 sec

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