cs & thread: fix problems with sync and breaks

This commit fixes two bugs:

 * `sync/enable-break` didn't implement the guarantee that an event is
   selected or a break exception raised, but not both; the problem was
   in the handling of making the break-enable state ignored after
   committing to an event

 * `sync` didn't cancel asynchronous pending commits when a break is
   received at certain points; the bug in `sync/enable-break` masked
   this bug for existing test cases

Closes #3574

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

@@ -1226,6 +1226,9 @@
       [did-post2 #f]
       [did-done #f]
       [break-on (lambda () (break-enabled #t))]
+      [sync-idle/break-thread (lambda (t)
+                                (sync (system-idle-evt))
+                                (break-thread t))]
       [sw semaphore-wait])
   (let ([mk-t
 	 (lambda (init ;; how to start
@@ -1273,8 +1276,8 @@
 			reset
 			(lambda ()
 			  (set! did-pre1 #t)
-			  (semaphore-post p)
 			  (pre-thunk)
+			  (semaphore-post p)
 			  (pre-semaphore-wait s)
 			  (set! did-pre2 #t))))
 		     (lambda () 
@@ -1282,8 +1285,8 @@
 			reset
 			(lambda ()
 			  (set! did-act1 #t)
-			  (semaphore-post p)
 			  (act-thunk)
+			  (semaphore-post p)
 			  (act-semaphore-wait s)
 			  (set! did-act2 #t))))
 		     (lambda ()
@@ -1291,8 +1294,8 @@
 			reset
 			(lambda ()
 			  (set! did-post1 #t)
-			  (semaphore-post p)
 			  (post-thunk)
+			  (semaphore-post p)
 			  (post-semaphore-wait s)
 			  (set! did-post2 #t)))))
 		 (set! did-done #t))))))])
@@ -1338,6 +1341,7 @@
 	(if should-pre-break?
 	    (begin
               (thread-wait t)
+              (test #t semaphore-try-wait? s)
 	      (test #f 'pre2 did-pre2))
 	    (if should-preact-break?
 		(begin
@@ -1365,6 +1369,7 @@
 			(if should-post-break?
 			    (begin
 			      (thread-wait t)
+                              (test #t semaphore-try-wait? s)
 			      (test #f 'post2 did-post2))
 			    (begin
 			      (thread-wait t)
@@ -1374,6 +1379,8 @@
      (lambda (mk-t)
        (for-each 
 	(lambda (nada)
+          (for-each
+           (lambda (break-thread)
              ;; Basic checks --- dynamic-wind thunks don't explicitly enable breaks
              (go mk-t #f  nada nada nada  sw sw sw  void #f #f void #f void #f #f)
              (go mk-t #f  nada nada nada  sw sw sw  break-thread #f 'pre-act void #f void #f #f)
@@ -1396,6 +1403,7 @@
              ;; Enable break in pre shouldn't affect act/done
              (go mk-t #t  break-on nada nada  sw sw sw  void #f #f break-thread #f void #f #f)
              (go mk-t #t  break-on nada nada  sw sw sw  void #f #f void #f break-thread #f #f))
+           (list break-thread sync-idle/break-thread)))
         (list void sleep)))
      ;; We'll make threads in three modes: normal, restore a continuation into pre,
      ;;  and restore a continuation into act
@@ -1576,6 +1584,40 @@
   (try (lambda (c) (choice-evt c (alarm-evt (+ 10000 (current-inexact-milliseconds)))))
        'ok-channel+alarm))
 
+;; ----------------------------------------
+;; Extra check that a sync/enable-break either succeeds or breaks,
+;; where we avoid explicit thread synchronization and spin to
+;; try to get different schedules.
+;; Based on an example from Bogdan.
+
+(for ([i 50])
+  (let ([succeeded? #f]
+        [broke? #f])
+    (define (spin-then thunk)
+      (let loop ([n (random 1000000)])
+        (cond
+          [(zero? n) (thunk)]
+          [else (loop (sub1 n))])))
+    (struct p ()
+      #:property prop:evt (unsafe-poller
+                           (lambda (self wakeups)
+                             ;; loop to try to be slow
+                             (spin-then
+                              (lambda ()
+                                (set! succeeded? #t)
+                                (values (list 'success) #f))))))
+    (define ready-sema (make-semaphore))
+    (define t (thread (lambda ()
+                        (parameterize-break #f
+                          (semaphore-post ready-sema)
+                          (with-handlers ([exn:break? (lambda (exn)
+                                                        (set! broke? #t))])
+                            (sync/enable-break (p)))))))
+    (semaphore-wait ready-sema)
+    (spin-then (lambda () (break-thread t)))
+    (sync t)
+    (test #t 'xor-on-success-and-break (if succeeded? (not broke?) broke?))))
+
 ;; ----------------------------------------
 ;; Make sure that suspending a thread that's blocked on a
 ;; semaphore works right when the semaphore becomes available

racket/src/cs/schemified/thread.scm

@@ -7010,8 +7010,7 @@
                          "attempt to deschedule the current thread in atomic mode"))
                       (void)))))))
              (loop_0))
-            (engine-block)
+            (engine-block))
-            (|#%app| 1/check-for-break))
           (void))))))
 (define thread-deschedule!
   (lambda (t_0 timeout-at_0 interrupt-callback_0)
@@ -7703,6 +7702,8 @@
                    (begin
                      (set-thread-pending-break! t_0 kind_0)
                      (thread-did-work!)
+                     (run-suspend/resume-callbacks t_0 car)
+                     (run-suspend/resume-callbacks t_0 cdr)
                      (if (thread-descheduled? t_0)
                        (if (thread-suspended? t_0)
                          (void)
@@ -9197,11 +9198,7 @@
            (go_0
             (|#%name|
              go
-             (lambda (enable-break?7_0
+             (lambda (enable-break?7_0 s_0 timeout10_0 thunk-result?38_0)
-                      local-break-cell_0
-                      s_0
-                      timeout10_0
-                      thunk-result?38_0)
                (begin
                  (dynamic-wind
                   (lambda ()
@@ -9239,11 +9236,6 @@
                       (start-atomic)
                       (thread-pop-suspend+resume-callbacks!)
                       (thread-pop-kill-callback!)
-                      (if local-break-cell_0
-                        (thread-remove-ignored-break-cell!
-                         (current-thread/in-atomic)
-                         local-break-cell_0)
-                        (void))
                       (|#%app| syncing-abandon! s_0)
                       (end-atomic))))))))
            (loop_0
@@ -9406,13 +9398,13 @@
                            push-authentic
                            break-enabled-key
                            local-break-cell_0
-                           (go_0
+                           (go_0 enable-break?7_0 s_0 timeout10_0 #t))))
-                            enable-break?7_0
+                     (begin
-                            local-break-cell_0
+                       (thread-remove-ignored-break-cell!
-                            s_0
+                        (current-thread/in-atomic)
-                            timeout10_0
+                        local-break-cell_0)
-                            #t))))
+                       (1/check-for-break)
-                     (begin (1/check-for-break) (|#%app| thunk_0)))
+                       (|#%app| thunk_0)))
                    (let ((temp52_0
                           (lambda (sched-info_0 polled-all?_0 no-wrappers?_0)
                             (if polled-all?_0
@@ -9423,26 +9415,15 @@
                                 (if (procedure? timeout10_0)
                                   (|#%app| timeout10_0)
                                   (if no-wrappers?_0
-                                    (go_0
+                                    (go_0 enable-break?7_0 s_0 timeout10_0 #f)
-                                     enable-break?7_0
-                                     local-break-cell_0
-                                     s_0
-                                     timeout10_0
-                                     #f)
                                     (|#%app|
                                      (go_0
                                       enable-break?7_0
-                                      local-break-cell_0
                                       s_0
                                       timeout10_0
                                       #t)))))
                               (|#%app|
-                               (go_0
+                               (go_0 enable-break?7_0 s_0 timeout10_0 #t))))))
-                                enable-break?7_0
-                                local-break-cell_0
-                                s_0
-                                timeout10_0
-                                #t))))))
                      (|#%app|
                       sync-poll.1
                       #f

racket/src/thread/sync.rkt

@@ -167,8 +167,6 @@
        (atomically
         (thread-pop-suspend+resume-callbacks!)
         (thread-pop-kill-callback!)
-        (when local-break-cell
-          (thread-remove-ignored-break-cell! (current-thread/in-atomic) local-break-cell))
         ;; On escape, post nacks, etc.:
         (syncing-abandon! s)
         (void)))))
@@ -181,6 +179,9 @@
                     break-enabled-key
                     local-break-cell
                     (go)))
+     ;; If we get here, the break wasn't triggered, and it must be currently ignored.
+     ;; (If the break was triggered so that we don't get here, it's not ignored.)
+     (thread-remove-ignored-break-cell! (current-thread/in-atomic) local-break-cell)
      ;; In case old break cell was meanwhile enabled:
      (check-for-break)
      ;; In tail position:

racket/src/thread/thread.rkt

@@ -440,8 +440,8 @@
           (if (force-atomic-timeout-callback)
               (loop)
               (internal-error "attempt to deschedule the current thread in atomic mode"))))
-      (engine-block)
+      ;; implies `(check-for-break)`:
-      (check-for-break))))
+      (engine-block))))
 
 ;; Extends `do-thread-deschdule!` where `t` is always `(current-thread)`.
 ;; The `interrupt-callback` is called if the thread receives a break
@@ -836,6 +836,10 @@
        (unless (thread-pending-break t)
          (set-thread-pending-break! t kind)
          (thread-did-work!)
+         (begin
+           ;; interrupt synchronization, if any
+           (run-suspend/resume-callbacks t car)
+           (run-suspend/resume-callbacks t cdr))
          (when (thread-descheduled? t)
            (unless (thread-suspended? t)
              (run-interrupt-callback t)