Skip to content

proxystore.stream.shims.queue

proxystore/stream/shims/queue.py

Python queue-based pub/sub implementation.

Warning

This implementation is meant for streaming between Python threads within the same process, or between Python processes on the same machine. Each queue topic may only have one subscriber.

QueuePublisher 

QueuePublisher(
    queues: Mapping[str, Queue[bytes] | Queue[bytes]],
    *,
    block: bool = True,
    timeout: float | None = None
)

Publisher built on Python queues.

Warning

Each topic can only have one subscriber.

Parameters:

  • queues (Mapping[str, Queue[bytes] | Queue[bytes]]) –

    Mapping of topic name to Python queue.

  • block (bool, default: True ) –

    Block until a free slot is available when sending a new message to the queue.

  • timeout (float | None, default: None ) –

    Block at most timeout seconds.

Source code in proxystore/stream/shims/queue.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
def __init__(
    self,
    queues: Mapping[
        str,
        multiprocessing.Queue[bytes] | queue.Queue[bytes],
    ],
    *,
    block: bool = True,
    timeout: float | None = None,
) -> None:
    self._queues = queues
    self._block = block
    self._timeout = timeout

close() 

close() -> None

Close this publisher.

Source code in proxystore/stream/shims/queue.py 
50
51
52
53
54
def close(self) -> None:
    """Close this publisher."""
    for q in self._queues.values():
        if isinstance(q, multiprocessing.queues.Queue):
            q.close()

send() 

send(topic: str, message: bytes) -> None

Publish a message to the stream.

Parameters:

  • topic (str) –

    Stream topic to publish message to.

  • message (bytes) –

    Message as bytes to publish to the stream.

Raises:

  • ValueError –

    if a queue with the name topic does not exist.

Source code in proxystore/stream/shims/queue.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def send(self, topic: str, message: bytes) -> None:
    """Publish a message to the stream.

    Args:
        topic: Stream topic to publish message to.
        message: Message as bytes to publish to the stream.

    Raises:
        ValueError: if a queue with the name `topic` does not exist.
    """
    if topic not in self._queues:
        raise ValueError(f'Unknown topic "{topic}".')
    self._queues[topic].put(
        message,
        block=self._block,
        timeout=self._timeout,
    )

QueueSubscriber 

QueueSubscriber(
    queue: Queue[bytes] | Queue[bytes],
    *,
    block: bool = True,
    timeout: float | None = None
)

Subscriber to a QueuePublisher topic.

Warning

Each topic can only have one subscriber.

Parameters:

  • queue (Queue[bytes] | Queue[bytes]) –

    Queue shared with the QueuePublisher to pull messages from.

  • block (bool, default: True ) –

    Block until the next message is available in the queue.

  • timeout (float | None, default: None ) –

    Block at most timeout seconds.

Source code in proxystore/stream/shims/queue.py 
89
90
91
92
93
94
95
96
97
98
def __init__(
    self,
    queue: multiprocessing.Queue[bytes] | queue.Queue[bytes],
    *,
    block: bool = True,
    timeout: float | None = None,
) -> None:
    self._queue = queue
    self._block = block
    self._timeout = timeout

close() 

close() -> None

Close this subscriber.

Source code in proxystore/stream/shims/queue.py 
111
112
113
def close(self) -> None:
    """Close this subscriber."""
    pass