fix mkdir (#52570)

* fix mkdir

* update

python/paddle/distributed/fleet/launch_utils.py

@@ -585,9 +585,9 @@ def start_local_trainers(
         fn = None
         pre_fn = None if os.name == 'nt' else os.setsid
         if log_dir is not None:
-            os.system(f"mkdir -p {log_dir}")
+            os.makedirs(log_dir, exist_ok=True)
             if os.path.exists("%s/endpoints.log" % log_dir):
-                os.system(f"rm -f {log_dir}/endpoints.log")
+                os.remove(f"{log_dir}/endpoints.log")
             with open("%s/endpoints.log" % log_dir, "w") as f:
                 f.write("PADDLE_TRAINER_ENDPOINTS: \n")
                 f.write("\n".join(cluster.trainers_endpoints()))
@@ -1694,7 +1694,7 @@ class ParameterServerLauncher:
                 )
 
             if args.log_dir is not None:
-                os.system(f"mkdir -p {args.log_dir}")
+                os.makedirs(args.log_dir, exist_ok=True)
                 fn = open("%s/serverlog.%d" % (args.log_dir, idx), "w")
                 self.log_fns["server"].append(fn)
                 proc = subprocess.Popen(
@@ -1802,7 +1802,7 @@ class ParameterServerLauncher:
                 )
 
             if args.log_dir is not None:
-                os.system(f"mkdir -p {args.log_dir}")
+                os.makedirs(args.log_dir, exist_ok=True)
                 fn = open("%s/workerlog.%d" % (args.log_dir, idx), "w")
                 self.log_fns["worker"].append(fn)
                 proc = subprocess.Popen(
@@ -1870,7 +1870,7 @@ class ParameterServerLauncher:
                 )
 
             if args.log_dir is not None:
-                os.system(f"mkdir -p {args.log_dir}")
+                os.makedirs(args.log_dir, exist_ok=True)
                 fn = open("%s/coordinator.%d" % (args.log_dir, idx), "w")
                 self.log_fns["coordinator"].append(fn)
                 proc = subprocess.Popen(
@@ -1961,7 +1961,7 @@ class ParameterServerLauncher:
                 )
 
             if args.log_dir is not None:
-                os.system(f"mkdir -p {args.log_dir}")
+                os.makedirs(args.log_dir, exist_ok=True)
                 fn = open("%s/heterlog.%d" % (args.log_dir, idx), "w")
                 self.log_fns["heter_worker"].append(fn)
                 proc = subprocess.Popen(

python/paddle/distributed/fleet/utils/fs.py

@@ -174,7 +174,7 @@ class LocalFS(FS):
         assert not os.path.isfile(fs_path), "{} is already a file".format(
             fs_path
         )
-        os.system(f"mkdir -p {fs_path}")
+        os.makedirs(fs_path, exist_ok=True)
 
     def rename(self, fs_src_path, fs_dst_path):
         """
@@ -319,8 +319,8 @@ class LocalFS(FS):
             if exist_ok:
                 return
             raise FSFileExistsError
-
-        os.system(f"touch {fs_path}")
+        with open(fs_path, 'a'):
+            pass
 
     def mv(self, src_path, dst_path, overwrite=False, test_exists=False):
         """

python/paddle/distributed/utils/launch_utils.py

@@ -485,7 +485,7 @@ def start_local_trainers(
 
         fn = None
         if log_dir is not None:
-            os.system(f"mkdir -p {log_dir}")
+            os.makedirs(log_dir, exist_ok=True)
             fn = open("%s/workerlog.%d" % (log_dir, idx), "a")
             proc = subprocess.Popen(cmd, env=current_env, stdout=fn, stderr=fn)
         else:

python/paddle/fluid/tests/unittests/test_dist_fleet_base.py

@@ -428,6 +428,9 @@ class TestFleetBase(unittest.TestCase):
                     basename
                 )
             )
+
+            if not os.path.isfile(logx):
+                raise FileNotFoundError(f"{logx} is not a file")
             os.system(f"cat {logx}")
             print(
                 "================== Error {} end =====================\n".format(