0

I would like to run tasks in parallel. At this time, I am using a very simple worker pool using a single concurrent queue shared by all the threads.

Every task has a non unique "tag" (an integer in my case). I would like tasks having the same tag not to run in parallel.

At this time, I have implemented it by having a Runner object for each tag. New tasks are sent to the Runner associated with the task tag.

The Runner then enqueue the task into an internal queue and the checks if it is already scheduled into the worker pool. If not, it schedules itself into the pool.

When run inside the pool, it will pop a task from its internal queue, run it and if the queue is not empty, reschedule itself.

Most operations (push, pop, already running check using a simple boolean and queue is not empty check) are "protected" using a mutex for each Runner instance.

My solution seems to be working on a few simple cases, but I fear it is somewhat fragile (for example, I'm almost certain that there is a race condition in the "queue is not empty" check) and uses a lot of locks (one for the main queue and for each tag).

Is there a cleaner solution, maybe less race condition-prone?

Here is a sample code illustrating my implementation (it is probably not working, as I'm replying from home, but I hope it is enough to get the idea):

class Task {
public:
        Task(int tag) : tag_(tag) {};
        void run() {
                // do something
        }

        int getTag() { return tag_; }

private:
        int tag_;
};

class Runner;

class WorkerPool {
public:
        void scheduleRunner(Runner *runner);
};

class TaskQueue {
public:
        void push(std::shared_ptr<Task> task);
        std::shared_ptr<Task> pop();
        bool isEmpty();
};

class Runner {
public:
    Runner(WorkerPool& pool) : pool_(pool) {};

    // schedule execution of the task.
    // called by producers
    void scheduleTask(std::shared_ptr<Task> task) {
        std::unique_lock<std::mutex> lock(lock_);
        queue_.push(std::move(task));
        if (!running_) {
            pool_.scheduleRunner(this);
            running_ = true;
        }
    }

    // run the task from the pool
    void runTaskFromPool() {
        std::unique_lock<std::mutex> lock(lock_);
        std::shared_ptr<Task> task = queue_.pop();
        // we can't leave it locked because the task may need
        // to enqueue another element
        lock.unlock();
        task->run();
        lock.lock();
        if (queue_.isEmpty())
            running_ = false;
        else
            pool_.scheduleRunner(this);
    }

private:
    WorkerPool& pool_;
    TaskQueue queue_;
    std::mutex lock_;
    bool running_;
};

class Dispatcher {
public:
    Dispatcher(WorkerPool& pool) : pool_(pool) {};

    void scheduleTask(std::shared_ptr<Task> task) {
        std::unique_lock<std::mutex> lock(lock_);
        createRunnerIfNotExists_(task->getTag());
        runners_[task->getTag()]->scheduleTask(std::move(task));
    }

private:
    std::map<int, std::unique_ptr<Runner>> runners_;
        WorkerPool& pool_;
    std::mutex lock_;
};
6
  • What language/platform are you currently using? Some platforms have mechanisms already built-in to handle this sort of thing. Mar 5, 2020 at 20:20
  • I'm using C++11. Mar 5, 2020 at 20:22
  • stackoverflow.com/q/15752659 Mar 5, 2020 at 20:23
  • 2
    @WillieTaylor Showing a more concise example of your code and tagging the question with the actual language used would be helpful to receive better answers here. Mar 5, 2020 at 20:43
  • 2
    If you have a small number of tags, just map each tag value to a particular thread. This will guarantee that items with the same tag are serialized. Mar 5, 2020 at 21:40

2 Answers 2

0

You can use a reactive library such as RxCpp.

Use the group_by operator to select on the tag for each piece of work. This gives you a collection of observables. Then perform each observable on its own executor. This process ensures serialization, and handles all the grotty details of pushing work across thread boundaries.

0

It's broadly fine, I've seen this design before and it works.

The only slight concerns are:

  1. you're looking up each Runner twice: first in createRunnerIfNotExists_ and again immediately afterwards. You're also using the logarithmic-time std::map, when it's not at all clear you need the runners to be ordered.

    Using std::unordered_map and just emplace()ing the runner every time would be faster.

  2. this section is not exception-safe:

        lock.unlock();
        task->run();
        lock.lock();
    

    (if the task throws, you never re-acquire the lock and the running_ state ends up wrong).

  3. this section is probably inefficient:

        if (queue_.isEmpty())
            running_ = false;
        else
            pool_.scheduleRunner(this);
    

    you have other threads to handle other tags, so there's no need to be fair. Just keep processing this runner's jobs in a loop until the queue is empty, and then return to the top-level scheduler.

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.