def test_coroutine1(self):
     loop = get_event_loop()
     d1 = Future()
     loop.call_later(0.2, d1.set_result, 1)
     a = yield c_summation(d1)
     self.assertEqual(a, 3)
     self.assertEqual(d1.result(), 1)

def wait_fd(fd, read=True):
    '''Wait for an event on file descriptor ``fd``.

    :param fd: file descriptor
    :param read=True: wait for a read event if ``True``, otherwise a wait
        for write event.

    This function must be invoked from a coroutine with parent, therefore
    invoking it from the main greenlet will raise an exception.
    Check how this function is used in the :func:`.psycopg2_wait_callback`
    function.
    '''
    current = greenlet.getcurrent()
    parent = current.parent
    assert parent, '"wait_fd" must be called by greenlet with a parent'
    try:
        fileno = fd.fileno()
    except AttributeError:
        fileno = fd
    loop = get_event_loop()
    future = Future(loop=loop)
    # When the event on fd occurs switch back to the current greenlet
    if read:
        loop.add_reader(fileno, _done_wait_fd, fileno, future, read)
    else:
        loop.add_writer(fileno, _done_wait_fd, fileno, future, read)
    # switch back to parent greenlet
    parent.switch(future)
    return future.result()

 def test_call_at(self):
     loop = get_event_loop()
     d1 = Future()
     d2 = Future()
     c1 = loop.call_at(loop.time()+1, lambda: d1.set_result(loop.time()))
     c2 = loop.call_later(1, lambda: d2.set_result(loop.time()))
     t1, t2 = yield pulsar.multi_async((d1, d2))
     self.assertTrue(t1 <= t2)

 def test_json_with_async_string2(self):
     d = Future()
     astr = wsgi.AsyncString(d)
     response = wsgi.Json({"bla": astr})
     self.assertEqual(len(response.children), 1)
     result = response.render()
     self.assertIsInstance(result, Future)
     d.set_result("ciao")
     result = yield from result
     self.assertEqual(result, json.dumps({"bla": "ciao"}))

 def test_call_soon(self):
     ioloop = get_event_loop()
     tid = yield loop_thread_id(ioloop)
     d = Future()
     callback = lambda: d.set_result(current_thread().ident)
     cbk = ioloop.call_soon(callback)
     self.assertEqual(cbk._callback, callback)
     self.assertEqual(cbk._args, ())
     # we should be able to wait less than a second
     result = yield d
     self.assertEqual(result, tid)

 def test_ping_pong_monitor(self):
     value = yield 3
     self.assertEqual(value, 3)
     try:
         future = Future()
         future.set_exception(ValueError('test'))
         yield future
     except ValueError:
         pass
     pong = yield send('monitor', 'ping')
     self.assertEqual(pong, 'pong')

def check_twisted(deferred, loop):
    """Binding for twisted.

    Added to pulsar asynchronous engine via the :func:`.add_async_binding`
    function.
    """
    if isinstance(deferred, Deferred):
        future = Future(loop=loop)

        deferred.addCallbacks(future.set_result, lambda failure: future.set_exception(failure.value))

        return future

 def _(*args, **kwargs):
     res = callable(*args, **kwargs)
     if isinstance(res, types.GeneratorType):
         res = _inlineCallbacks(None, res, Deferred())
     if isinstance(res, Deferred):
         future = Future()
         res.addCallbacks(
             future.set_result,
             lambda failure: future.set_exception(failure.value))
         future._deferred = res
         return future
     else:
         raise TypeError(
             "Callable %r should return a generator or a twisted Deferred"
             % callable)

 def shutdown(self, wait=True):
     with self._shutdown_lock:
         self._shutdown = True
         self._put(None)
         if wait:
             self._waiter = Future(loop=self._loop)
             return self._waiter

 def test_multi(self):
     d1 = Future()
     d2 = Future()
     d = multi_async([d1, d2, "bla"])
     self.assertFalse(d.done())
     d2.set_result("first")
     self.assertFalse(d.done())
     d1.set_result("second")
     result = yield from d
     self.assertEqual(result, ["second", "first", "bla"])

 def test_call_later(self):
     ioloop = get_event_loop()
     tid = yield loop_thread_id(ioloop)
     d = Future()
     timeout1 = ioloop.call_later(
         20, lambda: d.set_result(current_thread().ident))
     timeout2 = ioloop.call_later(
         10, lambda: d.set_result(current_thread().ident))
     # lets wake the ioloop
     self.assertTrue(has_callback(ioloop, timeout1))
     self.assertTrue(has_callback(ioloop, timeout2))
     timeout1.cancel()
     timeout2.cancel()
     self.assertTrue(timeout1._cancelled)
     self.assertTrue(timeout2._cancelled)
     timeout1 = ioloop.call_later(
         0.1, lambda: d.set_result(current_thread().ident))
     yield d
     self.assertTrue(d.done())
     self.assertEqual(d.result(), tid)
     self.assertFalse(has_callback(ioloop, timeout1))

class GreenPool(AsyncObject):
    """A pool of running greenlets.

    This pool maintains a group of greenlets to perform asynchronous
    tasks via the :meth:`submit` method.
    """
    worker_name = 'exec'

    def __init__(self, max_workers=None, loop=None):
        self._loop = loop or get_event_loop()
        self._max_workers = min(max_workers or _DEFAULT_WORKERS, _MAX_WORKERS)
        self._greenlets = set()
        self._available = set()
        self._queue = deque()
        self._shutdown = False
        self._waiter = None
        self._logger = logging.getLogger('pulsar.greenpool')
        self._shutdown_lock = threading.Lock()
        self.wait = wait

    @property
    def max_workers(self):
        return self._max_workers

    @max_workers.setter
    def max_workers(self, value):
        value = int(value)
        assert value > 0
        self._max_workers = value

    @property
    def in_green_worker(self):
        """True if the current greenlet is a green pool worker
        """
        return isinstance(getcurrent(), GreenletWorker)

    def submit(self, func, *args, **kwargs):
        """Equivalent to ``func(*args, **kwargs)``.

        This method create a new task for function ``func`` and adds it to
        the queue.
        Return a :class:`~asyncio.Future` called back once the task
        has finished.
        """
        with self._shutdown_lock:
            if self._shutdown:
                raise RuntimeError(
                    'cannot schedule new futures after shutdown')
            if self.in_green_worker:
                return wait(func(*args, **kwargs))
            else:
                future = Future(loop=self._loop)
                self._put((future, func, args, kwargs))
                return future

    def shutdown(self, wait=True):
        with self._shutdown_lock:
            self._shutdown = True
            self._put(None)
            if wait:
                self._waiter = Future(loop=self._loop)
                return self._waiter

    def getcurrent(self):
        return getcurrent()

    # INTERNALS
    def _adjust_greenlet_count(self):
        if (not self._shutdown and not self._available and
                len(self._greenlets) < self._max_workers):
            green = GreenletWorker(self._green_run)
            self._greenlets.add(green)
            self.logger.debug('Num greenlets: %d', len(self._greenlets))
            green.switch()
        return self._available

    def _put(self, task):
        # Run in the main greenlet of the evnet-loop thread
        self._queue.appendleft(task)
        self._check_queue()

    def _check_queue(self):
        # Run in the main greenlet of the event-loop thread
        if not self._adjust_greenlet_count():
            self.logger.debug('No greenlet available')
            return self._loop.call_soon(self._check_queue)
        try:
            task = self._queue.pop()
        except IndexError:
            return
        ensure_future(self._green_task(self._available.pop(), task),
                      loop=self._loop)

    async def _green_task(self, green, task):
        # Coroutine executing the in main greenlet
        # This coroutine is executed for every task put into the queue

        while task is not _DONE:
            # switch to the greenlet to start the task
            task = green.switch(task)
#.........这里部分代码省略.........

class GreenPool(AsyncObject):
    '''A pool of running greenlets.

    This pool maintains a group of greenlets to perform asynchronous
    tasks via the :meth:`submit` method.
    '''
    worker_name = 'exec'

    def __init__(self, max_workers=None, loop=None, maxtasks=None):
        self._loop = loop or get_event_loop()
        self._max_workers = min(max_workers or _DEFAULT_WORKERS, _MAX_WORKERS)
        self._greenlets = set()
        self._available = set()
        self._maxtasks = maxtasks
        self._queue = deque()
        self._shutdown = False
        self._waiter = None
        self._shutdown_lock = threading.Lock()

    def submit(self, func, *args, **kwargs):
        '''Equivalent to ``func(*args, **kwargs)``.

        This method create a new task for function ``func`` and adds it to
        the queue.
        Return a :class:`~asyncio.Future` called back once the task
        has finished.
        '''
        with self._shutdown_lock:
            if self._shutdown:
                raise RuntimeError(
                    'cannot schedule new futures after shutdown')
            future = Future(loop=self._loop)
            self._put((future, func, args, kwargs))
            return future

    def shutdown(self, wait=True):
        with self._shutdown_lock:
            self._shutdown = True
            self._put()
            if wait:
                self._waiter = Future(loop=self._loop)
                return self._waiter

    # INTERNALS
    def _adjust_greenlet_count(self):
        if not self._available and len(self._greenlets) < self._max_workers:
            greenlet = GreenletWorker(self._green_run)
            self._greenlets.add(greenlet)
            greenlet.switch()

    def _put(self, task=None):
        # Run in the main greenlet of the evnet-loop thread
        if task:
            self._adjust_greenlet_count()
        self._queue.appendleft(task)
        self._check_queue()

    def _check_queue(self):
        # Run in the main greenlet of the event-loop thread
        if not self._available:
            return
        try:
            task = self._queue.pop()
        except IndexError:
            return
        async(self._green_task(self._available.pop(), task), loop=self._loop)

    def _green_task(self, greenlet, task):
        # Run in the main greenlet of the event-loop thread

        while task is not _DONE:
            # switch to the greenlet to start the task
            task = greenlet.switch(task)

            # if an asynchronous result is returned, yield from
            while is_async(task):
                try:
                    task = yield from task
                except Exception as exc:
                    # This call can return an asynchronous component
                    task = greenlet.throw(exc)

    def _green_run(self):
        # The run method of a worker greenlet
        task = True
        while task:
            greenlet = getcurrent()
            parent = greenlet.parent
            assert parent
            self._available.add(greenlet)
            self._loop.call_soon(self._check_queue)
            task = parent.switch(_DONE)  # switch back to the main execution
            if task:
                # If a new task is available execute it
                # Here we are in the child greenlet
                future, func, args, kwargs = task
                try:
                    result = func(*args, **kwargs)
                except Exception as exc:
                    future.set_exception(exc)
#.........这里部分代码省略.........

 def __init__(self, headers, parser, transport=None):
     self.headers = headers
     self.parser = parser
     self.transport = transport
     self.buffer = b''
     self.on_message_complete = Future()