Python gevent.queue 模块，Empty() 实例源码

def _concurrent_execute(self, context, start_req, parser, pool, pool_size):
        queue = Queue()  # ????

        # ????????????
        for r in start_req:
            queue.put_nowait(r)

        if pool is None:
            pool = GeventPool(pool_size)

        greenlets = []

        while True:
            try:
                req = self._check_req(queue.get(timeout=1))
                if req.parser is None:
                    req.parser = parser
                greenlets.append(pool.spawn(req, context, queue))
            except Empty:
                break

        return [greenlet.get() for greenlet in greenlets]

def _messages_until_block(self, raiden, expiration_block):
        """ Returns the received messages up to the block `expiration_block`.
        """
        current_block = raiden.get_block_number()
        while current_block < expiration_block:
            try:
                response = self.response_queue.get(
                    timeout=DEFAULT_EVENTS_POLL_TIMEOUT,
                )
            except Empty:
                pass
            else:
                if response:
                    yield response

            current_block = raiden.get_block_number()

def _wait_for_ssh(queue, ssh, command, timeout=1, attempts=40):
    """Wait until a successful connection to the ssh endpoint can be made."""
    try:
        host, port = queue.get(timeout=timeout * attempts)
    except g_queue.Empty:
        cli.bad_exit("No SSH endpoint found.")

    for _ in six.moves.range(attempts):
        _LOGGER.debug('Checking SSH endpoint %s:%s', host, port)
        if checkout.connect(host, port):
            run_ssh(host, port, ssh, list(command))
            break  # if run_ssh doesn't end with os.execvp()...

        try:
            host, port = queue.get(timeout=timeout)
            queue.task_done()
        except g_queue.Empty:
            pass

    # Either all the connection attempts failed or we're after run_ssh
    # (not resulting in os.execvp) so let's "clear the queue" so the thread
    # can join
    queue.task_done()

def update_users(self):
        already_updated = set()

        while True:
            # Only update so many at a time
            if len(already_updated) > 10000:
                return

            try:
                user_id, data = self.user_updates.get_nowait()
            except Empty:
                return

            if user_id in already_updated:
                continue

            already_updated.add(user_id)

            try:
                User.update(**data).where(User.user_id == user_id).execute()
            except:
                self.log.exception('Failed to update user %s: ', user_id)

def _run(self):
        """
        Read from the work_queue, process it using an NNTPRequest object.
        """
        # block until we have an event to handle.
        # print "Worker %s ready!" % self
        while not self._exit.is_set():
            # Begin our loop
            try:
                request = self._work_queue.get()
                if request is StopIteration:
                    # during a close() call (defined below) we force
                    # a StopIteration into the queue to force an exit
                    # from a program level
                    return

                if request.is_set():
                    # Process has been aborted or is no longer needed
                    continue

            except StopIteration:
                # Got Exit
                return

            except EmptyQueueException:
                # Nothing available for us
                continue

            # Mark ourselves busy
            self._work_tracker.mark_busy(self)

            # If we reach here, we have a request to process
            request.run(connection=self._connection)

            # Mark ourselves available again
            self._work_tracker.mark_available(self)

        # Ensure our connection is closed before we exit
        self._connection.close()

def close(self):
        """
        closes out any open threads and cleans up NNTPManager
        gracefully.
        """
        while not self._work_queue.empty():
            try:
                self._work_queue.get_nowait()
            except EmptyQueueException:
                # Nothing available for us
                break

        for worker in self._workers:
            # Toggle Exit
            worker._exit.set()
            self._work_queue.put(StopIteration)

        for entry in self._pool:
            entry.close()

        for worker in self._workers:
            logger.info("Waiting for workers to exit.")
            worker.join()

        del self._pool
        del self._workers
        self._workers = []
        self._pool = []

def test_queue2(self):
        """?????size?????get/set????????"""
        _log.info('test_queue2222222222')
        task_queue = Queue(3)
        def worker(name):
            try:
                while True:
                    task = task_queue.get(timeout=1) # decrements queue size by 1
                    print('Worker %s got task %s' % (name, task))
                    gevent.sleep(0)
            except Empty:
                print('Quitting time!')

        def boss():
            """
            Boss will wait to hand out work until a individual worker is
            free since the maxsize of the task queue is 3.
            """

            for i in xrange(1,10):
                task_queue.put(i)
            print('Assigned all work in iteration 1')

            for i in xrange(10,20):
                task_queue.put(i)
            print('Assigned all work in iteration 2')

        gevent.joinall([
            gevent.spawn(boss),
            gevent.spawn(worker, 'steve'),
            gevent.spawn(worker, 'john'),
            gevent.spawn(worker, 'bob'),
        ])

def _send_and_wait_time(self, raiden, recipient, transfer, timeout):
        """ Utility to handle multiple messages for the same hashlock while
        properly handling expiration timeouts.
        """

        current_time = time.time()
        limit_time = current_time + timeout

        raiden.send_async(recipient, transfer)

        while current_time <= limit_time:
            # wait for a response message (not the Ack for the transfer)
            try:
                response = self.response_queue.get(
                    timeout=limit_time - current_time,
                )
            except Empty:
                yield TIMEOUT
                return

            yield response

            current_time = time.time()

        if log.isEnabledFor(logging.DEBUG):
            log.debug(
                'TIMED OUT %s %s',
                self.__class__,
                pex(transfer),
            )

def _send_and_wait_block(self, raiden, recipient, transfer, expiration_block):
        """ Utility to handle multiple messages and timeout on a block number. """
        raiden.send_async(recipient, transfer)

        current_block = raiden.get_block_number()
        while current_block < expiration_block:
            try:
                response = self.response_queue.get(
                    timeout=DEFAULT_EVENTS_POLL_TIMEOUT,
                )
            except Empty:
                pass
            else:
                if response:
                    yield response

            current_block = raiden.get_block_number()

        if log.isEnabledFor(logging.DEBUG):
            log.debug(
                'TIMED OUT ON BLOCK %s %s %s',
                current_block,
                self.__class__,
                pex(transfer),
                block_number=current_block,
            )

        yield TIMEOUT

def ajax_endpoint(environ, start_response):
    # WSGI????????????????????????????HTTP???
    # ??????WSGI?????HTTP?????????????environ——??????
    # HTTP?????dict???start_response——????HTTP??????
    status = '200 OK'
    headers = [
               ('Content-Type', 'application/json')
               ]
    start_response(status, headers)

    while True:
        try:
            datum = data_source.get(timeout=5)
            yield json.dumps(datum) + '\n'
        except Empty: pass

def poll(uid):
    try:
        msg = users[uid].queue.get(timeout=10)
    except queue.Empty:
        msg = []
    return json.dumps(msg)

def test_queue_empty_exception(self):
        from gevent.queue import Empty
        h = self._makeOne()
        h.start()
        ev = self._getEvent()()

        def func():
            ev.set()
            raise Empty()

        call1 = Callback('completion', func, ())
        h.dispatch_callback(call1)
        ev.wait()

def _create_greenlet_worker(self, queue):
        def greenlet_worker():
            while True:
                try:
                    func = queue.get()
                    if func is _STOP:
                        break
                    func()
                except Empty:
                    continue
                except Exception as exc:
                    log.warning("Exception in worker greenlet")
                    log.exception(exc)
        return gevent.spawn(greenlet_worker)

def save_search_result_with_queue(queue):
    while 1:
        try:
            p = queue.get(timeout=0)
        except Empty:
            break

        save_search_result(p, queue)
    print 'stopping crawler...'

def save_search_result_with_queue(queue):
    while 1:
        try:
            p = queue.get(timeout=0)
        except Empty:
            break

        print p, 'page'
        save_search_result(p, queue)
    print 'stopping crawler...'

def worker(n):
    try:
        while True:
            task = tasks.get(timeout=1) # decrements queue size by 1
            print('Worker %s got task %s' % (n, task))
            gevent.sleep(1.5)
    except Empty:
        print('Quitting time!')

def test_queue_empty_exception(self):
        from gevent.queue import Empty
        h = self._makeOne()
        h.start()
        ev = self._getEvent()()

        def func():
            ev.set()
            raise Empty()

        call1 = Callback('completion', func, ())
        h.dispatch_callback(call1)
        ev.wait()

def _create_greenlet_worker(self, queue):
        def greenlet_worker():
            while True:
                try:
                    func = queue.get()
                    if func is _STOP:
                        break
                    func()
                except Empty:
                    continue
                except Exception as exc:
                    log.warning("Exception in worker greenlet")
                    log.exception(exc)
        return gevent.spawn(greenlet_worker)

def _callback_wrapper(self, command_id, command_obj, callback, greenlet):
        assert callable(callback) or callback is None

        if not self._queue.empty():
            try:
                next_cmd_id = self._queue.get_nowait()
                log.d("Starting command id", next_cmd_id, " next in queue in service '{}'".format(self.name))
                self._start(next_cmd_id)
            except queue.Empty:
                pass

        command_obj.state = command.CommandState.finished
        try:
            greenlet.get()
        except BaseException:
            log.exception("Command", "{}({})".format(command_obj.__class__.__name__, command_id), "raised an exception")
            command_obj.state = command.CommandState.failed
            command_obj.exception = greenlet.exception
            if constants.dev:
                raise  # doesnt work
        if isinstance(greenlet.value, gevent.GreenletExit):
            command_obj.state = command.CommandState.stopped
            greenlet.value = None

        if command_id in self._decorators:
            greenlet.value = self._decorators[command_id](greenlet.value)

        log.d(
            "Command id", command_id, "in service '{}'".format(
                self.name), "has finished running with state:", str(
                command_obj.state))

        if callback:
            callback(greenlet.value)

def tracks():
    """Event Source endpoint for search queries."""
    query = request.args.get('q') or 'pink floyd'
    query = query.lower()

    artists = set()
    if Tag.is_tag(query):
        log.info('Query "%s" seems a tag', query)
        artists.update(Tag(name=query).top_artists)
    elif Artist.is_artist(query):
        log.info('Query "%s" seems an artist name', query)
        artists.update(Artist(name=query).similar_artists)
    else:
        log.info('Query "%s" might be tag or artist', query)
        artists.update(Artist(name=query).similar_artists)
        artists.update(Tag(name=query).top_artists)

    def fetch_artist(artist):
        """Fetch helper to run inside greenlet."""
        hit = artist.hit_track
        if hit:
            if not hit.youtube_id:
                log.info("Couldn't find youtube id, skipping track %s", hit)
                return
            hit_queue.put({
                'name': str(hit),
                'youtubeId': hit.youtube_id,
                'thumbnailUrl': hit.thumbnail_url,
            })

    hit_queue = queue.Queue()
    fetch_pool = pool.Pool(10)
    glet = fetch_pool.map_async(fetch_artist, artists)

    def gen():
        """Generate response by yielding to event source."""
        hits = []
        tried = 0
        while glet:
            try:
                hit = hit_queue.get(timeout=1)
            except queue.Empty:
                tried += 1
                if tried >= 30:
                    log.info('Query "%s" took too much, giving up', query)
                    fetch_pool.kill()
                    break
                if glet:
                    continue
                else:
                    break
            hits.append(hit)
            yield ('event: song\ndata: %s\n\n' % json.dumps(hit)).encode('utf-8')
        Track.save_hits(hits)
        yield 'event: finish\ndata: finish\n\n'.encode('utf-8')

    return Response(gen(), mimetype='text/event-stream')

def Call(self,_func, *params):
        if self.stoping:
            return None,"Process ready to stop!"
        ArgsType=[]
        index=0
        args=[]
        for arg in params:
            if arg==None:
                return None,"Parameter cannot be None!"
            elif isinstance(arg,(str,unicode)):
                ArgsType.append(public.STRING)
                args.append(arg)
            elif isinstance(arg,long):
                ArgsType.append(public.LONG)
                args.append(arg)
            elif isinstance(arg,float):
                ArgsType.append(public.DOUBLE)
                args.append(arg)
            elif isinstance(arg,bool):
                ArgsType.append(public.BOOL)
                args.append(arg)
            elif isinstance(arg,int):
                #int?????bool ??,???????????int
                ArgsType.append(public.INT)
                args.append(arg)
            elif isinstance(arg,Bytes):
                ArgsType.append(public.BYTES)
                #?base64
                args.append(arg.b64encode())
            elif isinstance(arg,dict):
                ArgsType.append(public.MAP)
                args.append(arg)
            else:
                return None,"args[%d] [%s] Types not allowed"%(index,type(arg))
            index+=1

        callInfo=CallInfo(_func,args,ArgsType,timeout=5)
        callback=Queue()
        if self.local_client!=None:
            err=self.local_client.Call(callInfo,callback)
        elif self.remote_client!=None:
            err=self.remote_client.Call(callInfo,callback)

        else:
            return None, "rpc service connection failed"

        if err!=None:
            return None,err
        try:
            resultInfo=callback.get()
            return resultInfo.Result, resultInfo.Error
        except Empty:
            return None,"Quitting time!"