source: benchmark/io/http/main.cfa @ d11d6eb

#define __USE_GNU

#include <errno.h>
#include <stdio.h>
#include <string.h>
#include <unistd.h>
extern "C" {
        #include <sys/socket.h>
        #include <netinet/in.h>
}

#include <kernel.hfa>
#include <stats.hfa>
#include <time.hfa>
#include <thread.hfa>

#include "channel.hfa"
#include "filecache.hfa"
#include "options.hfa"
#include "worker.hfa"

extern void register_fixed_files( cluster &, int *, unsigned count );

Duration default_preemption() {
        return 0;
}

//=============================================================================================
// Globals
//=============================================================================================
channel & wait_connect;

struct ServerProc {
        processor self;
};

void ?{}( ServerProc & this ) {
        /* paranoid */ assert( options.clopts.instance != 0p );
        (this.self){ "Benchmark Processor", *options.clopts.instance };

        #if !defined(__CFA_NO_STATISTICS__)
                if( options.clopts.procstats ) {
                        print_stats_at_exit( this.self, options.clopts.instance->print_stats );
                }
                if( options.clopts.viewhalts ) {
                        print_halts( this.self );
                }
        #endif
}

//=============================================================================================
// Main
//============================================================================================='
int main( int argc, char * argv[] ) {
        //===================
        // Parse args
        const char * path = parse_options(argc, argv);

        //===================
        // Open Files
        printf("Filling cache from %s\n", path);
        fill_cache( path );

        //===================
        // Open Socket
        printf("Listening on port %d\n", options.socket.port);
        int server_fd = socket(AF_INET, SOCK_STREAM, 0);
        if(server_fd < 0) {
                abort( "socket error: (%d) %s\n", (int)errno, strerror(errno) );
        }

        int ret = 0;
        struct sockaddr_in address;
        int addrlen = sizeof(address);
        memset( (char *)&address, '\0' );
        address.sin_family = AF_INET;
        address.sin_addr.s_addr = htonl(INADDR_ANY);
        address.sin_port = htons( options.socket.port );

        ret = bind( server_fd, (struct sockaddr *)&address, sizeof(address) );
        if(ret < 0) {
                abort( "bind error: (%d) %s\n", (int)errno, strerror(errno) );
        }

        ret = listen( server_fd, options.socket.backlog );
        if(ret < 0) {
                abort( "listen error: (%d) %s\n", (int)errno, strerror(errno) );
        }

        //===================
        // Run Server Cluster
        {
                cluster cl = { "Server Cluster", options.clopts.params };
                #if !defined(__CFA_NO_STATISTICS__)
                        print_stats_at_exit( cl, CFA_STATS_READY_Q | CFA_STATS_IO );
                #endif
                options.clopts.instance = &cl;

                channel chan = { options.clopts.chan_size };
                &wait_connect = &chan;

                int pipe_cnt = options.clopts.nworkers * 2;
                int pipe_off;
                int * fds;
                [fds, pipe_off] = filefds( pipe_cnt );
                for(i; 0 ~ pipe_cnt ~ 2) {
                        int ret = pipe(&fds[pipe_off + i]);
                        if( ret < 0 ) { abort( "pipe error: (%d) %s\n", (int)errno, strerror(errno) ); }
                }

                if(options.file_cache.fixed_fds) {
                        register_fixed_files(cl, fds, pipe_off);
                }

                {
                        ServerProc procs[options.clopts.nprocs];
                        {
                                Worker workers[options.clopts.nworkers];
                                for(i; options.clopts.nworkers) {
                                        // if( options.file_cache.fixed_fds ) {
                                        //      workers[i].pipe[0] = pipe_off + (i * 2) + 0;
                                        //      workers[i].pipe[1] = pipe_off + (i * 2) + 1;
                                        // }
                                        // else
                                        {
                                                workers[i].pipe[0] = fds[pipe_off + (i * 2) + 0];
                                                workers[i].pipe[1] = fds[pipe_off + (i * 2) + 1];
                                        }
                                        unpark( workers[i] __cfaabi_dbg_ctx2 );
                                }
                                printf("%d workers started on %d processors\n", options.clopts.nworkers, options.clopts.nprocs);
                                {
                                        Acceptor acceptor = { server_fd, (struct sockaddr *)&address, (socklen_t*)&addrlen, 0 };

                                        char buffer[128];
                                        while(!feof(stdin)) {
                                                fgets(buffer, 128, stdin);
                                        }

                                        printf("Shutting Down\n");
                                }
                                printf("Acceptor Closed\n");

                                // Clean-up the workers
                                for(options.clopts.nworkers) {
                                        put( wait_connect, -1 );
                                }
                        }
                        printf("Workers Closed\n");
                }

                for(i; pipe_cnt) {
                        ret = close( fds[pipe_off + i] );
                        if(ret < 0) {
                                abort( "close pipe error: (%d) %s\n", (int)errno, strerror(errno) );
                        }
                }
                free(fds);
        }

        //===================
        // Close Socket
        printf("Closing Socket\n");
        ret = close( server_fd );
        if(ret < 0) {
                abort( "close socket error: (%d) %s\n", (int)errno, strerror(errno) );
        }

        //===================
        // Close Files
        printf("Closing Files\n");
        close_cache();
}