Changeset ccf1d99

Timestamp:

Feb 2, 2023, 11:08:48 AM (22 months ago)

Author:

caparsons <caparson@…>

Branches:

ADT, ast-experimental, master

Children:

2d028039, 98a2b1dc

Parents:

a64137f

Message:

intermediate push so I can move to nasus to performance test some stuff

Files:

• libcfa/src/concurrency/actor.hfa (modified) (8 diffs)
• src/Concurrency/Actors.cpp (modified) (2 diffs)
• tests/concurrent/actors/executor.cfa (modified) (1 diff)

libcfa/src/concurrency/actor.hfa

@@ -24 +24 @@
 // Define if executor is created in a separate cluster
 #define __DEFAULT_EXECUTOR_SEPCLUS__ false
+
+// when you flip this make sure to recompile compiler and flip the appropriate flag there too in Actors.cpp
+#define __ALLOC 0
 
 // forward decls
@@ -58 +61 @@
 struct copy_queue {
     dlist( request ) list;
+    #if ! __ALLOC
     request * buffer;
     size_t count, buffer_size, index;
+    #endif
 };
 static inline void ?{}( copy_queue & this ) {}
 static inline void ?{}( copy_queue & this, size_t buf_size ) with(this) { 
     list{};
+    #if ! __ALLOC
     buffer_size = buf_size;
     buffer = aalloc( buffer_size );
     count = 0;
     index = 0;
-}
-static inline void ^?{}( copy_queue & this ) with(this) { adelete(buffer); }
+    #endif
+}
+static inline void ^?{}( copy_queue & this ) with(this) {
+    #if ! __ALLOC
+    adelete(buffer);
+    #endif
+}
 
 static inline void insert( copy_queue & this, request & elem ) with(this) {
+    #if ! __ALLOC
     if ( count < buffer_size ) { // fast path ( no alloc )
         buffer[count]{ elem };
@@ -80 +92 @@
     (*new_elem){ elem };
     insert_last( list, *new_elem );
+    #else
+    insert_last( list, elem );
+    #endif
 }
 
@@ -86 +101 @@
 // should_delete is an output param
 static inline request & remove( copy_queue & this, bool & should_delete ) with(this) {
+    #if ! __ALLOC
     if ( count > 0 ) {
         count--;
@@ -93 +109 @@
         return buffer[old_idx];
     }
+    #endif
     should_delete = true;
     return try_pop_front( list );
 }
 
-static inline bool isEmpty( copy_queue & this ) with(this) { return count == 0 && list`isEmpty; }
+static inline bool isEmpty( copy_queue & this ) with(this) {
+    #if ! __ALLOC
+    return count == 0 && list`isEmpty;
+    #else
+    return list`isEmpty;
+    #endif
+}
 
 static size_t __buffer_size = 10; // C_TODO: rework this to be passed from executor through ctors (no need for global)
 struct work_queue {
-    futex_mutex mutex_lock; 
+    __spinlock_t mutex_lock;
+    copy_queue owned_queue;
     copy_queue * c_queue; // C_TODO: try putting this on the stack with ptr juggling
+
 }; // work_queue
 static inline void ?{}( work_queue & this ) with(this) { 
-    c_queue = alloc();
-    (*c_queue){ __buffer_size }; // C_TODO: support passing copy buff size as arg to executor
-}
-static inline void ^?{}( work_queue & this ) with(this) { delete( c_queue ); }
+    // c_queue = alloc();
+    // (*c_queue){ __buffer_size };
+    owned_queue{ __buffer_size };
+    c_queue = &owned_queue;
+}
+// static inline void ^?{}( work_queue & this ) with(this) { delete( c_queue ); }
 
 static inline void insert( work_queue & this, request & elem ) with(this) {
-    lock( mutex_lock );
+    lock( mutex_lock __cfaabi_dbg_ctx2 );
     insert( *c_queue, elem );
     unlock( mutex_lock );
@@ -117 +144 @@
 
 static inline void transfer( work_queue & this, copy_queue ** transfer_to ) with(this) {
-    lock( mutex_lock );
+    lock( mutex_lock __cfaabi_dbg_ctx2 );
     // swap copy queue ptrs
     copy_queue * temp = *transfer_to;
@@ -126 +153 @@
 
 thread worker {
+    copy_queue owned_queue;
     work_queue * request_queues;
     copy_queue * current_queue;
@@ -135 +163 @@
     ((thread &)this){ clu };
     this.request_queues = request_queues;
-    this.current_queue = alloc();
-    (*this.current_queue){ __buffer_size };
+    // this.current_queue = alloc();
+    // (*this.current_queue){ __buffer_size };
+    this.owned_queue{ __buffer_size };
+    this.current_queue = &this.owned_queue;
     this.start = start;
     this.range = range;
 }
-static inline void ^?{}( worker & mutex this ) with(this) { delete( current_queue ); }
+// static inline void ^?{}( worker & mutex this ) with(this) { delete( current_queue ); }
 
 struct executor {

src/Concurrency/Actors.cpp

@@ -180 +180 @@
 };
 
-#define __ALLOC 0
+#define __ALLOC 0 // C_TODO: complete swap to no-alloc version
 
 struct GenReceiveDecls : public ast::WithDeclsToAdd<> {
@@ -225 +225 @@
                     return receiver;
                 }
-            */
+            */ // C_TODO: update this with new no alloc version
             CompoundStmt * sendBody = new CompoundStmt( decl->location );
 

tests/concurrent/actors/executor.cfa

@@ -86 +86 @@
 
     
-    executor e{ Processors, Processors, Processors == 1 ? 1 : Processors * 512, true, BufSize };
+    executor e{ Processors, Processors, Processors == 1 ? 1 : Processors * 512, true };
 
     printf("starting\n");