1 files changed, 394 insertions, 0 deletions
diff --git a/comphelper/source/misc/threadpool.cxx b/comphelper/source/misc/threadpool.cxx
new file mode 100644
index 000000000..f0a71eb05
--- /dev/null
+++ b/comphelper/source/misc/threadpool.cxx
@@ -0,0 +1,394 @@
+/* -*- Mode: C++; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+/*
+ * This file is part of the LibreOffice project.
+ *
+ * This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at http://mozilla.org/MPL/2.0/.
+ */
+
+#include <comphelper/threadpool.hxx>
+
+#include <com/sun/star/uno/Exception.hpp>
+#include <config_options.h>
+#include <o3tl/safeint.hxx>
+#include <sal/config.h>
+#include <sal/log.hxx>
+#include <salhelper/thread.hxx>
+#include <algorithm>
+#include <memory>
+#include <thread>
+#include <chrono>
+#include <cstddef>
+#include <comphelper/debuggerinfo.hxx>
+#include <utility>
+
+#if defined HAVE_VALGRIND_HEADERS
+#include <valgrind/memcheck.h>
+#endif
+
+#if defined(_WIN32)
+#define WIN32_LEAN_AND_MEAN
+#include <windows.h>
+#endif
+
+namespace comphelper {
+
+/** prevent waiting for a task from inside a task */
+#if defined DBG_UTIL && (defined LINUX || defined _WIN32)
+static thread_local bool gbIsWorkerThread;
+#endif
+
+// used to group thread-tasks for waiting in waitTillDone()
+class ThreadTaskTag
+{
+    std::mutex maMutex;
+    sal_Int32 mnTasksWorking;
+    std::condition_variable maTasksComplete;
+
+public:
+    ThreadTaskTag();
+    bool isDone();
+    void waitUntilDone();
+    void onTaskWorkerDone();
+    void onTaskPushed();
+};
+
+
+class ThreadPool::ThreadWorker : public salhelper::Thread
+{
+    ThreadPool *mpPool;
+public:
+
+    explicit ThreadWorker( ThreadPool *pPool ) :
+        salhelper::Thread("thread-pool"),
+        mpPool( pPool )
+    {
+    }
+
+    virtual void execute() override
+    {
+#if defined DBG_UTIL && (defined LINUX || defined _WIN32)
+        gbIsWorkerThread = true;
+#endif
+        std::unique_lock< std::mutex > aGuard( mpPool->maMutex );
+
+        while( !mpPool->mbTerminate )
+        {
+            std::unique_ptr<ThreadTask> pTask = mpPool->popWorkLocked( aGuard, true );
+            if( pTask )
+            {
+                std::shared_ptr<ThreadTaskTag> pTag(pTask->mpTag);
+                mpPool->incBusyWorker();
+                aGuard.unlock();
+
+                pTask->exec();
+                pTask.reset();
+
+                aGuard.lock();
+                mpPool->decBusyWorker();
+                pTag->onTaskWorkerDone();
+            }
+        }
+    }
+};
+
+ThreadPool::ThreadPool(std::size_t nWorkers)
+    : mbTerminate(true)
+    , mnMaxWorkers(nWorkers)
+    , mnBusyWorkers(0)
+{
+}
+
+ThreadPool::~ThreadPool()
+{
+    // note: calling shutdown from global variable dtor blocks forever on Win7
+    // note2: there isn't enough MSVCRT left on exit to call assert() properly
+    // so these asserts just print something to stderr but exit status is
+    // still 0, but hopefully they will be more helpful on non-WNT platforms
+    assert(mbTerminate);
+    assert(maTasks.empty());
+    assert(mnBusyWorkers == 0);
+}
+
+namespace {
+
+std::shared_ptr< ThreadPool >& GetStaticThreadPool()
+{
+    static std::shared_ptr< ThreadPool > POOL =
+    []()
+    {
+        const std::size_t nThreads = ThreadPool::getPreferredConcurrency();
+        return std::make_shared< ThreadPool >( nThreads );
+    }();
+    return POOL;
+}
+
+}
+
+ThreadPool& ThreadPool::getSharedOptimalPool()
+{
+    return *GetStaticThreadPool();
+}
+
+std::size_t ThreadPool::getPreferredConcurrency()
+{
+    static std::size_t ThreadCount = []()
+    {
+        const std::size_t nHardThreads = o3tl::clamp_to_unsigned<std::size_t>(
+            std::max(std::thread::hardware_concurrency(), 1U));
+        std::size_t nThreads = nHardThreads;
+        const char *pEnv = getenv("MAX_CONCURRENCY");
+        if (pEnv != nullptr)
+        {
+            // Override with user/admin preference.
+            nThreads = o3tl::clamp_to_unsigned<std::size_t>(rtl_str_toInt32(pEnv, 10));
+        }
+
+        nThreads = std::min(nHardThreads, nThreads);
+        return std::max<std::size_t>(nThreads, 1);
+    }();
+
+    return ThreadCount;
+}
+
+// Used to order shutdown, and to ensure there are no lingering
+// threads after LibreOfficeKit pre-init.
+void ThreadPool::shutdown()
+{
+//    if (mbTerminate)
+//        return;
+
+    std::unique_lock< std::mutex > aGuard( maMutex );
+    shutdownLocked(aGuard);
+}
+
+void ThreadPool::shutdownLocked(std::unique_lock<std::mutex>& aGuard)
+{
+    if( maWorkers.empty() )
+    { // no threads at all -> execute the work in-line
+        std::unique_ptr<ThreadTask> pTask;
+        while ( ( pTask = popWorkLocked(aGuard, false) ) )
+        {
+            std::shared_ptr<ThreadTaskTag> pTag(pTask->mpTag);
+            pTask->exec();
+            pTag->onTaskWorkerDone();
+        }
+    }
+    else
+    {
+        while( !maTasks.empty() )
+        {
+            maTasksChanged.wait( aGuard );
+            // In the (unlikely but possible?) case pushTask() gets called meanwhile,
+            // its notify_one() call is meant to wake a up a thread and process the task.
+            // But if this code gets woken up instead, it could lead to a deadlock.
+            // Pass on the notification.
+            maTasksChanged.notify_one();
+        }
+    }
+    assert( maTasks.empty() );
+
+    // coverity[missing_lock] - on purpose
+    mbTerminate = true;
+
+    maTasksChanged.notify_all();
+
+    decltype(maWorkers) aWorkers;
+    std::swap(maWorkers, aWorkers);
+    aGuard.unlock();
+
+    while (!aWorkers.empty())
+    {
+        rtl::Reference<ThreadWorker> xWorker = aWorkers.back();
+        aWorkers.pop_back();
+        assert(std::find(aWorkers.begin(), aWorkers.end(), xWorker)
+                == aWorkers.end());
+        {
+            xWorker->join();
+            xWorker.clear();
+        }
+    }
+}
+
+void ThreadPool::pushTask( std::unique_ptr<ThreadTask> pTask )
+{
+    std::scoped_lock< std::mutex > aGuard( maMutex );
+
+    mbTerminate = false;
+
+    // Worked on tasks are already removed from maTasks, so include the count of busy workers.
+    if (maWorkers.size() < mnMaxWorkers && maWorkers.size() <= maTasks.size() + mnBusyWorkers)
+    {
+        maWorkers.push_back( new ThreadWorker( this ) );
+        maWorkers.back()->launch();
+    }
+
+    pTask->mpTag->onTaskPushed();
+    maTasks.insert( maTasks.begin(), std::move(pTask) );
+
+    maTasksChanged.notify_one();
+}
+
+std::unique_ptr<ThreadTask> ThreadPool::popWorkLocked( std::unique_lock< std::mutex > & rGuard, bool bWait )
+{
+    do
+    {
+        if( !maTasks.empty() )
+        {
+            std::unique_ptr<ThreadTask> pTask = std::move(maTasks.back());
+            maTasks.pop_back();
+            return pTask;
+        }
+        else if (!bWait || mbTerminate)
+            return nullptr;
+
+        maTasksChanged.wait( rGuard );
+
+    } while (!mbTerminate);
+
+    return nullptr;
+}
+
+void ThreadPool::incBusyWorker()
+{
+    ++mnBusyWorkers;
+}
+
+void ThreadPool::decBusyWorker()
+{
+    assert(mnBusyWorkers >= 1);
+    --mnBusyWorkers;
+}
+
+void ThreadPool::waitUntilDone(const std::shared_ptr<ThreadTaskTag>& rTag, bool bJoin)
+{
+#if defined DBG_UTIL && (defined LINUX || defined _WIN32)
+    assert(!gbIsWorkerThread && "cannot wait for tasks from inside a task");
+#endif
+    {
+        std::unique_lock< std::mutex > aGuard( maMutex );
+
+        if( maWorkers.empty() )
+        { // no threads at all -> execute the work in-line
+            while (!rTag->isDone())
+            {
+                std::unique_ptr<ThreadTask> pTask = popWorkLocked(aGuard, false);
+                if (!pTask)
+                    break;
+                std::shared_ptr<ThreadTaskTag> pTag(pTask->mpTag);
+                pTask->exec();
+                pTag->onTaskWorkerDone();
+            }
+        }
+    }
+
+    rTag->waitUntilDone();
+
+    if (bJoin)
+        joinThreadsIfIdle();
+}
+
+void ThreadPool::joinThreadsIfIdle()
+{
+    std::unique_lock< std::mutex > aGuard( maMutex );
+    if (isIdle()) // check if there are still tasks from another tag
+    {
+        shutdownLocked(aGuard);
+    }
+}
+
+std::shared_ptr<ThreadTaskTag> ThreadPool::createThreadTaskTag()
+{
+    return std::make_shared<ThreadTaskTag>();
+}
+
+bool ThreadPool::isTaskTagDone(const std::shared_ptr<ThreadTaskTag>& pTag)
+{
+    return pTag->isDone();
+}
+
+ThreadTask::ThreadTask(std::shared_ptr<ThreadTaskTag> xTag)
+    : mpTag(std::move(xTag))
+{
+}
+
+void ThreadTask::exec()
+{
+    try {
+        doWork();
+    }
+    catch (const std::exception &e)
+    {
+        SAL_WARN("comphelper", "exception in thread worker while calling doWork(): " << e.what());
+    }
+    catch (const css::uno::Exception &e)
+    {
+        SAL_WARN("comphelper", "exception in thread worker while calling doWork(): " << e);
+    }
+    catch (...)
+    {
+        SAL_WARN("comphelper", "unknown exception in thread worker while calling doWork()");
+    }
+}
+
+ThreadTaskTag::ThreadTaskTag() : mnTasksWorking(0)
+{
+}
+
+void ThreadTaskTag::onTaskPushed()
+{
+    std::scoped_lock< std::mutex > aGuard( maMutex );
+    mnTasksWorking++;
+    assert( mnTasksWorking < 65536 ); // sanity checking
+}
+
+void ThreadTaskTag::onTaskWorkerDone()
+{
+    std::scoped_lock< std::mutex > aGuard( maMutex );
+    mnTasksWorking--;
+    assert(mnTasksWorking >= 0);
+    if (mnTasksWorking == 0)
+        maTasksComplete.notify_all();
+}
+
+bool ThreadTaskTag::isDone()
+{
+    std::scoped_lock< std::mutex > aGuard( maMutex );
+    return mnTasksWorking == 0;
+}
+
+void ThreadTaskTag::waitUntilDone()
+{
+    std::unique_lock< std::mutex > aGuard( maMutex );
+    while( mnTasksWorking > 0 )
+    {
+#if defined DBG_UTIL && !defined NDEBUG
+        // 10 minute timeout in debug mode, unless the code is built with
+        // sanitizers or debugged in valgrind or gdb, in which case the threads
+        // should not time out in the middle of a debugging session
+        int maxTimeout = 10 * 60;
+#if !ENABLE_RUNTIME_OPTIMIZATIONS
+        maxTimeout = 30 * 60;
+#endif
+#if defined HAVE_VALGRIND_HEADERS
+        if( RUNNING_ON_VALGRIND )
+            maxTimeout = 30 * 60;
+#endif
+        if( isDebuggerAttached())
+            maxTimeout = 300 * 60;
+        std::cv_status result = maTasksComplete.wait_for(
+            aGuard, std::chrono::seconds( maxTimeout ));
+        assert(result != std::cv_status::timeout);
+#else
+        // 10 minute timeout in production so the app eventually throws some kind of error
+        if (maTasksComplete.wait_for(
+                aGuard, std::chrono::seconds( 10 * 60 )) == std::cv_status::timeout)
+            throw std::runtime_error("timeout waiting for threadpool tasks");
+#endif
+    }
+}
+
+} // namespace comphelper
+
+/* vim:set shiftwidth=4 softtabstop=4 expandtab: */