CE & init use new streams IPC

aux/qemu_amd64.sh

@@ -2,4 +2,4 @@
 
 set -x
 
-qemu-system-x86_64 -M pc -m 4G -serial stdio -enable-kvm -cdrom mop3.iso -smp 1 -boot d $@
+qemu-system-x86_64 -M pc -m 4G -serial stdio -enable-kvm -cdrom mop3.iso -smp 4 -boot d $@

ce/interp.c

@@ -340,37 +340,34 @@ static void help (struct context* context) {
   cprintf (context, "quit\n");
 }
 
-/* static void cmd_cancel_proc (void* arg) { */
-/*   int pid = (int)(uintptr_t)arg; */
+static void cmd_cancel_proc (void* arg) {
+  int pid = (int)(uintptr_t)arg;
 
-/*   int ch = 0; */
+  uint8_t ch = 0;
+  for (;;) {
+    if (stream_read (process_get_pgid (), STREAM_IN, &ch, 1) > 0 && ch == KB_CTRL ('C'))
+      break;
+    else
+      sched ();
+  }
 
-/*   for (;;) { */
-/*     ch = kb_read_key_nonblock (); */
+  kill (pid);
+}
 
-/*     if (ch == KB_CTRL ('C')) */
-/*       break; */
-/*     else */
-/*       sched (); */
-/*   } */
+static void cmd_collect_proc (void* arg) {
+#define RECV_MAX (1024 * 16)
 
-/*   kill (pid); */
-/* } */
+  int pgid = (int)(uintptr_t)arg;
 
-/* static void cmd_collect_proc (void* arg) { */
-/* #define RECV_MAX (1024 * 16) */
-
-/*   (void)arg; */
-
-/*   char recv[RECV_MAX]; */
-/*   for (;;) { */
-/*     memset (recv, 0, sizeof (recv)); */
-/*     if (mail_receive_nonblock (&recv, sizeof (recv) - 1) == ST_OK) */
-/*       mail_send (process_get_exec_pgid (), recv, strlen (recv)); */
-/*     else */
-/*       sched (); */
-/*   } */
-/* } */
+  char recv[RECV_MAX];
+  int n;
+  for (;;) {
+    if ((n = stream_read (pgid, STREAM_OUT, (void*)recv, sizeof (recv) - 1)) > 0)
+      stream_write (process_get_pgid (), STREAM_OUT, (void*)recv, n);
+    else
+      sched ();
+  }
+}
 
 static void execute_cmd (struct ast_cmd* cmd, struct context* context) {
   if (strcmp (cmd->name, "echo") == 0) {
@@ -427,6 +424,8 @@ static void execute_cmd (struct ast_cmd* cmd, struct context* context) {
       return;
     }
 
+    int pgid = get_procgroup (pid);
+
     for (int i = 0; i < cmd->arg_count; i++) {
       char* arg = cmd->args[i];
       char *key, *value;
@@ -439,20 +438,20 @@ static void execute_cmd (struct ast_cmd* cmd, struct context* context) {
         key = arg;
         value = sep + 1;
 
-        env_set (pid, key, value, strlen (value) + 1);
+        env_set (pgid, key, value, strlen (value) + 1);
       }
     }
 
-    /* struct process_data* collect_pdata = process_spawn (&cmd_collect_proc, NULL); */
-    /* struct process_data* cancel_pdata = process_spawn (&cmd_cancel_proc, (void*)pid); */
+    struct process_data* collect_pdata = process_spawn (&cmd_collect_proc, (void*)(uintptr_t)pgid);
+    struct process_data* cancel_pdata = process_spawn (&cmd_cancel_proc, (void*)(uintptr_t)pid);
 
     exec_partial_fini (pid);
     wait_for_pid (pid);
 
-    /* kill (collect_pdata->pid); */
-    /* process_data_free (collect_pdata); */
-    /* kill (cancel_pdata->pid); */
-    /* process_data_free (cancel_pdata); */
+    kill (collect_pdata->pid);
+    process_data_free (collect_pdata);
+    kill (cancel_pdata->pid);
+    process_data_free (cancel_pdata);
   }
 }
 

init/init.c

@@ -19,8 +19,6 @@ void receiver (void* arg) {
   char recv[RECV_MAX];
   int n;
   for (;;) {
-    memset (recv, 0, sizeof (recv));
-
     n = stream_read (ce_pgid, STREAM_OUT, (void*)recv, RECV_MAX - 1);
 
     if (n > 0)