eessppeelllloo/ffmpeg
1
0
Fork 0
mirror of https://git.ffmpeg.org/ffmpeg.git synced 2026-06-16 04:32:47 +02:00
Code Issues Packages Projects Releases Wiki Activity

avfilter/dnn_backend_torch: handle lifecycle of worker thread

Browse Source 
Initialize async resources in dnn_load_model_th and ensure proper cleanup and thread joining in dnn_free_model_th.

Signed-off-by: Raja Rathour <imraja729@gmail.com>
Reviewed-by: Wenbin Chen <wenbin.chen@intel.com>
Reviewed-by: Guo Yejun <yejun.guo@intel.com>
This commit is contained in:
Raja Rathour
2026-01-01 13:37:22 +05:30
committed by Guo Yejun
parent 5e8ad5c39b
commit 06d799ce1a
1 changed files with 70 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
libavfilter/dnn/dnn_backend_torch.cpp
+70 -17
View File
@@ -127,27 +127,70 @@ static void dnn_free_model_th(DNNModel **model)
if (!model || !*model)
return;
th_model = (THModel *) (*model);
while (ff_safe_queue_size(th_model->request_queue) != 0) {
THRequestItem *item = (THRequestItem *)ff_safe_queue_pop_front(th_model->request_queue);
destroy_request_item(&item);
}
ff_safe_queue_destroy(th_model->request_queue);
th_model = (THModel *)(*model);
while (ff_queue_size(th_model->lltask_queue) != 0) {
LastLevelTaskItem *item = (LastLevelTaskItem *)ff_queue_pop_front(th_model->lltask_queue);
av_freep(&item);
/* 1. Stop and join the worker thread if it exists */
if (th_model->worker_thread) {
{
std::lock_guard<std::mutex> lock(*th_model->mutex);
th_model->worker_stop = true;
}
th_model->cond->notify_all();
th_model->worker_thread->join();
delete th_model->worker_thread;
th_model->worker_thread = NULL;
}
ff_queue_destroy(th_model->lltask_queue);
while (ff_queue_size(th_model->task_queue) != 0) {
TaskItem *item = (TaskItem *)ff_queue_pop_front(th_model->task_queue);
av_frame_free(&item->in_frame);
av_frame_free(&item->out_frame);
av_freep(&item);
/* 2. Safely delete C++ synchronization objects */
if (th_model->mutex) {
delete th_model->mutex;
th_model->mutex = NULL;
}
ff_queue_destroy(th_model->task_queue);
delete th_model->jit_model;
if (th_model->cond) {
delete th_model->cond;
th_model->cond = NULL;
}
/* 3. Clean up the pending queue */
if (th_model->pending_queue) {
while (ff_safe_queue_size(th_model->pending_queue) > 0) {
THRequestItem *item = (THRequestItem *)ff_safe_queue_pop_front(th_model->pending_queue);
destroy_request_item(&item);
}
ff_safe_queue_destroy(th_model->pending_queue);
}
/* 4. Clean up standard backend queues */
if (th_model->request_queue) {
while (ff_safe_queue_size(th_model->request_queue) != 0) {
THRequestItem *item = (THRequestItem *)ff_safe_queue_pop_front(th_model->request_queue);
destroy_request_item(&item);
}
ff_safe_queue_destroy(th_model->request_queue);
}
if (th_model->lltask_queue) {
while (ff_queue_size(th_model->lltask_queue) != 0) {
LastLevelTaskItem *item = (LastLevelTaskItem *)ff_queue_pop_front(th_model->lltask_queue);
av_freep(&item);
}
ff_queue_destroy(th_model->lltask_queue);
}
if (th_model->task_queue) {
while (ff_queue_size(th_model->task_queue) != 0) {
TaskItem *item = (TaskItem *)ff_queue_pop_front(th_model->task_queue);
av_frame_free(&item->in_frame);
av_frame_free(&item->out_frame);
av_freep(&item);
}
ff_queue_destroy(th_model->task_queue);
}
/* 5. Final model cleanup */
if (th_model->jit_model)
delete th_model->jit_model;
av_freep(&th_model);
*model = NULL;
}
@@ -513,6 +556,16 @@ static DNNModel *dnn_load_model_th(DnnContext *ctx, DNNFunctionType func_type, A
goto fail;
}
th_model->pending_queue = ff_safe_queue_create();
if (!th_model->pending_queue) {
goto fail;
}
th_model->mutex = new std::mutex();
th_model->cond = new std::condition_variable();
th_model->worker_stop = false;
th_model->worker_thread = new std::thread(th_worker_thread, th_model);
model->get_input = &get_input_th;
model->get_output = &get_output_th;
model->filter_ctx = filter_ctx;