examples/python/synchro-mutex/synchro-mutex.py

   1 from argparse import ArgumentParser
   2 from dataclasses import dataclass
   3 import sys
   4
   5 from simgrid import Actor, Engine, Host, Mutex, this_actor
   6
   7
   8 def create_parser() -> ArgumentParser:
   9     parser = ArgumentParser()
  10     parser.add_argument(
  11         '--platform',
  12         type=str,
  13         required=True,
  14         help='path to the platform description'
  15     )
  16     parser.add_argument(
  17         '--workers',
  18         type=int,
  19         default=6,
  20         help='number of workers to start'
  21     )
  22     parser.add_argument(
  23         '--trials-before-success',
  24         type=int,
  25         default=0,
  26         help='number of attempts each workers need to make before getting the correct answer'
  27              ' (i.e. number of simulated failures)'
  28     )
  29     return parser
  30
  31
  32 @dataclass
  33 class ResultHolder:
  34     value: int
  35
  36
  37 class CalculationError(RuntimeError):
  38     """ Fake calculation error
  39     """
  40     pass
  41
  42
  43 def worker_context_manager(mutex: Mutex, trials_before_success: int, result: ResultHolder):
  44     """ Worker that uses a context manager to acquire/release the shared mutex
  45     :param mutex: Shared mutex that guards read/write access to the shared result
  46     :param trials_before_success: Number of simulated calculation failures before success
  47     :param result: Shared result which will be updated by the worker
  48     """
  49     this_actor.info(f"I just started")
  50     for trial in range(trials_before_success + 1):
  51         try:
  52             with mutex:
  53                 this_actor.info(f"acquired the mutex with context manager")
  54                 this_actor.sleep_for(1)
  55                 if trial < trials_before_success:
  56                     raise CalculationError("did not manage to find the correct answer")
  57                 result.value += 1
  58                 this_actor.info(f"updated shared result, which is now {result.value}")
  59         except CalculationError as e:
  60             this_actor.warning(f"ran in trouble while calculating: {e}. Will retry shortly.")
  61         finally:
  62             this_actor.info(f"released the mutex after leaving the context manager")
  63     this_actor.info("Bye now!")
  64
  65
  66 def worker(mutex: Mutex, trials_before_success: int, result: ResultHolder):
  67     """ Worker that manually acquires/releases the shared mutex
  68     :param mutex: Shared mutex that guards read/write access to the shared result
  69     :param trials_before_success: Number of simulated calculation failures before success
  70     :param result: Shared result which will be updated by the worker
  71     """
  72     this_actor.info(f"I just started")
  73     for trial in range(trials_before_success + 1):
  74         try:
  75             mutex.lock()
  76             this_actor.info(f"acquired the mutex manually")
  77             this_actor.sleep_for(1)
  78             if trial < trials_before_success:
  79                 raise CalculationError("did not manage to find the correct answer")
  80             result.value += 1
  81             this_actor.info(f"updated shared result, which is now {result.value}")
  82         except CalculationError as e:
  83             this_actor.warning(f"ran in trouble while calculating: {e}. Will retry shortly.")
  84         finally:
  85             this_actor.info(f"released the mutex manually")
  86             mutex.unlock()
  87     this_actor.info("Bye now!")
  88
  89
  90 def master(settings):
  91     """ Spawns `--workers` workers and wait until they have all updated the shared result, then displays it before
  92         leaving. Alternatively spawns `worker_context_manager()` and `worker()` workers.
  93     :param settings: Simulation settings
  94     """
  95     result = ResultHolder(value=0)
  96     mutex = Mutex()
  97     actors = []
  98     for i in range(settings.workers):
  99         use_worker_context_manager = i % 2 == 0
 100         actors.append(
 101             Actor.create(
 102                 f"worker-{i}(mgr)" if use_worker_context_manager else f"worker-{i}",
 103                 Host.by_name("Jupiter" if use_worker_context_manager else "Tremblay"),
 104                 worker_context_manager if use_worker_context_manager else worker,
 105                 mutex,
 106                 settings.trials_before_success,
 107                 result
 108             )
 109         )
 110     [actor.join() for actor in actors]
 111     this_actor.info(f"The final result is: {result.value}")
 112
 113
 114 def main():
 115     settings = create_parser().parse_known_args()[0]
 116     e = Engine(sys.argv)
 117     e.load_platform(settings.platform)
 118     Actor.create("master", Host.by_name("Tremblay"), master, settings)
 119     e.run()
 120
 121
 122 if __name__ == "__main__":
 123     main()