Index: /trunk/Ohana/src/opihi/include/pantasks.h =================================================================== --- /trunk/Ohana/src/opihi/include/pantasks.h (revision 12403) +++ /trunk/Ohana/src/opihi/include/pantasks.h (revision 12404) @@ -196,6 +196,6 @@ void FreeJob (Job *job); -int CheckJobs (); -int CheckTasks (); +float CheckJobs (); +float CheckTasks (); int CheckSystem (); int CheckController (); @@ -225,5 +225,4 @@ int KillControllerJob (Job *job); int CheckControllerStatus (); -int TestElapsedCheck (); void gotsignal (int signum); int client_shell (int argc, char **argv); Index: /trunk/Ohana/src/opihi/lib.shell/multicommand.c =================================================================== --- /trunk/Ohana/src/opihi/lib.shell/multicommand.c (revision 12403) +++ /trunk/Ohana/src/opihi/lib.shell/multicommand.c (revision 12404) @@ -3,4 +3,5 @@ static int server = 0; +// XXX this is rather pantasks-specific... void multicommand_InitServer () { Index: /trunk/Ohana/src/opihi/pantasks/CheckController.c =================================================================== --- /trunk/Ohana/src/opihi/pantasks/CheckController.c (revision 12403) +++ /trunk/Ohana/src/opihi/pantasks/CheckController.c (revision 12404) @@ -37,5 +37,5 @@ p = buffer.buffer; - for (i = 0; (i < Njobs) && !TestElapsedCheck(); i++) { + for (i = 0; i < Njobs; i++) { q = strchr (p, '\n'); if (q == NULL) { @@ -56,7 +56,4 @@ if (VerboseMode()) gprint (GP_ERR, "clear %d exit jobs %f\n", i, TimerElapsed(TRUE)); - if (TestElapsedCheck()) goto finish; - /* this will prevent us from ever checking crashed jobs... */ - /*** check CRASH jobs ***/ FlushIOBuffer (&buffer); @@ -73,5 +70,5 @@ p = buffer.buffer; - for (i = 0; (i < Njobs) && !TestElapsedCheck(); i++) { + for (i = 0; i < Njobs; i++) { q = strchr (p, '\n'); if (q == NULL) { Index: /trunk/Ohana/src/opihi/pantasks/CheckJobs.c =================================================================== --- /trunk/Ohana/src/opihi/pantasks/CheckJobs.c (revision 12403) +++ /trunk/Ohana/src/opihi/pantasks/CheckJobs.c (revision 12404) @@ -1,5 +1,5 @@ # include "pantasks.h" -int CheckJobs () { +float CheckJobs () { FILE *f; @@ -10,7 +10,11 @@ char varname[64]; Queue *queue; + float time_running, next_timeout; // int Ncheck; // Ncheck = 0; + + // actual maximum delay is controlled in job_threads.c + next_timeout = 1.0; /** test all jobs: ready to test? finished? **/ @@ -21,5 +25,11 @@ /* check poll period (ready to ask for status?) */ - if (GetTaskTimer(job[0].last, FALSE) < task[0].poll_period) continue; + time_running = GetTaskTimer(job[0].last, FALSE); + // fprintf (stderr, "next: %f, poll: %f, run: %f\n", next_timeout, task[0].poll_period, time_running); + if (time_running < task[0].poll_period) { + next_timeout = MIN (next_timeout, task[0].poll_period - time_running); + continue; + } + next_timeout = 0.0; /* check current status */ @@ -139,5 +149,9 @@ */ if (job[0].mode == JOB_LOCAL) { - if (GetTaskTimer(job[0].start, FALSE) < task[0].timeout_period) continue; + if (GetTaskTimer(job[0].start, FALSE) < task[0].timeout_period) { + /* reset polling clock */ + SetTaskTimer (&job[0].last); + continue; + } if (VerboseMode()) gprint (GP_LOG, "timeout on %s\n", task[0].name); @@ -180,11 +194,7 @@ /* reset polling clock */ SetTaskTimer (&job[0].last); - if (TestElapsedCheck()) { - // fprintf (stderr, "check %d jobs\n", Ncheck); - return (TRUE); - } } // fprintf (stderr, "check %d jobs\n", Ncheck); - return (TRUE); + return (next_timeout); } Index: unk/Ohana/src/opihi/pantasks/CheckSystem.c =================================================================== --- /trunk/Ohana/src/opihi/pantasks/CheckSystem.c (revision 12403) +++ (revision ) @@ -1,45 +1,0 @@ -# include "pantasks.h" - -# define MAX_DELAY 0.1 -struct timeval start, last; -static int Ncheck = 0; - -int CheckSystem () { - - gettimeofday (&start, (void *) NULL); - - if (Ncheck < 100) { - CheckTasks (); - CheckJobs (); - Ncheck ++; - } else { - CheckController (); - CheckControllerOutput (); - Ncheck = 0; - } - /* ohana_memcheck (FALSE); */ - return (TRUE); -} - -int TestElapsedCheck () { - - struct timeval stop; - float dtime; - - gettimeofday (&stop, (void *) NULL); - dtime = DTIME (stop, start); - if (dtime > MAX_DELAY) return (TRUE); - return (FALSE); -} - - /* do we really need to check the controller on every call? - or perhaps alternate? - */ - -void SerialThreadLock () { - return; -} - -void SerialThreadUnlock () { - return; -} Index: /trunk/Ohana/src/opihi/pantasks/CheckTasks.c =================================================================== --- /trunk/Ohana/src/opihi/pantasks/CheckTasks.c (revision 12403) +++ /trunk/Ohana/src/opihi/pantasks/CheckTasks.c (revision 12404) @@ -1,36 +1,49 @@ # include "pantasks.h" -int CheckTasks () { +float CheckTasks () { Job *job; Task *task; int status; + float time_running, next_timeout; // struct timeval now; + + // actual maximum delay is controlled in job_threads.c + next_timeout = 1.0; /** test all tasks: ready to test? ready to run? **/ while ((task = NextTask ()) != NULL) { - if (!task[0].active) continue; + /*** test for all reasons we should skip this task ***/ - /* ready to test? : check exec period */ - if (GetTaskTimer(task[0].last, FALSE) < task[0].exec_period) continue; - - /* need to check if the current time is within valid/invalid periods */ + /* task has been de-activated by the user */ + if (!task[0].active) { + continue; + } + /* current time is not within valid/invalid periods */ if (!CheckTimeRanges (task[0].ranges, task[0].Nranges)) { - gettimeofday (&task[0].last, (void *) NULL); - continue; + continue; } + /* all allowed tasks have been run in this period */ if (task[0].Nmax && (task[0].Njobs >= task[0].Nmax)) { - gettimeofday (&task[0].last, (void *) NULL); - continue; + continue; } + /* too many outstanding jobs */ if (task[0].NpendingMax && (task[0].Npending >= task[0].NpendingMax)) { - // fprintf (stderr, "npending: %d, max npending: %d\n", task[0].Npending, task[0].NpendingMax); - gettimeofday (&task[0].last, (void *) NULL); - continue; + continue; } - // gettimeofday (&now, (void *) NULL); - // fprintf (stderr, "t0: %d %6d - \n", now.tv_sec, now.tv_usec); + /* ready to test? : check time since last exec */ + time_running = GetTaskTimer(task[0].last, FALSE); + if (time_running < task[0].exec_period) { + // is we are to ready to run, set time to timeout, if shortest of all tasks + next_timeout = MIN (next_timeout, task[0].exec_period - time_running); + continue; + } + + /* ready to try running the task : reset the timer */ + next_timeout = 0.0; + gettimeofday (&task[0].last, (void *) NULL); + // XXX here is where I should add a fuzz factor (fraction of exec_period) /* ready to run? : run task.exec macro */ @@ -38,6 +51,5 @@ status = exec_loop (task[0].exec); if (!status) { - gettimeofday (&task[0].last, (void *) NULL); - continue; + continue; } } @@ -48,6 +60,5 @@ /* check if there are errors with this task */ if (!ValidateTask (task, TRUE)) { - gettimeofday (&task[0].last, (void *) NULL); - continue; + continue; } @@ -67,6 +78,5 @@ // task[0].last.tv_sec, task[0].last.tv_usec); - /* reset timer on task (don't do this if Create/Submit fails) (why not??) */ - gettimeofday (&task[0].last, (void *) NULL); + /* increment job counters */ task[0].Njobs ++; task[0].Npending ++; @@ -77,7 +87,5 @@ /* increment Nrun for inclusive ranges with Nmax */ BumpTimeRanges (task[0].ranges, task[0].Nranges); - - if (TestElapsedCheck()) return (TRUE); } - return (TRUE); + return (next_timeout); } Index: unk/Ohana/src/opihi/pantasks/RunScheduler.c =================================================================== --- /trunk/Ohana/src/opihi/pantasks/RunScheduler.c (revision 12403) +++ (revision ) @@ -1,85 +1,0 @@ -# include "pantasks.h" - -/* the RunScheduler thread should continuously loop over: - - - tasks - - jobs - - occasionally, it should: - - check controller state - - check for new input files to be loaded - - alternatively, we could have separate threads for each of these - tasks and simply have them each grab a mutex before they go: - - CheckTasks: - - grab mutex - - loop over all tasks - - release mutex - - nanosleep? - - CheckJobs: - - grab mutex - - loop over all jobs - - release mutex - - nanosleep? - - CheckController: - - grab mutex - - check status - - check output - - release mutex - - usleep 10000 - - CheckInputs - - grab mutex - - check for inputs on list - - load inputs - - release mutex - - usleep 200000 - - this strategy seems fairly clean, with the somewhat odd design feature - that all of these 'body' threads are force to run in serial, not parallel, - by use of the mutexes. It is somewhat cleaner programming. The only - restriction on the commands allowed to the clients is that they must not - be allowed to change any of the pantasks internal state variables (variables, - macros, lists, etc). This is solved by restricting the clients to output - message functions only (reporting functions). - -*/ - -# define MAX_DELAY 0.1 -struct timeval start, last; -static int Ncheck = 0; - -int CheckSystem () { - - gettimeofday (&start, (void *) NULL); - - if (Ncheck < 100) { - CheckTasks (); - CheckJobs (); - Ncheck ++; - } else { - CheckController (); - CheckControllerOutput (); - Ncheck = 0; - } - /* ohana_memcheck (FALSE); */ - return (TRUE); -} - -int TestElapsedCheck () { - - struct timeval stop; - float dtime; - - gettimeofday (&stop, (void *) NULL); - dtime = DTIME (stop, start); - if (dtime > MAX_DELAY) return (TRUE); - return (FALSE); -} - - /* do we really need to check the controller on every call? - or perhaps alternate? - */ Index: unk/Ohana/src/opihi/pantasks/connect_to_server.c =================================================================== --- /trunk/Ohana/src/opihi/pantasks/connect_to_server.c (revision 12403) +++ (revision ) @@ -1,16 +1,0 @@ -# include "pantasks.h" - -int connect_to_server () { - - char hostname[256]; - - ScanConfig (GlobalConfig, "PANTASKS_SERVER", "%s", 0, hostname); - - server = GetClientSocket (hostname); - - /* here we can perform the security handshaking */ - SendCommand (BindSocket, strlen(PASSWORD), PASSWORD); - - return (server); - -} Index: /trunk/Ohana/src/opihi/pantasks/job_threads.c =================================================================== --- /trunk/Ohana/src/opihi/pantasks/job_threads.c (revision 12403) +++ /trunk/Ohana/src/opihi/pantasks/job_threads.c (revision 12404) @@ -14,4 +14,6 @@ void *CheckJobsThread (void *data) { + float next_timeout; + gprintInit (); // each thread needs to init the printing system while (1) { @@ -25,9 +27,12 @@ // one run of the task checker SerialThreadLock (); - CheckJobs (); + next_timeout = CheckJobs (); SerialThreadUnlock (); if (VerboseMode() == 2) fprintf (stderr, "J"); - // fprintf (stderr, "J"); - // usleep (10000); // allow other threads a chance to run + + next_timeout = MIN (next_timeout, 0.1); + if (next_timeout > 0.001) { + usleep ((int)(1000000*next_timeout)); // allow other threads a chance to run + } } } Index: unk/Ohana/src/opihi/pantasks/scheduler.c =================================================================== --- /trunk/Ohana/src/opihi/pantasks/scheduler.c (revision 12403) +++ (revision ) @@ -1,73 +1,0 @@ -# include "scheduler.h" - -# define opihi_name "scheduler" -# define opihi_prompt "sched: " -# define opihi_description "parallel task scheduler\n" -# define opihi_history ".sched" -# define opihi_rcfile ".schedrc" - -/* program-dependent initialization */ -void initialize (int argc, char **argv) { - - auto_break = TRUE; - - /* init functions required by libraries */ - /* -libopihi */ - InitCommands (); - InitMacros (); - InitBuffers (); - InitVectors (); - InitVariables (); - - /* -libdisplay */ - InitGraph (); - InitImage (); - - /* load the commands used by this implementation */ - InitBasic (); - InitData (); - InitSched (); - InitTasks (); - InitJobs (); - InitJobIDs (); - - rl_readline_name = opihi_name; - rl_attempted_completion_function = command_completer; - rl_event_hook = NULL; - rl_set_keyboard_input_timeout (1000000); - - set_str_variable ("HISTORY", opihi_history); - set_str_variable ("PROMPT", opihi_prompt); - set_str_variable ("RCFILE", opihi_rcfile); -# ifdef HELPDIR_DEFAULT - set_str_variable ("HELPDIR", MACRO_NAME(HELPDIR_DEFAULT)); -# endif - - { /* check history file permission */ - FILE *f; - f = fopen (opihi_history, "a"); - if (f == NULL) /* no current history file here */ - gprint (GP_ERR, "can't save history.\n"); - else - fclose (f); - stifle_history (200); - read_history (opihi_history); - } - - signal (SIGINT, SIG_IGN); - return; -} - -/* standard welcome message */ -void welcome () { - gprint (GP_LOG, "\n"); - gprint (GP_LOG, "Welcome to %s - %s\n\n", opihi_name, opihi_description); -} - -/* add program-dependent exit functions here */ -void cleanup () { - QuitImage (); - QuitGraph (); - QuitController (); - return; -} Index: unk/Ohana/src/opihi/pantasks/server_threads.c =================================================================== --- /trunk/Ohana/src/opihi/pantasks/server_threads.c (revision 12403) +++ (revision ) @@ -1,153 +1,0 @@ -# include "pantasks.h" - -/* this mutex is used by the serialized threads */ - -static pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER; - -void SerialThreadLock () { - pthread_mutex_lock (&mutex); -} - -void SerialThreadUnlock () { - pthread_mutex_unlock (&mutex); -} - -/* the threaded version of pantasks does not worry about limiting the - time spent in one of the test loops -- comms with the client are - in a separate thread anyway... -*/ - -int TestElapsedCheck () { - return (FALSE); -} - -/** things related to CheckTasks **/ - -static int CheckTasksRun = FALSE; - -void CheckTasksSetState (int state) { - CheckTasksRun = state; -} -int CheckTasksGetState () { - return (CheckTasksRun); -} - -void *CheckTasksThread (void *data) { - - gprintInit (); // each thread needs to init the printing system - while (1) { - - // check for thread suspend - if (!CheckTasksRun) { - usleep (100000); // idle if thread action is suspended - continue; - } - - // one run of the task checker - SerialThreadLock (); - CheckTasks (); - CheckJobs (); - SerialThreadUnlock (); - fprintf (stderr, "T"); - usleep (10000); // allow other threads a chance to run - } -} - -# if 0 -// XXX need to harvest jobs on the same thread as they were -// spawned. for the moment, put CheckTasks and CheckJobs in -// a single thread. -/** things related to CheckJobs **/ - -static int CheckJobsRun = FALSE; - -void CheckJobsSetState (int state) { - CheckJobsRun = state; -} -int CheckJobsGetState () { - return (CheckJobsRun); -} - -void *CheckJobsThread (void *data) { - - gprintInit (); // each thread needs to init the printing system - while (1) { - - // check for thread suspend - if (!CheckJobsRun) { - usleep (100000); // idle if thread action is suspended - continue; - } - - // one run of the task checker - SerialThreadLock (); - CheckJobs (); - SerialThreadUnlock (); - fprintf (stderr, "J"); - usleep (10000); // allow other threads a chance to run - } -} -# endif - -/** things related to CheckController **/ - -static int CheckControllerRun = FALSE; - -void CheckControllerSetState (int state) { - CheckControllerRun = state; -} -int CheckControllerGetState () { - return (CheckControllerRun); -} - -void *CheckControllerThread (void *data) { - - gprintInit (); // each thread needs to init the printing system - while (1) { - - // check for thread suspend - if (!CheckControllerRun) { - usleep (100000); // idle if thread action is suspended - continue; - } - - // one run of the task checker - SerialThreadLock (); - CheckController (); - CheckControllerOutput (); - SerialThreadUnlock (); - fprintf (stderr, "C"); - usleep (10000); // allow other threads a chance to run - } -} - -/** things related to CheckInputs **/ - -static int CheckInputsRun = TRUE; - -void CheckInputsSetState (int state) { - CheckInputsRun = state; -} -int CheckInputsGetState () { - return (CheckInputsRun); -} - -void *CheckInputsThread (void *data) { - - gprintInit (); // each thread needs to init the printing system - while (1) { - - // check for thread suspend - if (!CheckInputsRun) { - usleep (100000); // idle if thread action is suspended - continue; - } - - // one run of the task checker - SerialThreadLock (); - CheckInputs (); - SerialThreadUnlock (); - fprintf (stderr, "I"); - usleep (10000); // allow other threads a chance to run - } -} Index: /trunk/Ohana/src/opihi/pantasks/task_threads.c =================================================================== --- /trunk/Ohana/src/opihi/pantasks/task_threads.c (revision 12403) +++ /trunk/Ohana/src/opihi/pantasks/task_threads.c (revision 12404) @@ -14,4 +14,6 @@ void *CheckTasksThread (void *data) { + float next_timeout; + gprintInit (); // each thread needs to init the printing system while (1) { @@ -25,18 +27,16 @@ // one run of the task checker SerialThreadLock (); - CheckTasks (); + next_timeout = CheckTasks (); SerialThreadUnlock (); if (VerboseMode() == 2) fprintf (stderr, "T"); - // fprintf (stderr, "T"); - usleep (1000); // allow other threads a chance to run + + next_timeout = MIN (next_timeout, 0.1); + if (next_timeout > 0.001) { + usleep ((int)(1000000*next_timeout)); // allow other threads a chance to run + } } } -/* the threaded version of pantasks does not worry about limiting the - time spent in one of the test loops -- comms with the client are - in a separate thread anyway... -*/ - -int TestElapsedCheck () { - return (FALSE); -} +// I sleep for a small amount of time here, based on how long until the next expected +// timeout. this enforces a certain granularity in the task creation, but prevents the task +// thread from driving the load up to silly levels.