diff options
author | Andy Wingo <wingo@pobox.com> | 2012-02-06 20:12:29 +0100 |
---|---|---|
committer | Andy Wingo <wingo@pobox.com> | 2012-03-12 17:12:37 +0100 |
commit | 5f54be9b3d48c175bc488b2a9324da9df76bedd1 (patch) | |
tree | 8d76e4756db5a634bbcfa4c59d7c4cfe55348af6 | |
parent | c62bbee320ef8e48618bd4bb090d8a2a8ee70862 (diff) | |
download | guile-5f54be9b3d48c175bc488b2a9324da9df76bedd1.tar.gz |
add ice-9 thread-pool
* module/ice-9/thread-pool.scm: New file.
* module/Makefile.am: Add to build.
-rw-r--r-- | module/Makefile.am | 1 | ||||
-rw-r--r-- | module/ice-9/thread-pool.scm | 192 |
2 files changed, 193 insertions, 0 deletions
diff --git a/module/Makefile.am b/module/Makefile.am index 6f5583a09..fe688a0c7 100644 --- a/module/Makefile.am +++ b/module/Makefile.am @@ -231,6 +231,7 @@ ICE_9_SOURCES = \ ice-9/string-fun.scm \ ice-9/syncase.scm \ ice-9/threads.scm \ + ice-9/thread-pool.scm \ ice-9/top-repl.scm \ ice-9/buffered-input.scm \ ice-9/time.scm \ diff --git a/module/ice-9/thread-pool.scm b/module/ice-9/thread-pool.scm new file mode 100644 index 000000000..40dc6b452 --- /dev/null +++ b/module/ice-9/thread-pool.scm @@ -0,0 +1,192 @@ +;;; Thread pools + +;; Copyright (C) 2010, 2011, 2012 Free Software Foundation, Inc. + +;; This library is free software; you can redistribute it and/or +;; modify it under the terms of the GNU Lesser General Public +;; License as published by the Free Software Foundation; either +;; version 3 of the License, or (at your option) any later version. +;; +;; This library is distributed in the hope that it will be useful, +;; but WITHOUT ANY WARRANTY; without even the implied warranty of +;; MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU +;; Lesser General Public License for more details. +;; +;; You should have received a copy of the GNU Lesser General Public +;; License along with this library; if not, write to the Free Software +;; Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA +;; 02110-1301 USA + +;;; Commentary: +;;; +;;; A simple thread pool implementation. +;;; +;;; Code: + +(define-module (ice-9 thread-pool) + #:use-module (srfi srfi-9) + #:use-module (srfi srfi-9 gnu) + #:use-module (ice-9 threads) + #:export (make-thread-pool + start-thread-pool! + pause-thread-pool! + stop-thread-pool!)) + + +(define-record-type <worker> + (%make-worker mutex condvar thread handler state) + worker? + (mutex worker-mutex) + (condvar worker-condvar) + (thread worker-thread set-worker-thread!) + (handler worker-handler) + (state worker-state set-worker-state!)) + +(define (make-worker handler) + (%make-worker (make-mutex) (make-condition-variable) #f handler 'stopped)) + +(define* (worker-wait worker #:optional time) + (if time + (wait-condition-variable (worker-condvar worker) + (worker-mutex worker) + time) + (wait-condition-variable (worker-condvar worker) + (worker-mutex worker)))) + +(define (worker-stopping? worker) + (with-mutex (worker-mutex worker) + (let lp () + (case (worker-state worker) + ((running) + #f) + ((pausing) + (set-worker-state! worker 'paused) + (signal-condition-variable (worker-condvar worker)) + (lp)) + ((paused) + (worker-wait worker) + (lp)) + ((stopping) + #t) + (else + (error "bad state" worker)))))) + +(define (worker-loop worker) + (let lp () + (or (worker-stopping? worker) + (begin + ((worker-handler worker)) + (lp))))) + +(define (pause-worker worker) + (with-mutex (worker-mutex worker) + (let lp () + (case (worker-state worker) + ((running) + (set-worker-state! worker 'pausing) + (lp)) + ((pausing) + #f) + ((paused) + #t) + ((stopping) + (error "attempt to go stopping -> pausing" worker)) + ((stopped) + (set-worker-state! worker 'pausing) + (set-worker-thread! worker + (make-thread worker-loop worker)) + (lp)) + (else + (error "bad state" worker)))))) + +(define* (wait-for-paused worker #:optional time) + (with-mutex (worker-mutex worker) + (let lp () + (case (worker-state worker) + ((paused) + #t) + (else + (and (worker-wait worker time) + (lp))))))) + +(define (start-worker worker) + (with-mutex (worker-mutex worker) + (let lp () + (case (worker-state worker) + ((running) + #t) + ((stopping) + (error "attempt to go stopping -> running" worker)) + (else + (set-worker-state! worker 'running) + (if (worker-thread worker) + (signal-condition-variable (worker-condvar worker)) + (set-worker-thread! worker (make-thread worker-loop worker))) + (lp)))))) + +(define (stop-worker worker) + (with-mutex (worker-mutex worker) + (let lp () + (case (worker-state worker) + ((stopped) + #t) + ((stopping) + #f) + (else + (set-worker-state! worker 'stopping) + (lp)))))) + +(define* (wait-for-stopped worker #:optional time #:key cancel?) + (let ((thread + (with-mutex (worker-mutex worker) + (cond + ((eq? (worker-state worker) 'stopped) + #f) + ((thread-exited? (worker-thread worker)) + (set-worker-thread! worker #f) + (set-worker-state! worker 'stopped) + #f) + (else + (worker-thread worker)))))) + (or (not thread) + (begin + (if time + (join-thread (worker-thread worker) time) + (join-thread (worker-thread worker))) + (cond + ((thread-exited? thread) + (wait-for-stopped worker)) + (cancel? + (cancel-thread (worker-thread worker)) + ;; Assume it works. + (join-thread (worker-thread worker)) + (wait-for-stopped worker)) + (else + #f)))))) + +(define-record-type <thread-pool> + (%make-thread-pool size workers) + thread-pool? + (size thread-pool-size) + (workers thread-pool-workers)) + +;; Create a thread pool, and bring it to the "paused" state. +;; +(define (make-thread-pool size handler) + (let ((pool (%make-thread-pool size + (map (lambda (i) (make-worker handler)) + (iota size))))) + (pause-thread-pool! pool) + pool)) + +(define* (pause-thread-pool! pool #:optional time) + (for-each pause-worker (thread-pool-workers pool)) + (for-each (lambda (w) (wait-for-paused w time)) (thread-pool-workers pool))) + +(define (start-thread-pool! pool) + (for-each start-worker (thread-pool-workers pool))) + +(define* (stop-thread-pool! pool #:optional time #:key cancel?) + (for-each stop-worker (thread-pool-workers pool)) + (for-each (lambda (w) (wait-for-stopped w time #:cancel? cancel?)) + (thread-pool-workers pool))) |