1use std::collections::HashMap;
10use std::ops::Deref;
11use std::path::PathBuf;
12use std::sync::OnceLock;
13use std::time::Duration;
14
15use hyperactor::ActorHandle;
16use hyperactor::Endpoint as _;
17use hyperactor::Gateway;
18use hyperactor::Instance;
19use hyperactor::Proc;
20use hyperactor::channel::ChannelAddr;
21use hyperactor::id::Label;
22use hyperactor_mesh::ProcMeshRef;
23use hyperactor_mesh::bootstrap::BootstrapCommand;
24use hyperactor_mesh::bootstrap::ProcBind;
25use hyperactor_mesh::bootstrap::host;
26use hyperactor_mesh::host_mesh;
27use hyperactor_mesh::host_mesh::HostMesh;
28use hyperactor_mesh::host_mesh::HostMeshRef;
29use hyperactor_mesh::host_mesh::PerRankBootstrapFn;
30use hyperactor_mesh::host_mesh::host_agent::GetLocalProcClient;
31use hyperactor_mesh::host_mesh::host_agent::HostAgent;
32use hyperactor_mesh::host_mesh::host_agent::ShutdownHost;
33use hyperactor_mesh::mesh_admin::MeshAdminMessageClient;
34use hyperactor_mesh::mesh_id::ActorMeshId;
35use hyperactor_mesh::mesh_id::HostMeshId;
36use hyperactor_mesh::mesh_id::ProcMeshId;
37use hyperactor_mesh::proc_agent::GetProcClient;
38use hyperactor_mesh::proc_mesh::ProcRef;
39use hyperactor_mesh::proc_mesh::telemetry_actor_mesh_id;
40use hyperactor_mesh::shared_cell::SharedCell;
41use hyperactor_mesh::transport::default_bind_spec;
42use hyperactor_telemetry::hash_to_u64;
43use ndslice::View;
44use ndslice::view::RankedSliceable;
45use pyo3::IntoPyObjectExt;
46use pyo3::exceptions::PyException;
47use pyo3::exceptions::PyRuntimeError;
48use pyo3::exceptions::PyValueError;
49use pyo3::prelude::*;
50use pyo3::types::PyBytes;
51
52use crate::actor::PythonActor;
53use crate::actor::to_py_error;
54use crate::context::PyInstance;
55use crate::proc_mesh::PyProcMesh;
56use crate::pytokio::PyPythonTask;
57use crate::runtime::GilSite;
58use crate::runtime::monarch_with_gil;
59use crate::runtime::monarch_with_gil_blocking;
60use crate::shape::PyExtent;
61use crate::shape::PyPoint;
62use crate::shape::PyRegion;
63
64#[pyclass(
65 name = "BootstrapCommand",
66 module = "monarch._rust_bindings.monarch_hyperactor.host_mesh"
67)]
68#[derive(Clone)]
69pub struct PyBootstrapCommand {
70 #[pyo3(get, set)]
71 pub program: String,
72 #[pyo3(get, set)]
73 pub arg0: Option<String>,
74 #[pyo3(get, set)]
75 pub args: Vec<String>,
76 #[pyo3(get, set)]
77 pub env: HashMap<String, String>,
78}
79
80#[pymethods]
81impl PyBootstrapCommand {
82 #[new]
83 fn new(
84 program: String,
85 arg0: Option<String>,
86 args: Vec<String>,
87 env: HashMap<String, String>,
88 ) -> Self {
89 Self {
90 program,
91 arg0,
92 args,
93 env,
94 }
95 }
96
97 fn __repr__(&self) -> String {
98 format!(
99 "BootstrapCommand(program='{}', args={:?}, env={:?})",
100 self.program, self.args, self.env
101 )
102 }
103
104 fn with_env(&self, env: HashMap<String, String>) -> Self {
108 let mut new_env = self.env.clone();
109 new_env.extend(env);
110 Self {
111 program: self.program.clone(),
112 arg0: self.arg0.clone(),
113 args: self.args.clone(),
114 env: new_env,
115 }
116 }
117}
118
119impl PyBootstrapCommand {
120 pub fn to_rust(&self) -> BootstrapCommand {
121 BootstrapCommand {
122 program: PathBuf::from(&self.program),
123 arg0: self.arg0.clone(),
124 args: self.args.clone(),
125 env: self.env.clone(),
126 }
127 }
128}
129
130#[pyclass(
131 name = "HostMesh",
132 module = "monarch._rust_bindings.monarch_hyperactor.host_mesh"
133)]
134#[expect(
135 clippy::large_enum_variant,
136 reason = "PyO3 #[pyclass] enum; Box wrapping interacts with PyO3 codegen and Python interop — separate diff"
137)]
138pub(crate) enum PyHostMesh {
139 Owned(PyHostMeshImpl),
140 Ref(PyHostMeshRefImpl),
141}
142
143impl PyHostMesh {
144 pub(crate) fn new_owned(inner: HostMesh) -> Self {
145 Self::Owned(PyHostMeshImpl(SharedCell::from(inner)))
146 }
147
148 pub(crate) fn new_ref(inner: HostMeshRef) -> Self {
149 Self::Ref(PyHostMeshRefImpl(inner))
150 }
151
152 pub(crate) fn mesh_ref(&self) -> Result<HostMeshRef, anyhow::Error> {
153 match self {
154 PyHostMesh::Owned(inner) => Ok(inner.0.borrow()?.clone()),
155 PyHostMesh::Ref(inner) => Ok(inner.0.clone()),
156 }
157 }
158}
159
160#[pymethods]
161impl PyHostMesh {
162 #[pyo3(signature = (instance, name, per_host, proc_bind = None, per_rank_bootstrap = None))]
163 fn spawn_nonblocking(
164 &self,
165 _py: Python<'_>,
166 instance: &PyInstance,
167 name: String,
168 per_host: &PyExtent,
169 proc_bind: Option<Vec<HashMap<String, String>>>,
170 per_rank_bootstrap: Option<Py<PyAny>>,
171 ) -> PyResult<PyPythonTask> {
172 let host_mesh = self.mesh_ref()?.clone();
173 let per_rank_bootstrap: Option<Box<PerRankBootstrapFn>> = per_rank_bootstrap
174 .map(|callable| -> PyResult<Box<PerRankBootstrapFn>> {
175 Ok(Box::new(move |point| {
176 monarch_with_gil_blocking(GilSite::Bootstrap, |py| {
177 let result =
178 callable
179 .bind(py)
180 .call1((PyPoint::from(point),))
181 .map_err(|e| {
182 anyhow::anyhow!("per-rank bootstrap callable raised: {}", e)
183 })?;
184 let cmd: PyBootstrapCommand = result.extract().map_err(|e| {
185 anyhow::anyhow!(
186 "per-rank bootstrap callable did not return BootstrapCommand: {}",
187 e
188 )
189 })?;
190 Ok(cmd.to_rust())
191 })
192 }))
193 })
194 .transpose()?;
195 let instance = instance.clone();
196 let per_host = per_host.clone().into();
197 let proc_bind = proc_bind.map(|v| v.into_iter().map(ProcBind::from).collect());
198 let mesh_impl = async move {
199 let proc_mesh = host_mesh
200 .spawn(
201 instance.deref(),
202 &name,
203 per_host,
204 proc_bind,
205 per_rank_bootstrap,
206 )
207 .await
208 .map_err(to_py_error)?;
209 Ok(PyProcMesh::new_owned(proc_mesh))
210 };
211 PyPythonTask::new(mesh_impl)
212 }
213
214 fn with_bootstrap(&self, bootstrap_command: &PyBootstrapCommand) -> PyResult<Self> {
215 match self {
216 PyHostMesh::Owned(inner) => {
217 let cmd = bootstrap_command.to_rust();
218 inner
219 .0
220 .try_with_mut(|mesh| mesh.set_bootstrap(cmd))
221 .map_err(|e| PyException::new_err(e.to_string()))?;
222 Ok(Self::Owned(inner.clone()))
223 }
224 PyHostMesh::Ref(_) => Ok(Self::new_ref(
225 self.mesh_ref()?.with_bootstrap(bootstrap_command.to_rust()),
226 )),
227 }
228 }
229
230 fn sliced(&self, region: &PyRegion) -> PyResult<Self> {
231 Ok(Self::new_ref(
232 self.mesh_ref()?.sliced(region.as_inner().clone()),
233 ))
234 }
235
236 #[getter]
237 fn region(&self) -> PyResult<PyRegion> {
238 Ok(PyRegion::from(self.mesh_ref()?.region()))
239 }
240
241 fn __reduce__<'py>(&self, py: Python<'py>) -> PyResult<(Bound<'py, PyAny>, Bound<'py, PyAny>)> {
242 let mesh_ref = self.mesh_ref()?;
243 if crate::pickle::push_mesh_reference_if_active(crate::actor::MeshRef::Host(Box::new(
244 mesh_ref.clone(),
245 ))) {
246 let pop_fn = PyModule::import(py, "monarch._rust_bindings.monarch_hyperactor.pickle")?
247 .getattr("pop_mesh_reference")?;
248 return Ok((pop_fn, pyo3::types::PyTuple::empty(py).into_any()));
249 }
250 let bytes = bincode::serde::encode_to_vec(&mesh_ref, bincode::config::legacy())
251 .map_err(|e| PyErr::new::<PyValueError, _>(e.to_string()))?;
252 let py_bytes = (PyBytes::new(py, &bytes),).into_bound_py_any(py).unwrap();
253 let from_bytes =
254 PyModule::import(py, "monarch._rust_bindings.monarch_hyperactor.host_mesh")?
255 .getattr("py_host_mesh_from_bytes")?;
256 Ok((from_bytes, py_bytes))
257 }
258
259 fn __eq__(&self, other: &PyHostMesh) -> PyResult<bool> {
260 Ok(self.mesh_ref()? == other.mesh_ref()?)
261 }
262
263 fn shutdown(&self, instance: &PyInstance) -> PyResult<PyPythonTask> {
264 match self {
265 PyHostMesh::Owned(inner) => {
266 let instance = instance.clone();
267 let mesh_borrow = inner.0.clone();
268 let fut = async move {
269 match mesh_borrow.take().await {
270 Ok(mut mesh) => {
271 mesh.shutdown(instance.deref()).await?;
272 Ok(())
273 }
274 Err(_) => {
275 tracing::info!("shutdown was already called on host mesh");
278 Ok(())
279 }
280 }
281 };
282 PyPythonTask::new(fut)
283 }
284 PyHostMesh::Ref(_) => Err(PyRuntimeError::new_err(
285 "cannot shut down `HostMesh` that is a reference instead of owned",
286 )),
287 }
288 }
289
290 fn stop(&self, instance: &PyInstance) -> PyResult<PyPythonTask> {
291 match self {
292 PyHostMesh::Owned(inner) => {
293 let instance = instance.clone();
294 let mesh_borrow = inner.0.clone();
295 let fut = async move {
296 match mesh_borrow.take().await {
297 Ok(mut mesh) => {
298 mesh.stop(instance.deref()).await?;
299 Ok(())
300 }
301 Err(_) => {
302 tracing::info!("stop was already called on host mesh");
303 Ok(())
304 }
305 }
306 };
307 PyPythonTask::new(fut)
308 }
309 PyHostMesh::Ref(_) => Err(PyRuntimeError::new_err(
310 "cannot stop `HostMesh` that is a reference instead of owned",
311 )),
312 }
313 }
314}
315
316#[derive(Clone)]
317#[pyclass(
318 name = "HostMeshImpl",
319 module = "monarch._rust_bindings.monarch_hyperactor.host_mesh"
320)]
321pub(crate) struct PyHostMeshImpl(SharedCell<HostMesh>);
322
323#[derive(Debug, Clone)]
324#[pyclass(
325 name = "HostMeshRefImpl",
326 module = "monarch._rust_bindings.monarch_hyperactor.host_mesh"
327)]
328pub(crate) struct PyHostMeshRefImpl(HostMeshRef);
329
330impl PyHostMeshRefImpl {
331 fn __repr__(&self) -> PyResult<String> {
332 Ok(format!("<HostMeshRefImpl {:?}>", self.0))
333 }
334}
335
336static ROOT_CLIENT_INSTANCE_FOR_HOST: OnceLock<Instance<PythonActor>> = OnceLock::new();
338
339static HOST_MESH_AGENT_FOR_HOST: OnceLock<ActorHandle<HostAgent>> = OnceLock::new();
341
342static HOST_SHUTDOWN_HANDLE: OnceLock<
346 tokio::sync::Mutex<Option<hyperactor_mesh::bootstrap::HostShutdownHandle>>,
347> = OnceLock::new();
348
349#[pyfunction]
372#[pyo3(signature = (bootstrap_cmd, via=None))]
373fn bootstrap_host(
374 bootstrap_cmd: Option<PyBootstrapCommand>,
375 via: Option<&str>,
376) -> PyResult<PyPythonTask> {
377 let bootstrap_cmd = match bootstrap_cmd {
378 Some(cmd) => cmd.to_rust(),
379 None => BootstrapCommand::current().map_err(|e| PyException::new_err(e.to_string()))?,
380 };
381 let via_addr = via
382 .map(|s| {
383 ChannelAddr::from_zmq_url(s)
384 .map_err(|e| PyValueError::new_err(format!("via address: {}", e)))
385 })
386 .transpose()?;
387
388 PyPythonTask::new(async move {
389 let gateway = Gateway::global().clone();
397
398 let (host_mesh_agent, shutdown_handle) = host(
399 default_bind_spec().binding_addr(),
400 Some(bootstrap_cmd),
401 None,
402 false,
403 None,
404 gateway,
405 via_addr,
406 )
407 .await
408 .map_err(|e| PyException::new_err(e.to_string()))?;
409
410 HOST_MESH_AGENT_FOR_HOST.set(host_mesh_agent.clone()).ok();
412 HOST_SHUTDOWN_HANDLE.get_or_init(|| tokio::sync::Mutex::new(Some(shutdown_handle)));
413
414 let host_mesh_id = HostMeshId::singleton(Label::new("local").unwrap());
415 let host_mesh = HostMeshRef::from_host_agent(host_mesh_id, host_mesh_agent.bind())
416 .map_err(|e| PyException::new_err(e.to_string()))?;
417
418 hyperactor_mesh::global_context::register_client_host(host_mesh.clone());
421
422 let temp_proc = Proc::isolated();
424 let temp_instance = temp_proc.client("temp");
425
426 let local_proc_agent: hyperactor::ActorHandle<hyperactor_mesh::proc_agent::ProcAgent> =
427 host_mesh_agent
428 .get_local_proc(&temp_instance)
429 .await
430 .map_err(|e| PyException::new_err(e.to_string()))?;
431
432 let proc_mesh = ProcMeshRef::new_singleton(
433 ProcMeshId::singleton(Label::new("local").unwrap()),
434 ProcRef::new(
435 local_proc_agent.actor_addr().proc_addr(),
436 0,
437 local_proc_agent.bind(),
438 ),
439 )
440 .map_err(|e| PyException::new_err(e.to_string()))?;
441
442 let local_proc = local_proc_agent
443 .get_proc(&temp_instance)
444 .await
445 .map_err(|e| PyException::new_err(e.to_string()))?;
446
447 let (instance, _handle) = monarch_with_gil(GilSite::Bootstrap, |py| {
448 PythonActor::bootstrap_client_inner(py, local_proc, &ROOT_CLIENT_INSTANCE_FOR_HOST)
449 })
450 .await;
451
452 {
454 let now = std::time::SystemTime::now();
455
456 let host_name_str = host_mesh.id().to_string();
457 let host_mesh_id = hash_to_u64(host_mesh.id());
458 hyperactor_telemetry::notify_mesh_created(hyperactor_telemetry::MeshEvent {
459 id: host_mesh_id,
460 timestamp: now,
461 class: "Host".to_string(),
462 given_name: host_mesh
463 .id()
464 .display_label()
465 .map(|l| l.as_str())
466 .unwrap_or("unnamed")
467 .to_string(),
468 full_name: host_name_str,
469 shape_json: serde_json::to_string(&host_mesh.region().extent()).unwrap_or_default(),
470 parent_mesh_id: None,
471 parent_view_json: None,
472 });
473
474 let host_agent_addr = host_mesh_agent.actor_addr();
475 hyperactor_telemetry::notify_actor_created(hyperactor_telemetry::ActorEvent {
476 id: hyperactor_telemetry::hash_to_u64(host_agent_addr.id()),
477 timestamp: now,
478 mesh_id: host_mesh_id,
479 rank: 0,
480 full_name: host_agent_addr.to_string(),
481 display_name: None,
482 });
483
484 let proc_id_str = proc_mesh.id().to_string();
485 let proc_mesh_id = hash_to_u64(proc_mesh.id());
486 hyperactor_telemetry::notify_mesh_created(hyperactor_telemetry::MeshEvent {
487 id: proc_mesh_id,
488 timestamp: now,
489 class: "Proc".to_string(),
490 given_name: proc_mesh
491 .id()
492 .display_label()
493 .map(|l| l.as_str())
494 .unwrap_or("unnamed")
495 .to_string(),
496 full_name: proc_id_str,
497 shape_json: serde_json::to_string(&proc_mesh.region().extent()).unwrap_or_default(),
498 parent_mesh_id: Some(host_mesh_id),
499 parent_view_json: None,
500 });
501
502 let proc_agent_addr = local_proc_agent.actor_addr();
503 hyperactor_telemetry::notify_actor_created(hyperactor_telemetry::ActorEvent {
504 id: hyperactor_telemetry::hash_to_u64(proc_agent_addr.id()),
505 timestamp: now,
506 mesh_id: proc_mesh_id,
507 rank: 0,
508 full_name: proc_agent_addr.to_string(),
509 display_name: None,
510 });
511
512 let client_mesh_actor_id = ActorMeshId::singleton(Label::new("client").unwrap());
513 let client_mesh_name = format!("{}/client", proc_mesh.id());
514 let client_mesh_id = telemetry_actor_mesh_id(proc_mesh.id(), &client_mesh_actor_id);
515 hyperactor_telemetry::notify_mesh_created(hyperactor_telemetry::MeshEvent {
516 id: client_mesh_id,
517 timestamp: now,
518 class: <PythonActor as typeuri::Named>::typename().to_string(),
519 given_name: "client".to_string(),
520 full_name: client_mesh_name,
521 shape_json: serde_json::to_string(&proc_mesh.region().extent()).unwrap_or_default(),
522 parent_mesh_id: Some(proc_mesh_id),
523 parent_view_json: None,
524 });
525
526 hyperactor_telemetry::notify_actor_created(hyperactor_telemetry::ActorEvent {
527 id: hyperactor_telemetry::hash_to_u64(instance.self_addr().id()),
528 timestamp: now,
529 mesh_id: client_mesh_id,
530 rank: 0,
531 full_name: instance.self_addr().to_string(),
532 display_name: Some("<root>".to_string()),
533 });
534 }
535
536 Ok((
537 PyHostMesh::new_ref(host_mesh),
538 PyProcMesh::new_ref(proc_mesh),
539 PyInstance::from(instance),
540 ))
541 })
542}
543
544#[pyfunction]
545fn py_host_mesh_from_bytes(bytes: &Bound<'_, PyBytes>) -> PyResult<PyHostMesh> {
546 let r: PyResult<HostMeshRef> =
547 bincode::serde::decode_from_slice(bytes.as_bytes(), bincode::config::legacy())
548 .map(|(v, _)| v)
549 .map_err(|e| PyErr::new::<PyValueError, _>(e.to_string()));
550 r.map(PyHostMesh::new_ref)
551}
552
553#[pyfunction]
554fn shutdown_local_host_mesh() -> PyResult<PyPythonTask> {
555 let agent = HOST_MESH_AGENT_FOR_HOST
556 .get()
557 .ok_or_else(|| PyException::new_err("No local host mesh to shutdown"))?
558 .clone();
559
560 PyPythonTask::new(async move {
561 let temp_proc = hyperactor::Proc::isolated();
563 let instance = temp_proc.client("shutdown_requester");
564
565 tracing::info!(
566 "sending shutdown_host request to agent {}",
567 agent.actor_addr()
568 );
569 let (port, _) = instance.open_port::<usize>();
574 let mut port = port.bind();
575 port.return_undeliverable(false);
578 agent.post(
579 &instance,
580 ShutdownHost {
581 timeout: Duration::from_secs(10),
582 max_in_flight: 16,
583 rank: hyperactor_mesh::resource::Rank::new(0),
584 ack: port,
585 },
586 );
587
588 if let Some(lock) = HOST_SHUTDOWN_HANDLE.get()
591 && let Some(handle) = lock.lock().await.take()
592 {
593 handle.join().await;
594 }
595
596 Ok(())
597 })
598}
599
600#[pyclass(
605 name = "PyMeshAdminRef",
606 module = "monarch._rust_bindings.monarch_hyperactor.host_mesh"
607)]
608#[derive(Clone)]
609pub struct PyMeshAdminRef(hyperactor::ActorRef<hyperactor_mesh::mesh_admin::MeshAdminAgent>);
610
611impl PyMeshAdminRef {
612 pub fn actor_ref(&self) -> hyperactor::ActorRef<hyperactor_mesh::mesh_admin::MeshAdminAgent> {
613 self.0.clone()
614 }
615}
616
617#[pyfunction]
622fn _spawn_admin(
623 host_meshes: Vec<PyRef<'_, PyHostMesh>>,
624 instance: &PyInstance,
625 admin_addr: Option<String>,
626 telemetry_url: Option<String>,
627) -> PyResult<PyPythonTask> {
628 if host_meshes.is_empty() {
629 return Err(PyException::new_err("at least one mesh is required"));
630 }
631
632 let admin_addr = admin_addr
633 .map(|s| {
634 s.parse::<std::net::SocketAddr>()
635 .map_err(|e| PyException::new_err(format!("invalid admin_addr '{}': {}", s, e)))
636 })
637 .transpose()?;
638
639 let mesh_refs = host_meshes
640 .iter()
641 .map(|m| -> PyResult<HostMeshRef> { Ok(m.mesh_ref()?.clone()) })
642 .collect::<PyResult<Vec<HostMeshRef>>>()?;
643
644 let instance = instance.clone();
645 PyPythonTask::new(async move {
646 let admin_ref =
647 host_mesh::spawn_admin(&mesh_refs, instance.deref(), admin_addr, telemetry_url)
648 .await
649 .map_err(|e| PyException::new_err(e.to_string()))?;
650 let admin_url = admin_ref
651 .get_admin_addr(instance.deref())
652 .await
653 .map_err(|e| PyException::new_err(e.to_string()))?
654 .addr
655 .ok_or_else(|| PyException::new_err("mesh admin agent did not report an address"))?;
656 Ok((admin_url, PyMeshAdminRef(admin_ref)))
657 })
658}
659
660pub fn register_python_bindings(hyperactor_mod: &Bound<'_, PyModule>) -> PyResult<()> {
661 let f = wrap_pyfunction!(py_host_mesh_from_bytes, hyperactor_mod)?;
662 f.setattr(
663 "__module__",
664 "monarch._rust_bindings.monarch_hyperactor.host_mesh",
665 )?;
666 hyperactor_mod.add_function(f)?;
667
668 let f2 = wrap_pyfunction!(bootstrap_host, hyperactor_mod)?;
669 f2.setattr(
670 "__module__",
671 "monarch._rust_bindings.monarch_hyperactor.host_mesh",
672 )?;
673 hyperactor_mod.add_function(f2)?;
674
675 let f3 = wrap_pyfunction!(shutdown_local_host_mesh, hyperactor_mod)?;
676 f3.setattr(
677 "__module__",
678 "monarch._rust_bindings.monarch_hyperactor.host_mesh",
679 )?;
680 hyperactor_mod.add_function(f3)?;
681
682 let f4 = wrap_pyfunction!(_spawn_admin, hyperactor_mod)?;
683 f4.setattr(
684 "__module__",
685 "monarch._rust_bindings.monarch_hyperactor.host_mesh",
686 )?;
687 hyperactor_mod.add_function(f4)?;
688
689 hyperactor_mod.add_class::<PyHostMesh>()?;
690 hyperactor_mod.add_class::<PyBootstrapCommand>()?;
691 hyperactor_mod.add_class::<PyMeshAdminRef>()?;
692 Ok(())
693}