improve SGE system

dpdispatcher/machine.py

@@ -261,8 +261,7 @@ def gen_script_env(self, job):
 
         source_list = job.resources.source_list
         for ii in source_list:
-            line = f"{{ source {ii}; }} \n"
-            source_files_part += line
+            source_files_part += f"source {ii}\n"
 
         export_envs_part = ""
         envs = job.resources.envs

dpdispatcher/machines/pbs.py

@@ -181,10 +181,10 @@ def gen_script_header(self, job):
 
 sge_script_header_template = """
 #!/bin/bash
-#$ -N dpdispatcher_submit
-{select_node_line}
+#$ -S /bin/bash
 #$ -cwd
-
+#$ -N dp_job
+{select_node_line}
 """
 
 
@@ -209,21 +209,37 @@ def __init__(
         )
 
     def gen_script_header(self, job):
+        ### Ref:https://softpanorama.org/HPC/PBS_and_derivatives/Reference/pbs_command_vs_sge_commands.shtml
+        # resources.number_node is not used in SGE
         resources = job.resources
         sge_script_header_dict = {}
-        # resources.number_node is not used
         sge_script_header_dict["select_node_line"] = (
-            f"#$ -pe mpi {resources.cpu_per_node} "
+            f"#$ -pe {resources.sge_pe_name} {resources.cpu_per_node}\n"
         )
-        # resources.queue_name is not necessary
-        sge_script_header = sge_script_header_template.format(**sge_script_header_dict)
+        if resources.queue_name != "":
+            sge_script_header_dict["select_node_line"] += (
+                f"#$ -q {resources.queue_name}"
+            )
+        if (
+            resources["strategy"].get("customized_script_header_template_file")
+            is not None
+        ):
+            file_name = resources["strategy"]["customized_script_header_template_file"]
+            sge_script_header = customized_script_header_template(file_name, resources)
+        else:
+            sge_script_header = sge_script_header_template.format(
+                **sge_script_header_dict
+            )
         return sge_script_header
 
     def do_submit(self, job):
         script_file_name = job.script_file_name
         script_str = self.gen_script(job)
         job_id_name = job.job_hash + "_job_id"
         self.context.write_file(fname=script_file_name, write_str=script_str)
+        script_run_str = self.gen_script_command(job)
+        script_run_file_name = f"{job.script_file_name}.run"
+        self.context.write_file(fname=script_run_file_name, write_str=script_run_str)
         script_file_dir = self.context.remote_root
         stdin, stdout, stderr = self.context.block_checkcall(
             "cd {} && {} {}".format(script_file_dir, "qsub", script_file_name)

dpdispatcher/submission.py

@@ -961,6 +961,8 @@ class Resources:
         The queue name of batch job scheduler system.
     group_size : int
         The number of `tasks` in a `job`.
+    sge_pe_name : str
+        The parallel environment name of SGE.
     custom_flags : list of Str
         The extra lines pass to job submitting script header
     strategy : dict
@@ -991,6 +993,7 @@ def __init__(
         queue_name,
         group_size,
         *,
+        sge_pe_name="mpi",
         custom_flags=[],
         strategy=default_strategy,
         para_deg=1,
@@ -1011,6 +1014,7 @@ def __init__(
         self.group_size = group_size
 
         # self.extra_specification = extra_specification
+        self.sge_pe_name = sge_pe_name
         self.custom_flags = custom_flags
         self.strategy = strategy
         self.para_deg = para_deg
@@ -1057,6 +1061,7 @@ def serialize(self):
         resources_dict["queue_name"] = self.queue_name
         resources_dict["group_size"] = self.group_size
 
+        resources_dict["sge_pe_name"] = self.sge_pe_name
         resources_dict["custom_flags"] = self.custom_flags
         resources_dict["strategy"] = self.strategy
         resources_dict["para_deg"] = self.para_deg
@@ -1079,6 +1084,7 @@ def deserialize(cls, resources_dict):
             gpu_per_node=resources_dict.get("gpu_per_node", 0),
             queue_name=resources_dict.get("queue_name", ""),
             group_size=resources_dict["group_size"],
+            sge_pe_name=resources_dict.get("sge_pe_name", "mpi"),
             custom_flags=resources_dict.get("custom_flags", []),
             strategy=resources_dict.get("strategy", default_strategy),
             para_deg=resources_dict.get("para_deg", 1),
@@ -1127,6 +1133,7 @@ def arginfo(detail_kwargs=True):
         doc_gpu_per_node = "gpu numbers of each node assigned to each job."
         doc_queue_name = "The queue name of batch job scheduler system."
         doc_group_size = "The number of `tasks` in a `job`. 0 means infinity."
+        doc_sge_pe_name = "The parallel environment name of SGE."
         doc_custom_flags = "The extra lines pass to job submitting script header"
         doc_para_deg = "Decide how many tasks will be run in parallel."
         doc_source_list = "The env file to be sourced before the command execution."
@@ -1191,6 +1198,7 @@ def arginfo(detail_kwargs=True):
             ),
             Argument("queue_name", str, optional=True, doc=doc_queue_name, default=""),
             Argument("group_size", int, optional=False, doc=doc_group_size),
+            Argument("sge_pe_name", str, optional=True, doc=doc_sge_pe_name),
             Argument("custom_flags", List[str], optional=True, doc=doc_custom_flags),
             # Argument("strategy", dict, optional=True, doc=doc_strategy,default=default_strategy),
             strategy_format,

tests/sample_class.py

@@ -25,6 +25,7 @@ def get_sample_resources(cls):
             gpu_per_node=1,
             queue_name="T4_4_15",
             group_size=2,
+            sge_pe_name="mpi",
             custom_flags=[],
             strategy={"if_cuda_multi_devices": False},
             para_deg=1,
@@ -43,6 +44,7 @@ def get_sample_resources_dict(cls):
             "gpu_per_node": 1,
             "queue_name": "T4_4_15",
             "group_size": 2,
+            "sge_pe_name": "mpi",
             "custom_flags": [],
             "strategy": {"if_cuda_multi_devices": False, "ratio_unfinished": 0.0},
             "para_deg": 1,

tests/test_argcheck.py

@@ -60,6 +60,7 @@ def test_resources_argcheck(self):
             "para_deg": 1,
             "prepend_script": [],
             "queue_name": "haha",
+            "sge_pe_name": "mpi",
             "source_list": [],
             "strategy": {"if_cuda_multi_devices": False, "ratio_unfinished": 0.0},
             "wait_time": 0,