Context Navigation

main.cfa @ ac1aba4b

ADTast-experimentalpthread-emulationqualifiedEnum

Last change on this file since ac1aba4b was ce98816, checked in by Thierry Delisle <tdelisle@…>, 2 years ago
Stats printer now uses a condition variable to sleep so it exists immediately.
Property mode set to `100644`
File size: 10.4 KB

Rev	Line
[1e8b4b49]	1	#define _GNU_SOURCE
[0aec496]	2
	3	#include <errno.h>
[153570d]	4	#include <signal.h>
[0aec496]	5	#include <stdio.h>
	6	#include <string.h>
	7	#include <unistd.h>
	8	extern "C" {
[6117fc0]	9	#include <sched.h>
[c2df3031]	10	#include <signal.h>
[153570d]	11	#include <sys/eventfd.h>
[0aec496]	12	#include <sys/socket.h>
	13	#include <netinet/in.h>
	14	}
	15
[8c43d05]	16	#include <fstream.hfa>
[0aec496]	17	#include <kernel.hfa>
[ce98816]	18	#include <locks.hfa>
[153dc387]	19	#include <iofwd.hfa>
[0aec496]	20	#include <stats.hfa>
[d11d6eb]	21	#include <time.hfa>
[0aec496]	22	#include <thread.hfa>
	23
	24	#include "filecache.hfa"
	25	#include "options.hfa"
[c4b10e2]	26	#include "socket.hfa"
[0aec496]	27	#include "worker.hfa"
	28
[d11d6eb]	29	extern void register_fixed_files( cluster &, int *, unsigned count );
	30
	31	Duration default_preemption() {
	32	return 0;
	33	}
	34
[153dc387]	35	//=============================================================================================
	36	// Stats Printer
	37	//============================================================================================='
	38
[ef3c383]	39	thread StatsPrinter {
	40	Worker * workers;
	41	int worker_cnt;
[ce98816]	42	condition_variable(fast_block_lock) var;
[ef3c383]	43	};
[153dc387]	44
[348f81d5]	45	void ?{}( StatsPrinter & this, cluster & cl ) {
	46	((thread&)this){ "Stats Printer Thread", cl };
[ef3c383]	47	this.worker_cnt = 0;
[153dc387]	48	}
	49
[2cd784a]	50	void ^?{}( StatsPrinter & mutex this ) {}
	51
[ef3c383]	52	#define eng3(X) (ws(3, 3, unit(eng( X ))))
	53
[153dc387]	54	void main(StatsPrinter & this) {
	55	LOOP: for() {
	56	waitfor( ^?{} : this) {
	57	break LOOP;
	58	}
	59	or else {}
	60
[ce98816]	61	wait(this.var, 10`s);
[153dc387]	62
[348f81d5]	63	print_stats_now( *active_cluster(), CFA_STATS_READY_Q \| CFA_STATS_IO );
[ef3c383]	64	if(this.worker_cnt != 0) {
	65	uint64_t tries = 0;
	66	uint64_t calls = 0;
	67	uint64_t header = 0;
	68	uint64_t splcin = 0;
	69	uint64_t splcot = 0;
	70	struct {
	71	volatile uint64_t calls;
	72	volatile uint64_t bytes;
	73	} avgrd[zipf_cnts];
	74	memset(avgrd, 0, sizeof(avgrd));
	75
	76	for(i; this.worker_cnt) {
	77	tries += this.workers[i].stats.sendfile.tries;
	78	calls += this.workers[i].stats.sendfile.calls;
	79	header += this.workers[i].stats.sendfile.header;
	80	splcin += this.workers[i].stats.sendfile.splcin;
	81	splcot += this.workers[i].stats.sendfile.splcot;
	82	for(j; zipf_cnts) {
	83	avgrd[j].calls += this.workers[i].stats.sendfile.avgrd[j].calls;
	84	avgrd[j].bytes += this.workers[i].stats.sendfile.avgrd[j].bytes;
	85	}
	86	}
	87
	88	double ratio = ((double)tries) / calls;
	89
	90	sout \| "----- Worker Stats -----";
	91	sout \| "sendfile : " \| calls \| "calls," \| tries \| "tries (" \| ratio \| " try/call)";
	92	sout \| " " \| header \| "header," \| splcin \| "splice in," \| splcot \| "splice out";
	93	sout \| " - zipf sizes:";
	94	for(i; zipf_cnts) {
	95	double written = avgrd[i].calls > 0 ? ((double)avgrd[i].bytes) / avgrd[i].calls : 0;
	96	sout \| " " \| zipf_sizes[i] \| "bytes," \| avgrd[i].calls \| "shorts," \| written \| "written";
	97	}
	98	}
	99	else {
	100	sout \| "No Workers!";
	101	}
[348f81d5]	102	}
	103	}
	104
	105	//=============================================================================================
	106	// Globals
	107	//=============================================================================================
	108	struct ServerCluster {
	109	cluster self;
	110	processor * procs;
[2cd784a]	111	// io_context * ctxs;
[348f81d5]	112	StatsPrinter * prnt;
	113
	114	};
	115
	116	void ?{}( ServerCluster & this ) {
	117	(this.self){ "Server Cluster", options.clopts.params };
	118
[6117fc0]	119	cpu_set_t fullset;
	120	CPU_ZERO(&fullset);
	121	int ret = sched_getaffinity(getpid(), sizeof(fullset), &fullset);
	122	if( ret != 0 ) abort \| "sched_getaffinity failed with" \| errno \| strerror( errno );
	123	int cnt = CPU_COUNT(&fullset);
	124
[348f81d5]	125	this.procs = alloc(options.clopts.nprocs);
	126	for(i; options.clopts.nprocs) {
	127	(this.procs[i]){ "Benchmark Processor", this.self };
	128
[6117fc0]	129	int c = 0;
	130	int n = 1 + (i % cnt);
	131	for(int j = 0; j < CPU_SETSIZE; j++) {
	132	if(CPU_ISSET(j, &fullset)) n--;
	133	if(n == 0) {
	134	c = j;
	135	break;
	136	}
	137	}
	138	cpu_set_t localset;
	139	CPU_ZERO(&localset);
	140	CPU_SET(c, &localset);
	141	ret = pthread_setaffinity_np(this.procs[i].kernel_thread, sizeof(localset), &localset);
	142	if( ret != 0 ) abort \| "sched_getaffinity failed with" \| ret \| strerror( ret );
	143
[348f81d5]	144	#if !defined(__CFA_NO_STATISTICS__)
	145	if( options.clopts.procstats ) {
	146	print_stats_at_exit( *this.procs, this.self.print_stats );
	147	}
	148	if( options.clopts.viewhalts ) {
	149	print_halts( *this.procs );
	150	}
	151	#endif
	152	}
	153
[2cd784a]	154	if(options.stats) {
	155	this.prnt = alloc();
	156	(*this.prnt){ this.self };
	157	} else {
	158	this.prnt = 0p;
[348f81d5]	159	}
	160
	161	#if !defined(__CFA_NO_STATISTICS__)
	162	print_stats_at_exit( this.self, CFA_STATS_READY_Q \| CFA_STATS_IO );
	163	#endif
	164
	165	options.clopts.instance[options.clopts.cltr_cnt] = &this.self;
	166	options.clopts.cltr_cnt++;
[153dc387]	167	}
	168
[348f81d5]	169	void ^?{}( ServerCluster & this ) {
[2cd784a]	170	delete(this.prnt);
[348f81d5]	171
	172	for(i; options.clopts.nprocs) {
	173	^(this.procs[i]){};
	174	}
	175	free(this.procs);
	176
	177	^(this.self){};
	178	}
	179
	180	extern void init_protocol(void);
	181	extern void deinit_protocol(void);
	182
[153570d]	183	//=============================================================================================
	184	// Termination
	185	//=============================================================================================
	186
	187	int closefd;
	188	void cleanstop(int) {
	189	eventfd_t buffer = 1;
	190	char * buffer_s = (char*)&buffer;
	191	int ret = write(closefd, buffer_s, sizeof(buffer));
	192	if(ret < 0) abort( "eventfd write error: (%d) %s\n", (int)errno, strerror(errno) );
	193	return;
	194	}
	195
[0aec496]	196	//=============================================================================================
	197	// Main
	198	//============================================================================================='
	199	int main( int argc, char * argv[] ) {
[c4b10e2]	200	int ret;
[c2df3031]	201	__sighandler_t s = 1p;
	202	signal(SIGPIPE, s);
	203
[0aec496]	204	//===================
	205	// Parse args
[b57db73]	206	parse_options(argc, argv);
[0aec496]	207
[153570d]	208	//===================
	209	// Setup non-interactive termination
	210	if(!options.interactive) {
	211	closefd = eventfd(0, 0);
	212	if(closefd < 0) abort( "eventfd error: (%d) %s\n", (int)errno, strerror(errno) );
	213
	214	sighandler_t prev = signal(SIGTERM, cleanstop);
	215	intptr_t prev_workaround = (intptr_t) prev;
	216	// can't use SIG_ERR it crashes the compiler
	217	if(prev_workaround == -1) abort( "signal setup error: (%d) %s\n", (int)errno, strerror(errno) );
	218
	219	sout \| "Signal termination ready";
	220	}
	221
[0aec496]	222	//===================
	223	// Open Files
[b57db73]	224	if( options.file_cache.path ) {
	225	sout \| "Filling cache from" \| options.file_cache.path;
	226	fill_cache( options.file_cache.path );
	227	}
[0aec496]	228
	229	//===================
	230	// Open Socket
[8c43d05]	231	sout \| getpid() \| ": Listening on port" \| options.socket.port;
[0aec496]	232
	233	struct sockaddr_in address;
[c4b10e2]	234	int addrlen = prepaddr(address);
[0aec496]	235
[c4b10e2]	236	int server_fd = listener(address, addrlen);
[0aec496]	237
	238	//===================
	239	// Run Server Cluster
	240	{
[d9c2284]	241	int pipe_cnt = options.clopts.nworkers * 2;
	242	int pipe_off;
	243	int * fds;
	244	[fds, pipe_off] = filefds( pipe_cnt );
	245	for(i; 0 ~ pipe_cnt ~ 2) {
	246	int ret = pipe(&fds[pipe_off + i]);
	247	if( ret < 0 ) { abort( "pipe error: (%d) %s\n", (int)errno, strerror(errno) ); }
	248	}
	249
[4f762d3]	250	// if(options.file_cache.path && options.file_cache.fixed_fds) {
	251	// register_fixed_files(cl, fds, pipe_off);
	252	// }
[d11d6eb]	253
[0aec496]	254	{
[153570d]	255	// Stats printer makes a copy so this needs to persist longer than normal
	256	Worker * workers;
[348f81d5]	257	ServerCluster cl[options.clopts.nclusters];
[ece0e80]	258
	259	init_protocol();
[0aec496]	260	{
[153570d]	261	workers = anew(options.clopts.nworkers);
[ef3c383]	262	cl[0].prnt->workers = workers;
	263	cl[0].prnt->worker_cnt = options.clopts.nworkers;
[d9c2284]	264	for(i; options.clopts.nworkers) {
[d11d6eb]	265	// if( options.file_cache.fixed_fds ) {
	266	// workers[i].pipe[0] = pipe_off + (i * 2) + 0;
	267	// workers[i].pipe[1] = pipe_off + (i * 2) + 1;
	268	// }
	269	// else
	270	{
[d9c2284]	271	workers[i].pipe[0] = fds[pipe_off + (i * 2) + 0];
	272	workers[i].pipe[1] = fds[pipe_off + (i * 2) + 1];
[8e3034d]	273	workers[i].sockfd = server_fd;
	274	workers[i].addr = (struct sockaddr *)&address;
	275	workers[i].addrlen = (socklen_t*)&addrlen;
	276	workers[i].flags = 0;
[d9c2284]	277	}
[e235429]	278	unpark( workers[i] );
[d9c2284]	279	}
[348f81d5]	280	sout \| options.clopts.nworkers \| "workers started on" \| options.clopts.nprocs \| "processors /" \| options.clopts.nclusters \| "clusters";
	281	for(i; options.clopts.nclusters) {
	282	sout \| options.clopts.thrd_cnt[i] \| nonl;
	283	}
	284	sout \| nl;
[0aec496]	285	{
[153570d]	286	if(options.interactive) {
	287	char buffer[128];
	288	for() {
	289	int ret = cfa_read(0, buffer, 128, 0);
	290	if(ret == 0) break;
	291	if(ret < 0) abort( "main read error: (%d) %s\n", (int)errno, strerror(errno) );
	292	sout \| "User wrote '" \| "" \| nonl;
	293	write(sout, buffer, ret - 1);
	294	sout \| "'";
	295	}
	296	}
	297	else {
	298	char buffer[sizeof(eventfd_t)];
	299	int ret = cfa_read(closefd, buffer, sizeof(eventfd_t), 0);
[153dc387]	300	if(ret < 0) abort( "main read error: (%d) %s\n", (int)errno, strerror(errno) );
[e95a117]	301	}
	302
[8c43d05]	303	sout \| "Shutdown received";
[0aec496]	304	}
[ece0e80]	305
[b57db73]	306	sout \| "Notifying connections..." \| nonl; flush( sout );
[ece0e80]	307	for(i; options.clopts.nworkers) {
[481ee28]	308	workers[i].done = true;
[ece0e80]	309	}
[b57db73]	310	sout \| "done";
[ece0e80]	311
[b57db73]	312	sout \| "Shutting down socket..." \| nonl; flush( sout );
[ece0e80]	313	int ret = shutdown( server_fd, SHUT_RD );
[b57db73]	314	if( ret < 0 ) {
	315	abort( "shutdown error: (%d) %s\n", (int)errno, strerror(errno) );
	316	}
	317	sout \| "done";
[ece0e80]	318
	319	//===================
	320	// Close Socket
[b57db73]	321	sout \| "Closing Socket..." \| nonl; flush( sout );
[ece0e80]	322	ret = close( server_fd );
	323	if(ret < 0) {
	324	abort( "close socket error: (%d) %s\n", (int)errno, strerror(errno) );
	325	}
[0197418]	326	sout \| "done";
	327
[b57db73]	328	sout \| "Stopping connection threads..." \| nonl; flush( sout );
[153570d]	329	for(i; options.clopts.nworkers) {
[c4b10e2]	330	for(j; 2) {
	331	ret = close(workers[i].pipe[j]);
	332	if(ret < 0) abort( "close pipe %d error: (%d) %s\n", j, (int)errno, strerror(errno) );
	333	}
[153570d]	334	join(workers[i]);
	335	}
[0aec496]	336	}
[8c43d05]	337	sout \| "done";
[ece0e80]	338
[b57db73]	339	sout \| "Stopping protocol threads..." \| nonl; flush( sout );
[ece0e80]	340	deinit_protocol();
[8c43d05]	341	sout \| "done";
	342
[153570d]	343	sout \| "Stopping printer threads..." \| nonl; flush( sout );
	344	for(i; options.clopts.nclusters) {
	345	StatsPrinter * p = cl[i].prnt;
[ce98816]	346	if(p) {
	347	notify_one(p->var);
	348	join(*p);
	349	}
[153570d]	350	}
	351	sout \| "done";
	352
	353	// Now that the stats printer is stopped, we can reclaim this
	354	adelete(workers);
	355
[348f81d5]	356	sout \| "Stopping processors/clusters..." \| nonl; flush( sout );
[0aec496]	357	}
[8c43d05]	358	sout \| "done";
[d9c2284]	359
[c4b10e2]	360	// sout \| "Closing splice fds..." \| nonl; flush( sout );
	361	// for(i; pipe_cnt) {
	362	// ret = close( fds[pipe_off + i] );
	363	// if(ret < 0) {
	364	// abort( "close pipe error: (%d) %s\n", (int)errno, strerror(errno) );
	365	// }
	366	// }
[d9c2284]	367	free(fds);
[8c43d05]	368	sout \| "done";
[c3ee5f3]	369
[b57db73]	370	sout \| "Stopping processors..." \| nonl; flush( sout );
[0aec496]	371	}
[8c43d05]	372	sout \| "done";
[0aec496]	373
	374	//===================
	375	// Close Files
[b57db73]	376	if( options.file_cache.path ) {
	377	sout \| "Closing open files..." \| nonl; flush( sout );
	378	close_cache();
	379	sout \| "done";
	380	}
[ef3c383]	381	}
	382
	383	const size_t zipf_sizes[] = { 102, 204, 307, 409, 512, 614, 716, 819, 921, 1024, 2048, 3072, 4096, 5120, 6144, 7168, 8192, 9216, 10240, 20480, 30720, 40960, 51200, 61440, 71680, 81920, 92160, 102400, 204800, 307200, 409600, 512000, 614400, 716800, 819200, 921600 };
	384	static_assert(zipf_cnts == sizeof(zipf_sizes) / sizeof(zipf_sizes[0]));

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

Download in other formats:

Original Format