Rework of execution policy (multithreading)

PdfForQtLib/sources/pdfexecutionpolicy.cpp

@@ -19,13 +19,26 @@
 #include "pdfexecutionpolicy.h"
 
 #include <QThread>
+#include <QApplication>
 
 namespace pdf
 {
 
 struct PDFExecutionPolicyHolder
 {
+    PDFExecutionPolicyHolder()
+    {
+        qAddPostRoutine(&PDFExecutionPolicy::finalize);
+    }
+    ~PDFExecutionPolicyHolder()
+    {
+        auxiliary.waitForDone();
+        primary.waitForDone();
+    }
+
     PDFExecutionPolicy policy;
+    QThreadPool primary;
+    QThreadPool auxiliary;
 } s_execution_policy;
 
 void PDFExecutionPolicy::setStrategy(Strategy strategy)
@@ -50,13 +63,7 @@ bool PDFExecutionPolicy::isParallelizing(Scope scope)
                     return true; // We are parallelizing pages...
 
                 case Scope::Content:
-                {
-                    // Jakub Melka: this is a bit complicated. We must count number of content streams
-                    // being processed and if it is large enough, then do not parallelize.
-                    const size_t threadLimit = s_execution_policy.policy.m_threadLimit.load(std::memory_order_relaxed);
-                    const size_t contentStreamsCount = s_execution_policy.policy.m_contentStreamsCount.load(std::memory_order_seq_cst);
-                    return contentStreamsCount < threadLimit;
-                }
+                    return false;
             }
 
             break;
@@ -71,6 +78,34 @@ bool PDFExecutionPolicy::isParallelizing(Scope scope)
     return false;
 }
 
+int PDFExecutionPolicy::getActiveThreadCount(Scope scope)
+{
+    return getThreadPool(scope)->activeThreadCount();
+}
+
+int PDFExecutionPolicy::getMaxThreadCount(Scope scope)
+{
+    return getThreadPool(scope)->maxThreadCount();
+}
+
+void PDFExecutionPolicy::setMaxThreadCount(Scope scope, int count)
+{
+    // Sanitize value!
+    count = qMax(count, 1);
+    getThreadPool(scope)->setMaxThreadCount(count);
+}
+
+int PDFExecutionPolicy::getIdealThreadCount(Scope scope)
+{
+    Q_UNUSED(scope);
+    return QThread::idealThreadCount();
+}
+
+int PDFExecutionPolicy::getContentStreamCount()
+{
+    return s_execution_policy.policy.m_contentStreamsCount.load(std::memory_order_relaxed);
+}
+
 void PDFExecutionPolicy::startProcessingContentStream()
 {
     ++s_execution_policy.policy.m_contentStreamsCount;
@@ -81,9 +116,33 @@ void PDFExecutionPolicy::endProcessingContentStream()
     --s_execution_policy.policy.m_contentStreamsCount;
 }
 
+void PDFExecutionPolicy::finalize()
+{
+    s_execution_policy.auxiliary.waitForDone();
+    s_execution_policy.primary.waitForDone();
+}
+
+QThreadPool* PDFExecutionPolicy::getThreadPool(PDFExecutionPolicy::Scope scope)
+{
+    switch (scope)
+    {
+        case Scope::Page:
+        case Scope::Unknown:
+            return &s_execution_policy.primary;
+
+        case Scope::Content:
+            return &s_execution_policy.auxiliary;
+
+        default:
+            Q_ASSERT(false);
+            break;
+    }
+
+    return nullptr;
+}
+
 PDFExecutionPolicy::PDFExecutionPolicy() :
     m_contentStreamsCount(0),
-    m_threadLimit(QThread::idealThreadCount()),
     m_strategy(Strategy::PageMultithreaded)
 {
 

PdfForQtLib/sources/pdfexecutionpolicy.h

@@ -20,6 +20,9 @@
 
 #include "pdfglobal.h"
 
+#include <QSemaphore>
+#include <QThreadPool>
+
 #include <atomic>
 #include <execution>
 
@@ -53,15 +56,48 @@ public:
     static void setStrategy(Strategy strategy);
 
     /// Determines, if we should parallelize for scope
-    /// \param scope Scope for which we want to determine exectution policy
+    /// \param scope Scope for which we want to determine execution policy
     static bool isParallelizing(Scope scope);
 
+    template<typename ForwardIt, typename UnaryFunction>
+    class Runnable : public QRunnable
+    {
+    public:
+        explicit inline Runnable(ForwardIt it, UnaryFunction* function, QSemaphore* semaphore) :
+            m_forwardIt(qMove(it)),
+            m_function(function),
+            m_semaphore(semaphore)
+        {
+            setAutoDelete(true);
+        }
+
+        virtual void run() override
+        {
+            QSemaphoreReleaser semaphoreReleaser(m_semaphore);
+            (*m_function)(*m_forwardIt);
+        }
+
+    private:
+        ForwardIt m_forwardIt;
+        UnaryFunction* m_function;
+        QSemaphore* m_semaphore;
+    };
+
     template<typename ForwardIt, typename UnaryFunction>
     static void execute(Scope scope, ForwardIt first, ForwardIt last, UnaryFunction f)
     {
         if (isParallelizing(scope))
         {
-            std::for_each(std::execution::parallel_policy(), first, last, f);
+            QSemaphore semaphore(0);
+            int count = static_cast<int>(std::distance(first, last));
+
+            QThreadPool* pool = getThreadPool(scope);
+            for (auto it = first; it != last; ++it)
+            {
+                pool->start(new Runnable(it, &f, &semaphore));
+            }
+
+            semaphore.acquire(count);
         }
         else
         {
@@ -72,29 +108,45 @@ public:
     template<typename ForwardIt, typename Comparator>
     static void sort(Scope scope, ForwardIt first, ForwardIt last, Comparator f)
     {
-        if (isParallelizing(scope))
-        {
-            std::sort(std::execution::parallel_policy(), first, last, f);
-        }
-        else
-        {
-            std::sort(std::execution::sequenced_policy(), first, last, f);
-        }
+        Q_UNUSED(scope);
+
+        // We always sort by single thread
+        std::sort(std::execution::sequenced_policy(), first, last, f);
     }
 
+    /// Returns number of active threads for given scope
+    static int getActiveThreadCount(Scope scope);
+
+    /// Returns maximal number of threads for given scope
+    static int getMaxThreadCount(Scope scope);
+
+    /// Sets maximal number of threads for given scope
+    static void setMaxThreadCount(Scope scope, int count);
+
+    /// Returns ideal thread count for given scope
+    static int getIdealThreadCount(Scope scope);
+
+    /// Returns number of currently processed content streams
+    static int getContentStreamCount();
+
     /// Starts processing content stream
     static void startProcessingContentStream();
 
     /// Ends processing content stream
     static void endProcessingContentStream();
 
+    /// Finalize multithreading - must be called at the end of program
+    static void finalize();
+
 private:
     friend struct PDFExecutionPolicyHolder;
 
+    /// Returns thread pool based on scope
+    static QThreadPool* getThreadPool(Scope scope);
+
     explicit PDFExecutionPolicy();
 
-    std::atomic<size_t> m_contentStreamsCount;
-    std::atomic<size_t> m_threadLimit;
+    std::atomic<int> m_contentStreamsCount;
     std::atomic<Strategy> m_strategy;
 };