Concurrent execution deadlock investigation

TOKIO RUNTIME (multi-threaded, PID 1 inside guest VM)
┌──────────────────────────────────────────────────────────────┐
│                                                              │
│  Thread A (tokio-runtime-w)        Thread B (any tokio thrd) │
│  ┌───────────────────────────┐     ┌────────────────────────┐│
│  │ ExecService::exec()       │     │ (doing any work)       ││
│  │   ↓                       │     │   ↓                    ││
│  │ ContainerExecutor::spawn()│     │ Vec::push() / String   ││
│  │   ↓                       │     │ ::from() / format!()   ││
│  │ container.lock().await    │     │   ↓                    ││
│  │   ↓                       │     │ malloc()               ││
│  │ spawn_blocking {          │     │   ↓                    ││
│  │   builder.build()         │     │ lock(__malloc_lock) ◄━━━━━ HOLDS
│  │     ↓                    │     │ 0x16f6e68 = 0x02      ││
│  │   tenant_builder.build() │     │   ↓                    ││
│  │     ↓                    │     │ (memcpy, split, etc.)  ││
│  │   builder_impl.create()  │     │   ↓                    ││
│  │     ↓                    │     │ unlock(__malloc_lock)   ││
│  │   run_container()        │     └────────────────────────┘│
│  │     ↓                    │                                │
│  │   container_main_process │                                │
│  │     ↓                    │                                │
│  │   socketpair() x3        │  Creates 6 fds:               │
│  │   (SEQPACKET+CLOEXEC)    │    main_sender    (ms)        │
│  │     ↓                    │    main_receiver   (mr)        │
│  │                          │    inter_sender   (is)        │
│  │   ════════════════════════════════════════════════════    │
│  │   clone3()  ← FORK ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━    │
│  │   ════════════════════════════════════════════════════    │
│  │     ↓ (parent path)      │                                │
│  │   close(ms) ✓            │                                │
│  │   close(is) ✓            │                                │
│  │     ↓                    │                                │
│  │   mr.recv() ━━━━━━━━━━━━━━━━━━━━━ BLOCKS FOREVER        │
│  │   (recvmsg on SEQPACKET) │   (peer ms still open in     │
│  │   (syscall 212, aarch64) │    child process)             │
│  └───────────────────────────┘                               │
└──────────────────────────────────────────────────────────────┘

═══════════════════════════════ clone3() boundary ══════════════

INTERMEDIATE PROCESS (PID 248, single-threaded child)
┌──────────────────────────────────────────────────────────────┐
│                                                              │
│  Inherited from fork:                                        │
│  ┌────────────────────────────────────────────────────────┐  │
│  │ • ALL 6 channel fds (ms, mr, is, ir, xs, xr)          │  │
│  │ • __malloc_lock at 0x16f6e68 = 0x80000002 (LOCKED)     │  │
│  │ • Thread B DOES NOT EXIST in this process              │  │
│  │ • musl has NO pthread_atfork to reset __malloc_lock    │  │
│  └────────────────────────────────────────────────────────┘  │
│                                                              │
│  container_intermediate_process()                            │
│    ↓                                                         │
│  FIRST LINE OF CODE that does heap allocation:               │
│    Vec::new(), String::from(), format!(), PathBuf, Box, etc. │
│    ↓                                                         │
│  malloc() → lock(__malloc_lock) → futex(FUTEX_WAIT_PRIVATE)  │
│    ↓                                                         │
│  ═══════════ DEADLOCK ═══════════                            │
│  Lock owner (Thread B) does not exist in child.              │
│  futex will NEVER be woken.                                  │
│                                                              │
│  Consequences:                                               │
│    ✗ NEVER sends intermediate_ready to parent                │
│    ✗ NEVER forks init process                                │
│    ✗ NEVER closes inherited channel fds:                     │
│        fd=23 → main_sender   (peer of parent's mr)           │
│        fd=25 → inter_sender  (peer of parent's ir)           │
│    ✗ NEVER exits                                             │
│                                                              │
│  Parent's recv() consequence:                                │
│    poll(mr) → revents=0x0 (peer alive, no POLLHUP)           │
│    FIONREAD → 0 bytes available                              │
│    → recvmsg blocks forever                                  │
└──────────────────────────────────────────────────────────────┘

Thread B:    ──────[ malloc ]───────[ malloc ]───────[ malloc ]──
                      ▲                                 ▲
                  lock held                         lock held
                  (few μs)                          (few μs)

clone3():    ─────────┼───────────────┼──────────────────┼──────
                   DEADLOCK          SAFE             DEADLOCK

Upstream youki:   youki CLI (single-threaded) → clone3() → safe ✓
BoxLite guest:    tokio runtime (multi-threaded) → spawn_blocking → clone3() → DEADLOCK ✗

# Requires: pip install boxlite   (latest published version)
# Platform: Linux+KVM or macOS (Apple Hypervisor.framework)
# Purpose: concurrent exec regression smoke test — after the fix all should return reliably, with no hang
import asyncio

from boxlite import SimpleBox
from boxlite.errors import BoxliteError

# The deadlock was originally triggered at concurrency 8; raise the pressure slightly here
CONCURRENCY = 16

async def main() -> None:
    # SimpleBox is an async context manager; the box is created lazily and only starts on entering async with
    try:
        async with SimpleBox(image="alpine:latest") as box:
            async def one(i: int) -> str:
                # SimpleBox.exec timeout parameter is timeout (float, seconds); env takes a dict (not a list)
                result = await box.exec("echo", f"hello_{i}", timeout=30.0)
                # a non-zero exit code does not raise; check exit_code yourself
                if result.exit_code != 0:
                    raise RuntimeError(
                        f"task {i} exit_code={result.exit_code} stderr={result.stderr!r}"
                    )
                return result.stdout.strip()

            # Launch CONCURRENCY execs at once; before the fix this had a ~30-50% chance of hanging forever
            outputs = await asyncio.gather(*(one(i) for i in range(CONCURRENCY)))
            print(f"all {len(outputs)} execs returned, e.g. {outputs[:3]}")
    except BoxliteError as exc:
        # wrapper-layer error (parent of ExecError/TimeoutError/ParseError)
        print(f"BoxLite error: {exc}")
    except RuntimeError as exc:
        # Note: image pull failure / no virtualization raise a standard RuntimeError, not BoxliteError
        print(f"runtime error (image pull / virtualization?): {exc}")

if __name__ == "__main__":
    asyncio.run(main())

# Historical reproduction script (from that time); the current version is fixed, so it should pass reliably
# Inside the repository prefer make targets, not cargo directly
for i in $(seq 1 10); do
  cargo test -p boxlite --features krun \
    --test execution_shutdown test_concurrent_exec_high_concurrency \
    -- --nocapture 2>&1
done

src/guest/src/service/exec/executor.rs   → ContainerExecutor::spawn()
src/guest/src/container/command.rs       → build_and_spawn()
libcontainer/tenant_builder.rs           → TenantContainerBuilder::build()
libcontainer/builder_impl.rs             → ContainerBuilderImpl::create() → run_container()
libcontainer/container_main_process.rs   → container_main_process()
libcontainer/process/channel.rs          → channel pairs (SEQPACKET)
libcontainer/process/fork.rs             → clone3() without CLONE_FILES

let done = Arc::new(AtomicBool::new(false));
let done_clone = done.clone();
let parent_tid = nix::unistd::gettid().as_raw();

let watchdog = std::thread::spawn(move || {
    std::thread::sleep(Duration::from_secs(3));
    if done_clone.load(Ordering::Relaxed) { return; }
    eprintln!("[guest-diag] watchdog: build() still running after 3s");
    // ... diagnostic code ...
});

let wchan = std::fs::read_to_string(
    format!("/proc/self/task/{}/wchan", parent_tid)
);
let syscall = std::fs::read_to_string(
    format!("/proc/self/task/{}/syscall", parent_tid)
);

wchan   = __skb_wait_for_more_packets
syscall = 212 0x18 ...    (212 = recvmsg on aarch64, 0x18 = fd 24)

if let Ok(entries) = std::fs::read_dir("/proc/self/fd") {
    for fd in sorted_fds {
        let link = std::fs::read_link(format!("/proc/self/fd/{}", fd));
        eprintln!("[guest-diag]   fd={}: {:?}", fd, link);
    }
}

fd=22: socket:[1243]    ← NotifyListener (STREAM)
fd=24: socket:[1245]    ← BLOCKED (main_receiver, SEQPACKET)
fd=26: socket:[1247]    ← channel socket
fd=27: socket:[1248]    ← channel socket
fd=28: socket:[1249]    ← channel socket

let mut socket_inodes: Vec<String> = Vec::new();
// ... extract inodes from the readlink results ...

if let Ok(content) = std::fs::read_to_string("/proc/net/unix") {
    for line in content.lines() {
        if socket_inodes.iter().any(|ino| line.contains(ino)) {
            eprintln!("[guest-diag]   {}", line);
        }
    }
}

Inode  Type  St  RefCount
1243   0001  01  00000002   tenant-notify-*.sock (STREAM, listening)
1245   0005  03  00000003   SEQPACKET, connected
1247   0005  03  00000003   SEQPACKET, connected
1248   0005  03  00000003   SEQPACKET, connected
1249   0005  03  00000003   SEQPACKET, connected

if let Some(fd) = blocked_fd {
    let mut pfd = nix::libc::pollfd {
        fd, events: nix::libc::POLLIN | nix::libc::POLLHUP | nix::libc::POLLERR, revents: 0
    };
    let ret = unsafe { nix::libc::poll(&mut pfd, 1, 0) };
    // ... also ioctl(FIONREAD) ...
}

poll_ret=0  revents=0x0     ← no event, peer alive (no POLLHUP)
FIONREAD ioctl_ret=0 bytes=0   ← no data to read

if line.contains(" 0005 ")
    || socket_inodes.iter().any(|ino| line.contains(ino)) {
    eprintln!("[guest-diag]   {}", line);
}

Inode  RefCount  Notes
1244   00000003  ← new! peer socket (main_sender), not in our fd table
1245   00000003  ← our blocked main_receiver
1246   00000003  ← new! peer socket (inter_sender), not in our fd table
1247   00000003
1248   00000003
1249   00000003

if let Ok(proc_entries) = std::fs::read_dir("/proc") {
    for pe in proc_entries.flatten() {
        // ... for each numeric PID ...
        let fd_dir = pe.path().join("fd");
        if let Ok(fd_entries) = std::fs::read_dir(&fd_dir) {
            for fe in fd_entries.flatten() {
                if let Ok(link) = std::fs::read_link(fe.path()) {
                    // if link has the form "socket:[NNNN]", print pid/fd/link
                    eprintln!("pid=... fd=... -> {:?}", link);
                }
            }
        }
    }
}

pid=212 fd=24 -> socket:[1245]    ← our blocked main_receiver
pid=212 fd=26 -> socket:[1247]
pid=212 fd=27 -> socket:[1248]
pid=212 fd=28 -> socket:[1249]

pid=248 fd=22 -> socket:[1244]    ← extra! main_sender peer
pid=248 fd=23 -> socket:[1245]
pid=248 fd=24 -> socket:[1246]    ← extra! inter_sender peer
pid=248 fd=25 -> socket:[1247]
pid=248 fd=26 -> socket:[1248]
pid=248 fd=27 -> socket:[1249]

let comm = std::fs::read_to_string(proc_dir.join("comm"));
let wchan = std::fs::read_to_string(proc_dir.join("wchan"));
let syscall = std::fs::read_to_string(proc_dir.join("syscall"));
let stack = std::fs::read_to_string(proc_dir.join("stack"));

USERSPACE pid=248 ppid=211 state=S comm=tokio-runtime-w
  wchan=futex_wait_queue
  syscall=98 0x16f6e68 0x80 0xffffffff80000002
  stack:
    futex_wait_queue+0x6c/0x98
    __futex_wait+0xb4/0x12c
    futex_wait+0x64/0xcc
    do_futex+0xf8/0x1a0
    __arm64_sys_futex+0xd0/0x14c
    invoke_syscall+0x48/0x10c

llvm-nm --numeric-sort \
    target/aarch64-unknown-linux-musl/debug/boxlite-guest \
  | grep -B5 -A5 16f6e68

00000000016f6de8 B __libc
00000000016f6e50 B __hwcap
00000000016f6e58 B __eintr_valid_flag
00000000016f6e5c B __thread_list_lock
00000000016f6e60 B __abort_lock
00000000016f6e68 B __malloc_lock          ← exact match
00000000016f6e6c B __malloc_replaced
00000000016f6e70 B __aligned_alloc_replaced
00000000016f6e78 B __bss_end__

BEFORE clone3():
  Parent process memory:
  ┌─────────────────────────────────────┐
  │  0x16f6e68 (__malloc_lock) = LOCKED │ ← held by Thread B
  └─────────────────────────────────────┘
       Thread A         Thread B
       (calls clone3)   (doing malloc)

AFTER clone3():
  Parent (unchanged)          Child (COPY of parent memory)
  ┌───────────────────┐      ┌───────────────────┐
  │ __malloc_lock=LOCKED│      │ __malloc_lock=LOCKED│
  │ Thread B exists ✓  │      │ Thread B GONE ✗    │
  │ → will unlock      │      │ → stuck forever    │
  └───────────────────┘      └───────────────────┘

container_main_process() creates 3 channel pairs (6 SEQPACKET sockets):

  Parent side        Peer (child side)     Purpose
  ─────────────────────────────────────────────────────────────
  main_receiver  ←→  main_sender          children → parent readiness
  inter_sender   ←→  inter_receiver       parent → intermediate config
  init_sender    ←→  init_receiver        parent → init config

Fork lifecycle:
  1. Parent creates all 3 pairs (6 fds)
  2. clone3() → intermediate (inherits copies of all 6 fds)
  3. Parent closes main_sender, inter_sender (2 fds)
  4. Parent calls main_receiver.recv() —— waits for intermediate_ready
  5. Intermediate should: setup → send intermediate_ready → fork init → exit
  6. Init should: setup → send init_ready → exec() (CLOEXEC closes channel fds)

When intermediate deadlocks at step 5:
  - The main_sender copy held by intermediate stays open
  - Parent's recv() never reaches EOF (peer alive)
  - Parent blocks forever

src/guest/src/container/command.rs                               (MODIFIED — watchdog)
src/guest/src/service/exec/executor.rs                           (ContainerExecutor::spawn)
src/guest/src/service/exec/mod.rs
libcontainer/src/channel.rs                                      (base channel, socketpair)
libcontainer/src/process/channel.rs                              (Main/Inter/Init channels)
libcontainer/src/process/container_main_process.rs               (fork + channel lifecycle)
libcontainer/src/process/container_intermediate_process.rs       (intermediate setup)
libcontainer/src/process/fork.rs                                 (clone3, CLONE_PARENT)
libcontainer/src/container/tenant_builder.rs                     (build() entry point)
libcontainer/src/container/builder_impl.rs                       (run_container, NotifyListener)
libcontainer/src/notify_socket.rs                                (dangerous Clone impl)