Added a message passing system based on lock free queues for inter-thread communication. Channels now buffer on the sending side, and no longer require blocking when sending. Lots of other refactoring and bug fixes.

6 files changed, 183 insertions, 8 deletions

diff --git a/src/test/run-pass/task-comm-0.rs b/src/test/run-pass/task-comm-0.rs
new file mode 100644
index 00000000..5992ba5c
--- /dev/null
+++ b/src/test/run-pass/task-comm-0.rs
@@ -0,0 +1,19 @@
+io fn main() -> () {
+    test05();
+}
+
+io fn test05_start(chan[int] ch) {
+    ch <| 10;
+    ch <| 20;
+    ch <| 30;
+}
+
+io fn test05() {
+    let port[int] po = port();
+    let chan[int] ch = chan(po);
+    spawn test05_start(chan(po));
+    let int value <- po;
+    value <- po;
+    value <- po;
+    log value;
+}
diff --git a/src/test/run-pass/task-comm-1.rs b/src/test/run-pass/task-comm-1.rs
new file mode 100644
index 00000000..48983b71
--- /dev/null
+++ b/src/test/run-pass/task-comm-1.rs
@@ -0,0 +1,13 @@
+fn main() -> () {
+   test00(); 
+}
+
+fn start() {
+    log "Started / Finished Task.";
+}
+
+fn test00() {
+    let task t = spawn thread start();
+    join t;
+    log "Completing.";
+}
\ No newline at end of file
diff --git a/src/test/run-pass/task-comm-2.rs b/src/test/run-pass/task-comm-2.rs
new file mode 100644
index 00000000..9151c7b1
--- /dev/null
+++ b/src/test/run-pass/task-comm-2.rs
@@ -0,0 +1,34 @@
+fn main() -> () {
+   log "===== THREADS =====";
+   test00(true);
+   log "====== TASKS ======";
+   test00(false);
+}
+
+fn start(int task_number) {
+    log "Started task.";
+    let int i = 0;
+    while (i < 10000) {
+        i = i + 1;
+    }
+    log "Finished task.";
+}
+    
+fn test00(bool create_threads) {
+    let int number_of_tasks = 32;
+    
+    let int i = 0;
+    let vec[task] tasks = vec();
+    while (i < number_of_tasks) {
+        i = i + 1;
+        if (create_threads) {
+            tasks += vec(spawn thread start(i));
+        } else {
+            tasks += vec(spawn start(i));
+        }
+    }
+    
+    for (task t in tasks) {
+        join t;
+    }
+}
\ No newline at end of file
diff --git a/src/test/run-pass/task-comm-3.rs b/src/test/run-pass/task-comm-3.rs
new file mode 100644
index 00000000..6dd620cc
--- /dev/null
+++ b/src/test/run-pass/task-comm-3.rs
@@ -0,0 +1,59 @@
+io fn main() -> () {
+   log "===== THREADS =====";
+   test00(false);
+}
+
+io fn test00_start(chan[int] ch, int message, int count) {
+    log "Starting test00_start";
+    let int i = 0;
+    while (i < count) {
+        log "Sending Message";
+        ch <| message;
+        i = i + 1;
+    }
+    log "Ending test00_start";
+}
+
+io fn test00(bool is_multithreaded) {
+    let int number_of_tasks = 1;
+    let int number_of_messages = 0;
+    log "Creating tasks";
+    
+    let port[int] po = port();
+    let chan[int] ch = chan(po);
+    
+    let int i = 0;
+    
+    // Create and spawn tasks...
+    let vec[task] tasks = vec();
+    while (i < number_of_tasks) {
+        i = i + 1;
+        if (is_multithreaded) {
+            tasks += vec(
+                spawn thread test00_start(ch, i, number_of_messages));
+        } else {
+            tasks += vec(spawn test00_start(ch, i, number_of_messages));
+        }
+    }
+    
+    // Read from spawned tasks...
+    let int sum = 0;
+    for (task t in tasks) {
+        i = 0;
+        while (i < number_of_messages) {
+            let int value <- po;
+            sum += value;
+            i = i + 1;
+        }
+    }
+
+    // Join spawned tasks...
+    for (task t in tasks) {
+        join t;
+    }
+    
+    log "Completed: Final number is: ";
+    check (sum + 1 == number_of_messages * 
+           (number_of_tasks * number_of_tasks + number_of_tasks) / 2);
+    log sum;
+}
\ No newline at end of file
diff --git a/src/test/run-pass/task-comm-4.rs b/src/test/run-pass/task-comm-4.rs
new file mode 100644
index 00000000..42ba6992
--- /dev/null
+++ b/src/test/run-pass/task-comm-4.rs
@@ -0,0 +1,10 @@
+io fn main() -> () {
+   test00();
+}
+
+io fn test00() {
+    let port[int] p = port();
+    let chan[int] c = chan(p);
+    c <| 42;
+    let int r <- p;
+}
\ No newline at end of file
diff --git a/src/test/run-pass/task-comm.rs b/src/test/run-pass/task-comm.rs
index 4a21b4e4..ef71c6e1 100644
--- a/src/test/run-pass/task-comm.rs
+++ b/src/test/run-pass/task-comm.rs
@@ -1,17 +1,19 @@
 
-
-io fn main() -> () {
-    test00(true);
+fn main() -> () {
+    // test00(true);
     // test01();
     // test02();
     // test03();
     // test04();
+    // test05();
+    test06();
 }
 
 io fn test00_start(chan[int] ch, int message, int count) {
     log "Starting test00_start";
     let int i = 0;
     while (i < count) {
+        log "Sending Message";
         ch <| message;
         i = i + 1;
     }
@@ -19,7 +21,7 @@ io fn test00_start(chan[int] ch, int message, int count) {
 }
 
 io fn test00(bool is_multithreaded) {
-    let int number_of_tasks = 4;
+    let int number_of_tasks = 1;
     let int number_of_messages = 64;
     log "Creating tasks";
     
@@ -109,12 +111,50 @@ fn test04() {
     log "Finishing up.";
 }
 
+io fn test05_start(chan[int] ch) {
+    ch <| 10;
+    ch <| 20;
+    ch <| 30;
+    ch <| 30;
+    ch <| 30;    
+}
 
+io fn test05() {
+    let port[int] po = port();
+    let chan[int] ch = chan(po);
+    spawn thread test05_start(ch);
+    let int value <- po;
+    value <- po;
+    value <- po;
+    log value;
+}
 
-
-
-
-
+fn test06_start(int task_number) {
+    log "Started task.";
+    let int i = 0;
+    while (i < 100000000) {
+        i = i + 1;    
+    }
+    log "Finished task.";
+}
+    
+fn test06() {
+    let int number_of_tasks = 32;
+    log "Creating tasks";
+    
+    let int i = 0;
+    
+    let vec[task] tasks = vec();
+    while (i < number_of_tasks) {
+        i = i + 1;
+        tasks += vec(spawn thread test06_start(i));
+        // tasks += vec(spawn test06_start(i));
+    }
+    
+    for (task t in tasks) {
+        join t;
+    }
+}