Filling and saving shared buffer between threads

Question

I'm working with an API that retrieves I/Q data. Calling the function bbGetIQ(m_handle, &pkt);fills a buffer. This is a thread looping while the user hasn't input "stop". Pkt is a structure and the buffer used is pkt.iqData = &m_buffer[0]; which is a vector of float. The size of the vector is 5000 and each time we're looping the buffer is filled with 5000 values.

I want to save the data from the buffer into a file, and I was doing it right after a call to bbgetIQ but doing like so is a time consuming task, data wasn't retrieved fast enough resulting in the API dropping data so it can continue filling its buffer.

Here's what my code looked like :


void Acquisition::recordIQ(){

    int cpt = 0;
    ofstream myfile;


    while(1){

        while (keep_running)
        {   

            cpt++;

            if(cpt < 2)
                myfile.open ("/media/ssd/IQ_Data.txt");


            bbGetIQ(m_handle, &pkt); //Retrieve I/Q data


            //Writing content of buffer into the file.
            for(int i=0; i<m_buffer.size(); i++)
                myfile << m_buffer[i] << endl;


        }
        cpt = 0;
        myfile.close();
    }
}

Then i tried to only write into the file when we leave the loop :



void Acquisition::recordIQ(){

    int cpt = 0;
    ofstream myfile;
    int next=0;
    vector<float> data;


    while(1){

        while ( keep_running)
        {   
            if(keep_running == false){

                myfile.open ("/media/ssd/IQ_Data.txt");

                for(int i=0; i<data.size(); i++)
                    myfile << data[i] << endl;

                myfile.close();
                break;
            }

            cpt++;

            data.resize(next + m_buffer.size());

            bbGetIQ(m_handle, &pkt); //retrieve data

            std::copy(m_buffer.begin(), m_buffer.end(), data.begin() + next); //copy content of the buffer into final vector

            next += m_buffer.size(); //next index

        }

        cpt = 0;

    }
}

I am no longer getting data loss from the API, but the issue is that i'm limited by the size of data vector. For example, I can't let it retrieve data all night.

My idea is to make 2 threads. One will retrieve data and the other will write the data into a file. The 2 threads will share a circular buffer where the first thread will fill the buffer and the second thread will read the buffer and write the content to a file. As it is a shared buffer, i guess i should use mutexes.

I'm new to multi-threading and mutex, so would this be a good idea? I don't really know where to start and how the consumer thread can read the buffer while the producer will fill it. Will locking the buffer while reading cause data drop by the API ? (because it won't be able to write it into the circular buffer).

EDIT : As i want my record thread to run in background so i can do other stuff while it's recording, i detached it and the user can launch a record by setting the condition keep_running to true.


thread t1(&Acquisition::recordIQ, &acq);
t1.detach();

Radosław Cybulski Radosław Cybulski · Accepted Answer · 2019-06-18T14:28:32

You need to use something like this (https://en.cppreference.com/w/cpp/thread/condition_variable):

globals:

std::mutex m;
std::condition_variable cv;
std::vector<std::vector<float>> datas;
bool keep_running = true, start_running = false;

writing thread:

void writing_thread()
{
    myfile.open ("/media/ssd/IQ_Data.txt");

    while(1) {
        // Wait until main() sends data
        std::unique_lock<std::mutex> lk(m);
        cv.wait(lk, []{return keep_running && !datas.empty();});
        if (!keep_running) break;

        auto d = std::move(datas); 
        lk.unlock();

        for(auto &entry : d) {
            for(auto &e : entry)
                myfile << e << endl;             
        }
    }
}

sending thread:

void sending_thread() {
    while(1) {
        {
            std::unique_lock<std::mutex> lk(m);
            cv.wait(lk, []{return keep_running && start_running;});
            if (!keep_running) break;
        }

        bbGetIQ(m_handle, &pkt); //retrieve data

        std::vector<float> d = m_buffer;

        {
            std::lock_guard<std::mutex> lk(m);
            if (!keep_running) break;
            datas.push_back(std::move(d));
        }
        cv.notify_one();
    }
}
void start() {
    {
        std::unique_lock<std::mutex> lk(m);
        start_running = true;
    }
    cv.notify_all();
}
void stop() {
    {
        std::unique_lock<std::mutex> lk(m);
        start_running = false;
    }
    cv.notify_all();
}
void terminate() {
    {
        std::unique_lock<std::mutex> lk(m);
        keep_running = false;
    }
    cv.notify_all();

    thread1.join();
    thread2.join();
}

In short: Sending thread receives data from whatever it comes, locks mutex mt and moves data to datas storage. Then it uses cv condition variable to notify waiting threads, that there's something to do. Writing thread waits for condition variable to be signaled, then locks mutex mt, moves data from datas global variable to local, then releases mutex and proceed to write just received data to file. Key is to keep mutexed locked for least time possible.

EDIT: to terminate whole thing you need to set keep_running to false. Then call cv.notify_all(). Then join threads involved. Order is important. You need to join threads, because writing thread might be still in process of writing data.

EDIT2: added delayed start. Now create two threads, in one run sending_thread, in other writing_thread. Call start() to enable processing and stop() to stop it.

Filling and saving shared buffer between threads

1 Answers