hey, i want to join xine breakage party! (what do you wanna break today? tm)

- time to cleanup demuxer mess... no more pthread stuff on any demuxer.
  of course i haven't tested all demuxers, but they at least compile.
  (sorry Mike, we will need to move a few variables around to make most of
  your demuxers work again - not big deal i think)
- make api more consistent by providing xine_open/xine_close
  and xine_start/xine_stop.

CVS patchset: 3058
CVS date: 2002/10/28 03:24:42

3 files changed, 179 insertions, 26 deletions

diff --git a/src/xine-engine/demux.c b/src/xine-engine/demux.c
index 59339f0e2..1fb4dac2c 100644
--- a/src/xine-engine/demux.c
+++ b/src/xine-engine/demux.c
@@ -23,10 +23,16 @@
  * $id$ 
  */
 
+#include <stdio.h>
+#include <string.h>
+#include <pthread.h>
 #include "xine_internal.h"
 #include "demuxers/demux.h"
 #include "buffer.h"
 
+#define LOG
+
+
 /* internal use only - called from demuxers on seek/stop
  * warning: after clearing decoders fifos an absolute discontinuity
  *          indication must be sent. relative discontinuities are likely
@@ -121,3 +127,120 @@ void xine_demux_control_end( xine_stream_t *stream, uint32_t flags ) {
     stream->audio_fifo->put (stream->audio_fifo, buf);
   }
 }
+
+static void *demux_loop (void *stream_gen) {
+
+  xine_stream_t *stream = (xine_stream_t *)stream_gen;
+  int status;
+  
+#ifdef LOG
+  printf ("demux: loop starting...\n");
+#endif
+    
+  pthread_mutex_lock( &stream->demux_lock );
+
+  /* do-while needed to seek after demux finished */
+  do {
+
+    /* main demuxer loop */
+    status = stream->demux_plugin->get_status(stream->demux_plugin);
+    while(status == DEMUX_OK && stream->demux_thread_running) {
+
+      status = stream->demux_plugin->send_chunk(stream->demux_plugin);
+
+      /* someone may want to interrupt us */
+      pthread_mutex_unlock( &stream->demux_lock );
+      pthread_mutex_lock( &stream->demux_lock );
+    }
+
+#ifdef LOG
+    printf ("demux: main demuxer loop finished (status: %d)\n", status);
+#endif
+    /* wait before sending end buffers: user might want to do a new seek */
+    while(stream->demux_thread_running && 
+          (!stream->video_fifo || stream->video_fifo->size(stream->video_fifo)) &&
+          (!stream->audio_fifo || stream->audio_fifo->size(stream->audio_fifo)) &&
+          status != DEMUX_OK ){
+      pthread_mutex_unlock( &stream->demux_lock );
+      xine_usec_sleep(100000);
+      pthread_mutex_lock( &stream->demux_lock );
+      status = stream->demux_plugin->get_status(stream->demux_plugin);
+    }
+
+  } while( status == DEMUX_OK && stream->demux_thread_running );
+
+#ifdef LOG
+  printf ("demux: loop finished (status: %d)\n", status);
+#endif
+  
+  /* demux_thread_running is zero is demux loop has being stopped by user */
+  if (stream->demux_thread_running) {
+    xine_demux_control_end(stream, BUF_FLAG_END_STREAM);
+  } else {
+    xine_demux_control_end(stream, BUF_FLAG_END_USER);
+  }
+  
+  stream->demux_thread_running = 0;
+  pthread_mutex_unlock( &stream->demux_lock );
+  
+  pthread_exit(NULL);
+
+  return NULL;
+}
+
+int xine_demux_start_thread (xine_stream_t *stream) {
+
+  int err;
+  
+#ifdef LOG
+  printf ("demux: start thread called\n");
+#endif
+  
+  pthread_mutex_lock( &stream->demux_lock );
+  
+  if( !stream->demux_thread_running ) {
+    
+    stream->demux_thread_running = 1;
+    if ((err = pthread_create (&stream->demux_thread,
+			       NULL, demux_loop, (void *)stream)) != 0) {
+      printf ("demux: can't create new thread (%s)\n",
+	      strerror(err));
+      abort();
+    }
+  }
+  
+  pthread_mutex_unlock( &stream->demux_lock );
+  return 0;
+}
+
+int xine_demux_stop_thread (xine_stream_t *stream) {
+  
+  void *p;
+  
+#ifdef LOG
+  printf ("demux: stop thread called\n");
+#endif
+  
+  pthread_mutex_lock( &stream->demux_lock );
+  stream->demux_thread_running = 0;
+  pthread_mutex_unlock( &stream->demux_lock );
+
+#ifdef LOG
+  printf ("demux: joining thread %d\n", stream->demux_thread );
+#endif
+  
+  /* FIXME: counter_lock isn't meant to protect demux_thread update.
+     however we can't use demux_lock here. should we create a new lock? */
+  pthread_mutex_lock (&stream->counter_lock);
+  
+  /* <join; demux_thread = 0;> must be atomic */
+  if( stream->demux_thread )
+    pthread_join (stream->demux_thread, &p);
+  stream->demux_thread = 0;
+  
+  pthread_mutex_unlock (&stream->counter_lock);
+    
+  xine_demux_flush_engine(stream);
+  
+  return 0;
+}
diff --git a/src/xine-engine/xine.c b/src/xine-engine/xine.c
index a4560e658..8bf736c54 100644
--- a/src/xine-engine/xine.c
+++ b/src/xine-engine/xine.c
@@ -17,7 +17,7 @@
  * along with this program; if not, write to the Free Software
  * Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA  02111-1307, USA
  *
- * $Id: xine.c,v 1.177 2002/10/27 01:52:15 guenter Exp $
+ * $Id: xine.c,v 1.178 2002/10/28 03:24:43 miguelfreitas Exp $
  *
  * top-level xine functions
  *
@@ -67,6 +67,9 @@ void xine_handle_stream_end (xine_stream_t *stream, int non_user) {
   if (stream->status == XINE_STATUS_QUIT)
     return;
   stream->status = XINE_STATUS_STOP;
+  
+  /* join thread if needed to fix resource leaks */  
+  xine_demux_stop_thread( stream );
     
   if (non_user) {
     /* frontends will not be interested in receiving this event
@@ -160,8 +163,8 @@ static void xine_stop_internal (xine_stream_t *stream) {
   printf ("xine_stop: stopping demux\n");
 #endif
   if (stream->demux_plugin) {
-    stream->demux_plugin->dispose (stream->demux_plugin);
-    stream->demux_plugin = NULL;
+    
+    xine_demux_stop_thread( stream );
 
     /*
      * wait until engine has really stopped
@@ -183,15 +186,6 @@ static void xine_stop_internal (xine_stream_t *stream) {
   printf ("xine_stop: demux stopped\n");
 #endif
 
-  /*
-   * close input plugin
-   */
-
-  if (stream->input_plugin) {
-    stream->input_plugin->dispose(stream->input_plugin);
-    stream->input_plugin = NULL;
-  }
-
   /* remove buffered samples from the sound device driver */
   if (stream->audio_out)
     stream->audio_out->control (stream->audio_out, AO_CTRL_FLUSH_BUFFERS);
@@ -217,6 +211,40 @@ void xine_stop (xine_stream_t *stream) {
   pthread_mutex_unlock (&stream->frontend_lock);
 }
 
+
+static void xine_close_internal (xine_stream_t *stream) {
+
+  xine_stop_internal( stream );
+  
+#ifdef LOG
+  printf ("xine_close: disposing demux\n");
+#endif
+  if (stream->demux_plugin) {
+    
+    stream->demux_plugin->dispose (stream->demux_plugin);
+    stream->demux_plugin = NULL;
+  }
+
+  /*
+   * close input plugin
+   */
+
+  if (stream->input_plugin) {
+    stream->input_plugin->dispose(stream->input_plugin);
+    stream->input_plugin = NULL;
+  }
+}
+
+void xine_close (xine_stream_t *stream) {
+
+  pthread_mutex_lock (&stream->frontend_lock);
+
+  xine_close_internal (stream);
+  
+  pthread_mutex_unlock (&stream->frontend_lock);
+}
+
+
 xine_stream_t *xine_stream_new (xine_t *this, 
 				xine_ao_driver_t *ao, xine_vo_driver_t *vo) {
 
@@ -270,6 +298,7 @@ xine_stream_t *xine_stream_new (xine_t *this,
    * init mutexes and conditions
    */
 
+  pthread_mutex_init (&stream->demux_lock, NULL);
   pthread_mutex_init (&stream->frontend_lock, NULL);
   pthread_mutex_init (&stream->event_queues_lock, NULL);
   pthread_mutex_init (&stream->osd_lock, NULL);
@@ -335,7 +364,7 @@ static int xine_open_internal (xine_stream_t *stream, const char *mrl) {
    * stop engine if necessary
    */
 
-  xine_stop_internal (stream);
+  xine_close_internal (stream);
 
 #ifdef LOG
   printf ("xine: engine should be stopped now\n");
@@ -363,8 +392,6 @@ static int xine_open_internal (xine_stream_t *stream, const char *mrl) {
   if (!(stream->demux_plugin=find_demux_plugin (stream, stream->input_plugin))) {
     xine_log (stream->xine, XINE_LOG_MSG,
 	      _("xine: couldn't find demux for >%s<\n"), mrl);
-    stream->input_plugin->dispose (stream->input_plugin);
-    stream->input_plugin = NULL;
     stream->err = XINE_ERROR_NO_DEMUX_PLUGIN;
 
     /* remove buffered samples from the sound device driver */
@@ -483,14 +510,8 @@ static int xine_play_internal (xine_stream_t *stream, int start_pos, int start_t
     return 0;
   }    
   
-  if (stream->status == XINE_STATUS_STOP) {
-
-    demux_status = stream->demux_plugin->start (stream->demux_plugin,
-						    pos, start_time);
-  } else {
-    demux_status = stream->demux_plugin->seek (stream->demux_plugin,
+  demux_status = stream->demux_plugin->seek (stream->demux_plugin,
 						   pos, start_time);
-  }
 
   if (demux_status != DEMUX_OK) {
     xine_log (stream->xine, XINE_LOG_MSG, 
@@ -498,12 +519,10 @@ static int xine_play_internal (xine_stream_t *stream, int start_pos, int start_t
     
     stream->err = XINE_ERROR_DEMUX_FAILED;
     
-    if (stream->status == XINE_STATUS_STOP)   
-      stream->input_plugin->dispose(stream->input_plugin);
-  
     return 0;
     
   } else {
+    xine_demux_start_thread( stream );
     stream->status = XINE_STATUS_PLAY;
   }
 
@@ -754,9 +773,13 @@ int xine_get_speed (xine_stream_t *stream) {
  */
 
 static int xine_get_stream_length (xine_stream_t *stream) {
+  
+  pthread_mutex_lock( &stream->demux_lock );
 
   if (stream->demux_plugin)
     return stream->demux_plugin->get_stream_length (stream->demux_plugin);
+  
+  pthread_mutex_unlock( &stream->demux_lock );
 
   return 0;
 }
diff --git a/src/xine-engine/xine_internal.h b/src/xine-engine/xine_internal.h
index 74f0a1db2..0f7e8baec 100644
--- a/src/xine-engine/xine_internal.h
+++ b/src/xine-engine/xine_internal.h
@@ -17,7 +17,7 @@
  * along with this program; if not, write to the Free Software
  * Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA  02111-1307, USA
  *
- * $Id: xine_internal.h,v 1.107 2002/10/27 01:52:15 guenter Exp $
+ * $Id: xine_internal.h,v 1.108 2002/10/28 03:24:44 miguelfreitas Exp $
  *
  */
 
@@ -201,6 +201,11 @@ struct xine_stream_s {
   /* event mechanism */
   xine_list_t               *event_queues;
   pthread_mutex_t            event_queues_lock;
+  
+  /* demux thread stuff */
+  pthread_t                  demux_thread;
+  int                        demux_thread_running;
+  pthread_mutex_t            demux_lock;
 
   int                        err;
 };
@@ -238,6 +243,8 @@ void xine_demux_control_newpts       (xine_stream_t *stream, int64_t pts, uint32
 void xine_demux_control_headers_done (xine_stream_t *stream);
 void xine_demux_control_start        (xine_stream_t *stream);
 void xine_demux_control_end          (xine_stream_t *stream, uint32_t flags);
+int xine_demux_start_thread          (xine_stream_t *stream);
+int xine_demux_stop_thread           (xine_stream_t *stream);
 
 /* 
  * plugin_loader functions