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

Replace futures::executor::block_on with tokio::block_in_place

Browse Source
This commit is contained in:
revue_2_presse
2025-02-25 20:19:14 +01:00
parent 4fc4152eac
commit 5d468d3e19
4 changed files with 110 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files

109
src/machine/system_calls.rs
View File

@@ -91,6 +91,8 @@ use roxmltree;
use futures::future;
#[cfg(feature = "http")]
use reqwest::Url;
use tokio::runtime::Handle;
use tokio::task;
#[cfg(feature = "http")]
use warp::hyper::header::{HeaderName, HeaderValue};
#[cfg(feature = "http")]
@@ -4378,65 +4380,70 @@ impl Machine {
}
// do it!
match futures::executor::block_on(req.send()) {
Ok(resp) => {
// status code
let status = resp.status().as_u16();
self.machine_st
.unify_fixnum(Fixnum::build_with(status as i64), address_status);
// headers
let headers: Vec<HeapCellValue> = resp
.headers()
.iter()
.map(|(header_name, header_value)| {
let h = self.machine_st.heap.len();
task::block_in_place(move || {
match Handle::current().block_on(req.send()) {
Ok(resp) => {
// status code
let status = resp.status().as_u16();
self.machine_st
.unify_fixnum(Fixnum::build_with(status as i64), address_status);
// headers
let headers: Vec<HeapCellValue> = resp
.headers()
.iter()
.map(|(header_name, header_value)| {
let h = self.machine_st.heap.len();
let header_term = functor!(
AtomTable::build_with(
&self.machine_st.atom_tbl,
header_name.as_str()
),
[cell(string_as_cstr_cell!(AtomTable::build_with(
&self.machine_st.atom_tbl,
header_value.to_str().unwrap()
)))]
);
let header_term = functor!(
AtomTable::build_with(
&self.machine_st.atom_tbl,
header_name.as_str()
),
[cell(string_as_cstr_cell!(AtomTable::build_with(
&self.machine_st.atom_tbl,
header_value.to_str().unwrap()
)))]
);
self.machine_st.heap.extend(header_term);
str_loc_as_cell!(h)
})
.collect();
self.machine_st.heap.extend(header_term);
str_loc_as_cell!(h)
})
.collect();
let headers_list =
iter_to_heap_list(&mut self.machine_st.heap, headers.into_iter());
unify!(
self.machine_st,
heap_loc_as_cell!(headers_list),
self.machine_st.registers[6]
);
// body
let reader = futures::executor::block_on(resp.bytes()).unwrap().reader();
let headers_list =
iter_to_heap_list(&mut self.machine_st.heap, headers.into_iter());
let mut stream = Stream::from_http_stream(
AtomTable::build_with(&self.machine_st.atom_tbl, &address_string),
reader,
&mut self.machine_st.arena,
);
*stream.options_mut() = StreamOptions::default();
unify!(
self.machine_st,
heap_loc_as_cell!(headers_list),
self.machine_st.registers[6]
);
self.indices
.add_stream(stream, atom!("http_open"), 3)
.map_err(|stub_gen| stub_gen(&mut self.machine_st))?;
// body
let reader = futures::executor::block_on(resp.bytes()).unwrap().reader();
let stream = stream_as_cell!(stream);
let mut stream = Stream::from_http_stream(
AtomTable::build_with(&self.machine_st.atom_tbl, &address_string),
reader,
&mut self.machine_st.arena,
);
*stream.options_mut() = StreamOptions::default();
let stream_addr = self.deref_register(2);
self.machine_st.bind(stream_addr.as_var().unwrap(), stream);
self.indices
.add_stream(stream, atom!("http_open"), 3)
.map_err(|stub_gen| stub_gen(&mut self.machine_st))
.unwrap();
let stream = stream_as_cell!(stream);
let stream_addr = self.deref_register(2);
self.machine_st.bind(stream_addr.as_var().unwrap(), stream);
}
Err(_) => {
self.machine_st.fail = true;
}
}
Err(_) => {
self.machine_st.fail = true;
}
}
});
} else {
let err = self
.machine_st

23
tests-pl/issue-http_open-hanging.pl Normal file
View File

@@ -0,0 +1,23 @@
:- module(http_open_hanging, [submit_request/0]).
:- use_module(library(charsio)).
:- use_module(library(http/http_open)).
send_request :-
Options = [
method('get'),
status_code(StatusCode),
request_headers([]),
headers(_)
],
http_open("https://scryer.pl", _Stream, Options),
write_term('received response with status code':StatusCode, []), nl.
main :-
send_request,
send_request,
send_request,
send_request,
send_request.
:- initialization(main).

16
tests/scryer/helper.rs
View File

@@ -1,3 +1,5 @@
use scryer_prolog::MachineBuilder;
pub(crate) trait Expectable {
#[track_caller]
fn assert_eq(self, other: &[u8]);
@@ -31,3 +33,17 @@ pub(crate) fn load_module_test<T: Expectable>(file: &str, expected: T) {
let mut wam = MachineBuilder::default().build();
expected.assert_eq(wam.test_load_file(file).as_slice());
}
/// Same as `load_module_test` with tokio runtime
#[cfg(not(target_arch = "wasm32"))]
pub(crate) fn load_module_test_with_tokio_runtime<T: Expectable>(file: &str, expected: T) {
let runtime = tokio::runtime::Builder::new_multi_thread()
.enable_all()
.build()
.unwrap();
runtime.block_on(async move {
let mut wam = MachineBuilder::default().build();
expected.assert_eq(wam.test_load_file(file).as_slice())
});
}

13
tests/scryer/issues.rs
View File

@@ -1,4 +1,6 @@
use crate::helper::load_module_test;
#[cfg(not(target_arch = "wasm32"))]
use crate::helper::load_module_test_with_tokio_runtime;
use serial_test::serial;
// issue #831
@@ -43,3 +45,14 @@ fn load_context_unreachable() {
fn issue2725_dcg_without_module() {
load_module_test("tests-pl/issue2725.pl", "");
}
#[test]
#[cfg(feature = "http")]
#[cfg(not(target_arch = "wasm32"))]
#[cfg_attr(miri, ignore = "it takes too long to run")]
fn http_open_hanging() {
load_module_test_with_tokio_runtime(
"tests-pl/issue-http_open-hanging.pl",
"received response with status code:200\nreceived response with status code:200\nreceived response with status code:200\nreceived response with status code:200\nreceived response with status code:200\n"
);
}