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

Last change on this file was a0a949c, checked in by Peter A. Buhr <pabuhr@…>, 15 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
30extern void ?{}(processor &, const char[], cluster &, thread$ *);
31extern "C" {
32        extern void __cfactx_invoke_thread(void (*main)(void *), void * this);
33}
34
35extern Time __kernel_get_time();
36extern 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
223struct cfathread_object {
224        thread$ self;
225        void * (*themain)( void * );
226        void * arg;
227        void * ret;
228};
229void main(cfathread_object & this);
230void ^?{}(cfathread_object & mutex this);
231
232static inline thread$ * get_thread( cfathread_object & this ) { return &this.self; }
233
234typedef ThreadCancelled(cfathread_object) cfathread_exception;
235typedef vtable(ThreadCancelled(cfathread_object)) cfathread_vtable;
236
237void defaultResumptionHandler(ThreadCancelled(cfathread_object) & except) {
238        abort | "A thread was cancelled";
239}
240
241cfathread_vtable _cfathread_vtable_instance;
242
243cfathread_vtable & const _default_vtable = _cfathread_vtable_instance;
244
245cfathread_vtable const & get_exception_vtable(cfathread_exception *) {
246        return _cfathread_vtable_instance;
247}
248
249static 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
256void ^?{}(cfathread_object & mutex this) {
257        ^(this.self){};
258}
259
260void 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
270struct __cfainit {
271        thread$ self;
272        void (*init)( void * );
273        void * arg;
274};
275void main(__cfainit & this);
276void ^?{}(__cfainit & mutex this);
277
278static inline thread$ * get_thread( __cfainit & this ) { return &this.self; }
279
280typedef ThreadCancelled(__cfainit) __cfainit_exception;
281typedef vtable(ThreadCancelled(__cfainit)) __cfainit_vtable;
282
283void 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
293static 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
312void ^?{}(__cfainit & mutex this) {
313        ^(this.self){};
314}
315
316void 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
328extern "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
470extern "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.