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

Last change on this file was 0cee082, checked in by caparsons <caparson@…>, 3 weeks ago

refactored naming for lock to be more accurate and cleaned up REACQ nonsense in locks.hfa

  • 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
28extern "C" {
29                #include <string.h>
30                #include <errno.h>
31}
32
33extern void ?{}(processor &, const char[], cluster &, thread$ *);
34extern "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
39extern Time __kernel_get_time();
40extern 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
227struct cfathread_object {
228        thread$ self;
229        void * (*themain)( void * );
230        void * arg;
231        void * ret;
232};
233void main(cfathread_object & this);
234void ^?{}(cfathread_object & mutex this);
235
236static inline thread$ * get_thread( cfathread_object & this ) { return &this.self; }
237
238typedef ThreadCancelled(cfathread_object) cfathread_exception;
239typedef vtable(ThreadCancelled(cfathread_object)) cfathread_vtable;
240
241void defaultResumptionHandler(ThreadCancelled(cfathread_object) & except) {
242        abort | "A thread was cancelled";
243}
244
245cfathread_vtable _cfathread_vtable_instance;
246
247cfathread_vtable & const _default_vtable = _cfathread_vtable_instance;
248
249cfathread_vtable const & get_exception_vtable(cfathread_exception *) {
250        return _cfathread_vtable_instance;
251}
252
253static 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
260void ^?{}(cfathread_object & mutex this) {
261        ^(this.self){};
262}
263
264void 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
274struct __cfainit {
275        thread$ self;
276        void (*init)( void * );
277        void * arg;
278};
279void main(__cfainit & this);
280void ^?{}(__cfainit & mutex this);
281
282static inline thread$ * get_thread( __cfainit & this ) { return &this.self; }
283
284typedef ThreadCancelled(__cfainit) __cfainit_exception;
285typedef vtable(ThreadCancelled(__cfainit)) __cfainit_vtable;
286
287void 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
297static 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
316void ^?{}(__cfainit & mutex this) {
317        ^(this.self){};
318}
319
320void 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
332extern "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                exp_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(exp_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
476extern "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.