012 — CPython dbm 3.14.5：兩種都叫做「支援併發」的保證

{
  "name": "012-cpython-dbm",
  "upstream": "CPython Lib/dbm/",
  "examined_version": "3.14.5",
  "license": "PSF-2.0",
  "what_it_is": "The interchangeable key-value backends behind shelve, and what each of them does when two writers overlap.",

  "why_this_one": "Example 011 named concurrency as the first thing that would break it, and yesterday's archaeology measured shelve without ever asking what was underneath. This is what was underneath.",

  "the_finding": "Lost updates and index corruption are two different guarantees, and both get called 'safe for concurrent use'. Under the same written-down schedule, dbm.dumb and dbm.sqlite3 BOTH lose an update; only dbm.dumb corrupts. Real locking buys integrity and buys nothing at all on lost updates, because a lost update happens between two operations that are each perfectly atomic.",

  "the_second_finding": "The module that declares its gap is the one with the smaller gap on the axis anyone thinks about. dbm.dumb says 'support concurrent access (currently, if two processes take turns making updates, they can mess up the index)' - in a TO DO list, inside a docstring. dbm.sqlite3, which holds index integrity, says nothing about concurrency at all.",

  "the_scale_this_completes": "Three consecutive entries, one organisation's standard library, three places a mode or a gap is written down: archaeology 010 put the .pyc invalidation mode in HEADER FLAG BITS, machine-readable by anything; archaeology 011 put shelve's writeback mode in the open() CALL, readable only by whoever opened the file; archaeology 012 puts dbm.dumb's concurrency gap in a TO DO ITEM IN A DOCSTRING, readable by whoever reads the source, and dbm.sqlite3 does not write it anywhere.",

  "sets": {
    "FMS": "this file: the guarantee vocabulary, the two failures, and the scale the three entries make together",
    "SCL": "which schedule this deployment runs, and whether a declaration must match behaviour",
    "SMS": "the guarantee/requirement tables, schedule resolution, and the upstream probes",
    "TMS": "one file per schedule - each declares which failures it is capable of revealing, and reaches no sibling set",
    "DMS": "the two failures side by side, where each module declares its gap, and what none of it can see"
  },

  "non_goals": [
    "Saying dbm is broken. Every backend here does what it says; the problem is that dbm.open() gives the caller no way to ask what that is.",
    "Racing. The schedule is written down, so a lost update here is a fact about the code rather than about this machine's timing.",
    "Claiming anything about backends this machine does not have. dbm.gnu and dbm.ndbm are unavailable here and are printed as unavailable rather than assumed."
  ]
}

{
  "schedule": "interleaved",
  "declarations_must_match_behaviour": true,
  "_note": "main.py runs every schedule against every available backend regardless of what this says."
}

"""Which schedule this deployment runs, and what it refuses to let pass."""
import json
import pathlib

_C = json.loads((pathlib.Path(__file__).parent / "policy.json").read_text(encoding="utf-8"))

schedule = lambda: _C["schedule"]                                                    # noqa: E731
declarations_must_match_behaviour = lambda: bool(_C["declarations_must_match_behaviour"])  # noqa: E731

"""The re-cut: a backend declares what it guarantees, and a schedule declares
what it can reveal. Both declarations are checked by running them.

Upstream neither exists. `dbm.open` picks whatever is available and the caller
gets an interface with no way to ask what it promises.
"""
import importlib

SCHEDULES = ["interleaved", "one_at_a_time"]

# What each backend guarantees, as measured by main.py rather than as read from
# its documentation. A backend absent from this table is not assumed safe.
GUARANTEES = {
    "dbm.dumb": [],
    "dbm.sqlite3": ["index-integrity"],
    "dbm.gnu": ["index-integrity"],
    "dbm.ndbm": [],
}

REQUIREMENTS = {
    "read-modify-write": ["serialised-transaction"],
    "write-distinct-keys": ["index-integrity"],
}


def load_schedules():
    loaded, problems = {}, []
    for module_name in SCHEDULES:
        try:
            module = importlib.import_module(f"TMS.schedules.{module_name}")
        except ModuleNotFoundError:
            problems.append(f'schedule "{module_name}" has no module - fail closed')
            continue
        for attribute in ("NAME", "REVEALS", "order"):
            if not hasattr(module, attribute):
                problems.append(f"{module_name} does not declare {attribute}")
        loaded[module.NAME] = module
    return loaded, problems


def resolve(name, loaded):
    module = loaded.get(name)
    if module is None:
        return None, f'schedule "{name}" has no implementation - fail closed (known: {", ".join(sorted(loaded))})'
    return module, None


def unmet(operation, backend):
    """What the operation needs that the backend does not claim to give."""
    return [need for need in REQUIREMENTS[operation] if need not in GUARANTEES.get(backend, [])]

"""What the real dbm backends do under a written-down schedule.

Two failures are measured separately because they are separate guarantees that
both get called "safe for concurrent use":

  lost-update — two increments from 0 end at 1; the data is intact and wrong
  torn-index  — after interleaved writes of distinct keys, keys are missing
"""
import inspect
import os
import tempfile

BACKENDS = {}
for _name in ("dbm.dumb", "dbm.sqlite3", "dbm.gnu", "dbm.ndbm"):
    try:
        BACKENDS[_name] = __import__(_name, fromlist=["open"])
    except ImportError:
        pass


def _fresh():
    return os.path.join(tempfile.mkdtemp(), "store")


def lost_update(module, schedule_order):
    """Read-modify-write through two handles, following the given order.

    The order is a list of writer indices; each writer's steps are read, modify,
    write. Nothing is raced for, so the result is a fact about the code.
    """
    path = _fresh()
    with module.open(path, "c") as db:
        db[b"n"] = b"0"
    try:
        handles = [module.open(path, "w"), module.open(path, "w")]
        held = [None, None]
        cursor = [0, 0]
        steps = [
            lambda who: held.__setitem__(who, int(handles[who][b"n"])),
            lambda who: held.__setitem__(who, held[who] + 1),
            lambda who: handles[who].__setitem__(b"n", str(held[who]).encode()),
        ]
        for who in schedule_order:
            if cursor[who] < len(steps):
                steps[cursor[who]](who)
                cursor[who] += 1
        for who, handle in enumerate(handles):
            while cursor[who] < len(steps):
                steps[cursor[who]](who)
                cursor[who] += 1
            handle.close()
        with module.open(path, "r") as db:
            return {"final": int(db[b"n"]), "lost": 2 - int(db[b"n"]), "error": None}
    except Exception as error:                       # noqa: BLE001
        return {"final": None, "lost": None, "error": f"{type(error).__name__}: {error}"}


def torn_index(module, rounds=40):
    """Interleave writes of DIFFERENT keys through two handles, then count."""
    path = _fresh()
    with module.open(path, "c") as db:
        db[b"seed"] = b"1"
    try:
        first, second = module.open(path, "w"), module.open(path, "w")
        for index in range(rounds):
            (first if index % 2 == 0 else second)[f"k{index}".encode()] = b"v"
        first.close()
        second.close()
        with module.open(path, "r") as db:
            survived = len(db.keys())
        return {"survived": survived, "expected": rounds + 1,
                "missing": rounds + 1 - survived, "error": None}
    except Exception as error:                       # noqa: BLE001
        return {"survived": None, "expected": rounds + 1, "missing": None,
                "error": f"{type(error).__name__}: {error}"}


def declaration(module):
    """Where, if anywhere, the module says something about concurrent access."""
    source = inspect.getsource(module)
    doc = module.__doc__ or ""
    mentions = [line.strip() for line in doc.splitlines() if "concurrent" in line.lower()]
    return {
        "lines": len(source.splitlines()),
        "says_in_docstring": mentions[0] if mentions else None,
        "locking_primitives": [token for token in ("flock", "msvcrt", "LOCK_EX", "fcntl")
                               if token in source],
    }


def which_backend_shelve_uses():
    import dbm
    import shelve
    path = _fresh()
    with shelve.open(path) as shelf:
        shelf["k"] = 1
    return dbm.whichdb(path)

"""Two handles, taking turns. Written down, not raced for.

A schedule is a unit here because it is the thing that decides which failures
are visible at all. This one declares what it is capable of revealing; the
island test checks that declaration by running it.
"""
NAME = "interleaved"
REVEALS = ["lost-update", "torn-index"]


def order(writers, steps_each):
    return [who for _ in range(steps_each) for who in range(writers)]

"""One writer finishes before the next starts.

This is what a test suite with a single writer produces, and it reveals nothing
about concurrency - which is the finding rather than a shortcoming of the file.
An empty REVEALS is a claim, and the island test makes it fail if it is wrong.
"""
NAME = "one-at-a-time"
REVEALS = []


def order(writers, steps_each):
    return [who for who in range(writers) for _ in range(steps_each)]

"""The two failures side by side, and what none of it can see."""


def table(rows, out):
    out(f"\n  {'backend':<14} {'schedule':<15} {'two +1 from 0':<15} "
        f"{'lost':<6} {'40 distinct keys':<18} missing")
    for row in rows:
        lost = "-" if row["lost"] in (0, None) else str(row["lost"])
        survived = (f"{row['survived']} of {row['expected']}"
                    if row["survived"] is not None else "n/a")
        missing = "-" if not row["missing"] else str(row["missing"])
        out(f"  {row['backend']:<14} {row['schedule']:<15} {str(row['final']):<15} "
            f"{lost:<6} {survived:<18} {missing}")


def declarations(rows, out):
    out("\n  where each module says something about concurrent access:")
    for backend, info in rows.items():
        where = info["says_in_docstring"] or "nowhere"
        out(f"    {backend:<14} {info['lines']:>4} lines   {where[:64]}")
        if info["locking_primitives"]:
            out(f"    {'':<14} uses {', '.join(info['locking_primitives'])}")


def gaps(out):
    out("\n  measurable, not measured here:")
    out("    - how the backends behave under a real scheduler rather than this one")
    out("    - what the sqlite backend's locking costs in throughput")
    out("\n  not measurable by this entry at all:")
    out("    - whether any of this has ever bitten anyone. Reproducing a lost")
    out("      update says nothing about how often two writers meet.")
    out("    - the backends this machine does not have. dbm.gnu and dbm.ndbm are")
    out("      absent here, so every statement about them is a guess and is")
    out("      printed as unavailable rather than assumed.")

"""The island test, and the drill that lets the schedule's own claim fail.

    python src/island_test.py

Section 3 is the finding: a backend can hold index integrity and lose an update
under the same schedule, because a lost update happens BETWEEN two operations
that are each perfectly atomic.
"""
import pathlib
import re
import sys

HERE = pathlib.Path(__file__).parent
sys.path.insert(0, str(HERE))

from SCL import policy  # noqa: E402
from SMS import backends, upstream  # noqa: E402

FAILURES = []


def check(label, ok, detail=""):
    print(f"  {'PASS' if ok else 'FAIL'}  {label}{' - ' + detail if detail else ''}")
    if not ok:
        FAILURES.append(label)


SCHEDULES, PROBLEMS = backends.load_schedules()

print("\n== 1. every schedule is an island and declares what it can reveal")
check("both schedules loaded with no problems", not PROBLEMS, "; ".join(PROBLEMS))
schedule_dir = HERE / "TMS" / "schedules"
files = sorted(f for f in schedule_dir.iterdir() if f.suffix == ".py" and f.name != "__init__.py")
check("there are two schedule files", len(files) == 2, ", ".join(f.stem for f in files))
for f in files:
    reaches = re.findall(r"^\s*(?:from|import)\s+(\S+)", f.read_text(encoding="utf-8"), re.M)
    siblings = [r for r in reaches if r.split(".")[0] in {"TMS", "SMS", "SCL", "DMS", "FMS"}]
    check(f"{f.stem} reaches no sibling set", not siblings,
          ", ".join(reaches) or "no imports at all")
for name, module in sorted(SCHEDULES.items()):
    check(f"{name} declares what it reveals", isinstance(module.REVEALS, list), str(module.REVEALS))

print("\n== 2. that declaration is checked by running it")


def reveals_lost_update(order):
    return any(upstream.lost_update(module, order)["lost"] for module in upstream.BACKENDS.values())


for name, module in sorted(SCHEDULES.items()):
    measured = reveals_lost_update(module.order(2, 3))
    declared = "lost-update" in module.REVEALS
    check(f"{name}: declared lost-update={declared}, measured {measured}", measured == declared,
          "on the backends this machine has")

print("\n== 2b. the drill: a schedule that overclaims must be caught")


class Overclaiming:
    NAME = "sequential-but-claims-otherwise"
    REVEALS = ["lost-update"]

    @staticmethod
    def order(writers, steps_each):
        return [who for who in range(writers) for _ in range(steps_each)]


check("a sequential schedule declaring lost-update is caught",
      reveals_lost_update(Overclaiming.order(2, 3)) is False
      and "lost-update" in Overclaiming.REVEALS,
      "declared it reveals lost updates, revealed none")

print("\n== 3. two guarantees, and one backend holds exactly one of them")
interleaved = SCHEDULES["interleaved"].order(2, 3)
rows = {}
for name, module in sorted(upstream.BACKENDS.items()):
    rows[name] = {"lost": upstream.lost_update(module, interleaved)["lost"],
                  "missing": upstream.torn_index(module)["missing"]}
    print(f"        {name:<14} lost updates {rows[name]['lost']}   "
          f"keys missing {rows[name]['missing']}")
check("every available backend loses an update under this schedule",
      all(row["lost"] for row in rows.values()),
      ", ".join(f"{n}={r['lost']}" for n, r in rows.items()))
integrity = {name: row["missing"] == 0 for name, row in rows.items()}
check("and they do NOT agree on index integrity", len(set(integrity.values())) > 1,
      ", ".join(f"{n}={'holds' if ok else 'loses'}" for n, ok in integrity.items()))
check("so locking is real and does not address lost updates",
      rows.get("dbm.sqlite3", {}).get("missing") == 0
      and bool(rows.get("dbm.sqlite3", {}).get("lost")),
      "dbm.sqlite3 keeps every key and still ends at 1")

print("\n== 4. THE CONTROL: the sequential schedule reveals nothing, on any backend")
sequential = SCHEDULES["one-at-a-time"].order(2, 3)
finals = {name: upstream.lost_update(module, sequential)["final"]
          for name, module in sorted(upstream.BACKENDS.items())}
check("every backend ends at 2 under one-at-a-time", all(v == 2 for v in finals.values()),
      ", ".join(f"{n}={v}" for n, v in finals.items()))
check("so what the suite can produce, not what it asserts, decides what is visible",
      all(v == 2 for v in finals.values()) and all(row["lost"] for row in rows.values()))

print("\n== 5. where each module writes down its gap")
for name, module in sorted(upstream.BACKENDS.items()):
    info = upstream.declaration(module)
    print(f"        {name:<14} {info['lines']:>4} lines   "
          f"{'docstring TO DO' if info['says_in_docstring'] else 'nowhere'}   "
          f"locking: {', '.join(info['locking_primitives']) or 'none in this source'}")
declared = {name: bool(upstream.declaration(module)["says_in_docstring"])
            for name, module in upstream.BACKENDS.items()}
check("the backend that declares its gap is the one with the larger gap",
      declared.get("dbm.dumb") and not declared.get("dbm.sqlite3"),
      "dbm.dumb says it in a TO DO; dbm.sqlite3 says nothing and holds integrity")

print("\n== 6. fail closed")
_, problem = backends.resolve("random-order", SCHEDULES)
check("an unresolvable schedule stops the run", problem is not None, problem or "resolved anyway")
check("SCL names a schedule that exists", policy.schedule() in SCHEDULES, policy.schedule())
absent = [name for name in backends.GUARANTEES if name not in upstream.BACKENDS]
check("backends this machine lacks are reported, not assumed", bool(absent),
      f"untested here: {', '.join(sorted(absent))}")

print("\n== 7. what this entry cannot see")
print("        MEASURABLE, NOT MEASURED")
print("          - behaviour under a real scheduler rather than this written one")
print("          - what the sqlite backend's locking costs in throughput")
print("        NOT MEASURABLE HERE")
print("          - frequency. Reproducing a lost update says nothing about how")
print("            often two writers actually meet.")
print("          - the absent backends. Every statement about dbm.gnu and")
print("            dbm.ndbm would be a guess, so none is made.")

print(f"\n{len(FAILURES)} failure(s)" if FAILURES else "\nall checks passed")
for f in FAILURES:
    print(f"  - {f}")
sys.exit(1 if FAILURES else 0)

"""CPython's dbm backends under a written-down schedule.

    python src/main.py            the two failures, side by side, per backend
    python src/main.py --strict   exit 1 if a declaration disagrees with behaviour
"""
import json
import pathlib
import sys

HERE = pathlib.Path(__file__).parent
sys.path.insert(0, str(HERE))

from DMS import report  # noqa: E402
from SCL import policy  # noqa: E402
from SMS import backends, upstream  # noqa: E402

ARCH = json.loads((HERE / "FMS" / "architecture.json").read_text(encoding="utf-8"))


def main(argv):
    out = lambda line="": sys.stdout.write(line + "\n")  # noqa: E731
    schedules, problems = backends.load_schedules()
    if problems:
        for problem in problems:
            out(f"  !! {problem}")
        return 1

    out(f"\n== two failures that both get called 'concurrency'  "
        f"[python {sys.version.split()[0]}]")
    out(f"   shelve on this machine picks: {upstream.which_backend_shelve_uses()}")
    out(f"   backends available here:      {', '.join(sorted(upstream.BACKENDS)) or 'none'}")
    missing = [name for name in backends.GUARANTEES if name not in upstream.BACKENDS]
    if missing:
        out(f"   unavailable, so untested:     {', '.join(sorted(missing))}")

    rows = []
    for backend_name, module in sorted(upstream.BACKENDS.items()):
        for schedule_name, schedule in sorted(schedules.items()):
            lost = upstream.lost_update(module, schedule.order(2, 3))
            torn = upstream.torn_index(module) if schedule_name == "interleaved" else {
                "survived": None, "expected": None, "missing": None, "error": None}
            rows.append({"backend": backend_name, "schedule": schedule_name, **lost, **torn})
    report.table(rows, out)

    out("\n  The two columns are two different guarantees. A backend can hold the")
    out("  right-hand one and lose the left-hand one, and both are described as")
    out("  being safe for concurrent use.")

    report.declarations({name: upstream.declaration(module)
                         for name, module in sorted(upstream.BACKENDS.items())}, out)

    out("\n== what each schedule can reveal, checked by running it")
    wrong = []
    for schedule_name, schedule in sorted(schedules.items()):
        revealed = set()
        for module in upstream.BACKENDS.values():
            if upstream.lost_update(module, schedule.order(2, 3))["lost"]:
                revealed.add("lost-update")
        claimed = set(schedule.REVEALS) & {"lost-update"}
        agrees = revealed == claimed
        if not agrees:
            wrong.append(f"{schedule_name}: declares {sorted(claimed)}, revealed {sorted(revealed)}")
        out(f"    {'ok ' if agrees else 'NO '} {schedule_name:<15} declares {schedule.REVEALS}, "
            f"revealed {sorted(revealed) or 'nothing'}")

    out("\n== the requirement comparison the caller never gets to make")
    for operation in sorted(backends.REQUIREMENTS):
        for backend_name in sorted(upstream.BACKENDS):
            gap = backends.unmet(operation, backend_name)
            out(f"    {operation:<20} on {backend_name:<14} "
                f"{'unmet: ' + ', '.join(gap) if gap else 'satisfied'}")
    out("\n    dbm.open() gives no way to ask any of this. The table above is the")
    out("    re-cut's addition, and every row of it was measured, not read.")

    report.gaps(out)

    if "--strict" in argv and wrong and policy.declarations_must_match_behaviour():
        for line in wrong:
            out(f"\n  !! {line}")
        return 1
    return 0


if __name__ == "__main__":
    sys.exit(main(sys.argv[1:]))