counter.rkt

#lang racket/base

(require racket/async-channel)

(define-logger counter)

(struct waiter (threshold ch) #:transparent)
(struct counter (add-ch incr-ch thread))

(define (make-counter)
  (define add-ch (make-channel))
  (define incr-ch (make-async-channel))
  (define (service-waiting count waiting triggered)
    (define-values (new-waiting new-triggered)
      (for/fold ([new-waiting null]
                 [new-triggered triggered])
                ([w (in-list waiting)])
        (cond
          [(>= count (waiter-threshold w))
           (values new-waiting
                   (cons (channel-put-evt (waiter-ch w) #t)
                         new-triggered))]
          [else
           (values (cons w new-waiting)
                   new-triggered)])))
    (log-counter-debug "service: ~a ~s ~s" count new-waiting new-triggered)
    (run count new-waiting new-triggered))
  (define (run count waiting triggered)
    (define (add-waiter w)
      (log-counter-debug "adding waiter ~s" w)
      (service-waiting count (cons w waiting) triggered))
    (define (incr-count amount)
      (service-waiting (+ count amount) waiting triggered))      
    (define (remove-triggered evt)
      (log-counter-debug "removing triggered ~s" evt)
      (run count waiting (remq evt triggered)))
    (sync (handle-evt add-ch  add-waiter)
          (handle-evt incr-ch incr-count)
          (guard-evt
           (lambda ()
             (if (null? triggered)
                 never-evt
                 (handle-evt (apply choice-evt triggered) remove-triggered))))))
  (define work-thread (thread (lambda () (run 0 null null))))
  (counter add-ch incr-ch work-thread))

(define (counter-incr! a-counter [amount 1])
  (async-channel-put (counter-incr-ch a-counter) amount))

(define (counter-wait! a-counter amount)
  (define wait-ch (make-channel))
  (channel-put (counter-add-ch a-counter)
               (waiter amount wait-ch))
  (sync (wrap-evt wait-ch void)))

repl.rkt

> (define a-counter (make-counter))
> (counter-incr! a-counter 20)
> (thread
   (lambda ()
     (counter-wait! a-counter 10)
     (displayln 'ok)))
#<thread>
ok
> (thread
   (lambda ()
     (counter-wait! a-counter 20)
     (displayln 'ok2)))
#<thread>
ok2
>