From e971ad4582b6003d709e1a745b80b38224508a58 Mon Sep 17 00:00:00 2001 From: Denny Date: Sun, 15 Apr 2018 09:05:06 -0400 Subject: [PATCH] 39 [WIP] - restructured the polyphase filter process. Previous version was not computationally feasible with my 3rd Gen i7 processor running and airspy at 10 MHz. Restructured the polyphase channelizer to use vector iteration structures that the JDK 9 JIT compiler can translate into SIMD instructions. Separated the IFFT processing from the filter process so that they can be executed on separate threads and this allows the processing to occur real time. P25 C4FM decoding at 25 kHz channel sample rate using an airspy 10MHz sample stream seems to be working nicely with ~25% CPU usage and ~300 to 500 MB memory usage. Incremental (parallel) channel decoding should have a minimal increase in CPU loading since all of the 12.5 kHz channels are available in the polyphase channelizer and the only additional workload is the decoding of the 25 kHz sample stream. There is still an issue with stopping channels and the channelizer keeps processing, and occasionally when starting traffic channels something deadlocks and causes out of memory. --- .../AbstractComplexPolyphaseChannelizer.java | 19 +- .../ComplexPolyphaseChannelizerM2.java | 652 ++++-------------- .../ContinuousBufferProcessor.java | 23 +- .../channelizer/PolyphaseChannelManager.java | 28 +- .../channelizer/PolyphaseChannelSource.java | 3 +- .../output/ChannelOutputProcessor.java | 10 +- .../IPolyphaseChannelOutputProcessor.java | 4 +- .../output/OneChannelOutputProcessor.java | 8 +- .../output/TwoChannelOutputProcessor.java | 8 +- .../sample/OverflowableTransferQueue.java | 17 + .../sample/buffer/AbstractReusableBuffer.java | 151 ++++ .../buffer/AbstractReusableBufferQueue.java | 57 +- .../IReusableBufferDisposedListener.java | 2 +- .../sample/buffer/ReusableBuffer.java | 113 +-- .../sample/buffer/ReusableBufferQueue.java | 23 +- .../buffer/ReusableChannelResultsBuffer.java | 143 ++++ .../ReusableChannelResultsBufferQueue.java | 55 ++ .../buffer/ReusableComplexBufferQueue.java | 23 +- 18 files changed, 637 insertions(+), 702 deletions(-) create mode 100644 src/main/java/io/github/dsheirer/sample/buffer/AbstractReusableBuffer.java create mode 100644 src/main/java/io/github/dsheirer/sample/buffer/ReusableChannelResultsBuffer.java create mode 100644 src/main/java/io/github/dsheirer/sample/buffer/ReusableChannelResultsBufferQueue.java diff --git a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/AbstractComplexPolyphaseChannelizer.java b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/AbstractComplexPolyphaseChannelizer.java index e044b2af..8a7175bf 100644 --- a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/AbstractComplexPolyphaseChannelizer.java +++ b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/AbstractComplexPolyphaseChannelizer.java @@ -20,6 +20,8 @@ package io.github.dsheirer.dsp.filter.channelizer; import io.github.dsheirer.sample.Broadcaster; import io.github.dsheirer.sample.Listener; +import io.github.dsheirer.sample.buffer.ReusableChannelResultsBuffer; +import io.github.dsheirer.sample.buffer.ReusableChannelResultsBufferQueue; import io.github.dsheirer.sample.buffer.ReusableComplexBuffer; import io.github.dsheirer.source.ISourceEventListener; import io.github.dsheirer.source.SourceEvent; @@ -32,7 +34,7 @@ import java.util.concurrent.CopyOnWriteArrayList; public abstract class AbstractComplexPolyphaseChannelizer implements Listener, ISourceEventListener { private final static Logger mLog = LoggerFactory.getLogger(AbstractComplexPolyphaseChannelizer.class); - + private ReusableChannelResultsBufferQueue mBufferQueue = new ReusableChannelResultsBufferQueue(); private Broadcaster mSourceChangeBroadcaster = new Broadcaster(); private List mChannels = new CopyOnWriteArrayList<>(); private double mSampleRate; @@ -102,12 +104,15 @@ public abstract class AbstractComplexPolyphaseChannelizer implements Listener getSourceEventListener() { diff --git a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/ComplexPolyphaseChannelizerM2.java b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/ComplexPolyphaseChannelizerM2.java index 1293e153..a6be055a 100644 --- a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/ComplexPolyphaseChannelizerM2.java +++ b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/ComplexPolyphaseChannelizerM2.java @@ -22,6 +22,7 @@ import io.github.dsheirer.dsp.filter.FilterFactory; import io.github.dsheirer.dsp.filter.Window.WindowType; import io.github.dsheirer.dsp.filter.design.FilterDesignException; import io.github.dsheirer.sample.Listener; +import io.github.dsheirer.sample.buffer.ReusableChannelResultsBuffer; import io.github.dsheirer.sample.buffer.ReusableComplexBuffer; import io.github.dsheirer.sample.buffer.ReusableComplexBufferQueue; import io.github.dsheirer.sample.real.IOverflowListener; @@ -30,8 +31,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.Arrays; -import java.util.HashMap; -import java.util.Map; +import java.util.List; public class ComplexPolyphaseChannelizerM2 extends AbstractComplexPolyphaseChannelizer { @@ -39,46 +39,48 @@ public class ComplexPolyphaseChannelizerM2 extends AbstractComplexPolyphaseChann private static final int DEFAULT_MINIMUM_CHANNEL_BANDWIDTH = 12500; - private float[][] mSamples; - private float[][] mCoefficients; - private int[][] mSampleIndexMap; - private int mColumnPointer; - private int mRowPointer; - private int mBlockSize; - private int mSamplesPerBlock; + //Sized at 152 buffers a second where max = 5 seconds and reset = 2 seconds worth of buffers + private IFFTProcessor mIFFTProcessor = new IFFTProcessor((5 * 152), (2 * 152)); private FloatFFT_1D mFFT; - private PolyphaseChannelResultsBuffer mChannelResultsBuffer; - private int mChannelResultsBufferSize = 2500; - private SampleTimestampManager mTimestampManager; - private IFFTProcessor mIFFTProcessor = new IFFTProcessor(300, 50); - - private Map mSampleToFilterMap = new HashMap<>(); - - private int[][] mTopBlockAccumulatorIndexMap; - private int[][] mMiddleBlockAccumulatorIndexMap; - private int[] mTopBlockMap; - private int[] mMiddleBlockMap; private float[] mInlineSamples; private float[] mInlineFilter; private float[] mInlineInterimOutput; - private float[] mChannelAccumulator; - private int mSampleBufferPointer; - private int mTapsPerChannel; + private float[] mFilterAccumulator; private boolean mTopBlockIndicator = true; - + private int[] mTopBlockMap; + private int[] mMiddleBlockMap; + private int mSampleBufferPointer; + private int mSamplesPerBlock; + private int mTapsPerChannel; /** * Non-Maximally Decimated Polyphase Filter Bank (NMDPFB) channelizer that divides the input frequency band into * equal bandwidth channels that are each oversampled by 2x for output. * + * This polyphase channelizer is based off of the channelizer described by Fred Harris in Multirate Signal + * Processing for Communications Systems, p230-233. + * + * Samples are loaded into this filter one block at a time (1/2 channel count) and a filtered output is calculated + * to produce an overall 2x oversampled channel sample rate. Each sample block load is preceded by a serpentine + * shift of the existing sample blocks. We use the java System.arrayCopy() method which is able to leverage + * native processor intrinsics for efficiency. + * + * The prototype filter for the channelizer is rearranged to align with the structure of the sample buffer. + * + * Instead of using an array of channel filters as described in the Harris text, this filter and the sample buffer + * are arranged as a contiguous array to maximize Java's ability to leverage native processor Single Instruction + * Multiple Data (SIMD) intrinsics (since Java 8). The filter process is broken into three steps: + * -Multiply the inline array of samples and filter coefficients + * -Accumulate the results for each sub-channel + * -Rearrange the sub-channel results to correctly order the sub-channels + * * @param taps of a low-pass filter designed for the inbound sample rate with a cutoff frequency * equal to the channel bandwidth (sample rate / filters). If you need to synthesize (combine two or more - * filter outputs) a new bandwidth signal from the outputs of this filter, then the filter should be designed + * channel outputs) a new bandwidth signal from the outputs of this filter, then the filter should be designed * as a nyquist filter with -6 dB attenuation at the channel bandwidth cutoff frequency - * * @param sampleRate of the incoming sample stream - * @param channelCount - number of filters/channels to output. Since this filter bank oversamples each filter - * output, this number must be even (divisible by the 2x oversample rate). + * @param channelCount - number of filters/channels to output. Since this filter bank performs 2x oversampling for + * each channel output, this number must be even (divisible by 2). */ public ComplexPolyphaseChannelizerM2(float[] taps, int sampleRate, int channelCount) { @@ -89,8 +91,7 @@ public class ComplexPolyphaseChannelizerM2 extends AbstractComplexPolyphaseChann throw new IllegalArgumentException("Channel count must be an even multiple of the over-sample rate (2x)"); } - double oversampledChannelSampleRate = getChannelSampleRate() * 2.0; - mTimestampManager = new SampleTimestampManager(oversampledChannelSampleRate); + mTapsPerChannel = (int)Math.ceil((double)taps.length / (double)channelCount); init(taps); } @@ -99,22 +100,34 @@ public class ComplexPolyphaseChannelizerM2 extends AbstractComplexPolyphaseChann * Non-Maximally Decimated Polyphase Filter Bank (NMDPFB) channelizer that divides the input frequency band into * equal bandwidth channels that are each oversampled by 2x for output. * - * Note: this constructor uses a default minimum channel bandwidth of 12.5 kHz. The number of channels is calculate - * from the sample rate and this default minimum channel bandwidth. The resulting per-channel bandwidth will be - * between 12.5 and 25.0 kHz with the output per-channel sample rate of 2x the bandwidth. + * This polyphase channelizer is based off of the channelizer described by Fred Harris in Multirate Signal + * Processing for Communications Systems, p230-233. + * + * Samples are loaded into this filter one block at a time (1/2 channel count) and a filtered output is calculated + * to produce an overall 2x oversampled channel sample rate. Each sample block load is preceded by a serpentine + * shift of the existing sample blocks. We use the java System.arrayCopy() method which is able to leverage + * native processor intrinsics for efficiency. + * + * The prototype filter for the channelizer is rearranged to align with the structure of the sample buffer. + * + * Instead of using an array of channel filters as described in the Harris text, this filter and the sample buffer + * are arranged as a contiguous array to maximize Java's ability to leverage native processor Single Instruction + * Multiple Data (SIMD) intrinsics (since Java 8). The filter process is broken into three steps: + * -Multiply the inline array of samples and filter coefficients + * -Accumulate the results for each sub-channel + * -Rearrange the sub-channel results to correctly order the sub-channels * * @param sampleRate to be channelized. - * @param tapsPerFilter to use when designing the filter + * @param tapsPerChannel to use when designing the filter */ - public ComplexPolyphaseChannelizerM2(double sampleRate, int tapsPerFilter) throws FilterDesignException + public ComplexPolyphaseChannelizerM2(double sampleRate, int tapsPerChannel) throws FilterDesignException { super(sampleRate, getChannelCount(sampleRate)); - float[] filterTaps = FilterFactory.getSincM2Channelizer(getChannelSampleRate(), getChannelCount(), - tapsPerFilter, WindowType.BLACKMAN_HARRIS_7, true); + mTapsPerChannel = tapsPerChannel; - double oversampledChannelSampleRate = getChannelSampleRate() * 2.0; - mTimestampManager = new SampleTimestampManager(oversampledChannelSampleRate); + float[] filterTaps = FilterFactory.getSincM2Channelizer(getChannelSampleRate(), getChannelCount(), + mTapsPerChannel, WindowType.BLACKMAN_HARRIS_7, true); init(filterTaps); } @@ -161,7 +174,7 @@ public class ComplexPolyphaseChannelizerM2 extends AbstractComplexPolyphaseChann super.setSampleRate(sampleRate); float[] filterTaps = FilterFactory.getSincM2Channelizer(getChannelSampleRate(), getChannelCount(), - 15, WindowType.BLACKMAN_HARRIS_7, true); + mTapsPerChannel, WindowType.BLACKMAN_HARRIS_7, true); init(filterTaps); } @@ -172,46 +185,6 @@ public class ComplexPolyphaseChannelizerM2 extends AbstractComplexPolyphaseChann } } - /** - * Receives the complex sample buffer and processes the results through the channelizer. - */ - public void receiveOld(ReusableComplexBuffer reusableComplexBuffer) - { - long start = System.nanoTime(); - - //Use the buffer's reference timestamp to update our timestamp manager (for timestamping output buffers) - mTimestampManager.setReferenceTimestamp(reusableComplexBuffer.getTimestamp()); - - float[] samples = reusableComplexBuffer.getSamples(); - - int samplesPointer = 0; - - while(samplesPointer < samples.length) - { - int lengthToCopy = mBlockSize - (mRowPointer % mBlockSize); - - if((samplesPointer + lengthToCopy) > samples.length) - { - lengthToCopy = samples.length - samplesPointer; - } - - System.arraycopy(samples, samplesPointer, mSamples[mColumnPointer], mRowPointer, lengthToCopy); - samplesPointer += lengthToCopy; - mRowPointer += lengthToCopy; - - if(mRowPointer % mBlockSize == 0) - { - //if row pointer at max, then mRowPointer = 0 and mColumnPointer-- - processOld(); - } - } - - //Decrement the user count to let the originator know we're done with their buffer - reusableComplexBuffer.decrementUserCount(); - - mLog.debug("Duration: " + (System.nanoTime() - start)); - } - /** * Receives the complex sample buffer and processes the results through the channelizer. */ @@ -221,8 +194,8 @@ public class ComplexPolyphaseChannelizerM2 extends AbstractComplexPolyphaseChann long start = System.nanoTime(); long processing = 0; - //Use the buffer's reference timestamp to update our timestamp manager (for timestamping output buffers) - mTimestampManager.setReferenceTimestamp(reusableComplexBuffer.getTimestamp()); + ReusableChannelResultsBuffer channelResultsBuffer = getChannelResultsBuffer(); + channelResultsBuffer.setTimestamp(reusableComplexBuffer.getTimestamp()); float[] samples = reusableComplexBuffer.getSamples(); @@ -249,7 +222,7 @@ public class ComplexPolyphaseChannelizerM2 extends AbstractComplexPolyphaseChann if(mSampleBufferPointer >= mSamplesPerBlock) { //Filter buffered samples and produce a single sample across each of the polyphase channels - processing += process(); + processing += process(channelResultsBuffer); //Right-shift the samples in the buffer over to make room for a new block of samples //Note: since JDK 8, hotspot JIT compiler uses native processor intrinsics for efficiency @@ -258,78 +231,14 @@ public class ComplexPolyphaseChannelizerM2 extends AbstractComplexPolyphaseChann } } - mLog.debug("Duration: " + (System.nanoTime() - start) + " Processing:" + processing); + channelResultsBuffer.incrementUserCount(); + + //Enqueue the channel results buffer for IFFT processing and distribution on a different thread + mIFFTProcessor.receive(channelResultsBuffer); + +// mLog.debug("Duration: " + (System.nanoTime() - start) + " Processing:" + processing); //Decrement the user count to let the originator know we're done with their buffer reusableComplexBuffer.decrementUserCount(); - - } - - /** - * Creates a base accumulator index map that is preloaded with the indexes that have to be summed to produce - * each subchannel filtered value. This index map structure is designed for processing of polyphase filter - * samples in a contiguous/single sample array where the filter taps are also arranged to align with this - * inline sample array buffer. This method preloads the index values, but the first dimension of the array - * must be rearranged to align with either top-block or middle-block processing so that each of the - * polyphase filter arm outputs is assigned to each sub-channel (I/Q) correctly. - * - * @param channelCount - number of channels - * @param tapsPerChannel - number of filter taps per channel - * @return output index to filter accumulator index mapping - */ - private static int[][] getBaseAccumulatorIndexMap(int channelCount, int tapsPerChannel) - { - int subChannelCount = channelCount * 2; - - int[][] accumulatorIndexMap = new int[subChannelCount][tapsPerChannel]; - - int pointer = 0; - - //Step 1: load the interim product output indexes in sequential order. - for(int column = 0; column < tapsPerChannel; column++) - { - for(int subChannel = 0; subChannel < subChannelCount; subChannel++) - { - accumulatorIndexMap[subChannel][column] = pointer++; - } - } - - return accumulatorIndexMap; - } - - /** - * Creates a top-block processing accumulator map that maps each interim filter and sample index product - * to the corresponding final output index for the array that will feed the IFFT. - * - * @param channelCount - number of channels - * @param tapsPerChannel - number of filter taps per channel - * @return output index to filter accumulator index mapping - */ - private static int[][] getTopBlockAccumulatorIndexMap(int channelCount, int tapsPerChannel) - { - int[][] baseIndexMap = getBaseAccumulatorIndexMap(channelCount, tapsPerChannel); - - //SubChannel count is the number if I/Q channels (ie channels x 2) - int subChannelCount = channelCount * 2; - - //Reorder the subchannel arrays to the structure needed for top-block processing - int blockSize = channelCount / 2; - - int[][] reorderedMap = new int[subChannelCount][tapsPerChannel]; - - for(int channel = 0; channel < blockSize; channel++) - { - int newIndex = 2 * channel; - int originalIndex = 2 * (blockSize - channel - 1); - int offset = 2 * blockSize; - - reorderedMap[newIndex] = baseIndexMap[originalIndex]; - reorderedMap[newIndex + 1] = baseIndexMap[originalIndex + 1]; - - reorderedMap[offset + newIndex] = baseIndexMap[offset + originalIndex]; - reorderedMap[offset + newIndex + 1] = baseIndexMap[offset + originalIndex + 1]; - } - - return reorderedMap; } /** @@ -389,48 +298,12 @@ public class ComplexPolyphaseChannelizerM2 extends AbstractComplexPolyphaseChann return newMap; } - /** - * Creates a middle-block processing accumulator map that maps each interim filter and sample index product - * to the corresponding final output index for the array that will feed the IFFT. - * - * @param channelCount - number of channels - * @param tapsPerChannel - number of filter taps per channel - * @return output index to filter accumulator index mapping - */ - private static int[][] getMiddleBlockAccumulatorIndexMap(int channelCount, int tapsPerChannel) - { - int[][] baseIndexMap = getBaseAccumulatorIndexMap(channelCount, tapsPerChannel); - - //SubChannel count is the number if I/Q channels (ie channels x 2) - int subChannelCount = channelCount * 2; - - //Reorder the subchannel arrays to the structure needed for middle-block processing, where the outputs - //from the top and bottom blocks are inverted when compared to top-block processing - int blockSize = channelCount / 2; - - int[][] reorderedMap = new int[subChannelCount][tapsPerChannel]; - - for(int channel = 0; channel < blockSize; channel++) - { - int newIndex = 2 * channel; - int originalIndex = 2 * (blockSize - channel - 1); - int offset = 2 * blockSize; - - reorderedMap[newIndex] = baseIndexMap[offset + originalIndex]; - reorderedMap[newIndex + 1] = baseIndexMap[offset + originalIndex + 1]; - - reorderedMap[offset + newIndex] = baseIndexMap[originalIndex]; - reorderedMap[offset + newIndex + 1] = baseIndexMap[originalIndex + 1]; - } - - return reorderedMap; - } /** - * Arranges the filter coefficients for processing the polyphase filter bank using a contiguous array. - * @param coefficients - * @param channelCount - * @return + * Rearranges the filter coefficients to align with a contiguous sample buffer for processing efficiency. + * @param coefficients of the polyphase filter + * @param channelCount number of channels where each channel is an I/Q pair + * @return filter rearranged for inline sample buffer processing */ private static float[] getAlignedFilter(float[] coefficients, int channelCount, int tapsPerChannel) { @@ -467,370 +340,158 @@ public class ComplexPolyphaseChannelizerM2 extends AbstractComplexPolyphaseChann * Processes the sample buffer for each new block of sample data that is loaded and distributes the results to any * registered channel listeners. */ - private void processOld() + private long process(ReusableChannelResultsBuffer channelResultsBuffer) { - mSampleToFilterMap.clear(); - - - int sampleRowCount = getChannelCount() * 2; - - float[] processed = new float[sampleRowCount]; - - - //Top of block processing - if(mRowPointer >= sampleRowCount) - { - mLog.debug("Top Block"); - int[] sampleIndexMap = mSampleIndexMap[mColumnPointer]; - - for(int channel = 0; channel < getChannelCount(); channel++) - { - int outputIndex = 2 * (getChannelCount() - channel - 1); - int sampleIndex = 2 * channel; - - float[] filter = mCoefficients[channel]; - - processed[outputIndex] = filter(outputIndex, sampleIndex, mCoefficients[channel], sampleIndexMap); - processed[outputIndex + 1] = filter((outputIndex + 1), (sampleIndex + 1), mCoefficients[channel], sampleIndexMap); - } - - mRowPointer = 0; - mColumnPointer--; - - if(mColumnPointer < 0) - { - mColumnPointer = mSamples.length - 1; - } - } - //Mid-point processing - else - { - mLog.debug("Middle Block"); - int[] sampleIndexMap = mSampleIndexMap[mColumnPointer]; - - int half = getChannelCount() / 2; - - int outputOffset = getChannelCount() - 1; - int outputIndex; - int sampleIndex; - int filterIndex; - - for(int channel = 0; channel < half; channel++) - { - outputIndex = 2 * (outputOffset - channel); - sampleIndex = 2 * channel; - filterIndex = channel + half; - - processed[outputIndex] = filter(outputIndex, sampleIndex, mCoefficients[filterIndex], sampleIndexMap); - processed[outputIndex + 1] = filter((outputIndex + 1), (sampleIndex + 1), mCoefficients[filterIndex], sampleIndexMap); - } - - //The second half of the sample buffer was loaded using the previous column pointer - adjust for that now - sampleIndexMap = mSampleIndexMap[(mColumnPointer == (mSamples.length - 1) ? 0 : mColumnPointer + 1)]; - outputOffset = getChannelCount() - 1; - - for(int channel = half; channel < getChannelCount(); channel++) - { - outputIndex = 2 * (outputOffset - channel); - sampleIndex = 2 * channel; - filterIndex = channel - half; - - processed[outputIndex] = filter(outputIndex, sampleIndex, mCoefficients[filterIndex], sampleIndexMap); - processed[outputIndex + 1] = filter((outputIndex + 1), sampleIndex + 1, mCoefficients[filterIndex], sampleIndexMap); - } - } - - mLog.debug("Processed:" + Arrays.toString(processed)); - - mIFFTProcessor.receive(processed); - } - - private float[] getAccumulated() - { - return mChannelAccumulator; - } - - /** - * Processes the sample buffer for each new block of sample data that is loaded and distributes the results to any - * registered channel listeners. - */ - private long process() - { - //Calculate vector product of samples and filter coefficients using an iterative loop structure that the - //JIT compiler can easily recognize. - - //Note: since JDK 8, the hotspot JIT compiler can compile this loop to use vector-optimized native processor - //Single Instruction Multiple Data (SIMD) intrinsics, potentially making this loop iteration faster using - //32-bit float SIMD intrinsics (e.g. intel SSE/2/3/4(4x), AVX(8x), or AVX2(16x) instruction sets). + //Multiply each of the samples by the corresponding filter tap for(int x = 0; x < mInlineSamples.length; x++) { mInlineInterimOutput[x] = mInlineSamples[x] * mInlineFilter[x]; } - int offset = 0; - - Arrays.fill(mChannelAccumulator, 0.0f); - long start = System.nanoTime(); - //Accumulate the product results + Arrays.fill(mFilterAccumulator, 0.0f); + + int tapOffset = 0; + + //Accumulate the sample/filter product results into each of the I/Q sub-channels for(int tap = 0; tap < mTapsPerChannel; tap++) { - offset = tap * getSubChannelCount(); + tapOffset = tap * getSubChannelCount(); for(int channel = 0; channel < getSubChannelCount(); channel++) { - mChannelAccumulator[channel] += mInlineInterimOutput[offset + channel]; + mFilterAccumulator[channel] += mInlineInterimOutput[tapOffset + channel]; } } - float[] processed = new float[getSubChannelCount()]; + float[] processed = channelResultsBuffer.getEmptyBuffer(getSubChannelCount()); - //Rearrange the accumulated results if(mTopBlockIndicator) { for(int x = 0; x < getSubChannelCount(); x++) { - processed[x] = mChannelAccumulator[mTopBlockMap[x]]; + processed[x] = mFilterAccumulator[mTopBlockMap[x]]; } } else { for(int x = 0; x < getSubChannelCount(); x++) { - processed[x] = mChannelAccumulator[mMiddleBlockMap[x]]; + processed[x] = mFilterAccumulator[mMiddleBlockMap[x]]; } } - long elapsed = System.nanoTime() - start; - - //Accumulate the filtered channel results from the interim product array according to top/middle block - - //Top of block processing -// if(mTopBlockIndicator) -// { -//// mLog.debug("Top Block"); -// -// //Accumulate the product of each filter coefficient and sample into each sub-channel -// for(int subChannel = 0; subChannel < getSubChannelCount(); subChannel++) -// { -// for(int interimOutputIndex: mTopBlockAccumulatorIndexMap[subChannel]) -// { -// processed[subChannel] += mInlineInterimOutput[interimOutputIndex]; -// } -// } -// } -// //Mid-point processing -// else -// { -//// mLog.debug("Middle Block"); -// -// //Accumulate the product of each filter coefficient and sample into each sub-channel -// for(int subChannel = 0; subChannel < getSubChannelCount(); subChannel++) -// { -// for(int interimOutputIndex: mMiddleBlockAccumulatorIndexMap[subChannel]) -// { -// processed[subChannel] += mInlineInterimOutput[interimOutputIndex]; -// } -// } -// } -// + channelResultsBuffer.addChannelResults(processed); mTopBlockIndicator = !mTopBlockIndicator; -// mLog.debug("Processed:" + Arrays.toString(processed)); -// mIFFTProcessor.receive(processed); - mFFT.complexInverse(processed, true); - - processChannelResults(processed); + long elapsed = System.nanoTime() - start; return elapsed; } - /** - * Buffers the channel results and dispatches the channel results buffer to the channel output processors when full. - * @param channelResults - */ - private void processChannelResults(float[] channelResults) - { - if(mChannelResultsBuffer == null) - { - mChannelResultsBuffer = new PolyphaseChannelResultsBuffer(mTimestampManager.getCurrentTimestamp(), - mChannelResultsBufferSize); - } +// /** +// * Buffers the channel results and dispatches the channel results buffer to the channel output processors when full. +// * @param buffer containing filtered and IFFT'd channelResults with a corresponding timestamp +// */ +// private void processChannelResults(ReusableChannelResultsBuffer buffer) +// { +// if(mChannelResultsBuffer == null) +// { +// mChannelResultsBuffer = new PolyphaseChannelResultsBuffer(buffer.getTimestamp(), +// mChannelResultsBufferSize); +// } +// +// try +// { +// mChannelResultsBuffer.add(buffer.getSamplesCopy()); +// } +// catch(IllegalArgumentException iae) +// { +// //If the buffer is full (unlikely) or the channel results array length has changed (possible), flush the +// //current buffer, create a new one, and store the current results +// flushChannelResultsBuffer(); +// +// mChannelResultsBuffer = new PolyphaseChannelResultsBuffer(buffer.getTimestamp(), mChannelResultsBufferSize); +// +// mChannelResultsBuffer.add(buffer.getSamplesCopy()); +// } +// +// if(mChannelResultsBuffer.isFull()) +// { +// flushChannelResultsBuffer(); +// } +// +// buffer.decrementUserCount(); +// } - try - { - mChannelResultsBuffer.add(channelResults); - } - catch(IllegalArgumentException iae) - { - //If the buffer is full (unlikely) or the channel results array length has changed (possible), flush the - //current buffer, create a new one, and store the current results - flushChannelResultsBuffer(); - - mChannelResultsBuffer = new PolyphaseChannelResultsBuffer(mTimestampManager.getCurrentTimestamp(), - mChannelResultsBufferSize); - - mChannelResultsBuffer.add(channelResults); - } - - if(mChannelResultsBuffer.isFull()) - { - flushChannelResultsBuffer(); - } - - //Each channel results array is equivalent to one sample from our timestamp manager's perspective - mTimestampManager.increment(); - } +// /** +// * Dispatches the non-empty channel results buffer to the channel output processors and nullifies the reference to +// * the buffer so that a new buffer can be created upon receiving the next channel results. +// */ +// private void flushChannelResultsBuffer() +// { +// if(mChannelResultsBuffer != null && !mChannelResultsBuffer.isEmpty()) +// { +// dispatch(mChannelResultsBuffer); +// } +// +// mChannelResultsBuffer = null; +// } /** - * Dispatches the non-empty channel results buffer to the channel output processors and nullifies the reference to - * the buffer so that a new buffer can be created upon receiving the next channel results. - */ - private void flushChannelResultsBuffer() - { - if(mChannelResultsBuffer != null && !mChannelResultsBuffer.isEmpty()) - { - dispatch(mChannelResultsBuffer); - } - - mChannelResultsBuffer = null; - } - - /** - * Calculates the filtered output for polyphase filter arm. - * - * @param sampleRow index pointing to a row in the mSamples array to filter - * @param coefficients to use in filtering the samples - * @param sampleIndexMap to translate the column index offset based on the current sample column pointer - * @return filtered sample for the row - */ - private float filter(int outputIndex, int sampleRow, float[] coefficients, int[] sampleIndexMap) - { - StringBuilder sb = new StringBuilder(); - sb.append("Out[" + outputIndex + "] "); - float accumulator = 0.0f; - - for(int column = 0; column < mSamples.length; column++) - { - float sample = mSamples[sampleIndexMap[column]][sampleRow]; - float coefficient = coefficients[column]; - - mSampleToFilterMap.put((int)sample, (int)coefficient); - - sb.append("S").append((int)sample).append("*F").append((int)coefficient).append(" + "); - accumulator += mSamples[sampleIndexMap[column]][sampleRow] * coefficients[column]; - } - - mLog.debug(sb.toString()); - return accumulator; - } - - - /** - * Initializes the channelizer structures. Distributes the prototype filter coefficients to each of the polyphase - * filters, sets up the sample buffer array and pointers and initializes the FFT. - * - * These structures resemble those described by Fred Harris in Multirate Signal Processing for Communications - * Systems, p230-233. However, instead of swapping the input sample buffers and swapping the output sample array - * on every other processing block, we simply swap the filter coefficient banks and leave the input sample buffer - * and output array indexes aligned. However, the entire structure (sample buffer, coefficients and output array) - * is inverted (row zero index is M-1 and M-1 is zero index) to facilitate easy loading of data into the input - * sample buffer array using Java's efficient System.arrayCopy() utility. - * - * Note: the filter coefficients array is organized (row,column) for ease of accessing each filter arm row during - * convolution. The samples array is organized (column,row) for ease/efficiency in loading each row of data. Also, - * the input sample array implements a circular buffer where the mCoefficientsMap provides a lookup to each sample - * index based on the current value of the mColumnPointer. + * Initializes the channelizer filter structures. * * @param coefficients of the prototype filter for this channelizer */ private void init(float[] coefficients) { - int columnCount = (int)Math.ceil((double)coefficients.length / (double)getChannelCount()); - int filterRowCount = getChannelCount(); - int sampleRowCount = getChannelCount() * 2; - - //Sample and filter coefficient arrays are specifically setup as column/row and row/column - mSamples = new float[columnCount][sampleRowCount]; - mCoefficients = new float[filterRowCount][columnCount]; - - mColumnPointer = columnCount - 1; - mBlockSize = getChannelCount(); - mRowPointer = mBlockSize; - //Setup the FFT mFFT = new FloatFFT_1D(getChannelCount()); - //Setup the polyphase filters - - //Setup the index map for the sample array circular buffer - initIndexMap(columnCount); - - int coefficientIndex = 0; - - //Load filter columns left-2-right (samples are loaded in reverse column order, right-2-left, for easy convolution) - for(int columnIndex = 0; columnIndex < columnCount; columnIndex++) - { - //Load filter rows bottom to top, opposite of sample row loading and sample output ordering - for(int rowIndex = filterRowCount - 1; rowIndex >= 0; rowIndex--) - { - mCoefficients[rowIndex][columnIndex] = coefficients[coefficientIndex++]; - } - } - - mTapsPerChannel = (int)Math.ceil((double)coefficients.length / (double)getChannelCount()); int channelCount = getChannelCount(); int bufferLength = getSubChannelCount() * mTapsPerChannel; mSamplesPerBlock = getChannelCount(); //Same as subChannelCount / 2 - mTopBlockAccumulatorIndexMap = getTopBlockAccumulatorIndexMap(channelCount, mTapsPerChannel); - mMiddleBlockAccumulatorIndexMap = getMiddleBlockAccumulatorIndexMap(channelCount, mTapsPerChannel); mTopBlockMap = getTopBlockMap(channelCount); mMiddleBlockMap = getMiddleBlockMap(channelCount); mInlineFilter = getAlignedFilter(coefficients, channelCount, mTapsPerChannel); mInlineSamples = new float[bufferLength]; mInlineInterimOutput = new float[bufferLength]; - mChannelAccumulator = new float[getSubChannelCount()]; + mFilterAccumulator = new float[getSubChannelCount()]; } /** - * Generates a circular buffer index map to support lookup of the translated index based on the current column - * pointer and the desired sample index. + * Separate threaded processor to receive and enqueue filtered channel results buffers, perform IFFT on each array + * as required to align the phase of each polyphase channel, and then dispatch the results to any registered + * sample consumer channels. */ - private void initIndexMap(int size) - { - mSampleIndexMap = new int[size][size]; - - for(int row = 0; row < size; row++) - { - for(int column = 0;column < size;column++) - { - int offset = row + column; - - mSampleIndexMap[row][column] = ((offset < size) ? offset : offset - size); - } - } - } - - /** - * Processor to enqueue filtered channels, perform IFFT and dispatch the results - */ - public class IFFTProcessor extends ContinuousBufferProcessor + public class IFFTProcessor extends ContinuousBufferProcessor { public IFFTProcessor(int maximumSize, int resetThreshold) { super(maximumSize, resetThreshold); - setListener(new Listener() + //We create a listener interface to receive the buffers from the scheduled thread pool + //dispatcher thread that is part of this continuous buffer processor. We perform an IFFT on each + //channel results array contained in each results buffer and then dispatch the buffer + //so that it can be distributed to each channel listener. + setListener(new Listener>() { @Override - public void receive(float[] floatsToProcess) + public void receive(List buffers) { - //Rotate each of the channels to the correct phase using the IFFT - mFFT.complexInverse(floatsToProcess, true); - processChannelResults(floatsToProcess); + for(ReusableChannelResultsBuffer buffer: buffers) + { + for(float[] channelResults: buffer.getChannelResults()) + { + //Rotate each of the channels to the correct phase using the IFFT + mFFT.complexInverse(channelResults, true); + } + + dispatch(buffer); + } } }); @@ -860,21 +521,6 @@ public class ComplexPolyphaseChannelizerM2 extends AbstractComplexPolyphaseChann ComplexPolyphaseChannelizerM2 channelizer = new ComplexPolyphaseChannelizerM2(taps, sampleRate, channelCount); - int[][] top = channelizer.getTopBlockAccumulatorIndexMap(channelCount, tapsPerChannel); - int[][] middle = channelizer.getMiddleBlockAccumulatorIndexMap(channelCount, tapsPerChannel); - - StringBuilder sb = new StringBuilder(); - sb.append("Middle:\n"); - - for(int x = 0; x < middle.length; x++) - { - sb.append(x).append(": ").append(Arrays.toString(middle[x])).append("\n"); - } - - mLog.debug(sb.toString()); - - mLog.debug(Arrays.toString(getMiddleBlockMap(channelCount))); - ReusableComplexBufferQueue queue = new ReusableComplexBufferQueue(); int sampleCount = channelCount * 2 * tapsPerChannel; diff --git a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/ContinuousBufferProcessor.java b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/ContinuousBufferProcessor.java index 8c6ec27c..9c3a1791 100644 --- a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/ContinuousBufferProcessor.java +++ b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/ContinuousBufferProcessor.java @@ -33,7 +33,7 @@ public class ContinuousBufferProcessor implements Listener private final static Logger mLog = LoggerFactory.getLogger(ContinuousBufferProcessor.class); private OverflowableTransferQueue mQueue; - private Listener mListener; + private Listener> mListener; private ScheduledFuture mScheduledFuture; private AtomicBoolean mRunning = new AtomicBoolean(); @@ -89,7 +89,7 @@ public class ContinuousBufferProcessor implements Listener * Sets or changes the listener to receive buffers from this processor. * @param listener to receive buffers */ - public void setListener(Listener listener) + public void setListener(Listener> listener) { mListener = listener; } @@ -104,7 +104,6 @@ public class ContinuousBufferProcessor implements Listener @Override public void receive(E e) { -// mLog.debug(" Queuing: " + e.toString()); mQueue.offer(e); } @@ -157,28 +156,18 @@ public class ContinuousBufferProcessor implements Listener */ class Processor implements Runnable { - private List mBuffers = new ArrayList<>(); - @Override public void run() { + List buffers = new ArrayList<>(); + try { - mQueue.drainTo(mBuffers, 10); + mQueue.drainTo(buffers); if(mListener != null) { - while(!mBuffers.isEmpty()) - { - for(E buffer: mBuffers) - { -// mLog.debug("Dispatching: " + buffer.toString() + " Count:" + mBuffers.size()); - mListener.receive(buffer); - } - - mBuffers.clear(); - mQueue.drainTo(mBuffers, 10); - } + mListener.receive(buffers); } } catch(Throwable throwable) diff --git a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/PolyphaseChannelManager.java b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/PolyphaseChannelManager.java index 7de07312..dd11711f 100644 --- a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/PolyphaseChannelManager.java +++ b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/PolyphaseChannelManager.java @@ -546,7 +546,7 @@ public class PolyphaseChannelManager implements ISourceEventProcessor * that they can be processed on the buffer processor calling thread, avoiding unnecessary locks on the channelizer * and/or the channel sources and output processors. */ - public class BufferSourceEventMonitor implements Listener + public class BufferSourceEventMonitor implements Listener> { private Queue mQueuedSourceEvents = new ConcurrentLinkedQueue<>(); @@ -561,11 +561,11 @@ public class PolyphaseChannelManager implements ISourceEventProcessor } @Override - public void receive(ReusableComplexBuffer reusableComplexBuffer) + public void receive(List reusableComplexBuffers) { try { - //Process any queued source events before processing the buffer + //Process any queued source events before processing the buffers SourceEvent queuedSourceEvent = mQueuedSourceEvents.poll(); while(queuedSourceEvent != null) @@ -584,20 +584,26 @@ public class PolyphaseChannelManager implements ISourceEventProcessor queuedSourceEvent = mQueuedSourceEvents.poll(); } - if(mPolyphaseChannelizer != null) + for(ReusableComplexBuffer reusableComplexBuffer: reusableComplexBuffers) { - //User count management is handled by the channelizer - mPolyphaseChannelizer.receive(reusableComplexBuffer); - } - else - { - reusableComplexBuffer.decrementUserCount(); + if(mPolyphaseChannelizer != null) + { + //User count management is handled by the channelizer + mPolyphaseChannelizer.receive(reusableComplexBuffer); + } + else + { + reusableComplexBuffer.decrementUserCount(); + } } } catch(Throwable throwable) { mLog.error("Error", throwable); - reusableComplexBuffer.decrementUserCount(); + for(ReusableComplexBuffer buffer: reusableComplexBuffers) + { + buffer.decrementUserCount(); + } } } } diff --git a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/PolyphaseChannelSource.java b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/PolyphaseChannelSource.java index b6a6a99f..29a04521 100644 --- a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/PolyphaseChannelSource.java +++ b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/PolyphaseChannelSource.java @@ -18,6 +18,7 @@ package io.github.dsheirer.dsp.filter.channelizer; import io.github.dsheirer.dsp.filter.channelizer.output.IPolyphaseChannelOutputProcessor; import io.github.dsheirer.sample.Listener; import io.github.dsheirer.sample.buffer.ReusableBufferAssembler; +import io.github.dsheirer.sample.buffer.ReusableChannelResultsBuffer; import io.github.dsheirer.sample.buffer.ReusableComplexBuffer; import io.github.dsheirer.source.SourceEvent; import io.github.dsheirer.source.tuner.channel.TunerChannel; @@ -127,7 +128,7 @@ public class PolyphaseChannelSource extends TunerChannelSource * * @param channelResultsBuffer containing an array of polyphase channelizer outputs. */ - public void receiveChannelResults(PolyphaseChannelResultsBuffer channelResultsBuffer) + public void receiveChannelResults(ReusableChannelResultsBuffer channelResultsBuffer) { mReusableBufferAssembler.updateTimestamp(channelResultsBuffer.getTimestamp()); mPolyphaseChannelOutputProcessor.receiveChannelResults(channelResultsBuffer); diff --git a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/output/ChannelOutputProcessor.java b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/output/ChannelOutputProcessor.java index 60cfc9f7..69e7ec6f 100644 --- a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/output/ChannelOutputProcessor.java +++ b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/output/ChannelOutputProcessor.java @@ -18,11 +18,11 @@ ******************************************************************************/ package io.github.dsheirer.dsp.filter.channelizer.output; -import io.github.dsheirer.dsp.filter.channelizer.PolyphaseChannelResultsBuffer; import io.github.dsheirer.dsp.mixer.IOscillator; import io.github.dsheirer.dsp.mixer.Oscillator; import io.github.dsheirer.sample.OverflowableTransferQueue; import io.github.dsheirer.sample.buffer.ReusableBufferAssembler; +import io.github.dsheirer.sample.buffer.ReusableChannelResultsBuffer; import io.github.dsheirer.sample.real.IOverflowListener; import io.github.dsheirer.source.Source; import org.slf4j.Logger; @@ -35,8 +35,8 @@ public abstract class ChannelOutputProcessor implements IPolyphaseChannelOutputP { private final static Logger mLog = LoggerFactory.getLogger(ChannelOutputProcessor.class); - private OverflowableTransferQueue mChannelResultsQueue; - private List mChannelResultsToProcess = new ArrayList<>(); + private OverflowableTransferQueue mChannelResultsQueue; + private List mChannelResultsToProcess = new ArrayList<>(); private int mMaxResultsToProcess; private int mInputChannelCount; @@ -82,7 +82,7 @@ public abstract class ChannelOutputProcessor implements IPolyphaseChannelOutputP } @Override - public void receiveChannelResults(PolyphaseChannelResultsBuffer channelResults) + public void receiveChannelResults(ReusableChannelResultsBuffer channelResults) { mChannelResultsQueue.offer(channelResults); } @@ -116,7 +116,7 @@ public abstract class ChannelOutputProcessor implements IPolyphaseChannelOutputP * @param channelResults to process * @param reusableBufferAssembler to receive the processed channelizer results */ - public abstract void process(List channelResults, + public abstract void process(List channelResults, ReusableBufferAssembler reusableBufferAssembler); diff --git a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/output/IPolyphaseChannelOutputProcessor.java b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/output/IPolyphaseChannelOutputProcessor.java index 3726866e..c9856f25 100644 --- a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/output/IPolyphaseChannelOutputProcessor.java +++ b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/output/IPolyphaseChannelOutputProcessor.java @@ -18,8 +18,8 @@ ******************************************************************************/ package io.github.dsheirer.dsp.filter.channelizer.output; -import io.github.dsheirer.dsp.filter.channelizer.PolyphaseChannelResultsBuffer; import io.github.dsheirer.sample.buffer.ReusableBufferAssembler; +import io.github.dsheirer.sample.buffer.ReusableChannelResultsBuffer; import io.github.dsheirer.source.Source; import java.util.List; @@ -30,7 +30,7 @@ public interface IPolyphaseChannelOutputProcessor * Receive and enqueue output results from the polyphase analysis channelizer * @param channelResultsBuffer to enqueue */ - void receiveChannelResults(PolyphaseChannelResultsBuffer channelResultsBuffer); + void receiveChannelResults(ReusableChannelResultsBuffer channelResultsBuffer); /** * Process the channel output channel results queue and deliver the output to the listener diff --git a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/output/OneChannelOutputProcessor.java b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/output/OneChannelOutputProcessor.java index 1c5e534a..9684efa6 100644 --- a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/output/OneChannelOutputProcessor.java +++ b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/output/OneChannelOutputProcessor.java @@ -18,8 +18,8 @@ ******************************************************************************/ package io.github.dsheirer.dsp.filter.channelizer.output; -import io.github.dsheirer.dsp.filter.channelizer.PolyphaseChannelResultsBuffer; import io.github.dsheirer.sample.buffer.ReusableBufferAssembler; +import io.github.dsheirer.sample.buffer.ReusableChannelResultsBuffer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -76,10 +76,10 @@ public class OneChannelOutputProcessor extends ChannelOutputProcessor * @param reusableBufferAssembler to receive the extracted, frequency-translated channel results */ @Override - public void process(List channelResults, + public void process(List channelResults, ReusableBufferAssembler reusableBufferAssembler) { - for(PolyphaseChannelResultsBuffer channelResultsBuffer: channelResults) + for(ReusableChannelResultsBuffer channelResultsBuffer: channelResults) { try { @@ -96,6 +96,8 @@ public class OneChannelOutputProcessor extends ChannelOutputProcessor { mLog.error("Error extracting channel samples from polyphase channel results buffer"); } + + channelResultsBuffer.decrementUserCount(); } } } diff --git a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/output/TwoChannelOutputProcessor.java b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/output/TwoChannelOutputProcessor.java index d51096c2..44f9f8ea 100644 --- a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/output/TwoChannelOutputProcessor.java +++ b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/output/TwoChannelOutputProcessor.java @@ -15,10 +15,10 @@ ******************************************************************************/ package io.github.dsheirer.dsp.filter.channelizer.output; -import io.github.dsheirer.dsp.filter.channelizer.PolyphaseChannelResultsBuffer; import io.github.dsheirer.dsp.filter.channelizer.TwoChannelSynthesizerM2; import io.github.dsheirer.dsp.mixer.FS4DownConverter; import io.github.dsheirer.sample.buffer.ReusableBufferAssembler; +import io.github.dsheirer.sample.buffer.ReusableChannelResultsBuffer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -95,10 +95,10 @@ public class TwoChannelOutputProcessor extends ChannelOutputProcessor * @param reusableBufferAssembler to receive the extracted, frequency-translated channel results */ @Override - public void process(List channelResultsBuffers, + public void process(List channelResultsBuffers, ReusableBufferAssembler reusableBufferAssembler) { - for(PolyphaseChannelResultsBuffer buffer: channelResultsBuffers) + for(ReusableChannelResultsBuffer buffer: channelResultsBuffers) { float[] channel1 = buffer.getChannel(mChannelOffset1); float[] channel2 = buffer.getChannel(mChannelOffset2); @@ -113,6 +113,8 @@ public class TwoChannelOutputProcessor extends ChannelOutputProcessor getFrequencyCorrectionMixer().mixComplex(synthesized); reusableBufferAssembler.receive(synthesized); + + buffer.decrementUserCount(); } } } diff --git a/src/main/java/io/github/dsheirer/sample/OverflowableTransferQueue.java b/src/main/java/io/github/dsheirer/sample/OverflowableTransferQueue.java index c2b5a6d9..d3ad1541 100644 --- a/src/main/java/io/github/dsheirer/sample/OverflowableTransferQueue.java +++ b/src/main/java/io/github/dsheirer/sample/OverflowableTransferQueue.java @@ -115,6 +115,23 @@ public class OverflowableTransferQueue return drainCount; } + /** + * Retrieves elements from the queue into the collection up to the maximum number of elements specified + */ + public int drainTo(Collection collection) + { + int drainCount = mQueue.drainTo(collection); + + int size = mCounter.addAndGet(-drainCount); + + if(mOverflow.get() && size <= mResetThreshold) + { + setOverflow(false); + } + + return drainCount; + } + /** * Sets a listener to receive overflow state change events. */ diff --git a/src/main/java/io/github/dsheirer/sample/buffer/AbstractReusableBuffer.java b/src/main/java/io/github/dsheirer/sample/buffer/AbstractReusableBuffer.java new file mode 100644 index 00000000..d374e76a --- /dev/null +++ b/src/main/java/io/github/dsheirer/sample/buffer/AbstractReusableBuffer.java @@ -0,0 +1,151 @@ +/******************************************************************************* + * 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.sample.buffer; + +import java.util.concurrent.atomic.AtomicInteger; + +public abstract class AbstractReusableBuffer +{ + protected IReusableBufferDisposedListener mBufferDisposedListener; + protected long mTimestamp; + protected AtomicInteger mUserCount = new AtomicInteger(); + private String mDebugName; + + public AbstractReusableBuffer(IReusableBufferDisposedListener bufferDisposedListener, long timestamp) + { + mBufferDisposedListener = bufferDisposedListener; + mTimestamp = timestamp; + } + + public AbstractReusableBuffer(IReusableBufferDisposedListener bufferDisposedListener) + { + mBufferDisposedListener = bufferDisposedListener; + } + + public void setDebugName(String debugName) + { + mDebugName = debugName; + } + + public String name() + { + return (mDebugName != null ? mDebugName : "(null)") + " User Count:" + getUserCount(); + } + + public String toString() + { + return name(); + } + + /** + * Reference timestamp for this complex buffer. Timestamp is relative to the first sample in the buffer. + * @return timestamp as milliseconds since epoch + */ + public long getTimestamp() + { + return mTimestamp; + } + + public void setTimestamp(long timestamp) + { + mTimestamp = timestamp; + } + + /** + * Sets the number of users that will receive this buffer. Each user is expected to invoke the + * decrementUserCount() method to indicate that they are finished using this buffer so that this buffer can be + * reused once the user count is zero. + * + * @param userCount that will receive this reusable buffer + */ + public void setUserCount(int userCount) + { + mUserCount.set(userCount); + } + + /** + * Decrements the user count for this buffer to indicate that the user/receiver of this buffer is finished using + * the buffer. + * + * User count is established prior to sending this buffer to each listener and each listener invokes this method to + * indicate that they have finished processing the buffer so that when the user count reaches zero, this buffer + * can be reused. + * + * This method is thread-safe. + */ + public void decrementUserCount() + { + int currentCount = mUserCount.decrementAndGet(); + + if(currentCount == 0) + { + recycle(); + mBufferDisposedListener.disposed(this); + } + else if(currentCount < 0) + { + throw new IllegalStateException("User count is below zero. This indicates that this buffer's decrement" + + " user count was invoked by more than the expected user count"); + } + } + + /** + * Invoked just prior to notifying the owner that this buffer is ready for recycle. This method + * is intended for sub-class implementations to perform any recycle cleanup actions. + */ + protected void recycle() + { + //No-op in this abstract class + } + + /** + * Increments the user count to indicate that this buffer will be sent to another user. + * + * Users/receivers of this buffer that wish to forward the buffer to additional users should increment the expected + * user count prior to invoking the decrementUserCount() to indicate that the original receiver of the buffer is + * finished to avoid the user count reaching zero and the buffer being flagged as ready for reuse. + * + * This method is thread-safe. + */ + public void incrementUserCount() + { + mUserCount.incrementAndGet(); + } + + /** + * Increments the user count by the specified count to indicate that this buffer will be sent to additional users. + * + * Users/receivers of this buffer that wish to forward the buffer to additional users should increment the expected + * user count prior to invoking the decrementUserCount() to indicate that the original receiver of the buffer is + * finished to avoid the user count reaching zero and the buffer being flagged as ready for reuse. + * + * This method is thread-safe. + * + * @param additionalUserCount to add to the current user count + */ + public void incrementUserCount(int additionalUserCount) + { + mUserCount.addAndGet(additionalUserCount); + } + + /** + * Number of users currently registered for this buffer + */ + public int getUserCount() + { + return mUserCount.get(); + } +} diff --git a/src/main/java/io/github/dsheirer/sample/buffer/AbstractReusableBufferQueue.java b/src/main/java/io/github/dsheirer/sample/buffer/AbstractReusableBufferQueue.java index 5e5f3bd6..18b618e0 100644 --- a/src/main/java/io/github/dsheirer/sample/buffer/AbstractReusableBufferQueue.java +++ b/src/main/java/io/github/dsheirer/sample/buffer/AbstractReusableBufferQueue.java @@ -21,14 +21,14 @@ import org.slf4j.LoggerFactory; import java.util.Queue; import java.util.concurrent.LinkedTransferQueue; -public abstract class AbstractReusableBufferQueue implements IReusableBufferDisposedListener +public abstract class AbstractReusableBufferQueue + implements IReusableBufferDisposedListener { private final static Logger mLog = LoggerFactory.getLogger(AbstractReusableBufferQueue.class); private Queue mReusableBufferQueue = new LinkedTransferQueue<>(); private int mBufferCount = 0; private String mDebugName; - private int mBufferUserCounter; public AbstractReusableBufferQueue(String debugName) { @@ -54,48 +54,31 @@ public abstract class AbstractReusableBufferQueue impl public void disposed(T reusableBuffer) { mReusableBufferQueue.offer(reusableBuffer); - -// if(mDebugName != null && mDebugName.contentEquals("NativeBufferConverter")) -// { -// mLog.debug("Reclaimed: " + reusableBuffer.name()); -// } } /** - * Returns a reusable buffer from an internal recycling queue, or creates a new buffer if there are currently no - * buffers available for reuse. - * - * @param size of the samples that will be loaded into the buffer - * @return a reusable buffer + * Get a recycled buffer from the queue */ - public T getBuffer(int size) + protected T getRecycledBuffer() { - T buffer = mReusableBufferQueue.poll(); - - if(buffer == null) - { - buffer = createBuffer(size); - mBufferCount++; - buffer.setDebugName("Owner:" + mDebugName + " Number:" + mBufferUserCounter++ + " NEW BUFFER"); - - mLog.debug("Buffer Created - Count:" + mBufferCount + (mDebugName != null ? " [" + mDebugName + "]" : "")); - } - else - { - String previous = buffer.name(); - buffer.setDebugName("Owner:" + mDebugName + " Number:" + mBufferUserCounter++); - -// if(mDebugName != null && mDebugName.contentEquals("NativeBufferConverter")) -// { -// mLog.debug("Buffer Reused: " + previous + " New Name:" + buffer.name()); -// } - } - - return buffer; + return mReusableBufferQueue.poll(); } /** - * Create a new buffer of the specified size + * Increments the count of buffers managed by this queue. Note: this method is NOT thread safe */ - protected abstract T createBuffer(int size); + protected void incrementBufferCount() + { + mBufferCount++; + } + + protected int getBufferCount() + { + return mBufferCount; + } + + protected String getDebugName() + { + return mDebugName; + } } diff --git a/src/main/java/io/github/dsheirer/sample/buffer/IReusableBufferDisposedListener.java b/src/main/java/io/github/dsheirer/sample/buffer/IReusableBufferDisposedListener.java index b3baaffb..67cbf293 100644 --- a/src/main/java/io/github/dsheirer/sample/buffer/IReusableBufferDisposedListener.java +++ b/src/main/java/io/github/dsheirer/sample/buffer/IReusableBufferDisposedListener.java @@ -15,7 +15,7 @@ ******************************************************************************/ package io.github.dsheirer.sample.buffer; -public interface IReusableBufferDisposedListener +public interface IReusableBufferDisposedListener { /** * Notifies the listener that the timestamped buffer has been disposed and can now be reused. diff --git a/src/main/java/io/github/dsheirer/sample/buffer/ReusableBuffer.java b/src/main/java/io/github/dsheirer/sample/buffer/ReusableBuffer.java index d9434004..de1dd829 100644 --- a/src/main/java/io/github/dsheirer/sample/buffer/ReusableBuffer.java +++ b/src/main/java/io/github/dsheirer/sample/buffer/ReusableBuffer.java @@ -18,15 +18,10 @@ package io.github.dsheirer.sample.buffer; import org.apache.commons.lang3.Validate; import java.nio.FloatBuffer; -import java.util.concurrent.atomic.AtomicInteger; -public class ReusableBuffer +public class ReusableBuffer extends AbstractReusableBuffer { private float[] mSamples; - private IReusableBufferDisposedListener mBufferDisposedListener; - private long mTimestamp; - private AtomicInteger mUserCount = new AtomicInteger(); - private String mDebugName; /** * Creates a reusable, timestamped complex buffer using the specified time in milliseconds. @@ -41,26 +36,10 @@ public class ReusableBuffer */ ReusableBuffer(IReusableBufferDisposedListener bufferDisposedListener, float[] samples, long timestamp) { + super(bufferDisposedListener, timestamp); mSamples = samples; Validate.notNull(bufferDisposedListener, "Reusable Buffer Listener cannot be null"); - mBufferDisposedListener = bufferDisposedListener; - mTimestamp = timestamp; - } - - public void setDebugName(String debugName) - { - mDebugName = debugName; - } - - public String name() - { - return (mDebugName != null ? mDebugName : "(null)") + " User Count:" + getUserCount(); - } - - public String toString() - { - return name(); } /** @@ -105,20 +84,6 @@ public class ReusableBuffer this(bufferDisposedListener, samples, System.currentTimeMillis()); } - /** - * Reference timestamp for this complex buffer. Timestamp is relative to the first sample in the buffer. - * @return timestamp as milliseconds since epoch - */ - public long getTimestamp() - { - return mTimestamp; - } - - public void setTimestamp(long timestamp) - { - mTimestamp = timestamp; - } - /** * Reloads this buffer by copying the sample array into this buffer. Internal buffer length will be automatically * resized to fit the new sample array length. @@ -175,78 +140,4 @@ public class ReusableBuffer mTimestamp = timestamp; } - /** - * Sets the number of users that will receive this buffer. Each user is expected to invoke the - * decrementUserCount() method to indicate that they are finished using this buffer so that this buffer can be - * reused once the user count is zero. - * - * @param userCount that will receive this reusable buffer - */ - public void setUserCount(int userCount) - { - mUserCount.set(userCount); - } - - /** - * Decrements the user count for this buffer to indicate that the user/receiver of this buffer is finished using - * the buffer. - * - * User count is established prior to sending this buffer to each listener and each listener invokes this method to - * indicate that they have finished processing the buffer so that when the user count reaches zero, this buffer - * can be reused. - * - * This method is thread-safe. - */ - public void decrementUserCount() - { - int currentCount = mUserCount.decrementAndGet(); - - if(currentCount == 0) - { - mBufferDisposedListener.disposed(this); - } - else if(currentCount < 0) - { - throw new IllegalStateException("User count is below zero. This indicates that this buffer's decrement" + - " user count was invoked by more than the expected user count"); - } - } - - /** - * Increments the user count to indicate that this buffer will be sent to another user. - * - * Users/receivers of this buffer that wish to forward the buffer to additional users should increment the expected - * user count prior to invoking the decrementUserCount() to indicate that the original receiver of the buffer is - * finished to avoid the user count reaching zero and the buffer being flagged as ready for reuse. - * - * This method is thread-safe. - */ - public void incrementUserCount() - { - mUserCount.incrementAndGet(); - } - - /** - * Increments the user count by the specified count to indicate that this buffer will be sent to additional users. - * - * Users/receivers of this buffer that wish to forward the buffer to additional users should increment the expected - * user count prior to invoking the decrementUserCount() to indicate that the original receiver of the buffer is - * finished to avoid the user count reaching zero and the buffer being flagged as ready for reuse. - * - * This method is thread-safe. - * - * @param additionalUserCount to add to the current user count - */ - public void incrementUserCount(int additionalUserCount) - { - mUserCount.addAndGet(additionalUserCount); - } - - /** - * Number of users currently registered for this buffer - */ - public int getUserCount() - { - return mUserCount.get(); - } } diff --git a/src/main/java/io/github/dsheirer/sample/buffer/ReusableBufferQueue.java b/src/main/java/io/github/dsheirer/sample/buffer/ReusableBufferQueue.java index 85bb5efc..292e1033 100644 --- a/src/main/java/io/github/dsheirer/sample/buffer/ReusableBufferQueue.java +++ b/src/main/java/io/github/dsheirer/sample/buffer/ReusableBufferQueue.java @@ -31,9 +31,26 @@ public class ReusableBufferQueue extends AbstractReusableBufferQueue + * + ******************************************************************************/ +package io.github.dsheirer.sample.buffer; + +import java.util.LinkedList; +import java.util.List; + +public class ReusableChannelResultsBuffer extends AbstractReusableBuffer +{ + private LinkedList mEmptyBuffers = new LinkedList<>(); + private LinkedList mFilledBuffers = new LinkedList<>(); + private Integer mLength; + + /** + * Reusable buffer for storing polyphase channelizer results arrays. This buffer is NOT thread safe and is + * designed to be used by only one consumer. The consumer must invoke the recycle() method to indicate that this + * buffer can be reused. + * + * @param disposedListener to receive notification when this buffer is ready for recycle + */ + ReusableChannelResultsBuffer(IReusableBufferDisposedListener disposedListener) + { + super(disposedListener); + } + + /** + * Recycles this buffer for reuse. + */ + protected void recycle() + { + mEmptyBuffers.addAll(mFilledBuffers); + mFilledBuffers.clear(); + mLength = null; + } + + /** + * Gets an empty channel results array correctly sized to the requested length. This method should be used to + * obtain all empty channel results buffers to fill with information and add back to this buffer so that the results + * arrays can be reused. + * + * @param length of the requested array + * @return a float array of the requested length that may contain stale data from previous use. + */ + public float[] getEmptyBuffer(int length) + { + float[] emptyBuffer = mEmptyBuffers.poll(); + + if(emptyBuffer == null || emptyBuffer.length != length) + { + emptyBuffer = new float[length]; + } + + return emptyBuffer; + } + + /** + * Adds the channel results to this buffer. Note: you should only use channel results arrays obtained via the + * getEmptyBuffer() method and filled with data and then added back to this buffer via this method, so that array + * reuse can be accurately controlled. + * + * @param channelResults to add back to this buffer + */ + public void addChannelResults(float[] channelResults) + { + if(mLength == null) + { + mLength = channelResults.length; + } + else if(mLength != channelResults.length) + { + throw new IllegalArgumentException("Channel results length must be the same for all added results"); + } + + mFilledBuffers.offer(channelResults); + } + + /** + * Channel results array contained in this buffer + */ + public List getChannelResults() + { + return mFilledBuffers; + } + + /** + * Extracts a single I/Q interleaved sample buffer from this buffer of channel results arrays + * + * @param iChannelIndex to extract for the inphase sample. The q index is assumed to be one greater than i. + * @return float[] containing interleaved I and Q samples + * @throws IllegalArgumentException if this buffer doesn't contain any samples or if the requested channel is + * not within the channel range of the contained channel results. + */ + public float[] getChannel(int iChannelIndex) + { + if(mFilledBuffers.isEmpty()) + { + throw new IllegalArgumentException("Buffer is empty - cannot extract channel [" + iChannelIndex + "]"); + } + + if(!isValidChannelIndex(iChannelIndex)) + { + throw new IllegalArgumentException("Channel [" + iChannelIndex + "] is not valid for the contained channel " + + "results -- max channel is " + mLength); + } + + float[] samples = new float[mFilledBuffers.size() * 2]; + int pointer = 0; + + int qChannelIndex = iChannelIndex + 1; + + for(float[] channelResults: mFilledBuffers) + { + samples[pointer++] = channelResults[iChannelIndex]; + samples[pointer++] = channelResults[qChannelIndex]; + } + + return samples; + } + + /** + * Indicates if the channel argument is valid for the contained channel results. + * + * @param channelIndex number (0 to N) to check + * @return false if the buffer is empty or if the channel is greater than the contained channel results + */ + private boolean isValidChannelIndex(int channelIndex) + { + return (mLength != null) && (channelIndex <= mLength); + } +} diff --git a/src/main/java/io/github/dsheirer/sample/buffer/ReusableChannelResultsBufferQueue.java b/src/main/java/io/github/dsheirer/sample/buffer/ReusableChannelResultsBufferQueue.java new file mode 100644 index 00000000..9d6f7c16 --- /dev/null +++ b/src/main/java/io/github/dsheirer/sample/buffer/ReusableChannelResultsBufferQueue.java @@ -0,0 +1,55 @@ +/******************************************************************************* + * 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.sample.buffer; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class ReusableChannelResultsBufferQueue extends AbstractReusableBufferQueue +{ + private final static Logger mLog = LoggerFactory.getLogger(ReusableChannelResultsBufferQueue.class); + + public ReusableChannelResultsBufferQueue(String debugName) + { + super(debugName); + } + + public ReusableChannelResultsBufferQueue() + { + } + + /** + * Returns a reusable buffer from an internal recycling queue, or creates a new buffer if there are currently no + * buffers available for reuse. + * + * @return a reusable buffer + */ + public ReusableChannelResultsBuffer getBuffer() + { + ReusableChannelResultsBuffer buffer = getRecycledBuffer(); + + if(buffer == null) + { + buffer = new ReusableChannelResultsBuffer(this); + incrementBufferCount(); + + buffer.setDebugName("Owner:" + getDebugName()); + mLog.debug("Buffer Created - Count:" + getBufferCount() + (getDebugName() != null ? " [" + getDebugName() + "]" : "")); + } + + return buffer; + } +} diff --git a/src/main/java/io/github/dsheirer/sample/buffer/ReusableComplexBufferQueue.java b/src/main/java/io/github/dsheirer/sample/buffer/ReusableComplexBufferQueue.java index 0c52454e..4f0bcc88 100644 --- a/src/main/java/io/github/dsheirer/sample/buffer/ReusableComplexBufferQueue.java +++ b/src/main/java/io/github/dsheirer/sample/buffer/ReusableComplexBufferQueue.java @@ -31,9 +31,26 @@ public class ReusableComplexBufferQueue extends AbstractReusableBufferQueue