Хм. Зачем все так сложно мудрить, если есть уже давно простейшая рабочая реализация прописанная в букварях по программированию, остается ее только портировать на свой язык.
При чем тут еще от задачи зависит. Если это SPSC, тогда вообще все просто и без блокировок. Если MPSC/MPMC/SPMC - нужны уже блокировки.
Приведу простейший пример SPSC, написан был мной лет 8 назад(то есть просто реализован на ц++ алгоритм из букваря) и очень часто используется по сей день везде(драйверы, очередя сообщений, итд итп)
CODE
template<class T, int Size> class Fifo_spsc{
public:
static_assert((Size&(Size-1)) == 0, "Size must be power of 2");
Fifo_spsc(){
rx = 0;
wx = 0;
}
// Producer
bool push(const T &v){
uint32_t w = (wx+1)&(Size-1);
if(w==rx)return false; //ERR_FIFO_FULL;
data[wx] = v;
__DMB(); // just a memory barrier - ensure data has been completely stored
wx = w; // make item available for reading
return true;
}
// Consumer
bool pop(T *v){
uint32_t r = rx;
if(r==wx)return false; //ERR_FIFO_EMPTY;
*v = data[r];
__DMB(); // ensure data has been completely loaded
rx = (r+1)&(Size-1); // make item available for writing
return true;
}
// Consumer
void flush(){
__DMB();
rx = wx;
}
protected:
uint16_t rx, wx;
T data[Size];
};
Памяти занимает - собственно сами данные + 2 16-битных слова.
Хоть и c++, что позволяет использовать этот буфер для любых типов - как простого uint_8t, так и для сложных обьектов. Это базовая реализация для практически всех других более сложных очередей.
Один элемент всегда пустой - это делает код гораздо проще и быстрее, и как ни странно зачастую - меньше требовательным к памяти
Также, если у нас строго один писатель и один читатель, то можно сделать и чтение не по одному элементу, а сразу массивом.
CODE
// Consumer, use with caution
const T* getConsecutiveData_ptr()const{
if(rx==wx)return 0; // empty
return &data[rx];
}
int getConsecutiveData_len()const{
if(wx>rx)return wx-rx;
else return Size-rx;
}
void popConsecutiveData(unsigned n){
__DMB();
rx = (rx+n)&(Size-1);
}
Важно понимать, что связь между producer и consumer должна быть
только через функции этого буфера, без каких либо других переменных, вызовов функций итд в обход этого FIFO. Иначе однозначно рано или поздно схватите гонку.
То есть он подходит, например когда producer - это прерывание, а consumer - вечный цикл. Что для большинства задач достаточно.
Если у нас еvent-driven-движок (то есть без вечных циклов, все на событиях) - тогда нужна более сложная реализация. Но использовать нужно опять же только ее, без всяких сторонних связей.
Если интересно - расскажу и покажу как, я на этих циклических буферах(в том числе с обьектами не фиксированного размера) уже собаку сьел за много лет, и в своих программах использую только их - как самый простой, эффективный и безопасный интерфейс для связи между обьектами.
Но это все C++, зато позволяет эффективно и безопасно работать и писать очень простой код.
Например. Прерывание uart
Код
void U0RxCbk::readyRead(LPC_UART_TypeDef *uart){ // interrupt
while(Uart::isCharAvailable(uart)){
bool r = rxfifo.push(Uart::read_char(uart));
if(!r){
printf("U0RxCbk::readyRead(): rxfifo full!\n");
}
}
}
Приемник - просто функция, которая будет вызвана, когда на это появится время(в порядке приоритета и очереди).
Которая в свою очередь принимает текст NMEA и дальше парсит его.
Код
void Gps::start_receiver(){
// Устанавливаем обработчик события
rxfifo.set_onReadyRead([this](){ // собственно сам обработчик
uint8_t ch;
while(true){
bool r = rxfifo.pop(&ch); // читаем из очереди
parse(ch); // парсим
if(!r)break; // если очередь стала пустой - выходим
}
}, GPS_TASK_PRIORITY); // приоритет обработчика
}
Это более старый вариант, привел, чтобы понять как оно работает, скажем так, внутри.
Сейчас я пользуюсь в основном таким:
Код
class GpsReceiver : public UmFifoSingleReceiver<uint8_t, GpsReceiver, GPS_TASK_PRIORITY>{
protected:
void onItemReceived(uint8_t ch){
parse(ch);
}
};
То есть вся логика чтения из буфера реализована один раз отдельно и используется повторно сколько угодно раз.
Так же есть аналогичные реализации, когда приемник может обрабатывать не один элемент - а массив из нескольких(например отправка через DMA), физически - подряд лежащих в нашем ring-buffere.
Overhed-a нет, наоборот такая реализация на шаблонах работает быстрее, чем обычные сишные
Большинство работы делает компилятор, там даже вызова функции не будет - компилятор функцию parse встроит в недра самой очереди.
Проблема с Ring Buffer.
Aug 31 2016, 08:46
