Refactored interactivity to track with thread.

Hit a bit of a stopping point. Specifically, the way XAsync runs
currently is by forking the execution context and having the child run
the async code while the parent runs the remainder. The problem with
this architecture is it doesn't interact well with job control. When we
fork, we create a new process group. Thus the Xasync fork becomes the
new leader.

In short, our traversal of the parse tree as to be less "preorder" and
more "in order", i.e. from the leaves up. The "left" command of the
pipeline should be the "leader" of the process group.

1 files changed, 221 insertions, 110 deletions

diff --git a/sys/cmd/rc/exec.c b/sys/cmd/rc/exec.c
index 217f7ee..f9af866 100644
--- a/sys/cmd/rc/exec.c
+++ b/sys/cmd/rc/exec.c
@@ -10,6 +10,12 @@ struct Builtin{
 	void (*func)(void);
 };
 
+struct State {
+    int async;
+};
+
+static struct State state;
+
 // -----------------------------------------------------------------------
 // globals
 
@@ -138,6 +144,14 @@ path(char *w)
 
 static inline
 void
+undoredirs(void)
+{
+    while(runner->redir.end != runner->redir.start)
+        Xpopredir();
+}
+
+static inline
+void
 defaultsignal(void)
 {
     signal(SIGINT,  SIG_DFL);
@@ -150,15 +164,50 @@ defaultsignal(void)
 
 static inline
 void
-setpid(int pid)
+setpid(Thread *job, int pid)
+{
+    job->pid = pid;
+    if(job->pgid <= 0){
+        job->pgid = pid;
+        addjob(job);
+    }
+
+    setpgid(pid, job->pgid);
+}
+
+/* fork/execute helpers */
+
+static inline
+void
+initchild(Thread *job, int fg)
 {
-    runner->pid = pid;
-    if(runner->pgid <= 0){
-        runner->pgid = runner->pid;
-        addjob(runner);
+    int pid = getpid();
+    setpid(job, pid);
+
+    if(job->flag.user){
+        if(fg)
+            tcsetpgrp(0, job->pgid);
+        else
+            job->flag.user = 0;
+        defaultsignal();
     }
 
-    setpgid(pid, runner->pgid);
+    clearwait(job);
+}
+
+static inline
+void
+initparent(Thread *job, int pid, int fg)
+{
+    setpid(job, pid);
+
+    if(job->flag.user){
+        if(!fg){
+            tcsetpgrp(0, job->pgid);
+            job->flag.user = 0;
+        }
+    }
+    addwait(job, pid);
 }
 
 static
@@ -171,6 +220,7 @@ xx(void)
         return;
     }
 
+    redirect(runner->redir.end);
     execute(runner->args->word, path(runner->args->word->str));
     poplist();
 }
@@ -180,29 +230,20 @@ int
 xforkx(void)
 {
     int n, pid;
-    char buf[ERRMAX];
 
     switch(pid=fork()){
     case -1:
+        Xerror("try again\n");
         return -1;
     case 0: // child
-        clearwait(runner);
-        runner->link = nil;
-
-        if(shell.interactive){
-            setpid(getpid());
-            tcsetpgrp(0, runner->pgid);
-            defaultsignal();
-        }
+        initchild(runner, 1);
 
         pushword("exec");
         xx();
 
-        exit(2); // NOTE: xx does not return!
-        /*unreachable*/
+        exit(2); // NOTE: unreachable: xx does not return
     default: // parent
-        addwait(runner, pid);
-        setpid(pid); // ensure the state matches for parent and child
+        initparent(runner, pid, 0);
 
         return pid;
     }
@@ -217,15 +258,14 @@ pushredir(int type, int from, int to)
     r->type = type;
     r->from = from;
     r->to = to;
-    r->link = runner->redir;
 
-    runner->redir = r;
+    r->link = runner->redir.end, runner->redir.end = r;
 }
 
 /* byte code */
 static
 void
-run(Code *c, int pc, Var *local, int keeppid)
+run(Code *c, int pc, Var *local, int inherit)
 {
     Thread *new = emalloc(sizeof(*new));
 
@@ -237,10 +277,17 @@ run(Code *c, int pc, Var *local, int keeppid)
 
     new->args     = nil;
     new->local    = local;
-    new->redir    = nil;
 
-    new->flag.i   = 0;
-    new->flag.eof = 0;
+    new->flag.eof  = 0;
+    if(runner){
+        new->pid       = runner->pid;
+        new->flag.user = runner->flag.user;
+        new->redir.end = new->redir.start = runner->redir.end;
+    }else{
+        new->pid       = shell.pid;
+        new->flag.user = shell.interactive;
+        new->redir.end = new->redir.start = nil;
+    }
 
     new->wait.status = 0;
     new->wait.len    = 0;
@@ -248,14 +295,14 @@ run(Code *c, int pc, Var *local, int keeppid)
     new->wait.on     = nil;
 
     new->status = 0;
-    if(keeppid)
-        new->pgid = runner->pgid, new->pid = runner->pid;
+    if(inherit)
+        new->pgid = runner->pgid;
     else
-        new->pgid = new->pid = -1;
+        new->pgid = -1;
 
     new->line     = 0;
-    new->link     = runner;
-    new->next     = nil;
+    new->caller   = runner;
+    new->link     = nil;
 
     runner = new;
 }
@@ -354,9 +401,7 @@ xdot(void)
     }
 
     if(fd<0){
-        print(shell.err, "%s: ", base);
-        //setstatus("can't open");
-        Xerror(".: can't open\n");
+        print(shell.err, "failed open: %s: ", base);
         return;
     }
     /* set up for a new command loop */
@@ -432,7 +477,7 @@ xjob(void)
     int i;
     Thread *job;
 
-    for(i=0, job = shell.jobs; job; job = job->next, i++)
+    for(i=0, job = shell.jobs; job; job = job->link, i++)
         report(job,i);
 
     poplist();
@@ -454,16 +499,16 @@ xfg(void)
     }
 
     i = atoi(runner->args->word->str);
-    popword(); // [jobid]
+    popword(); // [pid|num]
 
-    for(job=shell.jobs; i > 0; job=job->next, --i)
+    for(job=shell.jobs; i > 0; job=job->link, --i)
         ;
 
     poplist(); // this goes here?
 
     wakeup(job);
-    job->link = runner, runner = job; // XXX: can this leave zombies?
     foreground(job, 1);
+    job->caller = runner, runner = job; // XXX: can this leave zombies?
 }
 
 void
@@ -493,6 +538,7 @@ xboot(int argc, char *argv[])
     bootstrap[i].i = 0;
 
     run(bootstrap, 1, nil, 0);
+    runner->pid = runner->pgid = shell.pid;
     pushlist(); // prime bootstrap argv
 
     argv0 = strdup(argv[0]);
@@ -516,54 +562,6 @@ Xmark(void)
 }
 
 void
-Xerror(char *msg)
-{
-    print(shell.err, "rc: %s", msg);
-    flush(shell.err);
-    while(!runner->flag.i)
-        Xreturn();
-}
-
-void
-Xreturn(void)
-{
-    Thread *run = runner;
-
-    switch(run->wait.status){
-    /*
-     * If our job is still running we must:
-     * 1. move program one step back to rerun Xreturn upon recall
-     * 2. return to our calling thread
-     * 3. don't free!
-     */
-     case Prun: case Pstop:
-        run->code.i--;
-        runner = run->link;
-        return;
-    /*
-     * If our job has finished:
-     * 1. remove from our list
-     * 2. continue to clean up its memory
-     */
-    case Pdone:
-        deljob(run);
-        /* fallthrough */
-    default:
-        ;
-    }
-
-    while(run->args)
-        poplist();
-    freecode(run->code.exe);
-    efree(run->wait.on);
-
-    runner = run->link;
-    efree(run);
-    if(!runner)
-        exit(0);
-}
-
-void
 Xword(void)
 {
     pushword(runner->code.exe[runner->code.i++].s);
@@ -703,21 +701,18 @@ Xasync(void)
         Xerror("fork failed: try again");
         break;
 
-    case 0: // child
-        clearwait(runner);
+    case 0: // child in background
+        initchild(runner,0);
         pushredir(Ropen, null, 0);
-        if(shell.interactive){ // NOTE: this needs to read off its caller's interactivity
-            setpid(getpid());
-            defaultsignal();
-        }
-        run(runner->code.exe, runner->code.i+1, runner->local, 1);
-        runner->link = nil;
+
+        run(runner->code.exe, runner->code.i+1, runner->local, 0);
+        runner->caller = nil;
+        runner->flag.user = 0;
         break;
 
-    default: // parent
+    default: // parent in foreground
+        initparent(runner,pid,1);
         close(null);
-        setpid(pid);
-        addwait(runner, pid);
 
         runner->code.i = runner->code.exe[runner->code.i].i; /* jump to end of async command */
         /* don't wait: continue running */
@@ -727,33 +722,80 @@ Xasync(void)
 void
 Xsubshell(void)
 {
-    int pid;
+    int pid, user;
 
+    user = runner->flag.user;
     switch(pid=fork()){
     case -1:
         Xerror("fork failed: try again");
         break;
 
     case 0: // child
-        clearwait(runner);
-        if(shell.interactive){ // NOTE: this needs to read off its caller's interactivity
-            setpid(getpid());
-            tcsetpgrp(0, runner->pgid);
-            defaultsignal();
-        }
+        initchild(runner, 1);
         run(runner->code.exe, runner->code.i+1, runner->local, 1);
-        runner->link = nil;
+        runner->caller = nil;
         break;
 
     default: // parent
-        setpid(pid);
-        addwait(runner, pid);
-        waitfor(runner, pid); // wait until child finishes
+        initparent(runner, pid, 0); // relinquish control
+        waitfor(runner, pid);       // wait until child finishes
+        if(user){
+            tcsetpgrp(0, shell.pid);
+            runner->flag.user = 1;  // take control
+        }
+
         runner->code.i = runner->code.exe[runner->code.i].i; // jump to end of subshell command and continue execution
     }
 }
 
+void
+Xpipewait(void)
+{
+    foreground(runner, 0);
+}
+
+void
+Xpipe(void)
+{
+    Thread *orig;
+    int pc, pid, lfd, rfd, pfd[2];
+
+    orig = runner;
+    pc   = orig->code.i;
+    lfd  = orig->code.exe[pc++].i;
+    rfd =  orig->code.exe[pc++].i;
+
+    if(pipe(pfd)<0){
+        Xerror("can't get pipe\n");
+        return;
+    }
+
+    switch(pid=fork()){
+    case -1:
+        Xerror("try again");
+        break;
+    case 0:  // child
+        initchild(runner,1);
+
+        /* child 0 (writer) forked process */
+        run(runner->code.exe, pc+2, runner->local, 0);
+        runner->caller = nil;
 
+        close(pfd[0]);
+        pushredir(Ropen, pfd[1], lfd);
+        break;
+
+    default: // parent
+        initparent(runner,pid,0);
+
+        /* child 1 (reader) subprocess*/
+        run(runner->code.exe, runner->code.exe[pc].i, runner->local, 0);
+
+        close(pfd[1]);
+        pushredir(Ropen, pfd[0], rfd);
+        break;
+    }
+}
 
 void
 Xbasic(void)
@@ -783,6 +825,7 @@ Xbasic(void)
     }
 
     /* if we are here then it's an external command */
+    // TODO: check for if we will exit, don't need to fork, just exec
 
     // run the external command
     if((pid = xforkx()) < 0) {
@@ -790,9 +833,6 @@ Xbasic(void)
         return;
     }
 
-    if(!shell.interactive)
-        waitall(runner);
-
     foreground(runner, 0); // waits for child
     poplist();
 }
@@ -865,7 +905,78 @@ Xbang(void)
 }
 
 void
+Xpopredir(void)
+{
+    Redir *r = runner->redir.end;
+    if(!r)
+        fatal("attempted to pop a nil redir\n");
+
+    runner->redir.end = runner->redir.end->link;
+    if(r->type==Ropen)
+        close(r->from);
+
+    efree(r);
+}
+
+void
+Xreturn(void)
+{
+    Thread *curr = runner;
+
+    switch(curr->wait.status){
+    /*
+     * If our job is still running or suspended we must:
+     * 1. move program one step back to rerun Xreturn upon recall
+     * 2. return to our calling thread
+     * 3. don't free!
+     */
+     case Prun:
+         report(curr, 0);
+         curr->flag.user = 0;
+     case Pstop:
+        curr->code.i--;
+        runner = curr->caller;
+        curr->caller = nil; // detach job
+        return;
+    /*
+     * If our job has finished:
+     * 1. remove from our list
+     * 2. continue to clean up its memory
+     */
+    case Pdone:
+        deljob(curr);
+        /* fallthrough */
+    default:
+        ;
+    }
+
+    undoredirs();
+
+    while(curr->args)
+        poplist();
+    freecode(curr->code.exe);
+    efree(curr->wait.on);
+
+    runner = curr->caller;
+    efree(curr);
+    if(!runner)
+        exit(0);
+}
+
+
+
+void
 Xexit(void)
 {
     exit(runner->status);
 }
+
+void
+Xerror(char *msg)
+{
+    print(shell.err, "rc: %s", msg);
+    flush(shell.err);
+    while(!runner->flag.user)
+        Xreturn();
+}
+