Using PMC for Pegasus Task Clustering¶

PMC can be used as the wrapper for executing clustered jobs in Pegasus. In this mode Pegasus groups several tasks together and submits them as a single clustered job to a remote system. PMC then executes the individual tasks in the cluster and returns the results.

PMC can be specified as the task manager for clustered jobs in Pegasus in three ways:

#PEGASUS PROPERTIES FILE
pegasus.clusterer.job.aggregator=mpiexec

<site handle="siteX" arch="x86" os="LINUX">


    ...


    <profile namespace="pegasus" key="job.aggregator">mpiexec</profile>
</site>

tr B {


    site siteX {


        pfn "/path/to/mytask"


        arch "x86"


        os "linux"


        type "INSTALLED"


        profile pegasus "clusters.size" "3"


        profile pegasus "job.aggregator" "mpiexec"


    }
}

It is usually necessary to have a pegasus::mpiexec entry in your transformation catalog that specifies a) the path to PMC on the remote site and b) the relevant globus profiles such as xcount, host_xcount and maxwalltime to control size of the MPI job. That entry would look like this:

tr pegasus::mpiexec {


    site siteX {


        pfn "/path/to/pegasus-mpi-cluster"


        arch "x86"


        os "linux"


        type "INSTALLED"


        profile globus "maxwalltime" "240"


        profile globus "host_xcount" "1"


        profile globus "xcount" "32"


    }
}

If this transformation catalog entry is not specified, Pegasus will attempt create a default path on the basis of the environment profile PEGASUS_HOME specified in the site catalog for the remote site.

PMC can be used with both horizontal and label-based clustering in Pegasus, but we recommend using label-based clustering so that entire sub-graphs of a Pegasus DAX can be clustered into a single PMC job, instead of only a single level of the workflow.

Pegasus Profiles for PMC¶

There are several Pegasus profiles that map to PMC task options:

pmc_request_memory

pmc_request_cpus

pmc_priority

These profiles are used by Pegasus when generating PMC’s input DAG when PMC is used as the task manager for clustered jobs in Pegasus.

The profiles can be specified in the DAX like this:

<job id="ID0000001" name="mytask">


    <arguments>-a 1 -b 2 -c 3</arguments>


    ...


    <profile namespace="pegasus" key="pmc_request_memory">1024</profile>


    <profile namespace="pegasus" key="pmc_request_cpus">4</profile>


    <profile namespace="pegasus" key="pmc_priority">10</profile>
</job>

This example specifies a PMC task that requires 1GB of memory and 4 cores, and has a priority of 10. It produces a task in the PMC DAG that looks like this:

TASK mytask_ID00000001 -m 1024 -c 4 -p 10 /path/to/mytask -a 1 -b 2 -c 3

Using PMC for the Entire Pegasus DAX¶

Pegasus can also be configured to run the entire workflow as a single PMC job. In this mode Pegasus will generate a single PMC DAG for the entire workflow as well as a PBS script that can be used to submit the workflow.

In contrast to using PMC as a task clustering tool, in this mode there are no jobs in the workflow executed without PMC. The entire workflow, including auxilliary jobs such as directory creation and file transfers, is managed by PMC. If Pegasus is configured in this mode, then DAGMan and Condor are not required.

To run in PMC-only mode, set the property "pegasus.code.generator" to "PMC" in the Pegasus properties file:

pegasus.code.generator=PMC

In order to submit the resulting PBS job you may need to make changes to the .pbs file generated by Pegasus to get it to work with your cluster. This mode is experimental and has not been used extensively.

Memory¶

Users can specify both the amount of memory required per task, and the amount of memory available per host. If the amount of memory required by any task exceeds the available memory of all the hosts, then the workflow will be aborted. By default, the host memory is determined automatically, however the user can specify --host-memory to "lie" to pegasus-mpi-cluster. The amount of memory required for each task is specified in the DAG using the -m/--request-memory argument (see DAG Files).

CPUs¶

Users can specify the number of CPUs required per task, and the total number of CPUs available on each host. If the number of CPUs required by a task exceeds the available CPUs on all hosts, then the workflow will be aborted. By default, the number of CPUs on a host is determined automatically, but the user can specify --host-cpus to over- or under-subscribe the host. The number of CPUs required for each task is specified in the DAG using the -c/--request-cpus argument (see DAG Files).

I/O forwarding using pipes¶

I/O forwarding with pipes works by having PMC worker processes collect data from each task using UNIX pipes. This approach is more efficient than the file-based approach, but it requires the code of the task to be changed so that the task writes to the pipe instead of a regular file.

In order to use I/O forwarding a PMC task just needs to specify the -f/--pipe-forward argument to specify the name of the file to forward data to, and the name of an environment variable through which the PMC worker process can inform it of the file descriptor for the pipe.

For example, if there is a task "mytask" that needs to forward data to two files: "myfile.a" and "myfile.b", it would look like this:

TASK mytask -f A=/tmp/myfile.a -f B=/tmp/myfile.b /bin/mytask

When the /bin/mytask process starts it will have two variables in its environment: "A=3" and "B=4", for example. The value of these variables is the file descriptor number of the corresponding files. In this case, if the task wants to write to "/tmp/myfile.a", it gets the value of environment variable "A", and calls write() on that descriptor number. In C the code for that looks like this:

char *A = getenv("A");
int fd = atoi(A);
char *message = "Hello, World\n";
write(fd, message, strlen(message));

In some programming languages it is not possible to write to a file descriptor directly. Fortran, for example, refers to files by unit number instead of using file descriptors. In these languages you can either link C I/O functions into your binary and call them from routines written in the other language, or you can open a special file in the Linux /proc file system to get another handle to the pipe you want to access. For the latter, the file you should open is "/proc/self/fd/NUMBER" where NUMBER is the file descriptor number you got from the environment variable. For the example above, the pipe for myfile.a (environment variable A) is "/proc/self/fd/3".

If you are using pegasus-kickstart, which is probably the case if you are using PMC for a Pegasus workflow, then there’s a trick you can do to avoid modifying your code. You use the /proc file system, as described above, but you let pegasus-kickstart handle the path construction. For example, if your application has an argument, -o, that allows you to specify the output file then you can write your task like this:

TASK mytask -f A=/tmp/myfile.a /bin/pegasus-kickstart /bin/mytask -o /proc/self/fd/$A

In this case, pegasus-kickstart will replace the $A in your application arguments with the file descriptor number you want. Your code can open that path normally, write to it, and then close it as if it were a regular file.

I/O forwarding using files¶

I/O forwarding with files works by having tasks write out data in files on the local disk. The PMC worker process reads these files and forwards the data to the master where it can be written to the desired output file. This approach may be much less efficient than using pipes because it involves the file system, which has more overhead than a pipe.

File forwarding can be enabled by giving the -F/--file-forward argument to a task.

Here’s an example:

TASK mytask -F /tmp/foo.0=/scratch/foo /bin/mytask -o /tmp/foo.0

In this case, the worker process will expect to find the file /tmp/foo.0 when mytask exits successfully. It reads the data from that file and sends it to the master to be written to the end of /scratch/foo. After /tmp/foo.0 is read it will be deleted by the worker process.

This approach works best on systems where the local disk is a RAM file system such as Cray XT machines. Alternatively, the task can use /dev/shm on a regular Linux cluster. It might also work relatively efficiently on a local disk if the file system cache is able to absorb all of the reads and writes.

I/O forwarding caveats¶

When using I/O forwarding it is important to consider a few caveats.

First, if the PMC job fails for any reason (including when the workflow is aborted for violating --max-wall-time), then the files containing forwarded I/O may be corrupted. They can include partial records, meaning that only part of the I/O from one or more tasks was written, and they can include duplicate records, meaning that the I/O was written, but the PMC job failed before the task could be marked as successful, and the workflow was restarted later. We make no guarantees about the contents of the data files in this case. It is up to the code that reads the files to a) detect and b) recover from such problems. To eliminate duplicates the records should include a unique identifier, and to eliminate partials the records should include a checksum.

Second, you should not use I/O forwarding if your task is going to write a lot of data to the file. Because the PMC worker is reading data off the pipe/file into memory and sending it in an MPI message, if you write too much, then the worker process will run the system out of memory. Also, all the data needs to fit in a single MPI message. In pipe forwarding there is no hard limit on the size, but in file forwarding the limit is 1MB. We haven’t benchmarked the performance on large I/O, but anything larger than about 1 MB is probably too much. At any rate, if your data is larger than 1MB, then I/O forwarding probably won’t have much of a performance benefit anyway.

Third, the I/O is not written to the file if the task returns a non-zero exitcode. We assume that if the task failed that you don’t want the data it produced.

Fourth, the data from different tasks is not interleaved. All of the data written by a given task will appear sequentially in the output file. Note that you can still get partial records, however, if any data from a task appears it will never be split among non-adjacent ranges in the output file. If you have 3 tasks that write: "I am a task" you can get:

I am a taskI am a taskI am a task

and:

I am a taskI amI am a task

but not:

I am a taskI amI am a task a task

Fifth, data from different tasks appears in arbitrary order in the output file. It depends on what order the tasks were executed by PMC, which may be arbitrary if there are no dependencies between the tasks. The data that is written should contain enough information that you are able to determine which task produced it if you require that. PMC does not add any headers or trailers to the data.

Sixth, a task will only be marked as successful if all of its I/O was successfully written. If the workflow completed successfully, then the I/O is guaranteed to have been written.

Seventh, if the master is not able to write to the output file for any reason (e.g. the master tries to write the I/O to the destination file, but the write() call returns an error) then the task is marked as failed even if the task produced a non-zero exitcode. In other words, you may get a non-zero kickstart record even when PMC marks the task failed.

Eighth, the pipes are write-only. If you need to read and write data from the file you should use file forwarding and not pipe forwarding.

Ninth, all files are opened by the master in append mode. This is so that, if the workflow fails and has to be restarted, or if a task fails and is retried, the data that was written previously is not lost. PMC never truncates the files. This is one of the reasons why you can have partial records and duplicate records in the output file.

Finally, in file forwarding the output file is removed when the task exits. You cannot rely on the file to be there when the next task runs even if you write it to a shared file system.

Resource Utilization¶

At the end of the workflow run, the master will report the resource utilization of the job. This is done by adding up the total runtimes of all the tasks executed (including failed tasks) and dividing by the total wall time of the job times N, where N is both the total number of processes including the master, and the total number of workers. These two resource utilization values are provided so that users can get an idea about how efficiently they are making use of the resources they allocated. Low resource utilization values suggest that the user should use fewer cores, and longer wall time, on future runs, while high resource utilization values suggest that the user could use more cores for future runs and get a shorter wall time.

fork() and exec()¶

In order for the worker processes to start tasks on the compute node the compute nodes must support the fork() and exec() system calls. If your target machine runs a stripped-down OS on the compute nodes that does not support these system calls, then pegasus-mpi-cluster will not work.

CPU Usage¶

Many MPI implementations are optimized so that message sends and receives do busy waiting (i.e. they spin/poll on a message send or receive instead of sleeping). The reasoning is that sleeping adds overhead and, since many HPC systems use space sharing on dedicated hardware, there are no other processes competing, so spinning instead of sleeping can produce better performance. On those implementations MPI processes will run at 100% CPU usage even when they are just waiting for a message. This is a big problem for multicore tasks in pegasus-mpi-cluster because idle slots consume CPU resources. In order to solve this problem pegasus-mpi-cluster processes sleep for a short period between checks for waiting messages. This reduces the load significantly, but causes a short delay in receiving messages. If you are using an MPI implementation that sleeps on message send and receive instead of doing busy waiting, then you can disable the sleep by specifying the --no-sleep-on-recv option. Note that the master will always sleep if --max-wall-time is specified because there is no way to interrupt or otherwise timeout a blocking call in MPI (e.g. SIGALRM does not cause MPI_Recv to return EINTR).