Context Navigation

main.cfa @ 329e26a

ADTast-experimentalpthread-emulationqualifiedEnum

Last change on this file since 329e26a was 329e26a, checked in by Thierry Delisle <tdelisle@…>, 23 months ago
Re-instated the isolate/multi-cluster option.
Property mode set to `100644`
File size: 11.1 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"
[137974a]	27	#include "printer.hfa"
[0aec496]	28	#include "worker.hfa"
	29
[d11d6eb]	30	extern void register_fixed_files( cluster &, int *, unsigned count );
	31
	32	Duration default_preemption() {
	33	return 0;
	34	}
	35
[348f81d]	36	//=============================================================================================
	37	// Globals
	38	//=============================================================================================
	39	void ?{}( ServerCluster & this ) {
	40	(this.self){ "Server Cluster", options.clopts.params };
	41
[6117fc0]	42	cpu_set_t fullset;
	43	CPU_ZERO(&fullset);
	44	int ret = sched_getaffinity(getpid(), sizeof(fullset), &fullset);
	45	if( ret != 0 ) abort \| "sched_getaffinity failed with" \| errno \| strerror( errno );
	46	int cnt = CPU_COUNT(&fullset);
	47
[348f81d]	48	this.procs = alloc(options.clopts.nprocs);
	49	for(i; options.clopts.nprocs) {
	50	(this.procs[i]){ "Benchmark Processor", this.self };
	51
[329e26a]	52	// int c = 0;
	53	// int n = 1 + (i % cnt);
	54	// for(int j = 0; j < CPU_SETSIZE; j++) {
	55	// if(CPU_ISSET(j, &fullset)) n--;
	56	// if(n == 0) {
	57	// c = j;
	58	// break;
	59	// }
	60	// }
	61	// cpu_set_t localset;
	62	// CPU_ZERO(&localset);
	63	// CPU_SET(c, &localset);
	64	// ret = pthread_setaffinity_np(this.procs[i].kernel_thread, sizeof(localset), &localset);
	65	// if( ret != 0 ) abort \| "sched_getaffinity failed with" \| ret \| strerror( ret );
[6117fc0]	66
[348f81d]	67	#if !defined(__CFA_NO_STATISTICS__)
	68	if( options.clopts.procstats ) {
	69	print_stats_at_exit( *this.procs, this.self.print_stats );
	70	}
	71	if( options.clopts.viewhalts ) {
	72	print_halts( *this.procs );
	73	}
	74	#endif
	75	}
	76
	77	#if !defined(__CFA_NO_STATISTICS__)
	78	print_stats_at_exit( this.self, CFA_STATS_READY_Q \| CFA_STATS_IO );
	79	#endif
	80
[329e26a]	81	options.clopts.instance[options.clopts.cltr_cnt] = &this.self;
	82	options.clopts.cltr_cnt++;
[153dc387]	83	}
	84
[348f81d]	85	void ^?{}( ServerCluster & this ) {
	86	for(i; options.clopts.nprocs) {
	87	^(this.procs[i]){};
	88	}
	89	free(this.procs);
	90
	91	^(this.self){};
	92	}
	93
	94	extern void init_protocol(void);
	95	extern void deinit_protocol(void);
	96
[7f0ac12]	97	//=============================================================================================
	98	// REUSEPORT
	99	//=============================================================================================
	100
	101	size_t sockarr_size;
	102	struct __attribute__((aligned(128))) Q {
	103	mpsc_queue(PendingRead) q;
	104	};
	105
[153570d]	106	//=============================================================================================
	107	// Termination
	108	//=============================================================================================
	109
	110	int closefd;
	111	void cleanstop(int) {
	112	eventfd_t buffer = 1;
	113	char * buffer_s = (char*)&buffer;
	114	int ret = write(closefd, buffer_s, sizeof(buffer));
	115	if(ret < 0) abort( "eventfd write error: (%d) %s\n", (int)errno, strerror(errno) );
	116	return;
	117	}
	118
[0aec496]	119	//=============================================================================================
	120	// Main
	121	//============================================================================================='
	122	int main( int argc, char * argv[] ) {
[c4b10e2]	123	int ret;
[c2df3031]	124	__sighandler_t s = 1p;
	125	signal(SIGPIPE, s);
	126
[0aec496]	127	//===================
	128	// Parse args
[b57db73]	129	parse_options(argc, argv);
[0aec496]	130
[153570d]	131	//===================
	132	// Setup non-interactive termination
	133	if(!options.interactive) {
	134	closefd = eventfd(0, 0);
	135	if(closefd < 0) abort( "eventfd error: (%d) %s\n", (int)errno, strerror(errno) );
	136
	137	sighandler_t prev = signal(SIGTERM, cleanstop);
	138	intptr_t prev_workaround = (intptr_t) prev;
	139	// can't use SIG_ERR it crashes the compiler
	140	if(prev_workaround == -1) abort( "signal setup error: (%d) %s\n", (int)errno, strerror(errno) );
	141
	142	sout \| "Signal termination ready";
	143	}
	144
[0aec496]	145	//===================
	146	// Open Files
[b57db73]	147	if( options.file_cache.path ) {
	148	sout \| "Filling cache from" \| options.file_cache.path;
	149	fill_cache( options.file_cache.path );
	150	}
[0aec496]	151
	152	//===================
	153	// Open Socket
[8c43d05]	154	sout \| getpid() \| ": Listening on port" \| options.socket.port;
[0aec496]	155
	156	struct sockaddr_in address;
[c4b10e2]	157	int addrlen = prepaddr(address);
[0aec496]	158
[86c12d65]	159	int server_fd;
[0aec496]	160
	161	//===================
	162	// Run Server Cluster
	163	{
[d9c2284]	164	int pipe_cnt = options.clopts.nworkers * 2;
	165	int pipe_off;
	166	int * fds;
	167	[fds, pipe_off] = filefds( pipe_cnt );
	168	for(i; 0 ~ pipe_cnt ~ 2) {
	169	int ret = pipe(&fds[pipe_off + i]);
	170	if( ret < 0 ) { abort( "pipe error: (%d) %s\n", (int)errno, strerror(errno) ); }
	171	}
	172
[4f762d3]	173	// if(options.file_cache.path && options.file_cache.fixed_fds) {
	174	// register_fixed_files(cl, fds, pipe_off);
	175	// }
[d11d6eb]	176
[0aec496]	177	{
[153570d]	178	// Stats printer makes a copy so this needs to persist longer than normal
[7f0ac12]	179	connection ** conns;
	180	AcceptWorker * aworkers = 0p;
	181	ChannelWorker * cworkers = 0p;
	182	Acceptor * acceptors = 0p;
	183	Q * queues = 0p;
[329e26a]	184	ServerCluster cl[options.clopts.nclusters];
[ece0e80]	185
[137974a]	186	if(options.stats) {
	187	stats_thrd = alloc();
[329e26a]	188	(*stats_thrd){ cl };
[137974a]	189	} else {
	190	stats_thrd = 0p;
	191	}
	192
[ece0e80]	193	init_protocol();
[0aec496]	194	{
[329e26a]	195	int nacceptors = options.clopts.nprocs * options.clopts.nclusters;
[7f0ac12]	196	conns = alloc(options.clopts.nworkers);
	197	if(options.socket.reuseport) {
[329e26a]	198	queues = alloc(nacceptors);
	199	acceptors = alloc(nacceptors);
	200	sout \| "Creating" \| nacceptors \| "Acceptors";
	201	for(i; nacceptors) {
	202	(acceptors[i]){ i % options.clopts.nclusters };
	203	}
	204	for(i; nacceptors) {
[7f0ac12]	205	(queues[i]){};
	206	{
	207	acceptors[i].sockfd = listener(address, addrlen);
	208	acceptors[i].addr = (struct sockaddr *)&address;
	209	acceptors[i].addrlen = (socklen_t*)&addrlen;
	210	acceptors[i].flags = 0;
	211	acceptors[i].queue = &queues[i].q;
	212	}
	213	unpark( acceptors[i] );
	214	}
	215
	216	cworkers = anew(options.clopts.nworkers);
	217	for(i; options.clopts.nworkers) {
	218	{
	219	cworkers[i].conn.pipe[0] = fds[pipe_off + (i * 2) + 0];
	220	cworkers[i].conn.pipe[1] = fds[pipe_off + (i * 2) + 1];
[329e26a]	221	cworkers[i].queue = &queues[i % nacceptors].q;
[7f0ac12]	222	conns[i] = &cworkers[i].conn;
	223	}
	224	unpark( cworkers[i] );
	225	}
	226	}
	227	else {
	228	server_fd = listener(address, addrlen);
	229	aworkers = anew(options.clopts.nworkers);
	230	for(i; options.clopts.nworkers) {
	231	// if( options.file_cache.fixed_fds ) {
	232	// workers[i].pipe[0] = pipe_off + (i * 2) + 0;
	233	// workers[i].pipe[1] = pipe_off + (i * 2) + 1;
	234	// }
	235	// else
	236	{
	237	aworkers[i].conn.pipe[0] = fds[pipe_off + (i * 2) + 0];
	238	aworkers[i].conn.pipe[1] = fds[pipe_off + (i * 2) + 1];
	239	aworkers[i].sockfd = server_fd;
	240	aworkers[i].addr = (struct sockaddr *)&address;
	241	aworkers[i].addrlen = (socklen_t*)&addrlen;
	242	aworkers[i].flags = 0;
	243	conns[i] = &aworkers[i].conn;
	244	}
	245	unpark( aworkers[i] );
[d9c2284]	246	}
	247	}
[329e26a]	248	sout \| options.clopts.nworkers \| "workers started on" \| options.clopts.nprocs \| "processors /" \| options.clopts.nclusters \| "clusters";
	249	for(i; options.clopts.nclusters) {
	250	sout \| options.clopts.thrd_cnt[i] \| nonl;
	251	}
[348f81d]	252	sout \| nl;
[0aec496]	253	{
[153570d]	254	if(options.interactive) {
	255	char buffer[128];
	256	for() {
	257	int ret = cfa_read(0, buffer, 128, 0);
	258	if(ret == 0) break;
	259	if(ret < 0) abort( "main read error: (%d) %s\n", (int)errno, strerror(errno) );
	260	sout \| "User wrote '" \| "" \| nonl;
	261	write(sout, buffer, ret - 1);
	262	sout \| "'";
	263	}
	264	}
	265	else {
	266	char buffer[sizeof(eventfd_t)];
	267	int ret = cfa_read(closefd, buffer, sizeof(eventfd_t), 0);
[153dc387]	268	if(ret < 0) abort( "main read error: (%d) %s\n", (int)errno, strerror(errno) );
[e95a117]	269	}
	270
[8c43d05]	271	sout \| "Shutdown received";
[0aec496]	272	}
[ece0e80]	273
[7f0ac12]	274	//===================
	275	// Close Socket and join
	276	if(options.socket.reuseport) {
	277	sout \| "Notifying connections..." \| nonl; flush( sout );
[329e26a]	278	for(i; nacceptors) {
[7f0ac12]	279	acceptors[i].done = true;
	280	}
	281	for(i; options.clopts.nworkers) {
	282	cworkers[i].done = true;
	283	}
	284	sout \| "done";
	285
	286	sout \| "Shutting down Socket..." \| nonl; flush( sout );
[329e26a]	287	for(i; nacceptors) {
[7f0ac12]	288	ret = shutdown( acceptors[i].sockfd, SHUT_RD );
	289	if( ret < 0 ) {
	290	abort( "shutdown1 error: (%d) %s\n", (int)errno, strerror(errno) );
	291	}
	292	}
	293	sout \| "done";
[ece0e80]	294
[7f0ac12]	295	sout \| "Closing Socket..." \| nonl; flush( sout );
[329e26a]	296	for(i; nacceptors) {
[7f0ac12]	297	ret = close( acceptors[i].sockfd );
	298	if( ret < 0) {
	299	abort( "close socket error: (%d) %s\n", (int)errno, strerror(errno) );
	300	}
	301	}
	302	sout \| "done";
	303
	304	sout \| "Stopping accept threads..." \| nonl; flush( sout );
[329e26a]	305	for(i; nacceptors) {
[7f0ac12]	306	join(acceptors[i]);
	307	}
	308	sout \| "done";
	309
	310	sout \| "Draining worker queues..." \| nonl; flush( sout );
[329e26a]	311	for(i; nacceptors) {
[7f0ac12]	312	PendingRead * p = 0p;
	313	while(p = pop(queues[i].q)) {
	314	fulfil(p->f, -ECONNRESET);
	315	}
	316	}
	317	sout \| "done";
	318
	319	sout \| "Stopping worker threads..." \| nonl; flush( sout );
[86c12d65]	320	for(i; options.clopts.nworkers) {
[7f0ac12]	321	for(j; 2) {
	322	ret = close(cworkers[i].conn.pipe[j]);
	323	if(ret < 0) abort( "close pipe %d error: (%d) %s\n", j, (int)errno, strerror(errno) );
	324	}
	325	join(cworkers[i]);
[86c12d65]	326	}
	327	}
	328	else {
[7f0ac12]	329	sout \| "Notifying connections..." \| nonl; flush( sout );
	330	for(i; options.clopts.nworkers) {
	331	aworkers[i].done = true;
	332	}
	333	sout \| "done";
	334
	335	sout \| "Shutting down Socket..." \| nonl; flush( sout );
[86c12d65]	336	ret = shutdown( server_fd, SHUT_RD );
	337	if( ret < 0 ) {
[7f0ac12]	338	abort( "shutdown2 error: (%d) %s\n", (int)errno, strerror(errno) );
[86c12d65]	339	}
[7f0ac12]	340	sout \| "done";
[ece0e80]	341
[7f0ac12]	342	sout \| "Closing Socket..." \| nonl; flush( sout );
[86c12d65]	343	ret = close( server_fd );
	344	if(ret < 0) {
	345	abort( "close socket error: (%d) %s\n", (int)errno, strerror(errno) );
	346	}
[7f0ac12]	347	sout \| "done";
[0197418]	348
[7f0ac12]	349	sout \| "Stopping connection threads..." \| nonl; flush( sout );
	350	for(i; options.clopts.nworkers) {
	351	for(j; 2) {
	352	ret = close(aworkers[i].conn.pipe[j]);
	353	if(ret < 0) abort( "close pipe %d error: (%d) %s\n", j, (int)errno, strerror(errno) );
	354	}
	355	join(aworkers[i]);
[c4b10e2]	356	}
[153570d]	357	}
[0aec496]	358	}
[8c43d05]	359	sout \| "done";
[ece0e80]	360
[b57db73]	361	sout \| "Stopping protocol threads..." \| nonl; flush( sout );
[ece0e80]	362	deinit_protocol();
[8c43d05]	363	sout \| "done";
	364
[153570d]	365	sout \| "Stopping printer threads..." \| nonl; flush( sout );
[137974a]	366	if(stats_thrd) {
	367	notify_one(stats_thrd->var);
[153570d]	368	}
[137974a]	369	delete(stats_thrd);
[153570d]	370	sout \| "done";
	371
	372	// Now that the stats printer is stopped, we can reclaim this
[7f0ac12]	373	adelete(aworkers);
	374	adelete(cworkers);
	375	adelete(acceptors);
	376	adelete(queues);
	377	free(conns);
[153570d]	378
[348f81d]	379	sout \| "Stopping processors/clusters..." \| nonl; flush( sout );
[0aec496]	380	}
[8c43d05]	381	sout \| "done";
[d9c2284]	382
	383	free(fds);
[c3ee5f3]	384
[b57db73]	385	sout \| "Stopping processors..." \| nonl; flush( sout );
[0aec496]	386	}
[8c43d05]	387	sout \| "done";
[0aec496]	388
	389	//===================
	390	// Close Files
[b57db73]	391	if( options.file_cache.path ) {
	392	sout \| "Closing open files..." \| nonl; flush( sout );
	393	close_cache();
	394	sout \| "done";
	395	}
[ef3c383]	396	}
	397
	398	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 };
	399	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