Async loading

bseq/__init__.py

@@ -3,7 +3,7 @@
 from .properties import BSEQ_scene_property, BSEQ_obj_property, BSEQ_mesh_property
 from .panels import BSEQ_UL_Obj_List, BSEQ_List_Panel, BSEQ_Settings, BSEQ_PT_Import, BSEQ_PT_Import_Child1, BSEQ_PT_Import_Child2, BSEQ_Globals_Panel, BSEQ_Advanced_Panel, BSEQ_Templates, BSEQ_UL_Att_List, draw_template
 from .messenger import subscribe_to_selected, unsubscribe_to_selected
-from .importer import update_obj
+from .callback import load_obj
 from .globals import *    
 
 import bpy
@@ -12,10 +12,10 @@
 
 @persistent
 def BSEQ_initialize(scene):
-    if update_obj not in bpy.app.handlers.frame_change_post:
+    if load_obj not in bpy.app.handlers.frame_change_post:
         # Insert at the beginning, so that it runs before other frame change handlers.
         # The other handlers don't need to be in the first position.
-        bpy.app.handlers.frame_change_post.insert(0, update_obj)
+        bpy.app.handlers.frame_change_post.insert(0, load_obj)
     if auto_refresh_active not in bpy.app.handlers.frame_change_post:
         bpy.app.handlers.frame_change_post.append(auto_refresh_active)
     if auto_refresh_all not in bpy.app.handlers.frame_change_post:

bseq/callback.py

@@ -1,6 +1,9 @@
 import bpy
 import fileseq
 import traceback
+import time
+from .frame_buffer import init as framebuffer_init, terminate as framebuffer_terminate, queue_load as framebuffer_queue_load, flush_buffer as framebuffer_flush_buffer
+from .importer import update_obj
 
 from .utils import show_message_box
 
@@ -60,4 +63,17 @@ def poll_material(self, material):
     return not material.is_grease_pencil
 
 def poll_edit_obj(self, object):
-    return object.BSEQ.init
+    return object.BSEQ.init
+
+def update_framebuffer(self, context) -> None:
+    if self.buffer_next_frame:
+        framebuffer_init()
+    else:
+        framebuffer_terminate()
+
+def load_obj(scene, depsgraph=None):
+    if scene.BSEQ.buffer_next_frame:
+        framebuffer_flush_buffer(scene, depsgraph)
+        framebuffer_queue_load(scene, depsgraph)
+        return
+    update_obj(scene, depsgraph)

bseq/frame_buffer.py

@@ -0,0 +1,172 @@
+import concurrent.futures
+import time
+import bpy
+import meshio
+from .importer import load_into_ram, update_obj, update_mesh, apply_transformation
+from bpy.app.handlers import persistent
+
+_executor: concurrent.futures.ThreadPoolExecutor
+_init = False
+
+# This needs to be persistent to keep the executor
+@persistent
+def init() -> None:
+    global _executor, _init
+    _executor = concurrent.futures.ThreadPoolExecutor(max_workers=5)
+    _executor.submit(pow, 2, 2)
+    _init = True
+    print("init")
+
+class Frame():
+    _future: concurrent.futures.Future
+    _loading_threads: list[concurrent.futures.Future]
+    _buffer_meshes: dict[str, meshio.Mesh]
+    _buffer_data: dict[str, bpy.types.Mesh]
+    _buffer_timings: dict[str, float]
+    _frame: int = -1
+    loading_complete: bool = False
+
+    def __init__(self):
+        self._buffer_meshes = {}
+        self._buffer_data = {}
+        self._buffer_timings = {}
+        self._loading_threads = []
+
+    def _load_data_into_buffer(self, meshio_mesh, object: bpy.types.Object):
+        """ Applies the meshio data to a copy of the object mesh """
+        start_time = time.perf_counter()
+
+        buffer_data = object.data.copy()
+        update_mesh(meshio_mesh, buffer_data)
+        self._buffer_data[object.name_full] = buffer_data
+
+        end_time = time.perf_counter()
+        print("update mesh() took ", (end_time - start_time) * 1000, " ms")
+
+    def _load_buffer_to_data(self, object: bpy.types.Object, meshio_mesh, depsgraph):
+        """ Swaps the object mesh with the buffered mesh """
+        if object.name_full in self._buffer_data:
+            old_mesh = object.data
+            object.data = self._buffer_data[object.name_full]
+            # We remove the old mesh data to prevent memory leaks
+            bpy.data.meshes.remove(old_mesh, do_unlink=False)
+            apply_transformation(meshio_mesh, object, depsgraph)
+
+    def _obj_load(self, obj, scene, depsgraph):
+        """ Buffering Obj Job for the executor """
+        start_time = time.perf_counter()
+        mesh = load_into_ram(obj, scene, depsgraph, target_frame=self._frame)
+        if isinstance(mesh, meshio.Mesh):
+            self._buffer_meshes[obj.name_full] = mesh
+            self._load_data_into_buffer(mesh, obj)
+            end_time = time.perf_counter()
+            # move to main threaded
+            self._buffer_timings[obj.name_full] = (end_time - start_time) * 1000
+
+    def _load_objs(self, scene, depsgraph):
+        """ Job which submits all object buffering jobs
+            Also updates the buffering status in the UI
+        """
+        start = time.perf_counter()
+        self._frame = scene.frame_current + scene.frame_step
+        self._buffer_meshes = {}
+        self._buffer_data = {}
+        self._buffer_timings = {}
+        self._loading_threads = []
+        n_loaded = 0
+        for obj in bpy.data.objects:
+            future = _executor.submit(self._obj_load, obj, scene, depsgraph)
+            self._loading_threads.append(future)
+        for future in concurrent.futures.as_completed(self._loading_threads):
+            n_loaded += 1
+            # Due to multithreading, Blender may forbid writing to the loading status while rendering
+            # In this case, we just skip the update
+            try:
+                scene.BSEQ.loading_status = f"{n_loaded}/{len(bpy.data.objects)}"
+            except Exception as e:
+                print("Skipped updating loading status")
+
+        concurrent.futures.wait(self._loading_threads)
+        scene.BSEQ.loading_status = "Complete"
+        self._loading_threads.clear()
+        end = time.perf_counter()
+        #print("load_objs() took ", (end - start) * 1000, " ms")
+
+    def _clear_buffer(self):
+        if not hasattr(self, "_buffer_meshes") or len(self._buffer_meshes) == 0:
+            print("buffer empty")
+            return
+        self._buffer_meshes.clear()
+        self._buffer_data.clear()
+        self._buffer_timings.clear()
+
+    def flush_buffer(self, scene, depsgraph, *, target_frame: int = -1):
+        """ Applies the buffer to the scene and clears the buffer afterwards
+
+            target_frame -- indicates the current frame which is to be loaded, 
+            '-1' indicates do not use buffered frame
+
+            If target_frame does not coincide with the buffered frame, the buffer will be
+            invalidated and the scene is loaded without pre-buffering  
+
+        """
+        start_time = time.perf_counter()
+        #print()
+        # if no target_frame is specified or if the target_frame does not coincide
+        # with the buffered frame, invalidate the buffer and update the scene serially
+        if target_frame == -1 or self._frame != target_frame:
+            self._frame = -1
+            update_obj(scene, depsgraph)            
+            print("invalidate buffer")
+            for _, obj in self._buffer_data.items():
+                bpy.data.meshes.remove(obj, do_unlink=False)
+            self._clear_buffer()
+            return
+        # Barrier to wait until loading is actually completed
+        concurrent.futures.wait([self._future])
+        for obj in bpy.data.objects:
+            if obj.name_full in self._buffer_meshes:
+                self._load_buffer_to_data(obj, self._buffer_meshes[obj.name_full], depsgraph)
+                obj.BSEQ.last_benchmark = self._buffer_timings[obj.name_full]
+
+        self._clear_buffer()
+        end_time = time.perf_counter()
+        #print("update_scene() took ", (end_time - start_time) * 1000, " ms")
+
+    def queue_load(self, scene, depsgraph):
+        """ Queues the next frame which is determined by the current frame and the set frame step
+
+            Also initialises the executor if not initialized already        
+        """
+        start = time.perf_counter()
+        global _executor, _init
+        if not _init:
+            init()
+
+        self._frame = scene.frame_current + scene.frame_step
+        start_queue = time.perf_counter()
+        self._future = _executor.submit(self._load_objs, scene, depsgraph)
+        scene.BSEQ.loading_status = "Queued"
+        end = time.perf_counter()
+        #print("queue_load():\n\ttotal: ", (end - start) * 1000, "\n\tcopy: ", (copy_end - copy_start) * 1000, "\n\tqueuing:", (end - start_queue) * 1000)
+
+_frame = Frame()
+
+def queue_load(scene, depsgraph=None) -> None:
+    start_time = time.perf_counter()
+    _frame.queue_load(scene, depsgraph)
+    #print("queue_load() took ", (time.perf_counter() - start_time) * 1000, " ms")
+
+def flush_buffer(scene, depsgraph) -> None:
+    _frame.flush_buffer(scene, depsgraph, target_frame=scene.frame_current)
+
+def terminate() -> None:
+    global _init
+    if not _init:
+        return
+    _executor.shutdown(wait=False, cancel_futures=True)
+    _init = False
+    for _, obj in _frame._buffer_data.items():
+        bpy.data.meshes.remove(obj, do_unlink=False)
+    _frame._clear_buffer()
+    print("terminated")