I'm building a dataflow engine in multithreaded C++ that is similar to GNU Radio but for performing audio processing instead of implementing software radios. The model is a graph of C++ objects that generate data and make it available via outputs that connect to the data inputs of another object. Once data is available on the inputs of an object it can execute and create data for more objects. Once data is made available on an output it is not modified by the generating object and data on an object input is never modified by the consuming object. The output of an object can connect to multiple inputs on other objects. For performance reasons I want to keep memory copying to a minimum.
I've implemented a C++ class that allocates memory for an audio buffer, the buffer is modified by the creating object, it is wrapped by a shared_ptr, then placed on a data output. Because the object is never modified after it is placed on a data output I did not use a mutex or atomic variables in it. This functions well aside from a problem where with certain graph topologies involving high levels of CPU usage the buffer is sometimes full of garbage when it is consumed.
I don't think the audio buffer is being corrupted after it has been created. I now suspect the problem originates from the lack of memory ordering guarantees due to the lack of a mutex or atomic variables. From the C++ reference:
Absent any constraints on a multi-core system, when multiple threads simultaneously read and write to several variables, one thread can observe the values change in an order different from the order another thread wrote them. Indeed, the apparent order of changes can even differ among multiple reader threads.
Here is the declaration of the buffer class:
struct buffer_t {
const size_t num_samples;
float *data = nullptr; // pointer to array of floats
buffer_t(const size_t num_samples_in) noexcept;
buffer_t(const buffer_t&) = delete;
buffer_t(const buffer_t&&) = delete;
~buffer_t();
const float * get_data() noexcept;
size_t get_num_samples() noexcept;
};
Edit: People were mentioning use after free bugs and noting there is no implementation. Here is the definition:
buffer_t::buffer_t(const size_t num_samples_in) noexcept
: num_samples(num_samples_in)
{
data = new float[num_samples_in];
}
buffer_t::~buffer_t() {
if (data != nullptr) {
delete [] data;
}
}
const float * buffer_t::get_data() noexcept {
assert(data != nullptr);
return data;
}
size_t buffer_t::get_num_samples() noexcept {
return num_samples;
}
I am not very experienced with concurrency in C/C++ programs. Here are my questions:
- Is it an invalid expectation that because the buffer is not modified after being shared that it can be used safely across threads with out a mutex or atomic variables?
- If my expectations were invalid what is the correct course of action?
Assuming I do need a mutex or atomic variables here are my thoughts; comments on them would be greatly appreciated.
- Though a mutex will enforce sequentially-consistent ordering across threads a std::mutex is not the right tool for the job as it would prevent parallelization that is available. A std::shared_mutex is a better fit but still seems to be the wrong thing to use.
- A std::atomic<std::shared_ptr> seems like it would work to enforce sequentially-consistent ordering but also does not seem like it is the right tool for the job as the pointer stored in the shared_ptr itself does not change.
- Making the data a pointer to an array of atomic floats is the most obvious answer but seems like it would have a heavy performance cost and would make it impossible to pass the array to functions expecting a standard array of floats which is a use case I do have.
- The correct solution then would be making the pointer to the array of floats itself atomic since dereferencing the pointer would then enforce sequentially-consistent memory ordering. The num_samples variable may also need to be made into an atomic size_t for correctness.
Thank you for helping me improve my understanding of what's going on.