01/06: Add (guix workers).

[Ludovic Courtès]

civodul pushed a commit to branch master
in repository guix.

commit 1563d6c79f759cb9a6cc5561463653f0f184cb36
Author: Ludovic Courtès <address@hidden>
Date:   Mon Apr 17 22:04:31 2017 +0200

    Add (guix workers).
    
    * guix/workers.scm, tests/workers.scm: New files.
    * Makefile.am (MODULES, SCM_TESTS): Add them.
    * .dir-locals.el: Add rule for 'eventually'.
---
 .dir-locals.el    |   1 +
 Makefile.am       |   2 +
 guix/workers.scm  | 123 ++++++++++++++++++++++++++++++++++++++++++++++++++++++
 tests/workers.scm |  45 ++++++++++++++++++++
 4 files changed, 171 insertions(+)

diff --git a/.dir-locals.el b/.dir-locals.el
index 917fd30..a2d1eb8 100644
--- a/.dir-locals.el
+++ b/.dir-locals.el
@@ -68,6 +68,7 @@
 
    (eval . (put 'call-with-container 'scheme-indent-function 1))
    (eval . (put 'container-excursion 'scheme-indent-function 1))
+   (eval . (put 'eventually 'scheme-indent-function 1))
 
    ;; Recognize '~', '+', and '$', as used for gexps, as quotation symbols.
    ;; This notably allows '(' in Paredit to not insert a space when the
diff --git a/Makefile.am b/Makefile.am
index c9671e2..46f9547 100644
--- a/Makefile.am
+++ b/Makefile.am
@@ -61,6 +61,7 @@ MODULES =                                     \
   guix/licenses.scm                            \
   guix/graph.scm                               \
   guix/cve.scm                                 \
+  guix/workers.scm                             \
   guix/zlib.scm                                        \
   guix/build-system.scm                                \
   guix/build-system/ant.scm                    \
@@ -296,6 +297,7 @@ SCM_TESTS =                                 \
   tests/graph.scm                              \
   tests/challenge.scm                          \
   tests/cve.scm                                        \
+  tests/workers.scm                            \
   tests/zlib.scm                               \
   tests/file-systems.scm                       \
   tests/system.scm                             \
diff --git a/guix/workers.scm b/guix/workers.scm
new file mode 100644
index 0000000..e3452d2
--- /dev/null
+++ b/guix/workers.scm
@@ -0,0 +1,123 @@
+;;; GNU Guix --- Functional package management for GNU
+;;; Copyright © 2017 Ludovic Courtès <address@hidden>
+;;;
+;;; This file is part of GNU Guix.
+;;;
+;;; GNU Guix is free software; you can redistribute it and/or modify it
+;;; under the terms of the GNU General Public License as published by
+;;; the Free Software Foundation; either version 3 of the License, or (at
+;;; your option) any later version.
+;;;
+;;; GNU Guix 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 General Public License for more details.
+;;;
+;;; You should have received a copy of the GNU General Public License
+;;; along with GNU Guix.  If not, see <http://www.gnu.org/licenses/>.
+
+(define-module (guix workers)
+  #:use-module (ice-9 threads)
+  #:use-module (ice-9 match)
+  #:use-module (ice-9 q)
+  #:use-module (srfi srfi-1)
+  #:use-module (srfi srfi-9)
+  #:use-module (srfi srfi-26)
+  #:export (pool?
+            make-pool
+            pool-enqueue!
+            pool-idle?
+            eventually))
+
+;;; Commentary:
+;;;
+;;; This module implements "worker pools".  Worker pools are the low-level
+;;; mechanism that's behind futures: there's a fixed set of threads
+;;; ("workers") that one can submit work to, and one of them will eventually
+;;; pick the submitted tasks.
+;;;
+;;; Unlike futures, these worker pools are meant to be used for tasks that
+;;; have a side-effect.  Thus, we never "touch" a task that was submitted like
+;;; we "touch" a future.  Instead, we simply assume that the task will
+;;; eventually complete.
+;;;
+;;; Code:
+
+(define-record-type <pool>
+  (%make-pool queue mutex condvar workers)
+  pool?
+  (queue    pool-queue)
+  (mutex    pool-mutex)
+  (condvar  pool-condition-variable)
+  (workers  pool-workers))
+
+(define-syntax-rule (without-mutex mutex exp ...)
+  (dynamic-wind
+    (lambda ()
+      (unlock-mutex mutex))
+    (lambda ()
+      exp ...)
+    (lambda ()
+      (lock-mutex mutex))))
+
+(define (worker-thunk mutex condvar pop-queue)
+  "Return the thunk executed by worker threads."
+  (define (loop)
+    (match (pop-queue)
+      (#f                                         ;empty queue
+       (wait-condition-variable condvar mutex))
+      ((? procedure? proc)
+       ;; Release MUTEX while executing PROC.
+       (without-mutex mutex
+         (catch #t proc
+           (lambda (key . args)
+             ;; XXX: In Guile 2.0 ports are not thread-safe, so this could
+             ;; crash (Guile 2.2 is fine).
+             (display-backtrace (make-stack #t) (current-error-port))
+             (print-exception (current-error-port)
+                              (stack-ref (make-stack #t) 0)
+                              key args))))))
+    (loop))
+
+  (lambda ()
+    (with-mutex mutex
+      (loop))))
+
+(define* (make-pool #:optional (count (current-processor-count)))
+  "Return a pool of COUNT workers."
+  (let* ((mutex   (make-mutex))
+         (condvar (make-condition-variable))
+         (queue   (make-q))
+         (procs   (unfold (cut >= <> count)
+                          (lambda (n)
+                            (worker-thunk mutex condvar
+                                          (lambda ()
+                                            (and (not (q-empty? queue))
+                                                 (q-pop! queue)))))
+                          1+
+                          0))
+         (threads (map (lambda (proc)
+                         (call-with-new-thread proc))
+                       procs)))
+    (%make-pool queue mutex condvar threads)))
+
+(define (pool-enqueue! pool thunk)
+  "Enqueue THUNK for future execution by POOL."
+  (with-mutex (pool-mutex pool)
+    (enq! (pool-queue pool) thunk)
+    (signal-condition-variable (pool-condition-variable pool))))
+
+(define (pool-idle? pool)
+  "Return true if POOL doesn't have any task in its queue."
+  (with-mutex (pool-mutex pool)
+    (q-empty? (pool-queue pool))))
+
+(define-syntax-rule (eventually pool exp ...)
+  "Run EXP eventually on one of the workers of POOL."
+  (pool-enqueue! pool (lambda () exp ...)))
+
+;;; Local Variables:
+;;; eval: (put 'without-mutex 'scheme-indent-function 1)
+;;; End:
+
+;;; workers.scm ends here
diff --git a/tests/workers.scm b/tests/workers.scm
new file mode 100644
index 0000000..44b882f
--- /dev/null
+++ b/tests/workers.scm
@@ -0,0 +1,45 @@
+;;; GNU Guix --- Functional package management for GNU
+;;; Copyright © 2017 Ludovic Courtès <address@hidden>
+;;;
+;;; This file is part of GNU Guix.
+;;;
+;;; GNU Guix is free software; you can redistribute it and/or modify it
+;;; under the terms of the GNU General Public License as published by
+;;; the Free Software Foundation; either version 3 of the License, or (at
+;;; your option) any later version.
+;;;
+;;; GNU Guix 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 General Public License for more details.
+;;;
+;;; You should have received a copy of the GNU General Public License
+;;; along with GNU Guix.  If not, see <http://www.gnu.org/licenses/>.
+
+(define-module (test-workers)
+  #:use-module (guix workers)
+  #:use-module (ice-9 threads)
+  #:use-module (srfi srfi-64))
+
+(test-begin "workers")
+
+(test-equal "enqueue"
+  4242
+  (let* ((pool   (make-pool))
+         (result 0)
+         (1+!    (let ((lock (make-mutex)))
+                   (lambda ()
+                     (with-mutex lock
+                       (set! result (+ result 1)))))))
+    (let loop ((i 4242))
+      (unless (zero? i)
+        (pool-enqueue! pool 1+!)
+        (loop (- i 1))))
+    (let poll ()
+      (unless (pool-idle? pool)
+        (pk 'busy result)
+        (sleep 1)
+        (poll)))
+    result))
+
+(test-end)