Changeset b6f2b21 for libcfa/src/concurrency

Timestamp:

May 5, 2020, 11:35:45 AM (4 years ago)

Author:

Thierry Delisle <tdelisle@…>

Branches:

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

Children:

4e74466

Parents:

f90d10f

Message:

Added the option to dynamically (at cluster creation time) enable/disable the user thread polling of I/O

Location:

libcfa/src/concurrency

Files:

• io.cfa (modified) (11 diffs)
• kernel.cfa (modified) (2 diffs)
• kernel.hfa (modified) (2 diffs)
• kernel_private.hfa (modified) (1 diff)

libcfa/src/concurrency/io.cfa

@@ -20 +20 @@
 
 #if !defined(HAVE_LINUX_IO_URING_H)
-        void __kernel_io_startup( cluster &, bool ) {
+        void __kernel_io_startup( cluster &, int, bool ) {
                 // Nothing to do without io_uring
         }
@@ -182 +182 @@
                 struct __submition_data submit_q;
                 struct __completion_data completion_q;
-                uint32_t flags;
+                uint32_t ring_flags;
+                int cltr_flags;
                 int fd;
                 semaphore submit;
@@ -199 +200 @@
 // I/O Startup / Shutdown logic
 //=============================================================================================
-        void __kernel_io_startup( cluster & this, bool main_cluster ) {
+        void __kernel_io_startup( cluster & this, int io_flags, bool main_cluster ) {
                 this.io = malloc();
 
@@ -294 +295 @@
 
                 // Update the global ring info
-                this.io->flags = params.flags;
-                this.io->fd    = fd;
-                this.io->done  = false;
+                this.io->ring_flags = params.flags;
+                this.io->cltr_flags = io_flags;
+                this.io->fd         = fd;
+                this.io->done       = false;
                 (this.io->submit){ min(*sq.num, *cq.num) };
 
@@ -314 +316 @@
 
         void __kernel_io_finish_start( cluster & this ) {
-                __cfadbg_print_safe(io_core, "Kernel I/O : Creating fast poller for cluter %p\n", &this);
-                (this.io->poller.fast){ this };
-                __thrd_start( this.io->poller.fast, main );
+                if( this.io->cltr_flags & CFA_CLUSTER_IO_POLLER_USER_THREAD ) {
+                        __cfadbg_print_safe(io_core, "Kernel I/O : Creating fast poller for cluter %p\n", &this);
+                        (this.io->poller.fast){ this };
+                        __thrd_start( this.io->poller.fast, main );
+                }
 
                 // Create the poller thread
@@ -339 +343 @@
                 __cfadbg_print_safe(io_core, "Kernel I/O : Slow poller stopped for cluster\n", &this);
 
-                #if defined(__CFA_IO_POLLING_USER__)
+                if( this.io->cltr_flags & CFA_CLUSTER_IO_POLLER_USER_THREAD ) {
                         verify( this.io->poller.fast.waiting );
                         verify( this.io->poller.fast.thrd.state == Blocked );
@@ -351 +355 @@
 
                         __cfadbg_print_safe(io_core, "Kernel I/O : Fast poller stopped for cluster\n", &this);
-                #endif
+                }
         }
 
@@ -473 +477 @@
                 __cfadbg_print_safe(io_core, "Kernel I/O : Slow poller for ring %p ready\n", &ring);
 
-                while(!__atomic_load_n(&ring.done, __ATOMIC_SEQ_CST)) {
-                        #if defined(__CFA_IO_POLLING_USER__)
-
+                if( ring.cltr_flags & CFA_CLUSTER_IO_POLLER_USER_THREAD ) {
+                        while(!__atomic_load_n(&ring.done, __ATOMIC_SEQ_CST)) {
                                 // In the user-thread approach drain and if anything was drained,
                                 // batton pass to the user-thread
@@ -491 +494 @@
                                         wait( ring.poller.sem );
                                 }
-
-                        #else
-
+                        }
+                }
+                else {
+                        while(!__atomic_load_n(&ring.done, __ATOMIC_SEQ_CST)) {
                                 //In the naive approach, just poll the io completion queue directly
                                 int count = __drain_io( ring, &mask, 1, true );
@@ -502 +506 @@
                                         ring.completion_q.stats.completed_avg.slow_cnt += 1;
                                 #endif
-
-                        #endif
+                        }
                 }
 
@@ -512 +515 @@
 
         void main( __io_poller_fast & this ) {
+                verify( this.ring->cltr_flags & CFA_CLUSTER_IO_POLLER_USER_THREAD );
+
                 // Start parked
                 park( __cfaabi_dbg_ctx );

libcfa/src/concurrency/kernel.cfa

@@ -254 +254 @@
 }
 
-void ?{}(cluster & this, const char name[], Duration preemption_rate) with( this ) {
+void ?{}(cluster & this, const char name[], Duration preemption_rate, int io_flags) with( this ) {
         this.name = name;
         this.preemption_rate = preemption_rate;
@@ -268 +268 @@
         threads{ __get };
 
-        __kernel_io_startup( this, &this == mainCluster );
+        __kernel_io_startup( this, io_flags, &this == mainCluster );
 
         doregister(this);

libcfa/src/concurrency/kernel.hfa

@@ -116 +116 @@
 struct __io_data;
 
+#define CFA_CLUSTER_IO_POLLER_USER_THREAD 1 << 0
+// #define CFA_CLUSTER_IO_POLLER_KERNEL_SIDE 1 << 1
+
 //-----------------------------------------------------------------------------
 // Cluster
@@ -156 +159 @@
 extern Duration default_preemption();
 
-void ?{} (cluster & this, const char name[], Duration preemption_rate);
+void ?{} (cluster & this, const char name[], Duration preemption_rate, int flags);
 void ^?{}(cluster & this);
 
-static inline void ?{} (cluster & this)                           { this{"Anonymous Cluster", default_preemption()}; }
-static inline void ?{} (cluster & this, Duration preemption_rate) { this{"Anonymous Cluster", preemption_rate}; }
-static inline void ?{} (cluster & this, const char name[])        { this{name, default_preemption()}; }
+static inline void ?{} (cluster & this)                                      { this{"Anonymous Cluster", default_preemption(), 0}; }
+static inline void ?{} (cluster & this, Duration preemption_rate)            { this{"Anonymous Cluster", preemption_rate, 0}; }
+static inline void ?{} (cluster & this, const char name[])                   { this{name, default_preemption(), 0}; }
+static inline void ?{} (cluster & this, int flags)                           { this{"Anonymous Cluster", default_preemption(), flags}; }
+static inline void ?{} (cluster & this, Duration preemption_rate, int flags) { this{"Anonymous Cluster", preemption_rate, flags}; }
+static inline void ?{} (cluster & this, const char name[], int flags)        { this{name, default_preemption(), flags}; }
 
 static inline [cluster *&, cluster *& ] __get( cluster & this ) __attribute__((const)) { return this.node.[next, prev]; }

libcfa/src/concurrency/kernel_private.hfa

@@ -77 +77 @@
 //-----------------------------------------------------------------------------
 // I/O
-void __kernel_io_startup     ( cluster &, bool );
+void __kernel_io_startup     ( cluster &, int, bool );
 void __kernel_io_finish_start( cluster & );
 void __kernel_io_prepare_stop( cluster & );