python类Empty()的实例源码

def fetch_batch(self):
        """ Fetch a batch of data without waiting"""
        inp, f = self.queue.get()
        nr_input_var = len(inp)
        batched, futures = [[] for _ in range(nr_input_var)], []
        for k in range(nr_input_var):
            batched[k].append(inp[k])
        futures.append(f)
        cnt = 1
        while cnt < self.batch_size:
            try:
                inp, f = self.queue.get_nowait()
                for k in range(nr_input_var):
                    batched[k].append(inp[k])
                futures.append(f)
            except queue.Empty:
                break
            cnt += 1
        return batched, futures

def events(self, start_date, end_date, frequency):
        running = True

        self.clock_engine_thread.start()
        self.quotation_engine_thread.start()

        while running:
            real_dt = datetime.datetime.now()
            while True:
                try:
                    dt, event_type = self.event_queue.get(timeout=1)
                    break
                except Empty:
                    continue

            system_log.debug("real_dt {}, dt {}, event {}", real_dt, dt, event_type)
            yield Event(event_type, real_dt, dt)

def run(self):
                """Run any background task scheduled for execution."""
                while self.__running:
                        try:
                                try:
                                        # A brief timeout here is necessary
                                        # to reduce CPU usage and to ensure
                                        # that shutdown doesn't wait forever
                                        # for a new task to appear.
                                        task, args, kwargs = \
                                            self.__q.get(timeout=.5)
                                except queue.Empty:
                                        continue
                                task(*args, **kwargs)
                                if hasattr(self.__q, "task_done"):
                                        # Task is done; mark it so.
                                        self.__q.task_done()
                        except:
                                self.bus.log("Failure encountered executing "
                                    "background task {0!r}.".format(self),
                                    traceback=True)

def run(self):
                """Run any background task scheduled for execution."""
                while self.__running:
                        try:
                                try:
                                        # A brief timeout here is necessary
                                        # to reduce CPU usage and to ensure
                                        # that shutdown doesn't wait forever
                                        # for a new task to appear.
                                        task, args, kwargs = \
                                            self.__q.get(timeout=.5)
                                except queue.Empty:
                                        continue
                                task(*args, **kwargs)
                                if hasattr(self.__q, "task_done"):
                                        # Task is done; mark it so.
                                        self.__q.task_done()
                                        if self.__q.unfinished_tasks == 0:
                                                self.__keep_busy = False
                        except Exception as e:
                                print("Failure encountered executing "
                                    "background task {0!r}.".format(self))

def run(self):
        while True:
            if self.is_shutdown:
                return

            try:
                while True:
                    run_at, i, task = self._queue.get(block=True, timeout=None)
                    if self.is_shutdown:
                        if task:
                            log.debug("Not executing scheduled task due to Scheduler shutdown")
                        return
                    if run_at <= time.time():
                        self._scheduled_tasks.discard(task)
                        fn, args, kwargs = task
                        kwargs = dict(kwargs)
                        future = self._executor.submit(fn, *args, **kwargs)
                        future.add_done_callback(self._log_if_failed)
                    else:
                        self._queue.put_nowait((run_at, i, task))
                        break
            except Queue.Empty:
                pass

            time.sleep(0.1)

def run(self):
        futures = queue.Queue(maxsize=121)

        self.start_profile()

        for i in range(self.num_queries):
            if i >= 120:
                old_future = futures.get_nowait()
                old_future.result()

            key = "{}-{}".format(self.thread_num, i)
            future = self.run_query(key)
            futures.put_nowait(future)

        while True:
            try:
                futures.get_nowait().result()
            except queue.Empty:
                break

        self.finish_profile

def run(self):
        futures = queue.Queue(maxsize=121)

        self.start_profile()

        for i in range(self.num_queries):
            if i > 0 and i % 120 == 0:
                # clear the existing queue
                while True:
                    try:
                        futures.get_nowait().result()
                    except queue.Empty:
                        break

            key = "{0}-{1}".format(self.thread_num, i)
            future = self.run_query(key)
            futures.put_nowait(future)

        while True:
            try:
                futures.get_nowait().result()
            except queue.Empty:
                break

        self.finish_profile()

def test_run_empty(self, m_count):
        events = [mock.sentinel.event1, mock.sentinel.event2]
        group = mock.sentinel.group
        m_queue = mock.Mock()
        m_queue.empty.return_value = True
        m_queue.get.side_effect = events + [six_queue.Empty()]
        m_handler = mock.Mock()
        m_count.return_value = list(range(5))
        async_handler = h_async.Async(m_handler, mock.Mock(), mock.Mock())

        with mock.patch('time.sleep'):
            async_handler._run(group, m_queue)

        m_handler.assert_has_calls([mock.call(event) for event in events])
        self.assertEqual(len(events), m_handler.call_count)

def test_run_stale(self, m_count):
        events = [mock.sentinel.event1, mock.sentinel.event2]
        group = mock.sentinel.group
        m_queue = mock.Mock()
        m_queue.empty.side_effect = [False, True, True]
        m_queue.get.side_effect = events + [six_queue.Empty()]
        m_handler = mock.Mock()
        m_count.return_value = list(range(5))
        async_handler = h_async.Async(m_handler, mock.Mock(), mock.Mock())

        with mock.patch('time.sleep'):
            async_handler._run(group, m_queue)

        m_handler.assert_called_once_with(mock.sentinel.event2)

def _run(self, group, queue):
        LOG.debug("Asynchronous handler started processing %s", group)
        for _ in itertools.count():
            # NOTE(ivc): this is a mock-friendly replacement for 'while True'
            # to allow more controlled environment for unit-tests (e.g. to
            # avoid tests getting stuck in infinite loops)
            try:
                event = queue.get(timeout=self._grace_period)
            except six_queue.Empty:
                break
            # FIXME(ivc): temporary workaround to skip stale events
            # If K8s updates resource while the handler is processing it,
            # when the handler finishes its work it can fail to update an
            # annotation due to the 'resourceVersion' conflict. K8sClient
            # was updated to allow *new* annotations to be set ignoring
            # 'resourceVersion', but it leads to another problem as the
            # Handler will receive old events (i.e. before annotation is set)
            # and will start processing the event 'from scratch'.
            # It has negative effect on handlers' performance (VIFHandler
            # creates ports only to later delete them and LBaaS handler also
            # produces some excess requests to Neutron, although with lesser
            # impact).
            # Possible solutions (can be combined):
            #  - use K8s ThirdPartyResources to store data/annotations instead
            #    of native K8s resources (assuming Kuryr-K8s will own those
            #    resources and no one else would update them)
            #  - use the resulting 'resourceVersion' received from K8sClient's
            #    'annotate' to provide feedback to Async to skip all events
            #    until that version
            #  - stick to the 'get-or-create' behaviour in handlers and
            #    also introduce cache for long operations
            time.sleep(STALE_PERIOD)
            while not queue.empty():
                event = queue.get()
                if queue.empty():
                    time.sleep(STALE_PERIOD)
            self._handler(event)

def pull_batch_from_queue(self):
        """
self explanatory:  take a rollout from the queue of the thread runner.
"""
        rollout = self.runner.queue.get(timeout=600.0)
        while not rollout.terminal:
            try:
                rollout.extend(self.runner.queue.get_nowait())
            except queue.Empty:
                break
        return rollout

def pull_batch_from_queue(self):
        """
self explanatory:  take a rollout from the queue of the thread runner.
"""
        rollout = self.runner.queue.get(timeout=600.0)
        while not rollout.terminal:
            try:
                rollout.extend(self.runner.queue.get_nowait())
            except queue.Empty:
                break
        return rollout

def _consume_queue(self, terminate_evt):
    """
    This is the main thread function that consumes functions that are
    inside the _queue object. To use, execute self._queue(fn), where fn
    is a function that performs some kind of network IO or otherwise
    benefits from threading and is independent.

    terminate_evt is automatically passed in on thread creation and 
    is a common event for this generation of threads. The threads
    will terminate when the event is set and the queue burns down.

    Returns: void
    """
    interface = self._initialize_interface()

    while not terminate_evt.is_set():
      try:
        fn = self._queue.get(block=True, timeout=0.01)
      except Queue.Empty:
        continue # periodically check if the thread is supposed to die

      fn = partial(fn, interface)

      try:
        self._consume_queue_execution(fn)
      except Exception as err:
        self._error_queue.put(err)

    self._close_interface(interface)

def _check_errors(self):
    try:
      err = self._error_queue.get(block=False) 
      self._error_queue.task_done()
      self.kill_threads()
      raise err
    except Queue.Empty:
      pass

def get_connection(self):    
        with self._lock:
            try:        
                conn = self.pool.get(block=False)
                self.pool.task_done()
            except Queue.Empty:
                conn = self._create_connection()
            finally:
                self.outstanding += 1

        return conn

def reset_pool(self):
        closefn = self._close_function()
        while True:
            if not self.pool.qsize():
                break
            try:
                conn = self.pool.get()
                closefn(conn)
                self.pool.task_done()
            except Queue.Empty:
                break

        with self._lock:
            self.outstanding = 0

def validate(self,
                 proxy_scanner,
                 expected_num=20,
                 queue_timeout=3,
                 val_timeout=5):
        """Target function of validation threads

        Args:
            proxy_scanner: A ProxyScanner object.
            expected_num: Max number of valid proxies to be scanned.
            queue_timeout: Timeout for getting a proxy from the queue.
            val_timeout: An integer passed to `is_valid` as argument `timeout`.
        """
        while self.proxy_num() < expected_num:
            try:
                candidate_proxy = proxy_scanner.proxy_queue.get(
                    timeout=queue_timeout)
            except queue.Empty:
                if proxy_scanner.is_scanning():
                    continue
                else:
                    break
            addr = candidate_proxy['addr']
            protocol = candidate_proxy['protocol']
            ret = self.is_valid(addr, protocol, val_timeout)
            if self.proxy_num() >= expected_num:
                self.logger.info('Enough valid proxies, thread {} exit.'
                                 .format(threading.current_thread().name))
                break
            if ret['valid']:
                self.add_proxy(Proxy(addr, protocol))
                self.logger.info('{} ok, {:.2f}s'.format(addr, ret[
                    'response_time']))
            else:
                self.logger.info('{} invalid, {}'.format(addr, ret['msg']))

def run(self):
        while 1:
            try:
                msg = self.input_q.get(True, 10.0)
                LOG.info("SyncThread: received message %s " % msg)
                self.proc_sync_msg(msg)
            except queue.Empty:
                LOG.debug("SyncThread: Queue timeout")
            except ValueError:
                LOG.error("Error processing sync message")
                break
        LOG.error("SyncThread exiting")
        SyncData.sync_thread_running = False

def stop(self):
        """ Stop queue and remove all connections from pool.
        """
        while True:
            try:
                ctx = self.queue.get_nowait()
                ctx.stop()
            except queue_six.Empty:
                break
        self.queue.queue.clear()
        self.queue = None

def generator(self):
        while not self.closed:
            # Use a blocking get() to ensure there's at least one chunk of
            # data, and stop iteration if the chunk is None, indicating the
            # end of the audio stream.
            chunk = self._buff.get()
            if chunk is None:
                return

            if self.isPause:
                continue

            data = [chunk]

            # Now consume whatever other data's still buffered.
            while True:
                try:
                    chunk = self._buff.get(block=False)
                    if chunk is None:
                        return
                    data.append(chunk)
                except queue.Empty:
                    break

            yield b''.join(data)

    #?? ??