Index: trunk/Ohana/src/addstar/src/find_matches_closest.c =================================================================== --- trunk/Ohana/src/addstar/src/find_matches_closest.c (revision 32631) +++ trunk/Ohana/src/addstar/src/find_matches_closest.c (revision 32632) @@ -132,4 +132,11 @@ } + /* this block will match a given detection to the closest object within range of that detection. + XXX note that this matches ALL detections within range of the single object to that same object + this is bad, but I cannot just go in linear order (ie, mark off each object as they are + used). I should make a list of all Nobj * Ndet pairs in range and choose the matches + based on their separations. UGH + */ + /* within match range; look for matches */ Jmin = -1; Index: trunk/Ohana/src/kapa2/src/DrawFrame.c =================================================================== --- trunk/Ohana/src/kapa2/src/DrawFrame.c (revision 32631) +++ trunk/Ohana/src/kapa2/src/DrawFrame.c (revision 32632) @@ -150,5 +150,6 @@ } -# define MIN_RANGE 1e-10 +// do I really want this? allow an override? +# define MIN_RANGE 1e-30 void AxisTickScale (Axis *axis, double *major, double *minor, int *nsignif) { Index: trunk/Ohana/src/libdvo/src/dvo_set_catdir.c =================================================================== --- trunk/Ohana/src/libdvo/src/dvo_set_catdir.c (revision 32631) +++ trunk/Ohana/src/libdvo/src/dvo_set_catdir.c (revision 32632) @@ -15,5 +15,6 @@ } if (catdir) { - current_catdir = strdup(catdir); + // need to keep this in the ohana memory system, so we cannot use strdup + current_catdir = strcreate(catdir); } } Index: trunk/Ohana/src/libohana/include/ohana.h =================================================================== --- trunk/Ohana/src/libohana/include/ohana.h (revision 32631) +++ trunk/Ohana/src/libohana/include/ohana.h (revision 32632) @@ -20,5 +20,5 @@ // comment this out to avoid the internal Ohana memory management code -// # define OHANA_MEMORY +# define OHANA_MEMORY // XXX I was including these before, but RHL claims they are not needed Index: trunk/Ohana/src/libohana/src/ohana_allocate.c =================================================================== --- trunk/Ohana/src/libohana/src/ohana_allocate.c (revision 32631) +++ trunk/Ohana/src/libohana/src/ohana_allocate.c (revision 32632) @@ -112,4 +112,7 @@ // memblock of supplied pointer old = (Memblock *) in - 1; + + if (old->startblock != OHANA_MEMMAGIC) ohana_memabort ("corrupt memory (%s, %d, %s)\n", file, line, func); + if (old->endblock != OHANA_MEMMAGIC) ohana_memabort ("corrupt memory (%s, %d, %s)\n", file, line, func); Nelem = MAX (1, Nelem); Index: trunk/Ohana/src/opihi/cmd.astro/Makefile =================================================================== --- trunk/Ohana/src/opihi/cmd.astro/Makefile (revision 32631) +++ trunk/Ohana/src/opihi/cmd.astro/Makefile (revision 32632) @@ -43,4 +43,5 @@ $(SRC)/galprofiles.$(ARCH).o \ $(SRC)/elliprofile.$(ARCH).o \ +$(SRC)/ringflux.$(ARCH).o \ $(SRC)/petrosian.$(ARCH).o \ $(SRC)/multifit.$(ARCH).o \ Index: trunk/Ohana/src/opihi/cmd.astro/init.c =================================================================== --- trunk/Ohana/src/opihi/cmd.astro/init.c (revision 32631) +++ trunk/Ohana/src/opihi/cmd.astro/init.c (revision 32632) @@ -29,4 +29,5 @@ int galradbins PROTO((int, char **)); int elliprofile PROTO((int, char **)); +int ringflux PROTO((int, char **)); int petrosian PROTO((int, char **)); int mkgauss PROTO((int, char **)); @@ -78,4 +79,5 @@ {1, "galradbins", galradbins, "generate radial vectors with interpolation along paths"}, {1, "elliprofile", elliprofile, "generate radial vectors with interpolation along paths"}, + {1, "ringflux", ringflux, "mean flux in a ring"}, {1, "petrosian", petrosian, "petrosian parameters given radial bins"}, {1, "multifit", multifit, "fit multi-order spectrum"}, Index: trunk/Ohana/src/opihi/cmd.astro/ringflux.c =================================================================== --- trunk/Ohana/src/opihi/cmd.astro/ringflux.c (revision 32632) +++ trunk/Ohana/src/opihi/cmd.astro/ringflux.c (revision 32632) @@ -0,0 +1,100 @@ +# include "astro.h" + +# define FLUX(BUF,X,Y) { \ + char valid = TRUE; \ + valid &= (X >= 0); \ + valid &= (Y >= 0); \ + valid &= (X < Nx); \ + valid &= (Y < Ny); \ + if (valid) { \ + out[Nvec] = BUF[(int)(X) + ((int)(Y))*Nx]; \ + Nvec++; \ + if (Nvec >= vec[0].Nelements) abort(); \ + } \ + } + +# define C1 0 + +int ringflux (int argc, char **argv) { + + int Nx, Ny, x, y, d, radius, Radius, dRadius, Nvec; + double *out; + float *in; + float Xo, Yo; + Buffer *buf; + Vector *vec; + + if (argc != 7) { + gprint (GP_ERR, "USAGE: ringflux (buffer) (vector) (Xo) (Yo) (Radius) (dRadius)\n"); + return (FALSE); + } + + /* select input / output buffers */ + if ((buf = SelectBuffer (argv[1], OLDBUFFER, TRUE)) == NULL) return (FALSE); + if ((vec = SelectVector (argv[2], ANYVECTOR, TRUE)) == NULL) return (FALSE); + + Nx = buf[0].header.Naxis[0]; + Ny = buf[0].header.Naxis[1]; + + Xo = atof(argv[3]); + Yo = atof(argv[4]); + + Radius = atoi (argv[5]); + dRadius = atoi (argv[6]); + + // just a rough guess at the number needed + ResetVector (vec, OPIHI_FLT, MAX (6*Radius*dRadius, 8)); + + Nvec = 0; + for (radius = Radius; radius < Radius + dRadius; radius ++) { + + x = 0; + y = radius; + +# if (C1) + d = 3 - 2*radius; +# else + d = 5 - 4*radius; +# endif + + in = (float *) buf[0].matrix.buffer; + out = vec[0].elements.Flt; + + while (x <= y) { + FLUX(in, (Xo + x), (Yo + y)); + FLUX(in, (Xo + x), (Yo - y)); + FLUX(in, (Xo + y), (Yo + x)); + FLUX(in, (Xo - y), (Yo + x)); + + if (x > 0) { + FLUX(in, (Xo - x), (Yo + y)); + FLUX(in, (Xo - x), (Yo - y)); + FLUX(in, (Xo - y), (Yo - x)); + FLUX(in, (Xo + y), (Yo - x)); + } + if (Nvec >= vec[0].Nelements - 8) { + // counting error here + ResetVector (vec, OPIHI_FLT, vec[0].Nelements + 64); + out = vec[0].elements.Flt; + } + + if (d < 0) { +# if (C1) + d = d + 4*x + 6; +# else + d = d + 8*x + 4; +# endif + } else { +# if (C1) + d = d + 4*(x-y) + 10; +# else + d = d + 8*(x-y) + 8; +# endif + y--; + } + x++; + } + } + vec[0].Nelements = Nvec; + return (TRUE); +} Index: trunk/Ohana/src/opihi/cmd.data/read_vectors.c =================================================================== --- trunk/Ohana/src/opihi/cmd.data/read_vectors.c (revision 32631) +++ trunk/Ohana/src/opihi/cmd.data/read_vectors.c (revision 32632) @@ -23,5 +23,5 @@ int read_vectors (int argc, char **argv) { - int i, j, Nskip, Nvec, *col, done, status, IsCSV; + int i, j, Nskip, Nvec, *col, done, status, IsCSV, VERBOSE; int Nbytes, nbytes, Nstart, NELEM, N, nread; char *colstr, *c0, *c1, *buffer, *extname; @@ -51,4 +51,10 @@ remove_argument (N, &argc, argv); IsCSV = TRUE; + } + + VERBOSE = FALSE; + if ((N = get_argument (argc, argv, "-v"))) { + remove_argument (N, &argc, argv); + VERBOSE = TRUE; } @@ -121,7 +127,4 @@ scan_line (f, buffer); } - - int Nfield = 0; - int Nline = 0; Nstart = 0; @@ -157,17 +160,28 @@ status = dparse (&value, col[i], c0); } - vec[i][0].elements.Flt[N] = value; if (status) { - Nfield ++; + vec[i][0].elements.Flt[N] = value; + } else { + vec[i][0].elements.Flt[N] = NAN; + if (VERBOSE) { + if (IsCSV) { + gprint (GP_ERR, "suspect field: %d (%s) in %s\n", col[i], argv[2*i+2], c0); + } else { + gprint (GP_ERR, "suspect field: %d in %s\n", col[i], c0); + } + } } - if (!status) vec[i][0].elements.Flt[N] = NAN; - } - if (status) N++; - } - Nline ++; - if (c1) { - // fprintf (stderr, "line %d, chars %ld, fields %d\n", Nline, (c1 - c0), Nfield); - } - Nfield = 0; + } + if (status) { + N++; + } else { + if (VERBOSE && FALSE) { + char temp[32]; + strncpy (temp, c0, 32); + temp[31] = 0; + gprint (GP_ERR, "skip line %s\n\n", temp); + } + } + } c0 = c1 + 1; if (N == NELEM) { @@ -177,5 +191,4 @@ } } - } } Index: trunk/Ohana/src/opihi/cmd.data/vstats.c =================================================================== --- trunk/Ohana/src/opihi/cmd.data/vstats.c (revision 32631) +++ trunk/Ohana/src/opihi/cmd.data/vstats.c (revision 32632) @@ -169,6 +169,6 @@ // we are going to do another pass: mark the entries to skip - pmin = max; - pmax = min; + pmin = min; + pmax = max; if (iter < Niter - 1) { if (vec[0].type == OPIHI_FLT) { @@ -208,4 +208,5 @@ skip: FREE(mask); + if (mask) mask = NULL; if (!Quiet) { Index: trunk/Ohana/src/opihi/dvo/skycoverage.c =================================================================== --- trunk/Ohana/src/opihi/dvo/skycoverage.c (revision 32631) +++ trunk/Ohana/src/opihi/dvo/skycoverage.c (revision 32632) @@ -202,5 +202,5 @@ V[ys*Nx + xs] = ShowDensity ? 0 : 2; } else { - V[ys*Nx + xs] = ShowDensity ? -1 : 0; + V[ys*Nx + xs] = ShowDensity ? NAN : 0; } } Index: trunk/Ohana/src/opihi/include/pantasks.h =================================================================== --- trunk/Ohana/src/opihi/include/pantasks.h (revision 32631) +++ trunk/Ohana/src/opihi/include/pantasks.h (revision 32632) @@ -40,4 +40,5 @@ TASK_END, TASK_HOST, + TASK_NICE, TASK_STDOUT, TASK_STDERR, @@ -115,4 +116,5 @@ int active; + int priority; } Task; @@ -159,4 +161,5 @@ JobMode mode; /* local or controller? */ + int priority; char *realhost; Index: trunk/Ohana/src/opihi/include/pcontrol.h =================================================================== --- trunk/Ohana/src/opihi/include/pcontrol.h (revision 32631) +++ trunk/Ohana/src/opihi/include/pcontrol.h (revision 32632) @@ -119,4 +119,5 @@ int exit_status; int Reset; + int priority; JobMode mode; JobStat state; @@ -152,4 +153,25 @@ } Host; +# if (USE_LLIST) +typedef struct StackItem { + StackItem *next; + StackItem *prev; + void *object; + char *name; + int id; +} StackItem; + +/* the Jobs and Hosts are managed in a set of Stacks which define their state */ +typedef struct { + StackItem *head; + StackItem *tail; // use this? + int Nobject; +# ifdef THREADED + pthread_mutex_t mutex; +# endif +} Stack; + +# else + /* the Jobs and Hosts are managed in a set of Stacks which define their state */ typedef struct { @@ -163,4 +185,5 @@ # endif } Stack; +# endif /* XXX if this is hard-wired, we can't change shell name in StartHost */ @@ -291,5 +314,5 @@ Job *PullJobByID (IDtype JobID, int *StackID); Job *PullJobFromStackByID (int StackID, int ID); -IDtype AddJob (char *hostname, JobMode mode, int timeout, int argc, char **argv, int Nxhosts, char **xhosts); +IDtype AddJob (char *hostname, JobMode mode, int timeout, int priority, int argc, char **argv, int Nxhosts, char **xhosts); void DelJob (Job *job); Host *UnlinkJobAndHost (Job *job); Index: trunk/Ohana/src/opihi/include/shell.h =================================================================== --- trunk/Ohana/src/opihi/include/shell.h (revision 32631) +++ trunk/Ohana/src/opihi/include/shell.h (revision 32632) @@ -184,7 +184,7 @@ /* socket functions */ -int InitServerSocket (SockAddress *Address); +int InitServerSocket (SockAddress *Address, char *hostname, char *portinfo); int WaitServerSocket (int InitSocket, SockAddress *Address); -int GetClientSocket (char *hostname); +int GetClientSocket (char *hostname, char *portinfo); int InitServerSocket_Named (char *hostname, SockAddress *Address); int DefineValidIP (void); Index: trunk/Ohana/src/opihi/lib.shell/SocketOps.c =================================================================== --- trunk/Ohana/src/opihi/lib.shell/SocketOps.c (revision 32631) +++ trunk/Ohana/src/opihi/lib.shell/SocketOps.c (revision 32632) @@ -1,3 +1,5 @@ # include "shell.h" +# include +# define HOST_NAME_MAX 256 # define MY_PORT 2000 @@ -11,34 +13,25 @@ static int *VALID; -int InitServerSocket (SockAddress *Address) { - +int GetPortRange (int *start, int *stop, char *portinfo); + +int InitServerSocket (SockAddress *Address, char *hostname, char *portinfo) { + + int start, stop; int status, InitSocket, length; - -# if (0) - struct hostent *host; - char tmpline[80], hostip[80]; - - host = gethostbyname (hostname); - bzero (hostip, 80); - for (i = 0; i < host[0].h_length; i++) { - sprintf (tmpline, "%u", (0xff & host[0].h_addr[i])); - strcat (hostip, tmpline); - if (i < host[0].h_length - 1) strcat (hostip, "."); - } -# endif - + char myHostname[HOST_NAME_MAX]; + + status = gethostname (myHostname, HOST_NAME_MAX); + + fprintf (stderr, "target host: %s, real host: %s\n", hostname, myHostname); + + GetPortRange (&start, &stop, portinfo); + + fprintf (stderr, "using port range %d - %d\n", start, stop); + Address[0].sin_family = AF_INET; - Address[0].sin_port = MY_PORT; + Address[0].sin_port = start; Address[0].sin_addr.s_addr = INADDR_ANY; // use this line to bind any address / port? retry_server: - -# if (0) - status = inet_aton (hostip, &Address[0].sin_addr); - if (!status) { - gprint (GP_ERR, "invalid address\n"); - exit (4); - } -# endif length = sizeof(Address[0]); @@ -74,5 +67,5 @@ if (errno == EADDRINUSE) { Address[0].sin_port ++; - if (Address[0].sin_port > MY_PORT + 10) { + if (Address[0].sin_port > stop) { fprintf (stderr, "failed to find a usable port\n"); exit (6); @@ -154,7 +147,49 @@ } -int GetClientSocket (char *hostname) { - - int i, status, InitSocket, length; +int GetPortRange (int *start, int *stop, char *portinfo) { + + // portinfo is a port or port range of the form NN or NN:MM + *start = MY_PORT; + *stop = *start + 10; + if (!portinfo) return TRUE; + if (*portinfo == 0) return TRUE; + + char *endptr; + *start = strtol (portinfo, &endptr, 0); + *stop = *start + 10; // default range of 10 + if (!endptr) { + gprint (GP_ERR, "error in port range parsing : %s\n", portinfo); + exit (20); + } + if (endptr == portinfo) { + gprint (GP_ERR, "error in port range (%s), must be in form NN:MM or NN\n", portinfo); + exit (20); + } + if (*endptr == 0) return TRUE; + + if (*endptr != ':') { + gprint (GP_ERR, "error in port range %s (wrong range separator, must be in form NN:MM)\n", portinfo); + exit (20); + } + char *ptr = endptr + 1; + *stop = strtol (ptr, &endptr, 0); + if (endptr == ptr) { + gprint (GP_ERR, "error in port range (%s), must be in form NN:MM or NN\n", portinfo); + exit (20); + } + if (*stop - *start > 20) { + gprint (GP_ERR, "error in port range (%s), range must be 20 or less\n", portinfo); + exit (20); + } + if (*stop < *start) { + gprint (GP_ERR, "error in port range (%s), stop must >= start\n", portinfo); + exit (20); + } + return TRUE; +} + +int GetClientSocket (char *hostname, char *portinfo) { + + int i, status, InitSocket, length, start, stop; SockAddress Address; struct hostent *host; @@ -169,10 +204,12 @@ } + GetPortRange (&start, &stop, portinfo); + if (DEBUG) { - gprint (GP_ERR, "trying %s (%s:%d)...", host[0].h_name, hostip, MY_PORT); + gprint (GP_ERR, "trying %s (%s:%d)...", host[0].h_name, hostip, start); } Address.sin_family = AF_INET; - Address.sin_port = MY_PORT; + Address.sin_port = start; retry_client: @@ -195,5 +232,5 @@ if (errno == ECONNREFUSED) { Address.sin_port ++; - if (Address.sin_port > MY_PORT + 10) exit (12); + if (Address.sin_port > stop) exit (12); goto retry_client; } Index: trunk/Ohana/src/opihi/lib.shell/multicommand.c =================================================================== --- trunk/Ohana/src/opihi/lib.shell/multicommand.c (revision 32631) +++ trunk/Ohana/src/opihi/lib.shell/multicommand.c (revision 32632) @@ -13,5 +13,5 @@ void multicommand_InitServer () { - char hostname[256], PASSWORD[256]; + char hostname[256], PASSWORD[256], portinfo[256]; if (server != 0) { @@ -27,6 +27,10 @@ } + /* is a port range defined? otherwise use the default */ + memset (portinfo, 0, 256); + VarConfig ("PANTASKS_SERVER_PORT", "%s", portinfo); + /* attempt to connect to the server */ - server = GetClientSocket (hostname); + server = GetClientSocket (hostname, portinfo); /* here we can perform the security handshaking */ Index: trunk/Ohana/src/opihi/pantasks/ControllerOps.c =================================================================== --- trunk/Ohana/src/opihi/pantasks/ControllerOps.c (revision 32631) +++ trunk/Ohana/src/opihi/pantasks/ControllerOps.c (revision 32632) @@ -258,4 +258,10 @@ sprintf (cmd, "job -host %s", job[0].task[0].host); } + } + + if (job[0].priority) { + char tmp[64]; + snprintf (tmp, 64, " -nice %d", job[0].priority); + strcat (cmd, tmp); } Index: trunk/Ohana/src/opihi/pantasks/JobOps.c =================================================================== --- trunk/Ohana/src/opihi/pantasks/JobOps.c (revision 32631) +++ trunk/Ohana/src/opihi/pantasks/JobOps.c (revision 32632) @@ -107,4 +107,5 @@ job[0].mode = JOB_CONTROLLER; } + job[0].priority = task[0].priority; /* we need our own copy of task[0].argv argc is the number of valid args, like the usual command line. we Index: trunk/Ohana/src/opihi/pantasks/LocalJob.c =================================================================== --- trunk/Ohana/src/opihi/pantasks/LocalJob.c (revision 32631) +++ trunk/Ohana/src/opihi/pantasks/LocalJob.c (revision 32632) @@ -161,4 +161,14 @@ if (VerboseMode()) gprint (GP_ERR, "local job launched\n"); + /* set nice level for the child process -- maybe I should not exit here... */ + if (job[0].priority) { + status = setpriority (PRIO_PROCESS, pid, job[0].priority); + if (status == -1) { + gprint (GP_ERR, "error setting priority\n"); + perror ("setpriority: "); + exit (2); + } + } + /* close the other ends of the pipes */ close (stdout_fd[1]); stdout_fd[1] = 0; Index: trunk/Ohana/src/opihi/pantasks/Makefile =================================================================== --- trunk/Ohana/src/opihi/pantasks/Makefile (revision 32631) +++ trunk/Ohana/src/opihi/pantasks/Makefile (revision 32632) @@ -83,4 +83,5 @@ $(SRC)/task.$(ARCH).o \ $(SRC)/task_host.$(ARCH).o \ +$(SRC)/task_nice.$(ARCH).o \ $(SRC)/task_nmax.$(ARCH).o \ $(SRC)/task_active.$(ARCH).o \ Index: trunk/Ohana/src/opihi/pantasks/TaskOps.c =================================================================== --- trunk/Ohana/src/opihi/pantasks/TaskOps.c (revision 32631) +++ trunk/Ohana/src/opihi/pantasks/TaskOps.c (revision 32632) @@ -487,4 +487,5 @@ NewTask[0].active = TRUE; + NewTask[0].priority = 0; return (NewTask); } @@ -595,4 +596,5 @@ if (!strcasecmp (command, "END")) hash = TASK_END; if (!strcasecmp (command, "HOST")) hash = TASK_HOST; + if (!strcasecmp (command, "NICE")) hash = TASK_NICE; if (!strcasecmp (command, "NMAX")) hash = TASK_NMAX; if (!strcasecmp (command, "ACTIVE")) hash = TASK_ACTIVE; Index: trunk/Ohana/src/opihi/pantasks/init.c =================================================================== --- trunk/Ohana/src/opihi/pantasks/init.c (revision 32631) +++ trunk/Ohana/src/opihi/pantasks/init.c (revision 32632) @@ -4,4 +4,5 @@ int task PROTO((int, char **)); int task_host PROTO((int, char **)); +int task_nice PROTO((int, char **)); int task_nmax PROTO((int, char **)); int task_npending PROTO((int, char **)); @@ -33,4 +34,5 @@ {1, "halt", halt, "halt the scheduler (no job harvesting)"}, {1, "host", task_host, "define host machine for a task"}, + {1, "nice", task_nice, "set nice priority level for a task"}, {1, "ipptool2book", ipptool2book, "convert queue with ipptool output to book"}, {1, "kill", kill_job, "kill job"}, Index: trunk/Ohana/src/opihi/pantasks/init_server.c =================================================================== --- trunk/Ohana/src/opihi/pantasks/init_server.c (revision 32631) +++ trunk/Ohana/src/opihi/pantasks/init_server.c (revision 32632) @@ -4,4 +4,5 @@ int task PROTO((int, char **)); int task_host PROTO((int, char **)); +int task_nice PROTO((int, char **)); int task_nmax PROTO((int, char **)); int task_npending PROTO((int, char **)); @@ -34,4 +35,5 @@ {1, "delete", delete_job, "delete job"}, {1, "host", task_host, "define host machine for a task"}, + {1, "nice", task_nice, "set nice priority level for a task"}, {1, "ipptool2book", ipptool2book, "convert queue with ipptool output to book"}, {1, "kill", kill_job, "kill job"}, Index: trunk/Ohana/src/opihi/pantasks/pantasks_server.c.in =================================================================== --- trunk/Ohana/src/opihi/pantasks/pantasks_server.c.in (revision 32631) +++ trunk/Ohana/src/opihi/pantasks/pantasks_server.c.in (revision 32632) @@ -18,4 +18,5 @@ int main (int argc, char **argv) { + char hostname[256], portinfo[256]; char log_stdout[1024], log_stderr[1024]; pthread_t JobsAndTasksThread; @@ -100,6 +101,16 @@ pass them to the ListenClient thread */ + /* find the defined server hostname */ + if (VarConfig ("PANTASKS_SERVER", "%s", hostname) == NULL) { + gprint (GP_ERR, "pantasks server host undefined\n"); + exit (31); + } + + /* is a port range defined? otherwise use the default */ + memset (portinfo, 0, 256); + VarConfig ("PANTASKS_SERVER_PORT", "%s", portinfo); + /* create the listening socket */ - InitSocket = InitServerSocket (&Address); + InitSocket = InitServerSocket (&Address, hostname, portinfo); InitPassword (); Index: trunk/Ohana/src/opihi/pantasks/task.c =================================================================== --- trunk/Ohana/src/opihi/pantasks/task.c (revision 32631) +++ trunk/Ohana/src/opihi/pantasks/task.c (revision 32632) @@ -72,4 +72,5 @@ case TASK_NMAX: case TASK_HOST: + case TASK_NICE: case TASK_EXIT: case TASK_EXEC: Index: trunk/Ohana/src/opihi/pantasks/task_nice.c =================================================================== --- trunk/Ohana/src/opihi/pantasks/task_nice.c (revision 32632) +++ trunk/Ohana/src/opihi/pantasks/task_nice.c (revision 32632) @@ -0,0 +1,34 @@ +# include "pantasks.h" + +int task_nice (int argc, char **argv) { + + char *endptr = NULL; + int priority; + Task *task; + + if (argc != 2) { + gprint (GP_ERR, "USAGE: nice (priority)\n"); + return (FALSE); + } + + JobTaskLock(); + task = GetNewTask (); + if (task == NULL) { + gprint (GP_ERR, "ERROR: not defining or running a task\n"); + JobTaskUnlock(); + return (FALSE); + } + + priority = strtol (argv[1], &endptr, 10); + if (*endptr) goto fail; + if (priority < 0) goto fail; + if (priority > 20) goto fail; + + task[0].priority = priority; + JobTaskUnlock(); + return (TRUE); + +fail: + gprint (GP_ERR, "ERROR: nice (priority) -- priority must be an integer 0 to 20\n"); + return (FALSE); +} Index: trunk/Ohana/src/opihi/pantasks/test/nice.sh =================================================================== --- trunk/Ohana/src/opihi/pantasks/test/nice.sh (revision 32632) +++ trunk/Ohana/src/opihi/pantasks/test/nice.sh (revision 32632) @@ -0,0 +1,70 @@ + +# a basic task which just runs 'sleep 10' +task nicetask_local + command sleep 10 + host local + nice 10 + + periods -poll 0.1 + periods -exec 1.0 + periods -timeout 20 + active true + npending 5 + + stdout tmp.txt + stderr tmp.txt + + task.exec + echo "nicetask_local start" + end + + # success + task.exit 0 + echo "nicetask_local stop" + end + + # default exit status + task.exit default + echo "failure: exit status: $EXIT" + end + + # operation times out? + task.exit timeout + echo "timeout" + end +end + +# a basic task which just runs 'sleep 10' +task meantask_local + command sleep 10 + host local + + periods -poll 0.1 + periods -exec 1.0 + periods -timeout 20 + active true + npending 5 + + stdout tmp.txt + stderr tmp.txt + + task.exec + echo "meantask_local start" + end + + # success + task.exit 0 + echo "meantask_local stop" + end + + # default exit status + task.exit default + echo "failure: exit status: $EXIT" + end + + # operation times out? + task.exit timeout + echo "timeout" + end +end + Index: trunk/Ohana/src/opihi/pantasks/test/nice_remote.sh =================================================================== --- trunk/Ohana/src/opihi/pantasks/test/nice_remote.sh (revision 32632) +++ trunk/Ohana/src/opihi/pantasks/test/nice_remote.sh (revision 32632) @@ -0,0 +1,70 @@ + +# a basic task which just runs 'sleep 10' +task nicetask_local + command sleep 10 + host anyhost + nice 10 + + periods -poll 0.1 + periods -exec 1.0 + periods -timeout 20 + active true + npending 5 + + stdout tmp.txt + stderr tmp.txt + + task.exec + echo "nicetask_local start" + end + + # success + task.exit 0 + echo "nicetask_local stop" + end + + # default exit status + task.exit default + echo "failure: exit status: $EXIT" + end + + # operation times out? + task.exit timeout + echo "timeout" + end +end + +# a basic task which just runs 'sleep 10' +task meantask_local + command sleep 10 + host anyhost + + periods -poll 0.1 + periods -exec 1.0 + periods -timeout 20 + active true + npending 5 + + stdout tmp.txt + stderr tmp.txt + + task.exec + echo "meantask_local start" + end + + # success + task.exit 0 + echo "meantask_local stop" + end + + # default exit status + task.exit default + echo "failure: exit status: $EXIT" + end + + # operation times out? + task.exit timeout + echo "timeout" + end +end + Index: trunk/Ohana/src/opihi/pclient/job.c =================================================================== --- trunk/Ohana/src/opihi/pclient/job.c (revision 32631) +++ trunk/Ohana/src/opihi/pclient/job.c (revision 32632) @@ -3,6 +3,13 @@ int job (int argc, char **argv) { - int i, pid, status; + int i, N, pid, status, priority; char **targv; + + priority = 0; + if ((N = get_argument (argc, argv, "-nice"))) { + remove_argument (N, &argc, argv); + priority = atoi (argv[N]); + remove_argument (N, &argc, argv); + } if (argc < 2) { @@ -54,4 +61,14 @@ } + /* set nice level for the child process */ + if (priority) { + status = setpriority (PRIO_PROCESS, pid, priority); + if (status == -1) { + gprint (GP_ERR, "error setting priority\n"); + perror ("setpriority: "); + exit (2); + } + } + /* free temporary arg list */ for (i = 0; i < argc - 1; i++) { Index: trunk/Ohana/src/opihi/pcontrol/CheckIdleHost.c =================================================================== --- trunk/Ohana/src/opihi/pcontrol/CheckIdleHost.c (revision 32631) +++ trunk/Ohana/src/opihi/pcontrol/CheckIdleHost.c (revision 32632) @@ -5,4 +5,27 @@ // a (temporary?) work-around for the problem that the remote pclient job tends to grow // too large over time. + +/**** + + queueing strategy: + + We have a problem when the queue contains many jobs of different processing times. If we + start with an equal number of jobs in two classes, fast and slow, eventually, we will end + up with all hosts running the slow jobs and the fast jobs completely blocked. If J1 takes + T1 and J2 takes T2, and jobs have equal probability to land in a slot... ?? + + It seems like we should boost the probability of the fast jobs over the slow jobs (of + course, the end result of that will be all fast jobs draining and the slow jobs hanging + around). + + There is also a problem related to LAP, in that chip stage has a huge number of tasks in + the queue (effectively infinite). Make the probability of the job being selected + proportional to something. In any case, we need a user-controllable way to change the + probability of selection + + As things are implemented below, the selection order depends on the queue order. The + easiest way to modify the probabilities is to re-sort based on something. + + ****/ static float MAX_WANTHOST_WAIT = 10.0; Index: trunk/Ohana/src/opihi/pcontrol/JobOps.c =================================================================== --- trunk/Ohana/src/opihi/pcontrol/JobOps.c (revision 32631) +++ trunk/Ohana/src/opihi/pcontrol/JobOps.c (revision 32632) @@ -195,5 +195,5 @@ } -IDtype AddJob (char *hostname, JobMode mode, int timeout, int argc, char **argv, int Nxhosts, char **xhosts) { +IDtype AddJob (char *hostname, JobMode mode, int timeout, int priority, int argc, char **argv, int Nxhosts, char **xhosts) { int JobID; @@ -220,4 +220,5 @@ job[0].mode = mode; + job[0].priority = priority; job[0].state = 0; @@ -270,2 +271,35 @@ FREE (job); } + +/*** ResortJobStack can be used to adjust priorities based on some info we supply. This + is not finished -- to finish this, I need to define the metric of interest and + arrange for that to be passed to the jobs in pcontrol + +void ResortJobStack (int StackID) { + + Stack *stack = GetJobStack (StackID); + LockStack (stack); + +# define SWAPFUNC(A,B){ \ + void *tmpObject = stack[0].object[A]; \ + stack[0].object[A] = stack[0].object[B]; \ + stack[0].object[B] = tmpObject; \ + void *tmpName = stack[0].name[A]; \ + stack[0].name[A] = stack[0].name[B]; \ + stack[0].name[B] = tmpName; \ + void *tmpID = stack[0].id[A]; \ + stack[0].id[A] = stack[0].id[B]; \ + stack[0].id[B] = tmpID; \ + } + +# define COMPARE(A,B)(((Job *)stack[0].object[A]).VALUE < ((Job *)stack[0].object[B]).VALUE) + + OHANA_SORT (stack[0].Nobject, COMPARE, SWAPFUNC); + +# undef SWAPFUNC +# undef COMPARE + + UnlockStack (stack); +} + +***/ Index: trunk/Ohana/src/opihi/pcontrol/StackOpsLL.c =================================================================== --- trunk/Ohana/src/opihi/pcontrol/StackOpsLL.c (revision 32632) +++ trunk/Ohana/src/opihi/pcontrol/StackOpsLL.c (revision 32632) @@ -0,0 +1,227 @@ +# include "pcontrol.h" + +/* 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 (linked list)\n", stack[0].Nobject); +} + +/* allocate stack: create head node (empty node?) */ +Stack *InitStack () { + + Stack *stack; + + ALLOCATE (stack, Stack, 1); + + stack->Nobject = 0; + stack->head = NULL; + stack->tail = NULL; + +# ifdef THREADED + pthread_mutex_init (&stack[0].mutex, NULL); +# endif + return (stack); +} + +void FreeStack (Stack *stack) { + + ASSERT (!stack->head, "stack items not freed"); + ASSERT (!stack->tail, "stack items not freed"); + free (stack); +} + +/* STACK_TOP == 0, STACK_BOTTOM == -1 + in this version, entries by seq number other than head and tail are not allowed + */ + +StackItem InitStackItem (void *object, char *name, int id) { + + StackItem *item; + ALLOCATE (item, StackItem, 1); + item->object = object; + item->name = name; + item->id = id; + + return item; +} + +/* push object on stack at given location */ +int PushStack (Stack *stack, int where, void *object, int id, char *name) { + + int i; + + ASSERT (where == STACK_TOP || where == STACK_BOTTOM, "invalid location"); + ASSERT (stack != NULL, "stack not set"); + + PrintStackInfo (stack, __func__); + LockStack (stack); + + // make this a StackItem generator + StackItem *item = InitStackItem (object, name, id); + ALLOCATE (item, StackItem, 1); + + /* there is a special case when the stack is empty. in this case, we have to update + both head and tail. + */ + if (!stack->head && !stack->tail) { + stack->head = item; + stack->tail = item; + stack->Nobject ++; + UnlockStack (stack); + return (TRUE); + } + + if (where == STACK_TOP) { + StackHead *oldHead = stack->head; + stack->head = item; + item->next = oldHead; + oldHead->prev = item; + } else { + StackHead *oldTail = stack->tail; + stack->tail = item; + item->prev = oldTail; + oldTail->next = item; + } + stack->Nobject ++; + UnlockStack (stack); + return (TRUE); +} + +/* get object from specified point in stack (negative == distance from end) */ +void *PullStackByLocation (Stack *stack, int where) { + + void *object; + + ASSERT (where == STACK_TOP || where == STACK_BOTTOM, "invalid location"); + ASSERT (stack != NULL, "stack not set"); + + PrintStackInfo (stack, __func__); + LockStack (stack); + + if ( + + 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; +} + +/*** I need a function to sort a stack by something. + I also need to replace these arrays with linked lists. + ***/ Index: trunk/Ohana/src/opihi/pcontrol/StartJob.c =================================================================== --- trunk/Ohana/src/opihi/pcontrol/StartJob.c (revision 32631) +++ trunk/Ohana/src/opihi/pcontrol/StartJob.c (revision 32632) @@ -25,4 +25,9 @@ bzero (line, Nline); strcpy (line, "job"); + if (job[0].priority) { + char tmp[64]; + snprintf (tmp, 64, " -nice %d", job[0].priority); + strcat (line, tmp); + } for (i = 0; i < job[0].argc; i++) { strcat (line, " "); Index: trunk/Ohana/src/opihi/pcontrol/job.c =================================================================== --- trunk/Ohana/src/opihi/pcontrol/job.c (revision 32631) +++ trunk/Ohana/src/opihi/pcontrol/job.c (revision 32632) @@ -5,5 +5,5 @@ char *Host = NULL; char **targv = NULL; - int i, N, Mode, targc, Timeout; + int i, N, Mode, targc, Timeout, priority; IDtype JobID; char **xhosts = NULL; @@ -43,4 +43,11 @@ } + priority = 0; + if ((N = get_argument (argc, argv, "-nice"))) { + remove_argument (N, &argc, argv); + priority = atoi (argv[N]); + remove_argument (N, &argc, argv); + } + xhosts = NULL; Nxhosts = 0; @@ -60,8 +67,5 @@ } - if (argc < 2) { - FREE (Host); - goto usage; - } + if (argc < 2) goto usage; targc = argc - 1; @@ -72,5 +76,5 @@ // a JobID < 0 mean the job was not accepted - JobID = AddJob (Host, Mode, Timeout, targc, targv, Nxhosts, xhosts); + JobID = AddJob (Host, Mode, Timeout, priority, targc, targv, Nxhosts, xhosts); gprint (GP_LOG, "JobID: %d\n", (int) JobID); return (TRUE); Index: trunk/Ohana/src/opihi/pcontrol/status.c =================================================================== --- trunk/Ohana/src/opihi/pcontrol/status.c (revision 32631) +++ trunk/Ohana/src/opihi/pcontrol/status.c (revision 32632) @@ -59,8 +59,9 @@ } + PrintID (GP_LOG, job[0].JobID); + gprint (GP_LOG, " %2d ", job[0].priority); 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"); }