Approach: use a fixed-size ring buffer protected by a lock for structural updates plus atomic counters for processed/dropped. offer() is non-blocking: if full, apply policy (drop_oldest, drop_new, reject) and update dropped counter atomically. Use conditionless semantics so producers never block; consumers poll with optional blocking pop() using a Condition.python
import threading
from collections import deque
from time import monotonic
class AtomicCounter:
def __init__(self):
self._v = 0
self._lock = threading.Lock()
def inc(self, n=1):
with self._lock:
self._v += n
def get(self):
with self._lock:
return self._v
class BoundedBuffer:
def __init__(self, capacity, drop_policy='drop_new'):
self.capacity = capacity
self.buf = deque()
self.lock = threading.Lock() # protects buf and size
self.not_empty = threading.Condition(self.lock)
self.dropped = AtomicCounter()
self.processed = AtomicCounter()
self.drop_policy = drop_policy # 'drop_new' | 'drop_oldest' | 'reject'
def offer(self, item):
# non-blocking offer
with self.lock:
if len(self.buf) < self.capacity:
self.buf.append(item)
self.not_empty.notify()
return True
# full -> policy handling
if self.drop_policy == 'drop_oldest':
self.buf.popleft()
self.buf.append(item)
self.dropped.inc()
self.not_empty.notify()
return True
if self.drop_policy == 'drop_new':
self.dropped.inc()
return False
if self.drop_policy == 'reject':
self.dropped.inc()
raise BufferError("buffer full")
return False
def pop(self, block=True, timeout=None):
with self.not_empty:
if not block:
if not self.buf:
return None
item = self.buf.popleft()
else:
end = None if timeout is None else monotonic() + timeout
while not self.buf:
remaining = None if end is None else end - monotonic()
if remaining is not None and remaining <= 0:
return None
self.not_empty.wait(remaining)
item = self.buf.popleft()
self.processed.inc()
return item
# inspection primitives
def size(self):
with self.lock: return len(self.buf)
def is_full(self):
with self.lock: return len(self.buf) >= self.capacity
def stats(self):
return {'size': self.size(), 'capacity': self.capacity,
'dropped': self.dropped.get(), 'processed': self.processed.get()}
