#define __USE_GNU #include #include #include #include extern "C" { #include #include } #include #include #include #include #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(); }