| 31 | # |
| 32 | |
| 33 | class Queue(object): |
| 34 | |
| 35 | def __init__(self, maxsize=0, *, ctx): |
| 36 | if maxsize <= 0: |
| 37 | # Can raise ImportError (see issues #3770 and #23400) |
| 38 | from .synchronize import SEM_VALUE_MAX as maxsize |
| 39 | self._maxsize = maxsize |
| 40 | self._reader, self._writer = connection.Pipe(duplex=False) |
| 41 | self._rlock = ctx.Lock() |
| 42 | self._opid = os.getpid() |
| 43 | if sys.platform == 'win32': |
| 44 | self._wlock = None |
| 45 | else: |
| 46 | self._wlock = ctx.Lock() |
| 47 | self._sem = ctx.BoundedSemaphore(maxsize) |
| 48 | # For use by concurrent.futures |
| 49 | self._ignore_epipe = False |
| 50 | self._reset() |
| 51 | |
| 52 | if sys.platform != 'win32': |
| 53 | register_after_fork(self, Queue._after_fork) |
| 54 | |
| 55 | def __getstate__(self): |
| 56 | context.assert_spawning(self) |
| 57 | return (self._ignore_epipe, self._maxsize, self._reader, self._writer, |
| 58 | self._rlock, self._wlock, self._sem, self._opid) |
| 59 | |
| 60 | def __setstate__(self, state): |
| 61 | (self._ignore_epipe, self._maxsize, self._reader, self._writer, |
| 62 | self._rlock, self._wlock, self._sem, self._opid) = state |
| 63 | self._reset() |
| 64 | |
| 65 | def _after_fork(self): |
| 66 | debug('Queue._after_fork()') |
| 67 | self._reset(after_fork=True) |
| 68 | |
| 69 | def _reset(self, after_fork=False): |
| 70 | if after_fork: |
| 71 | self._notempty._at_fork_reinit() |
| 72 | else: |
| 73 | self._notempty = threading.Condition(threading.Lock()) |
| 74 | self._buffer = collections.deque() |
| 75 | self._thread = None |
| 76 | self._jointhread = None |
| 77 | self._joincancelled = False |
| 78 | self._closed = False |
| 79 | self._close = None |
| 80 | self._send_bytes = self._writer.send_bytes |
| 81 | self._recv_bytes = self._reader.recv_bytes |
| 82 | self._poll = self._reader.poll |
| 83 | |
| 84 | def put(self, obj, block=True, timeout=None): |
| 85 | if self._closed: |
| 86 | raise ValueError(f"Queue {self!r} is closed") |
| 87 | if not self._sem.acquire(block, timeout): |
| 88 | raise Full |
| 89 | |
| 90 | with self._notempty: |