

def _execute_multiprocessing(self, start_run_idx, results):
    'Performs multiprocessing and signals expansion by postproc'
    n = start_run_idx
    total_runs = len(self._traj)
    expanded_by_postproc = False
    if ((self._wrap_mode == pypetconstants.WRAP_MODE_NONE) or self._storage_service.multiproc_safe):
        self._logger.info('I assume that your storage service is multiprocessing safe.')
    else:
        use_manager = ((self._wrap_mode == pypetconstants.WRAP_MODE_QUEUE) or self._immediate_postproc)
        self._multiproc_wrapper = MultiprocContext(self._traj, self._wrap_mode, full_copy=None, manager=None, use_manager=use_manager, lock=None, queue=None, queue_maxsize=self._queue_maxsize, port=self._url, timeout=self._timeout, gc_interval=self._gc_interval, log_config=self._logging_manager.log_config, log_stdout=self._logging_manager.log_stdout, graceful_exit=self._graceful_exit)
        self._multiproc_wrapper.start()
    try:
        if self._use_pool:
            self._logger.info(('Starting Pool with %d processes' % self._ncores))
            if self._freeze_input:
                self._logger.info('Freezing pool input')
                init_kwargs = self._make_kwargs()
                pool_full_copy = self._traj.v_full_copy
                self._traj.v_full_copy = True
                initializer = _configure_frozen_pool
                target = _frozen_pool_single_run
            else:
                pool_service = self._traj.v_storage_service
                self._traj.v_storage_service = None
                init_kwargs = dict(logging_manager=self._logging_manager, storage_service=pool_service, niceness=self._niceness)
                initializer = _configure_pool
                target = _pool_single_run
            try:
                iterator = self._make_iterator(start_run_idx)
                mpool = multip.Pool(self._ncores, initializer=initializer, initargs=(init_kwargs,))
                pool_results = mpool.imap(target, iterator)
                self._show_progress((n - 1), total_runs)
                for result in pool_results:
                    n = self._check_result_and_store_references(result, results, n, total_runs)
                mpool.close()
                mpool.join()
            finally:
                if self._freeze_input:
                    self._traj.v_full_copy = pool_full_copy
                else:
                    self._traj.v_storage_service = pool_service
            self._logger.info('Pool has joined, will delete it.')
            del mpool
        elif self._use_scoop:
            self._logger.info('Starting SCOOP jobs')
            if self._freeze_input:
                self._logger.info('Freezing SCOOP input')
                if hasattr(_frozen_scoop_single_run, 'kwargs'):
                    self._logger.warning('You already did run an experiment with SCOOP and a frozen input. Frozen input is realized as a shared constant, soover time your memory might get bloated. If you experience trouble, restart your python interpreter and SCOOP.')
                _frozen_scoop_single_run.kwargs = {
                    
                }
                scoop_full_copy = self._traj.v_full_copy
                self._traj.v_full_copy = True
                init_kwargs = self._make_kwargs()
                scoop_rev = ((self.name + '_') + str(time.time()).replace('.', '_'))
                shared.setConst(**{
                    scoop_rev: init_kwargs,
                })
                iterator = self._make_iterator(start_run_idx, copy_data=True, scoop_rev=scoop_rev)
                target = _frozen_scoop_single_run
            else:
                iterator = self._make_iterator(start_run_idx, copy_data=True)
                target = _scoop_single_run
            try:
                if scoop.IS_RUNNING:
                    scoop_results = futures.map(target, iterator, timeout=self._timeout)
                else:
                    self._logger.error('SCOOP is NOT running, I will use Python`s map function. To activate scoop, start your script via `python -m scoop your_script.py`.')
                    scoop_results = map(target, iterator)
                self._show_progress((n - 1), total_runs)
                for result in scoop_results:
                    n = self._check_result_and_store_references(result, results, init_kwargs, total_runs)
            finally:
                if self._freeze_input:
                    self._traj.v_full_copy = scoop_full_copy
        else:
            if self._immediate_postproc:
                maxsize = 0
            else:
                maxsize = total_runs
            start_result_length = len(results)
            result_queue = multip.Queue(maxsize=maxsize)
            iterator = self._make_iterator(start_run_idx, result_queue=result_queue)
            self._logger.info(('Starting multiprocessing with at most %d processes running at the same time.' % self._ncores))
            if self._check_usage:
                self._logger.info(('Monitoring usage statistics. I will not spawn new processes if one of the following cap thresholds is crossed, CPU: %.1f %%, RAM: %.1f %%, Swap: %.1f %%.' % (self._cpu_cap, self._memory_cap[0], self._swap_cap)))
            keep_running = True
            process_dict = {
                
            }
            cpu_usage_func = (lambda : self._estimate_cpu_utilization())
            memory_usage_func = (lambda : self._estimate_memory_utilization(process_dict))
            swap_usage_func = (lambda : psutil.swap_memory().percent)
            signal_cap = True
            max_signals = 10
            self._show_progress((n - 1), total_runs)
            while ((len(process_dict) > 0) or keep_running):
                for pid in compat.listkeys(process_dict):
                    proc = process_dict[pid]
                    if (not proc.is_alive()):
                        proc.join()
                        del process_dict[pid]
                        del proc
                no_cap = True
                if (self._check_usage and (self._ncores > len(process_dict) > 0)):
                    for (cap_name, cap_function, threshold) in (('CPU Cap', cpu_usage_func, self._cpu_cap), ('Memory Cap', memory_usage_func, self._memory_cap[0]), ('Swap Cap', swap_usage_func, self._swap_cap)):
                        cap_value = cap_function()
                        if (cap_value > threshold):
                            no_cap = False
                            if signal_cap:
                                if (cap_name == 'Memory Cap'):
                                    add_on_str = ' [including estimate]'
                                else:
                                    add_on_str = ''
                                self._logger.warning(('Could not start next process immediately [currently running %d process(es)]. %s reached, %.1f%% >= %.1f%%%s.' % (len(process_dict), cap_name, cap_value, threshold, add_on_str)))
                                signal_cap = False
                                max_signals -= 1
                                if (max_signals == 0):
                                    self._logger.warning('Maximum number of cap warnings reached. I will no longer notify about cap violations, but cap values are still applied silently in background.')
                            break
                if ((len(process_dict) < self._ncores) and keep_running and no_cap):
                    try:
                        task = next(iterator)
                        proc = multip.Process(target=_process_single_run, args=(task,))
                        proc.start()
                        process_dict[proc.pid] = proc
                        signal_cap = (max_signals > 0)
                    except StopIteration:
                        keep_running = False
                        if ((self._postproc is not None) and self._immediate_postproc):
                            if (self._wrap_mode == pypetconstants.WRAP_MODE_LOCAL):
                                reference_service = self._traj._storage_service
                                self._traj.v_storage_service = self._storage_service
                            try:
                                self._logger.info('Performing IMMEDIATE POSTPROCESSING.')
                                (keep_running, start_run_idx, new_runs) = self._execute_postproc(results)
                            finally:
                                if (self._wrap_mode == pypetconstants.WRAP_MODE_LOCAL):
                                    self._traj._storage_service = reference_service
                            if keep_running:
                                expanded_by_postproc = True
                                self._logger.info(('IMMEDIATE POSTPROCESSING expanded the trajectory and added %d new runs' % new_runs))
                                n = start_run_idx
                                total_runs = len(self._traj)
                                iterator = self._make_iterator(start_run_idx, result_queue=result_queue)
                        if (not keep_running):
                            self._logger.debug('All simulation runs have been started. No new runs will be started. The simulation will finish after the still active runs completed.')
                else:
                    time.sleep(0.001)
                n = self._get_results_from_queue(result_queue, results, n, total_runs)
            self._get_results_from_queue(result_queue, results, n, total_runs)
            result_queue.close()
            result_queue.join_thread()
            del result_queue
            result_sort(results, start_result_length)
    finally:
        if (self._multiproc_wrapper is not None):
            self._multiproc_wrapper.finalize()
            self._multiproc_wrapper = None
    return expanded_by_postproc

dataset/ETHPy150Open SmokinCaterpillar/pypet/pypet/environment.py Environment._execute_multiprocessing/VarMisuse@697/1244 `n`->`init_kwargs`