clocks.py 4.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152
  1. """Logical Clocks and Synchronization."""
  2. from __future__ import absolute_import, unicode_literals
  3. from threading import Lock
  4. from itertools import islice
  5. from operator import itemgetter
  6. from .five import python_2_unicode_compatible, zip
  7. __all__ = ('LamportClock', 'timetuple')
  8. R_CLOCK = '_lamport(clock={0}, timestamp={1}, id={2} {3!r})'
  9. @python_2_unicode_compatible
  10. class timetuple(tuple):
  11. """Tuple of event clock information.
  12. Can be used as part of a heap to keep events ordered.
  13. Arguments:
  14. clock (int): Event clock value.
  15. timestamp (float): Event UNIX timestamp value.
  16. id (str): Event host id (e.g. ``hostname:pid``).
  17. obj (Any): Optional obj to associate with this event.
  18. """
  19. __slots__ = ()
  20. def __new__(cls, clock, timestamp, id, obj=None):
  21. return tuple.__new__(cls, (clock, timestamp, id, obj))
  22. def __repr__(self):
  23. return R_CLOCK.format(*self)
  24. def __getnewargs__(self):
  25. return tuple(self)
  26. def __lt__(self, other):
  27. # 0: clock 1: timestamp 3: process id
  28. try:
  29. A, B = self[0], other[0]
  30. # uses logical clock value first
  31. if A and B: # use logical clock if available
  32. if A == B: # equal clocks use lower process id
  33. return self[2] < other[2]
  34. return A < B
  35. return self[1] < other[1] # ... or use timestamp
  36. except IndexError:
  37. return NotImplemented
  38. def __gt__(self, other):
  39. return other < self
  40. def __le__(self, other):
  41. return not other < self
  42. def __ge__(self, other):
  43. return not self < other
  44. clock = property(itemgetter(0))
  45. timestamp = property(itemgetter(1))
  46. id = property(itemgetter(2))
  47. obj = property(itemgetter(3))
  48. @python_2_unicode_compatible
  49. class LamportClock(object):
  50. """Lamport's logical clock.
  51. From Wikipedia:
  52. A Lamport logical clock is a monotonically incrementing software counter
  53. maintained in each process. It follows some simple rules:
  54. * A process increments its counter before each event in that process;
  55. * When a process sends a message, it includes its counter value with
  56. the message;
  57. * On receiving a message, the receiver process sets its counter to be
  58. greater than the maximum of its own value and the received value
  59. before it considers the message received.
  60. Conceptually, this logical clock can be thought of as a clock that only
  61. has meaning in relation to messages moving between processes. When a
  62. process receives a message, it resynchronizes its logical clock with
  63. the sender.
  64. See Also:
  65. * `Lamport timestamps`_
  66. * `Lamports distributed mutex`_
  67. .. _`Lamport Timestamps`: https://en.wikipedia.org/wiki/Lamport_timestamps
  68. .. _`Lamports distributed mutex`: https://bit.ly/p99ybE
  69. *Usage*
  70. When sending a message use :meth:`forward` to increment the clock,
  71. when receiving a message use :meth:`adjust` to sync with
  72. the time stamp of the incoming message.
  73. """
  74. #: The clocks current value.
  75. value = 0
  76. def __init__(self, initial_value=0, Lock=Lock):
  77. self.value = initial_value
  78. self.mutex = Lock()
  79. def adjust(self, other):
  80. with self.mutex:
  81. value = self.value = max(self.value, other) + 1
  82. return value
  83. def forward(self):
  84. with self.mutex:
  85. self.value += 1
  86. return self.value
  87. def sort_heap(self, h):
  88. """Sort heap of events.
  89. List of tuples containing at least two elements, representing
  90. an event, where the first element is the event's scalar clock value,
  91. and the second element is the id of the process (usually
  92. ``"hostname:pid"``): ``sh([(clock, processid, ...?), (...)])``
  93. The list must already be sorted, which is why we refer to it as a
  94. heap.
  95. The tuple will not be unpacked, so more than two elements can be
  96. present.
  97. Will return the latest event.
  98. """
  99. if h[0][0] == h[1][0]:
  100. same = []
  101. for PN in zip(h, islice(h, 1, None)):
  102. if PN[0][0] != PN[1][0]:
  103. break # Prev and Next's clocks differ
  104. same.append(PN[0])
  105. # return first item sorted by process id
  106. return sorted(same, key=lambda event: event[1])[0]
  107. # clock values unique, return first item
  108. return h[0]
  109. def __str__(self):
  110. return str(self.value)
  111. def __repr__(self):
  112. return '<LamportClock: {0.value}>'.format(self)