pandaharvester/harvestermonitor/dummy_mcore_monitor.py

0001 import os.path
0002 from concurrent.futures import ProcessPoolExecutor as Pool
0003
0004 from pandaharvester.harvestercore import core_utils
0005 from pandaharvester.harvestercore.plugin_base import PluginBase
0006 from pandaharvester.harvestercore.work_spec import WorkSpec
0007
0008 # logger
0009 baseLogger = core_utils.setup_logger("dummy_mcore_monitor")
0010
0011
0012 # check a worker
0013 def check_a_worker(workspec):
0014     # make logger
0015     tmpLog = core_utils.make_logger(baseLogger, f"workerID={workspec.workerID}", method_name="check_a_worker")
0016     dummyFilePath = os.path.join(workspec.get_access_point(), "status.txt")
0017     tmpLog.debug(f"look for {dummyFilePath}")
0018     newStatus = WorkSpec.ST_finished
0019     try:
0020         with open(dummyFilePath) as dummyFile:
0021             newStatus = dummyFile.readline()
0022             newStatus = newStatus.strip()
0023     except BaseException:
0024         pass
0025     tmpLog.debug(f"newStatus={newStatus}")
0026     return (newStatus, "")
0027
0028
0029 # dummy monitor with multi-cores
0030 class DummyMcoreMonitor(PluginBase):
0031     # constructor
0032     def __init__(self, **kwarg):
0033         PluginBase.__init__(self, **kwarg)
0034
0035     # check workers
0036     def check_workers(self, workspec_list):
0037         # make logger
0038         tmpLog = self.make_logger(baseLogger, method_name="check_workers")
0039         tmpLog.debug(f"start nWorkers={len(workspec_list)}")
0040         with Pool() as pool:
0041             retList = pool.map(check_a_worker, workspec_list)
0042         tmpLog.debug("done")
0043         return True, retList