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

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

Changed enable interrupts:

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