Python eventlet 模块，sleep() 实例源码

def make_command(command):
    """Create an command from a method signature."""
    # pylint: disable=missing-docstring
    @wraps(command)
    def actualcommand(self, *args, **kwds):
        data = command(self, *args, **kwds)
        name = command.__name__[3:]
        # pylint: disable=protected-access
        signal = '{uuid}{sep}{event}'.format(
            uuid=self._uuid,
            sep=SEPARATOR,
            event=name
        )
        if flask.has_request_context():
            emit(signal, {'data': pack(data)})
        else:
            sio = flask.current_app.extensions['socketio']
            sio.emit(signal, {'data': pack(data)})
        eventlet.sleep()

    return actualcommand

def save(key, value):
    """Store the key value pair.

    Parameters
    ----------
    key : str
        The key to determine where it's stored, you'll need this to load the value later.
    value : object
        The value to store in the cache.

    Returns
    -------
    None

    """
    signal = 'cache_save'
    if flask.has_request_context():
        emit(signal, {'key': pack(key), 'data': pack(value)})
    else:
        sio = flask.current_app.extensions['socketio']
        sio.emit(signal, {'key': pack(key), 'data': pack(value)})
    eventlet.sleep()

def _message(status, content):
    """Send message interface.

    Parameters
    ----------
    status : str
        The type of message
    content : str

    """
    event = 'message.{}'.format(status)
    if flask.has_request_context():
        emit(event, dict(data=pack(content)))
    else:
        sio = flask.current_app.extensions['socketio']
        sio.emit(event, dict(data=pack(content)))
    eventlet.sleep()

def _clean_loop(self):
        '''
        Cleans up garbage in brain so it will run faster.
        '''
        logger.info('Cleaner started')
        fmt = '%Y %b %d %H:%M:%S'
        period_seconds = 2
        obselete_seconds = 10
        while not self._cleaner_stop:
            obselete_pids = []
            long_ago = datetime.now() - timedelta(seconds=obselete_seconds)
            for pid, profile in self.pid_profiles.iteritems():
                last_seen = datetime.strptime(
                    profile['last_seen'], fmt)
                if last_seen <= long_ago and not is_alive(pid):
                    obselete_pids.append(pid)
            if obselete_pids:
                logger.debug('Cleaning obselete pids: %r...' % obselete_pids)
                for obselete_pid in obselete_pids:
                    try:
                        del self.pid_profiles[obselete_pid]
                    except KeyError:
                        pass
            eventlet.sleep(period_seconds)
        logger.info('Cleaner stopped')

def _event_loop(self):
        cls = type(self)
        while self._is_active:
            try:
                evt = cls.events.get(block=False)
            except eventlet.queue.Empty:
                eventlet.sleep(self._loop_period)
                continue
            except Exception as e:
                logger.exception(e)
                continue
            if evt == self._evt_stop:
                continue
            evt_cls = type(evt)
            # locate the handler method
            handlers = cls.handlers.get(evt_cls)
            if not handlers:
                raise Exception('%s did not register event: %s' %
                                (cls.__name__, evt_cls.__name__))
            # invoke the handler method
            for handler in handlers:
                handler(evt)

def stop(self, graceful=False):
        '''Stop any active threads belong to this threadgroup.'''
        # Try to stop all threads gracefully
        self.group.stop(graceful)
        self.group.wait()

        # Wait for link()ed functions (i.e. lock release)
        threads = self.group.threads[:]
        links_done = dict((th, False) for th in threads)

        def mark_done(gt, th):
            links_done[th] = True

        for th in threads:
            th.link(mark_done, th)

        while not all(links_done.values()):
            eventlet.sleep()

def run(self):
        acceptors = []
        for sock in self.sockets:
            gsock = GreenSocket(sock)
            gsock.setblocking(1)
            hfun = partial(self.handle, gsock)
            acceptor = eventlet.spawn(_eventlet_serve, gsock, hfun,
                                      self.worker_connections)

            acceptors.append(acceptor)
            eventlet.sleep(0.0)

        while self.alive:
            self.notify()
            eventlet.sleep(1.0)

        self.notify()
        try:
            with eventlet.Timeout(self.cfg.graceful_timeout) as t:
                [a.kill(eventlet.StopServe()) for a in acceptors]
                [a.wait() for a in acceptors]
        except eventlet.Timeout as te:
            if te != t:
                raise
            [a.kill() for a in acceptors]

def handleSubscriptions(app, socketio):
    eventHandler = SubscriptionEventHandler(app, socketio)
    while True:
        try:
            if app.sessionForSubscriptions is None:
                app.sessionForSubscriptions = openBloombergSession()
                app.allSubscriptions = {}

            event = app.sessionForSubscriptions.nextEvent(500)
            eventHandler.processEvent(event, app.sessionForSubscriptions)
        except Exception as e:
            traceback.print_exc()
            handleBrokenSession(app, e)
            eventlet.sleep(1)
        finally:
            eventlet.sleep()

def run(self):
        acceptors = []
        for sock in self.sockets:
            gsock = GreenSocket(sock)
            gsock.setblocking(1)
            hfun = partial(self.handle, gsock)
            acceptor = eventlet.spawn(_eventlet_serve, gsock, hfun,
                                      self.worker_connections)

            acceptors.append(acceptor)
            eventlet.sleep(0.0)

        while self.alive:
            self.notify()
            eventlet.sleep(1.0)

        self.notify()
        try:
            with eventlet.Timeout(self.cfg.graceful_timeout) as t:
                [a.kill(eventlet.StopServe()) for a in acceptors]
                [a.wait() for a in acceptors]
        except eventlet.Timeout as te:
            if te != t:
                raise
            [a.kill() for a in acceptors]

def run(self):
        acceptors = []
        for sock in self.sockets:
            gsock = GreenSocket(sock)
            gsock.setblocking(1)
            hfun = partial(self.handle, gsock)
            acceptor = eventlet.spawn(_eventlet_serve, gsock, hfun,
                                      self.worker_connections)

            acceptors.append(acceptor)
            eventlet.sleep(0.0)

        while self.alive:
            self.notify()
            eventlet.sleep(1.0)

        self.notify()
        try:
            with eventlet.Timeout(self.cfg.graceful_timeout) as t:
                [a.kill(eventlet.StopServe()) for a in acceptors]
                [a.wait() for a in acceptors]
        except eventlet.Timeout as te:
            if te != t:
                raise
            [a.kill() for a in acceptors]

def run(self):
        acceptors = []
        for sock in self.sockets:
            gsock = GreenSocket(sock)
            gsock.setblocking(1)
            hfun = partial(self.handle, gsock)
            acceptor = eventlet.spawn(_eventlet_serve, gsock, hfun,
                                      self.worker_connections)

            acceptors.append(acceptor)
            eventlet.sleep(0.0)

        while self.alive:
            self.notify()
            eventlet.sleep(1.0)

        self.notify()
        try:
            with eventlet.Timeout(self.cfg.graceful_timeout) as t:
                [a.kill(eventlet.StopServe()) for a in acceptors]
                [a.wait() for a in acceptors]
        except eventlet.Timeout as te:
            if te != t:
                raise
            [a.kill() for a in acceptors]

def wait(self, timeout=None, check_interval=0.01):
        # Instead of a blocking OS call, this version of wait() uses logic
        # borrowed from the eventlet 0.2 processes.Process.wait() method.
        if timeout is not None:
            endtime = time.time() + timeout
        try:
            while True:
                status = self.poll()
                if status is not None:
                    return status
                if timeout is not None and time.time() > endtime:
                    raise TimeoutExpired(self.args, timeout)
                eventlet.sleep(check_interval)
        except OSError as e:
            if e.errno == errno.ECHILD:
                # no child process, this happens if the child process
                # already died and has been cleaned up
                return -1
            else:
                raise

def watch_forever(self):
        """Watch a keyspace forevermore

        This may not exit - if there are errors they're logged (and in case
        they are persistent we pause).
        """

        while True:
            try:
                self.do_tick()
                if self.etcd_elector:
                    self.etcd_elector.wait_until_elected()
                self.do_watch()
            except Exception:
                LOG.exception('%s: etcd threw exception',
                              self.name)
                # In case of a dead etcd causing continuous
                # exceptions, the pause here avoids eating all the
                # CPU
                time.sleep(self.DANGER_PAUSE)

def run(self):
        context = pyudev.Context()
        monitor = pyudev.Monitor.from_netlink(context)
        monitor.filter_by(subsystem='net')
        monitor.start()

        # Initial replay on existing interfaces
        for device in context.list_devices(subsystem='net'):
            self._dev_add(device.sys_name)
            self.devices.add(device.sys_name)

        while True:
            # monitor.poll() is synchronized call, which will block eventlet
            # for certain time. So use select to avoid the stall.
            ready, _, _ = select.select([monitor.fileno()], [], [], timeout=10)
            if ready:
                device = monitor.poll(timeout=1)
                if device.action == 'add':
                    self._dev_add(device.sys_name)
                    self.devices.add(device.sys_name)
                elif device.action == 'remove':
                    self._dev_del(device.sys_name)
                    self.devices.remove(device.sys_name)
            else:
                eventlet.sleep()

def run(self):
        acceptors = []
        for sock in self.sockets:
            gsock = GreenSocket(sock)
            gsock.setblocking(1)
            hfun = partial(self.handle, gsock)
            acceptor = eventlet.spawn(_eventlet_serve, gsock, hfun,
                                      self.worker_connections)

            acceptors.append(acceptor)
            eventlet.sleep(0.0)

        while self.alive:
            self.notify()
            eventlet.sleep(1.0)

        self.notify()
        try:
            with eventlet.Timeout(self.cfg.graceful_timeout) as t:
                [a.kill(eventlet.StopServe()) for a in acceptors]
                [a.wait() for a in acceptors]
        except eventlet.Timeout as te:
            if te != t:
                raise
            [a.kill() for a in acceptors]

def run(self):
        acceptors = []
        for sock in self.sockets:
            gsock = GreenSocket(sock)
            gsock.setblocking(1)
            hfun = partial(self.handle, gsock)
            acceptor = eventlet.spawn(_eventlet_serve, gsock, hfun,
                                      self.worker_connections)

            acceptors.append(acceptor)
            eventlet.sleep(0.0)

        while self.alive:
            self.notify()
            eventlet.sleep(1.0)

        self.notify()
        try:
            with eventlet.Timeout(self.cfg.graceful_timeout) as t:
                [a.kill(eventlet.StopServe()) for a in acceptors]
                [a.wait() for a in acceptors]
        except eventlet.Timeout as te:
            if te != t:
                raise
            [a.kill() for a in acceptors]

def run(self):
        acceptors = []
        for sock in self.sockets:
            gsock = GreenSocket(sock)
            gsock.setblocking(1)
            hfun = partial(self.handle, gsock)
            acceptor = eventlet.spawn(_eventlet_serve, gsock, hfun,
                                      self.worker_connections)

            acceptors.append(acceptor)
            eventlet.sleep(0.0)

        while self.alive:
            self.notify()
            eventlet.sleep(1.0)

        self.notify()
        try:
            with eventlet.Timeout(self.cfg.graceful_timeout) as t:
                [a.kill(eventlet.StopServe()) for a in acceptors]
                [a.wait() for a in acceptors]
        except eventlet.Timeout as te:
            if te != t:
                raise
            [a.kill() for a in acceptors]

def _consume_loop(self):
        new_configs = {}
        while True:
            eventlet.sleep(5)
            client = etcd.Client(self.host, self.port)
            try:
                new_configs = self._readConfig(client)
                target_configs_list = self._detect_target_configs(new_configs)
                for target_config in target_configs_list:
                    uuid = target_config['id']
                    body = target_config['body']
                    version = body['version']
                    result = self.func(body)
                    if result == True:
                        self._writeState(client, uuid, version, self.resource, "ACTIVE")
                    elif result == False:
                        self._writeState(client, uuid, version, self.resource, "FAILED")
            except etcd.EtcdKeyNotFound:
                continue

def _monitor_neighbor(self):
        with hide('running', 'stdout'):
            while True:
                eventlet.sleep(1)
                try:
                    results = {}
                    cmd = 'gobgp -j neighbor -u {0}'.format(self.mgmt_addr)
                    output = local(cmd, capture=True)
                    ret = json.loads(output)
                    for i in range(len(ret)):
                        addr = ret[i]['conf']['remote_ip']
                        state = ret[i]['info']['bgp_state']
                        results[addr] = state
                    change_result_list = self._extract_change_state(results)
                    if change_result_list != []:
                        result_queue.put(change_result_list)
                except:
                    continue

def _monitoring_loop(self):
        while True:
            eventlet.sleep(1)
            if not result_queue.empty():
                change_result_list = result_queue.get()
                for change_result in change_result_list:
                    host_name = change_result['host_name']
                    neighbor_addr = change_result['neighbor_addr']
                    state = change_result['state']
                    if state == "BGP_FSM_ESTABLISHED":
                        status = 'UP'
                    elif state == "BGP_FSM_IDLE" or state == "BGP_FSM_ACTIVE":
                        status = 'DOWN'
                    while True:
                        eventlet.sleep(1)
                        uuid = self._search_uuid(host_name, neighbor_addr)
                        if uuid:
                            print "### Detect: ", host_name, neighbor_addr, uuid, status
                            self._writeState(uuid, self.resource, status)
                            break
                        else:
                            print ("### retry search uuid")

def _consume_loop(self):
        new_configs = {}
        while True:
            eventlet.sleep(5)
            client = etcd.Client(self.host, self.port)
            try:
                new_configs = self._readConfig(client)
                target_configs_list = self._detect_target_configs(new_configs)
                for target_config in target_configs_list:
                    uuid = target_config['id']
                    body = target_config['body']
                    version = body['version']
                    self._regist_neighbor(uuid, body)
                    self.uuids[uuid] = version
            except etcd.EtcdKeyNotFound:
                continue

def run(self):
        acceptors = []
        for sock in self.sockets:
            gsock = GreenSocket(sock)
            gsock.setblocking(1)
            hfun = partial(self.handle, gsock)
            acceptor = eventlet.spawn(_eventlet_serve, gsock, hfun,
                                      self.worker_connections)

            acceptors.append(acceptor)
            eventlet.sleep(0.0)

        while self.alive:
            self.notify()
            eventlet.sleep(1.0)

        self.notify()
        try:
            with eventlet.Timeout(self.cfg.graceful_timeout) as t:
                [a.kill(eventlet.StopServe()) for a in acceptors]
                [a.wait() for a in acceptors]
        except eventlet.Timeout as te:
            if te != t:
                raise
            [a.kill() for a in acceptors]

def run(self):
        acceptors = []
        for sock in self.sockets:
            gsock = GreenSocket(sock)
            gsock.setblocking(1)
            hfun = partial(self.handle, gsock)
            acceptor = eventlet.spawn(_eventlet_serve, gsock, hfun,
                                      self.worker_connections)

            acceptors.append(acceptor)
            eventlet.sleep(0.0)

        while self.alive:
            self.notify()
            eventlet.sleep(1.0)

        self.notify()
        try:
            with eventlet.Timeout(self.cfg.graceful_timeout) as t:
                [a.kill(eventlet.StopServe()) for a in acceptors]
                [a.wait() for a in acceptors]
        except eventlet.Timeout as te:
            if te != t:
                raise
            [a.kill() for a in acceptors]

def test_uri_length_limit(self):
        server = masakari.wsgi.Server("test_uri_length_limit", None,
            host="127.0.0.1", max_url_len=16384)
        server.start()

        uri = "http://127.0.0.1:%d/%s" % (server.port, 10000 * 'x')
        resp = requests.get(uri, proxies={"http": ""})
        eventlet.sleep(0)
        self.assertNotEqual(resp.status_code,
                            requests.codes.REQUEST_URI_TOO_LARGE)

        uri = "http://127.0.0.1:%d/%s" % (server.port, 20000 * 'x')
        resp = requests.get(uri, proxies={"http": ""})
        eventlet.sleep(0)
        self.assertEqual(resp.status_code,
                         requests.codes.REQUEST_URI_TOO_LARGE)
        server.stop()
        server.wait()

def test_schedule(self):
        hub = hubs.get_hub()
        # clean up the runloop, preventing side effects from previous tests
        # on this thread
        if hub.running:
            hub.abort()
            eventlet.sleep(0)
        called = []
        # t = timer.Timer(0, lambda: (called.append(True), hub.abort()))
        # t.schedule()
        # let's have a timer somewhere in the future; make sure abort() still works
        # (for pyevent, its dispatcher() does not exit if there is something scheduled)
        # XXX pyevent handles this, other hubs do not
        # hubs.get_hub().schedule_call_global(10000, lambda: (called.append(True), hub.abort()))
        hubs.get_hub().schedule_call_global(0, lambda: (called.append(True), hub.abort()))
        hub.default_sleep = lambda: 0.0
        hub.switch()
        assert called
        assert not hub.running

def parent(signal_path, pid):
    eventlet.Timeout(5)
    port = None
    while True:
        try:
            contents = open(signal_path, 'rb').read()
            port = int(contents.strip())
            break
        except Exception:
            eventlet.sleep(0.1)
    eventlet.connect(('127.0.0.1', port))
    while True:
        try:
            contents = open(signal_path, 'rb').read()
            result = contents.split()[1]
            break
        except Exception:
            eventlet.sleep(0.1)
    assert result == b'done', repr(result)
    print('pass')

def test_hub_exceptions(self):
        debug.hub_exceptions(True)
        server = eventlet.listen(('0.0.0.0', 0))
        client = eventlet.connect(('127.0.0.1', server.getsockname()[1]))
        client_2, addr = server.accept()

        def hurl(s):
            s.recv(1)
            {}[1]  # keyerror

        with capture_stderr() as fake:
            gt = eventlet.spawn(hurl, client_2)
            eventlet.sleep(0)
            client.send(b' ')
            eventlet.sleep(0)
            # allow the "hurl" greenlet to trigger the KeyError
            # not sure why the extra context switch is needed
            eventlet.sleep(0)
        self.assertRaises(KeyError, gt.wait)
        debug.hub_exceptions(False)
        # look for the KeyError exception in the traceback
        assert 'KeyError: 1' in fake.getvalue(), "Traceback not in:\n" + fake.getvalue()

def test_sleep(self):
        # even if there was an error in the mainloop, the hub should continue
        # to work
        start = time.time()
        eventlet.sleep(DELAY)
        delay = time.time() - start

        assert delay >= DELAY * \
            0.9, 'sleep returned after %s seconds (was scheduled for %s)' % (
                delay, DELAY)

        def fail():
            1 // 0

        hubs.get_hub().schedule_call_global(0, fail)

        start = time.time()
        eventlet.sleep(DELAY)
        delay = time.time() - start

        assert delay >= DELAY * \
            0.9, 'sleep returned after %s seconds (was scheduled for %s)' % (
                delay, DELAY)

def test_raised_multiple_readers(self):
        debug.hub_prevent_multiple_readers(True)

        def handle(sock, addr):
            sock.recv(1)
            sock.sendall(b"a")
            raise eventlet.StopServe()

        listener = eventlet.listen(('127.0.0.1', 0))
        eventlet.spawn(eventlet.serve, listener, handle)

        def reader(s):
            s.recv(1)

        s = eventlet.connect(('127.0.0.1', listener.getsockname()[1]))
        a = eventlet.spawn(reader, s)
        eventlet.sleep(0)
        self.assertRaises(RuntimeError, s.recv, 1)
        s.sendall(b'b')
        a.wait()

def test_pipe(self):
        r, w = os.pipe()
        rf = greenio.GreenPipe(r, 'rb')
        wf = greenio.GreenPipe(w, 'wb', 0)

        def sender(f, content):
            for ch in map(six.int2byte, six.iterbytes(content)):
                eventlet.sleep(0.0001)
                f.write(ch)
            f.close()

        one_line = b"12345\n"
        eventlet.spawn(sender, wf, one_line * 5)
        for i in range(5):
            line = rf.readline()
            eventlet.sleep(0.01)
            self.assertEqual(line, one_line)
        self.assertEqual(rf.readline(), b'')

def test_exhaustion(self):
        waiter = Queue(0)

        def consumer():
            gotten = None
            try:
                gotten = self.pool.get()
            finally:
                waiter.put(gotten)

        eventlet.spawn(consumer)

        one, two, three, four = (
            self.pool.get(), self.pool.get(), self.pool.get(), self.pool.get())
        self.assertEqual(self.pool.free(), 0)

        # Let consumer run; nothing will be in the pool, so he will wait
        eventlet.sleep(0)

        # Wake consumer
        self.pool.put(one)

        # wait for the consumer
        self.assertEqual(waiter.get(), one)

def test_create_contention(self):
        creates = [0]

        def sleep_create():
            creates[0] += 1
            eventlet.sleep()
            return "slept"

        p = pools.Pool(max_size=4, create=sleep_create)

        def do_get():
            x = p.get()
            self.assertEqual(x, "slept")
            p.put(x)

        gp = eventlet.GreenPool()
        for i in six.moves.range(100):
            gp.spawn_n(do_get)
        gp.waitall()
        self.assertEqual(creates[0], 4)

def test_no_leaking(self):
        refs = weakref.WeakKeyDictionary()
        my_local = corolocal.local()

        class X(object):
            pass

        def do_something(i):
            o = X()
            refs[o] = True
            my_local.foo = o

        p = eventlet.GreenPool()
        for i in six.moves.range(100):
            p.spawn(do_something, i)
        p.waitall()
        del p
        gc.collect()
        eventlet.sleep(0)
        gc.collect()
        # at this point all our coros have terminated
        self.assertEqual(len(refs), 1)

def test_recv_into_type():
    # make sure `_recv_loop` returns the correct value when `recv_meth` is of
    # foo_into type (fills a buffer and returns number of bytes, not the data)
    # Using threads like `test_recv_type` above.
    threading = eventlet.patcher.original('threading')
    addr = []

    def server():
        sock = eventlet.listen(('127.0.0.1', 0))
        addr[:] = sock.getsockname()
        eventlet.sleep(0.2)

    server_thread = threading.Thread(target=server)
    server_thread.start()
    eventlet.sleep(0.1)
    sock = eventlet.connect(tuple(addr))
    buf = array.array('B', b' ')
    res = sock.recv_into(buf, 1)
    assert isinstance(res, int)

def handle(ws):
    if ws.path == '/echo':
        while True:
            m = ws.wait()
            if m is None:
                break
            ws.send(m)
    elif ws.path == '/range':
        for i in range(10):
            ws.send("msg %d" % i)
            eventlet.sleep(0.01)
    elif ws.path == '/error':
        # some random socket error that we shouldn't normally get
        raise socket.error(errno.ENOTSOCK)
    else:
        ws.close()

def test_sending_messages_to_websocket_75(self):
        connect = [
            "GET /echo HTTP/1.1",
            "Upgrade: WebSocket",
            "Connection: Upgrade",
            "Host: %s:%s" % self.server_addr,
            "Origin: http://%s:%s" % self.server_addr,
            "WebSocket-Protocol: ws",
        ]
        sock = eventlet.connect(self.server_addr)

        sock.sendall(six.b('\r\n'.join(connect) + '\r\n\r\n'))
        sock.recv(1024)
        sock.sendall(b'\x00hello\xFF')
        result = sock.recv(1024)
        self.assertEqual(result, b'\x00hello\xff')
        sock.sendall(b'\x00start')
        eventlet.sleep(0.001)
        sock.sendall(b' end\xff')
        result = sock.recv(1024)
        self.assertEqual(result, b'\x00start end\xff')
        sock.shutdown(socket.SHUT_RDWR)
        sock.close()
        eventlet.sleep(0.01)

def _test_multiple_waiters(self, exception):
        evt = event.Event()
        results = []

        def wait_on_event(i_am_done):
            evt.wait()
            results.append(True)
            i_am_done.send()
            if exception:
                raise Exception()

        waiters = []
        count = 5
        for i in range(count):
            waiters.append(event.Event())
            eventlet.spawn_n(wait_on_event, waiters[-1])
        eventlet.sleep()  # allow spawns to start executing
        evt.send()

        for w in waiters:
            w.wait()

        self.assertEqual(len(results), count)

def test_semaphore_contention():
    g_mutex = eventlet.Semaphore()
    counts = [0, 0]

    def worker(no):
        while min(counts) < 200:
            with g_mutex:
                counts[no - 1] += 1
                eventlet.sleep(0.001)

    t1 = eventlet.spawn(worker, no=1)
    t2 = eventlet.spawn(worker, no=2)
    eventlet.sleep(0.5)
    t1.kill()
    t2.kill()

    assert abs(counts[0] - counts[1]) < int(min(counts) * 0.1), counts

def chunked_fail_app(environ, start_response):
    """http://rhodesmill.org/brandon/2013/chunked-wsgi/
    """
    headers = [('Content-Type', 'text/plain')]
    start_response('200 OK', headers)

    # We start streaming data just fine.
    yield b"The dwarves of yore made mighty spells,"
    yield b"While hammers fell like ringing bells"

    # Then the back-end fails!
    try:
        1 / 0
    except Exception:
        start_response('500 Error', headers, sys.exc_info())
        return

    # So rest of the response data is not available.
    yield b"In places deep, where dark things sleep,"
    yield b"In hollow halls beneath the fells."

def test_025_accept_errors(self):
        debug.hub_exceptions(True)
        listener = greensocket.socket()
        listener.bind(('localhost', 0))
        # NOT calling listen, to trigger the error
        with capture_stderr() as log:
            self.spawn_server(sock=listener)
            eventlet.sleep(0)  # need to enter server loop
            try:
                eventlet.connect(self.server_addr)
                self.fail("Didn't expect to connect")
            except socket.error as exc:
                self.assertEqual(support.get_errno(exc), errno.ECONNREFUSED)

        log_content = log.getvalue()
        assert 'Invalid argument' in log_content, log_content
        debug.hub_exceptions(False)

def test_ipv6(self):
        try:
            sock = eventlet.listen(('::1', 0), family=socket.AF_INET6)
        except (socket.gaierror, socket.error):  # probably no ipv6
            return
        log = six.StringIO()
        # first thing the server does is try to log the IP it's bound to

        def run_server():
            try:
                wsgi.server(sock=sock, log=log, site=Site())
            except ValueError:
                log.write(b'broken')

        self.spawn_thread(run_server)

        logval = log.getvalue()
        while not logval:
            eventlet.sleep(0.0)
            logval = log.getvalue()
        if 'broked' in logval:
            self.fail('WSGI server raised exception with ipv6 socket')

def test_001_trampoline_timeout(self):
        server_sock = eventlet.listen(('127.0.0.1', 0))
        bound_port = server_sock.getsockname()[1]

        def server(sock):
            client, addr = sock.accept()
            eventlet.sleep(0.1)
        server_evt = eventlet.spawn(server, server_sock)
        eventlet.sleep(0)
        try:
            desc = eventlet.connect(('127.0.0.1', bound_port))
            hubs.trampoline(desc, read=True, write=False, timeout=0.001)
        except eventlet.Timeout:
            pass  # test passed
        else:
            assert False, "Didn't timeout"

        server_evt.wait()
        check_hub()

def test_max_size(self):
        q = eventlet.Queue(2)
        results = []

        def putter(q):
            q.put('a')
            results.append('a')
            q.put('b')
            results.append('b')
            q.put('c')
            results.append('c')

        gt = eventlet.spawn(putter, q)
        eventlet.sleep(0)
        self.assertEqual(results, ['a', 'b'])
        self.assertEqual(q.get(), 'a')
        eventlet.sleep(0)
        self.assertEqual(results, ['a', 'b', 'c'])
        self.assertEqual(q.get(), 'b')
        self.assertEqual(q.get(), 'c')
        gt.wait()

def test_zero_max_size(self):
        q = eventlet.Queue(0)

        def sender(evt, q):
            q.put('hi')
            evt.send('done')

        def receiver(q):
            x = q.get()
            return x

        evt = event.Event()
        gt = eventlet.spawn(sender, evt, q)
        eventlet.sleep(0)
        assert not evt.ready()
        gt2 = eventlet.spawn(receiver, q)
        self.assertEqual(gt2.wait(), 'hi')
        self.assertEqual(evt.wait(), 'done')
        gt.wait()

def test_multiple_waiters(self):
        # tests that multiple waiters get their results back
        q = eventlet.Queue()

        sendings = ['1', '2', '3', '4']
        gts = [eventlet.spawn(q.get) for x in sendings]

        eventlet.sleep(0.01)  # get 'em all waiting

        q.put(sendings[0])
        q.put(sendings[1])
        q.put(sendings[2])
        q.put(sendings[3])
        results = set()
        for i, gt in enumerate(gts):
            results.add(gt.wait())
        self.assertEqual(results, set(sendings))

def test_channel_waiters(self):
        c = eventlet.Queue(0)
        w1 = eventlet.spawn(c.get)
        w2 = eventlet.spawn(c.get)
        w3 = eventlet.spawn(c.get)
        eventlet.sleep(0)
        self.assertEqual(c.getting(), 3)
        s1 = eventlet.spawn(c.put, 1)
        s2 = eventlet.spawn(c.put, 2)
        s3 = eventlet.spawn(c.put, 3)

        s1.wait()
        s2.wait()
        s3.wait()
        self.assertEqual(c.getting(), 0)
        # NOTE: we don't guarantee that waiters are served in order
        results = sorted([w1.wait(), w2.wait(), w3.wait()])
        self.assertEqual(results, [1, 2, 3])

def test_get_nowait_unlock(self):
        hub = hubs.get_hub()
        result = []
        q = eventlet.Queue(0)
        p = eventlet.spawn(q.put, 5)
        assert q.empty(), q
        assert q.full(), q
        eventlet.sleep(0)
        assert q.empty(), q
        assert q.full(), q
        hub.schedule_call_global(0, store_result, result, q.get_nowait)
        eventlet.sleep(0)
        assert q.empty(), q
        assert q.full(), q
        assert result == [5], result
        # TODO add ready to greenthread
        # assert p.ready(), p
        assert p.dead, p
        assert q.empty(), q

    # put_nowait must work from the mainloop

def test_put_nowait_unlock(self):
        hub = hubs.get_hub()
        result = []
        q = eventlet.Queue(0)
        eventlet.spawn(q.get)
        assert q.empty(), q
        assert q.full(), q
        eventlet.sleep(0)
        assert q.empty(), q
        assert q.full(), q
        hub.schedule_call_global(0, store_result, result, q.put_nowait, 10)
        # TODO ready method on greenthread
        # assert not p.ready(), p
        eventlet.sleep(0)
        assert result == [None], result
        # TODO ready method
        # assert p.ready(), p
        assert q.full(), q
        assert q.empty(), q

def handle(ws):
    if ws.path == '/echo':
        while True:
            m = ws.wait()
            if m is None:
                break
            ws.send(m)
    elif ws.path == '/range':
        for i in range(10):
            ws.send("msg %d" % i)
            eventlet.sleep(0.01)
    elif ws.path == '/error':
        # some random socket error that we shouldn't normally get
        raise socket.error(errno.ENOTSOCK)
    else:
        ws.close()

def test_send_recv_13(self):
        connect = [
            "GET /echo HTTP/1.1",
            "Upgrade: websocket",
            "Connection: Upgrade",
            "Host: %s:%s" % self.server_addr,
            "Origin: http://%s:%s" % self.server_addr,
            "Sec-WebSocket-Version: 13",
            "Sec-WebSocket-Key: d9MXuOzlVQ0h+qRllvSCIg==",
        ]
        sock = eventlet.connect(self.server_addr)
        sock.sendall(six.b('\r\n'.join(connect) + '\r\n\r\n'))
        sock.recv(1024)
        ws = websocket.RFC6455WebSocket(sock, {}, client=True)
        ws.send(b'hello')
        assert ws.wait() == b'hello'
        ws.send(b'hello world!\x01')
        ws.send(u'hello world again!')
        assert ws.wait() == b'hello world!\x01'
        assert ws.wait() == u'hello world again!'
        ws.close()
        eventlet.sleep(0.01)