Context Navigation

main.cfa@ 3e417bf

Visit:

ADT ast-experimental enum forall-pointer-decay pthread-emulation qualifiedEnum

Last change on this file since 3e417bf was 3a0ddb6, checked in by Thierry Delisle <tdelisle@…>, 4 years ago
Don't wait for the port if not in interactive mode.
Property mode set to `100644`
File size: 7.0 KB

Rev	Line
[0aec496]	1	#define __USE_GNU
	2
	3	#include <errno.h>
	4	#include <stdio.h>
	5	#include <string.h>
	6	#include <unistd.h>
	7	extern "C" {
[c2df3031]	8	#include <signal.h>
[0aec496]	9	#include <sys/socket.h>
	10	#include <netinet/in.h>
	11	}
	12
[8c43d05]	13	#include <fstream.hfa>
[0aec496]	14	#include <kernel.hfa>
[153dc387]	15	#include <iofwd.hfa>
[0aec496]	16	#include <stats.hfa>
[d11d6eb]	17	#include <time.hfa>
[0aec496]	18	#include <thread.hfa>
	19
	20	#include "filecache.hfa"
	21	#include "options.hfa"
	22	#include "worker.hfa"
	23
[d11d6eb]	24	extern void register_fixed_files( cluster &, int *, unsigned count );
	25
	26	Duration default_preemption() {
	27	return 0;
	28	}
	29
[153dc387]	30	//=============================================================================================
	31	// Stats Printer
	32	//============================================================================================='
	33
	34	thread StatsPrinter {};
	35
[348f81d5]	36	void ?{}( StatsPrinter & this, cluster & cl ) {
	37	((thread&)this){ "Stats Printer Thread", cl };
[153dc387]	38	}
	39
[2cd784a]	40	void ^?{}( StatsPrinter & mutex this ) {}
	41
[153dc387]	42	void main(StatsPrinter & this) {
	43	LOOP: for() {
	44	waitfor( ^?{} : this) {
	45	break LOOP;
	46	}
	47	or else {}
	48
	49	sleep(10`s);
	50
[348f81d5]	51	print_stats_now( *active_cluster(), CFA_STATS_READY_Q \| CFA_STATS_IO );
	52	}
	53	}
	54
	55	//=============================================================================================
	56	// Globals
	57	//=============================================================================================
	58	struct ServerCluster {
	59	cluster self;
	60	processor * procs;
[2cd784a]	61	// io_context * ctxs;
[348f81d5]	62	StatsPrinter * prnt;
	63
	64	};
	65
	66	void ?{}( ServerCluster & this ) {
	67	(this.self){ "Server Cluster", options.clopts.params };
	68
	69	this.procs = alloc(options.clopts.nprocs);
	70	for(i; options.clopts.nprocs) {
	71	(this.procs[i]){ "Benchmark Processor", this.self };
	72
	73	#if !defined(__CFA_NO_STATISTICS__)
	74	if( options.clopts.procstats ) {
	75	print_stats_at_exit( *this.procs, this.self.print_stats );
	76	}
	77	if( options.clopts.viewhalts ) {
	78	print_halts( *this.procs );
	79	}
	80	#endif
	81	}
	82
[2cd784a]	83	if(options.stats) {
	84	this.prnt = alloc();
	85	(*this.prnt){ this.self };
	86	} else {
	87	this.prnt = 0p;
[348f81d5]	88	}
	89
	90	#if !defined(__CFA_NO_STATISTICS__)
	91	print_stats_at_exit( this.self, CFA_STATS_READY_Q \| CFA_STATS_IO );
	92	#endif
	93
	94	options.clopts.instance[options.clopts.cltr_cnt] = &this.self;
	95	options.clopts.cltr_cnt++;
[153dc387]	96	}
	97
[348f81d5]	98	void ^?{}( ServerCluster & this ) {
[2cd784a]	99	delete(this.prnt);
[348f81d5]	100
	101	for(i; options.clopts.nprocs) {
	102	^(this.procs[i]){};
	103	}
	104	free(this.procs);
	105
	106	^(this.self){};
	107	}
	108
	109	extern void init_protocol(void);
	110	extern void deinit_protocol(void);
	111
[0aec496]	112	//=============================================================================================
	113	// Main
	114	//============================================================================================='
	115	int main( int argc, char * argv[] ) {
[c2df3031]	116	__sighandler_t s = 1p;
	117	signal(SIGPIPE, s);
	118
[0aec496]	119	//===================
	120	// Parse args
[b57db73]	121	parse_options(argc, argv);
[0aec496]	122
	123	//===================
	124	// Open Files
[b57db73]	125	if( options.file_cache.path ) {
	126	sout \| "Filling cache from" \| options.file_cache.path;
	127	fill_cache( options.file_cache.path );
	128	}
[0aec496]	129
	130	//===================
	131	// Open Socket
[8c43d05]	132	sout \| getpid() \| ": Listening on port" \| options.socket.port;
[0aec496]	133	int server_fd = socket(AF_INET, SOCK_STREAM, 0);
	134	if(server_fd < 0) {
	135	abort( "socket error: (%d) %s\n", (int)errno, strerror(errno) );
	136	}
	137
[7f389a5c]	138	int ret = 0;
[0aec496]	139	struct sockaddr_in address;
	140	int addrlen = sizeof(address);
	141	memset( (char *)&address, '\0' );
	142	address.sin_family = AF_INET;
	143	address.sin_addr.s_addr = htonl(INADDR_ANY);
[2ecbd7b]	144	address.sin_port = htons( options.socket.port );
[0aec496]	145
[ee913e0a]	146	int waited = 0;
	147	for() {
	148	ret = bind( server_fd, (struct sockaddr *)&address, sizeof(address) );
	149	if(ret < 0) {
[c3ee5f3]	150	if(errno == EADDRINUSE) {
[ee913e0a]	151	if(waited == 0) {
[3a0ddb6]	152	if(!options.interactive) abort \| "Port already in use in non-interactive mode. Aborting";
[8c43d05]	153	sout \| "Waiting for port";
[ee913e0a]	154	} else {
[8c43d05]	155	sout \| "\r" \| waited \| nonl;
	156	flush( sout );
[ee913e0a]	157	}
	158	waited ++;
	159	sleep( 1`s );
	160	continue;
	161	}
	162	abort( "bind error: (%d) %s\n", (int)errno, strerror(errno) );
	163	}
	164	break;
[0aec496]	165	}
	166
[2ecbd7b]	167	ret = listen( server_fd, options.socket.backlog );
[0aec496]	168	if(ret < 0) {
	169	abort( "listen error: (%d) %s\n", (int)errno, strerror(errno) );
	170	}
	171
	172	//===================
	173	// Run Server Cluster
	174	{
[d9c2284]	175	int pipe_cnt = options.clopts.nworkers * 2;
	176	int pipe_off;
	177	int * fds;
	178	[fds, pipe_off] = filefds( pipe_cnt );
	179	for(i; 0 ~ pipe_cnt ~ 2) {
	180	int ret = pipe(&fds[pipe_off + i]);
	181	if( ret < 0 ) { abort( "pipe error: (%d) %s\n", (int)errno, strerror(errno) ); }
	182	}
	183
[4f762d3]	184	// if(options.file_cache.path && options.file_cache.fixed_fds) {
	185	// register_fixed_files(cl, fds, pipe_off);
	186	// }
[d11d6eb]	187
[0aec496]	188	{
[348f81d5]	189	ServerCluster cl[options.clopts.nclusters];
[ece0e80]	190
	191	init_protocol();
[0aec496]	192	{
[3eb540fb]	193	Worker * workers = anew(options.clopts.nworkers);
[d9c2284]	194	for(i; options.clopts.nworkers) {
[d11d6eb]	195	// if( options.file_cache.fixed_fds ) {
	196	// workers[i].pipe[0] = pipe_off + (i * 2) + 0;
	197	// workers[i].pipe[1] = pipe_off + (i * 2) + 1;
	198	// }
	199	// else
	200	{
[d9c2284]	201	workers[i].pipe[0] = fds[pipe_off + (i * 2) + 0];
	202	workers[i].pipe[1] = fds[pipe_off + (i * 2) + 1];
[8e3034d]	203	workers[i].sockfd = server_fd;
	204	workers[i].addr = (struct sockaddr *)&address;
	205	workers[i].addrlen = (socklen_t*)&addrlen;
	206	workers[i].flags = 0;
[d9c2284]	207	}
[e235429]	208	unpark( workers[i] );
[d9c2284]	209	}
[348f81d5]	210	sout \| options.clopts.nworkers \| "workers started on" \| options.clopts.nprocs \| "processors /" \| options.clopts.nclusters \| "clusters";
	211	for(i; options.clopts.nclusters) {
	212	sout \| options.clopts.thrd_cnt[i] \| nonl;
	213	}
	214	sout \| nl;
[4087baf]	215	if(!options.interactive) park();
[0aec496]	216	{
[e95a117]	217	char buffer[128];
[2cd784a]	218	for() {
	219	int ret = cfa_read(0, buffer, 128, 0);
	220	if(ret == 0) break;
[153dc387]	221	if(ret < 0) abort( "main read error: (%d) %s\n", (int)errno, strerror(errno) );
[2cd784a]	222	sout \| "User wrote '" \| "" \| nonl;
	223	write(sout, buffer, ret - 1);
	224	sout \| "'";
[e95a117]	225	}
	226
[8c43d05]	227	sout \| "Shutdown received";
[0aec496]	228	}
[ece0e80]	229
[b57db73]	230	sout \| "Notifying connections..." \| nonl; flush( sout );
[ece0e80]	231	for(i; options.clopts.nworkers) {
[481ee28]	232	workers[i].done = true;
[ece0e80]	233	}
[b57db73]	234	sout \| "done";
[ece0e80]	235
[b57db73]	236	sout \| "Shutting down socket..." \| nonl; flush( sout );
[ece0e80]	237	int ret = shutdown( server_fd, SHUT_RD );
[b57db73]	238	if( ret < 0 ) {
	239	abort( "shutdown error: (%d) %s\n", (int)errno, strerror(errno) );
	240	}
	241	sout \| "done";
[ece0e80]	242
	243	//===================
	244	// Close Socket
[b57db73]	245	sout \| "Closing Socket..." \| nonl; flush( sout );
[ece0e80]	246	ret = close( server_fd );
	247	if(ret < 0) {
	248	abort( "close socket error: (%d) %s\n", (int)errno, strerror(errno) );
	249	}
[0197418]	250	sout \| "done";
	251
[b57db73]	252	sout \| "Stopping connection threads..." \| nonl; flush( sout );
[3eb540fb]	253	adelete(workers);
[0aec496]	254	}
[8c43d05]	255	sout \| "done";
[ece0e80]	256
[b57db73]	257	sout \| "Stopping protocol threads..." \| nonl; flush( sout );
[ece0e80]	258	deinit_protocol();
[8c43d05]	259	sout \| "done";
	260
[348f81d5]	261	sout \| "Stopping processors/clusters..." \| nonl; flush( sout );
[0aec496]	262	}
[8c43d05]	263	sout \| "done";
[d9c2284]	264
[b57db73]	265	sout \| "Closing splice fds..." \| nonl; flush( sout );
[d9c2284]	266	for(i; pipe_cnt) {
	267	ret = close( fds[pipe_off + i] );
	268	if(ret < 0) {
	269	abort( "close pipe error: (%d) %s\n", (int)errno, strerror(errno) );
	270	}
	271	}
	272	free(fds);
[8c43d05]	273	sout \| "done";
[c3ee5f3]	274
[b57db73]	275	sout \| "Stopping processors..." \| nonl; flush( sout );
[0aec496]	276	}
[8c43d05]	277	sout \| "done";
[0aec496]	278
	279	//===================
	280	// Close Files
[b57db73]	281	if( options.file_cache.path ) {
	282	sout \| "Closing open files..." \| nonl; flush( sout );
	283	close_cache();
	284	sout \| "done";
	285	}
[0aec496]	286	}

Note: See TracBrowser for help on using the repository browser.

Download in other formats:

Original Format