source: benchmark/io/http/main.cfa @ 013b028

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

Started doing preliminary work to use Fixed FDs. Starting with the thread pipes.

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