Introduce optimisation on multiprocessing

perf.py

+
+
+"""
+Produce performance measurement
+"""
+import argparse
+import multiprocessing
+import os
+
+import time
+from collections import OrderedDict
+
+from sandbox.perf.simulation import ComputeSubject
+from synergine2.config import Config
+from synergine2.cycle import CycleManager
+from synergine2.log import SynergineLogger
+from synergine2.simulation import Simulation, Subjects
+
+
+def simulate(complexity, subject_count, cycle_count, cores):
+    config = Config(dict(complexity=complexity, use_x_cores=cores))
+    simulation = Simulation(config)
+
+    subjects = Subjects(simulation=simulation)
+    for i in range(subject_count):
+        subjects.append(ComputeSubject(
+            config=config,
+            simulation=simulation,
+        ))
+
+    simulation.subjects = subjects
+
+    cycle_manager = CycleManager(
+        config,
+        SynergineLogger('perf'),
+        simulation=simulation,
+    )
+
+    for j in range(cycle_count):
+        cycle_manager.next()
+
+
+def main():
+    parser = argparse.ArgumentParser(description='Perf measures')
+    parser.add_argument(
+        '--max_cores',
+        type=int,
+        default=0,
+        help='number of used cores',
+    )
+
+    args = parser.parse_args()
+
+    host_cores = multiprocessing.cpu_count()
+    retry = 3
+    cycles = 200
+    subject_counts = [1, 10, 100, 1000, 5000]
+    complexities = [100, 2000]
+    max_cores = args.max_cores or host_cores
+
+    results = []
+    datas = OrderedDict()
+
+    for core_i in range(max_cores):
+        # if core_i == 0:
+        #     continue
+        core_count = core_i + 1
+        for subject_count in subject_counts:
+            for complexity in complexities:
+                print('COMPLEXITY: {}, SUBJECTS: {}, CORES: {}'.format(
+                    complexity,
+                    subject_count,
+                    core_count,
+                ), end='')
+
+                durations = []
+                for try_i in range(retry):
+                    start_time = time.time()
+                    simulate(complexity, subject_count, cycles, core_count)
+                    durations.append(time.time() - start_time)
+                duration = min(durations)
+
+                result = {
+                    'cores': core_count,
+                    'complexity': complexity,
+                    'subject_count': subject_count,
+                    'cycles': cycles,
+                    'duration': duration,
+                    'duration_cycle': duration / cycles,
+                    'duration_subj_complex': (duration / cycles) / (subject_count * complexity),
+                }
+                results.append(result)
+
+                print(': {}s, {}s/c, {}s/C'.format(
+                    result['duration'],
+                    result['duration_cycle'],
+                    result['duration_subj_complex'],
+                ))
+                datas.setdefault(complexity, {}).setdefault(subject_count, {})[core_count] = result['duration_cycle']
+
+    for d_complexity, c_values in sorted(datas.items(), key=lambda e: int(e[0])):
+        data_file_name = 'DATA_{}'.format(str(d_complexity))
+        try:
+            os.unlink(data_file_name)
+        except FileNotFoundError:
+            pass
+        with open(data_file_name, 'w+') as file:
+            file.writelines(['# (COMPLEXITY {}) SUBJECTS CORES_{}\n'.format(
+                str(d_complexity),
+                ' CORES_'.join(map(str, range(1, max_cores+1))),
+            )])
+            for d_subject_count, d_cores in c_values.items():
+                line = '{} {}\n'.format(
+                    str(d_subject_count),
+                    ' '.join(map(str, d_cores.values())),
+                )
+                file.writelines([line])
+
+        """
+        subj_core = []
+            for subj, core_v in c_values.items():
+                for core_nb in core_v.keys():
+                    subj_core.append((subj, core_nb))
+            file.writelines(['# (COMPLEXITY {}) SUBJECTS CORES_{}\n'.format(
+                str(d_complexity),
+                ' '.join([
+                    'SUBJ_{}_COR_{}'.format(
+                        subj, core_nb,
+                    ) for subj, core_nb in subj_core
+                ])
+            )])
+        """
+
+    for d_complexity, c_values in datas.items():
+        print('')
+        print('gnuplot -p -e "set title \\"COMPLEXITY_{}\\"; plot {}"'.format(
+            str(d_complexity),
+            ','.join([
+                ' \'DATA_{}\' using 1:{} title \'CORE_{}\' with lines'.format(
+                    d_complexity,
+                    d_core+1,
+                    d_core,
+                ) for d_core in range(1, max_cores+1)
+            ])
+        ))
+
+
+if __name__ == '__main__':
+    main()

sandbox/engulf/behaviour.py

 
         for position in positions_chunks[process_number]:
             arounds = get_around_positions_of_positions(position)
-            from_subject = self.simulation.subjects.grass_xyz[position]
+            from_subject = self.simulation.subjects.grass_xyz[position][0]
             around_data = {
                 'from_subject': from_subject,
                 'around': [],

sandbox/engulf/simulation.py

         if isinstance(value, Cell):
             try:
                 self.cell_xyz.get(value.position, []).remove(value)
+                if not self.cell_xyz[value.position]:
+                    del self.cell_xyz[value.position]
             except ValueError:
                 pass
 

sandbox/life_game/run.py

 
 import logging
 
-from synergine2.config import Config
-from synergine2.log import get_default_logger
-
 synergine2_ath = os.path.abspath(os.path.join(os.path.dirname(os.path.abspath(__file__)), '../../'))
 sys.path.append(synergine2_ath)
 
 import collections
 
+from synergine2.config import Config
+from synergine2.log import get_default_logger
 from sandbox.life_game.simulation import Cell, LotOfCellsSignalBehaviour, LifeGame, \
     EmptyPositionWithLotOfCellAroundEvent
 from sandbox.life_game.simulation import Empty

sandbox/perf/run.py

+from random import seed
+import logging
+import os
+import sys
+
+synergine2_ath = os.path.abspath(os.path.join(os.path.dirname(os.path.abspath(__file__)), '../../'))
+sys.path.append(synergine2_ath)
+
+
+from sandbox.perf.simulation import ComputeSubject
+from synergine2.config import Config
+from synergine2.core import Core
+from synergine2.cycle import CycleManager
+from synergine2.log import get_default_logger
+from synergine2.simulation import Simulation, Subjects
+from synergine2.terminals import TerminalManager
+
+
+def main():
+    seed(42)
+
+    config = Config(dict(complexity=10000))
+    logger = get_default_logger(level=logging.ERROR)
+
+    simulation = Simulation(config)
+    subjects = Subjects(simulation=simulation)
+    subjects.extend([ComputeSubject(
+        config=config,
+        simulation=simulation,
+    ) for i in range(500)])
+    simulation.subjects = subjects
+
+    core = Core(
+        config=config,
+        logger=logger,
+        simulation=simulation,
+        cycle_manager=CycleManager(
+            config=config,
+            logger=logger,
+            simulation=simulation,
+        ),
+        terminal_manager=TerminalManager(
+            config=config,
+            logger=logger,
+            terminals=[]
+        ),
+        cycles_per_seconds=1000000,
+    )
+    core.run()
+
+
+if __name__ == '__main__':
+    main()

sandbox/perf/simulation.py

+import random
+
+from synergine2.simulation import SubjectMechanism, SubjectBehaviour, Event, Subject
+
+
+def compute(complexity: int):
+    random_floats = []
+    result = 0.0
+
+    for i in range(complexity):
+        random_floats.append(random.uniform(0, 100))
+
+    for j, random_float in enumerate(random_floats):
+        if not j % 2:
+            result += random_float
+        else:
+            result -= random_float
+
+    return result
+
+
+class ComputeMechanism(SubjectMechanism):
+    def run(self):
+        complexity = self.config.get('complexity')
+        value = compute(complexity)
+
+        return {
+            'mechanism_value': value,
+            'complexity': complexity,
+        }
+
+
+class ComplexityEvent(Event):
+    def __init__(self, complexity, *args, **kwargs):
+        self.complexity = complexity
+
+
+class ComputeBehaviour(SubjectBehaviour):
+    use = [ComputeMechanism]
+
+    def run(self, data):
+        return not data.get(ComputeMechanism).get('mechanism_value') % 2
+
+    def action(self, data):
+        mechanism_value = data.get(ComputeMechanism).get('mechanism_value')
+        complexity = data.get(ComputeMechanism).get('complexity')
+
+        if not int(str(mechanism_value)[-1]) % 2:
+            compute(complexity)
+
+        return [Event(complexity)]
+
+
+class ComputeSubject(Subject):
+    behaviours_classes = [ComputeBehaviour]

synergine2/core.py

 # coding: utf-8
 import time
 
+from synergine2.base import BaseObject
 from synergine2.config import Config
 from synergine2.cycle import CycleManager
 from synergine2.log import SynergineLogger
 from synergine2.terminals import TerminalPackage
 
 
-class Core(object):
+class Core(BaseObject):
     def __init__(
         self,
         config: Config,

synergine2/cycle.py

 # coding: utf-8
 import multiprocessing
 
+from synergine2.base import BaseObject
 from synergine2.config import Config
 from synergine2.log import SynergineLogger
 from synergine2.processing import ProcessManager
 from synergine2.utils import ChunkManager
 
 
-class CycleManager(object):
+class CycleManager(BaseObject):
     def __init__(
             self,
             config: Config,
     ):
         if process_manager is None:
             process_manager = ProcessManager(
-                process_count=multiprocessing.cpu_count(),
+                process_count=config.get('use_x_cores', multiprocessing.cpu_count()),
                 chunk_manager=ChunkManager(multiprocessing.cpu_count()),
             )
 

synergine2/log.py

 # coding: utf-8
 import logging
-
 import sys
 import typing
 
+STREAM_HANDLER_TAG = '__STREAM_HANDLER__'
+
 
 class SynergineLogger(logging.getLoggerClass()):
     @property
     def is_debug(self) -> bool:
         return self.isEnabledFor(logging.DEBUG)
 
+    def __getstate__(self):
+        # Multiprocessing fail if stream handler present. Remove it if exist.
+        # TODO Bug when want to store STREAM_HANDLER_TAG instead stream handler. str still in handlers after
+        # __setstate__ ...
+        self.handlers = []
+
+        return self.__dict__.copy()
+
+    def __setstate__(self, state):
+        self.__dict__ = state
+        # TODO: This handler is hardcoded, it must depend of real context
+        self.handlers.append(logging.StreamHandler(sys.stdout))
+
 
 def get_default_logger(name: str='synergine', level: int=logging.ERROR) -> SynergineLogger:
     """

synergine2/processing.py

 # coding: utf-8
 import types
-from multiprocessing import Process
-from multiprocessing import Manager
+from multiprocessing import Pool
 
+from synergine2.base import BaseObject
 from synergine2.utils import ChunkManager
 
 
-class ProcessManager(object):
+class ProcessManager(BaseObject):
     def __init__(
             self,
             process_count: int,
     ):
         self._process_count = process_count
         self._chunk_manager = chunk_manager
+        self.pool = Pool(processes=self._process_count)
 
-    def chunk_and_execute_jobs(self, data: list, job_maker: types.FunctionType) -> list:
-        with Manager() as manager:
-            processes = list()
-            chunks = self._chunk_manager.make_chunks(data)
-            results = manager.dict()
-
-            for process_number in range(self._process_count):
-                processes.append(Process(
-                    target=self._job_maker_wrapper,
-                    args=(
-                        process_number,
-                        chunks[process_number],
-                        results,
-                        job_maker,
-                    )
-                ))
+    def __getstate__(self):
+        self_dict = self.__dict__.copy()
+        self_dict['pool'] = None
+        return self_dict
 
-            for process in processes:
-                process.start()
+    def chunk_and_execute_jobs(self, data: list, job_maker: types.FunctionType) -> list:
+        chunks = self._chunk_manager.make_chunks(data)
 
-            for process in processes:
-                process.join()
+        if self._process_count > 1:
+            results = self.pool.starmap(job_maker, [(chunk, i, self._process_count) for i, chunk in enumerate(chunks)])
+        else:
+            results = [job_maker(data, 0, 1)]
 
-            return results.values()
+        return results
 
     def execute_jobs(self, data: object, job_maker: types.FunctionType) -> list:
-        with Manager() as manager:
-            processes = list()
-            results = manager.dict()
-
-            for process_number in range(self._process_count):
-                processes.append(Process(
-                    target=self._job_maker_wrapper,
-                    args=(
-                        process_number,
-                        data,
-                        results,
-                        job_maker,
-                    )
-                ))
+        # TODO: Is there a reason to make multiprocessing here ? data is not chunked ...
+        if self._process_count > 1:
+            results = self.pool.starmap(job_maker, [(data, i, self._process_count) for i in range(self._process_count)])
+        else:
+            results = [job_maker(data, 0, 1)]
 
-            for process in processes:
-                process.start()
+        return results
 
-            for process in processes:
-                process.join()
-
-            return results.values()
-
-    def _job_maker_wrapper(
-            self,
-            process_number: int,
-            data: list,
-            results: dict,
-            job_maker: types.FunctionType,
-    ):
-        results[process_number] = job_maker(data, process_number, self._process_count)
+    def __del__(self):
+        if self.pool:
+            self.pool.terminate()

synergine2/simulation.py

 from synergine2.utils import get_mechanisms_classes
 
 
-class Subject(object):
+class Subject(BaseObject):
     collections = []
     behaviours_classes = []
     behaviour_selector_class = None  # type: typing.Type[SubjectBehaviourSelector]
             self.adds.append(p_object)
 
 
-class Simulation(object):
+class Simulation(BaseObject):
     accepted_subject_class = Subjects
     behaviours_classes = []
 

synergine2/terminals.py

         ))
 
 
-class Terminal(object):
+class Terminal(BaseObject):
     # Default behaviour is to do nothing.
     # DEFAULT_SLEEP is sleep time between each queue read
     default_sleep = 1
                         handler(event)
 
 
-class TerminalManager(object):
+class TerminalManager(BaseObject):
     def __init__(
         self,
         config: Config,

synergine2/utils.py

+from synergine2.base import BaseObject
 
 
-class ChunkManager(object):
+class ChunkManager(BaseObject):
     def __init__(self, chunks_numbers: int):
         self._chunks_numbers = chunks_numbers
 
     def make_chunks(self, data: list) -> list:
+        if self._chunks_numbers == 1:
+            return [data]
+
         i, j, x = len(data), 0, []
         for k in range(self._chunks_numbers):
             a, j = j, j + (i + k) // self._chunks_numbers

synergine2/xyz.py

 
         try:
             self.xyz.get(value.position, []).remove(value)
+            if not self.xyz[value.position]:
+                del self.xyz[value.position]
         except ValueError:
             pass
 

tests/test_processing.py

         # Goal is 4950
         assert final_result == 4950
 
+    def test_non_parallel_jobs_with_scalar(self):
+        chunk_manager = ChunkManager(1)
+        process_manager = ProcessManager(
+            process_count=1,
+            chunk_manager=chunk_manager,
+        )
+
+        data = list(range(100))
+        results = process_manager.chunk_and_execute_jobs(
+            data,
+            job_maker=self._make_job_with_scalar,
+        )
+        process_id, final_result = results[0]
+
+        assert len(results) == 1
+        assert process_id == os.getpid()
+        assert final_result == 4950
+
     def test_parallel_jobs_with_objects(self):
         chunk_manager = ChunkManager(4)
         process_manager = ProcessManager(