Repository: man-group/ArcticDB (language: python)

Issue:
Check TTL in try_lock method of the storage lock



Discussion / hints:

Fix patch:
diff --git a/cpp/arcticdb/util/storage_lock.hpp b/cpp/arcticdb/util/storage_lock.hpp
index b0a406a00c..a18acb60d9 100644
--- a/cpp/arcticdb/util/storage_lock.hpp
+++ b/cpp/arcticdb/util/storage_lock.hpp
@@ -117,7 +117,7 @@ class StorageLock {
         OnExit x{[that=this] () {
             that->mutex_.unlock();
         }};
-        if(!ref_key_exists(store)) {
+        if(!ref_key_exists(store) || ttl_expired(store)) {
             ts_= create_ref_key(store);
             auto lock_sleep = ConfigsMap::instance()->get_int("StorageLock.WaitMs", 200);
             std::this_thread::sleep_for(std::chrono::milliseconds(lock_sleep));
@@ -137,6 +137,10 @@ class StorageLock {
         }
     }
 
+    void _test_release_local_lock() {
+        mutex_.unlock();
+    }
+
   private:
     void do_lock(const std::shared_ptr<Store>& store, std::optional<size_t> timeout_ms = std::nullopt) {
         mutex_.lock();
@@ -151,15 +155,8 @@ class StorageLock {
             sleep_ms(wait_ms);
             total_wait += wait_ms;
             wait_ms *= 2;
-            if (auto read_ts = read_timestamp(store); read_ts) {
-                // check TTL
-                auto ttl = ConfigsMap::instance()->get_int("StorageLock.TTL", DEFAULT_TTL_INTERVAL);
-                if (ClockType::coarse_nanos_since_epoch() - *read_ts > ttl) {
-                    log::lock().warn("StorageLock {} taken for more than TTL (default 1 day). Force releasing", name_);
-                    force_release_lock(name_, store);
-                    break;
-                }
-            }
+            if (ttl_expired(store))
+                break;
             if (timeout_ms && total_wait > *timeout_ms) {
                 ts_ = 0;
                 log::lock().info("Lock timed out, giving up after {}", wait_ms);
@@ -227,6 +224,19 @@ class StorageLock {
             return std::nullopt;
         }
     }
+
+    bool ttl_expired(const std::shared_ptr<Store>& store) {
+        if (auto read_ts = read_timestamp(store); read_ts) {
+            // check TTL
+            auto ttl = ConfigsMap::instance()->get_int("StorageLock.TTL", DEFAULT_TTL_INTERVAL);
+            if (ClockType::coarse_nanos_since_epoch() - *read_ts > ttl) {
+                log::lock().warn("StorageLock {} taken for more than TTL (default 1 day). Force releasing", name_);
+                force_release_lock(name_, store);
+                return true;
+            }
+        }
+        return false;
+    }
 };
 
 class StorageLockWrapper {


Test patch:
diff --git a/cpp/arcticdb/util/test/test_storage_lock.cpp b/cpp/arcticdb/util/test/test_storage_lock.cpp
index 12dce26873..baef9ba281 100644
--- a/cpp/arcticdb/util/test/test_storage_lock.cpp
+++ b/cpp/arcticdb/util/test/test_storage_lock.cpp
@@ -52,10 +52,9 @@ struct LockData {
     std::shared_ptr<InMemoryStore> store_;
     volatile uint64_t vol_;
     std::atomic<uint64_t> atomic_;
-    std::mutex mutex_;
-    bool contended_;
+    std::atomic<bool> contended_;
     const size_t num_tests_;
-    bool timedout_;
+    std::atomic<bool> timedout_;
 
     LockData(size_t num_tests) :
     lock_name_("stress_test_lock"),
@@ -77,7 +76,6 @@ struct OptimisticLockTask {
     }
 
     folly::Future<folly::Unit> operator()() {
-        std::this_thread::sleep_for(std::chrono::milliseconds(1));
         StorageLock<> lock{data_->lock_name_};
 
         for (auto i = size_t(0); i < data_->num_tests_; ++i) {
@@ -86,8 +84,6 @@ struct OptimisticLockTask {
             }
             else {
                 ++data_->vol_;
-                // This should be unnecessary as we are already locked
-                std::lock_guard l{data_->mutex_};
                 ++data_->atomic_;
                 lock.unlock(data_->store_);
             }
@@ -108,11 +104,10 @@ TEST(StorageLock, Contention) {
     for(auto i = size_t{0}; i < 4; ++i) {
         futures.emplace_back(exec.addFuture(OptimisticLockTask{lock_data}));
     }
-    std::this_thread::sleep_for(std::chrono::milliseconds(500));
     collect(futures).get();
 
     ASSERT_EQ(lock_data->atomic_, lock_data->vol_);
-    //ASSERT_EQ(lock_data->contended_, true); Alas Jenkins is too rubbish for this, uncomment for tests on headnode
+    ASSERT_EQ(lock_data->contended_, true);
 }
 
 struct PessimisticLockTask {
@@ -125,7 +120,6 @@ struct PessimisticLockTask {
     }
 
     folly::Future<folly::Unit> operator()() {
-        std::this_thread::sleep_for(std::chrono::milliseconds(1));
         StorageLock<> lock{data_->lock_name_};
 
         for (auto i = size_t(0); i < data_->num_tests_; ++i) {
@@ -136,8 +130,6 @@ struct PessimisticLockTask {
                     lock.lock(data_->store_);
 
                 ++data_->vol_;
-                // This should be unnecessary as we are already locked
-                std::lock_guard l{data_->mutex_};
                 ++data_->atomic_;
                 lock.unlock(data_->store_);
             }
@@ -160,14 +152,11 @@ struct ForceReleaseLockTask {
     }
 
     folly::Future<folly::Unit> operator()() {
-        std::this_thread::sleep_for(std::chrono::milliseconds(1));
         StorageLock<> lock{data_->lock_name_};
 
         try {
             lock.lock_timeout(data_->store_, timeout_ms_);
             ++data_->vol_;
-            // This should be unnecessary as we are already locked
-            std::lock_guard l{data_->mutex_};
             ++data_->atomic_;
             // Dont unlock
         }
@@ -175,9 +164,43 @@ struct ForceReleaseLockTask {
             data_->timedout_ = true;
         }
 
-        std::this_thread::sleep_for(std::chrono::milliseconds(5));
         // Clean up locks to avoid "mutex destroyed while active" errors on Windows debug build
-        lock.unlock(data_->store_);
+        lock._test_release_local_lock();
+        return makeFuture(Unit{});
+    }
+};
+
+struct OptimisticForceReleaseLockTask {
+    std::shared_ptr<LockData> data_;
+    size_t timeout_ms_;
+    size_t retry_ms_;
+
+    OptimisticForceReleaseLockTask(std::shared_ptr<LockData> data, size_t timeout_ms, size_t retry_ms) :
+            data_(std::move(data)),
+            timeout_ms_(timeout_ms),
+            retry_ms_(retry_ms)
+    {
+    }
+
+    folly::Future<folly::Unit> operator()() {
+        StorageLock<> lock{data_->lock_name_};
+        bool contended = true;
+        size_t total_wait = 0;
+        while (contended && total_wait < timeout_ms_) {
+            contended = !lock.try_lock(data_->store_);
+            std::this_thread::sleep_for(std::chrono::milliseconds(retry_ms_));
+            total_wait += retry_ms_;
+        }
+        if (contended) {
+            data_->timedout_ = true;
+            data_->contended_ = true;
+        } else {
+            ++data_->vol_;
+            ++data_->atomic_;
+            // Dont unlock
+        }
+        // Clean up locks to avoid "mutex destroyed while active" errors on Windows debug build
+        lock._test_release_local_lock();
         return makeFuture(Unit{});
     }
 };
@@ -214,21 +237,38 @@ TEST(StorageLock, Timeouts) {
     ASSERT_TRUE(lock_data->timedout_);
 }
 
+int count_occurrences(std::string search, std::string pattern) {
+    if (search.size() < pattern.size()) return false;
+    int count = 0;
+    for (size_t pos = 0; pos <= search.size() - pattern.size(); pos++) {
+        if (search.substr(pos, pattern.size()) == pattern)
+            count++;
+    }
+    return count;
+}
+
 TEST(StorageLock, ForceReleaseLock) {
+    // Verify that lock() will take the lock when the TTL has expired.
+    // Then each thread simulates forgetting to release the lock, so that other threads need to rely
+    // on the TTL expiring to be able to acquire the lock.
+    // Initially take the lock, so that the first thread also has to wait for the TTL of that lock
+    // to expire.
     SKIP_MAC("StorageLock is not supported");
     using namespace arcticdb;
 
     auto lock_data = std::make_shared<LockData>(4);
     folly::FutureExecutor<folly::CPUThreadPoolExecutor> exec{4};
-    // This is set in nanoseconds => 1ms
-    ConfigsMap::instance()->set_int("StorageLock.TTL", 1000 * 1000);
-
-    std::vector<std::shared_ptr<StorageLock<>>> locks;
+    // This is set in milliseconds => 50ms for the preempting check
+    ConfigsMap::instance()->set_int("StorageLock.WaitMs", 50);
+    // This is set in nanoseconds => 200ms for the TTL
+    ConfigsMap::instance()->set_int("StorageLock.TTL", 200 * 1000 * 1000);
 
     // Create a first lock that the others will have to force release
     auto first_lock = StorageLock<>(lock_data->lock_name_);
     first_lock.lock(lock_data->store_);
 
+    testing::internal::CaptureStderr();
+    testing::internal::CaptureStdout();
     std::vector<Future<Unit>> futures;
     for(auto i = size_t{0}; i < 4; ++i) {
         futures.emplace_back(exec.addFuture(ForceReleaseLockTask{lock_data, 10 * 1000}));
@@ -239,6 +279,80 @@ TEST(StorageLock, ForceReleaseLock) {
     ASSERT_EQ(4u, lock_data->atomic_);
     ASSERT_EQ(4u, lock_data->vol_);
 
+    std::string stdout_str  = testing::internal::GetCapturedStdout();
+    std::string stderr_str = testing::internal::GetCapturedStderr();
+    std::string expected = "taken for more than TTL";
+
+    ASSERT_EQ(count_occurrences("abab", "ab"), 2u);
+    // If a lock is preempted, then it will still print the warning about having overridden the
+    // lock due to the TTL expiring, but will then have to retry, so there may be more than the expected
+    // number of log messages.
+    // Skip on Windows as capturing logs doesn't work. TODO: Configure the logger with the file output
+#ifndef _WIN32
+    ASSERT_TRUE(
+            count_occurrences(stdout_str, expected) >= 4 ||
+            count_occurrences(stderr_str, expected) >= 4
+    );
+#endif
+
     // Clean up locks to avoid "mutex destroyed while active" errors on Windows debug build
-    first_lock.unlock(lock_data->store_);
+    first_lock._test_release_local_lock();
 }
+
+TEST(StorageLock, OptimisticForceReleaseLock) {
+    // Verify that try_lock() will take the lock when the TTL has expired.
+    // Since this method does not retry automatically, the threads will retry periodically
+    // until the TTL expires, and then they should be able to take the lock. The threads
+    // will then simulate forgetting to release the lock, so that other threads need to rely
+    // on the TTL expiring to be able to acquire the lock.
+    // Initially take the lock using the lock() method, so that the first thread also has to
+    // wait for the TTL of that lock to expire.
+    SKIP_MAC("StorageLock is not supported");
+    using namespace arcticdb;
+
+    auto lock_data = std::make_shared<LockData>(4);
+    folly::FutureExecutor<folly::CPUThreadPoolExecutor> exec{4};
+    // This is set in milliseconds => 50ms for the preempting check
+    ConfigsMap::instance()->set_int("StorageLock.WaitMs", 50);
+    // This is set in nanoseconds => 200ms for the TTL
+    ConfigsMap::instance()->set_int("StorageLock.TTL", 200 * 1000 * 1000);
+
+    // Create a first lock that the others will have to force release
+    auto first_lock = StorageLock<>(lock_data->lock_name_);
+    first_lock.lock(lock_data->store_);
+
+    testing::internal::CaptureStderr();
+    testing::internal::CaptureStdout();
+    std::vector<Future<Unit>> futures;
+    for(auto i = size_t{0}; i < 4; ++i) {
+        futures.emplace_back(exec.addFuture(OptimisticForceReleaseLockTask{lock_data, 10 * 1000, 100}));
+    }
+
+    collect(futures).get();
+    ASSERT_FALSE(lock_data->timedout_);
+    ASSERT_FALSE(lock_data->contended_);
+    ASSERT_EQ(4u, lock_data->atomic_);
+    ASSERT_EQ(4u, lock_data->vol_);
+
+    std::string stdout_str  = testing::internal::GetCapturedStdout();
+    std::string stderr_str = testing::internal::GetCapturedStderr();
+    std::string expected = "taken for more than TTL";
+
+    std::cout << stdout_str << std::endl;
+    std::cout << stderr_str << std::endl;
+
+    ASSERT_EQ(count_occurrences("abab", "ab"), 2u);
+    // If a lock is preempted, then it will still print the warning about having overridden the
+    // lock due to the TTL expiring, but will then have to retry, so there may be more than the expected
+    // number of log messages.
+    // Skip on Windows as capturing logs doesn't work. TODO: Configure the logger with the file output
+#ifndef _WIN32
+    ASSERT_TRUE(
+        count_occurrences(stdout_str, expected) >= 4 ||
+        count_occurrences(stderr_str, expected) >= 4
+    );
+#endif
+
+    // Clean up locks to avoid "mutex destroyed while active" errors on Windows debug build
+    first_lock._test_release_local_lock();
+}
\ No newline at end of file