Should make fork join more stable

Author: ktakashi

sitelib/util/concurrent/fork-join-pool.scm

@@ -83,7 +83,7 @@
 	  shutdown)
   (protocol (lambda (p)
 	      (lambda (pool)
-		(p (make-notifier)
+		(p (make-event)
 		   pool
 		   (make-shared-queue)
 		   (make-atomic #f))))))
@@ -94,9 +94,9 @@
 (define (worker-queue-empty? wq) (shared-queue-empty? (worker-queue-queue wq)))
 (define (worker-queue-size wq) (shared-queue-size (worker-queue-queue wq)))
 (define (worker-queue-notify! wq)
-  (notifier-send-notification! (worker-queue-notifier wq)))
+  (event-set-event! (worker-queue-notifier wq)))
 (define (worker-queue-wait! wq . opts)
-  (apply notifier-wait-notification! (worker-queue-notifier wq) opts))
+  (apply event-receive! (worker-queue-notifier wq) opts))
 (define (worker-queue-shutdown! wq)
   (atomic-store! (worker-queue-shutdown wq) #t)
   (worker-queue-notify! wq))
@@ -157,7 +157,6 @@
 	  thread-count		  ;; # of worker threads
 	  idle-count		  ;; # of idle threads
 	  parameter
-	  notifier
 	  lock)
   (protocol (lambda (p)
 	      (lambda ((n (or integer? fork-join-pool-parameters?)))
@@ -173,7 +172,6 @@
 		   (make-atomic-fixnum 0)
 		   (make-atomic-fixnum 0)
 		   parameter
-		   (make-notifier)
 		   (make-mutex))))))
 
 (define (fork-join-pool-keep-alive pool)
@@ -265,13 +263,17 @@
   (fork-join-pool-thread-count-dec! pool))
 
 (define (fork-join-pool-run-worker pool wq)
-  (define (run-task task wq)
+  (define (run-task task root-wq wq)
     (when task
       (task)
-      (run-task (worker-queue-pop! wq 0 #f) wq)))
+      (run-task (or (worker-queue-pop! wq 0 #f)
+		    (worker-queue-pop! root-wq 0 #f))
+		root-wq
+		wq)))
   (define (->timeout pool) 
     (add-duration (current-time) (fork-join-pool-keep-alive pool)))
-  (run-task (worker-queue-pop! (fork-join-pool-worker-queue pool) 0 #f) wq)
+  (let ((queue (fork-join-pool-worker-queue pool)))
+    (run-task (worker-queue-pop! queue 0 #f) queue wq))
   (fork-join-pool-idle-count-inc! pool)
   (cond ((and (worker-queue-wait! wq (->timeout pool))
 	      (not (worker-queue-shutdown? wq)))

sitelib/util/concurrent/notifier.scm

@@ -34,7 +34,13 @@
 	    (rename (notifier <notifier>))
 	    notifier-waiting?
 	    notifier-send-notification!
-	    notifier-wait-notification!)
+	    notifier-wait-notification!
+
+	    event? make-event
+	    (rename (event <event>))
+	    event-set-event!
+	    event-receive!
+	    )
     (import (rnrs)
 	    (srfi :18)
 	    (util concurrent atomic))
@@ -55,9 +61,43 @@
 
 (define (notifier-wait-notification! notifier . maybe-timeout)
   (define to (and (not (null? maybe-timeout)) (car maybe-timeout)))
+  (define lock (notifier-lock notifier))
+  (mutex-lock! lock)
   (atomic-fixnum-inc! (notifier-waiter notifier))
-  (let ((r (mutex-unlock! (notifier-lock notifier) (notifier-cv notifier) to)))
+  (let ((r (mutex-unlock! lock (notifier-cv notifier) to)))
     (atomic-fixnum-dec! (notifier-waiter notifier))
     r))
 
+;; Similar to SetEvent on Win32 API
+;; It holds the value until the event is read
+(define-record-type event
+  (fields lock cv (mutable received))
+  (protocol (lambda (p)
+	      (lambda (:optional (set? #f))
+		(p (make-mutex "event-lock")
+		   (make-condition-variable)
+		   set?)))))
+
+(define (event-set-event! event)
+  (define lock (event-lock event))
+  (mutex-lock! lock)
+  (event-received-set! event #t)
+  (condition-variable-broadcast! (event-cv event))
+  (mutex-unlock! lock))
+
+(define (event-receive! event . maybe-timeout)
+  (define to (and (not (null? maybe-timeout)) (car maybe-timeout)))
+  (define lock (event-lock event))
+  (mutex-lock! lock)
+  (cond ((event-received event)
+	 (event-received-set! event #f)
+	 (mutex-unlock! lock))
+	((mutex-unlock! lock (event-cv event) to)
+	 (mutex-lock! lock)
+	 (let ((r (and (event-received event)
+		       (event-received-set! event #f))))
+	   (mutex-unlock! lock)
+	   r))
+	(else #f)))
+
 )