JobSystem now automatically free Jobs (#91)

* JobSystem now automatically free Jobs Until now Job allocation used a linear allocator strategy which required to “reset” the JobSystem periodically — typically once per frame in filament. This is no longer required. We use a pool allocator now, which doesn’t add much overhead. It does use a spin-lock for thread-safety though, since we assume very little contention, this shouldn’t be a problem. * Thread Safe Object Pool Allocator A lock-less, thread-safe object pool allocator, now used for storing JobSystem’s jobs allocations. This gets rid of the spin-lock introduced in the previous cl.
2018-08-13 22:25:15 -07:00
parent bb553bf53d
commit d9ba3998d2
4 changed files with 140 additions and 97 deletions
@@ -69,7 +69,7 @@ public:
    LinearAllocator(void* begin, void* end) noexcept;

    template <typename AREA>
-    LinearAllocator(const AREA& area) : LinearAllocator(area.begin(), area.end()) { }
+    explicit LinearAllocator(const AREA& area) : LinearAllocator(area.begin(), area.end()) { }

    // Allocators can't be copied
    LinearAllocator(const LinearAllocator& rhs) = delete;
@@ -118,6 +118,7 @@ public:

    void swap(LinearAllocator& rhs) noexcept;

+    void *base() noexcept { return mBegin; }

    // LinearAllocator shouldn't have a free() method
    // it's only needed to be compatible with STLAllocator<> below
@@ -141,7 +142,7 @@ public:
    HeapAllocator() noexcept = default;

    template <typename AREA>
-    HeapAllocator(const AREA&) { }
+    explicit HeapAllocator(const AREA&) { }

    // our allocator concept
    void* alloc(size_t size, size_t alignment = alignof(std::max_align_t), size_t extra = 0) {
@@ -162,14 +163,23 @@ public:
    HeapAllocator(HeapAllocator&& rhs) noexcept = default;
    HeapAllocator& operator=(HeapAllocator&& rhs) noexcept = default;

-    ~HeapAllocator() noexcept { }
+    ~HeapAllocator() noexcept = default;

    void swap(HeapAllocator& rhs) noexcept { }
 };

 // ------------------------------------------------------------------------------------------------

-class FreeList {
+class FreeListBase {
+public:
+    struct Node {
+        Node* next;
+    };
+    static Node* init(void* begin, void* end,
+            size_t elementSize, size_t alignment, size_t extra) noexcept;
+};
+
+class FreeList : private FreeListBase {
 public:
    FreeList() noexcept = default;
    FreeList(void* begin, void* end, size_t elementSize, size_t alignment, size_t extra) noexcept;
@@ -179,28 +189,28 @@ public:
    FreeList& operator=(FreeList&& rhs) noexcept = default;

    void* get() noexcept {
-        FreeList* const head = mNext;
-        if (head == nullptr) {
-            // no more space
-            return nullptr;
-        }
-        mNext = head->mNext;
+        Node* const head = mHead;
+        mHead = head ? head->next : nullptr;
        // this could indicate a use after free
-        assert(!mNext || mNext>=mBegin && mNext<mEnd);
+        assert(!mHead || mHead >= mBegin && mHead < mEnd);
        return head;
    }

    void put(void* p) noexcept {
        assert(p);
-        assert(p>=mBegin && p<mEnd);
+        assert(p >= mBegin && p < mEnd);
        // TODO: assert this is one of our pointer (i.e.: it's address match one of ours)
-        FreeList* const head = static_cast<FreeList*>(p);
-        head->mNext = mNext;
-        mNext = head;
+        Node* const head = static_cast<Node*>(p);
+        head->next = mHead;
+        mHead = head;
+    }
+
+    void *getCurrent() noexcept {
+        return mHead;
    }

 private:
-    FreeList* mNext = nullptr;
+    Node* mHead = nullptr;

 #ifndef NDEBUG
    // These are needed only for debugging...
@@ -209,9 +219,46 @@ private:
 #endif
 };

+class AtomicFreeList : private FreeListBase {
+public:
+    AtomicFreeList() noexcept = default;
+    AtomicFreeList(void* begin, void* end,
+            size_t elementSize, size_t alignment, size_t extra) noexcept;
+    AtomicFreeList(const FreeList& rhs) = delete;
+    AtomicFreeList& operator=(const FreeList& rhs) = delete;
+
+    void* get() noexcept {
+        Node* head = mHead.load(std::memory_order_relaxed);
+        while (head && !mHead.compare_exchange_weak(head, head->next,
+                std::memory_order_release, std::memory_order_relaxed)) {
+        }
+        return head;
+    }
+
+    void put(void* p) noexcept {
+        assert(p);
+        Node* head = static_cast<Node*>(p);
+        head->next = mHead.load(std::memory_order_relaxed);
+        while (!mHead.compare_exchange_weak(head->next, head,
+                std::memory_order_release, std::memory_order_relaxed)) {
+        }
+    }
+
+    void* getCurrent() noexcept {
+        return mHead.load(std::memory_order_relaxed);
+    }
+
+private:
+    std::atomic<Node*> mHead;
+};
+
 // ------------------------------------------------------------------------------------------------

-template <size_t ELEMENT_SIZE, size_t ALIGNMENT = alignof(std::max_align_t), size_t OFFSET = 0>
+template <
+        size_t ELEMENT_SIZE,
+        size_t ALIGNMENT = alignof(std::max_align_t),
+        size_t OFFSET = 0,
+        typename FREELIST = FreeList>
 class PoolAllocator {
    static_assert(ELEMENT_SIZE >= sizeof(void*), "ELEMENT_SIZE must accommodate at least a pointer");
 public:
@@ -235,7 +282,7 @@ public:
    }

    template <typename AREA>
-    PoolAllocator(const AREA& area) noexcept
+    explicit PoolAllocator(const AREA& area) noexcept
        : PoolAllocator(area.begin(), area.end()) {
    }

@@ -243,23 +290,29 @@ public:
    PoolAllocator(const PoolAllocator& rhs) = delete;
    PoolAllocator& operator=(const PoolAllocator& rhs) = delete;

-    // Allocators can be moved
-    PoolAllocator(PoolAllocator&& rhs) noexcept = default;
-    PoolAllocator& operator=(PoolAllocator&& rhs) noexcept = default;
-
    PoolAllocator() noexcept = default;
    ~PoolAllocator() noexcept = default;

+    // API specific to this allocator
+
+    void *getCurrent() noexcept {
+        return mFreeList.getCurrent();
+    }
+
 private:
-    FreeList mFreeList;
+    FREELIST mFreeList;
 };

 #define UTILS_MAX(a,b) ((a) > (b) ? (a) : (b))

 template <typename T, size_t OFFSET = 0>
-using ObjectPoolAllocator = PoolAllocator<sizeof(T), \
+using ObjectPoolAllocator = PoolAllocator<sizeof(T),
        UTILS_MAX(alignof(FreeList), alignof(T)), OFFSET>;

+template <typename T, size_t OFFSET = 0>
+using ThreadSafeObjectPoolAllocator = PoolAllocator<sizeof(T),
+        UTILS_MAX(alignof(FreeList), alignof(T)), OFFSET, AtomicFreeList>;
+

 // ------------------------------------------------------------------------------------------------
 // Areas
@@ -287,6 +340,7 @@ public:
    HeapArea(HeapArea&& rhs) noexcept = delete;
    HeapArea& operator=(HeapArea&& rhs) noexcept = delete;

+    void* data() const noexcept { return mBegin; }
    void* begin() const noexcept { return mBegin; }
    void* end() const noexcept { return mEnd; }
    size_t getSize() const noexcept { return uintptr_t(mEnd) - uintptr_t(mBegin); }
@@ -24,6 +24,7 @@
 #include <thread>
 #include <vector>

+#include <utils/Allocator.h>
 #include <utils/architecture.h>
 #include <utils/Condition.h>
 #include <utils/Log.h>
@@ -53,9 +54,9 @@ public:

    using JobFunc = void(*)(void*, JobSystem&, Job*);

-    class Job {
+    class alignas(CACHELINE_SIZE) Job { // NOLINT(cppcoreguidelines-pro-type-member-init)
    public:
-        Job() noexcept { }
+        Job() noexcept {} // = default;
        Job(const Job&) = delete;
        Job(Job&&) = delete;

@@ -92,18 +93,13 @@ public:
    // part of a Jobsystem.
    static JobSystem* getJobSystem() noexcept;

-    // Free-up all allocated jobs (without calling destructors)
-    // Make sure to call this when all call to wait() have returned.
-    // Also clears the master job
-    void reset() noexcept;
-
-    // resets a job
-    void reset(Job* job) noexcept;
-
    // If a parent is not specified when creating a job, that job will automatically take the
    // master job as a parent.
    // The master job is reset when calling reset()
-    Job* setMasterJob(Job* job) noexcept { return (mMasterJob = job); }
+    Job* setMasterJob(Job* job) noexcept { return mMasterJob = job; }
+
+    // Clears the master job
+    void reset() noexcept { mMasterJob = nullptr; }


    // NOTE: All methods below must be called from the same thread and that thread must be
@@ -241,7 +237,6 @@ public:
    void finish(Job* job) noexcept;

    // for debugging
-    size_t getJobWatermark() const noexcept { return mJobWaterMark; }
    friend utils::io::ostream& operator << (utils::io::ostream& out, JobSystem const& js);


@@ -305,27 +300,29 @@ private:
    void loop(ThreadState* threadState) noexcept;
    bool execute(JobSystem::ThreadState& state) noexcept;

-    void put(WorkQueue& workQueue, Job* job) noexcept { workQueue.push(jobToIndex(job)); }
-    Job* pop(WorkQueue& workQueue) noexcept { return indexToJob(workQueue.pop()); }
-    Job* steal(WorkQueue& workQueue) noexcept { return indexToJob(workQueue.steal()); }
-
-    Job* indexToJob(size_t index) {
-        assert(index <= MAX_JOB_COUNT);
-        return !index ? nullptr : (&mJobStorage[0] - 1) + index;
+    void put(WorkQueue& workQueue, Job* job) noexcept {
+        size_t index = job - mJobStorageBase;
+        assert(index >= 0 && index < MAX_JOB_COUNT);
+        workQueue.push(uint16_t(index + 1));
    }

-    uint16_t jobToIndex(Job const* job) const {
-        // we offset index by +1 so b/c workQueue returns 0 on failure
-        size_t index = job - (&mJobStorage[0] - 1);
-        assert(index > 0 && index <= MAX_JOB_COUNT);
-        return uint16_t(index);
+    Job* pop(WorkQueue& workQueue) noexcept {
+        size_t index = workQueue.pop();
+        assert(index <= MAX_JOB_COUNT);
+        return !index ? nullptr : (mJobStorageBase - 1) + index;
+    }
+
+    Job* steal(WorkQueue& workQueue) noexcept {
+        size_t index = workQueue.steal();
+        assert(index <= MAX_JOB_COUNT);
+        return !index ? nullptr : (mJobStorageBase - 1) + index;
    }

    // these have thread contention, keep them together
    utils::Mutex mLock;
    utils::Condition mCondition;
    std::atomic<uint32_t> mActiveJobs = { 0 };
-    std::atomic<uint16_t> mNextJobIndex = { 0 };
+    utils::Arena<utils::ThreadSafeObjectPoolAllocator<Job>, LockingPolicy::NoLock> mJobPool;

    template <typename T>
    using aligned_vector = std::vector<T, utils::STLAlignedAllocator<T>>;
@@ -340,13 +337,11 @@ private:
    aligned_vector<ThreadState> mThreadStates;          // actual data is stored offline
    std::atomic<bool> mExitRequested = { 0 };           // this one is almost never written
    std::atomic<uint16_t> mAdoptedThreads = { 0 };      // this one is almost never written
-    Job* mJobStorage = nullptr;
-    uint16_t mThreadCount = 0;
-    uint8_t mParallelSplitCount = 0;
+    Job* const mJobStorageBase;                         // Base for conversion to indices
+    uint16_t mThreadCount = 0;                          // total # of threads in the pool
+    uint8_t mParallelSplitCount = 0;                    // # of split allowable in parallel_for
    Job* mMasterJob = nullptr;

-    size_t mJobWaterMark = 0;
-
    static UTILS_DECLARE_TLS(ThreadState *) sThreadState;
 };

@@ -55,11 +55,8 @@ void LinearAllocator::swap(LinearAllocator& rhs) noexcept {
 // FreeList
 // ------------------------------------------------------------------------------------------------

-FreeList::FreeList(void* begin, void* end,
+FreeListBase::Node* FreeListBase::init(void* begin, void* end,
        size_t elementSize, size_t alignment, size_t extra) noexcept
-#ifndef NDEBUG
-            : mBegin(begin), mEnd(end)
-#endif
 {
    void* const p = pointermath::align(begin, alignment, extra);
    void* const n = pointermath::align(pointermath::add(p, elementSize), alignment, extra);
@@ -70,18 +67,33 @@ FreeList::FreeList(void* begin, void* end,
    const size_t num = (uintptr_t(end) - uintptr_t(p)) / d;

    // set first entry
-    mNext = static_cast<FreeList*>(p);
+    Node* head = static_cast<Node*>(p);

    // next entry
-    FreeList* cur = mNext;
+    Node* cur = head;
    for (size_t i=1 ; i<num ; ++i) {
-        FreeList* next = pointermath::add(cur, d);
-        cur->mNext = next;
+        Node* next = pointermath::add(cur, d);
+        cur->next = next;
        cur = next;
    }
    assert(cur < end);
    assert(pointermath::add(cur, d) <= end);
-    cur->mNext = nullptr;
+    cur->next = nullptr;
+    return head;
+}
+
+FreeList::FreeList(void* begin, void* end,
+        size_t elementSize, size_t alignment, size_t extra) noexcept
+        : mHead(init(begin, end, elementSize, alignment, extra))
+#ifndef NDEBUG
+        , mBegin(begin), mEnd(end)
+#endif
+{
+}
+
+AtomicFreeList::AtomicFreeList(void* begin, void* end,
+        size_t elementSize, size_t alignment, size_t extra) noexcept
+        : mHead(init(begin, end, elementSize, alignment, extra)) {
 }

 TrackingPolicy::HighWatermark::~HighWatermark() noexcept {
@@ -103,12 +103,12 @@ void JobSystem::setThreadAffinity(uint32_t mask) noexcept {
 #endif
 }

-JobSystem::JobSystem(size_t threadCount, size_t adoptableThreadsCount) noexcept {
+JobSystem::JobSystem(size_t threadCount, size_t adoptableThreadsCount) noexcept
+    : mJobPool("JobSystem Job pool", MAX_JOB_COUNT * sizeof(Job)),
+      mJobStorageBase(static_cast<Job *>(mJobPool.getAllocator().getCurrent()))
+{
    SYSTRACE_ENABLE();

-    void* p = aligned_alloc(MAX_JOB_COUNT * sizeof(Job), CACHELINE_SIZE);
-    mJobStorage = (Job*)p;
-
    if (threadCount == 0) {
        // default value, system dependant
        size_t hwThreads = std::thread::hardware_concurrency();
@@ -128,7 +128,6 @@ JobSystem::JobSystem(size_t threadCount, size_t adoptableThreadsCount) noexcept

    // this is pitty these are not compile-time checks (C++17 supports it apparently)
    assert(mExitRequested.is_lock_free());
-    assert(mNextJobIndex.is_lock_free());
    assert(Job().runningJobCount.is_lock_free());

    std::random_device rd;
@@ -158,8 +157,6 @@ JobSystem::~JobSystem() {
            state.thread.join();
        }
    }
-
-    aligned_free(mJobStorage);
 }

 JobSystem* JobSystem::getJobSystem() noexcept {
@@ -194,12 +191,7 @@ inline JobSystem::ThreadState& JobSystem::getState() noexcept {
 }

 JobSystem::Job* JobSystem::allocateJob() noexcept {
-    size_t index = mNextJobIndex.fetch_add(1, std::memory_order_relaxed);
-    if (UTILS_UNLIKELY(index >= MAX_JOB_COUNT)) {
-        mNextJobIndex.fetch_sub(1, std::memory_order_relaxed);
-        return nullptr;
-    }
-    return new(&mJobStorage[index]) Job();
+    return mJobPool.make<Job>();
 }

 inline JobSystem::ThreadState& JobSystem::getStateToStealFrom(JobSystem::ThreadState& state) noexcept {
@@ -270,8 +262,9 @@ JobSystem::Job* JobSystem::create(JobSystem::Job* parent, JobFunc func) noexcept
        if (parent) {
            // can't create a child job of a terminated parent
            assert(parent->runningJobCount.load(std::memory_order_relaxed) > 0);
+
            parent->runningJobCount.fetch_add(1, std::memory_order_relaxed);
-            index = parent - &mJobStorage[0];
+            index = parent - mJobStorageBase;
            assert(index < MAX_JOB_COUNT);
        }
        job->function = func;
@@ -281,20 +274,12 @@ JobSystem::Job* JobSystem::create(JobSystem::Job* parent, JobFunc func) noexcept
    return job;
 }

-void JobSystem::reset(JobSystem::Job* job) noexcept {
-    JobSystem::Job* parent = job->parent ? &mJobStorage[job->parent] : mMasterJob;
-    if (parent) {
-        assert(parent->runningJobCount.load(std::memory_order_relaxed) > 0);
-        parent->runningJobCount.fetch_add(1, std::memory_order_relaxed);
-    }
-    job->runningJobCount.store(1, std::memory_order_relaxed);
-}
-
 void JobSystem::finish(Job* job) noexcept {
    SYSTRACE_CALL();

    // terminate this job and notify its parent
-    Job* const storage = mJobStorage;
+    auto& jobPool = mJobPool;
+    Job* const storage = mJobStorageBase;
    do {
        // std::memory_order_release here is needed to synchronize with JobSystem::wait()
        // which needs to "see" all changes that happened before the job terminated.
@@ -304,13 +289,17 @@ void JobSystem::finish(Job* job) noexcept {
            // there is still work (e.g.: children), we're done.
            break;
        }
-        job = job->parent == 0x7FFF ? nullptr : &storage[job->parent];
+        Job* const parent = job->parent == 0x7FFF ? nullptr : &storage[job->parent];
+        // destroy this job...
+        jobPool.destroy(job);
+        // ... and check the parent
+        job = parent;
    } while (job);

 #if __ARM_ARCH_7A__
    // on ARMv7a SEL is needed
-        __dsb(0xA);     // ISHST = 0xA (b1010)
-        UTILS_SIGNAL_EVENT();
+    __dsb(0xA);     // ISHST = 0xA (b1010)
+    UTILS_SIGNAL_EVENT();
 #endif
 }

@@ -392,13 +381,6 @@ void JobSystem::emancipate() {
    sThreadState = nullptr;
 }

-void JobSystem::reset() noexcept {
-    assert(!mActiveJobs.load(std::memory_order_relaxed));
-    mJobWaterMark = std::max(mJobWaterMark, (size_t)mNextJobIndex.load(std::memory_order_relaxed));
-    mNextJobIndex.store(0, std::memory_order_relaxed);
-    mMasterJob = nullptr;
-}
-
 io::ostream& operator<<(io::ostream& out, JobSystem const& js) {
    for (auto const& item : js.mThreadStates) {
        out << size_t(std::log2f(item.mask)) << ": " << item.workQueue.getCount() << io::endl;