add a "direct" variant to ProcIds (#787)

Summary:
Pull Request resolved: #787

With this change, we can name proc ids by their channel address. This avoids having to manufacture world ids for bootstrapping purposes.
ghstack-source-id: 301740065
exported-using-ghexport

Reviewed By: shayne-fletcher, vidhyav

Differential Revision: D79761721

fbshipit-source-id: 7a8eb173ee8ccca36e6ec484a857e57bd7e9d58f

Author: mariusae

controller/src/lib.rs

@@ -175,7 +175,11 @@ impl ControllerActor {
     ) -> Result<(ActorHandle<ProcActor>, ActorRef<ControllerActor>), anyhow::Error> {
         let bootstrap = ProcActor::bootstrap(
             controller_id.proc_id().clone(),
-            controller_id.proc_id().world_id().clone(), // REFACTOR(marius): make world_id a parameter of ControllerActor::bootstrap
+            controller_id
+                .proc_id()
+                .world_id()
+                .expect("multiprocess supports only ranked procs")
+                .clone(), // REFACTOR(marius): make world_id a parameter of ControllerActor::bootstrap
             listen_addr,
             bootstrap_addr.clone(),
             supervision_update_interval,
@@ -685,7 +689,12 @@ mod tests {
                     world_size: 1,
                     comm_actor_ref: comm_handle.bind(),
                     worker_gang_ref: GangId(
-                        WorldId(proc.proc_id().world_name().to_string()),
+                        WorldId(
+                            proc.proc_id()
+                                .world_name()
+                                .expect("only ranked actors are supported in the controller tests")
+                                .to_string(),
+                        ),
                         "worker".to_string(),
                     )
                     .into(),
@@ -869,7 +878,12 @@ mod tests {
                     world_size: 1,
                     comm_actor_ref: comm_handle.bind(),
                     worker_gang_ref: GangId(
-                        WorldId(proc.proc_id().world_name().to_string()),
+                        WorldId(
+                            proc.proc_id()
+                                .world_name()
+                                .expect("only ranked actors are supported in the controller tests")
+                                .to_string(),
+                        ),
                         "worker".to_string(),
                     )
                     .into(),
@@ -975,7 +989,12 @@ mod tests {
             .await
             .unwrap();
 
-        let world_id = WorldId(proc.proc_id().world_name().to_string());
+        let world_id = WorldId(
+            proc.proc_id()
+                .world_name()
+                .expect("only ranked actors are supported in the controller tests")
+                .to_string(),
+        );
         let controller_handle = proc
             .spawn::<ControllerActor>(
                 "controller",
@@ -1500,7 +1519,12 @@ mod tests {
                 world_size: 1,
                 comm_actor_ref: ActorRef::attest(controller_id.proc_id().actor_id("comm", 0)),
                 worker_gang_ref: GangId(
-                    WorldId(proc_id.world_name().to_string()),
+                    WorldId(
+                        proc_id
+                            .world_name()
+                            .expect("only ranked actors are supported in the controller tests")
+                            .to_string(),
+                    ),
                     "worker".to_string(),
                 )
                 .into(),
@@ -1591,7 +1615,12 @@ mod tests {
                 world_size: 1,
                 comm_actor_ref: ActorRef::attest(controller_id.proc_id().actor_id("comm", 0)),
                 worker_gang_ref: GangId(
-                    WorldId(proc_id.world_name().to_string()),
+                    WorldId(
+                        proc_id
+                            .world_name()
+                            .expect("only ranked actors are supported in the controller tests")
+                            .to_string(),
+                    ),
                     "worker".to_string(),
                 )
                 .into(),
@@ -1692,7 +1721,12 @@ mod tests {
                 world_size: 1,
                 comm_actor_ref: ActorRef::attest(controller_id.proc_id().actor_id("comm", 0)),
                 worker_gang_ref: GangId(
-                    WorldId(proc_id.world_name().to_string()),
+                    WorldId(
+                        proc_id
+                            .world_name()
+                            .expect("only ranked actors are supported in the controller tests")
+                            .to_string(),
+                    ),
                     "worker".to_string(),
                 )
                 .into(),
@@ -1835,7 +1869,12 @@ mod tests {
                 world_size: 1,
                 comm_actor_ref: ActorRef::attest(controller_id.proc_id().actor_id("comm", 0)),
                 worker_gang_ref: GangId(
-                    WorldId(proc_id.world_name().to_string()),
+                    WorldId(
+                        proc_id
+                            .world_name()
+                            .expect("only ranked actors are supported in the controller tests")
+                            .to_string(),
+                    ),
                     "worker".to_string(),
                 )
                 .into(),

hyper/src/commands/demo.rs

@@ -61,7 +61,10 @@ impl DemoCommand {
 
                 let proc_actor = ProcActor::bootstrap(
                     proc_id.clone(),
-                    proc_id.0.clone(),
+                    proc_id
+                        .world_id()
+                        .expect("unranked proc not supported")
+                        .clone(),
                     addr,
                     system_addr,
                     Duration::from_secs(5),

hyperactor/src/mailbox.rs

@@ -19,7 +19,7 @@
 //! # use hyperactor::mailbox::Mailbox;
 //! # use hyperactor::reference::{ActorId, ProcId, WorldId};
 //! # tokio_test::block_on(async {
-//! # let proc_id = ProcId(WorldId("world".to_string()), 0);
+//! # let proc_id = ProcId::Ranked(WorldId("world".to_string()), 0);
 //! # let actor_id = ActorId(proc_id, "actor".to_string(), 0);
 //! let mbox = Mailbox::new_detached(actor_id);
 //! let (port, mut receiver) = mbox.open_port::<u64>();
@@ -36,7 +36,7 @@
 //! # use hyperactor::mailbox::Mailbox;
 //! # use hyperactor::reference::{ActorId, ProcId, WorldId};
 //! # tokio_test::block_on(async {
-//! # let proc_id = ProcId(WorldId("world".to_string()), 0);
+//! # let proc_id = ProcId::Ranked(WorldId("world".to_string()), 0);
 //! # let actor_id = ActorId(proc_id, "actor".to_string(), 0);
 //! let mbox = Mailbox::new_detached(actor_id);
 //!
@@ -2517,7 +2517,11 @@ mod tests {
     #[test]
     fn test_error() {
         let err = MailboxError::new(
-            ActorId(ProcId(WorldId("myworld".into()), 2), "myactor".into(), 5),
+            ActorId(
+                ProcId::Ranked(WorldId("myworld".to_string()), 2),
+                "myactor".to_string(),
+                5,
+            ),
             MailboxErrorKind::Closed,
         );
         assert_eq!(format!("{}", err), "myworld[2].myactor[5]: mailbox closed");

hyperactor/src/proc.rs

@@ -376,7 +376,7 @@ impl Proc {
     pub fn local() -> Self {
         // TODO: name these something that is ~ globally unique, e.g., incorporate
         // the hostname, some GUID, etc.
-        let proc_id = ProcId(id!(local), NEXT_LOCAL_RANK.fetch_add(1, Ordering::Relaxed));
+        let proc_id = ProcId::Ranked(id!(local), NEXT_LOCAL_RANK.fetch_add(1, Ordering::Relaxed));
         // TODO: make it so that local procs can talk to each other.
         Proc::new(proc_id, BoxedMailboxSender::new(PanickingMailboxSender))
     }