home
***
CD-ROM
|
disk
|
FTP
|
other
***
search
/
The World of Computer Software
/
World_Of_Computer_Software-02-385-Vol-1of3.iso
/
c
/
condor40.zip
/
CONDOR
/
src
/
util_lib
/
job_queue.c
< prev
next >
Wrap
C/C++ Source or Header
|
1989-05-15
|
13KB
|
602 lines
/*
** Copyright 1986, 1987, 1988, 1989 University of Wisconsin
**
** Permission to use, copy, modify, and distribute this software and its
** documentation for any purpose and without fee is hereby granted,
** provided that the above copyright notice appear in all copies and that
** both that copyright notice and this permission notice appear in
** supporting documentation, and that the name of the University of
** Wisconsin not be used in advertising or publicity pertaining to
** distribution of the software without specific, written prior
** permission. The University of Wisconsin makes no representations about
** the suitability of this software for any purpose. It is provided "as
** is" without express or implied warranty.
**
** THE UNIVERSITY OF WISCONSIN DISCLAIMS ALL WARRANTIES WITH REGARD TO
** THIS SOFTWARE, INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY AND
** FITNESS. IN NO EVENT SHALL THE UNIVERSITY OF WISCONSIN BE LIABLE FOR
** ANY SPECIAL, INDIRECT OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
** WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
** ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
** OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
**
** Authors: Allan Bricker and Michael J. Litzkow,
** University of Wisconsin, Computer Sciences Dept.
**
*/
/**********************************************************************
** These routines provide access to the CONDOR job queue. The queue consists
** of clusters of related processes. All processes in a cluster share the
** same executable, (and thus the same initial checkpoint), and are submitted
** as a group via a single command file. Individual processes are identified
** by a cluster_id/proc_id tuple. Cluster id's are unique for all time,
** and process id's are 0 - n where there are n processes in the cluster.
** Individual processes are not removed from a cluster, they are just marked
** as "completed", "deleted", "running", "idle", etc. Clusters are removed
** when all processes in the cluster are completed, but the cluster id's are
** not re-used.
**
** Operations:
**
** Q = OpenJobQueue( pathname, flags, mode )
** char *pathname;
** DBM *Q;
**
** CloseJobQueue( Q )
** DBM *Q;
**
** LockJobQueue( Q, operation )
** DBM *Q;
** int operation;
**
** ClusterId = CreateCluster( Q )
** DBM *Q;
** int ClusterId;
**
** StoreProc( Q, Proc )
** DBM *Q;
** PROC *Proc;
**
** Proc = FetchProc( Q, id )
** DBM *Q;
** PROC_ID *id;
** PROC *Proc;
**
** ScanJobQueue( Q, func )
** DBM *Q;
** int (*func)();
**
** ScanCluster( Q, cluster, func )
** DBM *Q;
** int cluster;
** int (*func)();
**
** TerminateCluster( Q, cluster, status )
** DBM *Q;
** int cluster;
** int status;
**
** TerminateProc( Q, pid, status )
** DBM *Q;
** PROC_ID *pid;
** int status;
**
** The low level database operations are accomplished by ndbm(3) using
** XDR to pack/unpack the Proc structs to/from contiguous areas of memory.
** The locking is accomplished by flock(3).
**************************************************************************/
#include <stdio.h>
#include <errno.h>
#include <sys/file.h>
#include <rpc/types.h>
#include <rpc/xdr.h>
#include <sys/param.h>
#include <sys/time.h>
#include <sys/resource.h>
#include "proc.h"
#include "debug.h"
#include "except.h"
#include "trace.h"
#include "clib.h"
#ifdef NDBM
#include <ndbm.h>
#else NDBM
#include "ndbm_fake.h"
#endif NDBM
static char *_FileName_ = __FILE__; /* Used by EXCEPT (see except.h) */
char *Spool;
char *malloc();
#define LIST_INCR 5
static PROC_ID ClusterId = { 0, 0 };
static datum ClusterKey = { (char *)&ClusterId, sizeof(ClusterId) };
/*
** Prepare the local job queue for reading/writing with these routines.
*/
DBM *
OpenJobQueue( path, flags, mode )
char *path;
{
return dbm_open(path,flags,mode);
}
/*
** Close the job queue.
*/
CloseJobQueue( Q )
DBM *Q;
{
dbm_close( Q );
}
/*
** Put a reader's/writer's lock on an open job queue.
*/
LockJobQueue( Q, op )
DBM *Q;
int op;
{
while( flock(Q->dbm_pagf,op) < 0 ) {
if( errno != EINTR ) {
EXCEPT("flock(%d,0%o) -- Q = 0x%x", Q->dbm_pagf, op, Q );
}
}
}
/*
** Allocate a new cluster id, and store it in the list of active
** cluster id's.
*/
CreateCluster( Q )
DBM *Q;
{
CLUSTER_LIST *list, *fetch_cluster_list(), *add_new_cluster();
int answer;
list = fetch_cluster_list( Q );
list = add_new_cluster( list );
answer = list->next_id - 1;
store_cluster_list( Q, list );
return answer;
}
/*
** Store a PROC in the database.
*/
StoreProc( Q, proc )
DBM *Q;
PROC *proc;
{
char buf[10 * MAXPATHLEN];
XDR xdr, *xdrs = &xdr;
datum key;
datum data;
/* Use XDR to pack the structure into a contiguous buffer */
xdrmem_create( xdrs, buf, sizeof(buf), XDR_ENCODE );
ASSERT( xdr_proc(xdrs,proc) );
/* Store it with dbm */
key.dptr = (char *)&proc->id;
key.dsize = sizeof( PROC_ID );
data.dptr = buf;
data.dsize = xdrs->x_private - xdrs->x_base;
return dbm_store(Q,key,data,DBM_REPLACE);
}
/*
** We are given a process structure with the id filled in. We look up
** the id in the database, and fill in the rest of the structure.
*/
FetchProc( Q, proc )
DBM *Q;
PROC *proc;
{
XDR xdr, *xdrs = &xdr;
datum key;
datum data;
/* Fetch the database entry */
key.dptr = (char *)&proc->id;
key.dsize = sizeof( PROC_ID );
data = dbm_fetch( Q, key );
if( !data.dptr ) {
return -1;
}
/* Use XDR to unpack the structure from the contiguous buffer */
xdrmem_create( xdrs, data.dptr, data.dsize, XDR_DECODE );
bzero( (char *)proc, sizeof(PROC) ); /* let XDR allocate all the space */
if( !xdr_proc(xdrs,proc) ) {
EXCEPT( "Can't unpack proc %d.%d", proc->id.cluster, proc->id.proc );
}
return 0;
}
/*
** Apply the given function to every process in the queue.
*/
ScanJobQueue( Q, func )
DBM *Q;
int (*func)();
{
int i;
CLUSTER_LIST *list;
list = fetch_cluster_list( Q );
for( i=0; i<list->n_ids; i++ ) {
ScanCluster( Q, list->id[i], func );
}
free( (char *)list );
}
/*
** Apply the function to every process in the specified cluster.
*/
ScanCluster( Q, cluster, func )
DBM *Q;
int cluster;
int (*func)();
{
PROC proc;
int proc_id;
proc.id.cluster = cluster;
for( proc_id = 0; ;proc_id++) {
proc.id.proc = proc_id;
if( FetchProc(Q,&proc) < 0 ) {
return;
}
func( &proc );
xdr_free_proc( &proc );
}
}
static DBM *CurJobQ;
static int CurStatus;
static XDR *CurHistory;
/*
** Terminate every process in the cluster with the given status. This
** includes storing the proc structs in the history file and removing
** the cluster from the active cluster list.
*/
TerminateCluster( Q, cluster, status )
DBM *Q;
int cluster;
int status;
{
int terminate();
XDR xdr, *OpenHistory();
int fd;
char ickpt_name[MAXPATHLEN];
CurJobQ = Q;
CurStatus = status;
CurHistory = OpenHistory( param("HISTORY"), &xdr, &fd );
(void)LockHistory( CurHistory, WRITER );
ScanCluster( Q, cluster, terminate );
CloseHistory( CurHistory );
delete_cluster( Q, cluster );
(void)sprintf( ickpt_name, "%s/job%06d.ickpt", Spool, cluster );
(void)unlink( ickpt_name );
CurJobQ = NULL;
}
static int EmptyCluster;
/*
** Terminate a particular process. We also check to see if this is the last
** active process in the cluster, and if so, we terminate the cluster.
*/
TerminateProc( Q, pid, status )
DBM *Q;
PROC_ID *pid;
int status;
{
PROC proc;
int check_empty_cluster();
char ckpt_name[MAXPATHLEN];
proc.id.cluster = pid->cluster;
proc.id.proc = pid->proc;
if( FetchProc(Q,&proc) != 0 ) {
printf( "Process %d.%d doesn't exist\n",
proc.id.cluster, proc.id.proc );
return -1;
}
proc.status = status;
if( StoreProc(Q,&proc) != 0 ) {
EXCEPT( "Can't store process struct %d.%d\n",
proc.id.cluster, proc.id.proc );
}
(void)sprintf( ckpt_name, "%s/job%06d.ckpt.%d",
Spool, proc.id.cluster, proc.id.proc );
(void)unlink( ckpt_name );
xdr_free_proc( &proc );
EmptyCluster = TRUE;
ScanCluster( Q, pid->cluster, check_empty_cluster );
if( EmptyCluster ) {
TerminateCluster( Q, pid->cluster, -1 );
}
return 0;
}
/*
** Store the given cluster list in the database. The list we are given
** is always malloc'd, so we free it here. ROUTINES WHICH CALL
** THIS ONE MUST NOT MAKE ANY USE OF THE LIST AFTERWARD!
*/
static
store_cluster_list( Q, list )
DBM *Q;
CLUSTER_LIST *list;
{
datum data;
data.dptr = (char *)list;
data.dsize = sizeof(CLUSTER_LIST) + (list->array_len - 1) * sizeof(int);
if( dbm_store(Q,ClusterKey,data,DBM_REPLACE) < 0 ) {
EXCEPT( "dbm_store" );
}
free( (char *)list );
}
/*
** Fetch a cluster list from the database. If none is there, create a
** new and empty one. In either case the list we give back is always
** malloc'd, and should be free'd later.
*/
CLUSTER_LIST *
fetch_cluster_list( Q )
DBM *Q;
{
datum data;
CLUSTER_LIST *answer, *make_cluster_list(), *copy_cluster_list();
data = dbm_fetch( Q, ClusterKey );
if( data.dptr ) {
answer = copy_cluster_list( (CLUSTER_LIST *)data.dptr );
} else {
answer = make_cluster_list( LIST_INCR );
}
return answer;
}
/*
** Increase the size of a cluster list by (incr). The list we are given
** is always malloc'd, so we can use realloc for this.
*/
static
CLUSTER_LIST *
grow_cluster_list( list, incr )
CLUSTER_LIST *list;
int incr;
{
list->array_len += incr;
return (CLUSTER_LIST *)realloc( (char *)list,
(unsigned)(sizeof(CLUSTER_LIST) + (list->array_len-1) * sizeof(int)) );
}
/*
** Return a new and empty cluster list in a malloc'd area.
*/
static
CLUSTER_LIST *
make_cluster_list( len )
int len;
{
CLUSTER_LIST *answer;
answer = (CLUSTER_LIST *)malloc( (unsigned)(sizeof(CLUSTER_LIST) +
(len - 1) * sizeof(int)) );
if( answer == NULL ) {
return NULL;
}
answer->n_ids = 0;
answer->array_len = len;
answer->next_id = 1;
return answer;
}
static
CLUSTER_LIST *
add_new_cluster( list )
CLUSTER_LIST *list;
{
CLUSTER_LIST *grow_cluster_list();
if( list->n_ids == list->array_len ) {
list = grow_cluster_list( list, LIST_INCR );
}
list->id[ list->n_ids++ ] = list->next_id++;
return list;
}
/*
** When we fetch a cluster list from the database, we get a pointer into
** the database's private buffer. Here we copy it into our own malloc'd
** space so we don't have to worry about the database overwriting it
** later.
*/
static
CLUSTER_LIST *
copy_cluster_list( list )
CLUSTER_LIST *list;
{
CLUSTER_LIST *answer;
/* Malloc space for the new one */
answer = (CLUSTER_LIST *)malloc( (unsigned)(sizeof(CLUSTER_LIST) +
(list->array_len - 1) * sizeof(int)) );
/* Copy in the old one */
bcopy( (char *)list, (char *)answer,
(unsigned)(sizeof(CLUSTER_LIST) + (list->array_len-1) * sizeof(int)) );
return answer;
}
static
terminate( p )
PROC *p;
{
datum key;
if( CurStatus >= 0 ) {
p->status = CurStatus;
}
if( p->status != SUBMISSION_ERR ) {
AppendHistory( CurHistory, p );
}
key.dptr = (char *)&p->id;
key.dsize = sizeof( PROC_ID );
if( dbm_delete(CurJobQ,key) != 0 ) {
fprintf( stderr, "Can't delete process %d.%d\n",
p->id.cluster, p->id.proc );
}
}
static
check_empty_cluster( p )
PROC *p;
{
switch( p->status ) {
case UNEXPANDED:
case IDLE:
case RUNNING:
EmptyCluster = FALSE;
break;
case REMOVED:
case COMPLETED:
default:
break;
}
}
/*
** Delete a cluster from the cluster list. This DOES NOT delete individual
** processes within the structure. That should be done by a higher level
** routine which also calls this one.
*/
static
delete_cluster( Q, cluster )
DBM *Q;
int cluster;
{
CLUSTER_LIST *list, *fetch_cluster_list();
int src, dst;
list = fetch_cluster_list( Q );
for( src = 0, dst = 0; src < list->n_ids; ) {
if( list->id[src] != cluster ) {
list->id[dst++] = list->id[src++];
} else {
src++;
}
}
list->n_ids = dst;
store_cluster_list( Q, list );
}
/*
** The current version of ULTRIX doesn't have ndbm. We'll build a subset
** of that interface here with some loss of functionality. We won't be
** able to handle multiple open databases, so we'll sacrifice the history
** keeping function for ULTRIX installations. If ULTRIX ever comes
** up with an ndbm type of database library, this can all go away.
*/
#ifndef NDBM
DBM *
dbm_open( file, flags, mode )
char *file;
int flags;
int mode;
{
static DBM dbm;
char name[MAXPATHLEN];
(void)sprintf( name, "%s.dir", file );
if( (dbm.dbm_dirf = open(name,flags,mode)) < 0 ) {
return NULL;
}
(void)sprintf( name, "%s.pag", file );
if( (dbm.dbm_pagf = open(name,flags,mode)) < 0 ) {
(void)close(dbm.dbm_dirf);
return NULL;
}
if( dbminit(file) < 0 ) {
return NULL;
}
return &dbm;
}
dbm_close( Q )
DBM *Q;
{
(void)close( Q->dbm_pagf );
(void)close( Q->dbm_dirf );
dbmclose();
}
/* ARGSUSED */
dbm_store(Q,key,data,flags)
DBM *Q;
datum key;
datum data;
int flags;
{
return store( key, data );
}
/* ARGSUSED */
datum
dbm_fetch( Q, key )
DBM *Q;
datum key;
{
return fetch( key );
}
/* ARGSUSED */
dbm_delete( Q, key )
DBM *Q;
datum key;
{
return delete( key );
}
#endif NOT NDBM