"""Reference solution — Memory-Efficient Object Pool. DO NOT READ BEFORE YOU HAVE RUN THE PROBLEM UNDER THE CLOCK. Teaching text: ../../../WARMUP.md § Chapter 3 and ../../python-internals/WARMUP.md § 5. Three things this problem is actually about. 1. __slots__ removes the per-instance __dict__, replacing hash-table attribute storage with fixed offsets in the object struct. The trap is that a SUBCLASS which omits __slots__ regains a __dict__ and most of the saving evaporates. Every class in the hierarchy must declare it. And you MEASURE it -- key sharing dictionaries (PEP 412) already narrowed the gap, so "we added __slots__" without a number is not an answer. 2. memoryview exposes the buffer protocol: slicing it allocates nothing, while slicing bytes copies. One bytearray carved into blocks means acquiring a buffer is free. Note that a live memoryview PINS the bytearray -- you cannot resize it while any view exists, which is why release() must drop the view. 3. weakref.finalize for leak detection, NOT __del__. A strong reference would make every leak undetectable by definition. And __del__ is the wrong tool regardless: finalization order within a cycle is undefined, exceptions in it are swallowed, and the object can resurrect itself. """ from __future__ import annotations import threading import time import tracemalloc import weakref class PoolExhausted(Exception): """No object was available within the timeout.""" # --------------------------------------------------------------------------- # Gate 2 — the slotted payload # --------------------------------------------------------------------------- class Pooled: """A slotted object the pool hands out. No __dict__, by construction. '__weakref__' must be in the slots. Declaring __slots__ removes weak- reference support along with the __dict__, and gate 4's leak detection is built on weakref.finalize -- so without it the whole design fails with "cannot create weak reference". That is the second half of the __slots__ trap and it is easy to miss until something downstream needs a weakref. """ __slots__ = ("payload", "generation", "__weakref__") def __init__(self, payload=None): self.payload = payload self.generation = 0 def reset(self): self.payload = None self.generation += 1 class PooledSubclassBroken(Pooled): """Omits __slots__ — and therefore regains a __dict__. The trap.""" class PooledSubclassFixed(Pooled): __slots__ = () def measure_footprint(cls, n=100_000, touch=False): """Bytes to hold n instances alive. Measured, never claimed. `touch` stores a NON-SLOT attribute on each instance. That matters more than it looks: since CPython 3.11 the instance dict is MANAGED and created lazily, so a subclass that omits __slots__ regains the ability to hold a __dict__ while costing nothing until something is actually put in it. sys.getsizeof reports the same number for both, and the trap only shows up under `touch`. Measured on 3.13, 100k instances: 7.2 MB slotted versus 41.6 MB once the dict is materialised — a 5.8x blowup that is invisible until it isn't. The reliable tell is therefore hasattr(instance, '__dict__'), not the size. """ tracemalloc.start() try: held = [] for _ in range(n): obj = cls() if touch: try: obj.extra_attribute = 1 except AttributeError: pass # properly slotted: refuses it held.append(obj) current, _peak = tracemalloc.get_traced_memory() finally: tracemalloc.stop() del held return current # --------------------------------------------------------------------------- # Gate 1, 4 — the object pool # --------------------------------------------------------------------------- class _Lease: """Context manager so release is guaranteed even if the body raises.""" __slots__ = ("_pool", "obj") def __init__(self, pool, obj): self._pool, self.obj = pool, obj def __enter__(self): return self.obj def __exit__(self, *exc): self._pool.release(self.obj) return False class Pool: """Bounded object pool with blocking acquire and weakref leak detection.""" def __init__(self, factory, size, clock=time.monotonic): if size <= 0: raise ValueError("size must be positive") self._factory = factory self.size = size self._clock = clock self._free = [] self._created = 0 self._lock = threading.Condition() # Weak references ONLY: a strong reference here would keep every # handed-out object alive, which makes leaks undetectable by design. self._out = weakref.WeakSet() self._finalizers = {} self._leaked = 0 self._sites = {} # ---- properties ------------------------------------------------------ @property def available(self): with self._lock: return len(self._free) + (self.size - self._created) @property def in_use(self): with self._lock: return self._created - len(self._free) # ---- acquire / release ---------------------------------------------- def acquire(self, timeout=None, tag=None): deadline = None if timeout is None else self._clock() + timeout with self._lock: while True: if self._free: obj = self._free.pop() break if self._created < self.size: obj = self._factory() self._created += 1 break remaining = None if deadline is None else deadline - self._clock() if remaining is not None and remaining <= 0: raise PoolExhausted( f"no object available within {timeout}s") if not self._lock.wait(remaining): raise PoolExhausted( f"no object available within {timeout}s") self._out.add(obj) key = id(obj) self._sites[key] = tag # If the caller drops its last reference without releasing, this # fires and records the leak — WITHOUT keeping the object alive. self._finalizers[key] = weakref.finalize(obj, self._on_lost, key) return _Lease(self, obj) def _on_lost(self, key): self._leaked += 1 self._sites.pop(key, None) self._finalizers.pop(key, None) def release(self, obj): with self._lock: key = id(obj) finalizer = self._finalizers.pop(key, None) if finalizer is not None: finalizer.detach() # a clean return is not a leak self._sites.pop(key, None) self._out.discard(obj) if hasattr(obj, "reset"): obj.reset() self._free.append(obj) self._lock.notify() # ---- leak reporting -------------------------------------------------- def leaked(self): return self._leaked def checkout_sites(self): with self._lock: return dict(self._sites) # --------------------------------------------------------------------------- # Gate 3 — the buffer pool # --------------------------------------------------------------------------- class _BufferLease: __slots__ = ("_pool", "_index", "view") def __init__(self, pool, index, view): self._pool, self._index, self.view = pool, index, view def __enter__(self): return self.view def __exit__(self, *exc): self._pool.release(self._index, self) return False class BufferPool: """One bytearray carved into fixed blocks, handed out as memoryviews.""" def __init__(self, block_size, count): if block_size <= 0 or count <= 0: raise ValueError("block_size and count must be positive") self.block_size = block_size self.count = count self.backing = bytearray(block_size * count) # allocated ONCE self._master = memoryview(self.backing) self._free = list(range(count)) self._lock = threading.Condition() @property def available(self): with self._lock: return len(self._free) def acquire(self, timeout=None): with self._lock: if not self._free and not self._lock.wait(timeout): raise PoolExhausted("no buffer available") if not self._free: raise PoolExhausted("no buffer available") index = self._free.pop() start = index * self.block_size # A SLICE of a memoryview allocates nothing — writes through it mutate # the backing bytearray directly. Slicing bytes would copy. return _BufferLease(self, index, self._master[start:start + self.block_size]) def release(self, index, lease=None): if lease is not None and lease.view is not None: # Release the view so the backing store can later be resized; a # live memoryview pins the bytearray and raises BufferError. lease.view.release() lease.view = None with self._lock: self._free.append(index) self._lock.notify() def close(self): self._master.release()