parse direct procs / actors (meta-pytorch#800)

Summary:
Pull Request resolved: meta-pytorch#800

This change introduces a new kind of identifier used to parse direct procs.

As we plan to phase out ranked procs, this will become the only kind of identifier in the system.

```
# The proc "proc-name" at tcp:[::1]:1234
tcp:[::1]:1234,proc-name

# The root actor "actor-name" at the above proc
tcp:[::1]:1234,proc-name,actor-name

# ... equivalent to:
tcp:[::1]:1234,proc-name,actor-name[0]

# A child actor of the above actor:
tcp:[::1]:1234,proc-name,actor-name[123]
```

## Alternatives considered

The central challenge with this syntax is satisfying the following simultaneously:

1) Unambiguous parsing: all while mixing channel addresses and procs, actor names, and PIDs
2) A shell-friendly syntax: we do not want to have to impose quoting or escaping in typical cases
3) Make it simple, drawing on only a few lexemes

For example, the first requirement precludes a straightforward URL syntax, consider:

```
tcp://[::1]:1234/proc-name/actor-name/pid
```

but what happens when you introduce unix sockets?

```
unix:///path/to/socket/proc-name/actor-name/pid
```

How do you delimit the socket address from the proc name?

One alternative is to introduce another delimiter, e.g.:

```
unix:///path/to/socket//proc-name/actor-name/pid
```

but now this looks rather like line noise! And it's hard to parse!

One alternative I liked is to quote the channel address: this sets it apart, and signifies a separate syntax.

```
"tcp:[::1]:1234".proc-name.actor-name[pid]
```

but the shell parses the quotes!

Similar issues exist for other alternatives, which can be parsed unambiguously, but would make the grammar more complicated:

```
# Brackets compete with IPv6 addresses, requiring recursive descent parsing
unix[/foo/bar].proc-name.actor-name[pid]

# "::" clash with IPv6 addresses, requiring additional context to parse correctly
unix:/foo/bar::proc-name::actor-name[pid]
```

And so on.

All of this being said: if you find a clearly superior alternative, I am all ears.
ghstack-source-id: 301740077
exported-using-ghexport

Reviewed By: shayne-fletcher

Differential Revision: D79856404

fbshipit-source-id: 60b479091af52c3f15f2c6bf446639fab05f60ee

Author: mariusae

hyperactor/src/parse.rs

@@ -38,6 +38,8 @@ pub enum Token<'a> {
     Colon,
     /// "."
     Dot,
+    /// ","
+    Comma,
 
     // Special token to denote an invalid element. It is used to poison
     // the parser.
@@ -56,7 +58,7 @@ impl<'a> Lexer<'a> {
         Self {
             // TODO: compose iterators directly; would be simpler with
             // existential type support.
-            tokens: chop(input, &["[", "]", ".", "@"]).collect(),
+            tokens: chop(input, &["[", "]", ".", "@", ","]).collect(),
         }
     }
 }
@@ -72,6 +74,7 @@ impl<'a> Iterator for Lexer<'a> {
             Some("@") => Some(Token::At),
             Some(":") => Some(Token::Colon),
             Some(".") => Some(Token::Dot),
+            Some(",") => Some(Token::Comma),
             Some(elem) => Some({
                 if let Ok(uint) = elem.parse::<usize>() {
                     Token::Uint(uint)
@@ -180,7 +183,7 @@ mod tests {
 
     #[test]
     fn test_lexer() {
-        let tokens = Lexer::new("foo.bar[123]");
+        let tokens = Lexer::new("foo.bar[123],baz");
         assert_eq!(
             tokens.collect::<Vec<Token>>(),
             vec![
@@ -189,7 +192,9 @@ mod tests {
                 Token::Elem("bar"),
                 Token::LeftBracket,
                 Token::Uint(123),
-                Token::RightBracket
+                Token::RightBracket,
+                Token::Comma,
+                Token::Elem("baz"),
             ]
         )
     }

hyperactor/src/reference.rs

@@ -302,7 +302,7 @@ macro_rules! id {
 pub use id;
 
 /// The type of error encountered while parsing references.
-#[derive(thiserror::Error, Debug, PartialEq, Eq)]
+#[derive(thiserror::Error, Debug)]
 pub enum ReferenceParsingError {
     /// The parser expected a token, but it reached the end of the token stream.
     #[error("expected token")]
@@ -323,45 +323,82 @@ pub enum ReferenceParsingError {
     /// The parser encountered the wrong reference type.
     #[error("wrong reference type: expected {0}")]
     WrongType(String),
+
+    /// An invalid channel address was encountered while parsing the reference.
+    #[error("invalid channel address {0}: {1}")]
+    InvalidChannelAddress(String, anyhow::Error),
 }
 
 impl FromStr for Reference {
     type Err = ReferenceParsingError;
 
     fn from_str(addr: &str) -> Result<Self, Self::Err> {
-        let reference = parse! {
-            Lexer::new(addr);
-
-            // world
-            Token::Elem(world) => Self::World(WorldId(world.into())),
-
-            // world[rank]
-            Token::Elem(world) Token::LeftBracket Token::Uint(rank) Token::RightBracket =>
-                Self::Proc(ProcId::Ranked(WorldId(world.into()), rank)),
-
-            // world[rank].actor  (implied pid=0)
-            Token::Elem(world) Token::LeftBracket Token::Uint(rank) Token::RightBracket
-                Token::Dot Token::Elem(actor) =>
-                Self::Actor(ActorId(ProcId::Ranked(WorldId(world.into()), rank), actor.into(), 0)),
-
-            // world[rank].actor[pid]
-            Token::Elem(world) Token::LeftBracket Token::Uint(rank) Token::RightBracket
-                Token::Dot Token::Elem(actor)
-                Token::LeftBracket Token::Uint(pid) Token::RightBracket =>
-                Self::Actor(ActorId(ProcId::Ranked(WorldId(world.into()), rank), actor.into(), pid)),
-
-            // world[rank].actor[pid][port]
-            Token::Elem(world) Token::LeftBracket Token::Uint(rank) Token::RightBracket
-                Token::Dot Token::Elem(actor)
-                Token::LeftBracket Token::Uint(pid) Token::RightBracket
-                Token::LeftBracket Token::Uint(index) Token::RightBracket =>
-                Self::Port(PortId(ActorId(ProcId::Ranked(WorldId(world.into()), rank), actor.into(), pid), index as u64)),
-
-            // world.actor
-            Token::Elem(world) Token::Dot Token::Elem(actor) =>
-                Self::Gang(GangId(WorldId(world.into()), actor.into())),
-        };
-        Ok(reference?)
+        // First, try to parse a "new style" reference:
+        // 1) If the reference contains a comma (anywhere), it is a new style reference;
+        //    commas were not a valid lexeme in the previous reference format.
+        // 2) This is a bit ugly, but we bypass the tokenizer prior to this comma,
+        //    try to parse a channel address, and then parse the remainder.
+
+        match addr.split_once(",") {
+            Some((channel_addr, rest)) => {
+                let channel_addr = channel_addr.parse().map_err(|err| {
+                    ReferenceParsingError::InvalidChannelAddress(channel_addr.to_string(), err)
+                })?;
+
+                Ok(parse! {
+                    Lexer::new(rest);
+
+                    // channeladdr,proc_name
+                    Token::Elem(proc_name) =>
+                    Self::Proc(ProcId::Direct(channel_addr, proc_name.to_string())),
+
+                    // channeladdr,proc_name,actor_name
+                    Token::Elem(proc_name) Token::Comma Token::Elem(actor_name) =>
+                    Self::Actor(ActorId(ProcId::Direct(channel_addr, proc_name.to_string()), actor_name.to_string(), 0)),
+
+                    // channeladdr,proc_name,actor_name[rank]
+                    Token::Elem(proc_name) Token::Comma Token::Elem(actor_name)
+                        Token::LeftBracket Token::Uint(rank) Token::RightBracket =>
+                        Self::Actor(ActorId(ProcId::Direct(channel_addr, proc_name.to_string()), actor_name.to_string(), rank)),
+                }?)
+            }
+
+            // "old style" / "ranked" reference
+            None => {
+                Ok(parse! {
+                    Lexer::new(addr);
+
+                    // world
+                    Token::Elem(world) => Self::World(WorldId(world.into())),
+
+                    // world[rank]
+                    Token::Elem(world) Token::LeftBracket Token::Uint(rank) Token::RightBracket =>
+                        Self::Proc(ProcId::Ranked(WorldId(world.into()), rank)),
+
+                    // world[rank].actor  (implied pid=0)
+                    Token::Elem(world) Token::LeftBracket Token::Uint(rank) Token::RightBracket
+                        Token::Dot Token::Elem(actor) =>
+                        Self::Actor(ActorId(ProcId::Ranked(WorldId(world.into()), rank), actor.into(), 0)),
+
+                    // world[rank].actor[pid]
+                    Token::Elem(world) Token::LeftBracket Token::Uint(rank) Token::RightBracket
+                        Token::Dot Token::Elem(actor)
+                        Token::LeftBracket Token::Uint(pid) Token::RightBracket =>
+                        Self::Actor(ActorId(ProcId::Ranked(WorldId(world.into()), rank), actor.into(), pid)),
+
+                    // world[rank].actor[pid][port]
+                    Token::Elem(world) Token::LeftBracket Token::Uint(rank) Token::RightBracket
+                        Token::Dot Token::Elem(actor)
+                        Token::LeftBracket Token::Uint(pid) Token::RightBracket
+                        Token::LeftBracket Token::Uint(index) Token::RightBracket =>
+                        Self::Port(PortId(ActorId(ProcId::Ranked(WorldId(world.into()), rank), actor.into(), pid), index as u64)),
+
+                    // world.actor
+                    Token::Elem(world) Token::Dot Token::Elem(actor) =>
+                        Self::Gang(GangId(WorldId(world.into()), actor.into())),
+                }?)
+            }
+        }
     }
 }
 
@@ -519,16 +556,6 @@ impl FromStr for ProcId {
     type Err = ReferenceParsingError;
 
     fn from_str(addr: &str) -> Result<Self, Self::Err> {
-        // We first try to parse the proc id as a channel address; otherwise
-        // as a ranked reference. These grammars are currently non-overlapping,
-        // but we need to be careful when we add new channel address types.
-        //
-        // Over time, we will deprecate ranked references and provide a robustly
-        // unambiguous syntax.
-        if let Ok(channel_addr) = addr.parse::<ChannelAddr>() {
-            // TODO: parse names
-            return Ok(ProcId::Direct(channel_addr, "".to_string()));
-        }
         match addr.parse()? {
             Reference::Proc(proc_id) => Ok(proc_id),
             _ => Err(ReferenceParsingError::WrongType("proc".into())),
@@ -1258,6 +1285,15 @@ mod tests {
                 "test.testactor",
                 GangId(WorldId("test".into()), "testactor".into()).into(),
             ),
+            (
+                "tcp:[::1]:1234,test,testactor[123]",
+                ActorId(
+                    ProcId::Direct("tcp:[::1]:1234".parse().unwrap(), "test".to_string()),
+                    "testactor".to_string(),
+                    123,
+                )
+                .into(),
+            ),
         ];
 
         for (s, expected) in cases {