Fix termination; add pty resizing

protocols/schema-bundle.bin

@@ -1,4 +1,4 @@
 ´³bundle·µ³pty„´³schema·³version°

³definitions·³Chunk´³atom³
-ByteString„³PtyInput´³rec´³lit³	pty-input„´³tupleµ´³named³id³any„´³named³data´³refµ„³Chunk„„„„„³	PtyOutput´³rec´³lit³
+ByteString„³PtySize´³rec´³lit³pty-size„´³tupleµ´³named³id³any„´³named³columns´³atom³
SignedInteger„„´³named³rows´³atom³
SignedInteger„„„„„³PtyInput´³rec´³lit³	pty-input„´³tupleµ´³named³id³any„´³named³data´³refµ„³Chunk„„„„„³	PtyOutput´³rec´³lit³
 pty-output„´³tupleµ´³named³id³any„´³named³data´³refµ„³Chunk„„„„„³
 PtySession´³rec´³lit³pty-session„´³tupleµ´³named³id³any„´³named³commandLine´³refµ„³CommandLine„„„„„³CommandLine´³tuplePrefixµ´³named³command³any„„´³named³args´³seqof³any„„„³PtySessionRunning´³rec´³lit³pty-session-running„´³tupleµ´³named³id³any„„„„„³embeddedType€„„„„

protocols/schemas/pty.prs

@@ -5,6 +5,9 @@ CommandLine = [@command any @args any ...] .
 
 PtySessionRunning = <pty-session-running @id any> .
 
+# Message, driver interprets it as a request to execute TIOCSWINSZ
+PtySize = <pty-size @id any @columns int @rows int> .
+
 # To the subprocess
 PtyInput = <pty-input @id any @data Chunk> .
 

src/main.rs

@@ -11,14 +11,15 @@ use syndicate::actor::ActorError;
 use syndicate::actor::ActorResult;
 use syndicate::actor::AnyValue;
 use syndicate::actor::Cap;
+use syndicate::actor::Field;
 use syndicate::actor::LinkedTaskTermination;
 use syndicate::dataspace::Dataspace;
 use syndicate::enclose;
 use syndicate::relay;
 use syndicate::value::NestedValue;
-
 use syndicate::value::NoEmbeddedDomainCodec;
 use syndicate::value::TextWriter;
+use syndicate::value::Value;
 use syndicate_macros::during;
 use syndicate_macros::on_message;
 
@@ -36,7 +37,46 @@ preserves_schema::define_language!(language(): Language<AnyValue> {
 use schemas::pty::{PtySession, PtySessionRunning, Chunk, PtyOutput};
 
 fn stringify(v: &AnyValue) -> Result<String, ActorError> {
-    Ok(TextWriter::encode(&mut NoEmbeddedDomainCodec, v)?)
+    match v.value() {
+        Value::String(s) => Ok(s.clone()),
+        Value::Symbol(s) => Ok(s.clone()),
+        _ => Ok(TextWriter::encode(&mut NoEmbeddedDomainCodec, v)?),
+    }
+}
+
+fn terminate_pid(t: &mut Activation, child_pid: &Field<Option<i32>>) -> ActorResult {
+    match std::mem::replace(t.get_mut(child_pid), None) {
+        None => Ok(()),
+        Some(pid) => {
+            tracing::trace!(?pid, "SIGHUP");
+            unsafe { libc::kill(pid, libc::SIGHUP); }
+            tokio::spawn(async move {
+                let mut status: i32 = 0;
+
+                tokio::time::sleep(std::time::Duration::from_secs(1)).await;
+                if unsafe { libc::waitpid(pid, &mut status, libc::WNOHANG) } != 0 {
+                    tracing::trace!(?pid, ?status, "SIGHUP was effective");
+                    return;
+                }
+
+                tracing::trace!(?pid, "SIGTERM");
+                unsafe { libc::kill(pid, libc::SIGTERM); }
+
+                tokio::time::sleep(std::time::Duration::from_secs(2)).await;
+                if unsafe { libc::waitpid(pid, &mut status, libc::WNOHANG) } != 0 {
+                    tracing::trace!(?pid, ?status, "SIGTERM was effective");
+                    return;
+                }
+
+                tracing::trace!(?pid, "SIGKILL");
+                unsafe { libc::kill(pid, libc::SIGKILL); }
+
+                unsafe { libc::waitpid(pid, &mut status, 0); }
+                tracing::trace!(?pid, ?status, "exited");
+            });
+            Ok(())
+        }
+    }
 }
 
 fn pty_session(t: &mut Activation, local_ds: Arc<Cap>, s: PtySession<AnyValue>) -> ActorResult {
@@ -52,15 +92,23 @@ fn pty_session(t: &mut Activation, local_ds: Arc<Cap>, s: PtySession<AnyValue>)
     match &fork {
         pty::prelude::Fork::Parent(child_pid, parent) => {
             tracing::info!(?child_pid);
-            let mut parent_w: std::fs::File = unsafe { FromRawFd::from_raw_fd(libc::dup(parent.as_raw_fd())) };
+            let parent_fd = parent.as_raw_fd();
-            let mut parent_r: tokio::fs::File = unsafe { FromRawFd::from_raw_fd(libc::dup(parent.as_raw_fd())) };
+            let child = pty::fork::Slave::new(parent.ptsname()?)?;
-            let child_pid = *child_pid;
+            let child_fd = unsafe { libc::dup(child.as_raw_fd()) };
+            let mut parent_w: std::fs::File = unsafe { FromRawFd::from_raw_fd(libc::dup(parent_fd)) };
+            let mut parent_r: tokio::fs::File = unsafe { FromRawFd::from_raw_fd(libc::dup(parent_fd)) };
+            let child_pid = t.named_field("child_pid", Some(*child_pid));
             let facet = t.facet.clone();
 
             local_ds.assert(t, language(), &PtySessionRunning {
                 id: s.id.clone(),
             });
 
+            t.on_stop(enclose!((child_pid) move |t| {
+                tracing::info!("terminating child on facet stop");
+                terminate_pid(t, &child_pid)
+            }));
+
             on_message!(t, local_ds, language(), <pty-input #(&s.id) $body: Chunk>, |_t| {
                 tracing::trace!(?body, "sending to child");
                 // TODO: use async here
@@ -68,18 +116,42 @@ fn pty_session(t: &mut Activation, local_ds: Arc<Cap>, s: PtySession<AnyValue>)
                 Ok(())
             });
 
-            t.linked_task(Some(AnyValue::symbol("waiter")), async move {
+            on_message!(t, local_ds, language(), <pty-size #(&s.id) $columns $rows>, |_t| {
-                let mut status: i32 = 0;
+                tracing::trace!(?columns, ?rows, "size change");
-                loop {
+                let result = unsafe {
-                    match unsafe { libc::waitpid(child_pid, &mut status, libc::WNOHANG) } {
+                    libc::ioctl(child_fd, libc::TIOCSWINSZ, &libc::winsize {
-                        0 => tokio::time::sleep(tokio::time::Duration::from_millis(1000)).await,
+                        ws_row: rows.value().as_u16().unwrap_or(24),
-                        -1 => Err(std::io::Error::last_os_error())?,
+                        ws_col: columns.value().as_u16().unwrap_or(80),
-                        _ => break,
+                        ws_xpixel: 0,
+                        ws_ypixel: 0,
+                    })
+                };
+                if result == 0 {
+                    tracing::trace!(?columns, ?rows, "size changed");
+                } else {
+                    tracing::error!(err=?std::io::Error::last_os_error(), ?columns, ?rows, ?child_fd, "size change failed");
+                }
+                Ok(())
+            });
+
+            t.every(tokio::time::Duration::from_secs(1), enclose!((child_pid) move |t| {
+                match t.get(&child_pid) {
+                    None => Ok(()),
+                    Some(pid) => {
+                        let mut status: i32 = 0;
+                        match unsafe { libc::waitpid(*pid, &mut status, libc::WNOHANG) } {
+                            0 => Ok(()), // hasn't exited yet
+                            -1 => Err(std::io::Error::last_os_error())?,
+                            _ => {
+                                tracing::info!(?status, "child exited");
+                                *t.get_mut(&child_pid) = None;
+                                t.stop();
+                                Ok(())
+                            }
+                        }
                     }
                 }
-                tracing::info!(?status, "child exited");
+            }))?;
-                Ok(LinkedTaskTermination::Normal)
-            });
 
             t.linked_task(Some(AnyValue::symbol("reader")), async move {
                 let read_account = Account::new(Some(AnyValue::symbol("reader_account")), None);
@@ -87,19 +159,22 @@ fn pty_session(t: &mut Activation, local_ds: Arc<Cap>, s: PtySession<AnyValue>)
                 loop {
                     let n = parent_r.read(&mut buf[..]).await?;
                     tracing::trace!(?n, buf=?&buf[0..n], "receiving from child");
-                    if n == 0 || !facet.activate(&read_account, None, |t| {
+                    if n == 0 {
-                        local_ds.message(t, language(), &PtyOutput {
+                        facet.activate(&read_account, None, |t| terminate_pid(t, &child_pid));
-                            id: s.id.clone(),
+                        break;
-                            data: Chunk(buf[0..n].to_vec()),
+                    } else {
-                        });
+                        if !facet.activate(&read_account, None, |t| {
-                        Ok(())
+                            local_ds.message(t, language(), &PtyOutput {
-                    }) {
+                                id: s.id.clone(),
-                        unsafe {
+                                data: Chunk(buf[0..n].to_vec()),
-                            libc::kill(child_pid, libc::SIGTERM);
+                            });
+                            Ok(())
+                        }) {
+                            break;
                         }
-                        return Ok(LinkedTaskTermination::Normal);
                     }
                 }
+                return Ok(LinkedTaskTermination::Normal);
             });
             Ok(())
         }