GitHub/sdrtrunk
1
0
Fork 0
mirror of https://github.com/DSheirer/sdrtrunk.git synced 2026-11-12 08:37:25 -05:00
Code Issues Packages Projects Releases 2 Wiki Activity

Merge pull request #319 from DSheirer/318-dft-processor-non-blocking-queue

Browse Source 
318 DFT Processor Non-Blocking Queue redesign
This commit is contained in:
Denny Sheirer
2018-01-15 04:54:40 -05:00
committed by GitHub
parent 69e9bed2af d5b377b1af
commit 3d9b38bef5
2 changed files with 314 additions and 223 deletions
Download Patch File Download Diff File Expand all files Collapse all files

335
src/main/java/io/github/dsheirer/spectrum/DFTProcessor.java
View File

@@ -1,19 +1,16 @@
/*******************************************************************************
* sdrtrunk
* Copyright (C) 2014-2017 Dennis Sheirer
* sdr-trunk
* Copyright (C) 2014-2018 Dennis Sheirer
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
* This program is free software: you can redistribute it and/or modify it under the terms of the GNU General Public
* License as published by the Free Software Foundation, either version 3 of the License, or (at your option) any
* later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
* This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied
* warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>
* You should have received a copy of the GNU General Public License along with this program.
* If not, see <http://www.gnu.org/licenses/>
*
******************************************************************************/
package io.github.dsheirer.spectrum;
@@ -21,7 +18,6 @@ package io.github.dsheirer.spectrum;
import io.github.dsheirer.dsp.filter.Window;
import io.github.dsheirer.dsp.filter.Window.WindowType;
import io.github.dsheirer.properties.SystemProperties;
import io.github.dsheirer.sample.Buffer;
import io.github.dsheirer.sample.Listener;
import io.github.dsheirer.sample.SampleType;
import io.github.dsheirer.sample.complex.ComplexBuffer;
@@ -33,60 +29,42 @@ import org.jtransforms.fft.FloatFFT_1D;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Arrays;
import java.io.IOException;
import java.util.Iterator;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
/**
* Processes both complex samples or float samples and dispatches a float array
* of DFT results, using configurable fft size and output dispatch timelines.
* Processes both complex samples or float samples and dispatches a float array of DFT results, using configurable fft
* size and output dispatch timelines.
*/
public class DFTProcessor implements Listener<ComplexBuffer>, IFrequencyChangeProcessor, IDFTWidthChangeProcessor
{
private final static Logger mLog = LoggerFactory.getLogger(DFTProcessor.class);
private CopyOnWriteArrayList<DFTResultsConverter> mListeners = new CopyOnWriteArrayList<DFTResultsConverter>();
//Fixed size sample queue. We set the size at 12 based on the highest
//sample rate tuner (10 MHz) producing 153 frames per second and the lowest
//DFT processing frame rate of 12 frames per second. This means that the
//maximum size needed is 153 / 12 = 11 frames per DFT processing iteration
//so we set it at 12.
private BlockingQueue<ComplexBuffer> mQueue = new ArrayBlockingQueue<>(12);
private ScheduledFuture<?> mProcessorTaskHandle;
public static final String FRAME_RATE_PROPERTY = "spectral.display.frame.rate";
private static final Logger mLog = LoggerFactory.getLogger(DFTProcessor.class);
private static final int BUFFER_QUEUE_MAX_SIZE = 12;
private static final int BUFFER_QUEUE_OVERFLOW_RESET_THRESHOLD = 6;
private static final String FRAME_RATE_PROPERTY = "spectral.display.frame.rate";
//The Cosine and Hanning windows seem to offer the best spectral display with minimal bin leakage/smearing
private WindowType mWindowType = Window.WindowType.HANNING;
private double[] mWindow;
private DFTSize mDFTSize = DFTSize.FFT04096;
private DFTSize mNewDFTSize = DFTSize.FFT04096;
private double[] mWindow;
/* The Cosine and Hanning windows seem to offer the best spectral display
* with minimal bin leakage/smearing */
private WindowType mWindowType = Window.WindowType.HANNING;
private FloatFFT_1D mFFT = new FloatFFT_1D(mDFTSize.getSize());
private int mFrameRate;
private int mSampleRate;
private int mFFTFloatsPerFrame;
private float mNewFloatsPerFrame;
private float mNewFloatResidual;
private float[] mPreviousFrame = new float[8192];
private float[] mCurrentBuffer;
private int mCurrentBufferPointer = 0;
private int mSampleRate = 2400000; //Initial high value until we receive update from tuner
private int mFrameSize;
private int mFrameFlushCount;
private int mFrameOverlapCount;
private SampleType mSampleType;
private AtomicBoolean mRunning = new AtomicBoolean();
private ScheduledFuture<?> mProcessorTaskHandle;
private CopyOnWriteArrayList<DFTResultsConverter> mListeners = new CopyOnWriteArrayList<DFTResultsConverter>();
private OverflowableBufferStream mOverflowableBufferStream = new OverflowableBufferStream(BUFFER_QUEUE_MAX_SIZE,
BUFFER_QUEUE_OVERFLOW_RESET_THRESHOLD, mDFTSize.getSize());
private float[] mPreviousSamples;
public DFTProcessor(SampleType sampleType)
{
@@ -101,9 +79,8 @@ public class DFTProcessor implements Listener<ComplexBuffer>, IFrequencyChangePr
stop();
mListeners.clear();
mQueue.clear();
mOverflowableBufferStream.clear();
mWindow = null;
mCurrentBuffer = null;
}
public WindowType getWindowType()
@@ -161,14 +138,10 @@ public class DFTProcessor implements Listener<ComplexBuffer>, IFrequencyChangePr
public void setFrameRate(int framesPerSecond)
{
//TODO: make sure frame rate & sample rate sample requirement doesn't
//expect overlap greater than the previous frame length
if(framesPerSecond < 1 || framesPerSecond > 1000)
{
throw new IllegalArgumentException("DFTProcessor cannot run "
+ "more than 1000 times per second -- requested setting:"
+ framesPerSecond);
throw new IllegalArgumentException("DFTProcessor cannot run more than 1000 times per second -- requested " +
"setting:" + framesPerSecond);
}
mFrameRate = framesPerSecond;
@@ -205,168 +178,46 @@ public class DFTProcessor implements Listener<ComplexBuffer>, IFrequencyChangePr
start();
}
public int getCalculationsPerSecond()
{
return mFrameRate;
}
/**
* Places the sample into a transfer queue for future processing.
*/
@Override
public void receive(ComplexBuffer sampleBuffer)
{
mQueue.offer(sampleBuffer);
}
private void getNextBuffer()
{
mCurrentBuffer = null;
try
{
Buffer buffer = mQueue.take();
mCurrentBuffer = buffer.getSamples();
}
catch(InterruptedException e)
{
mCurrentBuffer = null;
}
mCurrentBufferPointer = 0;
}
private float[] getSamples()
{
int remaining = (int) mFFTFloatsPerFrame;
float[] currentFrame = new float[remaining];
int currentFramePointer = 0;
float integralFloatsToConsume = mNewFloatsPerFrame + mNewFloatResidual;
int newFloatsToConsumeThisFrame = (int) integralFloatsToConsume;
mNewFloatResidual = integralFloatsToConsume - newFloatsToConsumeThisFrame;
/* If the number of required floats for the fft is greater than the
* consumption rate per frame, we have to reach into the previous
* frame to makeup the difference. */
if(newFloatsToConsumeThisFrame < remaining)
{
int previousFloatsRequired = remaining - newFloatsToConsumeThisFrame;
System.arraycopy(mPreviousFrame,
mPreviousFrame.length - previousFloatsRequired,
currentFrame,
currentFramePointer,
previousFloatsRequired);
remaining -= previousFloatsRequired;
currentFramePointer += previousFloatsRequired;
}
/* Fill the rest of the buffer with new samples */
while(mRunning.get() && remaining > 0)
{
if(mCurrentBuffer == null ||
mCurrentBufferPointer >= mCurrentBuffer.length)
{
getNextBuffer();
}
/* If we don't have new samples to use, send the current frame with
* the remaining values as zero */
if(mCurrentBuffer == null)
{
remaining = 0;
}
else
{
int samplesAvailable = mCurrentBuffer.length - mCurrentBufferPointer;
while(remaining > 0 && samplesAvailable > 0)
{
currentFrame[currentFramePointer++] =
(float) mCurrentBuffer[mCurrentBufferPointer++];
samplesAvailable--;
remaining--;
newFloatsToConsumeThisFrame--;
}
}
}
/* If the incoming float rate is greater than the fft consumption rate,
* then we have to purge some floats, otherwise, store the previous
* frame, because we have overlapping frames */
if(newFloatsToConsumeThisFrame > 0)
{
purge(newFloatsToConsumeThisFrame);
}
else
{
mPreviousFrame = Arrays.copyOf(currentFrame, currentFrame.length);
}
return currentFrame;
mOverflowableBufferStream.offer(sampleBuffer);
}
private void calculate()
{
float[] samples = getSamples();
Window.apply(mWindow, samples);
if(mSampleType == SampleType.REAL)
try
{
mFFT.realForward(samples);
}
else
{
mFFT.complexForward(samples);
}
dispatch(samples);
}
private void purge(int samplesToPurge)
{
if(samplesToPurge <= 0)
{
throw new IllegalArgumentException("DFTProcessor - cannot purge "
+ "negative sample amount");
}
while(mRunning.get() && samplesToPurge > 0)
{
if(mCurrentBuffer == null ||
mCurrentBufferPointer >= mCurrentBuffer.length)
if(mFrameFlushCount > 0)
{
getNextBuffer();
mOverflowableBufferStream.flush(mFrameFlushCount);
}
if(mCurrentBuffer != null)
{
int samplesAvailable = mCurrentBuffer.length - mCurrentBufferPointer;
//If this throws an IO exception, the buffer queue is (temporarily) empty and we return from the method
float[] samples = mOverflowableBufferStream.get(mFrameSize, mFrameOverlapCount);
if(samplesAvailable >= samplesToPurge)
{
mCurrentBufferPointer += samplesToPurge;
samplesToPurge = 0;
}
else
{
samplesToPurge -= samplesAvailable;
mCurrentBufferPointer = mCurrentBuffer.length;
}
Window.apply(mWindow, samples);
if(mSampleType == SampleType.REAL)
{
mFFT.realForward(samples);
}
else
{
samplesToPurge = 0;
mFFT.complexForward(samples);
}
dispatch(samples);
mPreviousSamples = samples;
}
catch(IOException ioe)
{
//No new data, dispatch the previous samples again
dispatch(mPreviousSamples);
}
}
@@ -389,11 +240,6 @@ public class DFTProcessor implements Listener<ComplexBuffer>, IFrequencyChangePr
mListeners.add(listener);
}
public void removeConverter(DFTResultsConverter listener)
{
mListeners.remove(listener);
}
private class DFTCalculationTask implements Runnable
{
@Override
@@ -433,22 +279,13 @@ public class DFTProcessor implements Listener<ComplexBuffer>, IFrequencyChangePr
setWindowType(mWindowType);
if(mSampleType == SampleType.COMPLEX)
{
mPreviousFrame = new float[mDFTSize.getSize() * 2];
}
else
{
mPreviousFrame = new float[mDFTSize.getSize()];
}
mFFT = new FloatFFT_1D(mDFTSize.getSize());
}
}
public void clearBuffer()
{
mQueue.clear();
mOverflowableBufferStream.clear();
}
@Override
@@ -466,17 +303,69 @@ public class DFTProcessor implements Listener<ComplexBuffer>, IFrequencyChangePr
}
/**
*
* Calculates the frame size, flush count and overlap count to use for each calculation cycle.
*/
private void calculateConsumptionRate()
{
mNewFloatResidual = 0.0f;
int floatsPerSample = mSampleType == SampleType.COMPLEX ? 2 : 1;
mNewFloatsPerFrame = ((float) mSampleRate / (float) mFrameRate) *
(mSampleType == SampleType.COMPLEX ? 2.0f : 1.0f);
mFrameSize = mDFTSize.getSize() * floatsPerSample;
mPreviousSamples = new float[mFrameSize];
mFFTFloatsPerFrame = (mSampleType == SampleType.COMPLEX ?
mDFTSize.getSize() * 2 :
mDFTSize.getSize());
int productionRate = mSampleRate * floatsPerSample;
int consumptionRate = mFrameRate * mFrameSize;
int residual = productionRate - consumptionRate;
if(residual < 0)
{
mFrameFlushCount = 0;
mFrameOverlapCount = -residual / mFrameRate;
if(mFrameOverlapCount * mFrameRate < -residual)
{
mFrameOverlapCount ++;
}
}
else if(residual > 0)
{
mFrameOverlapCount = 0;
mFrameFlushCount = residual / mFrameRate;
if(mFrameFlushCount * mFrameRate < residual)
{
mFrameFlushCount++;
}
}
//Ensure we're flushing or overlapping by even multiples of the sample size
if(mFrameOverlapCount % floatsPerSample != 0)
{
mFrameOverlapCount++;
}
if(mFrameFlushCount % floatsPerSample != 0)
{
mFrameFlushCount++;
}
//If the overlap size is greater than the frame size, we can't do that. Automatically decrease the frame rate
//until we reach a legitimate overlap value.
if(mFrameOverlapCount >= mFrameSize)
{
mLog.warn("Unable to provide frame rate [" + mFrameRate + "] for current DFT size [" + mDFTSize.getSize() +
"] - reducing frame rate");
mFrameRate--;
if(mFrameRate <= 0)
{
//Consider reducing the FFT size as a possible alternative corrective measure
throw new IllegalStateException("Unable to determine a viable frame rate based on incoming sample " +
"rate, DFT size and DFT frame rate");
}
//Recursively call this method to test the new frame rate
calculateConsumptionRate();
}
}
}

202
src/main/java/io/github/dsheirer/spectrum/OverflowableBufferStream.java Normal file
View File

@@ -0,0 +1,202 @@
/*******************************************************************************
* sdr-trunk
* Copyright (C) 2014-2018 Dennis Sheirer
*
* This program is free software: you can redistribute it and/or modify it under the terms of the GNU General Public
* License as published by the Free Software Foundation, either version 3 of the License, or (at your option) any
* later version.
*
* This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied
* warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License along with this program.
* If not, see <http://www.gnu.org/licenses/>
*
******************************************************************************/
package io.github.dsheirer.spectrum;
import io.github.dsheirer.sample.Buffer;
import io.github.dsheirer.sample.OverflowableTransferQueue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.nio.FloatBuffer;
import java.util.LinkedList;
public class OverflowableBufferStream extends OverflowableTransferQueue<Buffer>
{
private static final Logger mLog = LoggerFactory.getLogger(OverflowableBufferStream.class);
private int mFlushCount = 0;
private Buffer mCurrentBuffer;
private int mCurrentBufferPointer = 0;
private FloatBuffer mFloatBuffer;
private LinkedList<Buffer> mBufferList = new LinkedList<>();
private int mBufferFetchLimit;
/**
* Stream for receiving Buffers from a producer thread and providing access to a consumer thread to read the
* sample data. Extends the overflowable transfer queue to define maximum and reset thresholds to ensure the
* queue doesn't grow too large.
*
* This stream is designed for consistent sample size requests each time the get() samples method is invoked, such
* as might be needed for an FFT processor. It uses an internal float buffer that must be resized whenever the
* get(sample size) argument changes, so consider using another option if you anticipate that your sample size
* argument during get() operations will vary.
*
* @param maxSize of the internal buffer queue - defines the overflow threshold
* @param resetThreshold of the internal buffer queue - defines when the overflow state will be cleared once the
* queue size has reduced below this threshold
* @param arraySize of the anticipated samples to fetch using the get(x,y) method.
*/
public OverflowableBufferStream(int maxSize, int resetThreshold, int arraySize)
{
super(maxSize, resetThreshold);
mFloatBuffer = FloatBuffer.allocate(arraySize);
mBufferFetchLimit = maxSize = resetThreshold;
}
/**
* Retrieves the specified number of float samples from the stream. When a non-zero overlap is specified, a
* quantity of samples from the previous buffer are preloaded in the buffer and new samples are added to make a
* full array count.
*
* @param sampleCount of float samples to pull from the stream
* @param overlap to reuse part of the previous buffer's samples in this fetch - must be less than sample count
* @return an array of float samples from the stream
* @throws IOException if the buffer queue is/becomes empty and the samples cannot be provided
* @throws IllegalArgumentException if the overlap argument is not less than the sample count argument
*/
public float[] get(int sampleCount, int overlap) throws IOException
{
if(overlap >= sampleCount)
{
throw new IllegalArgumentException("Overlap must be less than the requested sample count");
}
if(mCurrentBuffer == null)
{
//This will throw an IO exception if the queue is empty
getNextBuffer();
}
//Resize the float buffer if the request size changes
if(mFloatBuffer.capacity() != sampleCount)
{
mFloatBuffer = FloatBuffer.allocate(sampleCount);
}
while(mFloatBuffer.hasRemaining())
{
int available = mCurrentBuffer.getSamples().length - mCurrentBufferPointer;
if(available <= 0)
{
getNextBuffer();
}
else
{
//Fill the buffer to capacity
int toCopy = mFloatBuffer.remaining();
if(available < toCopy)
{
toCopy = available;
}
if(toCopy > 0)
{
mFloatBuffer.put(mCurrentBuffer.getSamples(), mCurrentBufferPointer, toCopy);
mCurrentBufferPointer += toCopy;
}
}
}
//If we get to here, the float buffer is full. Get the samples and then refill with overlap if non-zero
mFloatBuffer.rewind();
float[] samples = new float[sampleCount];
mFloatBuffer.get(samples);
mFloatBuffer.clear();
if(overlap > 0)
{
mFloatBuffer.put(samples, samples.length - overlap, overlap);
}
try
{
performFlush();
}
catch(IOException ioe)
{
//Do nothing if we get an IO exception ... the buffer is empty
}
//Reset flush count to zero, even if we ran out of buffers
mFlushCount = 0;
return samples;
}
/**
* Throws away incoming sample data until mFlushCount is zeroized or the queue is empty.
*
* @throws IOException if the queue is emptied while performing flush.
*/
private void performFlush() throws IOException
{
//Flush sample data as requested
while(mFlushCount > 0)
{
int available = mCurrentBuffer.getSamples().length - mCurrentBufferPointer;
if(available <= mFlushCount)
{
mFlushCount -= available;
getNextBuffer();
}
else
{
mCurrentBufferPointer += mFlushCount;
mFlushCount = 0;
}
}
}
/**
* Flushes the specified number of samples from the incoming stream.
* @param sampleCount to throw away
*/
public void flush(int sampleCount)
{
mFlushCount += sampleCount;
}
/**
* Fetches the next buffer from the queue.
*
* @throws IOException if the queue is currently empty
*/
private void getNextBuffer() throws IOException
{
mCurrentBuffer = null;
mCurrentBufferPointer = 0;
if(mBufferList.isEmpty())
{
drainTo(mBufferList, mBufferFetchLimit);
}
if(!mBufferList.isEmpty())
{
mCurrentBuffer = mBufferList.poll();
}
if(mCurrentBuffer == null)
{
throw new IOException("Buffer queue is (currently) empty");
}
}
}