2
\$\begingroup\$

This code exists in my home repo and in general nobody else has really touched it. Living in such a vacuum has left me kind of uncertain of the actual code quality and correctness which is why I am approaching this site hopeful for a second set of eyes.

This is implemented and working, however what I am interested in is any feedback on:

  1. Thread safety. Have I messed anything up? If so, how do I fix it?
  2. Readability. Is it clear and concise? Do any parts trip you up?
  3. General correctness. Are there any bugs or gotchas you can observe upon review?
  4. Additional observations. I'm looking for critique so I can improve this, but I'd also like to know what works.

threadpool.h

#ifndef _MV_THREADPOOL_H_
#define _MV_THREADPOOL_H_

#include <thread>
#include <mutex>
#include <atomic>
#include <memory>
#include <vector>
#include <list>
#include <chrono>

#include "boost/asio.hpp"

namespace MV{

    class TaskStatus {
    public:
        TaskStatus(){}
        TaskStatus(std::shared_ptr<std::atomic<bool>> a_isFinished):
            isFinished(a_isFinished){
        }
        bool clear() {
            isFinished = nullptr;
        }
        bool active() const{
            return isFinished != nullptr;
        }
        bool finished(){
            return !isFinished || isFinished->load();
        }
        void join(){
            while(!finished()){
                std::this_thread::sleep_for(std::chrono::nanoseconds(100));
            }
        }
    private:
        std::shared_ptr<std::atomic<bool>> isFinished;
    };

    class ThreadPool {
    private:
        class ThreadTask {
        public:
            ThreadTask(const std::function<void()> &a_call);
            ThreadTask(const std::function<void()> &a_call, const std::function<void()> &a_onFinish);
            ThreadTask(ThreadTask&& a_rhs):
                groupCounter(std::move(a_rhs.groupCounter)),
                onGroupFinish(std::move(a_rhs.onGroupFinish)),
                isGroupFinished(std::move(a_rhs.isGroupFinished)),
                isRun(std::move(a_rhs.isRun)),
                isFinished(std::move(a_rhs.isFinished)),
                handled(std::move(a_rhs.handled)),
                call(std::move(a_rhs.call)),
                onFinish(std::move(a_rhs.onFinish)){
            }

            void operator()();

            bool finished();

            void group(const std::shared_ptr<std::atomic<size_t>> &a_groupCounter, const std::shared_ptr<std::function<void()>> &a_onGroupFinish, const std::shared_ptr<std::atomic<bool>> &a_isGroupFinished, bool a_groupFinishWaitForFrame);

            std::shared_ptr<std::atomic<bool>> isFinished;
            std::shared_ptr<std::atomic<bool>> isGroupFinished;
        private:
            ThreadTask(const ThreadTask& a_rhs) = delete;
            ThreadTask& operator=(const ThreadTask& a_rhs) = delete;

            std::shared_ptr<std::atomic<size_t>> groupCounter;
            std::shared_ptr<std::function<void()>> onGroupFinish;
            bool groupFinishWaitForFrame;
            std::unique_ptr<std::atomic<bool>> isRun;
            bool handled;
            std::function<void()> call;
            std::function<void()> onFinish;
        };

    public:
        struct TaskDefinition {
            TaskDefinition(const std::function<void()> &a_task);
            TaskDefinition(const std::function<void()> &a_task, const std::function<void()> &a_onComplete);
            std::function<void()> task;
            std::function<void()> onComplete;
        };

        ThreadPool();
        ThreadPool(std::size_t a_threads);
        ~ThreadPool();

        TaskStatus task(const std::function<void()> &a_task);
        TaskStatus task(const std::function<void()> &a_task, const std::function<void()> &a_onComplete);

        typedef std::vector<TaskDefinition> TaskList;
        TaskStatus tasks(const TaskList &a_tasks, const std::function<void()> &a_onGroupComplete, bool a_groupFinishWaitForFrame = true);

        size_t run();

        size_t threads() const{
            return totalThreads;
        }
    private:
        std::recursive_mutex lock;
        boost::asio::io_service service;
        using asio_worker = std::unique_ptr<boost::asio::io_service::work>;
        asio_worker working;
        size_t totalThreads;
        std::list<ThreadTask> runningTasks;
        std::vector<std::unique_ptr<std::thread>> workers;
    };
}

#endif

threadpool.cpp

#include "threadPool.h"

namespace MV{
    ThreadPool::ThreadTask::ThreadTask(const std::function<void()> &a_call):
        call(a_call),
        isRun(std::make_unique<std::atomic<bool>>(false)),
        isFinished(std::make_shared<std::atomic<bool>>(false)),
        handled(false){
    }

    ThreadPool::ThreadTask::ThreadTask(const std::function<void()> &a_call, const std::function<void()> &a_onFinish) :
        call(a_call),
        onFinish(a_onFinish),
        isRun(std::make_unique<std::atomic<bool>>(false)),
        isFinished(std::make_shared<std::atomic<bool>>(false)),
        handled(false){
    }

    bool ThreadPool::ThreadTask::finished() {
        if(isFinished->load()){
            if(onFinish && !handled){
                handled = true;
                onFinish();
                if(groupFinishWaitForFrame && (onGroupFinish && groupCounter && *groupCounter == 0)){
                    (*onGroupFinish)();
                }
            }
            return true;
        }
        return false;
    }

    void ThreadPool::ThreadTask::operator()() {
        bool isFalse = false;
        if(isRun->compare_exchange_strong(isFalse, true)){
            call();
            if((groupCounter && --(*groupCounter) == 0)){
                if(!groupFinishWaitForFrame && onGroupFinish){
                    (*onGroupFinish)();
                }
                *isGroupFinished = true;
            }
            *isFinished = true;
        }
    }

    void ThreadPool::ThreadTask::group(const std::shared_ptr<std::atomic<size_t>> &a_groupCounter, const std::shared_ptr<std::function<void()>> &a_onGroupFinish, const std::shared_ptr<std::atomic<bool>> &a_isGroupFinished, bool a_groupFinishWaitForFrame) {
        groupCounter = a_groupCounter;
        onGroupFinish = a_onGroupFinish;
        groupFinishWaitForFrame = a_groupFinishWaitForFrame;
        isGroupFinished = a_isGroupFinished;
    }

    ThreadPool::ThreadPool():
        ThreadPool(std::thread::hardware_concurrency()) {
    }

    ThreadPool::~ThreadPool() {
        service.stop();
        for(auto&& worker : workers){
            worker->join();
        }
    }

    ThreadPool::ThreadPool(size_t a_threads):
        totalThreads(a_threads < 2 ? 1 : a_threads - 1),
        service(),
        working(new asio_worker::element_type(service)) {

        std::cout << "Info: Generating ThreadPool [" << totalThreads << "]" << std::endl;

        for(size_t i = 0; i < totalThreads; ++i) {
            workers.emplace_back(new std::thread([this]{
                service.run();
            }));
        }
    }

    TaskStatus ThreadPool::task(const std::function<void()> &a_task) {
        std::lock_guard<std::recursive_mutex> guard(lock);

        runningTasks.push_back({a_task});
        auto thisTask = runningTasks.end();
        --thisTask;

        service.post([=](){(*thisTask)();});

        return{thisTask->isFinished};
    }

    TaskStatus ThreadPool::task(const std::function<void()> &a_task, const std::function<void()> &a_onComplete) {
        std::lock_guard<std::recursive_mutex> guard(lock);

        runningTasks.emplace_back(a_task, a_onComplete);
        auto thisTask = runningTasks.end();
        --thisTask;
        auto completed = std::make_shared<std::atomic<bool>>(false);
        std::weak_ptr<std::atomic<bool>> weakCompleted = completed;

        service.post([=](){(*thisTask)(); if(!weakCompleted.expired()){ *weakCompleted.lock() = true; }});

        return{thisTask->isFinished};
    }

    TaskStatus ThreadPool::tasks(const TaskList &a_tasks, const std::function<void()> &a_onGroupComplete, bool a_groupFinishWaitForFrame) {
        auto groupCounter = std::make_shared<std::atomic<size_t>>(a_tasks.size());
        auto onGroupComplete = std::make_shared<std::function<void()>>(a_onGroupComplete);
        auto isGroupComplete = std::make_shared<std::atomic<bool>>(false);

        std::lock_guard<std::recursive_mutex> guard(lock);
        for(auto&& currentTask : a_tasks){
            runningTasks.emplace_back(currentTask.task, currentTask.onComplete);
            auto thisTask = runningTasks.end();
            --thisTask;
            thisTask->group(groupCounter, onGroupComplete, isGroupComplete, a_groupFinishWaitForFrame);

            service.post([=](){(*thisTask)();});
        }
        return {isGroupComplete};
    }

    size_t ThreadPool::run() {
        std::lock_guard<std::recursive_mutex> guard(lock);

        runningTasks.remove_if([](ThreadTask& a_task){
            return a_task.finished();
        });

        return runningTasks.size();
    }


    ThreadPool::TaskDefinition::TaskDefinition(const std::function<void()> &a_task):
        task(a_task) {
    }

    ThreadPool::TaskDefinition::TaskDefinition(const std::function<void()> &a_task, const std::function<void()> &a_onComplete) :
        task(a_task),
        onComplete(a_onComplete) {
    }

}

Now the following serves as an example of the above. While review of all code is fair game, I am primarily interested in the ThreadPool itself! This is primarily to show a relatively complete use-case.

emitter.h

#ifndef _MV_SCENE_EMITTER_H_
#define _MV_SCENE_EMITTER_H_

#include "sprite.h"
#include <atomic>

namespace MV {
    namespace Scene {

        struct ParticleChangeValues {
            AxisAngles directionalChange;
            AxisAngles rotationalChange;

            float beginSpeed = 0.0f;
            float endSpeed = 0.0f;

            Scale beginScale;
            Scale endScale;

            Color beginColor;
            Color endColor;

            float maxLifespan = 1.0f;

            float gravityMagnitude = 0.0f;
            AxisAngles gravityDirection;

            float animationFramesPerSecond = 10.0f;

            template <class Archive>
            void serialize(Archive & archive) {
                archive(CEREAL_NVP(directionalChange), CEREAL_NVP(rotationalChange),
                    CEREAL_NVP(beginSpeed), CEREAL_NVP(endSpeed),
                    CEREAL_NVP(beginScale), CEREAL_NVP(endScale),
                    CEREAL_NVP(beginColor), CEREAL_NVP(endColor),
                    CEREAL_NVP(maxLifespan),
                    CEREAL_NVP(gravityMagnitude), CEREAL_NVP(gravityDirection),
                    CEREAL_NVP(animationFramesPerSecond)
                );
            }
        };

        struct Particle {
            //return true if dead.
            bool update(double a_dt) {
                float timeScale = static_cast<float>(a_dt);
                totalLifespan = std::min(totalLifespan + timeScale, change.maxLifespan);

                float mixValue = totalLifespan / change.maxLifespan;

                direction += change.directionalChange * timeScale;
                rotation += change.rotationalChange * timeScale;

                speed = mix(change.beginSpeed, change.endSpeed, mixValue);
                scale = mix(change.beginScale, change.endScale, mixValue);
                color = mix(change.beginColor, change.endColor, mixValue);

                Point<> distance(0.0f, -speed, 0.0f);
                rotatePoint(distance, direction);
                position += distance * timeScale;
                position += gravityConstant * timeScale;

                currentFrame = static_cast<int>(wrap(static_cast<float>(myTextures.size() * (change.animationFramesPerSecond / timeScale)), 0.0f, static_cast<float>(myTextures.size())));
                return totalLifespan == change.maxLifespan;
            }

            void reset() {
                totalLifespan = 0.0f;
            }

            Point<> position;
            float speed;
            AxisAngles direction;
            AxisAngles rotation;
            Scale scale;
            Color color;
            int currentFrame;

            float totalLifespan = 0.0f;

            ParticleChangeValues change;

            std::vector<std::shared_ptr<TextureHandle>> myTextures;

            void setGravity(float a_magnitude, const AxisAngles &a_direction = AxisAngles(0.0f, 180.0f, 0.0f)) {
                gravityConstant.locate(0.0f, -a_magnitude, 0.0f);
                rotatePoint(gravityConstant, a_direction);
            }
        private:
            Point<> gravityConstant;
        };

        struct EmitterSpawnProperties {
            uint32_t maximumParticles = std::numeric_limits<uint32_t>::max();

            float minimumSpawnRate = 0.0f;
            float maximumSpawnRate = 1.0f;

            Point<> maximumPosition;
            Point<> minimumPosition;

            AxisAngles maximumDirection;
            AxisAngles minimumDirection;

            AxisAngles maximumRotation;
            AxisAngles minimumRotation;

            ParticleChangeValues minimum;
            ParticleChangeValues maximum;

            template <class Archive>
            void serialize(Archive & archive) {
                archive(CEREAL_NVP(maximumParticles),
                    CEREAL_NVP(minimumSpawnRate), CEREAL_NVP(maximumSpawnRate),
                    CEREAL_NVP(minimumPosition), CEREAL_NVP(maximumPosition),
                    CEREAL_NVP(minimumDirection), CEREAL_NVP(maximumDirection),
                    CEREAL_NVP(minimumRotation), CEREAL_NVP(maximumRotation),
                    CEREAL_NVP(minimum), CEREAL_NVP(maximum)
                );
            }
        };

        EmitterSpawnProperties loadEmitterProperties(const std::string &a_file);

        class Emitter : public Drawable {
            friend Node;
            friend cereal::access;
        public:
            DrawableDerivedAccessors(Emitter)

            std::shared_ptr<Emitter> properties(const EmitterSpawnProperties &a_emitterProperties);

            const EmitterSpawnProperties& properties() const;

            EmitterSpawnProperties& properties();

            bool enabled() const;
            bool disabled() const;

            std::shared_ptr<Emitter> enable();
            std::shared_ptr<Emitter> disable();

            virtual void update(double a_dt) override;

            ~Emitter();

        protected:
            Emitter(const std::weak_ptr<Node> &a_owner, ThreadPool &a_pool);

            template <class Archive>
            void serialize(Archive & archive) {
                archive(
                    CEREAL_NVP(spawnProperties),
                    CEREAL_NVP(spawnParticles),
                    cereal::make_nvp("Drawable", cereal::base_class<Drawable>(this))
                );
            }

            template <class Archive>
            static void load_and_construct(Archive & archive, cereal::construct<Emitter> &construct) {
                ThreadPool *pool = nullptr;
                archive.extract(cereal::make_nvp("pool", pool));
                MV::require<PointerException>(pool != nullptr, "Null thread pool in Emitter::load_and_construct.");
                construct(std::shared_ptr<Node>(), *pool);
                archive(
                    cereal::make_nvp("spawnProperties", construct->spawnProperties),
                    cereal::make_nvp("spawnParticles", construct->spawnParticles),
                    cereal::make_nvp("Drawable", cereal::base_class<Drawable>(construct.ptr()))
                );
                construct->initialize();
            }

            virtual std::shared_ptr<Component> cloneImplementation(const std::shared_ptr<Node> &a_parent) {
                return cloneHelper(a_parent->attach<Emitter>(pool).self());
            }

            virtual std::shared_ptr<Component> cloneHelper(const std::shared_ptr<Component> &a_clone);

        private:
            virtual BoxAABB<> boundsImplementation();

            Point<> randomMix(const Point<> &a_rhs, const Point<> &a_lhs);

            Color randomMix(const Color &a_rhs, const Color &a_lhs);

            void spawnParticle(size_t a_groupIndex);

            void spawnParticlesOnMultipleThreads(double a_dt);

            void updateParticlesOnMultipleThreads(double a_dt);

            void loadParticlesToPoints(size_t a_groupIndex);

            void loadParticlePointsFromGroups();

            void Emitter::loadPointsFromBufferAndAllowUpdate();


            double accumulatedTimeDelta = 0.0f;

            EmitterSpawnProperties spawnProperties;

            size_t emitterThreads;

            struct ThreadData {
                std::vector<Particle> particles;
                std::vector<DrawPoint> points;
                std::vector<GLuint> vertexIndices;
            };
            std::vector<DrawPoint> pointBuffer;
            std::vector<GLuint> vertexIndexBuffer;

            std::vector<ThreadData> threadData;

            bool spawnParticles = true;
            static const double MAX_TIME_STEP;
            static const int MAX_PARTICLES_PER_FRAME = 2500;
            std::atomic<double> timeSinceLastParticle = 0.0;
            double nextSpawnDelta = 0.0;

            ThreadPool& pool;

            std::atomic<bool> updateInProgress;
        };
    }
}

#endif

emitter.cpp

#include "emitter.h"

CEREAL_REGISTER_TYPE(MV::Scene::Emitter);

namespace MV {
    namespace Scene {

        const double Emitter::MAX_TIME_STEP  = .25;

        Point<> Emitter::randomMix(const Point<>& a_rhs, const Point<>& a_lhs) {
            return{
                mix(a_rhs.x, a_lhs.x, randomNumber(0.0f, 1.0f)),
                mix(a_rhs.y, a_lhs.y, randomNumber(0.0f, 1.0f)),
                mix(a_rhs.z, a_lhs.z, randomNumber(0.0f, 1.0f))
            };
        }

        Color Emitter::randomMix(const Color & a_rhs, const Color & a_lhs) {
            return{
                mix(a_rhs.R, a_lhs.R, randomNumber(0.0f, 1.0f)),
                mix(a_rhs.G, a_lhs.G, randomNumber(0.0f, 1.0f)),
                mix(a_rhs.B, a_lhs.B, randomNumber(0.0f, 1.0f)),
                mix(a_rhs.A, a_lhs.A, randomNumber(0.0f, 1.0f))
            };
        }

        void Emitter::spawnParticle(size_t a_groupIndex) {
            Particle particle;

            particle.position = randomMix(spawnProperties.minimumPosition, spawnProperties.maximumPosition);
            particle.rotation = randomMix(spawnProperties.minimumRotation, spawnProperties.maximumRotation);
            particle.change.rotationalChange = randomMix(spawnProperties.minimum.rotationalChange, spawnProperties.maximum.rotationalChange);

            particle.direction = randomMix(spawnProperties.minimumDirection, spawnProperties.maximumDirection);
            particle.change.directionalChange = randomMix(spawnProperties.minimum.directionalChange, spawnProperties.maximum.directionalChange);

            particle.change.beginSpeed = mix(spawnProperties.minimum.beginSpeed, spawnProperties.maximum.beginSpeed, randomNumber(0.0f, 1.0f));
            particle.change.endSpeed = mix(spawnProperties.minimum.endSpeed, spawnProperties.maximum.endSpeed, randomNumber(0.0f, 1.0f));

            particle.change.beginScale = mix(spawnProperties.minimum.beginScale, spawnProperties.maximum.beginScale, randomNumber(0.0f, 1.0f));
            particle.change.endScale = mix(spawnProperties.minimum.endScale, spawnProperties.maximum.endScale, randomNumber(0.0f, 1.0f));

            particle.change.beginColor = randomMix(spawnProperties.minimum.beginColor, spawnProperties.maximum.beginColor);
            particle.change.endColor = randomMix(spawnProperties.minimum.endColor, spawnProperties.maximum.endColor);

            particle.change.animationFramesPerSecond = mix(spawnProperties.minimum.animationFramesPerSecond, spawnProperties.maximum.animationFramesPerSecond, randomNumber(0.0f, 1.0f));

            particle.change.maxLifespan = mix(spawnProperties.minimum.maxLifespan, spawnProperties.maximum.maxLifespan, randomNumber(0.0f, 1.0f));

            particle.setGravity(
                mix(spawnProperties.minimum.gravityMagnitude, spawnProperties.maximum.gravityMagnitude, randomNumber(0.0f, 1.0f)),
                randomMix(spawnProperties.minimum.gravityDirection, spawnProperties.maximum.gravityDirection)
                );

            particle.update(0.0f);

            threadData[a_groupIndex].particles.push_back(particle);
        }

        void Emitter::spawnParticlesOnMultipleThreads(double a_dt) {
            timeSinceLastParticle.store(timeSinceLastParticle.load() + a_dt);
            size_t particlesToSpawn = static_cast<size_t>(timeSinceLastParticle.load() / nextSpawnDelta);
            size_t totalParticles = std::min(std::accumulate(threadData.begin(), threadData.end(), static_cast<size_t>(0), [](size_t a_total, ThreadData& a_group) {return a_group.particles.size() + a_total; }), static_cast<size_t>(spawnProperties.maximumParticles));

            particlesToSpawn = std::min(particlesToSpawn + totalParticles, static_cast<size_t>(spawnProperties.maximumParticles)) - totalParticles;

            double maxTime = .001;

            if (particlesToSpawn >= emitterThreads) {
                ThreadPool::TaskList spawnTasks;
                for (size_t currentThread = 0; currentThread < emitterThreads; ++currentThread) {
                    spawnTasks.emplace_back([=]() {
                        MV::Stopwatch timer;
                        timer.start();
                        for (size_t count = 0; timer.check() < maxTime && count < particlesToSpawn / emitterThreads; ++count) {
                            spawnParticle(currentThread);
                        }
                    });
                }
                timeSinceLastParticle = 0.0f;
                nextSpawnDelta = randomNumber(spawnProperties.minimumSpawnRate, spawnProperties.maximumSpawnRate);

                pool.tasks(spawnTasks, [=]() {
                    updateParticlesOnMultipleThreads(a_dt);
                }, false);
            } else if (particlesToSpawn > 0) {
                auto randomOffset = randomNumber(static_cast<size_t>(0), emitterThreads);
                for (size_t count = 0; count < particlesToSpawn; ++count) {
                    spawnParticle((count + randomOffset) % emitterThreads);
                }

                timeSinceLastParticle = 0.0f;
                nextSpawnDelta = randomNumber(spawnProperties.minimumSpawnRate, spawnProperties.maximumSpawnRate);
                updateParticlesOnMultipleThreads(a_dt);
            } else {
                updateParticlesOnMultipleThreads(a_dt);
            }
        }

        void Emitter::updateParticlesOnMultipleThreads(double a_dt) {
            ThreadPool::TaskList spawnTasks;
            for (size_t currentThread = 0; currentThread < emitterThreads; ++currentThread) {
                spawnTasks.emplace_back([=]() {
                    threadData[currentThread].particles.erase(std::remove_if(threadData[currentThread].particles.begin(), threadData[currentThread].particles.end(), [&](Particle& a_particle) {
                        return a_particle.update(a_dt);
                    }), threadData[currentThread].particles.end());
                    loadParticlesToPoints(currentThread);
                });
            }
            auto tmpUpdate = pool.tasks(spawnTasks, [=]() {
                loadParticlePointsFromGroups();
            }, false);
        }

        void Emitter::loadParticlesToPoints(size_t a_groupIndex) {
            threadData[a_groupIndex].points.clear();
            threadData[a_groupIndex].vertexIndices.clear();

            std::vector<TexturePoint> texturePoints;
            if (ourTexture) {
                texturePoints.push_back({ static_cast<float>(ourTexture->percentBounds().minPoint.x), static_cast<float>(ourTexture->percentBounds().minPoint.y) });
                texturePoints.push_back({ static_cast<float>(ourTexture->percentBounds().minPoint.x), static_cast<float>(ourTexture->percentBounds().maxPoint.y) });
                texturePoints.push_back({ static_cast<float>(ourTexture->percentBounds().maxPoint.x), static_cast<float>(ourTexture->percentBounds().maxPoint.y) });
                texturePoints.push_back({ static_cast<float>(ourTexture->percentBounds().maxPoint.x), static_cast<float>(ourTexture->percentBounds().minPoint.y) });
            } else {
                texturePoints.push_back({ 0.0f, 0.0f });
                texturePoints.push_back({ 0.0f, 1.0f });
                texturePoints.push_back({ 1.0f, 1.0f });
                texturePoints.push_back({ 1.0f, 0.0f });
            }
            for (auto &&particle : threadData[a_groupIndex].particles) {
                BoxAABB<> bounds(toPoint(particle.scale / 2.0f), toPoint(particle.scale / -2.0f));

                for (size_t i = 0; i < 4; ++i) {
                    auto corner = bounds[i];
                    rotatePoint(corner, particle.rotation);
                    corner += particle.position;
                    threadData[a_groupIndex].points.push_back(DrawPoint(corner, particle.color, texturePoints[i]));
                }

                appendQuadVertexIndices(threadData[a_groupIndex].vertexIndices, static_cast<GLuint>(threadData[a_groupIndex].points.size()));
            }
        }

        void Emitter::loadParticlePointsFromGroups() {
            pointBuffer.clear();
            vertexIndexBuffer.clear();

            size_t pointCount = 0;
            size_t vertexCount = 0;
            for (int group = 0; group < emitterThreads; ++group) {
                pointCount += threadData[group].points.size();
                vertexCount += threadData[group].vertexIndices.size();
            }

            pointBuffer.resize(pointCount);
            vertexIndexBuffer.resize(vertexCount);

            size_t pointOffset = 0;
            size_t vertexOffset = 0;
            ThreadPool::TaskList copyTasks;
            for (int group = 0; group < emitterThreads; ++group) {
                copyTasks.emplace_back([=]() {
                    size_t indexSize = threadData[group].vertexIndices.size();
                    moveCopy(pointBuffer, threadData[group].points, pointOffset);
                    moveCopy(vertexIndexBuffer, threadData[group].vertexIndices, vertexOffset);
                    for (size_t index = vertexOffset; index < vertexOffset + indexSize; ++index) {
                        vertexIndexBuffer[index] += static_cast<GLuint>(pointOffset);
                    }
                });
                pointOffset += threadData[group].points.size();
                vertexOffset += threadData[group].vertexIndices.size();
            }
            pool.tasks(copyTasks, [=]() {
                loadPointsFromBufferAndAllowUpdate();
            }, false);
        }

        void Emitter::loadPointsFromBufferAndAllowUpdate() {
            std::lock_guard<std::recursive_mutex> guard(lock);
            points.clear();
            vertexIndices.clear();
            std::swap(points, pointBuffer);
            std::swap(vertexIndices, vertexIndexBuffer);
            updateInProgress.store(false);
        }

        std::shared_ptr<Emitter> Emitter::properties(const EmitterSpawnProperties & a_emitterProperties) {
            spawnProperties = a_emitterProperties;
            nextSpawnDelta = randomNumber(spawnProperties.minimumSpawnRate, spawnProperties.maximumSpawnRate);
            return std::static_pointer_cast<Emitter>(shared_from_this());
        }

        const EmitterSpawnProperties & Emitter::properties() const {
            return spawnProperties;
        }

        EmitterSpawnProperties & Emitter::properties() {
            nextSpawnDelta = 0.0f;
            return spawnProperties;
        }

        bool Emitter::enabled() const {
            return spawnParticles;
        }

        bool Emitter::disabled() const {
            return !spawnParticles;
        }

        std::shared_ptr<Emitter> Emitter::enable() {
            spawnParticles = true;
            return std::static_pointer_cast<Emitter>(shared_from_this());
        }

        std::shared_ptr<Emitter> Emitter::disable() {
            spawnParticles = false;
            return std::static_pointer_cast<Emitter>(shared_from_this());
        }

        void Emitter::update(double a_dt) {
            bool falseValue = false;
            accumulatedTimeDelta += a_dt;
            if (updateInProgress.compare_exchange_strong(falseValue, true)) {
                pool.task([&]() {
                    double dt = std::min(accumulatedTimeDelta, MAX_TIME_STEP);
                    accumulatedTimeDelta = 0.0;

                    if (nextSpawnDelta == 0.0) {
                        nextSpawnDelta = randomNumber(spawnProperties.minimumSpawnRate, spawnProperties.maximumSpawnRate);
                    }
                    if (enabled()) {
                        spawnParticlesOnMultipleThreads(dt);
                    } else {
                        updateParticlesOnMultipleThreads(dt);
                    }
                });
            }
        }

        Emitter::~Emitter() {
            while (updateInProgress.load()) {
                std::this_thread::sleep_for(std::chrono::nanoseconds(100));
            }
        }

        Emitter::Emitter(const std::weak_ptr<Node> &a_owner, ThreadPool &a_pool) :
            Drawable(a_owner),
            pool(a_pool),
            emitterThreads(a_pool.threads()),
            threadData(emitterThreads),
            updateInProgress(false){

            points.resize(4);
            clearTexturePoints(points);
            appendQuadVertexIndices(vertexIndices, 0);
        }

        BoxAABB<> Emitter::boundsImplementation() {
            return{ spawnProperties.minimumPosition, spawnProperties.maximumPosition };
        }

        std::shared_ptr<Component> Emitter::cloneHelper(const std::shared_ptr<Component> &a_clone) {
            Drawable::cloneHelper(a_clone);
            auto emitterClone = std::static_pointer_cast<Emitter>(a_clone);
            emitterClone->emitterThreads = emitterThreads;
            emitterClone->spawnProperties = spawnProperties;
            emitterClone->spawnParticles = spawnParticles;
            return a_clone;
        }

        MV::Scene::EmitterSpawnProperties loadEmitterProperties(const std::string &a_file) {
            try {
                std::ifstream file(a_file);
                std::shared_ptr<MV::Scene::Node> saveScene;
                cereal::JSONInputArchive archive(file);
                EmitterSpawnProperties EmitterProperties;
                archive(CEREAL_NVP(EmitterProperties));
                return EmitterProperties;
            } catch (::cereal::RapidJSONException &a_exception) {
                std::cerr << "Failed to load emitter: " << a_exception.what() << std::endl;
                return{};
            }
        }

    }
}
\$\endgroup\$
2
\$\begingroup\$
bool ThreadPool::ThreadTask::finished() {
    if(isFinished->load()){
        if(onFinish && !handled){
            handled = true;
            onFinish();
            if(groupFinishWaitForFrame && (onGroupFinish && groupCounter && *groupCounter == 0)){
                (*onGroupFinish)();
            }
        }
        return true;
    }
    return false; }
  • This is a minor stylistic thing. I prefer for code to return as soon as possible, so the main body of the program continues the normal execution path, and reducing the indent level. I'd write it this way:

    bool ThreadPool::ThreadTask::finished() {
        if(!isFinished->load()){
            return false;
        }
        if(onFinish && !handled){
            handled = true;
            onFinish();
            if(groupFinishWaitForFrame && (onGroupFinish && groupCounter && *groupCounter == 0)){
                (*onGroupFinish)();
            }
        }
        return true;
    }
    
  • In ThreadPool::ThreadPool(size_t a_threads) you're using std::cout to log. This violates two paradigms: your code that does the work shouldn't have any user interface stuff in it, and you should use a flexible logging package that can easily be switched on and off. How about Boost::Log?

  • ThreadPool::tasks() and ThreadPool::task() have some common code. How about factoring it out?
  • Emitter::randomMix has repeated randomNumber(0.0, 1.0). Make a helper function.
\$\endgroup\$
  • \$\begingroup\$ Thank you very much for the Boost::Log suggestion! I've been using cout and cerr (and have a thread-safe cout/cerr as well which ensure output is not mixed) I was planning on piping cerr to a log file, but Boost::Log probably has a less obtuse and more flexible method for toggling on/off/different levels. I'll investigate. \$\endgroup\$ – M2tM May 5 '15 at 18:19
  • \$\begingroup\$ I go back and forth on early return vs wrapping in an indent level. On one hand I agree I like early out, on the other, I usually prefer like 4 to 8 line functions which are typically fine with that indentation. I find if it gets much longer than that the early out is much much more valuable, but then you probably want to refactor to shrink the method anyway. Good call though. :) \$\endgroup\$ – M2tM May 5 '15 at 18:22
  • \$\begingroup\$ I may be able to clean task/tasks up a little, and I also noticed emplace_back is used in tasks, but not task. Not sure if there's a reason, but I doubt it. I'll investigate using emplace_back for both while sharing a bit better between the functions if possible, thanks for bringing it up! \$\endgroup\$ – M2tM May 5 '15 at 18:25
  • 1
    \$\begingroup\$ Thanks @glampert, I tried editing it three times and couldn't get it to recognize it as code. \$\endgroup\$ – Snowbody May 5 '15 at 18:46
  • 2
    \$\begingroup\$ @Snowbody, you need at least 8 spaces to enable code formatting when under a bullet-point or list item, it seems. \$\endgroup\$ – glampert May 5 '15 at 18:56

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.