Index: /trunk/Ohana/src/opihi/doc/pcontrol.txt =================================================================== --- /trunk/Ohana/src/opihi/doc/pcontrol.txt (revision 4449) +++ /trunk/Ohana/src/opihi/doc/pcontrol.txt (revision 4450) @@ -80,5 +80,4 @@ can set the CheckTask function to this hook (& unset it). - --- @@ -116,5 +115,41 @@ status -queues +pcontrol may be given a timeout for each job. pcontrol will monitor a +job and kill/crash it if the timeout expires. the timeout only +governs how long it is allowed to execute, not how long it can sit in +the queue. (the scheduler / operator should decide if a job has been +on pcontrol for too long -- this probably means there are no +appropriate machines ). + +pcontrol currently does not distinguish between multiple instances of +a single host. all have the same name. if you want to bring down a +host, you need to issue N host -down commands. perhaps this is +silly. a simple alternative would be for the host [-on -off -start +-stop] commands to apply to all defined entries which match the +hostname. In this case, a command like 'host foo -off' would find and +halt all connections to the machine 'foo', while 'host foo -on' would +restart them all (or rather, given the functionality of pcontrol, +would allow pcontrol to attempt to bring them on). + +It is not clear why a user should be able to execute 'start' (down -> +idle) or 'stop' (idle -> down). The transition down -> idle is +automatically performed by pcontrol for any machines which are +currently down, while the transition idle -> down is immediately +followed by an attempt by pcontrol to move the host from down -> idle. + +does it makes sense to kill all jobs on a host? this would only have +the effect of clearing the host for a moment until pcontrol decided to +start another job on that host. the desired effect is gained putting +the host to 'off'. +currently the command 'host (hostname)' puts the host in 'down' +state. pcontrol then immediately tries to connect to the host, moving +it to 'idle' state (and then 'busy' if any jobs are available). it +might be useful to be able to add a host in 'off' state as a starting +point. + +it is not obvious that the user should be able to run 'CheckHost', +unless this gets expanded to return state information on the host. + --- @@ -137,6 +172,7 @@ PENDING -> BUSY : StartJob PENDING -> DEL : DelJob -BUSY -> EXIT : CheckJob -BUSY -> CRASH : CheckJob | KillJob +BUSY -> DONE : CheckBusyJob | KillJob +DONE -> EXIT : CheckDoneJob +DONE -> CRASH : CheckDoneJob BUSY -> PENDING : CheckJob | CheckHost EXIT -> DEL : DelJob Index: /trunk/Ohana/src/opihi/doc/scheduler.txt =================================================================== --- /trunk/Ohana/src/opihi/doc/scheduler.txt (revision 4449) +++ /trunk/Ohana/src/opihi/doc/scheduler.txt (revision 4450) @@ -1,2 +1,14 @@ + +--- + +sched / pcontrol todo: + +- sched: validate task hosts with controller +- pcontrol: host (name) -check should return state +- sched: start / stop controller connection? +- pcontrol: add command to check status of a job +- pcontrol: add command to dump stdout/stderr for a job + +--- scheduler commands: @@ -24,5 +36,5 @@ (in task, command line is static; in task.macro, command line is expanded for each instance) -host (machine) +host (machine) [-required] - defines preferred host - may be in task or in task.macro (exit/exec) @@ -61,4 +73,26 @@ --- +scheduler / controller / local interactions + +scheduler can execute local jobs and jobs on the controller. +scheduler needs to initiate the connection to the controller, if the +controller is being used. it also needs to fork each local command. + +we need a command to define the controller hosts to be used. if a +task or job defines a host which is not known, how is this caught? +does sched need to keep a list of valid hosts, or should the host be +passed down to the controller. Also, the scheduler should not +initiate a connection to the controller unless it is requested. is +this implicit when defining a host? + +- controller host foobar + + this should setup the controller interface (if it does not currently + exist), and send the command 'host foobar'. scheduler should not + care if the connection is successful or not (we can know about hosts + which are currently off). + + +--- controller interaction: Index: /trunk/Ohana/src/opihi/include/pcontrol.h =================================================================== --- /trunk/Ohana/src/opihi/include/pcontrol.h (revision 4449) +++ /trunk/Ohana/src/opihi/include/pcontrol.h (revision 4450) @@ -72,4 +72,6 @@ int exit_status; int Reset; + int stdout_size; + int stderr_size; JobMode mode; JobStat state; @@ -128,4 +130,7 @@ Host *GetHost (int StackID, int where); int FindHost (IDtype HostID, int StackID); +Host *FindHostPtr (IDtype HostID, int StackID); +Host *FindHostStack (IDtype HostID); +Host *PullHost (IDtype HostID, int StackID); int FindNamedHost (char *name, int StackID); void AddHost (char *hostname); @@ -149,5 +154,5 @@ int CheckDoneHost (Host *host); int CheckDoneJob (Job *job); -int GetJobOutput (Job *job, char *channel); +int GetJobOutput (char *command, Host *host, IOBuffer *buffer, int Nbytes); int ResetJob (Job *job); int PclientCommand (Host *host, char *command, char *response, IOBuffer *buffer); @@ -159,6 +164,9 @@ Job *GetJob (int StackID, int where); int FindJob (IDtype JobID, int StackID); +Job *FindJobPtr (IDtype JobID, int StackID); +Job *FindJobStack (IDtype JobID); +Job *PullJob (IDtype JobID, int StackID); IDtype AddJob (char *hostname, JobMode mode, int timeout, int argc, char **argv); -void KillJob (Job *job); +int KillJob (Job *job); void DelJob (Job *job); Host *UnlinkJobAndHost (Job *job); Index: /trunk/Ohana/src/opihi/include/scheduler.h =================================================================== --- /trunk/Ohana/src/opihi/include/scheduler.h (revision 4449) +++ /trunk/Ohana/src/opihi/include/scheduler.h (revision 4450) @@ -1,4 +1,37 @@ +# include "external.h" +# include "shell.h" +# include "dvomath.h" +# include "convert.h" +# include "display.h" +# include +# include -enum {JOB_BUSY, JOB_EXIT, JOB_CRASH}; +typedef enum { + JOB_NONE, + JOB_BUSY, + JOB_EXIT, + JOB_CRASH, + JOB_PENDING, +} JobStat; + +typedef enum { + JOB_LOCAL, + JOB_CONTROLLER, +} JobMode; + +typedef enum { + CONTROLLER_HUNG = -1, + CONTROLLER_DOWN = 0, + CONTROLLER_GOOD = 1, +} ControllerStat; + +/* socket / pipe communication buffer */ +typedef struct { + char *buffer; + int Nalloc; + int Nreset; + int Nblock; + int Nbuffer; +} IOBuffer; /* a task is a description of the wrapping of a job */ @@ -16,4 +49,5 @@ char **argv; char *host; + int host_required; char *name; @@ -28,7 +62,5 @@ typedef struct { int JobID; /* internal ID for job */ - int PID; /* external ID for job */ - - Task task; + int pid; /* external ID for job */ struct timeval last; @@ -36,6 +68,20 @@ int state; int exit_status; + + int argc; + char **argv; + Task *task; + + IOBuffer stdout; /* stdout storage buffer */ + IOBuffer stderr; /* stderr storage buffer */ + JobMode mode; /* local or controller? */ + + int stdout_size; /* stdout pipe (local only) */ + int stderr_size; /* stderr pipe (local only) */ + int stdout_fd; /* stdout pipe (local only) */ + int stderr_fd; /* stderr pipe (local only) */ } Job; +# define CONTROLLER_PROMPT "pclient:" /* scheduler prototypes */ @@ -50,4 +96,5 @@ int TaskHash (char *input); int ShowTask (char *name); +char *memstr (char *m1, char *m2, int n); Job *NextJob (); @@ -65,2 +112,16 @@ void SetTaskTimer (struct timeval *timer); double GetTaskTimer (struct timeval start); + +int CheckJobs (); +int CheckTasks (); +int CheckSystem (); +int GetJobOutput (char *channel, int pid, IOBuffer *buffer, int Nbytes); +int CheckControllerJob (Job *job); +int CheckControllerJobStatus (Job *job); +int SubmitControllerJob (Job *job); +int StartController (); +int ControllerCommand (char *command, char *response, IOBuffer *buffer); +int SubmitLocalJob (Job *job); +int CheckLocalJob (Job *job); +int CheckLocalJobStatus (Job *job); +void InitTaskTimers (); Index: /trunk/Ohana/src/opihi/pantasks/CheckJobs.c =================================================================== --- /trunk/Ohana/src/opihi/pantasks/CheckJobs.c (revision 4450) +++ /trunk/Ohana/src/opihi/pantasks/CheckJobs.c (revision 4450) @@ -0,0 +1,74 @@ +# include "scheduler.h" + +int CheckJobs () { + + Job *job; + Task *task; + Macro *macro; + int i, found, status; + int Ntest; + + /** test all jobs: ready to test? finished? **/ + while ((job = NextJob ()) != NULL) { + + /* check for timeout */ + if (GetTaskTimer(job[0].start) >= job[0].task[0].timeout_period) { + fprintf (stderr, "timeout on %s\n", job[0].task[0].name); + /* run task[0].timeout macro, if it exists */ + if (job[0].task[0].timeout != NULL) { + exec_loop (job[0].task[0].timeout); + } + DeleteJob (job); + continue; + } + + /* check poll period (ready to run again?) */ + if (GetTaskTimer(job[0].last) < job[0].task[0].poll_period) continue; + + /* check current status */ + status = CheckJob (job); + switch (status) { + case JOB_BUSY: + fprintf (stderr, "job %s (%d) busy\n", job[0].task[0].name, job[0].JobID); + break; + + case JOB_CRASH: + fprintf (stderr, "job %s (%d) crash\n", job[0].task[0].name, job[0].JobID); + /* run task[0].crash macro, if it exists */ + /* set the stdout and stderr variables with job.stdout, job.stderr */ + /* XXX this will break on 0 values in output streams */ + set_str_variable ("stdout", job[0].stdout.buffer); + set_str_variable ("stderr", job[0].stderr.buffer); + if (job[0].task[0].crash != NULL) { + exec_loop (job[0].task[0].crash); + } + DeleteJob (job); + continue; + break; + + case JOB_EXIT: + fprintf (stderr, "job %s (%d) exit\n", job[0].task[0].name, job[0].JobID); + /* run corresponding task[0].exit macro, if it exists */ + macro = job[0].task[0].def; + for (i = 0; i < job[0].task[0].Nexit; i++) { + if (job[0].exit_status == atoi(job[0].task[0].exit[i][0].name)) { + macro = job[0].task[0].exit[i]; + break; + } + } + if (macro != NULL) exec_loop (macro); + DeleteJob (job); + continue; + break; + + default: + fprintf (stderr, "unknown exit status\n"); + /** do something more useful here ?? **/ + break; + } + + /* reset polling clock */ + SetTaskTimer (&job[0].last); + } + return (TRUE); +} Index: /trunk/Ohana/src/opihi/pantasks/CheckSystem.c =================================================================== --- /trunk/Ohana/src/opihi/pantasks/CheckSystem.c (revision 4450) +++ /trunk/Ohana/src/opihi/pantasks/CheckSystem.c (revision 4450) @@ -0,0 +1,8 @@ +# include "scheduler.h" + +int CheckSystem () { + + CheckTasks (); + CheckJobs (); + +} Index: /trunk/Ohana/src/opihi/pantasks/CheckTasks.c =================================================================== --- /trunk/Ohana/src/opihi/pantasks/CheckTasks.c (revision 4450) +++ /trunk/Ohana/src/opihi/pantasks/CheckTasks.c (revision 4450) @@ -0,0 +1,39 @@ +# include "scheduler.h" + +int CheckTasks () { + + Job *job; + Task *task; + Macro *macro; + int i, found, status; + int Ntest; + + /** test all tasks: ready to test? ready to run? **/ + while ((task = NextTask ()) != NULL) { + + /* ready to test? : check exec period */ + if (GetTaskTimer(task[0].last) < task[0].exec_period) continue; + + SetCurrentTask (task[0].name); + fprintf (stderr, "trying task %s\n", task[0].name); + + /* ready to run? : run task.exec macro */ + if (task[0].exec != NULL) { + status = exec_loop (task[0].exec); + if (!status) continue; + } + + /* is task valid? check state of task.(argc, argv) */ + /*** ADD CODE HERE ***/ + + /* construct job from task */ + job = CreateJob (task); + + /* execute job - XXX add status test */ + SubmitJob (job); + + /* reset timer on task (don't do this if Create/Submit fails)*/ + gettimeofday (&task[0].last, (void *) NULL); + } + return (TRUE); +} Index: /trunk/Ohana/src/opihi/pantasks/ControllerOps.c =================================================================== --- /trunk/Ohana/src/opihi/pantasks/ControllerOps.c (revision 4450) +++ /trunk/Ohana/src/opihi/pantasks/ControllerOps.c (revision 4450) @@ -0,0 +1,322 @@ +# include "scheduler.h" +# define CONTROLLER_TIMEOUT 20 +# define CONNECT_TIMEOUT 300 + +int status = FALSE; +int stdin_cntl, stdout_cntl, stderr_cntl; + +/* we read Nbytes from the host, then watch for the prompt */ +int GetJobOutput (char *cmd, int pid, IOBuffer *buffer, int Nbytes) { + + int i, status, Nstart; + char *line; + + /* flush any earlier messages */ + ReadtoIOBuffer (buffer, stdout_cntl); + FlushIOBuffer (buffer); + Nstart = buffer[0].Nbuffer; + + /* send command to get appropriate channel */ + ALLOCATE (line, char, MAX (1, strlen(cmd) + 15)); + sprintf (line, "%s %d\n", cmd, pid); + status = write (stdin_cntl, line, strlen(line)); + free (line); + + /* is pipe still open? */ + if ((status == -1) && (errno == EPIPE)) return (CONTROLLER_DOWN); + + /* read at least Nbytes, then watch for CONTROLLER_PROMPT */ + line = NULL; + status = -1; + for (i = 0; (i < CONTROLLER_TIMEOUT) && (status == -1) && (line == NULL); i++) { + status = ReadtoIOBuffer (buffer, stdout_cntl); + if ((buffer[0].Nbuffer - Nstart) >= Nbytes) { + line = memstr (buffer[0].buffer, CONTROLLER_PROMPT, buffer[0].Nbuffer); + } + if (status == -1) usleep (10000); + } + if (status == 0) return (CONTROLLER_DOWN); + if (status == -1) return (CONTROLLER_HUNG); + + fprintf (stderr, "message received (GetJobOutput : %s)\n", cmd); + /* drop extra bytes from pcontrol (not pclient:job) */ + buffer[0].Nbuffer = Nstart + Nbytes; + if (buffer[0].Nalloc > buffer[0].Nbuffer) { + bzero (buffer[0].buffer + buffer[0].Nbuffer, buffer[0].Nalloc - buffer[0].Nbuffer); + } + return (CONTROLLER_GOOD); +} + +int CheckControllerJob (Job *job) { + + CheckControllerJobStatus (job); + + if ((job[0].state == JOB_EXIT) || (job[0].state == JOB_CRASH)) { + GetJobOutput ("stdout", job[0].pid, &job[0].stdout, job[0].stdout_size); + GetJobOutput ("stderr", job[0].pid, &job[0].stderr, job[0].stderr_size); + } +} + +int CheckControllerJobStatus (Job *job) { + + int outstate; + char cmd[128], status_string[64]; + char *p; + IOBuffer buffer; + + + sprintf (cmd, "check job %d", job[0].pid); + InitIOBuffer (&buffer, 0x100); + status = ControllerCommand (cmd, CONTROLLER_PROMPT, &buffer); + + /* check on success of controller command */ + switch (status) { + case CONTROLLER_DOWN: + fprintf (stderr, "controller is down\n"); + FreeIOBuffer (&buffer); + return (FALSE); + + case CONTROLLER_HUNG: + fprintf (stderr, "controller is not responding\n"); + FreeIOBuffer (&buffer); + return (FALSE); + + case CONTROLLER_GOOD: + fprintf (stderr, "message received (CheckControllerJobStatus)\n"); + break; + + default: + fprintf (stderr, "unknown status for controller command: programming error\n"); + exit (1); + } + + /** was this a valid job? **/ + p = memstr (buffer.buffer, "job not found", buffer.Nbuffer); + if (p != NULL) { + fprintf (stderr, "unknown job %d\n", job[0].pid); + return (FALSE); + } + + /** parse status message **/ + p = memstr (buffer.buffer, "STATUS", buffer.Nbuffer); + sscanf (p, "%*s %s", status_string); + p = memstr (buffer.buffer, "EXITST", buffer.Nbuffer); + sscanf (p, "%*s %d", &job[0].exit_status); + p = memstr (buffer.buffer, "STDOUT", buffer.Nbuffer); + sscanf (p, "%*s %d", &job[0].stdout_size); + p = memstr (buffer.buffer, "STDERR", buffer.Nbuffer); + sscanf (p, "%*s %d", &job[0].stderr_size); + FreeIOBuffer (&buffer); + + /* possible exit status values */ + outstate = -1; + if (!strcmp(status_string, "BUSY")) outstate = JOB_BUSY; + if (!strcmp(status_string, "DONE")) outstate = JOB_BUSY; + if (!strcmp(status_string, "PENDING")) outstate = JOB_PENDING; + if (!strcmp(status_string, "EXIT")) outstate = JOB_EXIT; + if (!strcmp(status_string, "CRASH")) outstate = JOB_CRASH; + if (outstate == -1) { + fprintf (stderr, "programming error?\n"); + exit (1); + } + job[0].state = outstate; + return (TRUE); +} + +/* submitting a job to the controller automatically starts controller */ +int SubmitControllerJob (Job *job) { + + int i, Nchar; + char *cmd, *p, string[64]; + IOBuffer buffer; + + if (job[0].task[0].host == NULL) return (FALSE); + + StartController (); // XXX check for success + + /** construct the line to be sent to the controller **/ + + /* determine the total line length */ + Nchar = 0; + for (i = 0; i < job[0].argc; i++) { + Nchar += strlen (job[0].argv[i]) + 1; + } + if (job[0].task[0].host) { + Nchar += strlen (job[0].task[0].host) + 1; + } + Nchar += 10; + ALLOCATE (cmd, char, Nchar); + bzero (cmd, Nchar); + + /* construct the controller command portion */ + if (!strcmp (job[0].task[0].host, "NONE")) { + sprintf (cmd, "job", cmd); + } else { + if (job[0].task[0].host_required) { + sprintf (cmd, "job +host %s", job[0].task[0].host); + } else { + sprintf (cmd, "job -host %s", job[0].task[0].host); + } + } + + /* add the command arguments */ + for (i = 0; i < job[0].task[0].argc; i++) { + strcat (cmd, " "); + strcat (cmd, job[0].task[0].argv[i]); + } + + InitIOBuffer (&buffer, 0x100); + ControllerCommand (cmd, CONTROLLER_PROMPT, &buffer); + free (cmd); + + /* extract the job PID from the controller response */ + p = memstr (buffer.buffer, "STATUS", buffer.Nbuffer); + if (p == NULL) { + fprintf (stderr, "missing PID in pcontrol message : programming error\n"); + exit (1); + } + sscanf (p, "%*s %s", string); + job[0].pid = atoi (string); + FreeIOBuffer (&buffer); + return (TRUE); +} + +int StartController () { + + char *p; + char **argv, cmd[128]; + int i, pid; + int stdin_fd[2], stdout_fd[2], stderr_fd[2]; + IOBuffer buffer; + + if (status) return (TRUE); + + if (VarConfig ("CONTROLLER", "%s", cmd) == NULL) strcpy (cmd, "pcontrol"); + + bzero (stdin_fd, 2*sizeof(int)); + bzero (stdout_fd, 2*sizeof(int)); + bzero (stderr_fd, 2*sizeof(int)); + + if (pipe (stdin_fd) < 0) goto pipe_error; + if (pipe (stdout_fd) < 0) goto pipe_error; + if (pipe (stderr_fd) < 0) goto pipe_error; + + ALLOCATE (argv, char *, 2); + argv[0] = cmd; + argv[1] = 0; + + pid = fork (); + if (!pid) { /* must be child process */ + fprintf (stderr, "starting controller connection\n"); + + /* close the other ends of the pipes */ + close (stdin_fd[1]); + close (stdout_fd[0]); + close (stderr_fd[0]); + + /* tie our ends of the pipes to stdin, stdout, stderr */ + dup2 (stdin_fd[0], STDIN_FILENO); + dup2 (stdout_fd[1], STDOUT_FILENO); + dup2 (stderr_fd[1], STDERR_FILENO); + + /* set all three unblocking */ + setvbuf (stdin, (char *) NULL, _IONBF, BUFSIZ); + setvbuf (stdout, (char *) NULL, _IONBF, BUFSIZ); + setvbuf (stderr, (char *) NULL, _IONBF, BUFSIZ); + + status = execvp (argv[0], argv); + exit (1); + } + free (argv); + + /* close the other ends of the pipes */ + close (stdin_fd[0]); stdin_fd[0] = 0; + close (stdout_fd[1]); stdout_fd[1] = 0; + close (stderr_fd[1]); stderr_fd[1] = 0; + + /* make the pipes non-blocking */ + fcntl (stdin_fd[1], F_SETFL, O_NONBLOCK); + fcntl (stdout_fd[0], F_SETFL, O_NONBLOCK); + fcntl (stderr_fd[0], F_SETFL, O_NONBLOCK); + + /* perform handshake with controller to verify alive & running */ + /** this handshake is similar to ControllerCommand, but has important differences **/ + InitIOBuffer (&buffer, 0x100); + + /* send handshake command */ + sprintf (cmd, "echo CONNECTED\n"); + status = write (stdin_fd[1], cmd, strlen(cmd)); + if ((status == -1) && (errno == EPIPE)) goto pipe_error; + + /* try to get evidence connection is alive - wait upto a few seconds */ + p = NULL; + status = -1; + for (i = 0; (i < CONNECT_TIMEOUT) && (status == -1) && (p == NULL); i++) { + status = ReadtoIOBuffer (&buffer, stdout_fd[0]); + p = memstr (buffer.buffer, "CONNECTED", buffer.Nbuffer); + usleep (10000); + } + if (status == 0) goto pipe_error; + if (status == -1) goto io_error; + FreeIOBuffer (&buffer); + + /* set local static vars to pipe connections */ + stdin_cntl = stdin_fd[1]; + stdout_cntl = stdout_fd[0]; + stderr_cntl = stderr_fd[0]; + + fprintf (stderr, "Connected\n"); + return (TRUE); + +pipe_error: + perror ("pipe error:"); + goto close_pipes; + +io_error: + fprintf (stderr, "timeout while connecting\n"); + goto close_pipes; + +close_pipes: + if (stdin_fd[0] != 0) close (stdin_fd[0]); + if (stdin_fd[1] != 0) close (stdin_fd[1]); + if (stdout_fd[0] != 0) close (stdout_fd[0]); + if (stdout_fd[1] != 0) close (stdout_fd[1]); + if (stderr_fd[0] != 0) close (stderr_fd[0]); + if (stderr_fd[1] != 0) close (stderr_fd[1]); + return (FALSE); +} + +int ControllerCommand (char *cmd, char *response, IOBuffer *buffer) { + + int i; + int status; + char *line; + + ReadtoIOBuffer (buffer, stdout_cntl); + FlushIOBuffer (buffer); + + /* send command to client (adding on \n) */ + /* fprintf (stderr, "send: %s (%d)\n", command, buffer[0].Nbuffer); */ + ALLOCATE (line, char, MAX (1, strlen(cmd))); + sprintf (line, "%s\n", cmd); + status = write (stdin_cntl, line, strlen(line)); + free (line); + + /* is pipe still open? */ + if ((status == -1) && (errno == EPIPE)) return (CONTROLLER_DOWN); + + /* watch for response - wait up to 1 second */ + line = NULL; + status = -1; + for (i = 0; (i < CONTROLLER_TIMEOUT) && (status == -1) && (line == NULL); i++) { + status = ReadtoIOBuffer (buffer, stdout_cntl); + line = memstr (buffer[0].buffer, response, buffer[0].Nbuffer); + if (status == -1) usleep (10000); + } + if (status == 0) return (CONTROLLER_DOWN); + if (status == -1) return (CONTROLLER_HUNG); + /* fprintf (stderr, "buffer.buffer: %s\n", buffer[0].buffer); */ + return (CONTROLLER_GOOD); +} + +/* memstr returns a view, not an allocated string : don't free */ Index: /trunk/Ohana/src/opihi/pantasks/IOBufferOps.c =================================================================== --- /trunk/Ohana/src/opihi/pantasks/IOBufferOps.c (revision 4450) +++ /trunk/Ohana/src/opihi/pantasks/IOBufferOps.c (revision 4450) @@ -0,0 +1,68 @@ +# include "pclient.h" + +int InitIOBuffer (IOBuffer *buffer, int Nalloc) { + + buffer[0].Nalloc = Nalloc; + buffer[0].Nreset = Nalloc; + buffer[0].Nblock = Nalloc / 2; + buffer[0].Nbuffer = 0; + + ALLOCATE (buffer[0].buffer, char, buffer[0].Nalloc); + bzero (buffer[0].buffer, buffer[0].Nalloc); + + return (TRUE); +} + +int FlushIOBuffer (IOBuffer *buffer) { + + buffer[0].Nbuffer = 0; + buffer[0].Nalloc = buffer[0].Nreset; + REALLOCATE (buffer[0].buffer, char, buffer[0].Nalloc); + + return (TRUE); +} + +int ReadtoIOBuffer (IOBuffer *buffer, int fd) { + + int Nread, Nfree; + + if (fd == 0) { + fprintf (stderr, "pipe is closed (ReadIOBuffer)\n"); + return (0); + /* pipe is closed */ + } + + Nfree = buffer[0].Nalloc - buffer[0].Nbuffer; + if (Nfree < buffer[0].Nblock) { + buffer[0].Nalloc += 2*buffer[0].Nblock; + REALLOCATE (buffer[0].buffer, char, buffer[0].Nalloc); + Nfree = buffer[0].Nalloc - buffer[0].Nbuffer; + } + + Nread = read (fd, &buffer[0].buffer[buffer[0].Nbuffer], buffer[0].Nblock); + + if (Nread >= 0) { + buffer[0].Nbuffer += Nread; + return (Nread); + } + + if (Nread == -1) { + switch (errno) { + case EAGAIN: + case EIO: + /** no data available in pipe **/ + return (-1); + default: + /** error reading from pipe **/ + perror ("ReadtoIOBuffer read error"); + return (-2); + } + } +} + +void FreeIOBuffer (IOBuffer *buffer) { + + if (buffer[0].buffer != (char *) NULL) { + free (buffer[0].buffer); + } +} Index: /trunk/Ohana/src/opihi/pantasks/JobOps.c =================================================================== --- /trunk/Ohana/src/opihi/pantasks/JobOps.c (revision 4449) +++ /trunk/Ohana/src/opihi/pantasks/JobOps.c (revision 4450) @@ -1,3 +1,2 @@ -# include "opihi.h" # include "scheduler.h" @@ -9,6 +8,6 @@ static Job *jobs; -static int Njobs; -static int NJOBS; +static int Njobs; +static int NJOBS; static char *JobName = dot; @@ -65,5 +64,5 @@ for (i = 0; i < Njobs; i++) { - fprintf (stderr, "%d: %-15s %5d %20s (%x)\n", Njobs, jobs[i].task.name, jobs[i].JobID, jobs[i].task.argv[0], jobs[i].task.argv); + fprintf (stderr, "%d: %-15s %5d %20s (%x)\n", Njobs, jobs[i].task[0].name, jobs[i].JobID, jobs[i].argv[0], jobs[i].argv); } return; @@ -75,5 +74,5 @@ int i; - /* try for an exact match first */ + /* return job with matching JobID */ for (i = 0; i < Njobs; i++) { if (jobs[i].JobID == JobID) { @@ -90,13 +89,31 @@ jobs[Njobs].JobID = NextJobID (); - jobs[Njobs].PID = 0; - jobs[Njobs].task = task[0]; - - /* we need our own copy of task[0].argv */ - ALLOCATE (jobs[Njobs].task.argv, char *, MAX (jobs[Njobs].task.argc, 1)); - for (i = 0; i < jobs[Njobs].task.argc; i++) { - jobs[Njobs].task.argv[i] = strcreate (task[0].argv[i]); - } - jobs[Njobs].task.host = strcreate (task[0].host); + jobs[Njobs].pid = 0; + jobs[Njobs].mode = JOB_LOCAL; + if (task[0].host != NULL) { + jobs[Njobs].mode = JOB_CONTROLLER; + } + + /* we need our own copy of task[0].argv + * argc is the number of valid args, like the usual command line. + * we allocate one extra element, with value 0 to be passed to execvp + */ + jobs[Njobs].argc = task[0].argc; + ALLOCATE (jobs[Njobs].argv, char *, MAX (task[0].argc + 1, 1)); + for (i = 0; i < task[0].argc; i++) { + jobs[Njobs].argv[i] = strcreate (task[0].argv[i]); + } + jobs[Njobs].argv[i] = 0; + + /* other data from the task is needed by the job + we carry a pointer back to the task. this means we + cannot modify the task once a job is created, or the changes will + be applied to the existing jobs */ + + jobs[Njobs].task = task; + + /* if we decide we need to be able to dynamically set task qualities + (like host, timeouts, etc), the we will need to have matched + entries to these quantites in the job structure */ Njobs ++; @@ -108,4 +125,21 @@ } +void FreeJob (Job *job) { + + int i; + + if (job == NULL) return; + + if ((job[0].JobID >= 0) || (job[0].JobID < MAX_N_JOBS)) { + JobIDList[job[0].JobID] = FALSE; + } + + for (i = 0; i < job[0].argc; i++) { + free (job[0].argv[i]); + } + free (job[0].argv); + return; +} + void SetCurrentJob (char *name) { JobName = name; @@ -114,22 +148,4 @@ char *GetCurrentJob () { return (JobName); -} - -void FreeJob (Job *job) { - - int i; - - if (job == NULL) return; - - if ((job[0].JobID >= 0) || (job[0].JobID < MAX_N_JOBS)) { - JobIDList[job[0].JobID] = FALSE; - } - - for (i = 0; i < job[0].task.argc; i++) { - free (job[0].task.argv[i]); - } - free (job[0].task.argv); - free (job[0].task.host); - return; } @@ -161,29 +177,11 @@ } -/* this needs to: - 1) distinguish local from controller jobs - 2) fork the local jobs in the background - 3) send the controller jobs to the controller */ - int SubmitJob (Job *job) { - int i, Nchar; - char *string; - - Nchar = 0; - for (i = 0; i < job[0].task.argc; i++) { - Nchar += strlen (job[0].task.argv[i]) + 1; - } - ALLOCATE (string, char, Nchar); - bzero (string, Nchar); - - strcat (string, job[0].task.argv[0]); - for (i = 1; i < job[0].task.argc; i++) { - strcat (string, " "); - strcat (string, job[0].task.argv[i]); - } - - fprintf (stderr, "executing job %d ...%s...\n", job[0].JobID, string); - free (string); + if (job[0].mode == JOB_LOCAL) { + SubmitLocalJob (job); + } else { + SubmitControllerJob (job); + } /* reset clock for start and poll-test */ @@ -196,26 +194,11 @@ int CheckJob (Job *job) { - float f; - f = drand48 (); - - if ((0.00 < f) && (f < 0.25)) { - job[0].state = JOB_BUSY; - job[0].exit_status = -1; - return (JOB_BUSY); - } - if ((0.25 < f) && (f < 0.50)) { - job[0].state = JOB_BUSY; - job[0].exit_status = -1; - return (JOB_CRASH); - } - if ((0.50 < f) && (f < 0.75)) { - job[0].state = JOB_BUSY; - job[0].exit_status = 0; - return (JOB_EXIT); - } - if ((0.75 < f) && (f < 1.00)) { - job[0].state = JOB_BUSY; - job[0].exit_status = 1; - return (JOB_EXIT); - } -} + /* add checks for timeouts */ + + if (job[0].mode == JOB_LOCAL) { + CheckLocalJob (job); + } else { + CheckControllerJob (job); + } + return (job[0].state); +} Index: /trunk/Ohana/src/opihi/pantasks/LocalJob.c =================================================================== --- /trunk/Ohana/src/opihi/pantasks/LocalJob.c (revision 4449) +++ /trunk/Ohana/src/opihi/pantasks/LocalJob.c (revision 4450) @@ -1,17 +1,150 @@ +# include "scheduler.h" -/* local jobs are forked in the background */ +/* this could be written a just a one-way pipe */ +int SubmitLocalJob (Job *job) { -SubmitLocalJob (Job *job) { + int status, pid; + int stdout_fd[2], stderr_fd[2]; - /* - construct the command line - fork the command, get back the PID - increment local job counter - */ + bzero (stdout_fd, 2*sizeof(int)); + bzero (stderr_fd, 2*sizeof(int)); + if (pipe (stdout_fd) < 0) goto pipe_error; + if (pipe (stderr_fd) < 0) goto pipe_error; + + pid = fork (); + if (!pid) { /* must be child process */ + fprintf (stderr, "starting controller connection\n"); + + /* close the other ends of the pipes */ + close (stdout_fd[0]); + close (stderr_fd[0]); + + /* tie our ends of the pipes to stdin, stdout, stderr */ + dup2 (stdout_fd[1], STDOUT_FILENO); + dup2 (stderr_fd[1], STDERR_FILENO); + + /* set all three unblocking */ + setvbuf (stdout, (char *) NULL, _IONBF, BUFSIZ); + setvbuf (stderr, (char *) NULL, _IONBF, BUFSIZ); + + status = execvp (job[0].argv[0], job[0].argv); + exit (1); + } + + /* close the other ends of the pipes */ + close (stdout_fd[1]); stdout_fd[1] = 0; + close (stderr_fd[1]); stderr_fd[1] = 0; + + /* make the pipes non-blocking */ + fcntl (stdout_fd[0], F_SETFL, O_NONBLOCK); + fcntl (stderr_fd[0], F_SETFL, O_NONBLOCK); + + job[0].stdout_fd = stdout_fd[0]; + job[0].stderr_fd = stderr_fd[0]; + job[0].pid = pid; + + return (TRUE); + +pipe_error: + perror ("pipe error:"); + if (stdout_fd[0] != 0) close (stdout_fd[0]); + if (stdout_fd[1] != 0) close (stdout_fd[1]); + if (stderr_fd[0] != 0) close (stderr_fd[0]); + if (stderr_fd[1] != 0) close (stderr_fd[1]); + return (FALSE); } -CheckLocalJob (Job *job) { +/* update current state, drain stdout/stderr buffers */ +int CheckLocalJob (Job *job) { - /* - + int Nread; + + // XXX do something useful with exit status? + CheckLocalJobStatus (job); + + /* read stdout buffer */ + Nread = ReadtoIOBuffer (&job[0].stdout, job[0].stdout_fd); + switch (Nread) { + case -2: /* error in read (programming error? system level error?) */ + fprintf (stderr, "serious IO error\n"); + exit (2); + case -1: /* no data in pipe */ + break; + case 0: /* pipe is closed */ + /** change child state? **/ + break; + default: /* data in pipe */ + break; + } + + /* read stderr buffer */ + Nread = ReadtoIOBuffer (&job[0].stderr, job[0].stderr_fd); + switch (Nread) { + case -2: /* error in read (programming error? system level error?) */ + fprintf (stderr, "serious IO error\n"); + exit (2); + case -1: /* no data in pipe */ + break; + case 0: /* pipe is closed */ + /** change child state? **/ + break; + default: /* data in pipe */ + break; + } + return (TRUE); +} + +int CheckLocalJobStatus (Job *job) { + + int result, waitstatus; + + /* check local job status */ + result = waitpid (job[0].pid, &waitstatus, WNOHANG); + switch (result) { + case -1: /* error with waitpid */ + switch (errno) { + case ECHILD: + fprintf (stderr, "unknown PID, not a child proc\n"); + fprintf (stderr, "did process already exit? programming error?\n"); + job[0].state = JOB_NONE; + job[0].exit_status = 0; + return (FALSE); + case EINVAL: + fprintf (stderr, "error EINVAL (waitpid): programming error\n"); + exit (1); + case EINTR: + fprintf (stderr, "error EINTR (waitpid): programming error\n"); + exit (1); + default: + fprintf (stderr, "unknown error for waitpid (%d): programming error\n", errno); + exit (1); + } + break; + + case 0: /* process not exited */ + job[0].state = JOB_BUSY; + job[0].exit_status = 0; + return (TRUE); + + default: + if (result != job[0].pid) { + fprintf (stderr, "waitpid error: mis-matched PID (%d vs %d). programming error\n", result, job[0].pid); + exit (1); + } + + if (WIFEXITED(waitstatus)) { + job[0].state = JOB_EXIT; + job[0].exit_status = WEXITSTATUS(waitstatus); + } + if (WIFSIGNALED(waitstatus)) { + job[0].state = JOB_CRASH; + job[0].exit_status = WTERMSIG(waitstatus); + } + if (WIFSTOPPED(waitstatus)) { + fprintf (stderr, "waitpid returns 'stopped': programming error\n"); + exit (1); + } + } + return; +} Index: /trunk/Ohana/src/opihi/pantasks/Makefile =================================================================== --- /trunk/Ohana/src/opihi/pantasks/Makefile (revision 4449) +++ /trunk/Ohana/src/opihi/pantasks/Makefile (revision 4450) @@ -25,15 +25,25 @@ # sched user commands and support functions ######################## -sched = \ +funcs = \ +$(SDIR)/CheckJobs.$(ARCH).o \ +$(SDIR)/CheckSystem.$(ARCH).o \ +$(SDIR)/CheckTasks.$(ARCH).o \ +$(SDIR)/ControllerOps.$(ARCH).o \ +$(SDIR)/LocalJob.$(ARCH).o \ +$(SDIR)/JobOps.$(ARCH).o \ +$(SDIR)/TaskOps.$(ARCH).o \ +$(SDIR)/IOBufferOps.$(ARCH).o \ +$(SDIR)/memstr.$(ARCH).o \ +$(SDIR)/init.$(ARCH).o + +cmds = \ +$(SDIR)/controller.$(ARCH).o \ $(SDIR)/run.$(ARCH).o \ +$(SDIR)/scheduler.$(ARCH).o \ $(SDIR)/task.$(ARCH).o \ $(SDIR)/task_command.$(ARCH).o \ $(SDIR)/task_host.$(ARCH).o \ $(SDIR)/task_macros.$(ARCH).o \ -$(SDIR)/task_periods.$(ARCH).o \ -$(SDIR)/JobOps.$(ARCH).o \ -$(SDIR)/TaskOps.$(ARCH).o \ -$(SDIR)/init.$(ARCH).o \ -$(SDIR)/scheduler.$(ARCH).o +$(SDIR)/task_periods.$(ARCH).o libs = \ @@ -49,5 +59,5 @@ @echo done -$(BIN)/scheduler.$(ARCH) : $(sched) $(libs) +$(BIN)/scheduler.$(ARCH) : $(funcs) $(cmds) $(libs) install: $(DESTBIN)/scheduler Index: /trunk/Ohana/src/opihi/pantasks/TaskOps.c =================================================================== --- /trunk/Ohana/src/opihi/pantasks/TaskOps.c (revision 4449) +++ /trunk/Ohana/src/opihi/pantasks/TaskOps.c (revision 4450) @@ -108,4 +108,5 @@ tasks[Ntasks].host = NULL; + tasks[Ntasks].host_required = FALSE; tasks[Ntasks].argc = 0; @@ -124,4 +125,7 @@ tasks[Ntasks].poll_period = 1.0; tasks[Ntasks].timeout_period = 1.0; + + /* init task timer (is reset by 'run') */ + gettimeofday (&tasks[Ntasks].last, (void *) NULL); Ntasks ++; @@ -155,2 +159,12 @@ gettimeofday (timer, (void *) NULL); } + +/* start the clock for all tasks */ +void InitTaskTimers () { + + Task *task; + + while ((task = NextTask ()) != NULL) { + gettimeofday (&task[0].last, (void *) NULL); + } +} Index: /trunk/Ohana/src/opihi/pantasks/controller.c =================================================================== --- /trunk/Ohana/src/opihi/pantasks/controller.c (revision 4450) +++ /trunk/Ohana/src/opihi/pantasks/controller.c (revision 4450) @@ -0,0 +1,48 @@ +# include "scheduler.h" + +int controller (int argc, char **argv) { + + int status; + char command[1024]; + IOBuffer buffer; + + if (argc != 3) { + fprintf (stderr, "USAGE: controller host (hostname)\n"); + return (FALSE); + } + + if (strcmp (argv[1], "host")) { + fprintf (stderr, "invalid controller command\n"); + return (FALSE); + } + + /* start controller connection (if needed) */ + StartController (); + + sprintf (command, "host %s", argv[2]); + InitIOBuffer (&buffer, 0x100); + status = ControllerCommand (command, CONTROLLER_PROMPT, &buffer); + FreeIOBuffer (&buffer); + + /* check on success of controller command */ + switch (status) { + case CONTROLLER_DOWN: + fprintf (stderr, "controller is down\n"); + return (FALSE); + + case CONTROLLER_HUNG: + fprintf (stderr, "controller is not responding\n"); + return (FALSE); + + case CONTROLLER_GOOD: + fprintf (stderr, "controller command sent\n"); + return (TRUE); + + default: + fprintf (stderr, "unknown status for controller command: programming error\n"); + exit (1); + } + + fprintf (stderr, "programming error: should not reach here\n"); + exit (1); +} Index: /trunk/Ohana/src/opihi/pantasks/init.c =================================================================== --- /trunk/Ohana/src/opihi/pantasks/init.c (revision 4449) +++ /trunk/Ohana/src/opihi/pantasks/init.c (revision 4450) @@ -2,4 +2,5 @@ # include "scheduler.h" +int controller PROTO((int, char **)); int task PROTO((int, char **)); int task_host PROTO((int, char **)); @@ -8,13 +9,16 @@ int task_periods PROTO((int, char **)); int run PROTO((int, char **)); +int stop PROTO((int, char **)); static Command cmds[] = { - {"task", task, "define a schedulable task"}, - {"host", task_host, "define host machine for a task"}, - {"task.exit", task_macros, "define exit macros for a task"}, - {"task.exec", task_macros, "define pre-exec macro for a task"}, - {"command", task_command, "define executed command for a task"}, - {"periods", task_periods, "define time scales for a task"}, - {"run", run, "run the scheduler"}, + {"controller", controller, "controller commands"}, + {"task", task, "define a schedulable task"}, + {"host", task_host, "define host machine for a task"}, + {"task.exit", task_macros, "define exit macros for a task"}, + {"task.exec", task_macros, "define pre-exec macro for a task"}, + {"command", task_command, "define executed command for a task"}, + {"periods", task_periods, "define time scales for a task"}, + {"run", run, "run the scheduler"}, + {"stop", stop, "stop the scheduler"}, }; Index: /trunk/Ohana/src/opihi/pantasks/memstr.c =================================================================== --- /trunk/Ohana/src/opihi/pantasks/memstr.c (revision 4450) +++ /trunk/Ohana/src/opihi/pantasks/memstr.c (revision 4450) @@ -0,0 +1,13 @@ +# include "pcontrol.h" + +/* returns pointer to start of m2 in m1, or NULL if failure */ +char *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 (NULL); + return (m1); + +} Index: /trunk/Ohana/src/opihi/pantasks/run.c =================================================================== --- /trunk/Ohana/src/opihi/pantasks/run.c (revision 4449) +++ /trunk/Ohana/src/opihi/pantasks/run.c (revision 4450) @@ -1,12 +1,5 @@ -# include "basic.h" # include "scheduler.h" int run (int argc, char **argv) { - - Job *job; - Task *task; - Macro *macro; - int i, found, status; - int Ntest; if (argc != 1) { @@ -15,107 +8,21 @@ } - /* start the clock for all tasks */ - while ((task = NextTask ()) != NULL) { - gettimeofday (&task[0].last, (void *) NULL); + InitTaskTimers (); + rl_event_hook = CheckSystem; + rl_set_keyboard_input_timeout (1000000); + + return (TRUE); +} + +int stop (int argc, char **argv) { + + if (argc != 1) { + fprintf (stderr, "USAGE: stop\n"); + return (FALSE); } - Ntest = 0; + rl_event_hook = NULL; + rl_set_keyboard_input_timeout (1000000); - /* loop forever, checking for completed jobs and ready tasks */ - while (1) { - if (Ntest > 5) { - ListJobs (); - Ntest = 0; - } - usleep (10000); - Ntest ++; - - /** test all tasks: ready to test? ready to run? **/ - while ((task = NextTask ()) != NULL) { - - /* ready to test? : check exec period */ - if (GetTaskTimer(task[0].last) < task[0].exec_period) continue; - - SetCurrentTask (task[0].name); - fprintf (stderr, "trying task %s\n", task[0].name); - - /* ready to run? : run task.exec macro */ - if (task[0].exec != NULL) { - status = exec_loop (task[0].exec); - if (!status) continue; - } - - /* is task valid? check state of task.(argc, argv) */ - /*** ADD CODE HERE ***/ - - /* construct job from task */ - job = CreateJob (task); - - /* execute job - XXX add status test */ - SubmitJob (job); - - /* reset timer on task (don't do this if Create/Submit fails)*/ - gettimeofday (&task[0].last, (void *) NULL); - } - - /** test all jobs: ready to test? finished? **/ - while ((job = NextJob ()) != NULL) { - - /* check for timeout */ - if (GetTaskTimer(job[0].start) >= job[0].task.timeout_period) { - fprintf (stderr, "timeout on %s\n", job[0].task.name); - /* run task.timeout macro, if it exists */ - if (job[0].task.timeout != NULL) { - exec_loop (job[0].task.timeout); - } - DeleteJob (job); - continue; - } - - /* check poll period (ready to run again?) */ - if (GetTaskTimer(job[0].last) < job[0].task.poll_period) continue; - - /* check current status */ - status = CheckJob (job); - switch (status) { - case JOB_BUSY: - fprintf (stderr, "job %s (%d) busy\n", job[0].task.name, job[0].JobID); - break; - - case JOB_CRASH: - fprintf (stderr, "job %s (%d) crash\n", job[0].task.name, job[0].JobID); - /* run task.crash macro, if it exists */ - if (job[0].task.crash != NULL) { - exec_loop (job[0].task.crash); - } - DeleteJob (job); - continue; - break; - - case JOB_EXIT: - fprintf (stderr, "job %s (%d) exit\n", job[0].task.name, job[0].JobID); - /* run corresponding task.exit macro, if it exists */ - macro = job[0].task.def; - for (i = 0; i < job[0].task.Nexit; i++) { - if (job[0].exit_status == atoi(job[0].task.exit[i][0].name)) { - macro = job[0].task.exit[i]; - break; - } - } - if (macro != NULL) exec_loop (macro); - DeleteJob (job); - continue; - break; - - default: - fprintf (stderr, "unknown exit status\n"); - /** do something more useful here ?? **/ - break; - } - - /* reset polling clock */ - SetTaskTimer (&job[0].last); - } - } return (TRUE); } Index: /trunk/Ohana/src/opihi/pantasks/scheduler.c =================================================================== --- /trunk/Ohana/src/opihi/pantasks/scheduler.c (revision 4449) +++ /trunk/Ohana/src/opihi/pantasks/scheduler.c (revision 4450) @@ -49,4 +49,6 @@ rl_readline_name = opihi_name; rl_attempted_completion_function = command_completer; + rl_event_hook = NULL; + rl_set_keyboard_input_timeout (1000000); set_str_variable ("HISTORY", opihi_history); Index: /trunk/Ohana/src/opihi/pantasks/task_command.c =================================================================== --- /trunk/Ohana/src/opihi/pantasks/task_command.c (revision 4449) +++ /trunk/Ohana/src/opihi/pantasks/task_command.c (revision 4450) @@ -41,10 +41,2 @@ return (TRUE); } - - -/** - careful with this: the command is supposed to be realized - for the Job, not the Task (at execution) - the code is right, but who calls it when needs to be clarified. - **/ - Index: /trunk/Ohana/src/opihi/pantasks/task_host.c =================================================================== --- /trunk/Ohana/src/opihi/pantasks/task_host.c (revision 4449) +++ /trunk/Ohana/src/opihi/pantasks/task_host.c (revision 4450) @@ -4,10 +4,20 @@ int task_host (int argc, char **argv) { + int N, RequiredHost; Task *task; char *taskname; + RequiredHost = FALSE; + if (N = get_argument (argc, argv, "-required")) { + remove_argument (N, &argc, argv); + RequiredHost = TRUE; + } + if (argc != 2) { - fprintf (stderr, "USAGE: host \n"); - fprintf (stderr, " (define host machine for this task (or 'none'))\n"); + fprintf (stderr, "USAGE: host [-required]\n"); + fprintf (stderr, " define host machine for this task\n"); + fprintf (stderr, " -required flags indicates controller must use this host\n"); + fprintf (stderr, " value of 'local' for host indicates process not using controller\n"); + fprintf (stderr, " value of 'none' for host indicates controller may assign at will\n"); return (FALSE); } @@ -23,9 +33,10 @@ return (FALSE); } + task[0].host_required = RequiredHost; if (task[0].host != NULL) free (task[0].host); task[0].host = NULL; - if (!strcasecmp (argv[1], "NONE")) return (TRUE); + if (!strcasecmp (argv[1], "LOCAL")) return (TRUE); task[0].host = strcreate (argv[1]); Index: /trunk/Ohana/src/opihi/pclient/ChildOps.c =================================================================== --- /trunk/Ohana/src/opihi/pclient/ChildOps.c (revision 4449) +++ /trunk/Ohana/src/opihi/pclient/ChildOps.c (revision 4450) @@ -57,5 +57,4 @@ break; } - /* read stderr buffer */ Index: /trunk/Ohana/src/opihi/pclient/job.c =================================================================== --- /trunk/Ohana/src/opihi/pclient/job.c (revision 4449) +++ /trunk/Ohana/src/opihi/pclient/job.c (revision 4450) @@ -39,5 +39,5 @@ setvbuf (stdout, (char *) NULL, _IONBF, BUFSIZ); setvbuf (stderr, (char *) NULL, _IONBF, BUFSIZ); - fprintf (stderr, "child is spawned, executing command\n"); + /* fprintf (stderr, "child is spawned, executing command\n"); */ /* note that this message comes out on the other side of the stderr pipe */ Index: /trunk/Ohana/src/opihi/pclient/reset.c =================================================================== --- /trunk/Ohana/src/opihi/pclient/reset.c (revision 4449) +++ /trunk/Ohana/src/opihi/pclient/reset.c (revision 4450) @@ -60,2 +60,10 @@ return (TRUE); } + +/* exit conditions: + -1 - usage message / syntax error + 2 - unknown job + 0 - process hung + 1 - successful resetn +*/ + Index: /trunk/Ohana/src/opihi/pclient/stdout.c =================================================================== --- /trunk/Ohana/src/opihi/pclient/stdout.c (revision 4449) +++ /trunk/Ohana/src/opihi/pclient/stdout.c (revision 4450) @@ -9,5 +9,4 @@ } - fprintf (stdout, "NBYTES %d\n", child_stdout.Nbuffer); fwrite (child_stdout.buffer, 1, child_stdout.Nbuffer, stdout); fprintf (stdout, "STATUS %d\n", 0); @@ -24,5 +23,4 @@ } - fprintf (stdout, "NBYTES %d\n", child_stderr.Nbuffer); fwrite (child_stderr.buffer, 1, child_stderr.Nbuffer, stdout); fprintf (stdout, "STATUS %d\n", 0); Index: /trunk/Ohana/src/opihi/pcontrol/CheckBusyJob.c =================================================================== --- /trunk/Ohana/src/opihi/pcontrol/CheckBusyJob.c (revision 4449) +++ /trunk/Ohana/src/opihi/pcontrol/CheckBusyJob.c (revision 4450) @@ -5,4 +5,5 @@ int i; int status; + int outstate; char *p; char string[64]; @@ -11,6 +12,5 @@ /** must have a valid host : if not? **/ - host = job[0].host; - /*** why is this a type error? ***/ + host = (Host *) job[0].host; InitIOBuffer (&buffer, 0x100); @@ -34,5 +34,5 @@ case PCLIENT_GOOD: - fprintf (stderr, "message received (CheckBusyJob)\n"); + /* fprintf (stderr, "message received (CheckBusyJob)\n"); */ break; @@ -60,25 +60,26 @@ } - /** job has exited : move to DONE stack (host still BUSY) **/ - if (!strcmp(string, "EXIT")) { - p = memstr (buffer.buffer, "EXITST", buffer.Nbuffer); - sscanf (p, "%*s %d", &job[0].exit_status); - PutJob (job, PCONTROL_JOB_DONE, STACK_BOTTOM); - job[0].state = PCONTROL_JOB_EXIT; - FreeIOBuffer (&buffer); - return (TRUE); - } - /** job has crashed : move to DONE stack (host still BUSY) */ - if (!strcmp(string, "CRASH")) { - p = memstr (buffer.buffer, "EXITST", buffer.Nbuffer); - sscanf (p, "%*s %d", &job[0].exit_status); - PutJob (job, PCONTROL_JOB_DONE, STACK_BOTTOM); - job[0].state = PCONTROL_JOB_CRASH; - FreeIOBuffer (&buffer); - return (TRUE); + /* exit status better be either EXIT or CRASH */ + outstate = PCONTROL_JOB_BUSY; + if (!strcmp(string, "EXIT")) outstate = PCONTROL_JOB_EXIT; + if (!strcmp(string, "CRASH")) outstate = PCONTROL_JOB_CRASH; + if (outstate == PCONTROL_JOB_BUSY) { + fprintf (stderr, "programming error : should not reach here (CheckJob)\n"); + exit (1); } - fprintf (stderr, "programming error : should not reach here (CheckJob)\n"); - exit (1); + /* parse the exit status and sizes of output buffers */ + p = memstr (buffer.buffer, "EXITST", buffer.Nbuffer); + sscanf (p, "%*s %d", &job[0].exit_status); + p = memstr (buffer.buffer, "STDOUT", buffer.Nbuffer); + sscanf (p, "%*s %d", &job[0].stdout_size); + p = memstr (buffer.buffer, "STDERR", buffer.Nbuffer); + sscanf (p, "%*s %d", &job[0].stderr_size); + + /** job has exited : move to DONE stack (host still BUSY) **/ + PutJob (job, PCONTROL_JOB_DONE, STACK_BOTTOM); + job[0].state = outstate; + FreeIOBuffer (&buffer); + return (TRUE); } Index: /trunk/Ohana/src/opihi/pcontrol/CheckDoneJob.c =================================================================== --- /trunk/Ohana/src/opihi/pcontrol/CheckDoneJob.c (revision 4449) +++ /trunk/Ohana/src/opihi/pcontrol/CheckDoneJob.c (revision 4450) @@ -5,10 +5,11 @@ Host *host; - if (!GetJobOutput (job, "stdout")) { + if (!GetJobOutput ("stdout", (Host *) job[0].host, &job[0].stdout, job[0].stdout_size)) { + /* strip off first and last lines */ PutJob (job, PCONTROL_JOB_DONE, STACK_BOTTOM); return (FALSE); } - if (!GetJobOutput (job, "stderr")) { + if (!GetJobOutput ("stderr", (Host *) job[0].host, &job[0].stderr, job[0].stderr_size)) { PutJob (job, PCONTROL_JOB_DONE, STACK_BOTTOM); return (FALSE); Index: /trunk/Ohana/src/opihi/pcontrol/CheckHost.c =================================================================== --- /trunk/Ohana/src/opihi/pcontrol/CheckHost.c (revision 4449) +++ /trunk/Ohana/src/opihi/pcontrol/CheckHost.c (revision 4450) @@ -16,13 +16,14 @@ /* if host has a job, job is dead, push to Pending */ if (host[0].stack == PCONTROL_HOST_BUSY) { - job = host[0].job; - N = FindJob (job[0].JobID, PCONTROL_JOB_BUSY); - if (N < 0) { - fprintf (stderr, "error: job is not found in BUSY list\n"); - exit (2); + job = (Job *) host[0].job; + if (job != NULL) { + N = FindJob (job[0].JobID, PCONTROL_JOB_BUSY); + if (N < 0) { + fprintf (stderr, "programming error: job is not found in BUSY list\n"); + exit (2); + } + job[0].host = NULL; /* unlink host & job */ + PutJob (job, PCONTROL_JOB_PENDING, STACK_BOTTOM); } - job[0].host = NULL; /* unlink host & job */ - job = GetJob (PCONTROL_JOB_BUSY, N); - PutJob (job, PCONTROL_JOB_PENDING, STACK_BOTTOM); } host[0].job = NULL; Index: /trunk/Ohana/src/opihi/pcontrol/CheckIdleHost.c =================================================================== --- /trunk/Ohana/src/opihi/pcontrol/CheckIdleHost.c (revision 4449) +++ /trunk/Ohana/src/opihi/pcontrol/CheckIdleHost.c (revision 4450) @@ -13,5 +13,8 @@ job = (Job *) stack[0].object[i]; if (job[0].mode != PCONTROL_JOB_NEEDHOST) continue; - if (job[0].hostname == NULL) continue; + if (job[0].hostname == NULL) { + fprintf (stderr, "programming error: NEEDHOST hostname missing\n"); + exit (2); + } if (strcasecmp (job[0].hostname, host[0].hostname)) continue; LinkJobAndHost (job, host); @@ -24,5 +27,8 @@ job = (Job *) stack[0].object[i]; if (job[0].mode != PCONTROL_JOB_WANTHOST) continue; - if (job[0].hostname == NULL) continue; + if (job[0].hostname == NULL) { + fprintf (stderr, "programming error: WANTHOST hostname missing\n"); + exit (2); + } if (strcasecmp (job[0].hostname, host[0].hostname)) continue; LinkJobAndHost (job, host); Index: /trunk/Ohana/src/opihi/pcontrol/GetJobOutput.c =================================================================== --- /trunk/Ohana/src/opihi/pcontrol/GetJobOutput.c (revision 4449) +++ /trunk/Ohana/src/opihi/pcontrol/GetJobOutput.c (revision 4450) @@ -1,46 +1,56 @@ # include "pcontrol.h" +# define PCLIENT_TIMEOUT 20 -int GetJobOutput (Job *job, char *channel) { +/* we read Nbytes from the host, then watch for the prompt */ +int GetJobOutput (char *command, Host *host, IOBuffer *buffer, int Nbytes) { - int status; - IOBuffer *buffer; - Host *host; + int i, status, Nstart; + char *line; - buffer = NULL; - if (!strcasecmp (channel, "stdout")) buffer = &job[0].stdout; - if (!strcasecmp (channel, "stderr")) buffer = &job[0].stderr; - if (buffer == NULL) { - fprintf (stderr, "invalid output channel : programming error\n"); - exit (1); + /* flush any earlier messages */ + ReadtoIOBuffer (buffer, host[0].stdout); + FlushIOBuffer (buffer); + Nstart = buffer[0].Nbuffer; + + /* send command (stdout / stderr) */ + ALLOCATE (line, char, MAX (1, strlen(command) + 1)); + sprintf (line, "%s\n", command); + status = write (host[0].stdin, line, strlen(line)); + free (line); + + /* is pipe still open? */ + if ((status == -1) && (errno == EPIPE)) return (PCLIENT_DOWN); + + /* read at least Nbytes, then watch for PCLIENT_PROMPT */ + line = NULL; + status = -1; + for (i = 0; (i < PCLIENT_TIMEOUT) && (status == -1) && (line == NULL); i++) { + status = ReadtoIOBuffer (buffer, host[0].stdout); + if ((buffer[0].Nbuffer - Nstart) >= Nbytes) { + line = memstr (buffer[0].buffer, PCLIENT_PROMPT, buffer[0].Nbuffer); + } + if (status == -1) usleep (10000); } - - /* If we already have the output, just return, don't retry */ - if (buffer[0].Nbuffer) return (TRUE); - - /** must have a valid host : if not, move to pending? **/ - host = job[0].host; - status = PclientCommand (host, channel, PCLIENT_PROMPT, buffer); + if (status == 0) return (PCLIENT_DOWN); + if (status == -1) return (PCLIENT_HUNG); /* check on success of pclient command */ switch (status) { - case PCLIENT_DOWN: - /*** different behavior for ANYHOST, WANTHOST, NEEDHOST ***/ - buffer[0].Nbuffer = 0; + case -1: + fprintf (stderr, "host %s is not responding\n", host[0].hostname); + return (FALSE); + + case 0: fprintf (stderr, "host %s is down\n", host[0].hostname); return (FALSE); - case PCLIENT_HUNG: - /*** should we consider a HUNG host DOWN? ***/ - buffer[0].Nbuffer = 0; - fprintf (stderr, "host %s is not responding\n", host[0].hostname); - return (FALSE); - - case PCLIENT_GOOD: - fprintf (stderr, "message received (GetJobOutput : %s)\n", channel); + default: + fprintf (stderr, "message received (GetJobOutput : %s)\n", command); + /* drop extra bytes from pclient (not pclient:job) */ + buffer[0].Nbuffer = Nstart + Nbytes; + if (buffer[0].Nalloc > buffer[0].Nbuffer) { + bzero (buffer[0].buffer + buffer[0].Nbuffer, buffer[0].Nalloc - buffer[0].Nbuffer); + } return (TRUE); - - default: - fprintf (stderr, "unknown status for pclient command: programming error\n"); - exit (1); } Index: /trunk/Ohana/src/opihi/pcontrol/HostOps.c =================================================================== --- /trunk/Ohana/src/opihi/pcontrol/HostOps.c (revision 4449) +++ /trunk/Ohana/src/opihi/pcontrol/HostOps.c (revision 4450) @@ -33,4 +33,26 @@ fprintf (stderr, "error: unknown host stack : programming error\n"); exit (1); +} + +Host *FindHostStack (IDtype HostID) { + + Host *host; + + host = FindHostPtr (HostID, PCONTROL_HOST_IDLE); + if (host != NULL) return (host); + + host = FindHostPtr (HostID, PCONTROL_HOST_DOWN); + if (host != NULL) return (host); + + host = FindHostPtr (HostID, PCONTROL_HOST_DONE); + if (host != NULL) return (host); + + host = FindHostPtr (HostID, PCONTROL_HOST_BUSY); + if (host != NULL) return (host); + + host = FindHostPtr (HostID, PCONTROL_HOST_OFF); + if (host != NULL) return (host); + + return (NULL); } @@ -79,4 +101,38 @@ } +Host *FindHostPtr (IDtype HostID, int StackID) { + + int i; + Host *host; + Stack *stack; + + stack = GetHostStack (StackID); + if (stack == NULL) return (NULL); + + for (i = 0; i < stack[0].Nobject; i++) { + host = (Host *) stack[0].object[i]; + if (host[0].HostID == HostID) { + return (host); + } + } + return (NULL); +} + +Host *PullHost (IDtype HostID, int StackID) { + + int N; + Host *host; + + N = FindHost (HostID, StackID); + if (N < 0) return (NULL); + + host = GetHost (StackID, N); + if (host == NULL) { + fprintf (stderr, "programming error! host missing from stack\n"); + exit (1); + } + return (host); +} + int FindNamedHost (char *name, int StackID) { @@ -119,3 +175,2 @@ FREE (host); } - Index: /trunk/Ohana/src/opihi/pcontrol/JobOps.c =================================================================== --- /trunk/Ohana/src/opihi/pcontrol/JobOps.c (revision 4449) +++ /trunk/Ohana/src/opihi/pcontrol/JobOps.c (revision 4450) @@ -35,4 +35,27 @@ } +Job *FindJobStack (IDtype JobID) { + + Job *job; + + job = FindJobPtr (JobID, PCONTROL_JOB_PENDING); + if (job != NULL) return (job); + + job = FindJobPtr (JobID, PCONTROL_JOB_BUSY); + if (job != NULL) return (job); + + job = FindJobPtr (JobID, PCONTROL_JOB_EXIT); + if (job != NULL) return (job); + + job = FindJobPtr (JobID, PCONTROL_JOB_CRASH); + if (job != NULL) return (job); + + job = FindJobPtr (JobID, PCONTROL_JOB_DONE); + if (job != NULL) return (job); + + return (NULL); +} + +/* add job to position in stack */ int PutJob (Job *job, int StackID, int where) { @@ -51,4 +74,5 @@ } +/* remove job from position in stack */ Job *GetJob (int StackID, int where) { @@ -63,4 +87,5 @@ } +/* return stack position of job */ int FindJob (IDtype JobID, int StackID) { @@ -81,4 +106,40 @@ } +/* return pointer to job */ +Job *FindJobPtr (IDtype JobID, int StackID) { + + int i; + Job *job; + Stack *stack; + + stack = GetJobStack (StackID); + if (stack == NULL) return (NULL); + + for (i = 0; i < stack[0].Nobject; i++) { + job = (Job *) stack[0].object[i]; + if (job[0].JobID == JobID) { + return (job); + } + } + return (NULL); +} + +/* remove job from stack, return pointer */ +Job *PullJob (IDtype JobID, int StackID) { + + int N; + Job *job; + + N = FindJob (JobID, StackID); + if (N < 0) return (NULL); + + job = GetJob (StackID, N); + if (job == NULL) { + fprintf (stderr, "programming error! job missing from stack\n"); + exit (1); + } + return (job); +} + IDtype AddJob (char *hostname, JobMode mode, int timeout, int argc, char **argv) { @@ -89,5 +150,5 @@ job[0].argc = argc; job[0].argv = argv; - job[0].hostname = strcreate (hostname); + job[0].hostname = hostname; job[0].mode = mode; job[0].host = NULL; @@ -119,27 +180,5 @@ } -# if (0) -void KillJob (Job *job) { - - int status; - char line[64]; - Host *host; - - sprintf (line, "reset\n"); - - /* send command to client */ - status = write (host[0].stdin, line, strlen(line)); - if ((status == -1) && (errno == EPIPE)) { - fprintf (stderr, "host %s is down\n", host[0].hostname); - PutHost (host, PCONTROL_HOST_DOWN, STACK_BOTTOM); - PutJob (job, PCONTROL_JOB_PENDING, STACK_BOTTOM); - return (FALSE); - } - /* check for a response */ - - /** this needs to be cleaned up to handle the slow response case **/ -} -# endif - +/* unlink job and host, pull host from its stack */ Host *UnlinkJobAndHost (Job *job) { @@ -147,5 +186,9 @@ Host *host; - host = job[0].host; + host = (Host *) job[0].host; + if (host == NULL) { + fprintf (stderr, "programming error: job has no host\n"); + exit (2); + } /* unlink host & job */ @@ -153,11 +196,9 @@ host[0].job = NULL; - /*** need to pop host off of correct stack XXX ***/ - N = FindHost (host[0].HostID, host[0].stack); - if (N < 0) { + /* remove host from correct stack */ + if (PullHost (host[0].HostID, host[0].stack) == NULL) { fprintf (stderr, "programming error: host is not found in current stack\n"); exit (2); } - host = GetHost (host[0].stack, N); return (host); } @@ -166,27 +207,11 @@ int N; - job[0].host = host; - host[0].job = job; - -# if (0) - /*** need to pop host off of correct stack XXX ***/ - N = FindHost (host[0].HostID, host[0].stack); - if (N < 0) { - fprintf (stderr, "programming error: host is not found in current stack\n"); - exit (2); - } - host = GetHost (host[0].stack, N); -# endif - - /*** need to pop job off of correct stack XXX ***/ - N = FindJob (job[0].JobID, job[0].stack); - if (N < 0) { + job[0].host = (struct Host *) host; + host[0].job = (struct Job *) job; + + /* remove job from correct stack */ + if (PullJob (job[0].JobID, job[0].stack) == NULL) { fprintf (stderr, "programming error: job is not found in current stack\n"); exit (2); } - job = GetJob (job[0].stack, N); - - /*** this is fairly crazy : the only reason this works is cause I - get the same host ptr here and in CheckIdleHost - ***/ -} +} Index: /trunk/Ohana/src/opihi/pcontrol/KillJob.c =================================================================== --- /trunk/Ohana/src/opihi/pcontrol/KillJob.c (revision 4450) +++ /trunk/Ohana/src/opihi/pcontrol/KillJob.c (revision 4450) @@ -0,0 +1,73 @@ +# include "pcontrol.h" + +int KillJob (Job *job) { + + Host *host; + IOBuffer buffer; + int status; + char *p; + + /** must have a valid host : if not? **/ + host = (Host *) job[0].host; + + InitIOBuffer (&buffer, 0x100); + + status = PclientCommand (host, "reset", PCLIENT_PROMPT, &buffer); + + /* check on success of pclient command */ + switch (status) { + case PCLIENT_DOWN: + UnlinkJobAndHost (job); + PutHost (host, PCONTROL_HOST_DOWN, STACK_BOTTOM); + PutJob (job, PCONTROL_JOB_PENDING, STACK_BOTTOM); + FreeIOBuffer (&buffer); + return (FALSE); + + case PCLIENT_HUNG: + PutJob (job, PCONTROL_JOB_BUSY, STACK_BOTTOM); + job[0].state = PCONTROL_JOB_HUNG; + FreeIOBuffer (&buffer); + return (TRUE); + + case PCLIENT_GOOD: + fprintf (stderr, "message received (KillJob)\n"); + break; + + default: + fprintf (stderr, "unknown status for pclient command: programming error\n"); + exit (1); + } + + /** host is up, need to parse message **/ + p = memstr (buffer.buffer, "STATUS", buffer.Nbuffer); + if (p == NULL) { + fprintf (stderr, "missing STATUS in pclient message : programming error\n"); + exit (1); + } + fprintf (stderr, "client message: %s\n", buffer.buffer); + sscanf (p, "%*s %d", &status); + FreeIOBuffer (&buffer); + fprintf (stderr, "client status: %d\n", status); + + switch (status) { + case -1: + fprintf (stderr, "programming error (syntax error to pclient)\n"); + return (FALSE); + break; + case 0: + fprintf (stderr, "failure to kill child process\n"); + PutJob (job, PCONTROL_JOB_BUSY, STACK_BOTTOM); + return (FALSE); + case 1: + PutJob (job, PCONTROL_JOB_DONE, STACK_BOTTOM); + job[0].state = PCONTROL_JOB_CRASH; + /* this overrides the default assigned by PutJob */ + return (TRUE); + case 2: + fprintf (stderr, "programming error (client has no job)\n"); + return (FALSE); + } + + fprintf (stderr, "programming error : should not reach here (CheckJob)\n"); + exit (1); +} Index: /trunk/Ohana/src/opihi/pcontrol/Makefile =================================================================== --- /trunk/Ohana/src/opihi/pcontrol/Makefile (revision 4449) +++ /trunk/Ohana/src/opihi/pcontrol/Makefile (revision 4450) @@ -20,5 +20,5 @@ LIBS2 = -lbasiccmd -lshell -ldata LIBS = $(LIBS2) $(LIBS1) -CFLAGS = $(INCS) +CFLAGS = $(INCS) -Werror CCFLAGS = $(LIBS) @@ -46,9 +46,14 @@ $(SDIR)/StartHost.$(ARCH).o \ $(SDIR)/StopHosts.$(ARCH).o \ +$(SDIR)/KillJob.$(ARCH).o \ $(SDIR)/StartJob.$(ARCH).o cmds = \ +$(SDIR)/kill.$(ARCH).o \ +$(SDIR)/delete.$(ARCH).o \ +$(SDIR)/check.$(ARCH).o \ $(SDIR)/job.$(ARCH).o \ $(SDIR)/status.$(ARCH).o \ +$(SDIR)/stdout.$(ARCH).o \ $(SDIR)/host.$(ARCH).o Index: /trunk/Ohana/src/opihi/pcontrol/ResetJob.c =================================================================== --- /trunk/Ohana/src/opihi/pcontrol/ResetJob.c (revision 4449) +++ /trunk/Ohana/src/opihi/pcontrol/ResetJob.c (revision 4450) @@ -8,5 +8,5 @@ /** must have a valid host : if not, move to pending? **/ - host = job[0].host; + host = (Host *) job[0].host; InitIOBuffer (&buffer, 0x100); Index: /trunk/Ohana/src/opihi/pcontrol/StartJob.c =================================================================== --- /trunk/Ohana/src/opihi/pcontrol/StartJob.c (revision 4449) +++ /trunk/Ohana/src/opihi/pcontrol/StartJob.c (revision 4450) @@ -11,5 +11,5 @@ /* job must have assigned host */ - host = job[0].host; + host = (Host *) job[0].host; if (host == NULL) { fprintf (stderr, "no assigned host : programming error\n"); Index: /trunk/Ohana/src/opihi/pcontrol/StopHosts.c =================================================================== --- /trunk/Ohana/src/opihi/pcontrol/StopHosts.c (revision 4449) +++ /trunk/Ohana/src/opihi/pcontrol/StopHosts.c (revision 4450) @@ -114,4 +114,4 @@ CLOSE (host[0].stderr); host[0].job = NULL; - PutHost (host, PCONTROL_HOST_DOWN, STACK_BOTTOM); + PutHost (host, PCONTROL_HOST_OFF, STACK_BOTTOM); } Index: /trunk/Ohana/src/opihi/pcontrol/check.c =================================================================== --- /trunk/Ohana/src/opihi/pcontrol/check.c (revision 4450) +++ /trunk/Ohana/src/opihi/pcontrol/check.c (revision 4450) @@ -0,0 +1,44 @@ +# include "pcontrol.h" +char jobstate[6][32] = {"PENDING", "BUSY", "EXIT", "CRASH", "HUNG", "DONE"}; +char hoststate[5][32] = {"IDLE", "BUSY", "DOWN", "DONE", "OFF"}; + +int check (int argc, char **argv) { + + Job *job; + Host *host; + int JobID, HostID, N; + + if (argc != 3) { + fprintf (stderr, "USAGE: check (job) (JobID)\n"); + fprintf (stderr, "USAGE: check (host) (HostID)\n"); + return (FALSE); + } + + if (!strcasecmp (argv[1], "JOB")) { + JobID = atoi (argv[2]); + job = FindJobStack (JobID); + if (job == NULL) { + fprintf (stdout, "job not found\n"); + return (FALSE); + } + fprintf (stdout, "STATUS %s\n", jobstate[job[0].stack]); + fprintf (stdout, "EXITST %d\n", job[0].exit_status); + fprintf (stdout, "STDOUT %d\n", job[0].stdout_size); + fprintf (stdout, "STDERR %d\n", job[0].stderr_size); + return (TRUE); + } + + if (!strcasecmp (argv[1], "HOST")) { + HostID = atoi (argv[2]); + host = FindHostStack (HostID); + if (host == NULL) { + fprintf (stdout, "host not found\n"); + return (FALSE); + } + fprintf (stdout, "host %s\n", hoststate[host[0].stack]); + return (TRUE); + } + + fprintf (stderr, "unknown item to check\n"); + return (FALSE); +} Index: /trunk/Ohana/src/opihi/pcontrol/delete.c =================================================================== --- /trunk/Ohana/src/opihi/pcontrol/delete.c (revision 4450) +++ /trunk/Ohana/src/opihi/pcontrol/delete.c (revision 4450) @@ -0,0 +1,40 @@ +# include "pcontrol.h" + +int delete (int argc, char **argv) { + + Job *job; + int JobID, N; + + if (argc < 2) { + fprintf (stderr, "USAGE: delete (JobID)\n"); + return (FALSE); + } + JobID = atoi (argv[1]); + /* use a string interp to convert JobIDs to ints ? */ + + job = PullJob (JobID, PCONTROL_JOB_PENDING); + if (job != NULL) goto found; + + job = PullJob (JobID, PCONTROL_JOB_CRASH); + if (job != NULL) goto found; + + job = PullJob (JobID, PCONTROL_JOB_EXIT); + if (job != NULL) goto found; + + fprintf (stderr, "job %s not PENDING, CRASH, EXIT\n", argv[1]); + return (FALSE); + +found: + { + int j; + fprintf (stderr, "deleting job %s %d ", job[0].hostname, job[0].argc); + for (j = 0; j < job[0].argc; j++) { + fprintf (stderr, "%s ", job[0].argv[j]); + } + PrintID (stderr, job[0].JobID); + fprintf (stderr, "\n"); + } + DelJob (job); + + return (TRUE); +} Index: /trunk/Ohana/src/opihi/pcontrol/init.c =================================================================== --- /trunk/Ohana/src/opihi/pcontrol/init.c (revision 4449) +++ /trunk/Ohana/src/opihi/pcontrol/init.c (revision 4450) @@ -1,12 +1,22 @@ # include "opihi.h" -int job PROTO((int, char **)); -int host PROTO((int, char **)); -int status PROTO((int, char **)); +int kill_pc PROTO((int, char **)); +int delete PROTO((int, char **)); +int job PROTO((int, char **)); +int host PROTO((int, char **)); +int check PROTO((int, char **)); +int status PROTO((int, char **)); +int stdout_pc PROTO((int, char **)); +int stderr_pc PROTO((int, char **)); static Command cmds[] = { - {"job", job, "add job"}, - {"host", host, "add / delete / modify host"}, - {"status", status, "get system status"}, + {"kill", kill_pc, "kill job"}, + {"delete", delete, "delete job"}, + {"job", job, "add job"}, + {"host", host, "add / delete / modify host"}, + {"check", check, "get job or host status"}, + {"status", status, "get system status"}, + {"stdout", stdout_pc, "get stdout buffer for job"}, + {"stderr", stderr_pc, "get stderr buffer for job"}, }; Index: /trunk/Ohana/src/opihi/pcontrol/job.c =================================================================== --- /trunk/Ohana/src/opihi/pcontrol/job.c (revision 4449) +++ /trunk/Ohana/src/opihi/pcontrol/job.c (revision 4450) @@ -26,4 +26,5 @@ Mode = PCONTROL_JOB_NEEDHOST; } + if (Host == NULL) Host = strcreate ("anyhost"); Timeout = 100; @@ -47,5 +48,5 @@ JobID = AddJob (Host, Mode, Timeout, targc, targv); - FREE (Host); + fprintf (stdout, "JobID: %d\n", JobID); return (TRUE); } Index: /trunk/Ohana/src/opihi/pcontrol/kill.c =================================================================== --- /trunk/Ohana/src/opihi/pcontrol/kill.c (revision 4449) +++ /trunk/Ohana/src/opihi/pcontrol/kill.c (revision 4450) @@ -1,5 +1,5 @@ # include "pcontrol.h" -int kill (int argc, char **argv) { +int kill_pc (int argc, char **argv) { Job *job; @@ -12,46 +12,10 @@ JobID = atoi (argv[1]); - N = FindJob (JobID, PCONTROL_JOB_BUSY); - if (N < 0) { + job = PullJob (JobID, PCONTROL_JOB_BUSY); + if (job == NULL) { fprintf (stderr, "job %s not BUSY\n", argv[1]); return (FALSE); } - job = GetJob (PCONTROL_JOB_BUSY, N); KillJob (job); return (TRUE); } - -int delete (int argc, char **argv) { - - Job *job; - int JobID, N; - - if (argc < 2) { - fprintf (stderr, "USAGE: delete (JobID)\n"); - return (FALSE); - } - JobID = atoi (argv[1]); - /* use a string interp to convert JobIDs to ints ? */ - - N = FindJob (JobID, PCONTROL_JOB_PENDING); - if (N >= 0) { - job = GetJob (PCONTROL_JOB_PENDING, N); - DelJob (job); - return (TRUE); - } - N = FindJob (JobID, PCONTROL_JOB_CRASH); - if (N >= 0) { - job = GetJob (PCONTROL_JOB_CRASH, N); - DelJob (job); - return (TRUE); - } - N = FindJob (JobID, PCONTROL_JOB_EXIT); - if (N >= 0) { - job = GetJob (PCONTROL_JOB_EXIT, N); - DelJob (job); - return (TRUE); - } - - fprintf (stderr, "job %s not PENDING, CRASH, EXIT\n", argv[1]); - return (FALSE); -} Index: /trunk/Ohana/src/opihi/pcontrol/pclient.c =================================================================== --- /trunk/Ohana/src/opihi/pcontrol/pclient.c (revision 4449) +++ /trunk/Ohana/src/opihi/pcontrol/pclient.c (revision 4450) @@ -13,5 +13,5 @@ /* send command to client (adding on \n) */ /* fprintf (stderr, "send: %s (%d)\n", command, buffer[0].Nbuffer); */ - ALLOCATE (line, char, MAX (1, strlen(command))); + ALLOCATE (line, char, MAX (1, strlen(command) + 1)); sprintf (line, "%s\n", command); status = write (host[0].stdin, line, strlen(line)); Index: /trunk/Ohana/src/opihi/pcontrol/pcontrol.c =================================================================== --- /trunk/Ohana/src/opihi/pcontrol/pcontrol.c (revision 4449) +++ /trunk/Ohana/src/opihi/pcontrol/pcontrol.c (revision 4450) @@ -45,5 +45,5 @@ rl_attempted_completion_function = command_completer; rl_event_hook = CheckSystem; - /* rl_set_keyboard_input_timeout (2000000); */ + rl_set_keyboard_input_timeout (1000000); set_str_variable ("HISTORY", opihi_history); Index: /trunk/Ohana/src/opihi/pcontrol/rconnect.c =================================================================== --- /trunk/Ohana/src/opihi/pcontrol/rconnect.c (revision 4449) +++ /trunk/Ohana/src/opihi/pcontrol/rconnect.c (revision 4450) @@ -79,7 +79,8 @@ status = ReadtoIOBuffer (&buffer, stdout_fd[0]); p = memstr (buffer.buffer, "CONNECTED", buffer.Nbuffer); - fprintf (stderr, "%d %d %s\n", i, buffer.Nbuffer, buffer.buffer); + /* fprintf (stderr, "%d %d %s\n", i, buffer.Nbuffer, buffer.buffer); */ usleep (20000); } + fprintf (stderr, "%d cycles to connect\n", i); if (status == 0) goto pipe_error; if (status == -1) goto io_error; Index: /trunk/Ohana/src/opihi/pcontrol/stdout.c =================================================================== --- /trunk/Ohana/src/opihi/pcontrol/stdout.c (revision 4450) +++ /trunk/Ohana/src/opihi/pcontrol/stdout.c (revision 4450) @@ -0,0 +1,63 @@ +# include "pcontrol.h" + +int stdout_pc (int argc, char **argv) { + + int N, JobID, stack; + Job *job; + IOBuffer *buffer; + + if (argc != 2) { + fprintf (stderr, "USAGE: stdout (JobID)\n"); + fprintf (stdout, "STATUS %d\n", -1); + return (FALSE); + } + + /* find Job of interest (must be EXIT or CRASH) */ + JobID = atoi (argv[1]); + job = FindJobPtr (JobID, PCONTROL_JOB_EXIT); + if (job == NULL) { + job = FindJobPtr (JobID, PCONTROL_JOB_CRASH); + if (job == NULL) goto missing_stdout; + } + + buffer = &job[0].stdout; + fwrite (buffer[0].buffer, 1, buffer[0].Nbuffer, stdout); + fprintf (stdout, "STATUS %d\n", 0); + return (TRUE); + +missing_stdout: + fprintf (stderr, "job not found in EXIT or CRASH\n"); + fprintf (stdout, "STATUS %d\n", -2); + return (FALSE); +} + +int stderr_pc (int argc, char **argv) { + + int N, JobID; + Job *job; + IOBuffer *buffer; + + if (argc != 2) { + fprintf (stderr, "USAGE: stderr (JobID)\n"); + fprintf (stdout, "STATUS %d\n", -1); + return (FALSE); + } + + /* find Job of interest (must be EXIT or CRASH) */ + JobID = atoi (argv[1]); + job = FindJobPtr (JobID, PCONTROL_JOB_EXIT); + if (job == NULL) { + job = FindJobPtr (JobID, PCONTROL_JOB_CRASH); + if (job == NULL) goto missing_stderr; + } + + buffer = &job[0].stderr; + fwrite (buffer[0].buffer, 1, buffer[0].Nbuffer, stdout); + fprintf (stdout, "STATUS %d\n", 0); + return (TRUE); + +missing_stderr: + fprintf (stderr, "job not found in EXIT or CRASH\n"); + fprintf (stdout, "STATUS %d\n", -2); + return (FALSE); +}