Changeset 63be3387 for libcfa/src/concurrency/kernel

Timestamp:

Nov 14, 2022, 11:52:44 AM (3 years ago)

Author:

caparson <caparson@…>

Branches:

ADT, ast-experimental, master

Children:

7d9598d8

Parents:

b77f0e1 (diff), 19a8c40 (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/kernel

Files:

• cluster.cfa (modified) (1 diff)
• private.hfa (modified) (4 diffs)
• startup.cfa (modified) (17 diffs)

libcfa/src/concurrency/kernel/cluster.cfa

@@ -483 +483 @@
 
         // We add a boat-load of assertions here because the anchor code is very fragile
-        /* paranoid */ _Static_assert( offsetof( thread$, link ) == nested_offsetof(__intrusive_lane_t, l.anchor) );
-        /* paranoid */ verify( offsetof( thread$, link ) == nested_offsetof(__intrusive_lane_t, l.anchor) );
-        /* paranoid */ verify( ((uintptr_t)( mock_head(this) ) + offsetof( thread$, link )) == (uintptr_t)(&this.l.anchor) );
-        /* paranoid */ verify( &mock_head(this)->link.next == &this.l.anchor.next );
-        /* paranoid */ verify( &mock_head(this)->link.ts   == &this.l.anchor.ts   );
-        /* paranoid */ verify( mock_head(this)->link.next == 0p );
-        /* paranoid */ verify( mock_head(this)->link.ts   == MAX );
+        /* paranoid */ _Static_assert( offsetof( thread$, rdy_link ) == nested_offsetof(__intrusive_lane_t, l.anchor) );
+        /* paranoid */ verify( offsetof( thread$, rdy_link ) == nested_offsetof(__intrusive_lane_t, l.anchor) );
+        /* paranoid */ verify( ((uintptr_t)( mock_head(this) ) + offsetof( thread$, rdy_link )) == (uintptr_t)(&this.l.anchor) );
+        /* paranoid */ verify( &mock_head(this)->rdy_link.next == &this.l.anchor.next );
+        /* paranoid */ verify( &mock_head(this)->rdy_link.ts   == &this.l.anchor.ts   );
+        /* paranoid */ verify( mock_head(this)->rdy_link.next == 0p );
+        /* paranoid */ verify( mock_head(this)->rdy_link.ts   == MAX );
         /* paranoid */ verify( mock_head(this) == this.l.prev );
         /* paranoid */ verify( __alignof__(__intrusive_lane_t) == 64 );

libcfa/src/concurrency/kernel/private.hfa

@@ -20 +20 @@
 #endif
 
+#include <signal.h>
+
 #include "kernel.hfa"
 #include "thread.hfa"
@@ -39 +41 @@
 }
 
-// Defines whether or not we *want* to use io_uring_enter as the idle_sleep blocking call
-// #define CFA_WANT_IO_URING_IDLE
-
-// Defines whether or not we *can* use io_uring_enter as the idle_sleep blocking call
-#if defined(CFA_WANT_IO_URING_IDLE) && defined(CFA_HAVE_LINUX_IO_URING_H)
-        #if defined(CFA_HAVE_IORING_OP_READ) || (defined(CFA_HAVE_READV) && defined(CFA_HAVE_IORING_OP_READV))
-                #define CFA_WITH_IO_URING_IDLE
-        #endif
-#endif
-
 // #define READYQ_USE_LINEAR_AVG
 #define READYQ_USE_LOGDBL_AVG
@@ -63 +55 @@
 #endif
 
+extern "C" {
+        __attribute__((visibility("protected"))) int __cfaabi_pthread_create(pthread_t *_thread, const pthread_attr_t *attr, void *(*start_routine) (void *), void *arg);
+        __attribute__((visibility("protected"))) int __cfaabi_pthread_join(pthread_t _thread, void **retval);
+        __attribute__((visibility("protected"))) pthread_t __cfaabi_pthread_self(void);
+        __attribute__((visibility("protected"))) int __cfaabi_pthread_attr_init(pthread_attr_t *attr);
+        __attribute__((visibility("protected"))) int __cfaabi_pthread_attr_destroy(pthread_attr_t *attr);
+        __attribute__((visibility("protected"))) int __cfaabi_pthread_attr_setstack( pthread_attr_t *attr, void *stackaddr, size_t stacksize );
+        __attribute__((visibility("protected"))) int __cfaabi_pthread_attr_getstacksize( const pthread_attr_t *attr, size_t *stacksize );
+        __attribute__((visibility("protected"))) int __cfaabi_pthread_sigqueue(pthread_t _thread, int sig, const union sigval value);
+        __attribute__((visibility("protected"))) int __cfaabi_pthread_sigmask( int how, const sigset_t *set, sigset_t *oset);
+}
+
 //-----------------------------------------------------------------------------
 // Scheduler
@@ -153 +157 @@
 #define TICKET_RUNNING ( 0) // thread is running
 #define TICKET_UNBLOCK ( 1) // thread should ignore next block
+#define TICKET_DEAD    (0xDEAD) // thread should never be unparked
 
 //-----------------------------------------------------------------------------

libcfa/src/concurrency/kernel/startup.cfa

@@ -16 +16 @@
 #define __cforall_thread__
 #define _GNU_SOURCE
+
+// #define __CFA_DEBUG_PRINT_RUNTIME_CORE__
 
 // C Includes
@@ -113 +115 @@
 KERNEL_STORAGE(thread$,              mainThread);
 KERNEL_STORAGE(__stack_t,            mainThreadCtx);
-// KERNEL_STORAGE(__scheduler_RWLock_t, __scheduler_lock);
-KERNEL_STORAGE(eventfd_t,            mainIdleEventFd);
-KERNEL_STORAGE(io_future_t,          mainIdleFuture);
 #if !defined(__CFA_NO_STATISTICS__)
 KERNEL_STORAGE(__stats_t, mainProcStats);
@@ -222 +221 @@
                 ( this.runner ){};
                 init( this, "Main Processor", *mainCluster, 0p );
-                kernel_thread = pthread_self();
+                kernel_thread = __cfaabi_pthread_self();
 
                 runner{ &this };
@@ -232 +231 @@
         mainProcessor = (processor *)&storage_mainProcessor;
         (*mainProcessor){};
-
-        mainProcessor->idle_wctx.rdbuf = &storage_mainIdleEventFd;
-        mainProcessor->idle_wctx.ftr   = (io_future_t*)&storage_mainIdleFuture;
-        /* paranoid */ verify( sizeof(storage_mainIdleEventFd) == sizeof(eventfd_t) );
 
         __cfa_io_start( mainProcessor );
@@ -283 +278 @@
 }
 
+extern "C"{
+        void pthread_delete_kernel_threads_();
+}
+
+
 static void __kernel_shutdown(void) {
         if(!cfa_main_returned) return;
+
+        //delete kernel threads for pthread_concurrency
+        pthread_delete_kernel_threads_();
+
         /* paranoid */ verify( __preemption_enabled() );
         disable_interrupts();
@@ -327 +331 @@
 
                 /* paranoid */ verify( this.do_terminate == true );
-                __cfaabi_dbg_print_safe("Kernel : destroyed main processor context %p\n", &runner);
+                __cfadbg_print_safe(runtime_core, "Kernel : destroyed main processor context %p\n", &runner);
         }
 
@@ -373 +377 @@
         register_tls( proc );
 
-        // used for idle sleep when io_uring is present
-        io_future_t future;
-        eventfd_t idle_buf;
-        proc->idle_wctx.ftr = &future;
-        proc->idle_wctx.rdbuf = &idle_buf;
-
-
         // SKULLDUGGERY: We want to create a context for the processor coroutine
         // which is needed for the 2-step context switch. However, there is no reason
@@ -388 +385 @@
         (proc->runner){ proc, &info };
 
-        __cfaabi_dbg_print_safe("Coroutine : created stack %p\n", get_coroutine(proc->runner)->stack.storage);
+        __cfadbg_print_safe(runtime_core, "Coroutine : created stack %p\n", get_coroutine(proc->runner)->stack.storage);
 
         //Set global state
@@ -514 +511 @@
         self_mon.recursion = 1;
         self_mon_p = &self_mon;
-        link.next = 0p;
-        link.ts   = MAX;
+        rdy_link.next = 0p;
+        rdy_link.ts   = MAX;
         preferred = ready_queue_new_preferred();
         last_proc = 0p;
         random_state = __global_random_mask ? __global_random_prime : __global_random_prime ^ rdtscl();
         #if defined( __CFA_WITH_VERIFY__ )
+                executing = 0p;
                 canary = 0x0D15EA5E0D15EA5Ep;
         #endif
 
-        node.next = 0p;
-        node.prev = 0p;
         doregister(curr_cluster, this);
 
@@ -647 +643 @@
         #endif
 
-        threads{ __get };
+        threads{};
 
         io.arbiter = create();
         io.params = io_params;
+
+        managed.procs = 0p;
+        managed.cnt = 0;
 
         doregister(this);
@@ -667 +666 @@
 
 void ^?{}(cluster & this) libcfa_public {
+        set_concurrency( this, 0 );
+
         destroy(this.io.arbiter);
 
@@ -722 +723 @@
         lock      (cltr->thread_list_lock __cfaabi_dbg_ctx2);
         cltr->nthreads += 1;
-        push_front(cltr->threads, thrd);
+        insert_first(cltr->threads, thrd);
         unlock    (cltr->thread_list_lock);
 }
@@ -728 +729 @@
 void unregister( cluster * cltr, thread$ & thrd ) {
         lock  (cltr->thread_list_lock __cfaabi_dbg_ctx2);
-        remove(cltr->threads, thrd );
-        cltr->nthreads -= 1;
+        {
+                tytagref( dlink(thread$), dlink(thread$) ) ?`inner( thread$ & this ) = void;
+                with( DLINK_VIA( thread$, struct __thread_user_link ) )
+                        remove( thrd );
+                cltr->nthreads -= 1;
+        }
         unlock(cltr->thread_list_lock);
 }
@@ -777 +782 @@
         pthread_attr_t attr;
 
-        check( pthread_attr_init( &attr ), "pthread_attr_init" ); // initialize attribute
+        check( __cfaabi_pthread_attr_init( &attr ), "pthread_attr_init" ); // initialize attribute
 
         size_t stacksize = max( PTHREAD_STACK_MIN, DEFAULT_STACK_SIZE );
@@ -804 +809 @@
         #endif
 
-        check( pthread_attr_setstack( &attr, stack, stacksize ), "pthread_attr_setstack" );
-        check( pthread_create( pthread, &attr, start, arg ), "pthread_create" );
+        check( __cfaabi_pthread_attr_setstack( &attr, stack, stacksize ), "pthread_attr_setstack" );
+        check( __cfaabi_pthread_create( pthread, &attr, start, arg ), "pthread_create" );
         return stack;
 }
 
 void __destroy_pthread( pthread_t pthread, void * stack, void ** retval ) {
-        int err = pthread_join( pthread, retval );
+        int err = __cfaabi_pthread_join( pthread, retval );
         if( err != 0 ) abort("KERNEL ERROR: joining pthread %p caused error %s\n", (void*)pthread, strerror(err));
 
@@ -816 +821 @@
                 pthread_attr_t attr;
 
-                check( pthread_attr_init( &attr ), "pthread_attr_init" ); // initialize attribute
+                check( __cfaabi_pthread_attr_init( &attr ), "pthread_attr_init" ); // initialize attribute
 
                 size_t stacksize;
                 // default stack size, normally defined by shell limit
-                check( pthread_attr_getstacksize( &attr, &stacksize ), "pthread_attr_getstacksize" );
+                check( __cfaabi_pthread_attr_getstacksize( &attr, &stacksize ), "pthread_attr_getstacksize" );
                 assert( stacksize >= PTHREAD_STACK_MIN );
                 stacksize += __page_size;
@@ -838 +843 @@
 }
 
+unsigned set_concurrency( cluster & this, unsigned new ) libcfa_public {
+        unsigned old = this.managed.cnt;
+
+        __cfadbg_print_safe(runtime_core, "Kernel : resizing cluster from %u to %u\n", old, (unsigned)new);
+
+        // Delete all the old unneeded procs
+        if(old > new) for(i; (unsigned)new ~ old) {
+                __cfadbg_print_safe(runtime_core, "Kernel : destroying %u\n", i);
+                delete( this.managed.procs[i] );
+        }
+
+        // Allocate new array (uses realloc and memcpies the data)
+        this.managed.procs = alloc( new, this.managed.procs`realloc );
+        this.managed.cnt = new;
+
+        // Create the desired new procs
+        if(old < new) for(i; old ~ new) {
+                __cfadbg_print_safe(runtime_core, "Kernel : constructing %u\n", i);
+                (*(this.managed.procs[i] = alloc())){ this };
+        }
+
+        // return the old count
+        return old;
+}
+
 #if defined(__CFA_WITH_VERIFY__)
 static bool verify_fwd_bck_rng(void) {