rishi/scryer-prolog
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[WIP] add support to spawn new processes

Browse Source
This commit is contained in:
Bennet Bleßmann
2025-07-19 20:50:53 +02:00
parent f91aedcc2f
commit ae0baf4893
9 changed files with 326 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
src/arena.rs
View File

@@ -14,6 +14,8 @@ use ordered_float::OrderedFloat;
use std::fmt;
use std::fmt::Debug;
use std::hash::{Hash, Hasher};
use std::io::PipeReader;
use std::io::PipeWriter;
use std::mem;
use std::mem::ManuallyDrop;
use std::net::TcpListener;
@@ -71,7 +73,9 @@ pub enum ArenaHeaderTag {
TcpListener = 0b1000000,
HttpListener = 0b1000001,
HttpResponse = 0b1000010,
PipeWriter = 0b1000011,
Dropped = 0b1000100,
PipeReader = 0b1000101,
}
#[bitfield]
@@ -546,6 +550,12 @@ unsafe fn drop_slab_in_place(value: NonNull<AllocSlab>, tag: ArenaHeaderTag) {
ArenaHeaderTag::StandardErrorStream => {
drop_typed_slab_in_place!(StandardErrorStream, value);
}
ArenaHeaderTag::PipeReader => {
drop_typed_slab_in_place!(PipeReader, value);
}
ArenaHeaderTag::PipeWriter => {
drop_typed_slab_in_place!(PipeWriter, value);
}
ArenaHeaderTag::NullStream => {
unreachable!("NullStream is never arena allocated!");
}

14
src/lib/error.pl
View File

@@ -82,6 +82,9 @@ must_be_(octet_chars, Cs) :-
; true
).
must_be_(list, Term) :- check_(error:ilist, list, Term).
must_be_(list(Elem), Term) :-
must_be_(list, Term),
check_all(Elem, Term).
must_be_(type, Term) :- check_(error:type, type, Term).
must_be_(boolean, Term) :- check_(error:boolean, boolean, Term).
must_be_(pair, Term) :- check_(error:pair, pair, Term).
@@ -96,10 +99,12 @@ must_be_(term, Term) :-
% We cannot use maplist(must_be(character), Cs), because library(lists)
% uses library(error), so importing it would create a cyclic dependency.
all_characters([]).
all_characters([C|Cs]) :-
must_be(character, C),
all_characters(Cs).
check_all(_, []).
check_all(Type, [Head| Tail]) :-
must_be(Type, Head),
check_all(Type, Tail).
all_characters(Cs) :- check_all(character, Cs).
check_(Pred, Type, Term) :-
( var(Term) -> instantiation_error(must_be/2)
@@ -141,6 +146,7 @@ type(octet_character).
type(octet_chars).
type(chars).
type(list).
type(list(Type)) :- type(Type).
type(var).
type(boolean).
type(term).

52
src/lib/process.pl Normal file
View File

@@ -0,0 +1,52 @@
:- module(process, [process_create/3]).
:- use_module(library(error)).
:- use_module(library(iso_ext)).
:- use_module(library(lists), [append/3, member/2]).
process_create(Exe, Args, Options) :-
must_be(chars, Exe),
must_be(list(chars), Args),
must_be(list, Options),
check_option(Sin, find_stdio(Sin, stdin, Options), valid_stdio, [std], Stdin),
check_option(Sout, find_stdio(Sout, stdout, Options), valid_stdio, [std], Stdout),
check_option(Serr, find_stdio(Serr, stderr, Options), valid_stdio, [std], Stderr),
check_option(Envs, find_env(Envs, Options), valid_env, [environment, []], EnvVars),
check_option(P, member(process(P), Options), valid_pid, _, Pid),
check_option(C, member(cwd(C), Options), valid_cwd, _, Cwd),
'$process_create'(Exe, Args, Stdin, Stdout, Stderr, EnvVars, Cwd, Pid).
check_option(Template, Goal, Pred, Default, Choice) :-
findall(Template, Goal, Solutions),
check_option_(Solutions, Pred, Default, Choice).
check_option_([] , Pred , Default , Default ) :- call(Pred, Default).
check_option_([Choice] , Pred , _ , Choice ) :- call(Pred, Choice).
check_option_([X1,X2|Xs], _ , _ , _ ) :- throw(error(duplicate_option, process_create/3, [X1, X2 | Xs])).
find_stdio([std], Kind, Options ) :- Elem =.. [Kind, std], member(Elem, Options).
find_stdio([null], Kind, Options ) :- Elem =.. [Kind, null], member(Elem, Options).
find_stdio([pipe, Stream], Kind, Options) :- Elem =.. [Kind, pipe(Stream)], member(Elem, Options).
find_stdio([file, Path], Kind, Options ) :- Elem =.. [Kind, file(Path)], member(Elem, Options).
valid_stdio([std]).
valid_stdio([null]).
valid_stdio([pipe, Stream]) :- must_be(var, Stream).
valid_stdio([file, Path]) :- must_be(chars, Path).
find_env([env, E], Options) :- member(env(E), Options).
find_env([environment, E], Options) :- member(environment(E), Options).
valid_cwd(Cwd) :- must_be(chars, Cwd).
valid_env([env, E]) :- valid_env_(E).
valid_env([environment, E]) :- valid_env_(E).
valid_env_([]).
valid_env_([N=V|Es]) :-
must_be(chars, N),
must_be(chars, V),
valid_env_(Es).
valid_pid(Pid) :- must_be(var, Pid).

8
src/machine/dispatch.rs
View File

@@ -4787,6 +4787,14 @@ impl Machine {
self.shell();
step_or_fail!(self, self.machine_st.p = self.machine_st.cp);
}
&Instruction::CallProcessCreate => {
try_or_throw!(self.machine_st, self.process_create());
step_or_fail!(self, self.machine_st.p += 1);
}
&Instruction::ExecuteProcessCreate => {
try_or_throw!(self.machine_st, self.process_create());
step_or_fail!(self, self.machine_st.p = self.machine_st.cp);
}
&Instruction::CallPid => {
self.pid();
step_or_fail!(self, self.machine_st.p += 1);

81
src/machine/streams.rs
View File

@@ -23,6 +23,8 @@ use std::fmt::Debug;
use std::fs::{File, OpenOptions};
use std::hash::Hash;
use std::io;
use std::io::PipeReader;
use std::io::PipeWriter;
use std::io::{Cursor, ErrorKind, Read, Seek, SeekFrom, Write};
use std::mem::ManuallyDrop;
use std::net::{Shutdown, TcpStream};
@@ -588,6 +590,8 @@ arena_allocated_impl_for_stream!(StandardOutputStream, StandardOutputStream);
arena_allocated_impl_for_stream!(StandardErrorStream, StandardErrorStream);
arena_allocated_impl_for_stream!(CharReader<CallbackStream>, CallbackStream);
arena_allocated_impl_for_stream!(CharReader<InputChannelStream>, InputChannelStream);
arena_allocated_impl_for_stream!(CharReader<PipeReader>, PipeReader);
arena_allocated_impl_for_stream!(CharReader<PipeWriter>, PipeWriter);
#[derive(Debug, Copy, Clone)]
pub enum Stream {
@@ -608,6 +612,8 @@ pub enum Stream {
StandardError(TypedArenaPtr<StandardErrorStream>),
Callback(TypedArenaPtr<CallbackStream>),
InputChannel(TypedArenaPtr<InputChannelStream>),
PipeReader(TypedArenaPtr<PipeReader>),
PipeWriter(TypedArenaPtr<PipeWriter>),
}
impl From<TypedArenaPtr<ReadlineStream>> for Stream {
@@ -726,6 +732,8 @@ impl Stream {
Stream::StandardError(ptr) => ptr.header_ptr(),
Stream::Callback(ptr) => ptr.header_ptr(),
Stream::InputChannel(ptr) => ptr.header_ptr(),
Stream::PipeReader(ptr) => ptr.header_ptr(),
Stream::PipeWriter(ptr) => ptr.header_ptr(),
}
}
@@ -748,6 +756,8 @@ impl Stream {
Stream::StandardError(ref ptr) => &ptr.options,
Stream::Callback(ref ptr) => &ptr.options,
Stream::InputChannel(ref ptr) => &ptr.options,
Stream::PipeReader(ref ptr) => &ptr.options,
Stream::PipeWriter(ref ptr) => &ptr.options,
}
}
@@ -770,6 +780,8 @@ impl Stream {
Stream::StandardError(ref mut ptr) => &mut ptr.options,
Stream::Callback(ref mut ptr) => &mut ptr.options,
Stream::InputChannel(ref mut ptr) => &mut ptr.options,
Stream::PipeReader(ref mut ptr) => &mut ptr.options,
Stream::PipeWriter(ref mut ptr) => &mut ptr.options,
}
}
@@ -793,6 +805,8 @@ impl Stream {
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,
Stream::PipeReader(ptr) => ptr.lines_read += incr_num_lines_read,
Stream::PipeWriter(_) => {}
}
}
@@ -816,6 +830,8 @@ impl Stream {
Stream::StandardError(ptr) => ptr.lines_read = value,
Stream::Callback(ptr) => ptr.lines_read = value,
Stream::InputChannel(ptr) => ptr.lines_read = value,
Stream::PipeReader(ptr) => ptr.lines_read = value,
Stream::PipeWriter(_) => {}
}
}
@@ -839,6 +855,8 @@ impl Stream {
Stream::StandardError(ptr) => ptr.lines_read,
Stream::Callback(ptr) => ptr.lines_read,
Stream::InputChannel(ptr) => ptr.lines_read,
Stream::PipeReader(ptr) => ptr.lines_read,
Stream::PipeWriter(_) => 0,
}
}
}
@@ -856,6 +874,8 @@ impl CharRead for Stream {
Stream::StaticString(src) => (*src).peek_char(),
Stream::Byte(cursor) => (*cursor).peek_char(),
Stream::InputChannel(cursor) => (*cursor).peek_char(),
Stream::PipeReader(cursor) => (*cursor).peek_char(),
#[cfg(feature = "http")]
Stream::HttpWrite(_) => Some(Err(std::io::Error::new(
ErrorKind::PermissionDenied,
@@ -865,7 +885,8 @@ impl CharRead for Stream {
| Stream::StandardError(_)
| Stream::StandardOutput(_)
| Stream::Null(_)
| Stream::Callback(_) => Some(Err(std::io::Error::new(
| Stream::Callback(_)
| Stream::PipeWriter(_) => Some(Err(std::io::Error::new(
ErrorKind::PermissionDenied,
StreamError::ReadFromOutputStream,
))),
@@ -884,6 +905,7 @@ impl CharRead for Stream {
Stream::StaticString(src) => (*src).read_char(),
Stream::Byte(cursor) => (*cursor).read_char(),
Stream::InputChannel(cursor) => (*cursor).read_char(),
Stream::PipeReader(cursor) => (*cursor).read_char(),
#[cfg(feature = "http")]
Stream::HttpWrite(_) => Some(Err(std::io::Error::new(
ErrorKind::PermissionDenied,
@@ -893,7 +915,8 @@ impl CharRead for Stream {
| Stream::StandardError(_)
| Stream::StandardOutput(_)
| Stream::Null(_)
| Stream::Callback(_) => Some(Err(std::io::Error::new(
| Stream::Callback(_)
| Stream::PipeWriter(_) => Some(Err(std::io::Error::new(
ErrorKind::PermissionDenied,
StreamError::ReadFromOutputStream,
))),
@@ -911,13 +934,15 @@ impl CharRead for Stream {
Stream::Readline(rl_stream) => rl_stream.put_back_char(c),
Stream::StaticString(src) => src.put_back_char(c),
Stream::Byte(cursor) => cursor.put_back_char(c),
Stream::PipeReader(cursor) => cursor.put_back_char(c),
#[cfg(feature = "http")]
Stream::HttpWrite(_) => {}
Stream::OutputFile(_)
| Stream::StandardError(_)
| Stream::StandardOutput(_)
| Stream::Null(_)
| Stream::Callback(_) => {}
| Stream::Callback(_)
| Stream::PipeWriter(_) => {}
Stream::InputChannel(_) => {}
}
}
@@ -934,13 +959,15 @@ impl CharRead for Stream {
Stream::StaticString(ref mut src) => src.consume(nread),
Stream::Byte(ref mut cursor) => cursor.consume(nread),
Stream::InputChannel(ref mut cursor) => cursor.consume(nread),
Stream::PipeReader(ref mut cursor) => cursor.consume(nread),
#[cfg(feature = "http")]
Stream::HttpWrite(_) => {}
Stream::OutputFile(_)
| Stream::StandardError(_)
| Stream::StandardOutput(_)
| Stream::Null(_)
| Stream::Callback(_) => {}
| Stream::Callback(_)
| Stream::PipeWriter(_) => {}
}
}
}
@@ -959,6 +986,7 @@ impl Read for Stream {
Stream::StaticString(src) => (*src).read(buf),
Stream::Byte(cursor) => (*cursor).read(buf),
Stream::InputChannel(cursor) => (*cursor).read(buf),
Stream::PipeReader(cursor) => (*cursor).read(buf),
#[cfg(feature = "http")]
Stream::HttpWrite(_) => Err(std::io::Error::new(
ErrorKind::PermissionDenied,
@@ -967,7 +995,8 @@ impl Read for Stream {
Stream::OutputFile(_)
| Stream::StandardError(_)
| Stream::StandardOutput(_)
| Stream::Callback(_) => Err(std::io::Error::new(
| Stream::Callback(_)
| Stream::PipeWriter(_) => Err(std::io::Error::new(
ErrorKind::PermissionDenied,
StreamError::ReadFromOutputStream,
)),
@@ -989,6 +1018,7 @@ impl Write for Stream {
Stream::StandardError(stream) => stream.write(buf),
#[cfg(feature = "http")]
Stream::HttpWrite(ref mut stream) => stream.get_mut().write(buf),
Stream::PipeWriter(ref mut stream) => stream.get_mut().write(buf),
#[cfg(feature = "http")]
Stream::HttpRead(_) => Err(std::io::Error::new(
ErrorKind::PermissionDenied,
@@ -998,7 +1028,8 @@ impl Write for Stream {
Stream::StaticString(_)
| Stream::InputChannel(_)
| Stream::Readline(_)
| Stream::InputFile(..) => Err(std::io::Error::new(
| Stream::InputFile(..)
| Stream::PipeReader(_) => Err(std::io::Error::new(
ErrorKind::PermissionDenied,
StreamError::WriteToInputStream,
)),
@@ -1015,6 +1046,7 @@ impl Write for Stream {
Stream::Callback(ref mut callback_stream) => callback_stream.stream.get_mut().flush(),
Stream::StandardError(stream) => stream.stream.flush(),
Stream::StandardOutput(stream) => stream.stream.flush(),
Stream::PipeWriter(ref mut stream) => stream.stream.get_mut().flush(),
#[cfg(feature = "http")]
Stream::HttpWrite(ref mut stream) => stream.stream.get_mut().flush(),
#[cfg(feature = "http")]
@@ -1026,7 +1058,8 @@ impl Write for Stream {
Stream::StaticString(_)
| Stream::InputChannel(_)
| Stream::Readline(_)
| Stream::InputFile(_) => Err(std::io::Error::new(
| Stream::InputFile(_)
| Stream::PipeReader(_) => Err(std::io::Error::new(
ErrorKind::PermissionDenied,
StreamError::FlushToInputStream,
)),
@@ -1192,6 +1225,8 @@ impl 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,
Stream::PipeReader(stream) => stream.past_end_of_stream,
Stream::PipeWriter(stream) => stream.past_end_of_stream,
}
}
@@ -1220,6 +1255,8 @@ impl Stream {
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,
Stream::PipeReader(stream) => stream.past_end_of_stream = value,
Stream::PipeWriter(stream) => stream.past_end_of_stream = value,
}
}
@@ -1330,7 +1367,8 @@ impl Stream {
| Stream::InputChannel(_)
| Stream::Readline(_)
| Stream::StaticString(_)
| Stream::InputFile(..) => atom!("read"),
| Stream::InputFile(..)
| Stream::PipeReader(_) => atom!("read"),
Stream::NamedTcp(..) => atom!("read_append"),
Stream::OutputFile(file) if file.is_append => atom!("append"),
#[cfg(feature = "http")]
@@ -1338,7 +1376,8 @@ impl Stream {
Stream::OutputFile(_)
| Stream::StandardError(_)
| Stream::StandardOutput(_)
| Stream::Callback(_) => {
| Stream::Callback(_)
| Stream::PipeWriter(_) => {
atom!("write")
}
Stream::Null(_) => atom!(""),
@@ -1372,6 +1411,20 @@ impl Stream {
))
}
pub(crate) fn from_pipe_writer(writer: io::PipeWriter, arena: &mut Arena) -> Stream {
Stream::PipeWriter(arena_alloc!(
ManuallyDrop::new(StreamLayout::new(CharReader::new(writer))),
arena
))
}
pub(crate) fn from_pipe_reader(reader: io::PipeReader, arena: &mut Arena) -> Stream {
Stream::PipeReader(arena_alloc!(
ManuallyDrop::new(StreamLayout::new(CharReader::new(reader))),
arena
))
}
#[inline]
pub(crate) fn from_tcp_stream(address: Atom, tcp_stream: TcpStream, arena: &mut Arena) -> Self {
tcp_stream.set_read_timeout(None).unwrap();
@@ -1512,6 +1565,16 @@ impl Stream {
Ok(())
}
Stream::PipeReader(mut stream) => {
stream.drop_payload();
Ok(())
}
Stream::PipeWriter(mut stream) => {
stream.drop_payload();
Ok(())
}
Stream::Null(_) => Ok(()),
Stream::Readline(_) | Stream::StandardOutput(_) | Stream::StandardError(_) => {

168
src/machine/system_calls.rs
View File

@@ -56,6 +56,7 @@ use std::net::{SocketAddr, ToSocketAddrs};
use std::net::{TcpListener, TcpStream};
use std::num::NonZeroU32;
use std::process;
use std::process::Stdio;
#[cfg(feature = "http")]
use std::str::FromStr;
#[cfg(feature = "http")]
@@ -8393,6 +8394,173 @@ impl Machine {
};
}
pub(crate) fn process_create(&mut self) -> CallResult {
fn stub_gen() -> Vec<FunctorElement> {
functor_stub(atom!("process_create"), 3)
}
let exe_r = self.deref_register(1);
let args_r = self.deref_register(2);
let stdin_r = self.deref_register(3);
let stdout_r = self.deref_register(4);
let stderr_r = self.deref_register(5);
let env_r = self.deref_register(6);
let cwd_r = self.deref_register(7);
let pid_r = self.deref_register(8);
let exe = self.machine_st.value_to_str_like(exe_r).unwrap();
let args = self
.machine_st
.try_from_list(args_r, stub_gen)
.unwrap()
.into_iter()
.map(|arg| {
self.machine_st
.value_to_str_like(arg)
.unwrap()
.as_str()
.to_string()
})
.collect::<Vec<_>>();
let stdin_args = self.machine_st.try_from_list(stdin_r, stub_gen)?;
let stdin = self.handle_input_stream(stdin_args)?;
let stdout_args = self.machine_st.try_from_list(stdout_r, stub_gen)?;
let stdout = self.handle_output_stream(stdout_args)?;
let stderr_args = self.machine_st.try_from_list(stderr_r, stub_gen)?;
let stderr = self.handle_output_stream(stderr_args)?;
let env_args = self.machine_st.try_from_list(env_r, stub_gen)?;
let clear_env = match env_args[0].to_atom() {
Some(atom!("env")) => true,
Some(atom!("environment")) => false,
_ => panic!("Invalid value for clear_env"),
};
let env_names = self.machine_st.try_from_list(env_args[1], stub_gen)?;
let env_values = self.machine_st.try_from_list(env_args[2], stub_gen)?;
let envs = env_names
.into_iter()
.zip(env_values)
.map(|(name, value)| {
let name = self
.machine_st
.value_to_str_like(name)
.unwrap()
.as_str()
.to_string();
let value = self
.machine_st
.value_to_str_like(value)
.unwrap()
.as_str()
.to_string();
(name, value)
})
.collect::<Vec<_>>();
let cwd = self.machine_st.value_to_str_like(cwd_r);
let mut command = std::process::Command::new(&*exe.as_str());
command.args(args);
if let Some(cwd) = cwd {
command.current_dir(&*cwd.as_str());
}
if clear_env {
command.env_clear();
}
command
.envs(envs)
.stdin(stdin)
.stdout(stdout)
.stderr(stderr);
match command.spawn() {
Ok(child) => {
self.machine_st
.unify_fixnum(Fixnum::build_with(child.id()), pid_r);
Ok(())
}
Err(_) => {
self.machine_st.fail = true;
Ok(())
}
}
}
fn handle_output_stream(&mut self, args: Vec<HeapCellValue>) -> Result<Stdio, MachineStub> {
Ok(match args[0].to_atom() {
Some(atom!("std")) => Stdio::inherit(),
Some(atom!("null")) => Stdio::null(),
Some(atom!("pipe")) => {
// TODO handler Err
let (reader, writer) = std::io::pipe().unwrap();
let stream = Stream::from_pipe_reader(reader, &mut self.machine_st.arena);
self.indices
.add_stream(stream, atom!("process_create"), 3)
.map_err(|stub_gen| stub_gen(&mut self.machine_st))?;
self.machine_st
.bind(args[2].as_var().unwrap(), stream.into());
Stdio::from(writer)
}
Some(atom!("file")) => {
let path = self.machine_st.value_to_str_like(args[1]).unwrap();
// TODO handler Err
let file = std::fs::File::open(&*path.as_str()).unwrap();
Stdio::from(file)
}
_ => {
panic!("Invalid stdin tag")
}
})
}
fn handle_input_stream(&mut self, args: Vec<HeapCellValue>) -> Result<Stdio, MachineStub> {
Ok(match args[0].to_atom() {
Some(atom!("std")) => Stdio::inherit(),
Some(atom!("null")) => Stdio::null(),
Some(atom!("pipe")) => {
// TODO handler Err
let (reader, writer) = std::io::pipe().unwrap();
let stream = Stream::from_pipe_writer(writer, &mut self.machine_st.arena);
self.indices
.add_stream(stream, atom!("process_create"), 3)
.map_err(|stub_gen| stub_gen(&mut self.machine_st))
.unwrap();
self.machine_st
.bind(args[2].as_var().unwrap(), stream.into());
Stdio::from(reader)
}
Some(atom!("file")) => {
let path = self.machine_st.value_to_str_like(args[1]).unwrap();
// TODO handler Err
let file = std::fs::File::open(&*path.as_str()).unwrap();
Stdio::from(file)
}
_ => {
panic!("Invalid stdin tag")
}
})
}
#[inline(always)]
pub(crate) fn chars_base64(&mut self) -> CallResult {
let padding = cell_as_atom!(self.deref_register(3));