Changeset 5407cdc for tests/concurrent

Timestamp:

Apr 28, 2021, 4:56:50 PM (5 years ago)

Author:

Thierry Delisle <tdelisle@…>

Branches:

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

Children:

8d66610

Parents:

feacef9 (diff), b7fd2db6 (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:

tests/concurrent

Files:

• .expect/clib_tls.txt (added)
• .expect/semaphore.txt (added)
• .expect/spinaphore.txt (added)
• clib.c (modified) (6 diffs)
• clib_tls.c (added)
• coroutineYield.cfa (modified) (1 diff)
• futures/multi.cfa (modified) (4 diffs)
• semaphore.cfa (added)
• spinaphore.cfa (added)

tests/concurrent/clib.c

@@ -1 @@
-#include <clib/cfathread.h>
-
 #include <stdio.h>
 #include <stdlib.h>
+#include <clib/cfathread.h>
+#include <bits/defs.hfa>
+
+extern "C" {
+void _exit(int status);
+}
 
 thread_local struct drand48_data buffer = { 0 };
@@ -15 +19 @@
 cfathread_t volatile blocked[blocked_size];
 
-void Worker( cfathread_t this ) {
+void * Worker( void * ) {
         for(int i = 0; i < 1000; i++) {
                 int idx = myrand() % blocked_size;
@@ -22 +26 @@
                         cfathread_unpark( thrd );
                 } else {
-                        cfathread_t thrd = __atomic_exchange_n(&blocked[idx], this, __ATOMIC_SEQ_CST);
+                        cfathread_t thrd = __atomic_exchange_n(&blocked[idx], cfathread_self(), __ATOMIC_SEQ_CST);
                         cfathread_unpark( thrd );
                         cfathread_park();
@@ -28 +32 @@
         }
         printf("Done\n");
+        return NULL;
 }
 
 volatile bool stop;
-void Unparker( cfathread_t this ) {
+void * Unparker( void * ) {
         while(!stop) {
                 int idx = myrand() % blocked_size;
@@ -42 +47 @@
         }
         printf("Done Unparker\n");
+        return NULL;
 }
 
@@ -51 +57 @@
         }
 
-        cfathread_setproccnt( 4 );
-        cfathread_t u = cfathread_create( Unparker );
+        cfathread_cluster_t cl = cfathread_cluster_self();
+
+        cfathread_cluster_add_worker( cl, NULL, NULL, NULL );
+        cfathread_cluster_add_worker( cl, NULL, NULL, NULL );
+        cfathread_cluster_add_worker( cl, NULL, NULL, NULL );
+
+        cfathread_attr_t attr;
+        cfathread_attr_init(&attr);
+        cfathread_attr_setcluster(&attr, cl);
+
+        cfathread_t u;
+        cfathread_create( &u, &attr, Unparker, NULL );
         {
                 cfathread_t t[20];
                 for(int i = 0; i < 20; i++) {
-                        t[i] = cfathread_create( Worker );
+                        cfathread_create( &t[i], &attr, Worker, NULL );
                 }
                 for(int i = 0; i < 20; i++) {
-                        cfathread_join( t[i] );
+                        cfathread_join( t[i], NULL );
                 }
         }
         stop = true;
-        cfathread_join(u);
-        cfathread_setproccnt( 1 );
+        cfathread_join(u, NULL);
+        cfathread_attr_destroy(&attr);
+        fflush(stdout);
+        _exit(0);
 }

tests/concurrent/coroutineYield.cfa

@@ -38 +38 @@
 
 
+Coroutine c;
 int main(int argc, char* argv[]) {
-        Coroutine c;
         for(int i = 0; TEST(i < N); i++) {
                 #if !defined(TEST_FOREVER)

tests/concurrent/futures/multi.cfa

@@ -5 +5 @@
 
 thread Server {
-        int cnt, iteration;
+        int pending, done, iteration;
         multi_future(int) * request;
 };
 
 void ?{}( Server & this ) {
-        this.cnt = 0;
+        ((thread&)this){"Server Thread"};
+        this.pending = 0;
+        this.done = 0;
         this.iteration = 0;
         this.request = 0p;
@@ -16 +18 @@
 
 void ^?{}( Server & mutex this ) {
-        assert(this.cnt == 0);
-    this.request = 0p;
+        assert(this.pending == 0);
+        this.request = 0p;
 }
 
@@ -24 +26 @@
 }
 
-void process( Server & mutex this ) {
-        fulfil( *this.request, this.iteration );
-        this.iteration++;
+void call( Server & mutex this ) {
+        this.pending++;
 }
 
-void call( Server & mutex this ) {
-        this.cnt++;
+void finish( Server & mutex this ) {
+        this.done++;
 }
 
-void finish( Server & mutex this ) { }
-
 void main( Server & this ) {
+        MAIN_LOOP:
         for() {
                 waitfor( ^?{} : this ) {
                         break;
                 }
-                or when( this.cnt < NFUTURES ) waitfor( call: this ) {
-                        if (this.cnt == NFUTURES) {
-                                process(this);
+                or waitfor( call: this ) {
+                        if (this.pending != NFUTURES) { continue MAIN_LOOP; }
+
+                        this.pending = 0;
+                        fulfil( *this.request, this.iteration );
+                        this.iteration++;
+
+                        for(NFUTURES) {
+                                waitfor( finish: this );
                         }
-                }
-                or waitfor( finish: this ) {
-                        if (this.cnt == NFUTURES) {
-                                reset( *this.request );
-                                this.cnt = 0;
-                        }
+
+                        reset( *this.request );
+                        this.done = 0;
                 }
         }
@@ -57 +60 @@
 Server * the_server;
 thread Worker {};
+void ?{}(Worker & this) {
+        ((thread&)this){"Worker Thread"};
+}
+
 multi_future(int) * shared_future;