cs: fix a leak in plumbers

Who plumbs the plumbers?

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

@@ -1325,6 +1325,19 @@
       (plumber-flush-all c)
       (test 6 values done))))
 
+;; Make sure plumbers don't suffer a key-in-value leak:
+(unless (eq? 'cgc (system-type 'gc))
+  (define p (make-plumber))
+  (define fh
+    (plumber-add-flush! (current-plumber)
+                        (lambda (fh) p)
+                        ;; weak:
+                        #t))
+  (plumber-add-flush! p (lambda (fh2) fh))
+  (define wb (make-weak-box p))
+  (collect-garbage)
+  (test #f weak-box-value wb))
+
 ;; ----------------------------------------
 
 ;; Check that a terminated thread cleans up ownership

racket/src/thread/bootstrap.rkt

@@ -184,6 +184,9 @@
 
 (define place-done-prompt (make-continuation-prompt-tag 'place-done))
 
+(define-values (prop:unsafe-authentic-override unsafe-authentic-override? unsafe-authentic-override-ref)
+  (make-struct-type-property 'unsafe-authentic-override))
+
 ;; Beware that this implementation of `fork-place` doesn't support
 ;; rktio-based blocking in different places. So, be careful of the
 ;; preliminary tests that you might try with the "io" layer and
@@ -285,7 +288,8 @@
                   'mutex-release (lambda (s) (semaphore-post s))
                   'call-as-asynchronous-callback (lambda (thunk) (thunk))
                   'post-as-asynchronous-callback (lambda (thunk) (thunk))
-                  'continuation-current-primitive (lambda (k) #f)))
+                  'continuation-current-primitive (lambda (k) #f)
+                  'prop:unsafe-authentic-override prop:unsafe-authentic-override))
 
 ;; add dummy definitions that implement pthreads and conditions etc.
 ;; dummy definitions that error

racket/src/thread/demo.rkt

@@ -108,16 +108,6 @@
      (check (void) (sync/timeout 0 v)))
 
    ;; evt chaperone
-   (define e1 (make-semaphore 1))
-   (check #t (chaperone-of? (chaperone-evt e1 void) e1))
-   (check #f (chaperone-of? e1 (chaperone-evt e1 void)))
-   (let ([hit #f])
-     (check e1 (sync (chaperone-evt e1 (lambda (e)
-                                         (set! hit e)
-                                         (values e values)))))
-     (check e1 hit))
-   (check #t (semaphore? (chaperone-evt e1 void)))
-
    (check #t (chaperone-of? (chaperone-evt ch void) ch))
    (check #t (channel? (chaperone-evt ch void)))
    (check #t (channel? (chaperone-channel ch void void)))

racket/src/thread/plumber.rkt

@@ -26,17 +26,17 @@
                     v)
                   'current-plumber))
 
-(struct plumber-flush-handle (plumber))
+(struct plumber-flush-handle (plumber proc))
 
 (define/who (plumber-add-flush! p proc [weak? #f])
   (check who plumber? p)
   (check who (procedure-arity-includes/c 1) proc)
-  (define h (plumber-flush-handle p))
+  (define h (plumber-flush-handle p proc))
   (hash-set! (if weak?
                  (plumber-weak-callbacks p)
                  (plumber-callbacks p))
              h
-             proc)
+             #t)
   h)
 
 (define/who (plumber-flush-all p)
@@ -45,13 +45,13 @@
 
 (define (plumber-flush-all/wrap p app)
   ;; Spec requires getting all callbacks before running any
-  (define procs+hs
+  (define hs
     (for*/list ([cbs (in-list (list (plumber-callbacks p)
                                     (plumber-weak-callbacks p)))]
-                [(h proc) (in-hash cbs)])
-      (cons proc h)))
-  (for ([proc+h (in-list procs+hs)])
-    (app (car proc+h) (cdr proc+h))))
+                [h (in-hash-keys cbs)])
+      h))
+  (for ([h (in-list hs)])
+    (app (plumber-flush-handle-proc h) h)))
 
 (define/who (plumber-flush-handle-remove! h)
   (check who plumber-flush-handle? h)