Index: /tags/pantasks-0-5/Ohana/src/opihi/pantasks/CheckController.c =================================================================== --- /tags/pantasks-0-5/Ohana/src/opihi/pantasks/CheckController.c (revision 8568) +++ /tags/pantasks-0-5/Ohana/src/opihi/pantasks/CheckController.c (revision 8568) @@ -0,0 +1,120 @@ +# include "pantasks.h" + +static struct timeval start; +void TimerMark (); +float TimerElapsed (int reset); + +int CheckController () { + + char *p, *q; + int i, Njobs, status, JobID; + Job *job; + IOBuffer buffer; + + /* get the list of completed jobs (exit / crash), update the job status */ + if (!CheckControllerStatus()) return (TRUE); + + /*** check EXIT jobs ***/ + InitIOBuffer (&buffer, 0x100); + // TimerMark (); + // status = ControllerCommand ("stop", CONTROLLER_PROMPT, &buffer); + // if (VerboseMode()) gprint (GP_ERR, "stop controller %f\n", TimerElapsed(TRUE)); + + TimerMark (); + FlushIOBuffer (&buffer); + status = ControllerCommand ("jobstack exit", CONTROLLER_PROMPT, &buffer); + if (VerboseMode()) gprint (GP_ERR, "check exit stack %f\n", TimerElapsed(TRUE)); + if (!status) goto escape; + + /** programming error **/ + p = memstr (buffer.buffer, "USAGE: jobstack", buffer.Nbuffer); + if (p != NULL) goto escape; + + /** parse job list **/ + status = sscanf (buffer.buffer, "%*s %d", &Njobs); + if (status != 1) goto escape; + if (VerboseMode()) gprint (GP_ERR, "parse %d jobs on stack %f\n", Njobs, TimerElapsed(TRUE)); + + p = buffer.buffer; + for (i = 0; (i < Njobs) && !TestElapsedCheck(); i++) { + q = strchr (p, '\n'); + if (q == NULL) { + gprint (GP_ERR, "controller message error: incomplete job list\n"); + break; + } + p = q + 1; + status = sscanf (p, "%d", &JobID); + + job = FindControllerJob (JobID); + if (job == NULL) { + gprint (GP_ERR, "misplaced job? %d not in EXIT job list\n", JobID); + continue; + } + /* this checks the individual job status, grabs stdout/stderr */ + CheckControllerJob (job); + } + if (VerboseMode()) gprint (GP_ERR, "clear %d exit jobs %f\n", i, TimerElapsed(TRUE)); + + if (TestElapsedCheck()) goto finish; + /* this will prevent us from ever checking crashed jobs... */ + + /*** check CRASH jobs ***/ + FlushIOBuffer (&buffer); + status = ControllerCommand ("jobstack crash", CONTROLLER_PROMPT, &buffer); + if (!status) goto escape; + + p = memstr (buffer.buffer, "USAGE: jobstack", buffer.Nbuffer); + if (p != NULL) goto escape; + + /** parse job list **/ + status = sscanf (buffer.buffer, "%*s %d", &Njobs); + if (status != 1) goto escape; + if (VerboseMode()) gprint (GP_ERR, "check crash stack %f\n", TimerElapsed(TRUE)); + + p = buffer.buffer; + for (i = 0; (i < Njobs) && !TestElapsedCheck(); i++) { + q = strchr (p, '\n'); + if (q == NULL) { + gprint (GP_ERR, "controller message error: incomplete job list\n"); + break; + } + p = q + 1; + + status = sscanf (p, "%d", &JobID); + job = FindControllerJob (JobID); + if (job == NULL) { + gprint (GP_ERR, "misplaced job? %d not in CRASH job list\n", JobID); + continue; + } + /* this checks the individual job status, grabs stdout/stderr */ + CheckControllerJob (job); + } + if (VerboseMode()) gprint (GP_ERR, "clear %d crash jobs %f\n", i, TimerElapsed(TRUE)); + + finish: + FlushIOBuffer (&buffer); + // status = ControllerCommand ("run", CONTROLLER_PROMPT, &buffer); + FreeIOBuffer (&buffer); + return (TRUE); + + escape: + FlushIOBuffer (&buffer); + // status = ControllerCommand ("run", CONTROLLER_PROMPT, &buffer); + FreeIOBuffer (&buffer); + return (FALSE); +} + +void TimerMark () { + gettimeofday (&start, (void *) NULL); +} + +float TimerElapsed (int reset) { + + float dtime; + struct timeval stop; + + gettimeofday (&stop, (void *) NULL); + dtime = DTIME (stop, start); + if (reset) gettimeofday (&start, (void *) NULL); + return (dtime); +} Index: /tags/pantasks-0-5/Ohana/src/opihi/pantasks/CheckJobs.c =================================================================== --- /tags/pantasks-0-5/Ohana/src/opihi/pantasks/CheckJobs.c (revision 8568) +++ /tags/pantasks-0-5/Ohana/src/opihi/pantasks/CheckJobs.c (revision 8568) @@ -0,0 +1,155 @@ +# include "pantasks.h" + +int CheckJobs () { + + Job *job; + Macro *macro; + int i, status; + char varname[64]; + + /** test all jobs: ready to test? finished? **/ + while ((job = NextJob ()) != NULL) { + + /* check poll period (ready to ask for status?) */ + if (GetTaskTimer(job[0].last) < job[0].task[0].poll_period) continue; + + /* check current status */ + status = CheckJob (job); + switch (status) { + case JOB_PENDING: + /* if (VerboseMode()) gprint (GP_LOG, "job %s (%d) pending\n", job[0].task[0].name, job[0].JobID); */ + break; + + case JOB_BUSY: + /* if (VerboseMode()) gprint (GP_LOG, "job %s (%d) busy\n", job[0].task[0].name, job[0].JobID); */ + break; + + case JOB_CRASH: + case JOB_EXIT: + /* push output buffer data to the stdout and stderr queues */ + /* XXX this will break on 0 values in output streams */ + PushNamedQueue ("stdout", job[0].stdout.buffer); + PushNamedQueue ("stderr", job[0].stderr.buffer); + + /* set taskarg variables */ + for (i = 0; i < job[0].argc; i++) { + sprintf (varname, "taskarg:%d", i); + set_str_variable (varname, job[0].argv[i]); + } + set_int_variable ("taskarg:n", job[0].argc); + + /* set options variables */ + for (i = 0; i < job[0].optc; i++) { + sprintf (varname, "options:%d", i); + set_str_variable (varname, job[0].optv[i]); + } + set_int_variable ("options:n", job[0].optc); + + if (status == JOB_CRASH) { + /* XXX add an Ncrash element? */ + job[0].task[0].Nfailure ++; + + /* run task[0].crash macro, if it exists */ + /* perhaps define PushNamedQueueBuffer */ + + if (VerboseMode()) gprint (GP_LOG, "job %s (%d) crash\n", job[0].task[0].name, job[0].JobID); + if (job[0].task[0].crash != NULL) { + exec_loop (job[0].task[0].crash); + } + } + if (status == JOB_EXIT) { + /* update the exit status counters */ + if (job[0].exit_status) { + job[0].task[0].Nfailure ++; + } else { + job[0].task[0].Nsuccess ++; + } + + /* run corresponding task[0].exit macro, if it exists */ + if (VerboseMode()) gprint (GP_LOG, "job %s (%d) exit\n", job[0].task[0].name, job[0].JobID); + macro = job[0].task[0].defexit; + 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: + if (VerboseMode()) gprint (GP_LOG, "unknown exit status: %d\n", status); + /** do something more useful here ?? **/ + break; + } + + /* check for timeout - (local jobs only) + we only check timeout after a poll (forces at least one poll) + */ + if (job[0].mode == JOB_LOCAL) { + if (GetTaskTimer(job[0].start) < job[0].task[0].timeout_period) continue; + if (VerboseMode()) gprint (GP_LOG, "timeout on %s\n", job[0].task[0].name); + + /* update the timeout counter */ + job[0].task[0].Ntimeout ++; + + if (!KillLocalJob (job)) { + job[0].state = JOB_HUNG; + if (VerboseMode()) gprint (GP_LOG, "child process %d is hung, cannot kill\n", job[0].pid); + continue; + } + + /* set taskarg variables */ + for (i = 0; i < job[0].argc; i++) { + sprintf (varname, "taskarg:%d", i); + set_str_variable (varname, job[0].argv[i]); + } + set_int_variable ("taskarg:n", job[0].argc); + + /* set options variables */ + for (i = 0; i < job[0].optc; i++) { + sprintf (varname, "options:%d", i); + set_str_variable (varname, job[0].optv[i]); + } + set_int_variable ("options:n", job[0].optc); + + /* 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; + } + + /* reset polling clock */ + SetTaskTimer (&job[0].last); + if (TestElapsedCheck()) return (TRUE); + } + return (TRUE); +} + +/* + + job / task timeline: + + task: + 0 exec + start create + task clock new job + + job: + 0 1xpoll 2xpoll 3xpoll + start check check check + job clock status status status + + . . . timeout + run + timeout + + must be at least one poll before timeout + (timeout >= poll) +*/ Index: /tags/pantasks-0-5/Ohana/src/opihi/pantasks/CheckPassword.c =================================================================== --- /tags/pantasks-0-5/Ohana/src/opihi/pantasks/CheckPassword.c (revision 8568) +++ /tags/pantasks-0-5/Ohana/src/opihi/pantasks/CheckPassword.c (revision 8568) @@ -0,0 +1,34 @@ +# include "pantasks.h" +# define DEBUG 0 + +static char PASSWORD[256]; + +int InitPassword () { + + VarConfig ("PASSWORD", "%s", PASSWORD); + return (TRUE); +} + +int CheckPassword (int BindSocket) { + + IOBuffer message; + int status; + + if (DEBUG) gprint (GP_ERR, "waiting for password %s\n", PASSWORD); + + status = ExpectCommand (BindSocket, strlen(PASSWORD), 0.1, &message); + if (status != 0) { + if (DEBUG) gprint (GP_ERR, "failed connection\n"); + FreeIOBuffer (&message); + close (BindSocket); + return (FALSE); + } + if (strncmp (message.buffer, PASSWORD, strlen(PASSWORD))) { + if (DEBUG) gprint (GP_ERR, "invalid password\n"); + close (BindSocket); + return (FALSE); + } + if (DEBUG) gprint (GP_ERR, "accepted password (%s)\n", message.buffer); + + return (TRUE); +} Index: /tags/pantasks-0-5/Ohana/src/opihi/pantasks/CheckSystem.c =================================================================== --- /tags/pantasks-0-5/Ohana/src/opihi/pantasks/CheckSystem.c (revision 8568) +++ /tags/pantasks-0-5/Ohana/src/opihi/pantasks/CheckSystem.c (revision 8568) @@ -0,0 +1,45 @@ +# include "pantasks.h" + +# define MAX_DELAY 0.1 +struct timeval start, last; +static int Ncheck = 0; + +int CheckSystem () { + + gettimeofday (&start, (void *) NULL); + + if (Ncheck < 100) { + CheckTasks (); + CheckJobs (); + Ncheck ++; + } else { + CheckController (); + CheckControllerOutput (); + Ncheck = 0; + } + /* ohana_memcheck (FALSE); */ + return (TRUE); +} + +int TestElapsedCheck () { + + struct timeval stop; + float dtime; + + gettimeofday (&stop, (void *) NULL); + dtime = DTIME (stop, start); + if (dtime > MAX_DELAY) return (TRUE); + return (FALSE); +} + + /* do we really need to check the controller on every call? + or perhaps alternate? + */ + +void SerialThreadLock () { + return; +} + +void SerialThreadUnlock () { + return; +} Index: /tags/pantasks-0-5/Ohana/src/opihi/pantasks/CheckTasks.c =================================================================== --- /tags/pantasks-0-5/Ohana/src/opihi/pantasks/CheckTasks.c (revision 8568) +++ /tags/pantasks-0-5/Ohana/src/opihi/pantasks/CheckTasks.c (revision 8568) @@ -0,0 +1,42 @@ +# include "pantasks.h" + +int CheckTasks () { + + Job *job; + Task *task; + int status; + + /** 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; + + /* need to check if the current time is within valid/invalid periods */ + if (!CheckTimeRanges (task[0].ranges, task[0].Nranges)) continue; + if (task[0].Nmax && (task[0].Njobs >= task[0].Nmax)) continue; + + /* ready to run? : run task.exec macro */ + if (task[0].exec != NULL) { + status = exec_loop (task[0].exec); + if (!status) continue; + } + if (!ValidateTask (task, TRUE)) continue; + + /* 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) (why not??) */ + gettimeofday (&task[0].last, (void *) NULL); + task[0].Njobs ++; + + /* increment Nrun for inclusive ranges with Nmax */ + BumpTimeRanges (task[0].ranges, task[0].Nranges); + + if (TestElapsedCheck()) return (TRUE); + } + return (TRUE); +} Index: /tags/pantasks-0-5/Ohana/src/opihi/pantasks/CheckTimeRanges.c =================================================================== --- /tags/pantasks-0-5/Ohana/src/opihi/pantasks/CheckTimeRanges.c (revision 8568) +++ /tags/pantasks-0-5/Ohana/src/opihi/pantasks/CheckTimeRanges.c (revision 8568) @@ -0,0 +1,97 @@ +# include "pantasks.h" + +/* the tested time is saved by CheckTimeRanges for a following BumpTimeRanges + otherwise we could have an inconsistency between valid ranges and Nrun */ +static time_t daytime, weektime, abstime; + +/* test if we meet all time range qualifications */ +int CheckTimeRanges (TimeRange *ranges, int Nranges) { + + int i, intime, Ninclude, include, exclude, valid; + time_t testtime; + struct timeval now; + struct tm Now; + + /* get the current time */ + gettimeofday (&now, NULL); + gmtime_r (&now.tv_sec, &Now); + daytime = Now.tm_sec + Now.tm_min*60 + Now.tm_hour*3600; + weektime = Now.tm_sec + Now.tm_min*60 + Now.tm_hour*3600 + Now.tm_wday*86400; + abstime = now.tv_sec; + + Ninclude = 0; + include = FALSE; + exclude = FALSE; + + for (i = 0; i < Nranges; i++) { + if (ranges[i].include) Ninclude ++; + + switch (ranges[i].type) { + /* set the testtime */ + case RANGE_ABS: + testtime = abstime; + break; + case RANGE_DAY: + testtime = daytime; + break; + case RANGE_WEEK: + testtime = weektime; + break; + default: + abort (); + } + intime = (testtime >= ranges[i].start) && (testtime <= ranges[i].stop); + + /* check for more than max runs in time range */ + if (ranges[i].include && intime && ranges[i].Nmax) { + if (ranges[i].Nrun >= ranges[i].Nmax) return (FALSE); + } + /* reset Nrun if we are outside of intime */ + if (ranges[i].include && !intime && ranges[i].Nmax && ranges[i].Nrun) { + ranges[i].Nrun = 0; + } + + /* is this a valid time? */ + if ( ranges[i].include && intime) include = TRUE; + if (!ranges[i].include && !intime) exclude = TRUE; + } + + if (Ninclude == 0) include = TRUE; + valid = include && !exclude; + + return (valid); +} + +/* increment the number of runs for all inclusive time ranges with Nmax > 0 + (only call when we execute a task -- after CheckTimeRanges) */ +int BumpTimeRanges (TimeRange *ranges, int Nranges) { + + int i, intime; + time_t testtime; + + /* only increment the counter for ranges which are valid */ + for (i = 0; i < Nranges; i++) { + if (!ranges[i].Nmax) continue; + if (!ranges[i].include) continue; + + switch (ranges[i].type) { + /* set the testtime */ + case RANGE_ABS: + testtime = abstime; + break; + case RANGE_DAY: + testtime = daytime; + break; + case RANGE_WEEK: + testtime = weektime; + break; + default: + abort (); + } + intime = (testtime >= ranges[i].start) && (testtime <= ranges[i].stop); + + /* reset Nrun if we are outside of intime */ + if (intime) ranges[i].Nrun ++; + } + return (TRUE); +} Index: /tags/pantasks-0-5/Ohana/src/opihi/pantasks/ControllerOps.c =================================================================== --- /tags/pantasks-0-5/Ohana/src/opihi/pantasks/ControllerOps.c (revision 8568) +++ /tags/pantasks-0-5/Ohana/src/opihi/pantasks/ControllerOps.c (revision 8568) @@ -0,0 +1,476 @@ +# include "pantasks.h" +/* adding a new host can delay controller up to a second or so */ +# define CONTROLLER_TIMEOUT 500 +# define CONNECT_TIMEOUT 300 + +/* local static variables to hold the connection to the controller */ +static int ControllerStatus = FALSE; +static int stdin_cntl, stdout_cntl, stderr_cntl; +static IOBuffer stdout_buffer; +static IOBuffer stderr_buffer; +static int ControllerPID = 0; + +/* test if the controller is running */ +int CheckControllerStatus () { + return (ControllerStatus); +} + +/* check job / get output if done */ +int CheckControllerJob (Job *job) { + struct timeval start, stop; + float dtime; + + gettimeofday (&start, (void *) NULL); + CheckControllerJobStatus (job); + gettimeofday (&stop, (void *) NULL); + dtime = DTIME (stop, start); + /* if (VerboseMode()) gprint (GP_ERR, "check job status %f\n", dtime); */ + + if ((job[0].state == JOB_EXIT) || (job[0].state == JOB_CRASH)) { + gettimeofday (&start, (void *) NULL); + GetJobOutput ("stdout", job[0].pid, &job[0].stdout, job[0].stdout_size); + gettimeofday (&stop, (void *) NULL); + dtime = DTIME (stop, start); + /* if (VerboseMode()) gprint (GP_ERR, "get stdout %f\n", dtime); */ + + gettimeofday (&start, (void *) NULL); + GetJobOutput ("stderr", job[0].pid, &job[0].stderr, job[0].stderr_size); + gettimeofday (&stop, (void *) NULL); + dtime = DTIME (stop, start); + /* if (VerboseMode()) gprint (GP_ERR, "get stderr %f\n", dtime); */ + + gettimeofday (&start, (void *) NULL); + DeleteControllerJob (job); + gettimeofday (&stop, (void *) NULL); + dtime = DTIME (stop, start); + /* if (VerboseMode()) gprint (GP_ERR, "delete job %f\n", dtime); */ + } + return (TRUE); +} + +int DeleteControllerJob (Job *job) { + + int status; + char cmd[128]; + IOBuffer buffer; + + sprintf (cmd, "delete %d", job[0].pid); + InitIOBuffer (&buffer, 0x100); + status = ControllerCommand (cmd, CONTROLLER_PROMPT, &buffer); + FreeIOBuffer (&buffer); + return (status); +} + +/* ask controller about job status */ +int CheckControllerJobStatus (Job *job) { + + int outstate, status; + 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); + if (!status) { + FreeIOBuffer (&buffer); + return (FALSE); + } + + /** was this a valid job? **/ + p = memstr (buffer.buffer, "job not found", buffer.Nbuffer); + if (p != NULL) { + gprint (GP_ERR, "unknown job %d\n", job[0].pid); + FreeIOBuffer (&buffer); + 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) { + gprint (GP_ERR, "programming error?\n"); + exit (1); + } + job[0].state = outstate; + return (TRUE); +} + +/* 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; + struct timespec request, remain; + + /* avoid blocking on waitpid, test every 100 usec, up to 50 msec */ + request.tv_sec = 0; + request.tv_nsec = 100000; + + /* flush any earlier messages */ + ReadtoIOBuffer (buffer, stdout_cntl); + FlushIOBuffer (buffer); + Nstart = buffer[0].Nbuffer; + + /* send command to get appropriate channel */ + status = write_fmt (stdin_cntl, "%s %d\n", cmd, pid); + + /* 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 != 0) && (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) nanosleep (&request, &remain); + } + if (status == 0) return (CONTROLLER_DOWN); + if (status == -1) return (CONTROLLER_HUNG); + + /* if (VerboseMode()) gprint (GP_ERR, "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); +} + +/* submitting a job to the controller automatically starts controller */ +int SubmitControllerJob (Job *job) { + + int i, Nchar, status; + char *cmd, *p, string[64]; + IOBuffer buffer; + + if (job[0].task[0].host == NULL) return (FALSE); + + if (!StartController ()) { + gprint (GP_ERR, "failure to start pcontrol\n"); + return (FALSE); + } + + /** 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 (!strcasecmp (job[0].task[0].host, "ANYHOST")) { + sprintf (cmd, "job"); + } 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); + status = ControllerCommand (cmd, CONTROLLER_PROMPT, &buffer); + free (cmd); + + /* extract the job PID from the controller response */ + p = memstr (buffer.buffer, "JobID", buffer.Nbuffer); + if (p == NULL) { + gprint (GP_ERR, "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, status; + int stdin_fd[2], stdout_fd[2], stderr_fd[2]; + IOBuffer buffer; + + if (ControllerStatus) 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 */ + gprint (GP_LOG, "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 */ + status = write_fmt (stdin_fd[1], "echo CONNECTED\n"); + if ((status == -1) && (errno == EPIPE)) goto pipe_error; + + /* try to get evidence connection is alive - wait upto a few seconds */ + /* connection is likely slow; don't bother with nanosleep here */ + p = NULL; + status = -1; + for (i = 0; (i < CONNECT_TIMEOUT) && (status != 0) && (p == NULL); i++) { + status = ReadtoIOBuffer (&buffer, stdout_fd[0]); + p = memstr (buffer.buffer, "CONNECTED", buffer.Nbuffer); + usleep (10000); // wait for controller to start up + } + 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]; + + InitIOBuffer (&stdout_buffer, 0x100); + InitIOBuffer (&stderr_buffer, 0x100); + + ControllerPID = pid; + ControllerStatus = TRUE; + gprint (GP_LOG, "Connected\n"); + return (TRUE); + +pipe_error: + perror ("pipe error:"); + goto close_pipes; + +io_error: + gprint (GP_ERR, "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, status; + char *line; + struct timespec request, remain; + + /* avoid blocking on waitpid, test every 100 usec, up to 50 msec */ + request.tv_sec = 0; + request.tv_nsec = 100000; + + ReadtoIOBuffer (buffer, stdout_cntl); + FlushIOBuffer (buffer); + + /* send command, is pipe still open? */ + status = write_fmt (stdin_cntl, "%s\n", cmd); + if ((status == -1) && (errno == EPIPE)) { + StopController (); + if (VerboseMode()) gprint (GP_ERR, "controller is down\n"); + return (FALSE); + } + + /* watch for response - wait up to 1 second */ + line = NULL; + status = -1; + for (i = 0; (i < CONTROLLER_TIMEOUT) && (status != 0) && (line == NULL); i++) { + status = ReadtoIOBuffer (buffer, stdout_cntl); + line = memstr (buffer[0].buffer, response, buffer[0].Nbuffer); + if (status == -1) nanosleep (&request, &remain); + } + if (status == 0) { + StopController (); + if (VerboseMode()) gprint (GP_ERR, "controller is down\n"); + return (FALSE); + } + if (status == -1) { + if (VerboseMode()) gprint (GP_ERR, "controller is not responding\n"); + return (FALSE); + } + + /* need to strip off the prompt */ + line = memstr (buffer[0].buffer, response, buffer[0].Nbuffer); + if (line != NULL) { + buffer[0].Nbuffer = line - buffer[0].buffer; + bzero (buffer[0].buffer + buffer[0].Nbuffer, buffer[0].Nalloc - buffer[0].Nbuffer); + } + /* if (VerboseMode()) gprint (GP_ERR, "message received, %d cycles\n", i); */ + return (TRUE); +} + +int CheckControllerOutput () { + + int Nread; + + if (!ControllerStatus) return (TRUE); + + /* read stdout buffer */ + Nread = ReadtoIOBuffer (&stdout_buffer, stdout_cntl); + switch (Nread) { + case -2: /* error in read (programming error? system level error?) */ + gprint (GP_ERR, "serious IO error\n"); + exit (2); + case -1: /* no data in pipe */ + case 0: /* pipe is closed, change child state? **/ + default: /* data in pipe */ + break; + } + + /* read stderr buffer */ + Nread = ReadtoIOBuffer (&stderr_buffer, stderr_cntl); + switch (Nread) { + case -2: /* error in read (programming error? system level error?) */ + gprint (GP_ERR, "serious IO error\n"); + exit (2); + case -1: /* no data in pipe */ + case 0: /* pipe is closed, change child state? **/ + default: /* data in pipe */ + break; + } + return (TRUE); +} + +int PrintControllerOutput () { + + gprint (GP_LOG, "--- stdout ---\n"); + gwrite (stdout_buffer.buffer, 1, stdout_buffer.Nbuffer, GP_LOG); + gprint (GP_LOG, "--- stderr ---\n"); + gwrite (stderr_buffer.buffer, 1, stderr_buffer.Nbuffer, GP_LOG); + gprint (GP_LOG, "--- done ---\n"); + return (TRUE); +} + + +int KillControllerJob (Job *job) { + + int status; + char cmd[128]; + IOBuffer buffer; + + if (!ControllerStatus) return (TRUE); + + sprintf (cmd, "kill %d", job[0].pid); + InitIOBuffer (&buffer, 0x100); + status = ControllerCommand (cmd, CONTROLLER_PROMPT, &buffer); + FreeIOBuffer (&buffer); + return (status); + + /** need to interpret output message & free things **/ +} + +int QuitController () { + + int status; + char cmd[128]; + IOBuffer buffer; + + if (!ControllerStatus) return (TRUE); + + sprintf (cmd, "quit"); + InitIOBuffer (&buffer, 0x100); + status = ControllerCommand (cmd, "", &buffer); + FreeIOBuffer (&buffer); + + /* the quit command does not return a prompt, so we always + get an error on the controller here */ + StopController (); + return (TRUE); +} + +int StopController () { + + int i, waitstatus, result; + + if (!ControllerStatus) return (TRUE); + + ControllerStatus = FALSE; + result = waitpid (ControllerPID, &waitstatus, WNOHANG); + for (i = 0; (i < 10) && (result == 0); i++) { + usleep (10000); // wait for controller to exit + result = waitpid (ControllerPID, &waitstatus, WNOHANG); + } + ControllerPID = 0; + close (stdin_cntl); + close (stdout_cntl); + close (stderr_cntl); + FreeIOBuffer (&stdout_buffer); + FreeIOBuffer (&stderr_buffer); + return (TRUE); +} Index: /tags/pantasks-0-5/Ohana/src/opihi/pantasks/InputQueue.c =================================================================== --- /tags/pantasks-0-5/Ohana/src/opihi/pantasks/InputQueue.c (revision 8568) +++ /tags/pantasks-0-5/Ohana/src/opihi/pantasks/InputQueue.c (revision 8568) @@ -0,0 +1,74 @@ +# include "pantasks.h" +# define DEBUG 1 + +static int Ninputs = 0; +static int NINPUTS = 0; +static char **inputs; + +void InitInputs () { + + Ninputs = 0; + NINPUTS = 10; + ALLOCATE (inputs, char *, NINPUTS); +} + +/* add this input to the inputs table */ +void AddNewInput (char *input) { + + SerialThreadLock (); + if (DEBUG) gprint (GP_LOG, "adding a new input (%s)\n", input); + inputs[Ninputs] = input; + Ninputs ++; + if (Ninputs >= NINPUTS - 1) { + NINPUTS += 10; + REALLOCATE (inputs, char *, NINPUTS); + } + if (DEBUG) gprint (GP_LOG, "done new input (%s)\n", input); + SerialThreadUnlock (); +} + +/* remove this input from the inputs table */ +int DeleteInput (char *input) { + + int i, j; + + if (DEBUG) gprint (GP_LOG, "deleting an input (%s)\n", input); + for (i = 0; i < Ninputs; i++) { + if (inputs[i] == input) { + free (inputs[i]); + for (j = i; j < Ninputs - 1; j++) { + inputs[j] = inputs[j+1]; + } + Ninputs --; + if ((Ninputs > 10) && (Ninputs / 2 < NINPUTS)) { + NINPUTS = Ninputs + 10; + REALLOCATE (inputs, char *, NINPUTS); + } + if (DEBUG) gprint (GP_LOG, "deleted an input\n"); + return TRUE; + } + } + // did not find the input + return FALSE; +} + +/* if any inputs are pending, run one */ +void CheckInputs () { + + int Nbytes, status; + char *input, *line, *outline, tmp; + + if (Ninputs < 1) return; + + input = inputs[0]; + if (DEBUG) gprint (GP_LOG, "got an input (%s)\n", input); + + Nbytes = snprintf (&tmp, 0, "input %s", input); + ALLOCATE (line, char, Nbytes + 1); + snprintf (line, Nbytes + 1, "input %s", input); + + status = command (line, &outline, TRUE); + free (outline); + + DeleteInput (input); +} Index: /tags/pantasks-0-5/Ohana/src/opihi/pantasks/JobIDOps.c =================================================================== --- /tags/pantasks-0-5/Ohana/src/opihi/pantasks/JobIDOps.c (revision 8568) +++ /tags/pantasks-0-5/Ohana/src/opihi/pantasks/JobIDOps.c (revision 8568) @@ -0,0 +1,41 @@ +# include "pantasks.h" + +# define MAX_N_JOBS 1000 +static char *JobIDList; +static int JobIDPtr; + +int InitJobIDs () { + + JobIDPtr = 0; + ALLOCATE (JobIDList, char, MAX_N_JOBS); + bzero (JobIDList, MAX_N_JOBS*sizeof(char)); + return (TRUE); +} + +/* return next unique ID, recycle every MAX_N_JOBS */ +int NextJobID () { + + int Ntry; + + JobIDPtr ++; + if (JobIDPtr >= MAX_N_JOBS) JobIDPtr = 0; + + Ntry = 0; + while (JobIDList[JobIDPtr]) { + Ntry ++; + JobIDPtr ++; + if (JobIDPtr >= MAX_N_JOBS) JobIDPtr = 0; + if (Ntry == MAX_N_JOBS) return (-1); + } + JobIDList[JobIDPtr] = TRUE; + return (JobIDPtr); +} + +int FreeJobID (int JobID) { + + if (JobID < 0) return (FALSE); + if (JobID >= MAX_N_JOBS) return (FALSE); + + JobIDList[JobID] = FALSE; + return (TRUE); +} Index: /tags/pantasks-0-5/Ohana/src/opihi/pantasks/JobOps.c =================================================================== --- /tags/pantasks-0-5/Ohana/src/opihi/pantasks/JobOps.c (revision 8568) +++ /tags/pantasks-0-5/Ohana/src/opihi/pantasks/JobOps.c (revision 8568) @@ -0,0 +1,212 @@ +# include "pantasks.h" + +static Job **jobs; +static int Njobs; +static int NJOBS; + +/* counter marking job being visited by the run loop */ +static int ActiveJob; + +/* set up the jobs list */ +void InitJobs () { + NJOBS = 20; + Njobs = 0; + ALLOCATE (jobs, Job *, NJOBS); + ActiveJob = -1; +} + +/* provide a mechanism to loop over the list of jobs */ +Job *NextJob () { + + Job *job; + + ActiveJob ++; + if (ActiveJob < 0) ActiveJob = 0; + if (ActiveJob >= Njobs) { + ActiveJob = -1; + return (NULL); + } + job = jobs[ActiveJob]; + return (job); +} + +/* return job with given ID */ +Job *FindJob (int JobID) { + + int i; + + /* return job with matching JobID */ + for (i = 0; i < Njobs; i++) { + if (jobs[i][0].JobID == JobID) { + return (jobs[i]); + } + } + return (NULL); +} + +/* return job with given controller Job ID */ +Job *FindControllerJob (int JobID) { + + int i; + + /* return job with matching JobID */ + for (i = 0; i < Njobs; i++) { + if (jobs[i][0].mode == JOB_LOCAL) continue; + if (jobs[i][0].pid == JobID) { + return (jobs[i]); + } + } + return (NULL); +} + +/* list known jobs */ +void ListJobs () { + + int i; + + gprint (GP_LOG, "\n"); + if (Njobs == 0) { + gprint (GP_LOG, " no defined jobs\n"); + return; + } + + gprint (GP_LOG, " Jobs\n"); + for (i = 0; i < Njobs; i++) { + gprint (GP_LOG, " %d: %-15s %5d %20s (%lx)\n", Njobs, jobs[i][0].task[0].name, jobs[i][0].JobID, jobs[i][0].argv[0], (long) jobs[i][0].argv); + } + return; +} + +/* make a new job from a task */ +Job *CreateJob (Task *task) { + + int i; + Job *job; + + ALLOCATE (job, Job, 1); + + job[0].JobID = NextJobID (); + job[0].pid = 0; + job[0].mode = JOB_LOCAL; + if (task[0].host != NULL) { + job[0].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 + */ + job[0].argc = task[0].argc; + ALLOCATE (job[0].argv, char *, MAX (task[0].argc + 1, 1)); + for (i = 0; i < task[0].argc; i++) { + job[0].argv[i] = strcreate (task[0].argv[i]); + } + job[0].argv[i] = 0; + + /* we need our own copy of task[0].optv: optc is the number of valid opts. */ + job[0].optc = task[0].optc; + ALLOCATE (job[0].optv, char *, MAX (task[0].optc, 1)); + for (i = 0; i < job[0].optc; i++) { + job[0].optv[i] = strcreate (task[0].optv[i]); + } + + /* Other data from the task is needed by the job. We carry a pointer back to the task. Changes to an + executing task are applied to the existing jobs (exit macros, poll_period, timeout) */ + + job[0].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 */ + + InitIOBuffer (&job[0].stdout, 0x100); + InitIOBuffer (&job[0].stderr, 0x100); + + jobs[Njobs] = job; + Njobs ++; + if (Njobs == NJOBS) { + NJOBS += 20; + REALLOCATE (jobs, Job *, NJOBS); + } + return (jobs[Njobs-1]); +} + +void FreeJob (Job *job) { + + int i; + + if (job == NULL) return; + + FreeJobID (job[0].JobID); + + for (i = 0; i < job[0].argc; i++) { + free (job[0].argv[i]); + } + free (job[0].argv); + + for (i = 0; i < job[0].optc; i++) { + free (job[0].optv[i]); + } + free (job[0].optv); + + FreeIOBuffer (&job[0].stdout); + FreeIOBuffer (&job[0].stderr); + free (job); + return; +} + +/* delete the job from the job list & adjust ActiveJob counter */ +int DeleteJob (Job *job) { + + int i, Nm; + + Nm = -1; + for (i = 0; i < Njobs; i++) { + if (job == jobs[i]) { + Nm = i; + break; + } + } + if (Nm == -1) { + gprint (GP_ERR, "programming error: job not found\n"); + return (FALSE); + } + + FreeJob (jobs[Nm]); + for (i = Nm; i < Njobs - 1; i++) { + jobs[i] = jobs[i + 1]; + } + Njobs --; + + /* adjust active job number */ + if (ActiveJob >= Nm) { + ActiveJob --; + } + return (TRUE); +} + +int SubmitJob (Job *job) { + + if (job[0].mode == JOB_LOCAL) { + SubmitLocalJob (job); + } else { + SubmitControllerJob (job); + } + + /* reset clock for start and poll-test */ + gettimeofday (&job[0].start, (void *) NULL); + job[0].last = job[0].start; + job[0].state = JOB_PENDING; + return (TRUE); +} + +int CheckJob (Job *job) { + + /* add checks for timeouts */ + + if (job[0].mode == JOB_LOCAL) { + CheckLocalJob (job); + } else { + /* controller jobs are now checked en masse by CheckController */ + /* CheckControllerJob (job); */ + } + return (job[0].state); +} Index: /tags/pantasks-0-5/Ohana/src/opihi/pantasks/ListenClients.c =================================================================== --- /tags/pantasks-0-5/Ohana/src/opihi/pantasks/ListenClients.c (revision 8568) +++ /tags/pantasks-0-5/Ohana/src/opihi/pantasks/ListenClients.c (revision 8568) @@ -0,0 +1,162 @@ +# include "pantasks.h" +# define DEBUG 0 + +static int NCLIENTS; +static int Nclients; +static int *clients; +static IOBuffer **buffers; + +void InitClients () { + + Nclients = 0; + NCLIENTS = 10; + ALLOCATE (clients, int, NCLIENTS); + ALLOCATE (buffers, IOBuffer *, NCLIENTS); +} + +/* add this client to the client table, create an IOBuffer for it */ +void AddNewClient (int client) { + + if (DEBUG) gprint (GP_LOG, "adding a new client (%d)\n", client); + clients[Nclients] = client; + ALLOCATE (buffers[Nclients], IOBuffer, 1); + InitIOBuffer(buffers[Nclients], 256); + Nclients ++; + if (Nclients >= NCLIENTS - 1) { + NCLIENTS += 10; + REALLOCATE (clients, int, NCLIENTS); + REALLOCATE (buffers, IOBuffer *, NCLIENTS); + } +} + +/* add this client to the client table, create an IOBuffer for it */ +int DeleteClient (int client) { + + int i, j; + + if (DEBUG) gprint (GP_LOG, "deleting a client (%d)\n", client); + for (i = 0; i < Nclients; i++) { + if (clients[i] == client) { + FreeIOBuffer (buffers[i]); + free (buffers[i]); + close (clients[i]); + for (j = i; j < Nclients - 1; j++) { + clients[j] = clients[j+1]; + buffers[j] = buffers[j+1]; + } + Nclients --; + if ((Nclients > 10) && (Nclients / 2 < NCLIENTS)) { + NCLIENTS = Nclients + 10; + REALLOCATE (clients, int, NCLIENTS); + REALLOCATE (buffers, IOBuffer *, NCLIENTS); + } + return TRUE; + } + } + // did not find the client + return FALSE; +} + +/* select for messages from the current clients; wait for 0.5s before updating client list */ +void *ListenClients (void *data) { + + int i, Ncurrent, Nmax, status, Nread; + char *line; + fd_set fdSet; + struct timeval timeout; + IOBuffer *outbuffer; + + InitClients (); + gprintInit (); // each thread needs to init the printing system + + while (1) { + + /* Wait up to 0.5 second - need to timeout to update client list */ + /* timeout gets mucked: need to reset before each select */ + timeout.tv_sec = 0; + timeout.tv_usec = 500000; + + /* place all of the clients in the fdSet */ + Ncurrent = Nclients; + Nmax = 0; + FD_ZERO (&fdSet); + for (i = 0; i < Ncurrent; i++) { + Nmax = MAX (Nmax, clients[i]); + FD_SET (clients[i], &fdSet); + } + Nmax ++; + + /* block until we have some data on the pipes (or timeout) */ + if (DEBUG) gprint (GP_ERR, "listening to %d clients\n", Ncurrent); + status = select (Nmax, &fdSet, NULL, NULL, &timeout); + if (status == -1) { + perror("select()"); + return (FALSE); + } + + /* if no data, update client list, wait for another select */ + if (status <= 0) continue; + + /* loop over the clients with data */ + for (i = 0; i < Ncurrent; i++) { + /* if client has no data, skip it */ + if (!FD_ISSET(clients[i], &fdSet)) continue; + + /* read until the pipe is empty: 0 is closed, -1 is empty, -2 is error */ + Nread = 1; + while (Nread > 0) { + Nread = ReadtoIOBuffer (buffers[i], clients[i]); + } + if ((Nread == 0) || (Nread == -2)) { + /* error: do something */ + if (DEBUG && (Nread == 0)) gprint (GP_ERR, "socket is closed\n"); + if (DEBUG && (Nread == -2)) gprint (GP_ERR, "error reading from socket\n"); + DeleteClient (clients[i]); + continue; + } + + if (DEBUG) gprint (GP_ERR, "read %d total bytes\n", buffers[i][0].Nbuffer); + + /* see if we have a complete message waiting; if not, keep waiting for messages */ + line = CheckForMessage (buffers[i]); + if (line == NULL) continue; + + /* we now have a possible command from the client: run it */ + /* in this thread, we set the print output destination to be an + internal buffer, which we dump at the end of the execution */ + /* the commands sent to the server should not have ; */ + stripwhite (line); + if (*line) { + + /* set buffers for the output for this client */ + gprintSetBuffer (GP_LOG); + gprintSetBuffer (GP_ERR); + + /* run the command, return the exit status */ + status = multicommand (line); + SendMessage (clients[i], "STATUS %d", status); + + // return the stderr messages first + outbuffer = gprintGetBuffer (GP_ERR); + SendMessageFixed (clients[i], outbuffer[0].Nbuffer, outbuffer[0].buffer); + + // return the stdout messages first + outbuffer = gprintGetBuffer (GP_LOG); + SendMessageFixed (clients[i], outbuffer[0].Nbuffer, outbuffer[0].buffer); + + /* clear and reset the output buffers */ + gprintSetFile (GP_LOG, "stdout"); + gprintSetFile (GP_ERR, "stderr"); + } + free (line); + } + /* if Nread == -2, we probably need to drop the client */ + /* check if we need to drop / remove any clients */ + /* check if we need to shut down the thread */ + } +} + +/* the AddClient commands are issued by the parent thread + the value of Nclients may increase after we check it here. + only this thread is allowed to decrease Nclients and remove + a client from the table */ Index: /tags/pantasks-0-5/Ohana/src/opihi/pantasks/LocalJob.c =================================================================== --- /tags/pantasks-0-5/Ohana/src/opihi/pantasks/LocalJob.c (revision 8568) +++ /tags/pantasks-0-5/Ohana/src/opihi/pantasks/LocalJob.c (revision 8568) @@ -0,0 +1,185 @@ +# include "pantasks.h" + +/* local jobs are forked in the background + we might need to limit the maximum number of local jobs. + should we have a queue/stack of pending local jobs, much + like controller? */ + +/* update current state, drain stdout/stderr buffers */ +int CheckLocalJob (Job *job) { + + int Nread; + + // XXX do something useful with exit status? + CheckLocalJobStatus (job); + + if ((job[0].state == JOB_EXIT) || (job[0].state == JOB_CRASH)) { + EmptyIOBuffer (&job[0].stdout, 10, job[0].stdout_fd); + EmptyIOBuffer (&job[0].stderr, 10, job[0].stderr_fd); + close (job[0].stdout_fd); + close (job[0].stderr_fd); + } else { + /* read stdout buffer */ + Nread = ReadtoIOBuffer (&job[0].stdout, job[0].stdout_fd); + switch (Nread) { + case -2: /* error in read (programming error? system level error?) */ + gprint (GP_ERR, "serious IO error\n"); + exit (2); + case -1: /* no data in pipe */ + case 0: /* pipe is closed, change child state? **/ + 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?) */ + gprint (GP_ERR, "serious IO error\n"); + exit (2); + case -1: /* no data in pipe */ + case 0: /* pipe is closed, change child state? **/ + 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: + gprint (GP_ERR, "unknown PID, not a child proc\n"); + gprint (GP_ERR, "did process already exit? programming error?\n"); + job[0].state = JOB_NONE; + job[0].exit_status = 0; + return (FALSE); + case EINVAL: + gprint (GP_ERR, "error EINVAL (waitpid): programming error\n"); + exit (1); + case EINTR: + gprint (GP_ERR, "error EINTR (waitpid): programming error\n"); + exit (1); + default: + gprint (GP_ERR, "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) { + gprint (GP_ERR, "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)) { + gprint (GP_ERR, "waitpid returns 'stopped': programming error\n"); + exit (1); + } + } + return (FALSE); +} + +/* this could be written a just a one-way pipe */ +int SubmitLocalJob (Job *job) { + + int status, pid; + int stdout_fd[2], stderr_fd[2]; + + 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 */ + if (VerboseMode()) gprint (GP_ERR, "starting local job\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); +} + +/* should this function close the fd's? */ +int KillLocalJob (Job *job) { + + int i, result, waitstatus; + + if (job[0].state != JOB_BUSY) return (TRUE); + + /* send SIGTERM signal to job */ + kill (job[0].pid, SIGTERM); + result = 0; + for (i = 0; (i < 10) && (result == 0); i++) { + usleep (10000); // wait for job to exit + result = waitpid (job[0].pid, &waitstatus, WNOHANG); + } + if (result) return (TRUE); + + /* send SIGKILL signal to job */ + kill (job[0].pid, SIGKILL); + result = 0; + for (i = 0; (i < 10) && (result == 0); i++) { + usleep (10000); // wait for job to exit + result = waitpid (job[0].pid, &waitstatus, WNOHANG); + } + if (result) return (TRUE); + + /* total failure, don't reset */ + return (FALSE); +} + Index: /tags/pantasks-0-5/Ohana/src/opihi/pantasks/Makefile =================================================================== --- /tags/pantasks-0-5/Ohana/src/opihi/pantasks/Makefile (revision 8568) +++ /tags/pantasks-0-5/Ohana/src/opihi/pantasks/Makefile (revision 8568) @@ -0,0 +1,114 @@ +include ../../../Configure + +HOME = $(ROOT)/src/opihi +BIN = $(HOME)/bin +LIB = $(HOME)/lib +INC = $(HOME)/include +SDIR = $(HOME)/pantasks +# +DESTBIN = $(LBIN) +DESTLIB = $(LLIB) +DESTINC = $(LINC) +DESTMAN = $(LMAN) +DESTHLP = $(LHLP)/pantasks + +# compile flags +INCS = -I$(INC) -I$(LINC) -I$(XINC) +CFLAGS = $(INCS) -DHELPDIR_DEFAULT=$(DESTHLP) + +# link flags +LIBS = -L$(LIB) -L$(LLIB) -L$(XLIB) +LIBS1 = -lsocket -lnsl -lreadline $(TLIB) -ldvo -lkapa -lFITS -lohana -lX11 -lpthread -lm +LIBS2 = -lbasiccmd -ldatacmd -lastrocmd -lshell -ldata +LFLAGS = $(LIBS) $(LIBS2) $(LIBS1) + +# sched user commands and support functions ######################## + +single = \ +$(SDIR)/init.$(ARCH).o \ +$(SDIR)/run.$(ARCH).o \ +$(SDIR)/stop.$(ARCH).o \ +$(SDIR)/pantasks.$(ARCH).o \ +$(SDIR)/CheckSystem.$(ARCH).o + +funcs = \ +$(SDIR)/CheckJobs.$(ARCH).o \ +$(SDIR)/CheckController.$(ARCH).o \ +$(SDIR)/CheckTasks.$(ARCH).o \ +$(SDIR)/CheckTimeRanges.$(ARCH).o \ +$(SDIR)/ControllerOps.$(ARCH).o \ +$(SDIR)/LocalJob.$(ARCH).o \ +$(SDIR)/JobOps.$(ARCH).o \ +$(SDIR)/JobIDOps.$(ARCH).o \ +$(SDIR)/TaskOps.$(ARCH).o + +cmds = \ +$(SDIR)/pulse.$(ARCH).o \ +$(SDIR)/status.$(ARCH).o \ +$(SDIR)/kill.$(ARCH).o \ +$(SDIR)/delete.$(ARCH).o \ +$(SDIR)/verbose.$(ARCH).o \ +$(SDIR)/controller.$(ARCH).o \ +$(SDIR)/controller_host.$(ARCH).o \ +$(SDIR)/controller_exit.$(ARCH).o \ +$(SDIR)/controller_check.$(ARCH).o \ +$(SDIR)/controller_status.$(ARCH).o \ +$(SDIR)/controller_run.$(ARCH).o \ +$(SDIR)/controller_output.$(ARCH).o \ +$(SDIR)/controller_pulse.$(ARCH).o \ +$(SDIR)/task.$(ARCH).o \ +$(SDIR)/task_host.$(ARCH).o \ +$(SDIR)/task_nmax.$(ARCH).o \ +$(SDIR)/task_macros.$(ARCH).o \ +$(SDIR)/task_trange.$(ARCH).o \ +$(SDIR)/task_periods.$(ARCH).o \ +$(SDIR)/task_command.$(ARCH).o \ +$(SDIR)/task_options.$(ARCH).o \ +$(SDIR)/version.$(ARCH).o + +client = \ +$(SDIR)/pantasks_client.$(ARCH).o \ +$(SDIR)/client_shell.$(ARCH).o \ +$(SDIR)/invalid.$(ARCH).o \ +$(SDIR)/init_client.$(ARCH).o + +server = \ +$(SDIR)/pantasks_server.$(ARCH).o \ +$(SDIR)/server_threads.$(ARCH).o \ +$(SDIR)/server_run.$(ARCH).o \ +$(SDIR)/server_load.$(ARCH).o \ +$(SDIR)/InputQueue.$(ARCH).o \ +$(SDIR)/ListenClients.$(ARCH).o \ +$(SDIR)/server.$(ARCH).o \ +$(SDIR)/status_server.$(ARCH).o \ +$(SDIR)/init_server.$(ARCH).o \ +$(SDIR)/CheckPassword.$(ARCH).o + +libs = \ +$(DESTLIB)/libshell.a \ +$(DESTLIB)/libdata.a \ +$(DESTLIB)/libbasiccmd.a \ +$(DESTLIB)/libastrocmd.a \ +$(DESTLIB)/libdatacmd.a + +pantasks: $(BIN)/pantasks.$(ARCH) +$(BIN)/pantasks.$(ARCH) : $(single) $(funcs) $(cmds) $(libs) + +pantasks_client: $(BIN)/pantasks_client.$(ARCH) +$(BIN)/pantasks_client.$(ARCH) : $(client) $(libs) + +pantasks_server: $(BIN)/pantasks_server.$(ARCH) +$(BIN)/pantasks_server.$(ARCH) : $(server) $(funcs) $(cmds) $(libs) + @if [ ! -d $(BIN) ]; then mkdir -p $(BIN); fi + $(CC) -o $@ $^ $(LFLAGS) + +pantasks_client.install: $(DESTBIN)/pantasks_client +pantasks_server.install: $(DESTBIN)/pantasks_server + +install: $(DESTBIN)/pantasks $(DESTBIN)/pantasks_client $(DESTBIN)/pantasks_server help + +help: cmd.basic.help cmd.data.help cmd.astro.help pantasks.help + +.PHONY: pantasks pantasks_client pantasks_server + +include ../Makefile.Common Index: /tags/pantasks-0-5/Ohana/src/opihi/pantasks/RunScheduler.c =================================================================== --- /tags/pantasks-0-5/Ohana/src/opihi/pantasks/RunScheduler.c (revision 8568) +++ /tags/pantasks-0-5/Ohana/src/opihi/pantasks/RunScheduler.c (revision 8568) @@ -0,0 +1,85 @@ +# include "pantasks.h" + +/* the RunScheduler thread should continuously loop over: + + - tasks + - jobs + + occasionally, it should: + - check controller state + - check for new input files to be loaded + + alternatively, we could have separate threads for each of these + tasks and simply have them each grab a mutex before they go: + + CheckTasks: + - grab mutex + - loop over all tasks + - release mutex + - nanosleep? + + CheckJobs: + - grab mutex + - loop over all jobs + - release mutex + - nanosleep? + + CheckController: + - grab mutex + - check status + - check output + - release mutex + - usleep 10000 + + CheckInputs + - grab mutex + - check for inputs on list + - load inputs + - release mutex + - usleep 200000 + + this strategy seems fairly clean, with the somewhat odd design feature + that all of these 'body' threads are force to run in serial, not parallel, + by use of the mutexes. It is somewhat cleaner programming. The only + restriction on the commands allowed to the clients is that they must not + be allowed to change any of the pantasks internal state variables (variables, + macros, lists, etc). This is solved by restricting the clients to output + message functions only (reporting functions). + +*/ + +# define MAX_DELAY 0.1 +struct timeval start, last; +static int Ncheck = 0; + +int CheckSystem () { + + gettimeofday (&start, (void *) NULL); + + if (Ncheck < 100) { + CheckTasks (); + CheckJobs (); + Ncheck ++; + } else { + CheckController (); + CheckControllerOutput (); + Ncheck = 0; + } + /* ohana_memcheck (FALSE); */ + return (TRUE); +} + +int TestElapsedCheck () { + + struct timeval stop; + float dtime; + + gettimeofday (&stop, (void *) NULL); + dtime = DTIME (stop, start); + if (dtime > MAX_DELAY) return (TRUE); + return (FALSE); +} + + /* do we really need to check the controller on every call? + or perhaps alternate? + */ Index: /tags/pantasks-0-5/Ohana/src/opihi/pantasks/TaskOps.c =================================================================== --- /tags/pantasks-0-5/Ohana/src/opihi/pantasks/TaskOps.c (revision 8568) +++ /tags/pantasks-0-5/Ohana/src/opihi/pantasks/TaskOps.c (revision 8568) @@ -0,0 +1,454 @@ +# include "pantasks.h" + +static Task **tasks; +static int Ntasks; +static int NTASKS; + +/* counter marking task being visited by the run loop */ +static int ActiveTask; + +/* temporary holder for a new task */ +static Task *NewTask = NULL; + +/* set up the task list system */ +void InitTasks () { + NTASKS = 20; + Ntasks = 0; + ALLOCATE (tasks, Task *, NTASKS); + ActiveTask = -1; +} + +/* provide a mechanism to loop over the list of tasks */ +Task *NextTask () { + + Task *task; + + /* move to the next task and return it */ + ActiveTask ++; + if (ActiveTask < 0) ActiveTask = 0; + if (ActiveTask >= Ntasks) { + ActiveTask = -1; + return (NULL); + } + task = tasks[ActiveTask]; + return (task); +} + +/* return task with given name */ +Task *FindTask (char *name) { + + int i; + + /* try for an exact match first */ + for (i = 0; i < Ntasks; i++) { + if (!strcmp (tasks[i][0].name, name)) { + return (tasks[i]); + } + } + return (NULL); +} + +/* list known tasks */ +void ListTasks (int verbose) { + + int i, j, valid; + char *start, *stop; + + gprint (GP_LOG, "\n"); + if (Ntasks == 0) { + gprint (GP_LOG, " no defined tasks\n"); + return; + } + + gprint (GP_LOG, " Task Status\n"); + gprint (GP_LOG, " * Name Njobs Command\n"); + for (i = 0; i < Ntasks; i++) { + valid = CheckTimeRanges (tasks[i][0].ranges, tasks[i][0].Nranges); + if (verbose) gprint (GP_LOG, "\n"); + if (valid) { + gprint (GP_LOG, " + "); + } else { + gprint (GP_LOG, " - "); + } + if (tasks[i][0].argv == NULL) { + gprint (GP_LOG, "%-15s %5d %-20s\n", tasks[i][0].name, tasks[i][0].Njobs, "(dynamic)"); + } else { + gprint (GP_LOG, "%-15s %5d %-20s\n", tasks[i][0].name, tasks[i][0].Njobs, tasks[i][0].argv[0]); + } + if (verbose) { + gprint (GP_LOG, " spawn period: %f, polling period: %f, timeout period: %f\n", + tasks[i][0].exec_period, tasks[i][0].poll_period, tasks[i][0].timeout_period); + for (j = 0; j < tasks[i][0].Nranges; j++) { + switch (tasks[i][0].ranges[j].type) { + case RANGE_ABS: + start = sec_to_date (tasks[i][0].ranges[j].start); + stop = sec_to_date (tasks[i][0].ranges[j].stop); + break; + case RANGE_DAY: + start = sec_to_hms (tasks[i][0].ranges[j].start); + stop = sec_to_hms (tasks[i][0].ranges[j].stop); + break; + case RANGE_WEEK: + start = sec_to_day (tasks[i][0].ranges[j].start); + stop = sec_to_day (tasks[i][0].ranges[j].stop); + break; + default: + abort (); + } + if (tasks[i][0].ranges[j].include) { + gprint (GP_LOG, " active : %s - %s", start, stop); + if (tasks[i][0].ranges[j].Nmax) gprint (GP_LOG, " (%d of %d)", tasks[i][0].ranges[j].Nrun, tasks[i][0].ranges[j].Nmax); + gprint (GP_LOG, "\n"); + } else { + gprint (GP_LOG, " avoid : %s - %s\n", start, stop); + } + free (start); + free (stop); + } + if (tasks[i][0].host == NULL) { + gprint (GP_LOG, " task runs locally\n"); + continue; + } + if (!strcasecmp(tasks[i][0].host, "ANYHOST")) { + gprint (GP_LOG, " task host selected by controller\n"); + continue; + } + if (tasks[i][0].host_required) { + gprint (GP_LOG, " host %s (required)\n", tasks[i][0].host); + } else { + gprint (GP_LOG, " host %s (desired)\n", tasks[i][0].host); + } + } + } + return; +} + +/* show details of a task */ +int ShowTask (char *name) { + + int i, j; + char *start, *stop; + Task *task; + + task = FindTask (name); + if (task == NULL) { + gprint (GP_LOG, "task %s not found\n", name); + return (FALSE); + } + + gprint (GP_LOG, "\n macro %s\n", task[0].name); + + gprint (GP_LOG, "\n command: "); + for (i = 0; i < task[0].argc; i++) { + gprint (GP_LOG, "%s ", task[0].argv[i]); + } + gprint (GP_LOG, "\n\n"); + + gprint (GP_LOG, "\n options: "); + for (i = 0; i < task[0].optc; i++) { + gprint (GP_LOG, "%s ", task[0].optv[i]); + } + gprint (GP_LOG, "\n\n"); + + if (task[0].host == NULL) { + gprint (GP_LOG, " task runs locally\n"); + goto periods; + } + if (!strcasecmp(task[0].host, "ANYHOST")) { + gprint (GP_LOG, " task host selected by controller\n"); + goto periods; + } + if (task[0].host_required) { + gprint (GP_LOG, " host %s (required)\n", task[0].host); + } else { + gprint (GP_LOG, " host %s (desired)\n", task[0].host); + } + +periods: + gprint (GP_LOG, " time periods: exec: %f poll: %f timeout: %f\n", + task[0].exec_period, task[0].poll_period, task[0].timeout_period); + + for (j = 0; j < tasks[i][0].Nranges; j++) { + switch (tasks[i][0].ranges[j].type) { + case RANGE_ABS: + start = sec_to_date (tasks[i][0].ranges[j].start); + stop = sec_to_date (tasks[i][0].ranges[j].stop); + break; + case RANGE_DAY: + start = sec_to_hms (tasks[i][0].ranges[j].start); + stop = sec_to_hms (tasks[i][0].ranges[j].stop); + break; + case RANGE_WEEK: + start = sec_to_day (tasks[i][0].ranges[j].start); + stop = sec_to_day (tasks[i][0].ranges[j].stop); + break; + default: + abort (); + } + if (tasks[i][0].ranges[j].include) { + gprint (GP_LOG, " active : %s - %s", start, stop); + if (tasks[i][0].ranges[j].Nmax) gprint (GP_LOG, " (%d of %d)", tasks[i][0].ranges[j].Nrun, tasks[i][0].ranges[j].Nmax); + gprint (GP_LOG, "\n"); + } else { + gprint (GP_LOG, " avoid : %s - %s\n", start, stop); + } + free (start); + free (stop); + } + + gprint (GP_LOG, "\n pre-execute macro\n"); + ListMacro (task[0].exec); + + gprint (GP_LOG, "\n timeout macro\n"); + ListMacro (task[0].timeout); + + gprint (GP_LOG, "\n crash macro\n"); + ListMacro (task[0].crash); + + gprint (GP_LOG, "\n default exit macro\n"); + ListMacro (task[0].defexit); + + for (i = 0; i < task[0].Nexit; i++) { + gprint (GP_LOG, "\n exit macro (status == %d)\n", atoi(task[0].exit[i][0].name)); + ListMacro (task[0].exit[i]); + } + + return (TRUE); +} + +/* make a new named task */ +int FreeTask (Task *task) { + + int i; + + if (task == NULL) return (FALSE); + + if (task[0].name != NULL) free (task[0].name); + if (task[0].host != NULL) free (task[0].host); + if (task[0].argv != NULL) { + for (i = 0; i < task[0].argc; i++) { + free (task[0].argv[i]); + } + free (task[0].argv); + } + if (task[0].optv != NULL) { + for (i = 0; i < task[0].optc; i++) { + free (task[0].optv[i]); + } + free (task[0].optv); + } + if (task[0].exec != NULL) { + FreeMacro (task[0].exec); + free (task[0].exec); + } + if (task[0].crash != NULL) { + FreeMacro (task[0].crash); + free (task[0].crash); + } + if (task[0].timeout != NULL) { + FreeMacro (task[0].timeout); + free (task[0].timeout); + } + for (i = 0; i < task[0].Nexit; i++) { + if (task[0].exit[i] != NULL) { + FreeMacro (task[0].exit[i]); + } + free (task[0].exit[i]); + } + free (task[0].exit); + + if (task[0].ranges != NULL) { + free (task[0].ranges); + } + return (TRUE); +} + +/**** new task functions ***/ + +/* make a new named task */ +Task *CreateTask (char *name) { + + ALLOCATE (NewTask, Task, 1); + + NewTask[0].name = strcreate (name);; + + NewTask[0].host = NULL; + NewTask[0].host_required = FALSE; + + NewTask[0].argc = 0; + NewTask[0].argv = NULL; + + NewTask[0].optc = 0; + NewTask[0].optv = NULL; + + NewTask[0].exec = NULL; + NewTask[0].crash = NULL; + NewTask[0].timeout = NULL; + NewTask[0].defexit = NULL; + + NewTask[0].Nexit = 0; + NewTask[0].NEXIT = 10; + ALLOCATE (NewTask[0].exit, Macro *, NewTask[0].NEXIT); + /* don't free tasks[0].exit, keep at least 1 allocated */ + + NewTask[0].exec_period = 1.0; + NewTask[0].poll_period = 1.0; + NewTask[0].timeout_period = 1.0; + + NewTask[0].Nranges = 0; + ALLOCATE (NewTask[0].ranges, TimeRange, 1); + + /* init task timer (is reset by 'run') */ + gettimeofday (&NewTask[0].last, (void *) NULL); + NewTask[0].Nmax = 0; /* default value means 'no limit' */ + + NewTask[0].Njobs = 0; + NewTask[0].Nsuccess = 0; + NewTask[0].Nfailure = 0; + NewTask[0].Ntimeout = 0; + return (NewTask); +} + +/* remove the task from the task list */ +int RemoveTask (Task *task) { + + int i, Nt; + + /* find task in task list */ + Nt = -1; + for (i = 0; i < Ntasks; i++) { + if (task == tasks[i]) { + Nt = i; + break; + } + } + if (Nt == -1) { + gprint (GP_ERR, "programming error: task not found\n"); + return (FALSE); + } + for (i = Nt; i < Ntasks - 1; i++) { + tasks[i] = tasks[i+1]; + } + Ntasks --; + return (TRUE); +} + +int ValidateTask (Task *task, int RequireStatic) { + + int i, hash; + + /* is a static command defined? */ + if (task[0].argc != 0) { + if (task[0].argv == NULL) { + gprint (GP_ERR, "task command arguments not defined (programming error)\n"); + return (FALSE); + } + return (TRUE); + } + if (RequireStatic) { + gprint (GP_ERR, "task command not defined\n"); + return (FALSE); + } + + /* no static command; dynamic command? */ + if (task[0].exec != NULL) { + for (i = 0; i < task[0].exec[0].Nlines; i++) { + hash = TaskHash (task[0].exec[0].line[i]); + if (hash == TASK_COMMAND) return (TRUE); + } + } + gprint (GP_ERR, "task command not defined\n"); + return (FALSE); +} + +int RegisterNewTask () { + + int N; + + N = Ntasks; + Ntasks ++; + if (Ntasks == NTASKS) { + NTASKS += 20; + REALLOCATE (tasks, Task *, NTASKS); + } + tasks[N] = NewTask; + NewTask = NULL; + return (TRUE); +} + +int DeleteNewTask () { + if (NewTask != NULL) { + FreeTask (NewTask); + free (NewTask); + NewTask = NULL; + } + return (TRUE); +} + +Task *SetNewTask (Task *task) { + NewTask = task; + return (task); +} + +Task *GetNewTask () { + return (NewTask); +} + +Task *GetActiveTask () { + Task *task; + if (ActiveTask < 0) return (NULL); + task = tasks[ActiveTask]; + return (task); +} + +int TaskHash (char *input) { + + int hash; + char *command; + + hash = TASK_NONE; + + command = thisword (input); + if (command == NULL) return (TASK_EMPTY); + + if (command[0] == '#') hash = TASK_COMMENT; + if (!strcasecmp (command, "END")) hash = TASK_END; + if (!strcasecmp (command, "HOST")) hash = TASK_HOST; + if (!strcasecmp (command, "NMAX")) hash = TASK_NMAX; + if (!strcasecmp (command, "TRANGE")) hash = TASK_TRANGE; + if (!strcasecmp (command, "COMMAND")) hash = TASK_COMMAND; + if (!strcasecmp (command, "OPTIONS")) hash = TASK_OPTIONS; + if (!strcasecmp (command, "PERIODS")) hash = TASK_PERIODS; + if (!strcasecmp (command, "TASK.EXIT")) hash = TASK_EXIT; + if (!strcasecmp (command, "TASK.EXEC")) hash = TASK_EXEC; + + free (command); + return (hash); +} + +/*** task timer functions ***/ + +double GetTaskTimer (struct timeval start) { + + double dtime; + struct timeval now; + + gettimeofday (&now, (void *) NULL); + dtime = DTIME (now, start); + + return (dtime); +} + +void SetTaskTimer (struct timeval *timer) { + 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: /tags/pantasks-0-5/Ohana/src/opihi/pantasks/client_shell.c =================================================================== --- /tags/pantasks-0-5/Ohana/src/opihi/pantasks/client_shell.c (revision 8568) +++ /tags/pantasks-0-5/Ohana/src/opihi/pantasks/client_shell.c (revision 8568) @@ -0,0 +1,99 @@ +# include "opihi.h" + +/******************/ +int client_shell (int argc, char **argv) { + + int Nbad, status; + char *line, *prompt, *history; + pid_t ppid; + + general_init (&argc, argv); + program_init (&argc, argv); + startup (&argc, argv); + prompt = get_variable("PROMPT"); + history = get_variable("HISTORY"); + welcome (); + + /* attempt to connect to the pantasks server (exit on failure) */ + multicommand_InitServer (); + + Nbad = 0; + while (1) { /** must exit with command "exit" or "quit" */ + if (Nbad == 10) exit (3); + + line = readline (prompt); + + if (line == NULL) { + + ppid = getppid(); + if (ppid == 1) { + gprint (GP_ERR, "caught parent shutdown\n"); + exit (2); + } + if (!isatty (STDIN_FILENO)) exit (2); + gprint (GP_ERR, "Use \"quit\" to exit\n"); + Nbad ++; + continue; + } + Nbad = 0; + ohana_memregister (line); + + stripwhite (line); + + if (*line) { + status = multicommand (line); + add_history (line); + append_history (1, history); + } + free (line); + } +} + +/* + startup sequence: + + - general_init + - program_init + - startup (exit if non-interactive) + - welcome + +*/ + +/* client issues and questions: + +- need to identify the commands to be caught by the client +- pass each input line through the command parser first, then + send to server if not identified as a valid command +- this raises the question of variables: do we parse variables + at the client level? I would not think so. what about input? + if the input is to be parsed by the client, but executed on the server, + I'll need to re-work the input parsing system. would be better + for the server to parse the input command. in this case, the client + user needs to have access to the directories with input scripts. + this may be an advantage: it would allow some limitation on who can + install server scripts. +- variables, vectors and buffers: where are they valid (client or server?) + it seems they should all be defined locally on the server. this makes + graphics plotting a server-side action as well. this is probably easier, + but we do need to be careful about multiple clients trying to make plots + on the same window at the same time. Exporting the window can be done + for one client with the $KII variable, but could be tricky for more + complex interactions +- list of commands which clearly must be parsed by the client: + - quit / exit + - exec / ! + - ? + +- i may need an alternative version of command, modified to avoid parsing the + variables and vectors. I also need an alternative version of the Init commands + to just list the blocked ones. + +- the client now executes all of the basic opihi commands, and only passes + through the pantasks commands. The pantasks server interface should only + accept functions which immediately return, sending back the result to be + printed by the client. + + The server should not accept the 'task' commands from the client command-line. + To define and run macros or tasks on the server, these need to be added to + a script which is loaded with a varient on the input command. +*/ Index: /tags/pantasks-0-5/Ohana/src/opihi/pantasks/connect_to_server.c =================================================================== --- /tags/pantasks-0-5/Ohana/src/opihi/pantasks/connect_to_server.c (revision 8568) +++ /tags/pantasks-0-5/Ohana/src/opihi/pantasks/connect_to_server.c (revision 8568) @@ -0,0 +1,16 @@ +# include "pantasks.h" + +int connect_to_server () { + + char hostname[256]; + + ScanConfig (GlobalConfig, "PANTASKS_SERVER", "%s", 0, hostname); + + server = GetClientSocket (hostname); + + /* here we can perform the security handshaking */ + SendCommand (BindSocket, strlen(PASSWORD), PASSWORD); + + return (server); + +} Index: /tags/pantasks-0-5/Ohana/src/opihi/pantasks/controller.c =================================================================== --- /tags/pantasks-0-5/Ohana/src/opihi/pantasks/controller.c (revision 8568) +++ /tags/pantasks-0-5/Ohana/src/opihi/pantasks/controller.c (revision 8568) @@ -0,0 +1,64 @@ +# include "pantasks.h" + +int controller_host PROTO((int, char **)); +int controller_exit PROTO((int, char **)); +int controller_status PROTO((int, char **)); +int controller_run PROTO((int, char **)); +int controller_stop PROTO((int, char **)); +int controller_check PROTO((int, char **)); +int controller_output PROTO((int, char **)); +int controller_pulse PROTO((int, char **)); + +static Command controller_cmds[] = { + {"exit", controller_exit, "shutdown controller"}, + {"host", controller_host, "define host for controller"}, + {"check", controller_check, "check controller host/job"}, + {"run", controller_run, "start controller operation"}, + {"stop", controller_stop, "stop controller (no disconnect)"}, + {"status", controller_status, "check controller status"}, + {"output", controller_output, "print controller output"}, + {"pulse", controller_pulse, "set controller pulse"}, +}; + +int controller (int argc, char **argv) { + + int i, N, status; + CommandF *func; + + if (argc < 2) { + gprint (GP_ERR, "USAGE: controller (command) ... \n"); + return (FALSE); + } + + if (!strcasecmp (argv[1], "help")) { + N = sizeof (controller_cmds) / sizeof (Command); + + for (i = 0; i < N; i++) { + gprint (GP_LOG, "%-15s %s\n", controller_cmds[i].name, controller_cmds[i].help); + } + return (TRUE); + } + + func = FindControllerCommand (argv[1]); + if (func == NULL) { + gprint (GP_ERR, "invalid controller command\n"); + return (FALSE); + } + + status = (*func)(argc - 1, argv + 1); + return (status); +} + +CommandF *FindControllerCommand (char *cmd) { + + int i, N; + + N = sizeof (controller_cmds) / sizeof (Command); + + for (i = 0; i < N; i++) { + if (!strcmp (controller_cmds[i].name, cmd)) { + return (controller_cmds[i].func); + } + } + return (NULL); +} Index: /tags/pantasks-0-5/Ohana/src/opihi/pantasks/controller_check.c =================================================================== --- /tags/pantasks-0-5/Ohana/src/opihi/pantasks/controller_check.c (revision 8568) +++ /tags/pantasks-0-5/Ohana/src/opihi/pantasks/controller_check.c (revision 8568) @@ -0,0 +1,35 @@ +# include "pantasks.h" + +int controller_check (int argc, char **argv) { + + int status; + char command[1024]; + IOBuffer buffer; + + if (argc != 3) goto usage; + if (strcasecmp (argv[1], "JOB") && + strcasecmp (argv[1], "HOST")) goto usage; + + /* check if controller is running */ + status = CheckControllerStatus (); + if (!status) { + gprint (GP_LOG, "controller is not running\n"); + return (TRUE); + } + + sprintf (command, "check %s %s", argv[1], argv[2]); + InitIOBuffer (&buffer, 0x100); + status = ControllerCommand (command, CONTROLLER_PROMPT, &buffer); + if (VerboseMode()) { + gprint (GP_LOG, "controller command sent\n"); + gwrite (buffer.buffer, 1, buffer.Nbuffer, GP_LOG); + gprint (GP_LOG, "\n Nbytes received: %d\n", buffer.Nbuffer); + } + FreeIOBuffer (&buffer); + return (TRUE); + +usage: + gprint (GP_ERR, "USAGE: controller check job (jobID)\n"); + gprint (GP_ERR, "USAGE: controller check host (hostID)\n"); + return (FALSE); +} Index: /tags/pantasks-0-5/Ohana/src/opihi/pantasks/controller_exit.c =================================================================== --- /tags/pantasks-0-5/Ohana/src/opihi/pantasks/controller_exit.c (revision 8568) +++ /tags/pantasks-0-5/Ohana/src/opihi/pantasks/controller_exit.c (revision 8568) @@ -0,0 +1,14 @@ +# include "pantasks.h" + +int controller_exit (int argc, char **argv) { + + if (argc != 2) { + gprint (GP_ERR, "USAGE: controller exit TRUE\n"); + return (FALSE); + } + + if (strcasecmp (argv[1], "TRUE")) return (FALSE); + + QuitController (); + return (TRUE); +} Index: /tags/pantasks-0-5/Ohana/src/opihi/pantasks/controller_host.c =================================================================== --- /tags/pantasks-0-5/Ohana/src/opihi/pantasks/controller_host.c (revision 8568) +++ /tags/pantasks-0-5/Ohana/src/opihi/pantasks/controller_host.c (revision 8568) @@ -0,0 +1,37 @@ +# include "pantasks.h" + +int controller_host (int argc, char **argv) { + + int status; + char command[1024]; + IOBuffer buffer; + + if (argc != 3) { + gprint (GP_ERR, "USAGE: controller host (command) (hostname)\n"); + return (FALSE); + } + + /* start controller connection (if needed) */ + if (!StartController ()) { + gprint (GP_ERR, "failure to start pcontrol\n"); + return (FALSE); + } + + sprintf (command, "host %s %s", argv[1], argv[2]); + InitIOBuffer (&buffer, 0x100); + + SerialThreadLock (); + status = ControllerCommand (command, CONTROLLER_PROMPT, &buffer); + SerialThreadUnlock (); + + if (status) gwrite (buffer.buffer, 1, buffer.Nbuffer, GP_LOG); + + FreeIOBuffer (&buffer); + return (TRUE); +} + +/* should I keep an internal host table so I can reload the + hosts if the controller exits? + + alternatively, that could be a user-level choice +*/ Index: /tags/pantasks-0-5/Ohana/src/opihi/pantasks/controller_output.c =================================================================== --- /tags/pantasks-0-5/Ohana/src/opihi/pantasks/controller_output.c (revision 8568) +++ /tags/pantasks-0-5/Ohana/src/opihi/pantasks/controller_output.c (revision 8568) @@ -0,0 +1,13 @@ +# include "pantasks.h" + +int controller_output (int argc, char **argv) { + + if (argc != 1) { + gprint (GP_ERR, "USAGE: controller status\n"); + return (FALSE); + } + + CheckControllerOutput (); + PrintControllerOutput (); + return (TRUE); +} Index: /tags/pantasks-0-5/Ohana/src/opihi/pantasks/controller_pulse.c =================================================================== --- /tags/pantasks-0-5/Ohana/src/opihi/pantasks/controller_pulse.c (revision 8568) +++ /tags/pantasks-0-5/Ohana/src/opihi/pantasks/controller_pulse.c (revision 8568) @@ -0,0 +1,26 @@ +# include "pantasks.h" + +int controller_pulse (int argc, char **argv) { + + int status; + char command[1024]; + IOBuffer buffer; + + if (argc != 2) { + gprint (GP_ERR, "USAGE: controller pulse (usec)\n"); + return (FALSE); + } + + /* start controller connection (if needed) */ + if (!StartController ()) { + gprint (GP_ERR, "failure to start pcontrol\n"); + return (FALSE); + } + + sprintf (command, "pulse %d", atoi(argv[1])); + InitIOBuffer (&buffer, 0x100); + status = ControllerCommand (command, CONTROLLER_PROMPT, &buffer); + if (status) gwrite (buffer.buffer, 1, buffer.Nbuffer, GP_LOG); + FreeIOBuffer (&buffer); + return (TRUE); +} Index: /tags/pantasks-0-5/Ohana/src/opihi/pantasks/controller_run.c =================================================================== --- /tags/pantasks-0-5/Ohana/src/opihi/pantasks/controller_run.c (revision 8568) +++ /tags/pantasks-0-5/Ohana/src/opihi/pantasks/controller_run.c (revision 8568) @@ -0,0 +1,61 @@ +# include "pantasks.h" + +int controller_run (int argc, char **argv) { + + int status; + char command[1024]; + IOBuffer buffer; + + if (argc != 1) { + gprint (GP_ERR, "USAGE: controller run\n"); + return (FALSE); + } + + /* check if controller is running */ + status = CheckControllerStatus (); + if (!status) { + gprint (GP_LOG, "controller is not active\n"); + return (TRUE); + } + + sprintf (command, "run"); + InitIOBuffer (&buffer, 0x100); + status = ControllerCommand (command, CONTROLLER_PROMPT, &buffer); + if (status) { + gwrite (buffer.buffer, 1, buffer.Nbuffer, GP_LOG); + } else { + gprint (GP_LOG, "controller is not responding\n"); + } + FreeIOBuffer (&buffer); + return (TRUE); +} + +int controller_stop (int argc, char **argv) { + + int status; + char command[1024]; + IOBuffer buffer; + + if (argc != 1) { + gprint (GP_ERR, "USAGE: controller stop\n"); + return (FALSE); + } + + /* check if controller is running */ + status = CheckControllerStatus (); + if (!status) { + gprint (GP_LOG, "controller is not active\n"); + return (TRUE); + } + + sprintf (command, "stop"); + InitIOBuffer (&buffer, 0x100); + status = ControllerCommand (command, CONTROLLER_PROMPT, &buffer); + if (status) { + gwrite (buffer.buffer, 1, buffer.Nbuffer, GP_LOG); + } else { + gprint (GP_LOG, "controller is not responding\n"); + } + FreeIOBuffer (&buffer); + return (TRUE); +} Index: /tags/pantasks-0-5/Ohana/src/opihi/pantasks/controller_status.c =================================================================== --- /tags/pantasks-0-5/Ohana/src/opihi/pantasks/controller_status.c (revision 8568) +++ /tags/pantasks-0-5/Ohana/src/opihi/pantasks/controller_status.c (revision 8568) @@ -0,0 +1,36 @@ +# include "pantasks.h" + +int controller_status (int argc, char **argv) { + + int status; + char command[1024]; + IOBuffer buffer; + + if (argc != 1) { + gprint (GP_ERR, "USAGE: controller status\n"); + return (FALSE); + } + + /* check if controller is running */ + status = CheckControllerStatus (); + if (!status) { + gprint (GP_LOG, "controller is not active\n"); + return (TRUE); + } + + + sprintf (command, "status"); + InitIOBuffer (&buffer, 0x100); + + SerialThreadLock (); + status = ControllerCommand (command, CONTROLLER_PROMPT, &buffer); + SerialThreadUnlock (); + + if (status) { + gwrite (buffer.buffer, 1, buffer.Nbuffer, GP_LOG); + } else { + gprint (GP_LOG, "controller is not communicating\n"); + } + FreeIOBuffer (&buffer); + return (TRUE); +} Index: /tags/pantasks-0-5/Ohana/src/opihi/pantasks/delete.c =================================================================== --- /tags/pantasks-0-5/Ohana/src/opihi/pantasks/delete.c (revision 8568) +++ /tags/pantasks-0-5/Ohana/src/opihi/pantasks/delete.c (revision 8568) @@ -0,0 +1,18 @@ +# include "pantasks.h" + +int delete_job (int argc, char **argv) { + + int JobID; + + if (argc < 2) { + gprint (GP_ERR, "USAGE: delete (JobID)\n"); + return (FALSE); + } + JobID = atoi (argv[1]); + /* use a string interp to convert JobIDs to ints ? */ + + gprint (GP_ERR, "this function is not yet implemented\n"); + return (TRUE); +} + +/* find job from jobID, kill and delete */ Index: /tags/pantasks-0-5/Ohana/src/opihi/pantasks/help/controller =================================================================== --- /tags/pantasks-0-5/Ohana/src/opihi/pantasks/help/controller (revision 8568) +++ /tags/pantasks-0-5/Ohana/src/opihi/pantasks/help/controller (revision 8568) @@ -0,0 +1,29 @@ + + controller (command) ... + + Interact with the parallel controller 'pcontrol'. This command + groups together several controller sub-commands, listed below. + + + controller host (hostname) + + Define a new machine available to the parallel controller. + + + controller check (host) (HostID) + controller check (job) (JobID) + + Check on the status of a specific host or job known to the parallel + controller. + + + controller status + + Query the parallel controller for its current status + + + controller output + + Dump the contents of the parallel controller message buffers. + These are the messages from the controller itself, rather than the + messages generated by jobs run by the controller. Index: /tags/pantasks-0-5/Ohana/src/opihi/pantasks/help/task =================================================================== --- /tags/pantasks-0-5/Ohana/src/opihi/pantasks/help/task (revision 8568) +++ /tags/pantasks-0-5/Ohana/src/opihi/pantasks/help/task (revision 8568) @@ -0,0 +1,27 @@ + + task (name) + + Define or modify a task. A task is defined by issuing a series of + task commands, listed below (see their individual help pages). A + task describes the rules for constructing jobs. At a minimum, a + task must include a command, defining the UNIX command to run for + each job. Other task commands define optional concepts for the + task. These include: + + - a macro to be run before the job is constructed (task.exec). + - a set of macros to be run when the job is completed (task.exit) + - time ranges in which the task should be run (or not) (trange) + - required / desired parallel controller hosts (host) + - job contruction time interval (periods -exec) + - job polling time interval (periods -poll) + - job timeouts (periods -timeout) + - maximum number of jobs to construct + + See also: + task.command + task.exit + task.exec + task.trange + task.nmax + task.host + task.periods Index: /tags/pantasks-0-5/Ohana/src/opihi/pantasks/help/task.command =================================================================== --- /tags/pantasks-0-5/Ohana/src/opihi/pantasks/help/task.command (revision 8568) +++ /tags/pantasks-0-5/Ohana/src/opihi/pantasks/help/task.command (revision 8568) @@ -0,0 +1,15 @@ + + command (arg0) (arg1) ... + + Define the rule for constructing a job command for a task. The + arguments are UNIX command and the collection of command arguments. + The arguments may include references to Opihi variables. The task + command may be included at the top level of the task, or it may be + defined within the task exec macro. In the latter case, the + command is constructed only when the exec macro is run, every job + construction interval. Thus, the command arguments may depend on + the result of the exec macro operation. + + See also: + task + task.exec Index: /tags/pantasks-0-5/Ohana/src/opihi/pantasks/help/task.exec =================================================================== --- /tags/pantasks-0-5/Ohana/src/opihi/pantasks/help/task.exec (revision 8568) +++ /tags/pantasks-0-5/Ohana/src/opihi/pantasks/help/task.exec (revision 8568) @@ -0,0 +1,17 @@ + + task.exec + + A special macro to be executed before constructing the job command. + The macro is defined by commands following the 'task.exec' command, + ending with the word 'end'. If the exit status of the task.exec + macro is FALSE, then the job is NOT constructed. This allows the + construction of a job to be conditional on other, external + quantites. (Note: the exit status of the macro is false if 'break' + is called). The task command may be defined within the task.exec + macro, in which case the command is constructed only when the exec + macro is run, every job construction interval. Thus, the command + arguments may depend on the result of the exec macro operation. + + See also: + task + task.command Index: /tags/pantasks-0-5/Ohana/src/opihi/pantasks/init.c =================================================================== --- /tags/pantasks-0-5/Ohana/src/opihi/pantasks/init.c (revision 8568) +++ /tags/pantasks-0-5/Ohana/src/opihi/pantasks/init.c (revision 8568) @@ -0,0 +1,53 @@ +# include "pantasks.h" + +int controller PROTO((int, char **)); +int task PROTO((int, char **)); +int task_host PROTO((int, char **)); +int task_nmax PROTO((int, char **)); +int task_trange PROTO((int, char **)); +int task_macros PROTO((int, char **)); +int task_command PROTO((int, char **)); +int task_options PROTO((int, char **)); +int task_periods PROTO((int, char **)); +int run PROTO((int, char **)); +int stop PROTO((int, char **)); +int pulse PROTO((int, char **)); +int status_sys PROTO((int, char **)); +int kill_job PROTO((int, char **)); +int delete_job PROTO((int, char **)); +int verbose PROTO((int, char **)); +int version PROTO((int, char **)); + +static Command cmds[] = { + {"controller", controller, "controller commands"}, + {"task", task, "define a schedulable task"}, + {"host", task_host, "define host machine for a task"}, + {"nmax", task_nmax, "define maximum number of jobs for a task"}, + {"trange", task_trange, "define valid/invalid time periods 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"}, + {"options", task_options, "define optional variables associated with the job task"}, + {"periods", task_periods, "define time scales for a task"}, + {"run", run, "run the scheduler"}, + {"stop", stop, "stop the scheduler"}, + {"pulse", pulse, "set the scheduler update period"}, + {"status", status_sys, "get system status"}, + {"kill", kill_job, "kill job"}, + {"delete", delete_job, "delete job"}, + {"version", version, "show version information"}, + {"verbose", verbose, "set/toggle verbose mode"}, +}; + +void InitPantasks () { + + int i; + + InitTasks (); + InitJobs (); + InitJobIDs (); + + for (i = 0; i < sizeof (cmds) / sizeof (Command); i++) { + AddCommand (&cmds[i]); + } +} Index: /tags/pantasks-0-5/Ohana/src/opihi/pantasks/init_client.c =================================================================== --- /tags/pantasks-0-5/Ohana/src/opihi/pantasks/init_client.c (revision 8568) +++ /tags/pantasks-0-5/Ohana/src/opihi/pantasks/init_client.c (revision 8568) @@ -0,0 +1,23 @@ +# include "pantasks.h" + +int invalid PROTO((int, char **)); + +/* we list here commands which are not valid for the client, + but represent valid pantasks commands. these are caught + by the client and sent to the 'invalid' function */ + +static Command cmds[] = { + {"task", invalid, "define a schedulable task"}, + {"task.exit", invalid, "define exit macros for a task"}, + {"task.exec", invalid, "define pre-exec macro for a task"}, +}; + +void InitPantasksClient () { + + int i; + + for (i = 0; i < sizeof (cmds) / sizeof (Command); i++) { + AddCommand (&cmds[i]); + } + +} Index: /tags/pantasks-0-5/Ohana/src/opihi/pantasks/init_server.c =================================================================== --- /tags/pantasks-0-5/Ohana/src/opihi/pantasks/init_server.c (revision 8568) +++ /tags/pantasks-0-5/Ohana/src/opihi/pantasks/init_server.c (revision 8568) @@ -0,0 +1,51 @@ +# include "pantasks.h" + +int controller PROTO((int, char **)); +int task PROTO((int, char **)); +int task_host PROTO((int, char **)); +int task_nmax PROTO((int, char **)); +int task_trange PROTO((int, char **)); +int task_macros PROTO((int, char **)); +int task_command PROTO((int, char **)); +int task_options PROTO((int, char **)); +int task_periods PROTO((int, char **)); +int pulse PROTO((int, char **)); +int status_server PROTO((int, char **)); +int kill_job PROTO((int, char **)); +int delete_job PROTO((int, char **)); +int verbose PROTO((int, char **)); +int version PROTO((int, char **)); +int server PROTO((int, char **)); + +static Command cmds[] = { + {"controller", controller, "controller commands"}, + {"task", task, "define a schedulable task"}, + {"host", task_host, "define host machine for a task"}, + {"nmax", task_nmax, "define maximum number of jobs for a task"}, + {"trange", task_trange, "define valid/invalid time periods 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"}, + {"options", task_options, "define optional variables associated with the job task"}, + {"periods", task_periods, "define time scales for a task"}, + {"pulse", pulse, "set the scheduler update period"}, + {"server", server, "server-specific commands"}, + {"status", status_server, "get system status"}, + {"kill", kill_job, "kill job"}, + {"delete", delete_job, "delete job"}, + {"version", version, "show version information"}, + {"verbose", verbose, "set/toggle verbose mode"}, +}; + +void InitPantasksServer () { + + int i; + + InitTasks (); + InitJobs (); + InitJobIDs (); + + for (i = 0; i < sizeof (cmds) / sizeof (Command); i++) { + AddCommand (&cmds[i]); + } +} Index: /tags/pantasks-0-5/Ohana/src/opihi/pantasks/invalid.c =================================================================== --- /tags/pantasks-0-5/Ohana/src/opihi/pantasks/invalid.c (revision 8568) +++ /tags/pantasks-0-5/Ohana/src/opihi/pantasks/invalid.c (revision 8568) @@ -0,0 +1,8 @@ +# include "pantasks.h" + +int invalid (int argc, char **argv) { + + gprint (GP_ERR, "%s is not valid for the pantasks client\n", argv[0]); + return (TRUE); + +} Index: /tags/pantasks-0-5/Ohana/src/opihi/pantasks/job.c =================================================================== --- /tags/pantasks-0-5/Ohana/src/opihi/pantasks/job.c (revision 8568) +++ /tags/pantasks-0-5/Ohana/src/opihi/pantasks/job.c (revision 8568) @@ -0,0 +1,68 @@ +# include "pantasks.h" + +/* this will require a bit of care: to define a job, we need + to specify: + - the command + this is defined in the job command: job argv0 argv1 argv2... + - timeout (an option to the command) + - exit macros? + do nothing by default? + - stderr / stdout disposal? + save in a specific buffer by default? + send to a file? + + the command can look just like the controller equivalent one: + job [-host host] [-timeout timeout] args... + +*/ + +int job (int argc, char **argv) { + + char *Host, **targv; + int i, N, Mode, targc, Timeout; + IDtype JobID; + + Host = NULL; + Mode = PCONTROL_JOB_ANYHOST; + if ((N = get_argument (argc, argv, "-host"))) { + remove_argument (N, &argc, argv); + Host = strcreate (argv[N]); + remove_argument (N, &argc, argv); + Mode = PCONTROL_JOB_WANTHOST; + } + if ((N = get_argument (argc, argv, "+host"))) { + if (Mode == PCONTROL_JOB_WANTHOST) { + gprint (GP_ERR, "ERROR: -host and +host are incompatible\n"); + FREE (Host); + return (FALSE); + } + remove_argument (N, &argc, argv); + Host = strcreate (argv[N]); + remove_argument (N, &argc, argv); + Mode = PCONTROL_JOB_NEEDHOST; + } + if (Host == NULL) Host = strcreate ("anyhost"); + + Timeout = 100; + if ((N = get_argument (argc, argv, "-timeout"))) { + remove_argument (N, &argc, argv); + Timeout = atoi (argv[N]); + remove_argument (N, &argc, argv); + } + + if (argc < 2) { + gprint (GP_ERR, "USAGE: job [options] (arg0) (arg1) ... (argN)\n"); + FREE (Host); + return (FALSE); + } + + targc = argc - 1; + ALLOCATE (targv, char *, targc); + for (i = 1; i < argc; i++) { + targv[i-1] = strcreate (argv[i]); + } + + JobID = AddJob (Host, Mode, Timeout, targc, targv); + gprint (GP_LOG, "JobID: %d\n", JobID); + return (TRUE); +} Index: /tags/pantasks-0-5/Ohana/src/opihi/pantasks/kill.c =================================================================== --- /tags/pantasks-0-5/Ohana/src/opihi/pantasks/kill.c (revision 8568) +++ /tags/pantasks-0-5/Ohana/src/opihi/pantasks/kill.c (revision 8568) @@ -0,0 +1,32 @@ +# include "pantasks.h" + +int kill_job (int argc, char **argv) { + + Job *job; + int JobID; + + if (argc < 2) { + gprint (GP_ERR, "USAGE: kill (JobID)\n"); + return (FALSE); + } + JobID = atoi (argv[1]); + + job = FindJob (JobID); + + if (job[0].mode == JOB_LOCAL) { + if (!KillLocalJob (job)) { + job[0].state = JOB_HUNG; + if (VerboseMode()) gprint (GP_LOG, "child process %d is hung, cannot kill\n", job[0].pid); + return (FALSE); + } + } else { + if (!KillControllerJob (job)) { + job[0].state = JOB_HUNG; + if (VerboseMode()) gprint (GP_LOG, "child process %d is hung, cannot kill\n", job[0].pid); + return (FALSE); + } + } + + gprint (GP_ERR, "this function is not yet implemented\n"); + return (TRUE); +} Index: /tags/pantasks-0-5/Ohana/src/opihi/pantasks/notes.txt =================================================================== --- /tags/pantasks-0-5/Ohana/src/opihi/pantasks/notes.txt (revision 8568) +++ /tags/pantasks-0-5/Ohana/src/opihi/pantasks/notes.txt (revision 8568) @@ -0,0 +1,36 @@ + +PanTasks Client / Server design + +Client + +client uses readline to parse the incoming commands commands are only +validated by the client before being sent to the server: + +client loop: + + line = readline () + validate argv[0] + send to server + wait for response? + +Server + +Server does not use readline at all. server has two threads. first +thread listens for commands from the clients. how many clients? use +poll to wait for data? can we manage enough clients with a single +thread? + +the second thread runs the scheduler loop: this is the same background +check that is being run by the current model on the timeouts. + +server loop: + + check socket for new incoming client connection + check known clients for new messages + execute new command + send response to the client + run the scheduler for a single loop + +client / server communication is modelled after addstar client / +server? + Index: /tags/pantasks-0-5/Ohana/src/opihi/pantasks/pantasks.c =================================================================== --- /tags/pantasks-0-5/Ohana/src/opihi/pantasks/pantasks.c (revision 8568) +++ /tags/pantasks-0-5/Ohana/src/opihi/pantasks/pantasks.c (revision 8568) @@ -0,0 +1,64 @@ +# include "pantasks.h" + +# define opihi_name "pantasks" +# define opihi_prompt "pantasks: " +# define opihi_description "parallel task scheduler\n" +# define opihi_history ".pantasks" +# define opihi_rcfile ".pantasksrc" + +/* program-dependent initialization */ +void program_init (int *argc, char **argv) { + + auto_break = TRUE; + + /* load the commands used by this implementation */ + InitBasic (); + InitData (); + InitPantasks (); + + gprintInit (); + + rl_readline_name = opihi_name; + rl_attempted_completion_function = command_completer; + rl_event_hook = NULL; + rl_set_keyboard_input_timeout (1000); + + set_str_variable ("HISTORY", opihi_history); + set_str_variable ("PROMPT", opihi_prompt); + set_str_variable ("RCFILE", opihi_rcfile); + +# ifdef HELPDIR_DEFAULT + set_str_variable ("HELPDIR", MACRO_NAME(HELPDIR_DEFAULT)); +# endif + + signal (SIGPIPE, gotsignal); + signal (SIGTSTP, gotsignal); + signal (SIGTTIN, gotsignal); + return; +} + +/* standard welcome message */ +void welcome () { + gprint (GP_LOG, "\n"); + gprint (GP_LOG, "Welcome to %s - %s\n\n", opihi_name, opihi_description); +} + +/* add program-dependent exit functions here */ +void cleanup () { + QuitImage (); + QuitGraph (); + QuitController (); + return; +} + +void gotsignal (int signum) { + gprint (GP_ERR, "got signal : %d\n", signum); + return; +} + +/* call to opihi shell */ +int main (int argc, char **argv) { + int status; + status = opihi (argc, argv); + exit (status); +} Index: /tags/pantasks-0-5/Ohana/src/opihi/pantasks/pantasks_client.c =================================================================== --- /tags/pantasks-0-5/Ohana/src/opihi/pantasks/pantasks_client.c (revision 8568) +++ /tags/pantasks-0-5/Ohana/src/opihi/pantasks/pantasks_client.c (revision 8568) @@ -0,0 +1,64 @@ +# include "pantasks.h" + +# define opihi_name "pantasks client shell" +# define opihi_prompt "pantasks: " +# define opihi_description "parallel task scheduler\n" +# define opihi_history ".pantasks" +# define opihi_rcfile ".pantasksrc" + +/* program-dependent initialization */ +void program_init (int *argc, char **argv) { + + auto_break = TRUE; + + /* load the commands used by this implementation */ + InitBasic (); + InitData (); + InitAstro (); + InitPantasksClient (); + + gprintInit (); + + rl_readline_name = opihi_name; + rl_attempted_completion_function = command_completer; + + set_str_variable ("HISTORY", opihi_history); + set_str_variable ("PROMPT", opihi_prompt); + set_str_variable ("RCFILE", opihi_rcfile); + +# ifdef HELPDIR_DEFAULT + set_str_variable ("HELPDIR", MACRO_NAME(HELPDIR_DEFAULT)); +# endif + + signal (SIGPIPE, gotsignal); + signal (SIGTSTP, gotsignal); + signal (SIGTTIN, gotsignal); + return; +} + +/* standard welcome message */ +void welcome () { + gprint (GP_LOG, "\n"); + gprint (GP_LOG, "Welcome to %s - %s\n\n", opihi_name, opihi_description); +} + +/* add program-dependent exit functions here */ +void cleanup () { + QuitImage (); + QuitGraph (); + return; +} + +/* default signal handling */ +void gotsignal (int signum) { + gprint (GP_ERR, "got signal : %d\n", signum); + return; +} + +/* call to client_shell: this opihi tool uses a slightly different + shell function from the standard tools */ +int main (int argc, char **argv) { + int status; + status = client_shell (argc, argv); + exit (status); +} Index: /tags/pantasks-0-5/Ohana/src/opihi/pantasks/pantasks_server.c =================================================================== --- /tags/pantasks-0-5/Ohana/src/opihi/pantasks/pantasks_server.c (revision 8568) +++ /tags/pantasks-0-5/Ohana/src/opihi/pantasks/pantasks_server.c (revision 8568) @@ -0,0 +1,86 @@ +# include "pantasks.h" + +/* pantasks-server is not a readline program. + it runs three (or four) threads: + - one thread listens on the socket for incomming connections + and performs client validation + - one thread parses commands from the connected clients + - one thread runs the scheduler loop + - one thread may run the controller loop independently + + * note: the client shell traps certain commands and performs them + locally. at a minimum: help. perhaps: input, other I/O commands? +*/ + +/* program-dependent initialization */ +int main (int argc, char **argv) { + + pthread_t clientsThread; + pthread_t tasksThread; + // pthread_t jobsThread; + pthread_t inputsThread; + pthread_t controllerThread; + int InitSocket, BindSocket; + SockAddress Address; + + general_init (&argc, argv); + DefineValidIP (); + + /* XXX not sure what is a good setting for this in the server */ + auto_break = TRUE; + + /* load the commands used by this implementation */ + InitBasic (); + InitData (); + InitPantasksServer (); + + gprintInit (); // each thread needs to init the printing system + InitInputs (); + + signal (SIGPIPE, gotsignal); + // signal (SIGTSTP, gotsignal); + // signal (SIGTTIN, gotsignal); + signal (SIGINT, SIG_DFL); + + startup (&argc, argv); + + /* start up the background threads here */ + pthread_create (&clientsThread, NULL, &ListenClients, NULL); + pthread_create (&tasksThread, NULL, &CheckTasksThread, NULL); + // pthread_create (&jobsThread, NULL, &CheckJobsThread, NULL); + pthread_create (&controllerThread, NULL, &CheckControllerThread, NULL); + pthread_create (&inputsThread, NULL, &CheckInputsThread, NULL); + + /* in this loop, we listen for incoming connections, validate, and + pass them to the ListenClient thread */ + + /* create the listening socket */ + InitSocket = InitServerSocket (&Address); + + InitPassword (); + + while (1) { + + /* wait for clients to make connection */ + BindSocket = WaitServerSocket (InitSocket, &Address); + if (BindSocket == -1) continue; + + /* validate : wait for password */ + if (!CheckPassword (BindSocket)) continue; + + AddNewClient (BindSocket); + } +} + +/* add program-dependent exit functions here */ +void cleanup () { + // QuitImage (); + // QuitGraph (); + // QuitController (); + return; +} + +void gotsignal (int signum) { + gprint (GP_ERR, "got signal : %d\n", signum); + return; +} Index: /tags/pantasks-0-5/Ohana/src/opihi/pantasks/pulse.c =================================================================== --- /tags/pantasks-0-5/Ohana/src/opihi/pantasks/pulse.c (revision 8568) +++ /tags/pantasks-0-5/Ohana/src/opihi/pantasks/pulse.c (revision 8568) @@ -0,0 +1,16 @@ +# include "pantasks.h" + +int pulse (int argc, char **argv) { + + int Nusec; + + if (argc != 2) { + gprint (GP_ERR, "USAGE: pulse (microseconds)\n"); + return (FALSE); + } + + Nusec = atoi (argv[1]); + rl_set_keyboard_input_timeout (Nusec); + + return (TRUE); +} Index: /tags/pantasks-0-5/Ohana/src/opihi/pantasks/run.c =================================================================== --- /tags/pantasks-0-5/Ohana/src/opihi/pantasks/run.c (revision 8568) +++ /tags/pantasks-0-5/Ohana/src/opihi/pantasks/run.c (revision 8568) @@ -0,0 +1,21 @@ +# include "pantasks.h" + +// XXX for client/server, we need to simply start or stop the +// appropriate threads +// with one thread for each of the major actions, this would +// make it easy to keep the controller running and stop the +// scheduler (don't run CheckTasks, but run everything else +// until nothing is left... + +int run (int argc, char **argv) { + + if (argc != 1) { + gprint (GP_ERR, "USAGE: run\n"); + return (FALSE); + } + + InitTaskTimers (); + rl_event_hook = CheckSystem; + + return (TRUE); +} Index: /tags/pantasks-0-5/Ohana/src/opihi/pantasks/scheduler.c =================================================================== --- /tags/pantasks-0-5/Ohana/src/opihi/pantasks/scheduler.c (revision 8568) +++ /tags/pantasks-0-5/Ohana/src/opihi/pantasks/scheduler.c (revision 8568) @@ -0,0 +1,77 @@ +# include "scheduler.h" + +# define opihi_name "scheduler" +# define opihi_prompt "sched: " +# define opihi_description "parallel task scheduler\n" +# define opihi_history ".sched" +# define opihi_rcfile ".schedrc" + +/* program-dependent initialization */ +void initialize (int argc, char **argv) { + + auto_break = TRUE; + + Nlists = 0; + ALLOCATE (lists, List, 1); + + /* init functions required by libraries */ + /* -libopihi */ + InitCommands (); + InitMacros (); + InitBuffers (); + InitVectors (); + InitVariables (); + + /* -libdisplay */ + InitGraph (); + InitImage (); + + /* load the commands used by this implementation */ + InitBasic (); + InitData (); + InitOutfile (); + InitSched (); + InitTasks (); + InitJobs (); + InitJobIDs (); + + 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); + set_str_variable ("PROMPT", opihi_prompt); + set_str_variable ("RCFILE", opihi_rcfile); +# ifdef HELPDIR_DEFAULT + set_str_variable ("HELPDIR", MACRO_NAME(HELPDIR_DEFAULT)); +# endif + + { /* check history file permission */ + FILE *f; + f = fopen (opihi_history, "a"); + if (f == NULL) /* no current history file here */ + gprint (GP_ERR, "can't save history.\n"); + else + fclose (f); + stifle_history (200); + read_history (opihi_history); + } + + signal (SIGINT, SIG_IGN); + return; +} + +/* standard welcome message */ +void welcome () { + gprint (GP_LOG, "\n"); + gprint (GP_LOG, "Welcome to %s - %s\n\n", opihi_name, opihi_description); +} + +/* add program-dependent exit functions here */ +void cleanup () { + QuitImage (); + QuitGraph (); + QuitController (); + return; +} Index: /tags/pantasks-0-5/Ohana/src/opihi/pantasks/server.c =================================================================== --- /tags/pantasks-0-5/Ohana/src/opihi/pantasks/server.c (revision 8568) +++ /tags/pantasks-0-5/Ohana/src/opihi/pantasks/server.c (revision 8568) @@ -0,0 +1,68 @@ +# include "pantasks.h" + +/* most of these are defined in cmd.basic */ +int quit PROTO((int, char **)); +int input PROTO((int, char **)); +int server_load PROTO((int, char **)); +int server_run PROTO((int, char **)); +int server_stop PROTO((int, char **)); +int cd PROTO((int, char **)); +int pwd PROTO((int, char **)); +int output PROTO((int, char **)); + +CommandF *FindServerCommand (char *cmd); + +static Command server_cmds[] = { + {"exit", quit, "shutdown server"}, + {"quit", quit, "shutdown server"}, + {"input", input, "load input file on server"}, + {"load", server_load, "load input file on server"}, + {"run", server_run, "run scheduler"}, + {"stop", server_stop, "stop scheduler"}, + {"cd", cd, "set local directory"}, + {"pwd", pwd, "check local directory"}, + {"output", output, "set server output destinations"}, +}; + +int server (int argc, char **argv) { + + int i, N, status; + CommandF *func; + + if (argc < 2) { + gprint (GP_ERR, "USAGE: server (command) ... \n"); + return (FALSE); + } + + if (!strcasecmp (argv[1], "help")) { + N = sizeof (server_cmds) / sizeof (Command); + + for (i = 0; i < N; i++) { + gprint (GP_LOG, "%-15s %s\n", server_cmds[i].name, server_cmds[i].help); + } + return (TRUE); + } + + func = FindServerCommand (argv[1]); + if (func == NULL) { + gprint (GP_ERR, "invalid server command\n"); + return (FALSE); + } + + status = (*func)(argc - 1, argv + 1); + return (status); +} + +CommandF *FindServerCommand (char *cmd) { + + int i, N; + + N = sizeof (server_cmds) / sizeof (Command); + + for (i = 0; i < N; i++) { + if (!strcmp (server_cmds[i].name, cmd)) { + return (server_cmds[i].func); + } + } + return (NULL); +} Index: /tags/pantasks-0-5/Ohana/src/opihi/pantasks/server_load.c =================================================================== --- /tags/pantasks-0-5/Ohana/src/opihi/pantasks/server_load.c (revision 8568) +++ /tags/pantasks-0-5/Ohana/src/opihi/pantasks/server_load.c (revision 8568) @@ -0,0 +1,18 @@ +# include "pantasks.h" + +int server_load (int argc, char **argv) { + + char *input; + + if (argc != 2) { + gprint (GP_ERR, "USAGE: server load (file)\n"); + return (FALSE); + } + + // the allocated input string is eventually freed + // by the InputQueue system + input = strcreate (argv[1]); + AddNewInput (input); + + return (TRUE); +} Index: /tags/pantasks-0-5/Ohana/src/opihi/pantasks/server_run.c =================================================================== --- /tags/pantasks-0-5/Ohana/src/opihi/pantasks/server_run.c (revision 8568) +++ /tags/pantasks-0-5/Ohana/src/opihi/pantasks/server_run.c (revision 8568) @@ -0,0 +1,29 @@ +# include "pantasks.h" + +int server_run (int argc, char **argv) { + + if (argc != 1) { + gprint (GP_ERR, "USAGE: server run\n"); + return (FALSE); + } + + CheckTasksSetState (TRUE); + // CheckJobsSetState (TRUE); + CheckControllerSetState (TRUE); + CheckInputsSetState (TRUE); + return (TRUE); +} + +int server_stop (int argc, char **argv) { + + if (argc != 1) { + gprint (GP_ERR, "USAGE: server stop\n"); + return (FALSE); + } + + CheckTasksSetState (FALSE); + // CheckJobsSetState (FALSE); + CheckControllerSetState (FALSE); + CheckInputsSetState (FALSE); + return (TRUE); +} Index: /tags/pantasks-0-5/Ohana/src/opihi/pantasks/server_threads.c =================================================================== --- /tags/pantasks-0-5/Ohana/src/opihi/pantasks/server_threads.c (revision 8568) +++ /tags/pantasks-0-5/Ohana/src/opihi/pantasks/server_threads.c (revision 8568) @@ -0,0 +1,153 @@ +# include "pantasks.h" + +/* this mutex is used by the serialized threads */ + +static pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER; + +void SerialThreadLock () { + pthread_mutex_lock (&mutex); +} + +void SerialThreadUnlock () { + pthread_mutex_unlock (&mutex); +} + +/* the threaded version of pantasks does not worry about limiting the + time spent in one of the test loops -- comms with the client are + in a separate thread anyway... +*/ + +int TestElapsedCheck () { + return (FALSE); +} + +/** things related to CheckTasks **/ + +static int CheckTasksRun = FALSE; + +void CheckTasksSetState (int state) { + CheckTasksRun = state; +} +int CheckTasksGetState () { + return (CheckTasksRun); +} + +void *CheckTasksThread (void *data) { + + gprintInit (); // each thread needs to init the printing system + while (1) { + + // check for thread suspend + if (!CheckTasksRun) { + usleep (100000); // idle if thread action is suspended + continue; + } + + // one run of the task checker + SerialThreadLock (); + CheckTasks (); + CheckJobs (); + SerialThreadUnlock (); + fprintf (stderr, "T"); + usleep (10000); // allow other threads a chance to run + } +} + +# if 0 +// XXX need to harvest jobs on the same thread as they were +// spawned. for the moment, put CheckTasks and CheckJobs in +// a single thread. +/** things related to CheckJobs **/ + +static int CheckJobsRun = FALSE; + +void CheckJobsSetState (int state) { + CheckJobsRun = state; +} +int CheckJobsGetState () { + return (CheckJobsRun); +} + +void *CheckJobsThread (void *data) { + + gprintInit (); // each thread needs to init the printing system + while (1) { + + // check for thread suspend + if (!CheckJobsRun) { + usleep (100000); // idle if thread action is suspended + continue; + } + + // one run of the task checker + SerialThreadLock (); + CheckJobs (); + SerialThreadUnlock (); + fprintf (stderr, "J"); + usleep (10000); // allow other threads a chance to run + } +} +# endif + +/** things related to CheckController **/ + +static int CheckControllerRun = FALSE; + +void CheckControllerSetState (int state) { + CheckControllerRun = state; +} +int CheckControllerGetState () { + return (CheckControllerRun); +} + +void *CheckControllerThread (void *data) { + + gprintInit (); // each thread needs to init the printing system + while (1) { + + // check for thread suspend + if (!CheckControllerRun) { + usleep (100000); // idle if thread action is suspended + continue; + } + + // one run of the task checker + SerialThreadLock (); + CheckController (); + CheckControllerOutput (); + SerialThreadUnlock (); + fprintf (stderr, "C"); + usleep (10000); // allow other threads a chance to run + } +} + +/** things related to CheckInputs **/ + +static int CheckInputsRun = TRUE; + +void CheckInputsSetState (int state) { + CheckInputsRun = state; +} +int CheckInputsGetState () { + return (CheckInputsRun); +} + +void *CheckInputsThread (void *data) { + + gprintInit (); // each thread needs to init the printing system + while (1) { + + // check for thread suspend + if (!CheckInputsRun) { + usleep (100000); // idle if thread action is suspended + continue; + } + + // one run of the task checker + SerialThreadLock (); + CheckInputs (); + SerialThreadUnlock (); + fprintf (stderr, "I"); + usleep (10000); // allow other threads a chance to run + } +} Index: /tags/pantasks-0-5/Ohana/src/opihi/pantasks/status.c =================================================================== --- /tags/pantasks-0-5/Ohana/src/opihi/pantasks/status.c (revision 8568) +++ /tags/pantasks-0-5/Ohana/src/opihi/pantasks/status.c (revision 8568) @@ -0,0 +1,19 @@ +# include "pantasks.h" + +int status_sys (int argc, char **argv) { + + gprint (GP_LOG, "\n"); + if (rl_event_hook == NULL) { + gprint (GP_LOG, " Scheduler is stopped\n"); + } else { + gprint (GP_LOG, " Scheduler is running\n"); + } + if (CheckControllerStatus()) { + gprint (GP_LOG, " Controller is running\n"); + } else { + gprint (GP_LOG, " Controller is stopped\n"); + } + ListTasks (FALSE); + ListJobs (); + return (TRUE); +} Index: /tags/pantasks-0-5/Ohana/src/opihi/pantasks/status_server.c =================================================================== --- /tags/pantasks-0-5/Ohana/src/opihi/pantasks/status_server.c (revision 8568) +++ /tags/pantasks-0-5/Ohana/src/opihi/pantasks/status_server.c (revision 8568) @@ -0,0 +1,19 @@ +# include "pantasks.h" + +int status_server (int argc, char **argv) { + + gprint (GP_LOG, "\n"); + if (CheckTasksGetState()) { + gprint (GP_LOG, " Scheduler is running\n"); + } else { + gprint (GP_LOG, " Scheduler is stopped\n"); + } + if (CheckControllerGetState()) { + gprint (GP_LOG, " Controller is active\n"); + } else { + gprint (GP_LOG, " Controller is not active\n"); + } + ListTasks (FALSE); + ListJobs (); + return (TRUE); +} Index: /tags/pantasks-0-5/Ohana/src/opihi/pantasks/stop.c =================================================================== --- /tags/pantasks-0-5/Ohana/src/opihi/pantasks/stop.c (revision 8568) +++ /tags/pantasks-0-5/Ohana/src/opihi/pantasks/stop.c (revision 8568) @@ -0,0 +1,13 @@ +# include "pantasks.h" + +int stop (int argc, char **argv) { + + if (argc != 1) { + gprint (GP_ERR, "USAGE: stop\n"); + return (FALSE); + } + + rl_event_hook = NULL; + + return (TRUE); +} Index: /tags/pantasks-0-5/Ohana/src/opihi/pantasks/task.c =================================================================== --- /tags/pantasks-0-5/Ohana/src/opihi/pantasks/task.c (revision 8568) +++ /tags/pantasks-0-5/Ohana/src/opihi/pantasks/task.c (revision 8568) @@ -0,0 +1,115 @@ +# include "pantasks.h" +# define prompt "> " + +int task (int argc, char **argv) { + + int N, hash; + int ThisList, status; + char *input, *outline; + Task *task; + + if ((N = get_argument (argc, argv, "-list"))) { + remove_argument (N, &argc, argv); + ListTasks (FALSE); + return (TRUE); + } + + if ((N = get_argument (argc, argv, "-longlist"))) { + remove_argument (N, &argc, argv); + ListTasks (TRUE); + return (TRUE); + } + + if ((N = get_argument (argc, argv, "-show"))) { + remove_argument (N, &argc, argv); + if (argc != 2) goto usage; + ShowTask (argv[N]); + return (TRUE); + } + + if (argc != 2) goto usage; + + task = FindTask (argv[1]); + if (task == NULL) { /**** new task ****/ + task = CreateTask (argv[1]); + } else { + RemoveTask (task); + SetNewTask (task); + } + /* While a task is being defined, it is removed from the task list. The new task is added to the task list + when the definition process is complete. + XXX If an outstanding job has a task deleted, it will not be able to complete... */ + + /* read in task from appropriate source (keyboard or list) until end */ + + /* allowed tokens: command, host, stderr, periods, trange, nmax, task.exit, task.exec, end */ + + ThisList = Nlists; + while (1) { + + /* get the next line (from correct place) */ + if (ThisList == 0) + input = readline (prompt); + else + input = get_next_listentry (ThisList); + + if ((ThisList == 0) && (input == (char *) NULL)) { + gprint (GP_ERR, "end task with 'END'\n"); + continue; + } + if ((ThisList > 0) && (input == (char *) NULL)) { + gprint (GP_ERR, "missing 'END' in task definition\n"); + input = strcreate ("end"); + } + if (ThisList == 0) ohana_memregister (input); + + stripwhite (input); + hash = TaskHash (input); + switch (hash) { + + case TASK_EMPTY: + case TASK_COMMENT: + free (input); + break; + + case TASK_END: + /* I need to add in a test here to see if all task elements + have been defined. delete the task if not */ + free (input); + /* validate the new task: all mandatory elements defined? */ + if (!ValidateTask (task, FALSE)) { + DeleteNewTask (); + return (FALSE); + } + RegisterNewTask (); + return (TRUE); + break; + + case TASK_TRANGE: + case TASK_NMAX: + case TASK_HOST: + case TASK_EXIT: + case TASK_EXEC: + case TASK_COMMAND: + case TASK_OPTIONS: + case TASK_PERIODS: + status = command (input, &outline, TRUE); + if (outline != NULL) free (outline); + /* what to do if command is invalid? + if (!status) return (FALSE); */ + break; + + case TASK_NONE: + gprint (GP_ERR, "unknown task command %s\n", input); + break; + } + } + +usage: + gprint (GP_ERR, "USAGE: task -list\n"); + gprint (GP_ERR, "USAGE: task -longlist\n"); + gprint (GP_ERR, "USAGE: task -show (task)\n"); + gprint (GP_ERR, "USAGE: task \n"); + gprint (GP_ERR, " (enter commands & task functions; end with the word 'END')\n"); + return (FALSE); +} Index: /tags/pantasks-0-5/Ohana/src/opihi/pantasks/task_command.c =================================================================== --- /tags/pantasks-0-5/Ohana/src/opihi/pantasks/task_command.c (revision 8568) +++ /tags/pantasks-0-5/Ohana/src/opihi/pantasks/task_command.c (revision 8568) @@ -0,0 +1,38 @@ +# include "pantasks.h" + +int task_command (int argc, char **argv) { + + int i; + Task *task; + + if (argc < 2) { + gprint (GP_ERR, "USAGE: command . ..\n"); + gprint (GP_ERR, " (define command for this task)\n"); + return (FALSE); + } + + task = GetNewTask (); + if (task == NULL) { + task = GetActiveTask (); + if (task == NULL) { + gprint (GP_ERR, "ERROR: not defining or running a task\n"); + return (FALSE); + } + } + + /* free existing memory used by argv */ + if (task[0].argc != 0) { + for (i = 0; i < task[0].argc; i++) { + if (task[0].argv[i] != NULL) free (task[0].argv[i]); + } + free (task[0].argv); + } + + /* create new memory for argv */ + task[0].argc = argc - 1; + ALLOCATE (task[0].argv, char *, MAX (task[0].argc, 1)); + for (i = 0; i < task[0].argc; i++) { + task[0].argv[i] = strcreate (argv[i+1]); + } + return (TRUE); +} Index: /tags/pantasks-0-5/Ohana/src/opihi/pantasks/task_host.c =================================================================== --- /tags/pantasks-0-5/Ohana/src/opihi/pantasks/task_host.c (revision 8568) +++ /tags/pantasks-0-5/Ohana/src/opihi/pantasks/task_host.c (revision 8568) @@ -0,0 +1,42 @@ +# include "pantasks.h" + +int task_host (int argc, char **argv) { + + int N, RequiredHost; + Task *task; + + RequiredHost = FALSE; + if ((N = get_argument (argc, argv, "-required"))) { + remove_argument (N, &argc, argv); + RequiredHost = TRUE; + } + + if (argc != 2) { + gprint (GP_ERR, "USAGE: host [-required]\n"); + gprint (GP_ERR, " define host machine for this task\n"); + gprint (GP_ERR, " -required flags indicates controller must use this host\n"); + gprint (GP_ERR, " value of 'local' for host indicates process not using controller\n"); + gprint (GP_ERR, " value of 'anyhost' for host indicates controller may assign at will\n"); + return (FALSE); + } + + task = GetNewTask (); + if (task == NULL) { + task = GetActiveTask (); + if (task == NULL) { + gprint (GP_ERR, "ERROR: not defining or running a task\n"); + return (FALSE); + } + } + task[0].host_required = RequiredHost; + + if (task[0].host != NULL) free (task[0].host); + task[0].host = NULL; + + if (!strcasecmp (argv[1], "LOCAL")) return (TRUE); + + task[0].host = strcreate (argv[1]); + return (TRUE); +} + +/* apparently, local is the default! */ Index: /tags/pantasks-0-5/Ohana/src/opihi/pantasks/task_macros.c =================================================================== --- /tags/pantasks-0-5/Ohana/src/opihi/pantasks/task_macros.c (revision 8568) +++ /tags/pantasks-0-5/Ohana/src/opihi/pantasks/task_macros.c (revision 8568) @@ -0,0 +1,151 @@ +# include "pantasks.h" +# define D_NLINES 100 +# define prompt "> " + +int task_macros (int argc, char **argv) { + + int i, N, NLINES, done, depth, ThisList; + char *input; + Macro *macro; + Task *task; + + if (!strcmp (argv[0], "task.exec") && (argc != 1)) { + gprint (GP_ERR, "USAGE: task.exec\n"); + gprint (GP_ERR, " (define pre-exec macro for this task)\n"); + return (FALSE); + } + + if (!strcmp (argv[0], "task.exit") && (argc != 2)) { + gprint (GP_ERR, "USAGE: task.exit (state)\n"); + gprint (GP_ERR, " (define exit state macro for this task)\n"); + return (FALSE); + } + + task = GetNewTask (); + if (task == NULL) { + gprint (GP_ERR, "ERROR: not defining or running a task\n"); + return (FALSE); + } + + macro = NULL; + + /*** identify which Macro in Task this particular macro goes with ***/ + if (!strcmp (argv[0], "task.exec")) { + if (task[0].exec != NULL) { + FreeMacro (task[0].exec); + free (task[0].exec); + } + ALLOCATE (task[0].exec, Macro, 1); + macro = task[0].exec; + macro[0].name = strcreate ("exec"); + goto found; + } + if (!strcmp (argv[0], "task.exit") && !strcmp (argv[1], "timeout")) { + if (task[0].timeout != NULL) { + FreeMacro (task[0].timeout); + free (task[0].timeout); + } + ALLOCATE (task[0].timeout, Macro, 1); + macro = task[0].timeout; + macro[0].name = strcreate ("timeout"); + goto found; + } + if (!strcmp (argv[0], "task.exit") && !strcmp (argv[1], "crash")) { + if (task[0].crash != NULL) { + FreeMacro (task[0].crash); + free (task[0].crash); + } + ALLOCATE (task[0].crash, Macro, 1); + macro = task[0].crash; + macro[0].name = strcreate ("crash"); + goto found; + } + if (!strcmp (argv[0], "task.exit") && !strcmp (argv[1], "default")) { + if (task[0].defexit != NULL) { + FreeMacro (task[0].defexit); + free (task[0].defexit); + } + ALLOCATE (task[0].defexit, Macro, 1); + macro = task[0].defexit; + macro[0].name = strcreate ("default"); + goto found; + } + if (!strcmp (argv[0], "task.exit")) { /* generic exit function */ + for (i = 0; i < task[0].Nexit; i++) { /* find exiting exit function with same name */ + if (!strcmp (task[0].exit[i][0].name, argv[1])) { /*** replace this one ***/ + FreeMacro (task[0].exit[i]); + free (task[0].exit[i]); + ALLOCATE (task[0].exit[i], Macro, 1); + macro = task[0].exit[i]; + macro[0].name = strcreate (argv[1]); + goto found; + } + } + /** create a new exit status macro **/ + N = task[0].Nexit; + ALLOCATE (task[0].exit[N], Macro, 1); + task[0].Nexit ++; + if (task[0].Nexit == task[0].NEXIT) { + task[0].NEXIT += 10; + REALLOCATE (task[0].exit, Macro *, task[0].NEXIT); + } + macro = task[0].exit[N]; + macro[0].name = strcreate (argv[1]); + goto found; + } + +found: + + macro[0].Nlines = 0; + NLINES = D_NLINES; + ALLOCATE (macro[0].line, char *, NLINES); + + /* load macro lines from appropriate source (readline / list) */ + + depth = 0; + done = FALSE; + ThisList = Nlists; + while (!done) { + + /* get the next line (from correct place) */ + if (ThisList == 0) + input = readline (prompt); + else + input = get_next_listentry (ThisList); + + if ((ThisList == 0) && (input == (char *) NULL)) { + gprint (GP_ERR, "end macro with 'END'\n"); + continue; + } + if ((ThisList > 0) && (input == (char *) NULL)) { + gprint (GP_ERR, "missing 'END' in macro definition\n"); + input = strcreate ("end"); + } + if (ThisList == 0) ohana_memregister (input); + + stripwhite (input); + + /* test for new macro (or other list, in the future?) */ + if (is_list (input)) depth ++; + + /* test for end of nested list -- if not nested, END refers to this macro */ + if (!strncasecmp (input, "END", 3)) { + depth --; + if (depth < 0) { /* we hit the last "END", macro is done */ + free (input); + REALLOCATE (macro[0].line, char *, MAX (1, macro[0].Nlines)); + return (TRUE); + } + } + + if (*input) { + macro[0].line[macro[0].Nlines] = input; + macro[0].Nlines ++; + if (macro[0].Nlines == NLINES - 1) { + NLINES += D_NLINES; + REALLOCATE (macro[0].line, char *, NLINES); + } + } + } + return (TRUE); +} Index: /tags/pantasks-0-5/Ohana/src/opihi/pantasks/task_nmax.c =================================================================== --- /tags/pantasks-0-5/Ohana/src/opihi/pantasks/task_nmax.c (revision 8568) +++ /tags/pantasks-0-5/Ohana/src/opihi/pantasks/task_nmax.c (revision 8568) @@ -0,0 +1,21 @@ +# include "pantasks.h" + +int task_nmax (int argc, char **argv) { + + Task *task; + + if (argc != 2) goto usage; + + task = GetNewTask (); + if (task == NULL) { + gprint (GP_ERR, "ERROR: not defining or running a task\n"); + return (FALSE); + } + + task[0].Nmax = atoi (argv[1]); + return (TRUE); + +usage: + gprint (GP_ERR, "USAGE: nmax N\n"); + return (FALSE); +} Index: /tags/pantasks-0-5/Ohana/src/opihi/pantasks/task_options.c =================================================================== --- /tags/pantasks-0-5/Ohana/src/opihi/pantasks/task_options.c (revision 8568) +++ /tags/pantasks-0-5/Ohana/src/opihi/pantasks/task_options.c (revision 8568) @@ -0,0 +1,38 @@ +# include "pantasks.h" + +int task_options (int argc, char **argv) { + + int i; + Task *task; + + if (argc < 2) { + gprint (GP_ERR, "USAGE: options . ..\n"); + gprint (GP_ERR, " (define options for this task)\n"); + return (FALSE); + } + + task = GetNewTask (); + if (task == NULL) { + task = GetActiveTask (); + if (task == NULL) { + gprint (GP_ERR, "ERROR: not defining or running a task\n"); + return (FALSE); + } + } + + /* free existing memory used by optv */ + if (task[0].optc != 0) { + for (i = 0; i < task[0].optc; i++) { + if (task[0].optv[i] != NULL) free (task[0].optv[i]); + } + free (task[0].optv); + } + + /* create new memory for optv */ + task[0].optc = argc - 1; + ALLOCATE (task[0].optv, char *, MAX (task[0].optc, 1)); + for (i = 0; i < task[0].optc; i++) { + task[0].optv[i] = strcreate (argv[i+1]); + } + return (TRUE); +} Index: /tags/pantasks-0-5/Ohana/src/opihi/pantasks/task_periods.c =================================================================== --- /tags/pantasks-0-5/Ohana/src/opihi/pantasks/task_periods.c (revision 8568) +++ /tags/pantasks-0-5/Ohana/src/opihi/pantasks/task_periods.c (revision 8568) @@ -0,0 +1,56 @@ +# include "pantasks.h" + +int task_periods (int argc, char **argv) { + + int N, Poll, Exec, Timeout; + float PollValue, ExecValue, TimeoutValue; + Task *task; + + PollValue = 0; + Poll = FALSE; + if ((N = get_argument (argc, argv, "-poll"))) { + Poll = TRUE; + remove_argument (N, &argc, argv); + PollValue = atof(argv[N]); + remove_argument (N, &argc, argv); + } + + ExecValue = 0; + Exec = FALSE; + if ((N = get_argument (argc, argv, "-exec"))) { + Exec = TRUE; + remove_argument (N, &argc, argv); + ExecValue = atof(argv[N]); + remove_argument (N, &argc, argv); + } + + TimeoutValue = 0; + Timeout = FALSE; + if ((N = get_argument (argc, argv, "-timeout"))) { + Timeout = TRUE; + remove_argument (N, &argc, argv); + TimeoutValue = atof(argv[N]); + remove_argument (N, &argc, argv); + } + + if (argc != 1) { + gprint (GP_ERR, "USAGE: periods [-poll value] [-exec value] [-timeout value]\n"); + gprint (GP_ERR, " define time periods for this task\n"); + return (FALSE); + } + + task = GetNewTask (); + if (task == NULL) { + task = GetActiveTask (); + if (task == NULL) { + gprint (GP_ERR, "ERROR: not defining or running a task\n"); + return (FALSE); + } + } + + if (Poll) task[0].poll_period = PollValue; + if (Exec) task[0].exec_period = ExecValue; + if (Timeout) task[0].timeout_period = TimeoutValue; + + return (TRUE); +} Index: /tags/pantasks-0-5/Ohana/src/opihi/pantasks/task_trange.c =================================================================== --- /tags/pantasks-0-5/Ohana/src/opihi/pantasks/task_trange.c (revision 8568) +++ /tags/pantasks-0-5/Ohana/src/opihi/pantasks/task_trange.c (revision 8568) @@ -0,0 +1,93 @@ +# include "pantasks.h" + +int task_trange (int argc, char **argv) { + + int N; + Task *task; + TimeRange range; + + /* reset the tranges for the current task */ + if ((N = get_argument (argc, argv, "-reset"))) { + remove_argument (N, &argc, argv); + if (argc != 1) goto usage; + + task = GetNewTask (); + if (task == NULL) { + gprint (GP_ERR, "ERROR: not defining or running a task\n"); + return (FALSE); + } + + task[0].Nranges = 0; + REALLOCATE (task[0].ranges, TimeRange, 1); + return (TRUE); + } + + range.Nmax = 0; + range.Nrun = 0; + if ((N = get_argument (argc, argv, "-nmax"))) { + remove_argument (N, &argc, argv); + range.Nmax = atoi (argv[N]); + remove_argument (N, &argc, argv); + } + + /* keep = true means the time range defines a valid time range + keep = false means the time range defines an invalid range */ + range.include = TRUE; + if ((N = get_argument (argc, argv, "-exclude"))) { + remove_argument (N, &argc, argv); + range.include = FALSE; + } + + if (argc != 3) goto usage; + + task = GetNewTask (); + if (task == NULL) { + gprint (GP_ERR, "ERROR: not defining or running a task\n"); + return (FALSE); + } + + /* test for Mon[@HH:MM:SS] - both must match */ + if (day_to_sec (argv[1], &range.start)) { + if (!day_to_sec (argv[2], &range.stop)) { + gprint (GP_ERR, "invalid day/time %s\n", argv[2]); + goto usage; + } + range.type = RANGE_WEEK; + goto valid; + } + + /* test for HH:MM:SS */ + if (hms_to_sec (argv[1], &range.start)) { + if (!hms_to_sec (argv[2], &range.stop)) { + gprint (GP_ERR, "invalid time %s\n", argv[2]); + goto usage; + } + range.type = RANGE_DAY; + goto valid; + } + + /* this test does not fail sufficiently robustly for invalid inputs */ + /* test for YYYY/MM/DD - both must match */ + if (str_to_time (argv[1], &range.start)) { + if (!str_to_time (argv[2], &range.stop)) { + gprint (GP_ERR, "invalid date/time %s\n", argv[2]); + goto usage; + } + range.type = RANGE_ABS; + goto valid; + } + goto usage; + +valid: + N = task[0].Nranges; + task[0].Nranges ++; + REALLOCATE (task[0].ranges, TimeRange, task[0].Nranges); + + task[0].ranges[N] = range; + return (TRUE); + +usage: + gprint (GP_ERR, "USAGE: trange start end [-nmax N]\n"); + gprint (GP_ERR, "USAGE: trange -reset\n"); + return (FALSE); +} Index: /tags/pantasks-0-5/Ohana/src/opihi/pantasks/verbose.c =================================================================== --- /tags/pantasks-0-5/Ohana/src/opihi/pantasks/verbose.c (revision 8568) +++ /tags/pantasks-0-5/Ohana/src/opihi/pantasks/verbose.c (revision 8568) @@ -0,0 +1,37 @@ +# include "pantasks.h" + +static int VERBOSE = FALSE; + +int verbose (int argc, char **argv) { + + if (argc == 1) { + if (VERBOSE) { + gprint (GP_LOG, "verbose mode ON\n"); + } else { + gprint (GP_LOG, "verbose mode OFF\n"); + } + return (TRUE); + } + + if (argc == 2) { + if (!strcasecmp (argv[1], "ON")) { + VERBOSE = TRUE; + return (TRUE); + } + if (!strcasecmp (argv[1], "OFF")) { + VERBOSE = FALSE; + return (TRUE); + } + if (!strcasecmp (argv[1], "TOGGLE")) { + VERBOSE = ~VERBOSE; + return (TRUE); + } + } + + gprint (GP_ERR, "USAGE: verbose (on/off/toggle)\n"); + return (FALSE); +} + +int VerboseMode () { + return (VERBOSE); +} Index: /tags/pantasks-0-5/Ohana/src/opihi/pantasks/version.c =================================================================== --- /tags/pantasks-0-5/Ohana/src/opihi/pantasks/version.c (revision 8568) +++ /tags/pantasks-0-5/Ohana/src/opihi/pantasks/version.c (revision 8568) @@ -0,0 +1,17 @@ +# include "pantasks.h" +static char *name = "$Name: not supported by cvs2svn $"; + +int version (int argc, char **argv) { + + char *tmp; + + gprint (GP_LOG, "\n"); + gprint (GP_LOG, "pantasks version: %s\n", (tmp = strip_version (name))); free (tmp); + + gprint (GP_LOG, "opihi version: %s\n", (tmp = strip_version (opihi_version()))); free (tmp); + gprint (GP_LOG, "ohana version: %s\n", (tmp = strip_version (ohana_version()))); free (tmp); + gprint (GP_LOG, "gfits version: %s\n", (tmp = strip_version (gfits_version()))); free (tmp); + + gprint (GP_LOG, "compiled on %s %s\n", __DATE__, __TIME__); + return (TRUE); +}