re_viewer 0.30.0-alpha.2

The Rerun viewer
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100

use arrow::array::RecordBatch;

use re_chunk::{Chunk, TimeInt};
use re_entity_db::EntityDb;
use re_log_types::TimelinePoint;
use re_redap_client::{ApiResult, ConnectionClient};
use re_viewer_context::TimeControl;

pub fn prefetch_chunks_for_active_recording(
    egui_ctx: &egui::Context,
    memory_budget: re_memory::MemoryLimit,
    recording: &mut EntityDb,
    time_ctrl: &TimeControl,
    connection_registry: &re_redap_client::ConnectionRegistryHandle,
) -> Option<()> {
    re_tracing::profile_function!();

    let current_time = time_ctrl.time_i64()?;
    let timeline = *time_ctrl.timeline()?;

    let redap_uri = recording.redap_uri()?.clone();
    let origin = redap_uri.origin.clone();

    // Load data from slightly before the current time to give some room for latest-at.
    // This is a bit hacky, but works for now.
    let before_margin = match timeline.typ() {
        re_log_types::TimeType::Sequence => 30,
        re_log_types::TimeType::DurationNs | re_log_types::TimeType::TimestampNs => 1_000_000_000,
    };
    let start_time = TimeInt::new_temporal(current_time.saturating_sub(before_margin));
    let time_cursor = TimelinePoint::from((timeline, start_time));

    if !recording.can_fetch_chunks_from_redap() {
        return None;
    }

    let options = re_entity_db::ChunkPrefetchOptions {
        total_uncompressed_byte_budget: memory_budget.as_bytes(),
        ..Default::default()
    };

    let (rrd_manifest, storage_engine) = recording.rrd_manifest_index_mut_and_storage_engine();

    let load_chunks = &|rb| {
        egui_ctx.request_repaint();
        let connection_registry = connection_registry.clone();
        let origin = origin.clone();

        let fut = async move {
            let mut client = connection_registry.client(origin).await.map_err(|err| {
                re_log::warn_once!("Failed to connect to remote: {err}");
            })?;
            load_chunks(&mut client, &rb).await.map_err(|err| {
                re_log::warn_once!("{err}");
            })
        };

        // Annoying poll_promise API difference:
        cfg_if::cfg_if! {
            if #[cfg(target_arch = "wasm32")] {
                poll_promise::Promise::spawn_local(fut)
            } else {
                poll_promise::Promise::spawn_async(fut)
            }
        }
    };

    if let Err(err) =
        rrd_manifest.prefetch_chunks(storage_engine.store(), &options, time_cursor, load_chunks)
    {
        re_log::warn_once!("prefetch_chunks failed: {err}");
    }

    None
}

/// Takes a dataframe that looks like an [`re_log_encoding::RrdManifest`] (has a `chunk_key` column).
async fn load_chunks(client: &mut ConnectionClient, batch: &RecordBatch) -> ApiResult<Vec<Chunk>> {
    use tokio_stream::StreamExt as _;

    if batch.num_rows() == 0 {
        return Ok(vec![]);
    }

    re_log::trace!("Requesting {} chunk(s) from server…", batch.num_rows());

    let chunk_stream = client.fetch_segment_chunks_by_id(batch).await?;
    let mut chunk_stream =
        re_redap_client::fetch_chunks_response_to_chunk_and_segment_id(chunk_stream);
    let mut all_chunks = Vec::new();
    while let Some(chunks) = chunk_stream.next().await {
        for (chunk, _partition_id) in chunks? {
            all_chunks.push(chunk);
        }
    }

    re_log::trace!("Finished downloading {} chunk(s).", batch.num_rows());

    Ok(all_chunks)
}