_toil_worker

from toil.common import Toil
from toil.job import Job


def helloWorld(message, memory="1G", cores=1, disk="1G"):
    return f"Hello, world!, here's a message: {message}"


if __name__ == "__main__":
    parser = Job.Runner.getDefaultArgumentParser()
    options = parser.parse_args()
    options.clean = "always"
    with Toil(options) as toil:
        output = toil.start(Job.wrapFn(helloWorld, "You did it!"))
    print(output)

def main(options=None):
    if not options:
        # deal with command line arguments
        parser = ArgumentParser()
        Job.Runner.addToilOptions(parser)
        parser.add_argument('--numLines', default=defaultLines, help='Number of lines in file to sort.', type=int)
        parser.add_argument('--lineLength', default=defaultLineLen, help='Length of lines in file to sort.', type=int)
        parser.add_argument("--fileToSort", help="The file you wish to sort")
        parser.add_argument("--outputFile", help="Where the sorted output will go")
        parser.add_argument("--overwriteOutput", help="Write over the output file if it already exists.", default=True)
        parser.add_argument("--N", dest="N",
                            help="The threshold below which a serial sort function is used to sort file. "
                                 "All lines must of length less than or equal to N or program will fail",
                            default=10000)
        parser.add_argument('--downCheckpoints', action='store_true',
                            help='If this option is set, the workflow will make checkpoints on its way through'
                                 'the recursive "down" part of the sort')
        parser.add_argument("--sortMemory", dest="sortMemory",
                        help="Memory for jobs that sort chunks of the file.",
                        default=None)

        parser.add_argument("--mergeMemory", dest="mergeMemory",
                        help="Memory for jobs that collate results.",
                        default=None)

        options = parser.parse_args()
    if not hasattr(options, "sortMemory") or not options.sortMemory:
        options.sortMemory = sortMemory
    if not hasattr(options, "mergeMemory") or not options.mergeMemory:
        options.mergeMemory = sortMemory

    # do some input verification
    sortedFileName = options.outputFile or "sortedFile.txt"
    if not options.overwriteOutput and os.path.exists(sortedFileName):
        print(f'Output file {sortedFileName} already exists.  '
              f'Delete it to run the sort example again or use --overwriteOutput=True')
        exit()

    fileName = options.fileToSort
    if options.fileToSort is None:
        # make the file ourselves
        fileName = 'fileToSort.txt'
        if os.path.exists(fileName):
            print(f'Sorting existing file: {fileName}')
        else:
            print(f'No sort file specified. Generating one automatically called: {fileName}.')
            makeFileToSort(fileName=fileName, lines=options.numLines, lineLen=options.lineLength)
    else:
        if not os.path.exists(options.fileToSort):
            raise RuntimeError("File to sort does not exist: %s" % options.fileToSort)

    if int(options.N) <= 0:
        raise RuntimeError("Invalid value of N: %s" % options.N)

    # Now we are ready to run
    with Toil(options) as workflow:
        sortedFileURL = 'file://' + os.path.abspath(sortedFileName)
        if not workflow.options.restart:
            sortFileURL = 'file://' + os.path.abspath(fileName)
            sortFileID = workflow.importFile(sortFileURL)
            sortedFileID = workflow.start(Job.wrapJobFn(setup,
                                                        sortFileID,
                                                        int(options.N),
                                                        options.downCheckpoints,
                                                        options=options,
                                                        memory=sortMemory))
        else:
            sortedFileID = workflow.restart()
        workflow.exportFile(sortedFileID, sortedFileURL)

def setup(job, inputFile, N, downCheckpoints, options):
    """
    Sets up the sort.
    Returns the FileID of the sorted file
    """
    RealtimeLogger.info("Starting the merge sort")
    return job.addChildJobFn(down,
                             inputFile, N, 'root',
                             downCheckpoints,
                             options = options,
                             preemptible=True,
                             memory=sortMemory).rv()

def down(job, inputFileStoreID, N, path, downCheckpoints, options, memory=sortMemory):
    """
    Input is a file, a subdivision size N, and a path in the hierarchy of jobs.
    If the range is larger than a threshold N the range is divided recursively and
    a follow on job is then created which merges back the results else
    the file is sorted and placed in the output.
    """

    RealtimeLogger.info("Down job starting: %s" % path)

    # Read the file
    inputFile = job.fileStore.readGlobalFile(inputFileStoreID, cache=False)
    length = os.path.getsize(inputFile)
    if length > N:
        # We will subdivide the file
        RealtimeLogger.critical("Splitting file: %s of size: %s"
                % (inputFileStoreID, length))
        # Split the file into two copies
        midPoint = getMidPoint(inputFile, 0, length)
        t1 = job.fileStore.getLocalTempFile()
        with open(t1, 'w') as fH:
            fH.write(copySubRangeOfFile(inputFile, 0, midPoint+1))
        t2 = job.fileStore.getLocalTempFile()
        with open(t2, 'w') as fH:
            fH.write(copySubRangeOfFile(inputFile, midPoint+1, length))
        # Call down recursively. By giving the rv() of the two jobs as inputs to the follow-on job, up,
        # we communicate the dependency without hindering concurrency.
        result = job.addFollowOnJobFn(up,
                                    job.addChildJobFn(down, job.fileStore.writeGlobalFile(t1), N, path + '/0',
                                                      downCheckpoints, checkpoint=downCheckpoints, options=options,
                                                      preemptible=True, memory=options.sortMemory).rv(),
                                    job.addChildJobFn(down, job.fileStore.writeGlobalFile(t2), N, path + '/1',
                                                      downCheckpoints, checkpoint=downCheckpoints, options=options,
                                                      preemptible=True, memory=options.mergeMemory).rv(),
                                    path + '/up', preemptible=True, options=options, memory=options.sortMemory).rv()
    else:
        # We can sort this bit of the file
        RealtimeLogger.critical("Sorting file: %s of size: %s"
                % (inputFileStoreID, length))
        # Sort the copy and write back to the fileStore
        shutil.copyfile(inputFile, inputFile + '.sort')
        sort(inputFile + '.sort')
        result = job.fileStore.writeGlobalFile(inputFile + '.sort')

    RealtimeLogger.info("Down job finished: %s" % path)
    return result

def up(job, inputFileID1, inputFileID2, path, options, memory=sortMemory):
    """
    Merges the two files and places them in the output.
    """

    RealtimeLogger.info("Up job starting: %s" % path)

    with job.fileStore.writeGlobalFileStream() as (fileHandle, outputFileStoreID):
        fileHandle = codecs.getwriter('utf-8')(fileHandle)
        with job.fileStore.readGlobalFileStream(inputFileID1) as inputFileHandle1:
            inputFileHandle1 = codecs.getreader('utf-8')(inputFileHandle1)
            with job.fileStore.readGlobalFileStream(inputFileID2) as inputFileHandle2:
                inputFileHandle2 = codecs.getreader('utf-8')(inputFileHandle2)
                RealtimeLogger.info("Merging %s and %s to %s"
                    % (inputFileID1, inputFileID2, outputFileStoreID))
                merge(inputFileHandle1, inputFileHandle2, fileHandle)
        # Cleanup up the input files - these deletes will occur after the completion is successful.
        job.fileStore.deleteGlobalFile(inputFileID1)
        job.fileStore.deleteGlobalFile(inputFileID2)

        RealtimeLogger.info("Up job finished: %s" % path)

        return outputFileStoreID

if __name__ == '__main__'
    main()

$ python3 sort.py file:jobStore \
             --logLevel=critical \
             --overwriteOutput=True

def down(job, inputFileStoreID, N, downCheckpoints, memory=sortMemory):
    ...
    assert 1 == 2, "Test error!"

$ python3 sort.py file:jobStore
...
---TOIL WORKER OUTPUT LOG---
...
m/j/jobonrSMP    Traceback (most recent call last):
m/j/jobonrSMP      File "toil/src/toil/worker.py", line 340, in main
m/j/jobonrSMP        job._runner(jobGraph=jobGraph, jobStore=jobStore, fileStore=fileStore)
m/j/jobonrSMP      File "toil/src/toil/job.py", line 1270, in _runner
m/j/jobonrSMP        returnValues = self._run(jobGraph, fileStore)
m/j/jobonrSMP      File "toil/src/toil/job.py", line 1217, in _run
m/j/jobonrSMP        return self.run(fileStore)
m/j/jobonrSMP      File "toil/src/toil/job.py", line 1383, in run
m/j/jobonrSMP        rValue = userFunction(*((self,) + tuple(self._args)), **self._kwargs)
m/j/jobonrSMP      File "toil/example.py", line 30, in down
m/j/jobonrSMP        assert 1 == 2, "Test error!"
m/j/jobonrSMP    AssertionError: Test error!

$ python3 sort.py file:jobStore \
             --restart \
             --overwriteOutput=True

$ python3 sort.py file:jobStore \
             --retryCount 2 \
             --restart \
             --overwriteOutput=True

$ python3 sort.py file:jobStore \
             --restart \
             --overwriteOutput=True

$ toil example.py --help

options = Job.Runner.getDefaultOptions("./toilWorkflow") # Get the options object

with Toil(options) as toil:
    toil.start(Job())  # Run the script

options = Job.Runner.getDefaultOptions("./toilWorkflow") # Get the options object
options.logLevel = "DEBUG" # Set the log level to the debug level.
options.clean = "ALWAYS" # Always delete the jobStore after a run

with Toil(options) as toil:
    toil.start(Job())  # Run the script

parser = Job.Runner.getDefaultArgumentParser() # Get the parser
options = parser.parse_args() # Parse user args to create the options object

with Toil(options) as toil:
    toil.start(Job())  # Run the script

parser = Job.Runner.getDefaultArgumentParser() # Get the parser
options = parser.parse_args() # Parse user args to create the options object
options.logLevel = "DEBUG" # Set the log level to the debug level.
options.clean = "ALWAYS" # Always delete the jobStore after a run

with Toil(options) as toil:
    toil.start(Job())  # Run the script

$ toil debug-file file:path-to-jobstore-directory \
      --listFilesInJobStore

$ toil debug-file file:path-to-jobstore \
      --fetch overview.txt *.bam *.fastq \
      --localFilePath=/home/user/localpath

•

Your own institution

•

Amazon EKS

•

Microsoft Azure AKS

•

Google GKE

•

DigitalOcean Kubernetes

•

minikube

•

Configuring for Amazon EKS

•

Configuring for Microsoft Azure AKS

•

Configuring for Google GKE

•

Configuring for DigitalOcean Kubernetes Clusters

•

Configuring for minikube

1.

On the IAM Dashboard page, choose your account name in the navigation bar, and then choose My Security Credentials.

2.

Expand the Access keys (access key ID and secret access key) section.

3.

Choose Create New Access Key. Then choose Download Key File to save the access key ID and secret access key to a file on your computer. After you close the dialog box, you can't retrieve this secret access key again.

apiVersion: batch/v1
kind: Job
metadata:
  # It is good practice to include your username in your job name.
  # Also specify it in TOIL_KUBERNETES_OWNER
  name: demo-user-toil-test
# Do not try and rerun the leader job if it fails

spec:
 backoffLimit: 0
 template:
   spec:
     # Do not restart the pod when the job fails, but keep it around so the
     # log can be retrieved
     restartPolicy: Never
     volumes:
     - name: aws-credentials-vol
       secret:
         # Make sure the AWS credentials are available as a volume.
         # This should match TOIL_AWS_SECRET_NAME
         secretName: aws-credentials
     # You may need to replace this with a different service account name as
     # appropriate for your cluster.
     serviceAccountName: default
     containers:
     - name: main
       image: quay.io/ucsc_cgl/toil:5.5.0
       env:
       # Specify your username for inclusion in job names
       - name: TOIL_KUBERNETES_OWNER
         value: demo-user
       # Specify where to find the AWS credentials to access the job store with
       - name: TOIL_AWS_SECRET_NAME
         value: aws-credentials
       # Specify where per-host caches should be stored, on the Kubernetes hosts.
       # Needs to be set for Toil's caching to be efficient.
       - name: TOIL_KUBERNETES_HOST_PATH
         value: /data/scratch
       volumeMounts:
       # Mount the AWS credentials volume
       - mountPath: /root/.aws
         name: aws-credentials-vol
       resources:
         # Make sure to set these resource limits to values large enough
         # to accommodate the work your workflow does in the leader
         # process, but small enough to fit on your cluster.
         #
         # Since no request values are specified, the limits are also used
         # for the requests.
         limits:
           cpu: 2
           memory: "4Gi"
           ephemeral-storage: "10Gi"
       command:
       - /bin/bash
       - -c
       - |
         # This Bash script will set up Toil and the workflow to run, and run them.
         set -e
         # We make sure to create a work directory; Toil can't hot-deploy a
         # script from the root of the filesystem, which is where we start.
         mkdir /tmp/work
         cd /tmp/work
         # We make a virtual environment to allow workflow dependencies to be
         # hot-deployed.
         #
         # We don't really make use of it in this example, but for workflows
         # that depend on PyPI packages we will need this.
         #
         # We use --system-site-packages so that the Toil installed in the
         # appliance image is still available.
         virtualenv --python python3 --system-site-packages venv
         . venv/bin/activate
         # Now we install the workflow. Here we're using a demo workflow
         # script from Toil itself.
         wget https://raw.githubusercontent.com/DataBiosphere/toil/releases/4.1.0/src/toil/test/docs/scripts/tutorial_helloworld.py
         # Now we run the workflow. We make sure to use the Kubernetes batch
         # system and an AWS job store, and we set some generally useful
         # logging options. We also make sure to enable caching.
         python3 tutorial_helloworld.py \
             aws:us-west-2:demouser-toil-test-jobstore \
             --batchSystem kubernetes \
             --realTimeLogging \
             --logInfo

$ kubectl apply -f leader.yaml

$ kubectl get pods | grep demo-user-toil-test
demo-user-toil-test-g5496                                         1/1     Running     0          2m

$ kubectl get pods | grep demo-user

$ kubectl logs demo-user-toil-test-g5496

$ kubectl logs -f demo-user-toil-test-g5496

$ kubectl describe pod demo-user-toil-test-g5496

Type     Reason            Age                  From               Message
----     ------            ----                 ----               -------
Warning  FailedScheduling  13s (x79 over 100m)  default-scheduler  0/4 nodes are available: 1 Insufficient cpu, 1 Insufficient ephemeral-storage, 4 Insufficient memory.

$ kubectl exec -ti demo-user-toil-test-g5496 /bin/bash

$ kubectl get jobs | grep demo-user | cut -f1 -d' ' | xargs -n10 kubectl delete job

$ export TOIL_KUBERNETES_OWNER=demo-user  # This defaults to your local username if not set
$ export TOIL_AWS_SECRET_NAME=aws-credentials
$ export TOIL_KUBERNETES_HOST_PATH=/data/scratch
$ virtualenv --python python3 --system-site-packages venv
$ . venv/bin/activate
$ wget https://raw.githubusercontent.com/DataBiosphere/toil/releases/4.1.0/src/toil/test/docs/scripts/tutorial_helloworld.py
$ python3 tutorial_helloworld.py \
      aws:us-west-2:demouser-toil-test-jobstore \
      --batchSystem kubernetes \
      --realTimeLogging \
      --logInfo

$ export TOIL_KUBERNETES_OWNER=demo-user  # This defaults to your local username if not set
$ export TOIL_AWS_SECRET_NAME=aws-credentials
$ export TOIL_KUBERNETES_HOST_PATH=/data/scratch
$ virtualenv --python python3 --system-site-packages venv
$ . venv/bin/activate
$ pip install toil[kubernetes,cwl]==5.8.0
$ toil-cwl-runner  \
     --jobStore  aws:us-west-2:demouser-toil-test-jobstore \
     --batchSystem kubernetes \
     --realTimeLogging \
     --logInfo \
     --disableCaching \
     path/to/cwl/workflow \
     path/to/cwl/input/object

9.

Next, you need to create an AWS access key. First go to the IAM dashboard, again; for "us-west1", the example link would be here:

https://console.aws.amazon.com/iam/home?region=us-west-1#/home

10.

The directions (transcribed from: https://docs.aws.amazon.com/general/latest/gr/managing-aws-access-keys.html ) are now:

11.

Now you should have a newly generated "AWS Access Key ID" and "AWS Secret Access Key". We can now install the AWS CLI and make sure that it has the proper credentials:

$ pip install awscli --upgrade --user

12.

Now configure your AWS credentials with:

$ aws configure

13.

Add your "AWS Access Key ID" and "AWS Secret Access Key" from earlier and your region and output format:

" AWS Access Key ID [****************Q65Q]: "
" AWS Secret Access Key [****************G0ys]: "
" Default region name [us-west-1]: "
" Default output format [json]: "

14.

If not done already, install toil (example uses version 5.3.0, but we recommend the latest release):

$ virtualenv venv
$ source venv/bin/activate
$ pip install toil[all]==5.3.0

15.

Now that toil is installed and you are running a virtualenv, an example of launching a toil leader node would be the following (again, note that we set TOIL_APPLIANCE_SELF to toil version 5.3.0 in this example, but please set the version to the installed version that you are using if you're using a different version):

$ TOIL_APPLIANCE_SELF=quay.io/ucsc_cgl/toil:5.3.0 \
      toil launch-cluster clustername \
      --leaderNodeType t2.medium \
      --zone us-west-1a \
      --keyPairName id_rsa

$ python3 sort.py aws:us-west-2:my-aws-sort-jobstore

Choosing Toil Appliance Image

When using the Toil provisioner, the appliance image will be automatically chosen based on the pip-installed version of Toil on your system. That choice can be overridden by setting the environment variables TOIL_DOCKER_REGISTRY and TOIL_DOCKER_NAME or TOIL_APPLIANCE_SELF. See Environment Variables for more information on these variables. If you are developing with autoscaling and want to test and build your own appliance have a look at Developing with Docker.

(venv) $ toil launch-cluster my-cluster \
             --leaderNodeType t2.medium \
             --zone us-west-2a \
             --keyPairName <your-AWS-key-pair-name>

(venv) $ toil launch-cluster --help

(venv) $ toil launch-cluster my-cluster \
             --leaderNodeType t2.small -z us-west-2a \
             --keyPairName your-AWS-key-pair-name \
             --nodeTypes m3.large,t2.micro -w 1,4

(venv) $ toil rsync-cluster -z us-west-2a my-cluster toil-workflow.py :/

$ python3 my-toil-script.py --help

$ python /root/sort.py aws:us-west-2:<my-jobstore-name> \
      --provisioner aws \
      --nodeTypes c3.4xlarge:2.00 \
      --maxNodes 2 \
      --batchSystem mesos \
      --defaultPreemptible

Specify Preemptibility Carefully

Ensure that your choices for --nodeTypes and --maxNodes <> make sense for your workflow and won't cause it to hang. You should make sure the provisioner is able to create nodes large enough to run the largest job in the workflow, and that non-preemptible node types are allowed if there are non-preemptible jobs in the workflow.

TOIL_S3_HOST=127.0.0.1
TOIL_S3_PORT=9010
TOIL_S3_USE_SSL=False

(venv) $ toil ssh-cluster -z us-west-2a --grafana_port 8000 <cluster-name>

$ python3 sort.py google:my-project-id:my-google-sort-jobstore

toil launch-cluster --help

python3 discoverfiles.py file:my-jobstore --stats

import os
import subprocess

from toil.common import Toil
from toil.job import Job


class discoverFiles(Job):
    """Views files at a specified path using ls."""

    def __init__(self, path, *args, **kwargs):
        self.path = path
        super().__init__(*args, **kwargs)

    def run(self, fileStore):
        if os.path.exists(self.path):
            subprocess.check_call(["ls", self.path])


def main():
    options = Job.Runner.getDefaultArgumentParser().parse_args()
    options.clean = "always"

    job1 = discoverFiles(path="/sys/", displayName='sysFiles')
    job2 = discoverFiles(path=os.path.expanduser("~"), displayName='userFiles')
    job3 = discoverFiles(path="/tmp/")

    job1.addChild(job2)
    job2.addChild(job3)

    with Toil(options) as toil:
        if not toil.options.restart:
            toil.start(job1)
        else:
            toil.restart()


if __name__ == '__main__':
    main()

toil stats file:my-jobstore

Batch System: singleMachine
Default Cores: 1  Default Memory: 2097152K
Max Cores: 9.22337e+18
Total Clock: 0.56  Total Runtime: 1.01
Worker
    Count |                                    Time* |                                    Clock |                                     Wait |                                   Memory
        n |      min    med*     ave     max   total |      min     med     ave     max   total |      min     med     ave     max   total |      min     med     ave     max   total
        1 |     0.14    0.14    0.14    0.14    0.14 |     0.13    0.13    0.13    0.13    0.13 |     0.01    0.01    0.01    0.01    0.01 |      76K     76K     76K     76K     76K
Job
 Worker Jobs  |     min    med    ave    max
              |       3      3      3      3
    Count |                                    Time* |                                    Clock |                                     Wait |                                   Memory
        n |      min    med*     ave     max   total |      min     med     ave     max   total |      min     med     ave     max   total |      min     med     ave     max   total
        3 |     0.01    0.06    0.05    0.07    0.14 |     0.00    0.06    0.04    0.07    0.12 |     0.00    0.01    0.00    0.01    0.01 |      76K     76K     76K     76K    229K
 sysFiles
    Count |                                    Time* |                                    Clock |                                     Wait |                                   Memory
        n |      min    med*     ave     max   total |      min     med     ave     max   total |      min     med     ave     max   total |      min     med     ave     max   total
        1 |     0.01    0.01    0.01    0.01    0.01 |     0.00    0.00    0.00    0.00    0.00 |     0.01    0.01    0.01    0.01    0.01 |      76K     76K     76K     76K     76K
 userFiles
    Count |                                    Time* |                                    Clock |                                     Wait |                                   Memory
        n |      min    med*     ave     max   total |      min     med     ave     max   total |      min     med     ave     max   total |      min     med     ave     max   total
        1 |     0.06    0.06    0.06    0.06    0.06 |     0.06    0.06    0.06    0.06    0.06 |     0.01    0.01    0.01    0.01    0.01 |      76K     76K     76K     76K     76K
 discoverFiles
    Count |                                    Time* |                                    Clock |                                     Wait |                                   Memory
        n |      min    med*     ave     max   total |      min     med     ave     max   total |      min     med     ave     max   total |      min     med     ave     max   total
        1 |     0.07    0.07    0.07    0.07    0.07 |     0.07    0.07    0.07    0.07    0.07 |     0.00    0.00    0.00    0.00    0.00 |      76K     76K     76K     76K     76K

toil clean file:my-jobstore

python3 discoverfiles.py file:my-jobstore --stats

toil status file:my-jobstore

2018-01-11 19:31:29,739 - toil.lib.bioio - INFO - Root logger is at level 'INFO', 'toil' logger at level 'INFO'.
2018-01-11 19:31:29,740 - toil.utils.toilStatus - INFO - Parsed arguments
2018-01-11 19:31:29,740 - toil.utils.toilStatus - INFO - Checking if we have files for Toil
The root job of the job store is absent, the workflow completed successfully.

$ toil launch-cluster my-cluster \
      --leaderNodeType t2.small -z us-west-2a \
      --keyPairName your-AWS-key-pair-name \
      --nodeTypes m3.large,t2.micro -w 1,4

$ toil launch-cluster --help

$ toil ssh-cluster CLUSTER-NAME-HERE

$ script
$ screen

$ toil ssh-cluster CLUSTER-NAME-HERE remoteCommand

$ toil rsync-cluster CLUSTER-NAME-HERE \
   ~/localFile :/remoteDestination

$ toil destroy-cluster CLUSTER-NAME-HERE

toil kill file:my-jobstore

$ toil-cwl-runner example.cwl example-job.yml

docker: Error response from daemon: Mounts denied:
The paths /var/...tmp are not shared from OS X and are not known to Docker.

export TMPDIR=/tmp/docker_tmp

$ toil-cwl-runner -h

$ toil-cwl-runner \
    --singularity \
    --jobStore my_jobStore \
    --batchSystem lsf \
    --workDir `pwd` \
    --outdir `pwd` \
    --logFile cwltoil.log \
    --writeLogs `pwd` \
    --logLevel DEBUG \
    --retryCount 2 \
    --maxLogFileSize 20000000000 \
    --stats \
    standard_bam_processing.cwl \
    inputs.yaml

import os
import subprocess
import tempfile

from toil.common import Toil
from toil.job import Job


def initialize_jobs(job):
    job.fileStore.logToMaster('initialize_jobs')


def runQC(job, cwl_file, cwl_filename, yml_file, yml_filename, outputs_dir, output_num):
    job.fileStore.logToMaster("runQC")
    tempDir = job.fileStore.getLocalTempDir()

    cwl = job.fileStore.readGlobalFile(cwl_file, userPath=os.path.join(tempDir, cwl_filename))
    yml = job.fileStore.readGlobalFile(yml_file, userPath=os.path.join(tempDir, yml_filename))

    subprocess.check_call(["toil-cwl-runner", cwl, yml])

    output_filename = "output.txt"
    output_file = job.fileStore.writeGlobalFile(output_filename)
    job.fileStore.readGlobalFile(output_file, userPath=os.path.join(outputs_dir, "sample_" + output_num + "_" + output_filename))
    return output_file


if __name__ == "__main__":
    jobstore: str = tempfile.mkdtemp("tutorial_cwlexample")
    os.rmdir(jobstore)
    options = Job.Runner.getDefaultOptions(jobstore)
    options.logLevel = "INFO"
    options.clean = "always"
    with Toil(options) as toil:

        # specify the folder where the cwl and yml files live
        inputs_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "cwlExampleFiles")
        # specify where you wish the outputs to be written
        outputs_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "cwlExampleFiles")

        job0 = Job.wrapJobFn(initialize_jobs)

        cwl_filename = "hello.cwl"
        cwl_file = toil.importFile("file://" + os.path.abspath(os.path.join(inputs_dir, cwl_filename)))

        # add list of yml config inputs here or import and construct from file
        yml_files = ["hello1.yml", "hello2.yml", "hello3.yml"]
        i = 0
        for yml in yml_files:
            i = i + 1
            yml_file = toil.importFile("file://" + os.path.abspath(os.path.join(inputs_dir, yml)))
            yml_filename = yml
            job = Job.wrapJobFn(runQC, cwl_file, cwl_filename, yml_file, yml_filename, outputs_dir, output_num=str(i))
            job0.addChild(job)

        toil.start(job0)

cat cwltoil.log | grep jobVM1fIs

pcregrep -M "\[job .*\.cwl.*$\n(.*        .*$\n)*" cwltoil.log
#         ^allows for multiline matching

find . | grep -P '^./out_tmpdir.*_MD\.bam$'

cat log/cwltoil.log | grep -oP "\[job .*.cwl\]" | sort | uniq

cat log/cwltoil.log | grep -i "issued job"

$ toil status /home/johnsoni/TEST_RUNS_3/TEST_run/tmp/jobstore-09ae0acc-c800-11e8-9d09-70106fb1697e
<hostname> 2018-10-04 15:01:44,184 MainThread INFO toil.lib.bioio: Root logger is at level 'INFO', 'toil' logger at level 'INFO'.
<hostname> 2018-10-04 15:01:44,185 MainThread INFO toil.utils.toilStatus: Parsed arguments
<hostname> 2018-10-04 15:01:47,081 MainThread INFO toil.utils.toilStatus: Traversing the job graph gathering jobs. This may take a couple of minutes.

Of the 286 jobs considered, there are 179 jobs with children, 107 jobs ready to run, 0 zombie jobs, 0 jobs with services, 0 services, and 0 jobs with log files currently in file:/home/user/jobstore-09ae0acc-c800-11e8-9d09-70106fb1697e.

$ toil stats /path/to/jobstore

<hostname> 2018-10-15 12:06:19,003 MainThread INFO toil.lib.bioio: Root logger is at level 'INFO', 'toil' logger at level 'INFO'.
<hostname> 2018-10-15 12:06:19,004 MainThread INFO toil.utils.toilStats: Parsed arguments
<hostname> 2018-10-15 12:06:19,004 MainThread INFO toil.utils.toilStats: Checking if we have files for toil
<hostname> 2018-10-15 12:06:19,004 MainThread INFO toil.utils.toilStats: Checked arguments
Batch System: lsf
Default Cores: 1  Default Memory: 10485760K
Max Cores: 9.22337e+18
Total Clock: 106608.01  Total Runtime: 86634.11
Worker
    Count |                                       Time* |                                        Clock |                                              Wait |                                    Memory
        n |      min    med*     ave      max     total |      min     med      ave      max     total |        min      med       ave      max      total |      min     med     ave     max    total
     1659 |     0.00    0.80  264.87 12595.59 439424.40 |     0.00    0.46   449.05 42240.74 744968.80 |  -35336.69     0.16   -184.17  4230.65 -305544.39 |      48K    223K   1020K  40235K 1692300K
Job
 Worker Jobs  |     min    med    ave    max
              |    1077   1077   1077   1077
    Count |                                       Time* |                                        Clock |                                              Wait |                                    Memory
        n |      min    med*     ave      max     total |      min     med      ave      max     total |        min      med       ave      max      total |      min     med     ave     max    total
     1077 |     0.04    1.18  407.06 12593.43 438404.73 |     0.01    0.28   691.17 42240.35 744394.14 |  -35336.83     0.27   -284.11  4230.49 -305989.41 |     135K    268K   1633K  40235K 1759734K
 ResolveIndirect
    Count |                                       Time* |                                        Clock |                                              Wait |                                    Memory
        n |      min    med*     ave      max     total |      min     med      ave      max     total |        min      med       ave      max      total |      min     med     ave     max    total
      205 |     0.04    0.07    0.16     2.29     31.95 |     0.01    0.02     0.02     0.14      3.60 |       0.02     0.05      0.14     2.28      28.35 |     190K    266K    256K    314K   52487K
 CWLGather
    Count |                                       Time* |                                        Clock |                                              Wait |                                    Memory
        n |      min    med*     ave      max     total |      min     med      ave      max     total |        min      med       ave      max      total |      min     med     ave     max    total
       40 |     0.05    0.17    0.29     1.90     11.62 |     0.01    0.02     0.02     0.05      0.80 |       0.03     0.14      0.27     1.88      10.82 |     188K    265K    250K    316K   10039K
 CWLWorkflow
    Count |                                       Time* |                                        Clock |                                              Wait |                                    Memory
        n |      min    med*     ave      max     total |      min     med      ave      max     total |        min      med       ave      max      total |      min     med     ave     max    total
      205 |     0.09    0.40    0.98    13.70    200.82 |     0.04    0.15     0.16     1.08     31.78 |       0.04     0.26      0.82    12.62     169.04 |     190K    270K    257K    316K   52826K
 file:///home/johnsoni/pipeline_0.0.39/ACCESS-Pipeline/cwl_tools/expression_tools/group_waltz_files.cwl
    Count |                                       Time* |                                        Clock |                                              Wait |                                    Memory
        n |      min    med*     ave      max     total |      min     med      ave      max     total |        min      med       ave      max      total |      min     med     ave     max    total
       99 |     0.29    0.49    0.59     2.50     58.11 |     0.14    0.26     0.29     1.04     28.95 |       0.14     0.22      0.29     1.48      29.16 |     135K    135K    135K    136K   13459K
 file:///home/johnsoni/pipeline_0.0.39/ACCESS-Pipeline/cwl_tools/expression_tools/make_sample_output_dirs.cwl
    Count |                                       Time* |                                        Clock |                                              Wait |                                    Memory
        n |      min    med*     ave      max     total |      min     med      ave      max     total |        min      med       ave      max      total |      min     med     ave     max    total
       11 |     0.34    0.52    0.74     2.63      8.18 |     0.20    0.30     0.41     1.17      4.54 |       0.14     0.20      0.33     1.45       3.65 |     136K    136K    136K    136K    1496K
 file:///home/johnsoni/pipeline_0.0.39/ACCESS-Pipeline/cwl_tools/expression_tools/consolidate_files.cwl
    Count |                                       Time* |                                        Clock |                                              Wait |                                    Memory
        n |      min    med*     ave      max     total |      min     med      ave      max     total |        min      med       ave      max      total |      min     med     ave     max    total
        8 |     0.31    0.59    0.71     1.80      5.69 |     0.18    0.35     0.37     0.63      2.94 |       0.13     0.27      0.34     1.17       2.75 |     136K    136K    136K    136K    1091K
 file:///home/johnsoni/pipeline_0.0.39/ACCESS-Pipeline/cwl_tools/bwa-mem/bwa-mem.cwl
    Count |                                       Time* |                                        Clock |                                              Wait |                                    Memory
        n |      min    med*     ave      max     total |      min     med      ave      max     total |        min      med       ave      max      total |      min     med     ave     max    total
       22 |   895.76 3098.13 3587.34 12593.43  78921.51 |  2127.02 7910.31  8123.06 16959.13 178707.34 |  -11049.84 -3827.96  -4535.72    19.49  -99785.83 |    5659K   5950K   5854K   6128K  128807K

file:<path to cwl tool>.cwl_<job ID>.log

file:---home-johnsoni-pipeline_1.1.14-ACCESS--Pipeline-cwl_tools-marianas-ProcessLoopUMIFastq.cwl_I-O-jobfGsQQw000.log

{
"encode_mapping_workflow.fastqs": "Array[File]",
"encode_mapping_workflow.trimming_parameter": "String",
"encode_mapping_workflow.reference": "File"
}

{
"encode_mapping_workflow.fastqs": ["/path/to/unzipped/ENCODE_data/ENCFF000VOL_chr21.fq.gz"],
"encode_mapping_workflow.trimming_parameter": "native",
"encode_mapping_workflow.reference": "/path/to/unzipped/ENCODE_data/reference/GRCh38_chr21_bwa.tar.gz"
}

import os
import subprocess
import tempfile

from toil.common import Toil
from toil.job import Job



def initialize_jobs(job):
    job.fileStore.logToMaster("initialize_jobs")


def runQC(job, wdl_file, wdl_filename, json_file, json_filename, outputs_dir, jar_loc,output_num):
    job.fileStore.logToMaster("runQC")
    tempDir = job.fileStore.getLocalTempDir()

    wdl = job.fileStore.readGlobalFile(wdl_file, userPath=os.path.join(tempDir, wdl_filename))
    json = job.fileStore.readGlobalFile(json_file, userPath=os.path.join(tempDir, json_filename))

    subprocess.check_call(["java","-jar",jar_loc,"run",wdl,"--inputs",json])

    output_filename = "output.txt"
    output_file = job.fileStore.writeGlobalFile(outputs_dir + output_filename)
    job.fileStore.readGlobalFile(output_file, userPath=os.path.join(outputs_dir, "sample_" + output_num + "_" + output_filename))
    return output_file


if __name__ == "__main__":
    jobstore: str = tempfile.mkdtemp("tutorial_wdlexample")
    os.rmdir(jobstore)
    options = Job.Runner.getDefaultOptions(jobstore)
    options.logLevel = "INFO"
    options.clean = "always"

    with Toil(options) as toil:

        # specify the folder where the wdl and json files live
        inputs_dir = "wdlExampleFiles/"
        # specify where you wish the outputs to be written
        outputs_dir = "wdlExampleFiles/"
        # specify the location of your cromwell jar
        jar_loc = os.path.abspath("wdlExampleFiles/cromwell-35.jar")

        job0 = Job.wrapJobFn(initialize_jobs)

        wdl_filename = "hello.wdl"
        wdl_file = toil.importFile("file://" + os.path.abspath(os.path.join(inputs_dir, wdl_filename)))


        # add list of yml config inputs here or import and construct from file
        json_files = ["hello1.json", "hello2.json", "hello3.json"]
        i = 0
        for json in json_files:
            i = i + 1
            json_file = toil.importFile("file://" + os.path.join(inputs_dir, json))
            json_filename = json
            job = Job.wrapJobFn(runQC, wdl_file, wdl_filename, json_file, json_filename, outputs_dir, jar_loc, output_num=str(i))
            job0.addChild(job)

        toil.start(job0)

•

Scatter

•

Many Built-In Functions

•

Docker Calls

•

Handles Priority, and Output File Wrangling

•

Currently Handles Primitives and Arrays

•

Integrate Cloud Autoscaling Capacity More Robustly

•

WDL Files That "Import" Other WDL Files (Including URI Handling for ' http://' and 'https://')

$ toil server

http://localhost:8080/ga4gh/wes/v1

$ toil server --port 3000

$ toil server --help

# docker-compose.yml
version: "3.8"

services:
  rabbitmq:
    image: rabbitmq:3.9.5
    hostname: rabbitmq
  celery:
    image: ${TOIL_APPLIANCE_SELF}
    volumes:
      - /var/run/docker.sock:/var/run/docker.sock
      - /var/lib/docker:/var/lib/docker
      - /var/lib/toil:/var/lib/toil
      - /var/lib/cwl:/var/lib/cwl
      - /tmp/toil-workflows:/tmp/toil-workflows
    command: celery --broker=amqp://guest:guest@rabbitmq:5672// -A toil.server.celery_app worker --loglevel=INFO
    depends_on:
      - rabbitmq
  wes-server:
    image: ${TOIL_APPLIANCE_SELF}
    volumes:
      - /tmp/toil-workflows:/tmp/toil-workflows
    environment:
      - TOIL_WES_BROKER_URL=amqp://guest:guest@rabbitmq:5672//
    command: toil server --host 0.0.0.0 --port 8000 --work_dir /tmp/toil-workflows
    expose:
      - 8000
    labels:
      - "traefik.enable=true"
      - "traefik.http.routers.wes.rule=Host(`localhost`)"
      - "traefik.http.routers.wes.entrypoints=web"
      - "traefik.http.routers.wes.middlewares=auth"
      - "traefik.http.middlewares.auth.basicauth.users=test:$$2y$$12$$ci.4U63YX83CwkyUrjqxAucnmi2xXOIlEF6T/KdP9824f1Rf1iyNG"
      - "traefik.http.routers.wespublic.rule=Host(`localhost`) && Path(`/ga4gh/wes/v1/service-info`)"
    depends_on:
      - rabbitmq
      - celery
  traefik:
    image: traefik:v2.2
    command:
      - "--providers.docker"
      - "--providers.docker.exposedbydefault=false"
      - "--entrypoints.web.address=:8080"
    ports:
      - "8080:8080"
    volumes:
      - /var/run/docker.sock:/var/run/docker.sock

# example.cwl
cwlVersion: v1.0
class: CommandLineTool
baseCommand: echo
stdout: output.txt
inputs:
 message:
   type: string
   inputBinding:
     position: 1
outputs:
 output:
   type: stdout

$ curl --location --request POST 'http://localhost:8080/ga4gh/wes/v1/runs' \
    --user test:test \
    --form 'workflow_url="example.cwl"' \
    --form 'workflow_type="cwl"' \
    --form 'workflow_type_version="v1.0"' \
    --form 'workflow_params="{\"message\": \"Hello world!\"}"' \
    --form 'workflow_attachment=@"./toil_test_files/example.cwl"'
{
  "run_id": "4deb8beb24894e9eb7c74b0f010305d1"
}

$ curl --location --request POST 'http://localhost:8080/ga4gh/wes/v1/runs' \
    --user test:test \
    --form 'workflow_url="example.cwl"' \
    --form 'workflow_type="cwl"' \
    --form 'workflow_type_version="v1.0"' \
    --form 'workflow_params="{\"message\": \"Hello world!\"}"' \
    --form 'workflow_attachment=@"./toil_test_files/example.cwl"' \
    --form 'workflow_attachment=@"./toil_test_files/2.fasta";filename=inputs/test.fasta' \
    --form 'workflow_attachment=@"./toil_test_files/2.fastq";filename=inputs/test.fastq'

execution/
├── example.cwl
├── inputs
│     ├── test.fasta
|     └── test.fastq
└── wes_inputs.json

{"--logLevel": "INFO", "--workDir": "/tmp/"}

$ curl --user test:test http://localhost:8080/ga4gh/wes/v1/runs/4deb8beb24894e9eb7c74b0f010305d1/status
{
  "run_id": "4deb8beb24894e9eb7c74b0f010305d1",
  "state": "RUNNING"
}

$ curl --user test:test http://localhost:8080/ga4gh/wes/v1/runs/4deb8beb24894e9eb7c74b0f010305d1
{
  "run_id": "4deb8beb24894e9eb7c74b0f010305d1",
  "request": {
    "workflow_attachment": [
      "example.cwl"
    ],
    "workflow_url": "example.cwl",
    "workflow_type": "cwl",
    "workflow_type_version": "v1.0",
    "workflow_params": {
      "message": "Hello world!"
    }
  },
  "state": "RUNNING",
  "run_log": {
    "cmd": [
      "toil-cwl-runner --outdir=/home/toil/workflows/4deb8beb24894e9eb7c74b0f010305d1/outputs --jobStore=file:/home/toil/workflows/4deb8beb24894e9eb7c74b0f010305d1/toil_job_store /home/toil/workflows/4deb8beb24894e9eb7c74b0f010305d1/execution/example.cwl /home/workflows/4deb8beb24894e9eb7c74b0f010305d1/execution/wes_inputs.json"
    ],
    "start_time": "2021-08-30T17:35:50Z",
    "end_time": null,
    "stdout": null,
    "stderr": null,
    "exit_code": null
  },
  "task_logs": [],
  "outputs": {}
}

$ curl --location --request POST 'http://localhost:8080/ga4gh/wes/v1/runs/4deb8beb24894e9eb7c74b0f010305d1/cancel' \
      --user test:test
{
  "run_id": "4deb8beb24894e9eb7c74b0f010305d1"
}

import os
import tempfile

from toil.common import Toil
from toil.job import Job


def helloWorld(message, memory="2G", cores=2, disk="3G"):
    return f"Hello, world!, here's a message: {message}"


if __name__ == "__main__":
    jobstore: str = tempfile.mkdtemp("tutorial_quickstart")
    os.rmdir(jobstore)
    options = Job.Runner.getDefaultOptions(jobstore)
    options.logLevel = "OFF"
    options.clean = "always"

    hello_job = Job.wrapFn(helloWorld, "Woot")

    with Toil(options) as toil:
        print(toil.start(hello_job))  # prints "Hello, world!, ..."

from toil.job import Job

class HelloWorld(Job):
    def __init__(self, message):
        Job.__init__(self,  memory="2G", cores=2, disk="3G")
        self.message = message

    def run(self, fileStore):
        return f"Hello, world! Here's a message: {self.message}"

...
    def run(self, fileStore):
        self.log(f"Hello, world! Here's a message: {self.message}")

import os
import tempfile

from toil.common import Toil
from toil.job import Job


class HelloWorld(Job):
    def __init__(self, message):
        Job.__init__(self, memory="2G", cores=2, disk="3G")
        self.message = message

    def run(self, fileStore):
        return f"Hello, world!, here's a message: {self.message}"


if __name__ == "__main__":
    jobstore: str = tempfile.mkdtemp("tutorial_invokeworkflow")
    os.rmdir(jobstore)
    options = Job.Runner.getDefaultOptions(jobstore)
    options.logLevel = "OFF"
    options.clean = "always"

    hello_job = HelloWorld("Woot")

    with Toil(options) as toil:
        print(toil.start(hello_job))

import os
import tempfile

from toil.common import Toil
from toil.job import Job

class HelloWorld(Job):
    def __init__(self, message):
        Job.__init__(self,  memory="2G", cores=2, disk="3G")
        self.message = message

    def run(self, fileStore):
        return f"Hello, world!, I have a message: {self.message}"


if __name__ == "__main__":
    jobstore: str = tempfile.mkdtemp("tutorial_invokeworkflow2")
    os.rmdir(jobstore)
    options = Job.Runner.getDefaultOptions(jobstore)
    options.logLevel = "INFO"
    options.clean = "always"

    with Toil(options) as toil:
        if not toil.options.restart:
            job = HelloWorld("Woot!")
            output = toil.start(job)
        else:
            output = toil.restart()
    print(output)

from toil.common import Toil
from toil.job import Job


class HelloWorld(Job):
    def __init__(self, message):
        Job.__init__(self,  memory="2G", cores=2, disk="3G")
        self.message = message

    def run(self, fileStore):
        return "Hello, world!, here's a message: %s" % self.message


if __name__ == "__main__":
    parser = Job.Runner.getDefaultArgumentParser()
    options = parser.parse_args()
    options.logLevel = "OFF"
    options.clean = "always"

    hello_job = HelloWorld("Woot")

    with Toil(options) as toil:
        print(toil.start(hello_job))

import os
import tempfile

from toil.common import Toil
from toil.job import Job


def helloWorld(message, memory="2G", cores=2, disk="3G"):
    return f"Hello, world!, here's a message: {message}"


if __name__ == "__main__":
    jobstore: str = tempfile.mkdtemp("tutorial_quickstart")
    os.rmdir(jobstore)
    options = Job.Runner.getDefaultOptions(jobstore)
    options.logLevel = "OFF"
    options.clean = "always"

    hello_job = Job.wrapFn(helloWorld, "Woot")

    with Toil(options) as toil:
        print(toil.start(hello_job))  # prints "Hello, world!, ..."

Job.wrapFn(helloWorld, "Woot")

import os
import tempfile

from toil.common import Toil
from toil.job import Job


def helloWorld(job, message):
    job.log(f"Hello world, I have a message: {message}")


if __name__ == "__main__":
    jobstore: str = tempfile.mkdtemp("tutorial_jobfunctions")
    os.rmdir(jobstore)
    options = Job.Runner.getDefaultOptions(jobstore)
    options.logLevel = "INFO"
    options.clean = "always"

    hello_job = Job.wrapJobFn(helloWorld, "Woot!")

    with Toil(options) as toil:
        toil.start(hello_job)

hello_job = Job.wrapJobFn(helloWorld, "Woot")

from toil.common import Toil
from toil.job import Job


def helloWorld(job, message, memory="2G", cores=2, disk="3G"):
    job.log(f"Hello world, I have a message: {message}")


if __name__ == "__main__":
    parser = Job.Runner.getDefaultArgumentParser()
    options = parser.parse_args()
    options.logLevel = "INFO"
    options.clean = "always"

    j1 = Job.wrapJobFn(helloWorld, "first")
    j2 = Job.wrapJobFn(helloWorld, "second or third")
    j3 = Job.wrapJobFn(helloWorld, "second or third")
    j4 = Job.wrapJobFn(helloWorld, "last")

    j1.addChild(j2)
    j1.addChild(j3)
    j1.addFollowOn(j4)

    with Toil(options) as toil:
        toil.start(j1)

from toil.common import Toil
from toil.job import Job


def helloWorld(job, message, memory="2G", cores=2, disk="3G"):
    job.log(f"Hello world, I have a message: {message}")


if __name__ == "__main__":
    parser = Job.Runner.getDefaultArgumentParser()
    options = parser.parse_args()
    options.logLevel = "INFO"
    options.clean = "always"

    j1 = Job.wrapJobFn(helloWorld, "first")
    j2 = j1.addChildJobFn(helloWorld, "second or third")
    j3 = j1.addChildJobFn(helloWorld, "second or third")
    j4 = j1.addFollowOnJobFn(helloWorld, "last")

    with Toil(options) as toil:
        toil.start(j1)

from toil.common import Toil
from toil.job import Job


def helloWorld(job, message, memory="2G", cores=2, disk="3G"):
    job.log(f"Hello world, I have a message: {message}")


if __name__ == "__main__":
    parser = Job.Runner.getDefaultArgumentParser()
    options = parser.parse_args()
    options.logLevel = "INFO"
    options.clean = "always"

    j1 = Job.wrapJobFn(helloWorld, "first")
    j2 = j1.addChildJobFn(helloWorld, "second or third")
    j3 = j1.addChildJobFn(helloWorld, "second or third")
    j4 = j2.addChildJobFn(helloWorld, "last")
    j3.addChild(j4)

    with Toil(options) as toil:
        toil.start(j1)

import os
import tempfile

from toil.common import Toil
from toil.job import Job


def binaryStringFn(job, depth, message=""):
    if depth > 0:
        job.addChildJobFn(binaryStringFn, depth-1, message + "0")
        job.addChildJobFn(binaryStringFn, depth-1, message + "1")
    else:
        job.log(f"Binary string: {message}")


if __name__ == "__main__":
    jobstore: str = tempfile.mkdtemp("tutorial_dynamic")
    os.rmdir(jobstore)
    options = Job.Runner.getDefaultOptions(jobstore)
    options.logLevel = "INFO"
    options.clean = "always"

    with Toil(options) as toil:
        toil.start(Job.wrapJobFn(binaryStringFn, depth=5))

import os
import tempfile

from toil.common import Toil
from toil.job import Job


def fn(job, i):
    job.log("i is: %s" % i, level=100)
    return i + 1


if __name__ == "__main__":
    jobstore: str = tempfile.mkdtemp("tutorial_promises")
    os.rmdir(jobstore)
    options = Job.Runner.getDefaultOptions(jobstore)
    options.logLevel = "INFO"
    options.clean = "always"

    j1 = Job.wrapJobFn(fn, 1)
    j2 = j1.addChildJobFn(fn, j1.rv())
    j3 = j1.addFollowOnJobFn(fn, j2.rv())

    with Toil(options) as toil:
        toil.start(j1)

j2 = j1.addChildFn(fn, j1.rv())

def parent(job):
    indexable = Job.wrapJobFn(fn)
    job.addChild(indexable)
    job.addFollowOnFn(raiseWrap, indexable.rv(2))

def raiseWrap(arg):
    raise RuntimeError(arg) # raises "2"

def fn(job):
    return (0, 1, 2, 3)

import os
import tempfile

from toil.common import Toil
from toil.job import Job


def binaryStrings(job, depth, message=""):
    if depth > 0:
        s = [job.addChildJobFn(binaryStrings, depth - 1, message + "0").rv(),
             job.addChildJobFn(binaryStrings, depth - 1, message + "1").rv()]
        return job.addFollowOnFn(merge, s).rv()
    return [message]


def merge(strings):
    return strings[0] + strings[1]


if __name__ == "__main__":
    jobstore: str = tempfile.mkdtemp("tutorial_promises2")
    os.rmdir(jobstore)
    options = Job.Runner.getDefaultOptions(jobstore)
    options.loglevel = "OFF"
    options.clean = "always"

    with Toil(options) as toil:
        print(toil.start(Job.wrapJobFn(binaryStrings, depth=5)))

import os
import tempfile

from toil.common import Toil
from toil.job import Job, PromisedRequirement


def parentJob(job):
    downloadJob = Job.wrapJobFn(stageFn, "file://" + os.path.realpath(__file__), cores=0.1, memory='32M', disk='1M')
    job.addChild(downloadJob)

    analysis = Job.wrapJobFn(analysisJob,
                             fileStoreID=downloadJob.rv(0),
                             disk=PromisedRequirement(downloadJob.rv(1)))
    job.addFollowOn(analysis)


def stageFn(job, url, cores=1):
    importedFile = job.fileStore.import_file(url)
    return importedFile, importedFile.size


def analysisJob(job, fileStoreID, cores=2):
    # now do some analysis on the file
    pass


if __name__ == "__main__":
    jobstore: str = tempfile.mkdtemp("tutorial_requirements")
    os.rmdir(jobstore)
    options = Job.Runner.getDefaultOptions(jobstore)
    options.logLevel = "INFO"
    options.clean = "always"

    with Toil(options) as toil:
        toil.start(Job.wrapJobFn(parentJob))

def parentJob(job):
    aggregator = []
    for fileNum in range(0, 10):
        downloadJob = Job.wrapJobFn(stageFn, "file://" + os.path.realpath(__file__), cores=0.1, memory='32M', disk='1M')
        job.addChild(downloadJob)
        aggregator.append(downloadJob)

    analysis = Job.wrapJobFn(analysisJob,
                             fileStoreID=downloadJob.rv(0),
                             disk=PromisedRequirement(lambda xs: sum(xs), [j.rv(1) for j in aggregator]))
    job.addFollowOn(analysis)

Limitations

Just like regular promises, the return value must be determined prior to scheduling any job that depends on the return value. In our example above, notice how the dependent jobs were follow ons to the parent while promising jobs are children of the parent. This ordering ensures that all promises are properly fulfilled.

import os
import tempfile

from toil.common import Toil
from toil.job import Job


class LocalFileStoreJob(Job):
    def run(self, fileStore):
        # self.tempDir will always contain the name of a directory within the allocated disk space reserved for the job
        scratchDir = self.tempDir

        # Similarly create a temporary file.
        scratchFile = fileStore.getLocalTempFile()


if __name__ == "__main__":
    jobstore: str = tempfile.mkdtemp("tutorial_managing")
    os.rmdir(jobstore)
    options = Job.Runner.getDefaultOptions(jobstore)
    options.logLevel = "INFO"
    options.clean = "always"

    # Create an instance of FooJob which will have at least 2 gigabytes of storage space.
    j = LocalFileStoreJob(disk="2G")

    # Run the workflow
    with Toil(options) as toil:
        toil.start(j)

def localFileStoreJobFn(job):
    scratchDir = job.tempDir
    scratchFile = job.fileStore.getLocalTempFile()

import os
import tempfile

from toil.common import Toil
from toil.job import Job


def globalFileStoreJobFn(job):
    job.log("The following example exercises all the methods provided "
            "by the toil.fileStores.abstractFileStore.AbstractFileStore class")

    # Create a local temporary file.
    scratchFile = job.fileStore.getLocalTempFile()

    # Write something in the scratch file.
    with open(scratchFile, 'w') as fH:
        fH.write("What a tangled web we weave")

    # Write a copy of the file into the file-store; fileID is the key that can be used to retrieve the file.
    # This write is asynchronous by default
    fileID = job.fileStore.writeGlobalFile(scratchFile)

    # Write another file using a stream; fileID2 is the
    # key for this second file.
    with job.fileStore.writeGlobalFileStream(cleanup=True) as (fH, fileID2):
        fH.write(b"Out brief candle")

    # Now read the first file; scratchFile2 is a local copy of the file that is read-only by default.
    scratchFile2 = job.fileStore.readGlobalFile(fileID)

    # Read the second file to a desired location: scratchFile3.
    scratchFile3 = os.path.join(job.tempDir, "foo.txt")
    job.fileStore.readGlobalFile(fileID2, userPath=scratchFile3)

    # Read the second file again using a stream.
    with job.fileStore.readGlobalFileStream(fileID2) as fH:
        print(fH.read())  # This prints "Out brief candle"

    # Delete the first file from the global file-store.
    job.fileStore.deleteGlobalFile(fileID)

    # It is unnecessary to delete the file keyed by fileID2 because we used the cleanup flag,
    # which removes the file after this job and all its successors have run (if the file still exists)


if __name__ == "__main__":
    jobstore: str = tempfile.mkdtemp("tutorial_managing2")
    os.rmdir(jobstore)
    options = Job.Runner.getDefaultOptions(jobstore)
    options.logLevel = "INFO"
    options.clean = "always"

    with Toil(options) as toil:
        toil.start(Job.wrapJobFn(globalFileStoreJobFn))

import os
import tempfile

from toil.common import Toil
from toil.job import Job


class HelloWorld(Job):
    def __init__(self, id):
        Job.__init__(self, memory="2G", cores=2, disk="3G")
        self.inputFileID = id

    def run(self, fileStore):
        with fileStore.readGlobalFileStream(self.inputFileID, encoding='utf-8') as fi:
            with fileStore.writeGlobalFileStream(encoding='utf-8') as (fo, outputFileID):
                fo.write(fi.read() + 'World!')
        return outputFileID


if __name__ == "__main__":
    jobstore: str = tempfile.mkdtemp("tutorial_staging")
    os.rmdir(jobstore)
    options = Job.Runner.getDefaultOptions(jobstore)
    options.logLevel = "INFO"
    options.clean = "always"

    with Toil(options) as toil:
        if not toil.options.restart:
            ioFileDirectory = os.path.join(os.path.dirname(os.path.abspath(__file__)), "stagingExampleFiles")
            inputFileID = toil.importFile("file://" + os.path.abspath(os.path.join(ioFileDirectory, "in.txt")))
            outputFileID = toil.start(HelloWorld(inputFileID))
        else:
            outputFileID = toil.restart()

        toil.exportFile(outputFileID, "file://" + os.path.abspath(os.path.join(ioFileDirectory, "out.txt")))

dockerCall(job=job,
            tool='quay.io/ucsc_cgl/bwa',
            workDir=job.tempDir,
            parameters=['index', '/data/reference.fa'])

import os
import tempfile

from toil.common import Toil
from toil.job import Job
from toil.lib.docker import apiDockerCall

align = Job.wrapJobFn(apiDockerCall,
                      image='ubuntu',
                      working_dir=os.getcwd(),
                      parameters=['ls', '-lha'])

if __name__ == "__main__":
    jobstore: str = tempfile.mkdtemp("tutorial_docker")
    os.rmdir(jobstore)
    options = Job.Runner.getDefaultOptions(jobstore)
    options.logLevel = "INFO"
    options.clean = "always"

    with Toil(options) as toil:
        toil.start(align)

import os
import tempfile

from toil.common import Toil
from toil.job import Job


class DemoService(Job.Service):
    def start(self, fileStore):
        # Start up a database/service here
        # Return a value that enables another process to connect to the database
        return "loginCredentials"

    def check(self):
        # A function that if it returns False causes the service to quit
        # If it raises an exception the service is killed and an error is reported
        return True

    def stop(self, fileStore):
        # Cleanup the database here
        pass


j = Job()
s = DemoService()
loginCredentialsPromise = j.addService(s)


def dbFn(loginCredentials):
    # Use the login credentials returned from the service's start method to connect to the service
    pass


j.addChildFn(dbFn, loginCredentialsPromise)


if __name__ == "__main__":
    jobstore: str = tempfile.mkdtemp("tutorial_services")
    os.rmdir(jobstore)
    options = Job.Runner.getDefaultOptions(jobstore)
    options.logLevel = "INFO"
    options.clean = "always"

    with Toil(options) as toil:
        toil.start(j)

j = Job(checkpoint=True)

import os
import tempfile

from toil.common import Toil
from toil.job import Job

if __name__ == "__main__":
    # A is a job with children and follow-ons, for example:
    A = Job()
    A.addChild(Job())
    A.addFollowOn(Job())

    # B is a job which needs to run after A and its successors
    B = Job()

    # The way to do this without encapsulation is to make a parent of A, Ap, and make B a follow-on of Ap.
    Ap = Job()
    Ap.addChild(A)
    Ap.addFollowOn(B)

    jobstore: str = tempfile.mkdtemp("tutorial_encapsulations")
    os.rmdir(jobstore)
    options = Job.Runner.getDefaultOptions(jobstore)
    options.logLevel = "INFO"
    options.clean = "always"

    with Toil(options) as toil:
        print(toil.start(Ap))

import os
import tempfile

from toil.common import Toil
from toil.job import Job

if __name__ == "__main__":
    # A
    A = Job()
    A.addChild(Job())
    A.addFollowOn(Job())

    # Encapsulate A
    A = A.encapsulate()

    # B is a job which needs to run after A and its successors
    B = Job()

    # With encapsulation A and its successor subgraph appear to be a single job, hence:
    A.addChild(B)

    jobstore: str = tempfile.mkdtemp("tutorial_encapsulations2")
    os.rmdir(jobstore)
    options = Job.Runner.getDefaultOptions(jobstore)
    options.logLevel = "INFO"
    options.clean = "always"

    with Toil(options) as toil:
        print(toil.start(A))

__init__(options: Namespace) -> None

Initialize a Toil object from the given options. Note that this is very light-weight and that the bulk of the work is done when the context is entered.

start(rootJob: Job) -> Any

Invoke a Toil workflow with the given job as the root for an initial run. This method must be called in the body of a with Toil(...) as toil: statement. This method should not be called more than once for a workflow that has not finished.

restart() -> Any

Restarts a workflow that has been interrupted.

classmethod getJobStore(locator: str) -> AbstractJobStore

Create an instance of the concrete job store implementation that matches the given locator.

static createBatchSystem(config: Config) -> AbstractBatchSystem

Create an instance of the batch system specified in the given config.

import_file(src_uri: str, shared_file_name: str, symlink: bool = False) -> None

import_file(src_uri: str, shared_file_name: None = None, symlink: bool = False) -> FileID

Import the file at the given URL into the job store. See toil.jobStores.abstractJobStore.AbstractJobStore.importFile() for a full description

export_file(file_id: FileID, dst_uri: str) -> None

Export file to destination pointed at by the destination URL. See toil.jobStores.abstractJobStore.AbstractJobStore.exportFile() for a full description

static normalize_uri(uri: str, check_existence: bool = False) -> str

Given a URI, if it has no scheme, prepend "file:".

static getToilWorkDir(configWorkDir: Optional[str] = None) -> str

Return a path to a writable directory under which per-workflow directories exist. This directory is always required to exist on a machine, even if the Toil worker has not run yet. If your workers and leader have different temp directories, you may need to set TOIL_WORKDIR.

classmethod get_toil_coordination_dir(config_work_dir: Optional[str], config_coordination_dir: Optional[str]) -> str

Return a path to a writable directory, which will be in memory if convenient. Ought to be used for file locking and coordination.

classmethod getLocalWorkflowDir(workflowID: str, configWorkDir: Optional[str] = None) -> str

Return the directory where worker directories and the cache will be located for this workflow on this machine.

classmethod get_local_workflow_coordination_dir(workflow_id: str, config_work_dir: Optional[str], config_coordination_dir: Optional[ str]) -> str

Return the directory where coordination files should be located for this workflow on this machine. These include internal Toil databases and lock files for the machine. If an in-memory filesystem is available, it is used. Otherwise, the local workflow directory, which may be on a shared network filesystem, is used.

__init__(locator: str) -> None

Create an instance of the job store. The instance will not be fully functional until either initialize() or resume() is invoked. Note that the destroy() method may be invoked on the object with or without prior invocation of either of these two methods. Takes and stores the locator string for the job store, which will be accessible via self.locator.

initialize(config: Config) -> None

Initialize this job store. Create the physical storage for this job store, allocate a workflow ID and persist the given Toil configuration to the store.

write_config() -> None

Persists the value of the AbstractJobStore.config attribute to the job store, so that it can be retrieved later by other instances of this class.

resume() -> None

Connect this instance to the physical storage it represents and load the Toil configuration into the AbstractJobStore.config attribute.

property config: Config

Return the Toil configuration associated with this job store.

property locator: str

Get the locator that defines the job store, which can be used to connect to it.

setRootJob(rootJobStoreID: FileID) -> None

Set the root job of the workflow backed by this job store.

set_root_job(job_id: FileID) -> None

Set the root job of the workflow backed by this job store.

load_root_job() -> JobDescription

Loads the JobDescription for the root job in the current job store.

create_root_job(job_description: JobDescription) -> JobDescription

Create the given JobDescription and set it as the root job in this job store.

get_root_job_return_value() -> Any

Parse the return value from the root job. Raises an exception if the root job hasn't fulfilled its promise yet.

import_file(src_uri: str, shared_file_name: str, hardlink: bool = False, symlink: bool = False) -> None

import_file(src_uri: str, shared_file_name: None = None, hardlink: bool = False, symlink: bool = False) -> FileID

Imports the file at the given URL into job store. The ID of the newly imported file is returned. If the name of a shared file name is provided, the file will be imported as such and None is returned. If an executable file on the local filesystem is uploaded, its executability will be preserved when it is downloaded. Currently supported schemes are:

export_file(file_id: FileID, dst_uri: str) -> None

Exports file to destination pointed at by the destination URL. The exported file will be executable if and only if it was originally uploaded from an executable file on the local filesystem. Refer to AbstractJobStore.import_file() documentation for currently supported URL schemes. Note that the helper method _exportFile is used to read from the source and write to destination. To implement any optimizations that circumvent this, the _exportFile method should be overridden by subclasses of AbstractJobStore.

classmethod list_url(src_uri: str) -> List[ str]

List the directory at the given URL. Returned path components can be joined with '/' onto the passed URL to form new URLs. Those that end in '/' correspond to directories. The provided URL may or may not end with '/'. Currently supported schemes are:

classmethod get_is_directory(src_uri: str) -> bool

Return True if the thing at the given URL is a directory, and False if it is a file. The URL may or may not end in '/'.

classmethod read_from_url(src_uri: str, writable: IO[bytes]) -> Tuple[int, bool]

Read the given URL and write its content into the given writable stream.

abstract classmethod get_size(src_uri: ParseResult) -> None

Get the size in bytes of the file at the given URL, or None if it cannot be obtained.

abstract destroy() -> None

The inverse of initialize(), this method deletes the physical storage represented by this instance. While not being atomic, this method is at least idempotent, as a means to counteract potential issues with eventual consistency exhibited by the underlying storage mechanisms. This means that if the method fails (raises an exception), it may (and should be) invoked again. If the underlying storage mechanism is eventually consistent, even a successful invocation is not an ironclad guarantee that the physical storage vanished completely and immediately. A successful invocation only guarantees that the deletion will eventually happen. It is therefore recommended to not immediately reuse the same job store location for a new Toil workflow.

get_env() -> Dict[str, str]

Returns a dictionary of environment variables that this job store requires to be set in order to function properly on a worker.

clean(jobCache: Optional[Dict[Union[ str, TemporaryID], JobDescription]] = None) -> JobDescription

Function to cleanup the state of a job store after a restart. Fixes jobs that might have been partially updated. Resets the try counts and removes jobs that are not successors of the current root job.

abstract assign_job_id(job_description: JobDescription) -> None

Get a new jobStoreID to be used by the described job, and assigns it to the JobDescription. Files associated with the assigned ID will be accepted even if the JobDescription has never been created or updated.

batch() -> Iterator[None]

If supported by the batch system, calls to create() with this context manager active will be performed in a batch after the context manager is released.

abstract create_job(job_description: JobDescription) -> JobDescription

Writes the given JobDescription to the job store. The job must have an ID assigned already. Must call jobDescription.pre_update_hook()

abstract job_exists(job_id: str) -> bool

Indicates whether a description of the job with the specified jobStoreID exists in the job store

abstract get_public_url(file_name: str) -> str

Returns a publicly accessible URL to the given file in the job store. The returned URL may expire as early as 1h after its been returned. Throw an exception if the file does not exist.

abstract get_shared_public_url(shared_file_name: str) -> str

Differs from getPublicUrl() in that this method is for generating URLs for shared files written by writeSharedFileStream(). Returns a publicly accessible URL to the given file in the job store. The returned URL starts with 'http:', 'https:' or 'file:'. The returned URL may expire as early as 1h after its been returned. Throw an exception if the file does not exist.

abstract load_job(job_id: str) -> JobDescription

Loads the description of the job referenced by the given ID, assigns it the job store's config, and returns it. May declare the job to have failed (see toil.job.JobDescription.setupJobAfterFailure()) if there is evidence of a failed update attempt.

abstract update_job(job_description: JobDescription) -> None

Persists changes to the state of the given JobDescription in this store atomically. Must call jobDescription.pre_update_hook()

abstract delete_job(job_id: str) -> None

Removes the JobDescription from the store atomically. You may not then subsequently call load(), write(), update(), etc. with the same jobStoreID or any JobDescription bearing it. This operation is idempotent, i.e. deleting a job twice or deleting a non-existent job will succeed silently.

jobs() -> Iterator[JobDescription]

Best effort attempt to return iterator on JobDescriptions for all jobs in the store. The iterator may not return all jobs and may also contain orphaned jobs that have already finished successfully and should not be rerun. To guarantee you get any and all jobs that can be run instead construct a more expensive ToilState object

abstract write_file(local_path: str, job_id: Optional[ str] = None, cleanup: bool = False) -> str

Takes a file (as a path) and places it in this job store. Returns an ID that can be used to retrieve the file at a later time. The file is written in a atomic manner. It will not appear in the jobStore until the write has successfully completed.

abstract write_file_stream(job_id: Optional[str] = None, cleanup: bool = False, basename: Optional[str] = None, encoding: Optional[str] = None, errors: Optional[str] = None) -> Iterator[Tuple[ IO[bytes], str]]

Similar to writeFile, but returns a context manager yielding a tuple of 1) a file handle which can be written to and 2) the ID of the resulting file in the job store. The yielded file handle does not need to and should not be closed explicitly. The file is written in a atomic manner. It will not appear in the jobStore until the write has successfully completed.

abstract get_empty_file_store_id(job_id: Optional[ str] = None, cleanup: bool = False, basename: Optional[str] = None) -> str

Creates an empty file in the job store and returns its ID. Call to fileExists(getEmptyFileStoreID(jobStoreID)) will return True.

abstract read_file(file_id: str, local_path: str, symlink: bool = False) -> None

Copies or hard links the file referenced by jobStoreFileID to the given local file path. The version will be consistent with the last copy of the file written/updated. If the file in the job store is later modified via updateFile or updateFileStream, it is implementation-defined whether those writes will be visible at localFilePath. The file is copied in an atomic manner. It will not appear in the local file system until the copy has completed. The file at the given local path may not be modified after this method returns! Note! Implementations of readFile need to respect/provide the executable attribute on FileIDs.

abstract read_file_stream(file_id: Union[FileID, str], encoding: Literal[None] = None, errors: Optional[str] = None) -> ContextManager[ IO[bytes]]

abstract read_file_stream(file_id: Union[FileID, str], encoding: str, errors: Optional[ str] = None) -> ContextManager[IO[str]]

Similar to readFile, but returns a context manager yielding a file handle which can be read from. The yielded file handle does not need to and should not be closed explicitly.

abstract delete_file(file_id: str) -> None

Deletes the file with the given ID from this job store. This operation is idempotent, i.e. deleting a file twice or deleting a non-existent file will succeed silently.

fileExists(jobStoreFileID: str) -> bool

Determine whether a file exists in this job store.

abstract file_exists(file_id: str) -> bool

Determine whether a file exists in this job store.

getFileSize(jobStoreFileID: str) -> int

Get the size of the given file in bytes.

abstract get_file_size(file_id: str) -> int

Get the size of the given file in bytes, or 0 if it does not exist when queried. Note that job stores which encrypt files might return overestimates of file sizes, since the encrypted file may have been padded to the nearest block, augmented with an initialization vector, etc.

updateFile(jobStoreFileID: str, localFilePath: str) -> None

Replaces the existing version of a file in the job store.

abstract update_file(file_id: str, local_path: str) -> None

Replaces the existing version of a file in the job store. Throws an exception if the file does not exist.

abstract update_file_stream(file_id: str, encoding: Optional[str] = None, errors: Optional[str] = None) -> Iterator[IO[Any]]

Replaces the existing version of a file in the job store. Similar to writeFile, but returns a context manager yielding a file handle which can be written to. The yielded file handle does not need to and should not be closed explicitly.

abstract write_shared_file_stream(shared_file_name: str, encrypted: Optional[bool] = None, encoding: Optional[ str] = None, errors: Optional[str] = None) -> Iterator[IO[bytes]]

Returns a context manager yielding a writable file handle to the global file referenced by the given name. File will be created in an atomic manner.

abstract read_shared_file_stream(shared_file_name: str, encoding: Optional[str] = None, errors: Optional[ str] = None) -> Iterator[IO[bytes]]

Returns a context manager yielding a readable file handle to the global file referenced by the given name.

abstract write_logs(msg: str) -> None

Stores a message as a log in the jobstore.

abstract read_logs(callback: Callable[[...], Any], read_all: bool = False) -> int

Reads logs accumulated by the write_logs() method. For each log this method calls the given callback function with the message as an argument (rather than returning logs directly, this method must be supplied with a callback which will process log messages). Only unread logs will be read unless the read_all parameter is set.

write_leader_pid() -> None

Write the pid of this process to a file in the job store. Overwriting the current contents of pid.log is a feature, not a bug of this method. Other methods will rely on always having the most current pid available. So far there is no reason to store any old pids.

read_leader_pid() -> int

Read the pid of the leader process to a file in the job store.

write_leader_node_id() -> None

Write the leader node id to the job store. This should only be called by the leader.

read_leader_node_id() -> str

Read the leader node id stored in the job store.

write_kill_flag(kill: bool = False) -> None

Write a file inside the job store that serves as a kill flag. The initialized file contains the characters "NO". This should only be changed when the user runs the "toil kill" command. Changing this file to a "YES" triggers a kill of the leader process. The workers are expected to be cleaned up by the leader.

read_kill_flag() -> bool

Read the kill flag from the job store, and return True if the leader has been killed. False otherwise.

default_caching() -> bool

Jobstore's preference as to whether it likes caching or doesn't care about it. Some jobstores benefit from caching, however on some local configurations it can be flaky. see https://github.com/DataBiosphere/toil/issues/4218

__init__(userFunction, *args, **kwargs)

run(fileStore)

Override this function to perform work and dynamically create successor jobs.

run(fileStore)

Override this function to perform work and dynamically create successor jobs.

__init__(job, unitName=None)

addChild(childJob)

Add a childJob to be run as child of this job. Child jobs will be run directly after this job's toil.job.Job.run() method has completed.

addService(service, parentService=None)

Add a service. The toil.job.Job.Service.start() method of the service will be called after the run method has completed but before any successors are run. The service's toil.job.Job.Service.stop() method will be called once the successors of the job have been run. Services allow things like databases and servers to be started and accessed by jobs in a workflow.

addFollowOn(followOnJob)

Add a follow-on job. Follow-on jobs will be run after the child jobs and their successors have been run.

rv(*path) -> Promise

Create a promise (toil.job.Promise). The "promise" representing a return value of the job's run method, or, in case of a function-wrapping job, the wrapped function's return value.

prepareForPromiseRegistration(jobStore)

Set up to allow this job's promises to register themselves. Prepare this job (the promisor) so that its promises can register themselves with it, when the jobs they are promised to (promisees) are serialized. The promissee holds the reference to the promise (usually as part of the job arguments) and when it is being pickled, so will the promises it refers to. Pickling a promise triggers it to be registered with the promissor.

filesToDelete = {}

A set of IDs of files containing promised values when we know we won't need them anymore

__init__(job: Job, path: Any)

Initialize this promise.

__init__(valueOrCallable, *args)

Initialize this Promised Requirement.

getValue()

Return PromisedRequirement value.

static convertPromises(kwargs: Dict[str, Any]) -> bool

Return True if reserved resource keyword is a Promise or PromisedRequirement instance. Converts Promise instance to PromisedRequirement.

__init__(memory: Optional[Union[str, int]] = None, cores: Optional[ Union[str, int, float]] = None, disk: Optional[Union[str, int]] = None, accelerators: Optional[Union[str, int, Mapping[ str, Any], AcceleratorRequirement, Sequence[ Union[str, int, Mapping[str, Any], AcceleratorRequirement]]]] = None, preemptible: Optional[Union[str, int, bool]] = None, preemptable: Optional[Union[str, int, bool]] = None, unitName: Optional[str] = '', checkpoint: Optional[bool] = False, displayName: Optional[ str] = '', descriptionClass: Optional[str] = None) -> None

Job initializer. This method must be called by any overriding constructor.

property jobStoreID

Get the ID of this Job.

property description

Expose the JobDescription that describes this job.

property disk: int

The maximum number of bytes of disk the job will require to run.

property memory

The maximum number of bytes of memory the job will require to run.

property cores

property accelerators

property preemptible

Whether the job can be run on a preemptible node.

property checkpoint

Determine if the job is a checkpoint job or not.

assignConfig(config: Config)

Assign the given config object. It will be used by various actions implemented inside the Job class.

run(fileStore: AbstractFileStore) -> Any

Override this function to perform work and dynamically create successor jobs.

addChild(childJob: Job) -> Job

Add a childJob to be run as child of this job. Child jobs will be run directly after this job's toil.job.Job.run() method has completed.

hasChild(childJob: Job) -> bool

Check if childJob is already a child of this job.

addFollowOn(followOnJob: Job) -> Job

Add a follow-on job. Follow-on jobs will be run after the child jobs and their successors have been run.

hasPredecessor(job: Job) -> bool

Check if a given job is already a predecessor of this job.

hasFollowOn(followOnJob: Job) -> bool

Check if given job is already a follow-on of this job.

addService(service: Service, parentService: Optional[ Service] = None) -> Promise

Add a service. The toil.job.Job.Service.start() method of the service will be called after the run method has completed but before any successors are run. The service's toil.job.Job.Service.stop() method will be called once the successors of the job have been run. Services allow things like databases and servers to be started and accessed by jobs in a workflow.

hasService(service: Service) -> bool

Return True if the given Service is a service of this job, and False otherwise.

addChildFn(fn: Callable, *args, **kwargs) -> FunctionWrappingJob

Add a function as a child job.

addFollowOnFn(fn: Callable, *args, **kwargs) -> FunctionWrappingJob

Add a function as a follow-on job.

addChildJobFn(fn: Callable, *args, **kwargs) -> FunctionWrappingJob

Add a job function as a child job. See toil.job.JobFunctionWrappingJob for a definition of a job function.

addFollowOnJobFn(fn: Callable, *args, **kwargs) -> FunctionWrappingJob

Add a follow-on job function. See toil.job.JobFunctionWrappingJob for a definition of a job function.

property tempDir: str

Shortcut to calling job.fileStore.getLocalTempDir(). Temp dir is created on first call and will be returned for first and future calls :return: Path to tempDir. See job.fileStore.getLocalTempDir

log(text: str, level=20) -> None

Convenience wrapper for fileStore.logToMaster().

static wrapFn(fn, *args, **kwargs)

Makes a Job out of a function. Convenience function for constructor of toil.job.FunctionWrappingJob.

static wrapJobFn(fn, *args, **kwargs)

Makes a Job out of a job function. Convenience function for constructor of toil.job.JobFunctionWrappingJob.

encapsulate(name=None)

Encapsulates the job, see toil.job.EncapsulatedJob. Convenience function for constructor of toil.job.EncapsulatedJob.

rv(*path) -> Any

Create a promise (toil.job.Promise). The "promise" representing a return value of the job's run method, or, in case of a function-wrapping job, the wrapped function's return value.

prepareForPromiseRegistration(jobStore: AbstractJobStore) -> None

Set up to allow this job's promises to register themselves. Prepare this job (the promisor) so that its promises can register themselves with it, when the jobs they are promised to (promisees) are serialized. The promissee holds the reference to the promise (usually as part of the job arguments) and when it is being pickled, so will the promises it refers to. Pickling a promise triggers it to be registered with the promissor.

checkJobGraphForDeadlocks()

Ensures that a graph of Jobs (that hasn't yet been saved to the JobStore) doesn't contain any pathological relationships between jobs that would result in deadlocks if we tried to run the jobs. See toil.job.Job.checkJobGraphConnected(), toil.job.Job.checkJobGraphAcyclic() and toil.job.Job.checkNewCheckpointsAreLeafVertices() for more info.

getRootJobs() -> Set[Job]

Returns the set of root job objects that contain this job. A root job is a job with no predecessors (i.e. which are not children, follow-ons, or services). Only deals with jobs created here, rather than loaded from the job store.

checkJobGraphConnected()

checkJobGraphAcylic()

checkNewCheckpointsAreLeafVertices()

A checkpoint job is a job that is restarted if either it fails, or if any of its successors completely fails, exhausting their retries. A job is a leaf it is has no successors. A checkpoint job must be a leaf when initially added to the job graph. When its run method is invoked it can then create direct successors. This restriction is made to simplify implementation. Only works on connected components of jobs not yet added to the JobStore.

defer(function, *args, **kwargs)

Register a deferred function, i.e. a callable that will be invoked after the current attempt at running this job concludes. A job attempt is said to conclude when the job function (or the toil.job.Job.run() method for class-based jobs) returns, raises an exception or after the process running it terminates abnormally. A deferred function will be called on the node that attempted to run the job, even if a subsequent attempt is made on another node. A deferred function should be idempotent because it may be called multiple times on the same node or even in the same process. More than one deferred function may be registered per job attempt by calling this method repeatedly with different arguments. If the same function is registered twice with the same or different arguments, it will be called twice per job attempt. Examples for deferred functions are ones that handle cleanup of resources external to Toil, like Docker containers, files outside the work directory, etc.

getTopologicalOrderingOfJobs()

saveBody(jobStore)

Save the execution data for just this job to the JobStore, and fill in the JobDescription with the information needed to retrieve it. The Job's JobDescription must have already had a real jobStoreID assigned to it. Does not save the JobDescription.

saveAsRootJob(jobStore: AbstractJobStore) -> JobDescription

Save this job to the given jobStore as the root job of the workflow.

classmethod loadJob(jobStore: AbstractJobStore, jobDescription: JobDescription) -> Job

Retrieves a toil.job.Job instance from a JobStore

__init__(requirements: Mapping[str, Union[ int, str, bool]], jobName: str, unitName: str = '', displayName: str = '', command: Optional[ str] = None) -> None

Create a new JobDescription.

serviceHostIDsInBatches() -> Iterator[List[ str]]

Find all batches of service host job IDs that can be started at the same time. (in the order they need to start in)

successorsAndServiceHosts() -> Iterator[str]

Get an iterator over all child, follow-on, and service job IDs.

allSuccessors()

Get an iterator over all child and follow-on job IDs.

property services

Get a collection of the IDs of service host jobs for this job, in arbitrary order. Will be empty if the job has no unfinished services.

nextSuccessors() -> List[str]

Return the collection of job IDs for the successors of this job that are ready to run. If those jobs have multiple predecessor relationships, they may still be blocked on other jobs. Returns None when at the final phase (all successors done), and an empty collection if there are more phases but they can't be entered yet (e.g. because we are waiting for the job itself to run).

property stack: Tuple[Tuple[str, ...], ...]

Get IDs of successors that need to run still. Batches of successors are in reverse order of the order they need to run in. Some successors in each batch may have already been finished. Batches may be empty. Exists so that code that used the old stack list immutably can work still. New development should use nextSuccessors(), and all mutations should use filterSuccessors() (which automatically removes completed phases).

filterSuccessors(predicate: Callable[[str], bool]) -> None

Keep only successor jobs for which the given predicate function approves. The predicate function is called with the job's ID. Treats all other successors as complete and forgets them.

filterServiceHosts(predicate: Callable[[str], bool]) -> None

Keep only services for which the given predicate approves. The predicate function is called with the service host job's ID. Treats all other services as complete and forgets them.

clear_nonexistent_dependents(job_store: AbstractJobStore) -> None

Remove all references to child, follow-on, and associated service jobs that do not exist (i.e. have been completed and removed) in the given job store.

clear_dependents() -> None

Remove all references to child, follow-on, and associated service jobs.

is_subtree_done() -> bool

Return True if the job appears to be done, and all related child, follow-on, and service jobs appear to be finished and removed.

replace(other: JobDescription) -> None

Take on the ID of another JobDescription, retaining our own state and type. When updated in the JobStore, we will save over the other JobDescription. Useful for chaining jobs: the chained-to job can replace the parent job. Merges cleanup state from the job being replaced into this one.

addChild(childID: str) -> None

Make the job with the given ID a child of the described job.

addFollowOn(followOnID: str) -> None

Make the job with the given ID a follow-on of the described job.

addServiceHostJob(serviceID, parentServiceID=None)

Make the ServiceHostJob with the given ID a service of the described job. If a parent ServiceHostJob ID is given, that parent service will be started first, and must have already been added.

hasChild(childID: str) -> bool

Return True if the job with the given ID is a child of the described job.

hasFollowOn(followOnID: str) -> bool

Test if the job with the given ID is a follow-on of the described job.

hasServiceHostJob(serviceID) -> bool

Test if the ServiceHostJob is a service of the described job.

renameReferences(renames: Dict[TemporaryID, str]) -> None

Apply the given dict of ID renames to all references to jobs. Does not modify our own ID or those of finished predecessors. IDs not present in the renames dict are left as-is.

addPredecessor() -> None

Notify the JobDescription that a predecessor has been added to its Job.

onRegistration(jobStore: AbstractJobStore) -> None

Called by the Job saving logic when this JobDescription meets the JobStore and has its ID assigned. Overridden to perform setup work (like hooking up flag files for service jobs) that requires the JobStore.

setupJobAfterFailure(exit_status: Optional[int] = None, exit_reason: Optional[BatchJobExitReason] = None)

Reduce the remainingTryCount if greater than zero and set the memory to be at least as big as the default memory (in case of exhaustion of memory, which is common). Requires a configuration to have been assigned (see toil.job.Requirer.assignConfig()).

getLogFileHandle(jobStore)

Returns a context manager that yields a file handle to the log file. Assumes logJobStoreFileID is set.

property remainingTryCount

The try count set on the JobDescription, or the default based on the retry count from the config if none is set.

clearRemainingTryCount() -> bool

Clear remainingTryCount and set it back to its default value.

pre_update_hook() -> None

Called by the job store before pickling and saving a created or updated version of a job.

get_job_kind() -> str

Returns an identifier of the job for use with the message bus. Either the unit name, job name, or display name, which identifies the kind of job it is to toil. Otherwise returns Unknown Job in case no identifier is available

static getDefaultArgumentParser() -> ArgumentParser

Get argument parser with added toil workflow options.

static getDefaultOptions(jobStore: str) -> Namespace

Get default options for a toil workflow.

static addToilOptions(parser)

Adds the default toil options to an optparse or argparse parser object.

static startToil(job, options)

Run the toil workflow using the given options. Deprecated by toil.common.Toil.start. (see Job.Runner.getDefaultOptions and Job.Runner.addToilOptions) starting with this job. :param toil.job.Job job: root job of the workflow :raises: toil.leader.FailedJobsException if at the end of function their remain failed jobs. :return: The return value of the root job's run function. :rtype: Any

__init__(jobStore: AbstractJobStore, jobDesc: JobDescription, file_store_dir: str, waitForPreviousCommit: Callable[[], Any]) -> None

Create a new file store object.

static createFileStore(jobStore: AbstractJobStore, jobDesc: JobDescription, file_store_dir: str, waitForPreviousCommit: Callable[[], Any], caching: Optional[bool]) -> Union[NonCachingFileStore, CachingFileStore]

Create a concreate FileStore.

static shutdownFileStore(workflowID: str, config_work_dir: Optional[str], config_coordination_dir: Optional[ str]) -> None

Carry out any necessary filestore-specific cleanup. This is a destructive operation and it is important to ensure that there are no other running processes on the system that are modifying or using the file store for this workflow. This is the intended to be the last call to the file store in a Toil run, called by the batch system cleanup function upon batch system shutdown.

open(job: Job) -> Generator[None, None, None]

Create the context manager around tasks prior and after a job has been run. File operations are only permitted inside the context manager. Implementations must only yield from within with super().open(job):.

getLocalTempDir() -> str

Get a new local temporary directory in which to write files. The directory will only persist for the duration of the job.

getLocalTempFile(suffix: Optional[str] = None, prefix: Optional[str] = None) -> str

Get a new local temporary file that will persist for the duration of the job.

getLocalTempFileName(suffix: Optional[str] = None, prefix: Optional[str] = None) -> str

Get a valid name for a new local file. Don't actually create a file at the path.

abstract writeGlobalFile(localFileName: str, cleanup: bool = False) -> FileID

Upload a file (as a path) to the job store. If the file is in a FileStore-managed temporary directory (i.e. from toil.fileStores.abstractFileStore.AbstractFileStore.getLocalTempDir()), it will become a local copy of the file, eligible for deletion by toil.fileStores.abstractFileStore.AbstractFileStore.deleteLocalFile(). If an executable file on the local filesystem is uploaded, its executability will be preserved when it is downloaded again.

writeGlobalFileStream(cleanup: bool = False, basename: Optional[str] = None, encoding: Optional[str] = None, errors: Optional[str] = None) -> Iterator[Tuple[WriteWatchingStream, FileID]]

Similar to writeGlobalFile, but allows the writing of a stream to the job store. The yielded file handle does not need to and should not be closed explicitly.

logAccess(fileStoreID: Union[FileID, str], destination: Optional[str] = None) -> None

Record that the given file was read by the job. (to be announced if the job fails) If destination is not None, it gives the path that the file was downloaded to. Otherwise, assumes that the file was streamed. Must be called by readGlobalFile() and readGlobalFileStream() implementations.

abstract readGlobalFile(fileStoreID: str, userPath: Optional[str] = None, cache: bool = True, mutable: bool = False, symlink: bool = False) -> str

Make the file associated with fileStoreID available locally. If mutable is True, then a copy of the file will be created locally so that the original is not modified and does not change the file for other jobs. If mutable is False, then a link can be created to the file, saving disk resources. The file that is downloaded will be executable if and only if it was originally uploaded from an executable file on the local filesystem. If a user path is specified, it is used as the destination. If a user path isn't specified, the file is stored in the local temp directory with an encoded name. The destination file must not be deleted by the user; it can only be deleted through deleteLocalFile. Implementations must call logAccess() to report the download.

abstract readGlobalFileStream(fileStoreID: str, encoding: Optional[str] = None, errors: Optional[str] = None) -> ContextManager[Union[ IO[bytes], IO[str]]]

Read a stream from the job store; similar to readGlobalFile. The yielded file handle does not need to and should not be closed explicitly.

getGlobalFileSize(fileStoreID: Union[FileID, str]) -> int

Get the size of the file pointed to by the given ID, in bytes. If a FileID or something else with a non-None 'size' field, gets that. Otherwise, asks the job store to poll the file's size. Note that the job store may overestimate the file's size, for example if it is encrypted and had to be augmented with an IV or other encryption framing.

abstract deleteLocalFile(fileStoreID: Union[FileID, str]) -> None

Delete local copies of files associated with the provided job store ID. Raises an OSError with an errno of errno.ENOENT if no such local copies exist. Thus, cannot be called multiple times in succession. The files deleted are all those previously read from this file ID via readGlobalFile by the current job into the job's file-store-provided temp directory, plus the file that was written to create the given file ID, if it was written by the current job from the job's file-store-provided temp directory.

abstract deleteGlobalFile(fileStoreID: Union[FileID, str]) -> None

Delete local files and then permanently deletes them from the job store. To ensure that the job can be restarted if necessary, the delete will not happen until after the job's run method has completed.

logToMaster(text: str, level: int = 20) -> None

Send a logging message to the leader. The message will also be logged by the worker at the same level.

abstract startCommit(jobState: bool = False) -> None

Update the status of the job on the disk. May start an asynchronous process. Call waitForCommit() to wait on that process.

abstract waitForCommit() -> bool

Blocks while startCommit is running. This function is called by this job's successor to ensure that it does not begin modifying the job store until after this job has finished doing so. Might be called when startCommit is never called on a particular instance, in which case it does not block.

abstract classmethod shutdown(shutdown_info: Any) -> None

Shutdown the filestore on this node. This is intended to be called on batch system shutdown.

__init__(fileStoreID: str, size: int, executable: bool = False) -> None