[vlc-devel] [RFC 2/2] preparser: use the new executor API
Romain Vimont
rom1v at videolabs.io
Wed Aug 26 19:19:57 CEST 2020
Replace the background_worker by an executor.
---
src/preparser/preparser.c | 342 +++++++++++++++++++++-----------------
1 file changed, 187 insertions(+), 155 deletions(-)
diff --git a/src/preparser/preparser.c b/src/preparser/preparser.c
index 723feb96dc..c258963b1a 100644
--- a/src/preparser/preparser.c
+++ b/src/preparser/preparser.c
@@ -24,8 +24,9 @@
#include <vlc_common.h>
#include <vlc_atomic.h>
+#include <vlc_executor.h>
+#include <vlc_tick.h>
-#include "misc/background_worker.h"
#include "input/input_interface.h"
#include "input/input_internal.h"
#include "preparser.h"
@@ -35,218 +36,241 @@ struct input_preparser_t
{
vlc_object_t* owner;
input_fetcher_t* fetcher;
- struct background_worker* worker;
+ vlc_executor_t *executor;
+ vlc_tick_t default_timeout;
atomic_bool deactivated;
};
-typedef struct input_preparser_req_t
+struct input_preparser_task
{
+ input_preparser_t *preparser;
input_item_t *item;
input_item_meta_request_option_t options;
const input_preparser_callbacks_t *cbs;
void *userdata;
- vlc_atomic_rc_t rc;
-} input_preparser_req_t;
-typedef struct input_preparser_task_t
-{
- input_preparser_req_t *req;
- input_preparser_t* preparser;
- int preparse_status;
input_item_parser_id_t *parser;
- atomic_int state;
- atomic_bool done;
-} input_preparser_task_t;
-
-static input_preparser_req_t *ReqCreate(input_item_t *item,
- input_item_meta_request_option_t options,
- const input_preparser_callbacks_t *cbs,
- void *userdata)
+
+ vlc_mutex_t lock;
+ vlc_cond_t cond_ended;
+ bool preparse_ended;
+ int preparse_status;
+
+ bool fetch_ended;
+
+ atomic_bool interrupted;
+};
+
+static struct input_preparser_task *
+TaskNew(input_preparser_t *preparser, input_item_t *item,
+ input_item_meta_request_option_t options,
+ const input_preparser_callbacks_t *cbs, void *userdata)
{
- input_preparser_req_t *req = malloc(sizeof(*req));
- if (unlikely(!req))
+ struct input_preparser_task *task = malloc(sizeof(*task));
+ if (!task)
return NULL;
- req->item = item;
- req->options = options;
- req->cbs = cbs;
- req->userdata = userdata;
- vlc_atomic_rc_init(&req->rc);
+ task->preparser = preparser;
+ task->item = item;
+ task->options = options;
+ task->cbs = cbs;
+ task->userdata = userdata;
input_item_Hold(item);
- return req;
+ task->parser = NULL;
+ vlc_mutex_init(&task->lock);
+ vlc_cond_init(&task->cond_ended);
+ task->preparse_ended = false;
+ task->preparse_status = ITEM_PREPARSE_SKIPPED;
+ task->fetch_ended = false;
+
+ atomic_init(&task->interrupted, false);
+
+ return task;
}
-static void ReqHold(input_preparser_req_t *req)
+static void
+TaskDelete(struct input_preparser_task *task)
{
- vlc_atomic_rc_inc(&req->rc);
+ if (task->parser)
+ input_item_parser_id_Release(task->parser);
+ input_item_Release(task->item);
+ free(task);
}
-static void ReqRelease(input_preparser_req_t *req)
+static void
+OnParserEnded(input_item_t *item, int status, void *task_)
{
- if (vlc_atomic_rc_dec(&req->rc))
- {
- input_item_Release(req->item);
- free(req);
- }
+ VLC_UNUSED(item);
+ struct input_preparser_task *task = task_;
+
+ if (atomic_load(&task->interrupted))
+ /*
+ * The parsing interruption consists in calling
+ * input_item_parser_id_Release(), but this function happens to trigger
+ * this "parser ended" callback.
+ *
+ * We must ignore this case for two reasons:
+ * - the task lock is already held by the interrupt callback;
+ * - we want the interrupt callback to set the preparse_status anyway.
+ */
+ return;
+
+ vlc_mutex_lock(&task->lock);
+ assert(!task->preparse_ended);
+ task->preparse_status = status == VLC_SUCCESS ? ITEM_PREPARSE_DONE
+ : ITEM_PREPARSE_FAILED;
+ task->preparse_ended = true;
+ vlc_mutex_unlock(&task->lock);
+
+ vlc_cond_signal(&task->cond_ended);
}
-static void OnParserEnded(input_item_t *item, int status, void *task_)
+static void
+OnParserSubtreeAdded(input_item_t *item, input_item_node_t *subtree,
+ void *task_)
{
VLC_UNUSED(item);
- input_preparser_task_t* task = task_;
+ struct input_preparser_task *task = task_;
- atomic_store( &task->state, status );
- atomic_store( &task->done, true );
- background_worker_RequestProbe( task->preparser->worker );
+ if (task->cbs && task->cbs->on_subtree_added)
+ task->cbs->on_subtree_added(task->item, subtree, task->userdata);
}
-static void OnParserSubtreeAdded(input_item_t *item, input_item_node_t *subtree,
- void *task_)
+static void
+OnArtFetchEnded(input_item_t *item, bool fetched, void *userdata)
{
VLC_UNUSED(item);
- input_preparser_task_t* task = task_;
- input_preparser_req_t *req = task->req;
+ VLC_UNUSED(fetched);
+
+ struct input_preparser_task *task = userdata;
+
+ vlc_mutex_lock(&task->lock);
+ assert(!task->fetch_ended);
+ task->fetch_ended = true;
+ vlc_mutex_unlock(&task->lock);
- if (req->cbs && req->cbs->on_subtree_added)
- req->cbs->on_subtree_added(req->item, subtree, req->userdata);
+ vlc_cond_signal(&task->cond_ended);
}
-static int PreparserOpenInput( void* preparser_, void* req_, void** out )
-{
- input_preparser_t* preparser = preparser_;
- input_preparser_req_t *req = req_;
- input_preparser_task_t* task = malloc( sizeof *task );
+static const input_fetcher_callbacks_t input_fetcher_callbacks = {
+ .on_art_fetch_ended = OnArtFetchEnded,
+};
- if( unlikely( !task ) )
- goto error;
+static void
+RunnableRun(void *userdata)
+{
+ struct input_preparser_task *task = userdata;
static const input_item_parser_cbs_t cbs = {
.on_ended = OnParserEnded,
.on_subtree_added = OnParserSubtreeAdded,
};
- atomic_init( &task->state, VLC_ETIMEOUT );
- atomic_init( &task->done, false );
-
- task->preparser = preparser_;
- task->req = req;
- task->preparse_status = -1;
- task->parser = input_item_Parse( req->item, preparser->owner, &cbs,
- task );
- if( !task->parser )
- goto error;
-
- *out = task;
+ vlc_mutex_lock(&task->lock);
+ if (task->preparse_ended)
+ {
+ /* In case the task is already interrupted */
+ vlc_mutex_unlock(&task->lock);
+ return;
+ }
- return VLC_SUCCESS;
+ vlc_object_t *obj = task->preparser->owner;
+ task->parser = input_item_Parse(task->item, obj, &cbs, task);
+ if (!task->parser)
+ {
+ task->preparse_status = ITEM_PREPARSE_FAILED;
+ task->preparse_ended = true;
+ }
-error:
- free( task );
- if (req->cbs && req->cbs->on_preparse_ended)
- req->cbs->on_preparse_ended(req->item, ITEM_PREPARSE_FAILED, req->userdata);
- return VLC_EGENERIC;
-}
+ /* Wait until the end of parsing */
+ while (!task->preparse_ended)
+ vlc_cond_wait(&task->cond_ended, &task->lock);
-static int PreparserProbeInput( void* preparser_, void* task_ )
-{
- input_preparser_task_t* task = task_;
- return atomic_load( &task->done );
- VLC_UNUSED( preparser_ );
-}
-
-static void on_art_fetch_ended(input_item_t *item, bool fetched, void *userdata)
-{
- VLC_UNUSED(item);
- VLC_UNUSED(fetched);
- input_preparser_task_t *task = userdata;
- input_preparser_req_t *req = task->req;
+ if (task->fetch_ended)
+ {
+ /* Interrupted, do not request to fetch */
+ vlc_mutex_unlock(&task->lock);
+ return;
+ }
- input_item_SetPreparsed(req->item, true);
+ input_fetcher_t *fetcher = task->preparser->fetcher;
+ if (fetcher && (task->options & META_REQUEST_OPTION_FETCH_ANY))
+ {
+ int ret =
+ input_fetcher_Push(fetcher, task->item,
+ task->options & META_REQUEST_OPTION_FETCH_ANY,
+ &input_fetcher_callbacks, task);
+ if (ret != VLC_SUCCESS)
+ task->fetch_ended = true;
+
+ /* Wait until the end of fetching */
+ while (!task->fetch_ended)
+ vlc_cond_wait(&task->cond_ended, &task->lock);
+ }
- if (req->cbs && req->cbs->on_preparse_ended)
- req->cbs->on_preparse_ended(req->item, task->preparse_status, req->userdata);
+ vlc_mutex_unlock(&task->lock);
- ReqRelease(req);
- free(task);
+ input_item_SetPreparsed(task->item, true);
}
-static const input_fetcher_callbacks_t input_fetcher_callbacks = {
- .on_art_fetch_ended = on_art_fetch_ended,
-};
-
-static void PreparserCloseInput( void* preparser_, void* task_ )
+static void
+RunnableInterrupt(void *userdata)
{
- input_preparser_task_t* task = task_;
- input_preparser_req_t *req = task->req;
+ struct input_preparser_task *task = userdata;
- input_preparser_t* preparser = preparser_;
- input_item_t* item = req->item;
-
- int status;
- switch( atomic_load( &task->state ) )
- {
- case VLC_SUCCESS:
- status = ITEM_PREPARSE_DONE;
- break;
- case VLC_ETIMEOUT:
- status = ITEM_PREPARSE_TIMEOUT;
- break;
- default:
- status = ITEM_PREPARSE_FAILED;
- break;
- }
+ atomic_init(&task->interrupted, true);
- input_item_parser_id_Release( task->parser );
+ vlc_mutex_lock(&task->lock);
- if( preparser->fetcher && (req->options & META_REQUEST_OPTION_FETCH_ANY) )
- {
- task->preparse_status = status;
- ReqHold(task->req);
- if (!input_fetcher_Push(preparser->fetcher, item,
- req->options & META_REQUEST_OPTION_FETCH_ANY,
- &input_fetcher_callbacks, task))
- {
- return;
- }
- ReqRelease(task->req);
+ if (task->parser) {
+ input_item_parser_id_Release(task->parser);
+ task->parser = NULL;
}
- free(task);
+ task->preparse_status = ITEM_PREPARSE_TIMEOUT;
+ task->preparse_ended = true;
+ task->fetch_ended = true;
+ vlc_cond_signal(&task->cond_ended);
- input_item_SetPreparsed( item, true );
- if (req->cbs && req->cbs->on_preparse_ended)
- req->cbs->on_preparse_ended(req->item, status, req->userdata);
+ vlc_mutex_unlock(&task->lock);
}
-static void ReqHoldVoid(void *item) { ReqHold(item); }
-static void ReqReleaseVoid(void *item) { ReqRelease(item); }
+static void
+RunnableOnFinished(void *userdata)
+{
+ struct input_preparser_task *task = userdata;
+
+ if (task->cbs && task->cbs->on_preparse_ended)
+ task->cbs->on_preparse_ended(task->item, task->preparse_status,
+ task->userdata);
+
+ TaskDelete(task);
+}
input_preparser_t* input_preparser_New( vlc_object_t *parent )
{
input_preparser_t* preparser = malloc( sizeof *preparser );
+ if (!preparser)
+ return NULL;
- struct background_worker_config conf = {
- .default_timeout = VLC_TICK_FROM_MS(var_InheritInteger( parent, "preparse-timeout" )),
- .max_threads = var_InheritInteger( parent, "preparse-threads" ),
- .pf_start = PreparserOpenInput,
- .pf_probe = PreparserProbeInput,
- .pf_stop = PreparserCloseInput,
- .pf_release = ReqReleaseVoid,
- .pf_hold = ReqHoldVoid
- };
-
-
- if( likely( preparser ) )
- preparser->worker = background_worker_New( preparser, &conf );
+ int max_threads = var_InheritInteger(parent, "preparse-threads");
+ if (max_threads < 1)
+ max_threads = 1;
- if( unlikely( !preparser || !preparser->worker ) )
+ preparser->executor = vlc_executor_New(parent, max_threads);
+ if (!preparser->executor)
{
- free( preparser );
+ free(preparser);
return NULL;
}
+ preparser->default_timeout =
+ VLC_TICK_FROM_MS(var_InheritInteger(parent, "preparse-timeout"));
+ if (preparser->default_timeout < 0)
+ preparser->default_timeout = 0;
+
preparser->owner = parent;
preparser->fetcher = input_fetcher_New( parent );
atomic_init( &preparser->deactivated, false );
@@ -260,7 +284,7 @@ input_preparser_t* input_preparser_New( vlc_object_t *parent )
void input_preparser_Push( input_preparser_t *preparser,
input_item_t *item, input_item_meta_request_option_t i_options,
const input_preparser_callbacks_t *cbs, void *cbs_userdata,
- int timeout, void *id )
+ int timeout_ms, void *id )
{
if( atomic_load( &preparser->deactivated ) )
return;
@@ -287,14 +311,22 @@ void input_preparser_Push( input_preparser_t *preparser,
return;
}
- struct input_preparser_req_t *req = ReqCreate(item, i_options,
- cbs, cbs_userdata);
+ struct input_preparser_task *task =
+ TaskNew(preparser, item, i_options, cbs, cbs_userdata);
+ static const struct vlc_runnable runnable = {
+ .run = RunnableRun,
+ .interrupt = RunnableInterrupt,
+ .on_finished = RunnableOnFinished,
+ };
- if (background_worker_Push(preparser->worker, req, id, timeout))
- if (req->cbs && cbs->on_preparse_ended)
+ vlc_tick_t timeout = timeout_ms == -1 ? preparser->default_timeout
+ : VLC_TICK_FROM_MS(timeout_ms);
+ if (vlc_executor_Submit(preparser->executor, &runnable, task, id, timeout))
+ {
+ if (cbs && cbs->on_preparse_ended)
cbs->on_preparse_ended(item, ITEM_PREPARSE_FAILED, cbs_userdata);
-
- ReqRelease(req);
+ TaskDelete(task);
+ }
}
void input_preparser_fetcher_Push( input_preparser_t *preparser,
@@ -308,18 +340,18 @@ void input_preparser_fetcher_Push( input_preparser_t *preparser,
void input_preparser_Cancel( input_preparser_t *preparser, void *id )
{
- background_worker_Cancel( preparser->worker, id );
+ vlc_executor_Cancel(preparser->executor, id, true);
}
void input_preparser_Deactivate( input_preparser_t* preparser )
{
atomic_store( &preparser->deactivated, true );
- background_worker_Cancel( preparser->worker, NULL );
+ input_preparser_Cancel(preparser, NULL);
}
void input_preparser_Delete( input_preparser_t *preparser )
{
- background_worker_Delete( preparser->worker );
+ vlc_executor_Delete(preparser->executor);
if( preparser->fetcher )
input_fetcher_Delete( preparser->fetcher );
--
2.28.0
More information about the vlc-devel
mailing list