Merge pull request #90 from lsst/tickets/DM-50973

Use job name to map the quantum nodes

Author: zhaoyuyoung

config/bps_panda_qnode_map.yaml

@@ -0,0 +1,3 @@
+# enable qnode map to use job  name to map the long pseudo file name
+
+enableQnodeMap: true

doc/changes/DM-50973.feature.rst

@@ -0,0 +1 @@
+Use job name to map the quantum nodes

python/lsst/ctrl/bps/panda/edgenode/cmd_line_decoder.py

@@ -304,57 +304,87 @@ def replace_event_file(params, files):
     return ret_status, with_events, with_order_id_map, params_map
 
 
-deliver_input_files(sys.argv[3], sys.argv[4], sys.argv[5])
-cmd_line = str(binascii.unhexlify(sys.argv[1]).decode())
-data_params = sys.argv[2]
-cmd_line = replace_environment_vars(cmd_line)
-
-print(f"cmd_line: {cmd_line}")
-print(f"data_params: {data_params}")
-
-# If EventService is enabled, data_params will only contain event information.
-# So we need to convert the event information to LSST pseudo file names.
-# If EventService is not enabled, this part will not change data_params.
-ret_rep = replace_event_file(data_params, sys.argv[4])
-ret_event_status, with_events, with_order_id_map, event_params_map = ret_rep
-print(
-    f"ret_event_status: {ret_event_status}, with_events: {with_events} with_order_id_map: {with_order_id_map}"
-)
-if not ret_event_status:
-    print("failed to map EventService/orderIdMap parameters to original LSST pseudo file names")
-    exit_code = 1
-    sys.exit(exit_code)
+def use_map_file(input_file):
+    """Check whether the input file needs to be replaced
+    because enableQnodeMap is enabled.
 
-for event_param in event_params_map:
-    order_id = event_params_map[event_param]["order_id"]
-    pseudo_file_name = event_params_map[event_param]["order_id_map"][order_id]
-    print(f"replacing event {event_param} with order_id {order_id} to: {pseudo_file_name}")
-    cmd_line = cmd_line.replace(event_param, pseudo_file_name)
-    data_params = data_params.replace(event_param, pseudo_file_name)
+    Parameters
+    ----------
+    input_file : `str`
+        Input file either a pseudo file or job name.
 
-# If job name map is enabled, data_params will only contain order_id
-# information. Here we will convert order_id information to LSST pseudo
-# file names.
+    Returns
+    -------
+    use_qnode_map: `bool`
+        Whether qnode_map is used. There is a placeholder 'PH'
+    when enableQnodeMap is true.
+    """
+    parts = input_file.split(":")
+    use_qnode_map = len(parts) == 2 and parts[0] == "PH"
+    return use_qnode_map
+
+
+if __name__ == "__main__":
+    deliver_input_files(sys.argv[3], sys.argv[4], sys.argv[5])
+    cmd_line = str(binascii.unhexlify(sys.argv[1]).decode())
+    data_params = sys.argv[2]
+    cmd_line = replace_environment_vars(cmd_line)
+
+    print(f"cmd_line: {cmd_line}")
+    print(f"data_params: {data_params}")
+
+    # If EventService is enabled, data_params will only contain
+    # event information. So we need to convert the event information
+    # to LSST pseudo file names. If EventService is not enabled,
+    # this part will not change data_params.
+    ret_rep = replace_event_file(data_params, sys.argv[4])
+    ret_event_status, with_events, with_order_id_map, event_params_map = ret_rep
+    print(
+        f"ret_event_status: {ret_event_status}, "
+        f"with_events: {with_events} "
+        f"with_order_id_map: {with_order_id_map}"
+    )
+    if not ret_event_status:
+        print("failed to map EventService/orderIdMap parameters to original LSST pseudo file names")
+        exit_code = 1
+        sys.exit(exit_code)
+
+    for event_param in event_params_map:
+        order_id = event_params_map[event_param]["order_id"]
+        pseudo_file_name = event_params_map[event_param]["order_id_map"][order_id]
+        print(f"replacing event {event_param} with order_id {order_id} to: {pseudo_file_name}")
+        cmd_line = cmd_line.replace(event_param, pseudo_file_name)
+        data_params = data_params.replace(event_param, pseudo_file_name)
+
+    # If job name map is enabled, data_params will only contain order_id
+    # information. Here we will convert order_id information to LSST pseudo
+    # file names.
+
+    data_params = data_params.split("+")
+
+    """Replace the pipetask command line placeholders
+     with actual data provided in the script call
+     in form placeholder1:file1+placeholder2:file2:...
+    """
+    cmd_line = replace_files_placeholders(cmd_line, sys.argv[4])
 
-data_params = data_params.split("+")
+    jobname = data_params[0]
+    if use_map_file(jobname):
+        with open("qnode_map.json", encoding="utf-8") as f:
+            qnode_map = json.load(f)
+            data_params = qnode_map[jobname].split("+")
 
-"""Replace the pipetask command line placeholders
- with actual data provided in the script call
- in form placeholder1:file1+placeholder2:file2:...
-"""
-cmd_line = replace_files_placeholders(cmd_line, sys.argv[4])
-
-for key_value_pair in data_params[1:]:
-    (key, value) = key_value_pair.split(":")
-    cmd_line = cmd_line.replace("{" + key + "}", value)
-
-print("executable command line:")
-print(cmd_line)
-
-exit_status = os.system(cmd_line)
-exit_code = 1
-if os.WIFSIGNALED(exit_status):
-    exit_code = os.WTERMSIG(exit_status) + 128
-elif os.WIFEXITED(exit_status):
-    exit_code = os.WEXITSTATUS(exit_status)
-sys.exit(exit_code)
+    for key_value_pair in data_params[1:]:
+        (key, value) = key_value_pair.split(":")
+        cmd_line = cmd_line.replace("{" + key + "}", value)
+
+    print("executable command line:")
+    print(cmd_line)
+
+    exit_status = os.system(cmd_line)
+    exit_code = 1
+    if os.WIFSIGNALED(exit_status):
+        exit_code = os.WTERMSIG(exit_status) + 128
+    elif os.WIFEXITED(exit_status):
+        exit_code = os.WEXITSTATUS(exit_status)
+    sys.exit(exit_code)

python/lsst/ctrl/bps/panda/utils.py

@@ -37,6 +37,7 @@
 
 import binascii
 import concurrent.futures
+import json
 import logging
 import os
 import random
@@ -228,6 +229,7 @@ def _make_doma_work(
     max_payloads_per_panda_job=PANDA_DEFAULT_MAX_PAYLOADS_PER_PANDA_JOB,
     max_wms_job_wall_time=None,
     remote_filename=None,
+    qnode_map_filename=None,
 ):
     """Make the DOMA Work object for a PanDA task.
 
@@ -358,6 +360,9 @@ def _make_doma_work(
         if gwfile.job_access_remote:
             direct_io_files.add(gwfile.name)
 
+    if qnode_map_filename:
+        local_pfns.update(qnode_map_filename)
+
     submit_cmd = generic_workflow.run_attrs.get("bps_iscustom", False)
 
     if not direct_io_files:
@@ -590,6 +595,7 @@ def add_idds_work(config, generic_workflow, idds_workflow):
     """
     # event service
     _, enable_event_service = config.search("enableEventService", opt={"default": None})
+    _, enable_qnode_map = config.search("enableQnodeMap", opt={"default": None})
     _, max_payloads_per_panda_job = config.search(
         "maxPayloadsPerPandaJob", opt={"default": PANDA_DEFAULT_MAX_PAYLOADS_PER_PANDA_JOB}
     )
@@ -617,10 +623,11 @@ def add_idds_work(config, generic_workflow, idds_workflow):
     task_count = 0  # Task number/ID in idds workflow used for unique name
     remote_archive_filename = None
 
+    submit_path = config["submitPath"]
+
     submit_cmd = generic_workflow.run_attrs.get("bps_iscustom", False)
     if submit_cmd:
         files = generic_workflow.get_executables(data=False, transfer_only=True)
-        submit_path = config["submitPath"]
         archive_filename = f"jobO.{uuid.uuid4()}.tar.gz"
         archive_filename = create_archive_file(submit_path, archive_filename, files)
         remote_archive_filename = copy_files_to_pandacache(archive_filename)
@@ -636,7 +643,6 @@ def add_idds_work(config, generic_workflow, idds_workflow):
         enable_event_service = [i.strip() for i in enable_event_service]
     if enable_job_name_map:
         doma_tree = DomaTree(name=generic_workflow.name)
-        submit_path = config[".bps_defined.submitPath"]
         _, order_id_map_filename = config.search(
             "orderIdMapFilename", opt={"default": PANDA_DEFAULT_ORDER_ID_MAP_FILE}
         )
@@ -655,6 +661,14 @@ def add_idds_work(config, generic_workflow, idds_workflow):
     # catch dependency issues to loop through again later.
     jobs_with_dependency_issues = {}
 
+    # Initialize quantum node map
+    qnode_map = {}
+    qnode_map_filename = None
+    if enable_qnode_map:
+        qnode_map_file = os.path.join(submit_path, "qnode_map.json")
+        qnode_map_filename = {"qnodemap": qnode_map_file}
+        files_to_pre_stage.update(qnode_map_filename)
+
     # Assume jobs with same label share config values
     for job_label in generic_workflow.labels:
         _LOG.debug("job_label = %s", job_label)
@@ -709,6 +723,7 @@ def add_idds_work(config, generic_workflow, idds_workflow):
                     max_payloads_per_panda_job=max_payloads_per_panda_job,
                     max_wms_job_wall_time=max_wms_job_wall_time,
                     remote_filename=remote_archive_filename,
+                    qnode_map_filename=qnode_map_filename,
                 )
                 name_works[work.task_name] = work
                 files_to_pre_stage.update(files)
@@ -718,6 +733,12 @@ def add_idds_work(config, generic_workflow, idds_workflow):
 
             pseudo_filename = _make_pseudo_filename(config, gwjob)
             job_to_pseudo_filename[gwjob.name] = pseudo_filename
+
+            if enable_qnode_map:
+                job_name_PH = "PH:" + gwjob.name
+                job_to_pseudo_filename[gwjob.name] = job_name_PH
+                qnode_map[job_name_PH] = pseudo_filename
+
             job_to_task[gwjob.name] = work.get_work_name()
             deps = []
             missing_deps = False
@@ -742,7 +763,8 @@ def add_idds_work(config, generic_workflow, idds_workflow):
                         }
                     )
             if not missing_deps:
-                f_name = f"{job_label}:orderIdMap_{order_id}" if enable_job_name_map else pseudo_filename
+                j_name = job_to_pseudo_filename[gwjob.name]
+                f_name = f"{job_label}:orderIdMap_{order_id}" if enable_job_name_map else j_name
                 work.dependency_map.append(
                     {
                         "name": f_name,
@@ -757,6 +779,10 @@ def add_idds_work(config, generic_workflow, idds_workflow):
                     "label": job_label,
                 }
 
+    if enable_qnode_map:
+        with open(qnode_map_file, "w", encoding="utf-8") as f:
+            json.dump(qnode_map, f, indent=2)
+
     # If there were any issues figuring out dependencies through earlier loop
     if jobs_with_dependency_issues:
         _LOG.warning("Could not prepare workflow in single pass.  Please notify developers.")
@@ -790,7 +816,7 @@ def add_idds_work(config, generic_workflow, idds_workflow):
 
             work.dependency_map.append(
                 {
-                    "name": f"{job_label}:orderIdMap_{order_id}" if enable_job_name_map else pseudo_filename,
+                    "name": f"{job_label}:orderIdMap_{order_id}" if enable_job_name_map else job_name,
                     "order_id": order_id,
                     "dependencies": deps,
                 }

tests/test_cmd_line_decoder.py

@@ -0,0 +1,52 @@
+# This file is part of ctrl_bps_panda.
+#
+# Developed for the LSST Data Management System.
+# This product includes software developed by the LSST Project
+# (https://www.lsst.org).
+# See the COPYRIGHT file at the top-level directory of this distribution
+# for details of code ownership.
+#
+# This software is dual licensed under the GNU General Public License and also
+# under a 3-clause BSD license. Recipients may choose which of these licenses
+# to use; please see the files gpl-3.0.txt and/or bsd_license.txt,
+# respectively.  If you choose the GPL option then the following text applies
+# (but note that there is still no warranty even if you opt for BSD instead):
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+"""Unit tests for edgenode/cmd_line_decoder.py."""
+
+import unittest
+
+from lsst.ctrl.bps.panda.edgenode.cmd_line_decoder import use_map_file
+
+
+class TestCmdLineDecoder(unittest.TestCase):
+    """Test command line decoder functions."""
+
+    def test_valid_input(self):
+        self.assertTrue(use_map_file("PH:file12345"))
+
+    def test_invalid_prefix(self):
+        self.assertFalse(use_map_file("XX:file12345"))
+
+    def test_missing_colon(self):
+        self.assertFalse(use_map_file("PH12345"))
+
+    def test_too_many_parts(self):
+        self.assertFalse(use_map_file("PH:file123:456"))
+
+
+if __name__ == "__main__":
+    unittest.main()