added Latch utility

6 files changed, 178 insertions, 3 deletions

diff --git a/vespalib/CMakeLists.txt b/vespalib/CMakeLists.txt
index 8bd3dda4e5a..fb60cc66931 100644
--- a/vespalib/CMakeLists.txt
+++ b/vespalib/CMakeLists.txt
@@ -50,6 +50,7 @@ vespa_define_module(
     src/tests/host_name
     src/tests/io/fileutil
     src/tests/io/mapped_file_input
+    src/tests/latch
     src/tests/left_right_heap
     src/tests/make_fixture_macros
     src/tests/memory
diff --git a/vespalib/src/tests/latch/CMakeLists.txt b/vespalib/src/tests/latch/CMakeLists.txt
new file mode 100644
index 00000000000..57a67d74868
--- /dev/null
+++ b/vespalib/src/tests/latch/CMakeLists.txt
@@ -0,0 +1,8 @@
+# Copyright 2017 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+vespa_add_executable(vespalib_latch_test_app TEST
+    SOURCES
+    latch_test.cpp
+    DEPENDS
+    vespalib
+)
+vespa_add_test(NAME vespalib_latch_test_app COMMAND vespalib_latch_test_app)
diff --git a/vespalib/src/tests/latch/latch_test.cpp b/vespalib/src/tests/latch/latch_test.cpp
new file mode 100644
index 00000000000..f29b673e508
--- /dev/null
+++ b/vespalib/src/tests/latch/latch_test.cpp
@@ -0,0 +1,99 @@
+// Copyright 2018 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
+#include <vespa/vespalib/testkit/test_kit.h>
+#include <vespa/vespalib/util/gate.h>
+#include <vespa/vespalib/util/latch.h>
+
+using namespace vespalib;
+
+TEST("require that write then read works") {
+    Latch<int> latch;
+    EXPECT_TRUE(!latch.has_value());
+    latch.write(42);
+    EXPECT_TRUE(latch.has_value());
+    EXPECT_EQUAL(latch.read(), 42);
+    EXPECT_TRUE(!latch.has_value());
+}
+
+TEST_MT_FFF("require that read waits for write", 2, Latch<int>(), Gate(), TimeBomb(60)) {
+    if (thread_id == 0) {
+        EXPECT_TRUE(!f2.await(10));
+        f1.write(123);
+        EXPECT_TRUE(f2.await(60000));        
+    } else {
+        EXPECT_EQUAL(f1.read(), 123);
+        f2.countDown();
+    }
+}
+
+TEST_MT_FFF("require that write waits for read", 2, Latch<int>(), Gate(), TimeBomb(60)) {
+    if (thread_id == 0) {
+        f1.write(123);
+        f1.write(456);
+        f2.countDown();
+    } else {
+        EXPECT_TRUE(!f2.await(10));
+        EXPECT_EQUAL(f1.read(), 123);
+        EXPECT_TRUE(f2.await(60000));        
+        EXPECT_EQUAL(f1.read(), 456);
+    }
+}
+
+struct MyInt {
+    int value;
+    MyInt(int value_in) : value(value_in) {}
+    MyInt(MyInt &&rhs) = default;
+    MyInt(const MyInt &rhs) = delete;
+    MyInt &operator=(const MyInt &rhs) = delete;
+    MyInt &operator=(MyInt &&rhs) = delete;
+};
+
+TEST("require that un-assignable non-default-constructable move-only objects can be used") {
+    Latch<MyInt> latch;
+    latch.write(MyInt(1337));
+    EXPECT_EQUAL(latch.read().value, 1337);
+}
+
+struct MyObj {
+    static int total;
+    int *with_state;
+    MyObj(int &with_state_in) : with_state(&with_state_in) {}
+    MyObj(MyObj &&rhs) {
+        with_state = rhs.with_state;
+        rhs.with_state = nullptr;
+    }
+    void detach() { with_state = nullptr; }
+    ~MyObj() {
+        ++total;
+        if (with_state) {
+            ++(*with_state);
+        }
+    }
+    MyObj(const MyObj &rhs) = delete;
+    MyObj &operator=(const MyObj &rhs) = delete;
+    MyObj &operator=(MyObj &&rhs) = delete;
+};
+int MyObj::total = 0;
+
+TEST("require that latched objects are appropriately destructed") {
+    int with_state = 0;
+    int total_sample = 0;
+    {
+        Latch<MyObj> latch1;
+        Latch<MyObj> latch2;
+        Latch<MyObj> latch3;
+        latch2.write(MyObj(with_state));
+        latch3.write(MyObj(with_state));
+        latch2.read().detach();
+        EXPECT_TRUE(!latch1.has_value());
+        EXPECT_TRUE(!latch2.has_value());
+        EXPECT_TRUE(latch3.has_value());
+        EXPECT_EQUAL(with_state, 0);
+        EXPECT_GREATER_EQUAL(MyObj::total, 1);
+        total_sample = MyObj::total;
+    }
+    EXPECT_EQUAL(MyObj::total, total_sample + 1);
+    EXPECT_EQUAL(with_state, 1);
+}
+
+TEST_MAIN() { TEST_RUN_ALL(); }
diff --git a/vespalib/src/vespa/vespalib/util/CMakeLists.txt b/vespalib/src/vespa/vespalib/util/CMakeLists.txt
index 9ce28d62e18..fdf7524c82f 100644
--- a/vespalib/src/vespa/vespalib/util/CMakeLists.txt
+++ b/vespalib/src/vespa/vespalib/util/CMakeLists.txt
@@ -25,6 +25,7 @@ vespa_add_library(vespalib_vespalib_util OBJECT
     hdr_abort.cpp
     host_name.cpp
     joinable.cpp
+    latch.cpp
     left_right_heap.cpp
     lz4compressor.cpp
     md5.c
@@ -42,12 +43,12 @@ vespa_add_library(vespalib_vespalib_util OBJECT
     simple_thread_bundle.cpp
     slaveproc.cpp
     stash.cpp
-    stringfmt.cpp
     string_hash.cpp
-    thread_bundle.cpp
+    stringfmt.cpp
     thread.cpp
-    threadstackexecutorbase.cpp
+    thread_bundle.cpp
     threadstackexecutor.cpp
+    threadstackexecutorbase.cpp
     time_tracker.cpp
     valgrind.cpp
     zstdcompressor.cpp
diff --git a/vespalib/src/vespa/vespalib/util/latch.cpp b/vespalib/src/vespa/vespalib/util/latch.cpp
new file mode 100644
index 00000000000..3c86b4d214e
--- /dev/null
+++ b/vespalib/src/vespa/vespalib/util/latch.cpp
@@ -0,0 +1,7 @@
+// Copyright 2018 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
+#include "latch.h"
+
+namespace vespalib {
+
+} // namespace vespalib
diff --git a/vespalib/src/vespa/vespalib/util/latch.h b/vespalib/src/vespa/vespalib/util/latch.h
new file mode 100644
index 00000000000..73336111c0c
--- /dev/null
+++ b/vespalib/src/vespa/vespalib/util/latch.h
@@ -0,0 +1,59 @@
+// Copyright 2018 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
+#pragma once
+
+#include <mutex>
+#include <condition_variable>
+#include <cassert>
+
+namespace vespalib {
+
+/**
+ * A latch acts like a blocking queue where the maximum capacity is a
+ * single element. It enables directional exchange of data where reads
+ * and writes are alternating.
+ **/
+template <typename T>
+class Latch {
+private:
+    std::mutex              _lock;
+    std::condition_variable _cond;
+    char                    _space[sizeof(T)];
+    bool                    _has_value;
+
+    void *as_void() { return &_space[0]; }
+    T *as_value() { return (T*)as_void(); }
+public:
+    Latch() : _lock(), _cond(), _space(), _has_value(false) {}
+    ~Latch() {
+        if (_has_value) {
+            as_value()->~T();
+        }
+    }
+    bool has_value() {
+        std::lock_guard guard(_lock);
+        return _has_value;
+    }
+    T read() {
+        std::unique_lock guard(_lock);
+        while (!_has_value) {
+            _cond.wait(guard);
+        }
+        T value = std::move(*as_value());
+        as_value()->~T();
+        _has_value = false;
+        _cond.notify_all();
+        return value;
+    }
+    void write(T value) {
+        std::unique_lock guard(_lock);
+        while (_has_value) {
+            _cond.wait(guard);
+        }
+        new (as_void()) T(std::move(value));
+        _has_value = true;
+        _cond.notify_all();
+    }
+};
+
+} // namespace vespalib