source: libcfa/src/concurrency/clib/cfathread.cfa@ ad24245

ADT ast-experimental
Last change on this file since ad24245 was 2d028003, checked in by Peter A. Buhr <pabuhr@…>, 3 years ago

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