Context Navigation

worker.cfa @ 8bee858

ADTast-experimentalpthread-emulation

Last change on this file since 8bee858 was 3f95dab, checked in by Thierry Delisle <tdelisle@…>, 2 years ago
Committing hopefully last version of the webserver
Property mode set to `100644`
File size: 10.2 KB

Rev	Line
[0aec496]	1	#include "worker.hfa"
	2
	3	#include <errno.h>
	4	#include <stdio.h>
	5	#include <string.h>
[c82af9f]	6	#include <unistd.h>
[0aec496]	7
[8c43d05]	8	#include <fstream.hfa>
[0aec496]	9	#include <iofwd.hfa>
[7f0ac12]	10	#include <mutex_stmt.hfa>
[0aec496]	11
	12	#include "options.hfa"
	13	#include "protocol.hfa"
	14	#include "filecache.hfa"
	15
[3f95dab]	16	static const unsigned long long period = 50_000_000;
	17
[0aec496]	18	//=============================================================================================
[7f0ac12]	19	// Generic connection handling
[0aec496]	20	//=============================================================================================
[137974ae]	21	static void handle_connection( connection & this, volatile int & fd, char * buffer, size_t len, io_future_t * f, unsigned long long & last ) {
[7f0ac12]	22	REQUEST:
	23	for() {
	24	bool closed;
	25	HttpCode code;
	26	const char * file;
	27	size_t name_size;
	28
	29	// Read the http request
[10ba012]	30	if( options.log ) mutex(sout) sout \| "=== Reading request ===";
[7f0ac12]	31	[code, closed, file, name_size] = http_read(fd, buffer, len, f);
	32	f = 0p;
	33
	34	// if we are done, break out of the loop
	35	if( closed ) break REQUEST;
	36
	37	// If this wasn't a request retrun 400
	38	if( code != OK200 ) {
	39	sout \| "=== Invalid Request :" \| code_val(code) \| "===";
	40	answer_error(fd, code);
	41	continue REQUEST;
	42	}
[ef3c383]	43
[7f0ac12]	44	if(0 == strncmp(file, "plaintext", min(name_size, sizeof("plaintext") ))) {
[10ba012]	45	if( options.log ) mutex(sout) sout \| "=== Request for /plaintext ===";
[7f0ac12]	46
	47	int ret = answer_plaintext(fd);
	48	if( ret == -ECONNRESET ) break REQUEST;
	49
[10ba012]	50	if( options.log ) mutex(sout) sout \| "=== Answer sent ===";
[7f0ac12]	51	continue REQUEST;
	52	}
	53
	54	if(0 == strncmp(file, "ping", min(name_size, sizeof("ping") ))) {
[10ba012]	55	if( options.log ) mutex(sout) sout \| "=== Request for /ping ===";
[7f0ac12]	56
	57	// Send the header
	58	int ret = answer_empty(fd);
	59	if( ret == -ECONNRESET ) break REQUEST;
	60
[10ba012]	61	if( options.log ) mutex(sout) sout \| "=== Answer sent ===";
[7f0ac12]	62	continue REQUEST;
	63	}
	64
	65	if( options.log ) {
	66	sout \| "=== Request for file " \| nonl;
	67	write(sout, file, name_size);
	68	sout \| " ===";
	69	}
	70
	71	if( !options.file_cache.path ) {
	72	if( options.log ) {
	73	sout \| "=== File Not Found (" \| nonl;
	74	write(sout, file, name_size);
	75	sout \| ") ===";
	76	}
	77	answer_error(fd, E405);
	78	continue REQUEST;
	79	}
	80
	81	// Get the fd from the file cache
	82	int ans_fd;
	83	size_t count;
	84	[ans_fd, count] = get_file( file, name_size );
	85
	86	// If we can't find the file, return 404
	87	if( ans_fd < 0 ) {
	88	if( options.log ) {
	89	sout \| "=== File Not Found (" \| nonl;
	90	write(sout, file, name_size);
	91	sout \| ") ===";
	92	}
	93	answer_error(fd, E404);
	94	continue REQUEST;
	95	}
	96
	97	// Send the desired file
	98	int ret = answer_sendfile( this.pipe, fd, ans_fd, count, this.stats.sendfile );
[3f95dab]	99	if(ret < 0) {
	100	if( ret == -ECONNABORTED ) break REQUEST;
	101	if( ret == -ECONNRESET ) break REQUEST;
	102	if( ret == -EPIPE ) break REQUEST;
	103	abort( "sendfile error: %d (%d) %s\n", ret, (int)errno, strerror(errno) );
	104	}
[7f0ac12]	105
[10ba012]	106	if( options.log ) mutex(sout) sout \| "=== Answer sent ===";
[ef3c383]	107	}
[137974ae]	108
	109	if (stats_thrd) {
	110	unsigned long long next = rdtscl();
[3f95dab]	111	if(next > (last + period)) {
[10ba012]	112	if(try_lock(stats_thrd->stats.lock __cfaabi_dbg_ctx2)) {
[137974ae]	113	push(this.stats.sendfile, stats_thrd->stats.send);
	114	unlock(stats_thrd->stats.lock);
	115	last = next;
	116	}
	117	}
	118	}
[481ee28]	119	}
	120
[7f0ac12]	121	//=============================================================================================
	122	// Self Accepting Worker Thread
	123	//=============================================================================================
	124	void ?{}( AcceptWorker & this ) {
[329e26a]	125	size_t cli = rand() % options.clopts.cltr_cnt;
	126	((thread&)this){ "Server Worker Thread", *options.clopts.instance[cli], 64000 };
	127	options.clopts.thrd_cnt[cli]++;
[7f0ac12]	128	this.done = false;
[0aec496]	129	}
	130
[7f0ac12]	131	void main( AcceptWorker & this ) {
[e235429]	132	park();
[137974ae]	133	unsigned long long last = rdtscl();
[7f0ac12]	134	/* paranoid */ assert( this.conn.pipe[0] != -1 );
	135	/* paranoid */ assert( this.conn.pipe[1] != -1 );
[e95a117]	136	for() {
[10ba012]	137	if( options.log ) mutex(sout) sout \| "=== Accepting connection ===";
[7f0ac12]	138	int fd = cfa_accept4( this.sockfd, this.[addr, addrlen, flags], CFA_IO_LAZY );
[8e3034d]	139	if(fd < 0) {
	140	if( errno == ECONNABORTED ) break;
[ee59ede]	141	if( this.done && (errno == EINVAL \|\| errno == EBADF) ) break;
[8e3034d]	142	abort( "accept error: (%d) %s\n", (int)errno, strerror(errno) );
	143	}
[4f762d3]	144	if(this.done) break;
[e95a117]	145
[10ba012]	146	if( options.log ) mutex(sout) sout \| "=== New connection" \| fd \| "" \| ", waiting for requests ===";
[7f0ac12]	147	size_t len = options.socket.buflen;
	148	char buffer[len];
[137974ae]	149	handle_connection( this.conn, fd, buffer, len, 0p, last );
[3f95dab]	150	this.conn.stats.sendfile.maxfd = max(this.conn.stats.sendfile.maxfd, fd);
	151	this.conn.stats.sendfile.close++;
[0aec496]	152
[10ba012]	153	if( options.log ) mutex(sout) sout \| "=== Connection closed ===";
[7f0ac12]	154	}
	155	}
[0aec496]	156
[561dd26]	157
[7f0ac12]	158	//=============================================================================================
	159	// Channel Worker Thread
	160	//=============================================================================================
	161	void ?{}( ChannelWorker & this ) {
[329e26a]	162	size_t cli = rand() % options.clopts.cltr_cnt;
	163	((thread&)this){ "Server Worker Thread", *options.clopts.instance[cli], 64000 };
	164	options.clopts.thrd_cnt[cli]++;
[7f0ac12]	165	this.done = false;
	166	}
[561dd26]	167
[7f0ac12]	168	void main( ChannelWorker & this ) {
	169	park();
[137974ae]	170	unsigned long long last = rdtscl();
[7f0ac12]	171	/* paranoid */ assert( this.conn.pipe[0] != -1 );
	172	/* paranoid */ assert( this.conn.pipe[1] != -1 );
[3f95dab]	173	this.conn.stats.sendfile.maxfd = max(this.conn.pipe[0], this.conn.pipe[1]);
[7f0ac12]	174	for() {
	175	size_t len = options.socket.buflen;
	176	char buffer[len];
	177	PendingRead p;
[10ba012]	178	p.next = 0p;
[7f0ac12]	179	p.in.buf = (void*)buffer;
	180	p.in.len = len;
	181	push(*this.queue, &p);
[561dd26]	182
[10ba012]	183	if( options.log ) mutex(sout) sout \| "=== Waiting new connection ===";
[137974ae]	184	handle_connection( this.conn, p.out.fd, buffer, len, &p.f, last );
[3f95dab]	185	if(this.done) break;
	186	this.conn.stats.sendfile.maxfd = max(this.conn.stats.sendfile.maxfd, p.out.fd);
	187	this.conn.stats.sendfile.close++;
[561dd26]	188
[10ba012]	189	if( options.log ) mutex(sout) sout \| "=== Connection closed ===";
[7f0ac12]	190	}
[3f95dab]	191
	192	lock(stats_thrd->stats.lock __cfaabi_dbg_ctx2);
	193	push(this.conn.stats.sendfile, stats_thrd->stats.send);
	194	unlock(stats_thrd->stats.lock);
[7f0ac12]	195	}
[0aec496]	196
[7f0ac12]	197	extern "C" {
	198	extern int accept4(int sockfd, struct sockaddr addr, socklen_t addrlen, int flags);
	199	}
[97748ee]	200
[329e26a]	201	void ?{}( Acceptor & this, int cli ) {
	202	((thread&)this){ "Server Acceptor Thread", *options.clopts.instance[cli], 64000 };
	203	options.clopts.thrd_cnt[cli]++;
[7f0ac12]	204	this.done = false;
	205	}
[b57db73]	206
[07997cd]	207	static inline __s32 get_res( io_future_t & this ) {
	208	if( this.result < 0 ) {{
	209	errno = -this.result;
	210	return -1;
	211	}}
	212	return this.result;
	213	}
	214
	215	static inline void push_connection( Acceptor & this, int fd ) {
[3f95dab]	216	this.stats.accepts++;
[07997cd]	217	PendingRead * p = 0p;
	218	for() {
	219	if(this.done) return;
	220	p = pop(*this.queue);
	221	if(p) break;
	222	yield();
	223	this.stats.creates++;
	224	};
	225
	226	p->out.fd = fd;
	227	async_recv(p->f, p->out.fd, p->in.buf, p->in.len, 0, CFA_IO_LAZY);
	228	}
	229
	230	// #define ACCEPT_SPIN
[3f95dab]	231	#define ACCEPT_ONE
	232	// #define ACCEPT_MANY
[c25338d]	233
[7f0ac12]	234	void main( Acceptor & this ) {
	235	park();
[137974ae]	236	unsigned long long last = rdtscl();
[c25338d]	237
	238	#if defined(ACCEPT_SPIN)
[7f0ac12]	239	if( options.log ) sout \| "=== Accepting connection ===";
	240	for() {
	241	int fd = accept4(this.sockfd, this.[addr, addrlen, flags]);
	242	if(fd < 0) {
	243	if( errno == EWOULDBLOCK) {
[137974ae]	244	this.stats.eagains++;
[7f0ac12]	245	yield();
	246	continue;
[97748ee]	247	}
[7f0ac12]	248	if( errno == ECONNABORTED ) break;
	249	if( this.done && (errno == EINVAL \|\| errno == EBADF) ) break;
	250	abort( "accept error: (%d) %s\n", (int)errno, strerror(errno) );
	251	}
[137974ae]	252
[7f0ac12]	253	if(this.done) return;
[97748ee]	254
[7f0ac12]	255	if( options.log ) sout \| "=== New connection" \| fd \| "" \| ", waiting for requests ===";
[97748ee]	256
[07997cd]	257	if(fd) push_connection(this, fd);
[ee59ede]	258
[137974ae]	259	if (stats_thrd) {
	260	unsigned long long next = rdtscl();
[3f95dab]	261	if(next > (last + period)) {
	262	if(try_lock(stats_thrd->stats.lock)) {
	263	push(this.stats, stats_thrd->stats.accpt);
	264	unlock(stats_thrd->stats.lock);
	265	last = next;
	266	}
	267	}
	268	}
	269
	270	if( options.log ) sout \| "=== Accepting connection ===";
	271	}
	272
	273	#elif defined(ACCEPT_ONE)
	274	if( options.log ) sout \| "=== Accepting connection ===";
	275	for() {
	276	int fd = cfa_accept4(this.sockfd, this.[addr, addrlen, flags], 0);
	277	if(fd < 0) {
	278	if( errno == ECONNABORTED ) break;
	279	if( this.done && (errno == EINVAL \|\| errno == EBADF) ) break;
	280	abort( "accept error: (%d) %s\n", (int)errno, strerror(errno) );
	281	}
	282
	283	if(this.done) return;
	284
	285	if( options.log ) sout \| "=== New connection" \| fd \| "" \| ", waiting for requests ===";
	286
	287	if(fd) push_connection(this, fd);
	288
	289	if (stats_thrd) {
	290	unsigned long long next = rdtscl();
	291	if(next > (last + period)) {
[137974ae]	292	if(try_lock(stats_thrd->stats.lock)) {
	293	push(this.stats, stats_thrd->stats.accpt);
	294	unlock(stats_thrd->stats.lock);
	295	last = next;
	296	}
	297	}
	298	}
	299
[7f0ac12]	300	if( options.log ) sout \| "=== Accepting connection ===";
[0aec496]	301	}
[c25338d]	302
[07997cd]	303	#elif defined(ACCEPT_MANY)
[c25338d]	304	const int nacc = 10;
	305	io_future_t results[nacc];
	306
	307	for(i; nacc) {
	308	io_future_t & res = results[i];
	309	reset(res);
	310	/* paranoid */ assert(!available(res));
	311	if( options.log ) mutex(sout) sout \| "=== Re-arming accept no" \| i \| " ===";
	312	async_accept4(res, this.sockfd, this.[addr, addrlen, flags], CFA_IO_LAZY);
	313	}
	314
	315	for() {
	316	if (stats_thrd) {
	317	unsigned long long next = rdtscl();
[3f95dab]	318	if(next > (last + period)) {
[c25338d]	319	if(try_lock(stats_thrd->stats.lock __cfaabi_dbg_ctx2)) {
	320	push(this.stats, stats_thrd->stats.accpt);
	321	unlock(stats_thrd->stats.lock);
	322	last = next;
	323	}
	324	}
	325	}
	326
	327	for(i; nacc) {
	328	io_future_t & res = results[i];
	329	if(available(res)) {
	330	if( options.log ) mutex(sout) sout \| "=== Accept no " \| i \| "completed with result" \| res.result \| "===";
	331	int fd = get_res(res);
	332	reset(res);
	333	if(fd < 0) {
	334	if( errno == ECONNABORTED ) continue;
	335	if( this.done && (errno == EINVAL \|\| errno == EBADF) ) continue;
	336	abort( "accept error: (%d) %s\n", (int)errno, strerror(errno) );
	337	}
	338	push_connection( this, fd );
	339
	340	/* paranoid */ assert(!available(res));
	341	if( options.log ) mutex(sout) sout \| "=== Re-arming accept no" \| i \| " ===";
	342	async_accept4(res, this.sockfd, this.[addr, addrlen, flags], CFA_IO_LAZY);
	343	}
	344	}
	345	if(this.done) return;
	346
	347	if( options.log ) mutex(sout) sout \| "=== Waiting for any accept ===";
	348	this.stats.eagains++;
	349	wait_any(results, nacc);
	350
	351	if( options.log ) mutex(sout) {
	352	sout \| "=== Acceptor wake-up ===";
	353	for(i; nacc) {
	354	io_future_t & res = results[i];
	355	sout \| i \| "available:" \| available(res);
	356	}
	357	}
	358
	359	}
	360
	361	for(i; nacc) {
	362	wait(results[i]);
	363	}
	364	#else
	365	#error no accept algorithm specified
	366	#endif
[3f95dab]	367	lock(stats_thrd->stats.lock);
	368	push(this.stats, stats_thrd->stats.accpt);
	369	unlock(stats_thrd->stats.lock);
[7f0ac12]	370	}

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

Download in other formats:

Original Format