Changeset a1538cd for libcfa/src/concurrency/clib/cfathread.cfa

Timestamp:

Mar 12, 2021, 4:30:39 PM (2 years ago)

Author:

Thierry Delisle <tdelisle@…>

Branches:

ADT, arm-eh, enum, forall-pointer-decay, jacob/cs343-translation, master, new-ast-unique-expr, pthread-emulation, qualifiedEnum

Children:

df65c0c

Parents:

b1d83ba

Message:

Implemented part of the same C api for threads as libfibre.

File:

• libcfa/src/concurrency/clib/cfathread.cfa (modified) (4 diffs)

libcfa/src/concurrency/clib/cfathread.cfa

@@ -14 +14 @@
 //
 
+#include "fstream.hfa"
+#include "locks.hfa"
 #include "kernel.hfa"
 #include "thread.hfa"
-
-thread CRunner {
-        void (*themain)( CRunner * );
+#include "time.hfa"
+
+#include "cfathread.h"
+
+struct cfathread_object {
+        $thread self;
+        void * (*themain)( void * );
+        void * arg;
+        void * ret;
 };
-
-static void ?{}( CRunner & this, void (*themain)( CRunner * ) ) {
+void main(cfathread_object & this);
+void ^?{}(cfathread_object & mutex this);
+
+static inline $thread * get_thread( cfathread_object & this ) { return &this.self; }
+
+typedef ThreadCancelled(cfathread_object) cfathread_exception;
+typedef ThreadCancelled_vtable(cfathread_object) cfathread_vtable;
+
+void defaultResumptionHandler(ThreadCancelled(cfathread_object) & except) {
+        abort | "A thread was cancelled";
+}
+
+cfathread_vtable _cfathread_vtable_instance;
+
+cfathread_vtable const & get_exception_vtable(cfathread_exception *) {
+        return _cfathread_vtable_instance;
+}
+
+static void ?{}( cfathread_object & this, cluster & cl, void *(*themain)( void * ), void * arg ) {
         this.themain = themain;
-}
-
-void main( CRunner & this ) {
-        this.themain( &this );
+        this.arg = arg;
+        ((thread&)this){"C-thread", cl};
+        __thrd_start(this, main);
+}
+
+void ^?{}(cfathread_object & mutex this) {
+        ^(this.self){};
+}
+
+void main( cfathread_object & this ) {
+        __attribute__((unused)) void * const thrd_obj = (void*)&this;
+        __attribute__((unused)) void * const thrd_hdl = (void*)active_thread();
+        /* paranoid */ verify( thrd_obj == thrd_hdl );
+
+        this.ret = this.themain( this.arg );
 }
 
@@ -33 +69 @@
 
 extern "C" {
-        //--------------------
-        // Basic thread management
-        CRunner * cfathread_create( void (*main)( CRunner * ) ) {
-                return new( main );
-        }
-
-        void cfathread_join( CRunner * thrd ) {
+        int cfathread_cluster_create(cfathread_cluster_t * cl) __attribute__((nonnull(1))) {
+                *cl = new();
+                return 0;
+        }
+
+        cfathread_cluster_t cfathread_cluster_self(void) {
+                return active_cluster();
+        }
+
+        int cfathread_cluster_add_worker(cfathread_cluster_t cl, pthread_t* tid, void (*init_routine) (void *), void * arg) {
+                // processor * proc = new("C-processor", *cl, init_routine, arg);
+                processor * proc = alloc();
+                (*proc){ "C-processor", *cl, init_routine, arg };
+                if(tid) *tid = proc->kernel_thread;
+                return 0;
+        }
+
+        int cfathread_cluster_pause (cfathread_cluster_t) {
+                abort | "Pausing clusters is not supported";
+                exit(1);
+        }
+
+        int cfathread_cluster_resume(cfathread_cluster_t) {
+                abort | "Resuming clusters is not supported";
+                exit(1);
+        }
+
+        //--------------------
+        // Thread attributes
+        int cfathread_attr_init(cfathread_attr_t *attr) __attribute__((nonnull (1))) {
+                attr->cl = active_cluster();
+                return 0;
+        }
+
+        //--------------------
+        // Thread
+        int cfathread_create( cfathread_t * handle, cfathread_attr_t * attr, void *(*main)( void * ), void * arg ) __attribute__((nonnull (1))) {
+                cluster * cl = attr ? attr->cl : active_cluster();
+                cfathread_t thrd = alloc();
+                (*thrd){ *cl, main, arg };
+                *handle = thrd;
+                return 0;
+        }
+
+        int cfathread_join( cfathread_t thrd, void ** retval ) {
+                void * ret = join( *thrd ).ret;
                 delete( thrd );
+                if(retval) {
+                        *retval = ret;
+                }
+                return 0;
+        }
+
+        cfathread_t cfathread_self(void) {
+                return (cfathread_t)active_thread();
+        }
+
+        int cfathread_usleep(useconds_t usecs) {
+                sleep(usecs`us);
+                return 0;
+        }
+
+        int cfathread_sleep(unsigned int secs) {
+                sleep(secs`s);
+                return 0;
         }
 
@@ -47 +140 @@
         }
 
-        void cfathread_unpark( CRunner * thrd ) {
+        void cfathread_unpark( cfathread_t thrd ) {
                 unpark( *thrd );
         }
@@ -55 +148 @@
         }
 
-        //--------------------
-        // Basic kernel features
-        void cfathread_setproccnt( int ncnt ) {
-                assert( ncnt >= 1 );
-                adelete( procs );
-
-                proc_cnt = ncnt - 1;
-                procs = anew(proc_cnt);
-        }
-}
+        typedef struct cfathread_mutex * cfathread_mutex_t;
+
+        //--------------------
+        // Mutex
+        struct cfathread_mutex {
+                single_acquisition_lock impl;
+        };
+        int cfathread_mutex_init(cfathread_mutex_t *restrict mut, const cfathread_mutexattr_t *restrict) __attribute__((nonnull (1))) { *mut = new(); return 0; }
+        int cfathread_mutex_destroy(cfathread_mutex_t *mut) __attribute__((nonnull (1))) { delete( *mut ); return 0; }
+        int cfathread_mutex_lock   (cfathread_mutex_t *mut) __attribute__((nonnull (1))) { lock    ( (*mut)->impl ); return 0; }
+        int cfathread_mutex_trylock(cfathread_mutex_t *mut) __attribute__((nonnull (1))) { try_lock( (*mut)->impl ); return 0; }
+        int cfathread_mutex_unlock (cfathread_mutex_t *mut) __attribute__((nonnull (1))) { unlock  ( (*mut)->impl ); return 0; }
+
+        //--------------------
+        // Condition
+        struct cfathread_condition {
+                condition_variable(single_acquisition_lock) impl;
+        };
+        int cfathread_cond_init(cfathread_cond_t *restrict cond, const cfathread_condattr_t *restrict) __attribute__((nonnull (1))) { *cond = new(); return 0; }
+        int cfathread_cond_signal(cfathread_cond_t *cond) __attribute__((nonnull (1)))  { notify_one( (*cond)->impl ); return 0; }
+        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; }
+        int cfathread_cond_timedwait(cfathread_cond_t *restrict cond, cfathread_mutex_t *restrict mut, const struct timespec *restrict abstime) __attribute__((nonnull (1,2,3))) {
+                Time t = { *abstime };
+                if( wait( (*cond)->impl, (*mut)->impl, t ) ) {
+                        return 0;
+                }
+                errno = ETIMEDOUT;
+                return ETIMEDOUT;
+        }
+}
+
+#include <iofwd.hfa>
+
+extern "C" {
+        #include <unistd.h>
+        #include <sys/types.h>
+        #include <sys/socket.h>
+
+        //--------------------
+        // IO operations
+        int cfathread_socket(int domain, int type, int protocol) {
+                return socket(domain, type, protocol);
+        }
+        int cfathread_bind(int socket, const struct sockaddr *address, socklen_t address_len) {
+                return bind(socket, address, address_len);
+        }
+
+        int cfathread_listen(int socket, int backlog) {
+                return listen(socket, backlog);
+        }
+
+        int cfathread_accept(int socket, struct sockaddr *restrict address, socklen_t *restrict address_len) {
+                return cfa_accept4(socket, address, address_len, 0, CFA_IO_LAZY);
+        }
+
+        int cfathread_connect(int socket, const struct sockaddr *address, socklen_t address_len) {
+                return cfa_connect(socket, address, address_len, CFA_IO_LAZY);
+        }
+
+        int cfathread_dup(int fildes) {
+                return dup(fildes);
+        }
+
+        int cfathread_close(int fildes) {
+                return cfa_close(fildes, CFA_IO_LAZY);
+        }
+
+        ssize_t cfathread_sendmsg(int socket, const struct msghdr *message, int flags) {
+                return cfa_sendmsg(socket, message, flags, CFA_IO_LAZY);
+        }
+
+        ssize_t cfathread_write(int fildes, const void *buf, size_t nbyte) {
+                return cfa_write(fildes, buf, nbyte, CFA_IO_LAZY);
+        }
+
+        ssize_t cfathread_recvfrom(int socket, void *restrict buffer, size_t length, int flags, struct sockaddr *restrict address, socklen_t *restrict address_len)  {
+                struct iovec iov;
+                iov.iov_base = buffer;
+                iov.iov_len = length;
+
+                struct msghdr msg;
+                msg.msg_name = address;
+                msg.msg_namelen = address_len ? (socklen_t)*address_len : (socklen_t)0;
+                msg.msg_iov = &iov;
+                msg.msg_iovlen = 1;
+                msg.msg_control = 0p;
+                msg.msg_controllen = 0;
+
+                ssize_t ret = cfa_recvmsg(socket, &msg, flags, CFA_IO_LAZY);
+
+                if(address_len) *address_len = msg.msg_namelen;
+                return ret;
+        }
+
+        ssize_t cfathread_read(int fildes, void *buf, size_t nbyte) {
+                return cfa_read(fildes, buf, nbyte, CFA_IO_LAZY);
+        }
+
+        void cfathread_suspendFD(int) {
+                abort | "Suspending File Descriptors is not supported";
+        }
+
+        void cfathread_resumeFD (int) {
+                abort | "Resuming File Descriptors is not supported";
+        }
+
+}