Index: /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/CheckBusyJob.c =================================================================== --- /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/CheckBusyJob.c (revision 11539) +++ /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/CheckBusyJob.c (revision 11539) @@ -0,0 +1,97 @@ +# include "pcontrol.h" +# define DEBUG 0 + +int CheckBusyJob (Job *job, Host *host) { + + int status; + int outstate; + char *p; + char string[64]; + IOBuffer buffer; + + /* we are checking a job which is currently busy. it has been pulled from the + JOB_BUSY stack, and is linked to a host in the HOST_BUSY stack. + XXX need to check on state of HOST on return */ + + ASSERT (job, "job not set"); + ASSERT (host, "host not set"); + + ASSERT (host == (Host *) job[0].host, "invalid host"); + ASSERT (job == (Job *) host[0].job, "invalid job"); + + InitIOBuffer (&buffer, 0x100); + + status = PclientCommand (host, "status", PCLIENT_PROMPT, &buffer); + + /* check on success of pclient command */ + switch (status) { + case PCLIENT_DOWN: + HarvestHost (host[0].pid); + // unlink host & job + job[0].host = NULL; + host[0].job = NULL; + if (job[0].realhost) free (job[0].realhost); + job[0].realhost = NULL; + PutHost (host, PCONTROL_HOST_DOWN, STACK_BOTTOM); + PutJob (job, PCONTROL_JOB_PENDING, STACK_BOTTOM); + FreeIOBuffer (&buffer); + return (FALSE); + + case PCLIENT_HUNG: + // don't do anything drastic, just keep trying + if (DEBUG || VerboseMode()) gprint (GP_ERR, "client is busy, not responding"); + PutHost (host, PCONTROL_HOST_BUSY, STACK_BOTTOM); + PutJob (job, PCONTROL_JOB_BUSY, STACK_BOTTOM); + FreeIOBuffer (&buffer); + return (TRUE); + + case PCLIENT_GOOD: + if (DEBUG || VerboseMode()) gprint (GP_ERR, "message received (CheckBusyJob)"); + break; + + default: + ABORT ("unknown status for pclient command"); + } + + /** host is up, need to parse message **/ + p = memstr (buffer.buffer, "STATUS", buffer.Nbuffer); + ASSERT (p != NULL, "missing STATUS in pclient message"); + + sscanf (p, "%*s %s", string); + ASSERT (strcmp(string, "NONE"), "no current job\n"); + + /** no status change, return to BUSY stack **/ + if (!strcmp(string, "BUSY")) { + PutHost (host, PCONTROL_HOST_BUSY, STACK_BOTTOM); + PutJob (job, PCONTROL_JOB_BUSY, STACK_BOTTOM); + FreeIOBuffer (&buffer); + return (TRUE); + } + + /* exit status better be either EXIT or CRASH */ + outstate = PCONTROL_JOB_BUSY; + if (!strcmp(string, "EXIT")) outstate = PCONTROL_JOB_EXIT; + if (!strcmp(string, "CRASH")) outstate = PCONTROL_JOB_CRASH; + ASSERT (outstate != PCONTROL_JOB_BUSY, "should not reach here (CheckJob)"); + + /* parse the exit status and sizes of output buffers */ + p = memstr (buffer.buffer, "EXITST", buffer.Nbuffer); + sscanf (p, "%*s %d", &job[0].exit_status); + p = memstr (buffer.buffer, "STDOUT", buffer.Nbuffer); + sscanf (p, "%*s %d", &job[0].stdout_size); + p = memstr (buffer.buffer, "STDERR", buffer.Nbuffer); + sscanf (p, "%*s %d", &job[0].stderr_size); + + if (job[0].stdout_size > 100000) abort(); + if (job[0].stderr_size > 100000) abort(); + + // job has exited : move to DONE stack + // the host is still BUSY until job output is gathered (int CheckDoneJob) + // don't unlink job and host yet + PutHost (host, PCONTROL_HOST_BUSY, STACK_BOTTOM); + PutJobSetState (job, PCONTROL_JOB_DONE, STACK_BOTTOM, outstate); + gettimeofday (&job[0].stop, (void *) NULL); + job[0].dtime = DTIME(job[0].stop, job[0].start); + FreeIOBuffer (&buffer); + return (TRUE); +} Index: /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/CheckDoneHost.c =================================================================== --- /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/CheckDoneHost.c (revision 11539) +++ /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/CheckDoneHost.c (revision 11539) @@ -0,0 +1,71 @@ +# include "pcontrol.h" +# define DEBUG 0 + +int CheckDoneHost (Host *host) { + + int status; + char *p; + IOBuffer buffer; + + ASSERT (host, "host not set"); + + InitIOBuffer (&buffer, 0x100); + + status = PclientCommand (host, "reset", PCLIENT_PROMPT, &buffer); + + /* check on success of pclient command */ + switch (status) { + case PCLIENT_DOWN: + if (DEBUG || VerboseMode()) gprint (GP_ERR, "host %s is down\n", host[0].hostname); + /* DONE host does not have an incomplete job */ + HarvestHost (host[0].pid); + PutHost (host, PCONTROL_HOST_DOWN, STACK_BOTTOM); + FreeIOBuffer (&buffer); + return (FALSE); + // XXX do we need to close the connection? + + case PCLIENT_HUNG: + // don't do anything drastic, just try again later + PutHost (host, PCONTROL_HOST_DONE, STACK_BOTTOM); + if (DEBUG || VerboseMode()) gprint (GP_ERR, "host %s is not responding\n", host[0].hostname); + FreeIOBuffer (&buffer); + return (FALSE); + + case PCLIENT_GOOD: + if (VerboseMode()) gprint (GP_ERR, "message received (CheckDoneHost)\n"); + break; + + default: + ABORT ("unknown status for pclient command"); + } + + /** successful command, examine result **/ + p = memstr (buffer.buffer, "STATUS", buffer.Nbuffer); + ASSERT (p != NULL, "missing STATUS in pclient message (CheckDoneHost)"); + + sscanf (p, "%*s %d", &status); + switch (status) { + case -1: + ABORT ("reset syntax error"); + + case 0: + if (DEBUG || VerboseMode()) gprint (GP_ERR, "reset failed\n"); + PutHost (host, PCONTROL_HOST_DONE, STACK_BOTTOM); + FreeIOBuffer (&buffer); + return (FALSE); + + case 1: + case 2: + if (DEBUG || VerboseMode()) gprint (GP_ERR, "successful reset\n"); + PutHost (host, PCONTROL_HOST_IDLE, STACK_BOTTOM); + FreeIOBuffer (&buffer); + return (FALSE); + + default: + ABORT ("should not reach here (CheckDoneHost)"); + } + ABORT ("should not reach here (CheckDoneHost)"); +} + +/** probably need to flush the buffer before the command **/ +/** need to add timeout check here **/ Index: /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/CheckDoneJob.c =================================================================== --- /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/CheckDoneJob.c (revision 11539) +++ /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/CheckDoneJob.c (revision 11539) @@ -0,0 +1,35 @@ +# include "pcontrol.h" + +int CheckDoneJob (Job *job, Host *host) { + + int success; + + ASSERT (job, "job not set"); + ASSERT (host, "host not set"); + + ASSERT (host == (Host *) job[0].host, "invalid host"); + ASSERT (job == (Job *) host[0].job, "invalid job"); + + success = TRUE; + success &= GetJobOutput ("stdout", host, &job[0].stdout, job[0].stdout_size); + success &= GetJobOutput ("stderr", host, &job[0].stderr, job[0].stderr_size); + + if (!success) { + // XXX some kind of error? + // XXX try again later? + PutHost (host, PCONTROL_HOST_BUSY, STACK_BOTTOM); + PutJob (job, PCONTROL_JOB_DONE, STACK_BOTTOM); + return (FALSE); + } + + /* job's state is either EXIT or CRASH (verify?) */ + // unlink host & job + job[0].host = NULL; + host[0].job = NULL; + PutHost (host, PCONTROL_HOST_DONE, STACK_BOTTOM); + PutJob (job, job[0].state, STACK_BOTTOM); + + return (TRUE); +} + +/** need to add timeout check here **/ Index: /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/CheckHost.c =================================================================== --- /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/CheckHost.c (revision 11539) +++ /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/CheckHost.c (revision 11539) @@ -0,0 +1,47 @@ +# include "pcontrol.h" + +int CheckHost (Host *host) { + + int status; + IOBuffer buffer; + + ASSERT (host, "host not set"); + + if (host[0].job != NULL) return (TRUE); + + /* if this host has been marked to be turned off, do that and return */ + if (host[0].markoff) { + host[0].markoff = FALSE; + StopHost (host); + OffHost (host); + return (TRUE); + } + + InitIOBuffer (&buffer, 0x100); + + status = PclientCommand (host, "echo OK", PCLIENT_PROMPT, &buffer); + switch (status) { + case 0: + if (VerboseMode()) gprint (GP_ERR, "host %s is down\n", host[0].hostname); + HarvestHost (host[0].pid); + PutHost (host, PCONTROL_HOST_DOWN, STACK_BOTTOM); + FreeIOBuffer (&buffer); + return (FALSE); + + case -1: + if (VerboseMode()) gprint (GP_ERR, "host %s is not responding\n", host[0].hostname); + /*** do we mark this in some way (HUNG) ? ***/ + PutHost (host, host[0].stack, STACK_BOTTOM); + FreeIOBuffer (&buffer); + return (FALSE); + + default: + PutHost (host, host[0].stack, STACK_BOTTOM); + FreeIOBuffer (&buffer); + return (TRUE); + } + ABORT ("should not reach here (Check Host)"); +} + +// if the host has a job, we skip it (down or crash state will be caught elsewhere) +// in fact, just touch the IDLE hosts, not the BUSY hosts? Index: /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/CheckIdleHost.c =================================================================== --- /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/CheckIdleHost.c (revision 11539) +++ /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/CheckIdleHost.c (revision 11539) @@ -0,0 +1,102 @@ +# include "pcontrol.h" + +/* the supplied host is not on a stack: it cannot be taken by the other thread */ +int CheckIdleHost (Host *host) { + + int i; + Stack *stack; + Job *job; + + ASSERT (host, "host not set"); + + /* if this host has been marked to be turned off, do that and return */ + if (host[0].markoff) { + host[0].markoff = FALSE; + StopHost (host); + OffHost (host); + return (TRUE); + } + + /* search the JOB_PENDING stack for an appropriate job */ + stack = GetJobStack (PCONTROL_JOB_PENDING); + LockStack (stack); + + /* look for first NEEDHOST matching this host */ + for (i = 0; i < stack[0].Nobject; i++) { + job = (Job *) stack[0].object[i]; + if (job[0].mode != PCONTROL_JOB_NEEDHOST) continue; + ASSERT (job[0].hostname != NULL, "NEEDHOST hostname missing"); + if (strcasecmp (job[0].hostname, host[0].hostname)) continue; + + /* we have found an appropriate job; link it to the host and send to StartJob */ + job[0].host = (struct Host *) host; + host[0].job = (struct Job *) job; + + /* take the job off the stack and unlock the stack */ + RemoveStackEntry (stack, i); + UnlockStack (stack); + StartJob (job, host); + return (TRUE); + } + + /* no NEEDHOST entry, look for first WANTHOST matching this host */ + for (i = 0; i < stack[0].Nobject; i++) { + job = (Job *) stack[0].object[i]; + if (job[0].mode != PCONTROL_JOB_WANTHOST) continue; + ASSERT (job[0].hostname != NULL, "WANTHOST hostname missing"); + if (strcasecmp (job[0].hostname, host[0].hostname)) continue; + + /* we have found an appropriate job; link it to the host and send to StartJob */ + job[0].host = (struct Host *) host; + host[0].job = (struct Job *) job; + + /* take the job off the stack and unlock the stack */ + RemoveStackEntry (stack, i); + UnlockStack (stack); + StartJob (job, host); + return (TRUE); + } + + /* no WANTHOST entry, look for first ANYHOST matching this host */ + for (i = 0; i < stack[0].Nobject; i++) { + job = (Job *) stack[0].object[i]; + if (job[0].mode != PCONTROL_JOB_ANYHOST) continue; + + /* we have found an appropriate job; link it to the host and send to StartJob */ + job[0].host = (struct Host *) host; + host[0].job = (struct Job *) job; + + /* take the job off the stack and unlock the stack */ + RemoveStackEntry (stack, i); + UnlockStack (stack); + StartJob (job, host); + return (TRUE); + } + + /* no ANYHOST entry, look for first WANTHOST with old time */ + /* XXX perhaps I should add this to the conditions for ANYHOST instead of + running a separate loop? ie, WANTHOST && time > X == ANYHOST */ + for (i = 0; i < stack[0].Nobject; i++) { + job = (Job *) stack[0].object[i]; + if (job[0].mode != PCONTROL_JOB_WANTHOST) continue; + // XXX test the job age and skip if too young + + /* we have found an appropriate job; link it to the host and send to StartJob */ + job[0].host = (struct Host *) host; + host[0].job = (struct Job *) job; + + /* take the job off the stack and unlock the stack */ + RemoveStackEntry (stack, i); + UnlockStack (stack); + StartJob (job, host); + return (TRUE); + } + UnlockStack (stack); + + /* no jobs for host, put it back on IDLE stack */ + PutHost (host, PCONTROL_HOST_IDLE, STACK_BOTTOM); + return (TRUE); +} + +/** note : host and job popped off IDLE and PENDING stacks, + unless no job is available **/ Index: /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/CheckPoint.c =================================================================== --- /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/CheckPoint.c (revision 11539) +++ /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/CheckPoint.c (revision 11539) @@ -0,0 +1,69 @@ +# include "pcontrol.h" + +// one thread (user) interacts with the user and blocks for long periods on input. +// one thread (client) spins continuously and monitors the hosts and jobs +// in some cases, the user thread needs to set a check point on the client thread +// to ensure the all HOSTs and JOBs are on one of the stacks (nothing 'in flight'). +// these are not symmetric: the client thread should not call Set/Clear, the user +// thread should not call Test + +# ifdef THREADED +static pthread_mutex_t client = PTHREAD_MUTEX_INITIALIZER; +static pthread_mutex_t user = PTHREAD_MUTEX_INITIALIZER; +# endif + +// The user thread calls this to stop the client thread from shuffling the Host/Job stacks +int SetCheckPoint () { + +# ifdef THREADED + int status; + int Nwait; + + // set my lock + pthread_mutex_lock (&user); + + // wait until client thread sets its lock + Nwait = 0; + while (1) { + status = pthread_mutex_trylock (&client); + if (status == EBUSY) { + // client has reached the check-point + return (TRUE); + } + pthread_mutex_unlock (&client); + usleep (10000); // wait for client thread to set lock + Nwait ++; + } + // put in a timeout? (client thread not spinning...) + return (FALSE); +# else + return (TRUE); +# endif +} + +// The user thread calls this to allow the client thread to continue +int ClearCheckPoint () { + // clear my lock +# ifdef THREADED + pthread_mutex_unlock (&user); +# endif + return (TRUE); +} + +// The client thread calls in the thread loop somewhere the stacks are stable +// (ie, no jobs or hosts currently in flight) +int TestCheckPoint () { + +# ifdef THREADED + // set my lock + pthread_mutex_lock (&client); + + // try the user-thread lock + pthread_mutex_lock (&user); + pthread_mutex_unlock (&user); + + // clear my lock + pthread_mutex_unlock (&client); +# endif + return (TRUE); +} Index: /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/CheckSystem.c =================================================================== --- /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/CheckSystem.c (revision 11539) +++ /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/CheckSystem.c (revision 11539) @@ -0,0 +1,380 @@ +# include "pcontrol.h" +# define DEBUG 0 + +static struct timeval lastlive = {0, 0}; +static RunLevels RunLevel = PCONTROL_RUN_NONE; + +RunLevels SetRunLevel (RunLevels level) { + RunLevels oldlevel; + oldlevel = RunLevel; + RunLevel = level; + return oldlevel; +} + +RunLevels GetRunLevel () { + return RunLevel; +} + +int CheckSystem () { + + struct timeval now; + float dtime; + + /* we want to give each block a maximum allowed time */ + CheckIdleHosts(0.020); /* submit a new job */ + + CheckBusyJobs(0.020); /* get job status */ + CheckDoneJobs(0.020); /* harvest job stdout/stderr */ + CheckKillJobs(0.020); /* harvest job stdout/stderr */ + + CheckDoneHosts(0.020); /* reset the host */ + CheckDownHosts(0.100); /* launch the host */ + + /* always allow at least one test */ + /* most tests require about 2ms per host. + CheckDoneJobs must depend on the size of the output buffer */ + + gettimeofday (&now, (void *) NULL); + dtime = DTIME (now, lastlive); + if (dtime > 1.0) { + CheckLiveHosts(0.040); + lastlive = now; + } + + if (DEBUG) { + Stack *stack; + int Nidle, Ndown, Nbusy; + stack = GetHostStack (PCONTROL_HOST_IDLE); + Nidle = stack[0].Nobject; + stack = GetHostStack (PCONTROL_HOST_DOWN); + Ndown = stack[0].Nobject; + stack = GetHostStack (PCONTROL_HOST_BUSY); + Nbusy = stack[0].Nobject; + gprint (GP_ERR, "busy, idle, down: %2d %2d %2d\n", Nbusy, Nidle, Ndown); + } + + return (TRUE); +} + +void *CheckSystem_Threaded (void *data) { + + int Njobchecks, Nhostchecks, Nlivechecks; + + Nlivechecks = 0; + + gprintInit (); + + while (1) { + // stop here if the user-thread requests (no objects in flight) + TestCheckPoint (); + + // don't run the system checks if RunLevel is FALSE + // XXX stop should not suspend all checks: we should continue + // to harvest completed jobs and migrate idle machines to down + if (RunLevel == PCONTROL_RUN_NONE) { + usleep (100000); // idle if we are running nothing + continue; + } + + /* always allow at least one test */ + /* most tests require about 2ms per host. + CheckDoneJobs must depend on the size of the output buffer */ + /* the max delay times are fairly arbitrary and do not impact + the user interface. + */ + + Njobchecks = 0; + Nhostchecks = 0; + + if ((RunLevel == PCONTROL_RUN_ALL) || (RunLevel == PCONTROL_RUN_REAP)) { + Njobchecks += CheckBusyJobs(0.020); /* get job status */ + Njobchecks += CheckDoneJobs(0.020); /* harvest job stdout/stderr */ + Njobchecks += CheckKillJobs(0.020); /* harvest job stdout/stderr */ + TestCheckPoint (); + } + + if (RunLevel != PCONTROL_RUN_NONE) { + Nhostchecks += CheckDoneHosts(0.020); /* reset the host */ + Nhostchecks += CheckDownHosts(0.100); /* launch the host */ + TestCheckPoint (); + } + + if (RunLevel == PCONTROL_RUN_ALL) { + // we want to give each block a maximum allowed time + Nhostchecks += CheckIdleHosts(0.020); /* submit a new job */ + TestCheckPoint (); + } + + // there is nothing on the stacks. test the hosts and wait a bit + if (!Njobchecks && !Nhostchecks && (RunLevel != PCONTROL_RUN_NONE)) { + CheckLiveHosts(0.040); + usleep (100000); // idle if no jobs are waiting + } + + if (DEBUG) { + Stack *stack; + int Nidle, Ndown, Nbusy; + stack = GetHostStack (PCONTROL_HOST_IDLE); + Nidle = stack[0].Nobject; + stack = GetHostStack (PCONTROL_HOST_DOWN); + Ndown = stack[0].Nobject; + stack = GetHostStack (PCONTROL_HOST_BUSY); + Nbusy = stack[0].Nobject; + gprint (GP_ERR, "Njobchecks: %d, busy, idle, down: %2d %2d %2d\n", Njobchecks, Nbusy, Nidle, Ndown); + } + } + return (NULL); +} + +int CheckBusyJobs (float MaxDelay) { + + struct timeval start, stop; + int i, Nobject; + Stack *hoststack; + Stack *jobstack; + Job *job; + Host *host; + float dtime; + + /* Loop through objects on the stack, no more than once. Note that it is not important if the + stack size is modified by other threads or is changed by any of the actions performed during + this loop: the Nobject value is only used to get a rough number for the number of iterations. + */ + + hoststack = GetHostStack (PCONTROL_HOST_BUSY); + jobstack = GetJobStack (PCONTROL_JOB_BUSY); + Nobject = jobstack[0].Nobject; + + /* always allow at least one test */ + gettimeofday (&start, (void *) NULL); + dtime = 0.0; + for (i = 0; (i < Nobject) && (dtime < MaxDelay); i++) { + // pull both job and host from their stacks + LockStack (hoststack); + job = PullStackByLocation (jobstack, STACK_TOP); + if (job == NULL) { + UnlockStack (hoststack); + break; + } + host = (Host *) job[0].host; + ASSERT (host != NULL, "host is NULL"); + RemoveStackByID (hoststack, host[0].HostID); + UnlockStack (hoststack); + + CheckBusyJob (job, host); + gettimeofday (&stop, (void *) NULL); + dtime = DTIME (stop, start); + } + if (DEBUG && (Nobject > 0)) gprint (GP_ERR, "checked %d of %d jobs\n", i, Nobject); + return (i); +} + +int CheckDoneJobs (float MaxDelay) { + + struct timeval start, stop; + int i, Nobject; + Stack *hoststack; + Stack *jobstack; + Job *job; + Host *host; + float dtime; + + /* Loop through objects on the stack, no more than once. see note above */ + hoststack = GetHostStack (PCONTROL_HOST_BUSY); + jobstack = GetJobStack (PCONTROL_JOB_DONE); + Nobject = jobstack[0].Nobject; + + /* always allow at least one test */ + gettimeofday (&start, (void *) NULL); + dtime = 0.0; + for (i = 0; (i < Nobject) && (dtime < MaxDelay); i++) { + LockStack (hoststack); + job = PullStackByLocation (jobstack, STACK_TOP); + if (job == NULL) { + UnlockStack (hoststack); + break; + } + host = (Host *) job[0].host; + ASSERT (host, "host is NULL"); + + RemoveStackByID (hoststack, host[0].HostID); + UnlockStack (hoststack); + + CheckDoneJob (job, host); + gettimeofday (&stop, (void *) NULL); + dtime = DTIME (stop, start); + } + if (DEBUG && (Nobject > 0)) gprint (GP_ERR, "checked %d of %d jobs\n", i, Nobject); + return (i); +} + +int CheckKillJobs (float MaxDelay) { + + struct timeval start, stop; + int i, Nobject; + Stack *hoststack; + Stack *jobstack; + Job *job; + Host *host; + float dtime; + + /* Loop through objects on the stack, no more than once. see note above */ + hoststack = GetHostStack (PCONTROL_HOST_BUSY); + jobstack = GetJobStack (PCONTROL_JOB_KILL); + Nobject = jobstack[0].Nobject; + + /* always allow at least one test */ + gettimeofday (&start, (void *) NULL); + dtime = 0.0; + for (i = 0; (i < Nobject) && (dtime < MaxDelay); i++) { + LockStack (hoststack); + job = PullStackByLocation (jobstack, STACK_TOP); + if (job == NULL) { + UnlockStack (hoststack); + break; + } + host = (Host *) job[0].host; + ASSERT (host, "host is NULL"); + + RemoveStackByID (hoststack, host[0].HostID); + UnlockStack (hoststack); + + KillJob (job, host); + gettimeofday (&stop, (void *) NULL); + dtime = DTIME (stop, start); + } + if (DEBUG && (Nobject > 0)) gprint (GP_ERR, "checked %d of %d jobs\n", i, Nobject); + return (i); +} + +int CheckDoneHosts (float MaxDelay) { + + struct timeval start, stop; + int i, Nobject; + Stack *stack; + Host *host; + float dtime; + + /* Loop through objects on the stack, no more than once. see note above */ + stack = GetHostStack (PCONTROL_HOST_DONE); + Nobject = stack[0].Nobject; + + /* always allow at least one test */ + gettimeofday (&start, (void *) NULL); + dtime = 0.0; + for (i = 0; (i < Nobject) && (dtime < MaxDelay); i++) { + host = PullStackByLocation (stack, STACK_TOP); + if (host == NULL) break; + CheckDoneHost (host); + gettimeofday (&stop, (void *) NULL); + dtime = DTIME (stop, start); + } + if (DEBUG) gprint (GP_ERR, "checked %d hosts\n", i); + return (i); +} + +int CheckDownHosts (float MaxDelay) { + + int i, Nobject; + Stack *stack; + Host *host; + struct timeval start, stop; + float dtime; + + /* Loop through objects on the stack, no more than once. see note above */ + stack = GetHostStack (PCONTROL_HOST_DOWN); + Nobject = stack[0].Nobject; + + /* always allow at least one test */ + gettimeofday (&start, (void *) NULL); + dtime = 0.0; + for (i = 0; (i < Nobject) && (dtime < MaxDelay); i++) { + host = PullStackByLocation (stack, STACK_TOP); + if (host == NULL) break; + if (host[0].markoff) { + host[0].markoff = FALSE; + OffHost (host); + return (TRUE); + } + dtime = DTIME (host[0].nexttry, start); + if (dtime > 0) { + PutHost (host, PCONTROL_HOST_DOWN, STACK_BOTTOM); + } else { + StartHost (host); + } + gettimeofday (&stop, (void *) NULL); + dtime = DTIME (stop, start); + } + if (DEBUG) gprint (GP_ERR, "checked %d hosts\n", i); + return (i); +} + +int CheckIdleHosts (float MaxDelay) { + + struct timeval start, stop; + int i, Nobject; + Stack *stack; + Host *host; + float dtime; + + /* check if there are any pending jobs, otherwise skip step */ + stack = GetJobStack (PCONTROL_JOB_PENDING); + if (!stack[0].Nobject) return (0); + + /* Loop through objects on the stack, no more than once. see note above */ + stack = GetHostStack (PCONTROL_HOST_IDLE); + Nobject = stack[0].Nobject; + + /* always allow at least one test */ + gettimeofday (&start, (void *) NULL); + dtime = 0.0; + for (i = 0; (i < Nobject) && (dtime < MaxDelay); i++) { + host = PullStackByLocation (stack, STACK_TOP); + if (host == NULL) break; + CheckIdleHost (host); + gettimeofday (&stop, (void *) NULL); + dtime = DTIME (stop, start); + } + if (DEBUG) gprint (GP_ERR, "checked %d hosts\n", i); + return (i); +} + +/* this is just a heartbeat check (only IDLE hosts) */ +int CheckLiveHosts (float MaxDelay) { + + struct timeval start, stop; + int i, Nobject; + Stack *stack; + Host *host; + float dtime; + + /* Loop through objects on the stack, no more than once. see note above */ + stack = GetHostStack (PCONTROL_HOST_IDLE); + Nobject = stack[0].Nobject; + + gettimeofday (&start, (void *) NULL); + + dtime = 0.0; + for (i = 0; (i < Nobject) && (dtime < MaxDelay); i++) { + host = PullStackByLocation (stack, STACK_TOP); + if (host == NULL) break; + CheckHost (host); + gettimeofday (&stop, (void *) NULL); + dtime = DTIME (stop, start); + } + if (DEBUG) gprint (GP_ERR, "checked %d idle hosts\n", i); + return (TRUE); +} + +/* + + gettimeofday (&stop, (void *) NULL); + dtime = DTIME (stop, start); + if (VerboseMode()) gprint (GP_ERR, "check 4: %f seconds\n", dtime); + + gettimeofday (&start, (void *) NULL); +*/ + +/** All of the CheckFooBar entries cycle though their respective queues, popping from the top and + pushing to the bottom. if we stop before the loop is done there is no tendancy for bias because + we continue where we left off next round **/ Index: /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/GetJobOutput.c =================================================================== --- /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/GetJobOutput.c (revision 11539) +++ /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/GetJobOutput.c (revision 11539) @@ -0,0 +1,65 @@ +# include "pcontrol.h" +# define PCLIENT_TIMEOUT 500 + +/* we read Nbytes from the host, then watch for the prompt */ +int GetJobOutput (char *cmd, Host *host, IOBuffer *buffer, int Nbytes) { + + int i, status, Nstart; + char *line; + struct timespec request, remain; + + ASSERT (cmd, "cmd missing"); + ASSERT (host, "host missing"); + ASSERT (buffer, "buffer missing"); + + /* flush any earlier messages */ + ReadtoIOBuffer (buffer, host[0].stdout); + FlushIOBuffer (buffer); + Nstart = buffer[0].Nbuffer; + + /* avoid blocking on waitpid, test every 100 usec, up to 50 msec */ + request.tv_sec = 0; + request.tv_nsec = 100000; + + /* send cmd (stdout / stderr) */ + status = write_fmt (host[0].stdin, "%s\n", cmd); + + /* is pipe still open? */ + if ((status == -1) && (errno == EPIPE)) return (PCLIENT_DOWN); + + /* read at least Nbytes, then watch for PCLIENT_PROMPT */ + line = NULL; + status = -1; + for (i = 0; (i < PCLIENT_TIMEOUT) && (status != 0) && (line == NULL); i++) { + status = ReadtoIOBuffer (buffer, host[0].stdout); + if ((buffer[0].Nbuffer - Nstart) >= Nbytes) { + line = memstr (buffer[0].buffer, PCLIENT_PROMPT, buffer[0].Nbuffer); + } + if (status == -1) nanosleep (&request, &remain); + } + if (status == 0) return (PCLIENT_DOWN); + if (status == -1) return (PCLIENT_HUNG); + + /* check on success of pclient command */ + switch (status) { + case -1: + if (VerboseMode()) gprint (GP_ERR, "host %s is not responding\n", host[0].hostname); + return (FALSE); + + case 0: + if (VerboseMode()) gprint (GP_ERR, "host %s is down\n", host[0].hostname); + return (FALSE); + + default: + if (VerboseMode()) gprint (GP_ERR, "message received (GetJobOutput : %s)\n", cmd); + /* drop extra bytes from pclient (not pclient:job) */ + buffer[0].Nbuffer = Nstart + Nbytes; + if (buffer[0].Nalloc > buffer[0].Nbuffer) { + bzero (buffer[0].buffer + buffer[0].Nbuffer, buffer[0].Nalloc - buffer[0].Nbuffer); + } + return (TRUE); + } + + gprint (GP_ERR, "programming error: should not reach here (GetJobOutput)\n"); + exit (1); +} Index: /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/HostOps.c =================================================================== --- /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/HostOps.c (revision 11539) +++ /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/HostOps.c (revision 11539) @@ -0,0 +1,174 @@ +# include "pcontrol.h" + +Stack *HostPool_Idle; +Stack *HostPool_Busy; +Stack *HostPool_Done; +Stack *HostPool_Down; +Stack *HostPool_Off; + +void InitHostStacks () { + HostPool_Idle = InitStack (); + HostPool_Busy = InitStack (); + HostPool_Done = InitStack (); + HostPool_Down = InitStack (); + HostPool_Off = InitStack (); +} + +char *GetHostStackName (int StackID) { + switch (StackID) { + case PCONTROL_HOST_IDLE: return ("IDLE"); + case PCONTROL_HOST_DOWN: return ("DOWN"); + case PCONTROL_HOST_DONE: return ("DONE"); + case PCONTROL_HOST_BUSY: return ("BUSY"); + case PCONTROL_HOST_OFF: return ("OFF"); + } + gprint (GP_ERR, "error: unknown host stack : programming error\n"); + exit (1); +} + +Stack *GetHostStack (int StackID) { + switch (StackID) { + case PCONTROL_HOST_IDLE: return (HostPool_Idle); + case PCONTROL_HOST_DOWN: return (HostPool_Down); + case PCONTROL_HOST_DONE: return (HostPool_Done); + case PCONTROL_HOST_BUSY: return (HostPool_Busy); + case PCONTROL_HOST_OFF: return (HostPool_Off); + } + gprint (GP_ERR, "error: unknown host stack : programming error\n"); + exit (1); +} + +Stack *GetHostStackByName (char *name) { + if (!strcasecmp (name, "idle")) return (HostPool_Idle); + if (!strcasecmp (name, "down")) return (HostPool_Down); + if (!strcasecmp (name, "done")) return (HostPool_Done); + if (!strcasecmp (name, "busy")) return (HostPool_Busy); + if (!strcasecmp (name, "off")) return (HostPool_Off); + return (NULL); +} + +/* add host to position in stack */ +int PutHost (Host *host, int StackID, int where) { + + int stat; + Stack *stack; + + stack = GetHostStack (StackID); + if (stack == NULL) return (FALSE); + + host[0].stack = StackID; + stat = PushStack (stack, where, host, host[0].HostID, host[0].hostname); + // XXX need to handle the error conditions, or we drop the host & leak memory + return (stat); +} + +/* find the host by ID in the defined host stacks */ +Host *PullHostByID (IDtype HostID, int *StackID) { + + Host *host; + + *StackID = PCONTROL_HOST_IDLE; + host = PullHostFromStackByID (*StackID, HostID); + if (host != NULL) return (host); + + *StackID = PCONTROL_HOST_DOWN; + host = PullHostFromStackByID (*StackID, HostID); + if (host != NULL) return (host); + + *StackID = PCONTROL_HOST_DONE; + host = PullHostFromStackByID (*StackID, HostID); + if (host != NULL) return (host); + + *StackID = PCONTROL_HOST_BUSY; + host = PullHostFromStackByID (*StackID, HostID); + if (host != NULL) return (host); + + *StackID = PCONTROL_HOST_OFF; + host = PullHostFromStackByID (*StackID, HostID); + if (host != NULL) return (host); + + *StackID = -1; + return (NULL); +} + +/* find the host by ID in the defined host stacks */ +Host *PullHostByName (char *name, int *StackID) { + + Host *host; + + *StackID = PCONTROL_HOST_IDLE; + host = PullHostFromStackByName (*StackID, name); + if (host != NULL) return (host); + + *StackID = PCONTROL_HOST_DOWN; + host = PullHostFromStackByName (*StackID, name); + if (host != NULL) return (host); + + *StackID = PCONTROL_HOST_DONE; + host = PullHostFromStackByName (*StackID, name); + if (host != NULL) return (host); + + *StackID = PCONTROL_HOST_BUSY; + host = PullHostFromStackByName (*StackID, name); + if (host != NULL) return (host); + + *StackID = PCONTROL_HOST_OFF; + host = PullHostFromStackByName (*StackID, name); + if (host != NULL) return (host); + + *StackID = -1; + return (NULL); +} + +Host *PullHostFromStackByID (int StackID, IDtype ID) { + + Host *host; + Stack *stack; + + stack = GetHostStack (StackID); + if (stack == NULL) return (NULL); + + host = PullStackByID (stack, ID); + return (host); +} + +Host *PullHostFromStackByName (int StackID, char *name) { + + Host *host; + Stack *stack; + + stack = GetHostStack (StackID); + if (stack == NULL) return (NULL); + + host = PullStackByName (stack, name); + return (host); +} + +IDtype 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].HostID = NextHostID(); + + host[0].lasttry.tv_sec = 0; + host[0].lasttry.tv_usec = 0; + host[0].nexttry.tv_sec = 0; + host[0].nexttry.tv_usec = 0; + + host[0].markoff = FALSE; + host[0].job = NULL; + PutHost (host, PCONTROL_HOST_DOWN, STACK_BOTTOM); + return (host[0].HostID); +} + +void DelHost (Host *host) { + FREE (host[0].hostname); + FREE (host[0].job); + FREE (host); +} Index: /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/IDops.c =================================================================== --- /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/IDops.c (revision 11539) +++ /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/IDops.c (revision 11539) @@ -0,0 +1,42 @@ +# include "pcontrol.h" + +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 */ +void InitIDs () { + CurrentJobID = 0; + CurrentHostID = 0; +} + +IDtype NextJobID () { + + IDtype ID; + + ID = CurrentJobID; + CurrentJobID ++; + return (ID); +} + +/* only used by the User thread */ +IDtype NextHostID () { + + IDtype ID; + + ID = CurrentHostID; + CurrentHostID ++; + return (ID); +} + +void PrintID (gpDest dest, IDtype ID) { + + unsigned short int word0, word1, word2, word3; + + word0 = 0xffff & ID; + word1 = 0xffff & (ID >> 16); + word2 = 0xffff & (ID >> 32); + word3 = 0xffff & (ID >> 48); + + gprint (dest, "%x.%x.%x.%x", word3, word2, word1, word0); +} + Index: /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/JobID.c =================================================================== --- /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/JobID.c (revision 11539) +++ /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/JobID.c (revision 11539) @@ -0,0 +1,41 @@ +# include "pcontrol.h" + +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 */ +void 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 (gpDest dest, IDtype ID) { + + unsigned short int word0, word1, word2, word3; + + word0 = 0xffff & ID; + word1 = 0xffff & (ID >> 16); + word2 = 0xffff & (ID >> 32); + word3 = 0xffff & (ID >> 48); + + gprint (dest, "%x.%x.%x.%x", word3, word2, word1, word0); +} + Index: /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/JobOps.c =================================================================== --- /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/JobOps.c (revision 11539) +++ /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/JobOps.c (revision 11539) @@ -0,0 +1,173 @@ +# include "pcontrol.h" + +Stack *JobPool_Pending; +Stack *JobPool_Busy; +Stack *JobPool_Done; +Stack *JobPool_Kill; +Stack *JobPool_Exit; +Stack *JobPool_Crash; + +void InitJobStacks () { + JobPool_Pending = InitStack (); + JobPool_Busy = InitStack (); + JobPool_Done = InitStack (); + JobPool_Kill = InitStack (); + JobPool_Exit = InitStack (); + JobPool_Crash = InitStack (); +} + +char *GetJobStackName (int StackID) { + switch (StackID) { + case PCONTROL_JOB_PENDING: return ("PENDING"); + case PCONTROL_JOB_BUSY: return ("BUSY"); + case PCONTROL_JOB_DONE: return ("DONE"); + case PCONTROL_JOB_KILL: return ("KILL"); + case PCONTROL_JOB_EXIT: return ("EXIT"); + case PCONTROL_JOB_CRASH: return ("CRASH"); + } + gprint (GP_ERR, "error: unknown host stack : programming error\n"); + exit (1); +} + +Stack *GetJobStack (int StackID) { + switch (StackID) { + case PCONTROL_JOB_PENDING: return (JobPool_Pending); + case PCONTROL_JOB_BUSY: return (JobPool_Busy); + case PCONTROL_JOB_DONE: return (JobPool_Done); + case PCONTROL_JOB_KILL: return (JobPool_Kill); + case PCONTROL_JOB_EXIT: return (JobPool_Exit); + case PCONTROL_JOB_CRASH: return (JobPool_Crash); + } + gprint (GP_ERR, "error: unknown job stack : programming error\n"); + exit (1); +} + +Stack *GetJobStackByName (char *name) { + + if (!strcasecmp (name, "pending")) return (JobPool_Pending); + if (!strcasecmp (name, "busy")) return (JobPool_Busy); + if (!strcasecmp (name, "done")) return (JobPool_Done); + if (!strcasecmp (name, "exit")) return (JobPool_Exit); + if (!strcasecmp (name, "crash")) return (JobPool_Crash); + return (NULL); +} + +/* add job to position in stack, use StackID as default state */ +int PutJob (Job *job, int StackID, int where) { + + int stat; + Stack *stack; + + stack = GetJobStack (StackID); + if (stack == NULL) return (FALSE); + + /* by default, these are both the same - to override, use PutJobSetState */ + job[0].state = StackID; + job[0].stack = StackID; + stat = PushStack (stack, where, job, job[0].JobID, job[0].argv[0]); + // XXX need to handle the error conditions, or we drop the host & leak memory + return (stat); +} + +/* add job to position in stack. set state to 'state' */ +int PutJobSetState (Job *job, int StackID, int where, int state) { + + int stat; + Stack *stack; + + stack = GetJobStack (StackID); + if (stack == NULL) return (FALSE); + + /* alternate state specified by user */ + job[0].state = state; + job[0].stack = StackID; + stat = PushStack (stack, where, job, job[0].JobID, job[0].argv[0]); + // XXX need to handle the error conditions, or we drop the host & leak memory + return (stat); +} + +Job *PullJobByID (IDtype JobID, int *StackID) { + + Job *job; + + *StackID = PCONTROL_JOB_PENDING; + job = PullJobFromStackByID (*StackID, JobID); + if (job != NULL) return (job); + + *StackID = PCONTROL_JOB_BUSY; + job = PullJobFromStackByID (*StackID, JobID); + if (job != NULL) return (job); + + *StackID = PCONTROL_JOB_EXIT; + job = PullJobFromStackByID (*StackID, JobID); + if (job != NULL) return (job); + + *StackID = PCONTROL_JOB_CRASH; + job = PullJobFromStackByID (*StackID, JobID); + if (job != NULL) return (job); + + *StackID = PCONTROL_JOB_DONE; + job = PullJobFromStackByID (*StackID, JobID); + if (job != NULL) return (job); + + *StackID = PCONTROL_JOB_KILL; + job = PullJobFromStackByID (*StackID, JobID); + if (job != NULL) return (job); + + return (NULL); +} + +/* remove job from position in stack */ +Job *PullJobFromStackByID (int StackID, int ID) { + + Job *job; + Stack *stack; + + stack = GetJobStack (StackID); + if (stack == NULL) return (NULL); + + job = PullStackByID (stack, ID); + return (job); +} + +IDtype AddJob (char *hostname, JobMode mode, int timeout, int argc, char **argv) { + + int JobID; + Job *job; + + ALLOCATE (job, Job, 1); + + job[0].argc = argc; + job[0].argv = argv; + job[0].hostname = hostname; + job[0].mode = mode; + job[0].host = NULL; + job[0].JobID = NextJobID(); + job[0].Reset = FALSE; + job[0].realhost = NULL; + + /* do this step on start? */ + InitIOBuffer (&job[0].stdout, 0x1000); + InitIOBuffer (&job[0].stderr, 0x1000); + + JobID = job[0].JobID; + PutJob (job, PCONTROL_JOB_PENDING, STACK_BOTTOM); + if (VerboseMode()) gprint (GP_ERR, "added new job\n"); + return (JobID); +} + +void DelJob (Job *job) { + + int i; + + FREE (job[0].hostname); + for (i = 0; i < job[0].argc; i++) { + FREE (job[0].argv[i]); + } + FREE (job[0].argv); + + FreeIOBuffer (&job[0].stdout); + FreeIOBuffer (&job[0].stderr); + + FREE (job); +} Index: /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/KillJob.c =================================================================== --- /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/KillJob.c (revision 11539) +++ /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/KillJob.c (revision 11539) @@ -0,0 +1,75 @@ +# include "pcontrol.h" + +int KillJob (Job *job, Host *host) { + + IOBuffer buffer; + int status; + char *p; + + ASSERT (host != NULL, "host missing"); + ASSERT (job != NULL, "job missing"); + + ASSERT (host == (Host *) job[0].host, "invalid host"); + ASSERT (job == (Job *) host[0].job, "invalid job"); + + InitIOBuffer (&buffer, 0x100); + + status = PclientCommand (host, "reset", PCLIENT_PROMPT, &buffer); + + /* check on success of pclient command */ + switch (status) { + case PCLIENT_DOWN: + HarvestHost (host[0].pid); + // unlink host & job + job[0].host = NULL; + host[0].job = NULL; + PutHost (host, PCONTROL_HOST_DOWN, STACK_BOTTOM); + PutJob (job, PCONTROL_JOB_CRASH, STACK_BOTTOM); + FreeIOBuffer (&buffer); + return (FALSE); + + case PCLIENT_HUNG: + // don't do anything drastic, just keep trying + // XXX move to which stack?? + gprint (GP_ERR, "client is busy, not responding (KillJob)"); + FreeIOBuffer (&buffer); + return (TRUE); + + case PCLIENT_GOOD: + if (VerboseMode()) gprint (GP_ERR, "message received (KillJob)\n"); + break; + + default: + ABORT ("unknown status for pclient command"); + } + + /** host is up, need to parse message **/ + p = memstr (buffer.buffer, "STATUS", buffer.Nbuffer); + ASSERT (p != NULL, "missing STATUS in pclient message"); + if (VerboseMode()) gprint (GP_ERR, "client message: %s\n", buffer.buffer); + + sscanf (p, "%*s %d", &status); + FreeIOBuffer (&buffer); + gprint (GP_ERR, "client status: %d\n", status); + + switch (status) { + case -1: + ABORT ("syntax error to pclient"); + case 0: + gprint (GP_ERR, "failure to kill child process\n"); + PutHost (host, PCONTROL_HOST_BUSY, STACK_BOTTOM); + PutJob (job, PCONTROL_JOB_KILL, STACK_BOTTOM); + return (FALSE); + case 1: + gprint (GP_ERR, "killing job %s on %s\n", job[0].argv[0], host[0].hostname); + // unlink host & job + job[0].host = NULL; + host[0].job = NULL; + PutHost (host, PCONTROL_HOST_IDLE, STACK_BOTTOM); + PutJob (job, PCONTROL_JOB_CRASH, STACK_BOTTOM); + return (TRUE); + case 2: + ABORT ("client has no job"); + } + ABORT ("should not reach here (KillJob)"); +} Index: /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/Makefile =================================================================== --- /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/Makefile (revision 11539) +++ /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/Makefile (revision 11539) @@ -0,0 +1,76 @@ +include ../../../Configure +default: pcontrol + +HOME = $(ROOT)/src/opihi +BIN = $(HOME)/bin +LIB = $(HOME)/lib +INC = $(HOME)/include +SRC = $(HOME)/pcontrol +DATA = $(DESTDATA)/pcontrol + +# link flags +LIBS1 = -lkapa -lFITS -lohana $(LIBFLAGS) +LIBS2 = -lbasiccmd -lshell -ldata +LFLAGS = -L$(LIB) -L$(DESTLIB) $(LIBDIRS) $(LIBS2) $(LIBS1) + +# to build the non-threaded version, remove -lpthread and comment out +# the THREADED line in include/pcontrol.h + +# pcontrol user commands and support functions ######################## + +funcs = \ +$(SRC)/init.$(ARCH).o \ +$(SRC)/pcontrol.$(ARCH).o \ +$(SRC)/rconnect.$(ARCH).o \ +$(SRC)/CheckBusyJob.$(ARCH).o \ +$(SRC)/CheckDoneHost.$(ARCH).o \ +$(SRC)/CheckDoneJob.$(ARCH).o \ +$(SRC)/CheckHost.$(ARCH).o \ +$(SRC)/CheckIdleHost.$(ARCH).o \ +$(SRC)/CheckPoint.$(ARCH).o \ +$(SRC)/CheckSystem.$(ARCH).o \ +$(SRC)/GetJobOutput.$(ARCH).o \ +$(SRC)/HostOps.$(ARCH).o \ +$(SRC)/IDops.$(ARCH).o \ +$(SRC)/JobOps.$(ARCH).o \ +$(SRC)/StackOps.$(ARCH).o \ +$(SRC)/PclientCommand.$(ARCH).o \ +$(SRC)/ResetJob.$(ARCH).o \ +$(SRC)/StartHost.$(ARCH).o \ +$(SRC)/StopHosts.$(ARCH).o \ +$(SRC)/KillJob.$(ARCH).o \ +$(SRC)/StartJob.$(ARCH).o + +cmds = \ +$(SRC)/check.$(ARCH).o \ +$(SRC)/delete.$(ARCH).o \ +$(SRC)/host.$(ARCH).o \ +$(SRC)/job.$(ARCH).o \ +$(SRC)/jobstack.$(ARCH).o \ +$(SRC)/hoststack.$(ARCH).o \ +$(SRC)/kill.$(ARCH).o \ +$(SRC)/pulse.$(ARCH).o \ +$(SRC)/run.$(ARCH).o \ +$(SRC)/status.$(ARCH).o \ +$(SRC)/stdout.$(ARCH).o \ +$(SRC)/version.$(ARCH).o \ +$(SRC)/verbose.$(ARCH).o + +libs = \ +$(DESTLIB)/libbasiccmd.a \ +$(DESTLIB)/libshell.a \ +$(DESTLIB)/libdata.a + +pcontrol: $(BIN)/pcontrol.$(ARCH) +$(SRC)/pcontrol.$(ARCH).o : $(libs) +$(BIN)/pcontrol.$(ARCH) : $(cmds) $(funcs) + +$(cmds) $(funcs) : $(INC)/pcontrol.h + +install: $(DESTBIN)/pcontrol help + +help: clean-help cmd.basic.help pcontrol.help + +.PHONY: pcontrol + +include ../Makefile.Common Index: /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/PclientCommand.c =================================================================== --- /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/PclientCommand.c (revision 11539) +++ /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/PclientCommand.c (revision 11539) @@ -0,0 +1,52 @@ +# include "pcontrol.h" +# define PCLIENT_TIMEOUT 500 + +int PclientCommand (Host *host, char *command, char *response, IOBuffer *buffer) { + + int i; + int status; + char *line; + struct timespec request, remain; + + ASSERT (host != NULL, "host missing"); + ASSERT (buffer != NULL, "buffer missing"); + ASSERT (command != NULL, "command missing"); + ASSERT (response != NULL, "response missing"); + + /* avoid blocking on read, test every 100 usec, up to 50 msec */ + request.tv_sec = 0; + request.tv_nsec = 100000; + + // flush the stdout and stderr buffers here + ReadtoIOBuffer (buffer, host[0].stdout); + FlushIOBuffer (buffer); + ReadtoIOBuffer (buffer, host[0].stderr); + FlushIOBuffer (buffer); + + /* send command to client (adding on \n) */ + status = write_fmt (host[0].stdin, "%s\n", command); + + /* is pipe still open? */ + if ((status == -1) && (errno == EPIPE)) { + // gprint (GP_ERR, "pclient read gives pipe error for %s\n", command); + return (PCLIENT_DOWN); + } + + /* watch for response - wait up to 1 second */ + line = NULL; + status = -1; + for (i = 0; (i < PCLIENT_TIMEOUT) && (status != 0) && (line == NULL); i++) { + status = ReadtoIOBuffer (buffer, host[0].stdout); + line = memstr (buffer[0].buffer, response, buffer[0].Nbuffer); + if (status == -1) nanosleep (&request, &remain); + } + if (status == 0) { + // gprint (GP_ERR, "pclient read returns 0 for %s\n", command); + return (PCLIENT_DOWN); + } + if (status == -1) return (PCLIENT_HUNG); + /* gprint (GP_ERR, "buffer.buffer: %s\n", buffer[0].buffer); */ + return (PCLIENT_GOOD); +} + +/* memstr returns a view, not an allocated string : don't free */ Index: /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/QueueOps.c =================================================================== --- /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/QueueOps.c (revision 11539) +++ /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/QueueOps.c (revision 11539) @@ -0,0 +1,67 @@ +# include "pcontrol.h" + +/* get object from point in stack (negative == distance from end) */ +void *GetStack (Stack *stack, int where) { + + int i; + void *object; + + ASSERT (stack != NULL, "stack missing"); + + /* STACK_TOP == 0, STACK_BOTTOM == -1 */ + /* this code correctly handles the negative 'where' and Nobject == 0 */ + if (where < 0) where += stack[0].Nobject; + if (where < 0) return (NULL); + if (where >= stack[0].Nobject) return (NULL); + + object = stack[0].object[where]; + stack[0].Nobject --; + for (i = where; i < stack[0].Nobject; i++) { + stack[0].object[i] = stack[0].object[i+1]; + } + return (object); +} + +/* push object on top of stack */ +int PutStack (Stack *stack, int where, void *object) { + + int i; + + ASSERT (stack != NULL, "stack missing"); + + /* STACK_TOP == 0, STACK_BOTTOM == -1 */ + /* this code correctly handles the negative 'where' and Nobject == 0 */ + if (where < 0) where += stack[0].Nobject + 1; + if (where < 0) return (FALSE); + if (where > stack[0].Nobject) return (FALSE); + + /* extend stack as needed */ + if (stack[0].Nobject >= stack[0].NOBJECT) { + stack[0].NOBJECT += 100; + REALLOCATE (stack[0].object, void *, stack[0].NOBJECT); + } + + for (i = stack[0].Nobject; i > where; i--) { + stack[0].object[i] = stack[0].object[i-1]; + } + stack[0].object[where] = object; + stack[0].Nobject ++; + return (TRUE); +} + +/* allocate stack, setup with default values, allocate data */ +Stack *InitStack () { + + Stack *stack; + + ALLOCATE (stack, Stack, 1); + + stack[0].Nobject = 0; + stack[0].NOBJECT = 50; + ALLOCATE (stack[0].object, void *, stack[0].NOBJECT); + return (stack); +} + +/* these stacks are not super efficient, and should probably be replaced with linked lists, + but I find these easier to get my brain around +*/ Index: /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/ResetJob.c =================================================================== --- /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/ResetJob.c (revision 11539) +++ /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/ResetJob.c (revision 11539) @@ -0,0 +1,57 @@ +# include "pcontrol.h" + +int ResetJob (Job *job) { + + int status; + IOBuffer buffer; + Host *host; + + /** must have a valid host : if not, move to pending? **/ + ASSERT (job != NULL, "job missing"); + + host = (Host *) job[0].host; + ASSERT (job != NULL, "host missing"); + + InitIOBuffer (&buffer, 0x100); + + /* we have tried to reset the job; may not get status */ + job[0].Reset = TRUE; + + status = PclientCommand (host, "reset", PCLIENT_PROMPT, &buffer); + + /* check on success of pclient command */ + switch (status) { + case PCLIENT_DOWN: + /*** different behavior for ANYHOST, WANTHOST, NEEDHOST? ***/ + gprint (GP_ERR, "host %s is down\n", host[0].hostname); + HarvestHost (host[0].pid); + PutHost (host, PCONTROL_HOST_DOWN, STACK_BOTTOM); + FreeIOBuffer (&buffer); + return (FALSE); + + case PCLIENT_HUNG: + /*** should we consider a HUNG host DOWN? ***/ + gprint (GP_ERR, "host %s is not responding (ResetJob)\n", host[0].hostname); + FreeIOBuffer (&buffer); + return (FALSE); + + case PCLIENT_GOOD: + gprint (GP_ERR, "message received (ResetJob)\n"); + FreeIOBuffer (&buffer); + return (TRUE); + + default: + gprint (GP_ERR, "unknown status for pclient command: programming error\n"); + exit (1); + } + + gprint (GP_ERR, "programming error in GetJobOutput (should not reach here)\n"); + exit (1); +} + +/* if machine is down, return FALSE + this will place job back in BUSY state, + next check of job will catch state and + put machine down correctly + +*/ Index: /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/StackOps.c =================================================================== --- /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/StackOps.c (revision 11539) +++ /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/StackOps.c (revision 11539) @@ -0,0 +1,216 @@ +# include "pcontrol.h" + +/* these stacks are not super efficient, and should probably be replaced with linked lists, + but I find these easier to get my brain around. +*/ + +/* Stacks and thread locks: interacting with the Stacks needs to be thread-safe so that the user may + * perform operations which interact with the stacks at the same time that the background loops + * check the current status of the jobs and hosts in the different stacks. The simplest way in + * which the stacks are made thread safe is to lock them with a mutex before every interaction + */ + +# define DEBUG 0 + +void PrintStackInfo (Stack *stack, const char *func) { + + if (!DEBUG) return; + fprintf (stderr, "%s: %p ", func, stack); + fprintf (stderr, "objects: %p ", stack[0].object); + fprintf (stderr, "Nobjects: %d, NOBJECTS: %d\n", stack[0].Nobject, stack[0].NOBJECT); +} + +/* allocate stack, setup with default values, allocate data */ +Stack *InitStack () { + + Stack *stack; + + ALLOCATE (stack, Stack, 1); + + stack[0].Nobject = 0; + stack[0].NOBJECT = 50; + + ALLOCATE (stack[0].object, void *, stack[0].NOBJECT); + ALLOCATE (stack[0].name, char *, stack[0].NOBJECT); + ALLOCATE (stack[0].id, int, stack[0].NOBJECT); + +# ifdef THREADED + pthread_mutex_init (&stack[0].mutex, NULL); +# endif + return (stack); +} + +/* STACK_TOP == 0, STACK_BOTTOM == -1 */ +/* this code correctly handles the negative 'where' and Nobject == 0 */ + +/* push object on stack at given location */ +int PushStack (Stack *stack, int where, void *object, int id, char *name) { + + int i; + + PrintStackInfo (stack, __func__); + ASSERT (stack != NULL, "stack not set"); + LockStack (stack); + + if (where < 0) where += stack[0].Nobject + 1; + if (where < 0) { + UnlockStack (stack); + return (FALSE); + } + if (where > stack[0].Nobject) { + UnlockStack (stack); + return (FALSE); + } + + /* extend stack as needed */ + if (stack[0].Nobject >= stack[0].NOBJECT) { + stack[0].NOBJECT += 100; + REALLOCATE (stack[0].object, void *, stack[0].NOBJECT); + REALLOCATE (stack[0].name, char *, stack[0].NOBJECT); + REALLOCATE (stack[0].id, int, stack[0].NOBJECT); + } + + for (i = stack[0].Nobject; i > where; i--) { + stack[0].object[i] = stack[0].object[i-1]; + stack[0].name[i] = stack[0].name[i-1]; + stack[0].id[i] = stack[0].id[i-1]; + } + stack[0].object[where] = object; + stack[0].name[where] = name; + stack[0].id[where] = id; + stack[0].Nobject ++; + + UnlockStack (stack); + return (TRUE); +} + +/* get object from specified point in stack (negative == distance from end) */ +void *PullStackByLocation (Stack *stack, int where) { + + void *object; + + PrintStackInfo (stack, __func__); + ASSERT (stack != NULL, "stack not set"); + LockStack (stack); + + if (where < 0) where += stack[0].Nobject; + if (where < 0) { + UnlockStack (stack); + return (NULL); + } + if (where >= stack[0].Nobject) { + UnlockStack (stack); + return (NULL); + } + + object = stack[0].object[where]; + RemoveStackEntry (stack, where); + UnlockStack (stack); + return (object); +} + +/* get object from stack which matches name */ +void *PullStackByName (Stack *stack, char *name) { + + int i; + void *object; + + PrintStackInfo (stack, __func__); + ASSERT (stack != NULL, "stack not set"); + LockStack (stack); + + for (i = 0; i < stack[0].Nobject; i++) { + if (strcasecmp (stack[0].name[i], name)) continue; + + /* here is the element of interest */ + object = stack[0].object[i]; + RemoveStackEntry (stack, i); + UnlockStack (stack); + return (object); + } + UnlockStack (stack); + return (NULL); +} + +/* get object from point in stack (negative == distance from end) */ +void *PullStackByID (Stack *stack, int id) { + + int i; + void *object; + + PrintStackInfo (stack, __func__); + ASSERT (stack != NULL, "stack not set"); + LockStack (stack); + + for (i = 0; i < stack[0].Nobject; i++) { + if (stack[0].id[i] != id) continue; + + /* here is the element of interest */ + object = stack[0].object[i]; + RemoveStackEntry (stack, i); + UnlockStack (stack); + return (object); + } + UnlockStack (stack); + return (NULL); +} + +/* should only be called if you know where is a valid entry */ +int RemoveStackEntry (Stack *stack, int where) { + + int i; + + PrintStackInfo (stack, __func__); + ASSERT (stack != NULL, "stack not set"); + + if (where < 0) abort(); + if (where >= stack[0].Nobject) abort(); + if (stack[0].Nobject < 1) abort(); + + /* shift the remaining entries by one */ + /* XXX free associated memory */ + stack[0].Nobject --; + for (i = where; i < stack[0].Nobject; i++) { + stack[0].object[i] = stack[0].object[i+1]; + stack[0].name[i] = stack[0].name[i+1]; + stack[0].id[i] = stack[0].id[i+1]; + } + return (TRUE); +} + +/* should only be called if you manually lock the stack */ +void *RemoveStackByID (Stack *stack, int id) { + + int i; + void *object; + + PrintStackInfo (stack, __func__); + ASSERT (stack != NULL, "stack not set"); + for (i = 0; i < stack[0].Nobject; i++) { + if (stack[0].id[i] != id) continue; + + /* here is the element of interest */ + object = stack[0].object[i]; + RemoveStackEntry (stack, i); + return (object); + } + return (NULL); +} + +void LockStack (Stack *stack) { +# ifdef THREADED + PrintStackInfo (stack, __func__); + ASSERT (stack != NULL, "stack not set"); + pthread_mutex_lock (&stack[0].mutex); +# endif + return; +} + +void UnlockStack (Stack *stack) { +# ifdef THREADED + PrintStackInfo (stack, __func__); + ASSERT (stack != NULL, "stack not set"); + pthread_mutex_unlock (&stack[0].mutex); +# endif + return; +} Index: /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/StartHost.c =================================================================== --- /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/StartHost.c (revision 11539) +++ /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/StartHost.c (revision 11539) @@ -0,0 +1,47 @@ +# include "pcontrol.h" +# define RETRY_BASE 10.0 + +int StartHost (Host *host) { + + int pid; + int stdio[3]; + char command[64], shell[64]; + struct timeval now; + float delta; + + /* perhaps change the name of these config variables... */ + if (VarConfig ("COMMAND", "%s", command) == NULL) strcpy (command, "ssh"); + if (VarConfig ("SHELL", "%s", shell) == NULL) strcpy (shell, "pclient"); + + if (VerboseMode()) gprint (GP_ERR, "starting host within thread %d\n", pthread_self()); + + pid = rconnect (command, host[0].hostname, shell, stdio); + if (!pid) { + /** failure to start: extend retry period **/ + if (VerboseMode()) gprint (GP_ERR, "failure to start %s\n", host[0].hostname); + gettimeofday (&now, (void *) NULL); + if (ZTIME(host[0].nexttry) || ZTIME(host[0].lasttry)) { + /* reset retry period if either is zero */ + delta = RETRY_BASE; + } else { + delta = 2*DTIME (host[0].nexttry, host[0].lasttry); + } + host[0].nexttry.tv_sec = now.tv_sec + delta; + host[0].nexttry.tv_usec = now.tv_usec; + host[0].lasttry.tv_sec = now.tv_sec; + host[0].lasttry.tv_usec = now.tv_usec; + PutHost (host, PCONTROL_HOST_DOWN, STACK_BOTTOM); + return (FALSE); + } + host[0].nexttry.tv_sec = 0; + host[0].nexttry.tv_usec = 0; + host[0].lasttry.tv_sec = 0; + host[0].lasttry.tv_usec = 0; + + host[0].stdin = stdio[0]; + host[0].stdout = stdio[1]; + host[0].stderr = stdio[2]; + host[0].pid = pid; + PutHost (host, PCONTROL_HOST_IDLE, STACK_BOTTOM); + return (TRUE); +} Index: /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/StartJob.c =================================================================== --- /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/StartJob.c (revision 11539) +++ /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/StartJob.c (revision 11539) @@ -0,0 +1,102 @@ +# include "pcontrol.h" + +int StartJob (Job *job, Host *host) { + + int i, Nline, status; + char *line, *p; + IOBuffer buffer; + + InitIOBuffer (&buffer, 0x100); + + /* job must have assigned host */ + ASSERT (job != NULL, "missing job"); + ASSERT (host != NULL, "missing host"); + ASSERT (host == (Host *) job[0].host, "invalid host"); + ASSERT (job == (Job *) host[0].job, "invalid job"); + + /* construct command line : job arg0 arg1 ... argN\n */ + Nline = 10 + job[0].argc; + for (i = 0; i < job[0].argc; i++) { + Nline += strlen (job[0].argv[i]); + } + ALLOCATE (line, char, Nline); + bzero (line, Nline); + strcpy (line, "job"); + for (i = 0; i < job[0].argc; i++) { + strcat (line, " "); + strcat (line, job[0].argv[i]); + } + + status = PclientCommand (host, line, PCLIENT_PROMPT, &buffer); + free (line); + + /* check on success of pclient command */ + switch (status) { + case PCLIENT_DOWN: + if (VerboseMode()) gprint (GP_ERR, "host %s is down\n", host[0].hostname); + goto failure; + + case PCLIENT_HUNG: + // we need the job start to return a valid Job ID, + // give up on jobs which don't get started. + // XXX we are sensitive here to the time it takes pclient + // to fork the job. if this is slow, the client may appear to hang. + gprint (GP_ERR, "host %s is not responding (StartJob)\n", host[0].hostname); + if (VerboseMode()) gprint (GP_ERR, "host %s is not responding\n", host[0].hostname); + + // unlink host & job + job[0].host = NULL; + host[0].job = NULL; + if (job[0].realhost) free (job[0].realhost); + job[0].realhost = NULL; + PutHost (host, PCONTROL_HOST_DONE, STACK_BOTTOM); + PutJob (job, PCONTROL_JOB_PENDING, STACK_BOTTOM); + FreeIOBuffer (&buffer); + return (FALSE); + + case PCLIENT_GOOD: + if (VerboseMode()) gprint (GP_ERR, "message received (StartJob)\n"); + break; + + default: + ABORT ("unknown status for pclient command"); + } + + /* check on result of pclient command */ + p = memstr (buffer.buffer, "STATUS", buffer.Nbuffer); + ASSERT (p != NULL, "missing STATUS in pclient message"); + + sscanf (p, "%*s %d", &status); + switch (status) { + case -1: + if (VerboseMode()) gprint (GP_ERR, "error in pclient child\n"); + goto failure; + + case -2: + ABORT ("syntax error in pclient command"); + + case -3: + ABORT ("existing child on pclient"); + + default: + job[0].realhost = strcreate (host[0].hostname); + job[0].pid = status; + PutHost (host, PCONTROL_HOST_BUSY, STACK_BOTTOM); + PutJob (job, PCONTROL_JOB_BUSY, STACK_BOTTOM); + FreeIOBuffer (&buffer); + gettimeofday (&job[0].start, (void *) NULL); + return (TRUE); + } + /* we should never reach here */ + ABORT ("should not reach here (StartJob)"); + +failure: + // unlink host & job + job[0].host = NULL; + host[0].job = NULL; + HarvestHost (host[0].pid); + PutHost (host, PCONTROL_HOST_DOWN, STACK_BOTTOM); + PutJob (job, PCONTROL_JOB_PENDING, STACK_BOTTOM); + FreeIOBuffer (&buffer); + return (FALSE); +} Index: /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/StopHosts.c =================================================================== --- /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/StopHosts.c (revision 11539) +++ /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/StopHosts.c (revision 11539) @@ -0,0 +1,156 @@ +# include "pcontrol.h" + +void DownHost (Host *host) { + CLOSE (host[0].stdin); + CLOSE (host[0].stdout); + CLOSE (host[0].stderr); + host[0].job = NULL; + PutHost (host, PCONTROL_HOST_DOWN, STACK_BOTTOM); +} + +void OffHost (Host *host) { + CLOSE (host[0].stdin); + CLOSE (host[0].stdout); + CLOSE (host[0].stderr); + host[0].job = NULL; + PutHost (host, PCONTROL_HOST_OFF, STACK_BOTTOM); +} + +/* for use by shutdown: force machines which are up to go down + wait for a little while for the client thread to take care + of them +*/ + +int DownHosts () { + + int i, Nobject, Nwait; + Stack *stack; + Host *host; + + SetCheckPoint (); // ensure we can find the specified host + stack = GetHostStack (PCONTROL_HOST_IDLE); + ASSERT (stack != NULL, "stack missing"); + Nobject = stack[0].Nobject; + for (i = 0; i < Nobject; i++) { + host = PullStackByLocation (stack, STACK_TOP); + if (host == NULL) continue; + host[0].markoff = TRUE; + PutHost (host, PCONTROL_HOST_IDLE, STACK_BOTTOM); + } + + stack = GetHostStack (PCONTROL_HOST_BUSY); + ASSERT (stack != NULL, "stack missing"); + Nobject = stack[0].Nobject; + for (i = 0; i < Nobject; i++) { + host = PullStackByLocation (stack, STACK_TOP); + if (host == NULL) continue; + host[0].markoff = TRUE; + PutHost (host, PCONTROL_HOST_IDLE, STACK_BOTTOM); + } + ClearCheckPoint (); + + Nwait = 0; + stack = GetHostStack (PCONTROL_HOST_IDLE); + ASSERT (stack != NULL, "stack missing"); + + gprint (GP_ERR, "waiting for clients to exit"); + while ((Nwait < 15) && stack[0].Nobject) { + gprint (GP_ERR, "."); + usleep (100000); // wait for clients to exit + Nwait++; + } + gprint (GP_ERR, "\n"); + if (stack[0].Nobject) { + gprint (GP_ERR, "trouble shutting down all pclient instances: %d still alive\n", stack[0].Nobject); + } else { + gprint (GP_ERR, "done\n"); + } + return (TRUE); +} + +int StopHost (Host *host) { + + int status; + IOBuffer buffer; + + InitIOBuffer (&buffer, 0x100); + status = PclientCommand (host, "exit", "Goodbye", &buffer); + + /* check on success of pclient command */ + switch (status) { + case PCLIENT_DOWN: + break; + + case PCLIENT_HUNG: + gprint (GP_ERR, "host %s is not responding\n", host[0].hostname); + break; + + case PCLIENT_GOOD: + break; + + default: + gprint (GP_ERR, "unknown status for pclient command: programming error\n"); + exit (1); + } + HarvestHost (host[0].pid); + return (TRUE); +} + +/* the host is thought to be down; check for child exit status */ +int HarvestHost (int pid) { + + int i, result, waitstatus; + + gprint (GP_ERR, "harvesting within thread %d\n", pthread_self()); + gprint (GP_ERR, "child process %d is down, wait for exit status\n", pid); + + // Loop a few times waiting for child to exit + for (i = 0; i < 50; i++) { + result = waitpid (pid, &waitstatus, WNOHANG); + if ((result == -1) && (errno == ECHILD)) { + usleep (10000); // wait for child to exit + continue; + } else { + break; + } + } + 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"); + break; + case EINTR: + case EINVAL: + default: + perror ("unexpected error"); + ABORT ("(HarvestHost)"); + } + break; + + case 0: + gprint (GP_ERR, "child did not exit??"); + abort (); + /** put back in IDLE state? **/ + break; + + default: + if (result != pid) { + gprint (GP_ERR, "waitpid error: mis-matched PID (%d vs %d). programming error\n", result, pid); + exit (1); + } + + if (WIFEXITED(waitstatus)) { + if (VerboseMode()) gprint (GP_ERR, "child exited with status %d\n", WEXITSTATUS(waitstatus)); + } + if (WIFSIGNALED(waitstatus)) { + if (VerboseMode()) gprint (GP_ERR, "child crashed with status %d\n", WTERMSIG(waitstatus)); + } + if (WIFSTOPPED(waitstatus)) { + if (VerboseMode()) gprint (GP_ERR, "waitpid returns 'stopped': programming error\n"); + exit (1); + } + } + return (TRUE); +} Index: /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/check.c =================================================================== --- /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/check.c (revision 11539) +++ /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/check.c (revision 11539) @@ -0,0 +1,58 @@ +# include "pcontrol.h" + +int check (int argc, char **argv) { + + Job *job; + Host *host; + int JobID, HostID, StackID; + + if (argc != 3) { + gprint (GP_LOG, "USAGE: check job (JobID)\n"); + gprint (GP_LOG, "USAGE: check host (HostID)\n"); + return (FALSE); + } + + if (!strcasecmp (argv[1], "JOB")) { + JobID = atoi (argv[2]); + + SetCheckPoint (); // ensure the JOB is on one of the stacks + job = PullJobByID (JobID, &StackID); + if (job == NULL) { + gprint (GP_LOG, "job not found\n"); + ClearCheckPoint (); + return (FALSE); + } + gprint (GP_LOG, "STATUS %s\n", GetJobStackName(StackID)); + gprint (GP_LOG, "EXITST %d\n", job[0].exit_status); + gprint (GP_LOG, "STDOUT %d\n", job[0].stdout_size); + gprint (GP_LOG, "STDERR %d\n", job[0].stderr_size); + gprint (GP_LOG, "DTIME %lf\n", job[0].dtime); + if (job[0].realhost) { + gprint (GP_LOG, "HOSTNAME %s\n", job[0].realhost); + } else { + gprint (GP_LOG, "HOSTNAME NONE\n"); + } + PutJob (job, StackID, STACK_BOTTOM); + ClearCheckPoint (); + return (TRUE); + } + + if (!strcasecmp (argv[1], "HOST")) { + HostID = atoi (argv[2]); + + SetCheckPoint (); // ensure the HOST is on one of the stacks + host = PullHostByID (HostID, &StackID); + if (host == NULL) { + gprint (GP_LOG, "host not found\n"); + ClearCheckPoint (); + return (FALSE); + } + gprint (GP_LOG, "host %s\n", GetHostStackName(StackID)); + PutHost (host, StackID, STACK_BOTTOM); + ClearCheckPoint (); + return (TRUE); + } + + gprint (GP_LOG, "unknown item to check\n"); + return (FALSE); +} Index: /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/delete.c =================================================================== --- /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/delete.c (revision 11539) +++ /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/delete.c (revision 11539) @@ -0,0 +1,45 @@ +# include "pcontrol.h" + +int delete (int argc, char **argv) { + + Job *job; + 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 ? */ + + job = PullJobFromStackByID (PCONTROL_JOB_PENDING, JobID); + if (job != NULL) goto found; + + job = PullJobFromStackByID (PCONTROL_JOB_CRASH, JobID); + if (job != NULL) goto found; + + job = PullJobFromStackByID (PCONTROL_JOB_EXIT, JobID); + if (job != NULL) goto found; + + gprint (GP_ERR, "job %s not PENDING, CRASH, EXIT\n", argv[1]); + return (FALSE); + +found: + { + int j; + gprint (GP_LOG, "deleting job %s %d ", job[0].hostname, job[0].argc); + for (j = 0; j < job[0].argc; j++) { + gprint (GP_LOG, "%s ", job[0].argv[j]); + } + PrintID (GP_LOG, job[0].JobID); + gprint (GP_LOG, "\n"); + } + DelJob (job); + + return (TRUE); +} + +/**** at the moment, this function requires the job to be in the correct state + to be deleted. This should be changed to kill, then delete job if it is + not in the correct state +****/ Index: /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/help/host =================================================================== --- /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/help/host (revision 11539) +++ /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/help/host (revision 11539) @@ -0,0 +1,1 @@ +empty Index: /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/host.c =================================================================== --- /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/host.c (revision 11539) +++ /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/host.c (revision 11539) @@ -0,0 +1,85 @@ +# include "pcontrol.h" + +// we use CheckPoints in this function to prevent objects in flight from being missing. +int host (int argc, char **argv) { + + int StackID; + IDtype HostID; + Host *host; + + if (argc != 3) goto usage; + + if (!strcasecmp (argv[1], "ADD")) { + HostID = AddHost (argv[2]); + gprint (GP_LOG, "HostID: %d\n", (int) HostID); + return (TRUE); + } + if (!strcasecmp (argv[1], "ON")) { + host = PullHostFromStackByName (PCONTROL_HOST_OFF, argv[2]); + if (!host) { + gprint (GP_LOG, "host %s is not OFF\n", argv[2]); + return (FALSE); + } + host[0].markoff = FALSE; + DownHost (host); + return (TRUE); + } + if (!strcasecmp (argv[1], "RETRY")) { + // no need to use a check point [thief: CheckDownHost (DOWN->IDLE)] + host = PullHostFromStackByName (PCONTROL_HOST_DOWN, argv[2]); + if (!host) { + gprint (GP_LOG, "host %s is not DOWN\n", argv[2]); + return (FALSE); + } + /* reset time, place back on DOWN stack */ + host[0].nexttry.tv_sec = 0; + host[0].nexttry.tv_usec = 0; + host[0].lasttry.tv_sec = 0; + host[0].lasttry.tv_usec = 0; + PutHost (host, PCONTROL_HOST_DOWN, STACK_BOTTOM); + return (TRUE); + } + if (!strcasecmp (argv[1], "CHECK")) { + SetCheckPoint (); // ensure the host is on one of the stacks + host = PullHostByName (argv[2], &StackID); + if (host == NULL) { + gprint (GP_LOG, "host %s not found\n", argv[2]); + ClearCheckPoint (); + return (FALSE); + } + PutHost (host, StackID, STACK_BOTTOM); + ClearCheckPoint (); + + gprint (GP_LOG, "host %s is %s\n", argv[2], GetHostStackName (StackID)); + return (TRUE); + } + if (!strcasecmp (argv[1], "OFF")) { + SetCheckPoint (); // ensure we can find the specified host + host = PullHostByName (argv[2], &StackID); + if (host == NULL) { + gprint (GP_LOG, "host %s not found\n", argv[2]); + ClearCheckPoint (); + return (FALSE); + } + host[0].markoff = TRUE; + PutHost (host, StackID, STACK_BOTTOM); + ClearCheckPoint (); + return (TRUE); + } + + if (!strcasecmp (argv[1], "DELETE")) { + // a check point is not required: no possible thief + host = PullHostFromStackByName (PCONTROL_HOST_OFF, argv[2]); + if (!host) { + gprint (GP_LOG, "host %s is not OFF\n", argv[2]); + return (FALSE); + } + DelHost (host); + return (TRUE); + } + +usage: + gprint (GP_LOG, "USAGE: host (command) (hostname)\n"); + gprint (GP_ERR, " valid commands: add, on, retry, check, off, delete\n"); + return (FALSE); +} Index: /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/hoststack.c =================================================================== --- /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/hoststack.c (revision 11539) +++ /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/hoststack.c (revision 11539) @@ -0,0 +1,34 @@ +# include "pcontrol.h" + +int hoststack (int argc, char **argv) { + + int i; + Stack *stack; + Host *host; + + if (argc != 2) { + gprint (GP_ERR, "USAGE: hoststack (hoststack)\n"); + gprint (GP_ERR, " (hoststack) : idle, busy, done, down, off\n"); + return (FALSE); + } + + /* select hoststack */ + stack = GetHostStackByName (argv[1]); + if (stack == NULL) { + gprint (GP_ERR, "hoststack not found\n"); + return (FALSE); + } + + /* print list */ + LockStack (stack); + gprint (GP_LOG, "Nhosts: %d\n", stack[0].Nobject); + for (i = 0; i < stack[0].Nobject; i++) { + host = stack[0].object[i]; + gprint (GP_LOG, "%lld %s\n", host[0].HostID, host[0].hostname); + } + UnlockStack (stack); + + return (TRUE); +} + +// Safe with PTHREAD_MUTEX_INITIALIZER lock Index: /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/init.c =================================================================== --- /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/init.c (revision 11539) +++ /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/init.c (revision 11539) @@ -0,0 +1,47 @@ +# include "pcontrol.h" + +int check PROTO((int, char **)); +int delete PROTO((int, char **)); +int host PROTO((int, char **)); +int hoststack PROTO((int, char **)); +int job PROTO((int, char **)); +int jobstack PROTO((int, char **)); +int kill_pc PROTO((int, char **)); +int status PROTO((int, char **)); +int run PROTO((int, char **)); +int stderr_pc PROTO((int, char **)); +int stdout_pc PROTO((int, char **)); +int verbose PROTO((int, char **)); +int version PROTO((int, char **)); + +// pulse is only available in the un-threaded version +int pulse PROTO((int, char **)); + +static Command cmds[] = { + {"host", host, "add / delete / modify host"}, + {"hoststack", hoststack, "list hosts for a single stack"}, + {"status", status, "get system status"}, + {"stop", run, "stop controller processing"}, + {"run", run, "set controller runlevel"}, + {"verbose", verbose, "set the verbose mode for job"}, + {"version", version, "show version information"}, + {"job", job, "add job"}, + {"jobstack", jobstack, "list jobs for a single stack"}, + {"check", check, "get job or host status"}, + {"delete", delete, "delete job"}, + {"kill", kill_pc, "kill job"}, + {"stderr", stderr_pc, "get stderr buffer for job"}, + {"stdout", stdout_pc, "get stdout buffer for job"}, +# ifndef THREADED + {"pulse", pulse, "set system pulse"}, +# endif +}; + +void InitPcontrol () { + + int i; + + for (i = 0; i < sizeof (cmds) / sizeof (Command); i++) { + AddCommand (&cmds[i]); + } +} Index: /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/job.c =================================================================== --- /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/job.c (revision 11539) +++ /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/job.c (revision 11539) @@ -0,0 +1,52 @@ +# include "pcontrol.h" + +int job (int argc, char **argv) { + + char *Host, **targv; + int i, N, Mode, targc, Timeout; + IDtype JobID; + + if (get_argument (argc, argv, "-host") && get_argument (argc, argv, "+host")) { + gprint (GP_ERR, "ERROR: -host and +host are incompatible\n"); + return (FALSE); + } + + 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"))) { + 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", (int) JobID); + return (TRUE); +} Index: /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/jobstack.c =================================================================== --- /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/jobstack.c (revision 11539) +++ /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/jobstack.c (revision 11539) @@ -0,0 +1,36 @@ +# include "pcontrol.h" + +int jobstack (int argc, char **argv) { + + int i; + Stack *stack; + Job *job; + + if (argc != 2) { + gprint (GP_ERR, "USAGE: jobstack (jobstack)\n"); + gprint (GP_ERR, " (jobstack) : pending, busy, exit, crash, hung, done\n"); + return (FALSE); + } + + /* select jobstack */ + stack = GetJobStackByName (argv[1]); + if (stack == NULL) { + gprint (GP_ERR, "jobstack not found\n"); + return (FALSE); + } + + /* print list */ + LockStack (stack); + gprint (GP_LOG, "Njobs: %d\n", stack[0].Nobject); + for (i = 0; i < stack[0].Nobject; i++) { + job = stack[0].object[i]; + /* PrintID (GP_LOG, job[0].JobID); */ + gprint (GP_LOG, "%lld ", job[0].JobID); + gprint (GP_LOG, "%s %s\n", job[0].argv[0], job[0].hostname); + } + UnlockStack (stack); + + return (TRUE); +} + +// Safe with PTHREAD_MUTEX_INITIALIZER lock Index: /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/kill.c =================================================================== --- /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/kill.c (revision 11539) +++ /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/kill.c (revision 11539) @@ -0,0 +1,24 @@ +# include "pcontrol.h" + +int kill_pc (int argc, char **argv) { + + Job *job; + int JobID; + + if (argc < 2) { + gprint (GP_ERR, "USAGE: kill (JobID)\n"); + return (FALSE); + } + JobID = atoi (argv[1]); + + /* XXX this function should only fail if a process is hung */ + job = PullJobFromStackByID (PCONTROL_JOB_BUSY, JobID); + if (job == NULL) { + gprint (GP_ERR, "job %s not BUSY\n", argv[1]); + /* make output message more readable by scheduler */ + return (FALSE); + } + + PutJob (job, PCONTROL_JOB_KILL, STACK_BOTTOM); + return (TRUE); +} Index: /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/pcontrol.c =================================================================== --- /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/pcontrol.c (revision 11539) +++ /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/pcontrol.c (revision 11539) @@ -0,0 +1,92 @@ +# 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" + +/* program-dependent initialization */ +void program_init (int *argc, char **argv) { + +# ifdef THREADED + pthread_t clientsThread; +# endif + + auto_break = TRUE; + + /* load the commands used by this implementation */ + InitBasic (); + InitPcontrol (); + + InitJobStacks (); + InitHostStacks (); + + /* set global signal masks (these apply to all threads launched below) */ + // signal (SIGPIPE, gotsignal); + signal (SIGTSTP, gotsignal); + signal (SIGTTIN, gotsignal); + + rl_readline_name = opihi_name; + rl_attempted_completion_function = command_completer; +# ifdef THREADED + SetRunLevel (PCONTROL_RUN_ALL); + pthread_create (&clientsThread, NULL, &CheckSystem_Threaded, NULL); + rl_event_hook = NULL; + rl_set_keyboard_input_timeout (1000); +# else + rl_event_hook = CheckSystem; + rl_set_keyboard_input_timeout (1000); +# endif + + // set_str_variable ("HISTORY", opihi_history); + set_str_variable ("PROMPT", opihi_prompt); + set_str_variable ("RCFILE", opihi_rcfile); + +# ifdef DATADIR_DEFAULT + { + char *datadir; + char *helpdir; + char *modules; + datadir = MACRO_NAME(DATADIR_DEFAULT); + /* set_str_variable ("DATADIR", datadir); */ + ALLOCATE (helpdir, char, strlen(datadir) + strlen("/help") + 2); + sprintf (helpdir, "%s/help", datadir); + set_str_variable ("HELPDIR", helpdir); + free (helpdir); + ALLOCATE (modules, char, strlen(datadir) + strlen("/modules") + 2); + sprintf (modules, "%s/modules", datadir); + set_str_variable ("MODULES:0", modules); + set_int_variable ("MODULES:n", 1); + free (modules); + } +# endif + + /* ignore the history file. to change this, see, eg, mana.c */ + return; +} + +/* standard welcome message */ +void welcome () { + gprint (GP_ERR, "\n"); + gprint (GP_ERR, "Welcome to %s - %s\n\n", opihi_name, opihi_description); +} + +/* add program-dependent exit functions here */ +void cleanup () { + DownHosts (); + return; +} + +void gotsignal (int signum) { + // this message is lost if we are connected to a pantasks + fprintf (stderr, "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/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/pulse.c =================================================================== --- /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/pulse.c (revision 11539) +++ /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/pulse.c (revision 11539) @@ -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/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/rconnect.c =================================================================== --- /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/rconnect.c (revision 11539) +++ /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/rconnect.c (revision 11539) @@ -0,0 +1,148 @@ +# include "pcontrol.h" + +/* connection can take a while, allow up to 2 sec */ +# define CONNECT_TIMEOUT 200 + +/* 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; + int result, waitstatus; + pid_t pid; + char *p; + char **argv; + IOBuffer buffer; + struct timespec request, remain; + + ASSERT (command != NULL, "command is NULL"); + ASSERT (hostname != NULL, "hostname is NULL"); + ASSERT (shell != NULL, "shell is NULL"); + ASSERT (stdio != NULL, "stdio is NULL"); + + 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 */ + if (VerboseMode()) gprint (GP_ERR, "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); + gprint (GP_ERR, "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 */ + status = write_fmt (stdin_fd[1], "echo CONNECTED\n"); + if ((status == -1) && (errno == EPIPE)) goto connect_error; + + /* try to get evidence connection is alive - wait upto a few seconds */ + 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 client to be connected + } + if (status == 0) goto connect_error; + if (status == -1) goto connect_error; + if (VerboseMode()) gprint (GP_ERR, "%d cycles to connect\n", i); + FreeIOBuffer (&buffer); + + if (VerboseMode()) gprint (GP_ERR, "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; + +connect_error: + if (VerboseMode()) gprint (GP_ERR, "error while connecting\n"); + + /* avoid blocking on waitpid, test every 100 usec, up to 50 msec */ + request.tv_sec = 0; + request.tv_nsec = 100000; + + /* harvest the child process: kill & wait (< 100 ms) for exit */ + kill (pid, SIGKILL); + result = waitpid (pid, &waitstatus, WNOHANG); + for (i = 0; (i < 50) && (result == 0); i++) { + nanosleep (&request, &remain); + result = waitpid (pid, &waitstatus, WNOHANG); + } + + if ((result == -1) && (errno != ECHILD)) { + gprint (GP_ERR, "unexpected error from waitpid (%d): programming error\n", errno); + exit (1); + } + if (result == 0) { + if (VerboseMode()) gprint (GP_ERR, "child did not exit (rconnect)??"); + } + if (result > 0) { + if (result != pid) { + gprint (GP_ERR, "waitpid error: mis-matched PID (%d vs %d). programming error\n", result, pid); + exit (1); + } + if (WIFSTOPPED(waitstatus)) { + gprint (GP_ERR, "waitpid returns 'stopped': programming error\n"); + exit (1); + } + } + +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); +} Index: /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/run.c =================================================================== --- /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/run.c (revision 11539) +++ /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/run.c (revision 11539) @@ -0,0 +1,66 @@ +# include "pcontrol.h" + +int run (int argc, char **argv) { + + RunLevels level; + + if ((argc > 2) || + get_argument (argc, argv, "help") || + get_argument (argc, argv, "-h") || + get_argument (argc, argv, "--help")) + { + if (!strcmp (argv[0], "run")) { + gprint (GP_ERR, "USAGE: run [level]\n"); + gprint (GP_ERR, " allowed levels:\n"); + gprint (GP_ERR, " all (default) : manage machines, spawn jobs, harvest jobs\n"); + gprint (GP_ERR, " reap : manage machines, harvest jobs\n"); + gprint (GP_ERR, " hosts : manage machines, not jobs\n"); + gprint (GP_ERR, " none : all stop\n"); + } else { + gprint (GP_ERR, "USAGE: stop (immediate processing halt)\n"); + } + return (FALSE); + } + + level = PCONTROL_RUN_UNKNOWN; + if (argc == 1) { + if (!strcasecmp (argv[0], "run")) level = PCONTROL_RUN_ALL; + if (!strcasecmp (argv[0], "stop")) level = PCONTROL_RUN_NONE; + } else { + if (!strcasecmp (argv[1], "all")) level = PCONTROL_RUN_ALL; + if (!strcasecmp (argv[1], "reap")) level = PCONTROL_RUN_REAP; + if (!strcasecmp (argv[1], "host")) level = PCONTROL_RUN_HOSTS; + if (!strcasecmp (argv[1], "hosts")) level = PCONTROL_RUN_HOSTS; + if (!strcasecmp (argv[1], "none")) level = PCONTROL_RUN_NONE; + } + + if (level == PCONTROL_RUN_UNKNOWN) { + gprint (GP_ERR, " unknown run level %s\n", argv[1]); + gprint (GP_ERR, " allowed levels:\n"); + gprint (GP_ERR, " all (default) : manage machines, spawn jobs, harvest jobs\n"); + gprint (GP_ERR, " reap : manage machines, harvest jobs\n"); + gprint (GP_ERR, " hosts : manage machines, not jobs\n"); + gprint (GP_ERR, " none : all stop\n"); + return (FALSE); + } + +# ifdef THREADED + SetRunLevel (level); +# else + if (level == PCONTROL_RUN_NONE) { + rl_event_hook = NULL; + } else { + rl_event_hook = CheckSystem; + } +# endif + + return (TRUE); +} + +/* + run levels: + all (manage machines, spawn jobs, harvest jobs) + reap (manage machines, harvest jobs) + hosts (manage machines, not jobs) + none (all stop) +*/ Index: /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/status.c =================================================================== --- /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/status.c (revision 11539) +++ /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/status.c (revision 11539) @@ -0,0 +1,74 @@ +# include "pcontrol.h" + +int PrintJobStack (int Nstack); +int PrintHostStack (int Nstack); + +int status (int argc, char **argv) { + + PrintJobStack (PCONTROL_JOB_PENDING); + PrintJobStack (PCONTROL_JOB_BUSY); + PrintJobStack (PCONTROL_JOB_DONE); + PrintJobStack (PCONTROL_JOB_KILL); + PrintJobStack (PCONTROL_JOB_EXIT); + PrintJobStack (PCONTROL_JOB_CRASH); + + PrintHostStack (PCONTROL_HOST_OFF); + PrintHostStack (PCONTROL_HOST_DOWN); + PrintHostStack (PCONTROL_HOST_IDLE); + PrintHostStack (PCONTROL_HOST_BUSY); + PrintHostStack (PCONTROL_HOST_DONE); + return (TRUE); +} + +int PrintJobStack (int Nstack) { + + int i, j, Nobject; + Stack *stack; + Job *job; + + stack = GetJobStack (Nstack); + ASSERT (stack != NULL, "programming error"); + + LockStack (stack); + Nobject = stack[0].Nobject; + gprint (GP_LOG, "job stack %s: %d objects\n", GetJobStackName(Nstack), Nobject); + + for (i = 0; i < Nobject; i++) { + job = stack[0].object[i]; + ASSERT (job != NULL, "programming error"); + gprint (GP_LOG, "%d %s %d ", i, job[0].hostname, job[0].argc); + for (j = 0; j < job[0].argc; j++) { + gprint (GP_LOG, "%s ", job[0].argv[j]); + } + PrintID (GP_LOG, job[0].JobID); + gprint (GP_LOG, "\n"); + } + UnlockStack (stack); + + return (TRUE); +} + +int PrintHostStack (int Nstack) { + + int i, Nobject; + Stack *stack; + Host *host; + + stack = GetHostStack (Nstack); + + LockStack (stack); + Nobject = stack[0].Nobject; + gprint (GP_LOG, "host stack %s: %d objects\n", GetHostStackName(Nstack), Nobject); + + for (i = 0; i < Nobject; i++) { + host = stack[0].object[i]; + gprint (GP_LOG, "%d %s ", i, host[0].hostname); + PrintID (GP_LOG, host[0].HostID); + gprint (GP_LOG, "\n"); + } + UnlockStack (stack); + + return (TRUE); +} + +// Safe with PTHREAD_MUTEX_INITIALIZER lock Index: /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/stdout.c =================================================================== --- /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/stdout.c (revision 11539) +++ /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/stdout.c (revision 11539) @@ -0,0 +1,71 @@ +# include "pcontrol.h" + +int stdout_pc (int argc, char **argv) { + + int JobID, StackID; + Job *job; + IOBuffer *buffer; + + if (argc != 2) { + gprint (GP_ERR, "USAGE: stdout (JobID)\n"); + gprint (GP_LOG, "STATUS %d\n", -1); + return (FALSE); + } + + /* find Job of interest (must be EXIT or CRASH) */ + JobID = atoi (argv[1]); + + StackID = PCONTROL_JOB_EXIT; + job = PullJobFromStackByID (StackID, JobID); + if (job != NULL) goto found_stdout; + + StackID = PCONTROL_JOB_CRASH; + job = PullJobFromStackByID (StackID, JobID); + if (job != NULL) goto found_stdout; + + gprint (GP_ERR, "job not found in EXIT or CRASH\n"); + gprint (GP_LOG, "STATUS %d\n", -2); + return (FALSE); + +found_stdout: + buffer = &job[0].stdout; + fwrite (buffer[0].buffer, 1, buffer[0].Nbuffer, stdout); + gprint (GP_LOG, "STATUS %d\n", 0); + PutJob (job, StackID, STACK_BOTTOM); + return (TRUE); +} + +int stderr_pc (int argc, char **argv) { + + int JobID, StackID; + Job *job; + IOBuffer *buffer; + + if (argc != 2) { + gprint (GP_ERR, "USAGE: stderr (JobID)\n"); + gprint (GP_LOG, "STATUS %d\n", -1); + return (FALSE); + } + + /* find Job of interest (must be EXIT or CRASH) */ + JobID = atoi (argv[1]); + + StackID = PCONTROL_JOB_EXIT; + job = PullJobFromStackByID (StackID, JobID); + if (job != NULL) goto found_stderr; + + StackID = PCONTROL_JOB_CRASH; + job = PullJobFromStackByID (StackID, JobID); + if (job != NULL) goto found_stderr; + + gprint (GP_ERR, "job not found in EXIT or CRASH\n"); + gprint (GP_LOG, "STATUS %d\n", -2); + return (FALSE); + +found_stderr: + buffer = &job[0].stderr; + fwrite (buffer[0].buffer, 1, buffer[0].Nbuffer, stdout); + gprint (GP_LOG, "STATUS %d\n", 0); + PutJob (job, StackID, STACK_BOTTOM); + return (TRUE); +} Index: /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/stop.c =================================================================== --- /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/stop.c (revision 11539) +++ /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/stop.c (revision 11539) @@ -0,0 +1,17 @@ +# include "pcontrol.h" + +int stop (int argc, char **argv) { + + if (argc != 1) { + gprint (GP_ERR, "USAGE: stop\n"); + return (FALSE); + } + +# ifdef THREADED + SetRunSystem (FALSE); +# else + rl_event_hook = NULL; +# endif + + return (TRUE); +} Index: /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/verbose.c =================================================================== --- /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/verbose.c (revision 11539) +++ /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/verbose.c (revision 11539) @@ -0,0 +1,37 @@ +# include "pcontrol.h" + +static int VERBOSE = FALSE; + +int verbose (int argc, char **argv) { + + if (argc == 1) { + if (VERBOSE) { + gprint (GP_ERR, "verbose mode ON\n"); + } else { + gprint (GP_ERR, "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/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/version.c =================================================================== --- /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/version.c (revision 11539) +++ /tags/pcontrol-1-0-0/Ohana/src/opihi/pcontrol/version.c (revision 11539) @@ -0,0 +1,17 @@ +# include "pcontrol.h" +static char *name = "$Name: not supported by cvs2svn $"; + +int version (int argc, char **argv) { + + char *tmp; + + gprint (GP_LOG, "\n"); + gprint (GP_LOG, "pcontrol 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); +}