source: geniuswebcore/geniusweb/simplerunner/BlockingQueue.py@ 100

Last change on this file since 100 was 100, checked in by ruud, 19 months ago

python installs also wheel to avoid error messages

File size: 858 bytes
Line 
1from collections import deque
2import threading
3from typing import Generic, TypeVar
4
5
6T=TypeVar('T') # contained element type
7
8class BlockingQueue(Generic[T]):
9 '''
10 Like threading.Queue but using collections.queue
11 which allows us to extend the functioanlity
12 '''
13 def __init__(self, capacity: int):
14 self._pushing = threading.Semaphore(capacity)
15 self._pulling = threading.Semaphore(0)
16 self._data:deque = deque()
17
18 def put(self, element: T):
19 self._pushing.acquire()
20 self._data.append(element)
21 self._pulling.release()
22
23 def take(self) -> T:
24 self._pulling.acquire()
25 self._pushing.release()
26 return self._data.popleft()
27
28 def size(self) -> int:
29 return len(self._data)
30
31 def contains(self, elt:T) -> bool:
32 return elt in self._data
33
Note: See TracBrowser for help on using the repository browser.