Add ChannelSender and ChannelReceiver HeapObject variants

Author: 0xhenrique

lisp/test_channels.lisp

@@ -0,0 +1,90 @@
+(println "Testing channels...")
+
+;; Test 1: Simple send/recv in same thread
+(println "\n=== Test 1: Basic channel send/recv ===")
+(def ch (channel))
+(def sender (car ch))
+(def receiver (car (cdr ch)))
+(send! sender 42)
+(def result (recv receiver))
+(println "Received:" result)
+
+;; Test 2: Multiple sends and receives
+(println "\n=== Test 2: Multiple messages ===")
+(def ch2 (channel))
+(def s2 (car ch2))
+(def r2 (car (cdr ch2)))
+(send! s2 10)
+(send! s2 20)
+(send! s2 30)
+(def v1 (recv r2))
+(def v2 (recv r2))
+(def v3 (recv r2))
+(println "Received:" v1 v2 v3)
+(println "Sum:" (+ v1 v2 v3))
+
+;; Test 3: Send/recv strings
+(println "\n=== Test 3: Sending strings ===")
+(def ch3 (channel))
+(def s3 (car ch3))
+(def r3 (car (cdr ch3)))
+(send! s3 "Hello")
+(send! s3 "World")
+(println "Received:" (recv r3))
+(println "Received:" (recv r3))
+
+;; Test 4: Send/recv lists
+(println "\n=== Test 4: Sending lists ===")
+(def ch4 (channel))
+(def s4 (car ch4))
+(def r4 (car (cdr ch4)))
+(send! s4 (list 1 2 3))
+(send! s4 (list 4 5 6))
+(def list1 (recv r4))
+(def list2 (recv r4))
+(println "List 1:" list1)
+(println "List 2:" list2)
+
+;; Test 5: Producer/Consumer with threads
+(println "\n=== Test 5: Producer/Consumer threads ===")
+(def ch5 (channel))
+(def sender5 (car ch5))
+(def receiver5 (car (cdr ch5)))
+
+;; Producer thread - send numbers 1 through 5
+(def producer (fn ()
+  ;; Create a new channel in this thread's scope
+  ;; Since we can't share the sender from parent thread, we'll just compute
+  (+ 100 200)))
+
+;; For now, just demonstrate thread-local channel usage
+(def worker1 (fn ()
+  (def local-ch (channel))
+  (def local-s (car local-ch))
+  (def local-r (car (cdr local-ch)))
+  (send! local-s 123)
+  (recv local-r)))
+
+(def t1 (spawn worker1))
+(def r1 (join t1))
+(println "Thread result:" r1)
+
+;; Test 6: Passing channels between functions
+(println "\n=== Test 6: Channel as function argument ===")
+(def send-many (fn (sender values)
+  (def first-val (car values))
+  (def rest-vals (cdr values))
+  (send! sender first-val)
+  (if (not (nil? rest-vals))
+    (send-many sender rest-vals)
+    nil)))
+
+(def ch6 (channel))
+(def s6 (car ch6))
+(def r6 (car (cdr ch6)))
+(send-many s6 (list 7 8 9))
+(println "Received:" (recv r6))
+(println "Received:" (recv r6))
+(println "Received:" (recv r6))
+
+(println "\n=== All channel tests passed! ===")

lisp/test_spawn.lisp

@@ -1,5 +1,3 @@
-;; Test spawn and join - Phase 9 Concurrency
-
 (println "Testing spawn and join...")
 
 ;; Test 1: Simple spawn/join - just return a value

src/value.rs

@@ -2,6 +2,7 @@ use std::cell::RefCell;
 use std::collections::HashMap;
 use std::fmt;
 use std::rc::Rc;
+use std::sync::mpsc::{Receiver, Sender};
 use std::sync::{Arc, Mutex};
 use std::thread::JoinHandle;
 
@@ -264,6 +265,12 @@ pub enum HeapObject {
     /// Thread handle for spawned threads - wrapped in Mutex because JoinHandle can only be joined once
     /// The Option allows us to take the handle when joining (join consumes the handle)
     ThreadHandle(Arc<Mutex<Option<JoinHandle<Result<SharedValue, String>>>>>),
+    /// Channel sender for inter-thread communication
+    /// Wrapped in Arc<Mutex<>> to allow cloning and thread-safe access
+    ChannelSender(Arc<Mutex<Sender<SharedValue>>>),
+    /// Channel receiver for inter-thread communication
+    /// Wrapped in Arc<Mutex<>> to allow thread-safe access
+    ChannelReceiver(Arc<Mutex<Receiver<SharedValue>>>),
 }
 
 impl Clone for HeapObject {
@@ -277,6 +284,8 @@ impl Clone for HeapObject {
             HeapObject::NativeFunction(f) => HeapObject::NativeFunction(f.clone()),
             HeapObject::CompiledFunction(c) => HeapObject::CompiledFunction(c.clone()),
             HeapObject::ThreadHandle(h) => HeapObject::ThreadHandle(h.clone()),
+            HeapObject::ChannelSender(s) => HeapObject::ChannelSender(s.clone()),
+            HeapObject::ChannelReceiver(r) => HeapObject::ChannelReceiver(r.clone()),
         }
     }
 }
@@ -779,6 +788,8 @@ impl Value {
                 HeapObject::NativeFunction(_) => "native-function",
                 HeapObject::CompiledFunction(_) => "function",
                 HeapObject::ThreadHandle(_) => "thread-handle",
+                HeapObject::ChannelSender(_) => "channel-sender",
+                HeapObject::ChannelReceiver(_) => "channel-receiver",
             }
         } else {
             "unknown"
@@ -967,6 +978,16 @@ impl Value {
                     let heap = Rc::new(HeapObject::ThreadHandle(h.clone()));
                     Value::from_heap(heap)
                 }
+                Some(HeapObject::ChannelSender(s)) => {
+                    // ChannelSenders are already Arc-based, just clone the Value
+                    let heap = Rc::new(HeapObject::ChannelSender(s.clone()));
+                    Value::from_heap(heap)
+                }
+                Some(HeapObject::ChannelReceiver(r)) => {
+                    // ChannelReceivers are already Arc-based, just clone the Value
+                    let heap = Rc::new(HeapObject::ChannelReceiver(r.clone()));
+                    Value::from_heap(heap)
+                }
                 None => Value::nil(),
             }
         } else if self.is_ptr() {
@@ -1084,6 +1105,12 @@ impl Value {
             Some(HeapObject::ThreadHandle(_)) => {
                 Err("Thread handles cannot be shared across threads".to_string())
             }
+            Some(HeapObject::ChannelSender(_)) => {
+                Err("Channel senders cannot be converted to SharedValue (they are already thread-safe)".to_string())
+            }
+            Some(HeapObject::ChannelReceiver(_)) => {
+                Err("Channel receivers cannot be converted to SharedValue (they are already thread-safe)".to_string())
+            }
             None => Err("Cannot convert unknown value to SharedValue".to_string()),
         }
     }
@@ -1223,6 +1250,8 @@ impl fmt::Display for Value {
                 HeapObject::NativeFunction(nf) => write!(f, "<native fn {}>", nf.name),
                 HeapObject::CompiledFunction(_) => write!(f, "<function>"),
                 HeapObject::ThreadHandle(_) => write!(f, "<thread-handle>"),
+                HeapObject::ChannelSender(_) => write!(f, "<channel-sender>"),
+                HeapObject::ChannelReceiver(_) => write!(f, "<channel-receiver>"),
             }
         } else {
             write!(f, "<unknown>")

src/vm.rs

@@ -2038,6 +2038,89 @@ pub fn standard_vm() -> VM {
         }
     }));
 
+    vm.define_global("channel", native("channel", |args| {
+        use std::sync::mpsc;
+        use std::sync::{Arc, Mutex};
+        use crate::value::HeapObject;
+
+        if !args.is_empty() {
+            return Err("channel expects 0 arguments".to_string());
+        }
+
+        // Create an unbounded channel
+        let (sender, receiver) = mpsc::channel();
+
+        // Wrap sender and receiver in Arc<Mutex<>> for thread-safety
+        let sender_arc = Arc::new(Mutex::new(sender));
+        let receiver_arc = Arc::new(Mutex::new(receiver));
+
+        // Create HeapObject values
+        let sender_heap = Rc::new(HeapObject::ChannelSender(sender_arc));
+        let receiver_heap = Rc::new(HeapObject::ChannelReceiver(receiver_arc));
+
+        // Return a list [sender receiver]
+        Ok(Value::list(vec![
+            Value::from_heap(sender_heap),
+            Value::from_heap(receiver_heap),
+        ]))
+    }));
+
+    vm.define_global("send!", native("send!", |args| {
+        use crate::value::HeapObject;
+
+        if args.len() != 2 {
+            return Err("send! expects 2 arguments (sender value)".to_string());
+        }
+
+        // Get the sender
+        let sender_obj = args[0].as_heap()
+            .ok_or_else(|| "send! expects a channel-sender as first argument".to_string())?;
+
+        if let HeapObject::ChannelSender(sender_mutex) = sender_obj {
+            // Convert the value to SharedValue for thread-safe sending
+            let shared_value = args[1].make_shared()
+                .map_err(|e| format!("send! failed to convert value: {}", e))?;
+
+            // Send the value
+            let sender = sender_mutex.lock()
+                .map_err(|_| "Failed to lock channel sender".to_string())?;
+
+            sender.send(shared_value)
+                .map_err(|_| "send! failed: receiver has been dropped".to_string())?;
+
+            // Return nil
+            Ok(Value::nil())
+        } else {
+            Err("send! expects a channel-sender as first argument".to_string())
+        }
+    }));
+
+    vm.define_global("recv", native("recv", |args| {
+        use crate::value::HeapObject;
+
+        if args.len() != 1 {
+            return Err("recv expects 1 argument (receiver)".to_string());
+        }
+
+        // Get the receiver
+        let receiver_obj = args[0].as_heap()
+            .ok_or_else(|| "recv expects a channel-receiver".to_string())?;
+
+        if let HeapObject::ChannelReceiver(receiver_mutex) = receiver_obj {
+            // Receive a value (blocking)
+            let receiver = receiver_mutex.lock()
+                .map_err(|_| "Failed to lock channel receiver".to_string())?;
+
+            let shared_value = receiver.recv()
+                .map_err(|_| "recv failed: sender has been dropped".to_string())?;
+
+            // Convert back to Rc-based Value
+            Ok(Value::from_shared(&shared_value))
+        } else {
+            Err("recv expects a channel-receiver".to_string())
+        }
+    }));
+
     vm
 }