Index: /tags/elixir-1-5/Ohana/src/elixir/Makefile =================================================================== --- /tags/elixir-1-5/Ohana/src/elixir/Makefile (revision 8558) +++ /tags/elixir-1-5/Ohana/src/elixir/Makefile (revision 8558) @@ -0,0 +1,65 @@ +include ../../Configure +HOME = $(ROOT)/src/elixir +CONFIG = $(ROOT)/config +PROGRAM = elixir + +BIN = $(HOME)/bin +INC = $(HOME)/include +SRC = $(HOME)/src +MAN = $(HOME)/doc +DESTBIN = $(LBIN) +DESTLIB = $(LLIB) +DESTINC = $(LINC) +DESTMAN = $(LMAN) + +# +INCS = -I$(INC) -I$(LINC) -I$(XINC) +LIBS = -L$(LLIB) -lFITS -lohana -lm -lsocket -lnsl +CFLAGS = -o $*.$(ARCH).o $(INCS) +CCFLAGS = $(INCS) $(LIBS) + +ELIXIR = \ +$(SRC)/elixir.$(ARCH).o $(SRC)/ConfigInit.$(ARCH).o \ +$(SRC)/MachineOps.$(ARCH).o $(SRC)/FifoOps.$(ARCH).o \ +$(SRC)/QueueOps.$(ARCH).o $(SRC)/CheckCluster.$(ARCH).o \ +$(SRC)/CheckDepend.$(ARCH).o $(SRC)/CheckEndingState.$(ARCH).o \ +$(SRC)/CheckMachineStatus.$(ARCH).o $(SRC)/CheckProcess.$(ARCH).o \ +$(SRC)/InitProcess.$(ARCH).o $(SRC)/StartMachine.$(ARCH).o \ +$(SRC)/SockScan.$(ARCH).o $(SRC)/LoadPending.$(ARCH).o \ +$(SRC)/Photcodes.$(ARCH).o $(SRC)/rconnect.$(ARCH).o \ +$(SRC)/DefineProcesses.$(ARCH).o $(SRC)/ProcessOps.$(ARCH).o \ +$(SRC)/DumpStatus.$(ARCH).o $(SRC)/CheckMessages.$(ARCH).o \ +$(SRC)/MsgOps.$(ARCH).o $(SRC)/LogOpen.$(ARCH).o \ +$(SRC)/ConfigPID.$(ARCH).o $(SRC)/RemoteOps.$(ARCH).o + +default: elixir + +$(ELIXIR): $(INC)/controller.h + +# dependancy rules for binary code ########################## +$(PROGRAM): $(BIN)/$(PROGRAM).$(ARCH) + +$(BIN)/$(PROGRAM).$(ARCH): $(ELIXIR) + @if [ ! -d $(BIN) ]; then mkdir -p $(BIN); fi + $(CC) $(ELIXIR) -o $(BIN)/$(PROGRAM).$(ARCH) $(CCFLAGS) + +install: $(DESTBIN)/$(PROGRAM) + +$(DESTBIN)/$(PROGRAM): $(BIN)/$(PROGRAM).$(ARCH) + @if [ ! -d $(DESTBIN) ]; then mkdir -p $(DESTBIN); fi + rm -f $(DESTBIN)/$(PROGRAM) + cp $(BIN)/$(PROGRAM).$(ARCH) $(DESTBIN)/$(PROGRAM) + +# utilities ################################################# +clean: + rm -f $(BIN)/*.$(ARCH) + rm -f $(SRC)/*.o + rm -f `find . -name "*~"` + rm -f `find . -name "#*"` + +.SUFFIXES: .$(ARCH).o + +.c.$(ARCH).o: + $(CC) $(CFLAGS) -c $< + + Index: /tags/elixir-1-5/Ohana/src/elixir/bin/.cvsignore =================================================================== --- /tags/elixir-1-5/Ohana/src/elixir/bin/.cvsignore (revision 8558) +++ /tags/elixir-1-5/Ohana/src/elixir/bin/.cvsignore (revision 8558) @@ -0,0 +1,2 @@ +*.linux *.lin64 *.sol *.sun *.sid *.hp *.irix +*.linrh Index: /tags/elixir-1-5/Ohana/src/elixir/doc/ChangeLog.txt =================================================================== --- /tags/elixir-1-5/Ohana/src/elixir/doc/ChangeLog.txt (revision 8558) +++ /tags/elixir-1-5/Ohana/src/elixir/doc/ChangeLog.txt (revision 8558) @@ -0,0 +1,24 @@ + +- elixir-1-5 : 2006.08.23 + * converted to gfits APIs (forces libfits 1.6) + +elixir-1-4: + changed printing formats for pointers to %p + using updated version of mkdirhier (libohana 1.7) + +elixir-1-3: + libohana / libfits were modified, forcing the + addition of fitsio.h to controller.h + +elixir-1-2: + - changing setlockfile2 to fsetlockfile + (dropping use of fd version of APIs) + elixir should eventually be superceeded by sched/pcontrol, + but perhaps not at CFHT. + +elixir-1-1: + - minor cleanup of rconnect process + - ignore SIG_CHLD now + +elixir-1-0: + - created tag for first elixir distribution Index: /tags/elixir-1-5/Ohana/src/elixir/doc/README =================================================================== --- /tags/elixir-1-5/Ohana/src/elixir/doc/README (revision 8558) +++ /tags/elixir-1-5/Ohana/src/elixir/doc/README (revision 8558) @@ -0,0 +1,13 @@ + +There are two parts to elixir which are a bit dangerous (and limit the +possibilities). + +First, the number of arguments which describe an object is limited to +7. This is limited in LoadPending.c and is due to the dumb way I +parse these input lines. + +Second, in a similar way, the number of arguments to a command line +argument (the 'process.arg' entries) is limited to 5, again for about +the same reason. this is in 'ParseLine' in ProcessOps.c + + Index: /tags/elixir-1-5/Ohana/src/elixir/doc/notes.txt =================================================================== --- /tags/elixir-1-5/Ohana/src/elixir/doc/notes.txt (revision 8558) +++ /tags/elixir-1-5/Ohana/src/elixir/doc/notes.txt (revision 8558) @@ -0,0 +1,40 @@ + + +typedef struct { + char *buffer; + int Nalloc; + int Nmaxread; + int Nextra; + int Nlast; + int Nbuffer; +} Fifo; + +typedef struct { + int argc; char **argv; /* a list of words that define this object */ + struct timeval start, accum, timer; + int status; + char *logfile; + char *lastproc; +} Object; + +typedef struct { + Object **object; + int Nobject; + int NOBJECT; +} Queue; + +typedef struct { + char *hostname; + int rsock, wsock; + int status; /* idle, busy, etc... */ + struct timeval start, accum, timer; + Fifo fifo; + int code; + Object *object; +} Machine; + + +currently, the transport is /usr/bin/rsh, defined in InitMachines.c + +the shell on the remote machines is /bin/tcsh, defined by rconnect.c + Index: /tags/elixir-1-5/Ohana/src/elixir/include/controller.h =================================================================== --- /tags/elixir-1-5/Ohana/src/elixir/include/controller.h (revision 8558) +++ /tags/elixir-1-5/Ohana/src/elixir/include/controller.h (revision 8558) @@ -0,0 +1,173 @@ +# include +# include +# include +# include + +# define CR 0x0D +# define LF 0x0A + +# define IDLE 0x00 +# define BUSY 0x01 +# define DOWN 0x02 +# define DONE 0x04 +# define MESSAGE 0x08 +# define SUCCESS 0x10 +# define FAILURE 0x20 +# define JOBDONE 0x40 +# define WAITING 0x80 +# define ERROR 0x100 +# define CRASH 0x200 +# define TIMEOUT 0x400 + +# define DTIME(A,B) ((A.tv_sec - B.tv_sec) + 1e-6*(A.tv_usec - B.tv_usec)) + +/* represents a socketed connection to a remote host + which may or may not be currently running a process */ + +typedef struct { + char *buffer; + int Nalloc; + int Nmaxread; + int Nextra; + int Nlast; + int Nbuffer; +} Fifo; + +typedef struct { + int argc; char **argv; /* a list of words that define this object */ + struct timeval start, accum, timer; + int status; + char *logfile; + char *lastproc; +} Object; + +typedef struct { + Object **object; + int Nobject; + int NOBJECT; +} Queue; + +typedef struct { + char *hostname; + int rsock, wsock, pid; + int status; /* idle, busy, etc... */ + struct timeval start, accum, timer, quiet; + Fifo fifo; + int code; + Object *object; +} Machine; + +typedef struct { + Machine **machine; + int Nmachine; + int NMACHINE; +} Cluster; + +typedef struct { + char *name; + Queue *pending; + Queue *success; + Queue *failure; + Cluster *cluster; + int argc; + char **argv; +} Process; + +/* we have one ProcessTimer for each existing machine + each ProcessTimer has one element for each process */ +typedef struct { + struct timeval *timer; + double *timesum; + int *Njobs; + int *active; + Machine *machine; + Process **process; +} ProcessTimers; + +int VERBOSE; +char CONNECT[128]; + +/* prototypes */ + +Process **DefineProcesses (Process *global, int *nprocess, char *config); +Process **GetProcessInfo (Process **gb, int *np, int *no); +char *BaseFilename (char *file); +char *BuildCode (char *line); +char *BuildName (char *line); +char *ConfigInit (int *argc, char **argv); +Process *ConfigProcess (char *config, char *procname); +char *ExpandEntry (char *entry, int argc, char **argv); +Machine *GetMachine (Cluster *cluster); +Object *GetObject (Queue *queue); +char *GetPhotcode (char *file); +char *GetPhotcodeExt (char *file); +char *GetPhotcodeMef (char *file); +Machine *GrabMachine (); +Cluster *InitCluster (); +Process *InitProcess (char *name, Queue *pending, Queue *failure, int (*mkargs)()); +Queue *InitQueue (); +FILE *LogOpen (char *filename); +char *PathFilename (char *file); +char *RootFilename (char *file); +int CheckCluster (Cluster *cluster, Queue *success, Queue *failure, Queue *pending); +int CheckDepend (Object *object, int argc, char **argv, int *argd); +int CheckEndingState (Process *global, int Nobjects, int Dynamic); +int CheckMachineStatus (Machine *machine); +int CheckMessages (); +int CheckProcess (Process *process); +void CloseMachine (Machine *machine); +void ConfigPID (char *PIDFile); +int ConnectMachine (Machine *machine); +void DownMachine (Machine *machine); +void DumpFinished (Queue *queue, int Nstart, char *filename); +void DumpMachineStatus (FILE *f); +int DumpProcessTimes (char *filename); +int DumpStatus (char *filename); +void ElixirStop (); +int FlushFifo (Fifo *fifo); +void FreeArgs (int argc, char **argv, int *argd); +void FreeFifo (Fifo *fifo); +int GetDynamicState (); +double GetTimeout (); +void HaltElixir (char *pidfile); +int HalttoRestart (char *pidfile); +void IdleMachine (Machine *machine); +int InitFifo (Fifo *fifo, int Nalloc, int Nextra); +void InitMachines (char *config); +int InitMsgFile (char *file); +void InitProcessTimers (Process **process, int Nprocess); +void KillElixir (char *pidfile); +int LoadPID (char *file, pid_t *pid, char *username, char *machine); +int LoadPending (Process *global, char *inlist, int *state, int *dynamic); +int MakeArgs (Process *process, Object *object, int *cargc, char ***cargv, int **cargd); +void ParseLine (char *testline, int argc, char **argv, int *depend, char **outline); +void PushMachine (Machine *machine, Cluster *cluster); +void PushObject (Queue *queue, Object *object); +void PutMachine (Machine *machine, Cluster *cluster); +void PutObject (Queue *queue, Object *object); +int ReadMsg (char *fifo, char **message); +int ReadtoFifo (Fifo *fifo, int sock); +void RegisterTimeout (double value); +void RemovePID (); +void Restart (char **argv); +void RestartMachines (); +void SIG_DIE (int sig); +void SIG_MESSAGE (int sig); +void SIG_PIPE (int sig); +void SIG_RELOAD (int sig); +void SIG_STOP (int sig); +int SetExitTimer (); +void SetMessageFile (char *filename); +int ShiftFifo (Fifo *fifo); +void Shutdown (int status); +int SockScan (char *string, Fifo *fifo, int sock); +int StartMachine (Machine *machine, Object *object, int argc, char **argv); +void StartProcessTimer (Process *process, Machine *machine); +void StatusElixir (char *pidfile, char *msgfile); +void StopProcessTimer (Machine *machine); +int TestMachine (Machine *machine); +int WaitMsg (char *fifo, char **message, double maxdelay); +int WriteMsg (char *fifo, char *message); +int memstr (char *m1, char *m2, int n); +int rconnect (char *hostname, char *command, int *rsock, int *wsock); +void GetConfig (char *config, char *field, char *format, int N, void *ptr); Index: /tags/elixir-1-5/Ohana/src/elixir/src/CheckCluster.c =================================================================== --- /tags/elixir-1-5/Ohana/src/elixir/src/CheckCluster.c (revision 8558) +++ /tags/elixir-1-5/Ohana/src/elixir/src/CheckCluster.c (revision 8558) @@ -0,0 +1,113 @@ +# include "controller.h" + +int CheckCluster (Cluster *cluster, Queue *success, Queue *failure, Queue *pending) { + + int i, j, status; + Machine *machine, **tmachine; + Object *object; + FILE *logfile; + + /* check current status of all machines in cluster, idle those finished */ + for (i = 0; i < cluster[0].Nmachine; i++) { + + machine = cluster[0].machine[i]; + status = CheckMachineStatus (machine); + + object = machine[0].object; + logfile = LogOpen (object[0].logfile); + + if (status & MESSAGE) { + fprintf (logfile, "%s @ %s:", object[0].argv[0], machine[0].hostname); + fwrite (&machine[0].fifo.buffer[machine[0].fifo.Nlast], 1, machine[0].fifo.Nbuffer - machine[0].fifo.Nlast, logfile); + } + + if (status & DOWN) { + fprintf (stderr, "%s @ %s is down\n", object[0].argv[0], machine[0].hostname); + fprintf (logfile, "%s @ %s is down\n", object[0].argv[0], machine[0].hostname); + object[0].timer.tv_sec = 0; /* reset timer */ + StopProcessTimer (machine); + /* PutObject (failure, object); */ + /* if machine crashes, retry object later */ + PutObject (pending, object); + DownMachine (machine); + cluster[0].machine[i] = (Machine *) NULL; + goto escape; + } + if (status & DONE) { + object[0].timer.tv_sec = 0; /* reset timer */ + StopProcessTimer (machine); + fprintf (logfile, "%s @ %s is done\n", object[0].argv[0], machine[0].hostname); + if (status & SUCCESS) { + PutObject (success, object); + } + if (status & FAILURE) { + PutObject (failure, object); + } + IdleMachine (machine); + cluster[0].machine[i] = (Machine *) NULL; + goto escape; + } + if (status & CRASH) { + object[0].timer.tv_sec = 0; /* reset timer */ + StopProcessTimer (machine); + fprintf (logfile, "%s @ %s had process crash\n", object[0].argv[0], machine[0].hostname); + PutObject (failure, object); + IdleMachine (machine); + cluster[0].machine[i] = (Machine *) NULL; + goto escape; + } + if (status & ERROR) { + object[0].timer.tv_sec = 0; /* reset timer */ + StopProcessTimer (machine); + fprintf (logfile, "%s @ %s has an odd status\n", object[0].argv[0], machine[0].hostname); + PutObject (failure, object); + DownMachine (machine); + cluster[0].machine[i] = (Machine *) NULL; + goto escape; + } + if (status & TIMEOUT) { + object[0].timer.tv_sec = 0; /* reset timer */ + /* don't include TIMEOUT in timer stats StopProcessTimer (machine); */ + fprintf (logfile, "%s @ %s timed out, retrying\n", object[0].argv[0], machine[0].hostname); + PutObject (pending, object); + IdleMachine (machine); + cluster[0].machine[i] = (Machine *) NULL; + goto escape; + } + + escape: + if (logfile != stderr) fclose (logfile); + } + + /* remove idle machines from this cluster's list */ + ALLOCATE (tmachine, Machine *, cluster[0].NMACHINE); + for (j = i = 0; i < cluster[0].Nmachine; i++) { + if (cluster[0].machine[i] != (Machine *) NULL) { + tmachine[j] = cluster[0].machine[i]; + j++; + } + } + free (cluster[0].machine); + cluster[0].machine = tmachine; + cluster[0].Nmachine = j; + + return (TRUE); +} + + +/* + + possible ending states: + + DONE & SUCCESS = process finished normally & succeeded + DONE & FAILURE = process finished normally & failed + + CRASH = process crashed + DOWN = machine crashed + IDLE = no process (should never show up here!) + ERROR = unexpected state! + + whenever an object is done, in any of the possible states, the + arg dependency timer must be reset + +*/ Index: /tags/elixir-1-5/Ohana/src/elixir/src/CheckDepend.c =================================================================== --- /tags/elixir-1-5/Ohana/src/elixir/src/CheckDepend.c (revision 8558) +++ /tags/elixir-1-5/Ohana/src/elixir/src/CheckDepend.c (revision 8558) @@ -0,0 +1,52 @@ +# include "controller.h" + +int CheckDepend (Object *object, int argc, char **argv, int *argd) { + + int i; + struct stat filestat; + uid_t uid; + gid_t gid; + int status; + struct timeval now; + FILE *logfile; + double dtime; + + uid = getuid(); + gid = getgid(); + + for (i = 0; i < argc; i++) { + if (argd[i]) { + /* check permission to read file */ + status = stat (argv[i], &filestat); + + /* continue if file exists and is accessible */ + if (!status) { + if ((uid == filestat.st_uid) && (filestat.st_mode & S_IRUSR)) continue; + if ((gid == filestat.st_gid) && (filestat.st_mode & S_IRGRP)) continue; + if (filestat.st_mode & S_IROTH) continue; + } + + /* if the file doesn't exist or is inaccessible, check the timer on this object */ + if (object[0].timer.tv_sec == 0) { + gettimeofday (&object[0].timer, (void *) NULL); + return (0); + } + + gettimeofday (&now, (void *) NULL); + dtime = DTIME (now, object[0].timer); + if (dtime > argd[i]) { + logfile = LogOpen (object[0].logfile); + fprintf (logfile, "timeout on %s: %f > %d\n", argv[i], dtime, argd[i]); + if (logfile != stderr) fclose (logfile); + object[0].status &= TIMEOUT; + return (2); + } else { + return (0); + } + } + } + + return (TRUE); + +} + Index: /tags/elixir-1-5/Ohana/src/elixir/src/CheckEndingState.c =================================================================== --- /tags/elixir-1-5/Ohana/src/elixir/src/CheckEndingState.c (revision 8558) +++ /tags/elixir-1-5/Ohana/src/elixir/src/CheckEndingState.c (revision 8558) @@ -0,0 +1,80 @@ +# include "controller.h" + +static int Nsuccess = 0; +static int Nfailure = 0; +static double timeout; +struct timeval start; +static int Reload = FALSE; + +int CheckEndingState (Process *global, int Nobjects, int Dynamic) { + + int Ndone; + struct timeval now; + double dtime; + + /* check the success and failure queues for newly added objects */ + + if (global[0].success[0].Nobject > Nsuccess) { + DumpFinished (global[0].success, Nsuccess, global[0].argv[0]); + Nsuccess = global[0].success[0].Nobject; + } + if (global[0].failure[0].Nobject > Nfailure) { + DumpFinished (global[0].failure, Nfailure, global[0].argv[1]); + Nfailure = global[0].failure[0].Nobject; + } + + if (Dynamic) return (TRUE); + + Ndone = 0; + Ndone += global[0].success[0].Nobject; + Ndone += global[0].failure[0].Nobject; + + if (Reload) { + gettimeofday (&now, (void *) NULL); + dtime = DTIME (now, start); + if (dtime > timeout) return (FALSE); + } + + if (Ndone == Nobjects) + return (FALSE); + else + return (TRUE); +} + +void DumpFinished (Queue *queue, int Nstart, char *filename) { + + int i, j; + FILE *f; + + f = fopen (filename, "a"); + if (f == (FILE *) NULL) { + fprintf (stderr, "can't output status of finished objects to %s\n", filename); + f = stderr; + } + + for (i = Nstart; i < queue[0].Nobject; i++) { + + for (j = 0; j < queue[0].object[i][0].argc; j++) { + fprintf (f, "%s ", queue[0].object[i][0].argv[j]); + } + fprintf (f, "%08x %s\n", queue[0].object[i][0].status, queue[0].object[i][0].lastproc); + } + + if (f != stderr) fclose (f); + +} + +int SetExitTimer () { + + timeout = GetTimeout (); + gettimeofday (&start, (void *) NULL); + Reload = TRUE; + return (TRUE); +} + +/* + + global.success = global[0].argv[0] + global.failure = global[0].argv[1] + +*/ Index: /tags/elixir-1-5/Ohana/src/elixir/src/CheckMachineStatus.c =================================================================== --- /tags/elixir-1-5/Ohana/src/elixir/src/CheckMachineStatus.c (revision 8558) +++ /tags/elixir-1-5/Ohana/src/elixir/src/CheckMachineStatus.c (revision 8558) @@ -0,0 +1,131 @@ +# include "controller.h" +# define MAXTIME 120 + +static double timeout; + +void RegisterTimeout (double value) { + timeout = value; +} + +double GetTimeout () { + return (timeout); +} + +int CheckMachineStatus (Machine *machine) { + + unsigned int status; + int Nread; + struct timeval now; + FILE *logfile; + float dtime; + + status = machine[0].status; + + if (status == IDLE) return (status); + if (status == DONE) return (status); + if (status == DOWN) return (status); + + status &= ~MESSAGE; /* clear MESSAGE status */ + + /* read from socket into fifo */ + ShiftFifo (&machine[0].fifo); + + Nread = ReadtoFifo (&machine[0].fifo, machine[0].rsock); + + /* evaluate data in message */ + switch (Nread) { + case 0: + status = DOWN; + break; + case -1: + status |= BUSY; + gettimeofday (&now, (void *) NULL); + dtime = DTIME (now, machine[0].quiet); + if (dtime > timeout) { + logfile = LogOpen (machine[0].object[0].logfile); + fprintf (logfile, "%s: process is taking too long, giving up\n", machine[0].hostname); + if (logfile != stderr) fclose (logfile); + fprintf (stderr, "process on %s hung? killing and retrying...\n", machine[0].hostname); + /* interrupt process which is running */ + kill (machine[0].pid, SIGINT); + status = TIMEOUT; + /* status = (JOBDONE | FAILURE); */ + } + break; + default: + status |= MESSAGE; + gettimeofday (&machine[0].quiet, (void *) NULL); + if (memstr (machine[0].fifo.buffer, "ERROR", machine[0].fifo.Nbuffer)) + status |= FAILURE; + if (memstr (machine[0].fifo.buffer, "SUCCESS", machine[0].fifo.Nbuffer)) + status |= SUCCESS; + if (memstr (machine[0].fifo.buffer, "PROCESS DONE", machine[0].fifo.Nbuffer)) + status |= JOBDONE; + break; + } + + /* evaluate status completion-type status signals */ + switch (status & (JOBDONE | SUCCESS | FAILURE)) { + case FAILURE: + case SUCCESS: + case JOBDONE: + if (status & WAITING) break; + status |= WAITING; + gettimeofday (&machine[0].timer, (void *) NULL); + break; + case (JOBDONE | SUCCESS): + case (JOBDONE | FAILURE): + status |= DONE; + break; + case (FAILURE | SUCCESS): + case (JOBDONE | FAILURE | SUCCESS): + status |= ERROR; + break; + default: + break; + } + + /* check for completion timeout */ + if (status & WAITING) { + gettimeofday (&now, (void *) NULL); + if (DTIME (now, machine[0].timer) > MAXTIME) { + switch (status & (JOBDONE | SUCCESS | FAILURE)) { + case JOBDONE: + status |= CRASH; + break; + case SUCCESS: + case FAILURE: + default: + status |= ERROR; + break; + } + } + } + + machine[0].object[0].status = status; + machine[0].status = status; + + return (status); + +} + + +/* machine status: + + the machine can have several possible statuss, made up of specific bits + in the status variable: + + IDLE 0x00 - no process running on host + DOWN 0x01 - no connection to host + BUSY 0x02 - no message on socket + DONE 0x04 - process completed + + MESSAGE 0x08 - message on socket + FAILURE 0x10 - process reported 'failure' + SUCCESS 0x20 - process reported 'success' + JOBDONE 0x40 - process reported 'jobdone' + WAITING 0x80 - process reported 'jobdone' + + ERROR 0x010 - unexpected return values + CRASH 0x020 - process crashed +*/ Index: /tags/elixir-1-5/Ohana/src/elixir/src/CheckMessages.c =================================================================== --- /tags/elixir-1-5/Ohana/src/elixir/src/CheckMessages.c (revision 8558) +++ /tags/elixir-1-5/Ohana/src/elixir/src/CheckMessages.c (revision 8558) @@ -0,0 +1,68 @@ +# include "controller.h" + +static struct timeval then = {0,0}; +static char *MessageFile = (char *) NULL; + +void SetMessageFile (char *filename) { + MessageFile = filename; +} + +int CheckMessages () { + + int status, Message; + struct timeval now; + char *message, *p, *p2; + char file[256], cmd[64]; + + /* we should only do this check every 200ms or so */ + gettimeofday (&now, (void *) NULL); + if (DTIME (now, then) < 0.2) return (1); + + if (MessageFile == (char *) NULL) { + fprintf (stderr, "not ready for messages\n"); + return (FALSE); + } + + status = WaitMsg (MessageFile, &message, 0.1); + then = now; + if (!status) return (FALSE); + + Message = 0; + + /* loop over all lines in message */ + p = message; + while (strlen (p) > 0) { + p2 = strchr (p, '\n'); + if (p2 == (char *) NULL) { + p2 = p + strlen (p) - 1; + } else { + *p2 = 0; + } + + sscanf (p, "%s %s", cmd, file); + + if (!strcasecmp (cmd, "ALIVE")) { + WriteMsg (file, "BUSY"); + } + if (!strcasecmp (cmd, "TIMES")) { + DumpProcessTimes (file); + } + if (!strcasecmp (cmd, "STATUS")) { + DumpStatus (file); + } + if (!strcasecmp (cmd, "STOP")) { + WriteMsg (file, "Elixir will end when processes are done"); + ElixirStop (); + } + if (!strcasecmp (cmd, "ABORT")) { + WriteMsg (file, "Elixir is exiting without finishing"); + Shutdown (1); + } + + p = p2 + 1; + + } + + return (TRUE); + +} Index: /tags/elixir-1-5/Ohana/src/elixir/src/CheckProcess.c =================================================================== --- /tags/elixir-1-5/Ohana/src/elixir/src/CheckProcess.c (revision 8558) +++ /tags/elixir-1-5/Ohana/src/elixir/src/CheckProcess.c (revision 8558) @@ -0,0 +1,57 @@ +# include "controller.h" + +int CheckProcess (Process *process) { + + Machine *machine; + int cargc, *cargd; + char **cargv; + Object *object; + + CheckCluster (process[0].cluster, process[0].success, process[0].failure, process[0].pending); + if (drand48() > 0.5) return (FALSE); + + + if ((object = GetObject (process[0].pending)) == (Object *) NULL) return (FALSE); + object[0].lastproc = process[0].name; + + /* can't create arguments, some dependencies might not be ready */ + if (!MakeArgs (process, object, &cargc, &cargv, &cargd)) { + FreeArgs (cargc, cargv, cargd); + PutObject (process[0].pending, object); + return (FALSE); + } + + /* dependencies not ready */ + switch (CheckDepend (object, cargc, cargv, cargd)) { + case 0: + FreeArgs (cargc, cargv, cargd); + PutObject (process[0].pending, object); + return (FALSE); + case 2: + FreeArgs (cargc, cargv, cargd); + PutObject (process[0].failure, object); + return (FALSE); + case 1: + default: + break; + } + + /* no machine ready */ + if ((machine = GrabMachine ()) == (Machine *) NULL) { + /* need to free appropriate items from above */ + FreeArgs (cargc, cargv, cargd); + PushObject (process[0].pending, object); + return (FALSE); + } + + StartMachine (machine, object, cargc, cargv); + FreeArgs (cargc, cargv, cargd); + StartProcessTimer (process, machine); + + PutMachine (machine, process[0].cluster); + return (TRUE); +} + +/* objects that are not ready go to the bottom of the stack (PutObject) + objects that are ready, but have no machine free, go to top of stack (PushObject) +*/ Index: /tags/elixir-1-5/Ohana/src/elixir/src/ConfigInit.c =================================================================== --- /tags/elixir-1-5/Ohana/src/elixir/src/ConfigInit.c (revision 8558) +++ /tags/elixir-1-5/Ohana/src/elixir/src/ConfigInit.c (revision 8558) @@ -0,0 +1,187 @@ +# include "controller.h" + +void SIG_STOP (int sig) { + fprintf (stderr, "trapped signal %d, exiting when jobs are done\n", sig); + ElixirStop (); +} + +void SIG_MESSAGE (int sig) { + fprintf (stderr, "trapped signal %d, reading message\n", sig); + CheckMessages (); +} + +void SIG_DIE (int sig) { + fprintf (stderr, "trapped signal %d, exiting\n", sig); + Shutdown (1); +} + +void SIG_PIPE (int sig) { + fprintf (stderr, "pipe signal %d\n", sig); +} + +char *ConfigInit (int *argc, char **argv) { + + int N; + char *config, *file; + + /* use default settings */ + signal (SIGKILL, SIG_DFL); + signal (SIGCONT, SIG_DFL); + signal (SIGSTOP, SIG_DFL); + signal (SIGCHLD, SIG_DFL); + + /* exit on these signals */ + signal (SIGILL, SIG_DIE); + signal (SIGABRT, SIG_DIE); + signal (SIGFPE, SIG_DIE); + signal (SIGSEGV, SIG_DIE); + signal (SIGTERM, SIG_DIE); + signal (SIGBUS, SIG_DIE); + signal (SIGTRAP, SIG_DIE); + signal (SIGXCPU, SIG_DIE); + signal (SIGXFSZ, SIG_DIE); + signal (SIGIOT, SIG_DIE); + signal (SIGPWR, SIG_DIE); + signal (SIGHUP, SIG_IGN); + signal (SIGINT, SIG_DIE); + signal (SIGQUIT, SIG_IGN); + signal (SIGPIPE, SIG_PIPE); + signal (SIGALRM, SIG_IGN); + signal (SIGUSR1, SIG_STOP); + signal (SIGUSR2, SIG_MESSAGE); + signal (SIGTSTP, SIG_IGN); + signal (SIGTTIN, SIG_IGN); + signal (SIGTTOU, SIG_IGN); + signal (SIGPOLL, SIG_IGN); + signal (SIGPROF, SIG_IGN); + signal (SIGURG, SIG_IGN); + signal (SIGVTALRM, SIG_IGN); + signal (SIGIO, SIG_IGN); + signal (SIGWINCH, SIG_IGN); + + /* signals which are not always defined */ +# ifdef SIGSYS + signal (SIGSYS, SIG_DIE); +# endif +# ifdef SIGEMT + signal (SIGEMT, SIG_DIE); +# endif +# ifdef SIGSTKFLT + signal (SIGSTKFLT, SIG_DIE); +# endif +# ifdef SIGINFO + signal (SIGINFO, SIG_DIE); +# endif +# ifdef SIGCLD + signal (SIGCLD, SIG_DFL); +# endif +# ifdef SIGLOST + signal (SIGLOST, SIG_IGN); +# endif +# ifdef SIGUNUSED + signal (SIGUNUSED, SIG_IGN); +# endif + + /*** load configuration info ***/ + file = SelectConfigFile (argc, argv, "ptolemy"); + if (file == (char *) NULL) { + fprintf (stderr, "ERROR: can't choose configuration file\n"); + Shutdown (1); + } + config = LoadConfigFile (file); + if (config == (char *) NULL) { + fprintf (stderr, "ERROR: can't find configuration file %s\n", file); + if (file != (char *) NULL) free (file); + Shutdown (1); + } + if (VERBOSE) fprintf (stderr, "loaded config file: %s\n", file); + free (file); + + if ((N = get_argument (*argc, argv, "-restart"))) { + char pidfile[128]; + + remove_argument (N, argc, argv); + if (!ScanConfig (config, "global.pid", "%s", 0, pidfile)) { + fprintf (stderr, "pid file not defined: global.pid\n"); + exit (1); + } + HalttoRestart (pidfile); + } + + if ((*argc != 2) && (*argc != 1)) { + fprintf (stderr, "USAGE: elixir [list]\n"); + fprintf (stderr, " elixir [-kill]\n"); + fprintf (stderr, " elixir [-stop]\n"); + fprintf (stderr, " elixir [-status]\n"); + Shutdown (1); + } + + /* special command-line options */ + { + char pidfile[128], msgfile[128]; + + if (!ScanConfig (config, "global.pid", "%s", 0, pidfile)) { + fprintf (stderr, "pid file not defined: global.pid\n"); + exit (1); + } + + if (!ScanConfig (config, "global.msg", "%s", 0, msgfile)) { + fprintf (stderr, "msg file not defined: global.msg\n"); + exit (1); + } + + if (!ScanConfig (config, "CONNECT", "%s", 0, CONNECT)) { + sprintf (CONNECT, "/usr/bin/rsh"); + } + + if (get_argument (*argc, argv, "-kill")) KillElixir (pidfile); + if (get_argument (*argc, argv, "-stop")) HaltElixir (pidfile); + if (get_argument (*argc, argv, "-status")) StatusElixir (pidfile, msgfile); + } + + return (config); + +} + +/* + + SIGHUP 1 A Hangup detected on controlling terminal + or death of controlling process + SIGINT 2 A Interrupt from keyboard + SIGQUIT 3 C Quit from keyboard + SIGILL 4 C Illegal Instruction + SIGABRT 6 C Abort signal from abort(3) + SIGFPE 8 C Floating point exception + SIGKILL 9 AEF Kill signal + SIGSEGV 11 C Invalid memory reference + SIGPIPE 13 A Broken pipe: write to pipe with no readers + SIGALRM 14 A Timer signal from alarm(2) + SIGTERM 15 A Termination signal + SIGUSR1 30,10,16 A User-defined signal 1 + SIGUSR2 31,12,17 A User-defined signal 2 + SIGCHLD 20,17,18 B Child stopped or terminated + SIGCONT 19,18,25 Continue if stopped + SIGSTOP 17,19,23 DEF Stop process + SIGTSTP 18,20,24 D Stop typed at tty + SIGTTIN 21,21,26 D tty input for background process + SIGTTOU 22,22,27 D tty output for background process + SIGBUS 10,7,10 C Bus error (bad memory access) + SIGPOLL A Pollable event (Sys V). Synonym of SIGIO + SIGPROF 27,27,29 A Profiling timer expired + SIGSYS 12,-,12 C Bad argument to routine (SVID) + SIGTRAP 5 C Trace/breakpoint trap + SIGURG 16,23,21 B Urgent condition on socket (4.2 BSD) + SIGVTALRM 26,26,28 A Virtual alarm clock (4.2 BSD) + SIGXCPU 24,24,30 C CPU time limit exceeded (4.2 BSD) + SIGXFSZ 25,25,31 C File size limit exceeded (4.2 BSD) + SIGIOT 6 C IOT trap. A synonym for SIGABRT + SIGEMT 7,-,7 + SIGSTKFLT -,16,- A Stack fault on coprocessor + SIGIO 23,29,22 A I/O now possible (4.2 BSD) + SIGCLD -,-,18 A synonym for SIGCHLD + SIGPWR 29,30,19 A Power failure (System V) + SIGINFO 29,-,- A synonym for SIGPWR + SIGLOST -,-,- A File lock lost + SIGWINCH 28,28,20 B Window resize signal (4.3 BSD, Sun) + SIGUNUSED -,31,- A Unused signal (will be SIGSYS) +*/ Index: /tags/elixir-1-5/Ohana/src/elixir/src/ConfigPID.c =================================================================== --- /tags/elixir-1-5/Ohana/src/elixir/src/ConfigPID.c (revision 8558) +++ /tags/elixir-1-5/Ohana/src/elixir/src/ConfigPID.c (revision 8558) @@ -0,0 +1,63 @@ +# include "controller.h" + +static char *PIDMaster = (char *) NULL; + +void ConfigPID (char *PIDFile) { + + pid_t pid; + char *username, machine[256]; + FILE *f; + + f = fopen (PIDFile, "r"); + if (f == (FILE *) NULL) { + + pid = getpid (); + username = getenv ("USER"); + if (username == (char *) NULL) { + fprintf (stderr, "error getting username\n"); + exit (2); + } + bzero (machine, 256); + if (gethostname (machine, 256)) { + fprintf (stderr, "error getting hostname\n"); + exit (2); + } + + f = fopen (PIDFile, "w"); + if (f == (FILE *) NULL) { + fprintf (stderr, "can't write to PID file %s\n", PIDFile); + exit (2); + } + + fprintf (f, "PID: %d\n", pid); + fprintf (f, "USER: %-s\n", username); + fprintf (f, "MACHINE: %-s\n", machine); + fclose (f); + + PIDMaster = PIDFile; + return; + } + + ALLOCATE (username, char, 256); + fscanf (f, "%*s %d", &pid); + fscanf (f, "%*s %s", username); + fscanf (f, "%*s %s", machine); + fclose (f); + + fprintf (stderr, "elixir is apparently running:\n\n"); + fprintf (stderr, " machine: %s\n", machine); + fprintf (stderr, " user: %s\n", username); + fprintf (stderr, " PID: %d\n", pid); + fprintf (stderr, " remove %s if elixir has died unexpectedly\n", PIDFile); + Shutdown (1); +} + +void RemovePID () { + + if (PIDMaster == (char *) NULL) + return; + + if (unlink (PIDMaster)) { + fprintf (stderr, "error deleting PID File %s\n", PIDMaster); + } +} Index: /tags/elixir-1-5/Ohana/src/elixir/src/DefineProcesses.c =================================================================== --- /tags/elixir-1-5/Ohana/src/elixir/src/DefineProcesses.c (revision 8558) +++ /tags/elixir-1-5/Ohana/src/elixir/src/DefineProcesses.c (revision 8558) @@ -0,0 +1,135 @@ +# include "controller.h" + +Process **DefineProcesses (Process *global, int *nprocess, char *config) { + + int i, j; + char procname[256], entry[256], field[256]; + Process **process; + int Nprocess, NPROCESS; + double timeout; + + global[0].success = InitQueue (); + global[0].failure = InitQueue (); + + ALLOCATE (global[0].argv, char *, 8); + ALLOCATE (global[0].argv[0], char, 256); + ALLOCATE (global[0].argv[1], char, 256); + ALLOCATE (global[0].argv[2], char, 256); + ALLOCATE (global[0].argv[3], char, 256); + ALLOCATE (global[0].argv[4], char, 256); + ALLOCATE (global[0].argv[5], char, 256); + ALLOCATE (global[0].argv[6], char, 256); + ALLOCATE (global[0].argv[7], char, 256); + + GetConfig (config, "global.Nargs", "%d", 0, &global[0].argc); + GetConfig (config, "global.success", "%s", 0, global[0].argv[0]); + GetConfig (config, "global.failure", "%s", 0, global[0].argv[1]); + GetConfig (config, "global.pending", "%s", 0, global[0].argv[2]); + GetConfig (config, "global.logfile", "%s", 0, global[0].argv[3]); + GetConfig (config, "global.source", "%s", 0, global[0].argv[4]); + GetConfig (config, "global.msg", "%s", 0, global[0].argv[5]); + GetConfig (config, "global.end", "%s", 0, global[0].argv[6]); + GetConfig (config, "global.pid", "%s", 0, global[0].argv[7]); + + ScanConfig (config, "global.timeout", "%lf", 0, &timeout); + if (timeout < 1) { + fprintf (stderr, "global.timeout is absurd: %f\n", timeout); + Shutdown (1); + } + if (timeout < 100) { + fprintf (stderr, "**** global.timeout is very short (%f) ****\n", timeout); + } + RegisterTimeout (timeout); + + Nprocess = 0; + NPROCESS = 5; + ALLOCATE (process, Process *, NPROCESS); + + /* find all entries in config file labeled 'process' */ + for (i = 1; ScanConfig (config, "process", "%s", i, procname); i++) { + process[Nprocess] = ConfigProcess (config, procname); + Nprocess ++; + if (Nprocess == NPROCESS) { + NPROCESS += 5; + REALLOCATE (process, Process *, NPROCESS); + } + } + if (Nprocess == 0) { + fprintf (stderr, "no processes defined in config file\n"); + Shutdown (1); + } + + /* make links between processes */ + for (i = 0; i < Nprocess; i++) { + /* connect this process success queue */ + sprintf (field, "%s.success", process[i][0].name); + GetConfig (config, field, "%s", 0, entry); + if (!strcasecmp (entry, "global")) { + process[i][0].success = global[0].success; + goto stage1; + } + for (j = 0; j < Nprocess; j++) { + if (!strcasecmp (entry, process[j][0].name)) { + if (i == j) { + fprintf (stderr, "ERROR: can't connect a process to itself: %s\n", process[i][0].name); + Shutdown (1); + } + process[i][0].success = process[j][0].pending; + goto stage1; + } + } + fprintf (stderr, "ERROR: can't connect process %s to target %s\n", process[i][0].name, entry); + Shutdown (1); + + stage1: + /* connect this process failure queue */ + sprintf (field, "%s.failure", process[i][0].name); + GetConfig (config, field, "%s", 0, entry); + if (!strcasecmp (entry, "global")) { + process[i][0].failure = global[0].failure; + goto stage2; + } + for (j = 0; j < Nprocess; j++) { + if (!strcasecmp (entry, process[j][0].name)) { + if (i == j) { + fprintf (stderr, "ERROR: can't connect a process to itself: %s\n", process[i][0].name); + Shutdown (1); + } + process[i][0].failure = process[j][0].pending; + goto stage2; + } + } + fprintf (stderr, "ERROR: can't connect process %s to target %s\n", process[i][0].name, entry); + Shutdown (1); + + stage2: + continue; + } + + /* link global process to first process */ + for (j = 0; j < Nprocess; j++) { + if (!strcasecmp (global[0].argv[2], process[j][0].name)) { + global[0].pending = process[j][0].pending; + goto stage3; + } + } + fprintf (stderr, "ERROR: can't connect global process to target %s\n", global[0].argv[2]); + Shutdown (1); + +stage3: + *nprocess = Nprocess; + return (process); + +} + +void GetConfig (char *config, char *field, char *format, int N, void *ptr) { + + char *status; + + status = ScanConfig (config, field, format, N, ptr); + if (status == NULL) { + fprintf (stderr, "error in config, cannot find required field %s\n", field); + Shutdown (1); + } + return; +} Index: /tags/elixir-1-5/Ohana/src/elixir/src/DumpStatus.c =================================================================== --- /tags/elixir-1-5/Ohana/src/elixir/src/DumpStatus.c (revision 8558) +++ /tags/elixir-1-5/Ohana/src/elixir/src/DumpStatus.c (revision 8558) @@ -0,0 +1,78 @@ +# include "controller.h" + +int DumpStatus (char *filename) { + + FILE *f; + int i, state, total, mode; + Process *global, **process; + int Nprocess, Nobject, Dynamic; + + process = GetProcessInfo (&global, &Nprocess, &Nobject); + Dynamic = GetDynamicState (); + + if (filename == (char *) NULL) { + system ("tput clear"); + f = stderr; + } else { + /* lock file */ + f = fsetlockfile (filename, 0.1, LCK_XCLD, &state); + if (f == NULL) return (2); + fseek (f, 0, SEEK_END); + } + + fprintf (f, "processes status:\n"); + fprintf (f, " name pending success failure total\n"); + fprintf (f, "----------------------------------------------------------------\n"); + total = global[0].pending[0].Nobject + global[0].success[0].Nobject + global[0].failure[0].Nobject; + fprintf (f, "%16s %6d %6d %6d %4d\n", "global", + global[0].pending[0].Nobject, global[0].success[0].Nobject, global[0].failure[0].Nobject, total); + for (i = 0; i < Nprocess; i++) { + fprintf (f, "%16s %6d\n", process[i][0].name, process[i][0].pending[0].Nobject); + } + fprintf (f, "\n"); + + DumpMachineStatus (f); + + fprintf (f, "\n"); + fprintf (f, "objects loaded so far: %d\n\n", Nobject); + if (Dynamic) { + fprintf (f, "accepting input from FIFO\n"); + } else { + fprintf (f, "NOT accepting input from FIFO\n"); + } + + if (f != stderr) { + mode = S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP | S_IROTH | S_IWOTH; + chmod (filename, mode); + fclearlockfile (filename, f, LCK_XCLD, &state); + } + return (TRUE); +} + +/* + +processes status: + name pending success failure total +---------------------------------------------------------------- + global 1 0 4 5 + mkdir 1 + flatten 0 + dophot 0 + imclean 0 + gastro 0 + addstar 0 + rmfile 5 + +machine status: + name process status +---------------------------------------------------------------- + kiawe rmfile 1 + kiawe addstar 9 + milo addstar 1 + +objects loaded so far: 13 + +accepting input from FIFO + + +*/ Index: /tags/elixir-1-5/Ohana/src/elixir/src/FifoOps.c =================================================================== --- /tags/elixir-1-5/Ohana/src/elixir/src/FifoOps.c (revision 8558) +++ /tags/elixir-1-5/Ohana/src/elixir/src/FifoOps.c (revision 8558) @@ -0,0 +1,96 @@ +# include "controller.h" + +int InitFifo (Fifo *fifo, int Nalloc, int Nextra) { + + if (Nextra >= Nalloc) { + fprintf (stderr, "absurd fifo definition\n"); + return (FALSE); + } + + fifo[0].Nalloc = Nalloc; + fifo[0].Nextra = Nextra; + fifo[0].Nmaxread = Nalloc - Nextra; + fifo[0].Nlast = 0; + fifo[0].Nbuffer = 0; + + ALLOCATE (fifo[0].buffer, char, fifo[0].Nalloc); + + return (TRUE); + +} + +int FlushFifo (Fifo *fifo) { + + fifo[0].Nlast = 0; + fifo[0].Nbuffer = 0; + + return (TRUE); + +} + +/* after a shift, we can always read + fifo[0].Nmaxread + bytes into + &fifo[0].buffer[Nbuffer] + which is the byte after then end of existing data */ + +int ShiftFifo (Fifo *fifo) { + + int Nextra, Nshift; + + Nextra = fifo[0].Nextra; + Nshift = fifo[0].Nbuffer - fifo[0].Nextra; + if (Nshift <= 0) return (TRUE); + + memcpy (fifo[0].buffer, &fifo[0].buffer[Nshift], Nextra); + fifo[0].Nbuffer = Nextra; + fifo[0].Nlast = Nextra; + + return (TRUE); +} + +/* like a standard read, ReadtoFifo returns Nbytes read, + -1 for sock busy, or 0 for sock closed */ + +int ReadtoFifo (Fifo *fifo, int sock) { + + int Nread; + int Nbuffer, Nmaxread; + + if (sock == 0) { + fprintf (stderr, "error with socket?\n"); + return (0); + } + + fifo[0].Nlast = fifo[0].Nbuffer; + + Nbuffer = fifo[0].Nbuffer; + Nmaxread = fifo[0].Nmaxread; + Nread = read (sock, &fifo[0].buffer[Nbuffer], Nmaxread); + + if (Nread > 0) fifo[0].Nbuffer += Nread; + + if (Nread == -1) { + /* check for possible errors. anything other than EAGAIN + is bad and should indicate the connection is down */ + switch (errno) { + case EAGAIN: + case EIO: + Nread = -1; + break; + default: + fprintf (stderr, "read error: %d\n", errno); + Nread = 0; + break; + } + } + + return (Nread); +} + +void FreeFifo (Fifo *fifo) { + + if (fifo[0].buffer != (char *) NULL) { + free (fifo[0].buffer); + } +} Index: /tags/elixir-1-5/Ohana/src/elixir/src/InitProcess.c =================================================================== --- /tags/elixir-1-5/Ohana/src/elixir/src/InitProcess.c (revision 8558) +++ /tags/elixir-1-5/Ohana/src/elixir/src/InitProcess.c (revision 8558) @@ -0,0 +1,19 @@ +# include "controller.h" + +Process *InitProcess (char *name, Queue *pending, Queue *failure, int (*mkargs)()) { + + Process *process; + + ALLOCATE (process, Process, 1); + + process[0].success = InitQueue (); + process[0].failure = failure; + process[0].pending = pending; + + process[0].name = strcreate (name); + /* process[0].mkarg = mkargs; */ + + process[0].cluster = InitCluster (); + + return (process); +} Index: /tags/elixir-1-5/Ohana/src/elixir/src/LoadPending.c =================================================================== --- /tags/elixir-1-5/Ohana/src/elixir/src/LoadPending.c (revision 8558) +++ /tags/elixir-1-5/Ohana/src/elixir/src/LoadPending.c (revision 8558) @@ -0,0 +1,127 @@ +# include "controller.h" + +static struct timeval then = {0.0, 0.0}; + +/* LoadPending returns the number of objects loaded from inlist + + state can be: + 0 - success + 1 - file locked + 2 - file has EOF flag + 3 - error in input +*/ + +int LoadPending (Process *global, char *filename, int *state, int *dynamic) { + + int i, status, depend, Nobjects; + int lockstate, mode; + FILE *f; + char name[7][256], line[1024]; + struct timeval now; + Object *object; + + *state = 0; + Nobjects = 0; + + /* we should only do this check every 1sec or so */ + gettimeofday (&now, (void *) NULL); + if (DTIME (now, then) < 1.0) return (Nobjects); + then = now; + + /* if dynamic, we must lock the file first, but not if not dynamic */ + if (*dynamic) { + + /* check lockfile - don't remove list if locked */ + f = fsetlockfile (filename, 1.0, LCK_XCLD, &lockstate); + if (f == NULL) { + *state = 1; + return (Nobjects); + } + } else { + f = fopen (filename, "r"); + if (f == (FILE *) NULL) { + *state = 1; + return (Nobjects); + } + } + + /* read lines, parse each line */ + while (1) { + if (scan_line (f, line) == EOF) goto escape; + + switch (global[0].argc) { + case 1: + status = sscanf (line, "%s", name[0]); + break; + case 2: + status = sscanf (line, "%s%s", name[0], name[1]); + break; + case 3: + status = sscanf (line, "%s%s%s", name[0], name[1], name[2]); + break; + case 4: + status = sscanf (line, "%s%s%s%s", name[0], name[1], name[2], name[3]); + break; + case 5: + status = sscanf (line, "%s%s%s%s%s", name[0], name[1], name[2], name[3], name[4]); + break; + case 6: + status = sscanf (line, "%s%s%s%s%s%s", name[0], name[1], name[2], name[3], name[4], name[5]); + break; + case 7: + status = sscanf (line, "%s%s%s%s%s%s%s", name[0], name[1], name[2], name[3], name[4], name[5], name[6]); + break; + default: + fprintf (stderr, "ERROR: unexpected number of entries per line in input file\n"); + *state = 3; + goto escape; + } + + if (status == 0) continue; /* an empty or blank line */ + + if (!strcasecmp (name[0], "EOF")) { + /* dynamic -> static */ + *state = 2; + goto escape; + } + + /* silently ignore lines with the wrong number of args */ + if (status != global[0].argc) continue; + + ALLOCATE (object, Object, 1); + object[0].argc = status; + ALLOCATE (object[0].argv, char *, status); + for (i = 0; i < status; i++) { + object[0].argv[i] = strcreate (name[i]); + } + + /* convert the global.logfile description to a specific logfile for this object */ + ParseLine (global[0].argv[3], object[0].argc, object[0].argv, &depend, &object[0].logfile); + object[0].status = 0; + object[0].timer.tv_sec = 0; + object[0].lastproc = (char *) NULL; + PutObject (global[0].pending, object); + Nobjects ++; + } + + escape: + if (*dynamic) { + if (truncate (filename, 0)) { + fprintf (stderr, "failed to clear source %s (errno %d)\n", filename, errno); + Shutdown (1); + } + + /* clean up fifo - set mode to 666, unlock */ + mode = S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP | S_IROTH | S_IWOTH; + chmod (filename, mode); + fclearlockfile (filename, f, LCK_XCLD, &lockstate); + } else { + fclose (f); + } + + if ((*state == 2) && *dynamic) { + *dynamic = FALSE; + } + return (Nobjects); + +} Index: /tags/elixir-1-5/Ohana/src/elixir/src/LogOpen.c =================================================================== --- /tags/elixir-1-5/Ohana/src/elixir/src/LogOpen.c (revision 8558) +++ /tags/elixir-1-5/Ohana/src/elixir/src/LogOpen.c (revision 8558) @@ -0,0 +1,24 @@ +# include "controller.h" + +FILE *LogOpen (char *filename) { + + FILE *f; + char *path; + + f = fopen (filename, "a"); + + /* probably don't have needed directory */ + if ((f == (FILE *) NULL) && (errno == ENOENT)) { + path = pathname (filename); + if (!mkdirhier (path, S_IRWXU | S_IRWXG | S_IRWXO)) { + f = fopen (filename, "a"); + } + } + if (f == (FILE *) NULL) { + fprintf (stderr, "LogOpen could not open the log file %s, errno: %d\n", filename, errno); + f = stderr; + } + + return (f); + +} Index: /tags/elixir-1-5/Ohana/src/elixir/src/MachineOps.c =================================================================== --- /tags/elixir-1-5/Ohana/src/elixir/src/MachineOps.c (revision 8558) +++ /tags/elixir-1-5/Ohana/src/elixir/src/MachineOps.c (revision 8558) @@ -0,0 +1,484 @@ +# include "controller.h" + +/* this is in libohana -> should be in ohana.h */ +char *getcwd_cfht (char *path, int size); + +static Cluster *fullpool = (Cluster *) NULL; +static Cluster *idlepool = (Cluster *) NULL; +static Cluster *downpool = (Cluster *) NULL; +static char *ConfigFilename; + +/* put machine on top of cluster */ +void PushMachine (Machine *machine, Cluster *cluster) { + + int i; + + if (cluster[0].Nmachine == cluster[0].NMACHINE) { + cluster[0].NMACHINE += 5; + REALLOCATE (cluster[0].machine, Machine *, cluster[0].NMACHINE); + } + for (i = cluster[0].Nmachine; i > 0; i--) { + cluster[0].machine[i] = cluster[0].machine[i-1]; + } + cluster[0].machine[0] = machine; + cluster[0].Nmachine ++; +} + +/* put machine on bottom of cluster */ +void PutMachine (Machine *machine, Cluster *cluster) { + + if (cluster[0].Nmachine == cluster[0].NMACHINE) { + cluster[0].NMACHINE += 5; + REALLOCATE (cluster[0].machine, Machine *, cluster[0].NMACHINE); + } + cluster[0].machine[cluster[0].Nmachine] = machine; + cluster[0].Nmachine ++; +} + +/* Get a machine from top of cluster */ +Machine *GetMachine (Cluster *cluster) { + + int i; + Machine *machine; + + if (cluster[0].Nmachine == 0) return ((Machine *) NULL); + + machine = cluster[0].machine[0]; + cluster[0].Nmachine --; + for (i = 0; i < cluster[0].Nmachine; i++) { + cluster[0].machine[i] = cluster[0].machine[i+1]; + } + return (machine); +} + +void IdleMachine (Machine *machine) { + + machine[0].status = IDLE; + machine[0].object = (Object *) NULL; + PutMachine (machine, idlepool); + +} + +void DownMachine (Machine *machine) { + + machine[0].status = DOWN; + machine[0].object = (Object *) NULL; + gettimeofday (&machine[0].start, (void *) NULL); + gettimeofday (&machine[0].timer, (void *) NULL); + machine[0].timer.tv_sec += 5; + PutMachine (machine, downpool); + +} + +# define NRETRIES 50 +int TestMachine (Machine *machine) { + + int i, status; + char buffer[64]; + + FlushFifo (&machine[0].fifo); + + if ((machine[0].wsock == 0) || (machine[0].rsock == 0)) return (FALSE); + + /* writes are non-blocking. check for EPIPE in case pipe is closed */ + sprintf (buffer, "echo CONNECTION TEST\n"); + status = write (machine[0].wsock, buffer, strlen(buffer)); + if ((status == -1) && (errno == EPIPE)) { + fprintf (stderr, "socket unexpectedly closed in test\n"); + CloseMachine (machine); + return (FALSE); + } + + status = -1; + for (i = 0; (i < NRETRIES) && (status == -1); i++) { + status = SockScan ("CONNECTION TEST", &machine[0].fifo, machine[0].rsock); + if (status == 0) { + fprintf (stderr, "socket unexpectedly closed in test\n"); + CloseMachine (machine); + return (FALSE); + } + } + if (i == NRETRIES) { + fprintf (stderr, "no response from machine, shutting it down\n"); + CloseMachine (machine); + return (FALSE); + } + FlushFifo (&machine[0].fifo); + return (TRUE); +} + +Machine *GrabMachine () { + + Machine *machine; + + machine = GetMachine (idlepool); + if (machine == (Machine *) NULL) return (machine); + + if (!TestMachine (machine)) { + DownMachine (machine); + return ((Machine *) NULL); + } + return (machine); +} + +Cluster *InitCluster () { + + Cluster *cluster; + + ALLOCATE (cluster, Cluster, 1); + cluster[0].NMACHINE = 5; + cluster[0].Nmachine = 0; + ALLOCATE (cluster[0].machine, Machine *, cluster[0].NMACHINE); + + return (cluster); + +} + +void InitMachines (char *config) { + + int i, Nm, fd, Nout, status; + char name[256]; + Machine *machine; + char *cwd; + + fullpool = InitCluster (); + idlepool = InitCluster (); + downpool = InitCluster (); + + /* processes run on these machines need to have access to the + exact config file used by elixir. Therefore, we write the + config file to a unique filename in this directory and pass + that name to the processes which start machines. it is + crucial that the cwd be visible on the other machines, and have + the same path name. when we exit (Shutdown(n)), we will delete + this file (but not before!). */ + + /* get cwd and create config file here */ + if ((cwd = getcwd_cfht (NULL, 1024)) == NULL) { + fprintf (stderr, "error getting cwd\n"); + Shutdown (1); + } + sprintf (name, "%s/elixir.XXXXXX", cwd); + if ((fd = mkstemp (name)) == -1) { + fprintf (stderr, "can't store current config in cwd\n"); + Shutdown (1); + } + Nout = write (fd, config, strlen (config)); + if (Nout != strlen (config)) { + fprintf (stderr, "can't store current config in cwd\n"); + Shutdown (1); + } + status = close (fd); + if (status == -1) { + fprintf (stderr, "can't store current config in cwd\n"); + Shutdown (1); + } + ConfigFilename = strcreate (name); + + /* create connection to each of the machines */ + Nm = 0; + for (i = 1; ScanConfig (config, "MACHINE", "%s", i, name); i++) { + ALLOCATE (machine, Machine, 1); + machine[0].hostname = strcreate (name); + InitFifo (&machine[0].fifo, 0x4000, 0x1000); + if (ConnectMachine (machine)) { + IdleMachine (machine); + } else { + DownMachine (machine); + } + PutMachine (machine, fullpool); + Nm ++; + } + if (Nm == 0) { + fprintf (stderr, "no available machines, exiting\n"); + Shutdown (1); + } + +} + +int ConnectMachine (Machine *machine) { + + int rsock, wsock, pid; + char line[256]; + + pid = rconnect (machine[0].hostname, CONNECT, &rsock, &wsock); + if (pid) { + machine[0].rsock = rsock; + machine[0].wsock = wsock; + machine[0].pid = pid; + /* we can set up the shell here */ + sprintf (line, "setenv PTOLEMY %s\n", ConfigFilename); + write (wsock, line, strlen (line)); + sprintf (line, "umask 002\n"); + write (wsock, line, strlen (line)); + return (TRUE); + } else { + fprintf (stderr, "can't make connection to %s, skipping for now\n", machine[0].hostname); + machine[0].rsock = 0; + machine[0].wsock = 0; + machine[0].pid = 0; + return (FALSE); + } +} + +/* machine which are claimed as down need to be restarted. + first, check that they really are down, then restart as needed */ +void RestartMachines () { + + int i; + Machine *machine; + double dtime; + struct timeval now; + + for (i = 0; i < downpool[0].Nmachine; i++) { + machine = GetMachine (downpool); + if (machine == (Machine *) NULL) return; + + /* we only try to reconnect if timer is expired */ + gettimeofday (&now, (void *) NULL); + dtime = DTIME (machine[0].timer, now); + if (dtime > 0) { + PutMachine (machine, downpool); + continue; + } + + fprintf (stderr, "restarting machine %s\n", machine[0].hostname); + if (TestMachine (machine)) { + /* machine is still alive, return to idlepool */ + fprintf (stderr, "%s is alive\n", machine[0].hostname); + IdleMachine (machine); + continue; + } + + if (ConnectMachine (machine)) { + fprintf (stderr, "connection to %s successfully restarted\n", machine[0].hostname); + IdleMachine (machine); + } else { + /* advance dtime so restarts happen later and later */ + dtime = DTIME (machine[0].timer, machine[0].start); + dtime = dtime * 2; + dtime = MIN (600, dtime); + gettimeofday (&machine[0].start, (void *) NULL); + gettimeofday (&machine[0].timer, (void *) NULL); + machine[0].timer.tv_sec += dtime; + PutMachine (machine, downpool); + } + } + +} + +void Shutdown (int status) { + + int i; + + if (unlink (ConfigFilename) == -1) { + fprintf (stderr, "trouble deleting config file: %s\n", ConfigFilename); + } + + RemovePID (); + + if (fullpool == (Cluster *) NULL) exit (status); + + for (i = 0; i < fullpool[0].Nmachine; i++) { + CloseMachine (fullpool[0].machine[i]); + } + + exit (status); + +} + +void Restart (char **argv) { + + int i; + + if (unlink (ConfigFilename) == -1) { + fprintf (stderr, "trouble deleting config file: %s\n", ConfigFilename); + } + + RemovePID (); + + if (fullpool == (Cluster *) NULL) execvp (argv[0], argv); + + for (i = 0; i < fullpool[0].Nmachine; i++) { + CloseMachine (fullpool[0].machine[i]); + } + + execvp (argv[0], argv); +} + + +int InitMsgFile (char *file) { + + char line[256]; + int status; + struct stat filestat; + + if (stat (file, &filestat) == -1) return (TRUE); + sprintf (line, "mv -f %s %s~", file, file); + status = system (line); + return (status); +} + +void CloseMachine (Machine *machine) { + + char buffer[128]; + + sprintf (buffer, "exit\n"); + write (machine[0].wsock, buffer, strlen(buffer)); + close (machine[0].wsock); + close (machine[0].rsock); + fprintf (stderr, "shutdown machine %s, pid %d\n", machine[0].hostname, machine[0].pid); + +} + +void DumpMachineStatus (FILE *f) { + + int i; + + if (fullpool == (Cluster *) NULL) return; + + fprintf (f, "machine status:\n"); + fprintf (f, " name process status\n"); + fprintf (f, "----------------------------------------------------------------\n"); + for (i = 0; i < fullpool[0].Nmachine; i++) { + if (fullpool[0].machine[i][0].object == (Object *) NULL) { + fprintf (f, "%14s %8s %2d\n", + fullpool[0].machine[i][0].hostname, + "(none)", + fullpool[0].machine[i][0].status); + + } else { + fprintf (f, "%14s %8s %2d\n", + fullpool[0].machine[i][0].hostname, + fullpool[0].machine[i][0].object[0].lastproc, + fullpool[0].machine[i][0].status); + } + + } + +} + +static int Nprocs, Nmachine; +static ProcessTimers *proctimers; + +/* this must be called after machines have been loaded, registered */ +void InitProcessTimers (Process **process, int Nprocess) { + int i, j; + + Nmachine = fullpool[0].Nmachine; + Nprocs = Nprocess; + ALLOCATE (proctimers, ProcessTimers, Nmachine); + + for (i = 0; i < Nmachine; i++) { + ALLOCATE (proctimers[i].timer, struct timeval, Nprocs); + ALLOCATE (proctimers[i].timesum, double, Nprocs); + ALLOCATE (proctimers[i].Njobs, int, Nprocs); + ALLOCATE (proctimers[i].active, int, Nprocs); + ALLOCATE (proctimers[i].process, Process *, Nprocs); + + for (j = 0; j < Nprocess; j++) { + proctimers[i].timer[j].tv_sec = 0; + proctimers[i].timer[j].tv_usec = 0; + proctimers[i].timesum[j] = 0; + proctimers[i].Njobs[j] = 0; + proctimers[i].active[j] = FALSE; + proctimers[i].process[j] = process[j]; + } + + proctimers[i].machine = fullpool[0].machine[i]; + + } + +} + +void StartProcessTimer (Process *process, Machine *machine) { + + int i, j; + + /* identify the machine */ + for (i = 0; i < Nmachine; i++) { + if (proctimers[i].machine == machine) { + for (j = 0; j < Nprocs; j++) { + if (proctimers[i].process[j] == process) { + gettimeofday (&proctimers[i].timer[j], (void *) NULL); + proctimers[i].active[j] = TRUE; + return; + } + } + } + } + + fprintf (stderr, "can't find this machine, process combination!\n"); + fprintf (stderr, "machine: %p process: %p\n", machine, process); + return; +} + +void StopProcessTimer (Machine *machine) { + + int i, j; + struct timeval stop; + double dtime; + + /* identify the machine */ + for (i = 0; i < Nmachine; i++) { + if (proctimers[i].machine == machine) { + for (j = 0; j < Nprocs; j++) { + if (proctimers[i].active[j]) { + gettimeofday (&stop, (void *) NULL); + dtime = DTIME (stop, proctimers[i].timer[j]); + proctimers[i].timesum[j] += dtime; + proctimers[i].active[j] = FALSE; + proctimers[i].Njobs[j] ++; + return; + } + } + } + } + + fprintf (stderr, "can't find an active process for this machine!\n"); + fprintf (stderr, "machine: %p\n", machine); + +} + +int DumpProcessTimes (char *filename) { + + int i, j, state, mode; + FILE *f; + double dt; + + if (filename == (char *) NULL) { + system ("tput clear"); + f = stderr; + } else { + /* check lockfile */ + f = fsetlockfile (filename, 0.1, LCK_XCLD, &state); + if (f == NULL) return (2); + fseek (f, 0, SEEK_END); + } + + fprintf (f, " ."); + for (j = 0; j < Nprocs; j++) { + fprintf (f, " %-14s", proctimers[0].process[j][0].name); + } + fprintf (f, "\n"); + for (i = 0; i < Nmachine; i++) { + fprintf (f, "%16s ", proctimers[i].machine[0].hostname); + for (j = 0; j < Nprocs; j++) { + if (proctimers[i].Njobs[j] > 0) { + dt = proctimers[i].timesum[j] / proctimers[i].Njobs[j]; + } else { + dt = 0; + } + fprintf (f, "%7.3f %-6d ", dt, proctimers[i].Njobs[j]); + } + fprintf (f, "\n"); + } + + if (f != stderr) { + mode = S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP | S_IROTH | S_IWOTH; + chmod (filename, mode); + fclearlockfile (filename, f, LCK_XCLD, &state); + } + return (1); +} + Index: /tags/elixir-1-5/Ohana/src/elixir/src/MsgOps.c =================================================================== --- /tags/elixir-1-5/Ohana/src/elixir/src/MsgOps.c (revision 8558) +++ /tags/elixir-1-5/Ohana/src/elixir/src/MsgOps.c (revision 8558) @@ -0,0 +1,114 @@ +# include "controller.h" + +int WaitMsg (char *fifo, char **message, double maxdelay) { + + int status, Nsleep; + struct timeval now, then; + double dtime; + struct stat filestats; + + /* maxdelay is the longest we will wait if the file is locked. + if the file doesn't exist, or is empty, skip it */ + /* limit our reads to only 10 tries, waiting for a little while in between */ + Nsleep = 100000 * maxdelay; + + status = stat (fifo, &filestats); + if (status == -1) return (0); + if (filestats.st_size == 0) return (0); + + gettimeofday (&then, (void *) NULL); + while (TRUE) { + status = ReadMsg (fifo, message); + switch (status) { + case 1: + return (1); + default: + gettimeofday (&now, (void *) NULL); + dtime = DTIME (now, then); + if (dtime > maxdelay) return (0); + } + usleep (Nsleep); + } +} + +int ReadMsg (char *fifo, char **message) { + + int nbytes, Nbytes, NBYTES; + char *buffer; + int state, mode; + FILE *f; + + /* check lockfile */ + f = fsetlockfile (fifo, 0.1, LCK_XCLD, &state); + if (f == NULL) return (2); + + /* if file is empty, return 0 */ + if (state == LCK_EMPTY) { + mode = S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP | S_IROTH | S_IWOTH; + chmod (fifo, mode); + fclearlockfile (fifo, f, LCK_XCLD, &state); + return (0); + } + + /* read data from file */ + Nbytes = 0; + NBYTES = 0x1000; + ALLOCATE (buffer, char, NBYTES); + while (TRUE) { + nbytes = fread (&buffer[Nbytes], 1, 0x1000, f); + if (nbytes < 0) { + fprintf (stderr, "error in ReadMsg -- got -1 bytes\n"); + exit (0); + } + if (nbytes == 0) break; + Nbytes += nbytes; + NBYTES += 0x1000; + REALLOCATE (buffer, char, NBYTES); + } + buffer[Nbytes] = 0; + + /* file will remain until unlocked and fclosed below, + but cannot be written to because it is locked */ + if (truncate (fifo, 0)) { + fprintf (stderr, "failed to clear fifo file %s (errno: %d)\n", fifo, errno); + Shutdown (1); + } + + mode = S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP | S_IROTH | S_IWOTH; + chmod (fifo, mode); + fclearlockfile (fifo, f, LCK_XCLD, &state); + + if (Nbytes == 0) { + free (buffer); + return (0); + } + + *message = buffer; + return (1); + +} + +int WriteMsg (char *fifo, char *message) { + + int state, mode; + FILE *f; + + /* check lockfile */ + f = fsetlockfile (fifo, 0.1, LCK_XCLD, &state); + if (f == NULL) return (2); + + /* write message to end of file */ + fseek (f, 0, SEEK_END); + fprintf (f, "%s\n", message); + + mode = S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP | S_IROTH | S_IWOTH; + chmod (fifo, mode); + fclearlockfile (fifo, f, LCK_XCLD, &state); + return (1); +} + +/* possible return states: + 0 - no message (fifo file empty or non-existent) + 1 - message received + 2 - busy (fifo file locked) +*/ Index: /tags/elixir-1-5/Ohana/src/elixir/src/Photcodes.c =================================================================== --- /tags/elixir-1-5/Ohana/src/elixir/src/Photcodes.c (revision 8558) +++ /tags/elixir-1-5/Ohana/src/elixir/src/Photcodes.c (revision 8558) @@ -0,0 +1,256 @@ +# include "controller.h" + +char *GetPhotcode (char *file) { + + char *code; + Header header; + char detector[80], filter[80]; + int i, imageid; + + /* read in image header, open image data region */ + if (!gfits_read_header (file, &header)) { + fprintf (stderr, "ERROR: can't find image file %s (1)\n", file); + exit(0); + } + + /** WARNING: this should use the abstracted keyword names **/ + gfits_scan (&header, "DETECTOR", "%s", 1, detector); + gfits_scan (&header, "FILTER", "%s", 1, filter); + gfits_scan (&header, "IMAGEID", "%d", 1, &imageid); + + gfits_free_header (&header); + + for (i = 0; i < strlen(detector); i++) { detector[i] = toupper (detector[i]); } + for (i = 0; i < strlen(filter); i++) { filter[i] = toupper (filter[i]); } + + ALLOCATE (code, char, 256); + sprintf (code, "%s.%s.%02d", detector, filter, imageid); + return (code); + +} + +/** WARNING: this is pretty weak **/ +char *GetPhotcodeMef (char *file) { + + char *code; + Header header; + char detector[80], filter[80]; + int i, imageid; + char *filename, *p; + + /* file contains file and extend number */ + ALLOCATE (filename, char, strlen(file) + 1); + + p = file; + while ((p = strchr (p, ',')) != (char *) NULL) { *p = ' '; } + sscanf (file, "%s %d", filename, &imageid); + + /* read in image header, open image data region */ + if (!gfits_read_header (filename, &header)) { + fprintf (stderr, "ERROR: can't find image file %s (1)\n", file); + exit(0); + } + free (filename); + + gfits_scan (&header, "DETECTOR", "%s", 1, detector); + gfits_scan (&header, "FILTER", "%s", 1, filter); + gfits_free_header (&header); + + for (i = 0; i < strlen(detector); i++) { detector[i] = toupper (detector[i]); } + for (i = 0; i < strlen(filter); i++) { filter[i] = toupper (filter[i]); } + + ALLOCATE (code, char, 256); sprintf (code, "%s.%s.%02d", detector, filter, imageid); + return (code); + +} + +char *BuildCode (char *line) { + + char *code, *p; + Header header; + char fullname[256], detector[80], filter[80]; + char name[256], path[256], ccd[80], mode[80]; + int i; + + p = line; + while ((p = strchr (p, ',')) != (char *) NULL) { *p = ' '; } + sscanf (line, "%s %s %s %s", path, name, ccd, mode); + + if (!strcasecmp (mode, "MEF")) { + sprintf (fullname, "%s/%s.fits", path, name); + } else { + sprintf (fullname, "%s/%s/%s%s.fits", path, name, name, ccd); + } + + if (!gfits_read_header (fullname, &header)) { + fprintf (stderr, "ERROR: can't find image file %s (1)\n", fullname); + exit(0); + } + gfits_scan (&header, "DETECTOR", "%s", 1, detector); + gfits_scan (&header, "FILTER", "%s", 1, filter); + gfits_free_header (&header); + + for (i = 0; i < strlen(detector); i++) { detector[i] = toupper (detector[i]); } + for (i = 0; i < strlen(filter); i++) { filter[i] = toupper (filter[i]); } + + ALLOCATE (code, char, 256); + sprintf (code, "%s.%s.%s", detector, filter, ccd); + return (code); + +} + +char *BuildName (char *line) { + + char *fullname; + char name[256], path[256], ccd[80], mode[80]; + char *p; + + ALLOCATE (fullname, char, 256); + + p = line; + while ((p = strchr (p, ',')) != (char *) NULL) { *p = ' '; } + sscanf (line, "%s %s %s %s", path, name, ccd, mode); + + if (!strcasecmp (mode, "MEF")) { + sprintf (fullname, "%s/%s.fits", path, name); + } else { + sprintf (fullname, "%s/%s/%s%s.fits", path, name, name, ccd); + } + + return (fullname); + +} + +char *GetPhotcodeExt (char *file) { + + char *code; + Header header; + char detector[80], filter[80]; + int i, imageid, extend; + char *filename, *p; + + /* file contains file and extend number */ + ALLOCATE (filename, char, strlen(file) + 1); + + p = file; + while ((p = strchr (p, ',')) != (char *) NULL) { *p = ' '; } + sscanf (file, "%s %d", filename, &extend); + + /* read in image header, open image data region */ + if (!gfits_read_Xheader (filename, &header, extend)) { + fprintf (stderr, "ERROR: can't find image file %s (1)\n", file); + exit(0); + } + free (filename); + + gfits_scan (&header, "DETECTOR", "%s", 1, detector); + gfits_scan (&header, "FILTER", "%s", 1, filter); + gfits_scan (&header, "IMAGEID", "%d", 1, &imageid); + gfits_free_header (&header); + + for (i = 0; i < strlen(detector); i++) { detector[i] = toupper (detector[i]); } + for (i = 0; i < strlen(filter); i++) { filter[i] = toupper (filter[i]); } + + ALLOCATE (code, char, 256); + sprintf (code, "%s.%s.%02d", detector, filter, imageid); + return (code); + +} + +/* given path/filename.ext return filename */ +char *RootFilename (char *file) { + + int Nbyte; + char *root, *p1, *p2; + + p1 = strrchr (file, '/'); + if (p1 == (char *) NULL) p1 = file; + + p2 = strrchr (file, '.'); + if (p2 == (char *) NULL) p2 = p1 + strlen(p1); + Nbyte = p2-p1; + + ALLOCATE (root, char, Nbyte + 1); + strncpy (root, p1, Nbyte); + root[Nbyte] = 0; + + return (root); + +} + +/* given: path/filename + return: path or ./ if none */ +char *PathFilename (char *file) { + + int Nbyte; + char *path, *p1; + + p1 = strrchr (file, '/'); + if (p1 == (char *) NULL) { + ALLOCATE (p1, char, 2); + strcpy (p1, "."); + return (p1); + } + + Nbyte = p1-file; + ALLOCATE (path, char, Nbyte + 1); + strncpy (path, file, Nbyte); + path[Nbyte] = 0; + + return (path); + +} + +/* given: A/B/D/filename + return: D or ./ if none + if /filename, return '/' + if filename, return ./ +*/ +char *BaseFilename (char *file) { + + int Nbyte; + char *path, *p1, *p2; + + p1 = strrchr (file, '/'); + if (p1 == (char *) NULL) { + ALLOCATE (p1, char, 2); + strcpy (p1, "."); + return (p1); + } + if (p1 == file) { + ALLOCATE (p1, char, 2); + strcpy (p1, "/"); + return (p1); + } + + p2 = p1 - 1; + while ((p2 > file) && (*p2 != '/')) p2--; + if (p2 != file) { + p2 ++; + } + + Nbyte = (p1 - p2); + ALLOCATE (path, char, Nbyte + 1); + strncpy (path, p2, Nbyte); + path[Nbyte] = 0; + + return (path); + +} + + +/* + +There are two methods to implement the Photcode determination for MEF files. +The difference is in whether the IMAGEID is used to refer to the ccd, or the EXTNUM. +Originally, EXTNUM was used, which meant we needed to determine the IMAGEID for a given EXTNUM. +The old implementation of PhotCodeMef returned the photcode based on the input MEF filename +and the EXTNUM. This is now maintained as PhotCodeExt, but not used in elixir. + +As of 13/2/00, we've converted to only using the IMAGEID. this means that PhotCodeMef returns +the photcode, based on the MEF filename and the IMAGEID. To go along with this change, the +function flatten.mef operates on the given IMAGEID not EXTNUM as well. + +*/ + + Index: /tags/elixir-1-5/Ohana/src/elixir/src/ProcessOps.c =================================================================== --- /tags/elixir-1-5/Ohana/src/elixir/src/ProcessOps.c (revision 8558) +++ /tags/elixir-1-5/Ohana/src/elixir/src/ProcessOps.c (revision 8558) @@ -0,0 +1,240 @@ +# include "controller.h" + +/* take entry and return char with filled out value, + perform recursively */ + +char *ExpandEntry (char *entry, int argc, char **argv) { + + char *p1, *p2, *p3; + int Nbyte, N; + char *function, *operand, *value; + int Nout, Ncpy; + + if (entry[0] == '!') { + p1 = strchr (entry, '!'); p1++; + p2 = strchr (entry, '('); + p3 = strchr (entry, ')'); + if ((p2 == (char *) NULL) || (p3 == (char *) NULL)) { + fprintf (stderr, "syntax error in process defs\n"); + Shutdown (1); + } + Nbyte = p2 - p1; + function = strncreate (p1, Nbyte); + + p2 ++; + Nbyte = p3 - p2; + operand = strncreate (p2, Nbyte); + value = ExpandEntry (operand, argc, argv); + free (operand); + operand = value; + + value = (char *) NULL; + if (!strcasecmp (function, "buildname")) { + value = BuildName (operand); + } + if (!strcasecmp (function, "buildcode")) { + value = BuildCode (operand); + } + if (!strcasecmp (function, "photcode")) { + value = GetPhotcode (operand); + } + if (!strcasecmp (function, "photcodemef")) { + value = GetPhotcodeMef (operand); + } + if (!strcasecmp (function, "root")) { + value = RootFilename (operand); + } + if (!strcasecmp (function, "path")) { + value = PathFilename (operand); + } + if (!strcasecmp (function, "base")) { + value = BaseFilename (operand); + } + if (value == (char *) NULL) { + fprintf (stderr, "unknown process command %s\n", function); + Shutdown (1); + } + free (function); + free (operand); + } else { + Nbyte = strlen (entry) + 50; + ALLOCATE (value, char, Nbyte); + + Nout = 0; + p1 = entry; + while ((p2 = strchr (p1, '&')) != (char *) NULL) { + Ncpy = p2 - p1; + if (Nout + Ncpy >= Nbyte - 1) { + Nbyte = Nout + Ncpy + 50; + REALLOCATE (value, char, Nbyte); + } + strncpy (&value[Nout], p1, Ncpy); + Nout += Ncpy; + p2 ++; + N = strtod (p2, &p1); + if ((N > argc - 1) || (N < 0)) { + fprintf (stderr, "ERROR: command expects too many object arguments\n"); + Shutdown (1); + } + Ncpy = strlen (argv[N]); + if (Nout + Ncpy >= Nbyte - 1) { + Nbyte = Nout + Ncpy + 50; + REALLOCATE (value, char, Nbyte); + } + strcpy (&value[Nout], argv[N]); + Nout += Ncpy; + } + Ncpy = strlen (p1); + if (Nout + Ncpy >= Nbyte - 1) { + Nbyte = Nout + Ncpy + 50; + REALLOCATE (value, char, Nbyte); + } + strncpy (&value[Nout], p1, Ncpy); + Nout += Ncpy; + value[Nout] = 0; + } + + return (value); + +} + +Process *ConfigProcess (char *config, char *procname) { + + int j, NPAR; + char argname[256], argline[256]; + Process *process; + + NPAR = 10; + ALLOCATE (process, Process, 1); + ALLOCATE (process[0].argv, char *, NPAR); + + process[0].name = strcreate (procname); + + /* find all lines of form procname.arg - the argument lines */ + sprintf (argname, "%s.arg", procname); + for (j = 0; ScanConfig (config, argname, "%s", j+1, argline); j++) { + process[0].argv[j] = strcreate (argline); + if (j == NPAR) { + NPAR += 10; + REALLOCATE (process[0].argv, char *, NPAR); + } + } + process[0].argc = j; + REALLOCATE (process[0].argv, char *, MAX (j,1)); + if (j == 0) { + fprintf (stderr, "ERROR: process %s has no arguments defined\n", procname); + Shutdown (1); + } + + process[0].pending = InitQueue (); + process[0].failure = (Queue *) NULL; + process[0].success = (Queue *) NULL; + process[0].cluster = InitCluster (); + + return (process); + +} + +/* need a return value */ +int MakeArgs (Process *process, Object *object, int *cargc, char ***cargv, int **cargd) { + + int i; + int Cargc, *Cargd; + char **Cargv; + + Cargc = process[0].argc; + ALLOCATE (Cargv, char *, Cargc); + ALLOCATE (Cargd, int, Cargc); + + /* convert par to entry */ + for (i = 0; i < Cargc; i++) { + ParseLine (process[0].argv[i], object[0].argc, object[0].argv, &Cargd[i], &Cargv[i]); + } + *cargc = Cargc; + *cargv = Cargv; + *cargd = Cargd; + + return (TRUE); + +} + +/* 'process' carries around a list of strings which MakeArgs uses to create the command line + arguments and the dependencies (cargc, cargv, cargd) */ + +void FreeArgs (int argc, char **argv, int *argd) { + + int i; + + for (i = 0; i < argc; i++) { + free (argv[i]); + } + + free (argv); + free (argd); + +} + + +void ParseLine (char *testline, int argc, char **argv, int *depend, char **outline) { + + int i, k; + char format[256], *line, *value; + int NTERMS, Nterms; + char **terms; + + + NTERMS = 8; + ALLOCATE (terms, char *, NTERMS); + for (i = 0; i < NTERMS; i++) { + ALLOCATE (terms[i], char, 256); + } + + Nterms = sscanf (testline, "%d%s%s%s%s%s%s%s%s%s", depend, format, + terms[0], terms[1], terms[2], terms[3], terms[4], terms[5], terms[6], terms[7]); + Nterms -= 2; + + for (k = 0; k < Nterms; k++) { + value = ExpandEntry (terms[k], argc, argv); + strcpy (terms[k], value); + free (value); + } + + ALLOCATE (line, char, 256); + switch (Nterms) { + case 0: + sprintf (line, format); + break; + case 1: + sprintf (line, format, terms[0]); + break; + case 2: + sprintf (line, format, terms[0], terms[1]); + break; + case 3: + sprintf (line, format, terms[0], terms[1], terms[2]); + break; + case 4: + sprintf (line, format, terms[0], terms[1], terms[2], terms[3]); + break; + case 5: + sprintf (line, format, terms[0], terms[1], terms[2], terms[3], terms[4]); + break; + case 6: + sprintf (line, format, terms[0], terms[1], terms[2], terms[3], terms[4], terms[5]); + break; + case 7: + sprintf (line, format, terms[0], terms[1], terms[2], terms[3], terms[4], terms[5], terms[6]); + break; + case 8: + sprintf (line, format, terms[0], terms[1], terms[2], terms[3], terms[4], terms[5], terms[6], terms[7]); + break; + default: + fprintf (stderr, "too many terms for command line argument (lim = 5)\n"); + Shutdown (1); + } + *outline = line; + + for (i = 0; i < NTERMS; i++) { free (terms[i]); } + free (terms); + +} Index: /tags/elixir-1-5/Ohana/src/elixir/src/QueueOps.c =================================================================== --- /tags/elixir-1-5/Ohana/src/elixir/src/QueueOps.c (revision 8558) +++ /tags/elixir-1-5/Ohana/src/elixir/src/QueueOps.c (revision 8558) @@ -0,0 +1,60 @@ +# include "controller.h" + +/* get object from top of stack */ +Object *GetObject (Queue *queue) { + + int i; + Object *object; + + if (queue[0].Nobject == 0) return ((Object *) NULL); + + object = queue[0].object[0]; + queue[0].Nobject --; + for (i = 0; i < queue[0].Nobject; i++) { + queue[0].object[i] = queue[0].object[i+1]; + } + return (object); +} + +/* put object on bottom of stack */ +void PutObject (Queue *queue, Object *object) { + + if (queue[0].Nobject == queue[0].NOBJECT) { + queue[0].NOBJECT += 100; + REALLOCATE (queue[0].object, Object *, queue[0].NOBJECT); + } + queue[0].object[queue[0].Nobject] = object; + queue[0].Nobject ++; + +} + +/* push object on top of stack */ +void PushObject (Queue *queue, Object *object) { + + int i; + + if (queue[0].Nobject == queue[0].NOBJECT) { + queue[0].NOBJECT += 100; + REALLOCATE (queue[0].object, Object *, queue[0].NOBJECT); + } + for (i = queue[0].Nobject; i > 0; i--) { + queue[0].object[i] = queue[0].object[i-1]; + } + queue[0].object[0] = object; + queue[0].Nobject ++; + +} + +/* allocate queue, setup with default values, allocate data */ +Queue *InitQueue () { + + Queue *queue; + + ALLOCATE (queue, Queue, 1); + + queue[0].Nobject = 0; + queue[0].NOBJECT = 50; + ALLOCATE (queue[0].object, Object *, queue[0].NOBJECT); + return (queue); + +} Index: /tags/elixir-1-5/Ohana/src/elixir/src/RemoteOps.c =================================================================== --- /tags/elixir-1-5/Ohana/src/elixir/src/RemoteOps.c (revision 8558) +++ /tags/elixir-1-5/Ohana/src/elixir/src/RemoteOps.c (revision 8558) @@ -0,0 +1,189 @@ +# include "controller.h" + +void HaltElixir (char *pidfile) { + + pid_t pid; + char username[256], machine[256]; + char line[512]; + int i, wsock, rsock; + struct stat filestat; + + if (!LoadPID (pidfile, &pid, username, machine)) { + fprintf (stderr, "elixir is not running\n"); + exit (0); + } + + /* set signal to remote machine */ + if (!rconnect (machine, CONNECT, &rsock, &wsock)) { + fprintf (stderr, "can't make connection to machine %s to kill process\n", machine); + exit (1); + } + sprintf (line, "kill -USR1 %d\n", pid); + write (wsock, line, strlen (line)); + + for (i = 0; i < 100; i++) { + if (stat (pidfile, &filestat) == -1) exit (0); + usleep (100000); + } + fprintf (stderr, "elixir is still running\n"); + exit (2); +} + +void KillElixir (char *pidfile) { + + pid_t pid; + char username[256], machine[256]; + char line[512]; + int i, wsock, rsock; + struct stat filestat; + + if (!LoadPID (pidfile, &pid, username, machine)) { + fprintf (stderr, "elixir is not running\n"); + exit (0); + } + + /* send signal to remote machine */ + if (!rconnect (machine, CONNECT, &rsock, &wsock)) { + fprintf (stderr, "can't make connection to machine %s to kill process\n", machine); + exit (1); + } + sprintf (line, "kill -TERM %d\n", pid); + write (wsock, line, strlen (line)); + + for (i = 0; i < 100; i++) { + if (stat (pidfile, &filestat) == -1) exit (0); + usleep (100000); + } + fprintf (stderr, "elixir is still running\n"); + exit (2); +} + +void StatusElixir (char *pidfile, char *msgfile) { + + pid_t pid; + char username[256], machine[256], response[256], message[512]; + char *answer; + char line[512]; + int i, done, status, wsock, rsock; + struct stat filestat; + struct timeval now, then; + + if (!LoadPID (pidfile, &pid, username, machine)) { + fprintf (stderr, "elixir is not running\n"); + exit (0); + } + + sprintf (response, "%s.XXXXXX", msgfile); + mkstemp (response); + sprintf (message, "STATUS %s", response); + if (VERBOSE) fprintf (stderr, "sending message: %s\n", message); + WriteMsg (msgfile, message); + + /* send signal to remote machine */ + if (!rconnect (machine, CONNECT, &rsock, &wsock)) { + fprintf (stderr, "can't make connection to machine %s to signal process\n", machine); + exit (1); + } + sprintf (line, "kill -USR2 %d\n", pid); + write (wsock, line, strlen (line)); + + /* wait (2 sec) for file to exist, then try to read it */ + for (i = 0; ((status = stat (response, &filestat)) == -1) && (i < 20); i++) { + if (VERBOSE) fprintf (stderr, "waiting for response: %d\n", status); + usleep (100000); + } + if (i >= 20) { + fprintf (stderr, "no response\n"); + exit (2); + } + + done = FALSE; + gettimeofday (&then, (void *) NULL); + if (VERBOSE) fprintf (stderr, "reading message: %s\n", message); + while (!done) { + status = ReadMsg (response, &answer); + switch (status) { + case 1: + done = TRUE; + break; + default: + gettimeofday (&now, (void *) NULL); + if (DTIME (now, then) > 5.0) { + fprintf (stderr, "no response from elixir\n"); + exit (2); + } + } + usleep (10000); + } + fprintf (stderr, "%s\n", answer); + gettimeofday (&now, (void *) NULL); + if (VERBOSE) fprintf (stderr, "response in %f\n", DTIME (now, then)); + + unlink (response); + exit (0); +} + +int HalttoRestart (char *pidfile) { + + pid_t pid; + char username[256], machine[256]; + char line[512]; + int i, wsock, rsock; + struct stat filestat; + + if (!LoadPID (pidfile, &pid, username, machine)) { + fprintf (stderr, "previous elixir not running\n"); + return (TRUE); + } + /* check username matches: can only send signals to own process */ + + /* make connection to remote machine */ + if (!rconnect (machine, CONNECT, &rsock, &wsock)) { + fprintf (stderr, "can't make connection to machine %s to kill process\n", machine); + exit (1); + } + + /* send TERM signal */ + sprintf (line, "kill -TERM %d\n", pid); + write (wsock, line, strlen (line)); + + /* wait for cleanup to finish */ + for (i = 0; i < 100; i++) { + if (stat (pidfile, &filestat) == -1) { + fprintf (stderr, "previous elixir halted\n"); + goto success; + } + usleep (100000); + } + + /* kill meanly */ + sprintf (line, "kill -KILL %d", pid); + write (wsock, line, strlen (line)); + unlink (pidfile); + + /* test if process is still alive? */ + success: + sprintf (line, "exit\n"); + write (wsock, line, strlen(line)); + close (wsock); + close (rsock); + return (TRUE); + +} + +int LoadPID (char *file, pid_t *pid, char *username, char *machine) { + + FILE *f; + + f = fopen (file, "r"); + if (f == (FILE *) NULL) { + return (FALSE); + } + + fscanf (f, "%*s %d", pid); + fscanf (f, "%*s %s", username); + fscanf (f, "%*s %s", machine); + fclose (f); + + return (TRUE); +} Index: /tags/elixir-1-5/Ohana/src/elixir/src/SockScan.c =================================================================== --- /tags/elixir-1-5/Ohana/src/elixir/src/SockScan.c (revision 8558) +++ /tags/elixir-1-5/Ohana/src/elixir/src/SockScan.c (revision 8558) @@ -0,0 +1,53 @@ +# include "controller.h" +# include +# define MAXTIME 10 + +int SockScan (char *string, Fifo *fifo, int sock) { + + int i, done, status; + + done = 0; + for (i = 0; (i < MAXTIME) && !done; i++) { + ShiftFifo (fifo); + status = ReadtoFifo (fifo, sock); + switch (status) { + case 0: + break; + case -1: + usleep (1000); + break; + default: + done = memstr (fifo[0].buffer, string, fifo[0].Nbuffer); + break; + } + } + return (status); +} + +/* returns (offset to m2) + 1, or 0 if failure */ +int memstr (char *m1, char *m2, int n) { + + int i, N; + + N = strlen (m2); + for (i = 0; (i < n - N + 1) && memcmp (m1, m2, N); i++, m1++); + if (memcmp (m1, m2, N)) { + return (0); + } + else { + return (i+1); + } + +} + +/* SockScan reads from the given socket looking for the given string + SockScan stores data read from socket in the fifo buffer to ensure + it catches the given string. a single invocation of SockScan will + wait for up to 10 msec for data to come down the pipe before giving up. + SockScan can only search for one string in the output stream. + */ + + +/* need a way of signalling that the socket has closed... + maybe change the sock entry to 0? +*/ Index: /tags/elixir-1-5/Ohana/src/elixir/src/StartMachine.c =================================================================== --- /tags/elixir-1-5/Ohana/src/elixir/src/StartMachine.c (revision 8558) +++ /tags/elixir-1-5/Ohana/src/elixir/src/StartMachine.c (revision 8558) @@ -0,0 +1,35 @@ +# include "controller.h" + +int StartMachine (Machine *machine, Object *object, int argc, char **argv) { + + int i, Nbyte, Nout; + char command[1024]; + FILE *logfile; + + logfile = LogOpen (object[0].logfile); + machine[0].object = object; + + Nbyte = Nout = 0; + for (i = 0; i < argc; i++) { + Nout = sprintf (&command[Nbyte], "%s ", argv[i]); + Nbyte += Nout; + } + + fprintf (logfile, "%s @ %s: %s\n", object[0].argv[0], machine[0].hostname, command); + if (logfile != stderr) { + fflush (logfile); + fclose (logfile); + } + + Nout = sprintf (&command[Nbyte], "\n echo PROCESS DONE\n"); + write (machine[0].wsock, command, strlen(command)); + + gettimeofday (&machine[0].object[0].start, (void *) NULL); + gettimeofday (&machine[0].quiet, (void *) NULL); + + machine[0].status = BUSY; + return (TRUE); +} + +/* machine[0].quiet times since last message received from process. + if process is too quiet for too long, we assume it died */ Index: /tags/elixir-1-5/Ohana/src/elixir/src/elixir.c =================================================================== --- /tags/elixir-1-5/Ohana/src/elixir/src/elixir.c (revision 8558) +++ /tags/elixir-1-5/Ohana/src/elixir/src/elixir.c (revision 8558) @@ -0,0 +1,126 @@ +# include "controller.h" + +static int Dynamic; +static int Reload; + +static int Nobjects; +static int Nprocess; +static Process **process; +static Process global; + +int main (int argc, char **argv) { + + int i, Nnew, state; + char *config; + int nloop; + char **targv; + + /* save complete arglist incase we reload */ + ALLOCATE (targv, char *, argc + 1); + for (i = 0; i < argc; i++) { + targv[i] = strcreate (argv[i]); + } + targv[i] = (char *) NULL; + + config = ConfigInit (&argc, argv); + + process = DefineProcesses (&global, &Nprocess, config); + + /* PID filename stored in global.argv[7] */ + ConfigPID (global.argv[7]); + + InitMsgFile (global.argv[0]); /* success */ + InitMsgFile (global.argv[1]); /* failure */ + InitMsgFile (global.argv[5]); /* message */ + InitMsgFile (global.argv[6]); /* end stat */ + InitMachines (config); + free (config); + + SetMessageFile (global.argv[5]); + + InitProcessTimers (process, Nprocess); + + Nobjects = 0; + Reload = FALSE; + Dynamic = TRUE; + nloop = 0; + + /* any remaining entry is a file to load in static mode */ + if (argc == 2) { + Dynamic = FALSE; + Nobjects = LoadPending (&global, argv[1], &state, &Dynamic); + if (state) { + fprintf (stderr, "error with input file %s (state %d)\n", argv[1], state); + Shutdown (1); + } + } + + while (CheckEndingState (&global, Nobjects, Dynamic)) { + + if (Dynamic) { + Nnew = LoadPending (&global, global.argv[4], &state, &Dynamic); + Nobjects += Nnew; + } + + for (i = 0; i < Nprocess; i++) { + CheckProcess (process[i]); + } + + for (i = Nprocess - 1; i >= 0; i--) { + CheckProcess (process[i]); + } + + CheckMessages (); + /* CheckMessages (&global, process, Nprocess, &Dynamic, Nobjects); */ + RestartMachines (); + + usleep (1000000); + if (nloop > 3000) { + fprintf (stderr, "."); + nloop = 0; + } + nloop ++; + } + + DumpProcessTimes (global.argv[6]); + + /* DumpStatus (global.argv[6], &global, process, Nprocess, Dynamic, Nobjects); */ + DumpStatus (global.argv[6]); + + WriteMsg (global.argv[6], "DONE"); + + if (Reload) Restart (targv); + + Shutdown (0); + exit (0); +} + +void SIG_RELOAD (int sig) { + fprintf (stderr, "got signal RELOAD\n"); + SetExitTimer (); + Dynamic = FALSE; + Reload = TRUE; +} + +void ElixirStop () { + Dynamic = FALSE; +} + +int GetDynamicState () { + return (Dynamic); +} + +Process **GetProcessInfo (Process **gb, int *np, int *no) { + *np = Nprocess; + *no = Nobjects; + *gb = &global; + return (process); +} + +/* create 'reload' mode: + 0 - Reload = TRUE; + 1 - Dynamic = FALSE; + 2 - start timer (in CheckEndingState.c) + 3 - if timer expires, set exit state + 4 - if (Reload) exec (argc, arg) +*/ Index: /tags/elixir-1-5/Ohana/src/elixir/src/rconnect.c =================================================================== --- /tags/elixir-1-5/Ohana/src/elixir/src/rconnect.c (revision 8558) +++ /tags/elixir-1-5/Ohana/src/elixir/src/rconnect.c (revision 8558) @@ -0,0 +1,100 @@ +# include "controller.h" + +/* rconnect opens a remote shell on hostname and returns two file descriptors: + one for read, one for write */ + +int rconnect (char *hostname, char *command, int *rsock, int *wsock) { + + int i, rfd[2], wfd[2], status; + pid_t pid; + char buffer[0x4000]; + char *file; + Fifo fifo; + + InitFifo (&fifo, 0x4000, 0x1000); + + status = pipe (rfd); + if (status < 0) { + perror ("pipe"); + return (FALSE); + } + status = pipe (wfd); + if (status < 0) { + perror ("pipe"); + return (FALSE); + } + + file = filebasename (command); + pid = fork (); + if (!pid) { /* must be child process */ + fprintf (stderr, "starting remote connection to %s...", hostname); + /* close the excess sockets */ + close (wfd[1]); + close (rfd[0]); + dup2 (wfd[0], STDIN_FILENO); + dup2 (rfd[1], STDOUT_FILENO); + dup2 (rfd[1], STDERR_FILENO); + setvbuf (stdin, (char *) NULL, _IONBF, BUFSIZ); + setvbuf (stdout, (char *) NULL, _IONBF, BUFSIZ); + setvbuf (stderr, (char *) NULL, _IONBF, BUFSIZ); + fprintf (stderr, "child is spawned\n"); + + status = execl (command, file, hostname, "/bin/csh", 0); + fprintf (stderr, "error starting remote shell process\n"); + Shutdown (1); + } + *wsock = wfd[1]; + *rsock = rfd[0]; + close (wfd[0]); + close (rfd[1]); + + fcntl (*rsock, F_SETFL, O_NONBLOCK); + fcntl (*wsock, F_SETFL, O_NONBLOCK); + + sprintf (buffer, "echo PTOLEMY STARTED\n"); + status = write (*wsock, buffer, strlen(buffer)); + if ((status == -1) && (errno == EPIPE)) { + fprintf (stderr, "socket closed unexpectedly\n"); + close (*wsock); + close (*rsock); + return (FALSE); + } + + /* try to get evidence connection is alive - wait upto a few seconds */ + status = -1; + for (i = 0; (i < 300) && (status == -1); i++) { + fcntl (*rsock, F_SETFL, O_NONBLOCK); + status = SockScan ("PTOLEMY STARTED", &fifo, *rsock); + if (!(i % 30)) fprintf (stderr, "."); + if (status == 0) { + fprintf (stderr, "socket closed unexpectedly\n"); + close (*wsock); + close (*rsock); + return (FALSE); + } + } + if (status == -1) { + fprintf (stderr, "timeout while connecting\n"); + close (*wsock); + close (*rsock); + return (FALSE); + } + fprintf (stderr, "Connected\n"); + + /* the onintr command works with csh/tcsh type shells to + prevent trapping of SIGTERM, used to kill hung jobs. + for bash/sh type shells, you can use SIGQUIT instead + without setting onintr */ + sprintf (buffer, "onintr\n"); + write (*wsock, buffer, strlen(buffer)); + if ((status == -1) && (errno == EPIPE)) { + fprintf (stderr, "socket closed unexpectedly\n"); + close (*wsock); + close (*rsock); + return (FALSE); + } + + FreeFifo (&fifo); + return (pid); + +} Index: /tags/elixir-1-5/Ohana/src/elixir/test/README =================================================================== --- /tags/elixir-1-5/Ohana/src/elixir/test/README (revision 8558) +++ /tags/elixir-1-5/Ohana/src/elixir/test/README (revision 8558) @@ -0,0 +1,149 @@ + +This directory includes a very simple elixir script, test.pro, to test +that elixir is working correctly. + +To run it, create an input list with one element per line. The +elements can be any word; an example is provided in test.list. + +Make sure the machines listed above represent valid machine names in +your cluster. Also, be certain that you can rsh to these machines +without a password. + +Run the test by exectuting: + +elixir -c test.pro test.list + +the result of this test will be several files in your home directory: + +elixir-test.log +elixir-test.end +elixir-test.success + +elixir-test.log: + +The first file, elixir-test.log, shows the output from all of the +processing steps. In this test example, the log file is defined to be +a single file, but it is also possible to use the &0 style +command-line arguments to create a different file for each input line. +All output from each of the processes goes into this file in the order +received by the elixir program. The resulting lines may look a bit +confusing sometime as a result. In addition, elixir adds some extra +information to aid in debugging. Here are a few example lines: + +foo @ kiawe: ls foo ; echo SUCCESS +bar @ milo: ls bar ; echo SUCCESS +foo @ kiawe:ls: foo: No such file or directory +SUCCESS +PROCESS DONE +foo @ kiawe is done +bar @ milo:ls: bar: No such file or directory +SUCCESS +PROCESS DONE +bar @ milo is done + +Every output line from the processes are prepended with: + +(&0) @ (machine): + +In this example, I used the two machines, kiawe & milo. The first two +lines show the first two commands being executed: + +ls foo ; echo SUCCESS + +The next three lines show the output from the process executing on foo: + +ls: foo: No such file or directory +SUCCESS +PROCESS DONE + +The 'ls' command did not find a file 'foo' in the home directory - +note that all commands, since they are run remotely, execute in the +user's home directory. The word "SUCCESS" comes from the 'echo +SUCCESS' portion of the command. Currently, elixir uses the words +SUCCESS, ERROR, and PROCESS DONE for process monitoring. Ideally, the +program called by elixir should check its output state and echo either +SUCCESS or ERROR as appropriate. In this test example, we simply +force all processes to claim they succeeded with the 'echo SUCCESS' +command. The PROCESS DONE is added by elixir when it creates the +command. This lets elixir detect programs which crash and thereby +fail to send either SUCCESS or ERROR. + +Side Note: this SUCCESS / ERROR flow control construct will soon be +replaced by tests of the actual process exit status. This will make +elixir able to more cleanly support unix commnands. + +The 6th line in the example above is the output from elixir noting +that the process for 'foo' ended. The next few lines show the output +for the process on 'bar'. + +elixir-test.end: + +This file is created at the end of the elixir run and provides +information about the run. This same status information can be +retrieved from elixir while it is running by using the 'status' +program. Here is the output from the above example: + +################################################################## + . test1 test2 test3 + kiawe 0.120 3 5.056 2 0.130 1 + milo 0.110 1 5.062 2 0.109 3 +processes status: + name pending success failure total +---------------------------------------------------------------- + global 0 4 0 4 + test1 0 + test2 0 + test3 0 + +machine status: + name process status +---------------------------------------------------------------- + kiawe (none) 0 + milo (none) 0 + +objects loaded so far: 4 + +NOT accepting input from FIFO +DONE +################################################################## + +The first section shows a grid with processes across the top and +machines in the left column. The entries for each pair of machine & +process are the average execution time and the number of instances of +the given process which was run on the given machine. So, in this +example, the process 'test1' (which is the 'ls') was run on 'kiawe' 3 +times for an average of 0.120 sec and on 'milo' for an average of +0.110 sec. + +The second section shows the state of the different queues when the +elixir process ended. The left hand column shows the possible queues +(global, test1, test2, test3). For all nodes except the global node, +the success and failure queues are always linked to a queue of another +node. Jobs which land in the success or failure queue of a node are +instantly moved to the pending queue of another job. Therefore, we +only show the pending queue for the nodes except for 'global'. Each +number represents the number of jobs waiting in the given queue. +(note: the global.pending queue is identical to one of the other +pending queues and therefore represents a duplication). The 'total' +entry is the sum of jobs in each of the global queues. Note that this +table does not show the processes which are actually being executed, +as these are considered to be 'in' a node, not on the queue of a node. +They are represented in the next section. + +The third section lists the status of all machines available. The +first column is the machine name; the second is the name of the +process being executed, or '(none)' if the machine is idle; the third +is a numerical status indication (which is generally ignored). + +The next line shows the number of objects loaded by elixir. Since +elixir has the capability of loading new objects dynamically, this +value on this line may increment as time goes by. The next line shows +whether elixir is accepting input during runtime or not. + +elixir-test.success: + +This file lists the processes which elixir successfully finished. (An +equivalent file, elixir-test.failure, would include jobs which did not +succeed). Entries are added to this file as the jobs complete. In +this way, it can be used to monitor the process of an elixir run. + Index: /tags/elixir-1-5/Ohana/src/elixir/test/save.source =================================================================== --- /tags/elixir-1-5/Ohana/src/elixir/test/save.source (revision 8558) +++ /tags/elixir-1-5/Ohana/src/elixir/test/save.source (revision 8558) @@ -0,0 +1,4 @@ +msg1 +msg2 +msg3 +msg4 Index: /tags/elixir-1-5/Ohana/src/elixir/test/test.end =================================================================== --- /tags/elixir-1-5/Ohana/src/elixir/test/test.end (revision 8558) +++ /tags/elixir-1-5/Ohana/src/elixir/test/test.end (revision 8558) @@ -0,0 +1,21 @@ + . test1 test2 test3 + localhost 1.565 2 5.180 2 1.030 2 + localhost 1.575 2 5.180 2 1.030 2 +processes status: + name pending success failure total +---------------------------------------------------------------- + global 0 4 0 4 + test1 0 + test2 0 + test3 0 + +machine status: + name process status +---------------------------------------------------------------- + localhost (none) 0 + localhost (none) 0 + +objects loaded so far: 4 + +NOT accepting input from FIFO +DONE Index: /tags/elixir-1-5/Ohana/src/elixir/test/test.pro =================================================================== --- /tags/elixir-1-5/Ohana/src/elixir/test/test.pro (revision 8558) +++ /tags/elixir-1-5/Ohana/src/elixir/test/test.pro (revision 8558) @@ -0,0 +1,54 @@ +# Elixir test config file + +CONNECT /usr/bin/ssh +MACHINE localhost +MACHINE localhost +elixir test + +# FIFO files +global.success $elixir.success +global.failure $elixir.failure +global.source $elixir.source +global.msg $elixir.msg +global.end $elixir.end +global.pid $elixir.pid + +# process definition +global.Nargs 1 +global.logfile 0 %s $elixir.log +global.pending test1 +global.timeout 100 + +# test1: ls (will ls &0 user home dir) +process test1 +test1.arg 0 ls +test1.arg 0 %s &0 +test1.arg 0 ; +test1.arg 0 echo +test1.arg 0 SUCCESS +test1.success test2 +test1.failure global + +# test2 sleep 5 +process test2 +test2.arg 0 sleep +test2.arg 0 5 +test2.arg 0 ; +test2.arg 0 echo +test2.arg 0 SUCCESS +test2.success test3 +test2.failure global + +# test3 echo &0 +process test3 +test3.arg 0 echo +test3.arg 0 %s &0 +test3.arg 0 ; +test3.arg 0 echo +test3.arg 0 SUCCESS +test3.success global +test3.failure global + +# this is a very simple elixir script to test that elixir is working +# correctly. See the README file for instructions. + Index: /tags/elixir-1-5/Ohana/src/elixir/test/test.success =================================================================== --- /tags/elixir-1-5/Ohana/src/elixir/test/test.success (revision 8558) +++ /tags/elixir-1-5/Ohana/src/elixir/test/test.success (revision 8558) @@ -0,0 +1,4 @@ +msg1 0000005d test3 +msg2 0000005d test3 +msg3 0000005d test3 +msg4 0000005d test3