0

I have a function that has to process all files in a set of directories (anything between 5-300 files). The number of parallel threads to be used is user-specified (usually 4). The idea is to start the function in 4 separate threads. When one thread returns, I have to start processing the next (5th) file and so on till all files are complete.

On Windows, WaitForMultipleObjects() with bWaitAll=False helps me here. I have a structure that can be populated, and populated into an array

map<UINT, string>::iterator iter = m_FileList.begin();
string outputPath = GetOutputPath();
void ***threadArgs = (void***)malloc(sizeof(void**)*numThreads);
HANDLE *hdl = (HANDLE*)malloc(sizeof(HANDLE)*numThreads);
DWORD *thr = (DWORD*)malloc(sizeof(DWORD)*numThreads);

for (int t = 0; iter != m_FileList.end() && t < numThreads; t++, iter++)
{
    threadArgs[t] = prepThreadData(t, iter->second, opPath);
    printf("main: starting thread :%d %s outputPath: %s\n", t, iter->second.c_str(), threadArgs[t][2]);
    hdl[t] = CreateThread(NULL, 0, fileProc, (void*)threadArgs[t], 0, &thr[t]);
    if (hdl[t] == NULL)
    {
        err = GetLastError();
        printf("main: thread failed %x %x %s %s\n", err, iter->second.c_str(), threadArgs[t][2]);
    }
}

for (;iter != m_FileList.end(); iter++)
{
    int t = (int)WaitForMultipleObjects(numThreads, hdl, FALSE, INFINITE);
    if (t == WAIT_FAILED)
    {
        err = GetLastError();
        printf("main: thread failed %x %x\n", t, err);
    }
    if (t - WAIT_OBJECT_0 >= 0 && t - WAIT_OBJECT_0 < numThreads)
    {
        free(threadArgs[t][1]);
        free(threadArgs[t][2]);
        free(threadArgs[t]);
        threadArgs[t] = prepThreadData(t, iter->second, opPath);
        printf("main: starting thread :%d %s outputPath: %s\n", t, iter->second.c_str(), threadArgs[t][2]);
        hdl[t] = CreateThread(NULL, 0, fileProc, (void*)threadArgs[t], 0, &thr[t]);
        if (hdl[t] == NULL)
        {
            err = GetLastError();
            printf("main: thread failed %x %x %s %s\n", err, iter->second.c_str(), threadArgs[t][2]);
        }
    }
}
if (WAIT_FAILED == WaitForMultipleObjects(numThreads - 1, hdl, TRUE, INFINITE))     
{
    err = GetLastError();
    printf("main: thread failed %x %x\n", err);
}

My problem now is to get similar functionality using pthreads. The best I can think of is to use semaphores and when one of them is available, spawn a new thread, and instead of using threadArgs array, I will use just one pointer that is allocated memory for every thread spawn. Also, for ease of memory management, the memory allocated for the threadArgs[t] will then be owned by the spawned thread.

Is there a better solution? Or is there something similar to WaitForMutlipleObjects() with pthreads? To put it more concretely, if I replace CreateThread() with pthread_create(), what should I replace WaitForMultipleObjects() with?

1

It sounds like you want a work queue. You could populate that queue with the collection of files that need to be processed, with a function to dequeue an item from the queue that does the necessary locking to prevent races between threads. Then start howmany ever threads you want. Each thread will dequeue an item from the queue, process it, then dequeue the next item. When the queue becomes empty, the thread can either block waiting for more input, or if you know there will be no more input, the thread can then terminate.

Here's a simple example:

#include <cstdio>
#include <mutex>
#include <queue>
#include <thread>

template<typename T>
class ThreadSafeQueue {
public:
    void enqueue(const T& element)
    {
        std::lock_guard<std::mutex> lock(m_mutex);

        m_queue.push(element);
    }

    bool dequeue(T& value)
    {
        std::lock_guard<std::mutex> lock(m_mutex);

        if (m_queue.empty()) {
            return false;
        }

        value = m_queue.front();
        m_queue.pop();

        return true;
    }

private:
    std::mutex m_mutex;
    std::queue<T> m_queue;
};

static void threadEntry(const int threadNumber, ThreadSafeQueue<std::string>* const queue)
{
    std::string filename;

    while (queue->dequeue(filename)) {
        printf("Thread %d processing file '%s'\n", threadNumber, filename.c_str());
    }
}

int main()
{
    ThreadSafeQueue<std::string> queue;

    // Populate queue
    for (int i = 0; i < 100000; ++i) {
        queue.enqueue("filename_" + std::to_string(i) + ".txt");
    }

    const size_t NUM_THREADS = 4;

    // Spin up some threads
    std::thread threads[NUM_THREADS];
    for (int i = 0; i < NUM_THREADS; ++i) {
        threads[i] = std::thread(threadEntry, i, &queue);
    }

    // Wait for threads to finish
    for (int i = 0; i < NUM_THREADS; ++i) {
        threads[i].join();
    }

    return 0;
}

Compile with:

$ g++ example.cpp -pthread

The program defines ThreadSafeQueue -- a queue with internal locking to enable multiple threads to access it concurrently.

The main function begins by populating the queue. It then starts 4 threads. Each thread reads a value from the queue and "processes" it (here, by printing a message to standard output). When the queue is empty, the threads terminates. The main function waits for the threads to terminate before returning.

Note that this design assumes that all elements are populated in the queue before the threads start. With some changes, it could be extended to support handling new work while the threads are running.

| improve this answer | |
  • Your description perfectly fits the scenario, and I find myself stupid to not have thought of it in this way. As they say, our data structures define our solutions. But, even if I had I am fairly sure my solution wouldn't look so clean. Thanks for your solution! – tpb261 Sep 17 at 14:32

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.