From: Dan Dennedy Date: Fri, 26 Apr 2013 03:48:09 +0000 (-0700) Subject: Make mlt_consumer_purge() more thorough. (SF-187) X-Git-Url: https://git.sesse.net/?a=commitdiff_plain;h=560868ae10f58dfe8fb6b603dd46031dee9ede00;p=mlt Make mlt_consumer_purge() more thorough. (SF-187) This is applied to SDL consumers only at the moment since that is what most applications are using. Needs to be extended to other consumers. --- diff --git a/configure b/configure index 690b7fe8..a5ae2f72 100755 --- a/configure +++ b/configure @@ -1,7 +1,7 @@ #!/bin/sh export version=0.8.9 -export soversion=5 +export soversion=6 show_help() { diff --git a/src/framework/mlt_consumer.c b/src/framework/mlt_consumer.c index 3421cb67..866ebf2b 100644 --- a/src/framework/mlt_consumer.c +++ b/src/framework/mlt_consumer.c @@ -745,7 +745,15 @@ static void *consumer_read_ahead_thread( void *arg ) pthread_mutex_lock( &self->queue_mutex ); while( self->ahead && mlt_deque_count( self->queue ) >= buffer ) pthread_cond_wait( &self->queue_cond, &self->queue_mutex ); - mlt_deque_push_back( self->queue, frame ); + if ( self->is_purge ) + { + mlt_frame_close( frame ); + self->is_purge = 0; + } + else + { + mlt_deque_push_back( self->queue, frame ); + } pthread_cond_broadcast( &self->queue_cond ); pthread_mutex_unlock( &self->queue_mutex ); @@ -973,7 +981,7 @@ static void *consumer_worker_thread( void *arg ) pthread_cond_broadcast( &self->done_cond ); pthread_mutex_unlock( &self->done_mutex ); } - mlt_events_fire( properties, "consumer-thread-stopped" ); + mlt_events_fire( properties, "consumer-thread-stopped", NULL ); return NULL; } @@ -1202,16 +1210,44 @@ static void consumer_work_stop( mlt_consumer self ) void mlt_consumer_purge( mlt_consumer self ) { - if ( self && self->ahead ) + if ( self ) { + pthread_mutex_lock( &self->put_mutex ); + if ( self->put ) { + mlt_frame_close( self->put ); + self->put = NULL; + } + pthread_cond_broadcast( &self->put_cond ); + pthread_mutex_unlock( &self->put_mutex ); + if ( self->ahead && self->real_time ) pthread_mutex_lock( &self->queue_mutex ); + + if ( self->purge ) + self->purge( self ); + while ( mlt_deque_count( self->queue ) ) mlt_frame_close( mlt_deque_pop_back( self->queue ) ); - if ( self->ahead && self->real_time ) { + if ( self->ahead && self->real_time ) + { + self->is_purge = 1; pthread_cond_broadcast( &self->queue_cond ); pthread_mutex_unlock( &self->queue_mutex ); + if ( abs( self->real_time ) > 1 ) + { + pthread_mutex_lock( &self->done_mutex ); + pthread_cond_broadcast( &self->done_cond ); + pthread_mutex_unlock( &self->done_mutex ); + } } + + pthread_mutex_lock( &self->put_mutex ); + if ( self->put ) { + mlt_frame_close( self->put ); + self->put = NULL; + } + pthread_cond_broadcast( &self->put_cond ); + pthread_mutex_unlock( &self->put_mutex ); } } @@ -1270,33 +1306,37 @@ static mlt_frame worker_get_frame( mlt_consumer self, mlt_properties properties while ( self->ahead && mlt_deque_count( self->queue ) < buffer ) { frame = mlt_consumer_get_frame( self ); - if ( ! frame ) - return frame; - pthread_mutex_lock( &self->queue_mutex ); - mlt_deque_push_back( self->queue, frame ); - pthread_cond_signal( &self->queue_cond ); - pthread_mutex_unlock( &self->queue_mutex ); + if ( frame ) + { + pthread_mutex_lock( &self->queue_mutex ); + mlt_deque_push_back( self->queue, frame ); + pthread_cond_signal( &self->queue_cond ); + pthread_mutex_unlock( &self->queue_mutex ); + } } // Wait if not realtime. - mlt_frame head_frame = MLT_FRAME( mlt_deque_peek_front( self->queue ) ); - while ( self->ahead && self->real_time < 0 && - !( head_frame && mlt_properties_get_int( MLT_FRAME_PROPERTIES( head_frame ), "rendered" ) ) ) + while ( self->ahead && self->real_time < 0 && !self->is_purge && + !( mlt_properties_get_int( MLT_FRAME_PROPERTIES( MLT_FRAME( mlt_deque_peek_front( self->queue ) ) ), "rendered" ) ) ) { pthread_mutex_lock( &self->done_mutex ); pthread_cond_wait( &self->done_cond, &self->done_mutex ); pthread_mutex_unlock( &self->done_mutex ); } - + // Get the frame from the queue. pthread_mutex_lock( &self->queue_mutex ); frame = mlt_deque_pop_front( self->queue ); pthread_mutex_unlock( &self->queue_mutex ); + if ( ! frame ) { + self->is_purge = 0; + return frame; + } // Adapt the worker process head to the runtime conditions. if ( self->real_time > 0 ) { - if ( frame && mlt_properties_get_int( MLT_FRAME_PROPERTIES( frame ), "rendered" ) ) + if ( mlt_properties_get_int( MLT_FRAME_PROPERTIES( frame ), "rendered" ) ) { self->consecutive_dropped = 0; if ( self->process_head > threads && self->consecutive_rendered >= self->process_head ) @@ -1339,7 +1379,11 @@ static mlt_frame worker_get_frame( mlt_consumer self, mlt_properties properties } } } - + if ( self->is_purge ) { + self->is_purge = 0; + mlt_frame_close( frame ); + frame = NULL; + } return frame; } diff --git a/src/framework/mlt_consumer.h b/src/framework/mlt_consumer.h index c76c0e89..e4415516 100644 --- a/src/framework/mlt_consumer.h +++ b/src/framework/mlt_consumer.h @@ -102,6 +102,12 @@ struct mlt_consumer_s */ int ( *is_stopped )( mlt_consumer ); + /** Purge the consumer of buffered data (virtual function). + * + * \param mlt_consumer a consumer + */ + void ( *purge )( mlt_consumer ); + /** The destructor virtual function * * \param mlt_consumer a consumer @@ -124,6 +130,7 @@ struct mlt_consumer_s int put_active; mlt_event event_listener; mlt_position position; + int is_purge; /* additional fields added for the parallel work queue */ mlt_deque worker_threads; diff --git a/src/modules/sdl/consumer_sdl.c b/src/modules/sdl/consumer_sdl.c index 8ec04057..35015008 100644 --- a/src/modules/sdl/consumer_sdl.c +++ b/src/modules/sdl/consumer_sdl.c @@ -66,6 +66,7 @@ struct consumer_sdl_s SDL_Rect rect; uint8_t *buffer; int bpp; + int is_purge; }; /** Forward references to static functions. @@ -74,6 +75,7 @@ struct consumer_sdl_s static int consumer_start( mlt_consumer parent ); static int consumer_stop( mlt_consumer parent ); static int consumer_is_stopped( mlt_consumer parent ); +static void consumer_purge( mlt_consumer parent ); static void consumer_close( mlt_consumer parent ); static void *consumer_thread( void * ); static int consumer_get_dimensions( int *width, int *height ); @@ -147,6 +149,7 @@ mlt_consumer consumer_sdl_init( mlt_profile profile, mlt_service_type type, cons parent->start = consumer_start; parent->stop = consumer_stop; parent->is_stopped = consumer_is_stopped; + parent->purge = consumer_purge; // Register specific events mlt_events_register( self->properties, "consumer-sdl-event", ( mlt_transmitter )consumer_sdl_event ); @@ -316,6 +319,20 @@ int consumer_is_stopped( mlt_consumer parent ) return !self->running; } +void consumer_purge( mlt_consumer parent ) +{ + consumer_sdl self = parent->child; + if ( self->running ) + { + pthread_mutex_lock( &self->video_mutex ); + while ( mlt_deque_count( self->queue ) ) + mlt_frame_close( mlt_deque_pop_back( self->queue ) ); + self->is_purge = 1; + pthread_cond_broadcast( &self->video_cond ); + pthread_mutex_unlock( &self->video_mutex ); + } +} + static int sdl_lock_display( ) { pthread_mutex_lock( &mlt_sdl_mutex ); @@ -824,8 +841,17 @@ static void *consumer_thread( void *arg ) // Push this frame to the back of the queue pthread_mutex_lock( &self->video_mutex ); - mlt_deque_push_back( self->queue, frame ); - pthread_cond_broadcast( &self->video_cond ); + if ( self->is_purge ) + { + mlt_frame_close( frame ); + frame = NULL; + self->is_purge = 0; + } + else + { + mlt_deque_push_back( self->queue, frame ); + pthread_cond_broadcast( &self->video_cond ); + } pthread_mutex_unlock( &self->video_mutex ); // Calculate the next playtime diff --git a/src/modules/sdl/consumer_sdl_audio.c b/src/modules/sdl/consumer_sdl_audio.c index 31e71829..43f58322 100644 --- a/src/modules/sdl/consumer_sdl_audio.c +++ b/src/modules/sdl/consumer_sdl_audio.c @@ -57,6 +57,7 @@ struct consumer_sdl_s pthread_cond_t refresh_cond; pthread_mutex_t refresh_mutex; int refresh_count; + int is_purge; }; /** Forward references to static functions. @@ -65,6 +66,7 @@ struct consumer_sdl_s static int consumer_start( mlt_consumer parent ); static int consumer_stop( mlt_consumer parent ); static int consumer_is_stopped( mlt_consumer parent ); +static void consumer_purge( mlt_consumer parent ); static void consumer_close( mlt_consumer parent ); static void *consumer_thread( void * ); static void consumer_refresh_cb( mlt_consumer sdl, mlt_consumer self, char *name ); @@ -121,6 +123,7 @@ mlt_consumer consumer_sdl_audio_init( mlt_profile profile, mlt_service_type type parent->start = consumer_start; parent->stop = consumer_stop; parent->is_stopped = consumer_is_stopped; + parent->purge = consumer_purge; // Initialize the refresh handler pthread_cond_init( &self->refresh_cond, NULL ); @@ -220,6 +223,20 @@ int consumer_is_stopped( mlt_consumer parent ) return !self->running; } +void consumer_purge( mlt_consumer parent ) +{ + consumer_sdl self = parent->child; + if ( self->running ) + { + pthread_mutex_lock( &self->video_mutex ); + while ( mlt_deque_count( self->queue ) ) + mlt_frame_close( mlt_deque_pop_back( self->queue ) ); + self->is_purge = 1; + pthread_cond_broadcast( &self->video_cond ); + pthread_mutex_unlock( &self->video_mutex ); + } +} + static void sdl_fill_audio( void *udata, uint8_t *stream, int len ) { consumer_sdl self = udata; @@ -356,7 +373,6 @@ static int consumer_play_video( consumer_sdl self, mlt_frame frame ) mlt_properties properties = self->properties; if ( self->running && !mlt_consumer_is_stopped( &self->parent ) ) mlt_events_fire( properties, "consumer-frame-show", frame, NULL ); - return 0; } @@ -528,8 +544,16 @@ static void *consumer_thread( void *arg ) if ( self->running && speed ) { pthread_mutex_lock( &self->video_mutex ); - mlt_deque_push_back( self->queue, frame ); - pthread_cond_broadcast( &self->video_cond ); + if ( self->is_purge ) + { + mlt_frame_close( frame ); + self->is_purge = 0; + } + else + { + mlt_deque_push_back( self->queue, frame ); + pthread_cond_broadcast( &self->video_cond ); + } pthread_mutex_unlock( &self->video_mutex ); // Calculate the next playtime diff --git a/src/modules/sdl/consumer_sdl_preview.c b/src/modules/sdl/consumer_sdl_preview.c index 7e00007e..9710bdfe 100644 --- a/src/modules/sdl/consumer_sdl_preview.c +++ b/src/modules/sdl/consumer_sdl_preview.c @@ -59,6 +59,7 @@ struct consumer_sdl_s static int consumer_start( mlt_consumer parent ); static int consumer_stop( mlt_consumer parent ); static int consumer_is_stopped( mlt_consumer parent ); +static void consumer_purge( mlt_consumer parent ); static void consumer_close( mlt_consumer parent ); static void *consumer_thread( void * ); static void consumer_frame_show_cb( mlt_consumer sdl, mlt_consumer self, mlt_frame frame ); @@ -99,6 +100,7 @@ mlt_consumer consumer_sdl_preview_init( mlt_profile profile, mlt_service_type ty parent->start = consumer_start; parent->stop = consumer_stop; parent->is_stopped = consumer_is_stopped; + parent->purge = consumer_purge; self->joined = 1; mlt_events_listen( MLT_CONSUMER_PROPERTIES( self->play ), self, "consumer-frame-show", ( mlt_listener )consumer_frame_show_cb ); mlt_events_listen( MLT_CONSUMER_PROPERTIES( self->still ), self, "consumer-frame-show", ( mlt_listener )consumer_frame_show_cb ); @@ -274,6 +276,13 @@ static int consumer_is_stopped( mlt_consumer parent ) return !self->running; } +void consumer_purge( mlt_consumer parent ) +{ + consumer_sdl self = parent->child; + if ( self->running ) + mlt_consumer_purge( self->play ); +} + static void *consumer_thread( void *arg ) { // Identify the arg