From dd6533e76c3e41f3ffb2fe6782dcb41210debefa Mon Sep 17 00:00:00 2001 From: bakaq Date: Tue, 28 Jan 2025 17:58:40 -0300 Subject: [PATCH 1/9] Add callback streams --- src/arena.rs | 4 ++ src/machine/config.rs | 25 ++++++++++++- src/machine/streams.rs | 84 +++++++++++++++++++++++++++++++++++++++--- src/macros.rs | 1 + 4 files changed, 107 insertions(+), 7 deletions(-) diff --git a/src/arena.rs b/src/arena.rs index e5aceb4d..e3e6cbbd 100644 --- a/src/arena.rs +++ b/src/arena.rs @@ -181,6 +181,7 @@ pub enum ArenaHeaderTag { ReadlineStream = 0b110000, StaticStringStream = 0b110100, ByteStream = 0b111000, + CallbackStream = 0b111001, StandardOutputStream = 0b1100, StandardErrorStream = 0b11000, NullStream = 0b111100, @@ -841,6 +842,9 @@ unsafe fn drop_slab_in_place(value: NonNull, tag: ArenaHeaderTag) { ArenaHeaderTag::ByteStream => { drop_typed_slab_in_place!(ByteStream, value); } + ArenaHeaderTag::CallbackStream => { + drop_typed_slab_in_place!(CallbackStream, value); + } ArenaHeaderTag::LiveLoadState | ArenaHeaderTag::InactiveLoadState => { drop_typed_slab_in_place!(LiveLoadState, value); } diff --git a/src/machine/config.rs b/src/machine/config.rs index 2981899d..34529434 100644 --- a/src/machine/config.rs +++ b/src/machine/config.rs @@ -6,7 +6,8 @@ use crate::Machine; use super::{ bootstrapping_compile, current_dir, import_builtin_impls, libraries, load_module, Atom, - CompilationTarget, IndexStore, ListingSource, MachineArgs, MachineState, Stream, StreamOptions, + Callback, CompilationTarget, IndexStore, ListingSource, MachineArgs, MachineState, Stream, + StreamOptions, }; /// Describes how the streams of a [`Machine`](crate::Machine) will be handled. @@ -31,6 +32,13 @@ impl StreamConfig { inner: StreamConfigInner::Memory, } } + + /// Calls the given callbacks when the respective streams are written to. + pub fn with_callbacks(stdout: Option, stderr: Option) -> Self { + StreamConfig { + inner: StreamConfigInner::Callbacks { stdout, stderr }, + } + } } #[derive(Default)] @@ -38,6 +46,10 @@ enum StreamConfigInner { Stdio, #[default] Memory, + Callbacks { + stdout: Option, + stderr: Option, + }, } /// Describes how a [`Machine`](crate::Machine) will be configured. @@ -90,6 +102,17 @@ impl MachineBuilder { Stream::from_owned_string("".to_owned(), &mut machine_st.arena), Stream::stderr(&mut machine_st.arena), ), + StreamConfigInner::Callbacks { stdout, stderr } => ( + Stream::Null(StreamOptions::default()), + stdout.map_or_else( + || Stream::Null(StreamOptions::default()), + |x| Stream::from_callback(x, &mut machine_st.arena), + ), + stderr.map_or_else( + || Stream::Null(StreamOptions::default()), + |x| Stream::from_callback(x, &mut machine_st.arena), + ), + ), }; let mut wam = Machine { diff --git a/src/machine/streams.rs b/src/machine/streams.rs index b1ecba03..0c180d4a 100644 --- a/src/machine/streams.rs +++ b/src/machine/streams.rs @@ -24,6 +24,7 @@ use std::fs::{File, OpenOptions}; use std::hash::Hash; use std::io; use std::io::{Cursor, ErrorKind, Read, Seek, SeekFrom, Write}; +use std::mem::ManuallyDrop; use std::net::{Shutdown, TcpStream}; use std::ops::{Deref, DerefMut}; use std::path::PathBuf; @@ -375,6 +376,40 @@ impl Write for StandardErrorStream { } } +pub type Callback = Box>)>; + +pub struct CallbackStream { + pub(crate) inner: Cursor>, + callback: Callback, +} + +impl Debug for CallbackStream { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("CallbackStream") + .field("inner", &self.inner) + .finish() + } +} + +impl Write for CallbackStream { + #[inline] + fn write(&mut self, buf: &[u8]) -> std::io::Result { + let pos = self.inner.position(); + + self.inner.seek(SeekFrom::End(0))?; + let result = self.inner.write(buf); + self.inner.seek(SeekFrom::Start(pos))?; + + result + } + + #[inline] + fn flush(&mut self) -> std::io::Result<()> { + (self.callback)(&mut self.inner); + self.inner.flush() + } +} + #[bitfield] #[repr(u64)] #[derive(Debug, Copy, Clone, PartialEq, Eq, Hash)] @@ -500,6 +535,7 @@ arena_allocated_impl_for_stream!(ReadlineStream, ReadlineStream); arena_allocated_impl_for_stream!(StaticStringStream, StaticStringStream); arena_allocated_impl_for_stream!(StandardOutputStream, StandardOutputStream); arena_allocated_impl_for_stream!(StandardErrorStream, StandardErrorStream); +arena_allocated_impl_for_stream!(CharReader, CallbackStream); #[derive(Debug, Copy, Clone)] pub enum Stream { @@ -518,6 +554,7 @@ pub enum Stream { Readline(TypedArenaPtr), StandardOutput(TypedArenaPtr), StandardError(TypedArenaPtr), + Callback(TypedArenaPtr), } impl From> for Stream { @@ -581,6 +618,7 @@ impl Stream { ArenaHeaderTag::Dropped | ArenaHeaderTag::NullStream => { Stream::Null(StreamOptions::default()) } + ArenaHeaderTag::CallbackStream => Stream::Callback(unsafe { ptr.as_typed_ptr() }), _ => unreachable!(), } } @@ -617,6 +655,7 @@ impl Stream { Stream::Readline(ptr) => ptr.header_ptr(), Stream::StandardOutput(ptr) => ptr.header_ptr(), Stream::StandardError(ptr) => ptr.header_ptr(), + Stream::Callback(ptr) => ptr.header_ptr(), } } @@ -637,6 +676,7 @@ impl Stream { Stream::Readline(ref ptr) => &ptr.options, Stream::StandardOutput(ref ptr) => &ptr.options, Stream::StandardError(ref ptr) => &ptr.options, + Stream::Callback(ref ptr) => &ptr.options, } } @@ -657,6 +697,7 @@ impl Stream { Stream::Readline(ref mut ptr) => &mut ptr.options, Stream::StandardOutput(ref mut ptr) => &mut ptr.options, Stream::StandardError(ref mut ptr) => &mut ptr.options, + Stream::Callback(ref mut ptr) => &mut ptr.options, } } @@ -678,6 +719,7 @@ impl Stream { Stream::Readline(ptr) => ptr.lines_read += incr_num_lines_read, Stream::StandardOutput(ptr) => ptr.lines_read += incr_num_lines_read, Stream::StandardError(ptr) => ptr.lines_read += incr_num_lines_read, + Stream::Callback(ptr) => ptr.lines_read += incr_num_lines_read, } } @@ -699,6 +741,7 @@ impl Stream { Stream::Readline(ptr) => ptr.lines_read = value, Stream::StandardOutput(ptr) => ptr.lines_read = value, Stream::StandardError(ptr) => ptr.lines_read = value, + Stream::Callback(ptr) => ptr.lines_read = value, } } @@ -720,6 +763,7 @@ impl Stream { Stream::Readline(ptr) => ptr.lines_read, Stream::StandardOutput(ptr) => ptr.lines_read, Stream::StandardError(ptr) => ptr.lines_read, + Stream::Callback(ptr) => ptr.lines_read, } } } @@ -744,7 +788,8 @@ impl CharRead for Stream { Stream::OutputFile(_) | Stream::StandardError(_) | Stream::StandardOutput(_) - | Stream::Null(_) => Some(Err(std::io::Error::new( + | Stream::Null(_) + | Stream::Callback(_) => Some(Err(std::io::Error::new( ErrorKind::PermissionDenied, StreamError::ReadFromOutputStream, ))), @@ -770,7 +815,8 @@ impl CharRead for Stream { Stream::OutputFile(_) | Stream::StandardError(_) | Stream::StandardOutput(_) - | Stream::Null(_) => Some(Err(std::io::Error::new( + | Stream::Null(_) + | Stream::Callback(_) => Some(Err(std::io::Error::new( ErrorKind::PermissionDenied, StreamError::ReadFromOutputStream, ))), @@ -793,7 +839,8 @@ impl CharRead for Stream { Stream::OutputFile(_) | Stream::StandardError(_) | Stream::StandardOutput(_) - | Stream::Null(_) => {} + | Stream::Null(_) + | Stream::Callback(_) => {} } } @@ -813,7 +860,8 @@ impl CharRead for Stream { Stream::OutputFile(_) | Stream::StandardError(_) | Stream::StandardOutput(_) - | Stream::Null(_) => {} + | Stream::Null(_) + | Stream::Callback(_) => {} } } } @@ -839,7 +887,8 @@ impl Read for Stream { Stream::OutputFile(_) | Stream::StandardError(_) | Stream::StandardOutput(_) - | Stream::Null(_) => Err(std::io::Error::new( + | Stream::Null(_) + | Stream::Callback(_) => Err(std::io::Error::new( ErrorKind::PermissionDenied, StreamError::ReadFromOutputStream, )), @@ -855,6 +904,7 @@ impl Write for Stream { #[cfg(feature = "tls")] Stream::NamedTls(ref mut tls_stream) => tls_stream.get_mut().write(buf), Stream::Byte(ref mut cursor) => cursor.get_mut().write(buf), + Stream::Callback(ref mut callback_stream) => callback_stream.get_mut().write(buf), Stream::StandardOutput(stream) => stream.write(buf), Stream::StandardError(stream) => stream.write(buf), #[cfg(feature = "http")] @@ -881,6 +931,7 @@ impl Write for Stream { #[cfg(feature = "tls")] Stream::NamedTls(ref mut tls_stream) => tls_stream.stream.get_mut().flush(), Stream::Byte(ref mut cursor) => cursor.stream.get_mut().flush(), + Stream::Callback(ref mut callback_stream) => callback_stream.stream.get_mut().flush(), Stream::StandardError(stream) => stream.stream.flush(), Stream::StandardOutput(stream) => stream.stream.flush(), #[cfg(feature = "http")] @@ -1043,6 +1094,7 @@ impl Stream { Stream::Readline(stream) => stream.past_end_of_stream, Stream::StandardOutput(stream) => stream.past_end_of_stream, Stream::StandardError(stream) => stream.past_end_of_stream, + Stream::Callback(stream) => stream.past_end_of_stream, } } @@ -1069,6 +1121,7 @@ impl Stream { Stream::Readline(stream) => stream.past_end_of_stream = value, Stream::StandardOutput(stream) => stream.past_end_of_stream = value, Stream::StandardError(stream) => stream.past_end_of_stream = value, + Stream::Callback(stream) => stream.past_end_of_stream = value, } } @@ -1175,7 +1228,10 @@ impl Stream { Stream::OutputFile(file) if file.is_append => atom!("append"), #[cfg(feature = "http")] Stream::HttpWrite(_) => atom!("write"), - Stream::OutputFile(_) | Stream::StandardError(_) | Stream::StandardOutput(_) => { + Stream::OutputFile(_) + | Stream::StandardError(_) + | Stream::StandardOutput(_) + | Stream::Callback(_) => { atom!("write") } Stream::Null(_) => atom!(""), @@ -1198,6 +1254,17 @@ impl Stream { )) } + #[inline] + pub fn from_callback(callback: Callback, arena: &mut Arena) -> Self { + Stream::Callback(arena_alloc!( + ManuallyDrop::new(StreamLayout::new(CharReader::new(CallbackStream { + inner: Cursor::new(Vec::new()), + callback, + }))), + arena + )) + } + #[inline] pub(crate) fn from_tcp_stream(address: Atom, tcp_stream: TcpStream, arena: &mut Arena) -> Self { tcp_stream.set_read_timeout(None).unwrap(); @@ -1325,6 +1392,10 @@ impl Stream { stream.drop_payload(); Ok(()) } + Stream::Callback(mut stream) => { + stream.drop_payload(); + Ok(()) + } Stream::StaticString(mut stream) => { stream.drop_payload(); Ok(()) @@ -1370,6 +1441,7 @@ impl Stream { | Stream::StandardOutput(_) | Stream::NamedTcp(..) | Stream::Byte(_) + | Stream::Callback(_) | Stream::OutputFile(..) => true, _ => false, } diff --git a/src/macros.rs b/src/macros.rs index 30a863ca..9b2cbabe 100644 --- a/src/macros.rs +++ b/src/macros.rs @@ -305,6 +305,7 @@ macro_rules! match_untyped_arena_ptr_pat { | ArenaHeaderTag::ReadlineStream | ArenaHeaderTag::StaticStringStream | ArenaHeaderTag::ByteStream + | ArenaHeaderTag::CallbackStream | ArenaHeaderTag::StandardOutputStream | ArenaHeaderTag::StandardErrorStream }; From 7a6620b52df02f33690ecf3033b7920ae855c819 Mon Sep 17 00:00:00 2001 From: bakaq Date: Wed, 29 Jan 2025 11:35:04 -0300 Subject: [PATCH 2/9] Add input stream channel --- src/arena.rs | 4 +++ src/machine/config.rs | 57 ++++++++++++++++++++++++++++++++++++------ src/machine/streams.rs | 48 +++++++++++++++++++++++++++++++++++ src/macros.rs | 1 + 4 files changed, 103 insertions(+), 7 deletions(-) diff --git a/src/arena.rs b/src/arena.rs index e3e6cbbd..113ff6f0 100644 --- a/src/arena.rs +++ b/src/arena.rs @@ -182,6 +182,7 @@ pub enum ArenaHeaderTag { StaticStringStream = 0b110100, ByteStream = 0b111000, CallbackStream = 0b111001, + InputChannelStream = 0b111010, StandardOutputStream = 0b1100, StandardErrorStream = 0b11000, NullStream = 0b111100, @@ -845,6 +846,9 @@ unsafe fn drop_slab_in_place(value: NonNull, tag: ArenaHeaderTag) { ArenaHeaderTag::CallbackStream => { drop_typed_slab_in_place!(CallbackStream, value); } + ArenaHeaderTag::InputChannelStream => { + drop_typed_slab_in_place!(InputChannelStream, value); + } ArenaHeaderTag::LiveLoadState | ArenaHeaderTag::InactiveLoadState => { drop_typed_slab_in_place!(LiveLoadState, value); } diff --git a/src/machine/config.rs b/src/machine/config.rs index 34529434..06dbb77a 100644 --- a/src/machine/config.rs +++ b/src/machine/config.rs @@ -1,4 +1,7 @@ -use std::borrow::Cow; +use std::cell::RefCell; +use std::io::{Seek, SeekFrom, Write}; +use std::rc::Rc; +use std::{borrow::Cow, io::Cursor}; use rand::{rngs::StdRng, SeedableRng}; @@ -34,10 +37,45 @@ impl StreamConfig { } /// Calls the given callbacks when the respective streams are written to. - pub fn with_callbacks(stdout: Option, stderr: Option) -> Self { - StreamConfig { - inner: StreamConfigInner::Callbacks { stdout, stderr }, - } + /// + /// This also returns a handler to the stdin do the [`Machine`](crate::Machine). + pub fn with_callbacks(stdout: Option, stderr: Option) -> (UserInput, Self) { + let stdin = Rc::new(RefCell::new(Cursor::new(Vec::new()))); + ( + UserInput { + inner: stdin.clone(), + }, + StreamConfig { + inner: StreamConfigInner::Callbacks { + stdin, + stdout, + stderr, + }, + }, + ) + } +} + +/// A handler for the stdin of the [`Machine`](crate::Machine). +#[derive(Debug)] +pub struct UserInput { + inner: Rc>>>, +} + +impl Write for UserInput { + fn write(&mut self, buf: &[u8]) -> std::io::Result { + let mut inner = self.inner.borrow_mut(); + let pos = inner.position(); + + inner.seek(SeekFrom::End(0))?; + let result = inner.write(buf); + inner.seek(SeekFrom::Start(pos))?; + + result + } + + fn flush(&mut self) -> std::io::Result<()> { + self.inner.borrow_mut().flush() } } @@ -47,6 +85,7 @@ enum StreamConfigInner { #[default] Memory, Callbacks { + stdin: Rc>>>, stdout: Option, stderr: Option, }, @@ -102,8 +141,12 @@ impl MachineBuilder { Stream::from_owned_string("".to_owned(), &mut machine_st.arena), Stream::stderr(&mut machine_st.arena), ), - StreamConfigInner::Callbacks { stdout, stderr } => ( - Stream::Null(StreamOptions::default()), + StreamConfigInner::Callbacks { + stdin, + stdout, + stderr, + } => ( + Stream::input_channel(stdin, &mut machine_st.arena), stdout.map_or_else( || Stream::Null(StreamOptions::default()), |x| Stream::from_callback(x, &mut machine_st.arena), diff --git a/src/machine/streams.rs b/src/machine/streams.rs index 0c180d4a..ce0c8d05 100644 --- a/src/machine/streams.rs +++ b/src/machine/streams.rs @@ -16,6 +16,7 @@ pub use scryer_modular_bitfield::prelude::*; #[cfg(feature = "http")] use bytes::{buf::Reader as BufReader, Buf, Bytes}; +use std::cell::RefCell; use std::cmp::Ordering; use std::error::Error; use std::fmt; @@ -29,6 +30,7 @@ use std::net::{Shutdown, TcpStream}; use std::ops::{Deref, DerefMut}; use std::path::PathBuf; use std::ptr; +use std::rc::Rc; #[cfg(feature = "tls")] use native_tls::TlsStream; @@ -410,6 +412,18 @@ impl Write for CallbackStream { } } +#[derive(Debug)] +pub struct InputChannelStream { + pub(crate) inner: Rc>>>, +} + +impl Read for InputChannelStream { + #[inline] + fn read(&mut self, buf: &mut [u8]) -> std::io::Result { + self.inner.borrow_mut().read(buf) + } +} + #[bitfield] #[repr(u64)] #[derive(Debug, Copy, Clone, PartialEq, Eq, Hash)] @@ -536,6 +550,7 @@ arena_allocated_impl_for_stream!(StaticStringStream, StaticStringStream); arena_allocated_impl_for_stream!(StandardOutputStream, StandardOutputStream); arena_allocated_impl_for_stream!(StandardErrorStream, StandardErrorStream); arena_allocated_impl_for_stream!(CharReader, CallbackStream); +arena_allocated_impl_for_stream!(CharReader, InputChannelStream); #[derive(Debug, Copy, Clone)] pub enum Stream { @@ -555,6 +570,7 @@ pub enum Stream { StandardOutput(TypedArenaPtr), StandardError(TypedArenaPtr), Callback(TypedArenaPtr), + InputChannel(TypedArenaPtr), } impl From> for Stream { @@ -585,6 +601,14 @@ impl Stream { )) } + #[inline] + pub fn input_channel(cursor: Rc>>>, arena: &mut Arena) -> Stream { + Stream::InputChannel(arena_alloc!( + StreamLayout::new(CharReader::new(InputChannelStream { inner: cursor })), + arena + )) + } + #[inline] pub fn stdin(arena: &mut Arena, add_history: bool) -> Stream { Stream::Readline(arena_alloc!( @@ -619,6 +643,9 @@ impl Stream { Stream::Null(StreamOptions::default()) } ArenaHeaderTag::CallbackStream => Stream::Callback(unsafe { ptr.as_typed_ptr() }), + ArenaHeaderTag::InputChannelStream => { + Stream::InputChannel(unsafe { ptr.as_typed_ptr() }) + } _ => unreachable!(), } } @@ -656,6 +683,7 @@ impl Stream { Stream::StandardOutput(ptr) => ptr.header_ptr(), Stream::StandardError(ptr) => ptr.header_ptr(), Stream::Callback(ptr) => ptr.header_ptr(), + Stream::InputChannel(ptr) => ptr.header_ptr(), } } @@ -677,6 +705,7 @@ impl Stream { Stream::StandardOutput(ref ptr) => &ptr.options, Stream::StandardError(ref ptr) => &ptr.options, Stream::Callback(ref ptr) => &ptr.options, + Stream::InputChannel(ref ptr) => &ptr.options, } } @@ -698,6 +727,7 @@ impl Stream { Stream::StandardOutput(ref mut ptr) => &mut ptr.options, Stream::StandardError(ref mut ptr) => &mut ptr.options, Stream::Callback(ref mut ptr) => &mut ptr.options, + Stream::InputChannel(ref mut ptr) => &mut ptr.options, } } @@ -720,6 +750,7 @@ impl Stream { Stream::StandardOutput(ptr) => ptr.lines_read += incr_num_lines_read, Stream::StandardError(ptr) => ptr.lines_read += incr_num_lines_read, Stream::Callback(ptr) => ptr.lines_read += incr_num_lines_read, + Stream::InputChannel(ptr) => ptr.lines_read += incr_num_lines_read, } } @@ -742,6 +773,7 @@ impl Stream { Stream::StandardOutput(ptr) => ptr.lines_read = value, Stream::StandardError(ptr) => ptr.lines_read = value, Stream::Callback(ptr) => ptr.lines_read = value, + Stream::InputChannel(ptr) => ptr.lines_read = value, } } @@ -764,6 +796,7 @@ impl Stream { Stream::StandardOutput(ptr) => ptr.lines_read, Stream::StandardError(ptr) => ptr.lines_read, Stream::Callback(ptr) => ptr.lines_read, + Stream::InputChannel(ptr) => ptr.lines_read, } } } @@ -780,6 +813,7 @@ impl CharRead for Stream { Stream::Readline(rl_stream) => (*rl_stream).peek_char(), Stream::StaticString(src) => (*src).peek_char(), Stream::Byte(cursor) => (*cursor).peek_char(), + Stream::InputChannel(cursor) => (*cursor).peek_char(), #[cfg(feature = "http")] Stream::HttpWrite(_) => Some(Err(std::io::Error::new( ErrorKind::PermissionDenied, @@ -807,6 +841,7 @@ impl CharRead for Stream { Stream::Readline(rl_stream) => (*rl_stream).read_char(), Stream::StaticString(src) => (*src).read_char(), Stream::Byte(cursor) => (*cursor).read_char(), + Stream::InputChannel(cursor) => (*cursor).read_char(), #[cfg(feature = "http")] Stream::HttpWrite(_) => Some(Err(std::io::Error::new( ErrorKind::PermissionDenied, @@ -841,6 +876,7 @@ impl CharRead for Stream { | Stream::StandardOutput(_) | Stream::Null(_) | Stream::Callback(_) => {} + Stream::InputChannel(_) => {} } } @@ -855,6 +891,7 @@ impl CharRead for Stream { Stream::Readline(ref mut rl_stream) => rl_stream.consume(nread), Stream::StaticString(ref mut src) => src.consume(nread), Stream::Byte(ref mut cursor) => cursor.consume(nread), + Stream::InputChannel(ref mut cursor) => cursor.consume(nread), #[cfg(feature = "http")] Stream::HttpWrite(_) => {} Stream::OutputFile(_) @@ -879,6 +916,7 @@ impl Read for Stream { Stream::Readline(rl_stream) => (*rl_stream).read(buf), Stream::StaticString(src) => (*src).read(buf), Stream::Byte(cursor) => (*cursor).read(buf), + Stream::InputChannel(cursor) => (*cursor).read(buf), #[cfg(feature = "http")] Stream::HttpWrite(_) => Err(std::io::Error::new( ErrorKind::PermissionDenied, @@ -915,6 +953,7 @@ impl Write for Stream { StreamError::WriteToInputStream, )), Stream::StaticString(_) + | Stream::InputChannel(_) | Stream::Readline(_) | Stream::InputFile(..) | Stream::Null(_) => Err(std::io::Error::new( @@ -942,6 +981,7 @@ impl Write for Stream { StreamError::FlushToInputStream, )), Stream::StaticString(_) + | Stream::InputChannel(_) | Stream::Readline(_) | Stream::InputFile(_) | Stream::Null(_) => Err(std::io::Error::new( @@ -1095,6 +1135,7 @@ impl Stream { Stream::StandardOutput(stream) => stream.past_end_of_stream, Stream::StandardError(stream) => stream.past_end_of_stream, Stream::Callback(stream) => stream.past_end_of_stream, + Stream::InputChannel(stream) => stream.past_end_of_stream, } } @@ -1122,6 +1163,7 @@ impl Stream { Stream::StandardOutput(stream) => stream.past_end_of_stream = value, Stream::StandardError(stream) => stream.past_end_of_stream = value, Stream::Callback(stream) => stream.past_end_of_stream = value, + Stream::InputChannel(stream) => stream.past_end_of_stream = value, } } @@ -1221,6 +1263,7 @@ impl Stream { #[cfg(feature = "tls")] Stream::NamedTls(..) => atom!("read_append"), Stream::Byte(_) + | Stream::InputChannel(_) | Stream::Readline(_) | Stream::StaticString(_) | Stream::InputFile(..) => atom!("read"), @@ -1396,6 +1439,10 @@ impl Stream { stream.drop_payload(); Ok(()) } + Stream::InputChannel(mut stream) => { + stream.drop_payload(); + Ok(()) + } Stream::StaticString(mut stream) => { stream.drop_payload(); Ok(()) @@ -1423,6 +1470,7 @@ impl Stream { Stream::HttpRead(..) => true, Stream::NamedTcp(..) | Stream::Byte(_) + | Stream::InputChannel(_) | Stream::Readline(_) | Stream::StaticString(_) | Stream::InputFile(..) => true, diff --git a/src/macros.rs b/src/macros.rs index 9b2cbabe..547ccc00 100644 --- a/src/macros.rs +++ b/src/macros.rs @@ -306,6 +306,7 @@ macro_rules! match_untyped_arena_ptr_pat { | ArenaHeaderTag::StaticStringStream | ArenaHeaderTag::ByteStream | ArenaHeaderTag::CallbackStream + | ArenaHeaderTag::InputChannelStream | ArenaHeaderTag::StandardOutputStream | ArenaHeaderTag::StandardErrorStream }; From 4e032c8a285ee7a15bf2d9833bc332d64916674f Mon Sep 17 00:00:00 2001 From: bakaq Date: Wed, 29 Jan 2025 17:09:48 -0300 Subject: [PATCH 3/9] Test for callback streams --- src/machine/lib_machine/tests.rs | 36 +++++++++++++++++++++++++++++++- 1 file changed, 35 insertions(+), 1 deletion(-) diff --git a/src/machine/lib_machine/tests.rs b/src/machine/lib_machine/tests.rs index 5b89d67d..40c33225 100644 --- a/src/machine/lib_machine/tests.rs +++ b/src/machine/lib_machine/tests.rs @@ -1,5 +1,8 @@ +use std::io::Write; +use std::{cell::RefCell, io::Read, rc::Rc}; + use super::*; -use crate::MachineBuilder; +use crate::{MachineBuilder, StreamConfig}; #[test] #[cfg_attr(miri, ignore = "it takes too long to run")] @@ -608,3 +611,34 @@ fn errors_and_exceptions() { [Ok(LeafAnswer::Exception(Term::atom("a")))] ); } + +#[test] +#[cfg_attr(miri, ignore)] +fn callback_streams() { + let test_string = Rc::new(RefCell::new(String::new())); + let test_string2 = test_string.clone(); + + let (mut user_input, streams) = StreamConfig::with_callbacks( + Some(Box::new(move |x| { + x.read_to_string(&mut *test_string2.borrow_mut()).unwrap(); + })), + None, + ); + let mut machine = MachineBuilder::default().with_streams(streams).build(); + + write!(&mut user_input, "a(1,2,3).").unwrap(); + + let complete_answer: Vec<_> = machine + .run_query("read(A), write('asdf'), nl, flush_output.") + .collect(); + + assert_eq!( + complete_answer, + [Ok(LeafAnswer::from_bindings([( + "A", + Term::compound("a", [Term::integer(1), Term::integer(2), Term::integer(3)]) + ),]))] + ); + + assert_eq!(*test_string.borrow(), "asdf\n"); +} From baae1dca15080735f7b0a77093b86390f14f4e38 Mon Sep 17 00:00:00 2001 From: bakaq Date: Thu, 30 Jan 2025 06:14:38 -0300 Subject: [PATCH 4/9] Refactor UserInput to use channels --- src/machine/config.rs | 33 +++++++--------- src/machine/lib_machine/tests.rs | 2 +- src/machine/streams.rs | 65 +++++++++++++++++++++++++++++--- 3 files changed, 73 insertions(+), 27 deletions(-) diff --git a/src/machine/config.rs b/src/machine/config.rs index 06dbb77a..12e5a538 100644 --- a/src/machine/config.rs +++ b/src/machine/config.rs @@ -1,7 +1,6 @@ -use std::cell::RefCell; -use std::io::{Seek, SeekFrom, Write}; -use std::rc::Rc; -use std::{borrow::Cow, io::Cursor}; +use std::borrow::Cow; +use std::io::Write; +use std::sync::mpsc::{channel, Receiver, Sender}; use rand::{rngs::StdRng, SeedableRng}; @@ -40,14 +39,12 @@ impl StreamConfig { /// /// This also returns a handler to the stdin do the [`Machine`](crate::Machine). pub fn with_callbacks(stdout: Option, stderr: Option) -> (UserInput, Self) { - let stdin = Rc::new(RefCell::new(Cursor::new(Vec::new()))); + let (sender, receiver) = channel(); ( - UserInput { - inner: stdin.clone(), - }, + UserInput { inner: sender }, StreamConfig { inner: StreamConfigInner::Callbacks { - stdin, + stdin: receiver, stdout, stderr, }, @@ -59,23 +56,19 @@ impl StreamConfig { /// A handler for the stdin of the [`Machine`](crate::Machine). #[derive(Debug)] pub struct UserInput { - inner: Rc>>>, + inner: Sender>, } impl Write for UserInput { fn write(&mut self, buf: &[u8]) -> std::io::Result { - let mut inner = self.inner.borrow_mut(); - let pos = inner.position(); - - inner.seek(SeekFrom::End(0))?; - let result = inner.write(buf); - inner.seek(SeekFrom::Start(pos))?; - - result + self.inner + .send(buf.into()) + .map(|_| buf.len()) + .map_err(|_| std::io::ErrorKind::BrokenPipe.into()) } fn flush(&mut self) -> std::io::Result<()> { - self.inner.borrow_mut().flush() + Ok(()) } } @@ -85,7 +78,7 @@ enum StreamConfigInner { #[default] Memory, Callbacks { - stdin: Rc>>>, + stdin: Receiver>, stdout: Option, stderr: Option, }, diff --git a/src/machine/lib_machine/tests.rs b/src/machine/lib_machine/tests.rs index 40c33225..502f4d89 100644 --- a/src/machine/lib_machine/tests.rs +++ b/src/machine/lib_machine/tests.rs @@ -620,7 +620,7 @@ fn callback_streams() { let (mut user_input, streams) = StreamConfig::with_callbacks( Some(Box::new(move |x| { - x.read_to_string(&mut *test_string2.borrow_mut()).unwrap(); + x.read_to_string(&mut test_string2.borrow_mut()).unwrap(); })), None, ); diff --git a/src/machine/streams.rs b/src/machine/streams.rs index ce0c8d05..edb6756b 100644 --- a/src/machine/streams.rs +++ b/src/machine/streams.rs @@ -16,7 +16,6 @@ pub use scryer_modular_bitfield::prelude::*; #[cfg(feature = "http")] use bytes::{buf::Reader as BufReader, Buf, Bytes}; -use std::cell::RefCell; use std::cmp::Ordering; use std::error::Error; use std::fmt; @@ -30,7 +29,8 @@ use std::net::{Shutdown, TcpStream}; use std::ops::{Deref, DerefMut}; use std::path::PathBuf; use std::ptr; -use std::rc::Rc; +use std::sync::mpsc::Receiver; +use std::sync::mpsc::TryRecvError; #[cfg(feature = "tls")] use native_tls::TlsStream; @@ -414,13 +414,50 @@ impl Write for CallbackStream { #[derive(Debug)] pub struct InputChannelStream { - pub(crate) inner: Rc>>>, + pub(crate) inner: Cursor>, + pub eof: bool, + channel: Receiver>, } impl Read for InputChannelStream { #[inline] fn read(&mut self, buf: &mut [u8]) -> std::io::Result { - self.inner.borrow_mut().read(buf) + if self.eof { + return Ok(0); + } + + let to_read = buf.len(); + let mut total_read = 0; + + loop { + total_read += self.inner.read(&mut buf[total_read..])?; + + if total_read < to_read { + // We need to get more data to read + match self.channel.try_recv() { + Ok(data) => { + // Append into self.inner + let pos = self.inner.position(); + assert_eq!(pos as usize, self.inner.get_ref().len()); + self.inner.write_all(&data)?; + self.inner.seek(SeekFrom::Start(pos))?; + } + Err(TryRecvError::Empty) => { + // Data is pending + break; + } + Err(TryRecvError::Disconnected) => { + // The other end of the channel was closed so we are EOF + self.eof = true; + break; + } + } + } else { + assert_eq!(total_read, to_read); + break; + } + } + Ok(total_read) } } @@ -602,9 +639,14 @@ impl Stream { } #[inline] - pub fn input_channel(cursor: Rc>>>, arena: &mut Arena) -> Stream { + pub fn input_channel(channel: Receiver>, arena: &mut Arena) -> Stream { + let inner = Cursor::new(Vec::new()); Stream::InputChannel(arena_alloc!( - StreamLayout::new(CharReader::new(InputChannelStream { inner: cursor })), + StreamLayout::new(CharReader::new(InputChannelStream { + inner, + eof: false, + channel + })), arena )) } @@ -1239,6 +1281,13 @@ impl Stream { AtEndOfStream::Past } } + Stream::InputChannel(stream_layout) => { + if stream_layout.stream.get_ref().eof { + AtEndOfStream::At + } else { + AtEndOfStream::Not + } + } _ => AtEndOfStream::Not, } } @@ -1519,6 +1568,10 @@ impl Stream { readline_stream.reset(); true } + Stream::InputChannel(ref mut input_channel_stream) => { + input_channel_stream.stream.get_mut().inner.set_position(0); + true + } _ => false, } } From 0a2457943e428f5f3d3f5f34145fdf4630c4520f Mon Sep 17 00:00:00 2001 From: bakaq Date: Thu, 30 Jan 2025 09:20:28 -0300 Subject: [PATCH 5/9] Configure streams separately --- src/machine/config.rs | 135 +++++++++++++++++++++++++++--------------- 1 file changed, 86 insertions(+), 49 deletions(-) diff --git a/src/machine/config.rs b/src/machine/config.rs index 12e5a538..44d57945 100644 --- a/src/machine/config.rs +++ b/src/machine/config.rs @@ -7,22 +7,83 @@ use rand::{rngs::StdRng, SeedableRng}; use crate::Machine; use super::{ - bootstrapping_compile, current_dir, import_builtin_impls, libraries, load_module, Atom, + bootstrapping_compile, current_dir, import_builtin_impls, libraries, load_module, Arena, Atom, Callback, CompilationTarget, IndexStore, ListingSource, MachineArgs, MachineState, Stream, StreamOptions, }; -/// Describes how the streams of a [`Machine`](crate::Machine) will be handled. #[derive(Default)] +enum OutputStreamConfig { + #[default] + Null, + Memory, + Stdout, + Stderr, + Callback(Callback), +} + +impl std::fmt::Debug for OutputStreamConfig { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::Null => write!(f, "Null"), + Self::Memory => write!(f, "Memory"), + Self::Stdout => write!(f, "Stdout"), + Self::Stderr => write!(f, "Stderr"), + Self::Callback(_) => f.debug_tuple("Callback").field(&"").finish(), + } + } +} + +impl OutputStreamConfig { + fn into_stream(self, arena: &mut Arena) -> Stream { + match self { + OutputStreamConfig::Null => Stream::Null(StreamOptions::default()), + OutputStreamConfig::Memory => Stream::from_owned_string("".to_owned(), arena), + OutputStreamConfig::Stdout => Stream::stdout(arena), + OutputStreamConfig::Stderr => Stream::stderr(arena), + OutputStreamConfig::Callback(callback) => Stream::from_callback(callback, arena), + } + } +} + +#[derive(Debug, Default)] +enum InputStreamConfig { + #[default] + Null, + Stdin, + Channel(Receiver>), +} + +impl InputStreamConfig { + fn into_stream(self, arena: &mut Arena, add_history: bool) -> Stream { + match self { + InputStreamConfig::Null => Stream::Null(StreamOptions::default()), + InputStreamConfig::Stdin => Stream::stdin(arena, add_history), + InputStreamConfig::Channel(channel) => Stream::input_channel(channel, arena), + } + } +} + +/// Describes how the streams of a [`Machine`](crate::Machine) will be handled. pub struct StreamConfig { - inner: StreamConfigInner, + stdin: InputStreamConfig, + stdout: OutputStreamConfig, + stderr: OutputStreamConfig, +} + +impl Default for StreamConfig { + fn default() -> Self { + Self::in_memory() + } } impl StreamConfig { /// Binds the input, output and error streams to stdin, stdout and stderr. pub fn stdio() -> Self { StreamConfig { - inner: StreamConfigInner::Stdio, + stdin: InputStreamConfig::Stdin, + stdout: OutputStreamConfig::Stdout, + stderr: OutputStreamConfig::Stderr, } } @@ -31,7 +92,9 @@ impl StreamConfig { /// The input stream is ignored. pub fn in_memory() -> Self { StreamConfig { - inner: StreamConfigInner::Memory, + stdin: InputStreamConfig::Null, + stdout: OutputStreamConfig::Memory, + stderr: OutputStreamConfig::Stderr, } } @@ -43,14 +106,24 @@ impl StreamConfig { ( UserInput { inner: sender }, StreamConfig { - inner: StreamConfigInner::Callbacks { - stdin: receiver, - stdout, - stderr, - }, + stdin: InputStreamConfig::Channel(receiver), + stdout: stdout.map_or(OutputStreamConfig::Null, |x| { + OutputStreamConfig::Callback(x) + }), + stderr: stderr.map_or(OutputStreamConfig::Null, |x| { + OutputStreamConfig::Callback(x) + }), }, ) } + + fn into_streams(self, arena: &mut Arena, add_history: bool) -> (Stream, Stream, Stream) { + ( + self.stdin.into_stream(arena, add_history), + self.stdout.into_stream(arena), + self.stderr.into_stream(arena), + ) + } } /// A handler for the stdin of the [`Machine`](crate::Machine). @@ -72,18 +145,6 @@ impl Write for UserInput { } } -#[derive(Default)] -enum StreamConfigInner { - Stdio, - #[default] - Memory, - Callbacks { - stdin: Receiver>, - stdout: Option, - stderr: Option, - }, -} - /// Describes how a [`Machine`](crate::Machine) will be configured. pub struct MachineBuilder { pub(crate) streams: StreamConfig, @@ -123,33 +184,9 @@ impl MachineBuilder { let args = MachineArgs::new(); let mut machine_st = MachineState::new(); - let (user_input, user_output, user_error) = match self.streams.inner { - StreamConfigInner::Stdio => ( - Stream::stdin(&mut machine_st.arena, args.add_history), - Stream::stdout(&mut machine_st.arena), - Stream::stderr(&mut machine_st.arena), - ), - StreamConfigInner::Memory => ( - Stream::Null(StreamOptions::default()), - Stream::from_owned_string("".to_owned(), &mut machine_st.arena), - Stream::stderr(&mut machine_st.arena), - ), - StreamConfigInner::Callbacks { - stdin, - stdout, - stderr, - } => ( - Stream::input_channel(stdin, &mut machine_st.arena), - stdout.map_or_else( - || Stream::Null(StreamOptions::default()), - |x| Stream::from_callback(x, &mut machine_st.arena), - ), - stderr.map_or_else( - || Stream::Null(StreamOptions::default()), - |x| Stream::from_callback(x, &mut machine_st.arena), - ), - ), - }; + let (user_input, user_output, user_error) = self + .streams + .into_streams(&mut machine_st.arena, args.add_history); let mut wam = Machine { machine_st, From 5386c183d2946ed24862076f3f2f7811f8e4d5a1 Mon Sep 17 00:00:00 2001 From: bakaq Date: Fri, 31 Jan 2025 07:23:39 -0300 Subject: [PATCH 6/9] Make input and output stream configuration public --- src/machine/config.rs | 127 +++++++++++++++++++++++++++++++----------- 1 file changed, 96 insertions(+), 31 deletions(-) diff --git a/src/machine/config.rs b/src/machine/config.rs index 44d57945..395dbb2f 100644 --- a/src/machine/config.rs +++ b/src/machine/config.rs @@ -13,7 +13,7 @@ use super::{ }; #[derive(Default)] -enum OutputStreamConfig { +enum OutputStreamConfigInner { #[default] Null, Memory, @@ -22,7 +22,7 @@ enum OutputStreamConfig { Callback(Callback), } -impl std::fmt::Debug for OutputStreamConfig { +impl std::fmt::Debug for OutputStreamConfigInner { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { Self::Null => write!(f, "Null"), @@ -34,41 +34,110 @@ impl std::fmt::Debug for OutputStreamConfig { } } +/// Configuration for an output stream. +#[derive(Debug, Default)] +pub struct OutputStreamConfig { + inner: OutputStreamConfigInner, +} + impl OutputStreamConfig { + /// Ignores all output. + pub fn null() -> Self { + Self { + inner: OutputStreamConfigInner::Null, + } + } + /// Sends output to stdout. + pub fn stdout() -> Self { + Self { + inner: OutputStreamConfigInner::Stdout, + } + } + /// Sends output to stderr. + pub fn stderr() -> Self { + Self { + inner: OutputStreamConfigInner::Stderr, + } + } + /// Keeps output in a memory buffer. + pub fn memory() -> Self { + Self { + inner: OutputStreamConfigInner::Memory, + } + } + /// Calls a callback with the output whenever the stream is written to. + pub fn callback(callback: Callback) -> Self { + Self { + inner: OutputStreamConfigInner::Callback(callback), + } + } + fn into_stream(self, arena: &mut Arena) -> Stream { - match self { - OutputStreamConfig::Null => Stream::Null(StreamOptions::default()), - OutputStreamConfig::Memory => Stream::from_owned_string("".to_owned(), arena), - OutputStreamConfig::Stdout => Stream::stdout(arena), - OutputStreamConfig::Stderr => Stream::stderr(arena), - OutputStreamConfig::Callback(callback) => Stream::from_callback(callback, arena), + match self.inner { + OutputStreamConfigInner::Null => Stream::Null(StreamOptions::default()), + OutputStreamConfigInner::Memory => Stream::from_owned_string("".to_owned(), arena), + OutputStreamConfigInner::Stdout => Stream::stdout(arena), + OutputStreamConfigInner::Stderr => Stream::stderr(arena), + OutputStreamConfigInner::Callback(callback) => Stream::from_callback(callback, arena), } } } #[derive(Debug, Default)] -enum InputStreamConfig { +enum InputStreamConfigInner { #[default] Null, Stdin, Channel(Receiver>), } +/// Configuration for an input stream; +#[derive(Debug, Default)] +pub struct InputStreamConfig { + inner: InputStreamConfigInner, +} + impl InputStreamConfig { + /// Ignores all input. + pub fn null() -> Self { + Self { + inner: InputStreamConfigInner::Null, + } + } + /// Gets input from stdin. + pub fn stdin() -> Self { + Self { + inner: InputStreamConfigInner::Stdin, + } + } + /// Connects the input to the receiving end of a channel. + pub fn channel() -> (UserInput, Self) { + let (sender, receiver) = channel(); + ( + UserInput { inner: sender }, + Self { + inner: InputStreamConfigInner::Channel(receiver), + }, + ) + } + fn into_stream(self, arena: &mut Arena, add_history: bool) -> Stream { - match self { - InputStreamConfig::Null => Stream::Null(StreamOptions::default()), - InputStreamConfig::Stdin => Stream::stdin(arena, add_history), - InputStreamConfig::Channel(channel) => Stream::input_channel(channel, arena), + match self.inner { + InputStreamConfigInner::Null => Stream::Null(StreamOptions::default()), + InputStreamConfigInner::Stdin => Stream::stdin(arena, add_history), + InputStreamConfigInner::Channel(channel) => Stream::input_channel(channel, arena), } } } /// Describes how the streams of a [`Machine`](crate::Machine) will be handled. pub struct StreamConfig { - stdin: InputStreamConfig, - stdout: OutputStreamConfig, - stderr: OutputStreamConfig, + /// The configuration for the stdin of the [`Machine`](crate::Machine). + pub stdin: InputStreamConfig, + /// The configuration for the stdout of the [`Machine`](crate::Machine). + pub stdout: OutputStreamConfig, + /// The configuration for the stderr of the [`Machine`](crate::Machine). + pub stderr: OutputStreamConfig, } impl Default for StreamConfig { @@ -81,9 +150,9 @@ impl StreamConfig { /// Binds the input, output and error streams to stdin, stdout and stderr. pub fn stdio() -> Self { StreamConfig { - stdin: InputStreamConfig::Stdin, - stdout: OutputStreamConfig::Stdout, - stderr: OutputStreamConfig::Stderr, + stdin: InputStreamConfig::stdin(), + stdout: OutputStreamConfig::stdout(), + stderr: OutputStreamConfig::stderr(), } } @@ -92,9 +161,9 @@ impl StreamConfig { /// The input stream is ignored. pub fn in_memory() -> Self { StreamConfig { - stdin: InputStreamConfig::Null, - stdout: OutputStreamConfig::Memory, - stderr: OutputStreamConfig::Stderr, + stdin: InputStreamConfig::null(), + stdout: OutputStreamConfig::memory(), + stderr: OutputStreamConfig::stderr(), } } @@ -102,17 +171,13 @@ impl StreamConfig { /// /// This also returns a handler to the stdin do the [`Machine`](crate::Machine). pub fn with_callbacks(stdout: Option, stderr: Option) -> (UserInput, Self) { - let (sender, receiver) = channel(); + let (user_input, channel_stream) = InputStreamConfig::channel(); ( - UserInput { inner: sender }, + user_input, StreamConfig { - stdin: InputStreamConfig::Channel(receiver), - stdout: stdout.map_or(OutputStreamConfig::Null, |x| { - OutputStreamConfig::Callback(x) - }), - stderr: stderr.map_or(OutputStreamConfig::Null, |x| { - OutputStreamConfig::Callback(x) - }), + stdin: channel_stream, + stdout: stdout.map_or_else(OutputStreamConfig::null, OutputStreamConfig::callback), + stderr: stderr.map_or_else(OutputStreamConfig::null, OutputStreamConfig::callback), }, ) } From ad211d5e0557efeaf1a9784213377b43fcc017a7 Mon Sep 17 00:00:00 2001 From: bakaq Date: Fri, 31 Jan 2025 07:45:13 -0300 Subject: [PATCH 7/9] Disallow null streams in output --- src/machine/config.rs | 20 +++++++++----------- 1 file changed, 9 insertions(+), 11 deletions(-) diff --git a/src/machine/config.rs b/src/machine/config.rs index 395dbb2f..2e71ae26 100644 --- a/src/machine/config.rs +++ b/src/machine/config.rs @@ -15,7 +15,6 @@ use super::{ #[derive(Default)] enum OutputStreamConfigInner { #[default] - Null, Memory, Stdout, Stderr, @@ -25,7 +24,6 @@ enum OutputStreamConfigInner { impl std::fmt::Debug for OutputStreamConfigInner { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { - Self::Null => write!(f, "Null"), Self::Memory => write!(f, "Memory"), Self::Stdout => write!(f, "Stdout"), Self::Stderr => write!(f, "Stderr"), @@ -41,30 +39,27 @@ pub struct OutputStreamConfig { } impl OutputStreamConfig { - /// Ignores all output. - pub fn null() -> Self { - Self { - inner: OutputStreamConfigInner::Null, - } - } /// Sends output to stdout. pub fn stdout() -> Self { Self { inner: OutputStreamConfigInner::Stdout, } } + /// Sends output to stderr. pub fn stderr() -> Self { Self { inner: OutputStreamConfigInner::Stderr, } } + /// Keeps output in a memory buffer. pub fn memory() -> Self { Self { inner: OutputStreamConfigInner::Memory, } } + /// Calls a callback with the output whenever the stream is written to. pub fn callback(callback: Callback) -> Self { Self { @@ -74,7 +69,6 @@ impl OutputStreamConfig { fn into_stream(self, arena: &mut Arena) -> Stream { match self.inner { - OutputStreamConfigInner::Null => Stream::Null(StreamOptions::default()), OutputStreamConfigInner::Memory => Stream::from_owned_string("".to_owned(), arena), OutputStreamConfigInner::Stdout => Stream::stdout(arena), OutputStreamConfigInner::Stderr => Stream::stderr(arena), @@ -104,12 +98,14 @@ impl InputStreamConfig { inner: InputStreamConfigInner::Null, } } + /// Gets input from stdin. pub fn stdin() -> Self { Self { inner: InputStreamConfigInner::Stdin, } } + /// Connects the input to the receiving end of a channel. pub fn channel() -> (UserInput, Self) { let (sender, receiver) = channel(); @@ -176,8 +172,10 @@ impl StreamConfig { user_input, StreamConfig { stdin: channel_stream, - stdout: stdout.map_or_else(OutputStreamConfig::null, OutputStreamConfig::callback), - stderr: stderr.map_or_else(OutputStreamConfig::null, OutputStreamConfig::callback), + stdout: stdout + .map_or_else(OutputStreamConfig::memory, OutputStreamConfig::callback), + stderr: stderr + .map_or_else(OutputStreamConfig::memory, OutputStreamConfig::callback), }, ) } From 28926486e01e5227017e82f4e0dbeb4d5b778021 Mon Sep 17 00:00:00 2001 From: bakaq Date: Fri, 31 Jan 2025 13:14:45 -0300 Subject: [PATCH 8/9] More stream tests --- src/machine/config.rs | 28 ++++--- src/machine/lib_machine/mod.rs | 36 ++++++-- src/machine/lib_machine/tests.rs | 36 +------- src/machine/streams.rs | 140 ++++++++++++++++++++++++++++++- 4 files changed, 183 insertions(+), 57 deletions(-) diff --git a/src/machine/config.rs b/src/machine/config.rs index 2e71ae26..02480602 100644 --- a/src/machine/config.rs +++ b/src/machine/config.rs @@ -9,7 +9,6 @@ use crate::Machine; use super::{ bootstrapping_compile, current_dir, import_builtin_impls, libraries, load_module, Arena, Atom, Callback, CompilationTarget, IndexStore, ListingSource, MachineArgs, MachineState, Stream, - StreamOptions, }; #[derive(Default)] @@ -77,14 +76,19 @@ impl OutputStreamConfig { } } -#[derive(Debug, Default)] +#[derive(Debug)] enum InputStreamConfigInner { - #[default] - Null, + String(String), Stdin, Channel(Receiver>), } +impl Default for InputStreamConfigInner { + fn default() -> Self { + Self::String("".into()) + } +} + /// Configuration for an input stream; #[derive(Debug, Default)] pub struct InputStreamConfig { @@ -92,10 +96,10 @@ pub struct InputStreamConfig { } impl InputStreamConfig { - /// Ignores all input. - pub fn null() -> Self { + /// Gets input from string. + pub fn string(s: impl Into) -> Self { Self { - inner: InputStreamConfigInner::Null, + inner: InputStreamConfigInner::String(s.into()), } } @@ -119,7 +123,7 @@ impl InputStreamConfig { fn into_stream(self, arena: &mut Arena, add_history: bool) -> Stream { match self.inner { - InputStreamConfigInner::Null => Stream::Null(StreamOptions::default()), + InputStreamConfigInner::String(s) => Stream::from_owned_string(s, arena), InputStreamConfigInner::Stdin => Stream::stdin(arena, add_history), InputStreamConfigInner::Channel(channel) => Stream::input_channel(channel, arena), } @@ -152,14 +156,12 @@ impl StreamConfig { } } - /// Binds the output stream to a memory buffer, and the error stream to stderr. - /// - /// The input stream is ignored. + /// Binds the output and error streams to memory buffers and has an empty input. pub fn in_memory() -> Self { StreamConfig { - stdin: InputStreamConfig::null(), + stdin: InputStreamConfig::string(""), stdout: OutputStreamConfig::memory(), - stderr: OutputStreamConfig::stderr(), + stderr: OutputStreamConfig::memory(), } } diff --git a/src/machine/lib_machine/mod.rs b/src/machine/lib_machine/mod.rs index 24f22fe7..87b64074 100644 --- a/src/machine/lib_machine/mod.rs +++ b/src/machine/lib_machine/mod.rs @@ -6,11 +6,13 @@ use crate::heap_iter::{stackful_post_order_iter, NonListElider}; use crate::machine::machine_indices::VarKey; use crate::machine::mock_wam::CompositeOpDir; use crate::machine::{ - F64Offset, F64Ptr, Fixnum, Number, BREAK_FROM_DISPATCH_LOOP_LOC, LIB_QUERY_SUCCESS, + ArenaHeaderTag, F64Offset, F64Ptr, Fixnum, Number, BREAK_FROM_DISPATCH_LOOP_LOC, + LIB_QUERY_SUCCESS, }; use crate::parser::ast::{Var, VarPtr}; use crate::parser::parser::{Parser, Tokens}; use crate::read::{write_term_to_heap, TermWriteResult}; +use crate::types::UntypedArenaPtr; use dashu::{Integer, Rational}; use indexmap::IndexMap; @@ -280,11 +282,32 @@ impl Term { (HeapCellValueTag::Fixnum, n) => { term_stack.push(Term::Integer(n.into())); } - (HeapCellValueTag::Cons) => { - match Number::try_from(addr) { - Ok(Number::Integer(i)) => term_stack.push(Term::Integer((*i).clone())), - Ok(Number::Rational(r)) => term_stack.push(Term::Rational((*r).clone())), - _ => {} + (HeapCellValueTag::Cons, ptr) => { + if let Ok(n) = Number::try_from(addr) { + match n { + Number::Integer(i) => term_stack.push(Term::Integer((*i).clone())), + Number::Rational(r) => term_stack.push(Term::Rational((*r).clone())), + _ => { unreachable!() }, + } + } else { + match_untyped_arena_ptr!(ptr, + (ArenaHeaderTag::Stream, stream) => { + let stream_term = if let Some(alias) = stream.options().get_alias() { + Term::atom(alias.as_str().to_string()) + } else { + Term::compound("$stream", [ + Term::integer(stream.as_ptr() as usize) + ]) + }; + term_stack.push(stream_term); + } + (ArenaHeaderTag::Dropped, _stream) => { + term_stack.push(Term::atom("$dropped_value")); + } + _ => { + unreachable!(); + } + ); } } (HeapCellValueTag::CStr, s) => { @@ -394,6 +417,7 @@ impl Term { } */ _ => { + unreachable!(); } ); } diff --git a/src/machine/lib_machine/tests.rs b/src/machine/lib_machine/tests.rs index 502f4d89..5b89d67d 100644 --- a/src/machine/lib_machine/tests.rs +++ b/src/machine/lib_machine/tests.rs @@ -1,8 +1,5 @@ -use std::io::Write; -use std::{cell::RefCell, io::Read, rc::Rc}; - use super::*; -use crate::{MachineBuilder, StreamConfig}; +use crate::MachineBuilder; #[test] #[cfg_attr(miri, ignore = "it takes too long to run")] @@ -611,34 +608,3 @@ fn errors_and_exceptions() { [Ok(LeafAnswer::Exception(Term::atom("a")))] ); } - -#[test] -#[cfg_attr(miri, ignore)] -fn callback_streams() { - let test_string = Rc::new(RefCell::new(String::new())); - let test_string2 = test_string.clone(); - - let (mut user_input, streams) = StreamConfig::with_callbacks( - Some(Box::new(move |x| { - x.read_to_string(&mut test_string2.borrow_mut()).unwrap(); - })), - None, - ); - let mut machine = MachineBuilder::default().with_streams(streams).build(); - - write!(&mut user_input, "a(1,2,3).").unwrap(); - - let complete_answer: Vec<_> = machine - .run_query("read(A), write('asdf'), nl, flush_output.") - .collect(); - - assert_eq!( - complete_answer, - [Ok(LeafAnswer::from_bindings([( - "A", - Term::compound("a", [Term::integer(1), Term::integer(2), Term::integer(3)]) - ),]))] - ); - - assert_eq!(*test_string.borrow(), "asdf\n"); -} diff --git a/src/machine/streams.rs b/src/machine/streams.rs index edb6756b..ea082c91 100644 --- a/src/machine/streams.rs +++ b/src/machine/streams.rs @@ -2097,9 +2097,143 @@ impl MachineState { } #[cfg(test)] -mod test { - use super::*; - use crate::machine::config::*; +mod tests { + use crate::*; + use std::{cell::RefCell, io::Read, io::Write, rc::Rc}; + + fn succeeded(answer: Vec>) -> bool { + // Ideally this should be a method in QueryState or LeafAnswer. + matches!( + answer[0].as_ref(), + Ok(LeafAnswer::True) | Ok(LeafAnswer::LeafAnswer { .. }) + ) + } + + #[test] + #[cfg_attr(miri, ignore)] + fn user_input_string_stream() { + let streams = StreamConfig { + stdin: InputStreamConfig::string("a(1,2,3)."), + ..Default::default() + }; + + let mut machine = MachineBuilder::default().with_streams(streams).build(); + + let complete_answer: Vec<_> = machine + .run_query(r#"current_input(_), \+ at_end_of_stream."#) + .collect(); + + assert!(succeeded(complete_answer)); + + let complete_answer: Vec<_> = machine.run_query("read(A).").collect(); + + assert_eq!( + complete_answer, + [Ok(LeafAnswer::from_bindings([( + "A", + Term::compound("a", [Term::integer(1), Term::integer(2), Term::integer(3),]) + )]))] + ); + + let complete_answer: Vec<_> = machine.run_query(r#"at_end_of_stream."#).collect(); + + assert!(succeeded(complete_answer)); + } + + #[test] + #[cfg_attr(miri, ignore)] + fn user_input_channel_stream() { + let (mut user_input, channel_stream) = InputStreamConfig::channel(); + let streams = StreamConfig { + stdin: channel_stream, + ..Default::default() + }; + + let mut machine = MachineBuilder::default().with_streams(streams).build(); + + let complete_answer: Vec<_> = machine + .run_query(r#"current_input(_), \+ at_end_of_stream."#) + .collect(); + + assert!(succeeded(complete_answer)); + + write!(user_input, "a(1,2,3).").unwrap(); + + let complete_answer: Vec<_> = machine + .run_query(r#"\+ at_end_of_stream, read(A)."#) + .collect(); + + assert_eq!( + complete_answer, + [Ok(LeafAnswer::from_bindings([( + "A", + Term::compound("a", [Term::integer(1), Term::integer(2), Term::integer(3),]) + )]))] + ); + + // End-of-data but not end-of-stream; + let complete_answer: Vec<_> = machine + .run_query( + r#" + use_module(library(charsio)), + current_input(In), get_n_chars(In, N, C), + N == 0, \+ at_end_of_stream. + "#, + ) + .collect(); + + assert!(succeeded(complete_answer)); + + // Dropping the sender closes the input + drop(user_input); + + let complete_answer: Vec<_> = machine + .run_query( + r#" + current_input(In), get_n_chars(In, N, _), + N == 0, at_end_of_stream. + "#, + ) + .collect(); + + assert!(succeeded(complete_answer)); + } + + #[test] + #[cfg_attr(miri, ignore)] + fn user_output_callback_stream() { + let test_string = Rc::new(RefCell::new(String::new())); + + let streams = StreamConfig { + stdout: OutputStreamConfig::callback(Box::new({ + let test_string = test_string.clone(); + move |x| { + x.read_to_string(&mut test_string.borrow_mut()).unwrap(); + } + })), + ..Default::default() + }; + + let mut machine = MachineBuilder::default().with_streams(streams).build(); + + let complete_answer: Vec<_> = machine + .run_query(r#"current_output(Out), \+ at_end_of_stream(Out)."#) + .collect(); + + assert!(succeeded(complete_answer)); + + let complete_answer: Vec<_> = machine + .run_query(r#"write(asdf), nl, flush_output."#) + .collect(); + + assert!(succeeded(complete_answer)); + assert_eq!(test_string.borrow().as_str(), "asdf\n"); + + let complete_answer: Vec<_> = machine.run_query(r#"write(abcd), flush_output."#).collect(); + + assert!(succeeded(complete_answer)); + assert_eq!(test_string.borrow().as_str(), "asdf\nabcd"); + } #[test] #[cfg_attr(miri, ignore)] From 3fc5709c505c89de41444d8b692393b197ad3675 Mon Sep 17 00:00:00 2001 From: bakaq Date: Wed, 5 Feb 2025 12:07:28 -0300 Subject: [PATCH 9/9] Add builder style configuration of user input, output and error --- src/machine/config.rs | 55 +++++++++++++++++++++++++++--------------- src/machine/streams.rs | 20 +++++---------- 2 files changed, 41 insertions(+), 34 deletions(-) diff --git a/src/machine/config.rs b/src/machine/config.rs index 02480602..b9d3f703 100644 --- a/src/machine/config.rs +++ b/src/machine/config.rs @@ -132,12 +132,9 @@ impl InputStreamConfig { /// Describes how the streams of a [`Machine`](crate::Machine) will be handled. pub struct StreamConfig { - /// The configuration for the stdin of the [`Machine`](crate::Machine). - pub stdin: InputStreamConfig, - /// The configuration for the stdout of the [`Machine`](crate::Machine). - pub stdout: OutputStreamConfig, - /// The configuration for the stderr of the [`Machine`](crate::Machine). - pub stderr: OutputStreamConfig, + user_input: InputStreamConfig, + user_output: OutputStreamConfig, + user_error: OutputStreamConfig, } impl Default for StreamConfig { @@ -150,43 +147,61 @@ impl StreamConfig { /// Binds the input, output and error streams to stdin, stdout and stderr. pub fn stdio() -> Self { StreamConfig { - stdin: InputStreamConfig::stdin(), - stdout: OutputStreamConfig::stdout(), - stderr: OutputStreamConfig::stderr(), + user_input: InputStreamConfig::stdin(), + user_output: OutputStreamConfig::stdout(), + user_error: OutputStreamConfig::stderr(), } } /// Binds the output and error streams to memory buffers and has an empty input. pub fn in_memory() -> Self { StreamConfig { - stdin: InputStreamConfig::string(""), - stdout: OutputStreamConfig::memory(), - stderr: OutputStreamConfig::memory(), + user_input: InputStreamConfig::string(""), + user_output: OutputStreamConfig::memory(), + user_error: OutputStreamConfig::memory(), } } /// Calls the given callbacks when the respective streams are written to. /// - /// This also returns a handler to the stdin do the [`Machine`](crate::Machine). - pub fn with_callbacks(stdout: Option, stderr: Option) -> (UserInput, Self) { + /// This also returns a handler to the stdin of the [`Machine`](crate::Machine). + pub fn from_callbacks(stdout: Option, stderr: Option) -> (UserInput, Self) { let (user_input, channel_stream) = InputStreamConfig::channel(); ( user_input, StreamConfig { - stdin: channel_stream, - stdout: stdout + user_input: channel_stream, + user_output: stdout .map_or_else(OutputStreamConfig::memory, OutputStreamConfig::callback), - stderr: stderr + user_error: stderr .map_or_else(OutputStreamConfig::memory, OutputStreamConfig::callback), }, ) } + /// Configures the `user_input` stream. + pub fn with_user_input(self, user_input: InputStreamConfig) -> Self { + Self { user_input, ..self } + } + + /// Configures the `user_output` stream. + pub fn with_user_output(self, user_output: OutputStreamConfig) -> Self { + Self { + user_output, + ..self + } + } + + /// Configures the `user_error` stream. + pub fn with_user_error(self, user_error: OutputStreamConfig) -> Self { + Self { user_error, ..self } + } + fn into_streams(self, arena: &mut Arena, add_history: bool) -> (Stream, Stream, Stream) { ( - self.stdin.into_stream(arena, add_history), - self.stdout.into_stream(arena), - self.stderr.into_stream(arena), + self.user_input.into_stream(arena, add_history), + self.user_output.into_stream(arena), + self.user_error.into_stream(arena), ) } } diff --git a/src/machine/streams.rs b/src/machine/streams.rs index ea082c91..500ed679 100644 --- a/src/machine/streams.rs +++ b/src/machine/streams.rs @@ -2112,10 +2112,8 @@ mod tests { #[test] #[cfg_attr(miri, ignore)] fn user_input_string_stream() { - let streams = StreamConfig { - stdin: InputStreamConfig::string("a(1,2,3)."), - ..Default::default() - }; + let streams = + StreamConfig::default().with_user_input(InputStreamConfig::string("a(1,2,3).")); let mut machine = MachineBuilder::default().with_streams(streams).build(); @@ -2144,11 +2142,7 @@ mod tests { #[cfg_attr(miri, ignore)] fn user_input_channel_stream() { let (mut user_input, channel_stream) = InputStreamConfig::channel(); - let streams = StreamConfig { - stdin: channel_stream, - ..Default::default() - }; - + let streams = StreamConfig::default().with_user_input(channel_stream); let mut machine = MachineBuilder::default().with_streams(streams).build(); let complete_answer: Vec<_> = machine @@ -2204,15 +2198,13 @@ mod tests { fn user_output_callback_stream() { let test_string = Rc::new(RefCell::new(String::new())); - let streams = StreamConfig { - stdout: OutputStreamConfig::callback(Box::new({ + let streams = + StreamConfig::default().with_user_output(OutputStreamConfig::callback(Box::new({ let test_string = test_string.clone(); move |x| { x.read_to_string(&mut test_string.borrow_mut()).unwrap(); } - })), - ..Default::default() - }; + }))); let mut machine = MachineBuilder::default().with_streams(streams).build();