The Distributed Resource Management Application API

Frédéric is a member of the Market Development Engineering group at Sun Microsystems. He can be contacted at [email protected].

A key component of cluster grids is the Distributed Resource Management (DRM) software that queues, dispatches, and controls jobs in the grid. To date, DRM has provided either only proprietary interfaces for application integration or nothing at all, in which case, the command-line interface is used. This means that there are no standard mechanisms for programmers to integrate applications with a grid. As a result, grid integration has been left to end users, thereby limiting the adoption of grid technology to industries where the pain of integration is under control. Hence, we see little grid computing for enterprise applications. For DRM adoption to be more generically successful and grid computing to break into the enterprise, there needs to be a fundamental shift in the adoption pattern of DRM—the integration must move away from the hands of the end user to the hands of developers. For this to happen, there must be a standard and portable layer that you can program to, without the fear of being locked-in or of supporting excessive porting and deployment costs in the future.

There is light at the end of the tunnel. The Distributed Resource Management Application API (DRMAA) working group (http://www.drmaa.org/) has released the DRMAA 1.0 specification that offers a standardized API for application integration with C, Java, and Perl bindings. Altair, IBM, Sun, Intel, and other industry players participated in this working group, bringing the confidence that DRMAA is the consensus API. The availability of DRMAA is the key to enable the fundamental shift needed to drive grid adoption in commercial industries. And as of today, DRMAA 1.0 has already been implemented in Sun's Grid Engine 6 (http://wwws.sun.com/software/gridware/ and http://gridengine.sunsource.net/) and the University of Wisconsin-Madison's Condor 6.7 (http://www.cs.wisc.edu/condor/).

The DRMAA C API

The scope of the initial DRMAA specification is limited to job submission, job monitoring, and control, as well as retrieval of the finished job status. The interfaces are grouped in five categories:

• Init and exit.
• Job template handling.
• Job submission.
• Job monitoring and control.
• Auxiliary or informational routines.

Job-related routines require the initialization of a DRMAA session, auxiliary and informational routines do not. Every call to a DRMAA function returns an error code. If everything goes well, that code is DRMAA_ERRNO_SUCCESS; if not, an appropriate error code is returned.

Every DRMAA function also takes at least two parameters—a string to populate with an error message in case of an error, and an integer representing the maximum length of the error string. In this article, I use examples to examine the DRMAA C routines.

Compiling a DRMAA Program

At compile time, you must specify the path to the drmaa.h header file to be included in the various source files; for example:

cc -c -Ipath_to_drmaa_h filename.c

At link time, you need to link with the DRMAA library and specify the path where it sits. Because the DRMAA library typically won't reside in the standard library directory, it is also useful to set the runtime library path at link time using the -R flag, so end users won't have to play around with the LD_LIBRARY_PATH environment variable. Mind that the link time and runtime paths to the DRMAA library may not be the same. The link line eventually will look like this :

cc -o programname -Lpath_to_libdrmaa_so
-Rpath_to_libdrmaa_so -ldrmaa filename.o

Starting and Stopping Sessions

The drmaa_init() function (see Listing One) sets up the DRMAA session and must be called before most other DRMAA functions. If a job-related function is called before drmaa_init() returns, it returns the error code DRMAA_ERRNO_NO_ACTIVE_SESSION. Once drmaa_init() has been called successfully, it is the responsibility of the calling application to also call drmaa_exit() before terminating. If an application does not call drmaa_exit() before terminating, the session state may be left behind and performance of the DRM scheduler may be impacted. Similarly, if a job-related function is called after drmaa_exit() is called, it returns the error code DRMAA_ERRNO_NO_ACTIVE_SESSION.

Running Jobs

As you can see in Listing Two, the sequence for running a job consists of four major steps:

1. A new job is created using drmaa_allocate_job_template().
2. Attributes are then set using a call to drmaa_set_attribute() or drmaa_set_vector_attribute(); drmaa_set_vector_attribute() is used when the attribute is an array. At a minimum, the name of the program to run (DRMAA_REMOTE_COMMAND) and its arguments (DRMAA_V_ARGV) have to be set. Other attributes include the directory where the command will be run, an e-mail address of where to report completion of the job, the future start time for the job, and so on.
3. The job is run by a call to the function drmaa_run_job(). There is also a drmaa_run_bulk_jobs() function to start an array of jobs at once with a slightly different C prototype.
4. Finally, any memory allocated for the job must be freed by calling drmaa_delete_job_template(); this call has no effect on the submitted job, nor has drmaa_exit().

Getting Job Completion Status

To get the completion status of a job, you first need to wait until it completes, which is achieved by a call to drmaa_wait() or drmaa_synchronize() (see Listing Three). drmaa_wait() is used to wait on a single job; drmaa_synchronize() is used to wait on a bunch of jobs but requires a subsequent call to drmaa_wait() to get status information of each job.

Arguments to drmaa_wait() include the job we want to wait for, a specific job ID or DRMAA_JOB_IDS_SESSION_ANY for any job, along with a place to write the ID of the job for which you actually waited, and how long you are willing to wait for the job to finish, which can be a specific number of seconds, DRMAA_TIMEOUT_WAIT_FOREVER to forever wait or DRMAA_TIMEOUT_NO_WAIT to return immediately if the jobs is not complete.

The status information received from drmaa_wait() needs to be postprocessed by the various drmaa_w...() calls presented in the aforementioned example to learn whether a job has run, whether it has run to completion and its return code, and what may have provoked an abort. Once information about a job has been retrieved, it is disposed by the DRMAA library and can no longer be retrieved. It is the responsibility of the application to retrieve, or directly dispose using drmaa_synchronize(), the information for every job; not doing so creates a memory leak.

A Sample Wrapper DRMAA Library

Scalable applications tap into more CPU power by creating threads or forking processes inside large shared-memory processor (SMP) servers or over a cluster of dedicated smaller systems. With DRMAA, you aren't limited by the size of the box or cluster and can request CPU power and spin processes over to a grid they do not own beforehand.

However, for a complete solution, there is a missing piece in DRMAA 1.0—the retrieval of the job output. While you have to work out the appropriate solution for your project, I see three generic ways to communicate the result back to the master thread:

• Encode the result in the job return code, which can be retrieved by the DRMAA interfaces.
• Program the slave thread to directly communicate back to the master thread using sockets or more elaborate remote protocols.
• Use the DRMAA_TRANSFER_FILES job template attribute to transfer output files back to the master host and read the file.

Option 1 is simple but works only with a minimal output. The additional programming required by Option 2 can be minimized by using higher level communication protocols such as RPC (http://www.ietf.org/rfc/rfc1831.txt) in C or RMI (http://java.sun.com/j2se/) in Java. Finally, you may not be able to apply Option 3 as the DRMAA_TRANSFER_FILES attribute is optional in the 1.0 DRMAA specification; that is, DRM is not required to implement it to satisfy DRMAA 1.0.

As a more elaborate example, I present a sample distributed computing interface, called "dthread," that makes grid programming easy. It is built off DRMAA and mimics the POSIX pthread interface (http://www.opengroup.org/onlinepubs/ 009695399/) to ease the migration of existing applications to the grid. To make it more concrete, I use the example of capital markets. As investment firms face an ongoing challenge to ever-increasing product complexity and ever-decreasing settlement windows, they can use the grid to keep up with the growing computational requirements.

Listing Four details the interface. Similar to pthread_create(), the dthread_create() function creates a new thread on the grid with attributes previously initialized by dthread_attr_init(). Similar to pthread_join(), the dthread_join() function suspends processing of the calling process until the target thread completes and retrieves the return value of the thread. Using this dthread interface, a pthread application would be modified as in Listings Five and Six. Listing Seven is a sample implementation. For simplicity, it uses the return code to pass the output back to the master thread. Use this code to build the dthread library:

cc -KPIC -c -Ipath_to_drmaa_h dthread.c
ld -G -o libdthread.so -Lpath_to_libdrmaa_so
-Rpath_to_libdrmaa_so -ldrmaa -lc
dthread.o

Conclusion

DRMAA facilitates the development of grid-enabled apps by offering a standard interface upon which you can program without knowing in advance the deployment properties—the brand and configuration of the DRM, a problem that has so far prevented ISV from offering grid-aware software. As a result, DRMAA will ease the adoption of grid computing in the enterprise world where custom integration costs are high.

The dthread programming interface for distributed computing encapsulates the various DRMAA routines under four simple calls that mimic the pthread calls, which makes the migration of multithreaded apps to the grid straightforward. I recommend using dthread only for loosely coupled tasks; otherwise, the burden of programming data communication back and forth between the master and slave tasks is too high. In that context, it is best to use a message-based programming interface and treat the entire set of tasks as a single grid job.

Acknowledgment

I would like to thank Andreas Haas from the N1 Grid Engine engineering group at Sun Microsystems for his input and feedback.

DDJ

Listing One

#include <stdio.h>
#include "drmaa.h"
int main (int argc, char **argv) {
   char error[DRMAA_ERROR_STRING_BUFFER];
   if (drmaa_init (NULL, error, DRMAA_ERROR_STRING_BUFFER)
      != DRMAA_ERRNO_SUCCESS) {
   fprintf (stderr, "Could not initialize the DRMAA library: %s\n", error);
   return 1;
}
fprintf (stdout, "Successfully started the DRMAA library\n");
if (drmaa_exit (error, DRMAA_ERROR_STRING_BUFFER)
      != DRMAA_ERRNO_SUCCESS) {
   fprintf (stderr, "Could not shut down the DRMAA library: %s\n", error);
   return 1;
   }
   return 0;
}

#include <stdio.h>
#include "drmaa.h"
int main (int argc, char **argv) {
   char error[DRMAA_ERROR_STRING_BUFFER];
   drmaa_job_template_t *jt = NULL;
   const char *args[2] = {"jobargument", NULL};
   char jobid[DRMAA_JOBNAME_BUFFER];
   if (drmaa_init (NULL, error, DRMAA_ERROR_STRING_BUFFER)
       != DRMAA_ERRNO_SUCCESS) {
   fprintf (stderr, "Could not initialize the DRMAA library: %s\n", error);
   return 1;
}
if (drmaa_allocate_job_template (&jt, error, DRMAA_ERROR_STRING_BUFFER)
       != DRMAA_ERRNO_SUCCESS) {
   fprintf (stderr, "Could not create job template: %s\n", error);
   goto exit_drmaa;
}
if (drmaa_set_attribute (jt, DRMAA_REMOTE_COMMAND, "jobname",
                error, DRMAA_ERROR_STRING_BUFFER) != DRMAA_ERRNO_SUCCESS) {
   fprintf (stderr, "Could not set attribute \"%s\": %s\n",
                                              DRMAA_REMOTE_COMMAND, error);
   goto exit_drmaa;
}
if (drmaa_set_vector_attribute (jt, DRMAA_V_ARGV, args, error,
      DRMAA_ERROR_STRING_BUFFER) != DRMAA_ERRNO_SUCCESS) {
   fprintf (stderr, "Could not set attribute \"%s\": %s\n",
                                              DRMAA_REMOTE_COMMAND, error);
   goto exit_drmaa;
}
fprintf (stdout, "Your job has been submitted with id %s\n", jobid);
if (drmaa_delete_job_template (jt, error, DRMAA_ERROR_STRING_BUFFER)
      != DRMAA_ERRNO_SUCCESS)
   fprintf (stderr, "Could not delete job template: %s\n", error);
exit_drmaa:
   if (drmaa_exit (error, DRMAA_ERROR_STRING_BUFFER)
      != DRMAA_ERRNO_SUCCESS) {
   fprintf (stderr, "Could not shut down the DRMAA library: %s\n", error);
   return 1;
  }
  return 0;
}

(...)
int status = 0;
drmaa_attr_values_t *rusage = NULL;
if (drmaa_wait (DRMAA_JOB_IDS_SESSION_ANY, jobid, DRMAA_JOBNAME_BUFFER, 
        &status, DRMAA_TIMEOUT_WAIT_FOREVER, &rusage,
        error, DRMAA_ERROR_STRING_BUFFER) != DRMAA_ERRNO_SUCCESS) {
   fprintf (stderr, "Could not wait for job: %s\n", error);
} else {
   char usage[DRMAA_ERROR_STRING_BUFFER];
   int aborted = 0;
   drmaa_wifaborted(&aborted, status, NULL, 0);
   if (aborted == 1) {
      printf("Job %s never ran\n", jobid);
   } else {
   int exited = 0;
   drmaa_wifexited(&exited, status, NULL, 0);
   if (exited == 1) {
      int exit_status = 0;
      drmaa_wexitstatus(&exit_status, status, NULL, 0);
      printf("Job %s finished regularly with exit status %d\n", 
                                                    jobid, exit_status);
   } else {
      int signaled = 0;
      drmaa_wifsignaled(&signaled, status, NULL, 0);
      if (signaled == 1) {
         char termsig[DRMAA_SIGNAL_BUFFER+1];
         drmaa_wtermsig(termsig, DRMAA_SIGNAL_BUFFER, status, NULL, 0);
         printf("Job %s finished due to signal %s\n", jobid, termsig);
      } else {
         printf("Job %s finished with unclear conditions\n", jobid);
      }
    }
  }
}
drmaa_release_attr_values (rusage);
(...)

#ifndef __DTHREAD_H
#define __DTHREAD_H
#include "drmaa.h"
typedef char dthread_t[DRMAA_JOBNAME_BUFFER];
typedef drmaa_job_template_t *dthread_attr_t;
extern int dthread_create (dthread_t *,const dthread_attr_t *,char *,void *);
extern int dthread_join (dthread_t, void **);
extern int dthread_attr_init (dthread_attr_t *);
extern int dthread_attr_destroy (dthread_attr_t *);
#endif /* __DTHREAD_H */

#include <pthread.h>
(...)
pthread_t tid;
void *price_func(void *), *arg;
float price;
pthread_attr_t attr;
(...)
/* spawn off computation to another CPU */
pthread_attr_init (&attr);
pthread_attr_setscope (&attr, PTHREAD_SCOPE_SYSTEM);
pthread_create (&tid, &attr, price_func, arg);
(...)
/* retrieve result */
pthread_join (tid, &price);

#include "dthread.h"
(...)
dthread_t tid;
char *price_func, *arg;
float price;
dthread_attr_t attr;
(...)
/* spawn off computation to the grid */
dthread_attr_init (&attr);
dthread_create (&tid, &attr, price_func, arg);
(...)
/* retrieve result */
dthread_join (tid, &price);

#include <errno.h>
#include <stdio.h>
#include <stdlib.h>
#include "dthread.h"
void _init () {
   char error[DRMAA_ERROR_STRING_BUFFER];
   if (drmaa_init (NULL, error, DRMAA_ERROR_STRING_BUFFER)
      != DRMAA_ERRNO_SUCCESS)
   fprintf (stderr, "Could not initialize the dthread library: %s\n", error);
}
void _fini () {
   char error[DRMAA_ERROR_STRING_BUFFER];
   if (drmaa_exit (error, DRMAA_ERROR_STRING_BUFFER)
      != DRMAA_ERRNO_SUCCESS)
   fprintf (stderr, "Could not shut down the dthread library: %s\n", error);
}
int dthread_attr_init (dthread_attr_t *attr) {
   char error[DRMAA_ERROR_STRING_BUFFER];
   if (drmaa_allocate_job_template (attr, error, DRMAA_ERROR_STRING_BUFFER)
      != DRMAA_ERRNO_SUCCESS)
   return ELIBACC;
   if (drmaa_set_attribute (*attr, DRMAA_OUTPUT_PATH, ":/dev/null",
                 error, DRMAA_ERROR_STRING_BUFFER) != DRMAA_ERRNO_SUCCESS)
   return ELIBACC;
   if (drmaa_set_attribute (*attr, DRMAA_ERROR_PATH, ":/dev/null",
                 error, DRMAA_ERROR_STRING_BUFFER) != DRMAA_ERRNO_SUCCESS)
   return ELIBACC;
   return 0;
}
int dthread_attr_destroy (dthread_attr_t *attr) {
   char error[DRMAA_ERROR_STRING_BUFFER];
   drmaa_job_template_t *jt = *attr;
   if (drmaa_delete_job_template (jt, error, DRMAA_ERROR_STRING_BUFFER)
      != DRMAA_ERRNO_SUCCESS)
   return ELIBACC;
return 0;
}
int dthread_create (dthread_t *thread, const dthread_attr_t *attr,
                                            char *threadname, void *arg) {
   char error[DRMAA_ERROR_STRING_BUFFER];
   drmaa_job_template_t *jt = *attr;
   const char *args[2] = {arg, NULL};
   if (drmaa_set_attribute (jt, DRMAA_REMOTE_COMMAND, threadname,
      error, DRMAA_ERROR_STRING_BUFFER) != DRMAA_ERRNO_SUCCESS)
   return ELIBACC;
   if (drmaa_set_vector_attribute (jt, DRMAA_V_ARGV, args, error,
      DRMAA_ERROR_STRING_BUFFER) != DRMAA_ERRNO_SUCCESS)
   return ELIBACC;
   if (drmaa_run_job (thread, DRMAA_JOBNAME_BUFFER, jt, error,
      DRMAA_ERROR_STRING_BUFFER) != DRMAA_ERRNO_SUCCESS)
   return EAGAIN;
   return 0;
}
int dthread_join (dthread_t thread, void **returnvalue) {
   char error[DRMAA_ERROR_STRING_BUFFER];
   char jobid[DRMAA_JOBNAME_BUFFER];
   int status, exited;
   int *exit;
   drmaa_attr_values_t *rusage = NULL;
   char usage[DRMAA_ERROR_STRING_BUFFER];
   if (drmaa_wait (thread, jobid, DRMAA_JOBNAME_BUFFER,
             &status, DRMAA_TIMEOUT_WAIT_FOREVER, &rusage,
             error, DRMAA_ERROR_STRING_BUFFER) != DRMAA_ERRNO_SUCCESS)
   return ELIBACC;
   drmaa_wifexited (&exited, status, NULL, 0);
   if (exited != 1)
      return EINVAL;
   exit = (int *)malloc (sizeof (int));
   drmaa_wexitstatus (exit, status, NULL, 0);
   *returnvalue = exit;
   drmaa_release_attr_values (rusage);
   return 0;
}