work-stealy.rkt

#lang racket

(define-logger workin)

(struct queue (fore aft))

(define (enqueue q w)
  (match q
    [(queue a d) (queue a (cons w d))]))

(define (reorder q v)
  (define elems (append (queue-fore q) (reverse (queue-aft q))))
  (queue (cons v (remv v elems)) null))

(define (queue-empty? q)
  (match q
    [(queue (list) (list)) #t]
    [_ #f]))

(define (queue-top q)
  (match q
    [(queue (list) d) (car (reverse d))]
    [(queue (cons a _) _) a]))

(define (dequeue q)
  (match q
    [(queue (list) d) (dequeue (queue (reverse d) null))]
    [(queue (cons _ a) d) (queue a d)]))

(define req-ch (make-channel))
(define rpy-ch (make-channel))
(define work-ch (make-channel))

(define (run workq)
  (sync (handle-evt req-ch
                    (match-lambda
                      [`(enq ,work)
                       (channel-put rpy-ch work)
                       (run (enqueue workq work))]
                      [`(reorder ,id) (run (reorder workq id))]))
        (if (queue-empty? workq)
            never-evt
            (handle-evt (channel-put-evt work-ch (queue-top workq))
                        (λ (v)
                          (run (dequeue workq)))))))

(define (queue-work w)
  (channel-put req-ch `(enq ,w))
  (channel-get rpy-ch))

(define (reorder-work w)
  (channel-put req-ch `(reorder ,w)))

(define (dequeue-work)
  (channel-get work-ch))

(thread (λ () (run (queue null null))))

(thread
 (λ ()
   (for ([i 30])
     (log-workin-info "queuing work for ~a" i)
     (queue-work i)
     (sleep 1/10))))

(thread
 (λ ()
   (define (run)
     (define w (channel-get work-ch))
     (log-workin-info "working on ~a" w)
     (sleep 1)
     (run))
   (run)))

(sleep 5)
(log-workin-info "rescheduling")
(reorder-work 10)
(sleep 2)
(reorder-work 20)