| 355 | # |
| 356 | |
| 357 | class SimpleQueue(object): |
| 358 | |
| 359 | def __init__(self, *, ctx): |
| 360 | self._reader, self._writer = connection.Pipe(duplex=False) |
| 361 | self._rlock = ctx.Lock() |
| 362 | self._poll = self._reader.poll |
| 363 | if sys.platform == 'win32': |
| 364 | self._wlock = None |
| 365 | else: |
| 366 | self._wlock = ctx.Lock() |
| 367 | |
| 368 | def close(self): |
| 369 | self._reader.close() |
| 370 | self._writer.close() |
| 371 | |
| 372 | def empty(self): |
| 373 | return not self._poll() |
| 374 | |
| 375 | def __getstate__(self): |
| 376 | context.assert_spawning(self) |
| 377 | return (self._reader, self._writer, self._rlock, self._wlock) |
| 378 | |
| 379 | def __setstate__(self, state): |
| 380 | (self._reader, self._writer, self._rlock, self._wlock) = state |
| 381 | self._poll = self._reader.poll |
| 382 | |
| 383 | def get(self): |
| 384 | with self._rlock: |
| 385 | res = self._reader.recv_bytes() |
| 386 | # unserialize the data after having released the lock |
| 387 | return _ForkingPickler.loads(res) |
| 388 | |
| 389 | def put(self, obj): |
| 390 | # serialize the data before acquiring the lock |
| 391 | obj = _ForkingPickler.dumps(obj) |
| 392 | if self._wlock is None: |
| 393 | # writes to a message oriented win32 pipe are atomic |
| 394 | self._writer.send_bytes(obj) |
| 395 | else: |
| 396 | with self._wlock: |
| 397 | self._writer.send_bytes(obj) |
| 398 | |
| 399 | __class_getitem__ = classmethod(types.GenericAlias) |
no outgoing calls
no test coverage detected
searching dependent graphs…