/usr/lib/python2.7/dist-packages/zmq/sugar/tracker.py is in python-zmq 14.0.1-1build2.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 | """Tracker for zero-copy messages with 0MQ."""
#-----------------------------------------------------------------------------
# Copyright (C) 2013 Brian Granger, Min Ragan-Kelley
#
# This file is part of pyzmq
#
# Distributed under the terms of the New BSD License. The full license is in
# the file COPYING.BSD, distributed as part of this software.
#-----------------------------------------------------------------------------
import time
try:
# below 3.3
from threading import _Event as Event
except (ImportError, AttributeError):
# python throws ImportError, cython throws AttributeError
from threading import Event
from zmq.error import NotDone
from zmq.backend import Frame
class MessageTracker(object):
"""MessageTracker(*towatch)
A class for tracking if 0MQ is done using one or more messages.
When you send a 0MQ message, it is not sent immediately. The 0MQ IO thread
sends the message at some later time. Often you want to know when 0MQ has
actually sent the message though. This is complicated by the fact that
a single 0MQ message can be sent multiple times using different sockets.
This class allows you to track all of the 0MQ usages of a message.
Parameters
----------
*towatch : tuple of Event, MessageTracker, Message instances.
This list of objects to track. This class can track the low-level
Events used by the Message class, other MessageTrackers or
actual Messages.
"""
events = None
peers = None
def __init__(self, *towatch):
"""MessageTracker(*towatch)
Create a message tracker to track a set of mesages.
Parameters
----------
*towatch : tuple of Event, MessageTracker, Message instances.
This list of objects to track. This class can track the low-level
Events used by the Message class, other MessageTrackers or
actual Messages.
"""
self.events = set()
self.peers = set()
for obj in towatch:
if isinstance(obj, Event):
self.events.add(obj)
elif isinstance(obj, MessageTracker):
self.peers.add(obj)
elif isinstance(obj, Frame):
if not obj.tracker:
raise ValueError("Not a tracked message")
self.peers.add(obj.tracker)
else:
raise TypeError("Require Events or Message Frames, not %s"%type(obj))
@property
def done(self):
"""Is 0MQ completely done with the message(s) being tracked?"""
for evt in self.events:
if not evt.is_set():
return False
for pm in self.peers:
if not pm.done:
return False
return True
def wait(self, timeout=-1):
"""mt.wait(timeout=-1)
Wait for 0MQ to be done with the message or until `timeout`.
Parameters
----------
timeout : float [default: -1, wait forever]
Maximum time in (s) to wait before raising NotDone.
Returns
-------
None
if done before `timeout`
Raises
------
NotDone
if `timeout` reached before I am done.
"""
tic = time.time()
if timeout is False or timeout < 0:
remaining = 3600*24*7 # a week
else:
remaining = timeout
done = False
for evt in self.events:
if remaining < 0:
raise NotDone
evt.wait(timeout=remaining)
if not evt.is_set():
raise NotDone
toc = time.time()
remaining -= (toc-tic)
tic = toc
for peer in self.peers:
if remaining < 0:
raise NotDone
peer.wait(timeout=remaining)
toc = time.time()
remaining -= (toc-tic)
tic = toc
__all__ = ['MessageTracker']
|