yigid/openairinterface5g
1
0
Fork 0
mirror of https://gitlab.eurecom.fr/oai/openairinterface5g.git synced 2026-07-13 04:30:28 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(zmq): templatize ring_buffer and overflow_buffer

Browse Source 
Support float and int16 complex types

Signed-off-by: Bartosz Podrygajlo <bartosz.podrygajlo@openairinterface.org>
This commit is contained in:
Bartosz Podrygajlo
2026-06-30 15:31:04 +02:00
parent 75545fa2c1
commit 5fad2465f0
3 changed files with 51 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

62
radio/zmq/ring_buffer.cpp
View File

@@ -7,12 +7,14 @@
#include <iostream>
#include <algorithm>
ring_buffer::ring_buffer(size_t max_size) : max_size_(max_size)
template <typename T>
ring_buffer<T>::ring_buffer(size_t max_size) : max_size_(max_size)
{
buffer_ = std::make_unique<cf_t[]>(max_size);
buffer_ = std::make_unique<T[]>(max_size);
}
size_t ring_buffer::push_samples(const cf_t *samples, const size_t nsamps)
template <typename T>
size_t ring_buffer<T>::push_samples(const T *samples, const size_t nsamps)
{
size_t overflow = 0;
// if nsamps > max_size skip nsamps - max_size samples
@@ -31,12 +33,12 @@ size_t ring_buffer::push_samples(const cf_t *samples, const size_t nsamps)
}
size_t first_chunk = std::min(nsamps_left, max_size_ - head_);
memcpy(&buffer_[head_], samples, first_chunk * sizeof(cf_t));
memcpy(&buffer_[head_], samples, first_chunk * sizeof(T));
head_ = (head_ + first_chunk) % max_size_;
samples += first_chunk;
nsamps_left -= first_chunk;
if (nsamps_left > 0) {
memcpy(&buffer_[0], samples, nsamps_left * sizeof(cf_t));
memcpy(&buffer_[0], samples, nsamps_left * sizeof(T));
head_ = nsamps_left;
}
@@ -45,7 +47,8 @@ size_t ring_buffer::push_samples(const cf_t *samples, const size_t nsamps)
return overflow;
}
size_t ring_buffer::push_zeros(const size_t num_zeros)
template <typename T>
size_t ring_buffer<T>::push_zeros(const size_t num_zeros)
{
size_t overflow = 0;
// if nsamps > max_size skip nsamps - max_size samples
@@ -63,11 +66,11 @@ size_t ring_buffer::push_zeros(const size_t num_zeros)
}
size_t first_chunk = std::min(nsamps_left, max_size_ - head_);
memset(&buffer_[head_], 0, first_chunk * sizeof(cf_t));
memset(&buffer_[head_], 0, first_chunk * sizeof(T));
head_ = (head_ + first_chunk) % max_size_;
nsamps_left -= first_chunk;
if (nsamps_left > 0) {
memset(&buffer_[0], 0, nsamps_left * sizeof(cf_t));
memset(&buffer_[0], 0, nsamps_left * sizeof(T));
head_ = nsamps_left;
}
@@ -76,16 +79,17 @@ size_t ring_buffer::push_zeros(const size_t num_zeros)
return overflow;
}
size_t ring_buffer::pop_samples(cf_t *samples, size_t num_samples)
template <typename T>
size_t ring_buffer<T>::pop_samples(T *samples, size_t num_samples)
{
size_t samples_to_pop = std::min(size_, num_samples);
if (samples_to_pop > 0) {
if (tail_ + samples_to_pop > max_size_) {
size_t first_chunk = max_size_ - tail_;
memcpy(samples, &buffer_[tail_], first_chunk * sizeof(cf_t));
memcpy(samples + first_chunk, &buffer_[0], (samples_to_pop - first_chunk) * sizeof(cf_t));
memcpy(samples, &buffer_[tail_], first_chunk * sizeof(T));
memcpy(samples + first_chunk, &buffer_[0], (samples_to_pop - first_chunk) * sizeof(T));
} else {
memcpy(samples, &buffer_[tail_], samples_to_pop * sizeof(cf_t));
memcpy(samples, &buffer_[tail_], samples_to_pop * sizeof(T));
}
tail_ = (tail_ + samples_to_pop) % max_size_;
size_ -= samples_to_pop;
@@ -94,24 +98,28 @@ size_t ring_buffer::pop_samples(cf_t *samples, size_t num_samples)
return 0;
}
void ring_buffer::clear_samples()
template <typename T>
void ring_buffer<T>::clear_samples()
{
head_ = 0;
tail_ = 0;
size_ = 0;
}
void ring_buffer::reset()
template <typename T>
void ring_buffer<T>::reset()
{
clear_samples();
}
size_t ring_buffer::size() const
template <typename T>
size_t ring_buffer<T>::size() const
{
return size_;
}
size_t overflow_buffer::push_samples(const cf_t *samples, size_t nsamps)
template <typename T>
size_t overflow_buffer<T>::push_samples(const T *samples, size_t nsamps)
{
std::lock_guard<std::mutex> lock(mutex_);
size_t overflow = buffer_.push_samples(samples, nsamps);
@@ -119,7 +127,8 @@ size_t overflow_buffer::push_samples(const cf_t *samples, size_t nsamps)
return overflow;
}
size_t overflow_buffer::push_zeros(size_t num_zeros)
template <typename T>
size_t overflow_buffer<T>::push_zeros(size_t num_zeros)
{
std::lock_guard<std::mutex> lock(mutex_);
size_t overflow = buffer_.push_zeros(num_zeros);
@@ -127,13 +136,14 @@ size_t overflow_buffer::push_zeros(size_t num_zeros)
return overflow;
}
size_t overflow_buffer::pop_samples(cf_t *samples, size_t num_samples)
template <typename T>
size_t overflow_buffer<T>::pop_samples(T *samples, size_t num_samples)
{
std::lock_guard<std::mutex> lock(mutex_);
size_t samples_popped = 0;
if (zeros_to_send_ > 0) {
size_t num_zeros = std::min(zeros_to_send_, num_samples);
memset(samples, 0, num_zeros * sizeof(cf_t));
memset(samples, 0, num_zeros * sizeof(T));
zeros_to_send_ -= num_zeros;
samples += num_zeros;
num_samples -= num_zeros;
@@ -146,22 +156,30 @@ size_t overflow_buffer::pop_samples(cf_t *samples, size_t num_samples)
return samples_popped;
}
void overflow_buffer::reset()
template <typename T>
void overflow_buffer<T>::reset()
{
std::lock_guard<std::mutex> lock(mutex_);
buffer_.reset();
zeros_to_send_ = buffer_.size() / 2;
}
void overflow_buffer::clear_samples()
template <typename T>
void overflow_buffer<T>::clear_samples()
{
std::lock_guard<std::mutex> lock(mutex_);
buffer_.clear_samples();
zeros_to_send_ = 0;
}
size_t overflow_buffer::size()
template <typename T>
size_t overflow_buffer<T>::size()
{
std::lock_guard<std::mutex> lock(mutex_);
return buffer_.size() + zeros_to_send_;
}
template class ring_buffer<cf_t>;
template class ring_buffer<c16_t>;
template class overflow_buffer<cf_t>;
template class overflow_buffer<c16_t>;

16
radio/zmq/ring_buffer.h
View File

@@ -8,10 +8,11 @@
#include <memory>
#include "common/platform_types.h"
// A basic cirular sample buffer class
// A basic circular sample buffer class
template <typename T = cf_t>
class ring_buffer {
private:
std::unique_ptr<cf_t[]> buffer_;
std::unique_ptr<T[]> buffer_;
size_t head_ = 0;
size_t tail_ = 0;
size_t size_ = 0;
@@ -19,17 +20,18 @@ class ring_buffer {
public:
ring_buffer(size_t max_size = 614400);
size_t push_samples(const cf_t *samples, size_t nsamps);
size_t push_samples(const T *samples, size_t nsamps);
size_t push_zeros(size_t num_zeros);
size_t pop_samples(cf_t *samples, size_t num_samples);
size_t pop_samples(T *samples, size_t num_samples);
void reset();
void clear_samples();
size_t size() const;
};
// A thread-safe wrapper around ring_buffer that counts overflows
template <typename T = cf_t>
class overflow_buffer {
ring_buffer buffer_;
ring_buffer<T> buffer_;
size_t zeros_to_send_;
std::mutex mutex_;
@@ -37,9 +39,9 @@ class overflow_buffer {
overflow_buffer(size_t max_size = 614400) : buffer_(max_size), zeros_to_send_(0)
{
}
size_t push_samples(const cf_t *samples, size_t nsamps);
size_t push_samples(const T *samples, size_t nsamps);
size_t push_zeros(size_t num_zeros);
size_t pop_samples(cf_t *samples, size_t num_samples);
size_t pop_samples(T *samples, size_t num_samples);
void reset();
void clear_samples();
size_t size();

4
radio/zmq/zmq_imported.h
View File

@@ -17,7 +17,7 @@
class zmq_tx_channel {
public:
void *socket_;
overflow_buffer buffer_;
overflow_buffer<cf_t> buffer_;
std::atomic<uint64_t> sample_count_ = 0;
std::atomic<bool> is_tx_enabled_ = false;
std::mutex transmit_alignment_mutex_;
@@ -37,7 +37,7 @@ class zmq_tx_channel {
class zmq_rx_channel {
public:
void *socket_;
overflow_buffer buffer_;
overflow_buffer<cf_t> buffer_;
bool request_sent_;
std::atomic<bool> stopped_;
zmq_rx_channel(void *s, uint64_t buffer_size) : socket_(s), buffer_(buffer_size), stopped_(false)