Context Navigation

cfathread.cfa @ 29702ad

ADTast-experimental

Last change on this file since 29702ad was 95dab9e, checked in by Thierry Delisle <tdelisle@…>, 2 years ago
Changed real_pthread symbols (now cfaabi_pthread) to be protected in libcfathread
Property mode set to `100644`
File size: 16.4 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 "fstream.hfa"
19	#include "locks.hfa"
20	#include "kernel.hfa"
21	#include "stats.hfa"
22	#include "thread.hfa"
23	#include "time.hfa"
24	#include "stdlib.hfa"
25
26	#include "cfathread.h"
27
28	extern "C" {
29	#include <string.h>
30	#include <errno.h>
31	}
32
33	extern void ?{}(processor &, const char[], cluster &, thread$ *);
34	extern "C" {
35	extern void __cfactx_invoke_thread(void (main)(void ), void * this);
36	extern int accept4(int sockfd, struct sockaddr addr, socklen_t addrlen, int flags);
37	}
38
39	extern Time __kernel_get_time();
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;
132	unpark( thrd, UNPARK_REMOTE );
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
161	poller_cnt = 2;
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 = __cfaabi_pthread_attr_init(&attr); 0 != ret) {
175	abort \| "failed to create master epoll thread attr: " \| ret \| strerror(ret);
176	}
177
178	if (int ret = __cfaabi_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
198	int id = prng() % poller_cnt;
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
223
224	//================================================================================
225	// Thread run by the C Interface
226
227	struct cfathread_object {
228	thread$ self;
229	void * (themain)( void );
230	void * arg;
231	void * ret;
232	};
233	void main(cfathread_object & this);
234	void ^?{}(cfathread_object & mutex this);
235
236	static inline thread$ * get_thread( cfathread_object & this ) { return &this.self; }
237
238	typedef ThreadCancelled(cfathread_object) cfathread_exception;
239	typedef vtable(ThreadCancelled(cfathread_object)) cfathread_vtable;
240
241	void defaultResumptionHandler(ThreadCancelled(cfathread_object) & except) {
242	abort \| "A thread was cancelled";
243	}
244
245	cfathread_vtable _cfathread_vtable_instance;
246
247	cfathread_vtable & const _default_vtable = _cfathread_vtable_instance;
248
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 ) {
254	this.themain = themain;
255	this.arg = arg;
256	(this.self){"C-thread", cl};
257	__thrd_start(this, main);
258	}
259
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 );
270	}
271
272	//================================================================================
273	// Special Init Thread responsible for the initialization or processors
274	struct __cfainit {
275	thread$ self;
276	void (init)( void );
277	void * arg;
278	};
279	void main(__cfainit & this);
280	void ^?{}(__cfainit & mutex this);
281
282	static inline thread$ * get_thread( __cfainit & this ) { return &this.self; }
283
284	typedef ThreadCancelled(__cfainit) __cfainit_exception;
285	typedef vtable(ThreadCancelled(__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;
300	(this.self){"Processir Init"};
301
302	// Don't use __thrd_start! just prep the context manually
303	thread$ * this_thrd = get_thread(this);
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;
313	enable_interrupts();
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	}
327
328	#pragma GCC visibility push(default)
329
330	//================================================================================
331	// Main Api
332	extern "C" {
333	int cfathread_cluster_create(cfathread_cluster_t * cl) __attribute__((nonnull(1))) libcfa_public {
334	*cl = new();
335	return 0;
336	}
337
338	cfathread_cluster_t cfathread_cluster_self(void) libcfa_public {
339	return active_cluster();
340	}
341
342	int cfathread_cluster_print_stats( cfathread_cluster_t cl ) libcfa_public {
343	#if !defined(__CFA_NO_STATISTICS__)
344	print_stats_at_exit( *cl, CFA_STATS_READY_Q \| CFA_STATS_IO );
345	print_stats_now( *cl, CFA_STATS_READY_Q \| CFA_STATS_IO );
346	#endif
347	return 0;
348	}
349
350	int cfathread_cluster_add_worker(cfathread_cluster_t cl, pthread_t* tid, void (init_routine) (void ), void * arg) {
351	__cfainit * it = 0p;
352	if(init_routine) {
353	it = alloc();
354	(*it){init_routine, arg};
355	}
356	processor * proc = alloc();
357	(proc){ "C-processor", cl, get_thread(*it) };
358
359	// Wait for the init thread to return before continuing
360	if(it) {
361	^(*it){};
362	free(it);
363	}
364
365	if(tid) *tid = proc->kernel_thread;
366	return 0;
367	}
368
369	int cfathread_cluster_pause (cfathread_cluster_t) {
370	abort \| "Pausing clusters is not supported";
371	exit(1);
372	}
373
374	int cfathread_cluster_resume(cfathread_cluster_t) {
375	abort \| "Resuming clusters is not supported";
376	exit(1);
377	}
378
379	//--------------------
380	// Thread attributes
381	int cfathread_attr_init(cfathread_attr_t *attr) __attribute__((nonnull (1))) {
382	attr->cl = active_cluster();
383	return 0;
384	}
385
386	//--------------------
387	// Thread
388	int cfathread_create( cfathread_t * handle, const cfathread_attr_t * attr, void (main)( void * ), void * arg ) __attribute__((nonnull (1))) {
389	cluster * cl = attr ? attr->cl : active_cluster();
390	cfathread_t thrd = alloc();
391	(thrd){ cl, main, arg };
392	*handle = thrd;
393	return 0;
394	}
395
396	int cfathread_join( cfathread_t thrd, void ** retval ) {
397	void * ret = join( *thrd ).ret;
398	^( *thrd ){};
399	free(thrd);
400	if(retval) {
401	*retval = ret;
402	}
403	return 0;
404	}
405
406	int cfathread_get_errno(void) {
407	return errno;
408	}
409
410	cfathread_t cfathread_self(void) {
411	return (cfathread_t)active_thread();
412	}
413
414	int cfathread_usleep(useconds_t usecs) {
415	sleep(usecs`us);
416	return 0;
417	}
418
419	int cfathread_sleep(unsigned int secs) {
420	sleep(secs`s);
421	return 0;
422	}
423
424	void cfathread_park( void ) {
425	park();
426	}
427
428	void cfathread_unpark( cfathread_t thrd ) {
429	unpark( *thrd );
430	}
431
432	void cfathread_yield( void ) {
433	yield();
434	}
435
436	typedef struct cfathread_mutex * cfathread_mutex_t;
437
438	//--------------------
439	// Mutex
440	struct cfathread_mutex {
441	linear_backoff_then_block_lock impl;
442	};
443	int cfathread_mutex_init(cfathread_mutex_t restrict mut, const cfathread_mutexattr_t restrict) __attribute__((nonnull (1))) { *mut = new(); return 0; }
444	int cfathread_mutex_destroy(cfathread_mutex_t mut) __attribute__((nonnull (1))) { delete( mut ); return 0; }
445	int cfathread_mutex_lock (cfathread_mutex_t mut) __attribute__((nonnull (1))) { lock( (mut)->impl ); return 0; }
446	int cfathread_mutex_unlock (cfathread_mutex_t mut) __attribute__((nonnull (1))) { unlock( (mut)->impl ); return 0; }
447	int cfathread_mutex_trylock(cfathread_mutex_t *mut) __attribute__((nonnull (1))) {
448	bool ret = try_lock( (*mut)->impl );
449	if( ret ) return 0;
450	else return EBUSY;
451	}
452
453	//--------------------
454	// Condition
455	struct cfathread_condition {
456	condition_variable(linear_backoff_then_block_lock) impl;
457	};
458	int cfathread_cond_init(cfathread_cond_t restrict cond, const cfathread_condattr_t restrict) __attribute__((nonnull (1))) { *cond = new(); return 0; }
459	int cfathread_cond_signal(cfathread_cond_t cond) __attribute__((nonnull (1))) { notify_one( (cond)->impl ); return 0; }
460	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; }
461	int cfathread_cond_timedwait(cfathread_cond_t restrict cond, cfathread_mutex_t restrict mut, const struct timespec *restrict abstime) __attribute__((nonnull (1,2,3))) {
462	Time t = { *abstime };
463	timespec curr;
464	clock_gettime( CLOCK_REALTIME, &curr );
465	Time c = { curr };
466	if( wait( (cond)->impl, (mut)->impl, t - c ) ) {
467	return 0;
468	}
469	errno = ETIMEDOUT;
470	return ETIMEDOUT;
471	}
472	}
473
474	#include <iofwd.hfa>
475
476	extern "C" {
477	#include <unistd.h>
478	#include <sys/types.h>
479	#include <sys/socket.h>
480
481	//--------------------
482	// IO operations
483	int cfathread_socket(int domain, int type, int protocol) {
484	return socket(domain, type
485	#if defined(EPOLL_FOR_SOCKETS)
486	\| SOCK_NONBLOCK
487	#endif
488	, protocol);
489	}
490	int cfathread_bind(int socket, const struct sockaddr *address, socklen_t address_len) {
491	return bind(socket, address, address_len);
492	}
493
494	int cfathread_listen(int socket, int backlog) {
495	return listen(socket, backlog);
496	}
497
498	int cfathread_accept(int socket, struct sockaddr restrict address, socklen_t restrict address_len) {
499	#if defined(EPOLL_FOR_SOCKETS)
500	int ret;
501	for() {
502	yield();
503	ret = accept4(socket, address, address_len, SOCK_NONBLOCK);
504	if(ret >= 0) break;
505	if(errno != EAGAIN && errno != EWOULDBLOCK) break;
506
507	epoll_wait(socket, EPOLLIN);
508	}
509	return ret;
510	#else
511	return cfa_accept4(socket, address, address_len, 0, CFA_IO_LAZY);
512	#endif
513	}
514
515	int cfathread_connect(int socket, const struct sockaddr *address, socklen_t address_len) {
516	#if defined(EPOLL_FOR_SOCKETS)
517	int ret;
518	for() {
519	ret = connect(socket, address, address_len);
520	if(ret >= 0) break;
521	if(errno != EAGAIN && errno != EWOULDBLOCK) break;
522
523	epoll_wait(socket, EPOLLIN);
524	}
525	return ret;
526	#else
527	return cfa_connect(socket, address, address_len, CFA_IO_LAZY);
528	#endif
529	}
530
531	int cfathread_dup(int fildes) {
532	return dup(fildes);
533	}
534
535	int cfathread_close(int fildes) {
536	return cfa_close(fildes, CFA_IO_LAZY);
537	}
538
539	ssize_t cfathread_sendmsg(int socket, const struct msghdr *message, int flags) {
540	#if defined(EPOLL_FOR_SOCKETS)
541	ssize_t ret;
542	__STATS__( false, io.ops.sockwrite++; )
543	for() {
544	ret = sendmsg(socket, message, flags);
545	if(ret >= 0) break;
546	if(errno != EAGAIN && errno != EWOULDBLOCK) break;
547
548	__STATS__( false, io.ops.epllwrite++; )
549	epoll_wait(socket, EPOLLOUT);
550	}
551	#else
552	ssize_t ret = cfa_sendmsg(socket, message, flags, CFA_IO_LAZY);
553	#endif
554	return ret;
555	}
556
557	ssize_t cfathread_write(int fildes, const void *buf, size_t nbyte) {
558	// Use send rather then write for socket since it's faster
559	#if defined(EPOLL_FOR_SOCKETS)
560	ssize_t ret;
561	// __STATS__( false, io.ops.sockwrite++; )
562	for() {
563	ret = send(fildes, buf, nbyte, 0);
564	if(ret >= 0) break;
565	if(errno != EAGAIN && errno != EWOULDBLOCK) break;
566
567	// __STATS__( false, io.ops.epllwrite++; )
568	epoll_wait(fildes, EPOLLOUT);
569	}
570	#else
571	ssize_t ret = cfa_send(fildes, buf, nbyte, 0, CFA_IO_LAZY);
572	#endif
573	return ret;
574	}
575
576	ssize_t cfathread_recvfrom(int socket, void restrict buffer, size_t length, int flags, struct sockaddr restrict address, socklen_t *restrict address_len) {
577	struct iovec iov;
578	iov.iov_base = buffer;
579	iov.iov_len = length;
580
581	struct msghdr msg;
582	msg.msg_name = address;
583	msg.msg_namelen = address_len ? (socklen_t)*address_len : (socklen_t)0;
584	msg.msg_iov = &iov;
585	msg.msg_iovlen = 1;
586	msg.msg_control = 0p;
587	msg.msg_controllen = 0;
588
589	#if defined(EPOLL_FOR_SOCKETS)
590	ssize_t ret;
591	yield();
592	for() {
593	ret = recvmsg(socket, &msg, flags);
594	if(ret >= 0) break;
595	if(errno != EAGAIN && errno != EWOULDBLOCK) break;
596
597	epoll_wait(socket, EPOLLIN);
598	}
599	#else
600	ssize_t ret = cfa_recvmsg(socket, &msg, flags, CFA_IO_LAZY);
601	#endif
602
603	if(address_len) *address_len = msg.msg_namelen;
604	return ret;
605	}
606
607	ssize_t cfathread_read(int fildes, void *buf, size_t nbyte) {
608	// Use recv rather then read for socket since it's faster
609	#if defined(EPOLL_FOR_SOCKETS)
610	ssize_t ret;
611	__STATS__( false, io.ops.sockread++; )
612	yield();
613	for() {
614	ret = recv(fildes, buf, nbyte, 0);
615	if(ret >= 0) break;
616	if(errno != EAGAIN && errno != EWOULDBLOCK) break;
617
618	__STATS__( false, io.ops.epllread++; )
619	epoll_wait(fildes, EPOLLIN);
620	}
621	#else
622	ssize_t ret = cfa_recv(fildes, buf, nbyte, 0, CFA_IO_LAZY);
623	#endif
624	return ret;
625	}
626
627	}

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

Download in other formats:

Original Format