[WIP] Introduce barrier type

.typos.toml

@@ -1,3 +1,4 @@
 [default.extend-words]
 inout = "inout"
 ocur = "ocur"
+ba = "ba"

doc/autogen/reserved-keywords.txt

@@ -9,6 +9,7 @@ any
 assert
 assert-exception
 attribute
+barrier
 begin
 bitfield
 bool

doc/autogen/types/barrier.rst

@@ -0,0 +1,32 @@
+.. rubric:: Methods
+
+.. spicy:method:: barrier::abort barrier() abort False void ()
+
+    Aborts the barrier, causing any waiting parties to throw a
+    ``BarrierAborted`` exception. This has no effect if the barrier is
+    already released or aborted.
+
+.. spicy:method:: barrier::arrive barrier() arrive False void ()
+
+    Signals a party's arrival at the barrier, potentially releasing it if
+    the expected number of parties have been seen now. This has no effect
+    if the barrier is already released or aborted.
+
+.. spicy:method:: barrier::arrive_and_wait barrier() arrive_and_wait False void ()
+
+    Convenience method combining a `arrive()` with an immediately
+    following `wait()`.
+
+.. spicy:method:: barrier::wait barrier() wait False void ()
+
+    Blocks the caller until the barrier is released by the expected number
+    of parties arriving. If the barrier is already released, it will
+    return immediately. If the barrier gets aborted before or during the
+    wait, the method will throw a ``BarrierAborted`` exception.
+
+.. rubric:: Operators
+
+.. spicy:operator:: barrier::Call barrier() barrier(uint)
+
+    Creates a barrier that will wait for the given number of parties.
+

doc/programming/language/types.rst

@@ -25,6 +25,36 @@ This type supports the :ref:`pack/unpack operators <packing>`.
 
 .. include:: /autogen/types/address.rst
 
+.. _type_barrier:
+
+Barrier
+-------
+
+A ``barrier`` provides a synchronization point for multiple parsing
+stream, such as the two sides of a connection. It comes with an
+expected number ``N`` of parties to reach the synchronization point.
+Each party must call ``arrive()` to signal it has done so. Each party
+may call ``wait()`` to block until all parties have arrived.
+
+.. note::
+
+    It's possible to create dead-lock situations with barriers if you
+    aren't careful. The easiest way to do so is waiting for a barrier
+    on which nobody ever calls ``arrive()``. If a deadlock occurs,
+    it's up to the host application to eventually resolve the
+    situation. Zeek, for example, will eventually time out the
+    connection and clean up the parsing state.
+
+.. rubric:: Type
+
+- ``barrier(N)``
+
+.. rubric:: Constants
+
+- ``barrier(N)``, ``barrier()`` (untyped)
+
+.. include:: /autogen/types/barrier.rst
+
 .. _type_bitfield:
 
 Bitfield

hilti/runtime/CMakeLists.txt

@@ -35,6 +35,7 @@ set(SOURCES
     src/safe-math.cc
     src/type-info.cc
     src/types/address.cc
+    src/types/barrier.cc
     src/types/bytes.cc
     src/types/integer.cc
     src/types/port.cc
@@ -133,6 +134,7 @@ add_executable(
     src/tests/main.cc
     src/tests/address.cc
     src/tests/backtrace.cc
+    src/tests/barrier.cc
     src/tests/bytes.cc
     src/tests/context.cc
     src/tests/debug-logger.cc

hilti/runtime/include/exception.h

@@ -99,6 +99,12 @@ HILTI_EXCEPTION(AssertionFailure, RuntimeError)
  */
 HILTI_EXCEPTION(AttributeNotSet, RuntimeError)
 
+/**
+ * Exception triggered when a barrier synchronization is either explicitly or
+ * implicitly aborted.
+ */
+HILTI_EXCEPTION(BarrierAborted, RuntimeError)
+
 /**
  * Exception triggered when a division by zero is attempted.
  */

hilti/runtime/include/fiber.h

@@ -176,11 +176,17 @@ class Fiber {
     /** Returns the fiber's type. */
     auto type() { return _type; }
 
+    /**
+     * Returns true if the fiber is currently suspended due to waiting for a
+     * barrier.
+     */
+    auto atBarrier() const { return _state == State::YieldedAtBarrier; }
+
     /** Returns the fiber's stack buffer. */
     const auto& stackBuffer() const { return _stack_buffer; }
 
     void run();
-    void yield();
+    void yield(bool at_barrier = false);
     void resume();
     void abort();
 
@@ -189,7 +195,8 @@ class Fiber {
     bool isDone() {
         switch ( _state ) {
             case State::Running:
-            case State::Yielded: return false;
+            case State::Yielded:
+            case State::YieldedAtBarrier: return false;
 
             case State::Aborting:
             case State::Finished:
@@ -226,7 +233,7 @@ class Fiber {
     friend void ::__fiber_run_trampoline(void* argsp);
     friend void ::__fiber_switch_trampoline(void* argsp);
 
-    enum class State { Init, Running, Aborting, Yielded, Idle, Finished };
+    enum class State { Init, Running, Aborting, Yielded, YieldedAtBarrier, Idle, Finished };
 
     void _yield(const char* tag);
     void _activate(const char* tag);
@@ -275,7 +282,7 @@ class Fiber {
 
 std::ostream& operator<<(std::ostream& out, const Fiber& fiber);
 
-extern void yield();
+extern void yield(bool at_barrier = false);
 
 /**
  * Checks that the current fiber has sufficient stack space left for
@@ -328,6 +335,12 @@ class Resumable {
     /** Returns a handle to the currently running function. */
     resumable::Handle* handle() { return _fiber.get(); }
 
+    /**
+     * Returns true if the function is currently suspended due to waiting for a
+     * barrier.
+     */
+    auto atBarrier() const { return (! _done) && _fiber->atBarrier(); }
+
     /**
      * Returns true if the function has completed orderly and provided a result.
      * If so, `get()` can be used to retrieve the result.

hilti/runtime/include/type-info.h

@@ -13,6 +13,7 @@
 #include <hilti/rt/exception.h>
 #include <hilti/rt/fmt.h>
 #include <hilti/rt/types/address.h>
+#include <hilti/rt/types/barrier.h>
 #include <hilti/rt/types/bool.h>
 #include <hilti/rt/types/bytes.h>
 #include <hilti/rt/types/map.h>
@@ -378,6 +379,9 @@ class Address : public detail::AtomicType<hilti::rt::Address> {};
 /** Type information for type ``any`. */
 class Any : public detail::ValueLessType {};
 
+/** Type information for type ``barrier`. */
+class Barrier : public detail::AtomicType<hilti::rt::Barrier> {};
+
 /** Type information for type ``bool`. */
 class Bool : public detail::AtomicType<bool> {};
 
@@ -1148,6 +1152,7 @@ struct TypeInfo {
         Undefined,
         Address,
         Any,
+        Barrier,
         Bool,
         Bytes,
         BytesIterator,
@@ -1198,6 +1203,7 @@ struct TypeInfo {
     union {
         type_info::Address* address;
         type_info::Any* any;
+        type_info::Barrier* barrier;
         type_info::Bool* bool_;
         type_info::Bytes* bytes;
         type_info::BytesIterator* bytes_iterator;
@@ -1256,6 +1262,10 @@ struct TypeInfo {
             tag = Any;
             any = value;
         }
+        else if constexpr ( std::is_same_v<Type, type_info::Barrier> ) {
+            tag = Barrier;
+            barrier = value;
+        }
         else if constexpr ( std::is_same_v<Type, type_info::Bool> ) {
             tag = Bool;
             bool_ = value;
@@ -1450,6 +1460,10 @@ const Type* auxType(const type_info::Value& v) {
         assert(type.tag == TypeInfo::Any);
         return type.any;
     }
+    else if constexpr ( std::is_same_v<Type, type_info::Barrier> ) {
+        assert(type.tag == TypeInfo::Barrier);
+        return type.barrier;
+    }
     else if constexpr ( std::is_same_v<Type, type_info::Bool> ) {
         assert(type.tag == TypeInfo::Bool);
         return type.bool_;
@@ -1624,6 +1638,7 @@ const Type* auxType(const type_info::Value& v) {
 // Forward declare static built-in type information objects.
 extern TypeInfo address;
 extern TypeInfo any;
+extern TypeInfo barrier;
 extern TypeInfo bool_;
 extern TypeInfo bytes_iterator;
 extern TypeInfo bytes;

hilti/runtime/include/types/all.h

@@ -4,6 +4,7 @@
 
 #include <hilti/rt/types/address.h>
 #include <hilti/rt/types/any.h>
+#include <hilti/rt/types/barrier.h>
 #include <hilti/rt/types/bool.h>
 #include <hilti/rt/types/bytes.h>
 #include <hilti/rt/types/enum.h>

hilti/runtime/include/types/barrier.h

@@ -0,0 +1,105 @@
+// Copyright (c) 2020-2021 by the Zeek Project. See LICENSE for details.
+
+#pragma once
+
+#include <arpa/inet.h>
+
+#include <cinttypes>
+#include <string>
+
+#include <hilti/rt/extension-points.h>
+#include <hilti/rt/fiber.h>
+#include <hilti/rt/fmt.h>
+#include <hilti/rt/safe-int.h>
+
+namespace hilti::rt {
+
+/** Represents HILTI's `barrier` type. */
+class Barrier {
+public:
+    /**
+     * Constructs a barrier.
+     *
+     * @param parties number of parties that must arrive at the barrier before it is released
+     */
+    explicit Barrier(const hilti::rt::integer::safe<uint64_t>& expected_parties) : _expected(expected_parties) {}
+
+    /**
+     * Default constructor creating a barrier expecting no parties, meaning
+     * that it will start out as released already.
+     */
+    Barrier() = default;
+
+    Barrier(const Barrier&) = default;
+    Barrier(Barrier&&) noexcept = default;
+    ~Barrier() = default;
+
+    /**
+     * Blocks the caller until the barrier is released. If the barrier is
+     * already released, it will return immediately. If not, it will yield back
+     * to the runtime system, and re-check the barrier state when resumed.
+     *
+     * @throws BarrierAborted if the barrier is aborted either immediately at
+     * initial call time or at a later resume
+     */
+    void wait();
+
+    /**
+     * Signals a party's arrival at the barrier, potentially releasing it if
+     * the expected number of parties have been seen now. This has no effect if
+     * the barrier is already released or aborted.
+     */
+    void arrive();
+
+    /**
+     * Convenience method combining a `arrive()` with an immediately following
+     * `wait()`.
+     */
+    void arrive_and_wait() {
+        arrive();
+        wait();
+    }
+
+    /**
+     * Aborts operation of the barrier. That means that all parties waiting for
+     * it now or later, will receive a `BarrierAborted` exception. This method
+     * has no effect if the barrier has already been released.
+     */
+    void abort();
+
+    /**
+     * Returns true if the expected number of parties has arrived at the
+     * barrier.
+     */
+    bool isReleased() const { return _expected >= 0 && _expected == _arrived; }
+
+    /**
+     * Returns true if the barrier received an abort() before it could get
+     * released.
+     */
+    bool isAborted() const { return _expected < 0; }
+
+    /** Returns true if the barrier has been released. */
+    explicit operator bool() const { return isReleased(); }
+
+    /** Returns a printable representation of the barrier's current state. */
+    operator std::string() const;
+
+    Barrier& operator=(const Barrier&) = default;
+    Barrier& operator=(Barrier&&) noexcept = default;
+
+private:
+    integer::safe<int64_t> _expected = 0;
+    integer::safe<int64_t> _arrived = 0;
+};
+
+namespace detail::adl {
+inline std::string to_string(const hilti::rt::Barrier& x, adl::tag /*unused*/) { return std::string(x); }
+} // namespace detail::adl
+
+inline std::ostream& operator<<(std::ostream& out, const Barrier& x) {
+    out << to_string(x);
+    return out;
+}
+
+} // namespace hilti::rt

hilti/runtime/src/exception.cc

@@ -15,6 +15,7 @@ HILTI_EXCEPTION_IMPL(RecoverableFailure)
 
 HILTI_EXCEPTION_IMPL(AssertionFailure)
 HILTI_EXCEPTION_IMPL(AttributeNotSet)
+HILTI_EXCEPTION_IMPL(BarrierAborted)
 HILTI_EXCEPTION_IMPL(DivisionByZero)
 HILTI_EXCEPTION_IMPL(EnvironmentError)
 HILTI_EXCEPTION_IMPL(ExpiredReference)