4
\$\begingroup\$

Executive summary

The twist is that the consumers do not consume, they just read. The producer makes items continuously in one of two alternating slots. Readers read from the other slot. When producer comes up with a new item it waits until readers are done with their read slot (that would become the new write slot). The slots effectively alternate between write slot and read slot.

The long story

Background: solving a practical problem - watching my son overnight using an infrared Raspberry Pi camera. Came up with a basic C++ web server. One thread takes pictures (the producer). It takes ~6 secs for a picture! As explained above, the pictures are kept in memory in 2 picture slots - one in which the current picture is taken and the other one from which one can only read. The readers are web browsers (> 50 msecs to serve a picture).

The code below works (as much as tested) but doesn't look pretty. Locks are abused and notify() is called way too many times. Is there for sure no deadlock/race condition?

I looked at myself reasoning about the problem and saw myself adding a mutex/condition variable to solve this or that... notify() called too often? Well, adding 'if (no readers)' is tempting but things can change underneath... so protect that with a mutex... patching... I'd like an approach that's crystal clear.

Feel free to scrap the code and come up with a different approach. This wheel should have been already invented somewhere. Use semaphores if that makes things clearer. Should I ask for more than 2 slots (as latency becomes an issue on slow links/readers) :)? Maybe later.

The idea is that callbacks are called with an index (slot) argument. For the duration of producer callback there should be a guarantee that no readers are using that index, etc. main() is just the test harness.

#include <mutex>
#include <condition_variable>
#include <atomic>
#include <iostream>
#include <thread>
#include <unistd.h>

std::mutex global_mutex;
std::condition_variable cv[2];
std::mutex slot_mutex[2];
std::atomic_int reader_count[2];  // Number of readers
int read_slot{ 0 };
int write_slot{ 1 };


// When callback is called below only the producer uses the slot index
void produce(std::function<void(int index)>& callback)
{
    {   // 'Swap' the read/write slots
        std::lock_guard<std::mutex> guard(global_mutex);
        read_slot = write_slot;
        write_slot = (write_slot + 1) % 2;
    }

    // Wait until no readers in the write slot
    std::unique_lock<std::mutex> lock(slot_mutex[write_slot]);
    cv[write_slot].wait(lock, []{ return reader_count[write_slot] == 0;} );

    callback(write_slot);  // USE EXCLUSIVELY write_slot
}


// >1 reader (and no producer) can call callback below
void read(std::function<void(int index)>& callback)
{
    // Busy wait to get the read slot
    int read_slot_copy;
    while (1)
    {
        {
            std::lock_guard<std::mutex> guard(global_mutex);
            read_slot_copy = read_slot;
        }

        std::unique_lock<std::mutex> lock(slot_mutex[read_slot_copy], std::defer_lock);
        bool slot_acquired = lock.try_lock();
        if (slot_acquired)
        {
            ++reader_count[read_slot_copy];
            break;
        }
        else
        {            
            std::this_thread::yield();
        }
    }

    callback(read_slot_copy);  // SHARE read_slot_copy WITH OTHER CONSUMERS

    --reader_count[read_slot_copy];

    cv[read_slot_copy].notify_one();  // Should really notify only when read_slot_copy = 0...
}


int main()
{
    // 6 secs to take a picture
    std::function<void(int)> producerFct = [](int index) { /* use index */; sleep(6); };

    // Producer
    std::thread producerThread([&producerFct] { while (1) produce(producerFct); });

    // 50 msec to send the photo
    std::function<void(int)> readerFct = [](int index) { /* use index */; usleep(50000); };

    const int NUMBER_READERS = 10;
    const int NUMBER_READS_PER_READER = 1000;
    std::thread readers[NUMBER_READERS];

    for (int i = 0; i < NUMBER_READERS; ++i)
    {
        readers[i] = std::thread([&readerFct, i]
                        { int reads = NUMBER_READS_PER_READER;
                        while (--reads) read(readerFct); });
    }

    for (int i = 0; i < NUMBER_READERS; ++i) { readers[i].join(); }

    exit(0);  // On purpose quick exit
}
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.