Issue/619

README.md

@@ -292,11 +292,11 @@ Configure dbzero during initialization:
 
 ```python
 db0.init(
-    storage_dir="/path/to/data",
+    dbzero_root="/path/to/data",
     config={
         'autocommit': True,
-        'autocommit_interval': 1000,  # milliseconds
-        'cache_size': 1024 * 1024 * 100  # 100MB
+        'autocommit_interval': 367,  # milliseconds
+        'cache_size': 8 << 30 # 8GiB
     }
 )
 ```

dbzero/dbzero/initialization.py

@@ -24,7 +24,7 @@ def init(dbzero_root: str, **kwargs) -> None:
 
         Configure global dbzero behavior:
         * autocommit (bool, default True) to enable automatic commits
-        * autocommit_interval (int, default 250) for commit interval in milliseconds  
+        * autocommit_interval (int, default 367) for commit interval in milliseconds
         * cache_size (int, default 2 GiB) for main object cache size in bytes
         * lang_cache_size (int, default 1024) for language model data cache size
         * lock_flags (dict) to configure locking behavior when opening the prefix in read-write mode

python_tests/test_autocommit.py

@@ -12,9 +12,10 @@
 
 def test_db0_starts_autocommit_by_default(db0_fixture):
     object_1 = MemoTestClass(951)
+    commit_interval = db0.get_config()['autocommit_interval']
     state_1 = db0.get_state_num()
-    # auto-commit should happen no later than within 250ms    
-    time.sleep(0.3)
+    # wait as long as autocommit interval + 100ms margin
+    time.sleep(commit_interval / 1000.0 + 0.1)
     state_2 = db0.get_state_num()
     # state changed due to autocommit
     assert state_2 > state_1
@@ -136,7 +137,7 @@ def test_list_items_append(db0_autocommit_fixture):
 def test_autocommit_config(db0_fixture):
     cfg = db0.get_config()
     assert cfg['autocommit'] == True
-    assert cfg['autocommit_interval'] == 250
+    default_interval = cfg['autocommit_interval']
 
     db0.close()
     db0.init(DB0_DIR, autocommit=False, autocommit_interval=1000)
@@ -148,7 +149,7 @@ def test_autocommit_config(db0_fixture):
     db0.init(DB0_DIR, autocommit=False)
     cfg = db0.get_config()
     assert cfg['autocommit'] == False
-    assert cfg['autocommit_interval'] == 250
+    assert cfg['autocommit_interval'] == default_interval
 
     db0.close()
     with pytest.raises(Exception):

python_tests/test_copy_prefix.py

@@ -221,7 +221,7 @@ def validate_copy(copy_id, expected_len = None, expected_min_len = None):
 
     # in each 'epoch' we modify prefix while making copies
     # then drop the original prefix and restore if from the last copy    
-    epoch_count = 1
+    epoch_count = 3
     total_len = 0
     for epoch in range(epoch_count):
         print(f"=== Epoch {epoch} ===", flush=True)

python_tests/test_crash_recovery.py

@@ -26,14 +26,18 @@ def generator_process(op_size, op_count, crash_after):
         db0.init(DB0_DIR, autocommit=False)
         db0.open(px_name, "rw")
         root = MemoTestSingleton()
-        next_id = len(root.value)    
+        next_id = len(root.value)
         for i in range(op_count):
             for _ in range(op_size):
                 root.value.append(MemoTestClass(next_id))                
                 root.value_2.append(expected_values[next_id])
                 next_id += 1
-            if crash_after is not None and i == crash_after:
-                db0.dbg_crash_from_commit(3)
+            if crash_after is not None and i == crash_after[0]:
+                # activate the write poison to simulate a crash
+                if crash_after[1]:
+                    db0.set_test_params(dram_io_flush_poison = crash_after[1])
+                else:
+                    db0.set_test_params(write_poison = 3)
             db0.commit()
         # NOTE: db0.close is not called if the process crashes
         db0.close()
@@ -43,18 +47,20 @@ def generator_process(op_size, op_count, crash_after):
     db0.close()
 
     # NOTE: the 2nd process will crash during the 3rd commit
-    crash_after = [None, 3, None]
+    # the 3rd process will be killed during DRAM IO flush
+    crash_after = [None, (3, None), (2, 2), None]
     op_size = 50
     op_count = 5
-    for i in range(3):
-        p = multiprocessing.Process(target=generator_process, args=(op_size, op_count, crash_after[i]))
+    for poison in crash_after:
+        p = multiprocessing.Process(target=generator_process, args=(op_size, op_count, poison))
         p.start()
         p.join()
 
     db0.init(DB0_DIR)
     db0.open(px_name, "r")
-    # NOTE: we expect the 2 transactions to be discarded
-    expected_len = op_size * op_count * 3 - (2 * op_size)
+    # NOTE: we expect the 2 cycles to be fully completed
+    # and 2 to be partially completed
+    expected_len = op_size * op_count * 2 + (5 * op_size)
     assert len(MemoTestSingleton().value) == expected_len
     assert len(MemoTestSingleton().value_2) == expected_len
 

python_tests/test_durability.py

@@ -80,8 +80,8 @@ def open_prefix_then_crash():
     # end process with exception before commit / close
     raise Exception("Crash!")
 
-def test_opening_prefix_of_crashed_process(db0_no_default_fixture):
 
+def test_opening_prefix_of_crashed_process(db0_no_default_fixture):
     p = multiprocessing.Process(target=open_prefix_then_crash)
     p.start()
     p.join()
@@ -92,7 +92,6 @@ def test_opening_prefix_of_crashed_process(db0_no_default_fixture):
 
 
 def test_modify_prefix_of_crashed_process(db0_no_default_fixture):
-
     p = multiprocessing.Process(target=open_prefix_then_crash)
     p.start()
     p.join()
@@ -102,12 +101,14 @@ def test_modify_prefix_of_crashed_process(db0_no_default_fixture):
     db0.tags(MemoTestClass(123)).add("tag1", "tag2")    
     db0.commit()
 
+
 def rand_string(max_len):
     import random
     import string
     actual_len = random.randint(1, max_len)
     return ''.join(random.choice(string.ascii_letters) for i in range(actual_len))    
 
+
 def create_objects(append_count=1000):
     db0.open("new-prefix-1")        
     buf = db0.list()
@@ -123,23 +124,6 @@ def create_objects(append_count=1000):
     db0.commit()
     db0.close()
 
-# def test_durability_of_random_objects_issue1(db0_no_default_fixture):
-#     """
-#     This test was failing with an exception when opening the prefix:
-#     BDevStorage::findMutation: page_num 0 not found, state: 10
-#     Resolution: the problem was related to DirtyCache assuming page_size as pow 2 while DRAM_Prefix page size is not    
-#     """
-#     append_count = 1000
-
-#     p = multiprocessing.Process(target=create_objects, args=(append_count,))
-#     p.start()
-#     p.join()
-
-#     # read the created objects
-#     db0.open("new-prefix-1", "r")
-#     root = MemoTestSingleton()
-#     assert len(root.value) == append_count
-
 
 def test_dump_dram_io_map(db0_fixture):
     if 'D' in db0.build_flags():

src/dbzero/bindings/python/PyAPI.cpp

@@ -1424,30 +1424,7 @@ namespace db0::python
         db0::Settings::__dbg_logs = true;
         Py_RETURN_NONE;
     }
-
-    PyObject *tryCrashFromCommit(unsigned int op_count)
-    {
-        PyToolkit::getPyWorkspace().getWorkspace().setCrashFromCommit(op_count);
-        Py_RETURN_NONE;
-    }
-
-    PyObject *PyAPI_crashFromCommit(PyObject *self, PyObject *const *args, Py_ssize_t nargs)
-    {
-        PY_API_FUNC
-        if (nargs != 1) {
-            PyErr_SetString(PyExc_TypeError, "Function requires exactly 1 argument");
-            return NULL;
-        }
-
-        if (!PyLong_Check(args[0])) {
-            PyErr_SetString(PyExc_TypeError, "Argument must be an integer");
-            return NULL;
-        }
-
-        auto op_count = PyLong_AsUnsignedLong(args[0]);
-        return runSafe(tryCrashFromCommit, op_count);
-    }
-
+
     PyObject *PyAPI_breakpoint(PyObject *self, PyObject *const *args, Py_ssize_t nargs)
     {
         PY_API_FUNC

src/dbzero/bindings/python/PyAPI.hpp

@@ -183,9 +183,7 @@ namespace db0::python
 
     // For a specific prefix, extract page num -> state num mapping related with its DRAM_Prefix
     PyObject *getDRAM_IOMap(PyObject *, PyObject *args, PyObject *kwargs);
-
-    PyObject *PyAPI_crashFromCommit(PyObject *self, PyObject *const *args, Py_ssize_t nargs);    
-
+
     PyObject *PyAPI_breakpoint(PyObject *self, PyObject *const *args, Py_ssize_t nargs);
     PyObject *PyAPI_enableStorageValidation(PyObject *, PyObject *args, PyObject *kwargs);
     PyObject *PyAPI_setTestParams(PyObject *, PyObject *args, PyObject *kwargs);

src/dbzero/bindings/python/PyInternalAPI.cpp

@@ -1053,10 +1053,18 @@ namespace db0::python
     PyObject *trySetTestParams(PyObject *py_dict)
     {
         db0::Config config(py_dict);
-        db0::Settings::__sleep_interval = config.get<unsigned long long>("sleep_interval", 0);
+        if (config.hasKey("sleep_interval")) {
+            db0::Settings::__sleep_interval = config.get<unsigned long long>("sleep_interval", 0);            
+        }
+        if (config.hasKey("write_poison")) {
+            db0::Settings::__write_poison = config.get<unsigned int>("write_poison", 0);
+        }
+        if (config.hasKey("dram_io_flush_poison")) {
+            db0::Settings::__dram_io_flush_poison = config.get<unsigned int>("dram_io_flush_poison", 0);   
+        }
         Py_RETURN_NONE;
     }
-     
+
     PyObject *tryResetTestParams()
     {        
         db0::Settings::reset();

src/dbzero/bindings/python/PyToolkit.cpp

@@ -671,6 +671,20 @@ namespace db0::python
         }
         return PyLong_AsUnsignedLongLong(*py_value);
     }
+
+    std::optional<unsigned int> PyToolkit::getUnsignedInt(ObjectPtr py_object, const std::string &key)
+    {
+        auto py_value = Py_OWN(getValue(py_object, key));
+        if (!py_value) {
+            return std::nullopt;
+        }        
+
+        if (!PyLong_Check(*py_value)) {
+            THROWF(db0::InputException) << "Invalid type of: " << key << ". Integer expected but got: " 
+                << Py_TYPE(*py_value)->tp_name << THROWF_END;
+        }
+        return PyLong_AsUnsignedLong(*py_value);
+    }
 
     std::optional<bool> PyToolkit::getBool(ObjectPtr py_object, const std::string &key)
     {
@@ -696,7 +710,14 @@ namespace db0::python
         return std::string(PyUnicode_AsUTF8(*py_value));
     }
 
-    bool PyToolkit::compare(ObjectPtr py_object1, ObjectPtr py_object2) {
+    bool PyToolkit::hasKey(ObjectPtr py_object, const std::string &key)
+    {
+        auto py_value = Py_OWN(getValue(py_object, key));
+        return py_value.get() != nullptr;
+    }
+
+    bool PyToolkit::compare(ObjectPtr py_object1, ObjectPtr py_object2)
+    {
         auto result = PyObject_RichCompareBool(py_object1, py_object2, Py_EQ);
         if (result < 0) {
             // comparison failed

src/dbzero/bindings/python/PyToolkit.hpp

@@ -234,8 +234,11 @@ namespace db0::python
         // Extract keys (if present) from a Python dict object
         static std::optional<long> getLong(ObjectPtr py_object, const std::string &key);
         static std::optional<unsigned long long> getUnsignedLongLong(ObjectPtr py_object, const std::string &key);
+        static std::optional<unsigned int> getUnsignedInt(ObjectPtr py_object, const std::string &key);
         static std::optional<bool> getBool(ObjectPtr py_object, const std::string &key);
         static std::optional<std::string> getString(ObjectPtr py_object, const std::string &key);
+        // Check if key exists in a Python dict object
+        static bool hasKey(ObjectPtr py_object, const std::string &key);
 
         // Blocks until lock acquired
         static SafeRLock lockApi();

src/dbzero/bindings/python/dbzero.cpp

@@ -114,8 +114,7 @@ static PyMethodDef dbzero_methods[] =
     {"dbg_write_bytes", &py::writeBytes, METH_VARARGS, "Debug function"},
     {"dbg_free_bytes", &py::freeBytes, METH_VARARGS, "Debug function"},
     {"dbg_read_bytes", &py::readBytes, METH_VARARGS, "Debug function"},
-    {"dbg_start_logs", &py::PyAPI_startDebugLogs, METH_VARARGS, "Enable dbzeo debug logs"},
-    {"dbg_crash_from_commit", (PyCFunction)&py::PyAPI_crashFromCommit, METH_FASTCALL, "The function activates abrupt process terminate from storage::commit after specific number of DP writes (for testing purposes)"},
+    {"dbg_start_logs", &py::PyAPI_startDebugLogs, METH_VARARGS, "Enable dbzeo debug logs"},    
     {"get_base_lock_usage", &py::getResourceLockUsage, METH_VARARGS, "Debug function, retrieves total memory occupied by ResourceLocks"},
     {"get_dram_io_map", (PyCFunction)&py::getDRAM_IOMap, METH_VARARGS | METH_KEYWORDS, "Get page_num -> state_num mapping related with a specific DRAM_Prefix"},    
     {"breakpoint", (PyCFunction)&py::PyAPI_breakpoint, METH_FASTCALL, "Testing & debugging function "},

src/dbzero/core/memory/config.cpp

@@ -12,6 +12,8 @@ namespace db0
     bool Settings::__dbg_logs = false;
     bool Settings::__storage_validation = false;
     unsigned long long Settings::__sleep_interval = 0;
+    unsigned int Settings::__write_poison = 0;        
+    unsigned int Settings::__dram_io_flush_poison = 0;
 #endif 
 
     std::function<void()> Settings::m_decode_error = []() {
@@ -24,6 +26,8 @@ namespace db0
         __dbg_logs = false;
         __storage_validation = false;
         __sleep_interval = 0;
+        __write_poison = 0;
+        __dram_io_flush_poison = 0;
 #endif
     }
 

src/dbzero/core/memory/config.hpp

@@ -29,6 +29,10 @@ namespace db0
         static bool __storage_validation;
         // sleep interval for time-sensitive tests (e.g. copy_prefix) in milliseconds
         static unsigned long long __sleep_interval;
+        // the number of allowed writes before std::abort (or 0 = disabled)
+        static unsigned int __write_poison;
+        // the number of allowed DRAM_IO flush operations before std::abort (or 0 = disabled)
+        static unsigned int __dram_io_flush_poison;
 #endif
         // Function to throw the data decoding error (i.e. corrupt data detected)
         static std::function<void()> m_decode_error;

src/dbzero/core/memory/utils.cpp

@@ -48,5 +48,14 @@ namespace db0
         }
         std::cout << std::endl;
     }
+
+    void checkPoisonedOp(unsigned int &counter)
+    {
+        if (counter > 0) {            
+            if (--counter == 0) {
+                std::abort();
+            }
+        }
+    }
 
 }

src/dbzero/core/memory/utils.hpp

@@ -88,4 +88,7 @@ namespace db0
         return {};
     }
 
+    // std::abort or decrement the counter (nothing if counter is 0)
+    void checkPoisonedOp(unsigned int &counter);
+
 }