job/event: a simple reusable edge triggered event

CMakeLists.txt

@@ -84,6 +84,15 @@ list (
 )
 
 
+##-----------------------------------------------------------------------------
+if (LINUX)
+    list (APPEND UTIL_FILES
+        job/event_linux.cpp
+        )
+endif ()
+
+
+##-----------------------------------------------------------------------------
 if (NOT WINDOWS)
     list (
         APPEND UTIL_FILES
@@ -123,6 +132,7 @@ if (WINDOWS)
         win32/handle.hpp
         win32/registry.hpp
         win32/registry.cpp
+        job/event_win32.cpp
     )
 endif ()
 
@@ -254,6 +264,7 @@ list (
     io.cpp
     io.hpp
     iterator.hpp
+    job/event.hpp
     job/queue.cpp
     job/queue.hpp
     json/fwd.hpp
@@ -463,6 +474,7 @@ if (TESTS)
         hton
         introspection
         iterator
+        job/event
         job/queue
         json_types
         json2/event

job/event.hpp

@@ -0,0 +1,61 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * Copyright 2018 Danny Robson <danny@nerdcruft.net>
+ */
+
+#ifndef CRUFT_UTIL_JOB_EVENT_HPP
+#define CRUFT_UTIL_JOB_EVENT_HPP
+
+#include <atomic>
+
+namespace util::job {
+    /// a reusable synchronisation object that allows threads to wait until
+    /// notify is called.
+    ///
+    /// there is no internal state so it is easy to create races between wait
+    /// and notify calls. this makes the class mostly suitable for recurring
+    /// events.
+    ///
+    /// the user should ensure no callers are waiting at destruction time
+    /// otherwise they may remain blocked indefinitely.
+    ///
+    /// the address of the object is important so it must _never_ be
+    /// relocated in any manner if any caller may be waiting. it may be safe
+    /// to do so if there are no callers waiting (but the relevant functions
+    /// are deleted for safety anyway).
+    class event {
+    public:
+        event ();
+
+        event (const event&) = delete;
+        event (event&&) = delete;
+        event& operator= (const event&) = delete;
+        event& operator= (event&&) = delete;
+
+        /// block until notified
+        void wait (void);
+
+        /// wake all threads that are waiting
+        int notify (void);
+
+        /// wait `count' threads that are waiting
+        int notify (int count);
+
+    private:
+        alignas (4) std::atomic<int> value;
+    };
+}
+
+#endif
+

job/event_linux.cpp

@@ -0,0 +1,79 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * Copyright 2018 Danny Robson <danny@nerdcruft.net>
+ */
+
+#include "event.hpp"
+#include "../cast.hpp"
+
+#include <cerrno>
+#include <linux/futex.h>
+#include <unistd.h>
+#include <sys/syscall.h>
+#include <cruft/util/posix/except.hpp>
+#include <limits>
+
+using util::job::event;
+
+
+///////////////////////////////////////////////////////////////////////////////
+static long
+sys_futex (void *addr1, int op, int val1, struct timespec *timeout, void *addr2, int val3)
+{
+    return syscall (SYS_futex, addr1, op | FUTEX_PRIVATE_FLAG, val1, timeout, addr2, val3);
+}
+
+///////////////////////////////////////////////////////////////////////////////
+event::event ():
+    value (0)
+{ ; }
+
+
+///////////////////////////////////////////////////////////////////////////////
+void
+event::wait (void)
+{
+    for (auto observed = value.load (); observed == value.load (); ) {
+        auto res = sys_futex (&value, FUTEX_WAIT, observed, nullptr, nullptr, 0);
+
+        if (res < 0) {
+            switch (errno) {
+            case EAGAIN: return;
+            case EINTR:  continue;
+            }
+
+            posix::error::throw_code ();
+        }
+    }
+}
+
+
+///////////////////////////////////////////////////////////////////////////////
+int
+event::notify (void)
+{
+    return notify (std::numeric_limits<int>::max ());
+}
+
+
+//-----------------------------------------------------------------------------
+int
+event::notify (int count)
+{
+    ++value;
+    auto res = sys_futex (&value, FUTEX_WAKE, count, nullptr, nullptr, 0);
+    if (res < 0)
+        posix::error::throw_code ();
+    return util::cast::narrow<int> (res);
+}

test/job/event.cpp

@@ -0,0 +1,40 @@
+#include "job/event.hpp"
+#include "tap.hpp"
+
+#include <atomic>
+#include <thread>
+#include <iostream>
+
+
+int
+main ()
+{
+    util::TAP::logger tap;
+
+    // create an event which a thread will wait on. after it's been woken up
+    // it will modify the contents of val. this will be tested at a couple of
+    // points for consistency.
+    //
+    // the test isn't 100% deterministic (because we're attempting to create
+    // specific timings by just waiting). but it's a decent first check.
+    std::atomic<int> val = 0;
+    util::job::event a;
+
+    std::thread t ([&] () {
+        a.wait ();
+        ++val;
+    });
+
+    // block for hopefully long enough to allow the above thread to change
+    // the value of the integer.
+    std::this_thread::sleep_for (std::chrono::milliseconds (100));
+
+    tap.expect_eq (val, 0, "waiting actually blocks");
+
+    a.notify ();
+    t.join ();
+
+    tap.expect_eq (val, 1, "notification released the lock");
+
+    return tap.status ();
+}