

def pop_task_from_taskqueue(s3, task_queue, done_queue, due_queue, heartbeat_queue, output_dir, worker_kind):
    '\n    '
    with task_queue as db:
        task = task_queue.get()
        if (task is None):
            return
        _L.info('Got file {name} from task queue'.format(**task.data))
        passed_on_keys = ('job_id', 'file_id', 'name', 'url', 'content_b64', 'commit_sha', 'set_id')
        passed_on_kwargs = {k: task.data.get(k) for k in passed_on_keys}
        passed_on_kwargs['worker_id'] = _worker_id()
        interval = '{} seconds'.format((RUN_REUSE_TIMEOUT.seconds + (RUN_REUSE_TIMEOUT.days * 86400)))
        previous_run = get_completed_file_run(db, task.data.get('file_id'), interval)
        if previous_run:
            (previous_run_id, _, _) = previous_run
            copy_args = (passed_on_kwargs[k] for k in ('job_id', 'commit_sha', 'set_id'))
            passed_on_kwargs['run_id'] = copy_run(db, previous_run_id, *copy_args)
        else:
            passed_on_kwargs['run_id'] = add_run(db)
            due_task_data = dict(task_data=task.data, **passed_on_kwargs)
            due_queue.put(due_task_data, schedule_at=td2str((jobs.JOB_TIMEOUT + DUETASK_DELAY)))
    if previous_run:
        (run_id, state, status) = previous_run
        message = (MAGIC_OK_MESSAGE if status else 'Re-using failed previous run')
        result = dict(message=message, reused_run=run_id, output=state)
    else:
        from . import worker
        work_lock = threading.Lock()
        work_args = (work_lock, heartbeat_queue, worker_kind)
        work_wait = threading.Thread(target=_wait_for_work_lock, args=work_args)
        with work_lock:
            heartbeat_queue.put({
                'worker_id': _worker_id(),
                'worker_kind': worker_kind,
            })
            work_wait.start()
            (source_name, _) = splitext(relpath(passed_on_kwargs['name'], 'sources'))
            result = worker.do_work(s3, passed_on_kwargs['run_id'], source_name, passed_on_kwargs['content_b64'], output_dir)
        work_wait.join()
    done_task_data = dict(result=result, **passed_on_kwargs)
    done_queue.put(done_task_data, expected_at=td2str(timedelta(0)))
    _L.info('Done')
    sleep((WORKER_COOLDOWN.seconds + (WORKER_COOLDOWN.days * 86400)))

dataset/ETHPy150Open openaddresses/machine/openaddr/ci/__init__.py pop_task_from_taskqueue/original