def _clean(self):
        if not self._ongoing_requests:
            return
        with self._ongoing_requests_lock:
            ongoing_requests_uuids = set(six.iterkeys(self._ongoing_requests))
        waiting_requests = {}
        expired_requests = {}
        for request_uuid in ongoing_requests_uuids:
            try:
                request = self._ongoing_requests[request_uuid]
            except __HOLE__:
                # Guess it got removed before we got to it...
                pass
            else:
                if request.expired:
                    expired_requests[request_uuid] = request
                elif request.current_state == pr.WAITING:
                    waiting_requests[request_uuid] = request
        if expired_requests:
            with self._ongoing_requests_lock:
                while expired_requests:
                    request_uuid, request = expired_requests.popitem()
                    if self._handle_expired_request(request):
                        del self._ongoing_requests[request_uuid]
        if waiting_requests:
            finder = self._finder
            new_messages_processed = finder.messages_processed
            last_messages_processed = self._messages_processed['finder']
            if new_messages_processed > last_messages_processed:
                # Some new message got to the finder, so we can see
                # if any new workers match (if no new messages have been
                # processed we might as well not do anything).
                while waiting_requests:
                    _request_uuid, request = waiting_requests.popitem()
                    worker = finder.get_worker_for_task(request.task)
                    if (worker is not None and
                            request.transition_and_log_error(pr.PENDING,
                                                             logger=LOG)):
                        self._publish_request(request, worker)
                self._messages_processed['finder'] = new_messages_processed
dataset/ETHPy150Open openstack/taskflow/taskflow/engines/worker_based/executor.py/WorkerTaskExecutor._clean