feat(rc): prototype of async jobs

7 files changed, 125 insertions, 101 deletions

diff --git a/sys/cmd/rc/code.c b/sys/cmd/rc/code.c
index c98cde2..719e63b 100644
--- a/sys/cmd/rc/code.c
+++ b/sys/cmd/rc/code.c
@@ -30,7 +30,7 @@ grow(void)
 
 static
 void
-stashaddr(int a)
+storepc(int a)
 {
     if(interpreter.i <= a || a < 0)
         fatal("bad address %d in interpreter", a);
@@ -42,8 +42,8 @@ static
 void
 walk(Tree *node)
 {
-    int p;
     Tree *n;
+    int addr;
 
     if(!node)
         return;
@@ -72,17 +72,6 @@ walk(Tree *node)
         emitf(Xflat);
         break;
 
-    case Tindex:
-
-#if 0
-    case '&':
-        emitf(Xasync);
-        p = emiti(0);
-        emitf(Xexit);
-        stashaddr(p);
-        break;
-#endif
-
     case ';':
         walk(node->child[0]);
         walk(node->child[1]);
@@ -113,6 +102,14 @@ walk(Tree *node)
         emitf(Xbasic);
         break;
 
+    case '&':
+        emitf(Xasync);
+        addr = emiti(0);
+        walk(node->child[0]);
+        emitf(Xexit);
+        storepc(addr);
+        break;
+
     case Tword:
         emitf(Xword);
         emits(strdup(node->str));
@@ -173,18 +170,16 @@ int
 compile(Tree *node)
 {
     flush(shell.err);
-    //print(errio, "%t\n", node);
 
     interpreter.i    = 0;
     interpreter.code = emalloc(100*sizeof(*interpreter.code)); 
-
-    emiti(1); // reference count
+    emiti(0); // reference count: no thread owns code yet
 
     walk(node);
 
     emitf(Xreturn);
     emitf(nil);
 
-    compiled =  interpreter.code;
+    compiled = interpreter.code;
     return 0;
 }
diff --git a/sys/cmd/rc/exec.c b/sys/cmd/rc/exec.c
index 87b6bb7..37c289b 100644
--- a/sys/cmd/rc/exec.c
+++ b/sys/cmd/rc/exec.c
@@ -135,6 +135,31 @@ path(char *w)
     return path;
 }
 
+static inline
+void
+defaultsignal(void)
+{
+    signal(SIGINT,  SIG_DFL);
+    signal(SIGQUIT, SIG_DFL);
+    signal(SIGTSTP, SIG_DFL);
+    signal(SIGTTIN, SIG_DFL);
+    signal(SIGTTOU, SIG_DFL);
+    signal(SIGCHLD, SIG_DFL);
+}
+
+static inline
+void
+setpid(int pid)
+{
+    runner->pid = pid;
+    if(runner->pgid <= 0){
+        runner->pgid = runner->pid;
+        addjob(runner);
+    }
+
+    setpgid(pid, runner->pgid);
+}
+
 static
 void
 xx(void)
@@ -161,21 +186,12 @@ xforkx(void)
         return -1;
     case 0: // child
         clearwait(runner);
+        runner->link = nil;
 
         if(shell.interactive){
-            runner->pid = getpid();
-            if(runner->pgid <= 0)
-                runner->pgid = runner->pid;
-
-            setpgid(pid, runner->pgid);
+            setpid(getpid());
             tcsetpgrp(0, runner->pgid);
-
-            signal(SIGINT,  SIG_DFL);
-            signal(SIGQUIT, SIG_DFL);
-            signal(SIGTSTP, SIG_DFL);
-            signal(SIGTTIN, SIG_DFL);
-            signal(SIGTTOU, SIG_DFL);
-            signal(SIGCHLD, SIG_DFL);
+            defaultsignal();
         }
 
         pushword("exec");
@@ -185,12 +201,7 @@ xforkx(void)
         /*unreachable*/
     default: // parent
         addwait(runner, pid);
-
-        /* ensure the state matches for parent and child */
-        runner->pid = pid;
-        if(runner->pgid <= 0)
-            runner->pgid = pid;
-        setpgid(pid, runner->pgid);
+        setpid(pid); // ensure the state matches for parent and child
 
         return pid;
     }
@@ -213,7 +224,7 @@ pushredir(int type, int from, int to)
 /* byte code */
 static
 void
-run(Code *c, int pc, Var *local)
+run(Code *c, int pc, Var *local, int keeppid)
 {
     Thread *new = emalloc(sizeof(*new));
 
@@ -236,7 +247,10 @@ run(Code *c, int pc, Var *local)
     new->wait.on     = nil;
 
     new->status = 0;
-    new->pgid = new->pid = -1;
+    if(keeppid)
+        new->pgid = runner->pgid, new->pid = runner->pid;
+    else
+        new->pgid = new->pid = -1;
 
     new->line     = 0;
     new->link     = runner;
@@ -285,9 +299,9 @@ count(Word *w)
 // -----------------------------------------------------------------------
 // builtins
 
-static Code dotcmd[14] = 
+static Code dotcmd[14] =
 {
-    [0]  = {.i = 1},
+    [0]  = {.i = 0},
     [1]  = {.f = Xmark},
     [2]  = {.f = Xword},
     [3]  = {.s = "0"},
@@ -346,7 +360,7 @@ xdot(void)
     }
     /* set up for a new command loop */
     old = runner; // store pointer to old code
-    run(dotcmd, 1, nil);
+    run(dotcmd, 1, nil, 0);
 
     /* operations on new command stack */
     pushredir(Rclose, fd, 0);
@@ -434,7 +448,7 @@ xboot(int argc, char *argv[])
     bootstrap[i++].f = Xexit;
     bootstrap[i].i = 0;
 
-    run(bootstrap, 1, nil);
+    run(bootstrap, 1, nil, 0);
     pushlist(); // prime bootstrap argv
 
     argv0 = strdup(argv[0]);
@@ -478,7 +492,7 @@ Xreturn(void)
      * 2. return to our calling thread
      * 3. don't free!
      */
-    case PStop:
+     case Prun: case Pstop:
         run->code.i--;
         runner = run->link;
         return;
@@ -487,7 +501,7 @@ Xreturn(void)
      * 1. remove from our list
      * 2. continue to clean up its memory
      */
-    case PDone:
+    case Pdone:
         deljob(run);
         /* fallthrough */
     default:
@@ -589,7 +603,7 @@ Xreadcmd(void)
             --root->code.i;
     }else{
         --root->code.i;    /* re-execute Xreadcmd after codebuf runs */
-        run(compiled, 1, root->local);
+        run(compiled, 1, root->local, 0);
     }
 
     killzombies();
@@ -630,6 +644,43 @@ Xunlocal(void)
 }
 
 void
+Xasync(void)
+{
+    int pid;
+    int null = open("/dev/null", 0);
+    if(!null){
+        Xerror("can not open /dev/null\n");
+        return;
+    }
+
+    switch(pid=fork()){
+    case -1:
+        close(null);
+        Xerror("fork failed: try again");
+        break;
+
+    case 0: // child
+        clearwait(runner);
+        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;
+        break;
+
+    default: // parent
+        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 */
+    }
+}
+
+void
 Xbasic(void)
 {
     Var  *v;
@@ -642,7 +693,6 @@ Xbasic(void)
         Xerror("empty argument list\n");
         return;
     }
-    flush(shell.err);
 
     v = var(arg->str);
     if(v->func){
@@ -658,7 +708,6 @@ Xbasic(void)
     }
 
     /* if we are here then it's an external command */
-    addjob(runner);
 
     // run the external command
     if((pid = xforkx()) < 0) {
@@ -666,12 +715,11 @@ Xbasic(void)
         return;
     }
 
-    poplist();
-
     if(!shell.interactive)
         waitall(runner);
 
     foreground(runner, 0); // waits for child
+    poplist();
 }
 
 void
@@ -735,5 +783,5 @@ Xflat(void)
 void
 Xexit(void)
 {
-    exit(shell.status);
+    exit(runner->status);
 }
diff --git a/sys/cmd/rc/exec.h b/sys/cmd/rc/exec.h
index 566c4a3..0df01d0 100644
--- a/sys/cmd/rc/exec.h
+++ b/sys/cmd/rc/exec.h
@@ -15,6 +15,7 @@ void Xlocal(void);
 void Xreadcmd(void);
 void Xunlocal(void);
 void Xassign(void);
+void Xasync(void);
 void Xbasic(void);  // Xbasic(args) run command and wait for result
 void Xword(void);
 void Xcount(void);
diff --git a/sys/cmd/rc/job.c b/sys/cmd/rc/job.c
index 779ce6c..a8747e9 100644
--- a/sys/cmd/rc/job.c
+++ b/sys/cmd/rc/job.c
@@ -21,16 +21,16 @@ void
 report(Thread *job, int index)
 {
     switch(job->wait.status){
-        case PDone:
+        case Pdone:
             print(shell.err, "job %d [%d]: done\n",      index, job->pid);
             break;
-        case PStop:
+        case Pstop:
             print(shell.err, "job %d [%d]: suspended\n", index, job->pid);
             break;
-        case PAgain:
+        case Pagain:
             print(shell.err, "job %d [%d]: continued\n", index, job->pid);
             break;
-        case PRun:
+        case Prun:
             print(shell.err, "job %d [%d]: running\n",   index, job->pid);
             break;
         default:
@@ -42,10 +42,10 @@ void
 wakeup(Thread *t)
 {
     int i;
-    t->wait.status = PRun;
+    t->wait.status = Prun;
     for(i=0; i < t->wait.len; i++){
-        if(t->wait.on[i].status == PStop)
-            t->wait.on[i].status = PRun;
+        if(t->wait.on[i].status == Pstop)
+            t->wait.on[i].status = Prun;
     }
 }
 
@@ -68,6 +68,7 @@ addjob(Thread *job)
 {
     job->next = shell.jobs;
     shell.jobs = job;
+    job->wait.status = Prun;
 }
 
 void
diff --git a/sys/cmd/rc/rc.h b/sys/cmd/rc/rc.h
index ef7c18e..55a4eb2 100644
--- a/sys/cmd/rc/rc.h
+++ b/sys/cmd/rc/rc.h
@@ -95,12 +95,12 @@ struct Redir
 
 enum
 {
-    PNil,
-    PRun,
-    PStop,
-    PSig,
-    PAgain,
-    PDone,
+    Pnil,
+    Prun,
+    Pstop,
+    Psig,
+    Pagain,
+    Pdone,
 };
 
 struct WaitItem
diff --git a/sys/cmd/rc/sys.c b/sys/cmd/rc/sys.c
index 643f327..d286c88 100644
--- a/sys/cmd/rc/sys.c
+++ b/sys/cmd/rc/sys.c
@@ -88,5 +88,5 @@ execute(Word *cmd, Word *path)
         }
     }
     efree(argv);
-    fatal("failed to exec\n");
+    fatal("could not execute command: %s\n", path->str);
 }
diff --git a/sys/cmd/rc/wait.c b/sys/cmd/rc/wait.c
index 44ffb58..e3dcea1 100644
--- a/sys/cmd/rc/wait.c
+++ b/sys/cmd/rc/wait.c
@@ -41,7 +41,7 @@ await(int pid4, int opt, struct WaitMsg *msg)
             msg->time[1] = s;
             msg->time[2] = u+s;
             msg->status  = WEXITSTATUS(status);
-            msg->type    = PDone;
+            msg->type    = Pdone;
 
             return 1;
         }
@@ -52,7 +52,7 @@ await(int pid4, int opt, struct WaitMsg *msg)
             msg->time[1] = s;
             msg->time[2] = u+s;
             msg->status  = WCOREDUMP(status);
-            msg->type    = PSig;
+            msg->type    = Psig;
 
             return 1;
         }
@@ -63,7 +63,7 @@ await(int pid4, int opt, struct WaitMsg *msg)
             msg->time[1] = s;
             msg->time[2] = u+s;
             msg->status  = WSTOPSIG(status);
-            msg->type    = PStop;
+            msg->type    = Pstop;
 
             return 1;
         }
@@ -77,7 +77,7 @@ shouldwait(Thread *job)
     int i;
 
     for(i=0; i<job->wait.len; i++){
-        if(job->wait.on[i].status == PRun)
+        if(job->wait.on[i].status == Prun)
             return 1;
     }
 
@@ -112,7 +112,7 @@ addwait(Thread *job, int pid)
         job->wait.on  = erealloc(job->wait.on, job->wait.cap*sizeof(*job->wait.on));
     }
 
-    job->wait.on[job->wait.len++] = (struct WaitItem){.pid=pid, .status=PRun};
+    job->wait.on[job->wait.len++] = (struct WaitItem){.pid=pid, .status=Prun};
 }
 
 void
@@ -142,21 +142,22 @@ waitall(Thread *job)
 
         for(i=0; i < job->wait.len; i++){
             if(job->wait.on[i].pid == msg.pid){
+                job->status = msg.status;
                 switch(msg.type){
-                case PStop:
+                case Pstop:
                     print(shell.err, "%d: suspended\n", msg.pid);
-                    job->wait.status = PStop;
-                    job->wait.on[i].status = PStop;
+                    job->wait.status = Pstop;
+                    job->wait.on[i].status = Pstop;
                     break;
 
-                case PSig:
+                case Psig:
                     print(shell.err, "%d: terminated by signal %d\n", msg.pid, msg.status);
                     /* fallthrough */
-                case PDone:
-                    job->wait.on[i].status = PDone;
+                case Pdone:
+                    job->wait.on[i].status = Pdone;
                     delwait(job, msg.pid);
                     if(!job->wait.len)
-                        job->wait.status = PDone;
+                        job->wait.status = Pdone;
                     break;
 
                 default:
@@ -166,28 +167,6 @@ waitall(Thread *job)
             }
         }
     }
-#if 0
-        for(t=job; t; t=t->link){
-            if(t->pid == msg.pid){
-                t->status = msg.status;
-                if(msg.suspend){
-                    t->flag.stop = 1;
-                }else{
-                    t->flag.done = 1;
-                    if(msg.signal){
-                        print(shell.err, "%d: terminated by signal %d\n", msg.pid, msg.status);
-                    }
-                }
-                delwait(t, msg.pid);
-                goto outer;
-            }
-        }
-        perror("waitpid");
-        return 0;
-    outer:;
-    }
-#endif
-
     return 1;
 }
 
@@ -206,13 +185,13 @@ killzombies(void)
             perror("invalid pid");
 
         if(WIFEXITED(status))
-            job->wait.status = PDone;
+            job->wait.status = Pdone;
         if(WIFSTOPPED(status))
-            job->wait.status = PStop;
+            job->wait.status = Pstop;
         if(WIFCONTINUED(status))
-            job->wait.status = PAgain;
+            job->wait.status = Pagain;
 
-        if(job->wait.status == PDone){
+        if(job->wait.status == Pdone){
             report(job,index);
             deljob(job);
         }