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

allow to stop and start http server at will with SIGINT, stop the server on uncaught throws, display timing info for each request, use setup_call_cleanup in more places to avoid potential leaks, when handler for a path doesn't have a matching case, throw a detailed error instead of silently continuing

Browse Source
This commit is contained in:
Danil Platonov
2026-05-30 15:47:50 -07:00
parent 8b6d68a4cd
commit f8f7b3b7e2
6 changed files with 122 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
build/instructions_template.rs
View File

@@ -599,6 +599,8 @@ enum SystemClauseType {
HttpOpen,
#[strum_discriminants(strum(props(Arity = "5", Name = "$http_listen")))]
HttpListen,
#[strum_discriminants(strum(props(Arity = "1", Name = "$http_listen_stop")))]
HttpListenStop,
#[strum_discriminants(strum(props(Arity = "7", Name = "$http_accept")))]
HttpAccept,
#[strum_discriminants(strum(props(Arity = "4", Name = "$http_answer")))]

1
src/http.rs
View File

@@ -5,6 +5,7 @@ use warp::http;
pub struct HttpListener {
pub incoming: std::sync::mpsc::Receiver<HttpRequest>,
pub warp_shutdown: tokio::sync::mpsc::Sender<()>,
}
pub struct HttpRequest {

2
src/instructions.rs
View File

@@ -803,6 +803,7 @@ impl Instruction {
| &Instruction::CallDeterministicLengthRundown
| &Instruction::CallHttpOpen
| &Instruction::CallHttpListen
| &Instruction::CallHttpListenStop
| &Instruction::CallHttpAccept
| &Instruction::CallHttpAnswer
| &Instruction::CallLoadForeignLib
@@ -1062,6 +1063,7 @@ impl Instruction {
| &Instruction::ExecuteDeterministicLengthRundown
| &Instruction::ExecuteHttpOpen
| &Instruction::ExecuteHttpListen
| &Instruction::ExecuteHttpListenStop
| &Instruction::ExecuteHttpAccept
| &Instruction::ExecuteHttpAnswer
| &Instruction::ExecuteLoadForeignLib

70
src/lib/http/http_server.pl
View File

@@ -112,12 +112,29 @@ module_qualification(M, H0, H) :-
H0 =.. [Method, Path, Goal],
H =.. [Method, Path, M:Goal].
http_listen__(Addr, HttpListener, TLSKey, TLSCert, ContentLengthLimit) :-
'$http_listen'(Addr, HttpListener, TLSKey, TLSCert, ContentLengthLimit).
http_listen_stop_(HttpListener) :-
'$http_listen_stop'(HttpListener).
http_accept_(HttpListener, RequestMethod, RequestPath, RequestHeaders, RequestQuery, RequestStream, ResponseHandle) :-
'$http_accept'(HttpListener, RequestMethod, RequestPath, RequestHeaders, RequestQuery, RequestStream, ResponseHandle).
http_answer_(ResponseHandle, Code, Headers, ResponseStream) :-
'$http_answer'(ResponseHandle, Code, Headers, ResponseStream).
http_listen_(Port, Handlers, Options) :-
parse_options(Options, TLSKey, TLSCert, ContentLengthLimit),
phrase(format_("0.0.0.0:~d", [Port]), Addr),
'$http_listen'(Addr, HttpListener, TLSKey, TLSCert, ContentLengthLimit),!,
format("Listening at ~s\n", [Addr]),
http_loop(HttpListener, Handlers).
setup_call_cleanup(
(
http_listen__(Addr, HttpListener, TLSKey, TLSCert, ContentLengthLimit),
format("Listening at http://~s\n", [Addr])
),
http_loop(HttpListener, Handlers),
http_listen_stop_(HttpListener)
).
parse_options(Options, TLSKey, TLSCert, ContentLengthLimit) :-
member_option_default(tls_key, Options, "", TLSKey),
@@ -132,12 +149,12 @@ member_option_default(Key, List, Default, Default) :-
X =.. [Key, _],
\+ member(X, List).
http_loop(HttpListener, Handlers) :-
'$http_accept'(HttpListener, RequestMethod, RequestPath, RequestHeaders, RequestQuery, RequestStream, ResponseHandle),
time((
http_accept_(HttpListener, RequestMethod, RequestPath, RequestHeaders, RequestQuery, RequestStream, ResponseHandle),
current_time(Time),
phrase(format_time("%Y-%m-%d (%H:%M:%S)", Time), TimeString),
format("~s ~w ~s\n", [TimeString, RequestMethod, RequestPath]),
format("~s ~w ~s", [TimeString, RequestMethod, RequestPath]),
maplist(map_header_kv, RequestHeaders, RequestHeadersKV),
phrase(parse_queries(RequestQueries), RequestQuery),
(
@@ -145,23 +162,42 @@ http_loop(HttpListener, Handlers) :-
(
HttpRequest = http_request(RequestHeadersKV, stream(RequestStream), RequestQueries),
HttpResponse = http_response(_, _, _),
catch(
(call(Handler, HttpRequest, HttpResponse) ->
send_response(ResponseHandle, HttpResponse)
; (
'$http_answer'(ResponseHandle, 500, [], ResponseStream),
call_cleanup(format(ResponseStream, "Internal Server Error", []), close(ResponseStream)))
)
)
; (
'$http_answer'(ResponseHandle, 404, [], ResponseStream),
call_cleanup(format(ResponseStream, "Not Found", []), close(ResponseStream)))
;
setup_call_cleanup(
http_answer_(ResponseHandle, 500, [], ResponseStream),
format(ResponseStream, "Internal Server Error", []),
close(ResponseStream)
),
throw(handler_not_available(Handler, RequestMethod, RequestPath, RequestQuery, RequestHeaders))
),
HandlerError,
(
setup_call_cleanup(
http_answer_(ResponseHandle, 500, [], ResponseStream),
format(ResponseStream, "Internal Server Error", []),
close(ResponseStream)
),
throw(HandlerError)
)
)
)
;
setup_call_cleanup(
http_answer_(ResponseHandle, 404, [], ResponseStream),
format(ResponseStream, "Not Found", []),
close(ResponseStream)
)
)
)),
http_loop(HttpListener, Handlers).
send_response(ResponseHandle, http_response(StatusCode0, text(ResponseText), ResponseHeaders0)) :-
default(StatusCode0, 200, StatusCode),
maplist(map_header_kv_2, ResponseHeaders, ResponseHeaders0),
'$http_answer'(ResponseHandle, StatusCode, ResponseHeaders, ResponseStream0),
http_answer_(ResponseHandle, StatusCode, ResponseHeaders, ResponseStream0),
open(stream(ResponseStream0), write, ResponseStream, [type(text)]),
catch(
call_cleanup(format(ResponseStream, "~s", [ResponseText]),close(ResponseStream)),
@@ -172,7 +208,7 @@ send_response(ResponseHandle, http_response(StatusCode0, text(ResponseText), Res
send_response(ResponseHandle, http_response(StatusCode0, bytes(ResponseBytes), ResponseHeaders0)) :-
default(StatusCode0, 200, StatusCode),
maplist(map_header_kv_2, ResponseHeaders, ResponseHeaders0),
'$http_answer'(ResponseHandle, StatusCode, ResponseHeaders, ResponseStream),
http_answer_(ResponseHandle, StatusCode, ResponseHeaders, ResponseStream),
catch(
call_cleanup(format(ResponseStream, "~s", [ResponseBytes]),close(ResponseStream)),
error(existence_error(stream, _), _),
@@ -182,7 +218,7 @@ send_response(ResponseHandle, http_response(StatusCode0, bytes(ResponseBytes), R
send_response(ResponseHandle, http_response(StatusCode0, file(Filename), ResponseHeaders0)) :-
default(StatusCode0, 200, StatusCode),
maplist(map_header_kv_2, ResponseHeaders, ResponseHeaders0),
'$http_answer'(ResponseHandle, StatusCode, ResponseHeaders, ResponseStream),
http_answer_(ResponseHandle, StatusCode, ResponseHeaders, ResponseStream),
catch(
call_cleanup(
setup_call_cleanup(

10
src/machine/dispatch.rs
View File

@@ -4690,6 +4690,16 @@ impl Machine {
try_or_throw!(self.machine_st, self.http_listen(), continue);
step_or_fail!(self.machine_st, self.machine_st.p = self.machine_st.cp);
}
&Instruction::CallHttpListenStop => {
#[cfg(feature = "http")]
try_or_throw!(self.machine_st, self.http_listen_stop(), continue);
step_or_fail!(self.machine_st, self.machine_st.p += 1);
}
&Instruction::ExecuteHttpListenStop => {
#[cfg(feature = "http")]
try_or_throw!(self.machine_st, self.http_listen_stop(), continue);
step_or_fail!(self.machine_st, self.machine_st.p = self.machine_st.cp);
}
&Instruction::CallHttpAccept => {
#[cfg(feature = "http")]
try_or_throw!(self.machine_st, self.http_accept(), continue);

44
src/machine/system_calls.rs
View File

@@ -4593,6 +4593,10 @@ impl Machine {
let (tx, rx) = std::sync::mpsc::sync_channel(1024);
// warp shutdown channel
let (warp_shutdown_tx, mut warp_shutdown_rx) = tokio::sync::mpsc::channel(1);
let warp_shutdown_tx_timeout = warp_shutdown_tx.clone();
let runtime = tokio::runtime::Handle::current();
let _guard = runtime.enter();
@@ -4657,13 +4661,23 @@ impl Machine {
runtime.spawn(async move {
match ssl_server {
Some((key, cert)) => {
warp::serve(serve).tls().key(key).cert(cert).run(addr).await
let (_addr, server) = warp::serve(serve).tls().key(key).cert(cert).bind_with_graceful_shutdown(addr, async move {
warp_shutdown_rx.recv().await;
});
tokio::task::spawn(server);
}
None => {
let (_addr, server) = warp::serve(serve).bind_with_graceful_shutdown(addr, async move {
warp_shutdown_rx.recv().await;
});
tokio::task::spawn(server);
}
None => warp::serve(serve).run(addr).await,
}
});
let http_listener = HttpListener { incoming: rx };
let http_listener = HttpListener { incoming: rx, warp_shutdown: warp_shutdown_tx };
let http_listener: TypedArenaPtr<HttpListener> =
arena_alloc!(http_listener, &mut self.machine_st.arena);
@@ -4676,6 +4690,30 @@ impl Machine {
Ok(())
}
#[cfg(feature = "http")]
#[inline(always)]
pub(crate) fn http_listen_stop(&mut self) -> CallResult {
let culprit = self.deref_register(1);
read_heap_cell!(culprit,
(HeapCellValueTag::Cons, cons_ptr) => {
match_untyped_arena_ptr!(cons_ptr,
(ArenaHeaderTag::HttpListener, http_listener) => {
let _ = futures::executor::block_on(http_listener.warp_shutdown.send(()));
}
_ => {
unreachable!();
}
);
}
_ => {
unreachable!();
}
);
Ok(())
}
#[cfg(feature = "http")]
#[inline(always)]
pub(crate) fn http_accept(&mut self) -> CallResult {