gst/tcp/gstmultifdsink.*: Recover from a select with a bad file descriptor by removing the client.

Original commit message from CVS:
* gst/tcp/gstmultifdsink.c: (gst_multifdsink_class_init),
(gst_multifdsink_add), (gst_multifdsink_remove),
(gst_multifdsink_clear), (gst_multifdsink_get_stats),
(gst_multifdsink_client_remove),
(gst_multifdsink_handle_client_write),
(gst_multifdsink_queue_buffer), (gst_multifdsink_handle_clients):
* gst/tcp/gstmultifdsink.h:
Recover from a select with a bad file descriptor by removing
the client.

ChangeLog

@@ -1,3 +1,15 @@
+2004-07-30  Wim Taymans  <wim@fluendo.com>
+
+	* gst/tcp/gstmultifdsink.c: (gst_multifdsink_class_init),
+	(gst_multifdsink_add), (gst_multifdsink_remove),
+	(gst_multifdsink_clear), (gst_multifdsink_get_stats),
+	(gst_multifdsink_client_remove),
+	(gst_multifdsink_handle_client_write),
+	(gst_multifdsink_queue_buffer), (gst_multifdsink_handle_clients):
+	* gst/tcp/gstmultifdsink.h:
+	Recover from a select with a bad file descriptor by removing
+	the client.
+
 2004-07-30  Thomas Vander Stichele  <thomas at apestaart dot org>
 
 	* configure.ac:

gst/tcp/gstmultifdsink.c

@@ -302,9 +302,12 @@ gst_multifdsink_add (GstMultiFdSink * sink, int fd)
   GstTCPClient *client;
   GTimeVal now;
 
+  GST_DEBUG_OBJECT (sink, "adding client on fd %d", fd);
+
   /* create client datastructure */
   client = g_new0 (GstTCPClient, 1);
   client->fd = fd;
+  client->bad = FALSE;
   client->bufpos = -1;
   client->bufoffset = 0;
   client->sending = NULL;
@@ -340,6 +343,8 @@ gst_multifdsink_remove (GstMultiFdSink * sink, int fd)
 {
   GList *clients;
 
+  GST_DEBUG_OBJECT (sink, "removing client on fd %d", fd);
+
   g_mutex_lock (sink->clientslock);
   /* loop over the clients to find the one with the fd */
   for (clients = sink->clients; clients; clients = g_list_next (clients)) {
@@ -358,6 +363,8 @@ gst_multifdsink_remove (GstMultiFdSink * sink, int fd)
 void
 gst_multifdsink_clear (GstMultiFdSink * sink)
 {
+  GST_DEBUG_OBJECT (sink, "clearing all clients");
+
   g_mutex_lock (sink->clientslock);
   while (sink->clients) {
     GstTCPClient *client;
@@ -918,41 +925,70 @@ gst_multifdsink_handle_clients (GstMultiFdSink * sink)
 
     /* < 0 is an error, 0 just means a timeout happened */
     if (result < 0) {
-      GST_ELEMENT_ERROR (sink, RESOURCE, READ, (NULL),
+      GST_WARNING_OBJECT (sink, "select failed: %s", g_strerror (errno));
-          ("select failed: %s", g_strerror (errno)));
+      if (errno == EBADF) {
-      return;
+        /* ok, so one of the fds is invalid. We loop over them to find one
-    }
+         * that gives an error to the F_GETFL fcntl. 
+         */
+        g_mutex_lock (sink->clientslock);
+        for (clients = sink->clients; clients; clients = g_list_next (clients)) {
+          GstTCPClient *client;
+          int fd;
+          long flags;
+          int res;
 
-    GST_LOG_OBJECT (sink, "%d sockets had action", result);
+          client = (GstTCPClient *) clients->data;
-    GST_LOG_OBJECT (sink, "done select on server/client fds for reads");
+          fd = client->fd;
-    gst_multifdsink_debug_fdset (sink, &testreadfds);
-    GST_LOG_OBJECT (sink, "done select on client fds for writes");
-    gst_multifdsink_debug_fdset (sink, &testwritefds);
 
-    /* read all commands */
+          res = fcntl (fd, F_GETFL, &flags);
-    if (FD_ISSET (READ_SOCKET (sink), &testreadfds)) {
+          if (res == -1) {
-      while (TRUE) {
+            GST_WARNING_OBJECT (sink, "fnctl failed for %d, marking as bad: %s",
-        gchar command;
+                fd, g_strerror (errno));
-        int res;
+            if (errno == EBADF) {
-
+              client->bad = TRUE;
-        READ_COMMAND (sink, command, res);
+            }
-        if (res < 0) {
+          }
-          /* no more commands */
-          break;
         }
+        g_mutex_unlock (sink->clientslock);
+      } else if (errno == EINTR) {
+        try_again = TRUE;
+      } else {
+        GST_ELEMENT_ERROR (sink, RESOURCE, READ, (NULL),
+            ("select failed: %s", g_strerror (errno)));
+        return;
+      }
+    } else {
+      GST_LOG_OBJECT (sink, "%d sockets had action", result);
+      GST_LOG_OBJECT (sink, "done select on server/client fds for reads");
+      gst_multifdsink_debug_fdset (sink, &testreadfds);
+      GST_LOG_OBJECT (sink, "done select on client fds for writes");
+      gst_multifdsink_debug_fdset (sink, &testwritefds);
 
-        switch (command) {
+      /* read all commands */
-          case CONTROL_RESTART:
+      if (FD_ISSET (READ_SOCKET (sink), &testreadfds)) {
-            /* need to restart the select call as the fd_set changed */
+        while (TRUE) {
-            try_again = TRUE;
+          gchar command;
-            break;
+          int res;
-          case CONTROL_STOP:
+
-            /* stop this function */
+          READ_COMMAND (sink, command, res);
-            stop = TRUE;
+          if (res < 0) {
-            break;
+            /* no more commands */
-          default:
-            g_warning ("multifdsink: unknown control message received");
             break;
+          }
+
+          switch (command) {
+            case CONTROL_RESTART:
+              /* need to restart the select call as the fd_set changed */
+              try_again = TRUE;
+              break;
+            case CONTROL_STOP:
+              /* stop this function */
+              stop = TRUE;
+              break;
+            default:
+              g_warning ("multifdsink: unknown control message received");
+              break;
+          }
         }
       }
     }
@@ -971,6 +1007,11 @@ gst_multifdsink_handle_clients (GstMultiFdSink * sink)
     int fd;
 
     client = (GstTCPClient *) clients->data;
+    if (client->bad) {
+      error = g_list_prepend (error, client);
+      continue;
+    }
+
     fd = client->fd;
 
     if (FD_ISSET (fd, &testreadfds)) {

gst/tcp/gstmultifdsink.h

@@ -81,6 +81,8 @@ typedef struct {
   int fd;
   gint bufpos;                  /* position of this client in the global queue */
 
+  gboolean bad;
+
   GList *sending;               /* the buffers we need to send */
   gint bufoffset;               /* offset in the first buffer */