411 lines
14 KiB
Rust
411 lines
14 KiB
Rust
use super::*;
|
|
use std::collections::BTreeMap;
|
|
use std::os::unix::process::CommandExt;
|
|
|
|
fn sources() -> SourceMap {
|
|
let mut sources = SourceMap::default();
|
|
sources.add(
|
|
"media/native/port.dever",
|
|
"call(value: Int) (output: Int) fails app.NativeError",
|
|
);
|
|
sources.add(
|
|
"media/native/adapter.dever",
|
|
"external exec \"bin/worker\" {}",
|
|
);
|
|
sources.add(
|
|
"media/native/app.dever",
|
|
r#"type NativeError { error Rejected }
|
|
call(value: Int) (output: Int) { output = port.call(value) }
|
|
stall() () { output = port.call(-4) }
|
|
cancel() (output: Int) {
|
|
output = port.call(7)
|
|
pending = run(stall())
|
|
dever.task.sleep(100)
|
|
stop(pending)
|
|
}
|
|
expired() (output: Int) { output = 99 }
|
|
deadline() (output: Int) {
|
|
pending = run(call(-4))
|
|
output = timeout(pending, 100, expired)
|
|
}"#,
|
|
);
|
|
sources.add(
|
|
"media/native/api.dever",
|
|
"cmd call = app.call\ncmd cancel = app.cancel\ncmd deadline = app.deadline",
|
|
);
|
|
sources
|
|
}
|
|
|
|
#[test]
|
|
fn real_c_abi_fixture_checks_in_both_backends() {
|
|
let sources = sources();
|
|
let program = dever_core::check(&sources).unwrap_or_else(|errors| panic!("{errors:?}"));
|
|
dever_core::native::emit_project(
|
|
&program,
|
|
&sources,
|
|
dever_runtime::config::RuntimeProfile::default(),
|
|
)
|
|
.unwrap();
|
|
dever_core::llvm::emit_application(&program, &sources).unwrap();
|
|
}
|
|
|
|
struct Fixture {
|
|
directory: TemporaryDirectory,
|
|
worker: PathBuf,
|
|
executable: PathBuf,
|
|
hello: serde_json::Value,
|
|
rejected: String,
|
|
}
|
|
|
|
impl Fixture {
|
|
fn prepare() -> Self {
|
|
let compiler = PathBuf::from(std::env::var_os("DEVER_FFI_CC").expect("set DEVER_FFI_CC"));
|
|
let assets = PathBuf::from(
|
|
std::env::var_os("DEVER_PROCESS_RUNTIME").expect("set DEVER_PROCESS_RUNTIME"),
|
|
);
|
|
for path in [&compiler, &assets] {
|
|
assert!(
|
|
path.is_absolute(),
|
|
"fixture inputs must be explicit absolute paths"
|
|
);
|
|
}
|
|
let directory = TemporaryDirectory::new();
|
|
let sources = sources();
|
|
let program = dever_core::check(&sources).unwrap_or_else(|errors| panic!("{errors:?}"));
|
|
let contract = &program.external_worker_contracts()[0];
|
|
let rejected = contract.errors.keys().next().unwrap().clone();
|
|
let worker = directory.path().join(&contract.entry);
|
|
let libraries = worker.parent().unwrap().join("lib");
|
|
fs::create_dir_all(&libraries).unwrap();
|
|
let library_source = directory.path().join("library.c");
|
|
fs::write(
|
|
&library_source,
|
|
r#"#include <stdio.h>
|
|
#include <unistd.h>
|
|
long long fixture_call(long long value) {
|
|
if (value == -3) _exit(73);
|
|
if (value == -4) {
|
|
fputs("native-wait\n", stderr);
|
|
fflush(stderr);
|
|
sleep(60);
|
|
}
|
|
return value * 3;
|
|
}
|
|
"#,
|
|
)
|
|
.unwrap();
|
|
let output = Command::new(&compiler)
|
|
.args([
|
|
"-std=c11",
|
|
"-Wall",
|
|
"-Wextra",
|
|
"-Werror",
|
|
"-shared",
|
|
"-fPIC",
|
|
"-Wl,-soname,libfixture.so",
|
|
])
|
|
.arg(&library_source)
|
|
.arg("-o")
|
|
.arg(libraries.join("libfixture.so"))
|
|
.output()
|
|
.unwrap();
|
|
assert!(
|
|
output.status.success(),
|
|
"{}",
|
|
String::from_utf8_lossy(&output.stderr)
|
|
);
|
|
let adapter_source = directory.path().join("adapter.c");
|
|
fs::write(&adapter_source, format!(r#"#include "dever_ffi.h"
|
|
#include <stdio.h>
|
|
#include <string.h>
|
|
extern long long fixture_call(long long value);
|
|
static int invoke(void *context, dever_ffi_slice operation, dever_ffi_slice payload, dever_ffi_output *output) {{
|
|
(void)context;
|
|
if (operation.size != 4 || memcmp(operation.data, "call", 4) != 0 || payload.size >= 64) return DEVER_FFI_FAULT;
|
|
char input[64];
|
|
memcpy(input, payload.data, (size_t)payload.size);
|
|
input[payload.size] = 0;
|
|
long long value = 0;
|
|
if (sscanf(input, "{{\"value\":%lld}}", &value) != 1 || value > 1000000 || value < -1000000) return DEVER_FFI_FAULT;
|
|
if (value == -1) {{
|
|
const char *error = "{{\"error\":\"{rejected}\",\"payload\":{{}}}}";
|
|
memcpy(output->data, error, strlen(error));
|
|
output->size = strlen(error);
|
|
return DEVER_FFI_ERROR;
|
|
}}
|
|
if (value == -2) return DEVER_FFI_FAULT;
|
|
int size = snprintf(output->data, (size_t)output->capacity, "{{\"output\":%lld}}", fixture_call(value));
|
|
if (size < 0 || (uint64_t)size >= output->capacity) return DEVER_FFI_FAULT;
|
|
output->size = (uint64_t)size;
|
|
return DEVER_FFI_OK;
|
|
}}
|
|
static void closed(void *context) {{ (void)context; fputs("native-closed\n", stderr); }}
|
|
int main(void) {{
|
|
const dever_ffi_handler handler = {{DEVER_FFI_ABI_V1, NULL, NULL, invoke, closed}};
|
|
return dever_ffi_serve(&handler);
|
|
}}
|
|
"#)).unwrap();
|
|
let workspace = Path::new(env!("CARGO_MANIFEST_DIR")).join("../..");
|
|
let output = Command::new(&compiler)
|
|
.args(["-std=c11", "-Wall", "-Wextra", "-Werror", "-I"])
|
|
.arg(workspace.join("sdk/c"))
|
|
.arg(workspace.join("sdk/c/dever_ffi.c"))
|
|
.arg(&adapter_source)
|
|
.arg("-L")
|
|
.arg(&libraries)
|
|
.args(["-lfixture", "-Wl,-rpath,$ORIGIN/lib", "-o"])
|
|
.arg(&worker)
|
|
.output()
|
|
.unwrap();
|
|
assert!(
|
|
output.status.success(),
|
|
"{}",
|
|
String::from_utf8_lossy(&output.stderr)
|
|
);
|
|
let mut supplied = Vec::new();
|
|
for subdirectory in ["lib"] {
|
|
for entry in fs::read_dir(assets.join(subdirectory)).unwrap() {
|
|
let entry = entry.unwrap();
|
|
assert!(entry.file_type().unwrap().is_file());
|
|
let name = entry.file_name().into_string().unwrap();
|
|
let path = format!("process-runtime/{subdirectory}/{name}");
|
|
let input = entry.path();
|
|
let bytes = fs::read(input).unwrap();
|
|
supplied.push(EmbeddedResource {
|
|
sha256: sha(&bytes),
|
|
executable: dever_process::asset_executable(
|
|
path.strip_prefix("process-runtime/").unwrap(),
|
|
),
|
|
path,
|
|
bytes,
|
|
});
|
|
}
|
|
}
|
|
let mut entry = resource(contract.entry.clone(), fs::read(&worker).unwrap());
|
|
entry.executable = true;
|
|
supplied.push(entry);
|
|
let mut packaged = workers::prepare(directory.path(), &program, &supplied).unwrap();
|
|
let manifest = packaged
|
|
.iter()
|
|
.find(|file| file.path.ends_with("/contract.json"))
|
|
.unwrap();
|
|
fs::write(
|
|
worker.parent().unwrap().join("contract.json"),
|
|
&manifest.bytes,
|
|
)
|
|
.unwrap();
|
|
assert_eq!(
|
|
serde_json::from_slice::<serde_json::Value>(&manifest.bytes).unwrap(),
|
|
contract.sdk_manifest()
|
|
);
|
|
packaged.extend(
|
|
supplied
|
|
.into_iter()
|
|
.filter(|file| file.path.starts_with("process-runtime/")),
|
|
);
|
|
let native = dever_core::native::compile_project_with_resources(
|
|
&program,
|
|
&sources,
|
|
std::ffi::OsStr::new("rustc"),
|
|
dever_runtime::config::RuntimeProfile::default(),
|
|
&packaged,
|
|
)
|
|
.unwrap();
|
|
let executable = directory.path().join("application");
|
|
native.save(&executable).unwrap();
|
|
fs::create_dir(directory.path().join("config")).unwrap();
|
|
fs::write(
|
|
directory.path().join("config/setting.json"),
|
|
r#"{"adapter":{"media.native":{"use":"default"}}}"#,
|
|
)
|
|
.unwrap();
|
|
Self {
|
|
directory,
|
|
worker,
|
|
executable,
|
|
rejected,
|
|
hello: json!({"kind":"hello", "version":"dever-component-1", "port":contract.port,
|
|
"adapter":contract.adapter, "schema":contract.schema, "capabilities":contract.capabilities,
|
|
"operations":contract.operations, "setting":{}}),
|
|
}
|
|
}
|
|
|
|
fn capture(
|
|
&self,
|
|
command: &mut Command,
|
|
input: &[u8],
|
|
limit: Duration,
|
|
) -> (bool, Vec<u8>, String) {
|
|
let stdout = self.directory.path().join("stdout");
|
|
let stderr = self.directory.path().join("stderr");
|
|
let child = command
|
|
.env_clear()
|
|
.process_group(0)
|
|
.stdin(Stdio::piped())
|
|
.stdout(fs::File::create(&stdout).unwrap())
|
|
.stderr(fs::File::create(&stderr).unwrap())
|
|
.spawn()
|
|
.unwrap();
|
|
let mut owned = OwnedProcessGroup(child);
|
|
owned.0.stdin.take().unwrap().write_all(input).unwrap();
|
|
let mut descendants = Descendants::default();
|
|
let deadline = Instant::now() + limit;
|
|
let status = loop {
|
|
descendants.observe(owned.0.id());
|
|
if let Some(status) = owned.0.try_wait().unwrap() {
|
|
break status;
|
|
}
|
|
assert!(
|
|
Instant::now() < deadline,
|
|
"fixture exceeded its bounded process deadline"
|
|
);
|
|
thread::sleep(Duration::from_millis(5));
|
|
};
|
|
let cleanup_deadline = Instant::now() + Duration::from_secs(3);
|
|
loop {
|
|
let alive = descendants
|
|
.0
|
|
.iter()
|
|
.any(|(pid, started)| process_start(*pid) == Some(*started));
|
|
if !alive {
|
|
break;
|
|
}
|
|
assert!(
|
|
Instant::now() < cleanup_deadline,
|
|
"native Worker descendants were not reaped: {descendants:?}"
|
|
);
|
|
thread::sleep(Duration::from_millis(5));
|
|
}
|
|
(
|
|
status.success(),
|
|
fs::read(stdout).unwrap(),
|
|
fs::read_to_string(stderr).unwrap(),
|
|
)
|
|
}
|
|
}
|
|
|
|
#[derive(Default, Debug)]
|
|
struct Descendants(BTreeMap<u32, u64>);
|
|
|
|
impl Descendants {
|
|
fn observe(&mut self, parent: u32) {
|
|
let mut pending = vec![parent];
|
|
pending.extend(
|
|
self.0.iter().filter_map(|(pid, started)| {
|
|
(process_start(*pid) == Some(*started)).then_some(*pid)
|
|
}),
|
|
);
|
|
while let Some(parent) = pending.pop() {
|
|
let children = fs::read_to_string(format!("/proc/{parent}/task/{parent}/children"))
|
|
.unwrap_or_default();
|
|
for child in children
|
|
.split_whitespace()
|
|
.map(|pid| pid.parse::<u32>().unwrap())
|
|
{
|
|
if let Some(started) = process_start(child)
|
|
&& self.0.insert(child, started).is_none()
|
|
{
|
|
pending.push(child);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
impl Drop for Descendants {
|
|
fn drop(&mut self) {
|
|
for (pid, started) in &self.0 {
|
|
if process_start(*pid) == Some(*started)
|
|
&& let Some(pid) = rustix::process::Pid::from_raw(*pid as i32)
|
|
{
|
|
let _ = rustix::process::kill_process(pid, rustix::process::Signal::KILL);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
fn process_start(pid: u32) -> Option<u64> {
|
|
let stat = fs::read_to_string(format!("/proc/{pid}/stat")).ok()?;
|
|
stat.rsplit_once(") ")?
|
|
.1
|
|
.split_whitespace()
|
|
.nth(19)?
|
|
.parse()
|
|
.ok()
|
|
}
|
|
|
|
#[test]
|
|
#[ignore = "requires explicit DEVER_FFI_CC, DEVER_PROCESS_RUNTIME inputs"]
|
|
fn real_c_abi_packages_compiles_and_checks_the_generated_protocol() {
|
|
let fixture = Fixture::prepare();
|
|
for value in [7, -1, -2, -3] {
|
|
let input = [
|
|
frame(fixture.hello.clone()),
|
|
frame(json!({"kind":"call", "id":1, "operation":"call", "payload":{"value":value}})),
|
|
frame(json!({"kind":"shutdown"})),
|
|
]
|
|
.concat();
|
|
let (success, stdout, stderr) = fixture.capture(
|
|
Command::new(&fixture.worker).current_dir(fixture.worker.parent().unwrap()),
|
|
&input,
|
|
Duration::from_secs(5),
|
|
);
|
|
let mut frames = stdout.as_slice();
|
|
assert_eq!(response(&mut frames).unwrap()["kind"], "ready", "{stderr}");
|
|
if value >= -1 {
|
|
assert!(success, "{stderr}");
|
|
let reply = response(&mut frames).unwrap();
|
|
if value == 7 {
|
|
assert_eq!(reply["payload"]["output"], 21);
|
|
} else {
|
|
assert_eq!(reply["error"], fixture.rejected);
|
|
}
|
|
assert_eq!(response(&mut frames).unwrap()["kind"], "shutdown");
|
|
assert!(stderr.contains("native-closed"));
|
|
} else {
|
|
assert!(!success);
|
|
assert!(frames.is_empty());
|
|
assert_eq!(stderr.contains("native-closed"), value == -2);
|
|
}
|
|
assert!(frames.is_empty());
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
#[ignore = "requires explicit DEVER_FFI_CC and DEVER_PROCESS_RUNTIME inputs"]
|
|
fn real_c_abi_direct_worker_errors_deadlines_and_cancellation() {
|
|
let fixture = Fixture::prepare();
|
|
fs::remove_dir_all(fixture.worker.parent().unwrap()).unwrap();
|
|
for (entry, input, expected) in [
|
|
("call", "{\"value\":7}", Some(21)),
|
|
("call", "{\"value\":-1}", None),
|
|
("call", "{\"value\":-2}", None),
|
|
("call", "{\"value\":-3}", None),
|
|
("call", "{\"value\":-4}", None),
|
|
("cancel", "{}", None),
|
|
("deadline", "{}", None),
|
|
] {
|
|
let (success, stdout, stderr) = fixture.capture(
|
|
Command::new(&fixture.executable)
|
|
.current_dir(fixture.directory.path())
|
|
.args([&format!("media.native.{entry}"), input]),
|
|
&[],
|
|
Duration::from_secs(45),
|
|
);
|
|
if let Some(expected) = expected {
|
|
assert!(success, "{entry} {input}: {stderr}");
|
|
let reply: serde_json::Value = serde_json::from_slice(&stdout).unwrap();
|
|
assert_eq!(reply["code"], 0, "{reply}; {stderr}");
|
|
assert_eq!(reply["data"], expected, "{reply}; {stderr}");
|
|
} else {
|
|
assert!(!success, "{entry} {input}: {stderr}");
|
|
if input.contains("-1") {
|
|
assert!(stderr.contains("Rejected"), "{stderr}");
|
|
}
|
|
if entry != "call" || input.contains("-4") {
|
|
assert!(stderr.contains("timed out"), "{stderr}");
|
|
}
|
|
}
|
|
}
|
|
}
|