From d5b377b1af8d4af1775d9f3de3f0403003012803 Mon Sep 17 00:00:00 2001 From: Denny Date: Sun, 14 Jan 2018 12:05:39 -0500 Subject: [PATCH] Resolves #318. Redesigns DFT processor to use a non-blocking queue approach to access incoming buffer streams coupled with a overflowable queue that limits the quantity of complex buffers that are cached for the DFT processor. --- .../dsheirer/spectrum/DFTProcessor.java | 335 ++++++------------ .../spectrum/OverflowableBufferStream.java | 202 +++++++++++ 2 files changed, 314 insertions(+), 223 deletions(-) create mode 100644 src/main/java/io/github/dsheirer/spectrum/OverflowableBufferStream.java diff --git a/src/main/java/io/github/dsheirer/spectrum/DFTProcessor.java b/src/main/java/io/github/dsheirer/spectrum/DFTProcessor.java index b69628be..6e32e222 100644 --- a/src/main/java/io/github/dsheirer/spectrum/DFTProcessor.java +++ b/src/main/java/io/github/dsheirer/spectrum/DFTProcessor.java @@ -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 + * You should have received a copy of the GNU General Public License along with this program. + * If not, see * ******************************************************************************/ 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, IFrequencyChangeProcessor, IDFTWidthChangeProcessor { - private final static Logger mLog = LoggerFactory.getLogger(DFTProcessor.class); - - private CopyOnWriteArrayList mListeners = new CopyOnWriteArrayList(); - - //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 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 mListeners = new CopyOnWriteArrayList(); + 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, IFrequencyChangePr stop(); mListeners.clear(); - mQueue.clear(); + mOverflowableBufferStream.clear(); mWindow = null; - mCurrentBuffer = null; } public WindowType getWindowType() @@ -161,14 +138,10 @@ public class DFTProcessor implements Listener, 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, 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, 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, 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, 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(); + } } } diff --git a/src/main/java/io/github/dsheirer/spectrum/OverflowableBufferStream.java b/src/main/java/io/github/dsheirer/spectrum/OverflowableBufferStream.java new file mode 100644 index 00000000..0faec35c --- /dev/null +++ b/src/main/java/io/github/dsheirer/spectrum/OverflowableBufferStream.java @@ -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 + * + ******************************************************************************/ +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 +{ + 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 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"); + } + } +}