Changeset 7940

Timestamp:

Jul 19, 2006, 11:48:49 AM (20 years ago)

Author:

eugene

Message:

fixed some client/server I/O issues

Location:

trunk/Ohana/src/opihi

Files:

• include/pantasks.h (modified) (1 diff)
• lib.shell/multicommand.c (modified) (1 diff)
• pantasks/ListenClients.c (modified) (8 diffs)
• pantasks/pantasks_server.c (modified) (1 diff)

trunk/Ohana/src/opihi/include/pantasks.h

@@ -196 +196 @@
 void InitClients ();
 void AddNewClient (int client);
+int  DeleteClient (int client);
 void *ListenClients (void *data);
 

trunk/Ohana/src/opihi/lib.shell/multicommand.c

@@ -70 +70 @@
           }
 
+          // receive the command exit status
+          if (ExpectMessage (server, 2.0, &message)) {
+            switch (errno) {
+              case EPIPE:
+                gprint (GP_ERR, "server connection has died\n");
+                exit (1);
+              default:
+                gprint (GP_ERR, "I/O error sending server message\n");
+                exit (2);
+            }
+          } else {
+            sscanf (message.buffer, "STATUS %d", &status);
+          }
+
           // receive the resulting stderr
           if (ExpectMessage (server, 2.0, &message)) {

trunk/Ohana/src/opihi/pantasks/ListenClients.c

@@ -5 +5 @@
 static int Nclients;
 static int *clients;
-static IOBuffer *buffers;
+static IOBuffer **buffers;
 
 void InitClients () {
@@ -12 +12 @@
   NCLIENTS = 10;
   ALLOCATE (clients, int, NCLIENTS);
-  ALLOCATE (buffers, IOBuffer, NCLIENTS);
+  ALLOCATE (buffers, IOBuffer *, NCLIENTS);
 }
 
@@ -20 +20 @@
   if (DEBUG) gprint (GP_LOG, "adding a new client (%d)\n", client);
   clients[Nclients] = client;
-  InitIOBuffer(&buffers[Nclients], 256);
+  ALLOCATE (buffers[Nclients], IOBuffer, 1);
+  InitIOBuffer(buffers[Nclients], 256);
   Nclients ++;
   if (Nclients >= NCLIENTS - 1) {
     NCLIENTS += 10;
     REALLOCATE (clients, int, NCLIENTS);
-    REALLOCATE (buffers, IOBuffer, NCLIENTS);
+    REALLOCATE (buffers, IOBuffer *, NCLIENTS);
   }
+}
+
+/* add this client to the client table, create an IOBuffer for it */
+int DeleteClient (int client) {
+
+  int i, j;
+
+  if (DEBUG) gprint (GP_LOG, "deleting a client (%d)\n", client);
+  for (i = 0; i < Nclients; i++) {
+    if (clients[i] == client) {
+      FreeIOBuffer (buffers[i]);
+      free (buffers[i]);
+      close (clients[i]);
+      for (j = i; j < Nclients - 1; j++) {
+        clients[j] = clients[j+1];
+        buffers[j] = buffers[j+1];
+      }
+      Nclients --;
+      if ((Nclients > 10) && (Nclients / 2 < NCLIENTS)) {
+        NCLIENTS = Nclients + 10;
+        REALLOCATE (clients, int, NCLIENTS);
+        REALLOCATE (buffers, IOBuffer *, NCLIENTS);
+      }
+      return TRUE;
+    }
+  }
+  // did not find the client
+  return FALSE;
 }
 
@@ -56 +85 @@
       Nmax = MAX (Nmax, clients[i]);
       FD_SET (clients[i], &fdSet);
-      if (FD_ISSET(clients[i], &fdSet)) {
-        if (DEBUG) gprint (GP_ERR, "fd %d is set\n", clients[i]);
-      } else {
-        if (DEBUG) gprint (GP_ERR, "fd %d is not set\n", clients[i]);
-      }
     }    
     Nmax ++;
@@ -67 +91 @@
     if (DEBUG) gprint (GP_ERR, "listening to %d clients\n", Ncurrent);
     status = select (Nmax, &fdSet, NULL, NULL, &timeout);
-
-    if (DEBUG) gprint (GP_ERR, "messages from %d clients\n", status);
-
     if (status == -1) {
       perror("select()");
@@ -75 +96 @@
     }
 
-    /* if no data, go back for another select */
+    /* if no data, update client list, wait for another select */
     if (status <= 0) continue;
 
     /* loop over the clients with data */
     for (i = 0; i < Ncurrent; i++) {
+      /* if client has no data, skip it */
       if (!FD_ISSET(clients[i], &fdSet)) continue;
 
-      /* this client has data, handle it */
-      if (DEBUG) { 
-        gprint (GP_ERR, "data from client %d\n", clients[i]);
-        gprint (GP_ERR, "start: ", buffers[i].buffer);
-        gwrite (buffers[i].buffer, 1, buffers[i].Nbuffer, GP_ERR);
-        gprint (GP_ERR, "...\n");
+      /* read until the pipe is empty: 0 is closed, -1 is empty, -2 is error */
+      Nread = 1;
+      while (Nread > 0) {
+        Nread = ReadtoIOBuffer (buffers[i], clients[i]);        
       }
-        
-      /* 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) gprint (GP_ERR, "read %d bytes from socket\n", Nread);
+      if ((Nread == 0) || (Nread == -2)) {
+        /* error: do something */
+        if (DEBUG && (Nread == 0)) gprint (GP_ERR, "socket is closed\n");
+        if (DEBUG && (Nread == -2)) gprint (GP_ERR, "error reading from socket\n");
+        DeleteClient (clients[i]);
+        continue;
+      }
 
-        if (Nread == -2) {
-          /* error: do something */
-          if (1) gprint (GP_ERR, "error reading from socket\n");
-        }
-      }
-      if (DEBUG) gprint (GP_ERR, "read %d total bytes\n", buffers[i].Nbuffer);
+      if (DEBUG) gprint (GP_ERR, "read %d total bytes\n", buffers[i][0].Nbuffer);
 
-      if (DEBUG) { 
-        gprint (GP_ERR, "end: ", buffers[i].buffer);
-        gwrite (buffers[i].buffer, 1, buffers[i].Nbuffer, GP_ERR);
-        gprint (GP_ERR, "...\n");
-      }
-        
-      /* see if we have a complete message waiting */
-      line = CheckForMessage (&buffers[i]);
+      /* see if we have a complete message waiting; if not, keep waiting for messages */
+      line = CheckForMessage (buffers[i]);
       if (line == NULL) continue;
 
@@ -116 +126 @@
       /* in this thread, we set the print output destination to be an
          internal buffer, which we dump at the end of the execution */
-      /* XXX : we need to handle ; in the client-parsing of commands */
+      /* the commands sent to the server should not have ; */
       stripwhite (line);
       if (*line) {
 
-        if (1) fprintf (stderr, "got command: %s\n", line);
-
-        /* this works: we can send a simple message */
-        # if 0
-        SendMessage (clients[i], "first");
-        SendMessage (clients[i], "second");
-        # endif
-
-        /* this works: we can send a message using the IOBuffers */
-        # if 0
-        InitIOBuffer (&testbuffer, 64);
-        PrintIOBuffer (&testbuffer, "this is a test\n");
-        SendMessageFixed (clients[i], testbuffer.Nbuffer, testbuffer.buffer);
-        PrintIOBuffer (&testbuffer, "this is a second test\n");
-        SendMessageFixed (clients[i], testbuffer.Nbuffer, testbuffer.buffer);
-        # endif
-
-        /* this works: we can use the printing system to write a message */
-        # if 0
-        gprintSetBuffer (GP_LOG);
-        gprint (GP_LOG, "test is a test line\n");
-        outbuffer = gprintGetBuffer (GP_LOG);
-        SendMessageFixed (clients[i], outbuffer[0].Nbuffer, outbuffer[0].buffer);
-        
-        gprint (GP_LOG, "test is a second test line\n");
-        outbuffer = gprintGetBuffer (GP_LOG);
-        SendMessageFixed (clients[i], outbuffer[0].Nbuffer, outbuffer[0].buffer);
-        # endif
-
-        /* this works: we can use the gprint service */
-        # if 0
-        gprintSetBuffer (GP_LOG);
-        gprint (GP_LOG, "test is a test line\n");
-        status = multicommand (line);
-
-        outbuffer = gprintGetBuffer (GP_LOG);
-        SendMessageFixed (clients[i], outbuffer[0].Nbuffer, outbuffer[0].buffer);
-        gprintSetFile (GP_LOG, "stdout");
-        
-        SendMessage (clients[i], "second");
-        # endif
-
-        /* this only kind of works: it acts as if we hang */
-        # if 1
+        /* set buffers for the output for this client */
         gprintSetBuffer (GP_LOG);
         gprintSetBuffer (GP_ERR);
 
-        // gprint (GP_LOG, "test is a test stdout\n");
-        // gprint (GP_ERR, "test is a test stderr\n");
-
+        /* run the command, return the exit status */
         status = multicommand (line);
-        fprintf (stderr, "sending response\n");
+        SendMessage (clients[i], "STATUS %d", status);
 
         // return the stderr messages first
         outbuffer = gprintGetBuffer (GP_ERR);
         SendMessageFixed (clients[i], outbuffer[0].Nbuffer, outbuffer[0].buffer);
-        fprintf (stderr, "sent stderr\n");
 
+        // return the stdout messages first
         outbuffer = gprintGetBuffer (GP_LOG);
         SendMessageFixed (clients[i], outbuffer[0].Nbuffer, outbuffer[0].buffer);
-        fprintf (stderr, "sent stdout\n");
         
+        /* clear and reset the output buffers */
         gprintSetFile (GP_LOG, "stdout");
         gprintSetFile (GP_ERR, "stderr");
-        # endif
       }
       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 */
@@ -204 +162 @@
    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
-*/

trunk/Ohana/src/opihi/pantasks/pantasks_server.c

@@ -33 +33 @@
   gprintInit ();  // each thread needs to init the printing system
 
-  // signal (SIGPIPE, gotsignal);
+  signal (SIGPIPE, gotsignal);
   // signal (SIGTSTP, gotsignal);
   // signal (SIGTTIN, gotsignal);