Context Navigation

cfathread.cfa@ 0fa0201d

Visit:

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

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

Download in other formats:

Original Format