Context Navigation

cfathread.cfa @ 66b6f66

ADTast-experimental

Last change on this file since 66b6f66 was a0a949c, checked in by Peter A. Buhr <pabuhr@…>, 20 months ago
formatting, remove unnecessary #include files and code
Property mode set to `100644`
File size: 16.2 KB

Line
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
16	// #define EPOLL_FOR_SOCKETS
17
18	#include <string.h>
19
20	#include "fstream.hfa"
21	#include "locks.hfa"
22	#include "kernel.hfa"
23	#include "stats.hfa"
24	#include "thread.hfa"
25	#include "time.hfa"
26	#include "stdlib.hfa"
27	#include "iofwd.hfa"
28	#include "cfathread.h"
29
30	extern void ?{}(processor &, const char[], cluster &, thread$ *);
31	extern "C" {
32	extern void __cfactx_invoke_thread(void (main)(void ), void * this);
33	}
34
35	extern Time __kernel_get_time();
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;
128	unpark( thrd, UNPARK_REMOTE );
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
157	poller_cnt = 2;
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;
170	if (int ret = __cfaabi_pthread_attr_init(&attr); 0 != ret) {
171	abort \| "failed to create master epoll thread attr: " \| ret \| strerror(ret);
172	}
173
174	if (int ret = __cfaabi_pthread_create(&master_poller, &attr, master_epoll, 0p); 0 != ret) {
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
194	int id = prng() % poller_cnt;
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
219
220	//================================================================================
221	// Thread run by the C Interface
222
223	struct cfathread_object {
224	thread$ self;
225	void * (themain)( void );
226	void * arg;
227	void * ret;
228	};
229	void main(cfathread_object & this);
230	void ^?{}(cfathread_object & mutex this);
231
232	static inline thread$ * get_thread( cfathread_object & this ) { return &this.self; }
233
234	typedef ThreadCancelled(cfathread_object) cfathread_exception;
235	typedef vtable(ThreadCancelled(cfathread_object)) cfathread_vtable;
236
237	void defaultResumptionHandler(ThreadCancelled(cfathread_object) & except) {
238	abort \| "A thread was cancelled";
239	}
240
241	cfathread_vtable _cfathread_vtable_instance;
242
243	cfathread_vtable & const _default_vtable = _cfathread_vtable_instance;
244
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 ) {
250	this.themain = themain;
251	this.arg = arg;
252	(this.self){"C-thread", cl};
253	__thrd_start(this, main);
254	}
255
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 );
266	}
267
268	//================================================================================
269	// Special Init Thread responsible for the initialization or processors
270	struct __cfainit {
271	thread$ self;
272	void (init)( void );
273	void * arg;
274	};
275	void main(__cfainit & this);
276	void ^?{}(__cfainit & mutex this);
277
278	static inline thread$ * get_thread( __cfainit & this ) { return &this.self; }
279
280	typedef ThreadCancelled(__cfainit) __cfainit_exception;
281	typedef vtable(ThreadCancelled(__cfainit)) __cfainit_vtable;
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;
296	(this.self){"Processir Init"};
297
298	// Don't use __thrd_start! just prep the context manually
299	thread$ * this_thrd = get_thread(this);
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;
309	enable_interrupts();
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	}
323
324	#pragma GCC visibility push(default)
325
326	//================================================================================
327	// Main Api
328	extern "C" {
329	int cfathread_cluster_create(cfathread_cluster_t * cl) __attribute__((nonnull(1))) libcfa_public {
330	*cl = new();
331	return 0;
332	}
333
334	cfathread_cluster_t cfathread_cluster_self(void) libcfa_public {
335	return active_cluster();
336	}
337
338	int cfathread_cluster_print_stats( cfathread_cluster_t cl ) libcfa_public {
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
346	int cfathread_cluster_add_worker(cfathread_cluster_t cl, pthread_t* tid, void (init_routine) (void ), void * arg) {
347	__cfainit * it = 0p;
348	if(init_routine) {
349	it = alloc();
350	(*it){init_routine, arg};
351	}
352	processor * proc = alloc();
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
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
375	//--------------------
376	// Thread attributes
377	int cfathread_attr_init(cfathread_attr_t *attr) __attribute__((nonnull (1))) {
378	attr->cl = active_cluster();
379	return 0;
380	}
381
382	//--------------------
383	// Thread
384	int cfathread_create( cfathread_t * handle, const cfathread_attr_t * attr, void (main)( void * ), void * arg ) __attribute__((nonnull (1))) {
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;
394	^( *thrd ){};
395	free(thrd);
396	if(retval) {
397	*retval = ret;
398	}
399	return 0;
400	}
401
402	int cfathread_get_errno(void) {
403	return errno;
404	}
405
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;
418	}
419
420	void cfathread_park( void ) {
421	park();
422	}
423
424	void cfathread_unpark( cfathread_t thrd ) {
425	unpark( *thrd );
426	}
427
428	void cfathread_yield( void ) {
429	yield();
430	}
431
432	typedef struct cfathread_mutex * cfathread_mutex_t;
433
434	//--------------------
435	// Mutex
436	struct cfathread_mutex {
437	exp_backoff_then_block_lock impl;
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; }
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	}
448
449	//--------------------
450	// Condition
451	struct cfathread_condition {
452	condition_variable(exp_backoff_then_block_lock) impl;
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 };
459	timespec curr;
460	clock_gettime( CLOCK_REALTIME, &curr );
461	Time c = { curr };
462	if( wait( (cond)->impl, (mut)->impl, t - c ) ) {
463	return 0;
464	}
465	errno = ETIMEDOUT;
466	return ETIMEDOUT;
467	}
468	}
469
470	extern "C" {
471	//--------------------
472	// IO operations
473	int cfathread_socket(int domain, int type, int protocol) {
474	return socket(domain, type
475	#if defined(EPOLL_FOR_SOCKETS)
476	\| SOCK_NONBLOCK
477	#endif
478	, protocol);
479	}
480	int cfathread_bind(int socket, __CONST_SOCKADDR_ARG address, socklen_t address_len) {
481	return bind(socket, address, address_len);
482	}
483
484	int cfathread_listen(int socket, int backlog) {
485	return listen(socket, backlog);
486	}
487
488	int cfathread_accept(int socket, __SOCKADDR_ARG address, socklen_t *restrict address_len) {
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
503	}
504
505	int cfathread_connect(int socket, __CONST_SOCKADDR_ARG address, socklen_t address_len) {
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
519	}
520
521	int cfathread_dup(int fildes) {
522	return dup(fildes);
523	}
524
525	int cfathread_close(int fildes) {
526	return cfa_close(fildes, CFA_IO_LAZY);
527	}
528
529	ssize_t cfathread_sendmsg(int socket, const struct msghdr *message, int flags) {
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;
545	}
546
547	ssize_t cfathread_write(int fildes, const void *buf, size_t nbyte) {
548	// Use send rather then write for socket since it's faster
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;
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
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
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) {
598	// Use recv rather then read for socket since it's faster
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;
615	}
616
617	}

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

Download in other formats:

Original Format