def test_cleanReactor_delayed_calls_all_active(self):
        """
        It will cancel any delayed calls in the reactor queue.
        """
        reactor.callLater(1, self.ignoreFailure)
        reactor.callLater(2, self.ignoreFailure)
        self.assertIsNotEmpty(reactor.getDelayedCalls())

        self._cleanReactor()

        self.assertIsEmpty(reactor.getDelayedCalls())

    def test_delayedCallCleanup(self):
        """Checking to make sure Sessions do not leave extra DelayedCalls.
        """
        from twisted.internet import reactor

        delayedCallsBeforeSession = repr(reactor.getDelayedCalls())

        session = self.site.makeSession()
        session.touch()
        session.expire()

        self.failUnlessEqual(delayedCallsBeforeSession, repr(reactor.getDelayedCalls()))

    def umount(self):
        self.readonly = True
        if self.shutting_down:
            trace("shutdown", "called twice")
            return
        try:
            self.shutting_down = True

            trace("shutdown", "stopping services")
            yield self.services.stopService()
            trace("shutdown", "disconnect peers")
            self.disconnect_all()
            for k in self.remote.keys():
                del self.remote[k]
            n = 0
            for c in reactor.getDelayedCalls():
                n += 1
                trace("shutdown", "speed-up %s", c)
                c.reset(0)
            if n:
                trace("shutdown", "speed-up wait %d", n)
                yield reactor.callLater(n / 10, lambda: None)
            trace("shutdown", "run idle")
            yield IdleWorker.run()
            trace("shutdown", "flush inodes")
            yield self.db(flush_inodes)
            trace("shutdown", "super")
            yield super(SqlFuse, self).stop(False)
            trace("shutdown", "stop DB")
            yield self.db.stopService()
            trace("shutdown", "done")
        except Exception as e:
            log.err(e, "Shutting down")
            traceback.print_exc()

def shutdown():
    global main_xmlrpc_handler
    global _xmlrpc_listener
    global _xmlrpc_site
    try:
        site = _xmlrpc_site
        logging.info("shutting down, first closing listening ports...")
        print("Shutting down, hold on a moment...")
        yield stop_listening()

# This doesn't work, site.session is always empty
        logging.info("Ports closed, waiting for current sessions to close...")
        logging.debug("Clients still connected: {}".format(len(site.sessions)))
        while not len(site.sessions)==0:
            logging.debug("Waiting, {} sessions still active".format(len(site.sessions)))
            yield task.deferLater(reactor, 1, lambda _:0, 0)

        logging.info("No more sessions, waiting for locked hosts...")
        while not utils.none_waiting():
            logging.info("Waiting to shut down, {} hosts still blocked".format(utils.count_waiting()))
            yield task.deferLater(reactor, 1, lambda _:0, 0)
            logging.debug("reactor.getDelayedCalls: {}".format([c.func for c in reactor.getDelayedCalls()]))

        logging.info("All hosts unlocked, waiting 3 more seconds...")
        yield task.deferLater(reactor, 1, lambda _:0, 0)
        logging.debug("Waiting 2 more seconds...")
        yield task.deferLater(reactor, 1, lambda _:0, 0)
        logging.debug("Waiting 1 more second...")
        yield task.deferLater(reactor, 1, lambda _:0, 0)
        logging.info("Continuing shutdown")
    except:
        logging.exception("Error in shutdown callback")

def check_clean(_):
    delayed_calls = reactor.getDelayedCalls()
    if delayed_calls:
        print "DIRTY"
    else:
        print "CLEAN"
    reactor.stop()

 def tearDown(self):
     super(DNSQueryMonitoringProtocolTestCase, self).tearDown()
     # There doesn't seem to be any sane way to avoid the delayed call to
     # maybeParseConfig: https://twistedmatrix.com/trac/ticket/3745
     for call in reactor.getDelayedCalls():
         if call.func.func_name == 'maybeParseConfig':
             call.cancel()

def runtime_info():
    delayed = reactor.getDelayedCalls()
    readers = reactor.getReaders()
    writers = reactor.getWriters()
    servers = []
    clients = []
    other = []
    for reader in readers:
        if isinstance(reader, tcp.Server):
            servers.append({
                'transport': reader,
                'host': reader.getHost(),
                'peer': reader.getPeer()
            })
        elif isinstance(reader, tcp.Client):
            clients.append({
                'transport': reader,
                'host': reader.getHost(),
                'peer': reader.getPeer()
            })
        else:
            other.append(reader)
    return {
        'num_clients': len(clients),
        'num_servers': len(servers),
        'num_other': len(other),
        'num_writers': len(writers),
        'num_delayed': len(delayed),
        'clients': clients,
        'servers': servers,
        'other': other,
        'writers': writers,
        'delayed': delayed,
    }

def print_runtime_info(sig, frame):
    if sig in [signal.SIGUSR1, signal.SIGUSR2]:
        delayed = reactor.getDelayedCalls()
        readers = reactor.getReaders()
        writers = reactor.getWriters()
        clients = []
        http_conn_num = 0
        for reader in readers:
            if isinstance(reader, twisted.internet.tcp.Server):
                clients.append(reader.getPeer())
            if isinstance(reader, twisted.internet.tcp.Client):
                http_conn_num += 1
        log.msg(
            "[Clients: %(client_num)s] [HTTP Conns: %(http_conn_num)s] "
            "[Readers: %(reader_num)s] [Writers: %(writer_num)s] "
            "[DelayedCalls: %(delayed_num)s]"
            % {
                "client_num": len(clients),
                "http_conn_num": http_conn_num,
                "reader_num": len(readers),
                "writer_num": len(writers),
                "delayed_num": len(delayed),
            }
        )
        log.msg("[Connected Clients]: %s" % clients)
        if sig == signal.SIGUSR2:
            for d in delayed:
                log.msg("SIGUSR2[delayed]: %s" % d)

            for r in readers:
                log.msg("SIGUSR2[reader]: %s" % r)

            for w in writers:
                log.msg("SIGUSR2[writer]: %s" % w)

    def testQuestionLoadSpeed(self):
        "Check if trivia questions load fast enough."

        session = self.trivia_plugin.Session()

        #Create questions
        for i in range(40000): #There can be a lot of trivia questions
            question_string = generate_random_sentence(5, 10, 5, 20)
            answer_string = generate_random_sentence(1, 6, 1, 20)
            category_string = generate_random_word(10, 30)
            question = RegularQuestion(question = question_string, answer = answer_string, category = category_string)
            session.save_or_update(question)
        session.commit()

        #Load questions and check the amount of time it takes
        questions_asked = 5000
        before = datetime.now()
        for i in range(questions_asked):
            self.trivia_plugin.on_next_question(self.vtkbot, self.channel)
            timers = reactor.getDelayedCalls()
            for timer in timers:
                try:
                    timer.cancel()
                except:
                    pass
        after = datetime.now()
        diff = after - before
        self.assert_(diff.seconds < questions_asked/20) #Questions need to be asked as fast as possible, to avoid long delays for the trivia players

 def tearDown(self):
     """!Tear down a test
     """
     delayedCalls = reactor.getDelayedCalls()
     for call in delayedCalls:
         call.cancel()
     return self.dw.close()

 def checkSessionCreated(request):
     cookie = request.getCookie(fresources.COOKIE_NAME)
     self.failIf(cookie is None)
     sessionID = base64.b64decode(cookie).split(':')[0]
     session = self.site.sessions.get(sessionID, None)
     self.failIf(session is None)
     for d in reactor.getDelayedCalls():
         d.cancel()

 def tearDown(self):
     yield self.client.remove('test', CF)
     yield self.client.remove('test2', CF)
     yield self.client.remove('test', SCF)
     yield self.client.remove('test2', SCF)
     self.cmanager.shutdown()
     for c in reactor.getDelayedCalls():
         c.cancel()
     reactor.removeAll()

 def _reactorQueueToString(self):
     """
     Return a string representation of all delayed calls from reactor
     queue.
     """
     result = []
     for delayed in reactor.getDelayedCalls():
         result.append(str(delayed.func))
     return '\n'.join(result)

 def tearDown(self):
     pending = reactor.getDelayedCalls()
     active = bool(pending)
     for p in pending:
         print p
         if p.active():
             print "cancel"
             p.cancel()
     print "fini"

 def _delayed_calls_done(self):
     # We're done when the only remaining DelayedCalls fire after threshold.
     # (These will be associated with the test timeout, or else they *should*
     # cause an unclean reactor error because the test should have waited for
     # them.)
     threshold = time.time() + 10
     for delayed in reactor.getDelayedCalls():
         if delayed.getTime() < threshold:
             return False
     return True

def clean_reactor():
    """
    Cleans the reactor of any uncalled DelayedCall objects created by
    reactor.callLater().  This is useful for tests that run without the reactor
    and do not actually need the DelayedCall to run.
    """
    
    # clean reactor
    for call in reactor.getDelayedCalls():
        call.cancel()

 def checkSessionID(request):
     # The auth is not valid anymore and has been renewed,
     # but the session should stay the same
     cookie = request.getCookie(fresources.COOKIE_NAME)
     self.failIf(cookie is None)
     sessionID, authExpiracy, none = \
            base64.b64decode(cookie).split(':')
     self.assertEquals(authExpiracy, '0')
     self.assertEquals(sessionID, self.sessionID)
     for d in reactor.getDelayedCalls():
         d.cancel()

    def tearDown(self):
        """Cleanup method after each ``test_*`` method runs; removes timed out
        connections on the reactor and clears the :ivar:`transport`.
        """
        self.transport.clear()  # Clear bytes from the transport.

        for delay in reactor.getDelayedCalls():
            try:
                delay.cancel()
            except (AlreadyCalled, AlreadyCancelled):
                pass

 def die(self):        
     # we don't want zombie thread do we ? ;)
     print "Del SocksTools"
     # dirty but yeah
     for i in xrange(0,2):
         for calls in reactor.getDelayedCalls():
             if calls.func == self._twistedLoop and calls.active():
                 print "cancel", calls
                 try: calls.cancel()
                 except Exception as e: print e
         time.sleep(1)
     for port, values in self.proxies.items():
         print "close port ", port
         values[3].reply('Closing proxy on port '+str(port))
         values[5].stopListening()
         values[4].closeConnection()     
     for calls in reactor.getDelayedCalls():
         if calls.func == self._twistedLoop and calls.active():
             print "cancel", calls
             try: calls.cancel()
             except Exception as e: print e

    def handle_refuse(self, message):
        """Este método deve ser sobrescrito quando na implementação
            do protocolo, sendo executado sempre que o agente receber
            uma mensagem do tipo FIPA_REFUSE

            :param message: Mensagem FIPA-ACL
        """
        self.received_qtd += 1
        if self.received_qtd == self.cfp_qtd:
            delayed_calls = reactor.getDelayedCalls()
            for call in delayed_calls:
                call.cancel()