distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Thread-safe hash set.
2
3Ported from net.i2p.util.ConcurrentHashSet.
4"""
5
6from __future__ import annotations
7
8import threading
9from typing import Iterator
10
11
12class ConcurrentHashSet:
13 """Thread-safe set wrapper."""
14
15 def __init__(self) -> None:
16 self._lock = threading.Lock()
17 self._data: set = set()
18
19 def add(self, item) -> None:
20 with self._lock:
21 self._data.add(item)
22
23 def remove(self, item) -> None:
24 with self._lock:
25 self._data.remove(item)
26
27 def discard(self, item) -> None:
28 with self._lock:
29 self._data.discard(item)
30
31 def __contains__(self, item) -> bool:
32 with self._lock:
33 return item in self._data
34
35 def __len__(self) -> int:
36 with self._lock:
37 return len(self._data)
38
39 def __iter__(self) -> Iterator:
40 with self._lock:
41 return iter(list(self._data))