diff --git a/benchmarks/test_wal.py b/benchmarks/test_wal.py new file mode 100644 index 0000000..9742738 --- /dev/null +++ b/benchmarks/test_wal.py @@ -0,0 +1,84 @@ +"""Benchmarks for write-ahead log persistence.""" + +from pathlib import Path + +import pytest + +from seriousdb.wal import SetEntry, WriteAheadLog + +from ._support import WARMUP_ROUNDS + + +@pytest.mark.benchmark(group="wal") +def test_wal_append( + benchmark, + database_file: Path, + measured_rounds: int, +) -> None: + """Measure appending one entry to the WAL, including fsync.""" + + entry = SetEntry(key="key", value="value") + + def prepare_wal(): + database_file.unlink(missing_ok=True) + wal = WriteAheadLog(str(database_file)) + return (wal,), {} + + def append_entry(wal: WriteAheadLog) -> None: + wal.append(entry) + + def close_wal(wal: WriteAheadLog) -> None: + wal.close() + + benchmark.extra_info.update( + operation="WAL append", + persistence="write + flush + fsync", + key_bytes=len(entry.key.encode()), + value_bytes=len(entry.value.encode()), + ) + + benchmark.pedantic( + append_entry, + setup=prepare_wal, + teardown=close_wal, + rounds=50, + warmup_rounds=WARMUP_ROUNDS, + ) + + +@pytest.mark.benchmark(group="wal") +def test_wal_append_many( + benchmark, + database_file: Path, +) -> None: + """Measure repeated appends using one open WAL.""" + + entry = SetEntry(key="key", value="value") + + def prepare_wal(): + database_file.unlink(missing_ok=True) + wal = WriteAheadLog(str(database_file)) + return (wal,), {} + + def append_many(wal: WriteAheadLog) -> None: + for _ in range(1000): + wal.append(entry) + + def close_wal(wal: WriteAheadLog) -> None: + wal.close() + + benchmark.extra_info.update( + operation="1000 WAL appends", + persistence="write + flush + fsync", + entries=1000, + key_bytes=len(entry.key.encode()), + value_bytes=len(entry.value.encode()), + ) + + benchmark.pedantic( + append_many, + setup=prepare_wal, + teardown=close_wal, + rounds=10, + warmup_rounds=WARMUP_ROUNDS, + ) diff --git a/src/seriousdb/cache.py b/src/seriousdb/cache.py index 9c1d9c6..4a93a90 100644 --- a/src/seriousdb/cache.py +++ b/src/seriousdb/cache.py @@ -320,15 +320,23 @@ def load(self, filename: str) -> None: db = _write_default(filename) wal = WriteAheadLog(f"{filename}.wal") - replayed = wal.replay() - - for entry in replayed: - entry.apply(db) - + try: + replayed = wal.replay() + for entry in replayed: + entry.apply(db) + except Exception: + try: + wal.close() + except OSError: + pass + raise + old_wal = self.wal self.db = db self.filename = filename self.wal = wal self._writes_since_compact = len(replayed) + if old_wal is not None: + old_wal.close() def flush(self) -> None: """No-op, kept for backward compatibility. diff --git a/src/seriousdb/wal.py b/src/seriousdb/wal.py index 3acd9df..e832013 100644 --- a/src/seriousdb/wal.py +++ b/src/seriousdb/wal.py @@ -102,6 +102,19 @@ class WriteAheadLog: def __init__(self, filename: str): self.filename = filename self._offset = os.path.getsize(filename) if os.path.isfile(filename) else 0 + self._file = open(filename, "a+b") # noqa: SIM115 - kept open, closed in close() + + def close(self) -> None: + """Close the WAL file if it is open.""" + f, self._file = self._file, None + if f is not None: + f.close() + + def _ensure_open(self): + """Return the open WAL file, reopening it if necessary.""" + if self._file is None: + self._file = open(self.filename, "a+b") # noqa: SIM115 - reopened lazily, closed in close() + return self._file def append(self, entry: WalEntry) -> None: """Append `entry` to the log and fsync it. @@ -117,12 +130,24 @@ def append(self, entry: WalEntry) -> None: If the log file cannot be written. """ self._repair_torn_tail() - with open(self.filename, "ab") as f: - data = (json.dumps(entry.to_dict()) + "\n").encode() + f = self._ensure_open() + + data = (json.dumps(entry.to_dict()) + "\n").encode() + + try: f.write(data) f.flush() os.fsync(f.fileno()) - self._offset += len(data) + + self._offset += len(data) + except Exception: + try: + f.close() + except OSError: + pass + finally: + self._file = None + raise def replay(self) -> list[WalEntry]: """Return every entry durably written to the log, oldest first. @@ -145,40 +170,43 @@ def replay(self) -> list[WalEntry]: self._offset = 0 return [] + f = self._ensure_open() + entries: list[WalEntry] = [] good_offset = 0 found_bad_entry = False - with open(self.filename, "rb") as f: - for raw_line in f: - if not raw_line.endswith(b"\n"): + + f.seek(0) + + for raw_line in f: + if not raw_line.endswith(b"\n"): + found_bad_entry = True + break + line = raw_line.strip() + if line: + try: + entries.append(WalEntry.from_dict(json.loads(line.decode()))) + except ( + json.JSONDecodeError, + UnicodeDecodeError, + TypeError, + ValueError, + ) as e: + logger.warning( + "Corrupt entry in write-ahead log %s (%s)", + self.filename, + e, + ) found_bad_entry = True break - line = raw_line.strip() - if line: - try: - entries.append(WalEntry.from_dict(json.loads(line.decode()))) - except ( - json.JSONDecodeError, - UnicodeDecodeError, - TypeError, - ValueError, - ) as e: - logger.warning( - "Corrupt entry in write-ahead log %s (%s)", - self.filename, - e, - ) - found_bad_entry = True - break - good_offset += len(raw_line) + good_offset += len(raw_line) if found_bad_entry: logger.warning( "Truncating write-ahead log %s to its last known-good entry", self.filename, ) - with open(self.filename, "r+b") as f: - f.truncate(good_offset) + f.truncate(good_offset) self._offset = good_offset return entries @@ -194,12 +222,19 @@ def clear(self) -> None: dir_name = os.path.dirname(self.filename) or "." with tempfile.NamedTemporaryFile("wb", dir=dir_name, delete=False) as tmp_file: pass + try: + self.close() os.replace(tmp_file.name, self.filename) except OSError: - os.unlink(tmp_file.name) + try: + os.unlink(tmp_file.name) + except FileNotFoundError: + pass raise + self._offset = 0 + self._file = None def _repair_torn_tail(self) -> None: """Truncate any bytes left by a previous failed write.