More progress on jobmanager rewrite, port proxy task

src/bin/abstractprojectitem.cpp

@@ -48,7 +48,6 @@ AbstractProjectItem::AbstractProjectItem(PROJECTITEMTYPE type, QString id, const
     , m_usage(0)
     , m_rating(0)
     , m_clipStatus(FileStatus::StatusReady)
-    , m_clipJobProgress(0)
     , m_itemType(type)
     , m_lock(QReadWriteLock::Recursive)
     , m_isCurrent(false)
@@ -184,9 +183,8 @@ QVariant AbstractProjectItem::getData(DataType type) const
         break;
     case JobStatus:
         if (itemType() == ClipItem) {
-            if (m_clipJobProgress > 0) {
-                return QVariant::fromValue(JobManagerStatus::Running);
-            }
+            data = QVariant::fromValue(pCore->taskManager.jobStatus({ObjectType::BinClip, m_binId.toInt()}));
+            /*
             auto jobIds = pCore->jobManager()->getPendingJobsIds(clipId());
             if (jobIds.empty()) {
                 jobIds = pCore->jobManager()->getFinishedJobsIds(clipId());
@@ -195,18 +193,19 @@ QVariant AbstractProjectItem::getData(DataType type) const
                 data = QVariant::fromValue(pCore->jobManager()->getJobStatus(jobIds[0]));
             } else {
                 data = QVariant::fromValue(JobManagerStatus::NoJob);
-            }
+            }*/
         }
         break;
     case JobProgress:
         if (itemType() == ClipItem) {
-            return m_clipJobProgress;
+            data = pCore->taskManager.getJobProgressForClip({ObjectType::BinClip, m_binId.toInt()});
+            /*
             auto jobIds = pCore->jobManager()->getPendingJobsIds(clipId());
             if (jobIds.size() > 0) {
                 data = QVariant(pCore->jobManager()->getJobProgressForClip(jobIds[0], clipId()));
             } else {
                 data = QVariant(0);
-            }
+            }*/
         }
         break;
     case JobSuccess:

src/bin/abstractprojectitem.h

@@ -219,7 +219,6 @@ protected:
     uint m_rating;
     QString m_tags;
     FileStatus::ClipStatus m_clipStatus;
-    int m_clipJobProgress;
 
     PROJECTITEMTYPE m_itemType;
 

src/bin/bin.cpp

@@ -31,7 +31,7 @@ along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #include "effects/effectstack/model/effectstackmodel.hpp"
 #include "jobs/audiothumbjob.hpp"
 #include "jobs/transcodeclipjob.h"
-#include "jobs/jobmanager.h"
+#include "jobs/taskmanager.h"
 #include "jobs/thumbjob.hpp"
 #include "jobs/cliploadtask.h"
 #include "kdenlive_debug.h"
@@ -309,15 +309,15 @@ public:
                         reload.paint(painter, r);
                     }
                     int jobProgress = index.data(AbstractProjectItem::JobProgress).toInt();
-                    auto status = index.data(AbstractProjectItem::JobStatus).value<JobManagerStatus>();
-                    if (status == JobManagerStatus::Pending || status == JobManagerStatus::Running) {
+                    auto status = index.data(AbstractProjectItem::JobStatus).value<TaskManagerStatus>();
+                    if (status == TaskManagerStatus::Pending || status == TaskManagerStatus::Running) {
                         // Draw job progress bar
                         int progressWidth = option.fontMetrics.averageCharWidth() * 8;
                         int progressHeight = option.fontMetrics.ascent() / 4;
                         QRect progress(r1.x() + 1, opt.rect.bottom() - progressHeight - 2, progressWidth, progressHeight);
                         painter->setPen(Qt::NoPen);
                         painter->setBrush(Qt::darkGray);
-                        if (status == JobManagerStatus::Running) {
+                        if (status == TaskManagerStatus::Running) {
                             painter->drawRoundedRect(progress, 2, 2);
                             painter->setBrush((option.state & static_cast<int>((QStyle::State_Selected) != 0)) != 0 ? option.palette.text()
                                                                                                                     : option.palette.highlight());
@@ -1228,7 +1228,7 @@ Bin::Bin(std::shared_ptr<ProjectItemModel> model, QWidget *parent)
     // small info button for pending jobs
     m_infoLabel = new SmallJobLabel(this);
     m_infoLabel->setStyleSheet(SmallJobLabel::getStyleSheet(palette()));
-    connect(pCore->jobManager().get(), &JobManager::jobCount, m_infoLabel, &SmallJobLabel::slotSetJobCount);
+    connect(&pCore->taskManager, &TaskManager::jobCount, m_infoLabel, &SmallJobLabel::slotSetJobCount);
     QAction *infoAction = m_toolbar->addWidget(m_infoLabel);
     m_jobsMenu = new QMenu(this);
     // connect(m_jobsMenu, &QMenu::aboutToShow, this, &Bin::slotPrepareJobsMenu);
@@ -1247,14 +1247,15 @@ Bin::Bin(std::shared_ptr<ProjectItemModel> model, QWidget *parent)
     connect(m_discardCurrentClipJobs, &QAction::triggered, this, [&]() {
         const QString currentId = m_monitor->activeClipId();
         if (!currentId.isEmpty()) {
-            pCore->jobManager()->discardJobs(currentId);
+            pCore->taskManager.discardJobs({ObjectType::BinClip,currentId.toInt()});
         }
     });
     connect(m_cancelJobs, &QAction::triggered, [&]() {
-        pCore->jobManager()->slotCancelJobs();
+        pCore->taskManager.slotCancelJobs();
     });
     connect(m_discardPendingJobs, &QAction::triggered, [&]() {
-        pCore->jobManager()->slotCancelPendingJobs();
+        // TODO: implement pending only deletion
+        pCore->taskManager.slotCancelJobs();
     });
 
     // Hack, create toolbar spacer
@@ -1287,7 +1288,7 @@ Bin::Bin(std::shared_ptr<ProjectItemModel> model, QWidget *parent)
 
 Bin::~Bin()
 {
-    pCore->jobManager()->slotCancelJobs();
+    pCore->taskManager.slotCancelJobs();
     blockSignals(true);
     m_proxyModel->selectionModel()->blockSignals(true);
     setEnabled(false);
@@ -4051,7 +4052,7 @@ void Bin::reloadAllProducers(bool reloadThumbs)
         }
         if (!xml.isNull()) {
             clip->setClipStatus(FileStatus::StatusWaiting);
-            pCore->jobManager()->slotDiscardClipJobs(clip->clipId());
+            pCore->taskManager.discardJobs({ObjectType::BinClip, clip->clipId().toInt()});
             clip->discardAudioThumb();
             // We need to set a temporary id before all outdated producers are replaced;
             //int jobId = pCore->jobManager()->startJob<LoadJob>({clip->clipId()}, -1, QString(), xml);
@@ -4118,7 +4119,7 @@ void Bin::rebuildProxies()
         if (clp->hasProxy()) {
             toProxy << clp;
             // Abort all pending jobs
-            pCore->jobManager()->discardJobs(clp->clipId(), AbstractClipJob::PROXYJOB);
+            pCore->taskManager.discardJobs({ObjectType::BinClip,clp->clipId().toInt()}, AbstractTask::PROXYJOB);
             clp->deleteProxy();
         }
     }
@@ -4492,7 +4493,7 @@ void Bin::requestTranscoding(const QString &url, const QString &id)
         m_transcodingDialog = new TranscodeSeek(this);
         connect(m_transcodingDialog, &QDialog::accepted, this, [=] () {
             qDebug()<<"==== STARTING TCODE JOB: "<<m_transcodingDialog->ids().front()<<" = "<<m_transcodingDialog->params();
-            pCore->jobManager()->startJob<TranscodeJob>(m_transcodingDialog->ids(), -1, QString(), m_transcodingDialog->params(), true);
+            //pCore->jobManager()->startJob<TranscodeJob>(m_transcodingDialog->ids(), -1, QString(), m_transcodingDialog->params(), true);
             delete m_transcodingDialog;
             m_transcodingDialog = nullptr;
         });

src/bin/projectclip.cpp

@@ -30,6 +30,7 @@ along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #include "jobs/jobmanager.h"
 #include "jobs/audiolevelstask.h"
 #include "jobs/cliploadtask.h"
+#include "jobs/proxytask.h"
 #include "jobs/cachejob.hpp"
 #include "kdenlivesettings.h"
 #include "lib/audio/audioStreamInfo.h"
@@ -1247,7 +1248,8 @@ void ProjectClip::setProperties(const QMap<QString, QString> &properties, bool r
             // A proxy was requested, make sure to keep original url
             setProducerProperty(QStringLiteral("kdenlive:originalurl"), url());
             backupOriginalProperties();
-            emit pCore->jobManager()->startJob<ProxyJob>({clipId()}, -1, QString());
+            ProxyTask::start({ObjectType::BinClip,m_binId.toInt()}, this);
+            //emit pCore->jobManager()->startJob<ProxyJob>({clipId()}, -1, QString());
         }
     } else if (!reload) {
         const QList<QString> propKeys = properties.keys();
@@ -1308,7 +1310,8 @@ void ProjectClip::setProperties(const QMap<QString, QString> &properties, bool r
         if (hasProxy()) {
             pCore->jobManager()->discardJobs(clipId(), AbstractClipJob::PROXYJOB);
             setProducerProperty(QStringLiteral("_overwriteproxy"), 1);
-            emit pCore->jobManager()->startJob<ProxyJob>({clipId()}, -1, QString());
+            //emit pCore->jobManager()->startJob<ProxyJob>({clipId()}, -1, QString());
+            ProxyTask::start({ObjectType::BinClip,m_binId.toInt()}, this);
         } else {
             reloadProducer(refreshOnly, properties.contains(QStringLiteral("kdenlive:proxy")));
         }
@@ -2026,13 +2029,8 @@ void ProjectClip::updateTimelineOnReload()
     }
 }
 
-void ProjectClip::audioJobProgress(int progress)
+void ProjectClip::updateJobProgress()
 {
-    if (progress == 100) {
-        m_clipJobProgress = 0;
-    } else {
-        m_clipJobProgress = progress;
-    }
     if (auto ptr = m_model.lock()) {
         std::static_pointer_cast<ProjectItemModel>(ptr)->onItemUpdated(m_binId, AbstractProjectItem::JobProgress);
     }

src/bin/projectclip.h

@@ -269,7 +269,8 @@ public slots:
     void updateAudioThumbnail();
     /** @brief Delete the proxy file */
     void deleteProxy();
-    void audioJobProgress(int progress);
+    /** @brief A clip job progressed, update display */
+    void updateJobProgress();
 
     /** @brief Sets thumbnail for this clip. */
     void setThumbnail(const QImage &);

src/jobs/CMakeLists.txt

@@ -6,6 +6,7 @@ set(kdenlive_SRCS
   jobs/taskmanager.cpp
   jobs/audiolevelstask.cpp
   jobs/cliploadtask.cpp
+  jobs/proxytask.cpp
   jobs/jobmanager.cpp
   jobs/cachejob.cpp
   jobs/loadjob.cpp

src/jobs/abstracttask.cpp

@@ -40,6 +40,7 @@ AbstractTask::AbstractTask(const ObjectId &owner, JOBTYPE type, QObject* object)
     , m_progress(0)
     , m_isCanceled(false)
     , m_isForce(false)
+    , m_running(false)
     , m_type(type)
 {
     setAutoDelete(true);
@@ -63,6 +64,12 @@ AbstractTask::AbstractTask(const ObjectId &owner, JOBTYPE type, QObject* object)
     }
 }
 
+void AbstractTask::cancelJob()
+{
+    m_isCanceled = true;
+    emit jobCanceled();
+}
+
 const ObjectId AbstractTask::ownerId() const
 {
     return m_owner;
@@ -79,6 +86,5 @@ bool AbstractTask::operator==(AbstractTask &b)
 
 void AbstractTask::run()
 {
-    // 2 channels interleaved of uchar values
-    QMutexLocker lk(&m_runMutex);
+    qDebug()<<"============0\n\nABSTRACT TASKSTARTRING\n\n==================";
 }

src/jobs/abstracttask.h

@@ -25,8 +25,9 @@
 #include <QMutex>
 #include <QObject>
 
-class AbstractTask : public QRunnable
+class AbstractTask : public QObject, public QRunnable
 {
+    Q_OBJECT
     friend class TaskManager;
 
 public:
@@ -57,14 +58,19 @@ protected:
     bool m_successful;
     bool m_isCanceled;
     bool m_isForce;
+    bool m_running;
     void run() override;
+    QMutex m_runMutex;
     void cleanup();
 
 private:
     //QString cacheKey();
     JOBTYPE m_type;
     int m_priority;
-    QMutex m_runMutex;
+    void cancelJob();
+    
+signals:
+    void jobCanceled();
 };
 
 #endif // ABSTRACTTASK_H

src/jobs/audiolevelstask.cpp

@@ -46,10 +46,6 @@ AudioLevelsTask::AudioLevelsTask(const ObjectId &owner, QObject* object)
 {
 }
 
-AudioLevelsTask::~AudioLevelsTask()
-{
-}
-
 void AudioLevelsTask::start(const ObjectId &owner, QObject* object, bool force)
 {
     AudioLevelsTask* task = new AudioLevelsTask(owner, object);
@@ -67,6 +63,8 @@ void AudioLevelsTask::start(const ObjectId &owner, QObject* object, bool force)
 
 void AudioLevelsTask::run()
 {
+    QMutexLocker lk(&m_runMutex);
+    m_running = true;
     // 2 channels interleaved of uchar values
     if (m_isCanceled) {
         pCore->taskManager.taskDone(m_owner.second, this);
@@ -137,7 +135,6 @@ void AudioLevelsTask::run()
                     producer->set(key.toUtf8().constData(), levelsCopy, 0, (mlt_destructor) deleteQVariantList);
                     producer->unlock();
                     qDebug()<<"=== FINISHED PRODUCING AUDIO FOR: "<<key<<", SIZE: "<<levelsCopy->size();
-                    QMetaObject::invokeMethod(m_object, "updateAudioThumbnail");
                     pCore->taskManager.taskDone(m_owner.second, this);
                     return;
                 }
@@ -177,7 +174,7 @@ void AudioLevelsTask::run()
             int val = int(100.0 * z / lengthInFrames);
             if (m_progress != val) {
                 m_progress = val;
-                QMetaObject::invokeMethod(m_object, "audioJobProgress", Q_ARG(const int, val));
+                QMetaObject::invokeMethod(m_object, "updateJobProgress");
             }
             QScopedPointer<Mlt::Frame> mltFrame(audioProducer->get_frame());
             if ((mltFrame != nullptr) && mltFrame->is_valid() && (mltFrame->get_int("test_audio") == 0)) {
@@ -212,10 +209,9 @@ void AudioLevelsTask::run()
         for (double &v : mltLevels) {
             m_audioLevels << uchar(255 * v / maxLevel);
         }*/
-        
+        m_progress = 100;
+        QMetaObject::invokeMethod(m_object, "updateJobProgress");
         if (!m_isCanceled) {
-            m_progress = 100;
-            QMetaObject::invokeMethod(m_object, "audioJobProgress", Q_ARG(const int, 100));
             if (mltLevels.size() > 0 && !m_isCanceled) {
                 QVector <uint8_t>* levelsCopy = new  QVector <uint8_t>(mltLevels);
                 producer->lock();
@@ -247,4 +243,5 @@ void AudioLevelsTask::run()
         }
     }
     pCore->taskManager.taskDone(m_owner.second, this);
+    QMetaObject::invokeMethod(m_object, "updateJobProgress");
 }

src/jobs/audiolevelstask.h

@@ -29,7 +29,6 @@ class AudioLevelsTask : public AbstractTask
 {
 public:
     AudioLevelsTask(const ObjectId &owner, QObject* object);
-    virtual ~AudioLevelsTask();
     static void start(const ObjectId &owner, QObject* object, bool force = false);
 
 protected:

src/jobs/cliploadtask.cpp

@@ -351,6 +351,8 @@ void ClipLoadTask::run()
         pCore->taskManager.taskDone(m_owner.second, this);
         return;
     }
+    QMutexLocker lk(&m_runMutex);
+    m_running = true;
     pCore->getMonitor(Kdenlive::ClipMonitor)->resetPlayOrLoopZone(QString::number(m_owner.second));
     QString resource = Xml::getXmlProperty(m_xml, QStringLiteral("resource"));
     qDebug()<<"============STARTING LOAD TASK FOR: "<<resource<<"\n\n:::::::::::::::::::";

src/jobs/proxytask.h

+/***************************************************************************
+ *                                                                         *
+ *   Copyright (C) 2021 by Jean-Baptiste Mardelle (jb@kdenlive.org)        *
+ *                                                                         *
+ *   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          *
+ ***************************************************************************/
+
+#ifndef PROXYTASK_H
+#define PROXYTASK_H
+
+#include "abstracttask.h"
+
+class QProcess;
+
+class ProxyTask : public AbstractTask
+{
+public:
+    ProxyTask(const ObjectId &owner, QObject* object);
+    static void start(const ObjectId &owner, QObject* object, bool force = false);
+
+protected:
+    void run() override;
+
+private slots:
+    void processLogInfo();
+
+private:
+    int m_jobDuration;
+    bool m_isFfmpegJob;
+    std::unique_ptr<QProcess> m_jobProcess;
+    QString m_errorMessage;
+    QString m_logDetails;
+};
+
+
+#endif

src/jobs/taskmanager.cpp

@@ -57,14 +57,14 @@ void TaskManager::discardJobs(const ObjectId &owner, AbstractTask::JOBTYPE type)
     for (AbstractTask* t : taskList) {
         if (type == AbstractTask::NOJOBTYPE || type == t->m_type) {
             // If so, then just add ourselves to be notified upon completion.
-            t->m_isCanceled = true;
+            t->cancelJob();
             // Block until the task is finished
             t->m_runMutex.lock();
         }
     }
 }
 
-bool TaskManager::hasPendingJob(const ObjectId &owner, AbstractTask::JOBTYPE type)
+bool TaskManager::hasPendingJob(const ObjectId &owner, AbstractTask::JOBTYPE type) const
 {
     QReadLocker lk(&m_tasksListLock);
     if (type == AbstractTask::NOJOBTYPE) {
@@ -83,6 +83,22 @@ bool TaskManager::hasPendingJob(const ObjectId &owner, AbstractTask::JOBTYPE typ
     return false;
 }
 
+TaskManagerStatus TaskManager::jobStatus(const ObjectId &owner) const
+{
+    QReadLocker lk(&m_tasksListLock);
+    if (m_taskList.find(owner.second) == m_taskList.end()) {
+        // No job for this clip
+        return TaskManagerStatus::NoJob;
+    }
+    std::vector<AbstractTask*> taskList = m_taskList.at(owner.second);
+    for (AbstractTask* t : taskList) {
+        if (t->m_running) {
+            return TaskManagerStatus::Running;;
+        }
+    }
+    return TaskManagerStatus::Pending;
+}
+
 void TaskManager::updateJobCount()
 {
     QReadLocker lk(&m_tasksListLock);
@@ -96,24 +112,31 @@ void TaskManager::updateJobCount()
 
 void TaskManager::taskDone(int cid, AbstractTask *task)
 {
-    QReadLocker lk(&m_tasksListLock);
+    m_tasksListLock.lockForWrite();
+    Q_ASSERT(m_taskList.find(cid) != m_taskList.end());
     m_taskList[cid].erase(std::remove(m_taskList[cid].begin(), m_taskList[cid].end(), task), m_taskList[cid].end());
+    if (m_taskList[cid].size() == 0) {
+        m_taskList.erase(cid);
+    }
+    m_tasksListLock.unlock();
     updateJobCount();
 }
 
 
 void TaskManager::slotCancelJobs()
 {
-    QReadLocker lk(&m_tasksListLock);
+    m_tasksListLock.lockForWrite();
     // See if there is already a task for this MLT service and resource.
     for (auto task : m_taskList) {
         for (AbstractTask* t : task.second) {
             // If so, then just add ourselves to be notified upon completion.
-            t->m_isCanceled = true;
+            t->cancelJob();
             // Block until the task is finished
             t->m_runMutex.lock();
         }
     }
+    m_tasksListLock.unlock();
+    updateJobCount();
 }
 
 void TaskManager::startTask(int ownerId, AbstractTask *task)
@@ -140,11 +163,15 @@ int TaskManager::getJobProgressForClip(const ObjectId &owner) const
         return 100;
     }
     std::vector<AbstractTask*> taskList = m_taskList.at(owner.second);
+    int cnt = taskList.size();
+    if (cnt == 0) {
+        return 100;
+    }
     int total = 0;
     for (AbstractTask* t : taskList) {
         total += t->m_progress;
     }
-    total /= taskList.size();
+    total /= cnt;
     return total;
 }
 

src/jobs/taskmanager.h

@@ -61,7 +61,9 @@ public:
      *  @param owner the owner item for this task
      *  @param type The type of job that you want to query
      */
-    bool hasPendingJob(const ObjectId &owner, AbstractTask::JOBTYPE type = AbstractTask::NOJOBTYPE);
+    bool hasPendingJob(const ObjectId &owner, AbstractTask::JOBTYPE type = AbstractTask::NOJOBTYPE) const;
+    
+    TaskManagerStatus jobStatus(const ObjectId &owner) const;
 
     /** @brief return the progress of a given job on a given clip */
     int getJobProgressForClip(const ObjectId &owner) const;