¿Cómo se implementa un buffer circular en C?

Question

Necesito un buffer circular de tamaño fijo (seleccionable en tiempo de ejecución al crearlo, no en tiempo de compilación) que puede contener objetos de cualquier tipo y necesita ser muy de alto rendimiento. No creo que haya problemas de contención de recursos ya que, aunque se trata de un entorno integrado multitarea, es cooperativo, por lo que las tareas pueden lograrlo.

Mi idea inicial era almacenar una estructura simple en el búfer que contendría el tipo (enum/definición simple) y un puntero de vacío a la carga, pero quiero que sea lo más rápido posible, así que estoy abierto a sugerencias que implican eludir el montón.

En realidad, me alegro de omitir cualquiera de las bibliotecas estándar para velocidad bruta: por lo que he visto del código, no está muy optimizado para la CPU: parece que compiló código C para cosas como strcpy() y tal, no hay ensamblado codificado a mano.

Cualquier código o idea sería muy apreciada. Las operaciones requeridas son:

• crean un búfer con un tamaño específico.
• poner en la cola.
• obtener de la cabeza.
• devuelve el conteo.
• eliminar un búfer.

Answer 1

¿Puede enumerar los tipos necesarios en ¿el momento en que codifica el búfer o necesita poder agregar tipos en tiempo de ejecución a través de llamadas dinámicas? Si lo primero, crearía el búfer como una matriz de n estructuras asignadas en el montón, donde cada estructura consta de dos elementos: una etiqueta enum que identifica el tipo de datos y una unión de todos los tipos de datos. Lo que pierde en términos de almacenamiento adicional para elementos pequeños, lo compensa en términos de no tener que lidiar con la asignación/desasignación y la fragmentación de memoria resultante. Entonces solo necesita hacer un seguimiento de los índices de inicio y fin que definen los elementos de cabecera y cola del búfer, y asegurarse de calcular el mod n al incrementar/disminuir los índices.

Answer 2

La solución más sencilla sería la de no perder de vista el tamaño del artículo y el número de artículos, y luego crear una memoria intermedia del número adecuado de bytes:

typedef struct circular_buffer 
{ 
    void *buffer;  // data buffer 
    void *buffer_end; // end of data buffer 
    size_t capacity; // maximum number of items in the buffer 
    size_t count;  // number of items in the buffer 
    size_t sz;  // size of each item in the buffer 
    void *head;  // pointer to head 
    void *tail;  // pointer to tail 
} circular_buffer; 

void cb_init(circular_buffer *cb, size_t capacity, size_t sz) 
{ 
    cb->buffer = malloc(capacity * sz); 
    if(cb->buffer == NULL) 
     // handle error 
    cb->buffer_end = (char *)cb->buffer + capacity * sz; 
    cb->capacity = capacity; 
    cb->count = 0; 
    cb->sz = sz; 
    cb->head = cb->buffer; 
    cb->tail = cb->buffer; 
} 

void cb_free(circular_buffer *cb) 
{ 
    free(cb->buffer); 
    // clear out other fields too, just to be safe 
} 

void cb_push_back(circular_buffer *cb, const void *item) 
{ 
    if(cb->count == cb->capacity){ 
     // handle error 
    } 
    memcpy(cb->head, item, cb->sz); 
    cb->head = (char*)cb->head + cb->sz; 
    if(cb->head == cb->buffer_end) 
     cb->head = cb->buffer; 
    cb->count++; 
} 

void cb_pop_front(circular_buffer *cb, void *item) 
{ 
    if(cb->count == 0){ 
     // handle error 
    } 
    memcpy(item, cb->tail, cb->sz); 
    cb->tail = (char*)cb->tail + cb->sz; 
    if(cb->tail == cb->buffer_end) 
     cb->tail = cb->buffer; 
    cb->count--; 
} 

Answer 3

Una aplicación sencilla podría consistir en:

• Un buffer, implementado como una matriz de tamaño n, de cualquier tipo que necesita
• Un puntero o índice (el que sea más eficiente para su procesador) leer
• Un puntero de escritura o el índice de
• Un contador que indica la cantidad de datos en el búfer (derivable de la lectura y escritura de los punteros, pero más rápido para hacer un seguimiento por separado)

Cada vez que escribe datos, avanza el puntero de escritura e incrementa el contador. Cuando lee datos, aumenta el puntero de lectura y disminuye el contador. Si cualquiera de los punteros llega a n, establézcalo en cero.

No se puede escribir si counter = n. No puedes leer si counter = 0.

Answer 4

// Note power of two buffer size 
#define kNumPointsInMyBuffer 1024 

typedef struct _ringBuffer { 
    UInt32 currentIndex; 
    UInt32 sizeOfBuffer; 
    double data[kNumPointsInMyBuffer]; 
} ringBuffer; 

// Initialize the ring buffer 
ringBuffer *myRingBuffer = (ringBuffer *)calloc(1, sizeof(ringBuffer)); 
myRingBuffer->sizeOfBuffer = kNumPointsInMyBuffer; 
myRingBuffer->currentIndex = 0; 

// A little function to write into the buffer 
// N.B. First argument of writeIntoBuffer() just happens to have the 
// same as the one calloc'ed above. It will only point to the same 
// space in memory if the calloc'ed pointer is passed to 
// writeIntoBuffer() as an arg when the function is called. Consider 
// using another name for clarity 
void writeIntoBuffer(ringBuffer *myRingBuffer, double *myData, int numsamples) { 
    // -1 for our binary modulo in a moment 
    int buffLen = myRingBuffer->sizeOfBuffer - 1; 
    int lastWrittenSample = myRingBuffer->currentIndex; 

    int idx; 
    for (int i=0; i < numsamples; ++i) { 
     // modulo will automagically wrap around our index 
     idx = (i + lastWrittenSample) & buffLen; 
     myRingBuffer->data[idx] = myData[i]; 
    } 

    // Update the current index of our ring buffer. 
    myRingBuffer->currentIndex += numsamples; 
    myRingBuffer->currentIndex &= myRingBuffer->sizeOfBuffer - 1; 
} 

Mientras la longitud de su memoria cíclica es una potencia de dos, el increíblemente rápido binaria "&" operación envolverá alrededor de su índice para usted. Para mi aplicación, estoy mostrando un segmento de audio al usuario desde un buffer de anillo de audio adquirido desde un micrófono.

Siempre me aseguro de que la cantidad máxima de audio que se puede mostrar en la pantalla sea mucho menor que el tamaño del buffer de anillo. De lo contrario, podría estar leyendo y escribiendo desde el mismo fragmento. Esto probablemente te dará extraños artefactos de visualización.

Answer 5

Primero, el título. No necesita aritmética de módulo para envolver el búfer si usa bits de entrada para mantener los "indicadores" de cola & de cola, y dimensionarlos para que estén perfectamente sincronizados. IE: 4096 metido en una int sin firmar de 12 bits es 0 por sí mismo, sin que lo molesten de ninguna manera. La eliminación de la aritmética de módulo, incluso para potencias de 2, duplica la velocidad, casi exactamente.

10 millones de iteraciones de llenar y drenar un búfer 4096 de cualquier tipo de elementos de datos demoran 52 segundos en mi 3rd Gen i7 Dell XPS 8500 utilizando el compilador C++ de Visual Studio 2010 con alineación predeterminada y 1/8192nd de eso para dar servicio a un dato.

Rx volvería a escribir los bucles de prueba en main() para que ya no controlen el flujo, que es, y debería ser, controlado por los valores de retorno que indican que el búfer está lleno o vacío, y la interrupción de asistente; declaraciones. IE: el llenador y el escurridor deberían poder chocar entre sí sin corrupción ni inestabilidad. En algún momento espero enhebrar este código, por lo que ese comportamiento será crucial.

El QUEUE_DESC (descriptor de cola) y la función de inicialización fuerza a todos los almacenamientos intermedios en este código a una potencia de 2. El esquema anterior NO funcionará de otra manera. Mientras habla del tema, tenga en cuenta que QUEUE_DESC no está codificado, sino que utiliza una constante de manifiesto (#define BITS_ELE_KNT) para su construcción. (Supongo que una potencia de 2 es suficiente flexibilidad aquí)

Para hacer que el tamaño del buffer sea seleccionable en tiempo de ejecución, intenté diferentes enfoques (no se muestran aquí), y decidí usar USHRT para Head, Tail, EleKnt con capacidad de administrar un búfer FIFO [USHRT]. Para evitar la aritmética de módulo, creé una máscara a & & con Cabeza, Cola, pero esa máscara resultó ser (EleKnt -1), así que solo use eso. El uso de USHRTS en lugar de bits incrementó el rendimiento ~ 15% en una máquina silenciosa. Los núcleos de CPU de Intel siempre han sido más rápidos que sus buses, por lo que en una máquina compartida y ocupada, empacar sus estructuras de datos lo hace cargar y ejecutar por delante de otros hilos de la competencia. Compensaciones.

Tenga en cuenta que el almacenamiento real para el búfer se asigna en el montón con calloc(), y el puntero está en la base de la estructura, por lo que la estructura y el puntero tienen EXACTAMENTE la misma dirección. ES DECIR; no es necesario agregar ningún desplazamiento a la dirección de la estructura para unir los registros.

En la misma línea, todas las variables relacionadas con el mantenimiento del búfer son físicamente adyacentes al búfer, vinculadas a la misma estructura, por lo que el compilador puede crear un lenguaje de ensamblaje hermoso. Tendrás que eliminar la optimización en línea para ver cualquier ensamblaje, porque de lo contrario se aplastará en el olvido.

Para admitir el polimorfismo de cualquier tipo de datos, he usado memcpy() en lugar de asignaciones. Si solo necesita la flexibilidad para admitir un tipo de variable aleatoria por compilación, este código funciona perfectamente.

Para el polimorfismo, solo necesita saber el tipo y sus requisitos de almacenamiento. La matriz de descriptores DATA_DESC proporciona una forma de realizar un seguimiento de cada dato que se coloca en QUEUE_DESC.pBuffer para que pueda ser recuperado adecuadamente. Asignaría suficiente memoria pBuffer para contener todos los elementos del tipo de datos más grande, pero mantendría un registro de la cantidad de ese almacenamiento que un dato dado está usando realmente en DATA_DESC.dBytes. La alternativa es reinventar un administrador de montón.

Esto significa que el UCHAR * pBuffer de QUEUE_DESC tendría una matriz complementaria paralela para realizar un seguimiento del tipo de datos y el tamaño, mientras que la ubicación de almacenamiento de un datum en pBuffer se mantendría tal como está ahora. El nuevo miembro sería algo así como DATA_DESC * pDataDesc, o, tal vez, DATA_DESC DataDesc [2^BITS_ELE_KNT] si puede encontrar una manera de vencer a su compilador con esta referencia directa. Calloc() siempre es más flexible en estas situaciones.

Aún tendría memcpy() en Q_Put(), Q_Get, pero la cantidad de bytes realmente copiados estaría determinada por DATA_DESC.dBytes, no QUEUE_DESC.EleBytes. Los elementos son potencialmente de diferentes tipos/tamaños para cualquier put o get dado.

Creo que este código cumple con los requisitos de velocidad y tamaño de búfer, y se puede hacer para satisfacer el requisito de 6 tipos de datos diferentes. He dejado los muchos accesorios de prueba, en forma de instrucciones printf(), para que pueda asegurarse (o no) de que el código funciona correctamente. El generador de números aleatorios demuestra que el código funciona para cualquier combinación de cabeza/cola aleatoria.

enter code here 
// Queue_Small.cpp : Defines the entry point for the console application. 
// 
#include "stdafx.h" 
#include <stdio.h> 
#include <time.h> 
#include <limits.h> 
#include <stdlib.h> 
#include <malloc.h> 
#include <memory.h> 
#include <math.h> 

#define UCHAR unsigned char 
#define ULONG unsigned long 
#define USHRT unsigned short 
#define dbl double 
/* Queue structure */ 
#define QUEUE_FULL_FLAG 1 
#define QUEUE_EMPTY_FLAG -1 
#define QUEUE_OK 0 
// 
#define BITS_ELE_KNT 12 //12 bits will create 4.096 elements numbered 0-4095 
// 
//typedef struct { 
// USHRT dBytes:8;  //amount of QUEUE_DESC.EleBytes storage used by datatype 
// USHRT dType :3; //supports 8 possible data types (0-7) 
// USHRT dFoo :5; //unused bits of the unsigned short host's storage 
// } DATA_DESC; 
// This descriptor gives a home to all the housekeeping variables 
typedef struct { 
    UCHAR *pBuffer; // pointer to storage, 16 to 4096 elements 
    ULONG Tail :BITS_ELE_KNT; // # elements, with range of 0-4095 
    ULONG Head :BITS_ELE_KNT; // # elements, with range of 0-4095 
    ULONG EleBytes :8;  // sizeof(elements) with range of 0-256 bytes 
    // some unused bits will be left over if BITS_ELE_KNT < 12 
    USHRT EleKnt :BITS_ELE_KNT +1;// 1 extra bit for # elements (1-4096) 
    //USHRT Flags :(8*sizeof(USHRT) - BITS_ELE_KNT +1); // flags you can use 
    USHRT IsFull :1;  // queue is full 
    USHRT IsEmpty :1;  // queue is empty 
    USHRT Unused :1;  // 16th bit of USHRT 
} QUEUE_DESC; 

// --------------------------------------------------------------------------- 
// Function prototypes 
QUEUE_DESC *Q_Init(QUEUE_DESC *Q, int BitsForEleKnt, int DataTypeSz); 
int Q_Put(QUEUE_DESC *Q, UCHAR *pNew); 
int Q_Get(UCHAR *pOld, QUEUE_DESC *Q); 
// --------------------------------------------------------------------------- 
QUEUE_DESC *Q_Init(QUEUE_DESC *Q, int BitsForEleKnt, int DataTypeSz) { 
    memset((void *)Q, 0, sizeof(QUEUE_DESC));//init flags and bit integers to zero 
    //select buffer size from powers of 2 to receive modulo 
    //    arithmetic benefit of bit uints overflowing 
    Q->EleKnt = (USHRT)pow(2.0, BitsForEleKnt); 
    Q->EleBytes = DataTypeSz; // how much storage for each element? 
    // Randomly generated head, tail a test fixture only. 
    //  Demonstrates that the queue can be entered at a random point 
    //  and still perform properly. Normally zero 
    srand(unsigned(time(NULL))); // seed random number generator with current time 
    Q->Head = Q->Tail = rand(); // supposed to be set to zero here, or by memset 
    Q->Head = Q->Tail = 0; 
    // allocate queue's storage 
    if(NULL == (Q->pBuffer = (UCHAR *)calloc(Q->EleKnt, Q->EleBytes))) { 
     return NULL; 
    } else { 
     return Q; 
    } 
} 
// --------------------------------------------------------------------------- 
int Q_Put(QUEUE_DESC *Q, UCHAR *pNew) 
{ 
    memcpy(Q->pBuffer + (Q->Tail * Q->EleBytes), pNew, Q->EleBytes); 
    if(Q->Tail == (Q->Head + Q->EleKnt)) { 
     // Q->IsFull = 1; 
     Q->Tail += 1; 
     return QUEUE_FULL_FLAG; // queue is full 
    } 
    Q->Tail += 1; // the unsigned bit int MUST wrap around, just like modulo 
    return QUEUE_OK; // No errors 
} 
// --------------------------------------------------------------------------- 
int Q_Get(UCHAR *pOld, QUEUE_DESC *Q) 
{ 
    memcpy(pOld, Q->pBuffer + (Q->Head * Q->EleBytes), Q->EleBytes); 
    Q->Head += 1; // the bit int MUST wrap around, just like modulo 

    if(Q->Head == Q->Tail)  { 
     // Q->IsEmpty = 1; 
     return QUEUE_EMPTY_FLAG; // queue Empty - nothing to get 
    } 
    return QUEUE_OK; // No errors 
} 
// 
// --------------------------------------------------------------------------- 
int _tmain(int argc, _TCHAR* argv[]) { 
// constrain buffer size to some power of 2 to force faux modulo arithmetic 
    int LoopKnt = 1000000; // for benchmarking purposes only 
    int k, i=0, Qview=0; 
    time_t start; 
    QUEUE_DESC Queue, *Q; 
    if(NULL == (Q = Q_Init(&Queue, BITS_ELE_KNT, sizeof(int)))) { 
     printf("\nProgram failed to initialize. Aborting.\n\n"); 
     return 0; 
    } 

    start = clock(); 
    for(k=0; k<LoopKnt; k++) { 
     //printf("\n\n Fill'er up please...\n"); 
     //Q->Head = Q->Tail = rand(); 
     for(i=1; i<= Q->EleKnt; i++) { 
      Qview = i*i; 
      if(QUEUE_FULL_FLAG == Q_Put(Q, (UCHAR *)&Qview)) { 
       //printf("\nQueue is full at %i \n", i); 
       //printf("\nQueue value of %i should be %i squared", Qview, i); 
       break; 
      } 
      //printf("\nQueue value of %i should be %i squared", Qview, i); 
     } 
     // Get data from queue until completely drained (empty) 
     // 
     //printf("\n\n Step into the lab, and see what's on the slab... \n"); 
     Qview = 0; 
     for(i=1; i; i++) { 
      if(QUEUE_EMPTY_FLAG == Q_Get((UCHAR *)&Qview, Q)) { 
       //printf("\nQueue value of %i should be %i squared", Qview, i); 
       //printf("\nQueue is empty at %i", i); 
       break; 
      } 
      //printf("\nQueue value of %i should be %i squared", Qview, i); 
     } 
     //printf("\nQueue head value is %i, tail is %i\n", Q->Head, Q->Tail); 
    } 
    printf("\nQueue time was %5.3f to fill & drain %i element queue %i times \n", 
        (dbl)(clock()-start)/(dbl)CLOCKS_PER_SEC,Q->EleKnt, LoopKnt); 
    printf("\nQueue head value is %i, tail is %i\n", Q->Head, Q->Tail); 
    getchar(); 
    return 0; 
} 

Answer 6

Aquí hay una solución simple en C. Supongamos que las interrupciones se desactivan para cada función. Sin polimorfismo & cosas, solo sentido común.

---

#define BUFSIZE 128 
char buf[BUFSIZE]; 
char *pIn, *pOut, *pEnd; 
char full; 

// init 
void buf_init() 
{ 
    pIn = pOut = buf;  // init to any slot in buffer 
    pEnd = &buf[BUFSIZE]; // past last valid slot in buffer 
    full = 0;    // buffer is empty 
} 

// add char 'c' to buffer 
int buf_put(char c) 
{ 
    if (pIn == pOut && full) 
     return 0;   // buffer overrun 

    *pIn++ = c;    // insert c into buffer 
    if (pIn >= pEnd)  // end of circular buffer? 
     pIn = buf;   // wrap around 

    if (pIn == pOut)  // did we run into the output ptr? 
     full = 1;   // can't add any more data into buffer 
    return 1;    // all OK 
} 

// get a char from circular buffer 
int buf_get(char *pc) 
{ 
    if (pIn == pOut && !full) 
     return 0;   // buffer empty FAIL 

    *pc = *pOut++;    // pick up next char to be returned 
    if (pOut >= pEnd)  // end of circular buffer? 
     pOut = buf;   // wrap around 

    full = 0;    // there is at least 1 slot 
    return 1;    // *pc has the data to be returned 
} 

Answer 7

estilo C, tampón simple anillo de números enteros. Primero use init que use put y get. Si el buffer no contiene ningún dato, devuelve "0" cero.

//===================================== 
// ring buffer address based 
//===================================== 
#define cRingBufCount 512 
int  sRingBuf[cRingBufCount]; // Ring Buffer 
int  sRingBufPut;    // Input index address 
int  sRingBufGet;    // Output index address 
Bool sRingOverWrite; 

void GetRingBufCount(void) 
{ 
int  r; 
`  r= sRingBufPut - sRingBufGet; 
     if (r < cRingBufCount) r+= cRingBufCount; 
     return r; 
} 

void InitRingBuffer(void) 
{ 
     sRingBufPut= 0; 
     sRingBufGet= 0; 
}  

void PutRingBuffer(int d) 
{ 
     sRingBuffer[sRingBufPut]= d; 
     if (sRingBufPut==sRingBufGet)// both address are like ziro 
     { 
      sRingBufPut= IncRingBufferPointer(sRingBufPut); 
      sRingBufGet= IncRingBufferPointer(sRingBufGet); 
     } 
     else //Put over write a data 
     { 
      sRingBufPut= IncRingBufferPointer(sRingBufPut); 
      if (sRingBufPut==sRingBufGet) 
      { 
       sRingOverWrite= Ture; 
       sRingBufGet= IncRingBufferPointer(sRingBufGet); 
      } 
     } 
} 

int  GetRingBuffer(void) 
{ 
int  r; 
     if (sRingBufGet==sRingBufPut) return 0; 
     r= sRingBuf[sRingBufGet]; 
     sRingBufGet= IncRingBufferPointer(sRingBufGet); 
     sRingOverWrite=False; 
     return r; 
} 

int  IncRingBufferPointer(int a) 
{ 
     a+= 1; 
     if (a>= cRingBufCount) a= 0; 
     return a; 
}