some more handling

cogs/repl/repl-client.scm

@@ -7,6 +7,11 @@
 (define sender (channels-sender channels))
 (define receiver (channels-receiver channels))
 
+;; After every input if we should shutdown.
+(define shutdown (channels/new))
+(define shutdown-sender (channels-sender shutdown))
+(define shutdown-receiver (channels-receiver shutdown))
+
 (define MAGIC-START (string->bytes "%start-repl#%"))
 (define MAGIC-START-LENGTH (bytes-length MAGIC-START))
 
@@ -25,38 +30,43 @@
   (define reader (tcp-stream-reader stream))
   (define writer (tcp-stream-writer stream))
 
+  ;; Print out the startup message for the repl, and then
+  ;; we'll enter the event loop waiting for input.
+  (#%repl-display-startup)
   (define rl (#%create-repl))
 
   (define buffer (bytevector))
 
   (define (loop)
-
     (define next (read-char reader))
+    (cond
+      [(equal? next #\#)
+       (let ([maybe-magic (read-bytes MAGIC-START-LENGTH reader)])
+         (if (equal? maybe-magic MAGIC-START)
+             (let ([size (read-size buffer reader)])
 
-    (if (equal? next #\#)
-        (let ([maybe-magic (read-bytes MAGIC-START-LENGTH reader)])
-          (if (equal? maybe-magic MAGIC-START)
-              (let ([size (read-size buffer reader)])
+               (bytes-clear! buffer)
 
-                (bytes-clear! buffer)
+               ;; Read the next value
+               (define next-value (read-bytes size reader))
+               (define value-as-string (bytes->string/utf8 next-value))
 
-                ;; Read the next value
-                (define next-value (read-bytes size reader))
-                (define value-as-string (bytes->string/utf8 next-value))
+               (unless (equal? "#<void>" value-as-string)
+                 (display "=> ")
+                 (display value-as-string)
+                 (newline))
 
-                (unless (equal? "#<void>" value-as-string)
-                  (display "=> ")
-                  (display value-as-string)
-                  (newline))
+               (channel/send sender #t))
+             ;; Next should be the length, until the next newline
+             (begin
+               (write-char next (current-output-port))
+               (display (bytes->string/utf8 maybe-magic)))))]
 
-                (channel/send sender #t))
-              ;; Next should be the length, until the next newline
+      [(eof-object? next)
+       (displayln "Connection closed.")
+       (return! void)]
 
-              (begin
-                (write-char next (current-output-port))
-                (display (bytes->string/utf8 maybe-magic)))))
-
-        (write-char next (current-output-port)))
+      [else (write-char next (current-output-port))])
 
     ;; Remote repl... add bindings to readline?
     ;; That could help?
@@ -65,26 +75,28 @@
     (loop))
 
   (define (driver)
-    (with-handler (lambda (err) (displayln err)) (loop)))
+    (with-handler (lambda (err)
+                    (displayln err)
+                    (channel/send shutdown-sender #t))
+                  (loop)))
 
   (spawn-native-thread driver)
 
   ;; Read input, send on the stream
   (define (input-loop)
-    ;; If previous expression is still running:
-
-    ;; Do the thing
     (define input (#%read-line rl))
+    (#%repl-add-history-entry rl input)
 
     ;; Show the error, go again
-    (with-handler (lambda (err)
-                    (displayln err)
-                    (input-loop))
+    (with-handler (lambda (err) (displayln err))
                   (write (read (open-input-string input)) writer)
                   (newline writer)
-                  (channel/recv receiver)
-                  ;; Wait for acknowledgement?
-                  (input-loop)))
+                  ;; Selection. Wait on either an acknowledgement, or a shutdown?
+                  (define result (receivers-select receiver shutdown-receiver))
+                  (case result
+                    [(0) (input-loop)]
+                    [(1) (displayln "Shutting down")]
+                    [else void])))
   (input-loop))
 
 (define (main)

cogs/repl/repl.scm

@@ -3,6 +3,29 @@
 
 (provide repl-serve)
 
+;; Thread id -> TCP reader. Remove when finished.
+(define #%repl-manager (hash))
+
+(define (#%mark-connected tcp-stream)
+  (set! #%repl-manager (hash-insert #%repl-manager (current-thread-id) tcp-stream)))
+
+(define (#%set-thread-closed!)
+  (define connection (hash-get #%repl-manager (current-thread-id)))
+  ;; Close it all down
+  (tcp-shutdown! connection)
+  (set! #%repl-manager (hash-remove #%repl-manager (current-thread-id)))
+
+  ; (display "...Shutting down thread" (stdout))
+  ; (newline (stdout))
+  )
+
+(define (#%shutdown?)
+  (not (hash-contains? #%repl-manager (current-thread-id))))
+
+(define (quit)
+  ;; Attempt to set this thread closed no matter what.
+  (with-handler (lambda (_) void) (#%set-thread-closed!)))
+
 (define (repl-serve [port 8080] [thread-pool-size 2])
   (define listener (tcp-listen (string-append "0.0.0.0:" (int->string port))))
   (define tp (make-thread-pool thread-pool-size))
@@ -30,6 +53,12 @@
             ;; the stack trace is maintained?
             (define result (with-handler (lambda (err) (to-string err)) (eval expr)))
 
+            ;; Close the thread.
+            (when (#%shutdown?)
+              (close-output-port writer-port)
+              (close-input-port reader-port)
+              (return! #t))
+
             ;; TODO: Merge this into one display call.
             ;; It all has to come through in one fell swoop, such that
             ;; return values are atomic on the output stream.
@@ -42,6 +71,13 @@
              (string-append "#%start-repl#%" (int->string (string-length formatted)) "\n" formatted))
 
             (repl-loop))))
+
+      (#%mark-connected input-stream)
+
       ;; Set up the repl to also grab std out
       (parameterize ([current-output-port writer-port])
-        (repl-loop))))))
+        (repl-loop))
+
+      (displayln "Closing connection.")))))
+
+(repl-serve)

crates/steel-core/src/primitives/hashmaps.rs

@@ -36,7 +36,8 @@ pub(crate) fn hashmap_module() -> BuiltInModule {
         .register_native_fn_definition(VALUES_TO_VECTOR_DEFINITION)
         .register_native_fn_definition(CLEAR_DEFINITION)
         .register_native_fn_definition(HM_EMPTY_DEFINITION)
-        .register_native_fn_definition(HM_UNION_DEFINITION);
+        .register_native_fn_definition(HM_UNION_DEFINITION)
+        .register_native_fn_definition(HASH_REMOVE_DEFINITION);
     module
 }
 
@@ -108,6 +109,44 @@ pub fn hm_construct_keywords(args: &[SteelVal]) -> Result<SteelVal> {
     Ok(SteelVal::HashMapV(Gc::new(hm).into()))
 }
 
+/// Returns a new hashmap with the given key removed. Performs a functional
+/// update, so the old hash map is still available with the original key value pair.
+///
+/// (hash-remove map key) -> hash?
+///
+/// * map : hash?
+/// * key : any/c
+///
+/// # Examples
+/// ```scheme
+/// > (hash-remove (hash 'a 10 'b 20) 'a)
+///
+/// => '#hash(('b . 20))
+/// ```
+#[function(name = "hash-remove")]
+pub fn hash_remove(map: &mut SteelVal, key: SteelVal) -> Result<SteelVal> {
+    if key.is_hashable() {
+        if let SteelVal::HashMapV(SteelHashMap(ref mut m)) = map {
+            match Gc::get_mut(m) {
+                Some(m) => {
+                    m.remove(&key);
+                    Ok(std::mem::replace(map, SteelVal::Void))
+                }
+                None => {
+                    let mut m = m.unwrap();
+                    m.remove(&key);
+
+                    Ok(SteelVal::HashMapV(Gc::new(m).into()))
+                }
+            }
+        } else {
+            stop!(TypeMismatch => "hash-insert expects a hash map, found: {:?}", map);
+        }
+    } else {
+        stop!(TypeMismatch => "hash key not hashable: {:?}", key)
+    }
+}
+
 /// Returns a new hashmap with the additional key value pair added. Performs a functional update,
 /// so the old hash map is still accessible.
 ///

crates/steel-core/src/primitives/tcp.rs

@@ -23,6 +23,13 @@ pub fn tcp_connect(addr: SteelString) -> Result<SteelVal> {
     TcpStream::connect(addr.as_str())?.into_steelval()
 }
 
+#[function(name = "tcp-shutdown!")]
+pub fn tcp_close(stream: &SteelVal) -> Result<SteelVal> {
+    let writer = TcpStream::as_ref(stream)?.try_clone().unwrap();
+    writer.shutdown(std::net::Shutdown::Both)?;
+    Ok(SteelVal::Void)
+}
+
 #[function(name = "tcp-stream-writer")]
 pub fn tcp_input_port(stream: &SteelVal) -> Result<SteelVal> {
     let writer = TcpStream::as_ref(stream)?.try_clone().unwrap();
@@ -82,6 +89,7 @@ pub fn tcp_module() -> BuiltInModule {
 
     module
         .register_native_fn_definition(TCP_CONNECT_DEFINITION)
+        .register_native_fn_definition(TCP_CLOSE_DEFINITION)
         .register_native_fn_definition(TCP_INPUT_PORT_DEFINITION)
         .register_native_fn_definition(TCP_OUTPUT_PORT_DEFINITION)
         .register_native_fn_definition(TCP_BUFFERED_OUTPUT_PORT_DEFINITION)

crates/steel-core/src/steel_vm/vm.rs

@@ -1057,7 +1057,10 @@ impl ContinuationMark {
 
                 #[cfg(debug_assertions)]
                 {
-                    debug_assert_eq!(open.closed_continuation.stack, continuation.stack);
+                    debug_assert_eq!(
+                        open.closed_continuation.stack.len(),
+                        continuation.stack.len()
+                    );
                     debug_assert_eq!(
                         open.closed_continuation.stack_frames.len(),
                         continuation.stack_frames.len()
@@ -1170,7 +1173,10 @@ impl Continuation {
                                 ctx.instructions,
                                 open.closed_continuation.instructions
                             );
-                            debug_assert_eq!(ctx.thread.stack, open.closed_continuation.stack);
+                            debug_assert_eq!(
+                                ctx.thread.stack.len(),
+                                open.closed_continuation.stack.len()
+                            );
 
                             debug_assert_eq!(ctx.pop_count, open.closed_continuation.pop_count);
 

crates/steel-core/src/values/port.rs

@@ -272,7 +272,15 @@ impl SteelPortRepr {
             SteelPortRepr::ChildStdError(output) => output.read_exact(&mut byte),
             SteelPortRepr::StringInput(reader) => reader.read_exact(&mut byte),
             SteelPortRepr::DynReader(reader) => reader.read_exact(&mut byte),
-            SteelPortRepr::TcpStream(t) => t.read(&mut byte).map(|_| ()),
+            SteelPortRepr::TcpStream(t) => {
+                let amount = t.read(&mut byte)?;
+
+                if amount == 0 {
+                    stop!(Generic => "unexpected eof");
+                } else {
+                    Ok(())
+                }
+            }
             SteelPortRepr::FileOutput(_, _)
             | SteelPortRepr::StdOutput(_)
             | SteelPortRepr::StdError(_)

crates/steel-repl/Cargo.toml

@@ -17,6 +17,7 @@ steel-core = { workspace = true }
 steel-parser = { path = "../steel-parser", version = "0.6.0"}
 dirs = "5.0.1"
 ctrlc = "3.4.4"
+crossbeam = "0.8.4"
 
 [features]
 interrupt = ["steel-core/interrupt"]

crates/steel-repl/src/highlight.rs

@@ -2,7 +2,7 @@ extern crate rustyline;
 use colored::*;
 use steel_parser::parser::SourceId;
 
-use std::{cell::RefCell, rc::Rc};
+use std::sync::{Arc, Mutex};
 
 use rustyline::highlight::Highlighter;
 use rustyline::validate::{ValidationContext, ValidationResult, Validator};
@@ -25,18 +25,15 @@ impl Completer for RustylineHelper {
 
 #[derive(Helper)]
 pub struct RustylineHelper {
-    engine: Rc<RefCell<Engine>>,
-    bracket: std::cell::Cell<Option<(u8, usize)>>, // keywords: HashSet<&'static str>,
+    engine: Arc<Mutex<Engine>>,
+    bracket: crossbeam::atomic::AtomicCell<Option<(u8, usize)>>, // keywords: HashSet<&'static str>,
 }
 
-unsafe impl Send for RustylineHelper {}
-unsafe impl Sync for RustylineHelper {}
-
 impl RustylineHelper {
-    pub fn new(engine: Rc<RefCell<Engine>>) -> Self {
+    pub fn new(engine: Arc<Mutex<Engine>>) -> Self {
         Self {
             engine,
-            bracket: std::cell::Cell::new(None),
+            bracket: crossbeam::atomic::AtomicCell::new(None),
         }
     }
 }
@@ -198,7 +195,7 @@ impl Highlighter for RustylineHelper {
                 }
                 TokenType::Identifier(ident) => {
                     // If its a free identifier, nix it?
-                    if self.engine.borrow().global_exists(ident) {
+                    if self.engine.lock().unwrap().global_exists(ident) {
                         // println!("before length: {}", token.source().as_bytes().len());
                         let highlighted = format!("{}", token.source().bright_blue());
                         // println!("After length: {}", highlighted.as_bytes().len());
@@ -299,8 +296,8 @@ impl Highlighter for RustylineHelper {
 
     fn highlight_char(&self, line: &str, mut pos: usize, _: bool) -> bool {
         // will highlight matching brace/bracket/parenthesis if it exists
-        self.bracket.set(check_bracket(line, pos));
-        if self.bracket.get().is_some() {
+        self.bracket.store(check_bracket(line, pos));
+        if self.bracket.load().is_some() {
             return true;
         }
 
@@ -317,7 +314,7 @@ impl Highlighter for RustylineHelper {
                 _ => false,
             }
         } else {
-            self.bracket.get().is_some()
+            self.bracket.load().is_some()
         }
     }
 }