Index: /trunk/Ohana/src/opihi/Makefile =================================================================== --- /trunk/Ohana/src/opihi/Makefile (revision 7891) +++ /trunk/Ohana/src/opihi/Makefile (revision 7892) @@ -12,4 +12,7 @@ pclient: lib.data lib.shell cmd.basic pcontrol: lib.data lib.shell cmd.basic + +pantasks_server: lib.data lib.shell cmd.basic cmd.data cmd.astro +pantasks_client: lib.data lib.shell cmd.basic cmd.data cmd.astro LIBS = lib.data lib.shell cmd.basic cmd.data cmd.astro @@ -47,4 +50,10 @@ ############################################################# +pantasks_client pantasks_server: + if [ -d pantasks ]; then (cd pantasks && make $@); fi + +pantasks_client.install pantasks_server.install: + if [ -d pantasks ]; then (cd pantasks && make $@); fi + $(PROGRAM) $(LIBS) $(EXTRAS): if [ -d "$@" ]; then (cd $@ && make); fi Index: /trunk/Ohana/src/opihi/cmd.basic/init.c =================================================================== --- /trunk/Ohana/src/opihi/cmd.basic/init.c (revision 7891) +++ /trunk/Ohana/src/opihi/cmd.basic/init.c (revision 7892) @@ -89,2 +89,26 @@ } + +void InitBasic_PantasksClient () { + + int i; + + InitCommands (); + InitMacros (); + InitBuffers (); + InitVectors (); + InitVariables (); + InitLists (); + InitOutfile (); + + for (i = 0; i < sizeof (cmds) / sizeof (Command); i++) { + if (!strcmp (cmds[i].name, "quit")) goto valid; + if (!strcmp (cmds[i].name, "exit")) goto valid; + if (!strcmp (cmds[i].name, "exec")) goto valid; + if (!strcmp (cmds[i].name, "!")) goto valid; + continue; + + valid: + AddCommand (&cmds[i]); + } +} Index: /trunk/Ohana/src/opihi/doc/pantasks.txt =================================================================== --- /trunk/Ohana/src/opihi/doc/pantasks.txt (revision 7892) +++ /trunk/Ohana/src/opihi/doc/pantasks.txt (revision 7892) @@ -0,0 +1,25 @@ + +I need to do some serious work on the output messages in order +to support the client/server interactions. In client/server mode, +all of the output messages to be sent back to the client need to be +saved in a buffer and then pushed back to the client after the command +is executed. Currently, I can select only the output file pointer. +this is not sufficient. I need to supply a function which will write to +an internal buffer until a 'dump' function is called, or something +equivalent. + +in client/server mode, the commands issues by the client (except as noted +below) all return immediately. The output produced by those commands +is saved by the server until the command is finished, then the +resulting buffered data is pushed back to the client. There a few +commands which behave a bit differently. these commands are in the +background thread of the scheduler. the resulting output is sent to +the logger device of the scheduler rather than back to any specific +client. should there be a way for the client(s) to examine the +scheduler output? + +- pantasks input: this passes an 'input' command to the server, which + performs the input in a different thread. is this the same thread + as the scheduler loop? another option within the scheduler loop? + +- run Index: /trunk/Ohana/src/opihi/dvo/fitsed.c =================================================================== --- /trunk/Ohana/src/opihi/dvo/fitsed.c (revision 7891) +++ /trunk/Ohana/src/opihi/dvo/fitsed.c (revision 7892) @@ -1,4 +1,3 @@ # include "dvoshell.h" -# define NMIN_PTS 100 typedef struct { Index: /trunk/Ohana/src/opihi/include/basic.h =================================================================== --- /trunk/Ohana/src/opihi/include/basic.h (revision 7891) +++ /trunk/Ohana/src/opihi/include/basic.h (revision 7892) @@ -5,2 +5,3 @@ void cleanup (); void InitBasic (); +void InitBasic_PantasksClient (); Index: /trunk/Ohana/src/opihi/include/pantasks.h =================================================================== --- /trunk/Ohana/src/opihi/include/pantasks.h (revision 7891) +++ /trunk/Ohana/src/opihi/include/pantasks.h (revision 7892) @@ -1,4 +1,14 @@ # include "data.h" # include "basic.h" + +# include +# include +# include +# include +# include +# include +# include + +typedef struct sockaddr_in SockAddress; typedef enum { @@ -120,4 +130,5 @@ void InitPantasks (); +void InitPantasksClient (); void InitTasks (); Task *NextTask (); @@ -183,2 +194,10 @@ void gotsignal (int signum); +void InitClients (); +void AddNewClient (int client); +void *ListenClients (void *data); + +int InitServerSocket (SockAddress *Address); +int WaitServerSocket (int InitSocket, SockAddress *Address); +int GetClientSocket (char *hostname); +int InitServerSocket_Named (char *hostname, SockAddress *Address); Index: /trunk/Ohana/src/opihi/include/shell.h =================================================================== --- /trunk/Ohana/src/opihi/include/shell.h (revision 7891) +++ /trunk/Ohana/src/opihi/include/shell.h (revision 7892) @@ -49,4 +49,5 @@ int multicommand PROTO((char *line)); int command PROTO((char *, char **)); +int command_client PROTO((char *, char **)); char *expand_vars PROTO((char *line)); char *expand_vectors PROTO((char *line)); @@ -87,4 +88,5 @@ int ConfigInit PROTO((int *argc, char **argv)); char *VarConfig PROTO((char *keyword, char *mode, void *ptr)); +char *VarConfigEntry PROTO((char *keyword, char *mode, int entry, void *ptr)); FILE *GetOutfile PROTO((void)); void InitOutfile PROTO((void)); Index: /trunk/Ohana/src/opihi/lib.shell/ConfigInit.c =================================================================== --- /trunk/Ohana/src/opihi/lib.shell/ConfigInit.c (revision 7891) +++ /trunk/Ohana/src/opihi/lib.shell/ConfigInit.c (revision 7892) @@ -42,2 +42,24 @@ return (ptr); } + +char *VarConfigEntry (char *keyword, char *mode, int entry, void *ptr) { + + char *answer; + + answer = get_variable (keyword); + if (answer == (char *) NULL) { + answer = ScanConfig (GlobalConfig, keyword, mode, entry, ptr); + return (answer); + } + + if (!strcmp (mode, "%s")) strcpy ((char *) ptr, answer); + if (!strcmp (mode, "%d")) *(int *) ptr = atoi (answer); + if (!strcmp (mode, "%u")) *(unsigned *) ptr = atoi (answer); + if (!strcmp (mode, "%ld")) *(long *) ptr = atoi (answer); + if (!strcmp (mode, "%hd")) *(short *) ptr = atoi (answer); + if (!strcmp (mode, "%f")) *(float *) ptr = atof (answer); + if (!strcmp (mode, "%lf")) *(double *) ptr = atof (answer); + + free (answer); + return (ptr); +} Index: /trunk/Ohana/src/opihi/lib.shell/command_client.c =================================================================== --- /trunk/Ohana/src/opihi/lib.shell/command_client.c (revision 7892) +++ /trunk/Ohana/src/opihi/lib.shell/command_client.c (revision 7892) @@ -0,0 +1,43 @@ +# include "opihi.h" +# define VERBOSE 0 + +/* this function acts like the standard command parser, + but skips the expansion of variables and vectors */ +int command_client (char *line, char **outline) { + + int i, status, argc; + char **argv, **targv; + Command *cmd; + + /* force a space between ! and first word: !ls becomes ! ls */ + if (line[0] == '!') { + REALLOCATE (line, char, strlen(line) + 5); + memmove (&line[2], &line[1], strlen(&line[1]) + 1); + line[1] = ' '; + } + + /* we may have reallocated line, return new pointer */ + *outline = line; + + argv = parse_commands (line, &argc); + if (argc == 0) return (TRUE); /* empty command or assignment */ + + /* save the original values of argv since command may modify the array */ + ALLOCATE (targv, char *, argc); + for (i = 0; i < argc; i++) targv[i] = argv[i]; + + cmd = MatchCommand (argv[0], TRUE, FALSE); + if (cmd == NULL) { + status = FALSE; + } else { + free (argv[0]); + argv[0] = strcreate (cmd[0].name); + targv[0] = argv[0]; + status = (*cmd[0].func) (argc, argv); + } + for (i = 0; i < argc; i++) free (targv[i]); + free (targv); + free (argv); + + return (status); +} Index: /trunk/Ohana/src/opihi/lib.shell/opihi.c =================================================================== --- /trunk/Ohana/src/opihi/lib.shell/opihi.c (revision 7891) +++ /trunk/Ohana/src/opihi/lib.shell/opihi.c (revision 7892) @@ -2,5 +2,5 @@ /******************/ -int main (int argc, char **argv) { +int opihi (int argc, char **argv) { int Nbad, status; Index: /trunk/Ohana/src/opihi/mana/mana.c =================================================================== --- /trunk/Ohana/src/opihi/mana/mana.c (revision 7891) +++ /trunk/Ohana/src/opihi/mana/mana.c (revision 7892) @@ -44,2 +44,9 @@ return; } + +/* call to opihi shell */ +int main (int argc, char **argv) { + int status; + status = opihi (argc, argv); + exit (status); +} Index: /trunk/Ohana/src/opihi/pantasks/CheckPassword.c =================================================================== --- /trunk/Ohana/src/opihi/pantasks/CheckPassword.c (revision 7892) +++ /trunk/Ohana/src/opihi/pantasks/CheckPassword.c (revision 7892) @@ -0,0 +1,34 @@ +# include "pantasks.h" +# define DEBUG 0 + +static char PASSWORD[256]; + +int InitPassword () { + + VarConfig ("PASSWORD", "%s", PASSWORD); + +} + +int CheckPassword (int BindSocket) { + + IOBuffer message; + int status; + + if (DEBUG) fprintf (stderr, "waiting for password %s\n", PASSWORD); + + status = ExpectCommand (BindSocket, strlen(PASSWORD), 0.1, &message); + if (status != 0) { + if (DEBUG) fprintf (stderr, "failed connection\n"); + FreeIOBuffer (&message); + close (BindSocket); + return (FALSE); + } + if (strncmp (message.buffer, PASSWORD, strlen(PASSWORD))) { + if (DEBUG) fprintf (stderr, "invalid password\n"); + close (BindSocket); + return (FALSE); + } + if (DEBUG) fprintf (stderr, "accepted password (%s)\n", message.buffer); + + return (TRUE); +} Index: /trunk/Ohana/src/opihi/pantasks/ListenClients.c =================================================================== --- /trunk/Ohana/src/opihi/pantasks/ListenClients.c (revision 7892) +++ /trunk/Ohana/src/opihi/pantasks/ListenClients.c (revision 7892) @@ -0,0 +1,161 @@ +# include "pantasks.h" +# define DEBUG 0 + +static int NCLIENTS; +static int Nclients; +static int *clients; +static IOBuffer *buffers; + +/* outline of functions we need here */ + +void InitClients () { + + Nclients = 0; + NCLIENTS = 10; + ALLOCATE (clients, int, NCLIENTS); + ALLOCATE (buffers, IOBuffer, NCLIENTS); +} + +void AddNewClient (int client) { + + /* add this client to the client table */ + /* need to create an IOBuffer for each possible client? + we may not read a complete line in one pass. have to + wait until we get the complete lines... + */ + + if (DEBUG) fprintf (stderr, "adding a new client (%d)\n", client); + clients[Nclients] = client; + InitIOBuffer(&buffers[Nclients], 256); + Nclients ++; + if (Nclients >= NCLIENTS - 1) { + NCLIENTS += 10; + REALLOCATE (clients, int, NCLIENTS); + REALLOCATE (buffers, IOBuffer, NCLIENTS); + } +} + +void *ListenClients (void *data) { + + int i, Ncurrent, Nmax, status, Nread; + char *line; + fd_set fdSet; + struct timeval timeout; + + InitClients (); + + while (0) { + + Ncurrent = Nclients; + for (i = 0; i < Ncurrent; i++) { + ExpectMessage (clients[i], 0.25, &buffers[i]); + if (DEBUG) fprintf (stderr, "read %d from client %d\n", buffers[i].Nbuffer, i); + if (buffers[i].Nbuffer) { + if (DEBUG) fprintf (stderr, "%s\n", buffers[i].buffer); + } + } + usleep (250000); + } + + while (1) { + + /* Wait up to 0.5 second - need to timeout to update client list */ + /* timeout gets mucked: need to reset before each select */ + timeout.tv_sec = 1; + timeout.tv_usec = 000000; + + /* place all of the clients in the fdSet */ + Ncurrent = Nclients; + Nmax = 0; + FD_ZERO (&fdSet); + for (i = 0; i < Ncurrent; i++) { + Nmax = MAX (Nmax, clients[i]); + FD_SET (clients[i], &fdSet); + if (FD_ISSET(clients[i], &fdSet)) { + if (DEBUG) fprintf (stderr, "fd %d is set\n", clients[i]); + } else { + if (DEBUG) fprintf (stderr, "fd %d is not set\n", clients[i]); + } + } + Nmax ++; + + /* block until we have some data on the pipes (or timeout) */ + if (DEBUG) fprintf (stderr, "listening to %d clients\n", Ncurrent); + status = select (Nmax, &fdSet, NULL, NULL, &timeout); + + if (DEBUG) fprintf (stderr, "messages from %d clients\n", status); + + if (status == -1) { + perror("select()"); + return (FALSE); + } + + /* if no data, go back for another select */ + if (status <= 0) continue; + + /* loop over the clients with data */ + for (i = 0; i < Ncurrent; i++) { + if (!FD_ISSET(clients[i], &fdSet)) continue; + + /* this client has data, handle it */ + if (DEBUG) fprintf (stderr, "data from client %d\n", clients[i]); + + if (DEBUG) fprintf (stderr, "start: ", buffers[i].buffer); + fwrite (buffers[i].buffer, 1, buffers[i].Nbuffer, stderr); + if (DEBUG) fprintf (stderr, "...\n"); + + /* read until the pipe is empty: -1 is empty, -2 is error */ + Nread = 0; + while (Nread >= 0) { + Nread = ReadtoIOBuffer (&buffers[i], clients[i]); + if (DEBUG) fprintf (stderr, "read %d bytes from socket\n", Nread); + + if (Nread == -2) { + /* error: do something */ + if (1) fprintf (stderr, "error reading from socket\n"); + } + } + if (DEBUG) fprintf (stderr, "read %d total bytes\n", buffers[i].Nbuffer); + + if (DEBUG) { + fprintf (stderr, "end: ", buffers[i].buffer); + fwrite (buffers[i].buffer, 1, buffers[i].Nbuffer, stderr); + fprintf (stderr, "...\n"); + } + + /* see if we have a complete message waiting */ + line = CheckForMessage (&buffers[i]); + if (line == NULL) continue; + + /* we now have a possible command from the client: run it */ + stripwhite (line); + if (*line) { + if (1) fprintf (stderr, "got command: %s\n", line); + status = multicommand (line); + } + free (line); + + /* this function should return the output buffer + to the currently selected client */ + // DumpOutputBuffer (); + } + + /* if Nread == -2, we probably need to drop the client */ + /* check if we need to drop / remove any clients */ + /* check if we need to shut down the thread */ + } +} + +/* the AddClient commands are issued by the parent thread + the value of Nclients may increase after we check it here. + only this thread is allowed to decrease Nclients and remove + a client from the table */ + +/* - read data from fd + - if we have a complete line, request a block in the + scheduler loop + - execute the command + - send the response back to the client. + - all commands will have to send their output to a buffer, + to be sent back to the calling process +*/ Index: /trunk/Ohana/src/opihi/pantasks/Makefile =================================================================== --- /trunk/Ohana/src/opihi/pantasks/Makefile (revision 7891) +++ /trunk/Ohana/src/opihi/pantasks/Makefile (revision 7892) @@ -25,6 +25,8 @@ # sched user commands and support functions ######################## +single = \ +$(SDIR)/pantasks.$(ARCH).o + funcs = \ -$(SDIR)/pantasks.$(ARCH).o \ $(SDIR)/init.$(ARCH).o \ $(SDIR)/CheckJobs.$(ARCH).o \ @@ -63,4 +65,17 @@ $(SDIR)/version.$(ARCH).o +client = \ +$(SDIR)/pantasks_client.$(ARCH).o \ +$(SDIR)/client_shell.$(ARCH).o \ +$(SDIR)/invalid.$(ARCH).o \ +$(SDIR)/init_client.$(ARCH).o \ +$(SDIR)/SocketOps.$(ARCH).o + +server = \ +$(SDIR)/pantasks_server.$(ARCH).o \ +$(SDIR)/ListenClients.$(ARCH).o \ +$(SDIR)/CheckPassword.$(ARCH).o \ +$(SDIR)/SocketOps.$(ARCH).o + libs = \ $(DESTLIB)/libshell.a \ @@ -71,12 +86,22 @@ pantasks: $(BIN)/pantasks.$(ARCH) +$(BIN)/pantasks.$(ARCH) : $(single) $(funcs) $(cmds) $(libs) -$(BIN)/pantasks.$(ARCH) : $(funcs) $(cmds) $(libs) +pantasks_client: $(BIN)/pantasks_client.$(ARCH) +$(BIN)/pantasks_client.$(ARCH) : $(client) $(libs) -install: $(DESTBIN)/pantasks help +pantasks_server: $(BIN)/pantasks_server.$(ARCH) +$(BIN)/pantasks_server.$(ARCH) : $(server) $(funcs) $(cmds) $(libs) + @if [ ! -d $(BIN) ]; then mkdir -p $(BIN); fi + $(CC) -o $@ $^ $(LFLAGS) -lpthread + +pantasks_client.install: $(DESTBIN)/pantasks_client +pantasks_server.install: $(DESTBIN)/pantasks_server + +install: $(DESTBIN)/pantasks $(DESTBIN)/pantasks_client $(DESTBIN)/pantasks_server help help: cmd.basic.help cmd.data.help cmd.astro.help pantasks.help -.PHONY: pantasks +.PHONY: pantasks pantasks_client pantasks_server include ../Makefile.Common Index: /trunk/Ohana/src/opihi/pantasks/SocketOps.c =================================================================== --- /trunk/Ohana/src/opihi/pantasks/SocketOps.c (revision 7892) +++ /trunk/Ohana/src/opihi/pantasks/SocketOps.c (revision 7892) @@ -0,0 +1,245 @@ +# include "pantasks.h" + +# define MY_PORT 2000 +# define MY_WAIT 500 +# define DEBUG 0 + +static int NVALID; +static int Nvalid; +static int *VALID; + +int InitServerSocket (SockAddress *Address) { + + 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 + + Address[0].sin_family = AF_INET; + Address[0].sin_port = MY_PORT; + Address[0].sin_addr.s_addr = INADDR_ANY; // use this line to bind any address / port? + +# if (0) + status = inet_aton (hostip, &Address[0].sin_addr); + if (!status) { + fprintf (stderr, "invalid address\n"); + exit (2); + } +# endif + + length = sizeof(Address[0]); + + InitSocket = socket (PF_INET, SOCK_STREAM, 0); + if (InitSocket == -1) { + perror ("socket: "); + exit (2); + } + + if (DEBUG) fprintf (stderr, "init sock: %d, len: %d\n", InitSocket, length); + status = bind (InitSocket, (struct sockaddr *) Address, length); + if (status == -1) { + perror ("bind: "); + exit (2); + } + + status = listen (InitSocket, 10); + if (status == -1) { + perror ("listen: "); + exit (2); + } + return (InitSocket); +} + +int WaitServerSocket (int InitSocket, SockAddress *Address) { + + int i, BindSocket, length; + SockAddress Address_in; + u_int32_t addr; + + Address_in = Address[0]; + + length = sizeof(Address_in); + + /* this is a blocking wait; use in a separate thread */ + fcntl (InitSocket, F_SETFL, !O_NONBLOCK); + + if (DEBUG) fprintf (stderr, "init sock: %d, len: %d\n", InitSocket, length); + BindSocket = accept (InitSocket, (struct sockaddr *) &Address_in, &length); + if (DEBUG) fprintf (stderr, "bind sock: %d\n", BindSocket); + if (BindSocket == -1) { + perror ("accept: "); + exit (2); + } + + addr = Address_in.sin_addr.s_addr; + if (DEBUG) { + fprintf (stderr, "incoming connection from: "); + fprintf (stderr, " %u", (0xff & (addr >> 0))); + fprintf (stderr, ".%u", (0xff & (addr >> 8))); + fprintf (stderr, ".%u", (0xff & (addr >> 16))); + fprintf (stderr, ".%u", (0xff & (addr >> 24))); + fprintf (stderr, "\n"); + } + + if (Nvalid == 0) goto accepted; + + for (i = 0; i < Nvalid; i++) { + /* valid IP addresses may be machines (120.90.121.142) or + class C networks (120.90.121.0) */ + + /* for machine, address must match */ + if ((0xff & (VALID[i] >> 24)) != 0) { + if (addr == VALID[i]) goto accepted; + } + + /* for network, lower three bytes of address must match */ + if ((0xff & (VALID[i] >> 24)) == 0) { + if ((0x00ffffff & addr) == VALID[i]) goto accepted; + } + } + + if (DEBUG) fprintf (stderr, "connection rejected\n"); + close (BindSocket); + return (-1); + +accepted: + if (DEBUG) fprintf (stderr, "connection accepted\n"); + fcntl (BindSocket, F_SETFL, O_NONBLOCK); + return (BindSocket); +} + +int GetClientSocket (char *hostname) { + + int i, status, InitSocket, length; + SockAddress Address; + 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, "."); + } + + if (DEBUG) { + fprintf (stderr, "trying %s (%s:%d)...", host[0].h_name, hostip, MY_PORT); + } + + Address.sin_family = AF_INET; + Address.sin_port = MY_PORT; + status = inet_aton (hostip, &Address.sin_addr); + if (!status) { + fprintf (stderr, "invalid address\n"); + exit (2); + } + + length = sizeof(Address); + + InitSocket = socket (PF_INET, SOCK_STREAM, 0); + if (InitSocket == -1) { + perror ("socket: "); + exit (2); + } + + status = connect (InitSocket, (struct sockaddr *) &Address, length); + if (status == -1) { + perror ("connect: "); + exit (2); + } + + if (DEBUG) fprintf (stderr, "connected\n"); + fcntl (InitSocket, F_SETFL, O_NONBLOCK); + return (InitSocket); +} + +int InitServerSocket_Named (char *hostname, SockAddress *Address) { + + int i, status, InitSocket, length; + 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, "."); + } + + Address[0].sin_family = AF_INET; + Address[0].sin_port = MY_PORT; + status = inet_aton (hostip, &Address[0].sin_addr); + if (!status) { + fprintf (stderr, "invalid address\n"); + exit (2); + } + + length = sizeof(Address[0]); + + InitSocket = socket (PF_INET, SOCK_STREAM, 0); + if (InitSocket == -1) { + perror ("socket: "); + exit (2); + } + + if (DEBUG) fprintf (stderr, "init sock: %d, len: %d\n", InitSocket, length); + status = bind (InitSocket, (struct sockaddr *) Address, length); + if (status == -1) { + perror ("bind: "); + exit (2); + } + + status = listen (InitSocket, 10); + if (status == -1) { + perror ("listen: "); + exit (2); + } + + if (DEBUG) fprintf (stderr, "socket listening on %s (%s:%d)\n", host[0].h_name, hostip, MY_PORT); + return (InitSocket); +} + +/* load valid ip list */ +int DefineValidIP () { + + int i, Nvalid, ip1, ip2, ip3, ip4, test, status; + char string[80]; + + Nvalid = 0; + NVALID = 10; + ALLOCATE (VALID, int, NVALID); + for (i = 0; VarConfigEntry ("VALID_IP", "%s", i, string) != NULL; i++) { + status = sscanf (string, "%d.%d.%d.%d", &ip1, &ip2, &ip3, &ip4); + test = TRUE; + test &= (status == 4); + test &= ((ip1 > 0) && (ip1 < 256)); + test &= ((ip2 > 0) && (ip2 < 256)); + test &= ((ip3 > 0) && (ip3 < 256)); + test &= ((ip4 >=0) && (ip4 < 256)); + if (!test) { + fprintf (stderr, "invalid IP address %s\n", string); + exit (2); + } + VALID[Nvalid] = ip1 | (ip2 << 8) | (ip3 << 16) | (ip4 << 24); + Nvalid ++; + CHECK_REALLOCATE (VALID, int, NVALID, Nvalid, 10); + } + NVALID = Nvalid; + REALLOCATE (VALID, int, NVALID); + if (NVALID == 0) { + free (VALID); + VALID = NULL; + } +} Index: /trunk/Ohana/src/opihi/pantasks/client_shell.c =================================================================== --- /trunk/Ohana/src/opihi/pantasks/client_shell.c (revision 7892) +++ /trunk/Ohana/src/opihi/pantasks/client_shell.c (revision 7892) @@ -0,0 +1,124 @@ +# include "opihi.h" + +/******************/ +int client_shell (int argc, char **argv) { + + int Nbad, status, server; + char *line, *outline, *prompt, *history; + char hostname[256], PASSWORD[256]; + pid_t ppid; + IOBuffer message; + + general_init (&argc, argv); + program_init (&argc, argv); + startup (&argc, argv); + prompt = get_variable("PROMPT"); + history = get_variable("HISTORY"); + welcome (); + + /* connect to the server */ + if (VarConfig ("PANTASKS_SERVER", "%s", hostname) == NULL) { + fprintf (stderr, "pantasks server host undefined\n"); + exit (2); + } + server = GetClientSocket (hostname); + + /* here we can perform the security handshaking */ + VarConfig ("PASSWORD", "%s", PASSWORD); + SendCommand (server, strlen(PASSWORD), PASSWORD); + + Nbad = 0; + while (1) { /** must exit with command "exit" or "quit" */ + if (Nbad == 10) exit (3); + + line = readline (prompt); + + if (line == NULL) { + + ppid = getppid(); + if (ppid == 1) { + fprintf (stderr, "caught parent shutdown\n"); + exit (2); + } + if (!isatty (STDIN_FILENO)) exit (2); + fprintf (stdout, "Use \"quit\" to exit\n"); + Nbad ++; + continue; + } + Nbad = 0; + ohana_memregister (line); + + stripwhite (line); + + /* check for commands to be caught by client */ + /* use the standard command parser? */ + /* XXX the exit status of command does not allow us + to distinguish 'failed command' and 'command not found' */ + status = command (line, &outline); + if (outline == NULL) { + fprintf (stderr, "programming error: command_client returned NULL\n"); + exit (2); + } + + /* command was not caught by client, send to server */ + if (*outline && !status ) { + fprintf (stderr, "sending message to %d: %s\n", server, outline); + SendMessage (server, outline); + add_history (outline); + append_history (1, history); + + // XXX add this in and print to stdout + // status = ExpectMessage (server, 2.0, &message); + } + free (outline); + } +} + +/* + startup sequence: + + - general_init + - program_init + - startup (exit if non-interactive) + - welcome + +*/ + +/* client issues and questions: + +- need to identify the commands to be caught by the client +- pass each input line through the command parser first, then + send to server if not identified as a valid command +- this raises the question of variables: do we parse variables + at the client level? I would not think so. what about input? + if the input is to be parsed by the client, but executed on the server, + I'll need to re-work the input parsing system. would be better + for the server to parse the input command. in this case, the client + user needs to have access to the directories with input scripts. + this may be an advantage: it would allow some limitation on who can + install server scripts. +- variables, vectors and buffers: where are they valid (client or server?) + it seems they should all be defined locally on the server. this makes + graphics plotting a server-side action as well. this is probably easier, + but we do need to be careful about multiple clients trying to make plots + on the same window at the same time. Exporting the window can be done + for one client with the $KII variable, but could be tricky for more + complex interactions +- list of commands which clearly must be parsed by the client: + - quit / exit + - exec / ! + - ? + +- i may need an alternative version of command, modified to avoid parsing the + variables and vectors. I also need an alternative version of the Init commands + to just list the blocked ones. + +- the client now executes all of the basic opihi commands, and only passes + through the pantasks commands. The pantasks server interface should only + accept functions which immediately return, sending back the result to be + printed by the client. + + The server should not accept the 'task' commands from the client command-line. + To define and run macros or tasks on the server, these need to be added to + a script which is loaded with a varient on the input command. +*/ Index: /trunk/Ohana/src/opihi/pantasks/connect_to_server.c =================================================================== --- /trunk/Ohana/src/opihi/pantasks/connect_to_server.c (revision 7892) +++ /trunk/Ohana/src/opihi/pantasks/connect_to_server.c (revision 7892) @@ -0,0 +1,16 @@ +# include "pantasks.h" + +int connect_to_server () { + + char hostname[256]; + + ScanConfig (GlobalConfig, "PANTASKS_SERVER", "%s", 0, hostname); + + server = GetClientSocket (hostname); + + /* here we can perform the security handshaking */ + SendCommand (BindSocket, strlen(PASSWORD), PASSWORD); + + return (server); + +} Index: /trunk/Ohana/src/opihi/pantasks/init.c =================================================================== --- /trunk/Ohana/src/opihi/pantasks/init.c (revision 7891) +++ /trunk/Ohana/src/opihi/pantasks/init.c (revision 7892) @@ -34,5 +34,5 @@ {"kill", kill_job, "kill job"}, {"delete", delete_job, "delete job"}, - {"version", version, "show version information"}, + {"version", version, "show version information"}, {"verbose", verbose, "set/toggle verbose mode"}, }; Index: /trunk/Ohana/src/opihi/pantasks/init_client.c =================================================================== --- /trunk/Ohana/src/opihi/pantasks/init_client.c (revision 7892) +++ /trunk/Ohana/src/opihi/pantasks/init_client.c (revision 7892) @@ -0,0 +1,23 @@ +# include "pantasks.h" + +int invalid PROTO((int, char **)); + +/* we list here commands which are not valid for the client, + but represent valid pantasks commands. these are caught + by the client and sent to the 'invalid' function */ + +static Command cmds[] = { + {"task", invalid, "define a schedulable task"}, + {"task.exit", invalid, "define exit macros for a task"}, + {"task.exec", invalid, "define pre-exec macro for a task"}, +}; + +void InitPantasksClient () { + + int i; + + for (i = 0; i < sizeof (cmds) / sizeof (Command); i++) { + AddCommand (&cmds[i]); + } + +} Index: /trunk/Ohana/src/opihi/pantasks/invalid.c =================================================================== --- /trunk/Ohana/src/opihi/pantasks/invalid.c (revision 7892) +++ /trunk/Ohana/src/opihi/pantasks/invalid.c (revision 7892) @@ -0,0 +1,8 @@ +# include "pantasks.h" + +int invalid (int argc, char **argv) { + + fprintf (stderr, "%s is not valid for the pantasks client\n", argv[0]); + return (TRUE); + +} Index: /trunk/Ohana/src/opihi/pantasks/notes.txt =================================================================== --- /trunk/Ohana/src/opihi/pantasks/notes.txt (revision 7892) +++ /trunk/Ohana/src/opihi/pantasks/notes.txt (revision 7892) @@ -0,0 +1,36 @@ + +PanTasks Client / Server design + +Client + +client uses readline to parse the incoming commands commands are only +validated by the client before being sent to the server: + +client loop: + + line = readline () + validate argv[0] + send to server + wait for response? + +Server + +Server does not use readline at all. server has two threads. first +thread listens for commands from the clients. how many clients? use +poll to wait for data? can we manage enough clients with a single +thread? + +the second thread runs the scheduler loop: this is the same background +check that is being run by the current model on the timeouts. + +server loop: + + check socket for new incoming client connection + check known clients for new messages + execute new command + send response to the client + run the scheduler for a single loop + +client / server communication is modelled after addstar client / +server? + Index: /trunk/Ohana/src/opihi/pantasks/pantasks_client.c =================================================================== --- /trunk/Ohana/src/opihi/pantasks/pantasks_client.c (revision 7892) +++ /trunk/Ohana/src/opihi/pantasks/pantasks_client.c (revision 7892) @@ -0,0 +1,65 @@ +# include "pantasks.h" + +# define opihi_name "pantasks client shell" +# define opihi_prompt "pantasks: " +# define opihi_description "parallel task scheduler\n" +# define opihi_history ".pantasks" +# define opihi_rcfile ".pantasksrc" + +/* program-dependent initialization */ +void program_init (int *argc, char **argv) { + + /* load the commands used by this implementation */ + InitBasic (); + InitData (); + InitPantasksClient (); + + rl_readline_name = opihi_name; + rl_attempted_completion_function = command_completer; + rl_event_hook = NULL; + rl_set_keyboard_input_timeout (100000); + + set_str_variable ("HISTORY", opihi_history); + set_str_variable ("PROMPT", opihi_prompt); + set_str_variable ("RCFILE", opihi_rcfile); + + /* should we trap some commands and exec them locally? */ + /* help would be an obvious example... */ +# ifdef HELPDIR_DEFAULT + set_str_variable ("HELPDIR", MACRO_NAME(HELPDIR_DEFAULT)); +# endif + + signal (SIGPIPE, gotsignal); + signal (SIGTSTP, gotsignal); + signal (SIGTTIN, gotsignal); + return; +} + +/* standard welcome message */ +void welcome () { + fprintf (stderr, "\n"); + fprintf (stderr, "Welcome to %s - %s\n\n", opihi_name, opihi_description); +} + +/* add program-dependent exit functions here */ +/* unclear what we should do with the plotting functions: + do we keep the data only on the server, with all plotting + taking place on the server? or what? */ +void cleanup () { + // QuitImage (); + // QuitGraph (); + return; +} + +/* default signal handling */ +void gotsignal (int signum) { + fprintf (stderr, "got signal : %d\n", signum); + return; +} + +/* call to opihi shell */ +int main (int argc, char **argv) { + int status; + status = client_shell (argc, argv); + exit (status); +} Index: /trunk/Ohana/src/opihi/pantasks/pantasks_server.c =================================================================== --- /trunk/Ohana/src/opihi/pantasks/pantasks_server.c (revision 7892) +++ /trunk/Ohana/src/opihi/pantasks/pantasks_server.c (revision 7892) @@ -0,0 +1,81 @@ +# include "pantasks.h" + +/* pantasks-server is not a readline program. + it runs three (or four) threads: + - one thread listens on the socket for incomming connections + and performs client validation + - one thread parses commands from the connected clients + - one thread runs the scheduler loop + - one thread may run the controller loop independently + + * note: the client shell traps certain commands and performs them + locally. at a minimum: help. perhaps: input, other I/O commands? +*/ + +/* program-dependent initialization */ +int main (int argc, char **argv) { + + pthread_t thread; + int InitSocket, BindSocket; + SockAddress Address; + + general_init (&argc, argv); + DefineValidIP (); + + /* XXX not sure what is a good setting for this in the server */ + auto_break = TRUE; + + /* load the commands used by this implementation */ + InitBasic (); + InitData (); + InitPantasks (); + + // signal (SIGPIPE, gotsignal); + // signal (SIGTSTP, gotsignal); + // signal (SIGTTIN, gotsignal); + signal (SIGINT, SIG_DFL); + + // startup (&argc, argv); + + /* set up mechanisms for sharing data between the threads */ + // init_data_handling (); + + /* start up the background threads here */ + pthread_create (&thread, NULL, &ListenClients, NULL); + + // pthread_create (&thread, NULL, &RunScheduler, NULL); + // pthread_create (&thread, NULL, &pantasks_RunController, NULL); + + /* in this loop, we listen for incoming connections, validate, and + pass them to the ListenClient thread */ + + /* create the listening socket */ + InitSocket = InitServerSocket (&Address); + + InitPassword (); + + while (1) { + + /* wait for clients to make connection */ + BindSocket = WaitServerSocket (InitSocket, &Address); + if (BindSocket == -1) continue; + + /* validate : wait for password */ + if (!CheckPassword (BindSocket)) continue; + + AddNewClient (BindSocket); + } +} + +/* add program-dependent exit functions here */ +void cleanup () { + // QuitImage (); + // QuitGraph (); + // QuitController (); + return; +} + +void gotsignal (int signum) { + fprintf (stderr, "got signal : %d\n", signum); + return; +}