あらゆるタイプのオブジェクトを保持できる固定サイズ（コンパイル時ではなく、実行時に選択可能）の循環バッファーが必要であり、非常に高性能である必要があります。マルチタスクの組み込み環境ではありますが、タスク自体がそれを管理できるように協調的な環境であるため、リソース競合の問題は発生しないと思います。
私の最初の考えは、タイプ（単純な列挙型/定義）とペイロードへのvoidポインターを含む単純な構造体をバッファーに格納することでしたが、これをできるだけ速くしたいので、バイパスを含む提案を受け入れますヒープ。
実際、私は生の速度のために標準ライブラリのいずれかをバイパスすることを嬉しく思います-私が見たコードから、それはCPU用に高度に最適化されていません：彼らはちょうどそのようなもののためにCコードをコンパイルしたようですstrcpy()、手作業でコーディングされたアセンブリ。
コードやアイデアをいただければ幸いです。必要な操作は次のとおりです。
最も簡単な解決策は、アイテムのサイズとアイテムの数を追跡し、適切なバイト数のバッファーを作成することです。
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--;
}
// 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;
}
リングバッファの長さが2の累乗である限り、信じられないほど高速な2項「＆」演算がインデックスをラップアラウンドします。私のアプリケーションでは、マイクから取得したオーディオのリングバッファからユーザーにオーディオのセグメントを表示しています。
画面に表示できるオーディオの最大量が、リングバッファのサイズよりもはるかに少ないことを常に確認しています。そうしないと、同じチャンクから読み取りと書き込みを行う可能性があります。これにより、奇妙な表示アーティファクトが発生する可能性があります。
バッファーをコーディングするときに必要なタイプを列挙できますか、それとも実行時に動的呼び出しを介してタイプを追加できる必要がありますか？前者の場合、n個の構造体のヒープ割り当て配列としてバッファーを作成します。各構造体は、データ型を識別する列挙型タグとすべてのデータ型の共用体の2つの要素で構成されます。小さな要素の追加ストレージに関して失うものは、割り当て/割り当て解除とその結果として生じるメモリの断片化に対処する必要がないという点で補います。次に、バッファのヘッド要素とテール要素を定義する開始インデックスと終了インデックスを追跡し、インデックスをインクリメント/デクリメントするときにmodnを計算する必要があります。
まず、見出し。ビットintを使用してヘッドとテールの「ポインター」を保持し、完全に同期するようにサイズを変更する場合は、バッファーをラップするためにモジュロ演算は必要ありません。IE：12ビットのunsigned intに詰め込まれた4096は、それ自体が0であり、いかなる方法でも問題がありません。2の累乗であっても、モジュロ演算を排除すると、速度が2倍になります（ほぼ正確に）。
VisualStudio2010のC++コンパイラをデフォルトのインライン化で使用する第3世代i7DellXPS 8500では、あらゆるタイプのデータ要素の4096バッファの充填と排出を1,000万回繰り返すと、52秒かかり、その1/8192回でデータを処理します。
私はRXでmain（）のテストループを書き直して、フローを制御しなくなるようにします。これは、バッファがいっぱいまたは空であることを示す戻り値と、それに伴うブレークによって制御されます。ステートメント。IE：フィラーとドレイナーは、破損や不安定さなしに互いにぶつかることができる必要があります。ある時点で、このコードをマルチスレッド化することを望んでいます。その場合、その動作が重要になります。
QUEUE_DESC（キュー記述子）および初期化関数は、このコード内のすべてのバッファーを2の累乗にするように強制します。それ以外の場合、上記のスキームは機能しません。この件に関しては、QUEUE_DESCはハードコーディングされていないことに注意してください。その構築には、マニフェスト定数（#define BITS_ELE_KNT）が使用されます。（ここでは2の累乗で十分な柔軟性があると思います）
バッファサイズを実行時に選択可能にするために、さまざまなアプローチ（ここには示されていません）を試し、FIFOバッファ[USHRT]を管理できるHead、Tail、EleKntにUSHRTを使用することにしました。モジュロ演算を回避するために、Head、Tailを使用して&&のマスクを作成しましたが、そのマスクは（EleKnt -1）であることが判明したため、それを使用してください。ビットintの代わりにUSHRTSを使用すると、静かなマシンでパフォーマンスが約15％向上しました。Intel CPUコアは常にバスよりも高速であるため、ビジーな共有マシンでは、データ構造をパックすることで、他の競合するスレッドよりも先にロードして実行できます。トレードオフ。
バッファの実際のストレージはcalloc（）を使用してヒープに割り当てられ、ポインタは構造体のベースにあるため、構造体とポインタのアドレスはまったく同じであることに注意してください。IE; レジスタを拘束するために構造体アドレスにオフセットを追加する必要はありません。
同じように、バッファーのサービスに付随するすべての変数は、バッファーに物理的に隣接し、同じ構造体にバインドされているため、コンパイラーは美しいアセンブリ言語を作成できます。アセンブリを表示するには、インライン最適化を強制終了する必要があります。そうしないと、アセンブリが忘却に陥ってしまうためです。
あらゆるデータ型のポリモーフィズムをサポートするために、割り当ての代わりにmemcpy（）を使用しました。コンパイルごとに1つの確率変数タイプをサポートする柔軟性のみが必要な場合、このコードは完全に機能します。
ポリモーフィズムの場合は、タイプとそのストレージ要件を知っている必要があります。記述子のDATA_DESC配列は、QUEUE_DESC.pBufferに配置された各データを追跡して、適切に取得できるようにする方法を提供します。最大のデータ型のすべての要素を保持するのに十分なpBufferメモリを割り当てるだけですが、特定のデータがDATA_DESC.dBytesで実際に使用しているストレージの量を追跡します。別の方法は、ヒープマネージャーを再発明することです。
つまり、QUEUE_DESCのUCHAR * pBufferには、データ型とサイズを追跡するための並列コンパニオン配列がありますが、pBuffer内のデータの保存場所は現在のままです。新しいメンバーは、DATA_DESC * pDataDescのようなものになります。または、そのような前方参照を使用してコンパイラーを打ち負かして送信する方法を見つけることができれば、おそらくDATA_DESC DataDesc [2^BITS_ELE_KNT]になります。Calloc（）は、これらの状況で常により柔軟です。
Q_Put（）、Q_Getでもmemcpy（）を使用しますが、実際にコピーされるバイト数は、QUEUE_DESC.EleBytesではなくDATA_DESC.dBytesによって決定されます。要素は、特定のputまたはgetに対して、潜在的にすべて異なるタイプ/サイズです。
このコードは速度とバッファサイズの要件を満たし、6つの異なるデータ型の要件を満たすように作成できると思います。多くのテストフィクスチャをprintf（）ステートメントの形式で残しているので、コードが正しく機能することを（またはそうでなくても）満足させることができます。乱数ジェネレーターは、コードが任意のランダムなヘッド/テールコンボで機能することを示しています。
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;
}
これがCの簡単な解決策です。関数ごとに割り込みがオフになっていると仮定します。ポリモーフィズムなどはなく、常識です。
#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
}
単純な実装は次のもので構成されます。
データを書き込むたびに、書き込みポインタを進めてカウンタをインクリメントします。データを読み取るときは、読み取りポインターを増やし、カウンターをデクリメントします。いずれかのポインタがnに達した場合は、ゼロに設定します。
counter=nの場合は書き込みできません。counter = 0の場合、読み取ることはできません。
Cスタイル、整数用の単純なリングバッファ。put and getを使用するよりも、最初にinitを使用します。バッファにデータが含まれていない場合は、「0」のゼロが返されます。
//=====================================
// 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;
}
@Adam Rosenfieldのソリューションは、正しいものの、invloveやcircular_bufferを使用しないより軽量な構造で実装できます。countcapacity
構造体は、次の4つのポインターしか保持できませんでした。
buffer：メモリ内のバッファの開始を指します。buffer_end：メモリ内のバッファの終わりを指します。head：保存されたデータの終わりを指します。tail：保存されたデータの先頭を指します。szストレージの単位のパラメーター化を可能にする属性を保持できます。
countと値の両方はcapacity、上記のポインタを使用して導出可能である必要があります。
capacitybuffer_endポインタとポインタの間の距離bufferをストレージの単位で割ることで導出できるため、は簡単ですsz（以下のスニペットは擬似コードです）。
capacity = (buffer_end - buffer) / sz
しかし、数えれば、物事はもう少し複雑になります。たとえば、同じ場所headをtail指しているシナリオでは、バッファが空か満杯かを判断する方法はありません。
これに対処するには、バッファは追加の要素にメモリを割り当てる必要があります。たとえば、循環バッファの必要な容量がである場合、10 * szを割り当てる必要があり11 * szます。
容量の式は次のようになります（以下のスニペットは擬似コードです）。
capacity_bytes = buffer_end - buffer - sz
capacity = capacity_bytes / sz
この追加の要素セマンティクスにより、バッファーが空か満杯かを評価する条件を構築できます。
バッファを空にするために、headポインタはポインタと同じ場所をtail指します。
head == tail
上記がtrueと評価された場合、バッファは空です。
バッファがいっぱいになるには、headポインタがポインタの1要素後ろにある必要がありtailます。headしたがって、その場所からその場所にジャンプするためにカバーする必要のあるスペースは、tailに等しくなければなりません1 * sz。
tailより大きい場合head：tail - head == sz
上記がtrueと評価された場合、バッファはいっぱいです。
headより大きい場合tail：buffer_end - headheadからバッファの最後にジャンプするスペースを返します。tail - bufferバッファの先頭から`テールにジャンプするために必要なスペースを返します。head。tail1 * sz(buffer_end - head) + (tail - buffer) == sz
=> buffer_end - buffer - head + tail == sz
=> buffer_end - buffer - sz == head - tail
=> head - tail == buffer_end - buffer - sz
=> head - tail == capacity_bytes
上記がtrueと評価された場合、バッファはいっぱいです。
circular_buffer上記の構造を使用するように@AdamRosenfieldを変更します。
#include <string.h>
#define CB_SUCCESS 0 /* CB operation was successful */
#define CB_MEMORY_ERROR 1 /* Failed to allocate memory */
#define CB_OVERFLOW_ERROR 2 /* CB is full. Cannot push more items. */
#define CB_EMPTY_ERROR 3 /* CB is empty. Cannot pop more items. */
typedef struct circular_buffer {
void *buffer;
void *buffer_end;
size_t sz;
void *head;
void *tail;
} circular_buffer;
int cb_init(circular_buffer *cb, size_t capacity, size_t sz) {
const int incremented_capacity = capacity + 1; // Add extra element to evaluate count
cb->buffer = malloc(incremented_capacity * sz);
if (cb->buffer == NULL)
return CB_MEMORY_ERROR;
cb->buffer_end = (char *)cb->buffer + incremented_capacity * sz;
cb->sz = sz;
cb->head = cb->buffer;
cb->tail = cb->buffer;
return CB_SUCCESS;
}
int cb_free(circular_buffer *cb) {
free(cb->buffer);
return CB_SUCCESS;
}
const int _cb_length(circular_buffer *cb) {
return (char *)cb->buffer_end - (char *)cb->buffer;
}
int cb_push_back(circular_buffer *cb, const void *item) {
const int buffer_length = _cb_length(cb);
const int capacity_length = buffer_length - cb->sz;
if ((char *)cb->tail - (char *)cb->head == cb->sz ||
(char *)cb->head - (char *)cb->tail == capacity_length)
return CB_OVERFLOW_ERROR;
memcpy(cb->head, item, cb->sz);
cb->head = (char*)cb->head + cb->sz;
if(cb->head == cb->buffer_end)
cb->head = cb->buffer;
return CB_SUCCESS;
}
int cb_pop_front(circular_buffer *cb, void *item) {
if (cb->head == cb->tail)
return CB_EMPTY_ERROR;
memcpy(item, cb->tail, cb->sz);
cb->tail = (char*)cb->tail + cb->sz;
if(cb->tail == cb->buffer_end)
cb->tail = cb->buffer;
return CB_SUCCESS;
}
adam-rosenfieldのソリューションを拡張すると、マルチスレッドの単一プロデューサー-単一コンシューマーシナリオで次のことが機能すると思います。
int cb_push_back(circular_buffer *cb, const void *item)
{
void *new_head = (char *)cb->head + cb->sz;
if (new_head == cb>buffer_end) {
new_head = cb->buffer;
}
if (new_head == cb->tail) {
return 1;
}
memcpy(cb->head, item, cb->sz);
cb->head = new_head;
return 0;
}
int cb_pop_front(circular_buffer *cb, void *item)
{
void *new_tail = cb->tail + cb->sz;
if (cb->head == cb->tail) {
return 1;
}
memcpy(item, cb->tail, cb->sz);
if (new_tail == cb->buffer_end) {
new_tail = cb->buffer;
}
cb->tail = new_tail;
return 0;
}