[vlc-commits] [Git][videolan/vlc][master] 3 commits: qt: fix use after free in ThreadRunner
Jean-Baptiste Kempf (@jbk)
gitlab at videolan.org
Wed Feb 5 10:28:25 UTC 2025
Jean-Baptiste Kempf pushed to branch master at VideoLAN / VLC
Commits:
fa64272d by Pierre Lamot at 2025-02-05T09:45:09+00:00
qt: fix use after free in ThreadRunner
Some tasks are not exectuted from the UI thread (QQuickImageResponse for
instance). This ensures that the target still exists when calling the callback
- - - - -
53025657 by Pierre Lamot at 2025-02-05T09:45:09+00:00
qt: fix queued task leak and use after free in ThreadRunner
As the serial task queue doesn't autodelete. When the task was canceled while
running, this caused the serial task to be leaked.
Moreover, the TaksRunner wasn't traking these tasks explicitly, the serial task
may try to acces the TaskRunner after it was deleted.
Now we assing a task ID to serial tasks to track them as the regular ones.
- - - - -
aef81693 by Pierre Lamot at 2025-02-05T09:45:09+00:00
qt: add unit test for ThreadRunner
- - - - -
5 changed files:
- modules/gui/qt/Makefile.am
- modules/gui/qt/medialibrary/mlthreadpool.cpp
- modules/gui/qt/medialibrary/mlthreadpool.hpp
- modules/gui/qt/meson.build
- + modules/gui/qt/tests/test_thread_runner.cpp
Changes:
=====================================
modules/gui/qt/Makefile.am
=====================================
@@ -1658,6 +1658,25 @@ test_renderer_manager_model_LDFLAGS = $(QT_QTEST_COMMON_ldflags)
check_PROGRAMS += test_renderer_manager_model
TESTS += test_renderer_manager_model
+# test_thread_runner
+
+test_thread_runner_SOURCES = \
+ tests/test_thread_runner.cpp \
+ medialibrary/mlthreadpool.cpp medialibrary/mlthreadpool.hpp
+
+nodist_test_thread_runner_SOURCES = \
+ tests/test_thread_runner.moc \
+ medialibrary/mlthreadpool.moc.cpp
+
+BUILT_SOURCES += tests/test_thread_runner.moc
+CLEANFILES += tests/test_thread_runner.moc
+test_thread_runner_CPPFLAGS = $(QT_QTEST_COMMON_cppflags)
+test_thread_runner_CXXFLAGS = $(QT_QTEST_COMMON_cxxflags)
+test_thread_runner_LDADD = $(QT_QTEST_COMMON_ldadd)
+test_thread_runner_LDFLAGS = $(QT_QTEST_COMMON_ldflags)
+check_PROGRAMS += test_thread_runner
+TESTS += test_thread_runner
+
# test_vlc_dialog_model
test_vlc_dialog_model_SOURCES = \
=====================================
modules/gui/qt/medialibrary/mlthreadpool.cpp
=====================================
@@ -20,52 +20,32 @@
#include <QMutexLocker>
-MLThreadPoolSerialTask::MLThreadPoolSerialTask(MLThreadPool* parent, const QString& queueName)
- : m_parent(parent)
- , m_queueName(queueName)
-{
- assert(m_parent);
-}
-
-
-void MLThreadPoolSerialTask::run()
+ThreadRunner::ThreadRunner()
{
- QRunnable* task = m_parent->getNextTaskFromQueue(m_queueName);
- if (!task)
- {
- deleteLater();
- return;
- }
- task->run();
- if (task->autoDelete())
- delete task;
- m_parent->start(this, nullptr);
+ m_threadPool.setMaxThreadCount(4);
}
-MLThreadPool::MLThreadPool()
+ThreadRunner::~ThreadRunner()
{
+ assert(m_objectTasks.empty());
+ assert(m_runningTasks.empty());
}
-
-MLThreadPool::~MLThreadPool()
+void ThreadRunner::setMaxThreadCount(size_t threadCount)
{
+ m_threadPool.setMaxThreadCount(threadCount);
}
-void MLThreadPool::setMaxThreadCount(size_t poolsize)
-{
- m_threadpool.setMaxThreadCount(poolsize);
-}
-
-void MLThreadPool::start(QRunnable* task, const char* queue)
+void ThreadRunner::start(QRunnable* task, const char* queue)
{
if (queue == nullptr)
{
- m_threadpool.start(task);
+ m_threadPool.start(task);
}
else
{
- QMutexLocker lock(&m_lock);
+ QMutexLocker lock(&m_serialTaskLock);
if (m_serialTasks.contains(queue))
{
m_serialTasks[queue].push_back(task);
@@ -74,21 +54,19 @@ void MLThreadPool::start(QRunnable* task, const char* queue)
{
m_serialTasks[queue] = QQueue<QRunnable*>();
m_serialTasks[queue].push_back(task);
- auto serialTasks = new MLThreadPoolSerialTask(this, queue);
- serialTasks->setAutoDelete(false);
- m_threadpool.start(serialTasks);
+ processQueueLocked(queue);
}
}
}
-bool MLThreadPool::tryTake(QRunnable* task)
+bool ThreadRunner::tryTake(QRunnable* task)
{
- bool ret = m_threadpool.tryTake(task);
+ bool ret = m_threadPool.tryTake(task);
if (ret)
return true;
{
- QMutexLocker lock(&m_lock);
+ QMutexLocker lock(&m_serialTaskLock);
for (auto queueIt = m_serialTasks.begin(); queueIt != m_serialTasks.end(); ++queueIt)
{
auto& queue = queueIt.value();
@@ -106,9 +84,9 @@ bool MLThreadPool::tryTake(QRunnable* task)
}
-QRunnable* MLThreadPool::getNextTaskFromQueue(const QString& queueName)
+QRunnable* ThreadRunner::getNextTaskFromQueue(const QString& queueName)
{
- QMutexLocker lock(&m_lock);
+ QMutexLocker lock(&m_serialTaskLock);
auto& queue = m_serialTasks[queueName];
if (queue.empty())
{
@@ -120,19 +98,38 @@ QRunnable* MLThreadPool::getNextTaskFromQueue(const QString& queueName)
return task;
}
-ThreadRunner::ThreadRunner()
+void ThreadRunner::processQueueLocked(const QString& queueName)
{
- m_threadPool.setMaxThreadCount(4);
-}
+ if (!m_serialTasks.contains(queueName))
+ return;
-ThreadRunner::~ThreadRunner()
-{
- assert(m_objectTasks.empty());
- assert(m_runningTasks.empty());
+ auto taskId = m_taskId++;
+ struct Ctx{};
+ auto runnable = new RunOnThreadRunner<Ctx>(taskId, this,
+ [this, queueName](Ctx&){
+ QRunnable* task = getNextTaskFromQueue(queueName);
+ if (!task)
+ {
+ return;
+ }
+ task->run();
+ if (task->autoDelete())
+ delete task;
+ },
+ [this, queueName](quint64, const Ctx&){
+ QMutexLocker lock(&m_serialTaskLock);
+ processQueueLocked(queueName);
+ });
+ connect(runnable, &RunOnThreadBaseRunner::done, this, &ThreadRunner::runOnThreadDone);
+ m_runningTasks.insert(taskId, runnable);
+ m_objectTasks.insert(this, taskId);
+
+ m_threadPool.start(runnable);
}
void ThreadRunner::destroy()
{
+ QMutexLocker locker{&m_lock};
m_shuttingDown = true;
//try to cancel as many tasks as possible
for (auto taskIt = m_objectTasks.begin(); taskIt != m_objectTasks.end(); /**/)
@@ -140,7 +137,7 @@ void ThreadRunner::destroy()
const QObject* object = taskIt.key();
quint64 key = taskIt.value();
auto task = m_runningTasks.value(key, nullptr);
- if (m_threadPool.tryTake(task))
+ if (tryTake(task))
{
delete task;
m_runningTasks.remove(key);
@@ -161,12 +158,13 @@ void ThreadRunner::destroy()
void ThreadRunner::cancelTask(const QObject* object, quint64 taskId)
{
assert(taskId != 0);
+ QMutexLocker locker{&m_lock};
auto task = m_runningTasks.value(taskId, nullptr);
if (!task)
return;
task->cancel();
- bool removed = m_threadPool.tryTake(task);
+ bool removed = tryTake(task);
if (removed)
delete task;
m_runningTasks.remove(taskId);
@@ -177,39 +175,58 @@ void ThreadRunner::cancelTask(const QObject* object, quint64 taskId)
void ThreadRunner::runOnThreadDone(RunOnThreadBaseRunner* runner, quint64 target, const QObject* object, int status)
{
+ QMutexLocker locker{&m_lock};
+ if (!m_runningTasks.contains(target))
+ {
+ runner->deleteLater();
+ return;
+ }
+
+ m_runningTasks.remove(target);
+ m_objectTasks.remove(object, target);
+ if (m_objectTasks.count(object) == 0)
+ disconnect(object, &QObject::destroyed, this, &ThreadRunner::runOnThreadTargetDestroyed);
+
if (m_shuttingDown)
{
- if (m_runningTasks.contains(target))
- {
- m_runningTasks.remove(target);
- m_objectTasks.remove(object, target);
- if (m_objectTasks.count(object) == 0)
- disconnect(object, &QObject::destroyed, this, &ThreadRunner::runOnThreadTargetDestroyed);
- }
if (m_runningTasks.empty())
deleteLater();
+ runner->deleteLater();
+ return;
}
- else if (m_runningTasks.contains(target))
+
+ if (status == ML_TASK_STATUS_SUCCEED)
{
- if (status == ML_TASK_STATUS_SUCCEED)
+ if (object->thread() == this->thread())
+ {
+ locker.unlock();
runner->runUICallback();
- m_runningTasks.remove(target);
- m_objectTasks.remove(object, target);
- if (m_objectTasks.count(object) == 0)
- disconnect(object, &QObject::destroyed, this, &ThreadRunner::runOnThreadTargetDestroyed);
+ runner->deleteLater();
+ }
+ else
+ {
+ //run the callback in the object thread
+ QMetaObject::invokeMethod(
+ const_cast<QObject*>(object),
+ [runner](){
+ runner->runUICallback();
+ runner->deleteLater();
+ }
+ );
+ }
}
- runner->deleteLater();
}
void ThreadRunner::runOnThreadTargetDestroyed(QObject * object)
{
+ QMutexLocker locker{&m_lock};
if (m_objectTasks.contains(object))
{
for (auto taskId : m_objectTasks.values(object))
{
auto task = m_runningTasks.value(taskId, nullptr);
assert(task);
- bool removed = m_threadPool.tryTake(task);
+ bool removed = tryTake(task);
if (removed)
delete task;
m_runningTasks.remove(taskId);
=====================================
modules/gui/qt/medialibrary/mlthreadpool.hpp
=====================================
@@ -26,60 +26,6 @@
#include <QThreadPool>
#include <QMutex>
-class MLThreadPool;
-
-//internal task MLThreadPool
-class MLThreadPoolSerialTask : public QObject, public QRunnable
-{
- Q_OBJECT
-public:
- MLThreadPoolSerialTask(MLThreadPool* parent, const QString& queueName);
-
- void run() override;
-
-private:
- MLThreadPool* m_parent = nullptr;
- QString m_queueName;
-};
-
-/**
- * @brief The MLThreadPool act like a QThreadPool, with the difference that it allows tasks
- * to be run sequentially by specifying a queue name when starting the task.
- */
-class MLThreadPool
-{
-public:
- explicit MLThreadPool();
- ~MLThreadPool();
-
-
- void setMaxThreadCount(size_t threadCount);
-
- /**
- * @brief start enqueue a QRunnable to be executed on the threadpool
- * @param task is the task to enqueue
- * @param queue, the name of the queue, all task with the same queue name will be
- * ran sequentially, if queue is null, task will be run without additional specific
- * constraint (like on QThreadPool)
- */
- void start(QRunnable* task, const char* queue = nullptr);
-
- /**
- * @brief tryTake atempt to the specified task from the queue if the task has not started
- * @return true if the task has been removed from the queue and was not started yet
- */
- bool tryTake(QRunnable* task);
-
-private:
- friend class MLThreadPoolSerialTask;
-
- QRunnable* getNextTaskFromQueue(const QString& queueName);
-
- QMutex m_lock;
- QThreadPool m_threadpool;
- QMap<QString, QQueue<QRunnable*>> m_serialTasks;
-};
-
class RunOnThreadBaseRunner;
class ThreadRunner : public QObject
@@ -95,6 +41,8 @@ public:
ThreadRunner();
~ThreadRunner();
+ void setMaxThreadCount(size_t threadCount);
+
void destroy();
void cancelTask(const QObject* object, quint64 taskId);
@@ -108,13 +56,38 @@ private slots:
void runOnThreadDone(RunOnThreadBaseRunner* runner, quint64 target, const QObject* object, int status);
void runOnThreadTargetDestroyed(QObject * object);
+
+private:
+ /**
+ * @brief start enqueue a QRunnable to be executed on the threadpool
+ * @param task is the task to enqueue
+ * @param queue, the name of the queue, all task with the same queue name will be
+ * ran sequentially, if queue is null, task will be run without additional specific
+ * constraint (like on QThreadPool)
+ */
+ void start(QRunnable* task, const char* queue = nullptr);
+
+ /**
+ * @brief tryTake atempt to the specified task from the queue if the task has not started
+ * @return true if the task has been removed from the queue and was not started yet
+ */
+ bool tryTake(QRunnable* task);
+
+ QRunnable* getNextTaskFromQueue(const QString& queueName);
+ void processQueueLocked(const QString& queueName);
+
private:
- MLThreadPool m_threadPool;
+ QThreadPool m_threadPool;
+
+ friend class MLThreadPoolSerialTask;
+ QMutex m_serialTaskLock;
+ QMap<QString, QQueue<QRunnable*>> m_serialTasks;
bool m_shuttingDown = false;
quint64 m_taskId = 1;
QMap<quint64, RunOnThreadBaseRunner*> m_runningTasks;
QMultiMap<const QObject*, quint64> m_objectTasks;
+ QMutex m_lock;
};
class RunOnThreadBaseRunner : public QObject, public QRunnable
@@ -182,16 +155,18 @@ quint64 ThreadRunner::runOnThread(const QObject* obj,
std::function<void (quint64 taskId, Ctx&)> uiFun,
const char* queue)
{
+ QMutexLocker locker{&m_lock};
+
if (m_shuttingDown)
return 0;
auto taskId = m_taskId++;
auto runnable = new RunOnThreadRunner<Ctx>(taskId, obj, mlFun, uiFun);
connect(runnable, &RunOnThreadBaseRunner::done, this, &ThreadRunner::runOnThreadDone);
- connect(obj, &QObject::destroyed, this, &ThreadRunner::runOnThreadTargetDestroyed);
+ connect(obj, &QObject::destroyed, this, &ThreadRunner::runOnThreadTargetDestroyed, Qt::DirectConnection);
m_runningTasks.insert(taskId, runnable);
m_objectTasks.insert(obj, taskId);
- m_threadPool.start(runnable, queue);
+ start(runnable, queue);
return taskId;
}
=====================================
modules/gui/qt/meson.build
=====================================
@@ -1224,6 +1224,25 @@ if qt6_dep.found()
'dependencies': [qt6_dep, qt_extra_deps, qtest_qt6_dep]
}
+ vlc_tests += {
+ 'name': 'test_qt_thread_runner',
+ 'sources': files(
+ 'tests/test_thread_runner.cpp',
+ 'medialibrary/mlthreadpool.cpp',
+ 'medialibrary/mlthreadpool.hpp'
+ ),
+ 'moc_sources': files(
+ 'tests/test_thread_runner.cpp'
+ ),
+ 'moc_headers': files(
+ 'medialibrary/mlthreadpool.hpp'
+ ),
+ 'suite': ['qt'],
+ 'include_directories' : qt_include_dir,
+ 'dependencies': [qt6_dep, qtest_qt6_dep]
+ }
+
+
vlc_tests += {
'name': 'test_qt_dialog_model',
'sources': files(
=====================================
modules/gui/qt/tests/test_thread_runner.cpp
=====================================
@@ -0,0 +1,810 @@
+/*****************************************************************************
+ * Copyright (C) 2025 VLC authors and VideoLAN
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; either version 2 of the License, or
+ * ( at your option ) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston MA 02110-1301, USA.
+ *****************************************************************************/
+
+#include <QTest>
+#include <QSignalSpy>
+#include <QWaitCondition>
+#include "../medialibrary/mlthreadpool.hpp"
+#include "qtestcase.h"
+
+static const int TASK_TIMEOUT = 50; //MS
+
+class Dummy : public QObject
+{
+ Q_OBJECT
+};
+
+//simple context for testing
+struct Ctx {
+ int id;
+};
+
+struct Barrier {
+ Barrier(size_t count)
+ : count(count)
+ {}
+
+ bool wait(unsigned long timeout = 0) {
+ QMutexLocker lock{&mutex};
+ --count;
+ if (count > 0)
+ {
+ bool ret = condition.wait(&mutex, timeout);
+ if (!ret)
+ return false;
+ }
+ else
+ condition.wakeAll();
+ return true;
+ }
+private:
+ int count;
+ QMutex mutex;
+ QWaitCondition condition;
+};
+
+class RunnerInThread : public QThread
+{
+ Q_OBJECT
+
+public:
+ void run() override {
+ m_runner = new ThreadRunner();
+ emit ready();
+ exec();
+ if (m_runner)
+ {
+ QSignalSpy deletedSpy(m_runner.get(), &QObject::destroyed);
+ m_runner->destroy();
+ deletedSpy.wait(TASK_TIMEOUT);
+ }
+ }
+
+ QPointer<ThreadRunner> m_runner;
+
+signals:
+ void ready();
+};
+
+
+class TestThreadRunner : public QObject
+{
+ Q_OBJECT
+
+private:
+
+ ThreadRunner* getRunner()
+ {
+ QFETCH_GLOBAL(bool, threaded);
+ if (threaded)
+ return m_runnerThread->m_runner.get();
+ else
+ return m_runner.get();
+ }
+
+ enum TaskStatus {
+ PENDING,
+ FAILED,
+ SUCCESS
+ };
+
+#define CHECK_TASK_TIMEOUT(task) QCOMPARE(QTest::qWaitFor([&task](){ return task != PENDING;}, TASK_TIMEOUT), false);
+#define CHECK_TASK_COMPLETE(task) QCOMPARE(QTest::qWaitFor([&task](){ return task != PENDING;}, TASK_TIMEOUT), true); QCOMPARE(task, SUCCESS);
+
+private slots:
+ void init() {
+ m_runner = new ThreadRunner();
+ m_runnerThread = std::make_unique<RunnerInThread>();
+ QSignalSpy startSpy(m_runnerThread.get(), &RunnerInThread::ready);
+ m_runnerThread->start();
+ startSpy.wait(TASK_TIMEOUT);
+ }
+
+ void cleanup() {
+ if (m_runner) {
+ QSignalSpy deletedSpy(m_runner.get(), &QObject::destroyed);
+ m_runner->destroy();
+ deletedSpy.wait();
+ }
+ m_runnerThread->quit();
+ m_runnerThread->wait();
+ }
+
+ void initTestCase_data()
+ {
+ QTest::addColumn<bool>("threaded");
+ QTest::newRow("same thread") << false;
+ QTest::newRow("different thread") << true;
+ }
+
+ void testSimpleTask()
+ {
+ ThreadRunner* runner = getRunner();
+ std::unique_ptr<Dummy> dummy = std::make_unique<Dummy>();
+ TaskStatus ret = PENDING;
+ runner->runOnThread<Ctx>(dummy.get(), [](Ctx& ctx){
+ ctx.id = 42;
+ }, [&ret](size_t, const Ctx& ctx){
+ if (ctx.id == 42)
+ ret = SUCCESS;
+ else
+ ret = FAILED;
+ });
+ CHECK_TASK_COMPLETE(ret);
+ }
+
+ void testParallelTask()
+ {
+ ThreadRunner* runner = getRunner();
+
+ std::unique_ptr<Dummy> dummy = std::make_unique<Dummy>();
+
+ Barrier barrierPre(3);
+
+ TaskStatus ret1 = PENDING;
+ TaskStatus ret2 = PENDING;
+
+ runner->runOnThread<Ctx>(
+ dummy.get(),
+ //T1
+ [&barrierPre, &ret1](Ctx&) {
+ if (!barrierPre.wait(TASK_TIMEOUT))
+ {
+ ret1 = FAILED;
+ return;
+ }
+ },
+ [&ret1](size_t, const Ctx&){
+ if (ret1 != FAILED)
+ ret1 = SUCCESS;
+ }
+ );
+
+ runner->runOnThread<Ctx>(
+ dummy.get(),
+ //T2
+ [&barrierPre, &ret2](Ctx&) {
+ if (!barrierPre.wait(TASK_TIMEOUT))
+ {
+ ret2 = FAILED;
+ return;
+ }
+ },
+ [&ret2](size_t, const Ctx&){
+ if (ret2 != FAILED)
+ ret2 = SUCCESS;
+ }
+ );
+
+ QVERIFY(barrierPre.wait(TASK_TIMEOUT));
+
+ CHECK_TASK_COMPLETE(ret1);
+ CHECK_TASK_COMPLETE(ret2);
+ }
+
+
+ void testParallelTaskQueued()
+ {
+ ThreadRunner* runner = getRunner();
+
+ std::unique_ptr<Dummy> dummy = std::make_unique<Dummy>();
+ QMutex mutex;
+
+ Barrier barrierPre(3);
+
+ TaskStatus ret1 = PENDING;
+ TaskStatus ret2 = PENDING;
+
+ runner->runOnThread<Ctx>(
+ dummy.get(),
+ //T1
+ [&mutex, &barrierPre, &ret1](Ctx&) {
+ if (!barrierPre.wait(TASK_TIMEOUT))
+ {
+ ret1 = FAILED;
+ return;
+ }
+ },
+ [&ret1](size_t, const Ctx&){
+ if (ret1 != FAILED)
+ ret1 = SUCCESS;
+ },
+ "queue1"
+ );
+
+ runner->runOnThread<Ctx>(
+ dummy.get(),
+ //T2
+ [&barrierPre, &ret2](Ctx&) {
+ if (!barrierPre.wait(TASK_TIMEOUT))
+ {
+ ret2 = FAILED;
+ return;
+ }
+ },
+ [&ret2](size_t, const Ctx&){
+ if (ret2 != FAILED)
+ ret2 = SUCCESS;
+ },
+ "queue2" //not the same queue
+ );
+
+ //Both tasks should be running at the same time
+ QVERIFY(barrierPre.wait(TASK_TIMEOUT));
+
+ CHECK_TASK_COMPLETE(ret1);
+ CHECK_TASK_COMPLETE(ret2);
+ }
+
+ void testQueuedTask()
+ {
+ ThreadRunner* runner = getRunner();
+ std::unique_ptr<Dummy> dummy = std::make_unique<Dummy>();
+
+ //theses tasks should execute sequentially
+ int ret = 0;
+ runner->runOnThread<Ctx>(dummy.get(),
+ [](Ctx& ){
+ //N/A
+ }, [&ret](size_t, const Ctx&){
+ if (ret != 0)
+ ret = -1;
+ else
+ ret = 1;
+ }, "queue");
+
+ runner->runOnThread<Ctx>(dummy.get(),
+ [](Ctx& ){
+ //N/A
+ },
+ [&ret](size_t, const Ctx& ){
+ if (ret != 1)
+ ret = -2;
+ else
+ ret = 2;
+ }, "queue");
+
+ runner->runOnThread<Ctx>(dummy.get(),
+ [](Ctx& ){
+ //N/A
+ },
+ [&ret](size_t, const Ctx& ){
+ if (ret != 2)
+ ret = -3;
+ else
+ ret = 3;
+ }, "queue");
+
+ QTRY_COMPARE_WITH_TIMEOUT(ret, 3, TASK_TIMEOUT);
+ }
+
+
+ void testTargetDeletedDuringTask()
+ {
+ ThreadRunner* runner = getRunner();
+
+ std::unique_ptr<Dummy> dummy = std::make_unique<Dummy>();
+
+ Barrier barrierPre(2);
+ Barrier barrierPost(2);
+
+ TaskStatus ret = PENDING;
+
+ runner->runOnThread<Ctx>(
+ dummy.get(),
+ [&barrierPre, &barrierPost, &ret](Ctx&) {
+ if (!barrierPre.wait(TASK_TIMEOUT))
+ {
+ ret = FAILED;
+ return;
+ }
+
+ //dummy is destroyed here in the other thread
+
+ if (!barrierPost.wait(TASK_TIMEOUT))
+ {
+ ret = FAILED;
+ return;
+ }
+ },
+ [&ret](size_t, const Ctx&){
+ //this should not be executed
+ ret = FAILED;
+ });
+
+ {
+ //wait for the task to be started
+ QVERIFY(barrierPre.wait(TASK_TIMEOUT));
+ dummy.reset();
+ QVERIFY(barrierPost.wait(TASK_TIMEOUT));
+ }
+ CHECK_TASK_TIMEOUT(ret);
+ }
+
+ void testTargetDeletedDuringTaskQueued()
+ {
+ ThreadRunner* runner = getRunner();
+
+ std::unique_ptr<Dummy> dummy = std::make_unique<Dummy>();
+
+ Barrier barrierPre(2);
+ Barrier barrierPost(2);
+
+ TaskStatus ret = PENDING;
+
+ runner->runOnThread<Ctx>(
+ dummy.get(),
+ [&barrierPre, &barrierPost, &ret](Ctx&) {
+ if (!barrierPre.wait(TASK_TIMEOUT))
+ {
+ ret = FAILED;
+ return;
+ }
+
+ //dummy is destroyed here in the other thread
+
+ if (!barrierPost.wait(TASK_TIMEOUT))
+ {
+ ret = FAILED;
+ return;
+ }
+ },
+ [&ret](size_t, const Ctx&){
+ //this should not be executed
+ ret = FAILED;
+ }, "queued");
+
+ //wait for the task to be started
+ QVERIFY(barrierPre.wait(TASK_TIMEOUT));
+ dummy.reset();
+ QVERIFY(barrierPost.wait(TASK_TIMEOUT));
+
+ CHECK_TASK_TIMEOUT(ret);
+ }
+
+ void testTargetDeletedBeforeStart()
+ {
+ ThreadRunner* runner = getRunner();
+
+
+ std::unique_ptr<Dummy> dummy = std::make_unique<Dummy>();
+
+ Barrier barrierPre(2);
+ Barrier barrierPost(2);
+
+ TaskStatus task1ret = PENDING;
+ TaskStatus task1UIret = PENDING;
+ TaskStatus task2ret = PENDING;
+
+ //there is only 1 worker thread so tasks will be executed consecutivelly
+ runner->setMaxThreadCount(1);
+
+ runner->runOnThread<Ctx>(
+ dummy.get(),
+ [&barrierPre, &barrierPost, &task1ret](Ctx&) {
+ if (!barrierPre.wait(TASK_TIMEOUT))
+ {
+ task1ret = FAILED;
+ return;
+ }
+
+ //dummy is destroyed here in the other thread
+
+ if (!barrierPost.wait(TASK_TIMEOUT))
+ {
+ task1ret = FAILED;
+ return;
+ }
+ task1ret = SUCCESS;
+ },
+ [&task1UIret](size_t, const Ctx&){
+ //target is deleted this should not be executed
+ task1UIret = FAILED;
+ });
+
+ runner->runOnThread<Ctx>(
+ dummy.get(),
+ [&task2ret](Ctx&) {
+ //this should not be executed
+ task2ret = FAILED;
+ },
+ [&task2ret](size_t, const Ctx&){
+ //this should not be executed
+ task2ret = FAILED;
+ });
+
+ //wait for the task to be started
+ QVERIFY(barrierPre.wait(TASK_TIMEOUT));
+ dummy.reset();
+ QVERIFY(barrierPost.wait(TASK_TIMEOUT));
+
+ //task1 thread callback execute
+ CHECK_TASK_COMPLETE(task1ret);
+ //this should timeout
+ CHECK_TASK_TIMEOUT(task2ret);
+ QCOMPARE(task1UIret, PENDING);
+ }
+
+ void testTargetDeletedBeforeStartQueued()
+ {
+ ThreadRunner* runner = getRunner();
+
+ std::unique_ptr<Dummy> dummy = std::make_unique<Dummy>();
+
+ Barrier barrierPre(2);
+ Barrier barrierPost(2);
+
+ TaskStatus task1ret = PENDING;
+ TaskStatus task1UIret = PENDING;
+ TaskStatus task2ret = PENDING;
+
+ //task are on the same queue so they will execute consecutively
+ runner->runOnThread<Ctx>(
+ dummy.get(),
+ [&barrierPre, &barrierPost, &task1ret](Ctx&) {
+ if (!barrierPre.wait(TASK_TIMEOUT))
+ {
+ task1ret = FAILED;
+ return;
+ }
+
+ //dummy is destroyed here in the other thread
+
+ if (!barrierPost.wait(TASK_TIMEOUT))
+ {
+ task1ret = FAILED;
+ return;
+ }
+ task1ret = SUCCESS;
+ },
+ [&task1UIret](size_t, const Ctx&){
+ //target is deleted this should not be executed
+ task1UIret = FAILED;
+ },
+ "samequeue");
+
+ runner->runOnThread<Ctx>(
+ dummy.get(),
+ [&task2ret](Ctx&) {
+ //this should not be executed
+ task2ret = FAILED;
+ },
+ [&task2ret](size_t, const Ctx&){
+ //this should not be executed
+ task2ret = FAILED;
+ },
+ "samequeue");
+
+ //wait for the task to be started
+ QVERIFY(barrierPre.wait(TASK_TIMEOUT));
+ dummy.reset();
+ QVERIFY(barrierPost.wait(TASK_TIMEOUT));
+
+ //task1 thread callback execute
+ CHECK_TASK_COMPLETE(task1ret);
+ //this should timeout
+ CHECK_TASK_TIMEOUT(task2ret);
+ QCOMPARE(task1UIret, PENDING);
+ }
+
+
+ void testTaskCanceledBeforeStart()
+ {
+ ThreadRunner* runner = getRunner();
+
+ std::unique_ptr<Dummy> dummy = std::make_unique<Dummy>();
+
+ Barrier barrierPre(2);
+ Barrier barrierPost(2);
+
+ TaskStatus task1ret = PENDING;
+ TaskStatus task2ret = PENDING;
+
+ //task are on the same queue so they will execute consecutively
+ runner->runOnThread<Ctx>(
+ dummy.get(),
+ [&barrierPre, &barrierPost, &task1ret](Ctx&) {
+ if (!barrierPre.wait(TASK_TIMEOUT))
+ {
+ task1ret = FAILED;
+ return;
+ }
+
+ //task is canceled here in the other thread
+
+ if (!barrierPost.wait(TASK_TIMEOUT))
+ {
+ task1ret = FAILED;
+ return;
+ }
+ },
+ [&task1ret](size_t, const Ctx&){
+ if (task1ret != FAILED)
+ task1ret = SUCCESS;
+ },
+ "samequeue");
+
+ quint64 task2 = runner->runOnThread<Ctx>(
+ dummy.get(),
+ [&task2ret](Ctx&) {
+ //this should not be executed
+ task2ret = FAILED;
+ },
+ [&task2ret](size_t, const Ctx&){
+ //this should not be executed
+ task2ret = FAILED;
+ },
+ "samequeue");
+
+ //wait for the task to be started
+ QVERIFY(barrierPre.wait(TASK_TIMEOUT));
+ runner->cancelTask(dummy.get(), task2);
+ QVERIFY(barrierPost.wait(TASK_TIMEOUT));
+
+ CHECK_TASK_COMPLETE(task1ret);
+ CHECK_TASK_TIMEOUT(task2ret);
+ }
+
+ void testTaskCanceledDuringExecution()
+ {
+ ThreadRunner* runner = getRunner();
+
+ std::unique_ptr<Dummy> dummy = std::make_unique<Dummy>();
+
+ Barrier barrierPre(2);
+ Barrier barrierPost(2);
+
+ TaskStatus taskRet = PENDING;
+
+ quint64 taskId = runner->runOnThread<Ctx>(
+ dummy.get(),
+ [&barrierPre, &barrierPost, &taskRet](Ctx&) {
+ if (!barrierPre.wait(TASK_TIMEOUT))
+ {
+ taskRet = FAILED;
+ return;
+ }
+
+ //task is canceled here in the other thread
+
+ if (!barrierPost.wait(TASK_TIMEOUT))
+ {
+ taskRet = FAILED;
+ return;
+ }
+ },
+ [&taskRet](size_t, const Ctx&){
+ //this should not be executed
+ taskRet = FAILED;
+ });
+
+ QVERIFY(barrierPre.wait(TASK_TIMEOUT));
+ runner->cancelTask(dummy.get(), taskId);
+ QVERIFY(barrierPost.wait(TASK_TIMEOUT));
+
+ CHECK_TASK_TIMEOUT(taskRet);
+ }
+
+ void testTaskCanceledDuringExecutionQueued()
+ {
+ ThreadRunner* runner = getRunner();
+
+ std::unique_ptr<Dummy> dummy = std::make_unique<Dummy>();
+
+ Barrier barrierPre(2);
+ Barrier barrierPost(2);
+
+ TaskStatus taskRet = PENDING;
+
+ quint64 taskId = runner->runOnThread<Ctx>(
+ dummy.get(),
+ [&barrierPre, &barrierPost, &taskRet](Ctx&) {
+ if (!barrierPre.wait(TASK_TIMEOUT))
+ {
+ taskRet = FAILED;
+ return;
+ }
+
+ //task is canceled here in the other thread
+
+ if (!barrierPost.wait(TASK_TIMEOUT))
+ {
+ taskRet = FAILED;
+ return;
+ }
+ },
+ [&taskRet](size_t, const Ctx&){
+ //this should not be executed
+ taskRet = FAILED;
+ }, "queued");
+
+ QVERIFY(barrierPre.wait(TASK_TIMEOUT));
+ runner->cancelTask(dummy.get(), taskId);
+ QVERIFY(barrierPost.wait(TASK_TIMEOUT));
+
+ CHECK_TASK_TIMEOUT(taskRet);
+ }
+
+ void testRunnerDestroyedDuringExecution()
+ {
+ ThreadRunner* runner = getRunner();
+
+ std::unique_ptr<Dummy> dummy = std::make_unique<Dummy>();
+
+ Barrier barrierPre(2);
+ Barrier barrierPost(2);
+
+ TaskStatus taskRet = PENDING;
+
+ runner->runOnThread<Ctx>(
+ dummy.get(),
+ [&barrierPre, &barrierPost, &taskRet](Ctx&) {
+ if (!barrierPre.wait(TASK_TIMEOUT))
+ {
+ taskRet = FAILED;
+ return;
+ }
+
+ //runner is destroyed here in the other thread
+
+ if (!barrierPost.wait(TASK_TIMEOUT))
+ {
+ taskRet = FAILED;
+ return;
+ }
+ },
+ [&taskRet](size_t, const Ctx&){
+ //this should not be executed
+ taskRet = FAILED;
+ });
+
+ QVERIFY(barrierPre.wait(TASK_TIMEOUT));
+ //destroy runner in its thread
+ if (runner->thread() == this->thread())
+ runner->destroy();
+ else
+ QMetaObject::invokeMethod(runner, &ThreadRunner::destroy, Qt::BlockingQueuedConnection);
+ QVERIFY(barrierPost.wait(TASK_TIMEOUT));
+
+ CHECK_TASK_TIMEOUT(taskRet);
+ }
+
+ void testRunnerDestroyedDuringExecutionQueued()
+ {
+ ThreadRunner* runner = getRunner();
+
+ std::unique_ptr<Dummy> dummy = std::make_unique<Dummy>();
+
+ Barrier barrierPre(2);
+ Barrier barrierPost(2);
+
+ TaskStatus taskRet = PENDING;
+
+ runner->runOnThread<Ctx>(
+ dummy.get(),
+ [&barrierPre, &barrierPost, &taskRet](Ctx&) {
+ if (!barrierPre.wait(TASK_TIMEOUT))
+ {
+ taskRet = FAILED;
+ return;
+ }
+
+ //runner is destroyed here in the other thread
+
+ if (!barrierPost.wait(TASK_TIMEOUT))
+ {
+ taskRet = FAILED;
+ return;
+ }
+ },
+ [&taskRet](size_t, const Ctx&){
+ //this should not be executed
+ taskRet = FAILED;
+ }, "queue");
+
+ QVERIFY(barrierPre.wait(TASK_TIMEOUT));
+ //destroy runner in its thread
+ if (runner->thread() == this->thread())
+ runner->destroy();
+ else
+ QMetaObject::invokeMethod(runner, &ThreadRunner::destroy, Qt::BlockingQueuedConnection);
+ QVERIFY(barrierPost.wait(TASK_TIMEOUT));
+
+ CHECK_TASK_TIMEOUT(taskRet);
+ }
+
+
+ void testNestedTask()
+ {
+ ThreadRunner* runner = getRunner();
+
+ std::unique_ptr<Dummy> dummy = std::make_unique<Dummy>();
+
+ TaskStatus taskRet = PENDING;
+
+ runner->runOnThread<Ctx>(
+ dummy.get(),
+ [](Ctx& ctx) {
+ //run on worker thread
+ ctx.id = 1;
+ },
+ [runner, &taskRet, dummy = dummy.get()](size_t, const Ctx& ctx){
+ if (ctx.id != 1)
+ {
+ taskRet = FAILED;
+ return;
+ }
+ runner->runOnThread<Ctx>(
+ dummy,
+ [](Ctx& ctx) {
+ ctx.id = 2;
+ },
+ [&taskRet](size_t, const Ctx& ctx){
+ taskRet = (ctx.id == 2 ? SUCCESS : FAILED);
+ });
+ }
+ );
+ CHECK_TASK_COMPLETE(taskRet);
+ }
+
+ void testNestedTaskQueued()
+ {
+ ThreadRunner* runner = getRunner();
+
+ std::unique_ptr<Dummy> dummy = std::make_unique<Dummy>();
+
+ TaskStatus taskRet = PENDING;
+
+ runner->runOnThread<Ctx>(
+ dummy.get(),
+ [](Ctx& ctx) {
+ //run on worker thread
+ ctx.id = 1;
+ },
+ [runner, &taskRet, dummy = dummy.get()](size_t, const Ctx& ctx){
+ if (ctx.id != 1)
+ {
+ taskRet = FAILED;
+ return;
+ }
+ runner->runOnThread<Ctx>(
+ dummy,
+ [](Ctx& ctx) {
+ ctx.id = 2;
+ },
+ [&taskRet](size_t, const Ctx& ctx){
+ taskRet = (ctx.id == 2 ? SUCCESS : FAILED);
+ },
+ "queue"
+ );
+ },
+ "queue"
+ );
+ CHECK_TASK_COMPLETE(taskRet);
+ }
+
+private:
+ //A runner living in the main thread
+ QPointer<ThreadRunner> m_runner;
+
+ //A runner living in a different thread
+ std::unique_ptr<RunnerInThread> m_runnerThread;
+};
+
+QTEST_GUILESS_MAIN(TestThreadRunner)
+#include "test_thread_runner.moc"
View it on GitLab: https://code.videolan.org/videolan/vlc/-/compare/c6af4fe30801e1a37acaa478698c80e5141854f6...aef81693d2a9f48b1d1d2015ceb0959102ef4f09
--
View it on GitLab: https://code.videolan.org/videolan/vlc/-/compare/c6af4fe30801e1a37acaa478698c80e5141854f6...aef81693d2a9f48b1d1d2015ceb0959102ef4f09
You're receiving this email because of your account on code.videolan.org.
VideoLAN code repository instance
More information about the vlc-commits
mailing list