def worker_pool(popen_arg, num_workers, job_iterable,
        stop_when_jobs_done=True, stop_on_keyboard_interrupt=True,
        popen=None):
    if popen is None:
        popen = subprocess.Popen

    workers = []
    job_ids = []
    worker_fds = {}
    job_iter = iter(job_iterable)

    def start_job(worker=None):
        """
        assign a job to exiting or newly created worker subprocess.

        returns (job_id, worker) or (None, None) when no more jobs
        """
        job_id, job_str = next(job_iter, (None, None))
        if job_str is None:
            return None, None
        job_str = job_str.rstrip(b'\n') + b'\n'
        if not worker:
            worker = popen(
                popen_arg,
                stdin=subprocess.PIPE,
                stdout=subprocess.PIPE,
                )
        worker.stdin.write(job_str)
        worker.stdin.flush()
        return (job_id, worker)

    def assign_jobs():
        """
        start as many jobs as possible given maximum/idle workers
        and available jobs
        """
        while None in job_ids:
            wnum = job_ids.index(None)
            job_ids[wnum], w = start_job(workers[wnum])
            if w is None:
                return

        while len(workers) < num_workers:
            job_id, w = start_job()
            if w is None:
                return
            worker_fds[w.stdout] = len(workers)
            workers.append(w)
            job_ids.append(job_id)

    try:
        assign_jobs()
        while True:
            if all(i is None for i in job_ids):
                if stop_when_jobs_done:
                    return
                new_jobs = yield (None, None, None)
                # require new jobs to be submitted
                job_iter = iter(new_jobs)
                assign_jobs()
                continue

            try:
                readable, _, _ = select.select(worker_fds, [], [])
            except select.error as e:
                if e.args[0] == 10038:
                    # XXX: no many-worker support on windows yet
                    readable = list(worker_fds)[:1]
                else:
                    raise
            except KeyboardInterrupt:
                if stop_on_keyboard_interrupt:
                    return
                raise

            fd = readable[0]
            wnum = worker_fds[fd]
            w = workers[wnum]
            result = w.stdout.readline()
            finished = job_ids[wnum]
            job_ids[wnum], _ = start_job(w)

            new_jobs = yield (job_ids, finished, result)
            if new_jobs:
                job_iter = iter(new_jobs)
                assign_jobs()

    finally:
        for w in workers:
            w.stdin.close()
Decorator for marking evironment dependent filters.  The current
dataset/ETHPy150Open ckan/ckanapi/ckanapi/cli/workers.py/worker_pool--crossover with--dataset/ETHPy150Open IanLewis/kay/kay/lib/jinja2/filters.py/environmentfilter