Context Navigation

cfathread.cfa@ 05ffb7b

Visit:

ADT ast-experimental enum pthread-emulation qualifiedEnum

Last change on this file since 05ffb7b was 75965a6, checked in by Peter A. Buhr <pabuhr@…>, 4 years ago
replace thread_rand with prng, replace xorshift64 with xorshift_13_7_17
Property mode set to `100644`
File size: 16.3 KB

Rev	Line
[bb662027]	1	//
	2	// Cforall Version 1.0.0 Copyright (C) 2016 University of Waterloo
	3	//
	4	// The contents of this file are covered under the licence agreement in the
	5	// file "LICENCE" distributed with Cforall.
	6	//
	7	// clib/cfathread.cfa --
	8	//
	9	// Author : Thierry Delisle
	10	// Created On : Tue Sep 22 15:31:20 2020
	11	// Last Modified By :
	12	// Last Modified On :
	13	// Update Count :
	14	//
	15
[bc4a433]	16	// #define EPOLL_FOR_SOCKETS
[4500e52]	17
[a1538cd]	18	#include "fstream.hfa"
	19	#include "locks.hfa"
[bb662027]	20	#include "kernel.hfa"
[d971c8d]	21	#include "stats.hfa"
[bb662027]	22	#include "thread.hfa"
[a1538cd]	23	#include "time.hfa"
[75965a6]	24	#include "stdlib.hfa"
[bb662027]	25
[a1538cd]	26	#include "cfathread.h"
	27
[4500e52]	28	extern "C" {
	29	#include <string.h>
	30	#include <errno.h>
	31	}
	32
[e84ab3d]	33	extern void ?{}(processor &, const char[], cluster &, thread$ *);
[a5e7233]	34	extern "C" {
	35	extern void __cfactx_invoke_thread(void (main)(void ), void * this);
[4500e52]	36	extern int accept4(int sockfd, struct sockaddr addr, socklen_t addrlen, int flags);
[a5e7233]	37	}
	38
[c457dc41]	39	extern Time __kernel_get_time();
[4500e52]	40	extern unsigned register_proc_id( void );
	41
	42	//================================================================================
	43	// Epoll support for sockets
	44
	45	#if defined(EPOLL_FOR_SOCKETS)
	46	extern "C" {
	47	#include <sys/epoll.h>
	48	#include <sys/resource.h>
	49	}
	50
	51	static pthread_t master_poller;
	52	static int master_epollfd = 0;
	53	static size_t poller_cnt = 0;
	54	static int * poller_fds = 0p;
	55	static struct leaf_poller * pollers = 0p;
	56
	57	struct __attribute__((aligned)) fd_info_t {
	58	int pollid;
	59	size_t rearms;
	60	};
	61	rlim_t fd_limit = 0;
	62	static fd_info_t * volatile * fd_map = 0p;
	63
	64	void * master_epoll( __attribute__((unused)) void * args ) {
	65	unsigned id = register_proc_id();
	66
	67	enum { MAX_EVENTS = 5 };
	68	struct epoll_event events[MAX_EVENTS];
	69	for() {
	70	int ret = epoll_wait(master_epollfd, events, MAX_EVENTS, -1);
	71	if ( ret < 0 ) {
	72	abort \| "Master epoll error: " \| strerror(errno);
	73	}
	74
	75	for(i; ret) {
	76	thread$ * thrd = (thread$ *)events[i].data.u64;
	77	unpark( thrd );
	78	}
	79	}
	80
	81	return 0p;
	82	}
	83
	84	static inline int epoll_rearm(int epollfd, int fd, uint32_t event) {
	85	struct epoll_event eevent;
	86	eevent.events = event \| EPOLLET \| EPOLLONESHOT;
	87	eevent.data.u64 = (uint64_t)active_thread();
	88
	89	if(0 != epoll_ctl(epollfd, EPOLL_CTL_MOD, fd, &eevent))
	90	{
	91	if(errno == ENOENT) return -1;
	92	abort \| acquire \| "epoll" \| epollfd \| "ctl rearm" \| fd \| "error: " \| errno \| strerror(errno);
	93	}
	94
	95	park();
	96	return 0;
	97	}
	98
	99	thread leaf_poller {
	100	int epollfd;
	101	};
	102
	103	void ?{}(leaf_poller & this, int fd) { this.epollfd = fd; }
	104
	105	void main(leaf_poller & this) {
	106	enum { MAX_EVENTS = 1024 };
	107	struct epoll_event events[MAX_EVENTS];
	108	const int max_retries = 5;
	109	int retries = max_retries;
	110
	111	struct epoll_event event;
	112	event.events = EPOLLIN \| EPOLLET \| EPOLLONESHOT;
	113	event.data.u64 = (uint64_t)&(thread&)this;
	114
	115	if(0 != epoll_ctl(master_epollfd, EPOLL_CTL_ADD, this.epollfd, &event))
	116	{
	117	abort \| "master epoll ctl add leaf: " \| errno \| strerror(errno);
	118	}
	119
	120	park();
	121
	122	for() {
	123	yield();
	124	int ret = epoll_wait(this.epollfd, events, MAX_EVENTS, 0);
	125	if ( ret < 0 ) {
	126	abort \| "Leaf epoll error: " \| errno \| strerror(errno);
	127	}
	128
	129	if(ret) {
	130	for(i; ret) {
	131	thread$ * thrd = (thread$ *)events[i].data.u64;
[75c7252]	132	unpark( thrd, UNPARK_REMOTE );
[4500e52]	133	}
	134	}
	135	else if(0 >= --retries) {
	136	epoll_rearm(master_epollfd, this.epollfd, EPOLLIN);
	137	}
	138	}
	139	}
	140
	141	void setup_epoll( void ) __attribute__(( constructor ));
	142	void setup_epoll( void ) {
	143	if(master_epollfd) abort \| "Master epoll already setup";
	144
	145	master_epollfd = epoll_create1(0);
	146	if(master_epollfd == -1) {
	147	abort \| "failed to create master epoll: " \| errno \| strerror(errno);
	148	}
	149
	150	struct rlimit rlim;
	151	if(int ret = getrlimit(RLIMIT_NOFILE, &rlim); 0 != ret) {
	152	abort \| "failed to get nofile limit: " \| errno \| strerror(errno);
	153	}
	154
	155	fd_limit = rlim.rlim_cur;
	156	fd_map = alloc(fd_limit);
	157	for(i;fd_limit) {
	158	fd_map[i] = 0p;
	159	}
	160
[75c7252]	161	poller_cnt = 2;
[4500e52]	162	poller_fds = alloc(poller_cnt);
	163	pollers = alloc(poller_cnt);
	164	for(i; poller_cnt) {
	165	poller_fds[i] = epoll_create1(0);
	166	if(poller_fds[i] == -1) {
	167	abort \| "failed to create leaf epoll [" \| i \| "]: " \| errno \| strerror(errno);
	168	}
	169
	170	(pollers[i]){ poller_fds[i] };
	171	}
	172
	173	pthread_attr_t attr;
	174	if (int ret = pthread_attr_init(&attr); 0 != ret) {
	175	abort \| "failed to create master epoll thread attr: " \| ret \| strerror(ret);
	176	}
	177
	178	if (int ret = pthread_create(&master_poller, &attr, master_epoll, 0p); 0 != ret) {
	179	abort \| "failed to create master epoll thread: " \| ret \| strerror(ret);
	180	}
	181	}
	182
	183	static inline int epoll_wait(int fd, uint32_t event) {
	184	if(fd_map[fd] >= 1p) {
	185	fd_map[fd]->rearms++;
	186	epoll_rearm(poller_fds[fd_map[fd]->pollid], fd, event);
	187	return 0;
	188	}
	189
	190	for() {
	191	fd_info_t * expected = 0p;
	192	fd_info_t * sentinel = 1p;
	193	if(__atomic_compare_exchange_n( &(fd_map[fd]), &expected, sentinel, true, __ATOMIC_SEQ_CST, __ATOMIC_RELAXED)) {
	194	struct epoll_event eevent;
	195	eevent.events = event \| EPOLLET \| EPOLLONESHOT;
	196	eevent.data.u64 = (uint64_t)active_thread();
	197
[75965a6]	198	int id = prng() % poller_cnt;
[4500e52]	199	if(0 != epoll_ctl(poller_fds[id], EPOLL_CTL_ADD, fd, &eevent))
	200	{
	201	abort \| "epoll ctl add" \| poller_fds[id] \| fd \| fd_map[fd] \| expected \| "error: " \| errno \| strerror(errno);
	202	}
	203
	204	fd_info_t * ninfo = alloc();
	205	ninfo->pollid = id;
	206	ninfo->rearms = 0;
	207	__atomic_store_n( &fd_map[fd], ninfo, __ATOMIC_SEQ_CST);
	208
	209	park();
	210	return 0;
	211	}
	212
	213	if(expected >= 0) {
	214	fd_map[fd]->rearms++;
	215	epoll_rearm(poller_fds[fd_map[fd]->pollid], fd, event);
	216	return 0;
	217	}
	218
	219	Pause();
	220	}
	221	}
	222	#endif
[c457dc41]	223
[a5e7233]	224	//================================================================================
[4500e52]	225	// Thread run by the C Interface
[a5e7233]	226
[a1538cd]	227	struct cfathread_object {
[e84ab3d]	228	thread$ self;
[a1538cd]	229	void * (themain)( void );
	230	void * arg;
	231	void * ret;
[bb662027]	232	};
[a1538cd]	233	void main(cfathread_object & this);
	234	void ^?{}(cfathread_object & mutex this);
	235
[e84ab3d]	236	static inline thread$ * get_thread( cfathread_object & this ) { return &this.self; }
[a1538cd]	237
	238	typedef ThreadCancelled(cfathread_object) cfathread_exception;
	239	typedef ThreadCancelled_vtable(cfathread_object) cfathread_vtable;
	240
	241	void defaultResumptionHandler(ThreadCancelled(cfathread_object) & except) {
	242	abort \| "A thread was cancelled";
	243	}
[bb662027]	244
[a1538cd]	245	cfathread_vtable _cfathread_vtable_instance;
	246
[8edbe40]	247	cfathread_vtable & const _default_vtable = _cfathread_vtable_instance;
	248
[a1538cd]	249	cfathread_vtable const & get_exception_vtable(cfathread_exception *) {
	250	return _cfathread_vtable_instance;
	251	}
	252
	253	static void ?{}( cfathread_object & this, cluster & cl, void (themain)( void * ), void * arg ) {
[bb662027]	254	this.themain = themain;
[a1538cd]	255	this.arg = arg;
[4150779]	256	(this.self){"C-thread", cl};
[a1538cd]	257	__thrd_start(this, main);
[bb662027]	258	}
	259
[a1538cd]	260	void ^?{}(cfathread_object & mutex this) {
	261	^(this.self){};
	262	}
	263
	264	void main( cfathread_object & this ) {
	265	__attribute__((unused)) void * const thrd_obj = (void*)&this;
	266	__attribute__((unused)) void * const thrd_hdl = (void*)active_thread();
	267	/* paranoid */ verify( thrd_obj == thrd_hdl );
	268
	269	this.ret = this.themain( this.arg );
[bb662027]	270	}
	271
[a5e7233]	272	//================================================================================
	273	// Special Init Thread responsible for the initialization or processors
	274	struct __cfainit {
[e84ab3d]	275	thread$ self;
[a5e7233]	276	void (init)( void );
	277	void * arg;
	278	};
	279	void main(__cfainit & this);
	280	void ^?{}(__cfainit & mutex this);
	281
[e84ab3d]	282	static inline thread$ * get_thread( __cfainit & this ) { return &this.self; }
[a5e7233]	283
	284	typedef ThreadCancelled(__cfainit) __cfainit_exception;
	285	typedef ThreadCancelled_vtable(__cfainit) __cfainit_vtable;
	286
	287	void defaultResumptionHandler(ThreadCancelled(__cfainit) & except) {
	288	abort \| "The init thread was cancelled";
	289	}
	290
	291	__cfainit_vtable ___cfainit_vtable_instance;
	292
	293	__cfainit_vtable const & get_exception_vtable(__cfainit_exception *) {
	294	return ___cfainit_vtable_instance;
	295	}
	296
	297	static void ?{}( __cfainit & this, void (init)( void ), void * arg ) {
	298	this.init = init;
	299	this.arg = arg;
[4150779]	300	(this.self){"Processir Init"};
[a5e7233]	301
	302	// Don't use __thrd_start! just prep the context manually
[e84ab3d]	303	thread$ * this_thrd = get_thread(this);
[a5e7233]	304	void (*main_p)(__cfainit &) = main;
	305
	306	disable_interrupts();
	307	__cfactx_start(main_p, get_coroutine(this), this, __cfactx_invoke_thread);
	308
	309	this_thrd->context.[SP, FP] = this_thrd->self_cor.context.[SP, FP];
	310	/* paranoid */ verify( this_thrd->context.SP );
	311
	312	this_thrd->state = Ready;
[a3821fa]	313	enable_interrupts();
[a5e7233]	314	}
	315
	316	void ^?{}(__cfainit & mutex this) {
	317	^(this.self){};
	318	}
	319
	320	void main( __cfainit & this ) {
	321	__attribute__((unused)) void * const thrd_obj = (void*)&this;
	322	__attribute__((unused)) void * const thrd_hdl = (void*)active_thread();
	323	/* paranoid */ verify( thrd_obj == thrd_hdl );
	324
	325	this.init( this.arg );
	326	}
[bb662027]	327
[a5e7233]	328	//================================================================================
	329	// Main Api
[bb662027]	330	extern "C" {
[a1538cd]	331	int cfathread_cluster_create(cfathread_cluster_t * cl) __attribute__((nonnull(1))) {
	332	*cl = new();
	333	return 0;
	334	}
	335
	336	cfathread_cluster_t cfathread_cluster_self(void) {
	337	return active_cluster();
	338	}
	339
[d971c8d]	340	int cfathread_cluster_print_stats( cfathread_cluster_t cl ) {
	341	#if !defined(__CFA_NO_STATISTICS__)
	342	print_stats_at_exit( *cl, CFA_STATS_READY_Q \| CFA_STATS_IO );
	343	print_stats_now( *cl, CFA_STATS_READY_Q \| CFA_STATS_IO );
	344	#endif
	345	return 0;
	346	}
	347
[a1538cd]	348	int cfathread_cluster_add_worker(cfathread_cluster_t cl, pthread_t* tid, void (init_routine) (void ), void * arg) {
[a5e7233]	349	__cfainit * it = 0p;
	350	if(init_routine) {
	351	it = alloc();
	352	(*it){init_routine, arg};
	353	}
[a1538cd]	354	processor * proc = alloc();
[a5e7233]	355	(proc){ "C-processor", cl, get_thread(*it) };
	356
	357	// Wait for the init thread to return before continuing
	358	if(it) {
	359	^(*it){};
	360	free(it);
	361	}
	362
[a1538cd]	363	if(tid) *tid = proc->kernel_thread;
	364	return 0;
	365	}
	366
	367	int cfathread_cluster_pause (cfathread_cluster_t) {
	368	abort \| "Pausing clusters is not supported";
	369	exit(1);
	370	}
	371
	372	int cfathread_cluster_resume(cfathread_cluster_t) {
	373	abort \| "Resuming clusters is not supported";
	374	exit(1);
	375	}
	376
[bb662027]	377	//--------------------
[a1538cd]	378	// Thread attributes
	379	int cfathread_attr_init(cfathread_attr_t *attr) __attribute__((nonnull (1))) {
	380	attr->cl = active_cluster();
	381	return 0;
[bb662027]	382	}
	383
[a1538cd]	384	//--------------------
	385	// Thread
[9e27f69]	386	int cfathread_create( cfathread_t * handle, const cfathread_attr_t * attr, void (main)( void * ), void * arg ) __attribute__((nonnull (1))) {
[a1538cd]	387	cluster * cl = attr ? attr->cl : active_cluster();
	388	cfathread_t thrd = alloc();
	389	(thrd){ cl, main, arg };
	390	*handle = thrd;
	391	return 0;
	392	}
	393
	394	int cfathread_join( cfathread_t thrd, void ** retval ) {
	395	void * ret = join( *thrd ).ret;
[f03e11d]	396	^( *thrd ){};
	397	free(thrd);
[a1538cd]	398	if(retval) {
	399	*retval = ret;
	400	}
	401	return 0;
	402	}
	403
[9e27f69]	404	int cfathread_get_errno(void) {
	405	return errno;
	406	}
	407
[a1538cd]	408	cfathread_t cfathread_self(void) {
	409	return (cfathread_t)active_thread();
	410	}
	411
	412	int cfathread_usleep(useconds_t usecs) {
	413	sleep(usecs`us);
	414	return 0;
	415	}
	416
	417	int cfathread_sleep(unsigned int secs) {
	418	sleep(secs`s);
	419	return 0;
[bb662027]	420	}
	421
	422	void cfathread_park( void ) {
[e235429]	423	park();
[bb662027]	424	}
	425
[a1538cd]	426	void cfathread_unpark( cfathread_t thrd ) {
[e235429]	427	unpark( *thrd );
[bb662027]	428	}
	429
	430	void cfathread_yield( void ) {
	431	yield();
	432	}
	433
[a1538cd]	434	typedef struct cfathread_mutex * cfathread_mutex_t;
	435
	436	//--------------------
	437	// Mutex
	438	struct cfathread_mutex {
[420b498]	439	linear_backoff_then_block_lock impl;
[a1538cd]	440	};
	441	int cfathread_mutex_init(cfathread_mutex_t restrict mut, const cfathread_mutexattr_t restrict) __attribute__((nonnull (1))) { *mut = new(); return 0; }
	442	int cfathread_mutex_destroy(cfathread_mutex_t mut) __attribute__((nonnull (1))) { delete( mut ); return 0; }
[d27b6be]	443	int cfathread_mutex_lock (cfathread_mutex_t mut) __attribute__((nonnull (1))) { lock( (mut)->impl ); return 0; }
	444	int cfathread_mutex_unlock (cfathread_mutex_t mut) __attribute__((nonnull (1))) { unlock( (mut)->impl ); return 0; }
	445	int cfathread_mutex_trylock(cfathread_mutex_t *mut) __attribute__((nonnull (1))) {
	446	bool ret = try_lock( (*mut)->impl );
	447	if( ret ) return 0;
	448	else return EBUSY;
	449	}
[a1538cd]	450
	451	//--------------------
	452	// Condition
	453	struct cfathread_condition {
[420b498]	454	condition_variable(linear_backoff_then_block_lock) impl;
[a1538cd]	455	};
	456	int cfathread_cond_init(cfathread_cond_t restrict cond, const cfathread_condattr_t restrict) __attribute__((nonnull (1))) { *cond = new(); return 0; }
	457	int cfathread_cond_signal(cfathread_cond_t cond) __attribute__((nonnull (1))) { notify_one( (cond)->impl ); return 0; }
	458	int cfathread_cond_wait(cfathread_cond_t restrict cond, cfathread_mutex_t restrict mut) __attribute__((nonnull (1,2))) { wait( (cond)->impl, (mut)->impl ); return 0; }
	459	int cfathread_cond_timedwait(cfathread_cond_t restrict cond, cfathread_mutex_t restrict mut, const struct timespec *restrict abstime) __attribute__((nonnull (1,2,3))) {
	460	Time t = { *abstime };
[c457dc41]	461	timespec curr;
	462	clock_gettime( CLOCK_REALTIME, &curr );
	463	Time c = { curr };
	464	if( wait( (cond)->impl, (mut)->impl, t - c ) ) {
[a1538cd]	465	return 0;
	466	}
	467	errno = ETIMEDOUT;
	468	return ETIMEDOUT;
	469	}
	470	}
	471
	472	#include <iofwd.hfa>
	473
	474	extern "C" {
	475	#include <unistd.h>
	476	#include <sys/types.h>
	477	#include <sys/socket.h>
	478
[bb662027]	479	//--------------------
[a1538cd]	480	// IO operations
	481	int cfathread_socket(int domain, int type, int protocol) {
[4500e52]	482	return socket(domain, type
	483	#if defined(EPOLL_FOR_SOCKETS)
	484	\| SOCK_NONBLOCK
	485	#endif
	486	, protocol);
[a1538cd]	487	}
	488	int cfathread_bind(int socket, const struct sockaddr *address, socklen_t address_len) {
	489	return bind(socket, address, address_len);
	490	}
	491
	492	int cfathread_listen(int socket, int backlog) {
	493	return listen(socket, backlog);
	494	}
	495
	496	int cfathread_accept(int socket, struct sockaddr restrict address, socklen_t restrict address_len) {
[4500e52]	497	#if defined(EPOLL_FOR_SOCKETS)
	498	int ret;
	499	for() {
	500	yield();
	501	ret = accept4(socket, address, address_len, SOCK_NONBLOCK);
	502	if(ret >= 0) break;
	503	if(errno != EAGAIN && errno != EWOULDBLOCK) break;
	504
	505	epoll_wait(socket, EPOLLIN);
	506	}
	507	return ret;
	508	#else
	509	return cfa_accept4(socket, address, address_len, 0, CFA_IO_LAZY);
	510	#endif
[a1538cd]	511	}
	512
	513	int cfathread_connect(int socket, const struct sockaddr *address, socklen_t address_len) {
[4500e52]	514	#if defined(EPOLL_FOR_SOCKETS)
	515	int ret;
	516	for() {
	517	ret = connect(socket, address, address_len);
	518	if(ret >= 0) break;
	519	if(errno != EAGAIN && errno != EWOULDBLOCK) break;
	520
	521	epoll_wait(socket, EPOLLIN);
	522	}
	523	return ret;
	524	#else
	525	return cfa_connect(socket, address, address_len, CFA_IO_LAZY);
	526	#endif
[a1538cd]	527	}
	528
	529	int cfathread_dup(int fildes) {
	530	return dup(fildes);
	531	}
[bb662027]	532
[a1538cd]	533	int cfathread_close(int fildes) {
	534	return cfa_close(fildes, CFA_IO_LAZY);
[bb662027]	535	}
[a1538cd]	536
	537	ssize_t cfathread_sendmsg(int socket, const struct msghdr *message, int flags) {
[4500e52]	538	#if defined(EPOLL_FOR_SOCKETS)
	539	ssize_t ret;
	540	__STATS__( false, io.ops.sockwrite++; )
	541	for() {
	542	ret = sendmsg(socket, message, flags);
	543	if(ret >= 0) break;
	544	if(errno != EAGAIN && errno != EWOULDBLOCK) break;
	545
	546	__STATS__( false, io.ops.epllwrite++; )
	547	epoll_wait(socket, EPOLLOUT);
	548	}
	549	#else
	550	ssize_t ret = cfa_sendmsg(socket, message, flags, CFA_IO_LAZY);
	551	#endif
	552	return ret;
[a1538cd]	553	}
	554
	555	ssize_t cfathread_write(int fildes, const void *buf, size_t nbyte) {
[86dc95d]	556	// Use send rather then write for socket since it's faster
[4500e52]	557	#if defined(EPOLL_FOR_SOCKETS)
	558	ssize_t ret;
	559	// __STATS__( false, io.ops.sockwrite++; )
	560	for() {
	561	ret = send(fildes, buf, nbyte, 0);
	562	if(ret >= 0) break;
	563	if(errno != EAGAIN && errno != EWOULDBLOCK) break;
	564
	565	// __STATS__( false, io.ops.epllwrite++; )
	566	epoll_wait(fildes, EPOLLOUT);
	567	}
	568	#else
	569	ssize_t ret = cfa_send(fildes, buf, nbyte, 0, CFA_IO_LAZY);
	570	#endif
	571	return ret;
[a1538cd]	572	}
	573
	574	ssize_t cfathread_recvfrom(int socket, void restrict buffer, size_t length, int flags, struct sockaddr restrict address, socklen_t *restrict address_len) {
	575	struct iovec iov;
	576	iov.iov_base = buffer;
	577	iov.iov_len = length;
	578
	579	struct msghdr msg;
	580	msg.msg_name = address;
	581	msg.msg_namelen = address_len ? (socklen_t)*address_len : (socklen_t)0;
	582	msg.msg_iov = &iov;
	583	msg.msg_iovlen = 1;
	584	msg.msg_control = 0p;
	585	msg.msg_controllen = 0;
	586
[4500e52]	587	#if defined(EPOLL_FOR_SOCKETS)
	588	ssize_t ret;
	589	yield();
	590	for() {
	591	ret = recvmsg(socket, &msg, flags);
	592	if(ret >= 0) break;
	593	if(errno != EAGAIN && errno != EWOULDBLOCK) break;
	594
	595	epoll_wait(socket, EPOLLIN);
	596	}
	597	#else
	598	ssize_t ret = cfa_recvmsg(socket, &msg, flags, CFA_IO_LAZY);
	599	#endif
[a1538cd]	600
	601	if(address_len) *address_len = msg.msg_namelen;
	602	return ret;
	603	}
	604
	605	ssize_t cfathread_read(int fildes, void *buf, size_t nbyte) {
[86dc95d]	606	// Use recv rather then read for socket since it's faster
[4500e52]	607	#if defined(EPOLL_FOR_SOCKETS)
	608	ssize_t ret;
	609	__STATS__( false, io.ops.sockread++; )
	610	yield();
	611	for() {
	612	ret = recv(fildes, buf, nbyte, 0);
	613	if(ret >= 0) break;
	614	if(errno != EAGAIN && errno != EWOULDBLOCK) break;
	615
	616	__STATS__( false, io.ops.epllread++; )
	617	epoll_wait(fildes, EPOLLIN);
	618	}
	619	#else
	620	ssize_t ret = cfa_recv(fildes, buf, nbyte, 0, CFA_IO_LAZY);
	621	#endif
	622	return ret;
[a1538cd]	623	}
	624
[45444c3]	625	}

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

Download in other formats:

Original Format