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

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

httpforall now uses sout rather than printf

  • Property mode set to 100644
File size: 5.8 KB
RevLine 
[0aec496]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
[8c43d05]12#include <fstream.hfa>
[0aec496]13#include <kernel.hfa>
[153dc387]14#include <iofwd.hfa>
[0aec496]15#include <stats.hfa>
[d11d6eb]16#include <time.hfa>
[0aec496]17#include <thread.hfa>
18
19#include "filecache.hfa"
20#include "options.hfa"
21#include "worker.hfa"
22
[d11d6eb]23extern void register_fixed_files( cluster &, int *, unsigned count );
24
25Duration default_preemption() {
26        return 0;
27}
28
[0aec496]29//=============================================================================================
30// Globals
31//=============================================================================================
32struct ServerProc {
33        processor self;
34};
35
36void ?{}( ServerProc & this ) {
[2ecbd7b]37        /* paranoid */ assert( options.clopts.instance != 0p );
38        (this.self){ "Benchmark Processor", *options.clopts.instance };
[0aec496]39
40        #if !defined(__CFA_NO_STATISTICS__)
[2ecbd7b]41                if( options.clopts.procstats ) {
42                        print_stats_at_exit( this.self, options.clopts.instance->print_stats );
[0aec496]43                }
[2ecbd7b]44                if( options.clopts.viewhalts ) {
[0aec496]45                        print_halts( this.self );
46                }
47        #endif
48}
49
[c3ee5f3]50extern void init_protocol(void);
51extern void deinit_protocol(void);
52
[153dc387]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
[0aec496]76//=============================================================================================
77// Main
78//============================================================================================='
79int main( int argc, char * argv[] ) {
80        //===================
81        // Parse args
[2ecbd7b]82        const char * path = parse_options(argc, argv);
[0aec496]83
84        //===================
85        // Open Files
[8c43d05]86        sout | "Filling cache from" | path;
[481ee28]87        fill_cache( path );
[0aec496]88
89        //===================
90        // Open Socket
[8c43d05]91        sout | getpid() | ": Listening on port" | options.socket.port;
[0aec496]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
[7f389a5c]97        int ret = 0;
[0aec496]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);
[2ecbd7b]103        address.sin_port = htons( options.socket.port );
[0aec496]104
[ee913e0a]105        int waited = 0;
106        for() {
107                ret = bind( server_fd, (struct sockaddr *)&address, sizeof(address) );
108                if(ret < 0) {
[c3ee5f3]109                        if(errno == EADDRINUSE) {
[ee913e0a]110                                if(waited == 0) {
[8c43d05]111                                        sout | "Waiting for port";
[ee913e0a]112                                } else {
[8c43d05]113                                        sout | "\r" | waited | nonl;
114                                        flush( sout );
[ee913e0a]115                                }
116                                waited ++;
117                                sleep( 1`s );
118                                continue;
119                        }
120                        abort( "bind error: (%d) %s\n", (int)errno, strerror(errno) );
121                }
122                break;
[0aec496]123        }
124
[2ecbd7b]125        ret = listen( server_fd, options.socket.backlog );
[0aec496]126        if(ret < 0) {
127                abort( "listen error: (%d) %s\n", (int)errno, strerror(errno) );
128        }
129
130        //===================
131        // Run Server Cluster
132        {
[d11d6eb]133                cluster cl = { "Server Cluster", options.clopts.params };
[0aec496]134                #if !defined(__CFA_NO_STATISTICS__)
135                        print_stats_at_exit( cl, CFA_STATS_READY_Q | CFA_STATS_IO );
136                #endif
[2ecbd7b]137                options.clopts.instance = &cl;
[0aec496]138
[c3ee5f3]139
[d9c2284]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
[d11d6eb]149                if(options.file_cache.fixed_fds) {
150                        register_fixed_files(cl, fds, pipe_off);
151                }
152
[0aec496]153                {
[2ecbd7b]154                        ServerProc procs[options.clopts.nprocs];
[153dc387]155                        StatsPrinter printer;
[ece0e80]156
157                        init_protocol();
[0aec496]158                        {
[2ecbd7b]159                                Worker workers[options.clopts.nworkers];
[d9c2284]160                                for(i; options.clopts.nworkers) {
[d11d6eb]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                                        {
[d9c2284]167                                                workers[i].pipe[0] = fds[pipe_off + (i * 2) + 0];
168                                                workers[i].pipe[1] = fds[pipe_off + (i * 2) + 1];
[8e3034d]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;
[d9c2284]173                                        }
[e235429]174                                        unpark( workers[i] );
[d9c2284]175                                }
[8c43d05]176                                sout | options.clopts.nworkers | "workers started on" | options.clopts.nprocs | "processors";
[0aec496]177                                {
[e95a117]178                                        char buffer[128];
[7223dbf2]179                                        while(int ret = cfa_read(0, buffer, 128, 0, -1`s, 0p, 0p); ret != 0) {
[153dc387]180                                                if(ret < 0) abort( "main read error: (%d) %s\n", (int)errno, strerror(errno) );
[e95a117]181                                        }
182
[8c43d05]183                                        sout | "Shutdown received";
[0aec496]184                                }
[ece0e80]185
[8c43d05]186                                sout | "Notifying connections";
[ece0e80]187                                for(i; options.clopts.nworkers) {
[481ee28]188                                        workers[i].done = true;
[ece0e80]189                                        cancel(workers[i].cancel);
190                                }
191
[8c43d05]192                                sout | "Shutting down socket";
[ece0e80]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
[8c43d05]198                                sout | "Closing Socket";
[ece0e80]199                                ret = close( server_fd );
200                                if(ret < 0) {
201                                        abort( "close socket error: (%d) %s\n", (int)errno, strerror(errno) );
202                                }
[8c43d05]203                                sout | "Stopping connection threads..." | nonl;
[0aec496]204                        }
[8c43d05]205                        sout | "done";
[ece0e80]206
[8c43d05]207                        sout | "Stopping protocol threads..." | nonl;
[ece0e80]208                        deinit_protocol();
[8c43d05]209                        sout | "done";
210
211                        sout | "Stopping processors..." | nonl;
[0aec496]212                }
[8c43d05]213                sout | "done";
[d9c2284]214
[8c43d05]215                sout | "Closing splice fds..." | nonl;
[d9c2284]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);
[8c43d05]223                sout | "done";
[c3ee5f3]224
[8c43d05]225                sout | "Stopping processors..." | nonl;
[0aec496]226        }
[8c43d05]227        sout | "done";
[0aec496]228
229        //===================
230        // Close Files
[8c43d05]231        sout | "Closing open files..." | nonl;
[0aec496]232        close_cache();
[8c43d05]233        sout | "done";
[0aec496]234}
Note: See TracBrowser for help on using the repository browser.