Changes in libcfa/src/concurrency/kernel.hfa [431cd4f:454f478]

File:

• libcfa/src/concurrency/kernel.hfa (modified) (12 diffs)

libcfa/src/concurrency/kernel.hfa

@@ -28 +28 @@
 }
 
+//-----------------------------------------------------------------------------
+// Underlying Locks
 #ifdef __CFA_WITH_VERIFY__
         extern bool __cfaabi_dbg_in_kernel();
 #endif
 
-//-----------------------------------------------------------------------------
-// I/O
-struct cluster;
-struct $io_context;
-struct $io_arbiter;
-
-struct io_context_params {
-        int num_entries;
-};
-
-void  ?{}(io_context_params & this);
+extern "C" {
+        char * strerror(int);
+}
+#define CHECKED(x) { int err = x; if( err != 0 ) abort("KERNEL ERROR: Operation \"" #x "\" return error %d - %s\n", err, strerror(err)); }
+
+struct __bin_sem_t {
+        pthread_mutex_t         lock;
+        pthread_cond_t          cond;
+        int                     val;
+};
+
+static inline void ?{}(__bin_sem_t & this) with( this ) {
+        // Create the mutex with error checking
+        pthread_mutexattr_t mattr;
+        pthread_mutexattr_init( &mattr );
+        pthread_mutexattr_settype( &mattr, PTHREAD_MUTEX_ERRORCHECK_NP);
+        pthread_mutex_init(&lock, &mattr);
+
+        pthread_cond_init (&cond, (const pthread_condattr_t *)0p);  // workaround trac#208: cast should not be required
+        val = 0;
+}
+
+static inline void ^?{}(__bin_sem_t & this) with( this ) {
+        CHECKED( pthread_mutex_destroy(&lock) );
+        CHECKED( pthread_cond_destroy (&cond) );
+}
+
+static inline void wait(__bin_sem_t & this) with( this ) {
+        verify(__cfaabi_dbg_in_kernel());
+        CHECKED( pthread_mutex_lock(&lock) );
+                while(val < 1) {
+                        pthread_cond_wait(&cond, &lock);
+                }
+                val -= 1;
+        CHECKED( pthread_mutex_unlock(&lock) );
+}
+
+static inline bool post(__bin_sem_t & this) with( this ) {
+        bool needs_signal = false;
+
+        CHECKED( pthread_mutex_lock(&lock) );
+                if(val < 1) {
+                        val += 1;
+                        pthread_cond_signal(&cond);
+                        needs_signal = true;
+                }
+        CHECKED( pthread_mutex_unlock(&lock) );
+
+        return needs_signal;
+}
+
+#undef CHECKED
+
 
 //-----------------------------------------------------------------------------
@@ -51 +95 @@
 struct __processor_id_t {
         unsigned id:24;
+        bool full_proc:1;
 
         #if !defined(__CFA_NO_STATISTICS__)
@@ -69 +114 @@
         struct cluster * cltr;
 
-        // Ready Queue state per processor
-        struct {
-                unsigned short its;
-                unsigned short itr;
-                unsigned id;
-                unsigned target;
-                unsigned long long int cutoff;
-        } rdq;
-
         // Set to true to notify the processor should terminate
         volatile bool do_terminate;
@@ -89 +125 @@
         // Handle to pthreads
         pthread_t kernel_thread;
-
-        struct {
-                $io_context * ctx;
-                bool pending;
-                bool dirty;
-        } io;
 
         // Preemption data
@@ -104 +134 @@
 
         // Idle lock (kernel semaphore)
-        int idle;
+        __bin_sem_t idle;
 
         // Termination synchronisation (user semaphore)
@@ -114 +144 @@
         // Link lists fields
         DLISTED_MGD_IMPL_IN(processor)
-
-        // special init fields
-        // This is needed for memcached integration
-        // once memcached experiments are done this should probably be removed
-        // it is not a particularly safe scheme as it can make processors less homogeneous
-        struct {
-                $thread * thrd;
-        } init;
 
         #if !defined(__CFA_NO_STATISTICS__)
@@ -137 +159 @@
 void ^?{}(processor & this);
 
-static inline void  ?{}(processor & this)                        { this{ "Anonymous Processor", *mainCluster}; }
-static inline void  ?{}(processor & this, struct cluster & cltr) { this{ "Anonymous Processor", cltr}; }
-static inline void  ?{}(processor & this, const char name[])     { this{name, *mainCluster}; }
+static inline void  ?{}(processor & this)                    { this{ "Anonymous Processor", *mainCluster}; }
+static inline void  ?{}(processor & this, struct cluster & cltr)    { this{ "Anonymous Processor", cltr}; }
+static inline void  ?{}(processor & this, const char name[]) { this{name, *mainCluster }; }
 
 DLISTED_MGD_IMPL_OUT(processor)
 
 //-----------------------------------------------------------------------------
+// I/O
+struct __io_data;
+
+// IO poller user-thread
+// Not using the "thread" keyword because we want to control
+// more carefully when to start/stop it
+struct $io_ctx_thread {
+        struct __io_data * ring;
+        single_sem sem;
+        volatile bool done;
+        $thread self;
+};
+
+
+struct io_context {
+        $io_ctx_thread thrd;
+};
+
+struct io_context_params {
+        int num_entries;
+        int num_ready;
+        int submit_aff;
+        bool eager_submits:1;
+        bool poller_submits:1;
+        bool poll_submit:1;
+        bool poll_complete:1;
+};
+
+void  ?{}(io_context_params & this);
+
+void  ?{}(io_context & this, struct cluster & cl);
+void  ?{}(io_context & this, struct cluster & cl, const io_context_params & params);
+void ^?{}(io_context & this);
+
+struct io_cancellation {
+        __u64 target;
+};
+
+static inline void  ?{}(io_cancellation & this) { this.target = -1u; }
+static inline void ^?{}(io_cancellation &) {}
+bool cancel(io_cancellation & this);
+
+//-----------------------------------------------------------------------------
 // Cluster Tools
 
-// Intrusives lanes which are used by the ready queue
+// Intrusives lanes which are used by the relaxed ready queue
 struct __attribute__((aligned(128))) __intrusive_lane_t;
 void  ?{}(__intrusive_lane_t & this);
 void ^?{}(__intrusive_lane_t & this);
 
-// Aligned timestamps which are used by the relaxed ready queue
-struct __attribute__((aligned(128))) __timestamp_t;
-void  ?{}(__timestamp_t & this);
-void ^?{}(__timestamp_t & this);
+// Counter used for wether or not the lanes are all empty
+struct __attribute__((aligned(128))) __snzi_node_t;
+struct __snzi_t {
+        unsigned mask;
+        int root;
+        __snzi_node_t * nodes;
+};
+
+void  ?{}( __snzi_t & this, unsigned depth );
+void ^?{}( __snzi_t & this );
 
 //TODO adjust cache size to ARCHITECTURE
 // Structure holding the relaxed ready queue
 struct __ready_queue_t {
+        // Data tracking how many/which lanes are used
+        // Aligned to 128 for cache locality
+        __snzi_t snzi;
+
         // Data tracking the actual lanes
         // On a seperate cacheline from the used struct since
@@ -167 +242 @@
                 __intrusive_lane_t * volatile data;
 
-                // Array of times
-                __timestamp_t * volatile tscs;
-
                 // Number of lanes (empty or not)
                 volatile size_t count;
@@ -179 +251 @@
 
 // Idle Sleep
-struct __cluster_proc_list {
+struct __cluster_idles {
         // Spin lock protecting the queue
         volatile uint64_t lock;
@@ -190 +262 @@
 
         // List of idle processors
-        dlist(processor, processor) idles;
-
-        // List of active processors
-        dlist(processor, processor) actives;
+        dlist(processor, processor) list;
 };
 
@@ -209 +278 @@
 
         // List of idle processors
-        __cluster_proc_list procs;
+        __cluster_idles idles;
 
         // List of threads
@@ -223 +292 @@
 
         struct {
-                $io_arbiter * arbiter;
-                io_context_params params;
+                io_context * ctxs;
+                unsigned cnt;
         } io;