Changeset 75d874a for libcfa/src/concurrency

Timestamp:

Mar 24, 2023, 4:51:11 PM (21 months ago)

Author:

caparsons <caparson@…>

Branches:

ADT, ast-experimental, master

Children:

512d937c

Parents:

0e16a2d (diff), 1633e04 (diff)
Note: this is a merge changeset, the changes displayed below correspond to the merge itself.
Use the (diff) links above to see all the changes relative to each parent.

Message:

Merge branch 'master' of plg.uwaterloo.ca:software/cfa/cfa-cc

Location:

libcfa/src/concurrency

Files:

• channel.hfa (modified) (12 diffs)
• io.cfa (modified) (1 diff)
• io/call.cfa.in (modified) (11 diffs)
• locks.hfa (modified) (5 diffs)

libcfa/src/concurrency/channel.hfa

@@ -2 +2 @@
 
 #include <locks.hfa>
-
-struct no_reacq_lock {
-    inline exp_backoff_then_block_lock;
-};
-
-// have to override these by hand to get around plan 9 inheritance bug where resolver can't find the appropriate routine to call
-static inline void   ?{}( no_reacq_lock & this ) { ((exp_backoff_then_block_lock &)this){}; }
-static inline bool   try_lock(no_reacq_lock & this) { return try_lock(((exp_backoff_then_block_lock &)this)); }
-static inline void   lock(no_reacq_lock & this) { lock(((exp_backoff_then_block_lock &)this)); }
-static inline void   unlock(no_reacq_lock & this) { unlock(((exp_backoff_then_block_lock &)this)); }
-static inline void   on_notify(no_reacq_lock & this, struct thread$ * t ) { on_notify(((exp_backoff_then_block_lock &)this), t); }
-static inline size_t on_wait(no_reacq_lock & this) { return on_wait(((exp_backoff_then_block_lock &)this)); }
-// override wakeup so that we don't reacquire the lock if using a condvar
-static inline void   on_wakeup( no_reacq_lock & this, size_t recursion ) {}
-
-#define __PREVENTION_CHANNEL
+#include <list.hfa>
+
+#define __COOP_CHANNEL
 #ifdef __PREVENTION_CHANNEL
 forall( T ) {
 struct channel {
-    size_t size;
-    size_t front, back, count;
+    size_t size, count, front, back;
     T * buffer;
     thread$ * chair;
@@ -87 +73 @@
         return;
     }
-    else insert_( chan, elem );
+    insert_( chan, elem );
 
     unlock( mutex_lock );
@@ -110 +96 @@
 
     // wait if buffer is empty, work will be completed by someone else
-    if ( count == 0 ) { 
+    if ( count == 0 ) {
         chair = active_thread();
         chair_elem = &retval;
@@ -121 +107 @@
     memcpy((void *)&retval, (void *)&buffer[front], sizeof(T));
     count -= 1;
-    front = (front + 1) % size;
+    front++;
+    if ( front == size ) front = 0;
 
     if ( chair != 0p ) {
@@ -142 +129 @@
 
 #ifdef __COOP_CHANNEL
+
+// link field used for threads waiting on channel
+struct wait_link {
+    // used to put wait_link on a dl queue
+    inline dlink(wait_link);
+
+    // waiting thread
+    struct thread$ * t;
+
+    // shadow field
+    void * elem;
+};
+P9_EMBEDDED( wait_link, dlink(wait_link) )
+
+static inline void ?{}( wait_link & this, thread$ * t, void * elem ) {
+    this.t = t;
+    this.elem = elem;
+}
+
 forall( T ) {
+
 struct channel {
     size_t size;
     size_t front, back, count;
     T * buffer;
-    fast_cond_var( no_reacq_lock ) prods, cons;
-    no_reacq_lock mutex_lock;
+    dlist( wait_link ) prods, cons;
+    exp_backoff_then_block_lock mutex_lock;
 };
 
@@ -164 +171 @@
 static inline size_t get_count( channel(T) & chan ) with(chan) { return count; }
 static inline size_t get_size( channel(T) & chan ) with(chan) { return size; }
-static inline bool has_waiters( channel(T) & chan ) with(chan) { return !empty( cons ) || !empty( prods ); }
-static inline bool has_waiting_consumers( channel(T) & chan ) with(chan) { return !empty( cons ); }
-static inline bool has_waiting_producers( channel(T) & chan ) with(chan) { return !empty( prods ); }
+static inline bool has_waiters( channel(T) & chan ) with(chan) { return !cons`isEmpty || !prods`isEmpty; }
+static inline bool has_waiting_consumers( channel(T) & chan ) with(chan) { return !cons`isEmpty; }
+static inline bool has_waiting_producers( channel(T) & chan ) with(chan) { return !prods`isEmpty; }
 
 static inline void insert_( channel(T) & chan, T & elem ) with(chan) {
@@ -175 +182 @@
 }
 
+static inline void wake_one( dlist( wait_link ) & queue ) {
+    wait_link & popped = try_pop_front( queue );
+    unpark( popped.t );
+}
+
+static inline void block( dlist( wait_link ) & queue, void * elem_ptr, exp_backoff_then_block_lock & lock ) {
+    wait_link w{ active_thread(), elem_ptr };
+    insert_last( queue, w );
+    unlock( lock );
+    park();
+}
 
 static inline void insert( channel(T) & chan, T elem ) with(chan) {
@@ -180 +198 @@
 
     // have to check for the zero size channel case
-    if ( size == 0 && !empty( cons ) ) {
-        memcpy((void *)front( cons ), (void *)&elem, sizeof(T));
-        notify_one( cons );
+    if ( size == 0 && !cons`isEmpty ) {
+        memcpy(cons`first.elem, (void *)&elem, sizeof(T));
+        wake_one( cons );
         unlock( mutex_lock );
         return;
@@ -188 +206 @@
 
     // wait if buffer is full, work will be completed by someone else
-    if ( count == size ) { 
-        wait( prods, mutex_lock, (uintptr_t)&elem );
+    if ( count == size ) {
+        block( prods, &elem, mutex_lock );
         return;
     } // if
 
-    if ( count == 0 && !empty( cons ) )
-        // do waiting consumer work
-        memcpy((void *)front( cons ), (void *)&elem, sizeof(T)); 
-    else insert_( chan, elem );
+    if ( count == 0 && !cons`isEmpty ) {
+        memcpy(cons`first.elem, (void *)&elem, sizeof(T)); // do waiting consumer work
+        wake_one( cons );
+    } else insert_( chan, elem );
     
-    notify_one( cons );
     unlock( mutex_lock );
 }
@@ -207 +224 @@
 
     // have to check for the zero size channel case
-    if ( size == 0 && !empty( prods ) ) {
-        memcpy((void *)&retval, (void *)front( prods ), sizeof(T));
-        notify_one( prods );
+    if ( size == 0 && !prods`isEmpty ) {
+        memcpy((void *)&retval, (void *)prods`first.elem, sizeof(T));
+        wake_one( prods );
         unlock( mutex_lock );
         return retval;
@@ -215 +232 @@
 
     // wait if buffer is empty, work will be completed by someone else
-    if (count == 0) { 
-        wait( cons, mutex_lock, (uintptr_t)&retval );
+    if (count == 0) {
+        block( cons, &retval, mutex_lock );
         return retval;
     }
@@ -225 +242 @@
     front = (front + 1) % size;
 
-    if (count == size - 1 && !empty( prods ) ) 
-        insert_( chan, *((T *)front( prods )) );  // do waiting producer work
-
-    notify_one( prods );
+    if (count == size - 1 && !prods`isEmpty ) {
+        insert_( chan, *(T *)prods`first.elem );  // do waiting producer work
+        wake_one( prods );
+    }
+
     unlock( mutex_lock );
     return retval;
 }
-
 } // forall( T )
 #endif

libcfa/src/concurrency/io.cfa

@@ -295 +295 @@
                                 // make sure the target hasn't stopped existing since last time
                                 HELP: if(target < ctxs_count) {
-                                        // calculate it's age and how young it could be before we give ip on helping
+                                        // calculate it's age and how young it could be before we give up on helping
                                         const __readyQ_avg_t cutoff = calc_cutoff(ctsc, ctx->cq.id, ctxs_count, io.data, io.tscs, __shard_factor.io, false);
                                         const __readyQ_avg_t age = moving_average(ctsc, io.tscs[target].t.tv, io.tscs[target].t.ma, false);

libcfa/src/concurrency/io/call.cfa.in

@@ -89 +89 @@
 #if defined(CFA_HAVE_PREADV2)
         struct iovec;
-        extern ssize_t preadv2 (int fd, const struct iovec *iov, int iovcnt, off_t offset, int flags);
+        extern ssize_t preadv2 (int fd, const struct iovec * iov, int iovcnt, off_t offset, int flags);
 #endif
 #if defined(CFA_HAVE_PWRITEV2)
         struct iovec;
-        extern ssize_t pwritev2(int fd, const struct iovec *iov, int iovcnt, off_t offset, int flags);
+        extern ssize_t pwritev2(int fd, const struct iovec * iov, int iovcnt, off_t offset, int flags);
 #endif
 
@@ -108 +108 @@
         struct msghdr;
         struct sockaddr;
-        extern ssize_t sendmsg(int sockfd, const struct msghdr *msg, int flags);
-        extern ssize_t recvmsg(int sockfd, struct msghdr *msg, int flags);
-        extern ssize_t send(int sockfd, const void *buf, size_t len, int flags);
-        extern ssize_t recv(int sockfd, void *buf, size_t len, int flags);
+        extern ssize_t sendmsg(int sockfd, const struct msghdr * msg, int flags);
+        extern ssize_t recvmsg(int sockfd, struct msghdr * msg, int flags);
+        extern ssize_t send(int sockfd, const void * buf, size_t len, int flags);
+        extern ssize_t recv(int sockfd, void * buf, size_t len, int flags);
 
         extern int fallocate(int fd, int mode, off_t offset, off_t len);
         extern int posix_fadvise(int fd, off_t offset, off_t len, int advice);
-        extern int madvise(void *addr, size_t length, int advice);
-
-        extern int openat(int dirfd, const char *pathname, int flags, mode_t mode);
+        extern int madvise(void * addr, size_t length, int advice);
+
+        extern int openat(int dirfd, const char * pathname, int flags, mode_t mode);
         extern int close(int fd);
 
-        extern ssize_t read (int fd, void *buf, size_t count);
+        extern ssize_t read (int fd, void * buf, size_t count);
 
         struct epoll_event;
-        extern int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);
-
-        extern ssize_t splice(int fd_in, __off64_t *off_in, int fd_out, __off64_t *off_out, size_t len, unsigned int flags);
+        extern int epoll_ctl(int epfd, int op, int fd, struct epoll_event * event);
+
+        extern ssize_t splice(int fd_in, __off64_t * off_in, int fd_out, __off64_t * off_out, size_t len, unsigned int flags);
         extern ssize_t tee(int fd_in, int fd_out, size_t len, unsigned int flags);
 }
@@ -224 +224 @@
 calls = [
         # CFA_HAVE_IORING_OP_READV
-        Call('READV', 'ssize_t preadv2(int fd, const struct iovec *iov, int iovcnt, off_t offset, int flags)', {
+        Call('READV', 'ssize_t preadv2(int fd, const struct iovec * iov, int iovcnt, off_t offset, int flags)', {
                 'fd'  : 'fd',
+                'addr': '(typeof(sqe->addr))iov',
+                'len' : 'iovcnt',
                 'off' : 'offset',
-                'addr': '(uintptr_t)iov',
-                'len' : 'iovcnt',
+                'rw_flags' : 'flags'
         }, define = 'CFA_HAVE_PREADV2'),
         # CFA_HAVE_IORING_OP_WRITEV
-        Call('WRITEV', 'ssize_t pwritev2(int fd, const struct iovec *iov, int iovcnt, off_t offset, int flags)', {
+        Call('WRITEV', 'ssize_t pwritev2(int fd, const struct iovec * iov, int iovcnt, off_t offset, int flags)', {
                 'fd'  : 'fd',
+                'addr': '(typeof(sqe->addr))iov',
+                'len' : 'iovcnt',
                 'off' : 'offset',
-                'addr': '(uintptr_t)iov',
-                'len' : 'iovcnt'
+                'rw_flags' : 'flags'
         }, define = 'CFA_HAVE_PWRITEV2'),
         # CFA_HAVE_IORING_OP_FSYNC
@@ -242 +244 @@
         }),
         # CFA_HAVE_IORING_OP_EPOLL_CTL
-        Call('EPOLL_CTL', 'int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event)', {
+        Call('EPOLL_CTL', 'int epoll_ctl(int epfd, int op, int fd, struct epoll_event * event)', {
                 'fd': 'epfd',
+                'len': 'op',
                 'addr': 'fd',
-                'len': 'op',
-                'off': '(uintptr_t)event'
+                'off': '(typeof(sqe->off))event'
         }),
         # CFA_HAVE_IORING_OP_SYNC_FILE_RANGE
@@ -256 +258 @@
         }),
         # CFA_HAVE_IORING_OP_SENDMSG
-        Call('SENDMSG', 'ssize_t sendmsg(int sockfd, const struct msghdr *msg, int flags)', {
-                'fd': 'sockfd',
-                'addr': '(uintptr_t)(struct msghdr *)msg',
+        Call('SENDMSG', 'ssize_t sendmsg(int sockfd, const struct msghdr * msg, int flags)', {
+                'fd': 'sockfd',
+                'addr': '(typeof(sqe->addr))(struct msghdr *)msg',
                 'len': '1',
                 'msg_flags': 'flags'
         }),
         # CFA_HAVE_IORING_OP_RECVMSG
-        Call('RECVMSG', 'ssize_t recvmsg(int sockfd, struct msghdr *msg, int flags)', {
-                'fd': 'sockfd',
-                'addr': '(uintptr_t)(struct msghdr *)msg',
+        Call('RECVMSG', 'ssize_t recvmsg(int sockfd, struct msghdr * msg, int flags)', {
+                'fd': 'sockfd',
+                'addr': '(typeof(sqe->addr))(struct msghdr *)msg',
                 'len': '1',
                 'msg_flags': 'flags'
         }),
         # CFA_HAVE_IORING_OP_SEND
-        Call('SEND', 'ssize_t send(int sockfd, const void *buf, size_t len, int flags)', {
-                'fd': 'sockfd',
-                'addr': '(uintptr_t)buf',
+        Call('SEND', 'ssize_t send(int sockfd, const void * buf, size_t len, int flags)', {
+                'fd': 'sockfd',
+                'addr': '(typeof(sqe->addr))buf',
                 'len': 'len',
                 'msg_flags': 'flags'
         }),
         # CFA_HAVE_IORING_OP_RECV
-        Call('RECV', 'ssize_t recv(int sockfd, void *buf, size_t len, int flags)', {
-                'fd': 'sockfd',
-                'addr': '(uintptr_t)buf',
+        Call('RECV', 'ssize_t recv(int sockfd, void * buf, size_t len, int flags)', {
+                'fd': 'sockfd',
+                'addr': '(typeof(sqe->addr))buf',
                 'len': 'len',
                 'msg_flags': 'flags'
@@ -286 +288 @@
         Call('ACCEPT', 'int accept4(int sockfd, __SOCKADDR_ARG addr, socklen_t * restrict addrlen, int flags)', {
                 'fd': 'sockfd',
-                'addr': '(uintptr_t)&addr',
-                'addr2': '(uintptr_t)addrlen',
+                'addr': '(typeof(sqe->addr))&addr',
+                'addr2': '(typeof(sqe->addr2))addrlen',
                 'accept_flags': 'flags'
         }),
@@ -293 +295 @@
         Call('CONNECT', 'int connect(int sockfd, __CONST_SOCKADDR_ARG addr, socklen_t addrlen)', {
                 'fd': 'sockfd',
-                'addr': '(uintptr_t)&addr',
+                'addr': '(typeof(sqe->addr))&addr',
                 'off': 'addrlen'
         }),
@@ -299 +301 @@
         Call('FALLOCATE', 'int fallocate(int fd, int mode, off_t offset, off_t len)', {
                 'fd': 'fd',
-                'addr': '(uintptr_t)len',
                 'len': 'mode',
-                'off': 'offset'
+                'off': 'offset',
+                'addr': 'len'
         }),
         # CFA_HAVE_IORING_OP_FADVISE
@@ -311 +313 @@
         }),
         # CFA_HAVE_IORING_OP_MADVISE
-        Call('MADVISE', 'int madvise(void *addr, size_t length, int advice)', {
-                'addr': '(uintptr_t)addr',
+        Call('MADVISE', 'int madvise(void * addr, size_t length, int advice)', {
+                'addr': '(typeof(sqe->addr))addr',
                 'len': 'length',
                 'fadvise_advice': 'advice'
         }),
         # CFA_HAVE_IORING_OP_OPENAT
-        Call('OPENAT', 'int openat(int dirfd, const char *pathname, int flags, mode_t mode)', {
+        Call('OPENAT', 'int openat(int dirfd, const char * pathname, int flags, mode_t mode)', {
                 'fd': 'dirfd',
-                'addr': '(uintptr_t)pathname',
-                'len': 'mode',
-                'open_flags': 'flags;'
+                'addr': '(typeof(sqe->addr))pathname',
+                'open_flags': 'flags;',
+                'len': 'mode'
         }),
         # CFA_HAVE_IORING_OP_OPENAT2
-        Call('OPENAT2', 'int openat2(int dirfd, const char *pathname, struct open_how * how, size_t size)', {
+        Call('OPENAT2', 'int openat2(int dirfd, const char * pathname, struct open_how * how, size_t size)', {
                 'fd': 'dirfd',
-                'addr': 'pathname',
-                'len': 'sizeof(*how)',
-                'off': '(uintptr_t)how',
+                'addr': '(typeof(sqe->addr))pathname',
+                'off': '(typeof(sqe->off))how',
+                'len': 'sizeof(*how)'
         }, define = 'CFA_HAVE_OPENAT2'),
         # CFA_HAVE_IORING_OP_CLOSE
@@ -335 +337 @@
         }),
         # CFA_HAVE_IORING_OP_STATX
-        Call('STATX', 'int statx(int dirfd, const char *pathname, int flags, unsigned int mask, struct statx *statxbuf)', {
+        Call('STATX', 'int statx(int dirfd, const char * pathname, int flags, unsigned int mask, struct statx * statxbuf)', {
                 'fd': 'dirfd',
-                'off': '(uintptr_t)statxbuf',
-                'addr': 'pathname',
+                'addr': '(typeof(sqe->addr))pathname',
+                'statx_flags': 'flags',
                 'len': 'mask',
-                'statx_flags': 'flags'
+                'off': '(typeof(sqe->off))statxbuf'
         }, define = 'CFA_HAVE_STATX'),
         # CFA_HAVE_IORING_OP_READ
         Call('READ', 'ssize_t read(int fd, void * buf, size_t count)', {
                 'fd': 'fd',
-                'addr': '(uintptr_t)buf',
+                'addr': '(typeof(sqe->addr))buf',
                 'len': 'count'
         }),
@@ -351 +353 @@
         Call('WRITE', 'ssize_t write(int fd, void * buf, size_t count)', {
                 'fd': 'fd',
-                'addr': '(uintptr_t)buf',
+                'addr': '(typeof(sqe->addr))buf',
                 'len': 'count'
         }),
         # CFA_HAVE_IORING_OP_SPLICE
-        Call('SPLICE', 'ssize_t splice(int fd_in, __off64_t *off_in, int fd_out, __off64_t *off_out, size_t len, unsigned int flags)', {
+        Call('SPLICE', 'ssize_t splice(int fd_in, __off64_t * off_in, int fd_out, __off64_t * off_out, size_t len, unsigned int flags)', {
                 'splice_fd_in': 'fd_in',
-                'splice_off_in': 'off_in ? (__u64)*off_in : (__u64)-1',
+                'splice_off_in': 'off_in ? (typeof(sqe->splice_off_in))*off_in : (typeof(sqe->splice_off_in))-1',
                 'fd': 'fd_out',
-                'off': 'off_out ? (__u64)*off_out : (__u64)-1',
+                'off': 'off_out ? (typeof(sqe->off))*off_out : (typeof(sqe->off))-1',
                 'len': 'len',
                 'splice_flags': 'flags'

libcfa/src/concurrency/locks.hfa

@@ -253 +253 @@
 static inline void on_wakeup(clh_lock & this, size_t recursion ) { lock(this); }
 
-
 //-----------------------------------------------------------------------------
 // Exponential backoff then block lock
@@ -272 +271 @@
         this.lock_value = 0;
 }
-static inline void ^?{}( exp_backoff_then_block_lock & this ) {}
-// static inline void ?{}( exp_backoff_then_block_lock & this, exp_backoff_then_block_lock this2 ) = void;
-// static inline void ?=?( exp_backoff_then_block_lock & this, exp_backoff_then_block_lock this2 ) = void;
 
 static inline bool internal_try_lock(exp_backoff_then_block_lock & this, size_t & compare_val) with(this) {
-        if (__atomic_compare_exchange_n(&lock_value, &compare_val, 1, false, __ATOMIC_ACQUIRE, __ATOMIC_RELAXED)) {
-                return true;
-        }
-        return false;
+        return __atomic_compare_exchange_n(&lock_value, &compare_val, 1, false, __ATOMIC_ACQUIRE, __ATOMIC_RELAXED);
 }
 
@@ -286 +279 @@
 
 static inline bool try_lock_contention(exp_backoff_then_block_lock & this) with(this) {
-        if (__atomic_exchange_n(&lock_value, 2, __ATOMIC_ACQUIRE) == 0) {
-                return true;
-        }
-        return false;
+        return !__atomic_exchange_n(&lock_value, 2, __ATOMIC_ACQUIRE);
 }
 
 static inline bool block(exp_backoff_then_block_lock & this) with(this) {
-        lock( spinlock __cfaabi_dbg_ctx2 ); // TODO change to lockfree queue (MPSC)
-        if (lock_value != 2) {
-                unlock( spinlock );
-                return true;
-        }
-        insert_last( blocked_threads, *active_thread() );
-        unlock( spinlock );
+    lock( spinlock __cfaabi_dbg_ctx2 );
+    if (__atomic_load_n( &lock_value, __ATOMIC_SEQ_CST) != 2) {
+        unlock( spinlock );
+        return true;
+    }
+    insert_last( blocked_threads, *active_thread() );
+    unlock( spinlock );
         park( );
         return true;
@@ -307 +297 @@
         size_t compare_val = 0;
         int spin = 4;
+
         // linear backoff
         for( ;; ) {
@@ -324 +315 @@
 static inline void unlock(exp_backoff_then_block_lock & this) with(this) {
     if (__atomic_exchange_n(&lock_value, 0, __ATOMIC_RELEASE) == 1) return;
-        lock( spinlock __cfaabi_dbg_ctx2 );
-        thread$ * t = &try_pop_front( blocked_threads );
-        unlock( spinlock );
-        unpark( t );
+    lock( spinlock __cfaabi_dbg_ctx2 );
+    thread$ * t = &try_pop_front( blocked_threads );
+    unlock( spinlock );
+    unpark( t );
 }