Index: /trunk/Ohana/src/opihi/Makefile =================================================================== --- /trunk/Ohana/src/opihi/Makefile (revision 3186) +++ /trunk/Ohana/src/opihi/Makefile (revision 3187) @@ -11,4 +11,5 @@ dvo: lib.data lib.shell cmd.basic cmd.data cmd.astro sched: lib.data lib.shell cmd.basic cmd.data cmd.astro +pclient: lib.data lib.shell cmd.basic LIBS = cmd.basic cmd.data cmd.astro lib.data lib.shell @@ -16,5 +17,5 @@ PROGRAM = mana dvo -EXTRAS = dvo2 sched dimm +EXTRAS = dvo2 sched dimm pclient all: Index: /trunk/Ohana/src/opihi/include/pclient.h =================================================================== --- /trunk/Ohana/src/opihi/include/pclient.h (revision 3187) +++ /trunk/Ohana/src/opihi/include/pclient.h (revision 3187) @@ -0,0 +1,75 @@ +# include "external.h" +# include "shell.h" +# include "dvomath.h" +# include "convert.h" +# include "display.h" +# include +# include + +/** pclient global data **/ + +/** child status values **/ +enum { + PCLIENT_NONE, + PCLIENT_BUSY, + PCLIENT_EXIT, + PCLIENT_CRASH, +}; + +int ChildPID; /** current child PID **/ +int ChildStatus; /** current status of child **/ +int ChildExitStatus; /** recent exit status of child **/ + +/** child may be in the following states: + +PCLIENT_NONE +PCLIENT_BUSY +PCLIENT_EXIT +PCLIENT_CRASH + +allowed transitions: + +PCLIENT_NONE -> PCLIENT_BUSY (start a job) + +PCLIENT_BUSY -> PCLIENT_EXIT (job complete) +PCLIENT_BUSY -> PCLIENT_CRASH (job crashed) + +PCLIENT_EXIT -> PCLIENT_NONE (cleanup) +PCLIENT_CRASH -> PCLIENT_NONE (cleanup) + +**/ + +/** file descriptors for child I/O **/ +int child_stdin_fd[2]; +int child_stdout_fd[2]; +int child_stderr_fd[2]; + +typedef struct { + char *buffer; + int Nalloc; + int Nreset; + int Nblock; + int Nbuffer; +} IOBuffer; + +/** buffers for I/O storage **/ +IOBuffer child_stdout; +IOBuffer child_stderr; + +/** IOBuffers allow the following operations **/ + +/* create a new, empty IOBuffer with Nalloc free space */ +int InitIOBuffer (IOBuffer *buffer, int Nalloc); + +/* read from the file descriptor into IOBuffer, expanding as needed */ +int ReadtoIOBuffer (IOBuffer *buffer, int fd); + +/* empty the current IOBuffer, realloc to a minimal size */ +int FlushIOBuffer (IOBuffer *buffer); + +/* free memory associated with IOBuffer */ +void FreeIOBuffer (IOBuffer *buffer); + +int InitChild (); +int CheckChild (); +void CheckChildStatus (); Index: /trunk/Ohana/src/opihi/include/pcontrol.h =================================================================== --- /trunk/Ohana/src/opihi/include/pcontrol.h (revision 3187) +++ /trunk/Ohana/src/opihi/include/pcontrol.h (revision 3187) @@ -0,0 +1,107 @@ +# include "external.h" +# include "shell.h" +# include "dvomath.h" +# include "convert.h" +# include "display.h" +# include +# include + +/** pcontrol global data **/ + +/** job status values **/ +enum { + PCONTROL_JOB_PENDING, + PCONTROL_JOB_BUSY, + PCONTROL_JOB_EXIT, + PCONTROL_JOB_CRASH, +} JobStat; + +/** job status values **/ +enum { + PCONTROL_JOB_ANYHOST, + PCONTROL_JOB_WANTHOST, + PCONTROL_JOB_NEEDHOST, +} JobMode; + +/** host status values **/ +enum { + PCONTROL_HOST_IDLE, + PCONTROL_HOST_BUSY, + PCONTROL_HOST_DOWN, + PCONTROL_HOST_OFF, +} HostStat; + +typedef struct { + char *buffer; + int Nalloc; + int Nreset; + int Nblock; + int Nbuffer; +} IOBuffer; + +typedef struct { + char *buffer; + int Nalloc; + int Nmaxread; + int Nextra; + int Nlast; + int Nbuffer; +} Fifo; + +typedef struct timeval Ptime; + +/* data to define a job */ +typedef struct { + int argc; + char **argv; + char *hostname; + int exitstatus; + JobMode mode; + JobStat status; + IOBuffer stdin; + IOBuffer stdout; + IOBuffer stderr; + Ptime start; + Ptime accum; + Ptime timer; + struct Host *host; + int pid; + IDtype JobID; +} Job; + +/* data to define a host machine */ +typedef struct { + char *hostname; + int stdin; + int stdout; + int stderr; + int markoff; + HostStat status; + Ptime start; + Ptime accum; + Ptime timer; + struct Job *job; +} Host; + +typedef struct { + void **object; + int Nobject; + int NOBJECT; +} Queue; + +static char *CONNECT; /* connection method (ssh, rsh, etc) */ +static char *PCLIENT; /* shell on remote host (eg, pclient) */ + +/** IOBuffers allow the following operations (see pclient.h) **/ +int InitIOBuffer (IOBuffer *buffer, int Nalloc); +int ReadtoIOBuffer (IOBuffer *buffer, int fd); +int FlushIOBuffer (IOBuffer *buffer); +void FreeIOBuffer (IOBuffer *buffer); + +# define FREE(X) if (X != NULL) { free (X); } + +enum { + QUEUE_TOP = 0, + QUEUE_BOTTOM = -1, +} QueueWhere; + Index: /trunk/Ohana/src/opihi/lib.shell/command.c =================================================================== --- /trunk/Ohana/src/opihi/lib.shell/command.c (revision 3186) +++ /trunk/Ohana/src/opihi/lib.shell/command.c (revision 3187) @@ -32,5 +32,6 @@ /* this needs some work to use multicommand correctly */ - ALLOCATE (argv, char *, targc); + /* allocate extra space to have null terminated list */ + ALLOCATE (argv, char *, targc + 1); argc = 0; status = TRUE; @@ -43,4 +44,5 @@ if (argc == 0) continue; + argv[argc] = 0; cmd = MatchCommand (argv[0], TRUE, FALSE); Index: /trunk/Ohana/src/opihi/pclient/ChildOps.c =================================================================== --- /trunk/Ohana/src/opihi/pclient/ChildOps.c (revision 3187) +++ /trunk/Ohana/src/opihi/pclient/ChildOps.c (revision 3187) @@ -0,0 +1,112 @@ +# include "pclient.h" + +int InitChild () { + + ChildPID = 0; /** current child PID **/ + ChildStatus = PCLIENT_NONE; /** current status of child **/ + ChildExitStatus = 0; /** recent exit status of child **/ + + child_stdin_fd[0] = 0; + child_stdin_fd[1] = 0; + child_stdout_fd[0] = 0; + child_stdout_fd[1] = 0; + child_stderr_fd[0] = 0; + child_stderr_fd[1] = 0; + + InitIOBuffer (&child_stdout, 1024); + InitIOBuffer (&child_stderr, 1024); + return (TRUE); +} + +int CheckChild () { + + int Nread; + + CheckChildStatus (); + + /* read stdout buffer */ + Nread = ReadtoIOBuffer (&child_stdout, child_stdout_fd[0]); + switch (Nread) { + case -2: /* error in read (programming error? system level error?) */ + fprintf (stderr, "serious IO error\n"); + exit (2); + case -1: /* no data in pipe */ + break; + case 0: /* pipe is closed */ + /** change child state? **/ + break; + default: /* data in pipe */ + break; + } + + + /* read stderr buffer */ + Nread = ReadtoIOBuffer (&child_stderr, child_stderr_fd[0]); + switch (Nread) { + case -2: /* error in read (programming error? system level error?) */ + fprintf (stderr, "serious IO error\n"); + exit (2); + case -1: /* no data in pipe */ + break; + case 0: /* pipe is closed */ + /** change child state? **/ + break; + default: /* data in pipe */ + break; + } + +} + +void CheckChildStatus () { + + int result, waitstatus; + + if (ChildStatus != PCLIENT_BUSY) return; + + /* check current child status */ + result = waitpid (ChildPID, &waitstatus, WNOHANG); + switch (result) { + case -1: /* error with waitpid */ + switch (errno) { + case ECHILD: + fprintf (stderr, "unknown PID, not a child proc\n"); + fprintf (stderr, "did process already exit? programming error?\n"); + ChildStatus = PCLIENT_NONE; + break; + case EINVAL: + fprintf (stderr, "error EINVAL (waitpid): programming error\n"); + exit (1); + case EINTR: + fprintf (stderr, "error EINTR (waitpid): programming error\n"); + exit (1); + default: + fprintf (stderr, "unknown error for waitpid (%d): programming error\n", errno); + exit (1); + } + break; + + case 0: /* process not exited */ + ChildStatus = PCLIENT_BUSY; + break; + + default: + if (result != ChildPID) { + fprintf (stderr, "waitpid error: mis-matched PID (%d vs %d). programming error\n", result, ChildPID); + exit (1); + } + + if (WIFEXITED(waitstatus)) { + ChildStatus = PCLIENT_EXIT; + ChildExitStatus = WEXITSTATUS(waitstatus); + } + if (WIFSIGNALED(waitstatus)) { + ChildStatus = PCLIENT_CRASH; + ChildExitStatus = WTERMSIG(waitstatus); + } + if (WIFSTOPPED(waitstatus)) { + fprintf (stderr, "waitpid returns 'stopped': programming error\n"); + exit (1); + } + } + return; +} Index: /trunk/Ohana/src/opihi/pclient/IOBufferOps.c =================================================================== --- /trunk/Ohana/src/opihi/pclient/IOBufferOps.c (revision 3187) +++ /trunk/Ohana/src/opihi/pclient/IOBufferOps.c (revision 3187) @@ -0,0 +1,66 @@ +# include "pclient.h" + +int InitIOBuffer (IOBuffer *buffer, int Nalloc) { + + buffer[0].Nalloc = Nalloc; + buffer[0].Nreset = Nalloc; + buffer[0].Nblock = Nalloc / 2; + buffer[0].Nbuffer = 0; + + ALLOCATE (buffer[0].buffer, char, buffer[0].Nalloc); + + return (TRUE); +} + +int FlushIOBuffer (IOBuffer *buffer) { + + buffer[0].Nbuffer = 0; + buffer[0].Nalloc = buffer[0].Nreset; + REALLOCATE (buffer[0].buffer, char, buffer[0].Nalloc); + + return (TRUE); +} + +int ReadtoIOBuffer (IOBuffer *buffer, int fd) { + + int Nread, Nfree; + + if (fd == 0) { + /* pipe is closed */ + return (0); + } + + Nfree = buffer[0].Nalloc - buffer[0].Nbuffer; + if (Nfree < buffer[0].Nblock) { + buffer[0].Nalloc += 2*buffer[0].Nblock; + REALLOCATE (buffer[0].buffer, char, buffer[0].Nalloc); + Nfree = buffer[0].Nalloc - buffer[0].Nbuffer; + } + + Nread = read (fd, &buffer[0].buffer[buffer[0].Nbuffer], buffer[0].Nblock); + + if (Nread >= 0) { + buffer[0].Nbuffer += Nread; + return (Nread); + } + + if (Nread == -1) { + switch (errno) { + case EAGAIN: + case EIO: + /** no data available in pipe **/ + return (-1); + default: + /** error reading from pipe **/ + perror ("ReadtoIOBuffer read error"); + return (-2); + } + } +} + +void FreeIOBuffer (IOBuffer *buffer) { + + if (buffer[0].buffer != (char *) NULL) { + free (buffer[0].buffer); + } +} Index: /trunk/Ohana/src/opihi/pclient/check.c =================================================================== --- /trunk/Ohana/src/opihi/pclient/check.c (revision 3187) +++ /trunk/Ohana/src/opihi/pclient/check.c (revision 3187) @@ -0,0 +1,9 @@ +# include "pclient.h" + +int check (int argc, char **argv) { + + /* force a check */ + CheckChild (); + return (TRUE); + +} Index: /trunk/Ohana/src/opihi/pclient/init.c =================================================================== --- /trunk/Ohana/src/opihi/pclient/init.c (revision 3187) +++ /trunk/Ohana/src/opihi/pclient/init.c (revision 3187) @@ -0,0 +1,27 @@ +# include "opihi.h" + +int job PROTO((int, char **)); +int reset PROTO((int, char **)); +int check PROTO((int, char **)); +int status PROTO((int, char **)); +int stdout_pclient PROTO((int, char **)); +int stderr_pclient PROTO((int, char **)); + +static Command cmds[] = { + {"job", job, "start job"}, + {"reset", reset, "reset job"}, + {"check", check, "check job"}, + {"status", status, "check job status"}, + {"stdout", stdout_pclient, "get stdout buffer"}, + {"stderr", stderr_pclient, "get stderr buffer"}, +}; + +void InitPclient () { + + int i; + + for (i = 0; i < sizeof (cmds) / sizeof (Command); i++) { + AddCommand (&cmds[i]); + } + +} Index: /trunk/Ohana/src/opihi/pclient/job.c =================================================================== --- /trunk/Ohana/src/opihi/pclient/job.c (revision 3187) +++ /trunk/Ohana/src/opihi/pclient/job.c (revision 3187) @@ -0,0 +1,86 @@ +# include "pclient.h" + +int job (int argc, char **argv) { + + int pid, status; + + if (argc < 2) { + fprintf (stderr, "USAGE: job (arg0) (arg1) ... (argN)\n"); + fprintf (stdout, "STATUS %d\n", -2); + return (FALSE); + } + + if (ChildStatus != PCLIENT_NONE) { + fprintf (stderr, "need to clear existing child\n"); + fprintf (stdout, "STATUS %d\n", -3); + return (FALSE); + } + + if (pipe (child_stdin_fd) < 0) goto pipe_error; + if (pipe (child_stdout_fd) < 0) goto pipe_error; + if (pipe (child_stderr_fd) < 0) goto pipe_error; + + pid = fork (); + if (!pid) { /* must be child process */ + fprintf (stderr, "starting child process %s...\n", argv[1]); + + /* close the other ends of the pipes */ + close (child_stdin_fd[1]); + close (child_stdout_fd[0]); + close (child_stderr_fd[0]); + + /* tie our ends of the pipes to stdin, stdout, stderr */ + dup2 (child_stdin_fd[0], STDIN_FILENO); + dup2 (child_stdout_fd[1], STDOUT_FILENO); + dup2 (child_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); + fprintf (stderr, "child is spawned, executing command\n"); + /* note that this message comes out on the other side of the stderr pipe */ + + /* exec job */ + status = execvp (argv[1], &argv[1]); + fprintf (stderr, "error starting child process: %d\n", status); + exit (1); + } + + /* close the other ends of the pipes */ + close (child_stdin_fd[0]); + close (child_stdout_fd[1]); + close (child_stderr_fd[1]); + + /* make the pipes non-blocking */ + fcntl (child_stdin_fd[1], F_SETFL, O_NONBLOCK); + fcntl (child_stdout_fd[0], F_SETFL, O_NONBLOCK); + fcntl (child_stderr_fd[0], F_SETFL, O_NONBLOCK); + + ChildStatus = PCLIENT_BUSY; + ChildPID = pid; + + fprintf (stdout, "STATUS %d\n", ChildPID); + return (TRUE); + +pipe_error: + perror ("pipe error:"); + if (child_stdin_fd[0] != 0) close (child_stdin_fd[0]); + if (child_stdin_fd[1] != 0) close (child_stdin_fd[1]); + if (child_stdout_fd[0] != 0) close (child_stdout_fd[0]); + if (child_stdout_fd[1] != 0) close (child_stdout_fd[1]); + if (child_stderr_fd[0] != 0) close (child_stderr_fd[0]); + if (child_stderr_fd[1] != 0) close (child_stderr_fd[1]); + + fprintf (stdout, "STATUS %d\n", -1); + return (FALSE); +} + +/* possible responses: + +STATUS -1 - pipe error +STATUS -2 - syntax error +STATUS -3 - existing child +STATUS >0 - success (PID) + +*/ Index: /trunk/Ohana/src/opihi/pclient/pclient.c =================================================================== --- /trunk/Ohana/src/opihi/pclient/pclient.c (revision 3187) +++ /trunk/Ohana/src/opihi/pclient/pclient.c (revision 3187) @@ -0,0 +1,61 @@ +# include "pclient.h" + +# define opihi_name "PCLIENT" +# define opihi_prompt "pclient: " +# define opihi_description "pcontrol client shell\n" +# define opihi_history ".pclient" +# define opihi_rcfile ".pcontrolrc" + +void InitPclient (); +void InitBasic (); + +void welcome () { + fprintf (stderr, "\n"); + fprintf (stderr, "Welcome to %s - %s\n\n", opihi_name, opihi_description); +} + +/* program-dependent initialization */ +void initialize (int argc, char **argv) { + + FILE *f; + + auto_break = TRUE; + + Nlists = 0; + ALLOCATE (lists, List, 1); + + /* init functions required by libraries */ + /* -libshell */ + InitCommands (); + InitMacros (); + InitBuffers (); + InitVectors (); + InitVariables (); + + /* -libdata */ + InitGraph (); + InitImage (); + + /* load the commands used by this implementation */ + InitBasic (); + InitOutfile (); + InitPclient (); + InitChild (); + + rl_readline_name = opihi_name; + rl_attempted_completion_function = command_completer; + rl_event_hook = CheckChild; + + set_str_variable ("HISTORY", opihi_history); + set_str_variable ("PROMPT", opihi_prompt); + set_str_variable ("RCFILE", opihi_rcfile); + + /* ignore the history file. to change this, see, eg, mana.c */ + signal (SIGINT, SIG_IGN); + return; +} + +/* add program-dependent exit functions here */ +void cleanup () { + return; +} Index: /trunk/Ohana/src/opihi/pclient/reset.c =================================================================== --- /trunk/Ohana/src/opihi/pclient/reset.c (revision 3187) +++ /trunk/Ohana/src/opihi/pclient/reset.c (revision 3187) @@ -0,0 +1,61 @@ +# include "pclient.h" + +int reset (int argc, char **argv) { + + int i, result, waitstatus; + + if (argc != 1) { + fprintf (stderr, "USAGE: reset\n"); + fprintf (stdout, "STATUS %d\n", -2); + return (FALSE); + } + + if (ChildStatus == PCLIENT_NONE) { + fprintf (stderr, "no child process, cannot reset\n"); + fprintf (stdout, "STATUS 2\n"); + return (TRUE); + } + + if (ChildStatus == PCLIENT_BUSY) { + /* send SIGTERM signal to job */ + kill (ChildPID, SIGTERM); + result = 0; + for (i = 0; (i < 10) && (result == 0); i++) { + usleep (10000); /* 10 ms is min */ + result = waitpid (ChildPID, &waitstatus, WNOHANG); + } + if (result) goto reset_job; + + /* send SIGTERM signal to job */ + kill (ChildPID, SIGKILL); + result = 0; + for (i = 0; (i < 10) && (result == 0); i++) { + usleep (10000); /* 10 ms is min */ + result = waitpid (ChildPID, &waitstatus, WNOHANG); + } + if (result) goto reset_job; + + /* total failure, don't reset */ + fprintf (stderr, "child process %d is hung, cannot reset\n"); + fprintf (stdout, "STATUS 0\n"); + return (FALSE); + } + +reset_job: + /* close out ends of the pipes */ + close (child_stdin_fd[1]); + close (child_stdout_fd[0]); + close (child_stderr_fd[0]); + child_stdin_fd[1] = 0; + child_stdout_fd[0] = 0; + child_stderr_fd[0] = 0; + + FlushIOBuffer (&child_stdout); + FlushIOBuffer (&child_stderr); + ChildStatus = PCLIENT_NONE; + ChildPID = 0; + ChildExitStatus = 0; + + fprintf (stdout, "STATUS 1\n"); + return (TRUE); +} Index: /trunk/Ohana/src/opihi/pclient/status.c =================================================================== --- /trunk/Ohana/src/opihi/pclient/status.c (revision 3187) +++ /trunk/Ohana/src/opihi/pclient/status.c (revision 3187) @@ -0,0 +1,22 @@ +# include "pclient.h" + +int status (int argc, char **argv) { + + char status_string[64]; + + if (argc != 1) { + fprintf (stderr, "USAGE: status\n"); + return (FALSE); + } + + if (ChildStatus == PCLIENT_NONE) strcpy (status_string, "NONE"); + if (ChildStatus == PCLIENT_BUSY) strcpy (status_string, "BUSY"); + if (ChildStatus == PCLIENT_EXIT) strcpy (status_string, "EXIT"); + if (ChildStatus == PCLIENT_CRASH) strcpy (status_string, "CRASH"); + + fprintf (stdout, "STATUS %s\n", status_string); + fprintf (stdout, "EXITST %d\n", ChildExitStatus); + fprintf (stdout, "STDOUT %d\n", child_stdout.Nbuffer); + fprintf (stdout, "STDERR %d\n", child_stderr.Nbuffer); + return (TRUE); +} Index: /trunk/Ohana/src/opihi/pclient/stdout.c =================================================================== --- /trunk/Ohana/src/opihi/pclient/stdout.c (revision 3187) +++ /trunk/Ohana/src/opihi/pclient/stdout.c (revision 3187) @@ -0,0 +1,31 @@ +# include "pclient.h" + +int stdout_pclient (int argc, char **argv) { + + if (argc != 1) { + fprintf (stderr, "USAGE: stdout\n"); + fprintf (stdout, "STATUS %d\n", -1); + return (FALSE); + } + + fprintf (stdout, "NBYTES %d\n", child_stdout.Nbuffer); + fwrite (child_stdout.buffer, 1, child_stdout.Nbuffer, stdout); + fprintf (stdout, "STATUS %d\n", 0); + return (TRUE); + +} + +int stderr_pclient (int argc, char **argv) { + + if (argc != 1) { + fprintf (stderr, "USAGE: stderr\n"); + fprintf (stdout, "STATUS %d\n", -1); + return (FALSE); + } + + fprintf (stdout, "NBYTES %d\n", child_stderr.Nbuffer); + fwrite (child_stderr.buffer, 1, child_stderr.Nbuffer, stdout); + fprintf (stdout, "STATUS %d\n", 0); + return (TRUE); + +} Index: /trunk/Ohana/src/opihi/pcontrol/CheckHost.c =================================================================== --- /trunk/Ohana/src/opihi/pcontrol/CheckHost.c (revision 3187) +++ /trunk/Ohana/src/opihi/pcontrol/CheckHost.c (revision 3187) @@ -0,0 +1,48 @@ +# include "pcontrol.h" + +int CheckHost (Host *host) { + + int i, status; + IOBuffer buffer; + Job *job; + + InitIOBuffer (&buffer, 0x100); + + status = PclientCommand (host, "echo OK", &buffer); + switch (status) { + case 0: + fprintf (stderr, "host %s is down\n", host[0].hostname); + + /* if host has a job, job is dead, push to Pending */ + if (host[0].status == PCONTROL_HOST_BUSY) { + job = host[0].job; + N = FindJob (job[0].JobID, PCONTROL_JOB_BUSY); + if (N < 0) { + fprintf (stderr, "error: job is not found in BUSY list\n"); + exit (2); + } + job[0].host = NULL; /* unlink host & job */ + job = GetJob (PCONTROL_JOB_BUSY, N); + PutJob (job, PCONTROL_JOB_PENDING, QUEUE_BOTTOM); + } + host[0].job = NULL; + PutHost (host, PCONTROL_HOST_DOWN, QUEUE_BOTTOM); + FreeBuffer (&buffer); + return (FALSE); + + case -1: + fprintf (stderr, "host %s is not responding\n", host[0].hostname); + /*** do we mark this in some way (HUNG) ? ***/ + PutHost (host, host[0].status, QUEUE_BOTTOM); + FreeBuffer (&buffer); + return (FALSE); + + default: + fprintf (stderr, "host %s is alive\n", host[0].hostname); + PutHost (host, host[0].status, QUEUE_BOTTOM); + FreeBuffer (&buffer); + return (TRUE); + } + fprintf (stderr, "programming error (Check Host)\n"); + return (FALSE); +} Index: /trunk/Ohana/src/opihi/pcontrol/CheckJob.c =================================================================== --- /trunk/Ohana/src/opihi/pcontrol/CheckJob.c (revision 3187) +++ /trunk/Ohana/src/opihi/pcontrol/CheckJob.c (revision 3187) @@ -0,0 +1,82 @@ +# include "pcontrol.h" +# define CONNECT_TIMEOUT 20 + +int CheckJob (Job *job) { + + int i, status; + IOBuffer buffer; + Host *host; + + host = job[0].host; + + InitIOBuffer (&buffer, 0x100); + + status = PclientCommand (host, "status\n", &buffer); + + /* check on success of pclient command */ + switch (status) { + case DOWN: + fprintf (stderr, "host %s is down\n", host[0].hostname); + goto escape; + /*** different behavior for ANYHOST, WANTHOST, NEEDHOST ***/ + + case HUNG: + fprintf (stderr, "host %s is not responding\n", host[0].hostname); + goto escape; + /*** should we consider a HUNG host DOWN? ***/ + + case TRUE: + fprintf (stderr, "message received\n"); + break; + + default: + fprintf (stderr, "unknown status for pclient command: programming error\n"); + exit (1); + } + + /** host is up, need to parse message **/ + p = memstr (buffer.buffer, "STATUS", buffer.Nbuffer); + if (p == NULL) { + fprintf (stderr, "missing STATUS in pclient message : programming error\n"); + exit (1); + } + fscan (p, "%*s %s", string); + if (!strcmp(string, "NONE")) { + fprintf (stderr, "no current job : programming error\n"); + exit (2); + } + if (!strcmp(string, "BUSY")) { + /** no status change, return to BUSY queue **/ + PutJob (job, PCONTROL_JOB_BUSY, QUEUE_BOTTOM); + return (TRUE); + } + if (!strcmp(string, "EXIT")) { + job[0].status = PCONTROL_JOB_EXIT; + p = memstr (buffer.buffer, "EXITST", buffer.Nbuffer); + fscan (p, "%*s %d", &job[0].exitstatus); + return (TRUE); + } + if (!strcmp(string, "CRASH")) { + job[0].status = PCONTROL_JOB_CRASH; + p = memstr (buffer.buffer, "EXITST", buffer.Nbuffer); + fscan (p, "%*s %d", &job[0].exitstatus); + return (TRUE); + } + +escape: + job[0].host = NULL; /* unlink host & job */ + host[0].job = NULL; + + /*** need to pop host off of correct queue XXX ***/ + N = FindHost (host[0].ID, PCONTROL_HOST_BUSY); + if (N < 0) { + fprintf (stderr, "programming error: host is not found in BUSY list\n"); + exit (2); + } + host = GetHost (PCONTROL_HOST_BUSY, N); + PutHost (host, PCONTROL_HOST_DOWN, QUEUE_BOTTOM); + + PutJob (job, PCONTROL_JOB_PENDING, QUEUE_BOTTOM); + FreeBuffer (&buffer); + return (FALSE); +} Index: /trunk/Ohana/src/opihi/pcontrol/CheckSystem.c =================================================================== --- /trunk/Ohana/src/opihi/pcontrol/CheckSystem.c (revision 3187) +++ /trunk/Ohana/src/opihi/pcontrol/CheckSystem.c (revision 3187) @@ -0,0 +1,75 @@ +# include "pcontrol.h" + +int CheckSystem () { + + CheckBusyJobs(); + CheckLiveHosts(); + CheckDownHosts(); + CheckIdleHosts(); + +} + +CheckBusyJobs () { + + queue = GetJobQueue (PCONTROL_JOB_BUSY); + Nobjects = queue[0].Nobjects; + + for (i = 0; i < Nobjects; i++) { + job = GetObject (queue, QUEUE_TOP); + CheckJob (job); + } +} + +CheckLiveHosts () { + + queue = GetHostQueue (PCONTROL_HOST_IDLE); + Nobjects = queue[0].Nobjects; + + for (i = 0; i < Nobjects; i++) { + host = GetObject (queue, QUEUE_TOP); + CheckHost (host); + } + + queue = GetHostQueue (PCONTROL_HOST_BUSY); + Nobjects = queue[0].Nobjects; + + for (i = 0; i < Nobjects; i++) { + host = GetObject (queue, QUEUE_TOP); + CheckHost (host); + } +} + +CheckDownHosts () { + + queue = GetHostQueue (PCONTROL_HOST_DOWN); + Nobjects = queue[0].Nobjects; + + for (i = 0; i < Nobjects; i++) { + host = GetObject (queue, QUEUE_TOP); + StartHost (host); + } +} + +CheckIdleHosts () { + + hqueue = HostPool_Idle; + jqueue = JobPool_Pending; + + for (i = 0; i < hqueue[0].Nobjects; i++) { + host = (Host *) hqueue[0].object[i]; + /** need to add a timeout here **/ + for (j = 0; j < jqueue[0].Nobjects; j++) { + job = (Job *) jqueue[0].object[j]; + if ((job[0].mode == PCONTROL_JOB_ANYHOST) || + (!strcasecmp (job[0].hostname, host[0].hostname)) + + StartHost (host); + } +} + +/*** this section is tricky: need to have a plan on making WANTHOST likely + but not required (on average, every host is currently in use. some possible tricks: + + - search through PENDING for N entries looking for WANTHOST / matching host + - keep job through N tries waiting for WANTHOST +***/ Index: /trunk/Ohana/src/opihi/pcontrol/HostOps.c =================================================================== --- /trunk/Ohana/src/opihi/pcontrol/HostOps.c (revision 3187) +++ /trunk/Ohana/src/opihi/pcontrol/HostOps.c (revision 3187) @@ -0,0 +1,96 @@ +# include "pcontrol.h" + +Queue *HostPool_Idle; +Queue *HostPool_Busy; +Queue *HostPool_Down; +Queue *HostPool_Off; + +Queue *GetHostQueue (int QueueID) { + switch (QueueID) { + case PCONTROL_HOST_IDLE: + return (HostPool_Idle); + case PCONTROL_HOST_DOWN: + return (HostPool_Down); + case PCONTROL_HOST_BUSY: + return (HostPool_Busy); + case PCONTROL_HOST_OFF: + return (HostPool_Off); + default: + fprintf (stderr, "error: unknown host queue\n"); + return (NULL); + } + return (NULL); +} + +int PutHost (Host *host, int QueueID, int where) { + + int status; + Queue *queue; + + queue = GetHostQueue (QueueID); + if (queue == NULL) return (FALSE); + + host[0].status = QueueID; + status = PutObject (queue, where, host); + return (status); +} + +Host *GetHost (int QueueID, int where) { + + Host *host; + Queue *queue; + + queue = GetHostQueue (QueueID); + if (queue == NULL) return (NULL); + + host = GetObject (queue, where); + return (host); +} + +int FindHost (char *name, int QueueID) { + + Queue *queue; + + queue = GetHostQueue (QueueID); + if (queue == NULL) return (-2); + + for (i = 0; i < queue[0].Nobject; i++) { + host = (Host *) queue[0].object; + if (!strcasecmp (host[0].hostname, host)) { + return (i); + } + } + return (-1); +} + +AddHost (char *hostname) { + + Host *host; + + ALLOCATE (host, Host, 1); + + host[0].hostname = strcreate (hostname); + host[0].stdin = 0; + host[0].stdout = 0; + host[0].stderr = 0; + host[0].ID = NextHostID(); + + host[0].markoff = FALSE; + host[0].job = NULL; + DownHost (host); +} + +DelHost (Host *host) { + FREE (host[0].hostname); + FREE (host[0].job); + FREE (host); +} + +MarkOffHost (Host *host) { + if (host[0].status != PCONTROL_HOST_BUSY) { + return (FALSE); + } + host[0].markoff = TRUE; + PutObject (HostPool_Busy, host); /**** where are we removing hosts? **/ +} + Index: /trunk/Ohana/src/opihi/pcontrol/JobID.c =================================================================== --- /trunk/Ohana/src/opihi/pcontrol/JobID.c (revision 3187) +++ /trunk/Ohana/src/opihi/pcontrol/JobID.c (revision 3187) @@ -0,0 +1,41 @@ +# include "pcontrol.h" + +typedef unsigned long long IDtype; + +static IDtype CurrentJobID = 0; +static IDtype CurrentHostID = 0; + +/* for now, no persistence : we could use the date/time to seed the upper byte(s) if needed */ +InitIDs () { + CurrentJobID = 0; + CurrentHostID = 0; +} + +IDtype NextJobID () { + + IDtype ID; + + ID = CurrentJobID; + CurrentJobID ++; + return (ID); +} + +IDtype NextHostID () { + + IDtype ID; + + ID = CurrentHostID; + CurrentHostID ++; + return (ID); +} + +void PrintID (FILE *f, IDtype ID) { + + word0 = 0xffff & ID; + word1 = 0xffff & (ID >> 16); + word2 = 0xffff & (ID >> 32); + word3 = 0xffff & (ID >> 48); + + fprintf (f, "%x.%x.%x.%x", word3, word2, word1, word0); +} + Index: /trunk/Ohana/src/opihi/pcontrol/JobOps.c =================================================================== --- /trunk/Ohana/src/opihi/pcontrol/JobOps.c (revision 3187) +++ /trunk/Ohana/src/opihi/pcontrol/JobOps.c (revision 3187) @@ -0,0 +1,118 @@ +# include "pcontrol.h" + +Queue *JobPool_Pending; +Queue *JobPool_Busy; +Queue *JobPool_Exit; +Queue *JobPool_Crash; + +Queue *GetJobQueue (int QueueID) { + switch (QueueID) { + case PCONTROL_JOB_PENDING: + return (JobPool_Pending); + case PCONTROL_JOB_BUSY: + return (JobPool_Busy); + case PCONTROL_JOB_EXIT: + return (JobPool_Exit); + case PCONTROL_JOB_Crash: + return (JobPool_Crash); + default: + fprintf (stderr, "error: unknown job queue\n"); + return (NULL); + } + return (NULL); +} + +int PutJob (Job *job, int QueueID, int where) { + + int status; + Queue *queue; + + queue = GetJobQueue (QueueID); + if (queue == NULL) return (FALSE); + + job[0].status = QueueID; + status = PutObject (queue, where, job); + return (status); +} + +Job *GetJob (int QueueID, int where) { + + Job *job; + Queue *queue; + + queue = GetJobQueue (QueueID); + if (queue == NULL) return (NULL); + + job = GetObject (queue, where); + return (job); +} + +int FindJob (int JobID, int QueueID) { + + Job *job; + Queue *queue; + + queue = GetJobQueue (QueueID); + if (queue == NULL) return (-2); + + for (i = 0; i < queue[0].Nobject; i++) { + job = (Job *) queue[0].object; + if (job[0].JobID == JobID) { + return (i); + } + } + return (-1); +} + +AddJob (char *hostname, JobMode mode, int timeout, int argc, char **argv) { + + Job *job; + + ALLOCATE (job, Job, 1); + + job[0].argc = argc; + job[0].argv = argv; + job[0].hostname = strcreate (hostname); + job[0].mode = mode; + job[0].host = NULL; + job[0].ID = NextJobID(); + + InitIOBuffer (&job[0].stdin, 0x1000); + InitIOBuffer (&job[0].stdout, 0x1000); + InitIOBuffer (&job[0].stderr, 0x1000); + + PutJob (job, PCONTROL_JOB_PENDING, QUEUE_BOTTOM); +} + +DelJob (Job *job) { + + FREE (job[0].hostname); + for (i = 0; i < job[0].argc; i++) { + FREE (job[0].argv[i]); + } + FREE (job[0].argv); + + FreeIOBuffer (&job[0].stdin); + FreeIOBuffer (&job[0].stdout); + FreeIOBuffer (&job[0].stderr); + + FREE (job); +} + +KillJob (Job *job) { + + char line[64]; + + sprintf (line, "reset\n"); + + /* send command to client */ + status = write (host[0].stdin, line, strlen(line)); + if ((status == -1) && (errno == EPIPE)) { + fprintf (stderr, "host %s is down\n", host[0].hostname); + PutHost (host, PCONTROL_HOST_DOWN, QUEUE_BOTTOM); + PutJob (job, PCONTROL_JOB_PENDING, QUEUE_BOTTOM); + return (FALSE); + } + /* check for a response */ + +} Index: /trunk/Ohana/src/opihi/pcontrol/QueueOps.c =================================================================== --- /trunk/Ohana/src/opihi/pcontrol/QueueOps.c (revision 3187) +++ /trunk/Ohana/src/opihi/pcontrol/QueueOps.c (revision 3187) @@ -0,0 +1,63 @@ +# include "pcontrol.h" + +/* get object from point in stack (negative == distance from end) */ +void *GetQueue (Queue *queue, int where) { + + int i; + void *object; + + /* QUEUE_TOP == 0, QUEUE_BOTTOM == -1 */ + /* this code correctly handles the negative 'where' and Nobject == 0 */ + if (where < 0) where += queue[0].Nobject; + if (where < 0) return (NULL); + if (where >= queue[0].Nobject) return (NULL); + + object = queue[0].object[where]; + queue[0].Nobject --; + for (i = where; i < queue[0].Nobject; i++) { + queue[0].object[i] = queue[0].object[i+1]; + } + return (object); +} + +/* push object on top of stack */ +int PutQueue (Queue *queue, int where, void *object) { + + int i; + + /* QUEUE_TOP == 0, QUEUE_BOTTOM == -1 */ + /* this code correctly handles the negative 'where' and Nobject == 0 */ + if (where < 0) where += queue[0].Nobject + 1; + if (where < 0) return (FALSE); + if (where > queue[0].Nobject) return (FALSE); + + /* extend queue as needed */ + if (queue[0].Nobject >= queue[0].NOBJECT) { + queue[0].NOBJECT += 100; + REALLOCATE (queue[0].object, void *, queue[0].NOBJECT); + } + + for (i = queue[0].Nobject; i > where; i--) { + queue[0].object[i] = queue[0].object[i-1]; + } + queue[0].object[where] = object; + queue[0].Nobject ++; + return (TRUE); +} + +/* allocate queue, setup with default values, allocate data */ +Queue *InitQueue () { + + Queue *queue; + + ALLOCATE (queue, Queue, 1); + + queue[0].Nobject = 0; + queue[0].NOBJECT = 50; + ALLOCATE (queue[0].object, void *, queue[0].NOBJECT); + return (queue); +} + +/* these queues are not super efficient, and should probably be replaced with linked lists, + but I find these easier to get my brain around +*/ Index: /trunk/Ohana/src/opihi/pcontrol/StartHost.c =================================================================== --- /trunk/Ohana/src/opihi/pcontrol/StartHost.c (revision 3187) +++ /trunk/Ohana/src/opihi/pcontrol/StartHost.c (revision 3187) @@ -0,0 +1,19 @@ +# include "pcontrol.h" + +int StartHost (Host *host) { + + int stdio[3]; + + pid = rconnect (CONNECT, host[0].hostname, PCLIENT, stdio); + if (!pid) { /** failure to start **/ + fprintf (stderr, "failure to start %s\n", host[0].hostname); + PutHost (host, PCONTROL_HOST_DOWN, QUEUE_BOTTOM); + return (FALSE); + } + host[0].stdin = stdio[0]; + host[0].stdout = stdio[1]; + host[0].stderr = stdio[2]; + host[0].pid = pid; + PutHost (host, PCONTROL_HOST_IDLE, QUEUE_BOTTOM); + return (TRUE); +} Index: /trunk/Ohana/src/opihi/pcontrol/StartJob.c =================================================================== --- /trunk/Ohana/src/opihi/pcontrol/StartJob.c (revision 3187) +++ /trunk/Ohana/src/opihi/pcontrol/StartJob.c (revision 3187) @@ -0,0 +1,105 @@ +# include "pcontrol.h" + +int StartJob (Job *job) { + + int Nline; + char *line; + IOBuffer buffer; + + InitIOBuffer (&buffer); + + /* job must have assigned host */ + host = job[0].host; + if (host == NULL) { + fprintf (stderr, "no assigned host : programming error\n"); + exit (1); + } + + /* construct command line : job arg0 arg1 ... argN\n */ + Nline = 10 + argc; + for (i = 0; i < argc; i++) { + Nline += strlen (job[0].argv[i]); + } + ALLOCATE (line, char, Nline); + bzero (line, Nline); + strcpy (line, "job"); + for (i = 0; i < argc; i++) { + strcat (line, " "); + strcat (line, job[0].argv[i]); + } + strcat (line, "\n"); + + status = PclientCommand (host, line, &buffer); + free (line); + + /* check on success of pclient command */ + switch (status) { + case DOWN: + fprintf (stderr, "host %s is down\n", host[0].hostname); + goto escape; + /*** different behavior for ANYHOST, WANTHOST, NEEDHOST ***/ + + case HUNG: + fprintf (stderr, "host %s is not responding\n", host[0].hostname); + goto escape; + /*** should we consider a HUNG host DOWN? ***/ + + case TRUE: + fprintf (stderr, "message received\n"); + break; + + default: + fprintf (stderr, "unknown status for pclient command: programming error\n"); + exit (1); + } + + /* check on result of pclient command */ + p = memstr (buffer.buffer, "STATUS", buffer.Nbuffer); + if (p == NULL) { + fprintf (stderr, "missing STATUS in pclient message : programming error\n"); + exit (1); + } + fscan (p, "%*s %d", &status); + switch (status) { + case -1: + fprintf (stderr, "error in pclient child\n"); + goto escape; + /*** should we consider this host DOWN? ***/ + + case -2: + fprintf (stderr, "syntax error in pclient command : programming error\n"); + exit (1); + + case -3: + fprintf (stderr, "existing child on pclient : programming error\n"); + exit (1); + + default: + job[0].pid = status; + PutHost (host, PCONTROL_HOST_BUSY, QUEUE_BOTTOM); + PutJob (job, PCONTROL_JOB_BUSY, QUEUE_BOTTOM); + return (TRUE); + } + /* we should never reach here */ + fprintf (stderr, "pcontrol programming error\n"); + exit (1); + +escape: + job[0].host = NULL; /* unlink host & job */ + host[0].job = NULL; + /*** need to pop host from corresponding queue (BUSY) ***/ + PutHost (host, PCONTROL_HOST_DOWN, QUEUE_BOTTOM); + PutJob (job, PCONTROL_JOB_PENDING, QUEUE_BOTTOM); + FreeBuffer (&buffer); + return (FALSE); +} + +/* + this function sends a new command to the pclient. + before calling this function: + - grab the job from its queue (remove it from that queue with GetQueue) + - identify the target host (MUST already be IDLE) + - grab the host from its queue (remove it from that queue with GetQueue) + - link job and host (job[0].host = host, host[0].job = job) + - job must also have a valid command (argc, argv) +*/ Index: /trunk/Ohana/src/opihi/pcontrol/host.c =================================================================== --- /trunk/Ohana/src/opihi/pcontrol/host.c (revision 3187) +++ /trunk/Ohana/src/opihi/pcontrol/host.c (revision 3187) @@ -0,0 +1,72 @@ +# include "pcontrol.h" + +int host (int argc, char **argv) { + + int pid, status; + + /* this section needs some help: find the specified host in the queues */ + Delete = FALSE; + if ((N = get_argument (argc, argv, "-delete"))) { + remove_argument (N, &argc, argv); + Delete = TRUE; + } + + /* this section needs some help: find the specified host in the queues */ + Off = FALSE; + if ((N = get_argument (argc, argv, "-off"))) { + if (Delete) { + fprintf (stderr, "-delete and -off incompatible\n"); + return (FALSE); + } + remove_argument (N, &argc, argv); + Off = TRUE; + } + + if (argc < 2) { + fprintf (stderr, "USAGE: host (hostname) [-delete]\n"); + return (FALSE); + } + + if (Delete) { + host = FindHost (argv[1], PCONTROL_HOST_OFF); + if (host == NULL) { + fprintf (stderr, "host %s is not OFF\n", argv[1]); + return (FALSE); + } + DelHost (host); + return (TRUE); + } + + if (On) { + host = FindHost (argv[1], PCONTROL_HOST_OFF); + if (host == NULL) { + fprintf (stderr, "host %s is not OFF\n", argv[1]); + return (FALSE); + } + DownHost (host); + return (TRUE); + } + + if (Off) { + host = FindHost (argv[1], PCONTROL_HOST_IDLE); + if (host != NULL) { + OffHost (host); + return (TRUE); + } + host = FindHost (argv[1], PCONTROL_HOST_DOWN); + if (host != NULL) { + OffHost (host); + return (TRUE); + } + host = FindHost (argv[1], PCONTROL_HOST_BUSY); + if (host != NULL) { + MarkOffHost (host); + return (TRUE); + } + fprintf (stderr, "host %s is not BUSY, IDLE, or DOWN\n", argv[1]); + return (FALSE); + } + + AddHost (argv[1]); + return (TRUE); +} Index: /trunk/Ohana/src/opihi/pcontrol/job.c =================================================================== --- /trunk/Ohana/src/opihi/pcontrol/job.c (revision 3187) +++ /trunk/Ohana/src/opihi/pcontrol/job.c (revision 3187) @@ -0,0 +1,51 @@ +# include "pcontrol.h" + +int job (int argc, char **argv) { + + int pid, status; + + 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) { + fprintf (stderr, "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; + } + + 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) { + fprintf (stderr, "USAGE: job [options] (arg0) (arg1) ... (argN)\n"); + 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); + FREE (Host); + + fprintf (stdout, "STATUS %d\n", ChildPID); + return (TRUE); + +} Index: /trunk/Ohana/src/opihi/pcontrol/kill.c =================================================================== --- /trunk/Ohana/src/opihi/pcontrol/kill.c (revision 3187) +++ /trunk/Ohana/src/opihi/pcontrol/kill.c (revision 3187) @@ -0,0 +1,53 @@ +# include "pcontrol.h" + +int kill (int argc, char **argv) { + + Job *job; + int JobID; + + if (argc < 2) { + fprintf (stderr, "USAGE: kill (JobID)\n"); + return (FALSE); + } + JobID = atoi (argv[1]); + + job = FindJob (JobID, PCONTROL_JOB_BUSY); + if (job != NULL) { + KillJob (job); + return (TRUE); + } + + fprintf (stderr, "job %s not BUSY\n", argv[1]); + return (FALSE); +} + +int delete (int argc, char **argv) { + + Job *job; + int JobID; + + if (argc < 2) { + fprintf (stderr, "USAGE: delete (JobID)\n"); + return (FALSE); + } + JobID = atoi (argv[1]); + + job = FindJob (JobID, PCONTROL_JOB_PENDING); + if (job != NULL) { + DelJob (job); + return (TRUE); + } + job = FindJob (JobID, PCONTROL_JOB_CRASH); + if (job != NULL) { + DelJob (job); + return (TRUE); + } + job = FindJob (JobID, PCONTROL_JOB_EXIT); + if (job != NULL) { + DelJob (job); + return (TRUE); + } + + fprintf (stderr, "job %s not PENDING, CRASH, EXIT\n", argv[1]); + return (FALSE); +} Index: /trunk/Ohana/src/opihi/pcontrol/pclient.c =================================================================== --- /trunk/Ohana/src/opihi/pcontrol/pclient.c (revision 3187) +++ /trunk/Ohana/src/opihi/pcontrol/pclient.c (revision 3187) @@ -0,0 +1,25 @@ +# include "pcontrol.h" + +# define PCLIENT_TIMEOUT 20 +# define PCLIENT_PROMPT "pclient:" + +# define DOWN 0 +# define HUNG -1 + +int PclientCommand (Host *host, char *command, IOBuffer *buffer) { + + /* send command to client */ + status = write (host[0].stdin, command, strlen(command)); + if ((status == -1) && (errno == EPIPE)) return (DOWN); + + /* watch for response - wait up to 1 second */ + p = NULL; + status = -1; + for (i = 0; (i < PCLIENT_TIMEOUT) && (status == -1) && (p == NULL); i++) { + status = ReadtoIOBuffer (buffer, host[0].stdout); + p = memstr (buffer[0].buffer, PCLIENT_PROMPT, buffer[0].Nbuffer); + } + if (status == 0) return (DOWN); + if (status == -1) return (HUNG); + return (TRUE); +} Index: /trunk/Ohana/src/opihi/pcontrol/pcontrol.c =================================================================== --- /trunk/Ohana/src/opihi/pcontrol/pcontrol.c (revision 3187) +++ /trunk/Ohana/src/opihi/pcontrol/pcontrol.c (revision 3187) @@ -0,0 +1,61 @@ +# include "pcontrol.h" + +# define opihi_name "PCONTROL" +# define opihi_prompt "pcontrol: " +# define opihi_description "pcontrol client shell\n" +# define opihi_history ".pcontrol" +# define opihi_rcfile ".pcontrolrc" + +void InitPcontrol (); +void InitBasic (); + +void welcome () { + fprintf (stderr, "\n"); + fprintf (stderr, "Welcome to %s - %s\n\n", opihi_name, opihi_description); +} + +/* program-dependent initialization */ +void initialize (int argc, char **argv) { + + FILE *f; + + auto_break = TRUE; + + Nlists = 0; + ALLOCATE (lists, List, 1); + + /* init functions required by libraries */ + /* -libshell */ + InitCommands (); + InitMacros (); + InitBuffers (); + InitVectors (); + InitVariables (); + + /* -libdata */ + InitGraph (); + InitImage (); + + /* load the commands used by this implementation */ + InitBasic (); + InitOutfile (); + InitPcontrol (); + InitChild (); + + rl_readline_name = opihi_name; + rl_attempted_completion_function = command_completer; + rl_event_hook = CheckSystem; + + set_str_variable ("HISTORY", opihi_history); + set_str_variable ("PROMPT", opihi_prompt); + set_str_variable ("RCFILE", opihi_rcfile); + + /* ignore the history file. to change this, see, eg, mana.c */ + signal (SIGINT, SIG_IGN); + return; +} + +/* add program-dependent exit functions here */ +void cleanup () { + return; +} Index: /trunk/Ohana/src/opihi/pcontrol/rconnect.c =================================================================== --- /trunk/Ohana/src/opihi/pcontrol/rconnect.c (revision 3187) +++ /trunk/Ohana/src/opihi/pcontrol/rconnect.c (revision 3187) @@ -0,0 +1,109 @@ +# include "pcontrol.h" +# define CONNECT_TIMEOUT 300 + +/* connect to host, start the shell: ssh hostname pclient -> command hostname shell + stdio is an array of file descriptors (stdio[3]) +*/ + +int rconnect (char *command, char *hostname, char *shell, int *stdio) { + + int i, stdin_fd[2], stdout_fd[2], stderr_fd[2], status; + pid_t pid; + char command[0x100]; + char *p; + IOBuffer buffer; + + 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 *, 4); + argv[0] = command; + argv[1] = hostname; + argv[2] = shell; + argv[3] = 0; + + pid = fork (); + if (!pid) { /* must be child process */ + fprintf (stderr, "starting remote connection to %s...", hostname); + + /* 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); + fprintf (stderr, "error starting remote shell process\n"); + 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 pclient to verify alive & running */ + /** this handshake is similar to PclientCommand, but has important differences **/ + InitIOBuffer (&buffer, 0x100); + + /* send handshake command */ + sprintf (command, "echo CONNECTED\n"); + status = write (stdin_fd[1], command, strlen(command)); + if ((status == -1) && (errno == EPIPE)) goto pipe_error; + + /* try to get evidence connection is alive - wait upto a few seconds */ + p = NULL; + status = -1; + for (i = 0; (i < CONNECT_TIMEOUT) && (status == -1) && (p == NULL); i++) { + status = ReadtoIOBuffer (buffer, stdout_fd[0]); + p = memstr (buffer[0].buffer, "CONNECTED", buffer[0].Nbuffer); + } + if (status == 0) goto pipe_error; + if (status == -1) goto io_error; + FreeIOBuffer (&buffer); + + fprintf (stderr, "Connected\n"); + + stdio[0] = stdin_fd[1]; + stdio[1] = stdout_fd[0]; + stdio[2] = stderr_fd[0]; + + return (pid); + +pipe_error: + perror ("pipe error:"); + goto close_pipes; + +io_error: + fprintf (stderr, "timeout while connecting\n"); + goto close_pipes; + +close_pipes: + if (stdin_fd[0] != 0) close (stdin_fd[0]); + if (stdin_fd[1] != 0) close (stdin_fd[1]); + if (stdout_fd[0] != 0) close (stdout_fd[0]); + if (stdout_fd[1] != 0) close (stdout_fd[1]); + if (stderr_fd[0] != 0) close (stderr_fd[0]); + if (stderr_fd[1] != 0) close (stderr_fd[1]); + return (FALSE); +}