move known-done subprocess from custodian

The `call-with-deep-time-limit` function in `racket/sandbox` expects a
subprocess to be removed from its custodian when the subprocess is
done. CS wasn't doing that at all, leaving custodian removal to a
finalizer. BC was doing delaying a remove until `subprocess-status` is
used (which happened to work for existing uses of
`call-with-deep-time-limit`, apparently.)

Relevant to #3140

pkgs/racket-test-core/tests/racket/subprocess.rktl

@@ -415,6 +415,20 @@
 (parameterize ([current-subprocess-custodian-mode #f])
   (test #f current-subprocess-custodian-mode))
 
+;; Check that a subprocess is removed from its custodian as
+;; soon as it's known to be done:
+(let* ([c (make-custodian)]
+       [c2 (make-custodian c)])
+  (define-values (sp i o e)
+    (parameterize ([current-custodian c2]
+                   [current-subprocess-custodian-mode 'kill])
+      (subprocess #f #f #f self "-e" "(read-byte)")))
+  (test #t pair? (member sp (custodian-managed-list c2 c)))
+  (close-output-port o)
+  (subprocess-wait sp)
+  (test #f pair? (member sp (custodian-managed-list c2 c)))
+  (custodian-shutdown-all c))
+
 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
 ;;; process groups
 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;

racket/src/io/subprocess/main.rkt

@@ -34,14 +34,18 @@
   prop:evt
   (poller (lambda (sp ctx)
             (define v (rktio_poll_process_done rktio (subprocess-process sp)))
-            (if (eqv? v 0)
-                (begin
+            (cond
+              [(eqv? v 0)
                (sandman-poll-ctx-add-poll-set-adder!
                 ctx
                 (lambda (ps)
                   (rktio_poll_add_process rktio (subprocess-process sp) ps)))
-                  (values #f sp))
-                (values (list sp) #f)))))
+               (values #f sp)]
+              [else
+               ;; Unregister from the custodian as soon as the process is known
+               ;; to be stopped:
+               (no-custodian! sp)
+               (values (list sp) #f)]))))
 
 (define do-subprocess
   (let ()
@@ -199,6 +203,7 @@
      (end-atomic)
      'running]
     [else
+     (no-custodian! sp)
      (define v (rktio_status_result r))
      (rktio_free r)
      (end-atomic)
@@ -231,6 +236,12 @@
 
 ;; ----------------------------------------
 
+;; in atomic mode
+(define (no-custodian! sp)
+  (when (subprocess-cust-ref sp)
+    (unsafe-custodian-unregister sp (subprocess-cust-ref sp))
+    (set-subprocess-cust-ref! sp #f)))
+
 (define subprocess-will-executor (make-will-executor))
 
 (define (register-subprocess-finalizer sp)
@@ -240,9 +251,7 @@
                    (when (subprocess-process sp)
                      (rktio_process_forget rktio (subprocess-process sp))
                      (set-subprocess-process! sp #f))
-                   (when (subprocess-cust-ref sp)
-                     (unsafe-custodian-unregister sp (subprocess-cust-ref sp))
-                     (set-subprocess-cust-ref! sp #f))
+                   (no-custodian! sp)
                    #t)))
 
 (define (poll-subprocess-finalizations)

racket/src/racket/src/port.c

@@ -5966,10 +5966,16 @@ static void child_mref_done(Scheme_Subprocess *sp)
 static int subp_done(Scheme_Object *so)
 {
   Scheme_Subprocess *sp = (Scheme_Subprocess*)so;
+  int done;
 
   if (!sp->proc) return 1;
 
-  return rktio_poll_process_done(scheme_rktio, sp->proc);
+  done = rktio_poll_process_done(scheme_rktio, sp->proc);
+
+  if (done)
+    child_mref_done(sp);
+
+  return done;
 }
 
 static void subp_needs_wakeup(Scheme_Object *so, void *fds)