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

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

Every worker now does it's own accept.

  • Property mode set to 100644
File size: 4.3 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 <stats.hfa>
14#include <time.hfa>
15#include <thread.hfa>
16
17#include "channel.hfa"
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//=============================================================================================
31channel & wait_connect;
32
33struct ServerProc {
34        processor self;
35};
36
37void ?{}( ServerProc & this ) {
38        /* paranoid */ assert( options.clopts.instance != 0p );
39        (this.self){ "Benchmark Processor", *options.clopts.instance };
40
41        #if !defined(__CFA_NO_STATISTICS__)
42                if( options.clopts.procstats ) {
43                        print_stats_at_exit( this.self, options.clopts.instance->print_stats );
44                }
45                if( options.clopts.viewhalts ) {
46                        print_halts( this.self );
47                }
48        #endif
49}
50
51//=============================================================================================
52// Main
53//============================================================================================='
54int main( int argc, char * argv[] ) {
55        //===================
56        // Parse args
57        const char * path = parse_options(argc, argv);
58
59        //===================
60        // Open Files
61        printf("Filling cache from %s\n", path);
62        fill_cache( path );
63
64        //===================
65        // Open Socket
66        printf("Listening on port %d\n", options.socket.port);
67        int server_fd = socket(AF_INET, SOCK_STREAM, 0);
68        if(server_fd < 0) {
69                abort( "socket error: (%d) %s\n", (int)errno, strerror(errno) );
70        }
71
72        int ret = 0;
73        struct sockaddr_in address;
74        int addrlen = sizeof(address);
75        memset( (char *)&address, '\0' );
76        address.sin_family = AF_INET;
77        address.sin_addr.s_addr = htonl(INADDR_ANY);
78        address.sin_port = htons( options.socket.port );
79
80        ret = bind( server_fd, (struct sockaddr *)&address, sizeof(address) );
81        if(ret < 0) {
82                abort( "bind error: (%d) %s\n", (int)errno, strerror(errno) );
83        }
84
85        ret = listen( server_fd, options.socket.backlog );
86        if(ret < 0) {
87                abort( "listen error: (%d) %s\n", (int)errno, strerror(errno) );
88        }
89
90        //===================
91        // Run Server Cluster
92        {
93                cluster cl = { "Server Cluster", options.clopts.params };
94                #if !defined(__CFA_NO_STATISTICS__)
95                        print_stats_at_exit( cl, CFA_STATS_READY_Q | CFA_STATS_IO );
96                #endif
97                options.clopts.instance = &cl;
98
99                channel chan = { options.clopts.chan_size };
100                &wait_connect = &chan;
101
102                int pipe_cnt = options.clopts.nworkers * 2;
103                int pipe_off;
104                int * fds;
105                [fds, pipe_off] = filefds( pipe_cnt );
106                for(i; 0 ~ pipe_cnt ~ 2) {
107                        int ret = pipe(&fds[pipe_off + i]);
108                        if( ret < 0 ) { abort( "pipe error: (%d) %s\n", (int)errno, strerror(errno) ); }
109                }
110
111                if(options.file_cache.fixed_fds) {
112                        register_fixed_files(cl, fds, pipe_off);
113                }
114
115                {
116                        ServerProc procs[options.clopts.nprocs];
117                        {
118                                Worker workers[options.clopts.nworkers];
119                                for(i; options.clopts.nworkers) {
120                                        // if( options.file_cache.fixed_fds ) {
121                                        //      workers[i].pipe[0] = pipe_off + (i * 2) + 0;
122                                        //      workers[i].pipe[1] = pipe_off + (i * 2) + 1;
123                                        // }
124                                        // else
125                                        {
126                                                workers[i].pipe[0] = fds[pipe_off + (i * 2) + 0];
127                                                workers[i].pipe[1] = fds[pipe_off + (i * 2) + 1];
128                                                workers[i].sockfd  = server_fd;
129                                                workers[i].addr    = (struct sockaddr *)&address;
130                                                workers[i].addrlen = (socklen_t*)&addrlen;
131                                                workers[i].flags   = 0;
132                                        }
133                                        unpark( workers[i] __cfaabi_dbg_ctx2 );
134                                }
135                                printf("%d workers started on %d processors\n", options.clopts.nworkers, options.clopts.nprocs);
136                                {
137                                        char buffer[128];
138                                        while(!feof(stdin)) {
139                                                fgets(buffer, 128, stdin);
140                                        }
141
142                                        printf("Shutting Down\n");
143                                }
144                        }
145                        printf("Workers Closed\n");
146                }
147
148                for(i; pipe_cnt) {
149                        ret = close( fds[pipe_off + i] );
150                        if(ret < 0) {
151                                abort( "close pipe error: (%d) %s\n", (int)errno, strerror(errno) );
152                        }
153                }
154                free(fds);
155        }
156
157        //===================
158        // Close Socket
159        printf("Closing Socket\n");
160        ret = close( server_fd );
161        if(ret < 0) {
162                abort( "close socket error: (%d) %s\n", (int)errno, strerror(errno) );
163        }
164
165        //===================
166        // Close Files
167        printf("Closing Files\n");
168        close_cache();
169}
Note: See TracBrowser for help on using the repository browser.