{"record":{"id":"0dcf2be345aacea2","repo":"tracel-ai/burn","slug":"asyncpolicy-should-receive-queued-probabilities","errorCode":null,"errorMessage":"AsyncPolicy should receive queued probabilities.","messagePattern":"AsyncPolicy should receive queued probabilities\\.","errorType":"panic","errorClass":null,"httpStatus":null,"severity":"error","filePath":"crates/burn-rl/src/policy/async_policy.rs","lineNumber":274,"sourceCode":"    type ActionContext = P::ActionContext;\n    type PolicyState = P::PolicyState;\n\n    type Observation = P::Observation;\n    type ActionDistribution = P::ActionDistribution;\n    type Action = P::Action;\n\n    fn forward(&mut self, states: Self::Observation) -> Self::ActionDistribution {\n        let (action_sender, action_receiver) = std::sync::mpsc::channel();\n        let item = ForwardItem {\n            sender: action_sender,\n            inference_state: states,\n        };\n        self.inference_state_sender\n            .send(InferenceMessage::ForwardMessage(item))\n            .expect(\"Should be able to send message to inference_server\");\n        action_receiver\n            .recv()\n            .expect(\"AsyncPolicy should receive queued probabilities.\")\n    }\n\n    fn action(\n        &mut self,\n        states: Self::Observation,\n        deterministic: bool,\n    ) -> (Self::Action, Vec<Self::ActionContext>) {\n        let (action_sender, action_receiver) = std::sync::mpsc::channel();\n        let item = ActionItem {\n            sender: action_sender,\n            inference_state: states,\n            deterministic,\n        };\n        self.inference_state_sender\n            .send(InferenceMessage::ActionMessage(item))\n            .expect(\"should be able to send message to inference_server.\");\n        let action = action_receiver\n            .recv()","sourceCodeStart":256,"sourceCodeEnd":292,"githubUrl":"https://github.com/tracel-ai/burn/blob/d16f7ba2ed0d41408189384044cc886fb4c8f957/crates/burn-rl/src/policy/async_policy.rs#L256-L292","documentation":"After forwarding the observation, AsyncPolicy::forward blocks on action_receiver.recv() and panics if it returns Err — which only happens when every Sender for that reply channel was dropped without sending. The autobatcher's flush_logits drops the sender after send, so an Err means flush_logits never ran for this item: the server thread died (panicked or exited) before processing the queued ForwardMessage, leaving the caller stranded.","triggerScenarios":"The autobatcher thread panicked on an earlier message or inside inner_policy.forward while flushing a batch, so this ForwardItem's reply sender was dropped by unwinding without a value being sent; or the thread exited its loop entirely so the queued ForwardMessage is never processed and the item (with its sender) is leaked/dropped at process exit.","commonSituations":"Deadlock-then-panic patterns: an agent's request waits behind a batch that can never fill (num_agents misconfigured via increment/decrement), and an unrelated panic kills the server, converting every waiter's recv into this panic; GPU faults during batched forward aborting all pending requests.","solutions":["Check autobatcher-thread logs for the root panic (often inside flush_logits or the inner policy) and fix it.","Verify agent accounting: a wrong increment_agents/decrement_agents balance can stall batches forever; keep the counts consistent with the actual number of concurrent callers.","Add resilience: have the server flush pending items or reply with errors during shutdown instead of dropping senders silently.","Consider recv_timeout on the caller side with a clear error/timeout message rather than an opaque panic on a dead channel."],"exampleFix":"// before\naction_receiver\n    .recv()\n    .expect(\"AsyncPolicy should receive queued probabilities.\")\n// after\naction_receiver\n    .recv_timeout(std::time::Duration::from_secs(30))\n    .unwrap_or_else(|err| panic!(\"No probabilities returned from inference server (server may have died): {}\", err))","handlingStrategy":"retry","validationCode":"// Check server responsiveness before blocking indefinitely on a queued forward\nlet probe = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {\n    let mut p = policy.clone();\n    let _ = p.forward(sample_observation.clone());\n}));\nassert!(probe.is_ok(), \"inference server not responding\");","typeGuard":null,"tryCatchPattern":"match action_receiver.recv_timeout(Duration::from_secs(30)) {\n    Ok(dist) => dist,\n    Err(RecvTimeoutError::Timeout) => return Err(anyhow::anyhow!(\"inference timed out; batch possibly stalled\")),\n    Err(RecvTimeoutError::Disconnected) => return Err(anyhow::anyhow!(\"inference server died before returning probabilities\")),\n}","preventionTips":["Use recv_timeout instead of recv to distinguish slow inference from a dead server.","Keep increment_agents/decrement_agents counts accurate so batches always reach the flush threshold.","Fix root panics in flush_logits/inner_policy.forward that strand queued requests.","Prefer graceful server shutdown that flushes or rejects pending items instead of dropping senders."],"tags":["rust","mpsc-channel","panics","deadlock"],"backgroundTag":"channel-sender-dropped","analyzedSha":"d16f7ba2ed0d41408189384044cc886fb4c8f957","analyzedAt":"2026-09-05T13:19:14.260Z","contentChangedAt":"2026-09-05T13:19:14.260Z","schemaVersion":2},"datasetVersion":"2026-09-12T17:17:11.597Z"}