WorkerThread.cpp

Show log
Commit

Hash : c0386ad4
Author :
Date : 2024-04-05T11:56:17

AsyncWorkerPool releases shared_ptr<Closure> before notifying

Parallel compile (MainLinkLoadTask, Program::LinkingState)
is dependent on destructor getting called before the event is notified
Repro: https://crrev.com/c/5425924

More details on the parallel compile case, provided by syoussefi@:
"""
A race condition caused the worker pool to sometimes be destroyed from a
worker thread instead of the main thread.

The race condition triggered in the following scenario:

- The MainLinkLoadTask holds on to the worker pool
  - This is necessary for the main task to spawn further tasks
    asynchronously
- The reference to the worker pool in MainLinkLoadTask is released by
  its destructor
- The worker thread dequeues a task (i.e. MainLinkLoadTask) to execute
  and holds a reference to it.
- Once the task is run by the worker thread, the worker thread signals
  its completion
  - (1) At this point, the scope holding the reference to the task
    closes and the task is released.  However, this is done after
    signaling the task's completion.
- On program destruction, the program ensures that all its tasks are
  complete
  - This uses the signal coming from the worker thread
- (2) On display destruction, the worker pool is destroyed
  (by dereferencing it through the shared_ptr)
- The destructor of the worker pool waits for the worker thread, with
  the expectation that this wait is done in the main thread.

The race condition led to the assert firing when (2) was done before
(1). Because the task is already signaled complete, the main thread
considers it done and goes ahead with the destruction of the display.
However, until the scope of the worker thread closes, the task itself
is still not destroyed. Since the task is holding a reference to the
worker pool, that prevents the worker pool from getting destroyed too.

Once the display is destroyed, the worker thread closes its scope,
causing the task to be destroyed. In turn, this leads to the worker
pool itself to be destroyed. On destruction, the worker pool would wait
for the worker thread to end which is a deadlock. Fortunately, this was
caught earlier with an ASSERT that wanted to ensure destruction happens
on the main thread.

In this change, the worker thread ensures it releases the task before
signalling it complete, avoiding this issue. Other possible solutions
would have been:

- Release the worker pool from MainLinkLoadTask as soon as the subtasks
  are scheduled
- Explicitly call a "destroy" method on the pool, instead of relying on
  the destructor to clean up.
"""

Bug: angleproject:8661
Change-Id: I37c9bc8e8f05bce4062d794df449cc3d2c80a093
Reviewed-on: https://chromium-review.googlesource.com/c/angle/angle/+/5428806
Reviewed-by: Shahbaz Youssefi <syoussefi@chromium.org>
Commit-Queue: Roman Lavrov <romanl@google.com>

Download

src/common/WorkerThread.cpp

//
// Copyright 2016 The ANGLE Project Authors. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.
//
// WorkerThread:
//   Task running thread for ANGLE, similar to a TaskRunner in Chromium.
//   Might be implemented differently depending on platform.
//

#include "common/WorkerThread.h"

#include "common/angleutils.h"
#include "common/system_utils.h"

// Controls if our threading code uses std::async or falls back to single-threaded operations.
// Note that we can't easily use std::async in UWPs due to UWP threading restrictions.
#if !defined(ANGLE_STD_ASYNC_WORKERS) && !defined(ANGLE_ENABLE_WINDOWS_UWP)
#    define ANGLE_STD_ASYNC_WORKERS 1
#endif  // !defined(ANGLE_STD_ASYNC_WORKERS) && & !defined(ANGLE_ENABLE_WINDOWS_UWP)

#if ANGLE_DELEGATE_WORKERS || ANGLE_STD_ASYNC_WORKERS
#    include <future>
#    include <queue>
#    include <thread>
#endif  // ANGLE_DELEGATE_WORKERS || ANGLE_STD_ASYNC_WORKERS

namespace angle
{

WaitableEvent::WaitableEvent()  = default;
WaitableEvent::~WaitableEvent() = default;

void WaitableEventDone::wait() {}

bool WaitableEventDone::isReady()
{
    return true;
}

void AsyncWaitableEvent::markAsReady()
{
    std::lock_guard<std::mutex> lock(mMutex);
    mIsReady = true;
    mCondition.notify_all();
}

void AsyncWaitableEvent::wait()
{
    std::unique_lock<std::mutex> lock(mMutex);
    mCondition.wait(lock, [this] { return mIsReady; });
}

bool AsyncWaitableEvent::isReady()
{
    std::lock_guard<std::mutex> lock(mMutex);
    return mIsReady;
}

WorkerThreadPool::WorkerThreadPool()  = default;
WorkerThreadPool::~WorkerThreadPool() = default;

class SingleThreadedWorkerPool final : public WorkerThreadPool
{
  public:
    std::shared_ptr<WaitableEvent> postWorkerTask(const std::shared_ptr<Closure> &task) override;
    bool isAsync() override;
};

// SingleThreadedWorkerPool implementation.
std::shared_ptr<WaitableEvent> SingleThreadedWorkerPool::postWorkerTask(
    const std::shared_ptr<Closure> &task)
{
    // Thread safety: This function is thread-safe because the task is run on the calling thread
    // itself.
    (*task)();
    return std::make_shared<WaitableEventDone>();
}

bool SingleThreadedWorkerPool::isAsync()
{
    return false;
}

#if ANGLE_STD_ASYNC_WORKERS

class AsyncWorkerPool final : public WorkerThreadPool
{
  public:
    AsyncWorkerPool(size_t numThreads);

    ~AsyncWorkerPool() override;

    std::shared_ptr<WaitableEvent> postWorkerTask(const std::shared_ptr<Closure> &task) override;

    bool isAsync() override;

  private:
    void createThreads();

    using Task = std::pair<std::shared_ptr<AsyncWaitableEvent>, std::shared_ptr<Closure>>;

    // Thread's main loop
    void threadLoop();

    bool mTerminated = false;
    std::mutex mMutex;                 // Protects access to the fields in this class
    std::condition_variable mCondVar;  // Signals when work is available in the queue
    std::queue<Task> mTaskQueue;
    std::deque<std::thread> mThreads;
    size_t mDesiredThreadCount;
};

// AsyncWorkerPool implementation.

AsyncWorkerPool::AsyncWorkerPool(size_t numThreads) : mDesiredThreadCount(numThreads)
{
    ASSERT(numThreads != 0);
}

AsyncWorkerPool::~AsyncWorkerPool()
{
    {
        std::unique_lock<std::mutex> lock(mMutex);
        mTerminated = true;
    }
    mCondVar.notify_all();
    for (auto &thread : mThreads)
    {
        ASSERT(thread.get_id() != std::this_thread::get_id());
        thread.join();
    }
}

void AsyncWorkerPool::createThreads()
{
    if (mDesiredThreadCount == mThreads.size())
    {
        return;
    }
    ASSERT(mThreads.empty());

    for (size_t i = 0; i < mDesiredThreadCount; ++i)
    {
        mThreads.emplace_back(&AsyncWorkerPool::threadLoop, this);
    }
}

std::shared_ptr<WaitableEvent> AsyncWorkerPool::postWorkerTask(const std::shared_ptr<Closure> &task)
{
    // Thread safety: This function is thread-safe because access to |mTaskQueue| is protected by
    // |mMutex|.
    auto waitable = std::make_shared<AsyncWaitableEvent>();
    {
        std::lock_guard<std::mutex> lock(mMutex);

        // Lazily create the threads on first task
        createThreads();

        mTaskQueue.push(std::make_pair(waitable, task));
    }
    mCondVar.notify_one();
    return waitable;
}

void AsyncWorkerPool::threadLoop()
{
    angle::SetCurrentThreadName("ANGLE-Worker");

    while (true)
    {
        Task task;
        {
            std::unique_lock<std::mutex> lock(mMutex);
            mCondVar.wait(lock, [this] { return !mTaskQueue.empty() || mTerminated; });
            if (mTerminated)
            {
                return;
            }
            task = mTaskQueue.front();
            mTaskQueue.pop();
        }

        auto &waitable = task.first;
        auto &closure  = task.second;

        // Note: always add an ANGLE_TRACE_EVENT* macro in the closure.  Then the job will show up
        // in traces.
        (*closure)();
        // Release shared_ptr<Closure> before notifying the event to allow for destructor based
        // dependencies (example: anglebug.com/8661)
        task.second.reset();
        waitable->markAsReady();
    }
}

bool AsyncWorkerPool::isAsync()
{
    return true;
}

#endif  // ANGLE_STD_ASYNC_WORKERS

#if ANGLE_DELEGATE_WORKERS

class DelegateWorkerPool final : public WorkerThreadPool
{
  public:
    DelegateWorkerPool(PlatformMethods *platform) : mPlatform(platform) {}
    ~DelegateWorkerPool() override = default;

    std::shared_ptr<WaitableEvent> postWorkerTask(const std::shared_ptr<Closure> &task) override;

    bool isAsync() override;

  private:
    PlatformMethods *mPlatform;
};

// A function wrapper to execute the closure and to notify the waitable
// event after the execution.
class DelegateWorkerTask
{
  public:
    DelegateWorkerTask(const std::shared_ptr<Closure> &task,
                       std::shared_ptr<AsyncWaitableEvent> waitable)
        : mTask(task), mWaitable(waitable)
    {}
    DelegateWorkerTask()                     = delete;
    DelegateWorkerTask(DelegateWorkerTask &) = delete;

    static void RunTask(void *userData)
    {
        DelegateWorkerTask *workerTask = static_cast<DelegateWorkerTask *>(userData);
        (*workerTask->mTask)();
        workerTask->mWaitable->markAsReady();

        // Delete the task after its execution.
        delete workerTask;
    }

  private:
    ~DelegateWorkerTask() = default;

    std::shared_ptr<Closure> mTask;
    std::shared_ptr<AsyncWaitableEvent> mWaitable;
};

ANGLE_NO_SANITIZE_CFI_ICALL
std::shared_ptr<WaitableEvent> DelegateWorkerPool::postWorkerTask(
    const std::shared_ptr<Closure> &task)
{
    if (mPlatform->postWorkerTask == nullptr)
    {
        // In the unexpected case where the platform methods have been changed during execution and
        // postWorkerTask is no longer usable, simply run the task on the calling thread.
        (*task)();
        return std::make_shared<WaitableEventDone>();
    }

    // Thread safety: This function is thread-safe because the |postWorkerTask| platform method is
    // expected to be thread safe.  For Chromium, that forwards the call to the |TaskTracker| class
    // in base/task/thread_pool/task_tracker.h which is thread-safe.
    auto waitable = std::make_shared<AsyncWaitableEvent>();

    // The task will be deleted by DelegateWorkerTask::RunTask(...) after its execution.
    DelegateWorkerTask *workerTask = new DelegateWorkerTask(task, waitable);
    mPlatform->postWorkerTask(mPlatform, DelegateWorkerTask::RunTask, workerTask);

    return waitable;
}

bool DelegateWorkerPool::isAsync()
{
    return mPlatform->postWorkerTask != nullptr;
}
#endif

// static
std::shared_ptr<WorkerThreadPool> WorkerThreadPool::Create(size_t numThreads,
                                                           PlatformMethods *platform)
{
    const bool multithreaded = numThreads != 1;
    std::shared_ptr<WorkerThreadPool> pool(nullptr);

#if ANGLE_DELEGATE_WORKERS
    const bool hasPostWorkerTaskImpl = platform->postWorkerTask != nullptr;
    if (hasPostWorkerTaskImpl && multithreaded)
    {
        pool = std::shared_ptr<WorkerThreadPool>(new DelegateWorkerPool(platform));
    }
#endif
#if ANGLE_STD_ASYNC_WORKERS
    if (!pool && multithreaded)
    {
        pool = std::shared_ptr<WorkerThreadPool>(new AsyncWorkerPool(
            numThreads == 0 ? std::thread::hardware_concurrency() : numThreads));
    }
#endif
    if (!pool)
    {
        return std::shared_ptr<WorkerThreadPool>(new SingleThreadedWorkerPool());
    }
    return pool;
}
}  // namespace angle

Download