{"record":{"id":"776d7954f76f8da6","repo":"tracel-ai/burn","slug":"autobatcher-should-be-able-to-send-current-policy","errorCode":null,"errorMessage":"Autobatcher should be able to send current policy state.","messagePattern":"Autobatcher should be able to send current policy state\\.","errorType":"panic","errorClass":null,"httpStatus":null,"severity":"critical","filePath":"crates/burn-rl/src/policy/async_policy.rs","lineNumber":220,"sourceCode":"    ///\n    /// # Arguments\n    ///\n    /// * `autobatch_size` - Number of observations to accumulate before running a pass of inference.\n    /// * `inner_policy` - The policy used to take actions.\n    pub fn new(autobatch_size: usize, inner_policy: P) -> Self {\n        let (sender, receiver) = std::sync::mpsc::channel();\n        let mut autobatcher = PolicyInferenceServer::new(autobatch_size, inner_policy.clone());\n        spawn(move || {\n            loop {\n                match receiver.recv() {\n                    Ok(msg) => match msg {\n                        InferenceMessage::ActionMessage(item) => autobatcher.push_action(item),\n                        InferenceMessage::ForwardMessage(item) => autobatcher.push_logits(item),\n                        InferenceMessage::PolicyUpdate(update) => autobatcher.update_policy(update),\n                        InferenceMessage::ToDevice(device) => autobatcher.policy_to_device(&device),\n                        InferenceMessage::PolicyRequest(sender) => sender\n                            .send(autobatcher.state())\n                            .expect(\"Autobatcher should be able to send current policy state.\"),\n                        InferenceMessage::IncrementAgents(num) => autobatcher.increment_agents(num),\n                        InferenceMessage::DecrementAgents(num) => autobatcher.decrement_agents(num),\n                    },\n                    Err(err) => {\n                        log::error!(\"Error in AsyncPolicy : {}\", err);\n                        break;\n                    }\n                }\n            }\n        });\n\n        Self {\n            inference_state_sender: sender,\n        }\n    }\n\n    /// Increment the number of agents using the inference server.\n    pub fn increment_agents(&self, num: usize) {","sourceCodeStart":202,"sourceCodeEnd":238,"githubUrl":"https://github.com/tracel-ai/burn/blob/d16f7ba2ed0d41408189384044cc886fb4c8f957/crates/burn-rl/src/policy/async_policy.rs#L202-L238","documentation":"This panic fires inside the dedicated autobatcher thread when it handles a PolicyRequest: it calls autobatcher.state() and tries to send the resulting PolicyState back to the caller over the one-shot mpsc channel. The expect triggers when that receiver has already been dropped — i.e. the AsyncPolicy::state() caller gave up (its thread ended or the channel was dropped) before the autobatcher got around to answering. Because the expect is inside the server loop, this panic kills the whole inference thread, after which every other AsyncPolicy method will also panic.","triggerScenarios":"Calling AsyncPolicy::state() from a scope where the returned mpsc::Receiver can be dropped before the autobatcher processes the PolicyRequest — e.g. calling state() on a cloned AsyncPolicy inside a short-lived worker thread that is cancelled/timed out, or wrapping state() in a timeout that abandons the receiver while the inner policy's state() call (possibly a device copy) is still pending.","commonSituations":"RL training loops that snapshot policy state from rayon/tokio worker tasks with timeouts; dropping an AsyncPolicy clone while a PolicyRequest is still queued behind a large action batch, so the reply channel dies before the server reaches the request; panics or unwinding in the calling thread between send and recv.","solutions":["Ensure the thread that calls state() blocks on receiver.recv() until it gets a reply — do not drop the AsyncPolicy handle or unwind before recv() returns.","Wrap the autobatcher-side send so a dead reply channel does not kill the inference thread: replace .expect(...) with if let Err(err) = sender.send(...) { log::warn!(...); } since a dropped receiver is non-fatal for the server.","Check for earlier panics/logs from the autobatcher thread (e.g. inner_policy.state() failing) that delay or abort the reply.","If using timeouts around state(), poll recv_timeout on the caller side but keep the receiver alive, or re-issue the PolicyRequest after a timeout instead of abandoning it."],"exampleFix":"// before\nInferenceMessage::PolicyRequest(sender) => sender\n    .send(autobatcher.state())\n    .expect(\"Autobatcher should be able to send current policy state.\"),\n// after\nInferenceMessage::PolicyRequest(sender) => {\n    if let Err(err) = sender.send(autobatcher.state()) {\n        log::warn!(\"Policy requester dropped before receiving state: {}\", err);\n    }\n}","handlingStrategy":"try-catch","validationCode":"// Caller side: keep the receiver alive until the reply arrives\nlet (tx, rx) = std::sync::mpsc::channel();\npolicy_handle_send(PolicyRequest(tx)); // must not drop tx/rx before recv\nassert!(!std::thread::current().is_panicking());","typeGuard":null,"tryCatchPattern":"let state = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| policy.state()))\n    .map_err(|_| anyhow::anyhow!(\"inference server thread died while snapshotting state\"))?;","preventionTips":["Never drop the AsyncPolicy handle or return from the calling thread before receiver.recv() completes.","Avoid wrapping state() in fire-and-forget tasks or timeouts that abandon the reply channel.","Replace the autobatcher-side reply .expect with a logged error so a dropped receiver cannot kill the inference thread.","Monitor logs for earlier autobatcher-thread panics; they are the usual precursor to reply-channel failures."],"tags":["rust","mpsc-channel","panics","concurrency"],"backgroundTag":"mpsc-receiver-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"}