2
\$\begingroup\$

Since I had not found exactly what I was looking for I wrote some simple workqueue-based job system myself that is based on boost's lock-free queue. It is supposed to be used in some GUI application where calculation jobs should run in the background. At the moment, results are not conveyed and there is no signaling, when jobs are done, but this is fine with me. The aspect that has not been accounted for by many solutions is that jobs should be abortable. I have covered this by receiving some job ID that can be used as identifier to cancel the job.

I do not synchronize any access to some job, i.e. I set the abort flag to true and simultaneously it can be read. I do not see any issue here because it is ok if the worker thread works a bit longer on it. I hope there are no issues involved with this policy.

My question with the following code is, whether it is really thread-safe and I have no blunders in it. (Of course, I am also open to other suggestions)

#include <iostream>
#include <string>
#include <boost/lockfree/spsc_queue.hpp>
#include <boost/thread.hpp>
#include <boost/container/stable_vector.hpp>

using namespace std;

// interface

struct Job
{
    using JobFunction = std::function < void(bool& /* cancel */) > ;

    Job(int id, JobFunction&& func, bool abort) : id(id), func(std::move(func)), abort(abort) {}

    // only for comparison
    Job(int id) : id(id) {}

    bool operator<(const Job& j) const {return id < j.id;}

    Job(Job&& j) : id(j.id), func(std::move(j.func)), abort(j.abort) {} // msvc13 does not like = default
    Job& operator=(Job&& j) // msvc13 does not like = default
    {
        id = j.id;
        func = std::move(j.func);
        abort = j.abort;
    }

    Job(const Job&) = delete;
    Job& operator=(const Job&) = delete;

    int id;
    JobFunction func;
    mutable bool abort; // does not change sort order; actually, mutable does not make sense anymore, I have had a set for jobs before... I keep it here, if anybody criticizes it right now
};

class JobSystem
{
public:
    static const int CAPACITY = 1000;

    JobSystem() : thread(boost::bind(&JobSystem::worker, this)), jobQueue(CAPACITY), run(true)
    {
    }

    ~JobSystem()
    {
        run = false;
        thread.join();
        cout << "returned from join" << std::endl;
    }

    // not copyable, not assignable
    JobSystem(const JobSystem&) = delete;
    // not copyable, not assignable
    JobSystem& operator=(const JobSystem&) = delete;

    // add job and get job id
    int AddJob(Job::JobFunction&& func)
    {
        Job j(newJobId, std::move(func), false);

        jobs.push_back(std::move(j));

        jobQueue.push(&jobs.back());

        ++newJobId;

        return newJobId - 1;
    }

    // cancels job identified by id
    void AbortJob(int id)
    {
        auto it = std::lower_bound(jobs.begin(), jobs.end(), Job(id), [](const Job& lhs, const Job& rhs){return lhs.id < rhs.id;});
        if (it != jobs.end())
        {

            std::cout << "Job " << id << '/' << it->id << " aborted..." << std::endl;
            it->abort = true;
        }
    }

private:
    static int newJobId;

    // thread function, only one worker
    void worker()
    {
        while (run)
        {
            const Job* job;

            while (jobQueue.pop(job) && run)
            {
                if (!job->abort)
                    job->func(job->abort);
            }
        }

        std::cout << "worked through" << std::endl;
    }

    boost::lockfree::spsc_queue<const Job*> jobQueue;
    boost::container::stable_vector<Job> jobs;
    boost::thread thread;
    bool run;
};

int JobSystem::newJobId = 1;

I do not free up any jobs when they are finished from jobs because I would need to synchronize the stable_vector then, as well, but since it does not take much memory, I decided that freeing up everything at the end of the program is fine. The cout output was mostly for testing by myself, it can just be omitted.

Some simple way to work with this is shown below where some prime numbers are calculated:

    JobSystem jobSystem;

    int id = jobSystem.AddJob([](bool& abort){

        for (int n = 2; n < 10000000; ++n)
        {
            if (!abort)
            {
                bool prime = true;
                int m;
                for (m = 2; m * m <= n; ++m)
                {
                    if (n % m == 0)
                    {
                        prime = false;
                        break;
                    }
                }
                if (prime)
                {
                    cout << n << ' ';
                    boost::this_thread::sleep_for(boost::chrono::milliseconds(10));
                }
            }
        }

        if (abort)
            std::cout << "aborted" << std::endl;
    });

Now, we can wait for 2 seconds or something, and abort with:

jobSystem.AbortJob(id);

At least, all my tests are positive so far, but with multithreading... you never know, so a review might be more revealing.

Edit: Just learnt in another matter out that 'bool's should certainly be atomic, so this is one thing that need to be changed at least.

\$\endgroup\$
  • 1
    \$\begingroup\$ Welcome to code review. I hope you get some good answers. \$\endgroup\$ – pacmaninbw Aug 21 '16 at 14:58

Your Answer

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

Browse other questions tagged or ask your own question.