Gecko [ www.utparral.edu.mx ]

Name	Size	Permission
peg	[ DIR ]	drwxr-xr-x
and-let-star.scm	2.53 KB	-rw-r--r--
arrays.scm	2.63 KB	-rw-r--r--
atomic.scm	1.55 KB	-rw-r--r--
binary-ports.scm	1.99 KB	-rw-r--r--
boot-9.scm	143.94 KB	-rw-r--r--
buffered-input.scm	4.82 KB	-rw-r--r--
calling.scm	10.54 KB	-rw-r--r--
channel.scm	5.19 KB	-rw-r--r--
command-line.scm	18.2 KB	-rw-r--r--
common-list.scm	8.95 KB	-rw-r--r--
control.scm	4.08 KB	-rw-r--r--
curried-definitions.scm	1.79 KB	-rw-r--r--
debug.scm	1.09 KB	-rw-r--r--
deprecated.scm	2.95 KB	-rw-r--r--
documentation.scm	7.41 KB	-rw-r--r--
eval.scm	25.12 KB	-rw-r--r--
eval-string.scm	2.99 KB	-rw-r--r--
expect.scm	5.5 KB	-rw-r--r--
fdes-finalizers.scm	1.06 KB	-rw-r--r--
format.scm	74.37 KB	-rw-r--r--
ftw.scm	24.17 KB	-rw-r--r--
futures.scm	10.49 KB	-rw-r--r--
gap-buffer.scm	10.14 KB	-rw-r--r--
getopt-long.scm	16.49 KB	-rw-r--r--
hash-table.scm	1.77 KB	-rw-r--r--
hcons.scm	2.55 KB	-rw-r--r--
history.scm	2.29 KB	-rw-r--r--
i18n.scm	20.51 KB	-rw-r--r--
iconv.scm	3.65 KB	-rw-r--r--
lineio.scm	3.85 KB	-rw-r--r--
list.scm	1.29 KB	-rw-r--r--
local-eval.scm	9.96 KB	-rw-r--r--
ls.scm	3.2 KB	-rw-r--r--
mapping.scm	4.84 KB	-rw-r--r--
match.scm	2 KB	-rw-r--r--
match.upstream.scm	35.92 KB	-rw-r--r--
networking.scm	3.33 KB	-rw-r--r--
null.scm	1.13 KB	-rw-r--r--
occam-channel.scm	7.26 KB	-rw-r--r--
optargs.scm	15.75 KB	-rw-r--r--
peg.scm	1.64 KB	-rw-r--r--
poe.scm	3.3 KB	-rw-r--r--
poll.scm	5.79 KB	-rw-r--r--
popen.scm	6.82 KB	-rw-r--r--
ports.scm	18.89 KB	-rw-r--r--
posix.scm	2.73 KB	-rw-r--r--
pretty-print.scm	16.88 KB	-rw-r--r--
psyntax-pp.scm	180.55 KB	-rw-r--r--
psyntax.scm	148.7 KB	-rw-r--r--
q.scm	4.2 KB	-rw-r--r--
quasisyntax.scm	5.22 KB	-rw-r--r--
r5rs.scm	1.56 KB	-rw-r--r--
r6rs-libraries.scm	9.43 KB	-rw-r--r--
rdelim.scm	7.72 KB	-rw-r--r--
readline.scm	9.56 KB	-rw-r--r--
receive.scm	1.06 KB	-rw-r--r--
regex.scm	8.87 KB	-rw-r--r--
runq.scm	8.18 KB	-rw-r--r--
rw.scm	1.02 KB	-rw-r--r--
safe-r5rs.scm	3.72 KB	-rw-r--r--
safe.scm	1.25 KB	-rw-r--r--
sandbox.scm	34.23 KB	-rw-r--r--
save-stack.scm	2.15 KB	-rw-r--r--
scm-style-repl.scm	11.62 KB	-rw-r--r--
serialize.scm	3.78 KB	-rw-r--r--
session.scm	17.72 KB	-rw-r--r--
slib.scm	1.55 KB	-rw-r--r--
stack-catch.scm	1.94 KB	-rw-r--r--
streams.scm	5.86 KB	-rw-r--r--
string-fun.scm	8.59 KB	-rw-r--r--
suspendable-ports.scm	29.87 KB	-rw-r--r--
syncase.scm	1.52 KB	-rw-r--r--
textual-ports.scm	2.29 KB	-rw-r--r--
threads.scm	12.54 KB	-rw-r--r--
time.scm	2.07 KB	-rw-r--r--
top-repl.scm	2.75 KB	-rw-r--r--
unicode.scm	1005 B	-rw-r--r--
vlist.scm	21.56 KB	-rw-r--r--
weak-vector.scm	1.2 KB	-rw-r--r--

Code Editor : futures.scm

;;; -*- mode: scheme; coding: utf-8; -*-
;;;
;;; 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

(define-module (ice-9 futures)
  #:use-module (srfi srfi-1)
  #:use-module (srfi srfi-9)
  #:use-module (srfi srfi-9 gnu)
  #:use-module (srfi srfi-11)
  #:use-module (ice-9 q)
  #:use-module (ice-9 match)
  #:use-module (ice-9 control)
  #:use-module (ice-9 threads)
  #:export (future make-future future? touch))

;;; Author: Ludovic Courtès <ludo@gnu.org>
;;;
;;; Commentary:
;;;
;;; This module provides an implementation of futures, a mechanism for
;;; fine-grain parallelism.  Futures were first described by Henry Baker
;;; in ``The Incremental Garbage Collection of Processes'', 1977, and
;;; then implemented in MultiLisp (an implicit variant thereof, i.e.,
;;; without `touch'.)
;;;
;;; This modules uses a fixed thread pool, normally one per CPU core.
;;; Futures are off-loaded to these threads, when they are idle.
;;;
;;; Code:

;;;
;;; Futures.
;;;

(define-record-type <future>
  (%make-future thunk state mutex completion)
  future?
  (thunk        future-thunk  set-future-thunk!)
  (state        future-state  set-future-state!)  ; done | started | queued
  (result       future-result set-future-result!)
  (mutex        future-mutex)
  (completion   future-completion))               ; completion cond. var.

(set-record-type-printer!
 <future>
 (lambda (future port)
   (simple-format port "#<future ~a ~a ~s>"
                  (number->string (object-address future) 16)
                  (future-state future)
                  (future-thunk future))))

(define (make-future thunk)
  "Return a new future for THUNK.  Execution may start at any point
concurrently, or it can start at the time when the returned future is
touched."
  (create-workers!)
  (let ((future (%make-future thunk 'queued
                              (make-mutex) (make-condition-variable))))
    (register-future! future)
    future))

;;;
;;; Future queues.
;;;

;; Global queue of pending futures.
;; TODO: Use per-worker queues to reduce contention.
(define %futures (make-q))

;; Lock for %FUTURES and %FUTURES-WAITING.
(define %futures-mutex (make-mutex))
(define %futures-available (make-condition-variable))

;; A mapping of nested futures to futures waiting for them to complete.
(define %futures-waiting '())

;; Whether currently running within a future.
(define %within-future? (make-parameter #f))

(define-syntax-rule (with-mutex m e0 e1 ...)
  ;; Copied from (ice-9 threads) to avoid circular dependency.
  (let ((x m))
    (dynamic-wind
      (lambda () (lock-mutex x))
      (lambda () (begin e0 e1 ...))
      (lambda () (unlock-mutex x)))))

(define %future-prompt
  ;; The prompt futures abort to when they want to wait for another
  ;; future.
  (make-prompt-tag))

(define (register-future! future)
  ;; Register FUTURE as being processable.
  (lock-mutex %futures-mutex)
  (enq! %futures future)
  (signal-condition-variable %futures-available)
  (unlock-mutex %futures-mutex))

(define (process-future! future)
  "Process FUTURE.  When FUTURE completes, return #t and update its
result; otherwise, when FUTURE touches a nested future that has not
completed yet, then suspend it and return #f.  Suspending a future
consists in capturing its continuation, marking it as `queued', and
adding it to the waiter queue."
  (let/ec return
    (let* ((suspend
            (lambda (cont future-to-wait)
              ;; FUTURE wishes to wait for the completion of FUTURE-TO-WAIT.
              ;; At this point, FUTURE is unlocked and in `started' state,
              ;; and FUTURE-TO-WAIT is unlocked.
              (with-mutex %futures-mutex
                (with-mutex (future-mutex future)
                  (set-future-thunk! future cont)
                  (set-future-state! future 'queued))

(with-mutex (future-mutex future-to-wait)
                  ;; If FUTURE-TO-WAIT completed in the meantime, then
                  ;; reschedule FUTURE directly; otherwise, add it to the
                  ;; waiter queue.
                  (if (eq? 'done (future-state future-to-wait))
                      (begin
                        (enq! %futures future)
                        (signal-condition-variable %futures-available))
                      (set! %futures-waiting
                            (alist-cons future-to-wait future
                                        %futures-waiting))))

(return #f))))
           (thunk (lambda ()
                    (call-with-prompt %future-prompt
                                      (lambda ()
                                        (parameterize ((%within-future? #t))
                                          ((future-thunk future))))
                                      suspend))))
      (set-future-result! future
                          (catch #t
                            (lambda ()
                              (call-with-values thunk
                                (lambda results
                                  (lambda ()
                                    (apply values results)))))
                            (lambda args
                              (lambda ()
                                (apply throw args)))))
      #t)))

(define (process-one-future)
  "Attempt to pick one future from the queue and process it."
  ;; %FUTURES-MUTEX must be locked on entry, and is locked on exit.
  (or (q-empty? %futures)
      (let ((future (deq! %futures)))
        (lock-mutex (future-mutex future))
        (case (future-state future)
          ((done started)
           ;; Nothing to do.
           (unlock-mutex (future-mutex future)))
          (else
           ;; Do the actual work.

;; We want to release %FUTURES-MUTEX so that other workers can
           ;; progress.  However, to avoid deadlocks, we have to unlock
           ;; FUTURE as well, to preserve lock ordering.
           (unlock-mutex (future-mutex future))
           (unlock-mutex %futures-mutex)

(lock-mutex (future-mutex future))
           (if (eq? (future-state future) 'queued) ; lost the race?
               (begin                          ; no, so let's process it
                 (set-future-state! future 'started)
                 (unlock-mutex (future-mutex future))

(let ((done? (process-future! future)))
                   (when done?
                     (with-mutex %futures-mutex
                       (with-mutex (future-mutex future)
                         (set-future-state! future 'done)
                         (notify-completion future))))))
               (unlock-mutex (future-mutex future))) ; yes

(lock-mutex %futures-mutex))))))

(define (process-futures)
  "Continuously process futures from the queue."
  (lock-mutex %futures-mutex)
  (let loop ()
    (when (q-empty? %futures)
      (wait-condition-variable %futures-available
                               %futures-mutex))

(process-one-future)
    (loop)))

(define (notify-completion future)
  "Notify futures and callers waiting that FUTURE completed."
  ;; FUTURE and %FUTURES-MUTEX are locked.
  (broadcast-condition-variable (future-completion future))
  (let-values (((waiting remaining)
                (partition (match-lambda          ; TODO: optimize
                            ((waitee . _)
                             (eq? waitee future)))
                           %futures-waiting)))
    (set! %futures-waiting remaining)
    (for-each (match-lambda
               ((_ . waiter)
                (enq! %futures waiter)))
              waiting)))

(define (touch future)
  "Return the result of FUTURE, computing it if not already done."
  (define (work)
    ;; Do some work while waiting for FUTURE to complete.
    (lock-mutex %futures-mutex)
    (if (q-empty? %futures)
        (begin
          (unlock-mutex %futures-mutex)
          (with-mutex (future-mutex future)
            (unless (eq? 'done (future-state future))
              (wait-condition-variable (future-completion future)
                                       (future-mutex future)))))
        (begin
          (process-one-future)
          (unlock-mutex %futures-mutex))))

(let loop ()
    (lock-mutex (future-mutex future))
    (case (future-state future)
      ((done)
       (unlock-mutex (future-mutex future)))
      ((started)
       (unlock-mutex (future-mutex future))
       (if (%within-future?)
           (abort-to-prompt %future-prompt future)
           (begin
             (work)
             (loop))))
      (else
       (unlock-mutex (future-mutex future))
       (work)
       (loop))))
  ((future-result future)))

;;;
;;; Workers.
;;;

(define %worker-count
  (if (provided? 'threads)
      (- (current-processor-count) 1)
      0))

;; A dock of workers that stay here forever.

;; TODO
;; 1. Allow the pool to be shrunk, as in libgomp (though that we'd
;;    need semaphores, which aren't yet in libguile!).
;; 2. Provide a `worker-count' fluid.
(define %workers '())

(define (%create-workers!)
  (with-mutex
   %futures-mutex
   ;; Setting 'create-workers!' to a no-op is an optimization, but it is
   ;; still possible for '%create-workers!' to be called more than once
   ;; from different threads.  Therefore, to avoid creating %workers more
   ;; than once (and thus creating too many threads), we check to make
   ;; sure %workers is empty within the critical section.
   (when (null? %workers)
     (set! %workers
           (unfold (lambda (i) (>= i %worker-count))
                   (lambda (i) (call-with-new-thread process-futures))
                   1+
                   0))
     (set! create-workers! (lambda () #t)))))

(define create-workers!
  (lambda () (%create-workers!)))

;;;
;;; Syntax.
;;;

(define-syntax-rule (future body)
  "Return a new future for BODY."
  (make-future (lambda () body)))

;;; Local Variables:
;;; eval: (put 'with-mutex 'scheme-indent-function 1)
;;; End:

${this.title}

Code Editor : futures.scm