app: video streaming threads and generating draw calls realtime streamed loading

bin/app/Cargo.lock

@@ -345,6 +345,18 @@ dependencies = [
  "syn 1.0.109",
 ]
 
+[[package]]
+name = "async-broadcast"
+version = "0.7.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "435a87a52755b8f27fcf321ac4f04b2802e337c8c4872923137471ec39c37532"
+dependencies = [
+ "event-listener 5.4.1",
+ "event-listener-strategy",
+ "futures-core",
+ "pin-project-lite",
+]
+
 [[package]]
 name = "async-channel"
 version = "1.9.0"
@@ -1436,6 +1448,7 @@ name = "darkfi-app"
 version = "0.1.0"
 dependencies = [
  "android_logger",
+ "async-broadcast",
  "async-channel 2.5.0",
  "async-gen",
  "async-lock",

bin/app/Cargo.toml

@@ -33,6 +33,7 @@ async-channel = "2.5.0"
 easy-parallel = "3.3.1"
 rand = "0.8.5"
 async-lock = "3.4.1"
+async-broadcast = "0.7.2"
 futures = "0.3.31"
 async-recursion = "1.1.1"
 colored = "3.0.0"

bin/app/src/app/schema/test.rs

@@ -258,7 +258,7 @@ pub async fn make(app: &App, window: SceneNodePtr, i18n_fish: &I18nBabelFish) {
     node.set_property_str(atom, Role::App, "path", "assets/forest/forest_{frame}.png").unwrap();
     node.set_property_u32(atom, Role::App, "z_index", 1).unwrap();
     node.set_property_u32(atom, Role::App, "length", 357).unwrap();
-    let node = node.setup(|me| Video::new(me, app.render_api.clone())).await;
+    let node = node.setup(|me| Video::new(me, app.render_api.clone(), app.ex.clone())).await;
     layer_node.link(node);
 
     // Create some text

bin/app/src/gfx/anim.rs

@@ -17,70 +17,99 @@
  */
 
 use async_trait::async_trait;
-use darkfi_serial::{AsyncEncodable, AsyncWrite, Encodable, FutAsyncWriteExt, SerialEncodable};
+use darkfi_serial::{
+    AsyncEncodable, AsyncWrite, Encodable, FutAsyncWriteExt, SerialEncodable, VarInt,
+};
+use parking_lot::RwLock;
 use std::{
-    cell::RefCell,
     collections::HashMap,
     io::Write,
     sync::{
         atomic::{AtomicU32, Ordering},
-        Arc, RwLock,
+        Arc,
     },
 };
 
 use super::{BufferId, DrawCall, GfxDrawCall, TextureId};
 
-// This can be in instruction but also implement encodable
-// maybe just remove trax?
-
-/*
-type FrameOpt = Arc<RwLock<Option<SequenceAnimationFrame>>>;
-
-pub struct SequenceAnimBuffer {
-    frames: Vec<FrameOpt>
-}
-
-impl SequenceAnimBuffer {
-    pub fn new(len: usize) -> Self {
-    }
-}
-*/
-
-#[derive(Debug, Clone, SerialEncodable)]
-pub struct SequenceAnimation {
+#[derive(Debug, Clone)]
+pub struct SeqAnim {
     oneshot: bool,
-    frames: Vec<SequenceAnimationFrame>,
+    frames: Vec<Option<Frame>>,
+    recv_frames: async_channel::Receiver<(usize, Frame)>,
 }
 
-impl SequenceAnimation {
-    pub fn new(oneshot: bool, frames: Vec<SequenceAnimationFrame>) -> Self {
-        //let frames = frames.into_iter().map(|f| Arc::new(RwLock::new(
-        Self { oneshot, frames }
+impl SeqAnim {
+    pub fn new(
+        oneshot: bool,
+        frames: Vec<Option<Frame>>,
+        recv_frames: async_channel::Receiver<(usize, Frame)>,
+    ) -> Self {
+        Self { oneshot, frames, recv_frames }
     }
 
     pub(super) fn compile(
-        self: Self,
+        mut self: Self,
         textures: &HashMap<TextureId, miniquad::TextureId>,
         buffers: &HashMap<BufferId, miniquad::BufferId>,
-    ) -> GfxSequenceAnimation {
+    ) -> GfxSeqAnim {
         let mut frames = Vec::with_capacity(self.frames.len());
-        for gfxframe in self.frames {
-            let duration = std::time::Duration::from_millis(gfxframe.duration as u64);
-            let dc = gfxframe.dc.compile(textures, buffers, 0).unwrap();
-            frames.push(GfxGfxSequenceAnimationFrame { duration, dc });
+        for frame in self.frames {
+            let Some(frame) = frame else {
+                frames.push(None);
+                continue
+            };
+            let duration = std::time::Duration::from_millis(frame.duration as u64);
+            let dc = frame.dc.compile(textures, buffers, 0).unwrap();
+            frames.push(Some(GfxFrame { duration, dc }));
         }
-        GfxSequenceAnimation::new(self.oneshot, frames)
+        GfxSeqAnim::new(self.oneshot, frames, self.recv_frames)
+    }
+}
+
+impl Encodable for SeqAnim {
+    fn encode<S: Write>(&self, s: &mut S) -> std::result::Result<usize, std::io::Error> {
+        let mut len = 0;
+        len += self.oneshot.encode(s)?;
+        // Write frames array
+        /*
+        len += VarInt(self.frames.len() as u64).encode(s)?;
+        for frame in &self.frames {
+            let frame = frame.read();
+            frame.encode(s)?;
+        }
+        */
+        Ok(len)
+    }
+}
+#[async_trait]
+impl AsyncEncodable for SeqAnim {
+    async fn encode_async<W: AsyncWrite + Unpin + Send>(
+        &self,
+        w: &mut W,
+    ) -> std::io::Result<usize> {
+        let mut len = 0;
+        len += self.oneshot.encode_async(w).await?;
+        // Write frames array
+        /*
+        len += VarInt(self.frames.len() as u64).encode_async(w).await?;
+        for frame in &self.frames {
+            let frame = frame.read().clone();
+            frame.encode_async(w).await?;
+        }
+        */
+        Ok(len)
     }
 }
 
 #[derive(Debug, Clone)]
-pub struct SequenceAnimationFrame {
+pub struct Frame {
     /// Duration of this frame in ms
     duration: u32,
     dc: DrawCall,
 }
 
-impl SequenceAnimationFrame {
+impl Frame {
     pub fn new(duration: u32, dc: DrawCall) -> Self {
         Self { duration, dc }
     }
@@ -88,7 +117,7 @@ impl SequenceAnimationFrame {
 
 /// We have to implement this manually due to macro autism.
 /// Since it contains DrawCall that contains Instruction that can contain this.
-impl Encodable for SequenceAnimationFrame {
+impl Encodable for Frame {
     fn encode<S: Write>(&self, s: &mut S) -> std::result::Result<usize, std::io::Error> {
         let mut len = 0;
         len += self.duration.encode(s)?;
@@ -97,7 +126,7 @@ impl Encodable for SequenceAnimationFrame {
     }
 }
 #[async_trait]
-impl AsyncEncodable for SequenceAnimationFrame {
+impl AsyncEncodable for Frame {
     async fn encode_async<W: AsyncWrite + Unpin + Send>(
         &self,
         w: &mut W,
@@ -110,44 +139,63 @@ impl AsyncEncodable for SequenceAnimationFrame {
 }
 
 #[derive(Debug, Clone)]
-pub(super) struct GfxSequenceAnimation {
+pub(super) struct GfxSeqAnim {
     oneshot: bool,
-    frames: Vec<GfxGfxSequenceAnimationFrame>,
-    //incoming_frames: Vec<Arc<RwLock<Option<SequenceAnimationFrame>>>>,
-    state: RefCell<GfxSequenceAnimationState>,
+    frames: Vec<Option<GfxFrame>>,
+    /// Stream frames in
+    recv_frames: async_channel::Receiver<(usize, Frame)>,
+    /// Timer between frames
+    timer: std::time::Instant,
+    current_idx: usize,
 }
 
-impl GfxSequenceAnimation {
-    fn new(oneshot: bool, frames: Vec<GfxGfxSequenceAnimationFrame>) -> Self {
-        Self {
-            oneshot,
-            frames,
-            state: RefCell::new(GfxSequenceAnimationState { timer: None, current_idx: 0 }),
-        }
+impl GfxSeqAnim {
+    fn new(
+        oneshot: bool,
+        frames: Vec<Option<GfxFrame>>,
+        recv_frames: async_channel::Receiver<(usize, Frame)>,
+    ) -> Self {
+        Self { oneshot, frames, recv_frames, timer: std::time::Instant::now(), current_idx: 0 }
     }
 
-    pub fn tick(&self) -> GfxDrawCall {
-        let mut state = self.state.borrow_mut();
+    pub fn tick(
+        &mut self,
+        textures: &HashMap<TextureId, miniquad::TextureId>,
+        buffers: &HashMap<BufferId, miniquad::BufferId>,
+    ) -> Option<GfxDrawCall> {
+        while let Ok((frame_idx, frame)) = self.recv_frames.try_recv() {
+            let duration = std::time::Duration::from_millis(frame.duration as u64);
+            let dc = frame.dc.compile(textures, buffers, 0).unwrap();
+            self.frames[frame_idx] = Some(GfxFrame { duration, dc });
+        }
 
-        let elapsed = state.timer.get_or_insert_with(|| std::time::Instant::now()).elapsed();
-        assert!(state.current_idx < self.frames.len());
-        if elapsed >= self.frames[state.current_idx].duration {
-            state.current_idx = (state.current_idx + 1) % self.frames.len();
+        let elapsed = self.timer.elapsed();
+        assert!(self.current_idx < self.frames.len());
+        let frame = &self.frames[self.current_idx];
+        let Some(frame) = frame else {
+            assert_eq!(self.current_idx, 0);
+            return None
+        };
+
+        let curr_duration = frame.duration;
+        if elapsed >= curr_duration {
+            let next_idx = (self.current_idx + 1) % self.frames.len();
+            // Only advance when the next frame is Some
+            // Otherwise stay on the same frame
+            if self.frames[next_idx].is_some() {
+                self.current_idx = next_idx;
+                // Reset the timer now we changed frame
+                self.timer = std::time::Instant::now();
+            }
         }
 
-        self.frames[state.current_idx].dc.clone()
+        let curr_frame = self.frames[self.current_idx].clone().unwrap();
+        Some(curr_frame.dc)
     }
 }
 
 #[derive(Debug, Clone)]
-struct GfxGfxSequenceAnimationFrame {
+struct GfxFrame {
     duration: std::time::Duration,
     dc: GfxDrawCall,
 }
-
-#[derive(Debug, Clone)]
-struct GfxSequenceAnimationState {
-    /// Timer between frames
-    timer: Option<std::time::Instant>,
-    current_idx: usize,
-}

bin/app/src/gfx/mod.rs

@@ -31,6 +31,7 @@ use miniquad::{
 };
 use parking_lot::Mutex as SyncMutex;
 use std::{
+    cell::RefCell,
     collections::HashMap,
     fs::File,
     io::Write,
@@ -42,7 +43,7 @@ use std::{
 };
 
 pub mod anim;
-use anim::{GfxSequenceAnimation, SequenceAnimation};
+use anim::{GfxSeqAnim, SeqAnim};
 mod favico;
 mod linalg;
 pub use linalg::{Dimension, Point, Rectangle};
@@ -399,7 +400,7 @@ pub enum DrawInstruction {
     SetPos(Point),
     ApplyView(Rectangle),
     Draw(DrawMesh),
-    Animation(SequenceAnimation),
+    Animation(SeqAnim),
     EnableDebug,
 }
 
@@ -418,7 +419,9 @@ impl DrawInstruction {
             Self::Draw(mesh) => {
                 GfxDrawInstruction::Draw(mesh.compile(textures, buffers, debug_str)?)
             }
-            Self::Animation(anim) => GfxDrawInstruction::Animation(anim.compile(textures, buffers)),
+            Self::Animation(anim) => {
+                GfxDrawInstruction::Animation(RefCell::new(anim.compile(textures, buffers)))
+            }
             Self::EnableDebug => GfxDrawInstruction::EnableDebug,
         };
         Some(instr)
@@ -479,7 +482,7 @@ enum GfxDrawInstruction {
     SetPos(Point),
     ApplyView(Rectangle),
     Draw(GfxDrawMesh),
-    Animation(GfxSequenceAnimation),
+    Animation(RefCell<GfxSeqAnim>),
     EnableDebug,
 }
 
@@ -500,6 +503,10 @@ struct RenderContext<'a> {
     scale: f32,
     view: Rectangle,
     cursor: Point,
+
+    // Temp
+    textures: &'a HashMap<TextureId, miniquad::TextureId>,
+    buffers: &'a HashMap<BufferId, miniquad::BufferId>,
 }
 
 impl<'a> RenderContext<'a> {
@@ -633,8 +640,9 @@ impl<'a> RenderContext<'a> {
                     self.ctx.draw(0, mesh.num_elements, 1);
                 }
                 GfxDrawInstruction::Animation(anim) => {
-                    let dc = anim.tick();
-                    self.draw_call(&dc, indent + 1, is_debug);
+                    if let Some(dc) = anim.borrow_mut().tick(self.textures, self.buffers) {
+                        self.draw_call(&dc, indent + 1, is_debug);
+                    }
                 }
                 GfxDrawInstruction::EnableDebug => {
                     if !is_debug {
@@ -1427,6 +1435,8 @@ impl EventHandler for Stage {
             scale: 1.,
             view: Rectangle::from([0., 0., screen_w, screen_h]),
             cursor: Point::from([0., 0.]),
+            textures: &self.textures,
+            buffers: &self.buffers,
         };
         render_ctx.draw();
 

bin/app/src/ui/video.rs

@@ -18,7 +18,7 @@
 
 use async_trait::async_trait;
 use image::ImageReader;
-use parking_lot::Mutex as SyncMutex;
+use parking_lot::{Mutex as SyncMutex, RwLock};
 use rand::{rngs::OsRng, Rng};
 use std::{
     io::Cursor,
@@ -30,7 +30,7 @@ use std::{
 
 use crate::{
     gfx::{
-        anim::{SequenceAnimation, SequenceAnimationFrame},
+        anim::{Frame, SeqAnim},
         gfxtag, DrawCall, DrawInstruction, DrawMesh, ManagedTexturePtr, Rectangle, RenderApi,
     },
     mesh::{MeshBuilder, MeshInfo, COLOR_WHITE},
@@ -42,6 +42,8 @@ use crate::{
 
 use super::{DrawTrace, DrawUpdate, OnModify, UIObject};
 
+pub const N_LOADERS: usize = 4;
+
 macro_rules! t { ($($arg:tt)*) => { trace!(target: "ui::video", $($arg)*); } }
 
 pub type VideoPtr = Arc<Video>;
@@ -50,23 +52,29 @@ pub struct Video {
     node: SceneNodeWeak,
     render_api: RenderApi,
     tasks: SyncMutex<Vec<smol::Task<()>>>,
+    load_tasks: SyncMutex<Vec<smol::Task<()>>>,
+    ex: ExecutorPtr,
     stop_load: Arc<AtomicBool>,
-
-    textures: SyncMutex<Vec<ManagedTexturePtr>>,
     dc_key: u64,
 
+    textures_pub: async_broadcast::Sender<(usize, ManagedTexturePtr)>,
+    textures_sub: async_broadcast::Receiver<(usize, ManagedTexturePtr)>,
+    textures: Arc<SyncMutex<Vec<Option<ManagedTexturePtr>>>>,
+    // Do we need this?
+    _load_handles: SyncMutex<[Option<std::thread::JoinHandle<()>>; N_LOADERS]>,
+
     rect: PropertyRect,
     uv: PropertyRect,
     z_index: PropertyUint32,
     priority: PropertyUint32,
     path: PropertyStr,
-    len: PropertyUint32,
+    vid_len: PropertyUint32,
 
     parent_rect: SyncMutex<Option<Rectangle>>,
 }
 
 impl Video {
-    pub async fn new(node: SceneNodeWeak, render_api: RenderApi) -> Pimpl {
+    pub async fn new(node: SceneNodeWeak, render_api: RenderApi, ex: ExecutorPtr) -> Pimpl {
         t!("Video::new()");
 
         let node_ref = &node.upgrade().unwrap();
@@ -75,23 +83,30 @@ impl Video {
         let z_index = PropertyUint32::wrap(node_ref, Role::Internal, "z_index", 0).unwrap();
         let priority = PropertyUint32::wrap(node_ref, Role::Internal, "priority", 0).unwrap();
         let path = PropertyStr::wrap(node_ref, Role::Internal, "path", 0).unwrap();
-        let len = PropertyUint32::wrap(node_ref, Role::Internal, "length", 0).unwrap();
+        let vid_len = PropertyUint32::wrap(node_ref, Role::Internal, "length", 0).unwrap();
+
+        let (textures_pub, textures_sub) = async_broadcast::broadcast(1);
 
         let self_ = Arc::new(Self {
             node,
             render_api,
             tasks: SyncMutex::new(vec![]),
+            load_tasks: SyncMutex::new(vec![]),
+            ex,
             stop_load: Arc::new(AtomicBool::new(false)),
-
-            textures: SyncMutex::new(vec![]),
             dc_key: OsRng.gen(),
 
+            textures_pub,
+            textures_sub,
+            textures: Arc::new(SyncMutex::new(vec![])),
+            _load_handles: SyncMutex::new([const { None }; 4]),
+
             rect,
             uv,
             z_index,
             priority,
             path,
-            len,
+            vid_len,
 
             parent_rect: SyncMutex::new(None),
         });
@@ -105,37 +120,68 @@ impl Video {
     }
 
     fn load_textures(&self) {
-        let (sendr, recvr) = async_channel::bounded(1);
-        let len = self.len.get();
+        let vid_len = self.vid_len.get() as usize;
         let path_fmt = self.path.get();
-        let render_api = self.render_api.clone();
-        let stop_load = self.stop_load.clone();
-        let handle = std::thread::spawn(move || {
-            let mut textures = Vec::with_capacity(len as usize);
-            let instant = std::time::Instant::now();
-            for i in 0..len {
-                // Stop loading instantly
-                if stop_load.load(Ordering::Relaxed) {
-                    return
-                }
-                t!("i = {i}");
-                let path = path_fmt.replace("{frame}", &format!("{i:#03}"));
-                let texture = Self::load_texture(path, &render_api);
-                textures.push(texture);
-            }
-            t!("elapsed = {:?}", instant.elapsed());
-            sendr.send_blocking(textures).unwrap();
-        });
 
-        // Temp here
-        let Ok(textures) = recvr.recv_blocking() else {
-            let node_ref = &self.node.upgrade().unwrap();
-            t!("loading textures was stopped {node_ref:?}");
-            return
-        };
-        assert!(handle.is_finished());
+        // Starts N threads
+        // batch idxs across threads
+        //    0    1    2    3
+        //    4    5    6    7
+        //           ...
+        // load_texture:
+        //    read image
+        //    create texture
+        //    set texture slot
+        //    broadcast (idx, texture)
+        //
+        // draw_call:
+        //    create broadcast sub
+        //    load as many slots from mutex, then release
+        //    start task:
+        //        loop:
+        //            listen to broadcast
+        //            create draw_call
+        //            send to gfx
+
+        {
+            let mut textures = self.textures.lock();
+            *textures = vec![None; vid_len];
+            self.textures_pub.clone().set_capacity(vid_len);
+        }
 
-        *self.textures.lock() = textures;
+        let mut handles = [const { None }; 4];
+        for thread_idx in 0..N_LOADERS {
+            let path_fmt = path_fmt.clone();
+            let render_api = self.render_api.clone();
+            let stop_load = self.stop_load.clone();
+            let textures = self.textures.clone();
+            let textures_pub = self.textures_pub.clone();
+
+            let handle = std::thread::spawn(move || {
+                let mut frame_idx = thread_idx;
+                while frame_idx < vid_len {
+                    // Stop loading instantly
+                    if stop_load.load(Ordering::Relaxed) {
+                        return
+                    }
+                    t!("frame_idx = {frame_idx} [thread={thread_idx}]");
+                    let path = path_fmt.replace("{frame}", &format!("{frame_idx:#03}"));
+                    let texture = Self::load_texture(path, &render_api);
+                    // Make editing textures array and broadcasting an atomic op
+                    {
+                        let mut textures = textures.lock();
+                        // set texture slot
+                        textures[frame_idx] = Some(texture.clone());
+                        // broadcast
+                        textures_pub.try_broadcast((frame_idx, texture)).unwrap();
+                    }
+
+                    frame_idx += N_LOADERS;
+                }
+            });
+            handles[thread_idx] = Some(handle);
+        }
+        *self._load_handles.lock() = handles;
     }
 
     fn load_texture(path: String, render_api: &RenderApi) -> ManagedTexturePtr {
@@ -198,14 +244,67 @@ impl Video {
         self.uv.eval(atom, &rect).ok()?;
 
         let mesh = self.regen_mesh();
-        let textures = self.textures.lock().clone();
-        assert!(!textures.is_empty());
+        // Begin subscribing before we clone Mutex, but actually
+        // we need the length of the textures stored, so lets just hold it,
+        // do the clones THEN release.
+        let (textures, tsubs) = {
+            let textures = self.textures.lock();
+            // Possibly triggered by race condition.
+            // Check it anyway since generally should work.
+            assert_eq!(textures.len(), self.vid_len.get() as usize);
+            let tsubs = vec![self.textures_sub.clone(); textures.len()];
+            (textures.clone(), tsubs)
+        };
+        assert_eq!(textures.len(), tsubs.len());
+
+        // Only used in this function so fine to hold the entire time
+        let mut load_tasks = self.load_tasks.lock();
+        load_tasks.clear();
+
+        let (send_frames, recv_frames) = async_channel::bounded(textures.len());
 
         let mut frames = Vec::with_capacity(textures.len());
-        for texture in textures {
+        for (texture_idx, (mut texture, mut tsub)) in
+            textures.into_iter().zip(tsubs.into_iter()).enumerate()
+        {
+            let vertex_buffer = mesh.vertex_buffer.clone();
+            let index_buffer = mesh.index_buffer.clone();
+
+            let Some(texture) = texture.take() else {
+                frames.push(None);
+
+                let send_frames = send_frames.clone();
+                let task = self.ex.spawn(async move {
+                    while let Ok((frame_idx, texture)) = tsub.recv().await {
+                        if frame_idx != texture_idx {
+                            continue
+                        }
+
+                        let mesh = DrawMesh {
+                            vertex_buffer,
+                            index_buffer,
+                            texture: Some(texture),
+                            num_elements: mesh.num_elements,
+                        };
+                        let dc = DrawCall {
+                            instrs: vec![DrawInstruction::Draw(mesh)],
+                            dcs: vec![],
+                            z_index: 0,
+                            debug_str: "video",
+                        };
+
+                        // send here
+                        send_frames.send((frame_idx, Frame::new(40, dc)));
+                        break
+                    }
+                });
+                load_tasks.push(task);
+
+                continue
+            };
             let mesh = DrawMesh {
-                vertex_buffer: mesh.vertex_buffer.clone(),
-                index_buffer: mesh.index_buffer.clone(),
+                vertex_buffer,
+                index_buffer,
                 texture: Some(texture),
                 num_elements: mesh.num_elements,
             };
@@ -215,16 +314,17 @@ impl Video {
                 z_index: 0,
                 debug_str: "video",
             };
-            frames.push(SequenceAnimationFrame::new(40, dc));
+            frames.push(Some(Frame::new(40, dc)));
         }
-        let anim = SequenceAnimation::new(false, frames);
+        let anim = SeqAnim::new(false, frames, recv_frames);
 
         Some(DrawUpdate {
             key: self.dc_key,
             draw_calls: vec![(
                 self.dc_key,
                 DrawCall::new(
-                    vec![DrawInstruction::Move(rect.pos()), DrawInstruction::Animation(anim)],
+                    //vec![DrawInstruction::Move(rect.pos()), DrawInstruction::Animation(anim)],
+                    vec![],
                     vec![],
                     self.z_index.get(),
                     "vid",