admin/revng-revng
1
0
Fork 0
mirror of https://github.com/revng/revng synced 2026-06-21 14:07:57 +00:00
Code Issues Packages Projects Releases Wiki Activity

pypeline.storage: fix invalidation

Browse Source 
Fix the invalidation logic in both `InMemoryStorageProvider` and
`Sqlite3StorageProvider`, as it did not account for propagating the
invalidation to both children objects and parent objects.
This commit is contained in:
Giacomo Vercesi
2025-10-03 13:18:25 +02:00
parent d92f4cdca3
commit 5ea2cbdeaf
5 changed files with 172 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/revng/pypeline/object.py
+11
View File
@@ -252,6 +252,17 @@ class ObjectID(ABC):
"""Convert a bytes representation back to an ObjectID"""
raise NotImplementedError()
def is_related(self, to: ObjectID) -> bool:
def test(source: ObjectID, target: ObjectID):
test_object: ObjectID | None = target
while test_object is not None:
if test_object == source:
return True
test_object = test_object.parent()
return False
return test(self, to) or test(to, self)
def __eq__(self, other) -> bool:
return hash(self) == hash(other)
python/revng/pypeline/storage/memory.py
+44 -21
View File
@@ -10,16 +10,16 @@ from datetime import datetime
from typing import Iterable, Mapping
from revng.pypeline.container import ContainerID
from revng.pypeline.model import ModelPath, ModelPathSet
from revng.pypeline.model import ModelPathSet
from revng.pypeline.object import ObjectID
from revng.pypeline.task.task import ObjectDependencies
from .storage_provider import ConfigurationId, ContainerLocation, ProjectMetadata, SavepointID
from .storage_provider import SavePointsRange, StorageProvider
from .util import _REVNG_VERSION_PLACEHOLDER
from .util import _REVNG_VERSION_PLACEHOLDER, check_kind_structure
@dataclass
@dataclass(frozen=True)
class DependencyEntry:
savepoint_start: SavepointID
savepoint_end: SavepointID
@@ -34,6 +34,7 @@ class InMemoryStorageProvider(StorageProvider):
"""
def __init__(self):
check_kind_structure()
self.model = b""
self.storage: dict[ContainerLocation, dict[ObjectID, bytes]] = {}
self.dependencies: dict[str, list[DependencyEntry]] = defaultdict(list)
@@ -87,27 +88,49 @@ class InMemoryStorageProvider(StorageProvider):
self.storage[location][key] = value
self.last_change = datetime.now()
def _invalidate(self, path: ModelPath) -> None:
for key, entries in self.dependencies.items():
if key != path:
continue
def invalidate(self, invalidation_list: ModelPathSet):
invalidated: dict[ContainerLocation, set[ObjectID]] = defaultdict(set)
# TODO: this double loop is very inefficient
for entry in entries:
for container_loc, objects in self.storage.items():
if (
container_loc.savepoint_id < entry.savepoint_start
or container_loc.savepoint_id > entry.savepoint_end
or container_loc.container_id != entry.container_id
or container_loc.configuration_id != entry.configuration_id
):
continue
if entry.object_id in objects:
del objects[entry.object_id]
# Set of entries that will be collected from self.dependencies
object_to_delete: set[DependencyEntry] = set()
# Keys that exist on self.dependencies that will be deleted
paths_to_delete: list[str] = []
def invalidate(self, invalidation_list: ModelPathSet) -> None:
# Retrieve entries from self.dependencies that match the provided paths
for path in invalidation_list:
self._invalidate(path)
if path in self.dependencies:
paths_to_delete.append(path)
object_to_delete.update(self.dependencies[path])
# For each DependencyEntry, find the matching entries in self.storage
# TODO: this double loop is very inefficient
for entry in object_to_delete:
for container_loc, objects in self.storage.items():
# For an entry to match, the savepoint must be in the savepoint
# range and the configuration_id must match
if (
container_loc.savepoint_id < entry.savepoint_start
or container_loc.savepoint_id > entry.savepoint_end
or container_loc.configuration_id != entry.configuration_id
):
continue
# Check that the object is related (the same object, a parent
# or a child) to the invalidated object
for object_ in objects:
if entry.object_id.is_related(object_):
invalidated[container_loc].add(object_)
# Actually delete the data
for container_loc, objects_to_delete in invalidated.items():
self.storage[container_loc] = {
k: v for k, v in self.storage[container_loc].items() if k not in objects_to_delete
}
# Delete the paths that were previously found
for path in paths_to_delete:
del self.dependencies[path]
self.last_change = datetime.now()
def get_model(self) -> bytes:
python/revng/pypeline/storage/sqlite3.py
+60 -17
View File
@@ -7,7 +7,7 @@ from __future__ import annotations
import sqlite3
from datetime import datetime, timezone
from pathlib import Path
from typing import Iterable, Mapping
from typing import Collection, Mapping
from revng.pypeline.container import ConfigurationId
from revng.pypeline.model import ModelPathSet
@@ -16,7 +16,14 @@ from revng.pypeline.task.task import ObjectDependencies
from revng.pypeline.utils.registry import get_singleton
from .storage_provider import ContainerLocation, ProjectMetadata, SavePointsRange, StorageProvider
from .util import _REVNG_VERSION_PLACEHOLDER
from .util import _OBJECTID_MAXSIZE, _REVNG_VERSION_PLACEHOLDER, check_kind_structure
from .util import check_object_id_supported_by_sql
# This is a binary mask that will be used for invalidation, thanks to the
# binary structure of ObjectID, all children are guaranteed to have the parent
# prefixed, so, to check for all children the check will be
# target_object_id <= checked_object_id <= CONCAT(target_object_id, _OBJECTID_MASK) # noqa: E800
_OBJECTID_MASK = f"x'{"ff" * _OBJECTID_MAXSIZE}'"
CREATE_TABLES = """
CREATE TABLE IF NOT EXISTS project(
@@ -29,7 +36,7 @@ CREATE TABLE IF NOT EXISTS objects(
savepoint_id INT NOT NULL,
container_id TEXT NOT NULL,
configuration_hash TEXT NOT NULL,
object_id TEXT NOT NULL,
object_id BLOB NOT NULL,
content BLOB NOT NULL,
PRIMARY KEY (savepoint_id, container_id, configuration_hash, object_id)
) STRICT;
@@ -44,7 +51,7 @@ CREATE TABLE IF NOT EXISTS dependencies(
savepoint_id_end INT NOT NULL,
container_id TEXT NOT NULL,
configuration_hash TEXT NOT NULL,
object_id TEXT NOT NULL,
object_id BLOB NOT NULL,
model_path TEXT NOT NULL,
PRIMARY KEY (savepoint_id_start, savepoint_id_end, container_id,
configuration_hash, object_id, model_path)
@@ -74,6 +81,23 @@ REPLACE INTO objects(savepoint_id, container_id, configuration_hash, object_id,
VALUES (?, ?, ?, ?, ?)
"""
PUT_DEPENDENCIES_QUERY = "REPLACE INTO dependencies VALUES (?, ?, ?, ?, ?, ?)"
# Invalidation query, this retrieves the paths stored in `dependencies` that match.
# Given the records found, the following rules must all be met when selecting
# records from objects to be deleted:
# 1. Have the savepoint be in the savepoint range of the dependency
# 2. Have the configuration_hash match
# 3. Have the object_id be related to the dependency, this means that, either:
# 1. The objects.object_id is the same or a child of dependencies.object_id
# This is simplified by the structure of ObjectID, so this becomes a
# range comparison
# `dependencies.object_id <= objects.object_id <= (dependencies.object_id + mask)`
# 2. The object.object_id is a parent of dependencies.object_id. In the
# general case this would require generating the query dynamically. For
# the time being this is specialized for the current structure where
# there's only children of root, so the check becomes:
# `dependencies.object_id != x'' AND objects.object_id = x''`
INVALIDATE_QUERY = """
DELETE FROM objects
WHERE rowid IN (
@@ -81,7 +105,14 @@ WHERE rowid IN (
FROM objects
JOIN dependencies
WHERE dependencies.model_path IN ({model_paths})
AND objects.container_id = dependencies.container_id
AND (
(
objects.object_id >= dependencies.object_id
AND objects.object_id <= CAST((dependencies.object_id || {objectid_mask}) AS BLOB)
) OR (
dependencies.object_id != x'' AND objects.object_id = x''
)
)
AND objects.configuration_hash = dependencies.configuration_hash
AND objects.savepoint_id >= dependencies.savepoint_id_start
AND objects.savepoint_id <= dependencies.savepoint_id_end
@@ -116,6 +147,7 @@ class SQlite3StorageProvider(StorageProvider):
"""StorageProvider implementation with backing sqlite3 db"""
def __init__(self, db_path: str, model_path: str | Path):
check_kind_structure()
self._model_path = Path(model_path)
self._connection = sqlite3.connect(db_path, autocommit=False)
self._connection.commit()
@@ -137,11 +169,14 @@ class SQlite3StorageProvider(StorageProvider):
def has(
self,
location: ContainerLocation,
keys: Iterable[ObjectID],
) -> Iterable[ObjectID]:
keys: Collection[ObjectID],
) -> list[ObjectID]:
if len(keys) == 0:
return []
# NOTE: possible SQL injection, sqlite has a limit on parameters. If it
# can't be avoided chunk selects by 999 values
id_list = ",".join([f"'{key!s}'" for key in keys])
id_list = ",".join([f"x'{key.to_bytes().hex()}'" for key in keys])
with self._cursor() as cursor:
cursor.execute(
HAS_QUERY.format(id_list=id_list),
@@ -149,16 +184,19 @@ class SQlite3StorageProvider(StorageProvider):
)
result = cursor.fetchall()
obj_id_ty: type[ObjectID] = get_singleton(ObjectID) # type: ignore[type-abstract]
return [obj_id_ty.deserialize(x[0]) for x in result]
return [obj_id_ty.from_bytes(x[0]) for x in result]
def get(
self,
location: ContainerLocation,
keys: Iterable[ObjectID],
) -> Mapping[ObjectID, bytes]:
keys: Collection[ObjectID],
) -> dict[ObjectID, bytes]:
if len(keys) == 0:
return {}
# NOTE: possible SQL injection, sqlite has a limit on parameters. If it
# can't be avoided chunk selects by 999 values
id_list = ",".join([f"'{key!s}'" for key in keys])
id_list = ",".join([f"x'{key.to_bytes().hex()}'" for key in keys])
with self._cursor() as cursor:
cursor.execute(
GET_QUERY.format(id_list=id_list),
@@ -166,7 +204,7 @@ class SQlite3StorageProvider(StorageProvider):
)
result = cursor.fetchall()
obj_id_ty: type[ObjectID] = get_singleton(ObjectID) # type: ignore[type-abstract]
return {obj_id_ty.deserialize(x[0]): x[1] for x in result}
return {obj_id_ty.from_bytes(x[0]): x[1] for x in result}
def add_dependencies(
self,
@@ -176,14 +214,15 @@ class SQlite3StorageProvider(StorageProvider):
) -> None:
with self._cursor() as cursor:
for container_id, object_id, model_path in deps:
check_object_id_supported_by_sql(object_id)
cursor.execute(
"REPLACE INTO dependencies VALUES (?, ?, ?, ?, ?, ?)",
PUT_DEPENDENCIES_QUERY,
(
savepoint_range.start,
savepoint_range.end,
container_id,
configuration_id,
object_id.serialize(),
object_id.to_bytes(),
model_path,
),
)
@@ -202,17 +241,21 @@ class SQlite3StorageProvider(StorageProvider):
location.savepoint_id,
location.container_id,
location.configuration_id,
object_id.serialize(),
object_id.to_bytes(),
content,
),
)
self._write_metadata(cursor)
def invalidate(self, invalidation_list: ModelPathSet) -> None:
if len(invalidation_list) == 0:
return
with self._cursor() as cursor:
cursor.executescript(
INVALIDATE_QUERY.format(
model_paths=",".join(f"'{path}'" for path in invalidation_list)
model_paths=",".join(f"'{path}'" for path in invalidation_list),
objectid_mask=_OBJECTID_MASK,
)
)
self._write_metadata(cursor)
python/revng/pypeline/storage/storage_provider.py
+3 -3
View File
@@ -7,7 +7,7 @@ from __future__ import annotations
from abc import ABC, abstractmethod
from dataclasses import dataclass
from datetime import datetime
from typing import Annotated, Iterable, Mapping
from typing import Annotated, Collection, Iterable, Mapping
from revng.pypeline.container import ConfigurationId, ContainerID
from revng.pypeline.model import ModelPathSet
@@ -80,7 +80,7 @@ class StorageProvider(ABC):
def has(
self,
location: ContainerLocation,
keys: Iterable[ObjectID],
keys: Collection[ObjectID],
) -> Iterable[ObjectID]:
"""
Get the available objects from the storage.
@@ -91,7 +91,7 @@ class StorageProvider(ABC):
def get(
self,
location: ContainerLocation,
keys: Iterable[ObjectID],
keys: Collection[ObjectID],
) -> Mapping[ObjectID, bytes]:
"""
For each objects, return bytes that the container can ingest to
python/revng/pypeline/storage/util.py
+54
View File
@@ -2,4 +2,58 @@
# This file is distributed under the MIT License. See LICENSE.md for details.
#
import graphlib
from revng.pypeline.object import Kind, ObjectID
from revng.pypeline.utils.registry import get_singleton
__all__ = ("_REVNG_VERSION_PLACEHOLDER", "_OBJECTID_MAXSIZE", "check_kind_structure")
_REVNG_VERSION_PLACEHOLDER = "1.0.0"
# Number of bytes that the biggest ObjectID will contain when serialized to bytes
# Currently fixed to 17 because 1 byte for kind + 16 bytes for MetaAddress
_OBJECTID_MAXSIZE = 17
# Check that the kind structure is smaller than the `_OBJECTID_MAXSIZE`. This
# is a hack to avoid having storage providers be more flexible about the
# structure of Kinds. Currently the storage providers can assume that the revng
# kind structure is the one they are operating on and make stronger assumptions
# and optimizations. If needed this check can be lifted but it would require
# extra work and extra flexibility.
def check_kind_structure():
kind_type = get_singleton(Kind)
kind_root = kind_type.root()
kinds = kind_type.kinds()
# Check that we can store all Kinds in a byte
assert len(kinds) <= 2**8
# Sort kinds
sorter = graphlib.TopologicalSorter()
for kind in kinds:
if (parent := kind.parent()) is not None:
sorter.add(kind, parent)
else:
sorter.add(kind)
# Compute the serialized kind size
kind_sizes: dict[Kind, int] = {}
for kind in sorter.static_order():
if kind == kind_root:
kind_sizes[kind] = 0
else:
kind_sizes[kind] = kind_sizes[kind.parent()] + 1 + kind.byte_size()
# Check that the maximum kind size does not exceed the expected maximum size
assert max(kind_sizes.values()) <= _OBJECTID_MAXSIZE
# Generally SQL-like implementations require, in the general case, to
# dynamically generate the query to invalidate objects from paths. To simplify
# things we rely on the fact that currently objects have a maximum depth of
# 2. This function will be checked by SQL-like StorageProviders when adding
# object dependencies via `add_dependencies`.
def check_object_id_supported_by_sql(object_id: ObjectID):
if object_id.kind().rank() >= 2:
raise NotImplementedError("Objects with rank >= 2 are not supported")