A wrapper which defers writes on the given database to a background thread. Calls to :meth:`~hypothesis.database.ExampleDatabase.fetch` wait for any enqueued writes to finish before fetching from the database.
| 1164 | |
| 1165 | |
| 1166 | class BackgroundWriteDatabase(ExampleDatabase): |
| 1167 | """A wrapper which defers writes on the given database to a background thread. |
| 1168 | |
| 1169 | Calls to :meth:`~hypothesis.database.ExampleDatabase.fetch` wait for any |
| 1170 | enqueued writes to finish before fetching from the database. |
| 1171 | """ |
| 1172 | |
| 1173 | def __init__(self, db: ExampleDatabase) -> None: |
| 1174 | super().__init__() |
| 1175 | self._db = db |
| 1176 | self._queue: Queue[tuple[str, tuple[bytes, ...]]] = Queue() |
| 1177 | self._thread: Thread | None = None |
| 1178 | |
| 1179 | def _ensure_thread(self): |
| 1180 | if self._thread is None: |
| 1181 | self._thread = Thread(target=self._worker, daemon=True) |
| 1182 | self._thread.start() |
| 1183 | # avoid an unbounded timeout during gc. 0.1 should be plenty for most |
| 1184 | # use cases. |
| 1185 | weakref.finalize(self, self._join, 0.1) |
| 1186 | |
| 1187 | def __repr__(self) -> str: |
| 1188 | return f"BackgroundWriteDatabase({self._db!r})" |
| 1189 | |
| 1190 | def __eq__(self, other: object) -> bool: |
| 1191 | return isinstance(other, BackgroundWriteDatabase) and self._db == other._db |
| 1192 | |
| 1193 | def _worker(self) -> None: |
| 1194 | while True: |
| 1195 | method, args = self._queue.get() |
| 1196 | getattr(self._db, method)(*args) |
| 1197 | self._queue.task_done() |
| 1198 | |
| 1199 | def _join(self, timeout: float | None = None) -> None: |
| 1200 | # copy of Queue.join with a timeout. https://bugs.python.org/issue9634 |
| 1201 | with self._queue.all_tasks_done: |
| 1202 | while self._queue.unfinished_tasks: |
| 1203 | self._queue.all_tasks_done.wait(timeout) |
| 1204 | |
| 1205 | def fetch(self, key: bytes) -> Iterable[bytes]: |
| 1206 | self._join() |
| 1207 | return self._db.fetch(key) |
| 1208 | |
| 1209 | def save(self, key: bytes, value: bytes) -> None: |
| 1210 | self._ensure_thread() |
| 1211 | self._queue.put(("save", (key, value))) |
| 1212 | |
| 1213 | def delete(self, key: bytes, value: bytes) -> None: |
| 1214 | self._ensure_thread() |
| 1215 | self._queue.put(("delete", (key, value))) |
| 1216 | |
| 1217 | def move(self, src: bytes, dest: bytes, value: bytes) -> None: |
| 1218 | self._ensure_thread() |
| 1219 | self._queue.put(("move", (src, dest, value))) |
| 1220 | |
| 1221 | def _start_listening(self) -> None: |
| 1222 | self._db.add_listener(self._broadcast_change) |
| 1223 |
no outgoing calls