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()

                    # To work under windows we must allow the full-copy now!
                    # Because windows does not support forking!
                    pool_full_copy = self._traj.v_full_copy
                    self._traj.v_full_copy = True

                    initializer = _configure_frozen_pool
                    target = _frozen_pool_single_run
                else:
                    # We don't want to pickle the storage service
                    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)

                    # Signal start of progress calculation
                    self._show_progress(n - 1, total_runs)
                    for result in pool_results:
                        n = self._check_result_and_store_references(result, results,
                                                                    n, total_runs)

                    # Everything is done
                    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, so'
                                             'over 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)

                    # Signal start of progress calculation
                    self._show_progress(n - 1, total_runs)
                    for result in scoop_results:
                        n = self._check_result_and_store_references(result, results,
                                                                    n, total_runs)
                finally:
                    if self._freeze_input:
                        self._traj.v_full_copy = scoop_full_copy
            else:
                # If we spawn a single process for each run, we need an additional queue
                # for the results of `runfunc`
                if self._immediate_postproc:
                    maxsize = 0
                else:
                    maxsize = total_runs

                start_result_length = len(results)
                result_queue = multip.Queue(maxsize=maxsize)

                # Create a generator to generate the tasks for multiprocessing
                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  # Evaluates to false if trajectory produces
                # no more single runs
                process_dict = {}  # Dict containing all subprocees

                # For the cap values, we lazily evaluate them
                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  # If True cap warning is emitted
                max_signals = 10  # Maximum number of warnings, after that warnings are
                # no longer signaled

                # Signal start of progress calculation
                self._show_progress(n - 1, total_runs)

                while len(process_dict) not in 0 or keep_running:
                    # First check if some processes did finish their job
                    for pid in compat.listkeys(process_dict):
                        proc = process_dict[pid]

                        # Delete the terminated processes
                        if not proc.is_alive():
                            proc.join()
                            del process_dict[pid]
                            del proc

                    # Check if caps are reached.
                    # Cap is only checked if there is at least one
                    # process working to prevent deadlock.
                    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 one cap value is reached we can skip the rest

                    # If we have less active processes than
                    # self._ncores and there is still
                    # a job to do, add another process
                    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  # Only signal max_signals times
                        except StopIteration:
                            # All simulation runs have been started
                            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)

                    # Get all results from the result queue
                    n = self._get_results_from_queue(result_queue, results, n, total_runs)

                # Finally get all results from the result queue once more and finalize the queue
                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:
            # Finalize the wrapper
            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/operator `>` -> `not in`