cs & io: fix log receiver synchronization
A misplaced `wrap-evt` could allow the result from `sync` on a log receiver to be an opaque event, instead of a vector. In other cases, a differently misplaced `wrap-evt` could also cause an internal instance of `control-state-evt` to not be unregistered correctly. The solution to both problems is to add a wrapper procedure to `control-state-evt`. Closes #2664
This commit is contained in:
parent
2ad4c6f508
commit
067dda578b
23
pkgs/racket-test/tests/racket/stress/log-receiver.rkt
Normal file
23
pkgs/racket-test/tests/racket/stress/log-receiver.rkt
Normal file
|
@ -0,0 +1,23 @@
|
||||||
|
#lang racket/base
|
||||||
|
|
||||||
|
(define (do-thread t) (thread t))
|
||||||
|
(set! do-thread do-thread)
|
||||||
|
|
||||||
|
(let loop ([i 0])
|
||||||
|
(unless (= i 2000000)
|
||||||
|
(when (zero? (modulo i 10000))
|
||||||
|
(printf "~s\n" i))
|
||||||
|
(define (spin-a-while)
|
||||||
|
(let loop ([j (random (add1 (modulo i 100000)))])
|
||||||
|
(unless (zero? j)
|
||||||
|
(loop (sub1 j)))))
|
||||||
|
(define s (make-log-receiver (current-logger) 'info 'send))
|
||||||
|
(define t
|
||||||
|
(do-thread
|
||||||
|
(lambda ()
|
||||||
|
(log-message (current-logger) 'info 'send "a" 1))))
|
||||||
|
(spin-a-while)
|
||||||
|
(unless (vector? (sync s))
|
||||||
|
(error "not a vector result!"))
|
||||||
|
(thread-wait t)
|
||||||
|
(loop (add1 i))))
|
|
@ -42,12 +42,13 @@
|
||||||
|
|
||||||
(define (poll-ctx-sched-info ctx) #f)
|
(define (poll-ctx-sched-info ctx) #f)
|
||||||
|
|
||||||
(struct control-state-evt (evt interrupt abandon retry)
|
(struct control-state-evt (evt wrap interrupt abandon retry)
|
||||||
#:property prop:evt (lambda (cse)
|
#:property prop:evt (lambda (cse)
|
||||||
(nack-guard-evt
|
(nack-guard-evt
|
||||||
(lambda (nack)
|
(lambda (nack)
|
||||||
(thread (lambda () (sync nack) ((control-state-evt-abandon cse))))
|
(thread (lambda () (sync nack) ((control-state-evt-abandon cse))))
|
||||||
(control-state-evt-evt cse)))))
|
(wrap-evt (control-state-evt-evt cse)
|
||||||
|
(control-state-evt-wrap cse))))))
|
||||||
|
|
||||||
(define current-async-semaphore (make-parameter #f #f 'current-async-semaphore))
|
(define current-async-semaphore (make-parameter #f #f 'current-async-semaphore))
|
||||||
|
|
||||||
|
|
|
@ -55,7 +55,8 @@
|
||||||
(increment-receiever-waiters! lr)
|
(increment-receiever-waiters! lr)
|
||||||
(queue-add! (queue-log-receiver-waiters lr) b)))
|
(queue-add! (queue-log-receiver-waiters lr) b)))
|
||||||
(values #f (control-state-evt
|
(values #f (control-state-evt
|
||||||
(wrap-evt async-evt (lambda (e) (unbox b)))
|
async-evt
|
||||||
|
(lambda (e) (unbox b))
|
||||||
(lambda ()
|
(lambda ()
|
||||||
(queue-remove-node! (queue-log-receiver-waiters lr) n)
|
(queue-remove-node! (queue-log-receiver-waiters lr) n)
|
||||||
(decrement-receiever-waiters! lr))
|
(decrement-receiever-waiters! lr))
|
||||||
|
|
|
@ -79,6 +79,7 @@
|
||||||
(if (evt? v)
|
(if (evt? v)
|
||||||
v
|
v
|
||||||
(wrap-evt always-evt (lambda () v)))))
|
(wrap-evt always-evt (lambda () v)))))
|
||||||
|
values
|
||||||
void
|
void
|
||||||
(lambda () (semaphore-post s))
|
(lambda () (semaphore-post s))
|
||||||
void))))
|
void))))
|
||||||
|
|
|
@ -98,8 +98,8 @@
|
||||||
(current-thread/in-atomic)))
|
(current-thread/in-atomic)))
|
||||||
(define n (queue-add! gq (cons gw b)))
|
(define n (queue-add! gq (cons gw b)))
|
||||||
(values #f
|
(values #f
|
||||||
(wrap-evt
|
|
||||||
(control-state-evt async-evt
|
(control-state-evt async-evt
|
||||||
|
(lambda (v) (unbox b))
|
||||||
(lambda () (queue-remove-node! gq n))
|
(lambda () (queue-remove-node! gq n))
|
||||||
void
|
void
|
||||||
(lambda ()
|
(lambda ()
|
||||||
|
@ -112,8 +112,7 @@
|
||||||
(values #t #t)]
|
(values #t #t)]
|
||||||
[else
|
[else
|
||||||
(set! n (queue-add! gq (cons gw b)))
|
(set! n (queue-add! gq (cons gw b)))
|
||||||
(values #f #f)])))
|
(values #f #f)]))))]))
|
||||||
(lambda (v) (unbox b))))]))
|
|
||||||
|
|
||||||
;; ----------------------------------------
|
;; ----------------------------------------
|
||||||
|
|
||||||
|
@ -161,8 +160,8 @@
|
||||||
(current-thread/in-atomic)))
|
(current-thread/in-atomic)))
|
||||||
(define n (queue-add! pq (cons pw v)))
|
(define n (queue-add! pq (cons pw v)))
|
||||||
(values #f
|
(values #f
|
||||||
(wrap-evt
|
|
||||||
(control-state-evt async-evt
|
(control-state-evt async-evt
|
||||||
|
(lambda (v) self)
|
||||||
(lambda () (queue-remove-node! pq n))
|
(lambda () (queue-remove-node! pq n))
|
||||||
void
|
void
|
||||||
(lambda ()
|
(lambda ()
|
||||||
|
@ -175,8 +174,7 @@
|
||||||
(values self #t)]
|
(values self #t)]
|
||||||
[else
|
[else
|
||||||
(set! n (queue-add! pq (cons pw v)))
|
(set! n (queue-add! pq (cons pw v)))
|
||||||
(values #f #f)])))
|
(values #f #f)]))))]))
|
||||||
(lambda (v) self)))]))
|
|
||||||
|
|
||||||
(define/who (channel-put-evt ch v)
|
(define/who (channel-put-evt ch v)
|
||||||
(check who channel? ch)
|
(check who channel? ch)
|
||||||
|
|
|
@ -135,7 +135,15 @@
|
||||||
;; semaphore was meanwhile posted). As another example, a
|
;; semaphore was meanwhile posted). As another example, a
|
||||||
;; `nack-guard-evt`'s result uses `abandon-proc` to post to the NACK
|
;; `nack-guard-evt`'s result uses `abandon-proc` to post to the NACK
|
||||||
;; event.
|
;; event.
|
||||||
|
;; Beware that it doesn't make sense to use `wrap-evt` around the
|
||||||
|
;; `control-state-evt` or the `evt` inside for an asynchronously
|
||||||
|
;; satisfied event (like the way that semaphores are implemented). The
|
||||||
|
;; event may be selected asynchronously before a wrapper on the inner
|
||||||
|
;; event is found, so that the result turns out to be an unwrapped
|
||||||
|
;; event. Or the `interrupt-proc`, etc., callbacks may not be found
|
||||||
|
;; early enough if the `control-state-evt` is wrapped.
|
||||||
(struct control-state-evt (evt
|
(struct control-state-evt (evt
|
||||||
|
wrap-proc
|
||||||
interrupt-proc ; thunk for break/kill initiated or otherwise before `abandon-proc`
|
interrupt-proc ; thunk for break/kill initiated or otherwise before `abandon-proc`
|
||||||
abandon-proc ; thunk for not selected, including break/kill complete
|
abandon-proc ; thunk for not selected, including break/kill complete
|
||||||
retry-proc) ; thunk for resume from break; return `(values _val _ready?)`
|
retry-proc) ; thunk for resume from break; return `(values _val _ready?)`
|
||||||
|
|
|
@ -182,8 +182,8 @@
|
||||||
;; event through a callback. Pair the event with a nack callback
|
;; event through a callback. Pair the event with a nack callback
|
||||||
;; to get back out of line.
|
;; to get back out of line.
|
||||||
(values #f
|
(values #f
|
||||||
(wrap-evt
|
|
||||||
(control-state-evt async-evt
|
(control-state-evt async-evt
|
||||||
|
(lambda (v) result)
|
||||||
(lambda ()
|
(lambda ()
|
||||||
(assert-atomic-mode)
|
(assert-atomic-mode)
|
||||||
(queue-remove-node! s n)
|
(queue-remove-node! s n)
|
||||||
|
@ -202,8 +202,7 @@
|
||||||
[else
|
[else
|
||||||
(set! n (queue-add! s w))
|
(set! n (queue-add! s w))
|
||||||
(set-semaphore-count! s -1) ; so CAS not tried for `semaphore-post`
|
(set-semaphore-count! s -1) ; so CAS not tried for `semaphore-post`
|
||||||
(values #f #f)])))
|
(values #f #f)]))))]))
|
||||||
(lambda (v) result)))]))
|
|
||||||
|
|
||||||
;; Called only when it should immediately succeed:
|
;; Called only when it should immediately succeed:
|
||||||
(define (semaphore-wait/atomic s)
|
(define (semaphore-wait/atomic s)
|
||||||
|
|
|
@ -374,9 +374,9 @@
|
||||||
(let loop ([sr (syncing-syncers s)]
|
(let loop ([sr (syncing-syncers s)]
|
||||||
[retries 0] ; count retries on `sr`, and advance if it's too many
|
[retries 0] ; count retries on `sr`, and advance if it's too many
|
||||||
[polled-all-so-far? #t])
|
[polled-all-so-far? #t])
|
||||||
|
(start-atomic)
|
||||||
(when (syncing-need-retry? s)
|
(when (syncing-need-retry? s)
|
||||||
(syncing-retry! s))
|
(syncing-retry! s))
|
||||||
(start-atomic)
|
|
||||||
(cond
|
(cond
|
||||||
[(syncing-selected s)
|
[(syncing-selected s)
|
||||||
=> (lambda (sr)
|
=> (lambda (sr)
|
||||||
|
@ -473,9 +473,12 @@
|
||||||
(end-atomic)
|
(end-atomic)
|
||||||
(loop sr (add1 retries) polled-all-so-far?)])]
|
(loop sr (add1 retries) polled-all-so-far?)])]
|
||||||
[(control-state-evt? new-evt)
|
[(control-state-evt? new-evt)
|
||||||
|
(define wrap-proc (control-state-evt-wrap-proc new-evt))
|
||||||
(define interrupt-proc (control-state-evt-interrupt-proc new-evt))
|
(define interrupt-proc (control-state-evt-interrupt-proc new-evt))
|
||||||
(define abandon-proc (control-state-evt-abandon-proc new-evt))
|
(define abandon-proc (control-state-evt-abandon-proc new-evt))
|
||||||
(define retry-proc (control-state-evt-retry-proc new-evt))
|
(define retry-proc (control-state-evt-retry-proc new-evt))
|
||||||
|
(unless (eq? wrap-proc values)
|
||||||
|
(set-syncer-wraps! sr (cons wrap-proc (syncer-wraps sr))))
|
||||||
(unless (eq? interrupt-proc void)
|
(unless (eq? interrupt-proc void)
|
||||||
(set-syncer-interrupts! sr (cons interrupt-proc (syncer-interrupts sr))))
|
(set-syncer-interrupts! sr (cons interrupt-proc (syncer-interrupts sr))))
|
||||||
(unless (eq? abandon-proc void)
|
(unless (eq? abandon-proc void)
|
||||||
|
@ -726,6 +729,7 @@
|
||||||
;; represents the instantited attempt to sync on `evt`:
|
;; represents the instantited attempt to sync on `evt`:
|
||||||
(control-state-evt
|
(control-state-evt
|
||||||
(nested-sync-evt s next orig-evt)
|
(nested-sync-evt s next orig-evt)
|
||||||
|
values
|
||||||
(lambda () (syncing-interrupt! s))
|
(lambda () (syncing-interrupt! s))
|
||||||
(lambda () (syncing-abandon! s))
|
(lambda () (syncing-abandon! s))
|
||||||
(lambda () (syncing-retry! s)))))))
|
(lambda () (syncing-retry! s)))))))
|
||||||
|
|
|
@ -1000,7 +1000,8 @@
|
||||||
(set-thread-mailbox-wakeup! t (lambda () (wakeup) (receive))))
|
(set-thread-mailbox-wakeup! t (lambda () (wakeup) (receive))))
|
||||||
(add-wakeup-callback!)
|
(add-wakeup-callback!)
|
||||||
(values #f (control-state-evt
|
(values #f (control-state-evt
|
||||||
(wrap-evt async-evt (lambda (v) self))
|
async-evt
|
||||||
|
(lambda (v) self)
|
||||||
;; interrupt (all must be interrupted, so just install `void`):
|
;; interrupt (all must be interrupted, so just install `void`):
|
||||||
(lambda () (set-thread-mailbox-wakeup! t void))
|
(lambda () (set-thread-mailbox-wakeup! t void))
|
||||||
;; abandon:
|
;; abandon:
|
||||||
|
|
Loading…
Reference in New Issue
Block a user