a worker used by :func:`run_parallel`
350781
def _reliure_worker(wnum, Qin, Qout, pipeline, options={}):
    
    #pipeline = get_pipeline()
    logger = logging.getLogger("reliure.run_parallel.worker#%s" % wnum)
    logger.debug("worker created")
    if options is None:
        options = {}
    while True:
        chunk = Qin.get() # get an element (and wait for it if needed)
        logger.debug("Get %s elements to process" % len(chunk))
        res = [output for output in pipeline(chunk, **options)]
        logger.debug("processing done, results len = %s" % len(res))
        Qout.put(res)
        Qin.task_done()
844291
def _worker_fn(samples, batchify_fn, dataset=None):
    """"""
    # pylint: disable=unused-argument
    # it is required that each worker process has to fork a new MXIndexedRecordIO handle
    # preserving dataset as global variable can save tons of overhead and is safe in new process
    global _worker_dataset
    if isinstance(samples[0], (list, tuple)):
        batch = [batchify_fn([_worker_dataset[i] for i in shard]) for shard in samples]
    else:
        batch = batchify_fn([_worker_dataset[i] for i in samples])
    buf = io.BytesIO()
    ForkingPickler(buf, pickle.HIGHEST_PROTOCOL).dump(batch)
    return buf.getvalue()
211697
def _executor(self):
        '''
            
        '''
        if self.EXECUTE_PARALLEL is False:
            executor_path = "batch_requests.concurrent.executor.SequentialExecutor"
            executor_class = import_class(executor_path)
            return executor_class()
        else:
            executor_path = self.CONCURRENT_EXECUTOR
            executor_class = import_class(executor_path)
            return executor_class(self.NUM_WORKERS)
798595
def runner(view, parallel, dirs, config):
    
    def run(fn_name, items):
        setpath.prepend_bcbiopath()
        out = []
        fn, fn_name = (fn_name, fn_name.__name__) if callable(fn_name) else (_get_ipython_fn(fn_name, parallel), fn_name)
        items = [x for x in items if x is not None]
        items = diagnostics.track_parallel(items, fn_name)
        logger.info("ipython: %s" % fn_name)
        if len(items) > 0:
            items = [config_utils.add_cores_to_config(x, parallel["cores_per_job"], parallel) for x in items]
            if "wrapper" in parallel:
                wrap_parallel = {k: v for k, v in parallel.items() if k in set(["fresources"])}
                items = [[fn_name] + parallel.get("wrapper_args", []) + [wrap_parallel] + list(x) for x in items]
            items = zip_args([args for args in items])
            for data in view.map_sync(fn, items, track=False):
                if data:
                    out.extend(unzip_args(data))
        return out
    return run
1250316
def worker_thread(context):
    """"""
    queue = context.task_queue
    parameters = context.worker_parameters

    if parameters.initializer is not None:
        if not run_initializer(parameters.initializer, parameters.initargs):
            context.state = ERROR
            return

    for task in get_next_task(context, parameters.max_tasks):
        execute_next_task(task)
        queue.task_done()
963227
def bulk_invoke(func, args, nargs):
    
    # for comparison, simplest thing that works
    # for i in nargs:
    #    argv = list(args)
    #    argv.append(i)
    #    func.delay(*argv)

    # some variances between cpy and pypy, sniff detect
    for closure in func.delay.func_closure:
        if getattr(closure.cell_contents, 'queue', None):
            ctx = closure.cell_contents
            break
    q = Queue(ctx.queue, connection=connection)
    argv = list(args)
    argv.append(None)
    job = Job.create(
        func, args=argv, connection=connection,
        description="bucket-%s" % func.func_name,
        origin=q.name, status=JobStatus.QUEUED, timeout=ctx.timeout,
        result_ttl=0, ttl=ctx.ttl)

    for n in chunks(nargs, 100):
        job.created_at = datetime.utcnow()
        with connection.pipeline() as pipe:
            for s in n:
                argv[-1] = s
                job._id = unicode(uuid4())
                job.args = argv
                q.enqueue_job(job, pipeline=pipe)
            pipe.execute()
594409
function Parallel () {
  var actions = Array.prototype.slice.call(arguments);
  var tasksLeft;

  function _Parallel() {}

  function start() {
    tasksLeft = actions.length;
  }

  function done() {
    if (--tasksLeft === 0) {
      _Parallel.finish();
    }
  }

  function wait(action) {
    action.on('finish.parallel', function() {
      action.on('finish.parallel', null);
      done();
    });
  }


  _Parallel.enter = function() {
    start();
    for(var i = 0, len = actions.length; i < len; ++i) {
      var a = actions[i];
      if (a.enter) {
        if (a.enter()) {
          wait(a);
        } else {
          done();
        }
      }
    }
  };

  _Parallel.exit = function() {
    start();
    for(var i = actions.length - 1; i >= 0; --i) {
      var a = actions[i];
      if (a.exit) {
        if (a.exit()) {
          wait(a);
        } else {
          done();
        }
      }
    }
  };

  _Parallel.clear = function() {
    for(var i = 0, len = actions.length; i < len; ++i) {
      var a = actions[i];
      a.clear && a.clear();
    }
  }

  _Parallel = Action(_Parallel);
  return _Parallel;
}
94818
def _thread_worker(self):
        """"""
        while self._running:
            # Retrieve next cmd, or block
            packet = self._queue.get(True)
            if isinstance(packet, dict) and QS_CMD in packet:
                try:
                    self._callback_listen(packet)
                except Exception as err:  # pylint: disable=broad-except
                    _LOGGER.error("Exception in callback\nType: %s: %s",
                                  type(err), err)
            self._queue.task_done()
490586
def run_matrix_in_parallel(self, process_data):
        """"""
        worker_data = [{'matrix': entry, 'pipeline': process_data.pipeline,
                        'model': process_data.model, 'options': process_data.options,
                        'hooks': process_data.hooks} for entry in self.matrix
                       if Matrix.can_process_matrix(entry, process_data.options.matrix_tags)]
        output = []
        success = True
        with closing(multiprocessing.Pool(multiprocessing.cpu_count())) as pool:
            for result in pool.map(matrix_worker, worker_data):
                output += result['output']
                if not result['success']:
                    success = False
        return {'success': success, 'output': output}
556930
def _worker_fn(samples, batchify_fn, dataset=None):
    """"""
    # pylint: disable=unused-argument
    # it is required that each worker process has to fork a new MXIndexedRecordIO handle
    # preserving dataset as global variable can save tons of overhead and is safe in new process
    global _worker_dataset
    batch = batchify_fn([_worker_dataset[i] for i in samples])
    buf = io.BytesIO()
    ForkingPickler(buf, pickle.HIGHEST_PROTOCOL).dump(batch)
    return buf.getvalue()
253871
def _parallel_exec(self, hosts):
        '''  '''
        if not hosts:
            return
        p = multiprocessing.Pool(self.forks)
        results = []
        #results = p.map(multiprocessing_runner, hosts) # can't handle keyboard interrupt
        results = p.map_async(multiprocessing_runner, hosts).get(9999999)
        p.close()
        p.join()
        return results
642530
def _run_parallel_process_with_profiling(self, start_path, stop_path, queue, filename):
        
        runctx('Engine._run_parallel_process(self,  start_path, stop_path, queue)', globals(), locals(), filename)
402190
def run_parallel(workflow, n_threads):
    
    scheduler = Scheduler()
    threaded_worker = Queue() >> thread_pool(
        *repeat(worker, n_threads))

    return scheduler.run(threaded_worker, get_workflow(workflow))
1116725
function Worker(vfs) {
    Agent.call(this, {

        // Endpoints for writable streams at meta.stream (and meta.process.stdin)
        write: write,
        end: end,

        // Endpoint for readable stream at meta.stream (and meta.process.{stdout,stderr})
        destroy: destroy,

        // Endpoints for readable streams at options.stream
        onData: onData,
        onEnd: onEnd,

        // Endpoint for writable streams at options.stream
        onClose: onClose,

        // Endpoints for processes at meta.process
        kill: kill,

        // Endpoint for watchers at meta.watcher
        close: closeWatcher,

        // Endpoint for apis at meta.api
        call: call,

        // Endpoints for vfs itself
        subscribe: subscribe,
        unsubscribe: unsubscribe,
        emit: vfs.emit,

        // special vfs-socket api
        ping: ping,

        // Route other calls to the local vfs instance
        resolve:  route("resolve"),
        stat:     route("stat"),
        readfile: route("readfile"),
        readdir:  route("readdir"),
        mkfile:   route("mkfile"),
        mkdir:    route("mkdir"),
        rmfile:   route("rmfile"),
        rmdir:    route("rmdir"),
        rename:   route("rename"),
        copy:     route("copy"),
        symlink:  route("symlink"),
        watch:    route("watch"),
        connect:  route("connect"),
        spawn:    route("spawn"),
        execFile: route("execFile"),
        extend:   route("extend"),
        unextend: route("unextend"),
        use:      route("use")

    });

    var proxyStreams = {};
    var streams = {};
    var watchers = {};
    var processes = {};
    var apis = {};
    var handlers = {};
    var remote = this.remoteApi;

    function subscribe(name, callback) {
        handlers[name] = function (value) {
            remote.onEvent(name, value);
        };
        vfs.on(name, handlers[name], callback);
    }

    function unsubscribe(name, callback) {
        var handler = handlers[name];
        if (!handler) return;
        delete handlers[name];
        vfs.off(name, handler, callback);
    }

    // Resume readable streams that we paused when the channel drains
    // Forward drain events to all the writable proxy streams.
    this.on("drain", function () {
        Object.keys(streams).forEach(function (id) {
            var stream = streams[id];
            if (stream.readable && stream.resume) stream.resume();
        });
        Object.keys(proxyStreams).forEach(function (id) {
            var stream = proxyStreams[id];
            if (stream.writable) stream.emit("drain");
        });
    });

    // Cleanup streams, proxy streams, proxy processes, and proxy apis on disconnect.
    this.on("disconnect", function (err) {
        if (!err) {
            err = new Error("EDISCONNECT: vfs socket disconnected");
            err.code = "EDISCONNECT";
        }
        Object.keys(streams).forEach(function (id) {
            var stream = streams[id];
            stream.emit("close", err);
        });
        Object.keys(proxyStreams).forEach(onClose);
        Object.keys(processes).forEach(function (pid) {
            var process = processes[pid];
            process.kill();
            delete processes[pid];
        });
        Object.keys(watchers).forEach(function (id) {
            var watcher = watchers[id];
            delete watchers[id];
            watcher.close();
        });
    });
    
    function makeStreamProxy(token) {
        var stream = new Stream();
        var id = token.id;
        stream.id = id;
        proxyStreams[id] = stream;
        if (token.hasOwnProperty("readable")) stream.readable = token.readable;
        if (token.hasOwnProperty("writable")) stream.writable = token.writable;

        if (stream.writable) {
            stream.write = function (chunk) {
                return remote.write(id, chunk);
            };
            stream.end = function (chunk) {
                if (chunk) remote.end(id, chunk);
                else remote.end(id);
            };
        }
        if (stream.readable) {
            stream.destroy = function () {
                remote.destroy(id);
            };
        }

        return stream;
    }

    var nextStreamID = 1;
    function storeStream(stream) {
        nextStreamID = nextStreamID++ % 10000;
        while (streams.hasOwnProperty(nextStreamID)) { nextStreamID++ % 10000; }
        var id = nextStreamID;
        streams[id] = stream;
        stream.id = id;
        if (stream.readable) {
            stream.on("data", function (chunk) {
                if (remote.onData(id, chunk) === false) {
                    stream.pause && stream.pause();
                }
            });
            stream.on("end", function () {
                delete streams[id];
                remote.onEnd(id);
            });
        }
        stream.on("close", function () {
            delete streams[id];
            remote.onClose(id);
        });
        var token = {id: id};
        if (stream.hasOwnProperty("readable")) token.readable = stream.readable;
        if (stream.hasOwnProperty("writable")) token.writable = stream.writable;
        return token;
    }

    function storeProcess(process) {
        var pid = process.pid;
        processes[pid] = process;
        process.on("exit", function (code, signal) {
            delete processes[pid];
            remote.onExit(pid, code, signal);
        });
        process.on("close", function () {
            delete processes[pid];
            delete streams[process.stdout.id];
            delete streams[process.stderr.id];
            delete streams[process.stdin.id];
            remote.onProcessClose(pid);
        });

        process.kill = function(code) {
            killtree(pid, code);
        };

        var token = {pid: pid};
        token.stdin = storeStream(process.stdin);
        token.stdout = storeStream(process.stdout);
        token.stderr = storeStream(process.stderr);
        return token;
    }

    function killtree(pid, code) {
        childrenOfPid(pid, function(err, pidlist){
            if (err) {
                console.error(err.stack);
                return;
            }
            pidlist.forEach(function (pid) {
                try {
                    process.kill(pid, code || "SIGKILL");
                } catch(e) {
                    // kill may throw if the pid does not exist.
                }
            });
        });
    }

    function childrenOfPid(pid, callback) {
        execFile("ps", ["-A", "-oppid,pid"], function(code, stdout, stderr) {
            if (code)
                return callback(code + ":" + stderr);

            var parents = {};
            stdout.split("\n").slice(1).forEach(function(line) {
                var col = line.trim().split(/\s+/g);
                (parents[col[0]] || (parents[col[0]] = [])).push(col[1]);
            });

            function search(roots) {
            var res = roots.concat();
            for (var c, i = 0; i < roots.length; i++) {
                if ((c = parents[roots[i]]) && c.length)
                    res.push.apply(res, search(c));
                }
                return res;
            }
            callback(null, search([pid]));
        });
    }

    var nextWatcherID = 1;
    function storeWatcher(watcher) {
        while (watchers.hasOwnProperty(nextWatcherID)) { nextWatcherID++; }
        var id = nextWatcherID;
        watchers[id] = watcher;
        watcher.id = id;
        watcher.on("change", function (event, filename) {
            remote.onChange(id, event, filename);
        });
        var token = {id: id};
        return token;
    }

    function storeApi(api) {
        var name = api.name;
        apis[name] = api;
        var token = { name: name, names: api.names };
        return token;
    }

    // Remote side writing to our local writable streams
    function write(id, chunk) {
        // They want to write to our real stream
        var stream = streams[id];
        if (!stream) return;
        stream.write(chunk);
    }
    function destroy(id) {
        var stream = streams[id];
        if (!stream) return;
        delete streams[id];
        stream.destroy();
        nextStreamID = id;
    }
    function end(id, chunk) {
        var stream = streams[id];
        if (!stream) return;
        delete streams[id];
        if (chunk) stream.end(chunk);
        else stream.end();
        nextStreamID = id;
    }

    function kill(pid, code) {
        var process = processes[pid];
        if (!process) return;
        process.kill(code);
    }

    function closeWatcher(id) {
        var watcher = watchers[id];
        if (!watcher) return;
        delete watchers[id];
        watcher.close();
        nextWatcherID = id;
    }

    function call(name, fnName, args) {
        var api = apis[name];
        if (!api) return;

        // If the last arg is a function, assume it's a callback and process it.
        if (typeof args[args.length - 1] == "function") {
            var callback = args[args.length - 1];
            args[args.length - 1] = function (err, meta) {
                if (err || (meta && typeof meta === "object")) {
                    return processCallback(err, meta, callback);
                }
                callback(err, meta);
            };
        }

        api[fnName].apply(api, args);
    }

    function onData(id, chunk) {
        var stream = proxyStreams[id];
        if (!stream) return;
        stream.emit("data", chunk);
    }
    function onEnd(id) {
        var stream = proxyStreams[id];
        if (!stream) return;
        // TODO: not delete proxy if close is going to be called later.
        // but somehow do delete proxy if close won't be called later.
        delete proxyStreams[id];
        stream.emit("end");
    }
    function onClose(id) {
        var stream = proxyStreams[id];
        if (!stream) return;
        delete proxyStreams[id];
        stream.emit("close");
    }

    // Can be used for keepalive checks.
    function ping(callback) {
        callback();
    }

    function processCallback(err, meta, callback) {
        // Make error objects serializable
        if (err) {
            var nerr = {
                stack: process.pid + ": " + err.stack
            };
            if (err.hasOwnProperty("code")) nerr.code = err.code;
            if (err.hasOwnProperty("message")) nerr.message = err.message;
            return callback(nerr);
        }
        var token = {};
        var keys = Object.keys(meta);
        for (var i = 0, l = keys.length; i < l; i++) {
            var key = keys[i];
            switch (key) {
                case "stream": token.stream = storeStream(meta.stream); break;
                case "process": token.process = storeProcess(meta.process); break;
                case "watcher": token.watcher = storeWatcher(meta.watcher); break;
                case "api": token.api = storeApi(meta.api); break;
                default: token[key] = meta[key]; break;
            }
        }
        // Call the remote callback with the result
        callback(null, token);
    }

    function route(name) {
        return function wrapped(path, options, callback) {
            if (typeof callback !== "function") {
                throw new Error(name + ": callback must be function");
            }
            // Call the real local function, but intercept the callback
            if (options.stream) {
                options.stream = makeStreamProxy(options.stream);
            }
            vfs[name](path, options, function (err, meta) {
                processCallback(err, meta, callback);
            });
        };
    }
}
791258
def runInParallel(*fns):
        
        proc = []
        for fn in fns:
            p = Process(target=fn)
            p.start()
            proc.append(p)
        for p in proc:
            p.join()
503242
def runner(queue, parallel):
    
    def run(fn_name, items):
        logger.info("clusterk: %s" % fn_name)
        assert "wrapper" in parallel, "Clusterk requires bcbio-nextgen-vm wrapper"
        fn = getattr(__import__("{base}.clusterktasks".format(base=parallel["module"]),
                                fromlist=["clusterktasks"]),
                     parallel["wrapper"])
        wrap_parallel = {k: v for k, v in parallel.items() if k in set(["fresources", "pack"])}
        out = []
        for data in [fn(fn_name, queue, parallel.get("wrapper_args"), wrap_parallel, x) for x in items]:
            if data:
                out.extend(data)
        return out
    return run
243374
def _start_new_worker_process(self, server_socket):
        """"""
        from multiprocessing import Process
        p = Process(target=forked_child, args=self._get_child_args(server_socket))
        p.start()
        return p