Repository: deepmodeling/dpdispatcher

Issue:
sed: can't read script: No such file or directory,   when I submit the task using APEX  with LocalContext and Slurm.
---------------------------------------------------------------
░░░░░░█▐▓▓░████▄▄▄█▀▄▓▓▓▌█░░░░░░░░░░█▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀█░░░░░
░░░░░▄█▌▀▄▓▓▄▄▄▄▀▀▀▄▓▓▓▓▓▌█░░░░░░░░░█░░░░░░░░▓░░▓░░░░░░░░█░░░░░
░░░▄█▀▀▄▓█▓▓▓▓▓▓▓▓▓▓▓▓▀░▓▌█░░░░░░░░░█░░░▓░░░░░░░░░▄▄░▓░░░█░▄▄░░
░░█▀▄▓▓▓███▓▓▓███▓▓▓▄░░▄▓▐██░░░▄▀▀▄▄█░░░░░░░▓░░░░█░░▀▄▄▄▄▄▀░░█░
░█▌▓▓▓▀▀▓▓▓▓███▓▓▓▓▓▓▓▄▀▓▓▐█░░░█░░░░█░░░░░░░░░░░░█░░░░░░░░░░░█░
▐█▐██▐░▄▓▓▓▓▓▀▄░▀▓▓▓▓▓▓▓▓▓▌█░░░░▀▀▄▄█░░░░░▓░░░▓░█░░░█▒░░░░█▒░░█
█▌███▓▓▓▓▓▓▓▓▐░░▄▓▓███▓▓▓▄▀▐█░░░░░░░█░░▓░░░░▓░░░█░░░░░░░▀░░░░░█
█▐█▓▀░░▀▓▓▓▓▓▓▓▓▓██████▓▓▓▓▐█░░░░░▄▄█░░░░▓░░░░░░░█░░█▄▄█▄▄█░░█░
▌▓▄▌▀░▀░▐▀█▄▓▓██████████▓▓▓▌██░░░█░░░█▄▄▄▄▄▄▄▄▄▄█░█▄▄▄▄▄▄▄▄▄█░░
▌▓▓▓▄▄▀▀▓▓▓▀▓▓▓▓▓▓▓▓█▓█▓█▓▓▌██░░░█▄▄█░░█▄▄█░░░░░░█▄▄█░░█▄▄█░░░░
█▐▓▓▓▓▓▓▄▄▄▓▓▓▓▓▓█▓█▓█▓█▓▓▓▐█░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░
---------------------------------------------------------------
        AAAAA         PPPPPPPPP     EEEEEEEEEE  XXX       XXX
       AAA AAA        PPP     PPP   EEE           XXX   XXX
      AAA   AAA       PPP     PPP   EEE            XXX XXX
     AAAAAAAAAAA      PPPPPPPPP     EEEEEEEEE       XXXXX
    AAA       AAA     PPP           EEE            XXX XXX
   AAA         AAA    PPP           EEE           XXX   XXX
  AAA           AAA   PPP           EEEEEEEEEE  XXX       XXX
---------------------------------------------------------------
==>> Alloy Property EXplorer using simulations (v1.2.8)
Please cite DOI: 10.48550/arXiv.2404.17330
Li et al, An extendable cloud-native alloy property explorer (2024).
See https://github.com/deepmodeling/APEX for more information.
---------------------------------------------------------------
Checking input files...
-------Submit Workflow Mode-------
Running APEX calculation via lammps
Submitting joint workflow...
INFO:root:Working on: /work/home/acwrhohb19/jwcao/apex_eval/lammps_demo/GAP_demo
Workflow is running locally (ID: gap-demo-joint-pz26l)
INFO:root:Step RelaxLAMMPS-Cal with item 0 starts in process 84151
2024-08-28 13:47:44,362 - INFO : info:check_all_finished: False
2024-08-28 13:47:44,477 - INFO : job: fc71a240632e7109f0639ce201ec76327c076e03 submit; job_id is 9399253
2024-08-28 14:13:57,054 - INFO : job: fc71a240632e7109f0639ce201ec76327c076e03 9399253 terminated; fail_cout is 1; resubmitting job
2024-08-28 14:13:57,131 - INFO : job:fc71a240632e7109f0639ce201ec76327c076e03 re-submit after terminated; new job_id is 9399441
2024-08-28 14:13:57,393 - INFO : job:fc71a240632e7109f0639ce201ec76327c076e03 job_id:9399441 after re-submitting; the state now is <JobStatus.waiting: 2>
2024-08-28 14:24:32,132 - INFO : job: fc71a240632e7109f0639ce201ec76327c076e03 9399441 terminated; fail_cout is 2; resubmitting job
2024-08-28 14:24:32,217 - INFO : job:fc71a240632e7109f0639ce201ec76327c076e03 re-submit after terminated; new job_id is 9399740
2024-08-28 14:24:32,477 - INFO : job:fc71a240632e7109f0639ce201ec76327c076e03 job_id:9399740 after re-submitting; the state now is <JobStatus.waiting: 2>
2024-08-28 15:16:57,962 - INFO : job: fc71a240632e7109f0639ce201ec76327c076e03 9399740 terminated; fail_cout is 3; resubmitting job
Traceback (most recent call last):
  File "/work/home/acwrhohb19/anaconda3/envs/apex_gap/lib/python3.8/site-packages/dpdispatcher/submission.py", line 358, in handle_unexpected_submission_state
    job.handle_unexpected_job_state()
  File "/work/home/acwrhohb19/anaconda3/envs/apex_gap/lib/python3.8/site-packages/dpdispatcher/submission.py", line 862, in handle_unexpected_job_state
    raise RuntimeError(err_msg)
RuntimeError: job:fc71a240632e7109f0639ce201ec76327c076e03 9399740 failed 3 times.
Possible remote error message: ==> /work/home/acwrhohb19/jwcao/apex_eval/lammps_demo/GAP_demo/remote/3468e40db6124ec6e8be9f458b3aa37ac14d6bc6/./log <==
sed: can't read script: No such file or directory
sed: can't read script: No such file or directory
sed: can't read script: No such file or directory


The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/tmp/tmp1p82vt43/gap-demo-joint-pz26l/relaxflow--relax--run-0-lammps/script", line 95, in <module>
    submission.run_submission(clean=True)
    ……………………………………
   ---------------------------------------------------------Split line----------------------------------------------------------
   
  Run command:
    apex submit -d param_joint.json -c global_hpc_local.json
    
  ---------------------------------------------------------Split line----------------------------------------------------------
My  global_hpc.json file is as follows:

{
    "apex_image_name":"zhuoyli/apex_arm64",
    "run_image_name": "zhuoyli/apex_arm64",
    "run_command":"mpirun -np 8 lmp_mpi-in in.lammps",
    "group_size": 1,
    "batch_type": "Slurm",
    "context_type": "LocalContext",
    "machine": {
        "batch_type": "Slurm",
        "context_type": "LocalContext",
        "local_root":  "./",
        "remote_root": "/work/home/acwrhohb19/jwcao/apex_eval/lammps_demo/GAP_demo/remote",
        "clean_asynchronously": false
    },
    "resources":{
        "number_node": 1,
        "cpu_per_node": 8,
        "gpu_per_node": 0,
        "queue_name": "xhacnormalb",
        "group_size": 1,
        "module_purge": true,
        "module_list": ["compiler/devtoolset/7.3.1",
                        "mpi/hpcx/gcc-7.3.1",
                        "mathlib/fftw/3.3.8-gcc-7.3.1-double",
                        "compiler/cmake/3.23.3",
                        "mathlib/lapack/3.8.0-gcc-7.3.1"],
        "source_list": ["/work/home/acwrhohb19/jwcao/apex_eval/lammps_demo/GAP_demo/lammps-quip.sh"],
        "custom_flags": [
            "#SBATCH --job-name=GAP_demo",
            "#SBATCH --partition=xhacnormalb",
	        "#SBATCH --ntasks=8",
	        "#SBATCH --time=1-00:00:00"
        ]
    }
}

  ---------------------------------------------------------Split line----------------------------------------------------------

In the tmp dir:

the  *.sub.run file:


cd $REMOTE_ROOT
cd .
test $? -ne 0 && exit 1
if [ ! -f ec8c25739a7dfb1a7ddc1ef4cf92a0fc33d64998_task_tag_finished ] ;then
   ( sed -i "s#\$(pwd)#$(pwd)#g" script && python3 script ) 1>>log 2>>log 
  if test $? -eq 0; then touch ec8c25739a7dfb1a7ddc1ef4cf92a0fc33d64998_task_tag_finished; else echo 1 > $REMOTE_ROOT/218928143f7f92d20fcd0442ff2c75eab569b52d_flag_if_job_task_fail;tail -v -c 1000 $REMOTE_ROOT/./log > $REMOTE_ROOT/218928143f7f92d20fcd0442ff2c75eab569b52d_last_err_file;fi
fi &
wait 

So, why the sed can't find the script.  When I use:

ls $REMOTE_ROOT
ls -l script

echo "current workdir: $(pwd)" 
cat $(readlink -f script)

in the source file,  the slurm.out us as follows:

3468e40db6124ec6e8be9f458b3aa37ac14d6bc6.json
fc71a240632e7109f0639ce201ec76327c076e03_flag_if_job_task_fail
fc71a240632e7109f0639ce201ec76327c076e03_job_id
fc71a240632e7109f0639ce201ec76327c076e03_last_err_file
fc71a240632e7109f0639ce201ec76327c076e03.sub
fc71a240632e7109f0639ce201ec76327c076e03.sub.run
log
script
slurm-9399253.out
slurm-9399441.out
slurm-9399740.out
tmp

lrwxrwxr-x 1 acwrhohb19 acwrhohb19 87 Aug 28 13:47 script -> /tmp/tmp1p82vt43/gap-demo-joint-pz26l/relaxflow--relax--run-0-lammps/workdir/././script
current workdir: /work/home/acwrhohb19/jwcao/apex_eval/lammps_demo/GAP_demo/remote/3468e40db6124ec6e8be9f458b3aa37ac14d6bc6


the script seem to be a link file.

How to debug it?

Discussion:
@njzjz: In HPCs, usually different nodes have its own `/tmp` directory.

Solution patch:
diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
index cb1a8fa..364720c 100644
--- a/.pre-commit-config.yaml
+++ b/.pre-commit-config.yaml
@@ -18,7 +18,7 @@ repos:
   # Python
   - repo: https://github.com/astral-sh/ruff-pre-commit
     # Ruff version.
-    rev: v0.6.1
+    rev: v0.6.2
     hooks:
       - id: ruff
         args: ["--fix"]
diff --git a/doc/context.md b/doc/context.md
index c46a08b..544dcb5 100644
--- a/doc/context.md
+++ b/doc/context.md
@@ -16,7 +16,8 @@ Since [`bash -l`](https://www.gnu.org/software/bash/manual/bash.html#Invoking-Ba
 {dargs:argument}`context_type <machine/context_type>`: `Local`
 
 `Local` runs jobs in the local server, but in a different directory.
-Files will be copied to the remote directory before jobs start and copied back after jobs finish.
+Files will be symlinked to the remote directory before jobs start and copied back after jobs finish.
+If the local directory is not accessible with the [batch system](./batch.md), turn off {dargs:argument}`symlink <machine[SSHContext]/remote_profile/symlink>`, and then files on the local directory will be copied to the remote directory.
 
 Since [`bash -l`](https://www.gnu.org/software/bash/manual/bash.html#Invoking-Bash) is used in the shebang line of the submission scripts, the [login shell startup files](https://www.gnu.org/software/bash/manual/bash.html#Invoking-Bash) will be executed, potentially overriding the current environment variables. Therefore, it's advisable to explicitly set the environment variables using {dargs:argument}`envs <resources/envs>` or {dargs:argument}`source_list <resources/source_list>`.
 
diff --git a/dpdispatcher/contexts/local_context.py b/dpdispatcher/contexts/local_context.py
index 6ecbd64..eda0a7d 100644
--- a/dpdispatcher/contexts/local_context.py
+++ b/dpdispatcher/contexts/local_context.py
@@ -3,6 +3,9 @@ import shutil
 import subprocess as sp
 from glob import glob
 from subprocess import TimeoutExpired
+from typing import List
+
+from dargs import Argument
 
 from dpdispatcher.base_context import BaseContext
 from dpdispatcher.dlog import dlog
@@ -60,6 +63,7 @@ class LocalContext(BaseContext):
         self.temp_local_root = os.path.abspath(local_root)
         self.temp_remote_root = os.path.abspath(remote_root)
         self.remote_profile = remote_profile
+        self.symlink = remote_profile.get("symlink", True)
 
     @classmethod
     def load_from_dict(cls, context_dict):
@@ -83,6 +87,25 @@ class LocalContext(BaseContext):
             self.temp_remote_root, submission.submission_hash
         )
 
+    def _copy_from_local_to_remote(self, local_path, remote_path):
+        if not os.path.exists(local_path):
+            raise FileNotFoundError(
+                f"cannot find uploaded file {os.path.join(local_path)}"
+            )
+        if os.path.exists(remote_path):
+            os.remove(remote_path)
+        _check_file_path(remote_path)
+
+        if self.symlink:
+            # ensure the file exist
+            os.symlink(local_path, remote_path)
+        elif os.path.isfile(local_path):
+            shutil.copyfile(local_path, remote_path)
+        elif os.path.isdir(local_path):
+            shutil.copytree(local_path, remote_path)
+        else:
+            raise ValueError(f"Unknown file type: {local_path}")
+
     def upload(self, submission):
         os.makedirs(self.remote_root, exist_ok=True)
         for ii in submission.belonging_tasks:
@@ -103,14 +126,9 @@ class LocalContext(BaseContext):
                 file_list.extend(rel_file_list)
 
             for jj in file_list:
-                if not os.path.exists(os.path.join(local_job, jj)):
-                    raise FileNotFoundError(
-                        "cannot find upload file " + os.path.join(local_job, jj)
-                    )
-                if os.path.exists(os.path.join(remote_job, jj)):
-                    os.remove(os.path.join(remote_job, jj))
-                _check_file_path(os.path.join(remote_job, jj))
-                os.symlink(os.path.join(local_job, jj), os.path.join(remote_job, jj))
+                self._copy_from_local_to_remote(
+                    os.path.join(local_job, jj), os.path.join(remote_job, jj)
+                )
 
         local_job = self.local_root
         remote_job = self.remote_root
@@ -128,14 +146,9 @@ class LocalContext(BaseContext):
             file_list.extend(rel_file_list)
 
         for jj in file_list:
-            if not os.path.exists(os.path.join(local_job, jj)):
-                raise FileNotFoundError(
-                    "cannot find upload file " + os.path.join(local_job, jj)
-                )
-            if os.path.exists(os.path.join(remote_job, jj)):
-                os.remove(os.path.join(remote_job, jj))
-            _check_file_path(os.path.join(remote_job, jj))
-            os.symlink(os.path.join(local_job, jj), os.path.join(remote_job, jj))
+            self._copy_from_local_to_remote(
+                os.path.join(local_job, jj), os.path.join(remote_job, jj)
+            )
 
     def download(
         self, submission, check_exists=False, mark_failure=True, back_error=False
@@ -336,3 +349,31 @@ class LocalContext(BaseContext):
                 stdout = None
                 stderr = None
         return ret, stdout, stderr
+
+    @classmethod
+    def machine_subfields(cls) -> List[Argument]:
+        """Generate the machine subfields.
+
+        Returns
+        -------
+        list[Argument]
+            machine subfields
+        """
+        doc_remote_profile = "The information used to maintain the local machine."
+        return [
+            Argument(
+                "remote_profile",
+                dict,
+                optional=True,
+                doc=doc_remote_profile,
+                sub_fields=[
+                    Argument(
+                        "symlink",
+                        bool,
+                        optional=True,
+                        default=True,
+                        doc="Whether to use symbolic links to replace copy. This option should be turned off if the local directory is not accessible on the Batch system.",
+                    ),
+                ],
+            )
+        ]
diff --git a/dpdispatcher/machine.py b/dpdispatcher/machine.py
index 352eade..a78b61f 100644
--- a/dpdispatcher/machine.py
+++ b/dpdispatcher/machine.py
@@ -161,6 +161,9 @@ class Machine(metaclass=ABCMeta):
             machine_dict["remote_profile"] = self.context.remote_profile
         else:
             machine_dict["remote_profile"] = {}
+        # normalize the dict
+        base = self.arginfo()
+        machine_dict = base.normalize_value(machine_dict, trim_pattern="_*")
         return machine_dict
 
     def __eq__(self, other):


Test patch:
diff --git a/tests/test_argcheck.py b/tests/test_argcheck.py
index 2ef1883..b87f39f 100644
--- a/tests/test_argcheck.py
+++ b/tests/test_argcheck.py
@@ -23,7 +23,10 @@ class TestJob(unittest.TestCase):
             "context_type": "LocalContext",
             "local_root": "./",
             "remote_root": "/some/path",
-            "remote_profile": {},
+            "remote_profile": {
+                "symlink": True,
+            },
+            "clean_asynchronously": False,
         }
         self.assertDictEqual(norm_dict, expected_dict)
 
diff --git a/tests/test_run_submission.py b/tests/test_run_submission.py
index 51e9ec2..958eba7 100644
--- a/tests/test_run_submission.py
+++ b/tests/test_run_submission.py
@@ -291,6 +291,24 @@ class TestLocalContext(RunSubmission, unittest.TestCase):
         return super().test_async_run_submission()
 
 
+@unittest.skipIf(sys.platform == "win32", "Shell is not supported on Windows")
+class TestLocalContextCopy(RunSubmission, unittest.TestCase):
+    def setUp(self):
+        super().setUp()
+        self.temp_dir = tempfile.TemporaryDirectory()
+        self.machine_dict["context_type"] = "LocalContext"
+        self.machine_dict["remote_root"] = self.temp_dir.name
+        self.machine_dict["remote_profile"]["symlink"] = False
+
+    def tearDown(self):
+        super().tearDown()
+        self.temp_dir.cleanup()
+
+    @unittest.skip("It seems the remote file may be deleted")
+    def test_async_run_submission(self):
+        return super().test_async_run_submission()
+
+
 @unittest.skipIf(sys.platform == "win32", "Shell is not supported on Windows")
 class TestLazyLocalContext(RunSubmission, unittest.TestCase):
     def setUp(self):