guix/workers.scm

   1 ;;; GNU Guix --- Functional package management for GNU
   2 ;;; Copyright © 2017 Ludovic Courtès <ludo@gnu.org>
   3 ;;;
   4 ;;; This file is part of GNU Guix.
   5 ;;;
   6 ;;; GNU Guix is free software; you can redistribute it and/or modify it
   7 ;;; under the terms of the GNU General Public License as published by
   8 ;;; the Free Software Foundation; either version 3 of the License, or (at
   9 ;;; your option) any later version.
  10 ;;;
  11 ;;; GNU Guix is distributed in the hope that it will be useful, but
  12 ;;; WITHOUT ANY WARRANTY; without even the implied warranty of
  13 ;;; MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  14 ;;; GNU General Public License for more details.
  15 ;;;
  16 ;;; You should have received a copy of the GNU General Public License
  17 ;;; along with GNU Guix.  If not, see <http://www.gnu.org/licenses/>.
  18
  19 (define-module (guix workers)
  20   #:use-module (ice-9 threads)
  21   #:use-module (ice-9 match)
  22   #:use-module (ice-9 q)
  23   #:use-module (srfi srfi-1)
  24   #:use-module (srfi srfi-9)
  25   #:use-module (srfi srfi-26)
  26   #:use-module ((guix build syscalls) #:select (set-thread-name))
  27   #:export (pool?
  28             make-pool
  29             pool-enqueue!
  30             pool-idle?
  31             eventually))
  32
  33 ;;; Commentary:
  34 ;;;
  35 ;;; This module implements "worker pools".  Worker pools are the low-level
  36 ;;; mechanism that's behind futures: there's a fixed set of threads
  37 ;;; ("workers") that one can submit work to, and one of them will eventually
  38 ;;; pick the submitted tasks.
  39 ;;;
  40 ;;; Unlike futures, these worker pools are meant to be used for tasks that
  41 ;;; have a side-effect.  Thus, we never "touch" a task that was submitted like
  42 ;;; we "touch" a future.  Instead, we simply assume that the task will
  43 ;;; eventually complete.
  44 ;;;
  45 ;;; Code:
  46
  47 (define-record-type <pool>
  48   (%make-pool queue mutex condvar workers busy)
  49   pool?
  50   (queue    pool-queue)
  51   (mutex    pool-mutex)
  52   (condvar  pool-condition-variable)
  53   (workers  pool-workers)
  54   (busy     pool-busy))
  55
  56 (define-syntax-rule (without-mutex mutex exp ...)
  57   (dynamic-wind
  58     (lambda ()
  59       (unlock-mutex mutex))
  60     (lambda ()
  61       exp ...)
  62     (lambda ()
  63       (lock-mutex mutex))))
  64
  65 (define* (worker-thunk mutex condvar pop-queue
  66                        #:key idle busy (thread-name "guix worker"))
  67   "Return the thunk executed by worker threads."
  68   (define (loop)
  69     (match (pop-queue)
  70       (#f                                         ;empty queue
  71        (idle)
  72        (wait-condition-variable condvar mutex)
  73        (busy))
  74       ((? procedure? proc)
  75        ;; Release MUTEX while executing PROC.
  76        (without-mutex mutex
  77          (catch #t proc
  78            (const #f)
  79            (lambda (key . args)
  80              ;; XXX: In Guile 2.0 ports are not thread-safe, so this could
  81              ;; crash (Guile 2.2 is fine).
  82              (display-backtrace (make-stack #t) (current-error-port))
  83              (print-exception (current-error-port)
  84                               (and=> (make-stack #t)
  85                                      (cut stack-ref <> 0))
  86                               key args))))))
  87     (loop))
  88
  89   (lambda ()
  90     (catch 'system-error
  91       (lambda ()
  92         (set-thread-name thread-name))
  93       (const #f))
  94
  95     (with-mutex mutex
  96       (loop))))
  97
  98 (define* (make-pool #:optional (count (current-processor-count))
  99                     #:key (thread-name "guix worker"))
 100   "Return a pool of COUNT workers.  Use THREAD-NAME as the name of these
 101 threads as reported by the operating system."
 102   (let* ((mutex   (make-mutex))
 103          (condvar (make-condition-variable))
 104          (queue   (make-q))
 105          (busy    count)
 106          (procs   (unfold (cut >= <> count)
 107                           (lambda (n)
 108                             (worker-thunk mutex condvar
 109                                           (lambda ()
 110                                             (and (not (q-empty? queue))
 111                                                  (q-pop! queue)))
 112                                           #:busy (lambda ()
 113                                                    (set! busy (+ 1 busy)))
 114                                           #:idle (lambda ()
 115                                                    (set! busy (- busy 1)))
 116                                           #:thread-name thread-name))
 117                           1+
 118                           0))
 119          (threads (map (lambda (proc)
 120                          (call-with-new-thread proc))
 121                        procs)))
 122     (%make-pool queue mutex condvar threads (lambda () busy))))
 123
 124 (define (pool-enqueue! pool thunk)
 125   "Enqueue THUNK for future execution by POOL."
 126   (with-mutex (pool-mutex pool)
 127     (enq! (pool-queue pool) thunk)
 128     (signal-condition-variable (pool-condition-variable pool))))
 129
 130 (define (pool-idle? pool)
 131   "Return true if POOL doesn't have any task in its queue and all the workers
 132 are currently idle (i.e., waiting for a task)."
 133   (with-mutex (pool-mutex pool)
 134     (and (q-empty? (pool-queue pool))
 135          (zero? ((pool-busy pool))))))
 136
 137 (define-syntax-rule (eventually pool exp ...)
 138   "Run EXP eventually on one of the workers of POOL."
 139   (pool-enqueue! pool (lambda () exp ...)))
 140
 141 ;;; Local Variables:
 142 ;;; eval: (put 'without-mutex 'scheme-indent-function 1)
 143 ;;; End:
 144
 145 ;;; workers.scm ends here