VE7UWU/sdrangel
1
0
Fork 0
mirror of https://github.com/f4exb/sdrangel.git synced 2024-11-22 16:08:39 -05:00
Code Issues Projects Releases Wiki Activity

Sample MI FIFO fixes

Browse Source
This commit is contained in:
f4exb 2019-10-03 01:19:43 +02:00
parent ff617b6f7d
commit 7bc413f8a9
3 changed files with 308 additions and 461 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

66
sdrbase/dsp/dspdevicemimoengine.cpp
View File

@ -243,34 +243,51 @@ void DSPDeviceMIMOEngine::workSampleSinkFifos()
return; return;
} }
int iPart1Begin; unsigned int iPart1Begin;
int iPart1End; unsigned int iPart1End;
int iPart2Begin; unsigned int iPart2Begin;
int iPart2End; unsigned int iPart2End;
std::vector<SampleVector> data = sampleFifo->getData(); const std::vector<SampleVector>& data = sampleFifo->getData();
//unsigned int samplesDone = 0;
while (sampleFifo->fillSync() > 0) while ((sampleFifo->fillSync() > 0) && (m_inputMessageQueue.size() == 0))
{ {
unsigned int count = sampleFifo->readSync(sampleFifo->fillSync(), iPart1Begin, iPart1End, iPart2Begin, iPart2End); //unsigned int count = sampleFifo->readSync(sampleFifo->fillSync(), iPart1Begin, iPart1End, iPart2Begin, iPart2End);
sampleFifo->readSync(iPart1Begin, iPart1End, iPart2Begin, iPart2End);
for (unsigned int stream = 0; stream < data.size(); stream++) if (iPart1Begin != iPart1End)
{ {
SampleVector::const_iterator begin = data[stream].begin(); for (unsigned int stream = 0; stream < data.size(); stream++) {
workSamples(data[stream].begin() + iPart1Begin, data[stream].begin() + iPart1End, stream);
if (iPart1Begin != iPart1End) {
m_vectorBuffer.write(data[stream].begin() + iPart1Begin, data[stream].begin() + iPart1End, false);
} }
if (iPart2Begin != iPart2End) {
m_vectorBuffer.write(data[stream].begin() + iPart2Begin, data[stream].begin() + iPart2End);
}
SampleVector::iterator vbegin, vend;
m_vectorBuffer.read(vbegin, vend);
workSamples(vbegin, vend, stream);
} }
sampleFifo->readCommitSync(count); if (iPart2Begin != iPart2End)
{
for (unsigned int stream = 0; stream < data.size(); stream++) {
workSamples(data[stream].begin() + iPart2Begin, data[stream].begin() + iPart2End, stream);
}
}
// for (unsigned int stream = 0; stream < data.size(); stream++)
// {
// SampleVector::const_iterator begin = data[stream].begin();
// if (iPart1Begin != iPart1End) {
// m_vectorBuffer.write(data[stream].begin() + iPart1Begin, data[stream].begin() + iPart1End, false);
// }
// if (iPart2Begin != iPart2End) {
// m_vectorBuffer.write(data[stream].begin() + iPart2Begin, data[stream].begin() + iPart2End, false);
// }
// SampleVector::iterator vbegin, vend;
// m_vectorBuffer.read(vbegin, vend);
// workSamples(vbegin, vend, stream);
// }
//sampleFifo->readCommitSync(count);
//samplesDone += count;
} }
} }
@ -287,9 +304,10 @@ void DSPDeviceMIMOEngine::workSampleSinkFifo(unsigned int stream)
SampleVector::const_iterator part2begin; SampleVector::const_iterator part2begin;
SampleVector::const_iterator part2end; SampleVector::const_iterator part2end;
while (sampleFifo->fillAsync(stream) > 0) while ((sampleFifo->fillAsync(stream) > 0) && (m_inputMessageQueue.size() == 0))
{ {
unsigned int count = sampleFifo->readAsync(sampleFifo->fillAsync(stream), &part1begin, &part1end, &part2begin, &part2end, stream); //unsigned int count = sampleFifo->readAsync(sampleFifo->fillAsync(stream), &part1begin, &part1end, &part2begin, &part2end, stream);
sampleFifo->readAsync(&part1begin, &part1end, &part2begin, &part2end, stream);
if (part1begin != part1end) { // first part of FIFO data if (part1begin != part1end) { // first part of FIFO data
m_vectorBuffer.write(part1begin, part1end, false); m_vectorBuffer.write(part1begin, part1end, false);
@ -303,7 +321,7 @@ void DSPDeviceMIMOEngine::workSampleSinkFifo(unsigned int stream)
m_vectorBuffer.read(vbegin, vend); m_vectorBuffer.read(vbegin, vend);
workSamples(vbegin, vend, stream); workSamples(vbegin, vend, stream);
sampleFifo->readCommitAsync(count, stream); //sampleFifo->readCommitAsync(count, stream);
} }
} }

609
sdrbase/dsp/samplemififo.cpp
View File

@ -17,494 +17,325 @@
#include "samplemififo.h" #include "samplemififo.h"
#define MIN(x, y) (((x) < (y)) ? (x) : (y))
void SampleMIFifo::init(unsigned int nbStreams, unsigned int size) void SampleMIFifo::init(unsigned int nbStreams, unsigned int size)
{ {
m_nbStreams = nbStreams; m_nbStreams = nbStreams;
m_size = size; m_size = size;
m_fill = 0; m_fill = 0;
m_head = 0; m_head = 0;
m_tail = 0; m_data.resize(nbStreams);
m_data.resize(nbStreams); m_vFill.clear();
m_vfill.resize(nbStreams); m_vHead.clear();
m_vhead.resize(nbStreams);
m_vtail.resize(nbStreams);
for (unsigned int stream = 0; stream < nbStreams; stream++) for (unsigned int stream = 0; stream < nbStreams; stream++)
{ {
m_data[stream].resize(size); m_data[stream].resize(size);
m_vfill[stream] = 0; m_vFill.push_back(0);
m_vhead[stream] = 0; m_vHead.push_back(0);
m_vtail[stream] = 0;
} }
} }
SampleMIFifo::SampleMIFifo(QObject* parent) : void SampleMIFifo::reset()
QObject(parent)
{ {
m_suppressed = -1; QMutexLocker mutexLocker(&m_mutex);
m_size = 0;
m_fill = 0; m_fill = 0;
m_head = 0; m_head = 0;
m_tail = 0; for (unsigned int stream = 0; stream < m_nbStreams; stream++)
}
SampleMIFifo::SampleMIFifo(unsigned int nbStreams, unsigned int size, QObject* parent) :
QObject(parent)
{
init(nbStreams, size);
m_suppressed = -1;
for (unsigned int stream = 0; stream < nbStreams; stream++)
{ {
m_vsuppressed[stream] = -1; m_vFill[stream] = 0;
m_vmsgRateTimer.push_back(QTime()); m_vHead[stream] = 0;
} }
} }
unsigned int SampleMIFifo::writeSync(const quint8* data, unsigned int count) SampleMIFifo::SampleMIFifo(QObject *parent) :
QObject(parent),
m_nbStreams(0),
m_size(0),
m_fill(0),
m_head(0)
{ {
QMutexLocker mutexLocker(&m_mutex); }
unsigned int total;
unsigned int remaining;
unsigned int len;
unsigned int byteCount = count;
count /= sizeof(Sample);
total = std::min(count, m_size - m_fill); SampleMIFifo::SampleMIFifo(unsigned int nbStreams, unsigned int size, QObject *parent) :
QObject(parent)
{
init(nbStreams, size);
}
if (total < count) void SampleMIFifo::writeSync(const quint8* data, unsigned int count)
{
QMutexLocker mutexLocker(&m_mutex);
unsigned int spaceLeft = m_size - m_fill;
unsigned int size = count / sizeof(Sample);
for (unsigned int stream = 0; stream < m_data.size(); stream++)
{ {
if (m_suppressed < 0) if (size < spaceLeft)
{ {
m_suppressed = 0; std::copy(&data[stream*count], &data[stream*count] + count, m_data[stream].begin() + m_fill);
m_msgRateTimer.start(); m_fill += size;
qCritical("SampleMIFifo: overflow - dropping %u samples", count - total); }
}
else else
{ {
if (m_msgRateTimer.elapsed() > 2500) unsigned int remaining = size - spaceLeft;
{ unsigned int bytesLeft = spaceLeft*sizeof(Sample);
qCritical("SampleMIFifo: %u messages dropped", m_suppressed); std::copy(&data[stream*count], &data[stream*count] + bytesLeft, m_data[stream].begin() + m_fill);
qCritical("SampleMIFifo: overflow - dropping %u samples", count - total); std::copy(&data[stream*count] + bytesLeft, &data[stream*count] + count, m_data[stream].begin());
m_suppressed = -1; m_fill = remaining;
}
else
{
m_suppressed++;
}
}
}
remaining = total;
std::vector<const Sample*> vbegin;
vbegin.resize(m_nbStreams);
for (unsigned int stream = 0; stream < m_nbStreams; stream++) {
vbegin[stream] = (const Sample*) &data[stream*byteCount];
}
while (remaining > 0)
{
len = std::min(remaining, m_size - m_tail);
for (unsigned int stream = 0; stream < m_nbStreams; stream++)
{
std::copy(vbegin[stream], vbegin[stream] + len, m_data[stream].begin() + m_tail);
vbegin[stream] += len;
} }
m_tail += len;
m_tail %= m_size;
m_fill += len;
remaining -= len;
}
if (m_fill > 0) {
emit dataSyncReady();
} }
return total; emit dataSyncReady();
} }
unsigned int SampleMIFifo::writeSync(std::vector<SampleVector::const_iterator> vbegin, unsigned int count) void SampleMIFifo::writeSync(const std::vector<SampleVector::const_iterator>& vbegin, unsigned int size)
{ {
if ((vbegin.size() != m_nbStreams)) { if ((m_data.size() == 0) || (m_data.size() != vbegin.size())) {
return 0; return;
} }
QMutexLocker mutexLocker(&m_mutex); QMutexLocker mutexLocker(&m_mutex);
unsigned int total; unsigned int spaceLeft = m_size - m_fill;
unsigned int remaining;
unsigned int len;
total = std::min(count, m_size - m_fill); if (size < spaceLeft)
if (total < count)
{ {
if (m_suppressed < 0) for (unsigned int stream = 0; stream < m_data.size(); stream++) {
{ std::copy(vbegin[stream], vbegin[stream] + size, m_data[stream].begin() + m_fill);
m_suppressed = 0;
m_msgRateTimer.start();
qCritical("SampleMIFifo::writeSync: overflow - dropping %u samples", count - total);
}
else
{
if (m_msgRateTimer.elapsed() > 2500)
{
qCritical("SampleMIFifo::writeSync: %u messages dropped", m_suppressed);
qCritical("SampleMIFifo::writeSync: overflow - dropping %u samples", count - total);
m_suppressed = -1;
}
else
{
m_suppressed++;
}
}
}
remaining = total;
while (remaining > 0)
{
len = std::min(remaining, m_size - m_tail);
for (unsigned int stream = 0; stream < m_nbStreams; stream++)
{
std::copy(vbegin[stream], vbegin[stream] + len, m_data[stream].begin() + m_tail);
vbegin[stream] += len;
} }
m_tail += len; m_fill += size;
m_tail %= m_size; }
m_fill += len; else
remaining -= len; {
} unsigned int remaining = size - spaceLeft;
if (m_fill > 0) { for (unsigned int stream = 0; stream < m_data.size(); stream++)
emit dataSyncReady(); {
std::copy(vbegin[stream], vbegin[stream] + spaceLeft, m_data[stream].begin() + m_fill);
std::copy(vbegin[stream] + spaceLeft, vbegin[stream] + size, m_data[stream].begin());
}
m_fill = remaining;
} }
return total; emit dataSyncReady();
} }
unsigned int SampleMIFifo::readSync(unsigned int count, void SampleMIFifo::readSync(
std::vector<SampleVector::const_iterator*> vpart1Begin, std::vector<SampleVector::const_iterator*> vpart1End, std::vector<SampleVector::const_iterator*> vpart1Begin, std::vector<SampleVector::const_iterator*> vpart1End,
std::vector<SampleVector::const_iterator*> vpart2Begin, std::vector<SampleVector::const_iterator*> vpart2End) std::vector<SampleVector::const_iterator*> vpart2Begin, std::vector<SampleVector::const_iterator*> vpart2End
)
{ {
if ((vpart1Begin.size() != m_nbStreams) if (m_data.size() == 0) {
|| (vpart2Begin.size() != m_nbStreams) return;
|| (vpart1End.size() != m_nbStreams)
|| (vpart2End.size() != m_nbStreams))
{
return 0;
} }
QMutexLocker mutexLocker(&m_mutex); QMutexLocker mutexLocker(&m_mutex);
unsigned int total; std::vector<SampleVector>::iterator dataIt = m_data.begin();
unsigned int remaining; vpart1Begin.resize(m_nbStreams);
unsigned int len; vpart1End.resize(m_nbStreams);
unsigned int head = m_head; vpart2Begin.resize(m_nbStreams);
vpart2End.resize(m_nbStreams);
total = std::min(count, m_fill); if (m_head < m_fill)
if (total < count) {
qCritical("SampleMIFifo::readSync: underflow - missing %u samples", count - total);
}
remaining = total;
if (remaining > 0)
{ {
len = std::min(remaining, m_size - head); for (unsigned int stream = 0; stream < m_data.size(); stream++)
for (unsigned int stream = 0; stream < m_nbStreams; stream++)
{
*vpart1Begin[stream] = m_data[stream].begin() + head;
*vpart1End[stream] = m_data[stream].begin() + head + len;
}
head += len;
head %= m_size;
remaining -= len;
}
else
{
for (unsigned int stream = 0; stream < m_nbStreams; stream++)
{
*vpart1Begin[stream] = m_data[stream].end();
*vpart1End[stream] = m_data[stream].end();
}
}
if (remaining > 0)
{
len = std::min(remaining, m_size - head);
for (unsigned int stream = 0; stream < m_nbStreams; stream++)
{
*vpart2Begin[stream] = m_data[stream].begin() + head;
*vpart2End[stream] = m_data[stream].begin() + head + len;
}
}
else
{
for (unsigned int stream = 0; stream < m_nbStreams; stream++)
{ {
*vpart1Begin[stream] = m_data[stream].begin() + m_head;
*vpart1End[stream] = m_data[stream].begin() + m_fill;
*vpart2Begin[stream] = m_data[stream].end(); *vpart2Begin[stream] = m_data[stream].end();
*vpart2End[stream] = m_data[stream].end(); *vpart2End[stream] = m_data[stream].end();
} }
}
return total;
}
unsigned int SampleMIFifo::readSync(unsigned int count,
int& ipart1Begin, int& ipart1End,
int& ipart2Begin, int& ipart2End)
{
QMutexLocker mutexLocker(&m_mutex);
unsigned int total;
unsigned int remaining;
unsigned int len;
unsigned int head = m_head;
total = std::min(count, m_fill);
if (total < count) {
qCritical("SampleMIFifo::readSync: underflow - missing %u samples", count - total);
} }
remaining = total;
if (remaining > 0)
{
len = std::min(remaining, m_size - head);
ipart1Begin = head;
ipart1End =head + len;
head += len;
head %= m_size;
remaining -= len;
}
else else
{ {
ipart1Begin = m_size; for (unsigned int stream = 0; stream < m_data.size(); stream++)
ipart1End = m_size; {
} *vpart1Begin[stream] = m_data[stream].begin() + m_head;
*vpart1End[stream] = m_data[stream].end();
*vpart2Begin[stream] = m_data[stream].begin();
*vpart2End[stream] = m_data[stream].begin() + m_fill;
}
}
if (remaining > 0) m_head = m_fill;
}
void SampleMIFifo::readSync(
std::vector<unsigned int>& vpart1Begin, std::vector<unsigned int>& vpart1End,
std::vector<unsigned int>& vpart2Begin, std::vector<unsigned int>& vpart2End
)
{
if (m_data.size() == 0) {
return;
}
QMutexLocker mutexLocker(&m_mutex);
std::vector<SampleVector>::iterator dataIt = m_data.begin();
vpart1Begin.resize(m_nbStreams);
vpart1End.resize(m_nbStreams);
vpart2Begin.resize(m_nbStreams);
vpart2End.resize(m_nbStreams);
if (m_head < m_fill)
{ {
len = std::min(remaining, m_size - head); for (unsigned int stream = 0; stream < m_data.size(); stream++)
ipart2Begin = head; {
ipart2End = head + len; vpart1Begin[stream] = m_head;
} vpart1End[stream] = m_fill;
vpart2Begin[stream] = 0;
vpart2End[stream] = 0;
}
}
else else
{ {
ipart2Begin = m_size; for (unsigned int stream = 0; stream < m_data.size(); stream++)
ipart2End = m_size; {
} vpart1Begin[stream] = m_head;
vpart1End[stream] = m_size;
vpart2Begin[stream] = 0;
vpart2End[stream] = m_fill;
}
}
return total; m_head = m_fill;
} }
unsigned int SampleMIFifo::readCommitSync(unsigned int count) void SampleMIFifo::readSync(
unsigned int& ipart1Begin, unsigned int& ipart1End,
unsigned int& ipart2Begin, unsigned int& ipart2End
)
{ {
QMutexLocker mutexLocker(&m_mutex); if (m_data.size() == 0) {
return;
}
if (count > m_fill) if (m_head < m_fill)
{ {
qCritical("SampleMIFifo::readCommitSync: cannot commit more than available samples"); ipart1Begin = m_head;
count = m_fill; ipart1End = m_fill;
} ipart2Begin = 0;
ipart2End = 0;
}
else
{
ipart1Begin = m_head;
ipart1End = m_size;
ipart2Begin = 0;
ipart2End = m_fill;
}
m_head = (m_head + count) % m_size; m_head = m_fill;
m_fill -= count;
return count;
} }
unsigned int SampleMIFifo::writeAsync(const quint8* data, unsigned int count, unsigned int stream) void SampleMIFifo::writeAsync(const quint8* data, unsigned int count, unsigned int stream)
{ {
if (stream >= m_nbStreams) { if (stream >= m_nbStreams) {
return 0; return;
} }
QMutexLocker mutexLocker(&m_mutex); QMutexLocker mutexLocker(&m_mutex);
unsigned int total; unsigned int spaceLeft = m_size - m_vFill[stream];
unsigned int remaining; unsigned int size = count / sizeof(Sample);
unsigned int len;
const Sample* begin = (const Sample*) data;
count /= sizeof(Sample);
total = std::min(count, m_size - m_vfill[stream]);
if (total < count) if (size < spaceLeft)
{ {
if (m_vsuppressed[stream] < 0) std::copy(&data[stream*count], &data[stream*count] + count, m_data[stream].begin() + m_vFill[stream]);
{ m_vFill[stream] += size;
m_vsuppressed[stream] = 0; }
m_msgRateTimer.start(); else
qCritical("SampleSinkFifo::writeAsync[%u]: overflow - dropping %u samples", stream, count - total);
}
else
{
if (m_msgRateTimer.elapsed() > 2500)
{
qCritical("SampleSinkFifo::writeAsync[%u]: %u messages dropped", stream, m_vsuppressed[stream]);
qCritical("SampleSinkFifo::writeAsync[%u]: overflow - dropping %u samples", stream, count - total);
m_vsuppressed[stream] = -1;
}
else
{
m_vsuppressed[stream]++;
}
}
}
remaining = total;
while (remaining > 0)
{ {
len = std::min(remaining, m_size - m_vtail[stream]); unsigned int remaining = size - spaceLeft;
std::copy(begin, begin + len, m_data[stream].begin() + m_vtail[stream]); unsigned int bytesLeft = spaceLeft * sizeof(Sample);
begin += len; std::copy(&data[stream*count], &data[stream*count] + bytesLeft, m_data[stream].begin() + m_vFill[stream]);
m_vtail[stream] += len; std::copy(&data[stream*count] + bytesLeft, &data[stream*count] + count, m_data[stream].begin());
m_vtail[stream] %= m_size; m_vFill[stream] = remaining;
m_vfill[stream] += len;
remaining -= len;
}
if (m_vfill[stream] > 0) {
emit dataAsyncReady(stream);
} }
return total; emit dataAsyncReady(stream);
} }
unsigned int SampleMIFifo::writeAsync(SampleVector::const_iterator begin, unsigned int count, unsigned int stream) void SampleMIFifo::writeAsync(const SampleVector::const_iterator& begin, unsigned int size, unsigned int stream)
{ {
if (stream >= m_nbStreams) { if (stream >= m_nbStreams) {
return 0; return;
} }
QMutexLocker mutexLocker(&m_mutex); QMutexLocker mutexLocker(&m_mutex);
unsigned int total; int spaceLeft = m_data[stream].size() - m_vFill[stream];
unsigned int remaining;
unsigned int len;
total = std::min(count, m_size - m_vfill[stream]);
if (total < count) if (size < spaceLeft)
{ {
if (m_vsuppressed[stream] < 0) std::copy(begin, begin + size, m_data[stream].begin() + m_vFill[stream]);
{ m_vFill[stream] += size;
m_vsuppressed[stream] = 0; }
m_vmsgRateTimer[stream].start(); else
qCritical("SampleSinkFifo::writeAsync[%u]: overflow - dropping %u samples", stream, count - total);
}
else
{
if (m_vmsgRateTimer[stream].elapsed() > 2500)
{
qCritical("SampleSinkFifo::writeAsync[%u]: %u messages dropped", stream, m_vsuppressed[stream]);
qCritical("SampleSinkFifo::writeAsync[%u]: overflow - dropping %u samples", stream, count - total);
m_vsuppressed[stream] = -1;
}
else
{
m_vsuppressed[stream]++;
}
}
}
remaining = total;
while (remaining > 0)
{ {
len = std::min(remaining, m_size - m_vtail[stream]); int remaining = size - spaceLeft;
std::copy(begin, begin + len, m_data[stream].begin() + m_vtail[stream]); std::copy(begin, begin + spaceLeft, m_data[stream].begin() + m_vFill[stream]);
begin += len; std::copy(begin + spaceLeft, begin + size, m_data[stream].begin());
m_vtail[stream] += len; m_vFill[stream] = remaining;
m_vtail[stream] %= m_size;
m_vfill[stream] += len;
remaining -= len;
}
if (m_vfill[stream] > 0) {
emit dataAsyncReady(stream);
} }
return total; emit dataAsyncReady(stream);
} }
unsigned int SampleMIFifo::readAsync(unsigned int count,
SampleVector::const_iterator* part1Begin, SampleVector::const_iterator* part1End, void SampleMIFifo::readAsync(
SampleVector::const_iterator* part2Begin, SampleVector::const_iterator* part2End, SampleVector::const_iterator* part1Begin, SampleVector::const_iterator* part1End,
SampleVector::const_iterator* part2Begin, SampleVector::const_iterator* part2End,
unsigned int stream)
{
if (stream >= m_data.size()) {
return;
}
QMutexLocker mutexLocker(&m_mutex);
if (m_vHead[stream] < m_vFill[stream])
{
*part1Begin = m_data[stream].begin() + m_vHead[stream];
*part1End = m_data[stream].begin() + m_vFill[stream];
*part2Begin = m_data[stream].begin();
*part2End = m_data[stream].begin();
}
else
{
*part1Begin = m_data[stream].begin() + m_vHead[stream];
*part1End = m_data[stream].end();
*part2Begin = m_data[stream].begin();
*part2End = m_data[stream].begin() + m_vFill[stream];
}
m_vHead[stream] = m_vFill[stream];
}
void SampleMIFifo::readAsync(
unsigned int& ipart1Begin, unsigned int& ipart1End,
unsigned int& ipart2Begin, unsigned int& ipart2End,
unsigned int stream) unsigned int stream)
{ {
if (stream >= m_nbStreams) { if (stream >= m_data.size()) {
return 0; return;
} }
QMutexLocker mutexLocker(&m_mutex); QMutexLocker mutexLocker(&m_mutex);
unsigned int total;
unsigned int remaining;
unsigned int len;
unsigned int head = m_vhead[stream];
total = std::min(count, m_vfill[stream]);
if (total < count) { if (m_vHead[stream] < m_vFill[stream])
qCritical("SampleSinkFifo::readAsync[%u]: underflow - missing %u samples", stream, count - total);
}
remaining = total;
if (remaining > 0)
{ {
len = std::min(remaining, m_size - head); ipart1Begin = m_vHead[stream];
*part1Begin = m_data[stream].begin() + head; ipart1End = m_vFill[stream];
*part1End = m_data[stream].begin() + head + len; ipart2Begin = m_size;
head += len; ipart2End = m_size;
head %= m_size; }
remaining -= len;
}
else else
{ {
*part1Begin = m_data[stream].end(); ipart1Begin = m_vHead[stream];
*part1End = m_data[stream].end(); ipart1End = m_size;
} ipart2Begin = 0;
ipart2End = m_vFill[stream];
if (remaining > 0)
{
len = std::min(remaining, m_size - head);
*part2Begin = m_data[stream].begin() + head;
*part2End = m_data[stream].begin() + head + len;
}
else
{
*part2Begin = m_data[stream].end();
*part2End = m_data[stream].end();
}
return total;
}
unsigned int SampleMIFifo::readCommitAsync(unsigned int count, unsigned int stream)
{
if (stream >= m_nbStreams) {
return 0;
} }
QMutexLocker mutexLocker(&m_mutex); m_vHead[stream] = m_vFill[stream];
}
if (count > m_vfill[stream])
{
qCritical("SampleSinkFifo::readCommitAsync[%u]: cannot commit more than available samples", stream);
count = m_vfill[stream];
}
m_vhead[stream] = (m_vhead[stream] + count) % m_size;
m_vfill[stream] -= count;
return count;
}

94
sdrbase/dsp/samplemififo.h
View File

@ -20,7 +20,7 @@
#include <QObject> #include <QObject>
#include <QMutex> #include <QMutex>
#include <QTime> #include <vector>
#include "dsp/dsptypes.h" #include "dsp/dsptypes.h"
#include "export.h" #include "export.h"
@ -28,16 +28,46 @@ class SDRBASE_API SampleMIFifo : public QObject {
Q_OBJECT Q_OBJECT
public: public:
SampleMIFifo(QObject* parent = nullptr); SampleMIFifo(QObject *parent = nullptr);
SampleMIFifo(unsigned int nbStreams, unsigned int size, QObject* parent = nullptr); SampleMIFifo(unsigned int nbStreams, unsigned int size, QObject *parent = nullptr);
void init(unsigned int nbStreams, unsigned int size);
void reset();
void init(unsigned int nbStreams, unsigned int size); void writeSync(const quint8* data, unsigned int count); //!< de-interleaved data in input with count bytes for each stream
inline unsigned int size() const { return m_size; } void writeSync(const std::vector<SampleVector::const_iterator>& vbegin, unsigned int size);
void readSync(
std::vector<SampleVector::const_iterator*> vpart1Begin, std::vector<SampleVector::const_iterator*> vpart1End,
std::vector<SampleVector::const_iterator*> vpart2Begin, std::vector<SampleVector::const_iterator*> vpart2End
);
void readSync(
std::vector<unsigned int>& vPart1Begin, std::vector<unsigned int>& vPart1End,
std::vector<unsigned int>& vPart2Begin, std::vector<unsigned int>& vPart2End
);
void readSync(
unsigned int& ipart1Begin, unsigned int& ipart1End,
unsigned int& ipart2Begin, unsigned int& ipart2End
);
void writeAsync(const quint8* data, unsigned int count, unsigned int stream);
void writeAsync(const SampleVector::const_iterator& begin, unsigned int size, unsigned int stream);
void readAsync(
SampleVector::const_iterator* part1Begin, SampleVector::const_iterator* part1End,
SampleVector::const_iterator* part2Begin, SampleVector::const_iterator* part2End,
unsigned int stream);
void readAsync(
unsigned int& ipart1Begin, unsigned int& ipart1End,
unsigned int& ipart2Begin, unsigned int& ipart2End,
unsigned int stream);
const std::vector<SampleVector>& getData() { return m_data; }
const SampleVector& getData(unsigned int stream) { return m_data[stream]; }
unsigned int getNbStreams() const { return m_data.size(); }
inline unsigned int fillSync() inline unsigned int fillSync()
{ {
QMutexLocker mutexLocker(&m_mutex); QMutexLocker mutexLocker(&m_mutex);
unsigned int fill = m_fill; unsigned int fill = m_head <= m_fill ? m_fill - m_head : m_size - (m_head - m_fill);
return fill; return fill;
} }
@ -48,55 +78,23 @@ public:
} }
QMutexLocker mutexLocker(&m_mutex); QMutexLocker mutexLocker(&m_mutex);
unsigned int fill = m_vfill[stream]; unsigned int fill = m_vHead[stream] <= m_vFill[stream] ? m_vFill[stream] - m_vHead[stream] : m_size - (m_vHead[stream] - m_vFill[stream]);
return fill; return fill;
} }
const std::vector<SampleVector>& getData() { return m_data; }
unsigned int getNbStreams() const { return m_nbStreams; }
unsigned int writeSync(const quint8* data, unsigned int count); //!< de-interleaved data in input with count bytes for each stream
unsigned int writeSync(std::vector<SampleVector::const_iterator> vbegin, unsigned int count);
unsigned int readSync(unsigned int count,
std::vector<SampleVector::const_iterator*> vpart1Begin, std::vector<SampleVector::const_iterator*> vpart1End,
std::vector<SampleVector::const_iterator*> vpart2Begin, std::vector<SampleVector::const_iterator*> vpart2End);
unsigned int readSync(unsigned int count,
int& ipart1Begin, int& ipart1End,
int& ipart2Begin, int& ipart2End);
unsigned int readCommitSync(unsigned int count);
unsigned int writeAsync(const quint8* data, unsigned int count, unsigned int stream);
unsigned int writeAsync(SampleVector::const_iterator begin, unsigned int count, unsigned int stream);
unsigned int readAsync(unsigned int count,
SampleVector::const_iterator* part1Begin, SampleVector::const_iterator* part1End,
SampleVector::const_iterator* part2Begin, SampleVector::const_iterator* part2End,
unsigned int stream);
unsigned int readCommitAsync(unsigned int count, unsigned int stream);
signals: signals:
void dataSyncReady(); void dataSyncReady();
void dataAsyncReady(int streamIndex); void dataAsyncReady(int streamIndex);
private: private:
QMutex m_mutex; std::vector<SampleVector> m_data;
std::vector<SampleVector> m_data;
unsigned int m_nbStreams; unsigned int m_nbStreams;
unsigned int m_size; unsigned int m_size;
unsigned int m_fill; //!< Number of samples written from beginning of samples vector (sync)
// Synchronous unsigned int m_head; //!< Number of samples read from beginning of samples vector (sync)
int m_suppressed; std::vector<unsigned int> m_vFill; //!< Number of samples written from beginning of samples vector (async)
QTime m_msgRateTimer; std::vector<unsigned int> m_vHead; //!< Number of samples read from beginning of samples vector (async)
unsigned int m_fill; QMutex m_mutex;
unsigned int m_head;
unsigned int m_tail;
// Asynchronous
std::vector<int> m_vsuppressed;
std::vector<QTime> m_vmsgRateTimer;
std::vector<unsigned int> m_vfill;
std::vector<unsigned int> m_vhead;
std::vector<unsigned int> m_vtail;
}; };
#endif // INCLUDE_SAMPLEMIFIFO_H #endif // INCLUDE_SAMPLEMIFIFO_H