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