Высокоскоростная буферизация в C++

Question

Ive писал инструмент, который берет в буфер на высокой скорости с устройства SDR (10 миллионов сложных выборок в секунду (образцы короткие)). Но с кодом, который я написал, я теряю небольшие куски, когда я оглядываюсь на то, что было написано.

Способ, которым я попытался решить эту проблему, заключается в использовании двух буферов одинакового размера и обмена между ними, чтобы избежать пропусков каких-либо образцов. Куски пропадают, когда я просматриваю процесс замены буферов и выгружает образцы в задний буфер (размер которого в 3 раза превышает частоту дискретизации), и при необходимости вызывает новый поток для записи новых данных на диск.

Устройство SDR само рекламирует свой собственный внутренний размер буфера как нечто странное, как 2016, и дает два указателя на реальные и мнимые массивы образцов. Очевидно, я хотел избежать накладных расходов таких небольших массивов на этой частоте дискретизации, поэтому, реализуя обменные буферы с большим размером, скажем 65536, надеюсь, я мог бы надеяться избежать таких проблем, но безуспешно.

У меня есть контакт, указывающий на проблему, скорее всего, на функцию обратного вызова, поскольку, когда я уменьшаю размер подстановочных буферов, пропавшие куски становятся более частыми.

Я иду об этом неправильно или есть что-то более очевидное, которого я не вижу в своем решении, или я что-то не написал правильно?

Я избегал стандартной библиотеки как можно больше просто потому, что она слишком медленна для такого рода скорости передачи данных, следовательно, требуется memmove и memcpy. Единственными исключениями являются обмен указателем буфера и создание потоков.

Перестановка буферы реализованы в виде:

IQType<short>* bufferA; 
    IQType<short>* bufferB; 

IQType является:

template <class T> class IQType { 
public: 
     T inPhaseValue; 
     T quadraturePhaseValue; 

     IQType() : inPhaseValue(0), quadraturePhaseValue(0){}; 
     IQType(T i, T q) : inPhaseValue(i), quadraturePhaseValue(q){}; 
}; 

СДР устройство функции обратного вызова, который выгружает выборки данных SDR:

void MiricsDataSource::newSamplesCallBack(short *xi, short *xq, unsigned int firstSampleNum, int grChanged, int rfChanged, int fsChanged, unsigned int numSamples, unsigned int reset, void *cbContext) { 

    MiricsDataSource* mirCtx = static_cast<MiricsDataSource*>(cbContext); 

    for (int i = 0; i < numSamples; ++i) 
    { 
     mirCtx->bufferA[mirCtx->bufferCount] = IQType<short>(xi[i],xq[i]); 
     mirCtx->bufferCount++; 
     if(mirCtx->bufferCount == mirCtx->bufferSize-1) { 
      std::swap(mirCtx->bufferA,mirCtx->bufferB); 
      mirCtx->owner->backBuffer->write(mirCtx->bufferB,mirCtx->bufferSize); 
      mirCtx->bufferCount = 0; 
     } 
    } 
} 

BackBuffer записи и связанной с t_write функции:

void BackBuffer::write(const IQType<short>* buff, size_t bLength) { 
    std::thread dumpThread(&BackBuffer::t_write,this,buff,bLength); 
    dumpThread.detach(); 
} 

void BackBuffer::t_write(const IQType<short>* buff, size_t bLength) { 
    std::lock_guard<std::mutex> lck (bufferMutex); 
    memmove(&backBuffer[0],(&backBuffer[0])+bLength,(sizeof(IQType<short>*)*(length-bLength))); 
    memcpy(&backBuffer[length-bLength],buff,(sizeof(IQType<short>*)*(bLength))); 
    if(dumpToFile) { 
     IQType<short>* toWrite = new IQType<short>[bLength]; 
     memcpy(toWrite,buff,(sizeof(IQType<short>*)*(bLength))); 
     strmDmpMgr->write(toWrite,bLength); 
    } 
} 

Answer 1

1. Наибольшая стоимость - нереститься за нить в BackBuffer::write. Не делайте этого, просто запустите один постоянный фоновый поток и отправьте ему сообщение.

2. Существует опасность в вашей текущей настройке портирования выходных буферов (заполнение второго буфера до завершения потока первым буфером позволяет снова перезаписать первый буфер).Вы можете обрабатывать произвольное количество буферов, просто используя очередь полных буферов и очередь пустых буферов, чтобы циклировать их между вашими потоками.

   Сделать фоновый поток ответственным за создание новых буферов, если вы опуститесь ниже минимального уровня свободных, чтобы сохранить динамическое распределение из критического цикла.

3. Как говорит Voo, просто чтение непосредственно в ваш большой буфер (и избегание промежуточного memcpy и т. Д.) Еще проще. Он имеет меньшую эластичность, чем подход к списку буферов, но не очевидно, что вам нужна гибкость здесь.

Есть несколько небольших оптимизаций (например, не увеличивать кол-буфер через косвенность каждую итерацию, просто сохранить правильное значение один раз), но этот поток является основной проблемой.

Answer 2

Одним из возможных источников является создание нового потока для каждого «дампа данных». В зависимости от размеров буфера вы можете создавать тысячи потоков в секунду, что может привести к серьезному ухудшению производительности не только вашей программы, но и всего компьютера. Создание одного потока - дорогостоящая операция, не говоря уже о том, что операционная система должна перебирать все ваши потоки плюс все остальные потоки в системе.

Вместо этого я предлагаю другой дизайн, в котором у вас есть пул уже запущенных потоков (поиск c++ thread pools), который вы задаете, чтобы сбросить буфер. Тогда вы можете получить круговое кольцо буферов, по одному для каждого потока, плюс одно, где вы сейчас пишете.