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.

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 <http://www.gnu.org/licenses/>
+ *
+ ******************************************************************************/
+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();
+    }
+}

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<T extends ReusableBuffer> implements IReusableBufferDisposedListener<T>
+public abstract class AbstractReusableBufferQueue<T extends AbstractReusableBuffer>
+        implements IReusableBufferDisposedListener<T>
 {
     private final static Logger mLog = LoggerFactory.getLogger(AbstractReusableBufferQueue.class);
 
     private Queue<T> 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<T extends ReusableBuffer> 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;
+    }
 }

src/main/java/io/github/dsheirer/sample/buffer/IReusableBufferDisposedListener.java

@@ -15,7 +15,7 @@
  ******************************************************************************/
 package io.github.dsheirer.sample.buffer;
 
-public interface IReusableBufferDisposedListener<T extends ReusableBuffer>
+public interface IReusableBufferDisposedListener<T extends AbstractReusableBuffer>
 {
     /**
      * Notifies the listener that the timestamped buffer has been disposed and can now be reused.

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();
-    }
 }

src/main/java/io/github/dsheirer/sample/buffer/ReusableBufferQueue.java

@@ -31,9 +31,26 @@ public class ReusableBufferQueue extends AbstractReusableBufferQueue<ReusableBuf
     {
     }
 
-    @Override
-    protected ReusableBuffer createBuffer(int size)
+    /**
+     * 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
+     */
+    public ReusableBuffer getBuffer(int size)
     {
-        return new ReusableBuffer(this, new float[size]);
+        ReusableBuffer buffer = getRecycledBuffer();
+
+        if(buffer == null)
+        {
+            buffer = new ReusableBuffer(this, new float[size]);
+            incrementBufferCount();
+
+            buffer.setDebugName("Owner:" + getDebugName());
+            mLog.debug("Buffer Created - Count:" + getBufferCount() + (getDebugName() != null ? " [" + getDebugName() + "]" : ""));
+        }
+
+        return buffer;
     }
 }

src/main/java/io/github/dsheirer/sample/buffer/ReusableChannelResultsBuffer.java

@@ -0,0 +1,143 @@
+/*******************************************************************************
+ * sdr-trunk
+ * Copyright (C) 2014-2018 Dennis Sheirer
+ *
+ * This program is free software: you can redistribute it and/or modify it under the terms of the GNU General Public
+ * License as published by  the Free Software Foundation, either version 3 of the License, or  (at your option) any
+ * later version.
+ *
+ * This program is distributed in the hope that it will be useful,  but WITHOUT ANY WARRANTY; without even the implied
+ * warranty of  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License  along with this program.
+ * If not, see <http://www.gnu.org/licenses/>
+ *
+ ******************************************************************************/
+package io.github.dsheirer.sample.buffer;
+
+import java.util.LinkedList;
+import java.util.List;
+
+public class ReusableChannelResultsBuffer extends AbstractReusableBuffer
+{
+    private LinkedList<float[]> mEmptyBuffers = new LinkedList<>();
+    private LinkedList<float[]> 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<float[]> 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);
+    }
+}

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 <http://www.gnu.org/licenses/>
+ *
+ ******************************************************************************/
+package io.github.dsheirer.sample.buffer;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ReusableChannelResultsBufferQueue extends AbstractReusableBufferQueue<ReusableChannelResultsBuffer>
+{
+    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;
+    }
+}

src/main/java/io/github/dsheirer/sample/buffer/ReusableComplexBufferQueue.java

@@ -31,9 +31,26 @@ public class ReusableComplexBufferQueue extends AbstractReusableBufferQueue<Reus
     {
     }
 
-    @Override
-    protected ReusableComplexBuffer createBuffer(int size)
+    /**
+     * 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
+     */
+    public ReusableComplexBuffer getBuffer(int size)
     {
-        return new ReusableComplexBuffer(this, new float[size]);
+        ReusableComplexBuffer buffer = getRecycledBuffer();
+
+        if(buffer == null)
+        {
+            buffer = new ReusableComplexBuffer(this, new float[size]);
+            incrementBufferCount();
+
+            buffer.setDebugName("Owner:" + getDebugName());
+            mLog.debug("Buffer Created - Count:" + getBufferCount() + (getDebugName() != null ? " [" + getDebugName() + "]" : ""));
+        }
+
+        return buffer;
     }
 }