C++ fixed-size queue full/empty signaling

Question

I'm considering an application where a large data needs to be written to file often. I would like to use a queue and have a producer and consumer running on different threads. Additionally, I would like to have a fixed size queue as the data size can be very large. I've implemented a simple test of boost::lockfree::queue with boost::condition_variable to signal the state of the queue. I would like to avoid the mutex but for the exception where the queue is full (block producer) or empty (block consumer)

I would like to know (at risk of opinion based..) if I'm using conditionals properly or if there is a performance issue - compared to using other methods. Here is what I've done so far (small data)

#include <iostream>
#include <boost/thread/thread.hpp>
#include <boost/lockfree/queue.hpp>

#define N 6
#define QN 3   

struct testdata {
    int a;
    int b;
};

boost::lockfree::queue<testdata, boost::lockfree::fixed_size<true>> que(QN);
boost::condition_variable que_has_data, que_has_room;
boost::mutex que_mtx_hd, que_mtx_hr;

void producer(void) {
    testdata td;
    int i = 0;
    boost::mutex::scoped_lock lock(que_mtx_hr);
    boost::this_thread::sleep(boost::posix_time::seconds(1));
    for (;;) {
        td.a = i; td.b = i + 1;
        if (!que.push(td)) {
            std::cout << "producer waiting" << std::endl;
            que_has_room.wait(lock);
        } else {
            std::cout << "pushed " << td.a << std::endl;
            i += 1;
            que.has_data_notify_one();
        }
        if (i > N)
            break;
    }
}

void consumer(void) {
    testdata td;
    boost::mutex::scoped_lock lock(que_mtx_hd);
    for (;;) {
        if (que.pop(td)) {
            std::cout << "popped " << td.a << std::endl;
            if (td.a == N)
                break;
            que_has_room.notify_one();
        } else {
           std::cout << "consumer waiting" << std::endl;
           que_has_data.wait(lock);
        }
    }
    boost::this_thread::sleep(boost::posix_time::seconds(1));
}

int main(void) {
    boost::thread t1(&producer);
    boost::thread t2(&consumer);
    t1.join();
    t2.join();
    return 0;
}

This works (output):

consumer waiting
pushed 0
pushed 1
pushed 2
producer waiting
popped 0
pushed 3
producer waiting
popped 1
pushed 4
producer waiting
popped 2
pushed 5
producer waiting
popped 3
pushed 6
popped 4
popped 5
popped 6

I anticipate that for the most part, data will almost always be available but I want to block in the case of congestion (file write, network, etc). The interest in fixed size is the worry of massive data sets and dynamic allocation in the queue -

(This is more an experiment in what can be done. In reality, my data is updated at most about 20 Hz so just taking a lock on a std::queue that I manage the size of will work very well, too.)

Answer 1

Use a semaphore to cause the producers to sleep when the queue is full, and another semaphore to cause the consumers to sleep when the queue is empty. when the queue is neither full nor empty, the sem_post and sem_wait operations are nonblocking (in newer kernels)

#include <semaphore.h>

template<typename lock_free_container>
class blocking_lock_free
{
public:
    lock_free_queue_semaphore(size_t n) : container(n)
    {
        sem_init(&pop_semaphore, 0, 0);
        sem_init(&push_semaphore, 0, n);
    }

    ~lock_free_queue_semaphore()
    {
        sem_destroy(&pop_semaphore);
        sem_destroy(&push_semaphore);
    }

    bool push(const lock_free_container::value_type& v)
    {
        sem_wait(&push_semaphore);
        bool ret = container::bounded_push(v);
        ASSERT(ret);
        if (ret)
            sem_post(&pop_semaphore);
        else
            sem_post(&push_semaphore); // shouldn't happen
        return ret;
    }

    bool pop(lock_free_container::value_type& v)
    {
        sem_wait(&pop_semaphore);
        bool ret = container::pop(v);
        ASSERT(ret);
        if (ret)
            sem_post(&push_semaphore);
        else
            sem_post(&pop_semaphore); // shouldn't happen
        return ret;
    }
private:
    lock_free_container container;
    sem_t pop_semaphore;
    sem_t push_semaphore;
};

Answer 2

I would be very skeptical about the code containing a lockfree container, two mutexes, and a conditional variable to implement interthread blocking queue. Without even looking further.

I would probably start from the prototype below (maybe first checking if boost::interprocess has anything that I could use right away):

1. wrap boost::circular_buffer into facebook/Folly/Synchronized but with custom locker that does try_lock(), then spins 41 more times with try_lock(), then blocks on lock(), counting occurrences of all three scenarios, and with notify/wait on top
2. I would release that to production in pilot mode and check if I really need to bother with a lock-free container.

Answer 3

Boost provides a single-producer single-consumer queue that is lockfree I think, do you know about it? I think it fits your use case exactly.

http://www.boost.org/doc/libs/1_61_0/doc/html/boost/lockfree/spsc_queue.html

You can use fixed size and you will have blocked consumer if not data available, etc. without implementing it by yourself.