Argthtjtr

Correct Code Link: https://wandbox.org/permlink/JYr2XoaSxsS1QT14

I've been stuck for days because of this.. I tried to make a producer->ring buffer->consumer model. At first I used a mutex to make it, it works but it's not asynchronous. I want the consumer keeps reading without any stops, just like a video player, which I think I can't accomplish with a mutex.

Here's what I've done:

1. I make a FIFO structure based on fixed size dynamic array. (ring buffer)


2. I have a pair of pointers for both push_right and pop_left operations. So there won't have data race problems, if I understand correctly.


3. I make the producer to write several items ahead, then consumer starts to read and need to ensure that:
   
   
   
   
   • consumer reads speed <= producer writes speed （consumer read pointer < producer write pointer ）
   
   
   • A appropriate fixed array size, so that the producer won't override items that consumer haven't read.
   
   
   
   

My problem is that the output result is not as I expected, it's not synchronized.. And I have no idea how to debug this.

You can see the data write order (P ostream:) is not the same as the read order (O ostream:).

One possible output:

data size: 20

70 927 156 109 834 26 883 576 226 500 904 777 935 80 346 559 846 879 548 791 

********************

Consumer start working

791

548

879

846

26

346

109

156

927

70

500

226

576

883

26

834

109

156

927

70

Consumer done

********************

p_count: 20

c_count: 20

P ostream: 791 548 879 846 559 346 80 935 777 904 500 226 576 883 26 834 109 156 927 70 

C ostream: 791 548 879 846 26 346 109 156 927 70 500 226 576 883 26 834 109 156 927 70 

Code:

(It contains main.cpp, circular_array and a data txt file, so I
think visit the link would be less pain)

https://wandbox.org/permlink/ddQjNFdxABrjminQ

main.cpp

#include <iostream>

#include <thread>

#include <chrono>

#include <vector>

#include <fstream>

#include <sstream>

#include "circular_array.h"

using namespace ythlearn;

using namespace std;



int p_count = 0;

int c_count = 0;

int fileSize = 0;

ostringstream p_os, c_os;



void producer(CircularArray<int>* Ca, vector<int> &data){

    for(int i = 0; i < 5; i++){

        p_os << data.back() << " ";

        Ca->push_right(data.back());

        data.pop_back();

        p_count++;

    }

    while(!data.empty()){

        this_thread::sleep_for(chrono::seconds(1));    

        p_os << data.back() << " ";

        Ca->push_right(data.back());

        data.pop_back();

        p_count++;

    }



}



void consumer(CircularArray<int>* Ca){

    cout << "********************" << endl;

    cout << "Consumer start working" << endl;

    this_thread::sleep_for(chrono::seconds(5));

    while(c_count < fileSize){

        this_thread::sleep_for(chrono::seconds(1));    

        int re = Ca->pop_left();

        cout << re << endl;

        c_os << re << " ";

        c_count++;



    }

    cout << "Consumer done" << endl;

    cout << "********************" << endl;



}





void getInput(vector<int>& data){

    ifstream ifs("test.txt");

    int j;

    while(ifs >> j){

        data.push_back(j);

    }

}



int main(){

    cout << unitbuf;

    vector<int> data;

    getInput(data);



    CircularArray<int> Ca;



    ::fileSize = data.size();

    cout << "data size: " << ::fileSize << endl;

    for(const auto& s: data){

        cout << s << " ";

    }

    cout << endl;





    thread th_producer(producer, &Ca, std::ref(data));

    thread th_consumer(consumer, &Ca);



    th_consumer.join();

    th_producer.join();

    cout << "p_count: " << p_count << endl

         << "c_count: " << c_count << endl;

    cout << "P ostream: " << p_os.str() << endl;

    cout << "C ostream: " << c_os.str() << endl;

    return 0;

}

circular_array.h

#pragma once

#include <stdexcept>

#include <iostream>

namespace ythlearn{

    template<typename T>

    class CircularArray{

        public:

            CircularArray(int N = 10){

                head = tail = new T[N];

                past_end_ptr1 = past_end_ptr2 = head + N;

                start_ptr1 = start_ptr2 = head;

                _capacity = N;

                _size = 0;

            }



            void push_right(T elem){

                *tail = elem;

                if(tail + 1 == past_end_ptr1){

                    tail = start_ptr1;

                }else{

                    tail++;

                }

            }



            T pop_left(){

                T re = *head;

                if(head + 1 == past_end_ptr2){

                    head = start_ptr2;

                }else{

                    head++;

                }

                return re;

            }



            CircularArray& operator=(const CircularArray&) = delete;

            CircularArray(const CircularArray&) = delete;

            ~CircularArray(){

                delete start_ptr1;

            }



        private:

            T* head;

            T* tail;

            T* start_ptr1, *start_ptr2;

            T* past_end_ptr1, *past_end_ptr2;

            int _capacity;

            int _size;

    };

}

test.txt

70

927

156

109

834

26

883

576

226

500

904

777

935

80

346

559

846

879

548

791

Short version: The problem comes down to no implementation of the third thing you said you had done ("consumer read pointer < producer write pointer" and "the producer won't override items that consumer haven't read"). In particular, not checking for overwrite is being problematic. So good plan, not so good execution.

More details: You never check for the circular array being full. Since your test case lives on the border between the array being full and the array being overfull, you end up with a race condition. Sometimes the producer will overwrite the beginning of the array before the consumer reads it.

Here's a timeline (measured in seconds):

0: Producer writes 5 values to the array.
1: Producer writes a 6th value to the array.
2: Producer writes a 7th value to the array.
3: Producer writes an 8th value to the array.
4: Producer writes a 9th value to the array.
5: Producer writes a 10th value to the array (it's now at capacity), and Consumer starts its loop (but does not get far since the first step in each iteration is sleeping for a second).

5+n: Producer writes a value to the array, and Consumer reads a value from the array. If Producer goes first, the array's size temporarily goes up to 11, exceeding its capacity.

Look at the places where Consumer reads something different than what Producer wrote. Compare what Consumer read to what Producer wrote 10 steps later.

Now for the unasked-for general critique.

There are two aspects of your implementation of a circular array that look odd/wrong. First, there is the storage duplication, where you maintain two identical start pointers and two identical past-end pointers. Second, it looks like _size is always 0, which seems not helpful.

One aspect that is not necessarily wrong but that might be improvable is how you specify N. Have you considered making N a template parameter, similar to what was done for std::array? That could reduce your memory footprint (no need to store _capacity) and remove the need for dynamic memory management.

Addendum:

It has occurred to me that you have data members that should not be changed after construction, but that are not flagged const. You might want to address that, especially since flagging them const will make it clear that they cannot be involved in a race condition. So you could declare the circular array's data members more like:

private:

    int const _capacity;

    T * const start;

    T * const past_end;

    T* head;

    T* tail;

Then the constructor could be more like:

CircularArray(int N = 10) :

    _capacity(N),

    start(new T[N]),

    past_end(start + N)

{

    head = tail = start;

}

(Actually, you could use the initializer list for all the members; I just thought smaller changes would be more digestible.) Another style benefit of this is that the new and delete would both be applied to start, which looks better to someone checking the code.

Another simplification might be to use the expression start + N wherever you had been using past_end. The performance difference should be negligible, and you reduce the memory footprint.

Alternatively, my earlier suggestion of making N a template parameter would render the const question moot. Using N as a template parameter (still with a default value of 10), the circular array's data members could simply be:

private:

    T start[N]; // As a template parameter, N is a compile-time constant

    T* head; // Or an index into the array

    T* tail; // Or an index into the array

Switching to indices also gives more reason to drop the past-the-end member. The past-the-end index becomes N, which is a compile time constant -- no need to waste storage space on it.