fast3r/fast3r/viz /demo.py - 10a31a6 report "_thread.RLock"

[HengYuD]

Traceback (most recent call last):
File "D:\fast3r\fast3r\viz\demo_init.py", line 1164, in
main()
File "D:\fast3r\fast3r\viz\demo_init.py", line 1158, in main
demo = create_demo(args.checkpoint_dir, args.examples_dir, args.output_dir, device=device)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "D:\fast3r\fast3r\viz\demo_init.py", line 791, in create_demo
global_manager_req_queue, global_manager_resp_queue, manager_process = start_manager()
^^^^^^^^^^^^^^^
File "D:\fast3r\fast3r\viz\demo_init.py", line 255, in start_manager
manager_process.start()
File "D:\python\python311\Lib\multiprocessing\process.py", line 121, in start
self._popen = self._Popen(self)
^^^^^^^^^^^^^^^^^
File "D:\python\python311\Lib\multiprocessing\context.py", line 224, in _Popen
return _default_context.get_context().Process._Popen(process_obj)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "D:\python\python311\Lib\multiprocessing\context.py", line 336, in _Popen
return Popen(process_obj)
^^^^^^^^^^^^^^^^^^
File "D:\python\python311\Lib\multiprocessing\popen_spawn_win32.py", line 95, in init
reduction.dump(process_obj, to_child)
File "D:\python\python311\Lib\multiprocessing\reduction.py", line 60, in dump
ForkingPickler(file, protocol).dump(obj)
TypeError: cannot pickle '_thread.RLock' object

[jedyang97]

Here are some insights from ChatGPT:

This error arises because Windows uses a different process creation mechanism than Linux. On Linux, the default “fork” start method copies the process’s memory, so objects like thread locks (e.g., _thread.RLock) are simply inherited. On Windows, however, the default “spawn” method is used, which requires that all objects passed to the child process be picklable. Thread locks are inherently not picklable, which is why you’re seeing the error.

To address this, you might consider:
• Restructuring your code: Avoid passing objects (or globals) that contain thread locks to child processes.
• Delayed initialization: Create locks or shared resources after the process has started rather than during its creation.
• Using the proper guard: Ensure that your code is under an if name == 'main': block to avoid unintended code execution when the module is imported in a new process.

This is a common issue when porting multiprocessing code from Linux to Windows, and with some adjustments to how resources are shared or initialized, you can resolve it.

So it looks an issue that is specific to Windows. @HengYuD Can you do some digging following this lead? I current don’t have a Windows machine so it is hard for me to debug this. If you find a fix to make the code compatible with windows, please let us know either here or via a PR, and I can test if the fix still works on Linux.

[HengYuD]

Following deepseek's suggestion, I adjusted the following two functions
and it worked

class ViserServerManager:
    @staticmethod
    def run_worker(req_queue, resp_queue):
        """Worker process entry point (static method)"""
        from rich.console import Console 
        console = Console()
        console.log("[bold green]ViserServerManager started[/bold green]")

        servers = {}          # server_id -> server info
        session_servers = {}  # session_id -> list of server_ids
        next_server_id = 1

        while True:
            try:
                cmd = req_queue.get(timeout=1)
            except Exception:
                continue

            message_id = cmd.get("message_id")
            
            if cmd["cmd"] == "launch":
                server_id = next_server_id
                next_server_id += 1
                session_id = cmd.get("session_id", f"default_session_{server_id}")
                
                console.log(f"Launching viser server {server_id} for session {session_id}")
                try:
                    parent_conn, child_conn = mp.Pipe()
                    p = mp.Process(
                        target=run_viser_server,
                        args=(child_conn, cmd["output"], 
                              cmd.get("min_conf_thr_percentile", 10),
                              cmd.get("global_conf_thr_value_to_drop_view", 1.5),
                              cmd.get("point_size", 0.0004))
                    )
                    p.start()
                    child_conn.close()
                    
                    result = parent_conn.recv()
                    if "error" in result:
                        console.log(f"[red]Error: {result['error']}[/red]")
                        resp_queue.put({"cmd": "launch", "error": result["error"], "message_id": message_id})
                        p.terminate()
                    else:
                        servers[server_id] = {
                            "share_url": result["share_url"],
                            "process": p,
                            "session_id": session_id
                        }
                        session_servers.setdefault(session_id, []).append(server_id)
                        resp_queue.put({
                            "cmd": "launch",
                            "server_id": server_id,
                            "share_url": result["share_url"],
                            "message_id": message_id
                        })

                except Exception as e:
                    console.log(f"[red]Launch error: {e}[/red]")
                    resp_queue.put({"cmd": "launch", "error": str(e), "message_id": message_id})

            elif cmd["cmd"] == "terminate_server":
                server_id = cmd["server_id"]
                if server_id in servers:
                    proc = servers[server_id]["process"]
                    session_id = servers[server_id]["session_id"]
                    try:
                        proc.kill()
                        proc.join()
                        session_servers[session_id].remove(server_id)
                        if not session_servers[session_id]:
                            del session_servers[session_id]
                        del servers[server_id]
                        resp_queue.put({"cmd": "terminate_server", "server_id": server_id, "status": "terminated", "message_id": message_id})
                    except Exception as e:
                        resp_queue.put({"cmd": "terminate_server", "error": str(e), "message_id": message_id})
                else:
                    resp_queue.put({"cmd": "terminate_server", "error": "Server not found", "message_id": message_id})

            elif cmd["cmd"] == "terminate_session":
                session_id = cmd["session_id"]
                terminated = []
                errors = []
                if session_id in session_servers:
                    for sid in session_servers[session_id].copy():
                        if sid in servers:
                            try:
                                servers[sid]["process"].kill()
                                servers[sid]["process"].join()
                                del servers[sid]
                                terminated.append(sid)
                            except Exception as e:
                                errors.append(str(e))
                    del session_servers[session_id]
                    resp_queue.put({
                        "cmd": "terminate_session",
                        "terminated_servers": terminated,
                        "errors": errors,
                        "message_id": message_id
                    })
                else:
                    resp_queue.put({"cmd": "terminate_session", "error": "Session not found", "message_id": message_id})

            else:
                resp_queue.put({"cmd": "error", "error": "Unknown command", "message_id": message_id}) 

def start_manager():
        req_queue = mp.Queue()
        resp_queue = mp.Queue()
        manager_process = mp.Process(
            target=ViserServerManager.run_worker,
            args=(req_queue, resp_queue)
        )
        manager_process.start()
        return req_queue, resp_queue, manager_process

[HengYuD]

but windows env with codec error, I tried many ways but failed
seems happened within multiprocessing
I would try WSL2 later

Scene type detection:
- Found 7/7 frames with significant sky presence (>20% sky pixels)
- Scene classified as: outdoor, setting mask_sky to True
(viser) Share URL requested!
[16:58:21] Error launching server: 'gbk' codec can't decode byte   demo2.py:560
           0xa2 in position 71: illegal multibyte sequence   

[jedyang97]

Yeah it seems that some more work is needed to make the demo compatible with Windows. We currently don't have much bandwidth for this, but if you ever figure these out, please consider contribute a PR!

[zxy-bjtu]

I have met the same issue:

Scene type detection:
- Found 7/7 frames with significant sky presence (>20% sky pixels)
- Scene classified as: outdoor, setting mask_sky to True
(viser) Share URL requested!
[16:58:21] Error launching server: 'gbk' codec can't decode byte   demo2.py:560
           0xa2 in position 71: illegal multibyte sequence   

I have solved this issue by setting encoding=utf8 in function handle_feedback and process_images:

def handle_feedback(feedback_type, timestamp, output_dir):
    """
    Move data from no_feedback to the appropriate feedback folder.
    """
    print(f"Feedback type: {feedback_type} received for session: {timestamp}.")
    src_dir = os.path.join(output_dir, "no_feedback", timestamp)
    dst_dir = os.path.join(output_dir, feedback_type, timestamp)
    
    # Check if this is a regular upload (not an example)
    if os.path.exists(src_dir):
        # Move the directory to the appropriate feedback folder
        shutil.move(src_dir, dst_dir)
        
        # Update the metadata file
        metadata_path = os.path.join(dst_dir, "metadata.json")
        if os.path.exists(metadata_path):
            with open(metadata_path, 'r', encoding='utf-8') as f:
                metadata = json.load(f)
            metadata['feedback_type'] = feedback_type
            with open(metadata_path, 'w', encoding='utf-8') as f:
                json.dump(metadata, f, indent=2)
        # print that the feedback was saved
        print(f"Feedback saved to {dst_dir}")
        return True
    
    # Check if this is an example scene
    example_dir = os.path.join(output_dir, "example_scenes", timestamp)
    if os.path.exists(example_dir):
        # For example scenes, just update the metadata without moving
        metadata_path = os.path.join(example_dir, "metadata.json")
        if os.path.exists(metadata_path):
            with open(metadata_path, 'r', encoding='utf-8') as f:
                metadata = json.load(f)
            metadata['feedback_type'] = feedback_type
            with open(metadata_path, 'w', encoding='utf-8') as f:
                json.dump(metadata, f, indent=2)
        print(f"Feedback saved for example scene: {example_dir}")
        return True
        
    return False



def process_images(uploaded_files, video_file, state,
                   model, lit_module, device,
                   global_manager_req_queue, global_manager_resp_queue, 
                   output_dir, examples_dir,
                   image_size=512, rotate_clockwise_90=False, crop_to_landscape=False):
     ...
    with open(os.path.join(save_dir, 'metadata.json'), 'w', encoding='utf-8') as f:
                json.dump(metadata, f, indent=2)
    ...
    with open(os.path.join(example_dir, 'metadata.json'), 'w', encoding='utf-8') as f:
                json.dump(example_metadata, f, indent=2)

then, I got the normal result: