6
\$\begingroup\$

I'm looking for comments on the design, correctness and performance (not so much style) of a dynamic work queue for OpenMP worker threads.

I have an algorithm that can be thought of in terms of some number of work items (how many is not known beforehand), each of which is executed and may then spawn two child work items. I start with a root work item and continue until there are no more tasks, then terminate. Tasks that are not ancestor and descendant are independent (and therefore can be run in parallel).

I want to use OpenMP to parallelize the work by having threads pop work from a queue. The difficult bit is:

  • waiting for work to become available in an efficient way
  • terminating when there are no work items left, and new ones can't be created any more, but not before

I'm most concerned about the correctness of the termination condition and of my use of the OpenMP functions.

In the version below, the work items are dummies, but the idea is that you could drop in arbitrary work (as long as mutually non-ancestral work is independent).

#include <mutex>
#include <condition_variable>
#include <stack>

#include "omp.h"

using namespace std;

// Dummy nlogn algorithm work item
class Work {
public:
    Work() : n() {}
    Work(int n) : n(n) {}

    // Returns true iff child work created
    bool go(Work& childA, Work& childB) {
        if (n > 1) {
            childA = Work(n - 1);
            childB = Work(n - 1);
            return true;
        }
        return false;
    }

    int n;
};

int main() {
    Work rootWorkItem(20);

    int numWaiting = 0;
    std::mutex m;
    std::condition_variable cv;
    std::stack<Work> workStack;

    workStack.push(rootWorkItem);

#pragma omp parallel
    while (true) {
        std::unique_lock<std::mutex> lock(m);

        if (workStack.empty()) {
            // Register as waiting for work
            numWaiting++;

            // If everybody is now waiting, it means that no one has more work to push
            if (numWaiting >= omp_get_num_threads()) {
                cv.notify_all();
                break;
            }

            // Wait for one of two things:
            //  a) more work has been pushed to the stack
            //  b) everything's done
            while (true) {
                cv.wait(lock);
                if ((numWaiting >= omp_get_num_threads()) || !workStack.empty()) break;
            }

            // If it was b), exit
            if (numWaiting >= omp_get_num_threads()) break;

            // Otherwise, tell everybody that we're no longer waiting
            numWaiting--;
        }

        // Pop a work item to start executing
        Work work = workStack.top();
        workStack.pop();

        lock.unlock();

        // Execute work and, if it spawned children, push them
        Work childA, childB;
        if (work.go(childA, childB)) {
            // Push child work items and tell somebody to get working on them
            lock.lock();
            workStack.push(childA);
            workStack.push(childB);
            cv.notify_one();
            lock.unlock();
        }
    }

    return 0;
}
\$\endgroup\$
  • \$\begingroup\$ How attached are you to using OMP? \$\endgroup\$ – Dannnno Feb 6 '16 at 1:19
  • 1
    \$\begingroup\$ @Dannnno Is there a reason not to? Manually maintaining a thread pool is a lot of extra work. \$\endgroup\$ – alcedine Feb 6 '16 at 7:32
  • \$\begingroup\$ Look at Threading Building Blocks (TBB), or, if you're wedded to OpenMP, use OpenMP tasks, which can do exactly what you want without you having to do create your own implementation. \$\endgroup\$ – Jim Cownie Feb 8 '16 at 15:44
  • \$\begingroup\$ You can look at task constructs from OpenMP 3.0 (e.g stackoverflow.com/documentation/openmp/6930/…) \$\endgroup\$ – Harald Nov 3 '16 at 21:23
2
\$\begingroup\$

Firstly, I know you said you're not so interested in a style review, but:

using namespace std;

Apart from being harmful, this isn't even bringing any benefit. We can just drop it with no changes to the rest of the code.

And if OpenMP is correctly installed, the header should be on the include path (not in your source code location):

#include <omp.h>

The only thing we use from <omp.h> is omp_get_num_threads(); we can avoid needing that if we change the sense of the numWaiting variable, so we get to compare against zero instead:

    unsigned int num_working = 0;
#pragma omp parallel
{
#pragma omp single
    ++num_working;

    while (true) {
        std::unique_lock<std::mutex> lock(m);

        if (workStack.empty()) {
            --num_working;

            if (num_working == 0) {
                cv.notify_all();
                break;
            }

            while (true) {
                cv.wait(lock);
                if (num_working == 0 || !workStack.empty()) break;
            }

            if (num_working == 0) break;

            ++num_working;
        }

Instead of the #pragma omp single, we could make num_working be a std::atomic<unsigned> instead.


When we add two items to the work queue, we only wake up one worker thread. We need to wake up a thread for each child, potentially:

        workStack.push(childA);
        cv.notify_one();
        workStack.push(childB);
        cv.notify_one();

Possibly a style issue, but the while (true) loop looks better as a do ... while:

            do {
                cv.wait(lock);
            } while (num_working > 0 && workStack.empty());
\$\endgroup\$

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.