How to multithread queue processing

Question

C++ containers are supposed to be thread-safe by default. I must be using queue to multithread incorrectly because for this code:

#include <thread>
using std::thread;
#include <iostream>
using std::cout;
using std::endl;
#include <queue>
using std::queue;
#include <string>
using std::string;
using std::to_string;
#include <functional>
using std::ref;


void fillWorkQueue(queue<string>& itemQueue) {
    int size = 40000;
    for(int i = 0; i < size; i++)
        itemQueue.push(to_string(i));
}

void doWork(queue<string>& itemQueue) {
    while(!itemQueue.empty()) {
        itemQueue.pop();
    }   
}

void singleThreaded() {
    queue<string> itemQueue;
    fillWorkQueue(itemQueue);
    doWork(itemQueue);
    cout << "done\n";
}

void multiThreaded() {
    queue<string> itemQueue;
    fillWorkQueue(itemQueue);
    thread t1(doWork, ref(itemQueue));
    thread t2(doWork, ref(itemQueue));
    t1.join();
    t2.join();
    cout << "done\n";
}

int main() {
    cout << endl;

    // Single Threaded
    cout << "singleThreaded\n";
    singleThreaded();
    cout << endl;

    // Multi Threaded
    cout << "multiThreaded\n";
    multiThreaded();
    cout << endl;
}

I'm getting:

singleThreaded
done

multiThreaded
main(32429,0x10e530000) malloc: *** error for object 0x7fe4e3883e00: pointer being freed was not allocated
*** set a breakpoint in malloc_error_break to debug
make: *** [run] Abort trap: 6

What am I doing wrong here?

EDIT

Apparently I misread the link above. Is there a thread-safe queue implementation available that does what I am trying to do? I know this is a common thread organization strategy.

C++ containers are NOT thread safe by default. Quoting from the page you linked to: "Container operations that invalidate any iterators modify the container and cannot be executed concurrently with any operations on existing iterators even if those iterators are not invalidated." — Andrew Shepherd, May 13 '14 at 23:43
*C++ containers are supposed to be thread-safe by default* - whatever gave you that idea? They're only threadsafe when you have multiple threads calling `const` member functions, not if one or more of those threads are modifying the container. — Praetorian, May 13 '14 at 23:44
That's probably a question of wording: containers' methods are supposedly thread safe, but iterators operating on them are not. — didierc, May 13 '14 at 23:47
From the `C++ Concurrency in Action` book by Anthony Williams, page 209, you have to write a class that supports the concurrency operation. Section 7.2.6 (page 209) has the listing for a thread-safe queue without using locks. There are several other queue examples within the book. — CPlusPlus OOA and D, May 14 '14 at 00:00

score 7 · Answer 1 · answered May 14 '14 at 01:11

7

As pointed in the comments, STL containers are not thread-safe for read-write operations. Instead, try concurrent_queue class from TBB or PPL, e.g.:

void doWork(concurrent_queue<string>& itemQueue) {
    string result;
    while(itemQueue.try_pop(result)) {
        // you have `result`
    }   
}

answered May 14 '14 at 01:11

Anton

6,349
1
25
53

2

The other one would be a blockingQueue: http://stackoverflow.com/questions/12805041/c-equivalent-to-javas-blockingqueue – tofi9 May 14 '14 at 05:11

score 3 · Accepted Answer · edited May 23 '17 at 12:03

3

I ended up implementing a BlockingQueue, with the suggested fix to pop, here:

Creating a Blocking Queue

edited May 23 '17 at 12:03

Community

1
1

answered May 14 '14 at 18:24

Chris Redford

16,982
21
89
109

score 0 · Answer 3 · answered Oct 08 '18 at 11:04

C++ containers are definitely not thread safe. BlockingCollection is a C++11 thread safe collection class that is modeled after the .NET BlockingCollection class. It wraps std::deque to provide concurrent adding and taking of items from multiple threads to a queue. As well as stack and priority containers.

How to multithread queue processing

3 Answers3