Reopen dlio.log in non-fork reader_threads child processes (#130)

When reader_threads child processes are started with multiprocessing_context
set to either spawn or forkserver, the new child processes do not have the
dlio.log file open any longer.  Change worker_init() to reopen the dlio.log
file in those cases. When the context is set to fork, the dlio.log file
remains open in the child process and does not have to be reopened.

Author: krehm

dlio_benchmark/data_loader/torch_data_loader.py

@@ -54,6 +54,8 @@ def __init__(self, format_type, dataset_type, epoch, num_samples, num_workers, b
     @dlp.log
     def worker_init(self, worker_id):
         pickle.loads(self.serial_args)
+        _args = ConfigArguments.get_instance()
+        _args.configure_dlio_logging(True)
         logging.debug(f"{utcnow()} worker initialized {worker_id} with format {self.format_type}")
         self.reader = ReaderFactory.get_reader(type=self.format_type,
                                                dataset_type=self.dataset_type,

dlio_benchmark/main.py

@@ -73,15 +73,8 @@ def __init__(self, cfg):
         self.storage = StorageFactory().get_storage(self.args.storage_type, self.args.storage_root,
                                                     self.args.framework)
 
-        if self.args.output_folder is None:
-            try:
-                hydra_cfg = hydra.core.hydra_config.HydraConfig.get()
-                self.args.output_folder = hydra_cfg['runtime']['output_dir']
-            except:
-                self.args.output_folder = 'output/'
         self.output_folder = self.args.output_folder
         os.makedirs(self.args.output_folder, mode=0o755, exist_ok=True)
-        self.logfile = os.path.join(self.args.output_folder, self.args.log_file)
         dlp_trace = get_trace_name(self.args.output_folder)
         self.comm = DLIOMPI.get_instance().comm()
         self.my_rank = self.args.my_rank = DLIOMPI.get_instance().rank()
@@ -99,21 +92,11 @@ def __init__(self, cfg):
 
             # Delete previous logfile
             if self.my_rank == 0:
-                if os.path.isfile(self.logfile):
-                    os.remove(self.logfile)
+                if os.path.isfile(self.args.logfile_path):
+                    os.remove(self.args.logfile_path)
             self.comm.barrier()
             # Configure the logging library
-            log_level = logging.DEBUG if self.args.debug else logging.INFO
-            logging.basicConfig(
-                level=log_level,
-                force=True,
-                handlers=[
-                    logging.FileHandler(self.logfile, mode="a", encoding='utf-8'),
-                    logging.StreamHandler()
-                ],
-                format='[%(levelname)s] %(message)s [%(pathname)s:%(lineno)d]'
-                # logging's max timestamp resolution is msecs, we will pass in usecs in the message
-            )
+            self.args.configure_dlio_logging(False)
             if self.args.my_rank == 0:
                 logging.info(f"{utcnow()} Profiling DLIO {dlp_trace}")
                 logging.info(f"{utcnow()} Running DLIO with {self.args.comm_size} process(es)")

dlio_benchmark/utils/config.py

@@ -16,6 +16,7 @@
 """
 import importlib
 import inspect
+import hydra
 
 import logging
 from time import time
@@ -150,6 +151,23 @@ def get_instance():
             ConfigArguments()
         return ConfigArguments.__instance
 
+    def configure_dlio_logging(self, is_child=False):
+        # with "multiprocessing_context=fork" the log file remains open in the child process
+        if is_child and self.multiprocessing_context == "fork":
+            return
+        # Configure the logging library
+        log_level = logging.DEBUG if self.debug else logging.INFO
+        logging.basicConfig(
+            level=log_level,
+            force=True,
+            handlers=[
+                logging.FileHandler(self.logfile_path, mode="a", encoding='utf-8'),
+                logging.StreamHandler()
+            ],
+            format='[%(levelname)s] %(message)s [%(pathname)s:%(lineno)d]'
+            # logging's max timestamp resolution is msecs, we will pass in usecs in the message
+        )
+
     @dlp.log
     def validate(self):
         """ validate whether the parameters are set correctly"""
@@ -438,7 +456,15 @@ def LoadConfig(args, config):
             args.output_folder = config['output']['folder']
         if 'log_file' in config['output']:
             args.log_file = config['output']['log_file']
-            
+
+    if args.output_folder is None:
+        try:
+            hydra_cfg = hydra.core.hydra_config.HydraConfig.get()
+            args.output_folder = hydra_cfg['runtime']['output_dir']
+        except:
+            args.output_folder = 'output/'
+    args.logfile_path = os.path.join(args.output_folder, args.log_file)
+
     if 'workflow' in config:
         if 'generate_data' in config['workflow']:
             args.generate_data = config['workflow']['generate_data']