Running pipelines in parallel.
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}
971680
def runner(parallel, config):
    
    def run_parallel(fn_name, items):
        items = [x for x in items if x is not None]
        if len(items) == 0:
            return []
        items = diagnostics.track_parallel(items, fn_name)
        fn, fn_name = (fn_name, fn_name.__name__) if callable(fn_name) else (get_fn(fn_name, parallel), fn_name)
        logger.info("multiprocessing: %s" % fn_name)
        if "wrapper" in parallel:
            wrap_parallel = {k: v for k, v in parallel.items() if k in set(["fresources", "checkpointed"])}
            items = [[fn_name] + parallel.get("wrapper_args", []) + [wrap_parallel] + list(x) for x in items]
        return run_multicore(fn, items, config, parallel=parallel)
    return run_parallel
775883
public function pipe($stages): void
    {
        if ($stages instanceof PipelineInterface) {
            $stages = $stages->stages();
        }

        if (!\is_array($stages)) {
            $this->stages[] = $stages;

            return;
        }

        foreach ($stages as $stage) {
            $this->pipe($stage);
        }
    }
921366
def run(self):
		
		global parallel

		if parallel:
			download_parallel(self.url, self.directory, self.idx, 
							  self.min_file_size, self.max_file_size, self.no_redirects)
		else:
			download(self.url, self.directory, self.idx,  
					 self.min_file_size, self.max_file_size, self.no_redirects)
54223
def _multi_lpop_pipeline(self, pipe, queue, number):
        ''' '''
        pipe.lrange(queue, 0, number - 1)
        pipe.ltrim(queue, number, -1)
611021
public static void pipeline(KProgram program, KProcedure... procedures) {
		if (procedures.length < 2) {
			throw new IllegalArgumentException(MessageKind.E0003.format());
		}

		KProcedure kLastProcedure = procedures[procedures.length - 1];

		KProcedure previous = null;
		boolean isLastPipeline = false;
		for (KProcedure procedure : procedures) {

			if (procedure.equals(kLastProcedure)) {
				isLastPipeline = true;
			}

			if (previous != null) {
				new KPipeline(program, previous, procedure, isLastPipeline);
			}
			previous = procedure;
		}
	}
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()
1152994
@Override
    public void run() {
        synchronized (this) {
            // Save current event pipeline and currently handled event
            // because a SynchronousEventPipeline can be called while
            // handling an event (from another event processor).
            InternalEventPipeline currentPipeline = FeedBackPipelineFilter
                .getAssociatedPipeline();
            EventBase<?> currentEvent = newEventsParent.get();
            try {
                super.run();
            } finally {
                newEventsParent.set(currentEvent);
                FeedBackPipelineFilter.setAssociatedPipeline(currentPipeline);
            }
        }
    }
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
1291617
def process(self, document):
        """"""
        self.logger.info("Processing group of tasks (parallel=%s)", self.get_parallel_mode())
        self.pipeline.data.env_list[2] = {}

        output, shells = [], []
        result = Adapter({'success': True, 'output': []})
        for task_entry in document:
            key, entry = list(task_entry.items())[0]

            if (not self.parallel or key == 'env') and len(shells) > 0:
                result = Adapter(self.process_shells(shells))
                output += result.output
                shells = []
                if not result.success:
                    break

            if key == 'env':
                self.pipeline.data.env_list[2].update(entry)

            elif key in ['shell', 'docker(container)', 'docker(image)', 'python',
                         'packer', 'ansible(simple)']:
                self.prepare_shell_data(shells, key, entry)

        if result.success:
            result = Adapter(self.process_shells(shells))
            output += result.output

        self.event.delegate(result.success)
        return {'success': result.success, 'output': output}
390023
def _multi_rpush_pipeline(self, pipe, queue, values, bulk_size=0):
        ''' 
        '''
        cont = 0
        for value in values:
            pipe.rpush(queue, value)
            if bulk_size != 0 and cont % bulk_size == 0:
                pipe.execute()
413094
async def main():
    """"""
    for c in (1, 2, 4,):
        async with Sublemon(max_concurrency=c) as s:
            start = time.perf_counter()
            await asyncio.gather(one(s), two(s))
            end = time.perf_counter()
            print('Limiting to', c, 'concurrent subprocess(es) took',
                  end-start, 'seconds\n')
453164
def run(items, run_parallel):
    
    to_process = []
    extras = []
    for batch, cur_items in _group_by_batches(items).items():
        if _ready_for_het_analysis(cur_items):
            to_process.append((batch, cur_items))
        else:
            for data in cur_items:
                extras.append([data])
    processed = run_parallel("heterogeneity_estimate", ([xs, b, xs[0]["config"]] for b, xs in to_process))
    return _group_by_sample_and_batch(extras + processed)
1131907
def run(self, tokens):
        """"""
        for fn in self._stack:
            results = []
            for i, token in enumerate(tokens):
                # JS ignores additional arguments to the functions but we
                # force pipeline functions to declare (token, i, tokens)
                # or *args
                result = fn(token, i, tokens)
                if not result:
                    continue
                if isinstance(result, (list, tuple)):  # simulate Array.concat
                    results.extend(result)
                else:
                    results.append(result)
            tokens = results

        return tokens
1169798
def combine_calls_parallel(samples, run_parallel):
    
    batch_groups, extras = _group_by_batches(samples, _has_ensemble)
    out = []
    if batch_groups:
        processed = run_parallel("combine_calls", ((b, xs, xs[0]) for b, xs in batch_groups.items()))
        for batch_id, callinfo in processed:
            for data in batch_groups[batch_id]:
                data["variants"].insert(0, callinfo)
                out.append([data])
    return out + extras
335188
def evaluate(self, sequence, transformations):
        
        result = sequence
        parallel = partial(
            parallelize, processes=self.processes, partition_size=self.partition_size)
        staged = []
        for transform in transformations:
            strategies = transform.execution_strategies or {}
            if ExecutionStrategies.PARALLEL in strategies:
                staged.insert(0, transform.function)
            else:
                if staged:
                    result = parallel(compose(*staged), result)
                    staged = []
                if ExecutionStrategies.PRE_COMPUTE in strategies:
                    result = list(result)
                result = transform.function(result)
        if staged:
            result = parallel(compose(*staged), result)
        return iter(result)
69745
def multi_rpush(self, queue, values, bulk_size=0, transaction=False):
        '''  '''
        # Check that what we receive is iterable
        if hasattr(values, '__iter__'):
            self._multi_rpush_pipeline(self, queue, values, 0)
        else:
            raise ValueError('Expected an iterable')