def testNonZeroIsFalse(self):
     et = ExpiryTime(timedelta(milliseconds=10))
     self.assertFalse(et)
     self.assertFalse(bool(et))
     sleep(et.remainingSeconds())
     self.assertTrue(et)
     self.assertTrue(bool(et))

 def testNoneToUnExpiredComparison(self):
     et1 = ExpiryTime(None)
     et2 = ExpiryTime(timedelta(milliseconds=10))
     self.assertNotEqual(et1, et2)
     self.assertNotEqual(et2, et1)
     sleep(et2.remainingSeconds())
     self.assertNotEqual(et1, et2)
     self.assertNotEqual(et2, et1)

 def updateCapability(self, capabilityName, capabilityValue=None):
     self._updCAPFAILED = False
     attemptLimit = ExpiryTime(MAX_CAPABILITY_UPDATE_DELAY)
     self.transport.scheduleTransmit(
         None,
         TransmitIntent(self.adminAddr,
                        CapabilityUpdate(capabilityName, capabilityValue),
                        onError = self._updateCapsFailed))
     while not attemptLimit.expired():
         if not self.transport.run(TransmitOnly, attemptLimit.remaining()):
             break  # all transmits completed
     if self._updCAPFAILED or attemptLimit.expired():
         raise ActorSystemFailure("Could not update Actor System Admin capabilities.")

 def testNoneComparedToNonZero(self):
     et1 = ExpiryTime(None)
     et2 = ExpiryTime(timedelta(milliseconds=10))
     # None == forever, so it is greater than anything, although equal to itself
     self.assertGreater(et1, et2)
     self.assertLess(et2, et1)
     self.assertTrue(et1 > et2)
     self.assertTrue(et2 < et1)
     sleep(et2.remainingSeconds())
     self.assertGreater(et1, et2)
     self.assertLess(et2, et1)
     self.assertTrue(et1 > et2)
     self.assertTrue(et2 < et1)

 def unloadActorSource(self, sourceHash):
     self._LOADFAILED = None
     loadLimit = ExpiryTime(MAX_LOAD_SOURCE_DELAY)
     self.transport.scheduleTransmit(None,
                                     TransmitIntent(self.adminAddr,
                                                    ValidateSource(sourceHash, None),
                                                    onError = self._loadReqFailed))
     while not loadLimit.expired():
         if not self.transport.run(TransmitOnly, loadLimit.remaining()):
             break  # all transmits completed
     if self._LOADFAILED or loadLimit.expired():
         raise ActorSystemFailure('Unload source failed due to ' +
                                  ('failure response' if self._LOADFAILED else
                                   'timeout (%s)'%str(loadLimit)))

    def run(self, incomingHandler, maximumDuration=None):
        """Core scheduling method; called by the current Actor process when
           idle to await new messages (or to do background
           processing).
        """
        self._max_runtime = ExpiryTime(maximumDuration)

        while not self._max_runtime.expired():
            now = datetime.now()
            self.run_time = min([ExpiryTime(P - now) for P in self._pendingWakeups] +
                                [self._max_runtime])
            rval = self._runWithExpiry(incomingHandler)
            if rval is not None:
                return rval

            if not self._realizeWakeups():
                # No wakeups were processed, and the inner run
                # returned, so assume there's nothing to do and exit
                return rval

            while self._activeWakeups:
                w = self._activeWakeups.pop()
                if incomingHandler in (None, TransmitOnly):
                    return w
                if not incomingHandler(w):
                    return None

        return None

 def sendWithHysteresis(self, intent):
     if self._hysteresis_until.expired():
         self._current_hysteresis = self._hysteresis_min_period
         self._sender(intent)
     else:
         dups = self._keepIf(lambda M: (M.targetAddr != intent.targetAddr or
                                              type(M.message) != type(intent.message)))
         # The dups are duplicate sends to the new intent's target; complete them when
         # the actual message is finally sent with the same result
         if dups:
             intent.addCallback(self._dupSentGood(dups), self._dupSentFail(dups))
         self._hysteresis_queue.append(intent)
         self._current_hysteresis = min(
             (self._hysteresis_min_period
              if (self._current_hysteresis is None or
                  self._current_hysteresis < self._hysteresis_min_period) else
              self._current_hysteresis * self._hysteresis_rate),
             self._hysteresis_max_period)
     self._hysteresis_until = ExpiryTime(
         timedelta(seconds=0)
         if not self._current_hysteresis else
         (self._current_hysteresis -
          (timedelta(seconds=0)
           if not self._hysteresis_until else
           self._hysteresis_until.remaining())))

    def _checkConvention(self):
        if self.isConventionLeader():
            missing = []
            for each in self._conventionMembers:
                if self._conventionMembers[each].registryValid.expired():
                    missing.append(each)
            for each in missing:
                thesplog('%s missed %d checkins (%s); assuming it has died',
                         str(self._conventionMembers[each]),
                         CONVENTION_REGISTRATION_MISS_MAX,
                         str(self._conventionMembers[each].registryValid),
                         level=logging.WARNING, primary=True)
                self._remoteSystemCleanup(self._conventionMembers[each].remoteAddress)
            self._conventionRegistration = ExpiryTime(CONVENTION_REREGISTRATION_PERIOD)
        else:
            # Re-register with the Convention if it's time
            if self._conventionAddress and self._conventionRegistration.expired():
                self.setupConvention()

        for each in self._conventionMembers:
            member = self._conventionMembers[each]
            if member.preRegistered and \
               member.preRegistered.pingValid.expired() and \
               not member.preRegistered.pingPending:
                member.preRegistered.pingPending = True
                member.preRegistered.pingValid = ExpiryTime(CONVENTION_RESTART_PERIOD
                                                            if member.registryValid.expired()
                                                            else CONVENTION_REREGISTRATION_PERIOD)
                self._hysteresisSender.sendWithHysteresis(
                    TransmitIntent(member.remoteAddress, ConventionInvite(),
                                   onSuccess = self._preRegQueryNotPending,
                                   onError = self._preRegQueryNotPending))

 def __init__(self, *args, **kw):
     super(ConventioneerAdmin, self).__init__(*args, **kw)
     self._conventionMembers = {} # key=Remote Admin Addr, value=ConventionMemberData
     self._conventionRegistration = ExpiryTime(timedelta(seconds=0))
     self._conventionNotificationHandler = None
     self._conventionAddress = None  # Not a member; still could be a leader
     self._pendingSources = {}  # key = sourceHash, value is array of PendingActor requests
     self._hysteresisSender = HysteresisDelaySender(self._send_intent)

    def __init__(self, system, logDefs = None):
        self._numPrimaries = 0
        # Expects self.transport has already been set by subclass __init__
        self.adminAddr = self.transport.getAdminAddr(system.capabilities)
        tryingTime = ExpiryTime(MAX_SYSTEM_SHUTDOWN_DELAY + timedelta(seconds=1))
        while not tryingTime.expired():
            if not self.transport.probeAdmin(self.adminAddr):
                self._startAdmin(self.adminAddr,
                                 self.transport.myAddress,
                                 system.capabilities,
                                 logDefs)
            if self._verifyAdminRunning(): return
            import time
            time.sleep(0.5)  # Previous version may have been exiting

        if not self._verifyAdminRunning():
            raise InvalidActorAddress(self.adminAddr,
                                          'not a valid or useable ActorSystem Admin')

 def tell(self, anActor, msg):
     attemptLimit = ExpiryTime(MAX_TELL_PERIOD)
     import socket
     for attempt in range(5000):
         try:
             self.transport.scheduleTransmit(
                 None,
                 TransmitIntent(anActor, msg, onError=self._tellFailed))
             while not attemptLimit.expired():
                 if not self.transport.run(TransmitOnly, attemptLimit.remaining()):
                     break  # all transmits completed
             return
         except socket.error as ex:
             import errno
             if errno.EMFILE == ex.errno:
                 import time
                 time.sleep(0.1)
             else:
                 raise

 def loadActorSource(self, fname):
     self._LOADFAILED = None
     loadLimit = ExpiryTime(MAX_LOAD_SOURCE_DELAY)
     f = fname if hasattr(fname, 'read') else open(fname, 'rb')
     try:
         d = f.read()
         import hashlib
         hval = hashlib.md5(d).hexdigest()
         self.transport.scheduleTransmit(None,
                                         TransmitIntent(self.adminAddr,
                                                        ValidateSource(hval, d),
                                                        onError = self._loadReqFailed))
         while not loadLimit.expired():
             if not self.transport.run(TransmitOnly, loadLimit.remaining()):
                 break  # all transmits completed
         if self._LOADFAILED or loadLimit.expired():
             raise ActorSystemFailure('Load source failed due to ' +
                                      ('failure response (%s)'%self._LOADFAILED
                                       if self._LOADFAILED else
                                       'timeout (%s)'%str(loadLimit)))
         return hval
     finally:
         f.close()

 def checkSends(self):
     if self.delay.expired():
         hsends = self._hysteresis_queue
         self._hysteresis_queue = []
         self._current_hysteresis = (
             None
             if (self._current_hysteresis is None or
                 self._current_hysteresis < self._hysteresis_min_period) else
             self._current_hysteresis / self._hysteresis_rate)
         self._hysteresis_until = ExpiryTime(self._current_hysteresis
                                             if self._current_hysteresis else
                                             timedelta(seconds=0))
         for intent in hsends:
             self._sender(intent)

 def shutdown(self):
     thesplog('ActorSystem shutdown requested.', level=logging.INFO)
     time_to_quit = ExpiryTime(MAX_SYSTEM_SHUTDOWN_DELAY)
     self.transport.scheduleTransmit(
         None,
         TransmitIntent(self.adminAddr, SystemShutdown(),
                        onError=self._shutdownSendFailed))
     while not time_to_quit.expired():
         response = self.transport.run(None, time_to_quit.remaining())
         if getattr(self, '_TASF', False):
             thesplog('Could not send shutdown request to Admin'
                      '; aborting but not necessarily stopped',
                      level=logging.WARNING)
             return
         if response:
             if isinstance(response.message, SystemShutdownCompleted):
                 break
             else:
                 thesplog('Expected shutdown completed message, got: %s', response.message,
                          level=logging.WARNING)
         else:
             thesplog('No response to Admin shutdown request; Actor system not completely shutdown',
                      level=logging.ERROR)
     thesplog('ActorSystem shutdown complete.')

 def _checkConvention(self):
     if self.isConventionLeader():
         missing = []
         for each in self._conventionMembers:
             if self._conventionMembers[each].registryValid.expired():
                 missing.append(each)
         for each in missing:
             thesplog('%s missed %d checkins (%s); assuming it has died',
                      str(self._conventionMembers[each]),
                      CONVENTION_REGISTRATION_MISS_MAX,
                      str(self._conventionMembers[each].registryValid),
                      level=logging.WARNING, primary=True)
             self._remoteSystemCleanup(self._conventionMembers[each].remoteAddress)
         self._conventionRegistration = ExpiryTime(CONVENTION_REREGISTRATION_PERIOD)
     else:
         # Re-register with the Convention if it's time
         if self._conventionAddress and self._conventionRegistration.expired():
             self.setupConvention()

 def setupConvention(self):
     if self.isShuttingDown(): return
     if not self._conventionAddress:
         gCA = getattr(self.transport, 'getConventionAddress', lambda c: None)
         self._conventionAddress = gCA(self.capabilities)
         if self._conventionAddress == self.myAddress:
             self._conventionAddress = None
     if self._conventionAddress:
         thesplog('Admin registering with Convention @ %s (%s)',
                  self._conventionAddress,
                  'first time' if getattr(self, '_conventionLeaderIsGone', True) else
                  're-registering',
                  level=logging.INFO, primary=True)
         self._hysteresisSender.sendWithHysteresis(
             TransmitIntent(self._conventionAddress,
                            ConventionRegister(self.myAddress,
                                               self.capabilities,
                                               getattr(self, '_conventionLeaderIsGone', True)),
                            onSuccess = self._setupConventionCBGood,
                            onError = self._setupConventionCBError))
         self._conventionRegistration = ExpiryTime(CONVENTION_REREGISTRATION_PERIOD)

 def drainTransmits(self):
     drainLimit = ExpiryTime(MAX_SHUTDOWN_DRAIN_PERIOD)
     while not drainLimit.expired():
         if not self.transport.run(TransmitOnly, drainLimit.remaining()):
             break  # no transmits left

 def testNoneExpired(self):
     et = ExpiryTime(None)
     self.assertFalse(et.expired())

 def testZeroExpired(self):
     et = ExpiryTime(timedelta(seconds=0))
     self.assertTrue(et.expired())

 def testZeroRemaining(self):
     et = ExpiryTime(timedelta(seconds=0))
     self.assertEqual(timedelta(days=0), et.remaining())