{"record":{"id":"697486dd2f12f2ac","repo":"tracel-ai/burn","slug":"can-send-message-to-autobatcher","errorCode":null,"errorMessage":"Can send message to autobatcher.","messagePattern":"Can send message to autobatcher\\.","errorType":"panic","errorClass":null,"httpStatus":null,"severity":"error","filePath":"crates/burn-rl/src/policy/async_policy.rs","lineNumber":241,"sourceCode":"                    },\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) {\n        self.inference_state_sender\n            .send(InferenceMessage::IncrementAgents(num))\n            .expect(\"Can send message to autobatcher.\")\n    }\n\n    /// Decrement the number of agents using the inference server.\n    pub fn decrement_agents(&self, num: usize) {\n        self.inference_state_sender\n            .send(InferenceMessage::DecrementAgents(num))\n            .expect(\"Can send message to autobatcher.\")\n    }\n}\n\nimpl<P> Policy for AsyncPolicy<P>\nwhere\n    P: Policy + Send + 'static,\n{\n    type ActionContext = P::ActionContext;\n    type PolicyState = P::PolicyState;\n\n    type Observation = P::Observation;","sourceCodeStart":223,"sourceCodeEnd":259,"githubUrl":"https://github.com/tracel-ai/burn/blob/d16f7ba2ed0d41408189384044cc886fb4c8f957/crates/burn-rl/src/policy/async_policy.rs#L223-L259","documentation":"AsyncPolicy::increment_agents sends an IncrementAgents message to the autobatcher thread via a std::sync::mpsc channel and panics if send fails. send only fails when the receiving side of the channel has been closed, which happens once the autobatcher thread has exited — either because all senders were dropped (it saw a RecvError and broke out of its loop) or because it panicked on an earlier message. After this panic the AsyncPolicy handle is permanently unusable.","triggerScenarios":"Calling increment_agents on an AsyncPolicy after the inference thread has terminated: an earlier expect panic in the autobatcher loop (e.g. a failed action reply send or a panic in inner_policy.action/forward), or every clone of the AsyncPolicy being dropped and recreated so the old thread exited with RecvError while a stale clone is still used.","commonSituations":"Multi-threaded RL environments where one worker's panic poisons the shared inference server and remaining workers then panic on their next increment_agents/action call; holding an AsyncPolicy clone across a restart of the training loop; a previous operation panicked inside the inner policy (e.g. shape mismatch) killing the server thread.","solutions":["Check logs for the original autobatcher-thread panic or 'Error in AsyncPolicy' message; fix the root cause that killed the thread first.","Recreate the AsyncPolicy (AsyncPolicy::new) once the server thread has died — a dead handle cannot be revived.","Avoid panics in the inner policy (validate observation shapes/batching) so the server thread never unwinds.","If a panic in the server thread is expected to be possible, wrap the thread body in catch_unwind or replace expects with error logging plus a restart mechanism."],"exampleFix":"// before\nlet policy = policy.clone(); // stale clone from previous run; server thread already dead\npolicy.increment_agents(1);\n// after\nlet policy = AsyncPolicy::new(autobatch_size, inner_policy.clone()); // rebuild after server death\npolicy.increment_agents(1);","handlingStrategy":"fallback","validationCode":"// Before using a long-lived AsyncPolicy, verify the server is responsive\nlet probe = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {\n    let mut p = policy.clone();\n    p.increment_agents(0);\n}));\nif probe.is_err() { rebuild_policy(); }","typeGuard":null,"tryCatchPattern":"let ok = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| policy.increment_agents(1)));\nif ok.is_err() {\n    log::warn!(\"inference server dead; rebuilding AsyncPolicy\");\n    policy = AsyncPolicy::new(autobatch_size, inner_policy.clone());\n}","preventionTips":["Fix the root panic that killed the autobatcher thread before reusing any clone of the handle.","Keep at least one AsyncPolicy alive for the whole server lifetime to prevent unintended thread shutdown.","Keep inner-policy inference panic-free (validate observation shapes and devices).","Wrap the server thread body in catch_unwind and restart it on failure."],"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"}