improve SGE system (#446)

- add new key `sge_pe_name` into `resources`
- allow `customized_script_header_template` in SGE system
- ...

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

- **New Features**
- Enhanced PBS script header generation with updated job name, node
selection, and custom script headers.

- **Bug Fixes**
- Corrected shell script sourcing commands in test scripts by removing
unnecessary double curly braces.

- **Improvements**
  - Simplified source file handling in job submission scripts.
- Added support for SGE parallel environment names in resource
configurations.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

---------

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>

Author: thangckt

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

@@ -1,4 +1,7 @@
 import shlex
+from typing import List
+
+from dargs import Argument
 
 from dpdispatcher.dlog import dlog
 from dpdispatcher.machine import Machine
@@ -181,10 +184,10 @@ def gen_script_header(self, job):
 
 sge_script_header_template = """
 #!/bin/bash
-#$ -N dpdispatcher_submit
-{select_node_line}
+#$ -S /bin/bash
 #$ -cwd
-
+#$ -N DPjob
+{select_node_line}
 """
 
 
@@ -209,21 +212,38 @@ 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_pe_name = resources.kwargs.get("sge_pe_name", "mpi")
         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 {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)
@@ -281,3 +301,35 @@ def check_status(self, job):
     def check_finish_tag(self, job):
         job_tag_finished = job.job_hash + "_job_tag_finished"
         return self.context.check_file_exists(job_tag_finished)
+
+    @classmethod
+    def resources_subfields(cls) -> List[Argument]:
+        """Generate the resources subfields.
+
+            sge_pe_name : str
+        The parallel environment name of SGE.
+
+        Returns
+        -------
+        list[Argument]
+            resources subfields
+        """
+        doc_sge_pe_name = "The parallel environment name of SGE."
+
+        return [
+            Argument(
+                "kwargs",
+                dict,
+                [
+                    Argument(
+                        "sge_pe_name",
+                        str,
+                        optional=True,
+                        default="mpi",
+                        doc=doc_sge_pe_name,
+                    ),
+                ],
+                optional=False,
+                doc="Extra arguments.",
+            )
+        ]

tests/test_lsf_script_generation.py

@@ -108,8 +108,8 @@ def test_shell_trival(self):
             module load use.own
             module load deepmd/1.3
 
-            {{ source /data/home/ypliu/scripts/avail_gpu.sh; }} 
-            {{ source /data/home/ypliu/dprun/tf_envs.sh; }} 
+            source /data/home/ypliu/scripts/avail_gpu.sh
+            source /data/home/ypliu/dprun/tf_envs.sh
 
             export DP_DISPATCHER_EXPORT=test_foo_bar_baz