mq: Define envelope data type, again.

The new envelope data type can be used without
fibers or multi-threading.

* Makefile.am (modules): Remove replaced
  gnu/gnunet/message/envelope.scm.
  (%.go: %.scm): Do not unset GUILE_LOAD_COMPILED_PATH as that
  would interfere with guile-pfds.
* README.org (Modules): Remove the obsolete
  gnu/gnunet/message/envelope.scm.
  (Message queues): Document new envelope module. Adjust
  message queue blurb for the future.
* gnu/gnunet/mq/envelope.scm: Define new envelope module.
* gnu/gnunet/message/envelope.scm: Delete.
* tests/envelope.scm: Test the new envelope module.

1 files changed, 195 insertions, 0 deletions

diff --git a/gnu/gnunet/mq/envelope.scm b/gnu/gnunet/mq/envelope.scm
new file mode 100644
index 0000000..e0c94a2
--- /dev/null
+++ b/gnu/gnunet/mq/envelope.scm
@@ -0,0 +1,195 @@
+;; This file is part of GNUnet.
+;; Copyright (C) 2012-2019 GNUnet e.V.
+;; Copyright (C) 2021 Maxime Devos (<maximedevos@telenet.be>)
+;;
+;; 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.
+;;
+;; 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
+
+;; Author: Florian Dold
+;; Author: Maxime Devos
+;; C file: util/mq.c
+;; Scheme module: (gnu gnunet mq envelope)
+;;
+;; Limitation: the format of messages is still in flux,
+;; so no type checks there.
+(define-library (gnu gnunet mq envelope)
+  (export <envelope> make-envelope envelope?
+          attempt-cancel! attempt-irrevocable-sent!)
+  (import (gnu gnunet utils hat-let)
+          (only (guile) define* lambda* exact-integer?)
+          (only (ice-9 match) match)
+          (only (ice-9 atomic)
+                make-atomic-box atomic-box-ref
+                atomic-box-compare-and-swap!)
+          (only (rnrs base)
+                lambda assert letrec let begin define
+                syntax-rules let-syntax define-syntax
+                procedure? eq? >= = <= < if quote ...
+                identifier-syntax values and let*
+                vector vector-ref vector? vector-length)
+          (only (rnrs records syntactic) define-record-type))
+  (begin
+    (define-record-type (<envelope> make-envelope envelope?)
+      ;; Atomic box:
+      ;;    #t: cancelled
+      ;;    #f: to late to cancel, message has been irrevocabily sent!
+      ;;
+      ;;        (Unless you play tricks like pulling out the Ethernet
+      ;;        cable before the message is received by the router)
+      ;;    #(message prio notify-sent! cancel!)
+      (fields (immutable state %cancellation-state))
+      (protocol
+       (lambda (%make)
+         (lambda* (cancel! message #:key (priority 0) (notify-sent! values))
+           "Make a message envelope; i.e., a record containing the message
+(@var{message}, @var{priority}) and information on how to cancel the sending
+of the message (@var{cancel!}) and who should be notified when the message
+cannot be unsent anymore (@var{notify-sent!}).
+
+Once marked as cancelled or irrevocabily sent, the record drops its
+references to @var{message}, @var{cancel!} and @var{notify-sent!}.
+When being marked as cancelled, the thunk @var{cancel!} is called."
+           (assert (and (procedure? cancel!) (procedure? notify-sent!)
+                        (exact-integer? priority)
+                        (<= 0 priority 511)))
+           (%make (make-atomic-box
+                   (vector message priority notify-sent! cancel!)))))))
+
+    (define (%attempt-irrevocable-sent! envelope already-sent go cancelled)
+      (bind-atomic-boxen
+       ((state (%cancellation-state envelope) swap!))
+       (let spin ((old state))
+         (match old
+           ;; See comment at %attempt-cancel! for
+           ;; why we don't do #(message prio notify-sent! cancel!)
+           ((? vector?)
+            (if (eq? old (swap! old #f))
+                (let^ ((!! (= (vector-length old) 4))
+                       (! message (vector-ref old 0))
+                       (! prio (vector-ref old 1))
+                       (! notify-sent! (vector-ref old 2)))
+                      (notify-sent!)
+                      (go message prio))
+                (spin state)))
+           (#t (cancelled))
+           (#f (already-sent))))))
+
+    (define-syntax attempt-irrevocable-sent!
+      (syntax-rules (go cancelled already-sent)
+        "If @var{envelope} is not cancelled and has not yet been sent,
+mark the message as irrevocably sent, call the notify-sent callback and
+evaluate @var{exp/go} in an environment where the message @var{message}
+and its priority @var{priority} are bound.
+
+If the message has already been marked as irrevocabily sent,
+evaluate @var{exp/already-sent} instead. If the message is cancelled,
+evaluate @var{exp/cancelled} instead.
+
+Even if this operation (and perhaps @code{attempt-cancel!}) is used concurrently
+on the same @var{envelope}, whether by multi-threading, asynchronicities
+(via @code{system-async-mark}) or by recursion, the following properties hold:
+
+@begin itemize
+@item the notify-sent callback of @var{envelope} is called at most once
+@item the notify-sent callback is never called if @var{envelope} is cancelled
+  at any point in time
+@item likewise, the code in @var{exp/go} is at most evaluated once
+@end itemize"
+        ((_ envelope
+            ((go message priority) . exp/go)
+            ((cancelled) . exp/cancelled)
+            ((already-sent) . exp/already-sent))
+         (%attempt-irrevocable-sent! envelope
+                                     (lambda () . exp/already-sent)
+                                     (lambda (message priority) . exp/go)
+                                     (lambda () . exp/cancelled)))))
+
+    (define (%attempt-cancel! envelope now-cancelled already-cancelled
+                              already-sent)
+      (bind-atomic-boxen
+       ((state (%cancellation-state envelope) swap!))
+       (let spin ((old state))
+         (match old
+           ;; Do _not_ use #(message prio notify-sent! cancel!)
+           ;; here! Instead, delay the bounds check and accessing
+           ;; the elements of the vector after the swap!. That way:
+           ;;
+           ;; Premature optimisation.
+           ;;   We save a little time in case two threads try to concurrently
+           ;;   @var{state}.
+           ;;
+           ;; Meager excuse: self-healing (in case of memory corruption).
+           ;;   Suppose a cosmic ray flipped a few bits and now
+           ;;   @var{state} contains another vector, of different length.
+           ;;   Then by performing the swap before the bounds check,
+           ;;   the envelope is brought into a valid state. (And an
+           ;;   exception will still result.)
+           ((? vector?)
+            (if (eq? old (swap! old #t))
+                (let^ ((!! (= (vector-length old) 4))
+                       (! cancel! (vector-ref old 3)))
+                      (cancel!)
+                      (now-cancelled))
+                (spin state)))
+           (#t (already-cancelled))
+           ;; XXX maybe make the meager excuse less meager
+           ;; and add a 'default' case where @var{state} is
+           ;; set to #f when bad (and an exception is raised)?
+           ;; Seems like some dedicated exception types for
+           ;; memory corruption are required then ...
+           ;; And tests.
+           (#f (already-sent))))))
+
+    (define-syntax attempt-cancel!
+      (syntax-rules (now-cancelled already-cancelled already-sent)
+        "If @var{envelope} is not yet marked as cancelled or sent,
+mark it as cancelled, call the corresponding cancellation callback
+and evaluate @var{exp/now-cancelled}.
+
+If @var{envelope} is already marked as cancelled, do not mutate
+anything or call any callback and evaluate @var{exp/already-cancelled}.
+Likewise, if @var{envelope} is marked as irrevocably sent, evaluate
+@var{exp/already-sent} instead.
+
+If this operation is interrupted before @var{exp/now-cancelled} is
+evaluated, the envelope may be marked as cancelled even if the
+cancellation callback has not yet been called or has not yet returned.
+
+However, by tolerating this limitation, it can be (and is) guaranteed
+that the cancellation callback is called at most once. Likewise, the
+code in @var{exp/now-cancelled} is only be called at most once.
+Also, the cancellation callback and is never called (and @var{exp/now-cancelled}
+never evaluated) if @var{envelope} is marked as sent at any point in time."
+        ((_ envelope
+            ((now-cancelled) . exp/now-cancelled)
+            ((already-cancelled) . exp/already-cancelled)
+            ((already-sent) . exp/already-sent))
+         (%attempt-cancel! envelope
+                           (lambda () . exp/now-cancelled)
+                           (lambda () . exp/already-cancelled)
+                           (lambda () . exp/already-sent)))))
+
+    (define-syntax bind-atomic-boxen
+      (syntax-rules ()
+        ((_ () exp exp* ...)
+         (let () exp exp* ...))
+        ((_ ((variable box swap!) . etc) exp exp* ...)
+         (let ((stashed-box box))
+           (let-syntax ((variable (identifier-syntax
+                                   (atomic-box-ref box))))
+             (let ((swap! (lambda (expected desired)
+                            (atomic-box-compare-and-swap! box expected
+                                                          desired))))
+               (bind-atomic-boxen etc exp exp* ...)))))))))