1 files changed, 107 insertions, 0 deletions

diff --git a/gnu/gnunet/concurrency/update.scm b/gnu/gnunet/concurrency/update.scm
new file mode 100644
index 0000000..6a487cf
--- /dev/null
+++ b/gnu/gnunet/concurrency/update.scm
@@ -0,0 +1,107 @@
+;; This file is part of scheme-GNUnet.
+;; Copyright (C) 2021 Maxime Devos
+;;
+;; scheme-GNUnet is free software: you can redistribute it and/or modify it
+;; under the terms of the GNU Affero General Public License as published
+;; by the Free Software Foundation, either version 3 of the License,
+;; or (at your option) any later version.
+;;
+;; scheme-GNUnet 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
+;; Affero General Public License for more details.
+;;
+;; You should have received a copy of the GNU Affero General Public License
+;; along with this program.  If not, see <http://www.gnu.org/licenses/>.
+;;
+;; SPDX-License-Identifier: AGPL3.0-or-later
+
+;; (Can be relicensed for use in guile-fibers on request)
+
+;; A stream of values, of which the next value might not
+;; yet be determined, but can be waited upon.  All can wait
+;; for an update and read the new value, but only the creator
+;; of the original update object can add new values.
+
+;; TODO implement a time machine for exploring alternate
+;; time-lines (maybe with a @code{time-machine} parameter,
+;; if time machines can be nested this way).
+
+;; Old updates are reclaimed by the garbage collector.
+;; TODO an implementation *not* relying on a garbage collector,
+;; at the cost of only allowing access to the latest value,
+;; would be nice to compare with.
+
+;; Example uses:
+;;  - network-size estimation client (TODO)
+
+(define-library (gnu gnunet concurrency update)
+  (export make-update update? update-value wait-for-update-operation
+          next-update next-update-peek
+
+          &double-update double-update? make-double-update-violation)
+  (import (rnrs records syntactic)
+          (rnrs conditions)
+          (rnrs base)
+          (rnrs exceptions)
+          (srfi srfi-8)
+          (fibers conditions)
+          (fibers operations)
+          (ice-9 atomic))
+  (begin
+    (define-condition-type &double-update &violation
+      make-double-update-violation double-update?)
+
+    (define (double-update-violation)
+      (raise (condition
+              (make-who-condition 'update!)
+              (make-double-update-violation)
+              (make-message-condition "An update already exists!"))))
+
+    (define-record-type (<update> make-update update?)
+      (fields (immutable value update-value)
+              ;; value in box is #f if not yet updated,
+              ;; otherwise it is an <update>
+              (immutable next next-update-box)
+              (immutable when-next next-update-condition))
+      (protocol
+       (lambda (%make)
+         (lambda (initial)
+           "Create an update object, initialised to @var{initial}.
+Two values are returned: the update object and the update procedure."
+           (let ((update (%make initial (make-atomic-box #f)
+                                (make-condition))))
+             (define (update! next-value)
+               "Update the update object to the value @var{next-value}.
+
+If the update object was already updated, raise a @code{&double-update}
+instead.  If the object was updated successfully, return the next update
+object and updater."
+               (receive (next-update next-update!) (make-update next-value)
+                 (case (atomic-box-compare-and-swap!
+                        (next-update-box update) #f next-update)
+                   ((#f)
+                    (signal-condition! (next-update-condition update))
+                    (values next-update next-update!))
+                   (else (double-update-violation)))))
+             (values update update!)))))
+      (opaque #t)
+      (sealed #t))
+
+    (define (wait-for-update-operation update)
+      "Return an operation for waiting for the next value
+of the update @var{update}.  The return value of the
+operation is the next @code{update?}."
+      (wrap-operation
+       (wait-operation (next-update-condition update))
+       (lambda ()
+         (atomic-box-ref (next-update-box update)))))
+
+    (define (next-update update)
+      "Return the next update of @var{update}."
+      (perform-operation (wait-for-update-operation update)))
+
+    (define (next-update-peek update)
+      "If the next update of @var{update} is known, return it,
+otherwise return @code{#f}."
+      (atomic-box-ref (next-update-box update)))))