Python asyncore 模块，poll() 实例源码

def start(self):
        """The main loop keeping the bot alive"""
        self.connect(self.current_server)
        while self.running:
            # No need to busy-wait
            time.sleep(0.2)
            self.schedule.execute()
            # Skip polling etc. if we're reconnecting
            if self.reconnecting and not self.connected:
                continue
            try:
                asyncore.poll(timeout=10, map={self.socket: self})
            except socket.error as e:
                log.error(e)
                self.reconnect_with_delay()
                continue
            dc_channels = self.disconnected_channels()
            if dc_channels:
                self.join(dc_channels, delay=5)

def seconds_until_wakeup(cls):
    """
    Calculate delay until next timer expires, or None if no timers are
    set and we should wait indefinitely.  Rounds up to avoid spinning
    in select() or poll().  We could calculate fractional seconds in
    the right units instead, but select() and poll() don't even take
    the same units (argh!), and we're not doing anything that
    hair-triggered, so rounding up is simplest.
    """
    if not timer_queue:
      return None
    now = rpki.sundial.now()
    if now >= timer_queue[0].when:
      return 0
    delay = timer_queue[0].when - now
    seconds = delay.convert_to_seconds()
    if delay.microseconds:
      seconds += 1
    return seconds

def seconds_until_wakeup(cls):
    """
    Calculate delay until next timer expires, or None if no timers are
    set and we should wait indefinitely.  Rounds up to avoid spinning
    in select() or poll().  We could calculate fractional seconds in
    the right units instead, but select() and poll() don't even take
    the same units (argh!), and we're not doing anything that
    hair-triggered, so rounding up is simplest.
    """
    if not timer_queue:
      return None
    now = rpki.sundial.now()
    if now >= timer_queue[0].when:
      return 0
    delay = timer_queue[0].when - now
    seconds = delay.convert_to_seconds()
    if delay.microseconds:
      seconds += 1
    return seconds

def seconds_until_wakeup(cls):
    """
    Calculate delay until next timer expires, or None if no timers are
    set and we should wait indefinitely.  Rounds up to avoid spinning
    in select() or poll().  We could calculate fractional seconds in
    the right units instead, but select() and poll() don't even take
    the same units (argh!), and we're not doing anything that
    hair-triggered, so rounding up is simplest.
    """
    if not timer_queue:
      return None
    now = rpki.sundial.now()
    if now >= timer_queue[0].when:
      return 0
    delay = timer_queue[0].when - now
    seconds = delay.convert_to_seconds()
    if delay.microseconds:
      seconds += 1
    return seconds

def seconds_until_wakeup(cls):
    """
    Calculate delay until next timer expires, or None if no timers are
    set and we should wait indefinitely.  Rounds up to avoid spinning
    in select() or poll().  We could calculate fractional seconds in
    the right units instead, but select() and poll() don't even take
    the same units (argh!), and we're not doing anything that
    hair-triggered, so rounding up is simplest.
    """
    if not timer_queue:
      return None
    now = rpki.sundial.now()
    if now >= timer_queue[0].when:
      return 0
    delay = timer_queue[0].when - now
    seconds = delay.convert_to_seconds()
    if delay.microseconds:
      seconds += 1
    return seconds

def seconds_until_wakeup(cls):
    """
    Calculate delay until next timer expires, or None if no timers are
    set and we should wait indefinitely.  Rounds up to avoid spinning
    in select() or poll().  We could calculate fractional seconds in
    the right units instead, but select() and poll() don't even take
    the same units (argh!), and we're not doing anything that
    hair-triggered, so rounding up is simplest.
    """
    if not timer_queue:
      return None
    now = rpki.sundial.now()
    if now >= timer_queue[0].when:
      return 0
    delay = timer_queue[0].when - now
    seconds = delay.convert_to_seconds()
    if delay.microseconds:
      seconds += 1
    return seconds

def seconds_until_wakeup(cls):
    """
    Calculate delay until next timer expires, or None if no timers are
    set and we should wait indefinitely.  Rounds up to avoid spinning
    in select() or poll().  We could calculate fractional seconds in
    the right units instead, but select() and poll() don't even take
    the same units (argh!), and we're not doing anything that
    hair-triggered, so rounding up is simplest.
    """
    if not timer_queue:
      return None
    now = rpki.sundial.now()
    if now >= timer_queue[0].when:
      return 0
    delay = timer_queue[0].when - now
    seconds = delay.convert_to_seconds()
    if delay.microseconds:
      seconds += 1
    return seconds

def seconds_until_wakeup(cls):
    """
    Calculate delay until next timer expires, or None if no timers are
    set and we should wait indefinitely.  Rounds up to avoid spinning
    in select() or poll().  We could calculate fractional seconds in
    the right units instead, but select() and poll() don't even take
    the same units (argh!), and we're not doing anything that
    hair-triggered, so rounding up is simplest.
    """
    if not timer_queue:
      return None
    now = rpki.sundial.now()
    if now >= timer_queue[0].when:
      return 0
    delay = timer_queue[0].when - now
    seconds = delay.convert_to_seconds()
    if delay.microseconds:
      seconds += 1
    return seconds

def signal_handler(signal, frame):
  global server
  global must_exit
  logging.error('Exiting...')
  must_exit = True
  del server


# Wrapper around the asyncore loop that lets us poll the in/out pipes every 1ms

def debugging_server(serv, serv_evt, client_evt):
    serv_evt.set()

    try:
        if hasattr(select, 'poll'):
            poll_fun = asyncore.poll2
        else:
            poll_fun = asyncore.poll

        n = 1000
        while asyncore.socket_map and n > 0:
            poll_fun(0.01, asyncore.socket_map)

            # when the client conversation is finished, it will
            # set client_evt, and it's then ok to kill the server
            if client_evt.is_set():
                serv.close()
                break

            n -= 1

    except socket.timeout:
        pass
    finally:
        if not client_evt.is_set():
            # allow some time for the client to read the result
            time.sleep(0.5)
            serv.close()
        asyncore.close_all()
        serv_evt.set()

def test_readwriteexc(self):
        # Check exception handling behavior of read, write and _exception

        # check that ExitNow exceptions in the object handler method
        # bubbles all the way up through asyncore read/write/_exception calls
        tr1 = exitingdummy()
        self.assertRaises(asyncore.ExitNow, asyncore.read, tr1)
        self.assertRaises(asyncore.ExitNow, asyncore.write, tr1)
        self.assertRaises(asyncore.ExitNow, asyncore._exception, tr1)

        # check that an exception other than ExitNow in the object handler
        # method causes the handle_error method to get called
        tr2 = crashingdummy()
        asyncore.read(tr2)
        self.assertEqual(tr2.error_handled, True)

        tr2 = crashingdummy()
        asyncore.write(tr2)
        self.assertEqual(tr2.error_handled, True)

        tr2 = crashingdummy()
        asyncore._exception(tr2)
        self.assertEqual(tr2.error_handled, True)

    # asyncore.readwrite uses constants in the select module that
    # are not present in Windows systems (see this thread:
    # http://mail.python.org/pipermail/python-list/2001-October/109973.html)
    # These constants should be present as long as poll is available

def test_send(self):
        evt = threading.Event()
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(3)
        port = support.bind_port(sock)

        cap = BytesIO()
        args = (evt, cap, sock)
        t = threading.Thread(target=capture_server, args=args)
        t.start()
        try:
            # wait a little longer for the server to initialize (it sometimes
            # refuses connections on slow machines without this wait)
            time.sleep(0.2)

            data = b"Suppose there isn't a 16-ton weight?"
            d = dispatcherwithsend_noread()
            d.create_socket(socket.AF_INET, socket.SOCK_STREAM)
            d.connect((HOST, port))

            # give time for socket to connect
            time.sleep(0.1)

            d.send(data)
            d.send(data)
            d.send(b'\n')

            n = 1000
            while d.out_buffer and n > 0:
                asyncore.poll()
                n -= 1

            evt.wait()

            self.assertEqual(cap.getvalue(), data*2)
        finally:
            t.join()

def debugging_server(serv, serv_evt, client_evt):
    serv_evt.set()

    try:
        if hasattr(select, 'poll'):
            poll_fun = asyncore.poll2
        else:
            poll_fun = asyncore.poll

        n = 1000
        while asyncore.socket_map and n > 0:
            poll_fun(0.01, asyncore.socket_map)

            # when the client conversation is finished, it will
            # set client_evt, and it's then ok to kill the server
            if client_evt.is_set():
                serv.close()
                break

            n -= 1

    except socket.timeout:
        pass
    finally:
        if not client_evt.is_set():
            # allow some time for the client to read the result
            time.sleep(0.5)
            serv.close()
        asyncore.close_all()
        serv_evt.set()

def test_readwriteexc(self):
        # Check exception handling behavior of read, write and _exception

        # check that ExitNow exceptions in the object handler method
        # bubbles all the way up through asyncore read/write/_exception calls
        tr1 = exitingdummy()
        self.assertRaises(asyncore.ExitNow, asyncore.read, tr1)
        self.assertRaises(asyncore.ExitNow, asyncore.write, tr1)
        self.assertRaises(asyncore.ExitNow, asyncore._exception, tr1)

        # check that an exception other than ExitNow in the object handler
        # method causes the handle_error method to get called
        tr2 = crashingdummy()
        asyncore.read(tr2)
        self.assertEqual(tr2.error_handled, True)

        tr2 = crashingdummy()
        asyncore.write(tr2)
        self.assertEqual(tr2.error_handled, True)

        tr2 = crashingdummy()
        asyncore._exception(tr2)
        self.assertEqual(tr2.error_handled, True)

    # asyncore.readwrite uses constants in the select module that
    # are not present in Windows systems (see this thread:
    # http://mail.python.org/pipermail/python-list/2001-October/109973.html)
    # These constants should be present as long as poll is available

def test_send(self):
        evt = threading.Event()
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(3)
        port = test_support.bind_port(sock)

        cap = StringIO()
        args = (evt, cap, sock)
        t = threading.Thread(target=capture_server, args=args)
        t.start()
        try:
            # wait a little longer for the server to initialize (it sometimes
            # refuses connections on slow machines without this wait)
            time.sleep(0.2)

            data = "Suppose there isn't a 16-ton weight?"
            d = dispatcherwithsend_noread()
            d.create_socket(socket.AF_INET, socket.SOCK_STREAM)
            d.connect((HOST, port))

            # give time for socket to connect
            time.sleep(0.1)

            d.send(data)
            d.send(data)
            d.send('\n')

            n = 1000
            while d.out_buffer and n > 0:
                asyncore.poll()
                n -= 1

            evt.wait()

            self.assertEqual(cap.getvalue(), data*2)
        finally:
            t.join()

def debugging_server(serv, serv_evt, client_evt):
    serv_evt.set()

    try:
        if hasattr(select, 'poll'):
            poll_fun = asyncore.poll2
        else:
            poll_fun = asyncore.poll

        n = 1000
        while asyncore.socket_map and n > 0:
            poll_fun(0.01, asyncore.socket_map)

            # when the client conversation is finished, it will
            # set client_evt, and it's then ok to kill the server
            if client_evt.is_set():
                serv.close()
                break

            n -= 1

    except socket.timeout:
        pass
    finally:
        if not client_evt.is_set():
            # allow some time for the client to read the result
            time.sleep(0.5)
            serv.close()
        asyncore.close_all()
        serv_evt.set()

def test_readwriteexc(self):
        # Check exception handling behavior of read, write and _exception

        # check that ExitNow exceptions in the object handler method
        # bubbles all the way up through asyncore read/write/_exception calls
        tr1 = exitingdummy()
        self.assertRaises(asyncore.ExitNow, asyncore.read, tr1)
        self.assertRaises(asyncore.ExitNow, asyncore.write, tr1)
        self.assertRaises(asyncore.ExitNow, asyncore._exception, tr1)

        # check that an exception other than ExitNow in the object handler
        # method causes the handle_error method to get called
        tr2 = crashingdummy()
        asyncore.read(tr2)
        self.assertEqual(tr2.error_handled, True)

        tr2 = crashingdummy()
        asyncore.write(tr2)
        self.assertEqual(tr2.error_handled, True)

        tr2 = crashingdummy()
        asyncore._exception(tr2)
        self.assertEqual(tr2.error_handled, True)

    # asyncore.readwrite uses constants in the select module that
    # are not present in Windows systems (see this thread:
    # http://mail.python.org/pipermail/python-list/2001-October/109973.html)
    # These constants should be present as long as poll is available

def test_send(self):
        evt = threading.Event()
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(3)
        port = test_support.bind_port(sock)

        cap = StringIO()
        args = (evt, cap, sock)
        t = threading.Thread(target=capture_server, args=args)
        t.start()
        try:
            # wait a little longer for the server to initialize (it sometimes
            # refuses connections on slow machines without this wait)
            time.sleep(0.2)

            data = "Suppose there isn't a 16-ton weight?"
            d = dispatcherwithsend_noread()
            d.create_socket(socket.AF_INET, socket.SOCK_STREAM)
            d.connect((HOST, port))

            # give time for socket to connect
            time.sleep(0.1)

            d.send(data)
            d.send(data)
            d.send('\n')

            n = 1000
            while d.out_buffer and n > 0:
                asyncore.poll()
                n -= 1

            evt.wait()

            self.assertEqual(cap.getvalue(), data*2)
        finally:
            t.join()

def debugging_server(serv, serv_evt, client_evt):
    serv_evt.set()

    try:
        if hasattr(select, 'poll'):
            poll_fun = asyncore.poll2
        else:
            poll_fun = asyncore.poll

        n = 1000
        while asyncore.socket_map and n > 0:
            poll_fun(0.01, asyncore.socket_map)

            # when the client conversation is finished, it will
            # set client_evt, and it's then ok to kill the server
            if client_evt.is_set():
                serv.close()
                break

            n -= 1

    except socket.timeout:
        pass
    finally:
        if not client_evt.is_set():
            # allow some time for the client to read the result
            time.sleep(0.5)
            serv.close()
        asyncore.close_all()
        serv_evt.set()

def test_readwriteexc(self):
        # Check exception handling behavior of read, write and _exception

        # check that ExitNow exceptions in the object handler method
        # bubbles all the way up through asyncore read/write/_exception calls
        tr1 = exitingdummy()
        self.assertRaises(asyncore.ExitNow, asyncore.read, tr1)
        self.assertRaises(asyncore.ExitNow, asyncore.write, tr1)
        self.assertRaises(asyncore.ExitNow, asyncore._exception, tr1)

        # check that an exception other than ExitNow in the object handler
        # method causes the handle_error method to get called
        tr2 = crashingdummy()
        asyncore.read(tr2)
        self.assertEqual(tr2.error_handled, True)

        tr2 = crashingdummy()
        asyncore.write(tr2)
        self.assertEqual(tr2.error_handled, True)

        tr2 = crashingdummy()
        asyncore._exception(tr2)
        self.assertEqual(tr2.error_handled, True)

    # asyncore.readwrite uses constants in the select module that
    # are not present in Windows systems (see this thread:
    # http://mail.python.org/pipermail/python-list/2001-October/109973.html)
    # These constants should be present as long as poll is available

def test_send(self):
        evt = threading.Event()
        sock = socket.socket()
        sock.settimeout(3)
        port = support.bind_port(sock)

        cap = BytesIO()
        args = (evt, cap, sock)
        t = threading.Thread(target=capture_server, args=args)
        t.start()
        try:
            # wait a little longer for the server to initialize (it sometimes
            # refuses connections on slow machines without this wait)
            time.sleep(0.2)

            data = b"Suppose there isn't a 16-ton weight?"
            d = dispatcherwithsend_noread()
            d.create_socket()
            d.connect((HOST, port))

            # give time for socket to connect
            time.sleep(0.1)

            d.send(data)
            d.send(data)
            d.send(b'\n')

            n = 1000
            while d.out_buffer and n > 0:
                asyncore.poll()
                n -= 1

            evt.wait()

            self.assertEqual(cap.getvalue(), data*2)
        finally:
            t.join()

def debugging_server(serv, serv_evt, client_evt):
    serv_evt.set()

    try:
        if hasattr(select, 'poll'):
            poll_fun = asyncore.poll2
        else:
            poll_fun = asyncore.poll

        n = 1000
        while asyncore.socket_map and n > 0:
            poll_fun(0.01, asyncore.socket_map)

            # when the client conversation is finished, it will
            # set client_evt, and it's then ok to kill the server
            if client_evt.is_set():
                serv.close()
                break

            n -= 1

    except socket.timeout:
        pass
    finally:
        if not client_evt.is_set():
            # allow some time for the client to read the result
            time.sleep(0.5)
            serv.close()
        asyncore.close_all()
        serv_evt.set()

def test_readwriteexc(self):
        # Check exception handling behavior of read, write and _exception

        # check that ExitNow exceptions in the object handler method
        # bubbles all the way up through asyncore read/write/_exception calls
        tr1 = exitingdummy()
        self.assertRaises(asyncore.ExitNow, asyncore.read, tr1)
        self.assertRaises(asyncore.ExitNow, asyncore.write, tr1)
        self.assertRaises(asyncore.ExitNow, asyncore._exception, tr1)

        # check that an exception other than ExitNow in the object handler
        # method causes the handle_error method to get called
        tr2 = crashingdummy()
        asyncore.read(tr2)
        self.assertEqual(tr2.error_handled, True)

        tr2 = crashingdummy()
        asyncore.write(tr2)
        self.assertEqual(tr2.error_handled, True)

        tr2 = crashingdummy()
        asyncore._exception(tr2)
        self.assertEqual(tr2.error_handled, True)

    # asyncore.readwrite uses constants in the select module that
    # are not present in Windows systems (see this thread:
    # http://mail.python.org/pipermail/python-list/2001-October/109973.html)
    # These constants should be present as long as poll is available

def test_send(self):
        evt = threading.Event()
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(3)
        port = test_support.bind_port(sock)

        cap = StringIO()
        args = (evt, cap, sock)
        t = threading.Thread(target=capture_server, args=args)
        t.start()
        try:
            # wait a little longer for the server to initialize (it sometimes
            # refuses connections on slow machines without this wait)
            time.sleep(0.2)

            data = "Suppose there isn't a 16-ton weight?"
            d = dispatcherwithsend_noread()
            d.create_socket(socket.AF_INET, socket.SOCK_STREAM)
            d.connect((HOST, port))

            # give time for socket to connect
            time.sleep(0.1)

            d.send(data)
            d.send(data)
            d.send('\n')

            n = 1000
            while d.out_buffer and n > 0:
                asyncore.poll()
                n -= 1

            evt.wait()

            self.assertEqual(cap.getvalue(), data*2)
        finally:
            t.join()

def run(self):
        while not self._stop:
            asyncore.poll(timeout=0.1)

        if self.nf_queue_started:
            self.printDebug("Stopping QueueAsyncThread ...")
            self.printDebug("Unbind nf_queue")
            try:
                self.q.unbind(AF_INET6)
                self.printDebug("Close nf_queue {}".format(self.queue_number))
                self.q.close()
            except:
                pass
            self.nf_queue_started=False
            self.printDebug("QueueAsyncThread stopped")

def debugging_server(serv, serv_evt, client_evt):
    serv_evt.set()

    try:
        if hasattr(select, 'poll'):
            poll_fun = asyncore.poll2
        else:
            poll_fun = asyncore.poll

        n = 1000
        while asyncore.socket_map and n > 0:
            poll_fun(0.01, asyncore.socket_map)

            # when the client conversation is finished, it will
            # set client_evt, and it's then ok to kill the server
            if client_evt.is_set():
                serv.close()
                break

            n -= 1

    except socket.timeout:
        pass
    finally:
        if not client_evt.is_set():
            # allow some time for the client to read the result
            time.sleep(0.5)
            serv.close()
        asyncore.close_all()
        serv_evt.set()

def test_readwriteexc(self):
        # Check exception handling behavior of read, write and _exception

        # check that ExitNow exceptions in the object handler method
        # bubbles all the way up through asyncore read/write/_exception calls
        tr1 = exitingdummy()
        self.assertRaises(asyncore.ExitNow, asyncore.read, tr1)
        self.assertRaises(asyncore.ExitNow, asyncore.write, tr1)
        self.assertRaises(asyncore.ExitNow, asyncore._exception, tr1)

        # check that an exception other than ExitNow in the object handler
        # method causes the handle_error method to get called
        tr2 = crashingdummy()
        asyncore.read(tr2)
        self.assertEqual(tr2.error_handled, True)

        tr2 = crashingdummy()
        asyncore.write(tr2)
        self.assertEqual(tr2.error_handled, True)

        tr2 = crashingdummy()
        asyncore._exception(tr2)
        self.assertEqual(tr2.error_handled, True)

    # asyncore.readwrite uses constants in the select module that
    # are not present in Windows systems (see this thread:
    # http://mail.python.org/pipermail/python-list/2001-October/109973.html)
    # These constants should be present as long as poll is available

def test_send(self):
        evt = threading.Event()
        sock = socket.socket()
        sock.settimeout(3)
        port = support.bind_port(sock)

        cap = BytesIO()
        args = (evt, cap, sock)
        t = threading.Thread(target=capture_server, args=args)
        t.start()
        try:
            # wait a little longer for the server to initialize (it sometimes
            # refuses connections on slow machines without this wait)
            time.sleep(0.2)

            data = b"Suppose there isn't a 16-ton weight?"
            d = dispatcherwithsend_noread()
            d.create_socket()
            d.connect((support.HOST, port))

            # give time for socket to connect
            time.sleep(0.1)

            d.send(data)
            d.send(data)
            d.send(b'\n')

            n = 1000
            while d.out_buffer and n > 0:
                asyncore.poll()
                n -= 1

            evt.wait()

            self.assertEqual(cap.getvalue(), data*2)
        finally:
            t.join(timeout=TIMEOUT)
            if t.is_alive():
                self.fail("join() timed out")

def debugging_server(serv, serv_evt, client_evt):
    serv_evt.set()

    try:
        if hasattr(select, 'poll'):
            poll_fun = asyncore.poll2
        else:
            poll_fun = asyncore.poll

        n = 1000
        while asyncore.socket_map and n > 0:
            poll_fun(0.01, asyncore.socket_map)

            # when the client conversation is finished, it will
            # set client_evt, and it's then ok to kill the server
            if client_evt.is_set():
                serv.close()
                break

            n -= 1

    except socket.timeout:
        pass
    finally:
        if not client_evt.is_set():
            # allow some time for the client to read the result
            time.sleep(0.5)
            serv.close()
        asyncore.close_all()
        serv_evt.set()

def test_readwriteexc(self):
        # Check exception handling behavior of read, write and _exception

        # check that ExitNow exceptions in the object handler method
        # bubbles all the way up through asyncore read/write/_exception calls
        tr1 = exitingdummy()
        self.assertRaises(asyncore.ExitNow, asyncore.read, tr1)
        self.assertRaises(asyncore.ExitNow, asyncore.write, tr1)
        self.assertRaises(asyncore.ExitNow, asyncore._exception, tr1)

        # check that an exception other than ExitNow in the object handler
        # method causes the handle_error method to get called
        tr2 = crashingdummy()
        asyncore.read(tr2)
        self.assertEqual(tr2.error_handled, True)

        tr2 = crashingdummy()
        asyncore.write(tr2)
        self.assertEqual(tr2.error_handled, True)

        tr2 = crashingdummy()
        asyncore._exception(tr2)
        self.assertEqual(tr2.error_handled, True)

    # asyncore.readwrite uses constants in the select module that
    # are not present in Windows systems (see this thread:
    # http://mail.python.org/pipermail/python-list/2001-October/109973.html)
    # These constants should be present as long as poll is available

def test_send(self):
        evt = threading.Event()
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(3)
        port = test_support.bind_port(sock)

        cap = StringIO()
        args = (evt, cap, sock)
        t = threading.Thread(target=capture_server, args=args)
        t.start()
        try:
            # wait a little longer for the server to initialize (it sometimes
            # refuses connections on slow machines without this wait)
            time.sleep(0.2)

            data = "Suppose there isn't a 16-ton weight?"
            d = dispatcherwithsend_noread()
            d.create_socket(socket.AF_INET, socket.SOCK_STREAM)
            d.connect((HOST, port))

            # give time for socket to connect
            time.sleep(0.1)

            d.send(data)
            d.send(data)
            d.send('\n')

            n = 1000
            while d.out_buffer and n > 0:
                asyncore.poll()
                n -= 1

            evt.wait()

            self.assertEqual(cap.getvalue(), data*2)
        finally:
            t.join()

def debugging_server(serv, serv_evt, client_evt):
    serv_evt.set()

    try:
        if hasattr(select, 'poll'):
            poll_fun = asyncore.poll2
        else:
            poll_fun = asyncore.poll

        n = 1000
        while asyncore.socket_map and n > 0:
            poll_fun(0.01, asyncore.socket_map)

            # when the client conversation is finished, it will
            # set client_evt, and it's then ok to kill the server
            if client_evt.is_set():
                serv.close()
                break

            n -= 1

    except socket.timeout:
        pass
    finally:
        if not client_evt.is_set():
            # allow some time for the client to read the result
            time.sleep(0.5)
            serv.close()
        asyncore.close_all()
        serv_evt.set()

def test_readwriteexc(self):
        # Check exception handling behavior of read, write and _exception

        # check that ExitNow exceptions in the object handler method
        # bubbles all the way up through asyncore read/write/_exception calls
        tr1 = exitingdummy()
        self.assertRaises(asyncore.ExitNow, asyncore.read, tr1)
        self.assertRaises(asyncore.ExitNow, asyncore.write, tr1)
        self.assertRaises(asyncore.ExitNow, asyncore._exception, tr1)

        # check that an exception other than ExitNow in the object handler
        # method causes the handle_error method to get called
        tr2 = crashingdummy()
        asyncore.read(tr2)
        self.assertEqual(tr2.error_handled, True)

        tr2 = crashingdummy()
        asyncore.write(tr2)
        self.assertEqual(tr2.error_handled, True)

        tr2 = crashingdummy()
        asyncore._exception(tr2)
        self.assertEqual(tr2.error_handled, True)

    # asyncore.readwrite uses constants in the select module that
    # are not present in Windows systems (see this thread:
    # http://mail.python.org/pipermail/python-list/2001-October/109973.html)
    # These constants should be present as long as poll is available

def test_send(self):
        evt = threading.Event()
        sock = socket.socket()
        sock.settimeout(3)
        port = support.bind_port(sock)

        cap = BytesIO()
        args = (evt, cap, sock)
        t = threading.Thread(target=capture_server, args=args)
        t.start()
        try:
            # wait a little longer for the server to initialize (it sometimes
            # refuses connections on slow machines without this wait)
            time.sleep(0.2)

            data = b"Suppose there isn't a 16-ton weight?"
            d = dispatcherwithsend_noread()
            d.create_socket()
            d.connect((support.HOST, port))

            # give time for socket to connect
            time.sleep(0.1)

            d.send(data)
            d.send(data)
            d.send(b'\n')

            n = 1000
            while d.out_buffer and n > 0:
                asyncore.poll()
                n -= 1

            evt.wait()

            self.assertEqual(cap.getvalue(), data*2)
        finally:
            t.join(timeout=TIMEOUT)
            if t.is_alive():
                self.fail("join() timed out")

def lifetime_loop():
    # The main loop. Stay in here until we need to shutdown
    map = asyncore.socket_map
    timeout = 30.0
    while map and _shutdown_phase == 0:
        asyncore.poll(timeout, map)

def graceful_shutdown_loop():
    # The shutdown loop. Allow various services to shutdown gradually.
    global _shutdown_phase
    timestamp = time.time()
    timeout = 1.0
    map = asyncore.socket_map
    while map and _shutdown_phase < 4:
        time_in_this_phase = time.time() - timestamp
        veto = 0
        for fd, obj in map.items():
            try:
                fn = getattr(obj, 'clean_shutdown_control')
            except AttributeError:
                pass
            else:
                try:
                    veto = veto or fn(_shutdown_phase, time_in_this_phase)
                except:
                    obj.handle_error()
        if veto and time_in_this_phase < _shutdown_timeout:
            # Any open socket handler can veto moving on to the next shutdown
            # phase.  (but not forever)
            asyncore.poll(timeout, map)
        else:
            # No vetos? That is one step closer to shutting down
            _shutdown_phase += 1
            timestamp = time.time()

def go(self, timeout=30.0, granularity=15):
        global socket_map
        last_event_check = 0
        while socket_map:
            now = int(time.time())
            if (now - last_event_check) >= granularity:
                last_event_check = now
                fired = []
                # yuck. i want my lisp.
                i = j = 0
                while i < len(self.events):
                    when, what = self.events[i]
                    if now >= when:
                        fired.append(what)
                        j = i + 1
                    else:
                        break
                    i = i + 1
                if fired:
                    self.events = self.events[j:]
                    for what in fired:
                        what(self, now)
                        # sample the number of channels
            n = len(asyncore.socket_map)
            self.num_channels = n
            if n > self.max_channels:
                self.max_channels = n
            asyncore.poll(timeout)

def run(self):
        while self.continue_running:
            asyncore.poll()

def run_loop():
  global must_exit
  global in_pipe
  global out_pipe
  global needs_flush
  global flush_pipes
  global last_activity
  winmm = None

  # increase the windows timer resolution to 1ms
  if platform.system() == "Windows":
    try:
      import ctypes
      winmm = ctypes.WinDLL('winmm')
      winmm.timeBeginPeriod(1)
    except:
      pass

  last_activity = time.clock()
  last_check = time.clock()
  # disable gc to avoid pauses during traffic shaping/proxying
  gc.disable()
  while not must_exit:
    # Tick every 1ms if traffic-shaping is enabled and we have data or are doing background dns lookups, every 1 second otherwise
    lock.acquire()
    tick_interval = 0.001
    if background_activity_count == 0:
      if in_pipe.next_message is None and in_pipe.queue.empty() and out_pipe.next_message is None and out_pipe.queue.empty():
        tick_interval = 1.0
      elif in_pipe.kbps == .0 and in_pipe.latency == 0 and out_pipe.kbps == .0 and out_pipe.latency == 0:
        tick_interval = 1.0
    lock.release()
    asyncore.poll(tick_interval, asyncore.socket_map)
    if needs_flush:
      flush_pipes = True
      needs_flush = False
    out_pipe.tick()
    in_pipe.tick()
    if flush_pipes:
      PrintMessage('OK')
      flush_pipes = False
    # Every 500 ms check to see if it is a good time to do a gc
    now = time.clock()
    if now - last_check > 0.5:
      last_check = now
      # manually gc after 5 seconds of idle
      if now - last_activity >= 5:
        last_activity = now
        logging.debug("Triggering manual GC")
        gc.collect()

  if winmm is not None:
    winmm.timeEndPeriod(1)

def event_loop(catch_signals = (signal.SIGINT, signal.SIGTERM)):
  """
  Replacement for asyncore.loop(), adding timer and signal support.
  """
  old_signal_handlers = {}
  while True:
    save_sigs = len(old_signal_handlers) == 0
    try:
      for sig in catch_signals:
        old = signal.signal(sig, _raiseExitNow)
        if save_sigs:
          old_signal_handlers[sig] = old
      while asyncore.socket_map or timer_queue:
        t = timer.seconds_until_wakeup()
        if debug_event_timing:
          logger.debug("Dismissing to asyncore.poll(), t = %s, q = %r", t, timer_queue)
        asyncore.poll(t, asyncore.socket_map)
        timer.runq()
        if timer.gc_debug:
          gc.collect()
          if gc.garbage:
            for i in gc.garbage:
              logger.debug("GC-cycle %r", i)
            del gc.garbage[:]
    except ExitNow:
      break
    except SystemExit:
      raise
    except ValueError, e:
      if str(e) == "filedescriptor out of range in select()":
        logger.error("Something is badly wrong, select() thinks we gave it a bad file descriptor.")
        logger.error("Content of asyncore.socket_map:")
        for fd in sorted(asyncore.socket_map.iterkeys()):
          logger.error("  fd %s obj %r", fd, asyncore.socket_map[fd])
        logger.error("Not safe to continue due to risk of spin loop on select().  Exiting.")
        sys.exit(1)
      logger.exception("event_loop() exited with exception %r, this is not supposed to happen, restarting")
    except Exception, e:
      logger.exception("event_loop() exited with exception %r, this is not supposed to happen, restarting")
    else:
      break
    finally:
      for sig in old_signal_handlers:
        signal.signal(sig, old_signal_handlers[sig])

def event_loop(catch_signals = (signal.SIGINT, signal.SIGTERM)):
  """
  Replacement for asyncore.loop(), adding timer and signal support.
  """
  old_signal_handlers = {}
  while True:
    save_sigs = len(old_signal_handlers) == 0
    try:
      for sig in catch_signals:
        old = signal.signal(sig, _raiseExitNow)
        if save_sigs:
          old_signal_handlers[sig] = old
      while asyncore.socket_map or timer_queue:
        t = timer.seconds_until_wakeup()
        if debug_event_timing:
          logger.debug("Dismissing to asyncore.poll(), t = %s, q = %r", t, timer_queue)
        asyncore.poll(t, asyncore.socket_map)
        timer.runq()
        if timer.gc_debug:
          gc.collect()
          if gc.garbage:
            for i in gc.garbage:
              logger.debug("GC-cycle %r", i)
            del gc.garbage[:]
    except ExitNow:
      break
    except SystemExit:
      raise
    except ValueError, e:
      if str(e) == "filedescriptor out of range in select()":
        logger.error("Something is badly wrong, select() thinks we gave it a bad file descriptor.")
        logger.error("Content of asyncore.socket_map:")
        for fd in sorted(asyncore.socket_map.iterkeys()):
          logger.error("  fd %s obj %r", fd, asyncore.socket_map[fd])
        logger.error("Not safe to continue due to risk of spin loop on select().  Exiting.")
        sys.exit(1)
      logger.exception("event_loop() exited with exception %r, this is not supposed to happen, restarting")
    except Exception, e:
      logger.exception("event_loop() exited with exception %r, this is not supposed to happen, restarting")
    else:
      break
    finally:
      for sig in old_signal_handlers:
        signal.signal(sig, old_signal_handlers[sig])

def event_loop(catch_signals = (signal.SIGINT, signal.SIGTERM)):
  """
  Replacement for asyncore.loop(), adding timer and signal support.
  """
  old_signal_handlers = {}
  while True:
    save_sigs = len(old_signal_handlers) == 0
    try:
      for sig in catch_signals:
        old = signal.signal(sig, _raiseExitNow)
        if save_sigs:
          old_signal_handlers[sig] = old
      while asyncore.socket_map or timer_queue:
        t = timer.seconds_until_wakeup()
        if debug_event_timing:
          logger.debug("Dismissing to asyncore.poll(), t = %s, q = %r", t, timer_queue)
        asyncore.poll(t, asyncore.socket_map)
        timer.runq()
        if timer.gc_debug:
          gc.collect()
          if gc.garbage:
            for i in gc.garbage:
              logger.debug("GC-cycle %r", i)
            del gc.garbage[:]
    except ExitNow:
      break
    except SystemExit:
      raise
    except ValueError, e:
      if str(e) == "filedescriptor out of range in select()":
        logger.error("Something is badly wrong, select() thinks we gave it a bad file descriptor.")
        logger.error("Content of asyncore.socket_map:")
        for fd in sorted(asyncore.socket_map.iterkeys()):
          logger.error("  fd %s obj %r", fd, asyncore.socket_map[fd])
        logger.error("Not safe to continue due to risk of spin loop on select().  Exiting.")
        sys.exit(1)
      logger.exception("event_loop() exited with exception %r, this is not supposed to happen, restarting")
    except Exception, e:
      logger.exception("event_loop() exited with exception %r, this is not supposed to happen, restarting")
    else:
      break
    finally:
      for sig in old_signal_handlers:
        signal.signal(sig, old_signal_handlers[sig])

def event_loop(catch_signals = (signal.SIGINT, signal.SIGTERM)):
  """
  Replacement for asyncore.loop(), adding timer and signal support.
  """
  old_signal_handlers = {}
  while True:
    save_sigs = len(old_signal_handlers) == 0
    try:
      for sig in catch_signals:
        old = signal.signal(sig, _raiseExitNow)
        if save_sigs:
          old_signal_handlers[sig] = old
      while asyncore.socket_map or timer_queue:
        t = timer.seconds_until_wakeup()
        if debug_event_timing:
          logger.debug("Dismissing to asyncore.poll(), t = %s, q = %r", t, timer_queue)
        asyncore.poll(t, asyncore.socket_map)
        timer.runq()
        if timer.gc_debug:
          gc.collect()
          if gc.garbage:
            for i in gc.garbage:
              logger.debug("GC-cycle %r", i)
            del gc.garbage[:]
    except ExitNow:
      break
    except SystemExit:
      raise
    except ValueError, e:
      if str(e) == "filedescriptor out of range in select()":
        logger.error("Something is badly wrong, select() thinks we gave it a bad file descriptor.")
        logger.error("Content of asyncore.socket_map:")
        for fd in sorted(asyncore.socket_map.iterkeys()):
          logger.error("  fd %s obj %r", fd, asyncore.socket_map[fd])
        logger.error("Not safe to continue due to risk of spin loop on select().  Exiting.")
        sys.exit(1)
      logger.exception("event_loop() exited with exception %r, this is not supposed to happen, restarting")
    except Exception, e:
      logger.exception("event_loop() exited with exception %r, this is not supposed to happen, restarting")
    else:
      break
    finally:
      for sig in old_signal_handlers:
        signal.signal(sig, old_signal_handlers[sig])

def event_loop(catch_signals = (signal.SIGINT, signal.SIGTERM)):
  """
  Replacement for asyncore.loop(), adding timer and signal support.
  """
  old_signal_handlers = {}
  while True:
    save_sigs = len(old_signal_handlers) == 0
    try:
      for sig in catch_signals:
        old = signal.signal(sig, _raiseExitNow)
        if save_sigs:
          old_signal_handlers[sig] = old
      while asyncore.socket_map or timer_queue:
        t = timer.seconds_until_wakeup()
        if debug_event_timing:
          logger.debug("Dismissing to asyncore.poll(), t = %s, q = %r", t, timer_queue)
        asyncore.poll(t, asyncore.socket_map)
        timer.runq()
        if timer.gc_debug:
          gc.collect()
          if gc.garbage:
            for i in gc.garbage:
              logger.debug("GC-cycle %r", i)
            del gc.garbage[:]
    except ExitNow:
      break
    except SystemExit:
      raise
    except ValueError, e:
      if str(e) == "filedescriptor out of range in select()":
        logger.error("Something is badly wrong, select() thinks we gave it a bad file descriptor.")
        logger.error("Content of asyncore.socket_map:")
        for fd in sorted(asyncore.socket_map.iterkeys()):
          logger.error("  fd %s obj %r", fd, asyncore.socket_map[fd])
        logger.error("Not safe to continue due to risk of spin loop on select().  Exiting.")
        sys.exit(1)
      logger.exception("event_loop() exited with exception %r, this is not supposed to happen, restarting")
    except Exception, e:
      logger.exception("event_loop() exited with exception %r, this is not supposed to happen, restarting")
    else:
      break
    finally:
      for sig in old_signal_handlers:
        signal.signal(sig, old_signal_handlers[sig])