/ - Diff - snf-ganeti - Greek Research and Technology Network's projects

Revision 34cb5617

       return wrap
     class _SingleActionPipeConditionWaiter(object):
       """Callable helper class for _SingleActionPipeCondition.
     class _SingleNotifyPipeConditionWaiter(object):
       """Helper class for SingleNotifyPipeCondition
       """
       __slots__ = [
         "_cond",
         "_fd",
         "_poller",
+        ]
       def __init__(self, cond, poller, fd):
         """Initializes this class.
       def __init__(self, poller, fd):
         """Constructor for _SingleNotifyPipeConditionWaiter
         @type cond: L{_SingleActionPipeCondition}
         @param cond: Parent condition
         @type poller: select.poll
         @param poller: Poller object
         @type fd: int
-...
         """
         object.__init__(self)
         self._cond = cond
         self._poller = poller
         self._fd = fd
-...
           raise RuntimeError("cannot work with un-aquired lock")
     class _SingleActionPipeCondition(object):
       """Wrapper around a pipe for usage inside conditions.
       This class contains a POSIX pipe(2) and a poller to poll it. The pipe is
       always allocated when constructing the class. Extra care is taken to always
       close the file descriptors.
       An additional class, L{_SingleActionPipeConditionWaiter}, is used to wait for
       notifications.
     class SingleNotifyPipeCondition(_BaseCondition):
       """Condition which can only be notified once.
       Warning: This class is designed to be used as the underlying component of a
       locking condition, but is not by itself thread safe, and needs to be
       protected by an external lock.
       This condition class uses pipes and poll, internally, to be able to wait for
       notification with a timeout, without resorting to polling. It is almost
       compatible with Python's threading.Condition, with the following differences:
         - notifyAll can only be called once, and no wait can happen after that
         - notify is not supported, only notifyAll
       """
       __slots__ = [
       __slots__ = _BaseCondition.__slots__ + [
         "_poller",
         "_read_fd",
         "_write_fd",
         "_nwaiters",
         "_notified",
+        ]
       _waiter_class = _SingleActionPipeConditionWaiter
       _waiter_class = _SingleNotifyPipeConditionWaiter
       def __init__(self):
         """Initializes this class.
       def __init__(self, lock):
         """Constructor for SingleNotifyPipeCondition
         """
         object.__init__(self)
         _BaseCondition.__init__(self, lock)
         self._nwaiters = 0
         self._notified = False
         self._read_fd = None
         self._write_fd = None
         self._poller = None
         # Just assume the unpacking is successful, otherwise error handling gets
         # very complicated.
         (self._read_fd, self._write_fd) = os.pipe()
         try:
           # The poller looks for closure of the write side
           poller = select.poll()
           poller.register(self._read_fd, select.POLLHUP)
           self._poller = poller
         except:
           if self._read_fd is not None:
             os.close(self._read_fd)
           if self._write_fd is not None:
             os.close(self._write_fd)
           raise
         # There should be no code here anymore, otherwise the pipe file descriptors
         # may be not be cleaned up properly in case of errors.
       def StartWaiting(self):
         """Return function to wait for notification.
       def _check_unnotified(self):
         if self._notified:
           raise RuntimeError("cannot use already notified condition")
         @rtype: L{_SingleActionPipeConditionWaiter}
         @return: Function to wait for notification
       def _Cleanup(self):
         """Cleanup open file descriptors, if any.
         """
         assert self._nwaiters >= 0
         if self._poller is None:
           raise RuntimeError("Already cleaned up")
         # Create waiter function and increase number of waiters
         wait_fn = self._waiter_class(self, self._poller, self._read_fd)
         self._nwaiters += 1
         return wait_fn
         if self._read_fd is not None:
           os.close(self._read_fd)
           self._read_fd = None
       def DoneWaiting(self):
         """Decrement number of waiters and automatic cleanup.
         if self._write_fd is not None:
           os.close(self._write_fd)
           self._write_fd = None
         self._poller = None
         Must be called after waiting for a notification.
       def wait(self, timeout=None):
         """Wait for a notification.
         @rtype: bool
         @return: Whether this was the last waiter
         @type timeout: float or None
         @param timeout: Waiting timeout (can be None)
         """
         assert self._nwaiters > 0
         self._nwaiters -= 1
         self._check_owned()
         self._check_unnotified()
         if self._nwaiters == 0:
           self._Cleanup()
           return True
         self._nwaiters += 1
         try:
           if self._poller is None:
             (self._read_fd, self._write_fd) = os.pipe()
             self._poller = select.poll()
             self._poller.register(self._read_fd, select.POLLHUP)
         return False
           wait_fn = self._waiter_class(self._poller, self._read_fd)
           self.release()
           try:
             # Wait for notification
             wait_fn(timeout)
           finally:
             # Re-acquire lock
             self.acquire()
         finally:
           self._nwaiters -= 1
           if self._nwaiters == 0:
             self._Cleanup()
       def notifyAll(self):
         """Close the writing side of the pipe to notify all waiters.
         """
         if self._write_fd is None:
           raise RuntimeError("Can only notify once")
         os.close(self._write_fd)
         self._write_fd = None
       def _Cleanup(self):
         """Close all file descriptors.
         """
         if self._read_fd is not None:
           os.close(self._read_fd)
           self._read_fd = None
         self._check_owned()
         self._check_unnotified()
         self._notified = True
         if self._write_fd is not None:
           os.close(self._write_fd)
           self._write_fd = None
         self._poller = None
       def __del__(self):
         """Called on object deletion.
         Ensure no file descriptors are left open.
         """
         self._Cleanup()
     class _PipeCondition(_BaseCondition):
     class PipeCondition(_BaseCondition):
       """Group-only non-polling condition with counters.
       This condition class uses pipes and poll, internally, to be able to wait for
-...
       """
       __slots__ = _BaseCondition.__slots__ + [
         "_nwaiters",
         "_pipe",
         "_single_condition",
+        ]
       _pipe_class = _SingleActionPipeCondition
       _single_condition_class = SingleNotifyPipeCondition
       def __init__(self, lock):
         """Initializes this class.
-...
         """
         _BaseCondition.__init__(self, lock)
         self._nwaiters = 0
         self._pipe = None
         self._single_condition = self._single_condition_class(self._lock)
       def wait(self, timeout=None):
         """Wait for a notification.
-...
         """
         self._check_owned()
         if not self._pipe:
           self._pipe = self._pipe_class()
         # Keep local reference to the pipe. It could be replaced by another thread
         # notifying while we're waiting.
         pipe = self._pipe
         my_condition = self._single_condition
         assert self._nwaiters >= 0
         self._nwaiters += 1
         try:
           # Get function to wait on the pipe
           wait_fn = pipe.StartWaiting()
           try:
             # Release lock while waiting
             self.release()
             try:
               # Wait for notification
               wait_fn(timeout)
             finally:
               # Re-acquire lock
               self.acquire()
           finally:
             # Destroy pipe if this was the last waiter and the current pipe is
             # still the same. The same pipe cannot be reused after cleanup.
             if pipe.DoneWaiting() and pipe == self._pipe:
               self._pipe = None
           my_condition.wait(timeout)
         finally:
           assert self._nwaiters > 0
           self._nwaiters -= 1
-...
         """
         self._check_owned()
         # Notify and forget pipe. A new one will be created on the next call to
         # wait.
         if self._pipe is not None:
           self._pipe.notifyAll()
           self._pipe = None
         self._single_condition.notifyAll()
         self._single_condition = self._single_condition_class(self._lock)
       def has_waiting(self):
         """Returns whether there are active waiters.
-...
         """Waits for the condition to be notified.
         @type timeout: float or None
         @param timeout: Timeout in seconds
         @param timeout: Waiting timeout (can be None)
         """
         assert self._nwaiters >= 0
-...
         "__shr",
+        ]
       __condition_class = _PipeCondition
       __condition_class = PipeCondition
       def __init__(self):
         """Construct a new SharedLock.

         self.assert_(not self.cond._is_owned())
     class TestSingleNotifyPipeCondition(_ConditionTestCase):
       """SingleNotifyPipeCondition tests"""
       def setUp(self):
         _ConditionTestCase.setUp(self, locking.SingleNotifyPipeCondition)
       def testAcquireRelease(self):
         self._testAcquireRelease()
       def testNotification(self):
         self._testNotification()
       def testWaitReuse(self):
         self.cond.acquire()
         self.cond.wait(0)
         self.cond.wait(0.1)
         self.cond.release()
       def testNoNotifyReuse(self):
         self.cond.acquire()
         self.cond.notifyAll()
         self.assertRaises(RuntimeError, self.cond.wait)
         self.assertRaises(RuntimeError, self.cond.notifyAll)
         self.cond.release()
     class TestPipeCondition(_ConditionTestCase):
       """_PipeCondition tests"""
       """PipeCondition tests"""
       def setUp(self):
         _ConditionTestCase.setUp(self, locking._PipeCondition)
         _ConditionTestCase.setUp(self, locking.PipeCondition)
       def testAcquireRelease(self):
         self._testAcquireRelease()
-...
         self.assertRaises(Queue.Empty, self.done.get_nowait)
     class TestSingleActionPipeCondition(unittest.TestCase):
       """_SingleActionPipeCondition tests"""
       def setUp(self):
         self.cond = locking._SingleActionPipeCondition()
       def testInitialization(self):
         self.assert_(self.cond._read_fd is not None)
         self.assert_(self.cond._write_fd is not None)
         self.assert_(self.cond._poller is not None)
         self.assertEqual(self.cond._nwaiters, 0)
       def testUsageCount(self):
         self.cond.StartWaiting()
         self.assert_(self.cond._read_fd is not None)
         self.assert_(self.cond._write_fd is not None)
         self.assert_(self.cond._poller is not None)
         self.assertEqual(self.cond._nwaiters, 1)
         # use again
         self.cond.StartWaiting()
         self.assertEqual(self.cond._nwaiters, 2)
         # there is more than one user
         self.assert_(not self.cond.DoneWaiting())
         self.assert_(self.cond._read_fd is not None)
         self.assert_(self.cond._write_fd is not None)
         self.assert_(self.cond._poller is not None)
         self.assertEqual(self.cond._nwaiters, 1)
         self.assert_(self.cond.DoneWaiting())
         self.assertEqual(self.cond._nwaiters, 0)
         self.assert_(self.cond._read_fd is None)
         self.assert_(self.cond._write_fd is None)
         self.assert_(self.cond._poller is None)
       def testNotify(self):
         wait1 = self.cond.StartWaiting()
         wait2 = self.cond.StartWaiting()
         self.assert_(self.cond._read_fd is not None)
         self.assert_(self.cond._write_fd is not None)
         self.assert_(self.cond._poller is not None)
         self.cond.notifyAll()
         self.assert_(self.cond._read_fd is not None)
         self.assert_(self.cond._write_fd is None)
         self.assert_(self.cond._poller is not None)
         self.assert_(not self.cond.DoneWaiting())
         self.assert_(self.cond._read_fd is not None)
         self.assert_(self.cond._write_fd is None)
         self.assert_(self.cond._poller is not None)
         self.assert_(self.cond.DoneWaiting())
         self.assert_(self.cond._read_fd is None)
         self.assert_(self.cond._write_fd is None)
         self.assert_(self.cond._poller is None)
       def testReusage(self):
         self.cond.StartWaiting()
         self.assert_(self.cond._read_fd is not None)
         self.assert_(self.cond._write_fd is not None)
         self.assert_(self.cond._poller is not None)
         self.assert_(self.cond.DoneWaiting())
         self.assertRaises(RuntimeError, self.cond.StartWaiting)
         self.assert_(self.cond._read_fd is None)
         self.assert_(self.cond._write_fd is None)
         self.assert_(self.cond._poller is None)
       def testNotifyTwice(self):
         self.cond.notifyAll()
         self.assertRaises(RuntimeError, self.cond.notifyAll)
     class TestSharedLock(_ThreadedTestCase):
       """SharedLock tests"""

Also available in: Unified diff

Synnefo » snf-ganeti

Revision 34cb5617