Implemented a single qgraph file per workflow #30
Conversation
There was a problem hiding this comment.
Shouldn't the example yaml be updated as well (or another example written if PanDA plugin supports both small job files as well as the new single run file)?
Some questions or recommendations for minor changes. As well as a couple comments that the code will break with future changes, but we agreed it was ok for this ticket.
| @@ -2,6 +2,7 @@ | |||
| # This is a starter script needed to initialize basic Rubin software environment inside the container and execute | |||
| # the actual command line after decoding from hexed strings | |||
| cd /tmp; | |||
| ls /; | |||
There was a problem hiding this comment.
Just double checking that this was intended to be committed (as opposed to accidental committing of a temporary debugging line).
There was a problem hiding this comment.
Right, this line is removed.
| @@ -1,6 +1,7 @@ | |||
| """ | |||
|
|
|||
| """ | |||
| import ntpath | |||
There was a problem hiding this comment.
Why ntpath instead of os.path or pathlib?
| @@ -128,17 +130,20 @@ def add_dependencies(self, tasks, tasks_dependency_map): | |||
| """ | |||
| for task in tasks: | |||
| jobs = tasks_dependency_map[task.step] | |||
| dependencies = [] | |||
| task_dependencies = [] | |||
There was a problem hiding this comment.
Why not just use task.dependencies?
| dependency_map.setdefault(self.get_pseudo_input_file_name(edge[1]), []).\ | ||
| append(self.get_pseudo_input_file_name(edge[0])) | ||
| self.jobs_steps[self.get_pseudo_input_file_name(edge[1])] = \ | ||
| self.bps_workflow.nodes.get(edge[1]).get('job').label |
There was a problem hiding this comment.
I didn't quite follow this code, but wondering if this could be self.bps_workflow.get_job(edge[1]).label
There was a problem hiding this comment.
Thanks for the tip. Fixed.
| dependency_map.setdefault(node, []) | ||
| dependency_map.setdefault(self.get_pseudo_input_file_name(node), []) | ||
| self.jobs_steps[self.get_pseudo_input_file_name(node)] = \ | ||
| self.bps_workflow.nodes.get(node).get('job').label |
There was a problem hiding this comment.
Same question about get_job.
| qgraph_node_ids = self.bps_workflow.nodes.get(jobname).get("job").qgraph_node_ids | ||
| if qgraph_node_ids: | ||
| pseudo_input_file_name = self.qgraph_file+"+"+jobname + "+" + qgraph_node_ids[0].buildId + \ | ||
| "+" + str(qgraph_node_ids[0].number) |
There was a problem hiding this comment.
We agreed this was fine for this ticket, but just pointing it out for future. By limiting the node_ids to a single value, this code works today but definitely will not work as soon as a single GenericWorkflow job executes more than one Quantum.
| @@ -196,8 +201,7 @@ def from_generic_workflow(cls, config, generic_workflow, out_prefix, service_cla | |||
| idds_workflow.generated_tasks = workflow_generator.define_tasks() | |||
| cloud_prefix = config['bucket'] + '/' + \ | |||
| config['payload_folder'] + '/' + config['workflowName'] + '/' | |||
| for task in idds_workflow.generated_tasks: | |||
| cls.copy_pickles_into_cloud(task.local_pfns, cloud_prefix) | |||
| cls.copy_pickles_into_cloud([config['bps_defined']['run_qgraph_file']], cloud_prefix) | |||
There was a problem hiding this comment.
This is very fragile code as opposed to checking the inputs for the jobs. I expect that this section will be revisited in the upcoming execution butler work. We should talk then to figure out if there are GenericWorkflow issues that need to be fixed so that this can be written more generically.
| @@ -225,3 +234,12 @@ def get_input_file(self, job_name): | |||
| quantum graph file name | |||
| """ | |||
| return next(iter(self.bps_workflow.nodes.get(job_name).get("inputs"))) | |||
|
|
|||
| def get_pseudo_input_file_name(self, jobname): | |||
There was a problem hiding this comment.
"get" sounds like a lookup. Perhaps change to something like "create"
There was a problem hiding this comment.
Also, method needs docstring.
There was a problem hiding this comment.
Docstring added, function name updated.
| @@ -7,5 +7,9 @@ | |||
| import sys | |||
| import binascii | |||
| cmdline = str(binascii.unhexlify(sys.argv[1]).decode()) | |||
| cmdline = cmdline.replace("${IN/L}", sys.argv[2]) | |||
| dataparams = sys.argv[2].split(":") | |||
There was a problem hiding this comment.
Do dataparams come from the pseudo input filename? If yes, I didn't follow why ":" here, but "+" in get_pseudo_input_file_name(). Could use a couple short comments to make code easier to understand.
No description provided.