Context Navigation

main.cfa@ de31a1d

Visit:

ADT ast-experimental enum forall-pointer-decay pthread-emulation qualifiedEnum stuck-waitfor-destruct

Last change on this file since de31a1d was 3eb540fb, checked in by Thierry Delisle <tdelisle@…>, 5 years ago
Modified httpforall not to use VLA of threads.
Property mode set to `100644`
File size: 6.9 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) {
[8c43d05]	152	sout \| "Waiting for port";
[ee913e0a]	153	} else {
[8c43d05]	154	sout \| "\r" \| waited \| nonl;
	155	flush( sout );
[ee913e0a]	156	}
	157	waited ++;
	158	sleep( 1`s );
	159	continue;
	160	}
	161	abort( "bind error: (%d) %s\n", (int)errno, strerror(errno) );
	162	}
	163	break;
[0aec496]	164	}
	165
[2ecbd7b]	166	ret = listen( server_fd, options.socket.backlog );
[0aec496]	167	if(ret < 0) {
	168	abort( "listen error: (%d) %s\n", (int)errno, strerror(errno) );
	169	}
	170
	171	//===================
	172	// Run Server Cluster
	173	{
[d9c2284]	174	int pipe_cnt = options.clopts.nworkers * 2;
	175	int pipe_off;
	176	int * fds;
	177	[fds, pipe_off] = filefds( pipe_cnt );
	178	for(i; 0 ~ pipe_cnt ~ 2) {
	179	int ret = pipe(&fds[pipe_off + i]);
	180	if( ret < 0 ) { abort( "pipe error: (%d) %s\n", (int)errno, strerror(errno) ); }
	181	}
	182
[4f762d3]	183	// if(options.file_cache.path && options.file_cache.fixed_fds) {
	184	// register_fixed_files(cl, fds, pipe_off);
	185	// }
[d11d6eb]	186
[0aec496]	187	{
[348f81d5]	188	ServerCluster cl[options.clopts.nclusters];
[ece0e80]	189
	190	init_protocol();
[0aec496]	191	{
[3eb540fb]	192	Worker * workers = anew(options.clopts.nworkers);
[d9c2284]	193	for(i; options.clopts.nworkers) {
[d11d6eb]	194	// if( options.file_cache.fixed_fds ) {
	195	// workers[i].pipe[0] = pipe_off + (i * 2) + 0;
	196	// workers[i].pipe[1] = pipe_off + (i * 2) + 1;
	197	// }
	198	// else
	199	{
[d9c2284]	200	workers[i].pipe[0] = fds[pipe_off + (i * 2) + 0];
	201	workers[i].pipe[1] = fds[pipe_off + (i * 2) + 1];
[8e3034d]	202	workers[i].sockfd = server_fd;
	203	workers[i].addr = (struct sockaddr *)&address;
	204	workers[i].addrlen = (socklen_t*)&addrlen;
	205	workers[i].flags = 0;
[d9c2284]	206	}
[e235429]	207	unpark( workers[i] );
[d9c2284]	208	}
[348f81d5]	209	sout \| options.clopts.nworkers \| "workers started on" \| options.clopts.nprocs \| "processors /" \| options.clopts.nclusters \| "clusters";
	210	for(i; options.clopts.nclusters) {
	211	sout \| options.clopts.thrd_cnt[i] \| nonl;
	212	}
	213	sout \| nl;
[4087baf]	214	if(!options.interactive) park();
[0aec496]	215	{
[e95a117]	216	char buffer[128];
[2cd784a]	217	for() {
	218	int ret = cfa_read(0, buffer, 128, 0);
	219	if(ret == 0) break;
[153dc387]	220	if(ret < 0) abort( "main read error: (%d) %s\n", (int)errno, strerror(errno) );
[2cd784a]	221	sout \| "User wrote '" \| "" \| nonl;
	222	write(sout, buffer, ret - 1);
	223	sout \| "'";
[e95a117]	224	}
	225
[8c43d05]	226	sout \| "Shutdown received";
[0aec496]	227	}
[ece0e80]	228
[b57db73]	229	sout \| "Notifying connections..." \| nonl; flush( sout );
[ece0e80]	230	for(i; options.clopts.nworkers) {
[481ee28]	231	workers[i].done = true;
[ece0e80]	232	}
[b57db73]	233	sout \| "done";
[ece0e80]	234
[b57db73]	235	sout \| "Shutting down socket..." \| nonl; flush( sout );
[ece0e80]	236	int ret = shutdown( server_fd, SHUT_RD );
[b57db73]	237	if( ret < 0 ) {
	238	abort( "shutdown error: (%d) %s\n", (int)errno, strerror(errno) );
	239	}
	240	sout \| "done";
[ece0e80]	241
	242	//===================
	243	// Close Socket
[b57db73]	244	sout \| "Closing Socket..." \| nonl; flush( sout );
[ece0e80]	245	ret = close( server_fd );
	246	if(ret < 0) {
	247	abort( "close socket error: (%d) %s\n", (int)errno, strerror(errno) );
	248	}
[0197418]	249	sout \| "done";
	250
[b57db73]	251	sout \| "Stopping connection threads..." \| nonl; flush( sout );
[3eb540fb]	252	adelete(workers);
[0aec496]	253	}
[8c43d05]	254	sout \| "done";
[ece0e80]	255
[b57db73]	256	sout \| "Stopping protocol threads..." \| nonl; flush( sout );
[ece0e80]	257	deinit_protocol();
[8c43d05]	258	sout \| "done";
	259
[348f81d5]	260	sout \| "Stopping processors/clusters..." \| nonl; flush( sout );
[0aec496]	261	}
[8c43d05]	262	sout \| "done";
[d9c2284]	263
[b57db73]	264	sout \| "Closing splice fds..." \| nonl; flush( sout );
[d9c2284]	265	for(i; pipe_cnt) {
	266	ret = close( fds[pipe_off + i] );
	267	if(ret < 0) {
	268	abort( "close pipe error: (%d) %s\n", (int)errno, strerror(errno) );
	269	}
	270	}
	271	free(fds);
[8c43d05]	272	sout \| "done";
[c3ee5f3]	273
[b57db73]	274	sout \| "Stopping processors..." \| nonl; flush( sout );
[0aec496]	275	}
[8c43d05]	276	sout \| "done";
[0aec496]	277
	278	//===================
	279	// Close Files
[b57db73]	280	if( options.file_cache.path ) {
	281	sout \| "Closing open files..." \| nonl; flush( sout );
	282	close_cache();
	283	sout \| "done";
	284	}
[0aec496]	285	}

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

Download in other formats:

Original Format