source: libcfa/src/concurrency/clib/cfathread.cfa @ 5c2b454

ADTarm-ehast-experimentalenumforall-pointer-decayjacob/cs343-translationnew-ast-unique-exprpthread-emulationqualifiedEnum
Last change on this file since 5c2b454 was a5e7233, checked in by Thierry Delisle <tdelisle@…>, 3 years ago

C interface now runs worker init routine in dedicated thread.
Also added test for this case.

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