Refactor, still poc

Author: ognyanstoimenov

cpp/arcticdb/storage/failure_simulation.hpp

@@ -23,7 +23,7 @@ namespace arcticdb {
 enum class FailureType : int {
     WRITE = 0,
     READ,
-    WRITE_SLOW,
+    WRITE_LOCK,
     ITERATE,
     DELETE,
 };
@@ -176,7 +176,7 @@ class StorageFailureSimulator {
         Params default_actions = {
             {WRITE, {action_factories::fault(cfg.write_failure_prob())}},
             {READ, {action_factories::fault(cfg.read_failure_prob())}},
-            {WRITE_SLOW, {action_factories::slow_action(cfg.write_slowdown_prob(), cfg.slow_down_min_ms(), cfg.slow_down_max_ms())}}
+            {WRITE_LOCK, {action_factories::slow_action(cfg.write_slowdown_prob(), cfg.slow_down_min_ms(), cfg.slow_down_max_ms())}}
         };
 
         configure(default_actions);

cpp/arcticdb/util/storage_lock.hpp

@@ -168,8 +168,8 @@ class StorageLock {
             auto read_ts = read_timestamp(store);
             auto duration = ClockType::coarse_nanos_since_epoch() - start;
             auto duration_in_ms = duration / ONE_MILLISECOND;
-            ARCTICDB_INFO(log::lock(), "Took {} ms", duration_in_ms);
-            ARCTICDB_INFO(log::lock(), "Max is {} ms", 1.5 * lock_sleep_ms);
+            ARCTICDB_DEBUG(log::lock(), "Took {} ms", duration_in_ms);
+            ARCTICDB_DEBUG(log::lock(), "Max is {} ms", 1.5 * lock_sleep_ms);
             if (duration > 1.5 * lock_sleep_ms * ONE_MILLISECOND) {
                 ARCTICDB_DEBUG(log::lock(), "Took too long to read and write the lock. Aborting.");
                 return false;
@@ -194,7 +194,7 @@ class StorageLock {
 
     timestamp create_ref_key(const std::shared_ptr<Store>& store) {
         auto ts =  ClockType::nanos_since_epoch();
-        StorageFailureSimulator::instance()->go(FailureType::WRITE_SLOW);
+        StorageFailureSimulator::instance()->go(FailureType::WRITE_LOCK);
         store->write_sync(KeyType::LOCK, name_, lock_segment(name_, ts));
         ARCTICDB_DEBUG(log::lock(), "Created lock with timestamp {}", ts);
         return ts;

cpp/arcticdb/util/test/test_storage_lock.cpp

@@ -365,69 +365,76 @@ TEST(StorageLock, OptimisticForceReleaseLock) {
     first_lock._test_release_local_lock();
 }
 
-TEST(StorageLock, ConcurrentWrites) {
-    log::lock().set_level(spdlog::level::debug);
-    const StorageFailureSimulator::ParamActionSequence SLOW_ON_2ND_CALL = {
-        action_factories::slow_action(1, 10, 10), action_factories::slow_action(1, 1700, 1700)
+
+class StorageLockWithSlowWrites : public ::testing::TestWithParam<std::tuple<int, int>> {
+protected:
+    void SetUp() override {
+        log::lock().set_level(spdlog::level::debug);
+        StorageFailureSimulator::instance()->reset();
+    }
+};
+
+TEST_P(StorageLockWithSlowWrites, ConcurrentWrites) {
+    const int first_delay = std::get<0>(GetParam());
+    const int second_delay = std::get<1>(GetParam());
+
+    const StorageFailureSimulator::ParamActionSequence SLOW_ACTIONS = {
+        action_factories::slow_action(1, first_delay, first_delay),
+        action_factories::slow_action(1, second_delay, second_delay)
     };
+
     constexpr size_t num_writers = 2;
+    FutureExecutor<CPUThreadPoolExecutor> exec{num_writers};
+
+    StorageFailureSimulator::instance()->configure({{FailureType::WRITE_LOCK, SLOW_ACTIONS}});
 
-    using StorageLockType = StorageLock<>;
-    std::vector<std::unique_ptr<StorageLockType>> locks;
-    locks.reserve(num_writers);
-    while (locks.size() < num_writers) { locks.emplace_back(std::make_unique<StorageLockType>("test")); }
-    auto store = std::make_shared<InMemoryStore>();
-    folly::FutureExecutor<folly::CPUThreadPoolExecutor> exec{num_writers};
-    StorageFailureSimulator::instance()->configure({{FailureType::WRITE_SLOW, SLOW_ON_2ND_CALL}});
     std::vector<Future<Unit>> futures;
-    std::vector<bool> writer_has_lock;
-    writer_has_lock.resize(num_writers);
-    for(size_t i = 0; i < num_writers; ++i) {
-        futures.emplace_back(exec.addFuture([&store, &lock = locks[i], i, &writer_has_lock] {
-            writer_has_lock[i] = lock->try_lock(store);
-        }));
+    auto lock_data = std::make_shared<LockData>(num_writers);
+    lock_data->store_ = std::make_shared<InMemoryStore>();
+
+    for (size_t i = 0; i < num_writers; ++i) {
+        futures.emplace_back(exec.addFuture(OptimisticLockTask(lock_data)));
     }
     collect(futures).get();
 
-    const long have_lock = std::ranges::count(writer_has_lock, true);
-    ASSERT_EQ(have_lock, 0); // Both should fail
+    const int expected_locks = second_delay < 1000 ? 0 : 1;
+    ASSERT_EQ(lock_data->atomic_, expected_locks); // Both should fail
 }
 
-TEST(StorageLock, ConcurrentWrites2) {
-    log::lock().set_level(spdlog::level::debug);
-    log::lock().set_level(spdlog::level::debug);
-    const StorageFailureSimulator::ParamActionSequence SLOW_ON_2ND_CALL = {
-        action_factories::slow_action(1, 10, 10), action_factories::slow_action(1, 500, 600)
+TEST_P(StorageLockWithSlowWrites, ConcurrentWritesWithRetrying) {
+    const int first_delay = std::get<0>(GetParam());
+    const int second_delay = std::get<1>(GetParam());
+
+    const StorageFailureSimulator::ParamActionSequence SLOW_ACTIONS = {
+        action_factories::slow_action(1, first_delay, first_delay),
+        action_factories::slow_action(1, second_delay, second_delay)
     };
     constexpr size_t num_writers = 2;
 
-    using StorageLockType = StorageLock<>;
-    std::vector<std::unique_ptr<StorageLockType>> locks;
-    locks.reserve(num_writers);
-    while (locks.size() < num_writers) { locks.emplace_back(std::make_unique<StorageLockType>("test")); }
-    auto store = std::make_shared<InMemoryStore>();
-    folly::FutureExecutor<folly::CPUThreadPoolExecutor> exec{num_writers};
-    StorageFailureSimulator::instance()->configure({{FailureType::WRITE_SLOW, SLOW_ON_2ND_CALL}});
+    auto lock_data = std::make_shared<LockData>(num_writers);
+    lock_data->store_ = std::make_shared<InMemoryStore>();
+    FutureExecutor<CPUThreadPoolExecutor> exec{num_writers};
+    StorageFailureSimulator::instance()->configure({{FailureType::WRITE_LOCK, SLOW_ACTIONS}});
     std::vector<Future<Unit>> futures;
     std::vector<bool> writer_has_lock;
     writer_has_lock.resize(num_writers);
     for(size_t i = 0; i < num_writers; ++i) {
-        futures.emplace_back(exec.addFuture([&store, &lock = locks[i], i, &writer_has_lock] {
-            try {
-                lock->lock_timeout(store, 2000);
-                writer_has_lock[i] = true;
-            }
-            catch(const StorageLockTimeout&) {
-                writer_has_lock[i] = false;
-            }
-        }));
+        futures.emplace_back(exec.addFuture(PessimisticLockTask(lock_data, 3000)));
     }
     collect(futures).get();
 
-    const long have_lock = std::ranges::count(writer_has_lock, true);
-    ASSERT_EQ(have_lock, 1);
+    ASSERT_EQ(lock_data->atomic_, 1);
 }
 
+INSTANTIATE_TEST_SUITE_P(
+    DelayPairs,
+    StorageLockWithSlowWrites,
+    ::testing::Values(
+        std::make_tuple(10, 800),
+        std::make_tuple(10, 1700),
+        std::make_tuple(10, 2000)
+    )
+);
 
 TEST(StorageLock, SlowWrites) {
     const auto current_lock_sleep_wait_ms = ConfigsMap::instance()->get_int("StorageLock.WaitMs", StorageLock<>::DEFAULT_WAIT_MS);
@@ -436,7 +443,7 @@ TEST(StorageLock, SlowWrites) {
     const StorageFailureSimulator::ParamActionSequence SLOW_WRITE = {
         action_factories::slow_action(1, min_ms, max_ms)
     };
-    StorageFailureSimulator::instance()->configure({{FailureType::WRITE_SLOW, SLOW_WRITE}});
+    StorageFailureSimulator::instance()->configure({{FailureType::WRITE_LOCK, SLOW_WRITE}});
     auto lock = StorageLock("test");
     ASSERT_FALSE(lock.try_lock(std::make_shared<InMemoryStore>()));
 }
@@ -455,7 +462,7 @@ TEST(StorageLock, DISABLED_LockSameTimestamp) { // Not yet implemented
     while (locks.size() < num_writers) { locks.emplace_back(std::make_unique<StorageLockType>("test")); };
     auto store = std::make_shared<InMemoryStore>();
     folly::FutureExecutor<folly::CPUThreadPoolExecutor> exec{num_writers};
-    StorageFailureSimulator::instance()->configure({{FailureType::WRITE_SLOW, SLOW_ON_2ND_CALL}});
+    StorageFailureSimulator::instance()->configure({{FailureType::WRITE_LOCK, SLOW_ON_2ND_CALL}});
     std::vector<Future<Unit>> futures;
     std::vector<bool> writer_has_lock;
     writer_has_lock.resize(num_writers);