concurrency: implement an ‘update stream’

This will be used for communicating NSE values.

From README.org:

  a box with a value, that can be updated,
  resulting in a new box.  Updates can be
  waited upon.

* README.org (Modules): describe new module.
  (Purpose): expand purpose.
* gnu/gnunet/concurrency/update.scm: implement new module.
* tests/update.scm: test new module.

3 files changed, 223 insertions, 0 deletions

diff --git a/README.org b/README.org
index bece9a3..6309f1d 100644
--- a/README.org
+++ b/README.org
@@ -30,6 +30,7 @@
 * Purposes
   + for use by Guix and disarchive
   + bit-for-bit reproducibility in directory creation
+  + a nice Scheme interface to GNUnet!
 * Modules
   + gnu/gnunet/directory.scm: directory construction
   + gnu/gnunet/util/mq.scm and friends: message queues for
@@ -37,6 +38,9 @@
     each message type.
   + gnu/gnunet/message/envelope.scm: some program data around
     message types (e.g. priority, notify on sent hook)
+  + gnu/gnunet/concurrency/update.scm: a box with a value,
+    that can be updated, resulting in a new box.  Updates
+    can be waited upon.
 * Conventions
 ** Fiddling with options
    Options like ‘priority’, ‘anonymity’, ‘replication’
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)))))
diff --git a/tests/update.scm b/tests/update.scm
new file mode 100644
index 0000000..a9b7928
--- /dev/null
+++ b/tests/update.scm
@@ -0,0 +1,112 @@
+;; 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
+
+(use-modules (gnu gnunet concurrency update)
+             (srfi srfi-8)
+             (srfi srfi-11)
+             (srfi srfi-26)
+             (fibers operations)
+             (fibers timers)
+             (fibers))
+
+(test-begin "update")
+
+;; Tests without concurrency
+(test-equal "make-update result types"
+  '(#t . #t)
+  (receive (update update!)
+      (make-update 0)
+    (cons (update? update)
+          (procedure? update!))))
+
+(test-equal "update! and next-update-peek"
+  '(new #t #t)
+  (let*-values (((update update!) (make-update 'old))
+                ((next-update next-update!) (update! 'new)))
+    (receive (next-update-peeked) (next-update-peek update)
+      (list (update-value next-update-peeked)
+            (procedure? next-update!)
+            (eq? next-update-peeked next-update)))))
+
+(test-eq "no update! and next-update-peek"
+  #f
+  (next-update-peek (make-update 0)))
+
+(test-error "update! twice -> &double-update"
+  &double-update
+  (receive (next-update next-update!)
+      (make-update 0)
+    (next-update! next-update)
+    (next-update! next-update)))
+
+
+;; Tests with operations
+
+;; Unfortunately, fibers does not not have
+;; a ‘run this operation -- unless it would
+;; block’ procedure, and using a combination
+;; of wrap-operation and sleep-operation/
+;; timer-operation turns out to be racy.
+;;
+;; Our approach:
+;;  * if an operation is expected to block,
+;;    include a short timer-operation
+;;    for testing detecting blocking.
+;;    (Short to ensure tests still pass
+;;    quickly.)
+;;
+;;    A false ‘PASS’ could occassionally
+;;    result, but no false ‘FAIL’ will
+;;    be created.
+;;  * if a test is expected *not* to block,
+;;    just perform the operation.
+;;
+;;    If the test terminates, it's a PASS,
+;;    if it loops forever, that would be a FAIL.
+
+(define expected-blocking-operation
+  (wrap-operation (sleep-operation 1e-4)
+                  (lambda () 'blocking)))
+
+(test-eq "no update -> blocking next-update"
+  'blocking
+  (perform-operation
+   (choice-operation
+    (wrap-operation (wait-for-update-operation (make-update #f))
+                    (lambda (_) 'nonblocking))
+    expected-blocking-operation)))
+
+(test-eq "updated -> non-blocking next-update"
+  'nonblocking
+  (perform-operation
+   (receive (update update!)
+       (make-update 'old)
+     (update! 'new)
+     (wrap-operation (wait-for-update-operation update)
+                     (lambda (update) 'nonblocking)))))
+
+(receive (update update!)
+    (make-update 'old)
+  (let ((new (update! 'new)))
+    (test-eq "updated -> correct non-blocking next-update"
+      new
+      ;; For unknown reasons, using choice-operation
+      ;; and blocking-operation leads to a FAIL.
+      (perform-operation (wait-for-update-operation update)))))
+
+(test-end "update")