Skip to content

Lsfscriptadapter

LSF Scheduler interface implementation.

LSFScriptAdapter

Bases: SchedulerScriptAdapter

A ScriptAdapter class for interfacing with the LSF cluster scheduler.

Source code in maestrowf/interfaces/script/lsfscriptadapter.py
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
class LSFScriptAdapter(SchedulerScriptAdapter):
    """A ScriptAdapter class for interfacing with the LSF cluster scheduler."""

    NOJOB_REGEX = re.compile(r"^No\s")
    key = "lsf"

    def __init__(self, **kwargs):
        """
        Initialize an instance of the SlurmScriptAdapter.

        The SlurmScriptAdapter is this package's interface to the Slurm
        scheduler. This adapter constructs Slurm scripts for a StudyStep based
        on user set defaults and local settings present in each step.

        The expected keyword arguments that are expected when the Slurm adapter
        is instantiated are as follows:
        * host: The cluster to execute scripts on.
        * bank: The account to charge computing time to.
        * queue: Scheduler queue scripts should be submitted to.
        * tasks: The number of compute nodes to be reserved for computing.

        :param **kwargs: A dictionary with default settings for the adapter.
        """
        super(LSFScriptAdapter, self).__init__()

        # NOTE: Host doesn't seem to matter for LSF
        self.add_batch_parameter("host", kwargs.pop("host"))
        self.add_batch_parameter("bank", kwargs.pop("bank"))
        self.add_batch_parameter("queue", kwargs.pop("queue"))
        self.add_batch_parameter("nodes", kwargs.pop("nodes", "1"))

        reservation = kwargs.get("reservation", None)
        if reservation:
            self.add_batch_parameter("reservation", reservation)

        self._header = {
            "nodes": "#BSUB -nnodes {nodes}",
            "queue": "#BSUB -q {queue}",
            "bank": "#BSUB -G {bank}",
            "walltime": "#BSUB -W {walltime}",
            "job-name": "#BSUB -J {job-name}",
            "output": "#BSUB -o {output}",
            "reservation": "#BSUB -U {reservation}",
            "error": "#BSUB -e {error}",
        }

        self._cmd_flags = {
            "cmd":          "jsrun",
            "rs per node":  "-r",
            "ntasks":       "--nrs",
            "tasks per rs": "-a",
            "gpus":         "-g",
            "cpus per rs":  "-c",
            "reservation":  "-J",
            "bind":         "-b",
            "bind gpus":    "-B",
        }

        self._extension = "lsf.sh"

    def get_header(self, step):
        """
        Generate the header present at the top of LSF execution scripts.

        :param step: A StudyStep instance.
        :returns: A string of the header based on internal batch parameters and
                  the parameter step.
        """
        batch_header = dict(self._batch)
        batch_header["nodes"] = step.run.get("nodes", self._batch["nodes"])
        batch_header["job-name"] = step.name.replace(" ", "_")
        batch_header["output"] = "{}.%J.out".format(batch_header["job-name"])
        batch_header["error"] = "{}.%J.err".format(batch_header["job-name"])

        # Updte the batch header with the values from the step's resources
        batch_header.update(
            {
                resource: value for (resource, value) in step.run.items()
                if value
            }
        )

        # LSF requires an hour and minutes format. We need to attempt to split
        # and correct if we get something that's coming in as HH:MM:SS
        walltime = step.run.get("walltime")
        wt_split = walltime.split(":")
        if len(wt_split) == 3:
            # If wall time is specified in three parts, we'll just calculate
            # the minutes off of the seconds and then shift up to hours if
            # needed.
            seconds_minutes = ceil(float(wt_split[2])/60)
            total_minutes = int(wt_split[1]) + seconds_minutes
            hours = int(wt_split[0]) + int(total_minutes/60)
            total_minutes %= 60
            walltime = "{:02d}:{:02d}".format(hours, int(total_minutes))

        batch_header["walltime"] = walltime

        modified_header = ["#!{}".format(self._exec)]
        for key, value in self._header.items():
            if key in batch_header:
                modified_header.append(value.format(**batch_header))

        return "\n".join(modified_header)

    def get_parallelize_command(self, procs, nodes=None, **kwargs):
        """
        Generate the LSF parallelization segement of the command line.

        :param procs: Number of processors to allocate to the parallel call.
        :param nodes: Number of nodes to allocate to the parallel call
                      (default = 1).
        :returns: A string of the parallelize command configured using nodes
                  and procs.
        """

        args = [self._cmd_flags["cmd"]]

        # Processors segment, checking
        # Need to account for processors per rs -> tasks per rs
        rs_per_node = kwargs.get("rs per node", 1)
        tasks_per_rs = kwargs.get("tasks per rs", 1)

        if int(procs) > int((int(rs_per_node)*int(nodes)*int(tasks_per_rs))):

            LOGGER.error("Resource Specification Error: 'procs' (%s)"
                         " must be a multiple of "
                         "'rs per node' * 'nodes' * 'tasks per rs' (%s)"
                         " where 'rs per node' = %s, 'nodes' = %s, and"
                         " 'tasks per rs' = %s",
                         procs,
                         rs_per_node*nodes*tasks_per_rs,
                         rs_per_node,
                         nodes,
                         tasks_per_rs)

        if nodes:
            rs_tasks = int(rs_per_node)*int(nodes)*int(tasks_per_rs)
            if (int(procs) > rs_tasks) or (int(procs) % rs_tasks) > 0:
                LOGGER.error("Resource Specification Error: 'procs' (%s)"
                             " must be a multiple of "
                             "'rs per node' * 'nodes' * 'tasks per rs' (%s)"
                             " where 'rs per node' = %s, 'nodes' = %s, and"
                             " 'tasks per rs' = %s",
                             procs,
                             int(rs_per_node)*int(nodes)*int(tasks_per_rs),
                             rs_per_node,
                             nodes,
                             tasks_per_rs)

        else:
            # NOTE: is this case even allowed on lsf allocations? will it auto
            #       compute the number of nodes on the allocation if scheduling
            #       this to a reservation?  If so, might want to revisit
            rs_tasks = int(rs_per_node)*int(tasks_per_rs)
            if int(procs) > rs_tasks or int(procs) % rs_tasks > 0:
                LOGGER.error("Resource Specification Error: 'procs' (%s)"
                             " must be a multiple of "
                             "'rs per node' * 'tasks per rs' (%s)"
                             " where 'rs per node' = {}, and"
                             " 'tasks per rs' = %s",
                             procs,
                             int(rs_per_node)*int(tasks_per_rs),
                             rs_per_node,
                             tasks_per_rs)

        args += [
            self._cmd_flags["ntasks"],
            str(procs)
        ]

        # Binding
        bind = kwargs.get("bind", "rs")
        args += [
            self._cmd_flags["bind"],
            str(bind)
        ]

        # If we have GPUs being requested, add them to the command.
        gpus = kwargs.get("gpus", 0)
        if not gpus:     # Initialized to "" in study step, FIX THIS
            gpus = 0
        if gpus:
            args += [
                self._cmd_flags["gpus"],
                str(gpus)
            ]

        # Optional gpu binding -> LSF 10.1+
        bind_gpus = kwargs.get("bind gpus", None)
        if bind_gpus:
            args += [
                self._cmd_flags["bind gpus"],
                str(bind_gpus)
            ]

        # handle mappings from node/procs to tasks/rs/nodes

        cpus_per_rs = kwargs.get("cpus per rs", 1)
        if not cpus_per_rs:     # Initialized to "" in study step, FIX THIS
            cpus_per_rs = 1

        args += [self._cmd_flags['tasks per rs'],
                 str(tasks_per_rs)]

        args += [self._cmd_flags['rs per node'],
                 str(rs_per_node)]

        args += [self._cmd_flags['cpus per rs'],
                 str(cpus_per_rs)]

        return " ".join(args)

    def submit(self, step, path, cwd, job_map=None, env=None):
        """
        Submit a script to the LSF scheduler.

        :param step: The StudyStep instance this submission is based on.
        :param path: Local path to the script to be executed.
        :param cwd: Path to the current working directory.
        :param job_map: A dictionary mapping step names to their job
                        identifiers.
        :param env: A dict containing a modified environment for execution.
        :returns: The return status of the submission command and job
                  identiifer.
        """
        args = ["bsub"]
        args += ["-cwd", cwd, "<", path]
        cmd = " ".join(args)
        LOGGER.debug("cwd = %s", cwd)
        LOGGER.debug("Command to execute: %s", cmd)
        p = Popen(cmd, shell=True, stdout=PIPE, stderr=PIPE, cwd=cwd, env=env)
        output, err = p.communicate()
        retcode = p.wait()
        output = output.decode("utf-8")

        # TODO: We need to check for dependencies here. The sbatch is where
        # dependent batch jobs are specified. If we're trying to launch
        # everything at once then that should happen here.

        if retcode == 0:
            LOGGER.info("Submission returned status OK.")
            return SubmissionRecord(
                SubmissionCode.OK, retcode,
                re.search('[0-9]+', output).group(0))
        else:
            LOGGER.warning("Submission returned an error.")
            return SubmissionRecord(SubmissionCode.ERROR, retcode, -1)

    def check_jobs(self, joblist):
        """
        For the given job list, query execution status.

        This method uses the scontrol show job <jobid> command and does a
        regex search for job information.

        :param joblist: A list of job identifiers to be queried.
        :returns: The return code of the status query, and a dictionary of job
                  identifiers to their status.
        """
        # TODO: This method needs to be updated to use sacct.
        # squeue options:
        # -u = username to search queues for.
        # -t = list of job states to search for. 'all' for all states.
        # -o = status output formatting
        o_format = "jobid:7 stat:5 exit_code:10 exit_reason:50 delimiter='|'"
        stat_cmd = "bjobs -a -u $USER -o \"{}\""
        cmd = stat_cmd.format(o_format)
        LOGGER.debug("bjobs cmd = \"%s\"", cmd)
        p = Popen(cmd, shell=True, stdout=PIPE, stderr=PIPE)
        output, err = p.communicate()
        retcode = p.wait()
        output = output.decode("utf-8")

        status = {}
        for jobid in joblist:
            LOGGER.debug("Looking for jobid %s", jobid)
            status[jobid] = None

        state_index = 1
        jobid_index = 0
        term_reason = 3
        if retcode == 0:
            # It seems that LSF may still return 0 even if it found nothing.
            # We'll explicitly check for a "^No " regex in the event that the
            # system is configured to return 0.
            no_jobs = re.search(self.NOJOB_REGEX, output)
            if no_jobs:
                LOGGER.warning("User '%s' has no jobs executing. Returning.",
                               getpass.getuser())
                return JobStatusCode.NOJOBS, {}

            # Otherwise, we can just process as normal.
            for job in output.split("\n")[1:]:
                LOGGER.debug("Job Entry: %s", job)
                # The squeue command output is split with the following indices
                # used for specific information:
                # 0 - Job Identifier
                # 1 - Status of the job
                # 2 - Exit code application terminated with
                # 3 - Reason for termination (if applicable)
                job_split = [x.strip() for x in job.split("|")]
                LOGGER.debug("Entry split: %s", job_split)
                if len(job_split) < 4:
                    LOGGER.debug(
                        "Entry has less than 4 fields. Skipping.",
                        job_split)
                    continue

                while job_split[0] == "":
                    LOGGER.debug("Removing blank entry from head of status.")
                    job_split = job_split[1:]

                if not job_split:
                    LOGGER.debug("Continuing...")
                    continue

                if job_split[jobid_index] in status:
                    if job_split[state_index] == "EXIT":
                        if "TERM_RUNLIMIT" in job_split[term_reason]:
                            _j_state = "TIMEOUT"
                        elif "TERM_OWNER" in job_split[term_reason]:
                            _j_state = "CANCELLED"
                        else:
                            _j_state = job_split[state_index]
                    else:
                        _j_state = job_split[state_index]
                    _state = self._state(_j_state)
                    LOGGER.debug("ID Found. %s -- %s",
                                 job_split[state_index],
                                 _state)
                    status[job_split[jobid_index]] = _state

            return JobStatusCode.OK, status
        # NOTE: We're keeping this here for now since we could see it in the
        # future...
        elif retcode == 255:
            LOGGER.warning("User '%s' has no jobs executing. Returning.",
                           getpass.getuser())
            return JobStatusCode.NOJOBS, status
        else:
            LOGGER.error("Error code '%s' seen. Unexpected behavior "
                         "encountered.", retcode)
            return JobStatusCode.ERROR, status

    def cancel_jobs(self, joblist):
        """
        For the given job list, cancel each job.

        :param joblist: A list of job identifiers to be cancelled.
        :returns: The return code to indicate if jobs were cancelled.
        """
        # If we don't have any jobs to check, just return status OK.
        if not joblist:
            return CancelCode.OK

        cmd = "bkill {}".format(" ".join(joblist))
        p = Popen(cmd, shell=True, stdout=PIPE, stderr=PIPE)
        output, err = p.communicate()
        retcode = p.wait()

        if retcode == 0:
            return CancellationRecord(CancelCode.OK, retcode)
        else:
            LOGGER.error("Error code '%s' seen. Unexpected behavior "
                         "encountered.", retcode)
            return CancellationRecord(CancelCode.ERROR, retcode)

    def _state(self, lsf_state):
        """
        Map a scheduler specific job state to a Study.State enum.

        :param slurm_state: String representation of scheduler job status.
        :returns: A Study.State enum corresponding to parameter job_state.
        """
        # NOTE: fdinatale -- If I'm understanding this correctly, there are
        # four naturally occurring states (excluding states of suspension.)
        # This is somewhat problematic because we don't actually get a time out
        # status here. We probably need to start considering what to do with
        # the post and pre monikers in steps.
        LOGGER.debug("Received LSF State -- %s", lsf_state)
        if lsf_state == "RUN":
            return State.RUNNING
        elif lsf_state == "PEND":
            return State.PENDING
        elif lsf_state == "DONE":
            return State.FINISHED
        elif lsf_state == "CANCELLED":
            return State.CANCELLED
        elif lsf_state == "EXIT":
            return State.FAILED
        elif lsf_state == "TIMEOUT":
            return State.TIMEDOUT
        elif lsf_state == "WAIT" or lsf_state == "PROV":
            return State.WAITING
        elif lsf_state == "UNKWN":
            return State.UNKNOWN
        else:
            return State.UNKNOWN

    def _write_script(self, ws_path, step):
        """
        Write a LSF script to the workspace of a workflow step.

        The job_map optional parameter is a map of workflow step names to job
        identifiers. This parameter so far is only planned to be used when a
        study is configured to be launched in one go (more or less a script
        chain using a scheduler's dependency setting). The functionality of
        the parameter may change depending on both future intended use.

        :param ws_path: Path to the workspace directory of the step.
        :param step: An instance of a StudyStep.
        :returns: Boolean value (True if to be scheduled), the path to the
                  written script for run["cmd"], and the path to the script
                  written for run["restart"] (if it exists).
        """
        to_be_scheduled, cmd, restart = self.get_scheduler_command(step)

        fname = "{}.{}".format(step.name, self._extension)
        script_path = os.path.join(ws_path, fname)
        with open(script_path, "w") as script:
            if to_be_scheduled:
                script.write(self.get_header(step))
            else:
                script.write("#!{}".format(self._exec))

            cmd = "\n\n{}\n".format(cmd)
            script.write(cmd)

        if restart:
            rname = "{}.restart.{}".format(step.name, self._extension)
            restart_path = os.path.join(ws_path, rname)

            with open(restart_path, "w") as script:
                if to_be_scheduled:
                    script.write(self.get_header(step))
                else:
                    script.write("#!{}".format(self._exec))

                cmd = "\n\n{}\n".format(restart)
                script.write(cmd)
        else:
            restart_path = None

        return to_be_scheduled, script_path, restart_path

    @property
    def extension(self):
        return self._extension

__init__(**kwargs)

Initialize an instance of the SlurmScriptAdapter.

The SlurmScriptAdapter is this package's interface to the Slurm scheduler. This adapter constructs Slurm scripts for a StudyStep based on user set defaults and local settings present in each step.

The expected keyword arguments that are expected when the Slurm adapter is instantiated are as follows: * host: The cluster to execute scripts on. * bank: The account to charge computing time to. * queue: Scheduler queue scripts should be submitted to. * tasks: The number of compute nodes to be reserved for computing.

Parameters:

Name Type Description Default
**kwargs

A dictionary with default settings for the adapter.

{}
Source code in maestrowf/interfaces/script/lsfscriptadapter.py
def __init__(self, **kwargs):
    """
    Initialize an instance of the SlurmScriptAdapter.

    The SlurmScriptAdapter is this package's interface to the Slurm
    scheduler. This adapter constructs Slurm scripts for a StudyStep based
    on user set defaults and local settings present in each step.

    The expected keyword arguments that are expected when the Slurm adapter
    is instantiated are as follows:
    * host: The cluster to execute scripts on.
    * bank: The account to charge computing time to.
    * queue: Scheduler queue scripts should be submitted to.
    * tasks: The number of compute nodes to be reserved for computing.

    :param **kwargs: A dictionary with default settings for the adapter.
    """
    super(LSFScriptAdapter, self).__init__()

    # NOTE: Host doesn't seem to matter for LSF
    self.add_batch_parameter("host", kwargs.pop("host"))
    self.add_batch_parameter("bank", kwargs.pop("bank"))
    self.add_batch_parameter("queue", kwargs.pop("queue"))
    self.add_batch_parameter("nodes", kwargs.pop("nodes", "1"))

    reservation = kwargs.get("reservation", None)
    if reservation:
        self.add_batch_parameter("reservation", reservation)

    self._header = {
        "nodes": "#BSUB -nnodes {nodes}",
        "queue": "#BSUB -q {queue}",
        "bank": "#BSUB -G {bank}",
        "walltime": "#BSUB -W {walltime}",
        "job-name": "#BSUB -J {job-name}",
        "output": "#BSUB -o {output}",
        "reservation": "#BSUB -U {reservation}",
        "error": "#BSUB -e {error}",
    }

    self._cmd_flags = {
        "cmd":          "jsrun",
        "rs per node":  "-r",
        "ntasks":       "--nrs",
        "tasks per rs": "-a",
        "gpus":         "-g",
        "cpus per rs":  "-c",
        "reservation":  "-J",
        "bind":         "-b",
        "bind gpus":    "-B",
    }

    self._extension = "lsf.sh"

cancel_jobs(joblist)

For the given job list, cancel each job.

Parameters:

Name Type Description Default
joblist

A list of job identifiers to be cancelled.

required

Returns:

Type Description

The return code to indicate if jobs were cancelled.

Source code in maestrowf/interfaces/script/lsfscriptadapter.py
def cancel_jobs(self, joblist):
    """
    For the given job list, cancel each job.

    :param joblist: A list of job identifiers to be cancelled.
    :returns: The return code to indicate if jobs were cancelled.
    """
    # If we don't have any jobs to check, just return status OK.
    if not joblist:
        return CancelCode.OK

    cmd = "bkill {}".format(" ".join(joblist))
    p = Popen(cmd, shell=True, stdout=PIPE, stderr=PIPE)
    output, err = p.communicate()
    retcode = p.wait()

    if retcode == 0:
        return CancellationRecord(CancelCode.OK, retcode)
    else:
        LOGGER.error("Error code '%s' seen. Unexpected behavior "
                     "encountered.", retcode)
        return CancellationRecord(CancelCode.ERROR, retcode)

check_jobs(joblist)

For the given job list, query execution status.

This method uses the scontrol show job command and does a regex search for job information.

Parameters:

Name Type Description Default
joblist

A list of job identifiers to be queried.

required

Returns:

Type Description

The return code of the status query, and a dictionary of job identifiers to their status.

Source code in maestrowf/interfaces/script/lsfscriptadapter.py
def check_jobs(self, joblist):
    """
    For the given job list, query execution status.

    This method uses the scontrol show job <jobid> command and does a
    regex search for job information.

    :param joblist: A list of job identifiers to be queried.
    :returns: The return code of the status query, and a dictionary of job
              identifiers to their status.
    """
    # TODO: This method needs to be updated to use sacct.
    # squeue options:
    # -u = username to search queues for.
    # -t = list of job states to search for. 'all' for all states.
    # -o = status output formatting
    o_format = "jobid:7 stat:5 exit_code:10 exit_reason:50 delimiter='|'"
    stat_cmd = "bjobs -a -u $USER -o \"{}\""
    cmd = stat_cmd.format(o_format)
    LOGGER.debug("bjobs cmd = \"%s\"", cmd)
    p = Popen(cmd, shell=True, stdout=PIPE, stderr=PIPE)
    output, err = p.communicate()
    retcode = p.wait()
    output = output.decode("utf-8")

    status = {}
    for jobid in joblist:
        LOGGER.debug("Looking for jobid %s", jobid)
        status[jobid] = None

    state_index = 1
    jobid_index = 0
    term_reason = 3
    if retcode == 0:
        # It seems that LSF may still return 0 even if it found nothing.
        # We'll explicitly check for a "^No " regex in the event that the
        # system is configured to return 0.
        no_jobs = re.search(self.NOJOB_REGEX, output)
        if no_jobs:
            LOGGER.warning("User '%s' has no jobs executing. Returning.",
                           getpass.getuser())
            return JobStatusCode.NOJOBS, {}

        # Otherwise, we can just process as normal.
        for job in output.split("\n")[1:]:
            LOGGER.debug("Job Entry: %s", job)
            # The squeue command output is split with the following indices
            # used for specific information:
            # 0 - Job Identifier
            # 1 - Status of the job
            # 2 - Exit code application terminated with
            # 3 - Reason for termination (if applicable)
            job_split = [x.strip() for x in job.split("|")]
            LOGGER.debug("Entry split: %s", job_split)
            if len(job_split) < 4:
                LOGGER.debug(
                    "Entry has less than 4 fields. Skipping.",
                    job_split)
                continue

            while job_split[0] == "":
                LOGGER.debug("Removing blank entry from head of status.")
                job_split = job_split[1:]

            if not job_split:
                LOGGER.debug("Continuing...")
                continue

            if job_split[jobid_index] in status:
                if job_split[state_index] == "EXIT":
                    if "TERM_RUNLIMIT" in job_split[term_reason]:
                        _j_state = "TIMEOUT"
                    elif "TERM_OWNER" in job_split[term_reason]:
                        _j_state = "CANCELLED"
                    else:
                        _j_state = job_split[state_index]
                else:
                    _j_state = job_split[state_index]
                _state = self._state(_j_state)
                LOGGER.debug("ID Found. %s -- %s",
                             job_split[state_index],
                             _state)
                status[job_split[jobid_index]] = _state

        return JobStatusCode.OK, status
    # NOTE: We're keeping this here for now since we could see it in the
    # future...
    elif retcode == 255:
        LOGGER.warning("User '%s' has no jobs executing. Returning.",
                       getpass.getuser())
        return JobStatusCode.NOJOBS, status
    else:
        LOGGER.error("Error code '%s' seen. Unexpected behavior "
                     "encountered.", retcode)
        return JobStatusCode.ERROR, status

get_header(step)

Generate the header present at the top of LSF execution scripts.

Parameters:

Name Type Description Default
step

A StudyStep instance.

required

Returns:

Type Description

A string of the header based on internal batch parameters and the parameter step.

Source code in maestrowf/interfaces/script/lsfscriptadapter.py
def get_header(self, step):
    """
    Generate the header present at the top of LSF execution scripts.

    :param step: A StudyStep instance.
    :returns: A string of the header based on internal batch parameters and
              the parameter step.
    """
    batch_header = dict(self._batch)
    batch_header["nodes"] = step.run.get("nodes", self._batch["nodes"])
    batch_header["job-name"] = step.name.replace(" ", "_")
    batch_header["output"] = "{}.%J.out".format(batch_header["job-name"])
    batch_header["error"] = "{}.%J.err".format(batch_header["job-name"])

    # Updte the batch header with the values from the step's resources
    batch_header.update(
        {
            resource: value for (resource, value) in step.run.items()
            if value
        }
    )

    # LSF requires an hour and minutes format. We need to attempt to split
    # and correct if we get something that's coming in as HH:MM:SS
    walltime = step.run.get("walltime")
    wt_split = walltime.split(":")
    if len(wt_split) == 3:
        # If wall time is specified in three parts, we'll just calculate
        # the minutes off of the seconds and then shift up to hours if
        # needed.
        seconds_minutes = ceil(float(wt_split[2])/60)
        total_minutes = int(wt_split[1]) + seconds_minutes
        hours = int(wt_split[0]) + int(total_minutes/60)
        total_minutes %= 60
        walltime = "{:02d}:{:02d}".format(hours, int(total_minutes))

    batch_header["walltime"] = walltime

    modified_header = ["#!{}".format(self._exec)]
    for key, value in self._header.items():
        if key in batch_header:
            modified_header.append(value.format(**batch_header))

    return "\n".join(modified_header)

get_parallelize_command(procs, nodes=None, **kwargs)

Generate the LSF parallelization segement of the command line.

Parameters:

Name Type Description Default
procs

Number of processors to allocate to the parallel call.

required
nodes

Number of nodes to allocate to the parallel call (default = 1).

None

Returns:

Type Description

A string of the parallelize command configured using nodes and procs.

Source code in maestrowf/interfaces/script/lsfscriptadapter.py
def get_parallelize_command(self, procs, nodes=None, **kwargs):
    """
    Generate the LSF parallelization segement of the command line.

    :param procs: Number of processors to allocate to the parallel call.
    :param nodes: Number of nodes to allocate to the parallel call
                  (default = 1).
    :returns: A string of the parallelize command configured using nodes
              and procs.
    """

    args = [self._cmd_flags["cmd"]]

    # Processors segment, checking
    # Need to account for processors per rs -> tasks per rs
    rs_per_node = kwargs.get("rs per node", 1)
    tasks_per_rs = kwargs.get("tasks per rs", 1)

    if int(procs) > int((int(rs_per_node)*int(nodes)*int(tasks_per_rs))):

        LOGGER.error("Resource Specification Error: 'procs' (%s)"
                     " must be a multiple of "
                     "'rs per node' * 'nodes' * 'tasks per rs' (%s)"
                     " where 'rs per node' = %s, 'nodes' = %s, and"
                     " 'tasks per rs' = %s",
                     procs,
                     rs_per_node*nodes*tasks_per_rs,
                     rs_per_node,
                     nodes,
                     tasks_per_rs)

    if nodes:
        rs_tasks = int(rs_per_node)*int(nodes)*int(tasks_per_rs)
        if (int(procs) > rs_tasks) or (int(procs) % rs_tasks) > 0:
            LOGGER.error("Resource Specification Error: 'procs' (%s)"
                         " must be a multiple of "
                         "'rs per node' * 'nodes' * 'tasks per rs' (%s)"
                         " where 'rs per node' = %s, 'nodes' = %s, and"
                         " 'tasks per rs' = %s",
                         procs,
                         int(rs_per_node)*int(nodes)*int(tasks_per_rs),
                         rs_per_node,
                         nodes,
                         tasks_per_rs)

    else:
        # NOTE: is this case even allowed on lsf allocations? will it auto
        #       compute the number of nodes on the allocation if scheduling
        #       this to a reservation?  If so, might want to revisit
        rs_tasks = int(rs_per_node)*int(tasks_per_rs)
        if int(procs) > rs_tasks or int(procs) % rs_tasks > 0:
            LOGGER.error("Resource Specification Error: 'procs' (%s)"
                         " must be a multiple of "
                         "'rs per node' * 'tasks per rs' (%s)"
                         " where 'rs per node' = {}, and"
                         " 'tasks per rs' = %s",
                         procs,
                         int(rs_per_node)*int(tasks_per_rs),
                         rs_per_node,
                         tasks_per_rs)

    args += [
        self._cmd_flags["ntasks"],
        str(procs)
    ]

    # Binding
    bind = kwargs.get("bind", "rs")
    args += [
        self._cmd_flags["bind"],
        str(bind)
    ]

    # If we have GPUs being requested, add them to the command.
    gpus = kwargs.get("gpus", 0)
    if not gpus:     # Initialized to "" in study step, FIX THIS
        gpus = 0
    if gpus:
        args += [
            self._cmd_flags["gpus"],
            str(gpus)
        ]

    # Optional gpu binding -> LSF 10.1+
    bind_gpus = kwargs.get("bind gpus", None)
    if bind_gpus:
        args += [
            self._cmd_flags["bind gpus"],
            str(bind_gpus)
        ]

    # handle mappings from node/procs to tasks/rs/nodes

    cpus_per_rs = kwargs.get("cpus per rs", 1)
    if not cpus_per_rs:     # Initialized to "" in study step, FIX THIS
        cpus_per_rs = 1

    args += [self._cmd_flags['tasks per rs'],
             str(tasks_per_rs)]

    args += [self._cmd_flags['rs per node'],
             str(rs_per_node)]

    args += [self._cmd_flags['cpus per rs'],
             str(cpus_per_rs)]

    return " ".join(args)

submit(step, path, cwd, job_map=None, env=None)

Submit a script to the LSF scheduler.

Parameters:

Name Type Description Default
step

The StudyStep instance this submission is based on.

required
path

Local path to the script to be executed.

required
cwd

Path to the current working directory.

required
job_map

A dictionary mapping step names to their job identifiers.

None
env

A dict containing a modified environment for execution.

None

Returns:

Type Description

The return status of the submission command and job identiifer.

Source code in maestrowf/interfaces/script/lsfscriptadapter.py
def submit(self, step, path, cwd, job_map=None, env=None):
    """
    Submit a script to the LSF scheduler.

    :param step: The StudyStep instance this submission is based on.
    :param path: Local path to the script to be executed.
    :param cwd: Path to the current working directory.
    :param job_map: A dictionary mapping step names to their job
                    identifiers.
    :param env: A dict containing a modified environment for execution.
    :returns: The return status of the submission command and job
              identiifer.
    """
    args = ["bsub"]
    args += ["-cwd", cwd, "<", path]
    cmd = " ".join(args)
    LOGGER.debug("cwd = %s", cwd)
    LOGGER.debug("Command to execute: %s", cmd)
    p = Popen(cmd, shell=True, stdout=PIPE, stderr=PIPE, cwd=cwd, env=env)
    output, err = p.communicate()
    retcode = p.wait()
    output = output.decode("utf-8")

    # TODO: We need to check for dependencies here. The sbatch is where
    # dependent batch jobs are specified. If we're trying to launch
    # everything at once then that should happen here.

    if retcode == 0:
        LOGGER.info("Submission returned status OK.")
        return SubmissionRecord(
            SubmissionCode.OK, retcode,
            re.search('[0-9]+', output).group(0))
    else:
        LOGGER.warning("Submission returned an error.")
        return SubmissionRecord(SubmissionCode.ERROR, retcode, -1)