source: libcfa/src/concurrency/clib/cfathread.cfa @ 2b910f9

ADTast-experimentalenumforall-pointer-decayjacob/cs343-translationnew-ast-unique-exprpthread-emulationqualifiedEnum
Last change on this file since 2b910f9 was c457dc41, checked in by Thierry Delisle <tdelisle@…>, 3 years ago

Fix sequential handling of timers

  • Property mode set to 100644
File size: 9.7 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#include "fstream.hfa"
17#include "locks.hfa"
18#include "kernel.hfa"
19#include "stats.hfa"
20#include "thread.hfa"
21#include "time.hfa"
22
23#include "cfathread.h"
24
25extern void ?{}(processor &, const char[], cluster &, $thread *);
26extern "C" {
27      extern void __cfactx_invoke_thread(void (*main)(void *), void * this);
28}
29
30extern Time __kernel_get_time();
31
32//================================================================================
33// Thread run y the C Interface
34
35struct cfathread_object {
36        $thread self;
37        void * (*themain)( void * );
38        void * arg;
39        void * ret;
40};
41void main(cfathread_object & this);
42void ^?{}(cfathread_object & mutex this);
43
44static inline $thread * get_thread( cfathread_object & this ) { return &this.self; }
45
46typedef ThreadCancelled(cfathread_object) cfathread_exception;
47typedef ThreadCancelled_vtable(cfathread_object) cfathread_vtable;
48
49void defaultResumptionHandler(ThreadCancelled(cfathread_object) & except) {
50        abort | "A thread was cancelled";
51}
52
53cfathread_vtable _cfathread_vtable_instance;
54
55cfathread_vtable & const _default_vtable = _cfathread_vtable_instance;
56
57cfathread_vtable const & get_exception_vtable(cfathread_exception *) {
58        return _cfathread_vtable_instance;
59}
60
61static void ?{}( cfathread_object & this, cluster & cl, void *(*themain)( void * ), void * arg ) {
62        this.themain = themain;
63        this.arg = arg;
64        (this.self){"C-thread", cl};
65        __thrd_start(this, main);
66}
67
68void ^?{}(cfathread_object & mutex this) {
69        ^(this.self){};
70}
71
72void main( cfathread_object & this ) {
73        __attribute__((unused)) void * const thrd_obj = (void*)&this;
74        __attribute__((unused)) void * const thrd_hdl = (void*)active_thread();
75        /* paranoid */ verify( thrd_obj == thrd_hdl );
76
77        this.ret = this.themain( this.arg );
78}
79
80//================================================================================
81// Special Init Thread responsible for the initialization or processors
82struct __cfainit {
83        $thread self;
84        void (*init)( void * );
85        void * arg;
86};
87void main(__cfainit & this);
88void ^?{}(__cfainit & mutex this);
89
90static inline $thread * get_thread( __cfainit & this ) { return &this.self; }
91
92typedef ThreadCancelled(__cfainit) __cfainit_exception;
93typedef ThreadCancelled_vtable(__cfainit) __cfainit_vtable;
94
95void defaultResumptionHandler(ThreadCancelled(__cfainit) & except) {
96        abort | "The init thread was cancelled";
97}
98
99__cfainit_vtable ___cfainit_vtable_instance;
100
101__cfainit_vtable const & get_exception_vtable(__cfainit_exception *) {
102        return ___cfainit_vtable_instance;
103}
104
105static void ?{}( __cfainit & this, void (*init)( void * ), void * arg ) {
106        this.init = init;
107        this.arg = arg;
108        (this.self){"Processir Init"};
109
110        // Don't use __thrd_start! just prep the context manually
111        $thread * this_thrd = get_thread(this);
112        void (*main_p)(__cfainit &) = main;
113
114        disable_interrupts();
115        __cfactx_start(main_p, get_coroutine(this), this, __cfactx_invoke_thread);
116
117        this_thrd->context.[SP, FP] = this_thrd->self_cor.context.[SP, FP];
118        /* paranoid */ verify( this_thrd->context.SP );
119
120        this_thrd->state = Ready;
121        enable_interrupts();
122}
123
124void ^?{}(__cfainit & mutex this) {
125        ^(this.self){};
126}
127
128void main( __cfainit & this ) {
129        __attribute__((unused)) void * const thrd_obj = (void*)&this;
130        __attribute__((unused)) void * const thrd_hdl = (void*)active_thread();
131        /* paranoid */ verify( thrd_obj == thrd_hdl );
132
133        this.init( this.arg );
134}
135
136//================================================================================
137// Main Api
138extern "C" {
139        int cfathread_cluster_create(cfathread_cluster_t * cl) __attribute__((nonnull(1))) {
140                *cl = new();
141                return 0;
142        }
143
144        cfathread_cluster_t cfathread_cluster_self(void) {
145                return active_cluster();
146        }
147
148        int cfathread_cluster_print_stats( cfathread_cluster_t cl ) {
149                #if !defined(__CFA_NO_STATISTICS__)
150                        print_stats_at_exit( *cl, CFA_STATS_READY_Q | CFA_STATS_IO );
151                        print_stats_now( *cl, CFA_STATS_READY_Q | CFA_STATS_IO );
152                #endif
153                return 0;
154        }
155
156        int cfathread_cluster_add_worker(cfathread_cluster_t cl, pthread_t* tid, void (*init_routine) (void *), void * arg) {
157                __cfainit * it = 0p;
158                if(init_routine) {
159                        it = alloc();
160                        (*it){init_routine, arg};
161                }
162                processor * proc = alloc();
163                (*proc){ "C-processor", *cl, get_thread(*it) };
164
165                // Wait for the init thread to return before continuing
166                if(it) {
167                        ^(*it){};
168                        free(it);
169                }
170
171                if(tid) *tid = proc->kernel_thread;
172                return 0;
173        }
174
175        int cfathread_cluster_pause (cfathread_cluster_t) {
176                abort | "Pausing clusters is not supported";
177                exit(1);
178        }
179
180        int cfathread_cluster_resume(cfathread_cluster_t) {
181                abort | "Resuming clusters is not supported";
182                exit(1);
183        }
184
185        //--------------------
186        // Thread attributes
187        int cfathread_attr_init(cfathread_attr_t *attr) __attribute__((nonnull (1))) {
188                attr->cl = active_cluster();
189                return 0;
190        }
191
192        //--------------------
193        // Thread
194        int cfathread_create( cfathread_t * handle, const cfathread_attr_t * attr, void *(*main)( void * ), void * arg ) __attribute__((nonnull (1))) {
195                cluster * cl = attr ? attr->cl : active_cluster();
196                cfathread_t thrd = alloc();
197                (*thrd){ *cl, main, arg };
198                *handle = thrd;
199                return 0;
200        }
201
202        int cfathread_join( cfathread_t thrd, void ** retval ) {
203                void * ret = join( *thrd ).ret;
204                ^( *thrd ){};
205                free(thrd);
206                if(retval) {
207                        *retval = ret;
208                }
209                return 0;
210        }
211
212        int cfathread_get_errno(void) {
213                return errno;
214        }
215
216        cfathread_t cfathread_self(void) {
217                return (cfathread_t)active_thread();
218        }
219
220        int cfathread_usleep(useconds_t usecs) {
221                sleep(usecs`us);
222                return 0;
223        }
224
225        int cfathread_sleep(unsigned int secs) {
226                sleep(secs`s);
227                return 0;
228        }
229
230        void cfathread_park( void ) {
231                park();
232        }
233
234        void cfathread_unpark( cfathread_t thrd ) {
235                unpark( *thrd );
236        }
237
238        void cfathread_yield( void ) {
239                yield();
240        }
241
242        typedef struct cfathread_mutex * cfathread_mutex_t;
243
244        //--------------------
245        // Mutex
246        struct cfathread_mutex {
247                fast_lock impl;
248        };
249        int cfathread_mutex_init(cfathread_mutex_t *restrict mut, const cfathread_mutexattr_t *restrict) __attribute__((nonnull (1))) { *mut = new(); return 0; }
250        int cfathread_mutex_destroy(cfathread_mutex_t *mut) __attribute__((nonnull (1))) { delete( *mut ); return 0; }
251        int cfathread_mutex_lock   (cfathread_mutex_t *mut) __attribute__((nonnull (1))) { lock( (*mut)->impl ); return 0; }
252        int cfathread_mutex_unlock (cfathread_mutex_t *mut) __attribute__((nonnull (1))) { unlock( (*mut)->impl ); return 0; }
253        int cfathread_mutex_trylock(cfathread_mutex_t *mut) __attribute__((nonnull (1))) {
254                bool ret = try_lock( (*mut)->impl );
255                if( ret ) return 0;
256                else return EBUSY;
257        }
258
259        //--------------------
260        // Condition
261        struct cfathread_condition {
262                condition_variable(fast_lock) impl;
263        };
264        int cfathread_cond_init(cfathread_cond_t *restrict cond, const cfathread_condattr_t *restrict) __attribute__((nonnull (1))) { *cond = new(); return 0; }
265        int cfathread_cond_signal(cfathread_cond_t *cond) __attribute__((nonnull (1)))  { notify_one( (*cond)->impl ); return 0; }
266        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; }
267        int cfathread_cond_timedwait(cfathread_cond_t *restrict cond, cfathread_mutex_t *restrict mut, const struct timespec *restrict abstime) __attribute__((nonnull (1,2,3))) {
268                Time t = { *abstime };
269                timespec curr;
270                clock_gettime( CLOCK_REALTIME, &curr );
271                Time c = { curr };
272                if( wait( (*cond)->impl, (*mut)->impl, t - c ) ) {
273                        return 0;
274                }
275                errno = ETIMEDOUT;
276                return ETIMEDOUT;
277        }
278}
279
280#include <iofwd.hfa>
281
282extern "C" {
283        #include <unistd.h>
284        #include <sys/types.h>
285        #include <sys/socket.h>
286
287        //--------------------
288        // IO operations
289        int cfathread_socket(int domain, int type, int protocol) {
290                return socket(domain, type, protocol);
291        }
292        int cfathread_bind(int socket, const struct sockaddr *address, socklen_t address_len) {
293                return bind(socket, address, address_len);
294        }
295
296        int cfathread_listen(int socket, int backlog) {
297                return listen(socket, backlog);
298        }
299
300        int cfathread_accept(int socket, struct sockaddr *restrict address, socklen_t *restrict address_len) {
301                return cfa_accept4(socket, address, address_len, 0, CFA_IO_LAZY);
302        }
303
304        int cfathread_connect(int socket, const struct sockaddr *address, socklen_t address_len) {
305                return cfa_connect(socket, address, address_len, CFA_IO_LAZY);
306        }
307
308        int cfathread_dup(int fildes) {
309                return dup(fildes);
310        }
311
312        int cfathread_close(int fildes) {
313                return cfa_close(fildes, CFA_IO_LAZY);
314        }
315
316        ssize_t cfathread_sendmsg(int socket, const struct msghdr *message, int flags) {
317                return cfa_sendmsg(socket, message, flags, CFA_IO_LAZY);
318        }
319
320        ssize_t cfathread_write(int fildes, const void *buf, size_t nbyte) {
321                // Use send rather then write for socket since it's faster
322                return cfa_send(fildes, buf, nbyte, 0, CFA_IO_LAZY);
323        }
324
325        ssize_t cfathread_recvfrom(int socket, void *restrict buffer, size_t length, int flags, struct sockaddr *restrict address, socklen_t *restrict address_len)  {
326                struct iovec iov;
327                iov.iov_base = buffer;
328                iov.iov_len = length;
329
330                struct msghdr msg;
331                msg.msg_name = address;
332                msg.msg_namelen = address_len ? (socklen_t)*address_len : (socklen_t)0;
333                msg.msg_iov = &iov;
334                msg.msg_iovlen = 1;
335                msg.msg_control = 0p;
336                msg.msg_controllen = 0;
337
338                ssize_t ret = cfa_recvmsg(socket, &msg, flags, CFA_IO_LAZY);
339
340                if(address_len) *address_len = msg.msg_namelen;
341                return ret;
342        }
343
344        ssize_t cfathread_read(int fildes, void *buf, size_t nbyte) {
345                // Use recv rather then read for socket since it's faster
346                return cfa_recv(fildes, buf, nbyte, 0, CFA_IO_LAZY);
347        }
348
349}
Note: See TracBrowser for help on using the repository browser.