source: libcfa/src/concurrency/clib/cfathread.cfa@ 14f6a3cb

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

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