fixed a segfault in Orchid
also - fixed a panic in Orchid because the logging channel was discarded outside a logger - added pre-buffering to checklings - polished log view in checkers - started on tree view in checkers -
This commit is contained in:
lbfalvy committed
1 parent
bb31e0daf1
commit
82992ead16
19 files changed
+831
-534
No files matched your search
Generated
+8
-104
@@ -31,7 +31,7 @@ dependencies = [
|
||||
"futures-util",
|
||||
"log 0.4.33",
|
||||
"once_cell",
|
||||
"smallvec 1.15.2",
|
||||
"smallvec",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -66,7 +66,7 @@ dependencies = [
|
||||
"pin-project-lite",
|
||||
"rand 0.10.2",
|
||||
"sha1 0.11.0",
|
||||
"smallvec 1.15.2",
|
||||
"smallvec",
|
||||
"tokio",
|
||||
"tokio-util",
|
||||
"tracing",
|
||||
@@ -181,7 +181,7 @@ dependencies = [
|
||||
"serde",
|
||||
"serde_json",
|
||||
"serde_urlencoded",
|
||||
"smallvec 1.15.2",
|
||||
"smallvec",
|
||||
"socket2 0.6.4",
|
||||
"time 0.3.53",
|
||||
"tracing",
|
||||
@@ -818,7 +818,6 @@ dependencies = [
|
||||
"checklings",
|
||||
"dioxus",
|
||||
"dioxus-stores",
|
||||
"facet-xdr",
|
||||
"futures",
|
||||
"gloo-net 0.7.0",
|
||||
"itertools",
|
||||
@@ -956,12 +955,6 @@ dependencies = [
|
||||
"memchr",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "const-fnv1a-hash"
|
||||
version = "1.1.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "32b13ea120a812beba79e34316b3942a857c86ec1593cb34f27bb28272ce2cca"
|
||||
|
||||
[[package]]
|
||||
name = "const-oid"
|
||||
version = "0.10.2"
|
||||
@@ -1927,83 +1920,6 @@ dependencies = [
|
||||
"num-traits",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "facet-core"
|
||||
version = "0.46.5"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "c33cf9d80307e9a29f93e5a65ddb65b227b5fa082d649c65d4083345f5d398d0"
|
||||
dependencies = [
|
||||
"autocfg 1.5.1",
|
||||
"const-fnv1a-hash",
|
||||
"impls",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "facet-dessert"
|
||||
version = "0.46.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "d4505ec8ef086d776fd99b86fb17e3d1c711188d27aa72845e64a30146226e3b"
|
||||
dependencies = [
|
||||
"facet-core",
|
||||
"facet-reflect",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "facet-format"
|
||||
version = "0.47.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "f2353807588cc4057a5e31bf459e7304ea84729ad77080e267928b1e09d36235"
|
||||
dependencies = [
|
||||
"facet-core",
|
||||
"facet-dessert",
|
||||
"facet-path",
|
||||
"facet-reflect",
|
||||
"facet-solver",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "facet-path"
|
||||
version = "0.46.5"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "20670a58be24eb604b76ed3a9ca719e5cc5e11b0885dc3847ebb899aa95bf1bc"
|
||||
dependencies = [
|
||||
"facet-core",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "facet-reflect"
|
||||
version = "0.46.5"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "399876f2bded4a572385dc414336aad0002ce4b4154cd0d1cfe9d1881243c400"
|
||||
dependencies = [
|
||||
"facet-core",
|
||||
"facet-path",
|
||||
"hashbrown 0.17.1",
|
||||
"smallvec 2.0.0-alpha.12",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "facet-solver"
|
||||
version = "0.46.5"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "86cab172249e0d68bf57f804e97b18e375e3fc4a46300f2a4ea6e31e6c1fa4df"
|
||||
dependencies = [
|
||||
"facet-core",
|
||||
"facet-reflect",
|
||||
"strsim",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "facet-xdr"
|
||||
version = "0.46.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "ae8e9e05df7b635664f258863ce2115ab95c9c64732628f6bfd3252cdf65c48e"
|
||||
dependencies = [
|
||||
"facet-core",
|
||||
"facet-format",
|
||||
"facet-reflect",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "fake-simd"
|
||||
version = "0.1.2"
|
||||
@@ -2551,7 +2467,7 @@ dependencies = [
|
||||
"httparse",
|
||||
"itoa",
|
||||
"pin-project-lite",
|
||||
"smallvec 1.15.2",
|
||||
"smallvec",
|
||||
"tokio",
|
||||
"want",
|
||||
]
|
||||
@@ -2658,7 +2574,7 @@ dependencies = [
|
||||
"icu_normalizer_data",
|
||||
"icu_properties",
|
||||
"icu_provider",
|
||||
"smallvec 1.15.2",
|
||||
"smallvec",
|
||||
"zerovec",
|
||||
]
|
||||
|
||||
@@ -2727,7 +2643,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "3b0875f23caa03898994f6ddc501886a45c7d3d62d04d2d90788d47be1b1e4de"
|
||||
dependencies = [
|
||||
"idna_adapter",
|
||||
"smallvec 1.15.2",
|
||||
"smallvec",
|
||||
"utf8_iter",
|
||||
]
|
||||
|
||||
@@ -2747,12 +2663,6 @@ version = "0.3.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "35a84fd5aa25fae5c0f4a33d9cac2ca017fc622cbd089be2229993514990f870"
|
||||
|
||||
[[package]]
|
||||
name = "impls"
|
||||
version = "1.0.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "7a46645bbd70538861a90d0f26c31537cdf1e44aae99a794fb75a664b70951bc"
|
||||
|
||||
[[package]]
|
||||
name = "include_dir"
|
||||
version = "0.7.4"
|
||||
@@ -3780,7 +3690,7 @@ dependencies = [
|
||||
"cfg-if",
|
||||
"libc",
|
||||
"redox_syscall",
|
||||
"smallvec 1.15.2",
|
||||
"smallvec",
|
||||
"windows-link",
|
||||
]
|
||||
|
||||
@@ -4954,12 +4864,6 @@ version = "1.15.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "8ed6a63f02c8539c91a8685a86f4099661ba3da017932f6ebbea6de3f0fa7c90"
|
||||
|
||||
[[package]]
|
||||
name = "smallvec"
|
||||
version = "2.0.0-alpha.12"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "ef784004ca8777809dcdad6ac37629f0a97caee4c685fcea805278d81dd8b857"
|
||||
|
||||
[[package]]
|
||||
name = "socket2"
|
||||
version = "0.5.10"
|
||||
@@ -5495,7 +5399,7 @@ dependencies = [
|
||||
"once_cell",
|
||||
"regex-automata",
|
||||
"sharded-slab",
|
||||
"smallvec 1.15.2",
|
||||
"smallvec",
|
||||
"thread_local",
|
||||
"tracing",
|
||||
"tracing-core",
|
||||
|
||||
@@ -129,7 +129,7 @@ async fn read(
|
||||
let payload = (topic.backlog.lock().unwrap()[range].iter())
|
||||
.flat_map(|msg| msg.data.iter().copied())
|
||||
.collect::<Vec<_>>();
|
||||
for chunk in payload.chunks(8192) {
|
||||
for chunk in payload.chunks(2 << 15) {
|
||||
if let Err(e) = session.binary(chunk.to_vec()).await {
|
||||
eprintln!("Error on receiving binary: {e}");
|
||||
break 'sock_loop None;
|
||||
|
||||
@@ -7,7 +7,6 @@ edition = "2024"
|
||||
checklings = { version = "0.1.0", path = "../checklings" }
|
||||
dioxus = { version = "0.7.10", features = ["web"] }
|
||||
dioxus-stores = "0.7.10"
|
||||
facet-xdr = "0.46.1"
|
||||
futures = "0.3.32"
|
||||
gloo-net = "0.7.0"
|
||||
itertools = "0.15.0"
|
||||
|
||||
@@ -1,13 +1,38 @@
|
||||
body {
|
||||
background-color: #0f1116;
|
||||
background-color: #282828;
|
||||
color: #ffffff;
|
||||
font-family: 'Segoe UI', Tahoma, Geneva, Verdana, sans-serif;
|
||||
margin: 0;
|
||||
padding: 0;
|
||||
/* Log view */
|
||||
--bg-hovered: #666;
|
||||
--bg-related: #444;
|
||||
--gutter-bg: #555;
|
||||
--gutter-bg-emphasis: #777;
|
||||
--gutter-markings: #aaa;
|
||||
/* Inputs and buttons */
|
||||
--bg-btn: #444;
|
||||
--bg-btn-emphasis: #55a;
|
||||
--input-under: #444;
|
||||
--input-under: #888;
|
||||
--input-under-focus: #55a;
|
||||
|
||||
/* client-colors */
|
||||
--client-0: oklch(0.87 0.180 164);
|
||||
--client-1: oklch(0.86 0.241 147);
|
||||
--client-2: oklch(0.81 0.161 322);
|
||||
--client-3: oklch(0.81 0.158 74);
|
||||
--client-4: oklch(0.81 0.140 217);
|
||||
--client-5: oklch(0.86 0.091 139);
|
||||
--client-6: oklch(0.81 0.105 14);
|
||||
--client-7: oklch(0.81 0.146 122);
|
||||
--client-8: oklch(0.81 0.153 174);
|
||||
--client-9: oklch(0.82 0.091 284);
|
||||
--client-a: oklch(0.87 0.139 101);
|
||||
--client-b: oklch(0.87 0.097 331);
|
||||
--client-c: oklch(0.86 0.104 191);
|
||||
--client-d: oklch(0.82 0.180 144);
|
||||
--client-e: oklch(0.82 0.091 93);
|
||||
--client-f: oklch(0.87 0.220 128);
|
||||
}
|
||||
|
||||
input {
|
||||
@@ -30,4 +55,33 @@ button {
|
||||
border: 0;
|
||||
color: white;
|
||||
background: var(--bg-btn);
|
||||
}
|
||||
|
||||
.message-list {
|
||||
display: grid;
|
||||
grid-template-columns: 2em 8ch auto;
|
||||
grid-template-rows: repeat(auto, auto);
|
||||
|
||||
.event-gutter-item {
|
||||
grid-column: 1;
|
||||
border-bottom: 1px solid var(--gutter-markings);
|
||||
overflow: hidden;
|
||||
|
||||
.event-gutter-label {
|
||||
display: block;
|
||||
text-align: center;
|
||||
font-size: calc(60%);
|
||||
top: 5px;
|
||||
bottom: 5px;
|
||||
position: sticky;
|
||||
}
|
||||
}
|
||||
|
||||
.message-id {
|
||||
grid-column: 2,
|
||||
}
|
||||
|
||||
.message-details {
|
||||
grid-column: 3,
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,84 @@
|
||||
use std::pin::pin;
|
||||
use std::time::Duration;
|
||||
|
||||
use dioxus::prelude::*;
|
||||
use futures::StreamExt;
|
||||
|
||||
use crate::state::Message;
|
||||
use crate::web_bindings::{get_local_storage, set_local_storage, sleep};
|
||||
use crate::ws_reader::{WsEvent, with_ws};
|
||||
|
||||
#[component]
|
||||
pub fn ConnectionPanel(onreset: Callback<(), ()>, onmessage: Callback<Message, ()>) -> Element {
|
||||
let mut remote = use_signal(|| {
|
||||
get_local_storage("remote").unwrap_or_else(|| "wss://localhost:8170".to_string())
|
||||
});
|
||||
use_effect(move || set_local_storage("remote", &remote()));
|
||||
|
||||
let mut status = use_signal(String::new);
|
||||
use_resource(move || async move {
|
||||
if !status.read().is_empty() {
|
||||
sleep(Duration::from_secs(3)).await;
|
||||
status.set(String::new());
|
||||
}
|
||||
});
|
||||
|
||||
let mut topic = use_signal(|| get_local_storage("topic").unwrap_or_else(|| "main".to_string()));
|
||||
use_effect(move || set_local_storage("topic", &topic()));
|
||||
|
||||
let connect = use_coroutine(move |mut cx: UnboundedReceiver<()>| async move {
|
||||
loop {
|
||||
futures::future::join(
|
||||
pin!(async { cx.next().await }),
|
||||
pin!(with_ws(remote(), topic(), move |action| match action {
|
||||
WsEvent::Status(text) => status.set(text),
|
||||
WsEvent::Set(vec) => {
|
||||
onreset(());
|
||||
for item in vec {
|
||||
onmessage(item);
|
||||
}
|
||||
},
|
||||
WsEvent::Push(item) => onmessage(item),
|
||||
})),
|
||||
)
|
||||
.await;
|
||||
}
|
||||
});
|
||||
|
||||
use_effect(move || {
|
||||
let _ = (remote(), topic());
|
||||
connect.send(());
|
||||
});
|
||||
|
||||
rsx! {
|
||||
div {
|
||||
display: "grid",
|
||||
grid_template: "
|
||||
'remote topic '
|
||||
'status status'
|
||||
",
|
||||
input {
|
||||
grid_area: "remote",
|
||||
placeholder: "remote",
|
||||
min_width: "200px",
|
||||
value: remote(),
|
||||
oninput: move |event| {
|
||||
remote.set(event.value());
|
||||
}
|
||||
}
|
||||
input {
|
||||
grid_area: "topic",
|
||||
placeholder: "topic",
|
||||
value: topic(),
|
||||
oninput: move |event| {
|
||||
topic.set(event.value());
|
||||
}
|
||||
}
|
||||
div {
|
||||
grid_area: "status",
|
||||
display: if status.is_empty() { "none" } else { "block" },
|
||||
"{status}"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
+90
-47
@@ -1,36 +1,10 @@
|
||||
use std::rc::Rc;
|
||||
|
||||
use checklings::{TraceInfo, TracedEvent};
|
||||
use dioxus::prelude::*;
|
||||
use itertools::Itertools;
|
||||
|
||||
use crate::state::{LogItem, Message};
|
||||
use crate::virtual_scroll::{PinEnd, VirtualScroll};
|
||||
|
||||
#[derive(Clone, Debug, Hash, PartialEq, Eq)]
|
||||
pub enum LogItem {
|
||||
Event(TracedEvent),
|
||||
Message(Rc<String>),
|
||||
}
|
||||
impl LogItem {
|
||||
pub fn msg(str: impl AsRef<str>) -> Self { Self::Message(Rc::new(str.as_ref().to_string())) }
|
||||
}
|
||||
|
||||
#[component]
|
||||
pub fn LogLine(id: u64, line: LogItem) -> Element {
|
||||
rsx! {
|
||||
div {
|
||||
span {
|
||||
display: "inline-block",
|
||||
width: "5ex",
|
||||
"{id}",
|
||||
}
|
||||
{match line {
|
||||
LogItem::Message(msg) => rsx! { span { margin_left: "10px", pre { display: "inline-block", margin: 0, "{msg}" } } },
|
||||
LogItem::Event(event) => rsx! { EventLine { event: event.clone() } },
|
||||
}}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn mk_path(info: &TraceInfo) -> String {
|
||||
match &info.parent {
|
||||
None => info.label.clone(),
|
||||
@@ -39,8 +13,8 @@ fn mk_path(info: &TraceInfo) -> String {
|
||||
}
|
||||
|
||||
#[component]
|
||||
fn EventLine(event: TracedEvent) -> Element {
|
||||
let TracedEvent { client, info, kind } = event;
|
||||
fn EventLine(event: Message) -> Element {
|
||||
let Message { client, traced: TracedEvent { info, kind } } = event;
|
||||
let path = info.as_deref().map(mk_path);
|
||||
rsx! {
|
||||
span {
|
||||
@@ -67,22 +41,41 @@ fn EventLine(event: TracedEvent) -> Element {
|
||||
}
|
||||
|
||||
#[component]
|
||||
pub fn LogView(rows: ReadSignal<Vec<LogItem>>) -> Element {
|
||||
pub fn LogView(
|
||||
rows: ReadSignal<Vec<LogItem>>,
|
||||
selection: Option<usize>,
|
||||
onselect: Callback<usize, ()>,
|
||||
) -> Element {
|
||||
let bound = move |idx: i64| 0.max(idx.min(rows.len() as i64 - 1));
|
||||
|
||||
let mut first_index = use_signal(|| 0.max(rows.len() as i64 - 2));
|
||||
let mut last_index = use_signal(|| 0.max(rows.len() as i64 - 1));
|
||||
|
||||
let mut follow = use_signal(|| true);
|
||||
let mut hovered_row = use_signal(|| None::<usize>);
|
||||
|
||||
let rendered_rows = if rows.is_empty() {
|
||||
Vec::new()
|
||||
} else {
|
||||
if follow() && last_index() + 1 < rows.len() as i64 {
|
||||
last_index.set(rows.len() as i64 - 1);
|
||||
}
|
||||
rows.read()[first_index() as usize..=bound(last_index()) as usize].to_vec()
|
||||
};
|
||||
let content_memo = use_memo(move || {
|
||||
let rendered_rows: Vec<LogItem> = if rows.is_empty() {
|
||||
Vec::new()
|
||||
} else {
|
||||
if follow() && last_index() + 1 < rows.len() as i64 {
|
||||
last_index.set(rows.len() as i64 - 1);
|
||||
}
|
||||
rows.read()[first_index() as usize..=bound(last_index()) as usize].to_vec()
|
||||
};
|
||||
let events = (rendered_rows.iter().map(|row| row.ev))
|
||||
.dedup_with_count()
|
||||
.scan(0, |end, (count, id)| {
|
||||
let start = *end;
|
||||
*end += count;
|
||||
Some((start, *end, id))
|
||||
})
|
||||
.collect::<Vec<_>>();
|
||||
(rendered_rows, events)
|
||||
});
|
||||
let (rendered_rows, events) = &*content_memo.read();
|
||||
|
||||
let hovered_event = hovered_row().and_then(|idx| Some(rendered_rows.get(idx)?.ev));
|
||||
|
||||
rsx! {
|
||||
div {
|
||||
@@ -90,16 +83,17 @@ pub fn LogView(rows: ReadSignal<Vec<LogItem>>) -> Element {
|
||||
min_height: "5em",
|
||||
height: "100%",
|
||||
position: "relative",
|
||||
onmouseleave: move |_| hovered_row.set(None),
|
||||
button {
|
||||
position: "absolute",
|
||||
bottom: "10px",
|
||||
right: "10px",
|
||||
onclick: move |_| follow.toggle(),
|
||||
onmouseenter: move |_| hovered_row.set(None),
|
||||
background_color: if follow() { "var(--bg-btn-emphasis)" } else { "var(--bg-btn)" },
|
||||
{if follow() { rsx!{ "Stop following" } } else { rsx!{ "Follow" } }}
|
||||
}
|
||||
VirtualScroll {
|
||||
pin_end: if follow() { Some(PinEnd::End) } else { None },
|
||||
can_extend_start: 0 < first_index(),
|
||||
set_start: move |start: i64| {
|
||||
let new = bound(start);
|
||||
@@ -114,14 +108,63 @@ pub fn LogView(rows: ReadSignal<Vec<LogItem>>) -> Element {
|
||||
last_index.set(new)
|
||||
}
|
||||
},
|
||||
current: first_index()..=last_index(),
|
||||
overscroll: 3..=30,
|
||||
{rendered_rows.iter().enumerate().map(|(i, l)| rsx!{
|
||||
LogLine {
|
||||
id: first_index() as u64 + i as u64,
|
||||
line: l.clone(),
|
||||
}
|
||||
})}
|
||||
pin_end: if follow() { Some(PinEnd::End) } else { None },
|
||||
on_pin_contest: move |()| follow.set(false),
|
||||
current: first_index()..=last_index(),
|
||||
div {
|
||||
class: "message-list",
|
||||
{events.iter().copied().map(|(start, end, id)| rsx!{
|
||||
div {
|
||||
class: "event-gutter-item",
|
||||
grid_row_start: start,
|
||||
grid_row_end: end,
|
||||
background_color: if Some(id) == hovered_event {
|
||||
"var(--gutter-bg-emphasis)"
|
||||
} else {
|
||||
"var(--gutter-bg)"
|
||||
},
|
||||
span {
|
||||
class: "event-gutter-label",
|
||||
"{id}"
|
||||
}
|
||||
}
|
||||
})}
|
||||
{rendered_rows.iter().enumerate().map(|(i, l)| {
|
||||
let LogItem { ev, idx, msg } = l.clone();
|
||||
let background = if Some(l.idx) == selection {
|
||||
"var(--bg-btn-emphasis)"
|
||||
} else if Some(i) == hovered_row() {
|
||||
"var(--bg-hovered)"
|
||||
} else if Some(ev) == hovered_event {
|
||||
"var(--bg-related)"
|
||||
} else {
|
||||
"unset"
|
||||
};
|
||||
let onmouseenter = move |_| hovered_row.set(Some(i));
|
||||
let onclick = move |_| onselect(idx);
|
||||
rsx!{
|
||||
div {
|
||||
class: "message-id",
|
||||
grid_row: i,
|
||||
background,
|
||||
onclick,
|
||||
onmouseenter,
|
||||
"{first_index() as u64 + i as u64}"
|
||||
}
|
||||
div {
|
||||
class: "message-details",
|
||||
grid_row: i,
|
||||
background,
|
||||
onclick,
|
||||
onmouseenter,
|
||||
EventLine {
|
||||
event: msg,
|
||||
}
|
||||
}
|
||||
}
|
||||
})}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+57
-56
@@ -1,60 +1,39 @@
|
||||
use std::pin::pin;
|
||||
use std::ops::Rem;
|
||||
use std::sync::OnceLock;
|
||||
|
||||
use dioxus::prelude::*;
|
||||
use futures::prelude::*;
|
||||
|
||||
use crate::tree::{AggregateExt as _, AggregateStoreExt};
|
||||
use crate::connection_panel::ConnectionPanel;
|
||||
use crate::state::{AggregateExt as _, TargetStateStoreExt};
|
||||
|
||||
mod connection_panel;
|
||||
mod debug;
|
||||
mod log;
|
||||
pub mod tree;
|
||||
pub mod state;
|
||||
mod tree;
|
||||
mod virtual_scroll;
|
||||
mod web_bindings;
|
||||
mod ws_reader;
|
||||
|
||||
use crate::log::LogView;
|
||||
use crate::tree::Aggregate;
|
||||
use crate::web_bindings::{get_local_storage, set_local_storage};
|
||||
use crate::ws_reader::{LogAction, with_ws};
|
||||
use crate::state::TargetState;
|
||||
use crate::tree::TreeView;
|
||||
|
||||
const MAIN_CSS: Asset = asset!("/assets/main.css");
|
||||
|
||||
fn main() { dioxus::launch(App); }
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Hash)]
|
||||
pub enum Selection {
|
||||
None,
|
||||
Message(usize),
|
||||
}
|
||||
|
||||
#[component]
|
||||
fn App() -> Element {
|
||||
let mut remote = use_signal(|| {
|
||||
get_local_storage("remote").unwrap_or_else(|| "wss://localhost:8170".to_string())
|
||||
});
|
||||
use_effect(move || set_local_storage("remote", &remote()));
|
||||
let mut global_selection = use_signal(|| Selection::None);
|
||||
|
||||
let mut topic = use_signal(|| get_local_storage("topic").unwrap_or_else(|| "main".to_string()));
|
||||
use_effect(move || set_local_storage("topic", &topic()));
|
||||
|
||||
let mut state = use_store(Aggregate::default);
|
||||
|
||||
let connect = use_coroutine(move |mut cx: UnboundedReceiver<()>| async move {
|
||||
loop {
|
||||
futures::future::join(
|
||||
pin!(async { cx.next().await }),
|
||||
pin!(with_ws(remote(), topic(), move |action| match action {
|
||||
LogAction::Set(vec) => {
|
||||
state.set(Aggregate::default());
|
||||
for item in vec {
|
||||
state.update(item);
|
||||
}
|
||||
},
|
||||
LogAction::Push(item) => state.update(item),
|
||||
})),
|
||||
)
|
||||
.await;
|
||||
}
|
||||
});
|
||||
|
||||
use_effect(move || {
|
||||
let _ = (remote(), topic());
|
||||
connect.send(());
|
||||
});
|
||||
let mut state = use_store(TargetState::default);
|
||||
|
||||
rsx! {
|
||||
// Global app resources
|
||||
@@ -65,35 +44,57 @@ fn App() -> Element {
|
||||
width: "100vw",
|
||||
display: "grid",
|
||||
grid_template: "
|
||||
\"menu \" auto
|
||||
\"msg_log\" 1fr
|
||||
/ auto
|
||||
'menu menu ' auto
|
||||
'tree msg_log' 1fr
|
||||
/ auto 1fr
|
||||
",
|
||||
div {
|
||||
grid_area: "menu",
|
||||
input {
|
||||
placeholder: "remote",
|
||||
min_width: "200px",
|
||||
value: remote(),
|
||||
oninput: move |event| {
|
||||
remote.set(event.value());
|
||||
}
|
||||
}
|
||||
input {
|
||||
placeholder: "topic",
|
||||
value: topic(),
|
||||
oninput: move |event| {
|
||||
topic.set(event.value());
|
||||
}
|
||||
ConnectionPanel {
|
||||
onreset: move |()| state.set(TargetState::default()),
|
||||
onmessage: move |message| state.update(message),
|
||||
}
|
||||
}
|
||||
div {
|
||||
grid_area: "tree",
|
||||
overflow_y: "hidden",
|
||||
FutureTreePanel { state, selection: global_selection }
|
||||
}
|
||||
div {
|
||||
grid_area: "msg_log",
|
||||
overflow_y: "hidden",
|
||||
LogView {
|
||||
rows: state.messages()
|
||||
rows: state.messages(),
|
||||
selection: if let Selection::Message(idx) = global_selection() { Some(idx) } else { None },
|
||||
onselect: move |idx| global_selection.set(Selection::Message(idx)),
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub fn client_color(client_id: u32) -> &'static str {
|
||||
static COLOR_TABLE: OnceLock<Vec<String>> = OnceLock::new();
|
||||
let color_table =
|
||||
COLOR_TABLE.get_or_init(|| (0..16).map(|i| format!("var(--client-{i:x})")).collect());
|
||||
&color_table[client_id.rem(16) as usize]
|
||||
}
|
||||
|
||||
#[component]
|
||||
fn FutureTreePanel(state: Store<TargetState>, mut selection: WriteSignal<Selection>) -> Element {
|
||||
let msg_id = match selection() {
|
||||
Selection::None => state.messages().len().checked_sub(1),
|
||||
Selection::Message(id) => Some(id),
|
||||
};
|
||||
let Some(msg_id) = msg_id else { return rsx! { "Waiting for messages..."} };
|
||||
let forests = state.get_active_forests(msg_id);
|
||||
rsx! {
|
||||
{forests.iter().map(|forest| {
|
||||
rsx!{
|
||||
TreeView {
|
||||
trees: forest.trees.clone()
|
||||
}
|
||||
}
|
||||
})}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,391 @@
|
||||
use std::ops::Deref;
|
||||
use std::range::RangeInclusive;
|
||||
use std::rc::Rc;
|
||||
|
||||
use checklings::{FutureEventKind, TracedEvent, TracedEventKind, WakerEventKind};
|
||||
use dioxus::signals::{ReadableExt, ReadableVecExt, WritableExt};
|
||||
use dioxus::stores::{Store, store};
|
||||
use itertools::Itertools;
|
||||
|
||||
#[derive(Clone, Debug, Hash, PartialEq, Eq)]
|
||||
pub struct Message {
|
||||
pub client: u32,
|
||||
pub traced: TracedEvent,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug)]
|
||||
pub struct LogItem {
|
||||
pub idx: usize,
|
||||
pub msg: Message,
|
||||
pub ev: usize,
|
||||
}
|
||||
|
||||
impl Eq for LogItem {}
|
||||
impl PartialEq for LogItem {
|
||||
fn eq(&self, other: &Self) -> bool { self.idx == other.idx }
|
||||
}
|
||||
impl std::hash::Hash for LogItem {
|
||||
fn hash<H: std::hash::Hasher>(&self, state: &mut H) { self.idx.hash(state) }
|
||||
}
|
||||
|
||||
/// Information about the transition a future went through within a timespan
|
||||
///
|
||||
/// This is fieldwise-compared, but Tree is ref-compared, so equality is not
|
||||
/// deep
|
||||
#[derive(Clone, Debug, PartialEq)]
|
||||
pub struct TreeNode {
|
||||
/// ID assigned to this future in target and messages
|
||||
pub id: u64,
|
||||
/// index of future in client's catalogue
|
||||
pub index: usize,
|
||||
/// Label passed to section() when the future was created
|
||||
pub name: Rc<String>,
|
||||
/// True if the future was created
|
||||
pub is_new: bool,
|
||||
/// True if the future was dropped
|
||||
pub is_drop: bool,
|
||||
/// True if the future was polled
|
||||
pub was_polled: bool,
|
||||
/// True if the future woke other futures
|
||||
pub did_wake: bool,
|
||||
/// child relationships
|
||||
pub children: Vec<Tree>,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug)]
|
||||
pub struct Tree(Rc<TreeNode>);
|
||||
impl PartialEq for Tree {
|
||||
fn eq(&self, other: &Self) -> bool { Rc::ptr_eq(&self.0, &other.0) }
|
||||
}
|
||||
impl Deref for Tree {
|
||||
type Target = TreeNode;
|
||||
fn deref(&self) -> &Self::Target { &self.0 }
|
||||
}
|
||||
impl Tree {
|
||||
fn deep_update(
|
||||
&self,
|
||||
path: impl IntoIterator<Item = usize>,
|
||||
update: impl FnOnce(TreeNode) -> TreeNode,
|
||||
) -> Self {
|
||||
let mut path_iter = path.into_iter();
|
||||
if let Some(index) = path_iter.next() {
|
||||
let (i, child) =
|
||||
self.children.iter().find_position(|ch| ch.index == index).expect("Invalid path");
|
||||
let new = child.deep_update(path_iter, update);
|
||||
if child != &new {
|
||||
let mut node = self.0.deref().clone();
|
||||
node.children[i] = new;
|
||||
Self(Rc::new(node))
|
||||
} else {
|
||||
self.clone()
|
||||
}
|
||||
} else {
|
||||
let new_node = update(self.0.deref().clone());
|
||||
if new_node != *self.0 { Self(Rc::new(new_node)) } else { self.clone() }
|
||||
}
|
||||
}
|
||||
fn stabilize(&self) -> Option<Self> {
|
||||
let mut node = self.0.deref().clone();
|
||||
node.did_wake = false;
|
||||
node.is_new = false;
|
||||
node.was_polled = false;
|
||||
node.children = (node.children.into_iter()).filter_map(|child| child.stabilize()).collect();
|
||||
if node.is_drop && node.children.is_empty() {
|
||||
None
|
||||
} else if node == **self {
|
||||
Some(self.clone())
|
||||
} else {
|
||||
Some(Self(Rc::new(node)))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn deep_update_vec(
|
||||
children: &mut Vec<Tree>,
|
||||
path: impl IntoIterator<Item = usize>,
|
||||
update: impl FnOnce(TreeNode) -> TreeNode,
|
||||
) -> bool {
|
||||
let mut path_iter = path.into_iter();
|
||||
let Some(name) = path_iter.next() else { panic!("Empty path used to index a void root tree") };
|
||||
let (idx, child) =
|
||||
children.iter().find_position(|ch| ch.index == name).expect("Invalid root name");
|
||||
let new = child.deep_update(path_iter, update);
|
||||
let did_change = new != *child;
|
||||
if did_change {
|
||||
children[idx] = new;
|
||||
}
|
||||
did_change
|
||||
}
|
||||
|
||||
/// A logical group of wake events suspected to be originating from the same
|
||||
/// wake call. This can be built conservatively from nested wakes, or through
|
||||
/// heuristics
|
||||
#[derive(Clone, Debug)]
|
||||
pub struct EventRecord {
|
||||
/// wake that is suspected to have caused the poll. In stepping mode this is
|
||||
/// discoverable because we take control of all wakes, but the conservative
|
||||
/// nested wake strategy does not discover it.
|
||||
pub cause: Option<usize>,
|
||||
/// Message ranges associated with this event. Since at a minimum nested wakes
|
||||
/// are grouped which always have a distinct start and end message, this is an
|
||||
/// ordered list of disjoint ranges
|
||||
pub message_windows: Vec<RangeInclusive<usize>>,
|
||||
}
|
||||
|
||||
/// A single call to the poll() of a Future
|
||||
#[derive(Clone, Debug, Hash)]
|
||||
struct PollRecord {
|
||||
/// The event that this wake, all of its parents, and all of its children
|
||||
/// belong to
|
||||
event: usize,
|
||||
/// The future that was polled
|
||||
future: usize,
|
||||
/// The enclosing future's PollCall, if there is one
|
||||
parent: Option<usize>,
|
||||
/// PollCalls nested within this PollCall on the stack
|
||||
children: Vec<usize>,
|
||||
/// Polls believed to have been caused within this PollCall but outside its
|
||||
/// children
|
||||
woke: Vec<usize>,
|
||||
/// Message when the future was pushed on the stack
|
||||
start_msg: usize,
|
||||
/// Message when the future was popped from the stack
|
||||
end_msg: Option<usize>,
|
||||
}
|
||||
|
||||
pub struct Wake {
|
||||
/// Message corresponding to this wake
|
||||
msg: usize,
|
||||
/// Waker that registered the wake
|
||||
waker: usize,
|
||||
/// Poll record in which wake was called
|
||||
poll: Option<usize>,
|
||||
/// Event activated by this waker. This may be None if
|
||||
///
|
||||
/// - stepping mode was not active so events were never connected to their
|
||||
/// wakers
|
||||
/// - its effect was grouped with a different waker by the executor
|
||||
/// - the waker did not wake anything, because its target got dropped before
|
||||
/// it could be polled
|
||||
event: Option<usize>,
|
||||
}
|
||||
|
||||
/// A long-lived Waker copied out of a Future
|
||||
#[derive(Clone, Debug)]
|
||||
struct Waker {
|
||||
id: u64,
|
||||
/// Future this Waker should wake
|
||||
future: usize,
|
||||
/// Message where this waker was constructed
|
||||
new_msg: usize,
|
||||
/// Message where this waker was dropped
|
||||
drop_msg: Option<usize>,
|
||||
/// Polls believed to have been caused by this Waker. All of these have Some
|
||||
/// cause pointing back to the waker, and if the target is correct, all of
|
||||
/// them include a wake of [Self::future]
|
||||
polls: Vec<usize>,
|
||||
}
|
||||
|
||||
/// Record of a single future across time
|
||||
#[derive(Clone, Debug)]
|
||||
struct FutureRecord {
|
||||
/// Unique sequential ID for this future with respect to this debug target
|
||||
id: u64,
|
||||
/// Name provided to the section in annotated code
|
||||
section: Rc<String>,
|
||||
/// Index of the message that created this future
|
||||
new_msg: usize,
|
||||
/// Index of the message that dropped this future
|
||||
drop_msg: Option<usize>,
|
||||
/// Every PollCall to this future's poll(). We fundamentally don't know which
|
||||
/// waker caused these. Since poll() borrows the future mutably, these are
|
||||
/// non-overlapping unless the client has an unsafe violation
|
||||
polled: Vec<usize>,
|
||||
/// Every waker belonging to this future
|
||||
wakers: Vec<usize>,
|
||||
/// All futures ever polled under this one
|
||||
children: Vec<usize>,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq)]
|
||||
pub struct ForestSnapshot {
|
||||
pub msg: usize,
|
||||
pub client_id: u32,
|
||||
pub trees: Vec<Tree>,
|
||||
}
|
||||
|
||||
#[derive(Store, Default)]
|
||||
pub struct ClientState {
|
||||
client_id: u32,
|
||||
futures: Vec<FutureRecord>,
|
||||
wakers: Vec<Waker>,
|
||||
wakes: Vec<Wake>,
|
||||
polls: Vec<PollRecord>,
|
||||
stack: Vec<usize>,
|
||||
forest_states: Vec<ForestSnapshot>,
|
||||
}
|
||||
impl ClientState {
|
||||
fn find_future(&self, id: u64) -> usize {
|
||||
self.futures.binary_search_by_key(&id, |f| f.id).expect("Future not found")
|
||||
}
|
||||
fn find_waker(&self, id: u64) -> usize {
|
||||
self.wakers.binary_search_by_key(&id, |f| f.id).expect("Waker not found")
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Store, Default)]
|
||||
pub struct TargetState {
|
||||
pub clients: Vec<ClientState>,
|
||||
pub events: Vec<EventRecord>,
|
||||
pub messages: Vec<LogItem>,
|
||||
}
|
||||
|
||||
#[store(pub name = AggregateExt)]
|
||||
impl<Lens> Store<TargetState, Lens> {
|
||||
fn update(&mut self, msg: Message) {
|
||||
let mut agg = self.write();
|
||||
let TargetState { clients, events, messages } = &mut *agg;
|
||||
let msg_idx = messages.len();
|
||||
|
||||
// determine which client this belongs to
|
||||
let client = match clients.iter_mut().find(|c| c.client_id == msg.client) {
|
||||
Some(client) => client,
|
||||
None => {
|
||||
// create client if missing
|
||||
let client = ClientState { client_id: msg.client, ..ClientState::default() };
|
||||
clients.push(client);
|
||||
clients.last_mut().unwrap()
|
||||
},
|
||||
};
|
||||
|
||||
// try to find the current event
|
||||
let parent_poll_idx = client.stack.last().copied();
|
||||
let ev_idx = if let Some(parent_poll_idx) = parent_poll_idx {
|
||||
let ev_idx = client.polls[parent_poll_idx].event;
|
||||
let msg_wins = &mut events[ev_idx].message_windows;
|
||||
// extend current event to include current message
|
||||
if let Some(end) = msg_wins.last_mut().filter(|r| r.last + 1 == msg_idx) {
|
||||
end.last = msg_idx;
|
||||
} else {
|
||||
msg_wins.push((msg_idx..=msg_idx).into());
|
||||
}
|
||||
ev_idx
|
||||
} else {
|
||||
// start a new event
|
||||
// TODO: configurable eventing strategy. This is a good fallback.
|
||||
let ev_idx = events.len();
|
||||
events.push(EventRecord {
|
||||
cause: None,
|
||||
message_windows: vec![RangeInclusive { start: msg_idx, last: msg_idx }],
|
||||
});
|
||||
ev_idx
|
||||
};
|
||||
|
||||
let trees = (client.forest_states.last())
|
||||
.map(|last| last.trees.iter().filter_map(|t| t.stabilize()).collect())
|
||||
.unwrap_or_default();
|
||||
|
||||
match &msg.traced.kind {
|
||||
TracedEventKind::Log(_) => (/* No specific action in relation to log messages */),
|
||||
TracedEventKind::Waker(wak_ev) => match &wak_ev.kind {
|
||||
WakerEventKind::New { future_id } => {
|
||||
let fut_idx = client.find_future(*future_id);
|
||||
client.wakers.push(Waker {
|
||||
polls: Vec::new(),
|
||||
drop_msg: None,
|
||||
id: wak_ev.id,
|
||||
new_msg: msg_idx,
|
||||
future: fut_idx,
|
||||
});
|
||||
},
|
||||
WakerEventKind::Drop => {
|
||||
let wak_idx = client.find_waker(wak_ev.id);
|
||||
client.wakers[wak_idx].drop_msg = Some(msg_idx);
|
||||
},
|
||||
WakerEventKind::Wake { by_ref } => {
|
||||
// TODO: in stepping mode, create the event here and use it to collect all
|
||||
// future wakes
|
||||
client.wakes.push(Wake {
|
||||
event: None,
|
||||
msg: msg_idx,
|
||||
poll: client.stack.last().copied(),
|
||||
waker: client.find_waker(wak_ev.id),
|
||||
});
|
||||
},
|
||||
},
|
||||
TracedEventKind::Future(fut_ev) => match fut_ev.kind {
|
||||
FutureEventKind::Poll => {
|
||||
let fut_idx = client.find_future(fut_ev.id);
|
||||
let poll_idx = client.polls.len();
|
||||
// try to find the parent poll
|
||||
if let Some(parent_poll_idx) = parent_poll_idx {
|
||||
// insert future as child of parent poll's future if missing (future and child
|
||||
// indices may appear out of order)
|
||||
let parent_fut_idx = client.polls[parent_poll_idx].future;
|
||||
let children_of_parent = &mut client.futures[parent_fut_idx].children;
|
||||
if let Err(parent_fut_child_idx) = children_of_parent.binary_search(&fut_idx) {
|
||||
children_of_parent.insert(parent_fut_child_idx, fut_idx);
|
||||
}
|
||||
// insert poll as child of parent poll and copy its event
|
||||
client.polls[parent_poll_idx].children.push(poll_idx);
|
||||
}
|
||||
client.polls.push(PollRecord {
|
||||
children: Vec::new(),
|
||||
woke: Vec::new(),
|
||||
end_msg: None,
|
||||
future: fut_idx,
|
||||
parent: parent_poll_idx,
|
||||
event: ev_idx,
|
||||
start_msg: msg_idx,
|
||||
});
|
||||
client.stack.push(poll_idx);
|
||||
},
|
||||
FutureEventKind::PollEnd => {
|
||||
let fut_idx = client.find_future(fut_ev.id);
|
||||
let Some(last_poll_idx) = client.stack.pop() else {
|
||||
eprintln!("{fut_ev:?} could not find a matching start in the aggregate");
|
||||
return;
|
||||
};
|
||||
let last_poll = &mut client.polls[last_poll_idx];
|
||||
if last_poll.future != fut_idx {
|
||||
eprintln!("{fut_ev:?} is for a different future than {last_poll:?}");
|
||||
}
|
||||
last_poll.end_msg = Some(msg_idx);
|
||||
},
|
||||
FutureEventKind::Trampoline => {},
|
||||
FutureEventKind::Drop => {
|
||||
let fut_idx = client.find_future(fut_ev.id);
|
||||
client.futures[fut_idx].drop_msg = Some(msg_idx);
|
||||
},
|
||||
FutureEventKind::New => client.futures.push(FutureRecord {
|
||||
children: Vec::new(),
|
||||
polled: Vec::new(),
|
||||
wakers: Vec::new(),
|
||||
drop_msg: None,
|
||||
id: fut_ev.id,
|
||||
new_msg: msg_idx,
|
||||
section: match &msg.traced.info {
|
||||
None => Rc::new("MISSING NAME".to_string()),
|
||||
Some(trace) => Rc::new(trace.label.clone()),
|
||||
},
|
||||
}),
|
||||
},
|
||||
}
|
||||
|
||||
client.forest_states.push(ForestSnapshot { msg: msg_idx, client_id: msg.client, trees });
|
||||
messages.push(LogItem { idx: msg_idx, ev: ev_idx, msg });
|
||||
}
|
||||
fn get_active_forests(&self, msg_idx: usize) -> Vec<ForestSnapshot> {
|
||||
self
|
||||
.clients()
|
||||
.iter()
|
||||
.filter_map(|c| {
|
||||
match c.forest_states().read().binary_search_by_key(&msg_idx, |state| state.msg) {
|
||||
Err(0) => None,
|
||||
Err(idx) => c.forest_states().read().get(idx - 1).cloned(),
|
||||
Ok(idx) => c.forest_states().read().get(idx).cloned(),
|
||||
}
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
}
|
||||
+16
-209
@@ -1,216 +1,23 @@
|
||||
use std::range::RangeInclusive;
|
||||
use std::rc::Rc;
|
||||
|
||||
use checklings::{FutureEvent, FutureEventKind, TracedEvent, TracedEventKind, WakerEventKind};
|
||||
use dioxus::signals::{ReadableExt, WritableExt};
|
||||
use dioxus::stores::{Store, store};
|
||||
use dioxus::prelude::*;
|
||||
use itertools::Itertools;
|
||||
|
||||
use crate::log::LogItem;
|
||||
use crate::web_bindings::Console;
|
||||
use crate::state::{Tree, TreeNode};
|
||||
|
||||
/// A logical group of wake events suspected to be originating from the same
|
||||
/// wake call. This can be built conservatively from nested wakes, or through
|
||||
/// heuristics
|
||||
#[derive(Clone, Debug)]
|
||||
pub struct EventRecord {
|
||||
/// wake that is suspected to have caused the poll. In stepping mode this is
|
||||
/// discoverable because we take control of all wakes, but the conservative
|
||||
/// nested wake strategy does not discover it.
|
||||
cause: Option<usize>,
|
||||
/// Message ranges associated with this event. Since at a minimum nested wakes
|
||||
/// are grouped which always have a distinct start and end message, this is an
|
||||
/// ordered list of disjoint ranges
|
||||
message_windows: Vec<RangeInclusive<usize>>,
|
||||
}
|
||||
|
||||
/// A single call to the poll() of a Future
|
||||
#[derive(Clone, Debug, Hash)]
|
||||
struct PollRecord {
|
||||
/// The event that this wake, all of its parents, and all of its children
|
||||
/// belong to
|
||||
event: usize,
|
||||
/// The future that was polled
|
||||
future: usize,
|
||||
/// The enclosing future's PollCall, if there is one
|
||||
parent: Option<usize>,
|
||||
/// PollCalls nested within this PollCall on the stack
|
||||
children: Vec<usize>,
|
||||
/// Polls believed to have been caused within this PollCall but outside its
|
||||
/// children
|
||||
woke: Vec<usize>,
|
||||
/// Message when the future was pushed on the stack
|
||||
start_msg: usize,
|
||||
/// Message when the future was popped from the stack
|
||||
end_msg: Option<usize>,
|
||||
}
|
||||
|
||||
pub struct Wake {
|
||||
/// Message corresponding to this wake
|
||||
msg: usize,
|
||||
/// Waker that registered the wake
|
||||
waker: usize,
|
||||
/// Poll record in which wake was called
|
||||
poll: Option<usize>,
|
||||
/// Event activated by this waker. This may be None if
|
||||
///
|
||||
/// - stepping mode was not active so events were never connected to their
|
||||
/// wakers
|
||||
/// - its effect was grouped with a different waker by the executor
|
||||
/// - the waker did not wake anything, because its target got dropped before
|
||||
/// it could be polled
|
||||
event: Option<usize>,
|
||||
}
|
||||
|
||||
/// A long-lived Waker copied out of a Future
|
||||
#[derive(Clone, Debug)]
|
||||
struct Waker {
|
||||
/// Future this Waker should wake
|
||||
future: u64,
|
||||
/// Message where this waker was constructed
|
||||
new_msg: usize,
|
||||
/// Message where this waker was dropped
|
||||
drop_msg: usize,
|
||||
/// Polls believed to have been caused by this Waker. All of these have Some
|
||||
/// cause pointing back to the waker, and if the target is correct, all of
|
||||
/// them include a wake of [Self::future]
|
||||
polls: Vec<usize>,
|
||||
}
|
||||
|
||||
/// Record of a single future across time
|
||||
#[derive(Clone, Debug)]
|
||||
struct FutureRecord {
|
||||
/// Unique sequential ID for this future with respect to this debug target
|
||||
id: u64,
|
||||
/// Name provided to the section in annotated code
|
||||
section: Rc<String>,
|
||||
/// Index of the message that created this future
|
||||
new_msg: usize,
|
||||
/// Index of the message that dropped this future
|
||||
drop_msg: Option<usize>,
|
||||
/// Every PollCall to this future's poll(). We fundamentally don't know which
|
||||
/// waker caused these. Since poll() borrows the future mutably, these are
|
||||
/// non-overlapping unless the client has an unsafe violation
|
||||
polled: Vec<usize>,
|
||||
/// Every waker belonging to this future
|
||||
wakers: Vec<usize>,
|
||||
/// All futures ever polled under this one
|
||||
children: Vec<usize>,
|
||||
}
|
||||
|
||||
#[derive(Store, Default)]
|
||||
pub struct ClientRecord {
|
||||
client_id: u32,
|
||||
futures: Vec<FutureRecord>,
|
||||
wakers: Vec<Waker>,
|
||||
wakes: Vec<Wake>,
|
||||
polls: Vec<PollRecord>,
|
||||
stack: Vec<usize>,
|
||||
}
|
||||
impl ClientRecord {
|
||||
fn find_future(&self, id: u64) -> usize {
|
||||
self.futures.binary_search_by_key(&id, |f| f.id).expect("Future not found")
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Store, Default)]
|
||||
pub struct Aggregate {
|
||||
pub clients: Vec<ClientRecord>,
|
||||
pub events: Vec<EventRecord>,
|
||||
pub messages: Vec<LogItem>,
|
||||
}
|
||||
|
||||
#[store(pub name = AggregateExt)]
|
||||
impl<Lens> Store<Aggregate, Lens> {
|
||||
fn update(&mut self, item: LogItem) {
|
||||
let msg_idx = self.messages().len();
|
||||
let mut agg = self.write();
|
||||
let Aggregate { clients, events, messages } = &mut *agg;
|
||||
messages.push(item.clone());
|
||||
let LogItem::Event(event) = item else { return };
|
||||
let client = match clients.iter_mut().find(|c| c.client_id == event.client) {
|
||||
Some(client) => client,
|
||||
None => {
|
||||
let client = ClientRecord { client_id: event.client, ..ClientRecord::default() };
|
||||
clients.push(client);
|
||||
clients.last_mut().unwrap()
|
||||
},
|
||||
};
|
||||
match event.kind {
|
||||
TracedEventKind::Log(_) => (),
|
||||
TracedEventKind::Waker(wak_ev) => match wak_ev.kind {
|
||||
WakerEventKind::New { future_id } => todo!(),
|
||||
WakerEventKind::Wake { by_ref } => todo!(),
|
||||
WakerEventKind::Drop => todo!(),
|
||||
},
|
||||
TracedEventKind::Future(fut_ev) => match fut_ev.kind {
|
||||
FutureEventKind::Poll => {
|
||||
let fut_idx = client.find_future(fut_ev.id);
|
||||
let poll_idx = client.polls.len();
|
||||
// try to find the parent poll
|
||||
let parent_idx = client.stack.last().copied();
|
||||
let ev_idx = if let Some(parent_idx) = parent_idx {
|
||||
// insert future as child of parent poll's future if missing (future and child
|
||||
// indices may appear out of order)
|
||||
let parent_fut_idx = client.polls[parent_idx].future;
|
||||
if let Err(parent_fut_child_idx) =
|
||||
client.futures[parent_fut_idx].children.binary_search(&parent_idx)
|
||||
{
|
||||
client.futures[parent_fut_idx].children.insert(parent_fut_child_idx, fut_idx);
|
||||
}
|
||||
// inesrt poll as child of parent poll and copy its event
|
||||
client.polls[parent_idx].children.push(poll_idx);
|
||||
client.polls[parent_idx].event
|
||||
} else {
|
||||
// start a new event
|
||||
// TODO: configurable eventing strategy. This is a good fallback.
|
||||
let ev_idx = events.len();
|
||||
events.push(EventRecord {
|
||||
cause: None,
|
||||
message_windows: vec![RangeInclusive { start: msg_idx, last: msg_idx }],
|
||||
});
|
||||
ev_idx
|
||||
};
|
||||
client.polls.push(PollRecord {
|
||||
children: Vec::new(),
|
||||
woke: Vec::new(),
|
||||
end_msg: None,
|
||||
future: fut_idx,
|
||||
parent: parent_idx,
|
||||
event: ev_idx,
|
||||
start_msg: msg_idx,
|
||||
});
|
||||
client.stack.push(poll_idx);
|
||||
},
|
||||
FutureEventKind::PollEnd => {
|
||||
let fut_idx = client.find_future(fut_ev.id);
|
||||
let Some(last_poll_idx) = client.stack.pop() else {
|
||||
eprintln!("{fut_ev:?} could not find a matching start in the aggregate");
|
||||
return;
|
||||
};
|
||||
let last_poll = &mut client.polls[last_poll_idx];
|
||||
if last_poll.future != fut_idx {
|
||||
eprintln!("{fut_ev:?} is for a different future than {last_poll:?}");
|
||||
}
|
||||
last_poll.end_msg = Some(msg_idx);
|
||||
},
|
||||
FutureEventKind::Trampoline => todo!(),
|
||||
FutureEventKind::Drop => {
|
||||
let fut_idx = client.find_future(fut_ev.id);
|
||||
client.futures[fut_idx].drop_msg = Some(msg_idx);
|
||||
},
|
||||
FutureEventKind::New => client.futures.push(FutureRecord {
|
||||
children: Vec::new(),
|
||||
polled: Vec::new(),
|
||||
wakers: Vec::new(),
|
||||
drop_msg: None,
|
||||
id: fut_ev.id,
|
||||
new_msg: msg_idx,
|
||||
section: match &event.info {
|
||||
None => Rc::new("MISSING NAME".to_string()),
|
||||
Some(trace) => Rc::new(trace.label.clone()),
|
||||
},
|
||||
}),
|
||||
},
|
||||
#[component]
|
||||
pub fn TreeView(trees: Vec<Tree>) -> Element {
|
||||
rsx! {
|
||||
div {
|
||||
"This is a tree view"
|
||||
}
|
||||
{trees.iter().map(|tree| rsx!{ TreeBranch { tree: tree.clone() } })}
|
||||
}
|
||||
}
|
||||
|
||||
#[component]
|
||||
pub fn TreeBranch(tree: Tree) -> Element {
|
||||
rsx! {
|
||||
div { "{tree.name}"}
|
||||
}
|
||||
}
|
||||
@@ -24,26 +24,34 @@ impl PinEnd {
|
||||
/// with
|
||||
///
|
||||
/// when we call a set_* callback, we place a request. The host can signal
|
||||
/// completion by updating current, and block further attempts by setting
|
||||
/// can_extend_$1 to false. Retries must be expressed in the host.
|
||||
///
|
||||
/// For ideal experience, overscroll.start must be big enough for the fetch
|
||||
/// function to complete before a very fast scrolling user gets through it.
|
||||
///
|
||||
/// For correctness, `overscroll.end - overscroll.start` must be at least 3x
|
||||
/// times as large as the largest possible element. Failure to ensure this can
|
||||
/// cause the scroller to freeze up while repeatedly extending and shrinking the
|
||||
/// list
|
||||
/// completion by updating current in the same render where it updates children,
|
||||
/// and block further attempts by setting can_extend_$1 to false. Retries must
|
||||
/// be expressed in the host.
|
||||
#[component]
|
||||
pub fn VirtualScroll(
|
||||
pin_end: ReadSignal<Option<PinEnd>>,
|
||||
set_start: Callback<i64, ()>,
|
||||
can_extend_start: ReadSignal<bool>,
|
||||
set_end: Callback<i64, ()>,
|
||||
can_extend_end: ReadSignal<bool>,
|
||||
current: ReadSignal<RangeInclusive<i64>>,
|
||||
children: Element,
|
||||
/// For ideal experience, overscroll.start must be big enough for the fetch
|
||||
/// function to complete before a very fast scrolling user gets through it.
|
||||
///
|
||||
/// For correctness, `overscroll.end - overscroll.start` must be at least 3x
|
||||
/// times as large as the largest possible element. Failure to ensure this can
|
||||
/// cause the scroller to freeze up while repeatedly extending and shrinking
|
||||
/// the list
|
||||
overscroll: RangeInclusive<i8>,
|
||||
/// pin_end can be used to force the scroller to one or the other end of the
|
||||
/// spectrum. In this mode the component will not move the pinned end, and
|
||||
/// only use the other end to maintain the overscroll region, thereby acting
|
||||
/// as a view-box-sized paginator.
|
||||
pin_end: ReadSignal<Option<PinEnd>>,
|
||||
/// If pin_end is in use, on_pin_contest can be used
|
||||
/// to respond to a scroll action away from the pinned edge. The most obvious
|
||||
/// use of this event is to disable the pinning.
|
||||
on_pin_contest: Option<Callback<(), ()>>,
|
||||
) -> Element {
|
||||
let (oscr_low, oscr_high) = overscroll.into_inner();
|
||||
|
||||
@@ -172,6 +180,16 @@ pub fn VirtualScroll(
|
||||
end.scroll_to(el)
|
||||
}
|
||||
},
|
||||
onwheel: move |ev| {
|
||||
if let (Some(end), Some(on_contest)) = (pin_end(), on_pin_contest) {
|
||||
let offset = ev.data().delta().strip_units().y;
|
||||
match (end, offset) {
|
||||
(PinEnd::Start, 0f64..) => on_contest(()),
|
||||
(PinEnd::End, ..0f64) => on_contest(()),
|
||||
_ => {},
|
||||
}
|
||||
}
|
||||
},
|
||||
onscroll: move |_| {
|
||||
if let Some(end) = pin_end() && let Some(el) = view_area() {
|
||||
end.scroll_to(el)
|
||||
|
||||
@@ -4,31 +4,31 @@ use std::time::Duration;
|
||||
|
||||
use checklings::TracedEvent;
|
||||
use futures::prelude::*;
|
||||
use gloo_net::websocket::Message;
|
||||
use gloo_net::websocket::Message as WSMessage;
|
||||
use gloo_net::websocket::futures::WebSocket;
|
||||
use orchid_api_traits::Decode as _;
|
||||
use scoped_log::{NullLogger, with_logger};
|
||||
|
||||
use crate::log::LogItem;
|
||||
use crate::state::Message;
|
||||
use crate::web_bindings::{Console, sleep};
|
||||
|
||||
#[derive(Clone, Debug, Hash, PartialEq, Eq)]
|
||||
pub enum LogAction {
|
||||
Push(LogItem),
|
||||
Set(Vec<LogItem>),
|
||||
pub enum WsEvent {
|
||||
Push(Message),
|
||||
Set(Vec<Message>),
|
||||
Status(String),
|
||||
}
|
||||
|
||||
pub async fn with_ws(url: String, topic: String, mut log_action: impl FnMut(LogAction)) {
|
||||
pub async fn with_ws(url: String, topic: String, mut log_action: impl FnMut(WsEvent)) {
|
||||
loop {
|
||||
sleep(Duration::from_millis(100)).await;
|
||||
let url = format!("{}/topic/{}/read", url, topic);
|
||||
let Ok(mut socket) = WebSocket::open(&url) else {
|
||||
log_action(LogAction::Set(vec![LogItem::msg(format!(
|
||||
"Failed to construct websocket at {url}"
|
||||
))]));
|
||||
log_action(WsEvent::Status(format!("Failed to construct websocket at {url}")));
|
||||
break;
|
||||
};
|
||||
log_action(LogAction::Set(vec![LogItem::msg(format!("Connected to {url}"))]));
|
||||
log_action(WsEvent::Status(format!("Connected to {url}")));
|
||||
log_action(WsEvent::Set(Vec::new()));
|
||||
let log_action = &RefCell::new(&mut log_action);
|
||||
let (mut writer, reader) = unsync_pipe::pipe(1024);
|
||||
let (retry, ()) = futures::future::join(
|
||||
@@ -37,17 +37,14 @@ pub async fn with_ws(url: String, topic: String, mut log_action: impl FnMut(LogA
|
||||
match msg {
|
||||
Err(e) => {
|
||||
writeln!(Console, "Failed to connect websocket, {e}");
|
||||
log_action.borrow_mut()(LogAction::Push(LogItem::msg(format!(
|
||||
"connection error: {e}"
|
||||
))));
|
||||
log_action.borrow_mut()(WsEvent::Status(format!("connection error: {e}")));
|
||||
sleep(Duration::from_secs(3)).await;
|
||||
return true;
|
||||
},
|
||||
Ok(Message::Text(txt)) if txt == "reset" =>
|
||||
log_action.borrow_mut()(LogAction::Set(Vec::new())),
|
||||
Ok(Message::Text(txt)) =>
|
||||
log_action.borrow_mut()(LogAction::Push(LogItem::msg(format!("text: {txt}")))),
|
||||
Ok(Message::Bytes(bytes)) =>
|
||||
Ok(WSMessage::Text(txt)) if txt == "reset" =>
|
||||
log_action.borrow_mut()(WsEvent::Set(Vec::new())),
|
||||
Ok(WSMessage::Text(txt)) => log_action.borrow_mut()(WsEvent::Status(txt)),
|
||||
Ok(WSMessage::Bytes(bytes)) =>
|
||||
if let Err(e) = writer.write_all(&bytes).await {
|
||||
writeln!(Console, "{e}");
|
||||
break;
|
||||
@@ -58,16 +55,19 @@ pub async fn with_ws(url: String, topic: String, mut log_action: impl FnMut(LogA
|
||||
}),
|
||||
with_logger(NullLogger, async move {
|
||||
let mut source = pin!(reader);
|
||||
loop {
|
||||
match TracedEvent::decode(source.as_mut()).await {
|
||||
Ok(ev) => log_action.borrow_mut()(LogAction::Push(LogItem::Event(ev))),
|
||||
Err(e) => {
|
||||
writeln!(Console, "{e}");
|
||||
log_action.borrow_mut()(LogAction::Push(LogItem::msg(e.to_string())));
|
||||
break;
|
||||
},
|
||||
}
|
||||
}
|
||||
let err = loop {
|
||||
let client = match u32::decode(source.as_mut()).await {
|
||||
Ok(c) => c,
|
||||
Err(e) => break e,
|
||||
};
|
||||
let msg = match TracedEvent::decode(source.as_mut()).await {
|
||||
Ok(ev) => ev,
|
||||
Err(e) => break e,
|
||||
};
|
||||
log_action.borrow_mut()(WsEvent::Push(Message { client, traced: msg }));
|
||||
};
|
||||
writeln!(Console, "{err}");
|
||||
log_action.borrow_mut()(WsEvent::Status(err.to_string()));
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
|
||||
+43
-30
@@ -3,16 +3,13 @@ use std::borrow::Cow;
|
||||
use std::cell::RefCell;
|
||||
use std::pin::Pin;
|
||||
use std::rc::Rc;
|
||||
#[cfg(feature = "telemetry")]
|
||||
use std::sync::Mutex;
|
||||
use std::sync::atomic::{AtomicU32, AtomicU64, Ordering};
|
||||
use std::sync::{Arc, OnceLock};
|
||||
use std::sync::atomic::{AtomicU64, Ordering};
|
||||
use std::sync::{Arc, Mutex, OnceLock};
|
||||
use std::task::{Context, Poll, RawWaker, RawWakerVTable, Wake, Waker};
|
||||
#[cfg(feature = "telemetry")]
|
||||
use std::time::Duration;
|
||||
|
||||
use orchid_api_derive::Coding;
|
||||
#[cfg(feature = "telemetry")]
|
||||
use orchid_api_traits::Encode;
|
||||
use scoped_value::{ScopeExt, WithRef};
|
||||
#[cfg(feature = "telemetry")]
|
||||
@@ -22,13 +19,10 @@ use websocket::stream::sync::NetworkStream;
|
||||
#[cfg(feature = "telemetry")]
|
||||
use websocket::sync::Client;
|
||||
#[cfg(feature = "telemetry")]
|
||||
use websocket::{ClientBuilder, OwnedMessage};
|
||||
use websocket::{ClientBuilder, Message};
|
||||
|
||||
#[derive(Debug, Coding, Clone, Hash, PartialEq, Eq)]
|
||||
pub struct TracedEvent {
|
||||
/// Initialized from the checkers server. If using a custom hook via
|
||||
/// [init_trace], must be initialized in the hook. Defaults to zero.
|
||||
pub client: u32,
|
||||
/// Where the event was triggered
|
||||
pub info: Option<Arc<TraceInfo>>,
|
||||
/// The event that actually occurred
|
||||
@@ -98,7 +92,8 @@ pub struct TraceInfo {
|
||||
pub label: String,
|
||||
}
|
||||
|
||||
pub static ON_TRACE: OnceLock<Box<dyn Fn(TracedEvent) + Send + Sync>> = OnceLock::new();
|
||||
type TraceCallback = Box<dyn Fn(&[u8]) + Send + Sync>;
|
||||
pub static ON_TRACE: OnceLock<TraceCallback> = OnceLock::new();
|
||||
|
||||
#[cfg(feature = "telemetry")]
|
||||
static UREQ_AGENT: OnceLock<Agent> = OnceLock::new();
|
||||
@@ -109,11 +104,23 @@ struct WebsocketState {
|
||||
}
|
||||
#[cfg(feature = "telemetry")]
|
||||
static WEBSOCKET_CONN: OnceLock<Mutex<WebsocketState>> = OnceLock::new();
|
||||
static CLIENT_ID: AtomicU32 = AtomicU32::new(0);
|
||||
static CLIENT_ID: OnceLock<u32> = OnceLock::new();
|
||||
static BACKLOG: Mutex<Vec<Vec<u8>>> = Mutex::new(Vec::new());
|
||||
|
||||
pub fn init_trace<F: Fn(TracedEvent) + Send + Sync + 'static>(if_not_ready: bool, on_trace: F) {
|
||||
let new = ON_TRACE.set(Box::new(on_trace)).is_ok();
|
||||
assert!(new || if_not_ready, "Trace handler already configured");
|
||||
fn init_trace<F: Fn(&[u8]) + Send + Sync + 'static>(if_not_ready: bool, on_trace: F) {
|
||||
let new = ON_TRACE.set(Box::new(on_trace));
|
||||
assert!(new.is_ok() || if_not_ready, "Trace handler already configured");
|
||||
let backlog_items = std::mem::take(&mut *BACKLOG.lock().unwrap());
|
||||
for message in backlog_items {
|
||||
let mut with_client_id = Vec::new();
|
||||
CLIENT_ID.get().unwrap().encode_vec(&mut with_client_id);
|
||||
with_client_id.extend(message);
|
||||
(ON_TRACE.get().unwrap())(&with_client_id);
|
||||
}
|
||||
}
|
||||
|
||||
pub fn disable_trace() {
|
||||
let _ = ON_TRACE.set(Box::new(|_| panic!("Trace is disabled; this should not be called")));
|
||||
}
|
||||
|
||||
#[cfg(feature = "telemetry")]
|
||||
@@ -146,27 +153,22 @@ pub fn init_telemetry(url: impl AsRef<str>, topic: impl AsRef<str>, role: Role)
|
||||
let http_url =
|
||||
url.strip_prefix("ws").map_or(Cow::Borrowed(url), |s| Cow::Owned(format!("http{s}")));
|
||||
if role == Role::Leader {
|
||||
assert!(CLIENT_ID.get().is_none(), "Leader initialized in already-initialized address space");
|
||||
(get_ureq().post(format!("{http_url}/topic/{topic}/clear")).send_empty())
|
||||
.unwrap_or_else(|e| panic!("Could not clear {topic} on {http_url}; {e}"));
|
||||
}
|
||||
let id = {
|
||||
let id_str = (get_ureq().post(format!("{http_url}/topic/{topic}/next_id")).send_empty())
|
||||
CLIENT_ID.get_or_init(|| {
|
||||
let id_rep = (get_ureq().post(format!("{http_url}/topic/{topic}/next_id")).send_empty())
|
||||
.and_then(|res| res.into_body().read_to_string())
|
||||
.unwrap_or_else(|e| panic!("Could not acquire client ID from {topic} on {http_url}; {e}"));
|
||||
(id_str.trim().parse::<u32>())
|
||||
(id_rep.trim().parse::<u32>())
|
||||
.unwrap_or_else(|e| panic!("Server did not return a number from next_id as expected; {e}"))
|
||||
};
|
||||
CLIENT_ID.store(id, std::sync::atomic::Ordering::Relaxed);
|
||||
fn encode_event(ev: &TracedEvent) -> Vec<u8> {
|
||||
let mut vec = Vec::new();
|
||||
ev.encode_vec(&mut vec);
|
||||
vec
|
||||
}
|
||||
});
|
||||
if url.starts_with("http") {
|
||||
let post_url = format!("{url}/topic/{topic}/push");
|
||||
let ureq = get_ureq();
|
||||
init_trace(role == Role::Library, move |ev| {
|
||||
ureq.post(&post_url).send(&encode_event(&ev)).unwrap();
|
||||
init_trace(role == Role::Library, move |data| {
|
||||
ureq.post(&post_url).send(data).unwrap();
|
||||
});
|
||||
return;
|
||||
};
|
||||
@@ -175,7 +177,7 @@ pub fn init_telemetry(url: impl AsRef<str>, topic: impl AsRef<str>, role: Role)
|
||||
let client = builder.connect(None).unwrap();
|
||||
let _ = WEBSOCKET_CONN.set(Mutex::new(WebsocketState { builder, client }));
|
||||
init_trace(role == Role::Library, move |ev| {
|
||||
let data = OwnedMessage::Binary(encode_event(&ev));
|
||||
let data = Message::binary(ev);
|
||||
let WebsocketState { builder, client } = &mut *WEBSOCKET_CONN.get().unwrap().lock().unwrap();
|
||||
for _attempt in 0..3 {
|
||||
match client.send_message(&data) {
|
||||
@@ -195,10 +197,21 @@ thread_local! {
|
||||
}
|
||||
|
||||
fn emit(kind: TracedEventKind) {
|
||||
let info = TRACE_INFO.get();
|
||||
let client = CLIENT_ID.load(std::sync::atomic::Ordering::Relaxed);
|
||||
if let Some(cb) = ON_TRACE.get() {
|
||||
(cb)(TracedEvent { client, info, kind })
|
||||
if let Some(client) = CLIENT_ID.get().copied() {
|
||||
let info = TRACE_INFO.get();
|
||||
let mut buf = Vec::new();
|
||||
client.encode_vec(&mut buf);
|
||||
TracedEvent { info, kind }.encode_vec(&mut buf);
|
||||
(cb)(&buf)
|
||||
} else {
|
||||
// do nothing, trace is disabled
|
||||
}
|
||||
} else {
|
||||
let info = TRACE_INFO.get();
|
||||
let mut buf = Vec::new();
|
||||
TracedEvent { info, kind }.encode_vec(&mut buf);
|
||||
BACKLOG.lock().unwrap().push(buf);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -1,20 +1,21 @@
|
||||
use std::pin::Pin;
|
||||
use std::task::{Context, Poll};
|
||||
|
||||
use futures::FutureExt;
|
||||
|
||||
/// Future returned by [cancel_cleanup]
|
||||
pub struct CancelCleanup<Fut: Future + Unpin, Fun: FnOnce(Fut)> {
|
||||
pub struct CancelCleanup<Fut: Future + Unpin, Fun: FnOnce(Fut) + Unpin> {
|
||||
/// Set to None when Ready
|
||||
fut: Option<Fut>,
|
||||
/// Only set to None in Drop
|
||||
on_drop: Option<Fun>,
|
||||
}
|
||||
impl<Fut: Future + Unpin, Fun: FnOnce(Fut)> Future for CancelCleanup<Fut, Fun> {
|
||||
impl<Fut: Future + Unpin, Fun: FnOnce(Fut) + Unpin> Future for CancelCleanup<Fut, Fun> {
|
||||
type Output = Fut::Output;
|
||||
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
|
||||
let Self { fut, .. } = unsafe { self.get_unchecked_mut() };
|
||||
let Self { fut, .. } = self.get_mut();
|
||||
if let Some(future) = fut {
|
||||
let future = unsafe { Pin::new_unchecked(future) };
|
||||
let poll = future.poll(cx);
|
||||
let poll = future.poll_unpin(cx);
|
||||
if poll.is_ready() {
|
||||
*fut = None;
|
||||
}
|
||||
@@ -24,7 +25,7 @@ impl<Fut: Future + Unpin, Fun: FnOnce(Fut)> Future for CancelCleanup<Fut, Fun> {
|
||||
}
|
||||
}
|
||||
}
|
||||
impl<Fut: Future + Unpin, Fun: FnOnce(Fut)> Drop for CancelCleanup<Fut, Fun> {
|
||||
impl<Fut: Future + Unpin, Fun: FnOnce(Fut) + Unpin> Drop for CancelCleanup<Fut, Fun> {
|
||||
fn drop(&mut self) {
|
||||
if let Some(fut) = self.fut.take() {
|
||||
(self.on_drop.take().unwrap())(fut)
|
||||
@@ -35,7 +36,7 @@ impl<Fut: Future + Unpin, Fun: FnOnce(Fut)> Drop for CancelCleanup<Fut, Fun> {
|
||||
/// Handle a Future's Drop. The callback is only called if the future has not
|
||||
/// yet returned and would be cancelled, and it receives the future as an
|
||||
/// argument
|
||||
pub fn cancel_cleanup<Fut: Future + Unpin, Fun: FnOnce(Fut)>(
|
||||
pub fn cancel_cleanup<Fut: Future + Unpin, Fun: FnOnce(Fut) + Unpin>(
|
||||
fut: Fut,
|
||||
on_drop: Fun,
|
||||
) -> CancelCleanup<Fut, Fun> {
|
||||
|
||||
@@ -91,13 +91,9 @@ impl<A: OwnedAtom> AtomOps for OwnedAtomOps<A> {
|
||||
}
|
||||
fn call(&self, AtomCtx(_, id): AtomCtx, arg: Expr) -> LocalBoxFuture<'_, GExpr> {
|
||||
Box::pin(async move {
|
||||
eprintln!("before debug msg for {}", type_name::<A>());
|
||||
let fun_s = take_first(&AtomReadGuard::new(id.unwrap()).await.dyn_print().await, false);
|
||||
eprintln!("after arg1 {arg:?}");
|
||||
let arg_s = take_first(&arg.print(&FmtCxImpl::default()).await, true);
|
||||
eprintln!("after arg2");
|
||||
writeln!(log("msg"), "owned call {} {}", fun_s, arg_s,);
|
||||
eprintln!("after debug msg");
|
||||
take_atom(id.unwrap()).await.dyn_call(arg).await
|
||||
})
|
||||
}
|
||||
|
||||
@@ -130,10 +130,13 @@ impl ExtensionBuilder {
|
||||
/// on the other side drops the extension. Extension authors would typically
|
||||
/// pass the prepared builder into some function that is responsible for
|
||||
/// managing the [ExtPort]
|
||||
pub async fn run(self, mut ctx: ExtPort) {
|
||||
let spawner = ctx.spawn.clone();
|
||||
pub async fn run(
|
||||
self,
|
||||
ExtPort { mut input, mut output, spawn: spawner, log: extport_log }: ExtPort,
|
||||
) {
|
||||
with_logger(NullLogger, async { std::mem::drop(extport_log) }).await;
|
||||
let host_header =
|
||||
with_logger(NullLogger, api::HostHeader::decode(ctx.input.as_mut())).await.unwrap();
|
||||
with_logger(NullLogger, api::HostHeader::decode(input.as_mut())).await.unwrap();
|
||||
if let Some(url) = host_header.checkers_url {
|
||||
init_telemetry(&url, "orchid", Role::Library);
|
||||
}
|
||||
@@ -143,13 +146,13 @@ impl ExtensionBuilder {
|
||||
.collect_vec();
|
||||
let ext_header = api::ExtensionHeader { name: self.name.to_string(), systems: decls.clone() };
|
||||
with_logger(NullLogger, async {
|
||||
ext_header.encode(ctx.output.as_mut()).await.unwrap();
|
||||
ctx.output.as_mut().flush().await.unwrap();
|
||||
ext_header.encode(output.as_mut()).await.unwrap();
|
||||
output.as_mut().flush().await.unwrap();
|
||||
})
|
||||
.await;
|
||||
let logger1 = LoggerImpl::from_api(&host_header.logger);
|
||||
let logger2 = logger1.clone();
|
||||
let IoComm { client, cx: comm_ctx, srv } = io_comm(ctx.output, ctx.input, None);
|
||||
let IoComm { client, cx: comm_ctx, srv } = io_comm(output, input, None);
|
||||
let client = Rc::new(client);
|
||||
let client2 = client.clone();
|
||||
let spawner2 = spawner.clone();
|
||||
@@ -398,10 +401,8 @@ impl ExtensionBuilder {
|
||||
eprintln!("Blocked on serialize of {}", ret_txt);
|
||||
})
|
||||
};
|
||||
eprintln!("Beginning serialize");
|
||||
let api_expr = serialize(ret).await;
|
||||
bgtask.abort();
|
||||
eprintln!("surpassed serialize");
|
||||
mem::drop(expr_handle);
|
||||
expr_store.dispose().await;
|
||||
handle.reply(call, api_expr).await
|
||||
|
||||
@@ -19,7 +19,7 @@ use orchid_base::deadlock_probe::ThreadlockProbe;
|
||||
use orchid_base::{
|
||||
AtomRepr, Client, ClientExt, CommCx, FmtCxImpl, FmtUnit, Format, IStr, IStrv, IoComm,
|
||||
MsgReaderExt, Pos, ReqHandleExt, ReqReaderExt, ReqWait, SpawnExt, Sym, TaskHandle, Witness, es,
|
||||
ev, io_comm, is, iv, stash, take_first, with_stash,
|
||||
ev, io_comm, is, iv, stash, with_stash,
|
||||
};
|
||||
use scoped_log::log;
|
||||
|
||||
@@ -255,31 +255,13 @@ impl Extension {
|
||||
.await
|
||||
},
|
||||
api::ExprReq::ExprPrint(prt @ api::ExprPrint { target }) => {
|
||||
eprintln!("recvd ExprPrint");
|
||||
let receipt = match cx.exprs.get_expr(target) {
|
||||
match cx.exprs.get_expr(target) {
|
||||
None => handle.reply(&prt, FmtUnit::from("EXPIRED_TICKET").to_api()).await,
|
||||
Some(expr) => {
|
||||
let bgtask = {
|
||||
use std::time::Duration;
|
||||
|
||||
use orchid_base::SpawnExt;
|
||||
let text = format!("{expr:?}");
|
||||
let spawner2 = cx.spawn();
|
||||
cx.spawn().spawn(async move {
|
||||
spawner2.sleep(Duration::from_secs(5)).await;
|
||||
eprintln!("Blocked on exprPrint of {}", text);
|
||||
})
|
||||
};
|
||||
let fmt_unit = expr.print(&FmtCxImpl::default()).await;
|
||||
eprintln!("Responding to exprPrint with {fmt_unit:?}");
|
||||
eprintln!("Which evaluates to {}", take_first(&fmt_unit, true));
|
||||
let receipt = handle.reply(&prt, fmt_unit.to_api()).await;
|
||||
bgtask.abort();
|
||||
receipt
|
||||
handle.reply(&prt, fmt_unit.to_api()).await
|
||||
},
|
||||
};
|
||||
eprintln!("Fulfilled exprPrint");
|
||||
receipt
|
||||
}
|
||||
},
|
||||
api::ExprReq::Create(cre) => {
|
||||
let req = Witness::of(&cre);
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
#![allow(refining_impl_trait)]
|
||||
#![allow(clippy::single_range_in_vec_init)]
|
||||
mod macros;
|
||||
mod std;
|
||||
|
||||
|
||||
@@ -93,7 +93,6 @@ pub async fn gen_match_macro_lib() -> Vec<GenMember> {
|
||||
cx.recur_call(mactree!("pattern::_row" "push" line_mac.own().await)).await?;
|
||||
rule_atoms.push((matcher, body));
|
||||
}
|
||||
eprintln!("Before base case created");
|
||||
let base_case = lam(async |_| {
|
||||
bot(mk_errv(is!("No branches match"), "None of the patterns matches this value", [
|
||||
rules.pos(),
|
||||
@@ -102,7 +101,6 @@ pub async fn gen_match_macro_lib() -> Vec<GenMember> {
|
||||
.await
|
||||
.create()
|
||||
.await;
|
||||
eprintln!("after base case created");
|
||||
let match_chain = stream::iter(rule_atoms.into_iter().rev())
|
||||
.fold(mactree!("Val" base_case), async |tail, (mat, body)| {
|
||||
mactree!("l" "pattern::x" (
|
||||
@@ -115,7 +113,6 @@ pub async fn gen_match_macro_lib() -> Vec<GenMember> {
|
||||
))
|
||||
})
|
||||
.await;
|
||||
eprintln!("after match chain created");
|
||||
Ok(mactree!("push" match_chain; "push" value))
|
||||
},
|
||||
)
|
||||
|
||||
@@ -58,11 +58,16 @@ impl<F: Future, D: Decorator> Future for DecoratedFuture<F, D> {
|
||||
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
|
||||
let Self { inner, decorator } = unsafe { Pin::get_unchecked_mut(self) };
|
||||
let inner = inner.as_mut().expect("This is only unset on drop");
|
||||
// SAFETY: project pin to inner, this needs to be respected in drop too
|
||||
decorator.run(|| unsafe { Pin::new_unchecked(inner) }.poll(cx))
|
||||
}
|
||||
}
|
||||
impl<F, D: Decorator> Drop for DecoratedFuture<F, D> {
|
||||
fn drop(&mut self) { self.decorator.run(|| std::mem::drop(self.inner.take())) }
|
||||
fn drop(&mut self) {
|
||||
// SAFETY: since inner may have been pinned, it's important that we don't move
|
||||
// it with something like Option::take
|
||||
self.decorator.run(|| self.inner = None)
|
||||
}
|
||||
}
|
||||
|
||||
/// Borrow a type from another with a stackframe in between to hold any guards.
|
||||
|
||||
Reference in new issue
Block a user