def _schedule(self):
        """Internal use only.
        """
        while not self._quit:
            # process I/O events
            self._notifier.poll(0)
            self._lock.acquire()
            if not self._scheduled:
                if self._timeouts:
                    timeout = self._timeouts[0][0] - _time()
                else:
                    timeout = None
                self._polling = True
                self._lock.release()
                self._notifier.poll(timeout)
                self._lock.acquire()
                self._polling = False
            if self._timeouts:
                # wake up timed suspends; pollers may timeout slightly
                # earlier, so give a bit of slack
                now = _time() + 0.0001
                while self._timeouts and self._timeouts[0][0] <= now:
                    timeout, cid, alarm_value = heappop(self._timeouts)
                    assert timeout <= now
                    coro = self._coros.get(cid, None)
                    if not coro or coro._timeout != timeout:
                        continue
                    if coro._state not in (AsynCoro._AwaitIO_, AsynCoro._Suspended,
                                           AsynCoro._AwaitMsg_):
                        logger.warning('coro %s/%s is in state %s for resume; ignored',
                                       coro._name, coro._id, coro._state)
                        continue
                    coro._timeout = None
                    self._suspended.discard(cid)
                    self._scheduled.add(cid)
                    coro._state = AsynCoro._Scheduled
                    coro._value = alarm_value
            scheduled = [self._coros[cid] for cid in self._scheduled]
            self._lock.release()

            for coro in scheduled:
                coro._state = AsynCoro._Running
                AsynCoro.__cur_coro = coro

                try:
                    if coro._exceptions:
                        exc = coro._exceptions.pop(0)
                        if exc[0] == GeneratorExit:
                            # assert str(exc[1]) == 'close'
                            coro._generator.close()
                            retval = coro._value
                        else:
                            retval = coro._generator.throw(*exc)
                    else:
                        retval = coro._generator.send(coro._value)
                except:
                    self._lock.acquire()
                    exc = sys.exc_info()
                    if exc[0] == StopIteration:
                        v = exc[1].args
                        if v:
                            if len(v) == 1:
                                coro._value = v[0]
                            else:
                                coro._value = v
                        coro._exceptions = []
                    elif exc[0] == HotSwapException:
                        v = exc[1].args
                        if isinstance(v, tuple) and len(v) == 1 and inspect.isgenerator(v[0]) and \
                           coro._hot_swappable and not coro._callers:
                            try:
                                coro._generator.close()
                            except:
                                logger.warning('closing %s/%s raised exception: %s',
                                               coro._name, coro._id, traceback.format_exc())
                            coro._generator = v[0]
                            coro._name = coro._generator.__name__
                            coro._exceptions = []
                            coro._value = None
                            # coro._msgs is not reset, so new
                            # coroutine can process pending messages
                            coro._state = AsynCoro._Scheduled
                        else:
                            logger.warning('invalid HotSwapException from %s/%s ignored',
                                           coro._name, coro._id)
                        self._lock.release()
                        continue
                    else:
                        coro._exceptions.append(exc)

                    if coro._callers:
                        # return to caller
                        caller = coro._callers.pop(-1)
                        coro._generator = caller[0]
                        if coro._swap_generator and not coro._callers and coro._hot_swappable:
                            coro._exceptions.append((HotSwapException,
                                                     HotSwapException(coro._swap_generator)))
                            coro._swap_generator = None
                            coro._state = AsynCoro._Scheduled
                        elif coro._exceptions:
                            # exception in callee, restore saved value
                            coro._value = caller[1]
                            self._suspended.discard(coro._id)
                            self._scheduled.add(coro._id)
                            coro._state = AsynCoro._Scheduled
                        elif coro._state == AsynCoro._Running:
                            coro._state = AsynCoro._Scheduled
                    else:
                        if coro._exceptions:
                            exc = coro._exceptions[0]
                            assert isinstance(exc, tuple)
                            if len(exc) == 2:
                                exc = ''.join(traceback.format_exception_only(*exc))
                            else:
                                exc = ''.join(traceback.format_exception(*exc))
                            logger.warning('uncaught exception in %s:\n%s', coro, exc)
                            try:
                                coro._generator.close()
                            except:
                                logger.warning('closing %s raised exception: %s',
                                               coro._name, traceback.format_exc())
                        # delete this coro
                        if coro._state not in (AsynCoro._Scheduled, AsynCoro._Running):
                            logger.warning('coro "%s" is in state: %s', coro._name, coro._state)
                        monitors = list(coro._monitors)
                        for monitor in monitors:
                            if monitor._location == self._location:
                                if coro._exceptions:
                                    exc = MonitorException(coro, coro._exceptions[0])
                                else:
                                    exc = MonitorException(coro, (StopIteration, coro._value))
                                if self._coros.get(monitor._id, None) == monitor:
                                    monitor.send(exc)
                                else:
                                    logger.warning('monitor for %s/%s has gone away!',
                                                   coro._name, coro._id)
                                    coro._monitors.discard(monitor)
                            else:
                                # remote monitor; prepare serializable data
                                if coro._exceptions:
                                    exc = coro._exceptions[0][:2]
                                    try:
                                        serialize(exc[1])
                                    except pickle.PicklingError:
                                        # send only the type
                                        exc = (exc[0], type(exc[1].args[0]))
                                    exc = MonitorException(coro, exc)
                                    coro._exceptions = []
                                else:
                                    exc = coro._value
                                    try:
                                        serialize(exc)
                                    except pickle.PicklingError:
                                        exc = type(exc)
                                    exc = MonitorException(coro, (StopIteration, exc))
                                monitor.send(exc)
                        if not coro._monitors or not coro._exceptions:
                            coro._msgs.clear()
                            coro._monitors.clear()
                            coro._exceptions = []
                            if self._coros.pop(coro._id, None) != coro:
                                logger.warning('invalid coro: %s, %s', coro._id, coro._state)
                            if coro._daemon is True:
                                self._daemons -= 1
                        elif coro._monitors:
                            # a (local) monitor can restart it with hot_swap
                            coro._hot_swappable = True
                            coro._exceptions = []
                        coro._state = None
                        coro._generator = None
                        if coro._complete:
                            coro._complete.set()
                        else:
                            coro._complete = 0
                        self._scheduled.discard(coro._id)
                        if len(self._coros) == self._daemons:
                            self._complete.set()
                    self._lock.release()
                else:
                    self._lock.acquire()
                    if coro._state == AsynCoro._Running:
                        coro._state = AsynCoro._Scheduled
                        # if this coroutine is suspended, don't update
                        # the value; when it is resumed, it will be
                        # updated with the 'update' value
                        coro._value = retval

                    if isinstance(retval, types.GeneratorType):
                        # push current generator onto stack and activate
                        # new generator
                        coro._callers.append((coro._generator, coro._value))
                        coro._generator = retval
                        coro._value = None
                    self._lock.release()
            AsynCoro.__cur_coro = None

        self._lock.acquire()
        for coro in self._coros.itervalues():
            logger.debug('terminating Coro %s/%s%s', coro._name, coro._id,
                         ' (daemon)' if coro._daemon else '')
            AsynCoro.__cur_coro = coro
            coro._state = AsynCoro._Running
            while coro._generator:
                try:
                    coro._generator.close()
                except:
                    logger.warning('closing %s raised exception: %s',
                                   coro._generator.__name__, traceback.format_exc())
                if coro._callers:
                    coro._generator, coro._value = coro._callers.pop(-1)
                else:
                    coro._generator = None
            if coro._complete:
                coro._complete.set()
            else:
                coro._complete = 0
        self._scheduled = set()
        self._suspended = set()
        self._timeouts = []
        self._coros = {}
        self._channels = {}
        self.__class__.__instance = None
        self._quit = True
        self._notifier.terminate()
        self._lock.release()
        logger.debug('AsynCoro %sterminated', ('%s ' % self._location) if self._location else '')
        while self._atexit:
            func, fargs, fkwargs = self._atexit.pop()
            try:
                func(*fargs, **fkwargs)
            except:
                logger.warning('running %s failed:', func.__name__)
                logger.warning(traceback.format_exc())
        self._complete.set()
        logger.shutdown()
        time.sleep(0.1)
dataset/ETHPy150Open pgiri/asyncoro/py2/asyncoro/__init__.py/AsynCoro._schedule/original