Merge pull request #1695 from DSheirer/1694-sdrplay-rsp-tuner-memory-leaks

#1694 SDRPlay RSP Memory Leak & Audio Playback Delays

src/main/java/io/github/dsheirer/dsp/filter/channelizer/ComplexPolyphaseChannelizerM2.java

@@ -68,8 +68,8 @@ public class ComplexPolyphaseChannelizerM2 extends AbstractComplexPolyphaseChann
      */
     private static final int PROCESSED_CHANNEL_RESULTS_THRESHOLD = 1024;
 
-    //Sized to process 1/20th batch of 152 buffers a second,  at 40 times per second
+    //Sized to process 40 times per second
-    private IFFTProcessorDispatcher mIFFTProcessorDispatcher = new IFFTProcessorDispatcher(152 / 20, 25);
+    private IFFTProcessorDispatcher mIFFTProcessorDispatcher = new IFFTProcessorDispatcher(25);
     private FloatFFT_1D mFFT;
     private float[] mInlineSamples;
     private float[] mInlineFilter;
@@ -406,9 +406,9 @@ public class ComplexPolyphaseChannelizerM2 extends AbstractComplexPolyphaseChann
      */
     public class IFFTProcessorDispatcher extends Dispatcher<List<float[]>>
     {
-        public IFFTProcessorDispatcher(int batchSize, long interval)
+        public IFFTProcessorDispatcher(long interval)
         {
-            super("sdrtrunk polyphase ifft processor", batchSize, interval);
+            super("sdrtrunk polyphase ifft processor", interval);
 
             //We create a listener interface to receive the batched channel results arrays from the scheduled thread pool
             //dispatcher thread that is part of this continuous buffer processor.  We perform an IFFT on each

src/main/java/io/github/dsheirer/dsp/filter/channelizer/PolyphaseChannelManager.java

@@ -110,7 +110,7 @@ public class PolyphaseChannelManager implements ISourceEventProcessor
         }
 
         mChannelCalculator = new ChannelCalculator(sampleRate, channelCount, frequency, CHANNEL_OVERSAMPLING);
-        mBufferDispatcher = new Dispatcher("sdrtrunk polyphase buffer processor", 50, 10);
+        mBufferDispatcher = new Dispatcher("sdrtrunk polyphase buffer processor", 10);
         mBufferDispatcher.setListener(mNativeBufferReceiver);
     }
 

src/main/java/io/github/dsheirer/dsp/filter/channelizer/output/ChannelOutputProcessor.java

@@ -50,8 +50,7 @@ public abstract class ChannelOutputProcessor implements IPolyphaseChannelOutputP
         mInputChannelCount = inputChannelCount;
         //Process 1/10th of the sample rate per second at a rate of 20 times a second (200% of anticipated rate)
         mHeartbeatManager = heartbeatManager;
-        mChannelResultsDispatcher = new Dispatcher("sdrtrunk polyphase channel", (int)(sampleRate / 10),
+        mChannelResultsDispatcher = new Dispatcher("sdrtrunk polyphase channel",50, mHeartbeatManager);
-                50, mHeartbeatManager);
         mChannelResultsDispatcher.setListener(floats -> {
             try
             {

src/main/java/io/github/dsheirer/record/binary/BinaryRecorder.java

@@ -49,7 +49,7 @@ public class BinaryRecorder extends Module implements IByteBufferListener
     private final static Logger mLog = LoggerFactory.getLogger(BinaryRecorder.class);
     private static final int MAX_RECORDING_BYTE_SIZE = 524288;  //500 kB
 
-    private Dispatcher<ByteBuffer> mBufferProcessor = new Dispatcher<>("sdrtrunk binary recorder", 200, 250);
+    private Dispatcher<ByteBuffer> mBufferProcessor = new Dispatcher<>("sdrtrunk binary recorder", 250);
     private AtomicBoolean mRunning = new AtomicBoolean();
     private Path mBaseRecordingPath;
     private String mRecordingIdentifier;

src/main/java/io/github/dsheirer/record/wave/ComplexSamplesWaveRecorder.java

@@ -44,7 +44,7 @@ public class ComplexSamplesWaveRecorder extends Module implements IComplexSample
 {
     private final static Logger mLog = LoggerFactory.getLogger(ComplexSamplesWaveRecorder.class);
 
-    private Dispatcher<ComplexSamples> mBufferProcessor = new Dispatcher<>("sdrtrunk complex wave recorder", 100, 250);
+    private Dispatcher<ComplexSamples> mBufferProcessor = new Dispatcher<>("sdrtrunk complex wave recorder", 250);
     private AtomicBoolean mRunning = new AtomicBoolean();
     private BufferWaveWriter mWriter;
     private String mFilePrefix;

src/main/java/io/github/dsheirer/record/wave/NativeBufferWaveRecorder.java

@@ -47,7 +47,7 @@ public class NativeBufferWaveRecorder extends Module implements Listener<INative
     private static final Logger mLog = LoggerFactory.getLogger(ComplexSamplesWaveRecorder.class);
     private static final long STATUS_UPDATE_BYTE_INTERVAL = 1_048_576;
     private static final long MAX_RECORDING_SIZE = (long)Integer.MAX_VALUE * 2l;
-    private Dispatcher<INativeBuffer> mBufferProcessor = new Dispatcher<>("sdrtrunk native buffer wave recorder", 100, 250);
+    private Dispatcher<INativeBuffer> mBufferProcessor = new Dispatcher<>("sdrtrunk native buffer wave recorder", 250);
 
     private AtomicBoolean mRunning = new AtomicBoolean();
     private NativeBufferWaveWriter mWriter;

src/main/java/io/github/dsheirer/source/tuner/channel/HalfBandTunerChannelSource.java

@@ -74,8 +74,7 @@ public class HalfBandTunerChannelSource<T extends INativeBuffer> extends TunerCh
         mQDecimationFilter = DecimationFilterFactory.getRealDecimationFilter(decimation);
 
         //Set dispatcher to process 1/10 of estimated sample arrival rate, 20 times per second (up to 200% per interval)
-        mBufferDispatcher = new Dispatcher("sdrtrunk heterodyne channel " + tunerChannel.getFrequency(),
+        mBufferDispatcher = new Dispatcher("sdrtrunk heterodyne channel " + tunerChannel.getFrequency(), 50, getHeartbeatManager());
-                (int)(sampleRate / 10), 50, getHeartbeatManager());
         mBufferDispatcher.setListener(new NativeBufferProcessor());
 
         //Setup the frequency mixer to the current source frequency

src/main/java/io/github/dsheirer/source/tuner/channel/PassThroughChannelSource.java

@@ -55,7 +55,7 @@ public class PassThroughChannelSource extends TunerChannelSource implements ISou
         super(listener, tunerChannel);
         mTunerController = tunerController;
         mBufferDispatcher = new Dispatcher<>("sdrtrunk pass-through channel " + tunerChannel.getFrequency(),
-                250, 50, getHeartbeatManager());
+                50, getHeartbeatManager());
         mBufferDispatcher.setListener(new BufferProcessor());
     }
 

src/main/java/io/github/dsheirer/source/tuner/sdrplay/api/SDRplay.java

@@ -70,7 +70,7 @@ public class SDRplay
     private static final Logger mLog = LoggerFactory.getLogger(SDRplay.class);
 
     /**
-     * Foreign memory arenaallocation resource scope
+     * Foreign memory arena allocation resource scope
      */
     private final Arena mArena = Arena.openShared();
 

src/main/java/io/github/dsheirer/source/tuner/sdrplay/api/callback/CallbackFunctions.java

@@ -37,24 +37,24 @@ public class CallbackFunctions
 
     /**
      * Constructs a callback functions for single-tuner use.
-     * @param arena for native memory allocation
+     * @param longLifecycleArena for native memory allocation for long lifecycle objects.
      * @param deviceEventListener for device events
      * @param streamListener for streaming samples
      * @param streamCallbackListener for streaming events
      */
-    public CallbackFunctions(Arena arena, IDeviceEventListener deviceEventListener,
+    public CallbackFunctions(Arena longLifecycleArena, IDeviceEventListener deviceEventListener,
                              IStreamListener streamListener, IStreamCallbackListener streamCallbackListener)
     {
         //Create the event callback function
-        mDeviceEventAdapter = new DeviceEventAdapter(arena, deviceEventListener);
+        mDeviceEventAdapter = new DeviceEventAdapter(deviceEventListener);
-        MemorySegment eventFunction = sdrplay_api_EventCallback_t.allocate(mDeviceEventAdapter, arena.scope());
+        MemorySegment eventFunction = sdrplay_api_EventCallback_t.allocate(mDeviceEventAdapter, longLifecycleArena.scope());
 
         //Create the stream A callback function
-        mStreamACallbackAdapter = new StreamCallbackAdapter(arena, streamListener, streamCallbackListener);
+        mStreamACallbackAdapter = new StreamCallbackAdapter(streamListener, streamCallbackListener);
-        MemorySegment streamAFunction = sdrplay_api_StreamCallback_t.allocate(mStreamACallbackAdapter, arena.scope());
+        MemorySegment streamAFunction = sdrplay_api_StreamCallback_t.allocate(mStreamACallbackAdapter, longLifecycleArena.scope());
 
         //Create the callback functions union and populate the callback functions
-        mCallbackFunctionsMemorySegment = sdrplay_api_CallbackFnsT.allocate(arena);
+        mCallbackFunctionsMemorySegment = sdrplay_api_CallbackFnsT.allocate(longLifecycleArena);
         sdrplay_api_CallbackFnsT.EventCbFn$set(mCallbackFunctionsMemorySegment, eventFunction);
         sdrplay_api_CallbackFnsT.StreamACbFn$set(mCallbackFunctionsMemorySegment, streamAFunction);
     }
@@ -72,15 +72,15 @@ public class CallbackFunctions
                              IStreamCallbackListener streamCallbackListener)
     {
         //Create the event callback function
-        mDeviceEventAdapter = new DeviceEventAdapter(arena, deviceEventListener);
+        mDeviceEventAdapter = new DeviceEventAdapter(deviceEventListener);
         MemorySegment eventFunction = sdrplay_api_EventCallback_t.allocate(mDeviceEventAdapter, arena.scope());
 
         //Create the stream A callback function
-        mStreamACallbackAdapter = new StreamCallbackAdapter(arena, streamAListener, streamCallbackListener);
+        mStreamACallbackAdapter = new StreamCallbackAdapter(streamAListener, streamCallbackListener);
         MemorySegment streamAFunction = sdrplay_api_StreamCallback_t.allocate(mStreamACallbackAdapter, arena.scope());
 
         //Create the stream B callback function
-        mStreamBCallbackAdapter = new StreamCallbackAdapter(arena, streamBListener, streamCallbackListener);
+        mStreamBCallbackAdapter = new StreamCallbackAdapter(streamBListener, streamCallbackListener);
         MemorySegment streamBFunction = sdrplay_api_StreamCallback_t.allocate(mStreamBCallbackAdapter, arena.scope());
 
         //Create the callback functions union and populate the callback functions

src/main/java/io/github/dsheirer/source/tuner/sdrplay/api/callback/DeviceEventAdapter.java

@@ -36,24 +36,14 @@ import org.slf4j.LoggerFactory;
 public class DeviceEventAdapter implements sdrplay_api_EventCallback_t
 {
     private static final Logger mLog = LoggerFactory.getLogger(DeviceEventAdapter.class);
-
-    private Arena mArena;
     private IDeviceEventListener mDeviceEventListener;
 
     /**
      * Constructs an instance.
-     * @param arena to use in creating foreign memory segments to access foreign structures.
      * @param listener to receive translated device events.
      */
-    public DeviceEventAdapter(Arena arena, IDeviceEventListener listener)
+    public DeviceEventAdapter(IDeviceEventListener listener)
     {
-        if(arena == null)
-        {
-            throw new IllegalArgumentException("Resource scope must be non-null");
-        }
-
-        mArena = arena;
-
         setListener(listener);
     }
 
@@ -75,35 +65,44 @@ public class DeviceEventAdapter implements sdrplay_api_EventCallback_t
     public void apply(int eventTypeId, int tunerSelectId, MemorySegment eventParametersPointer,
                       MemorySegment callbackContext)
     {
-        MemorySegment memorySegment = sdrplay_api_EventParamsT.ofAddress(eventParametersPointer, mArena.scope());
+        try(Arena arena = Arena.openConfined())
-        EventType eventType = EventType.fromValue(eventTypeId);
-        TunerSelect tunerSelect = TunerSelect.fromValue(tunerSelectId);
-
-        switch(eventType)
         {
-            case GAIN_CHANGE -> {
+            MemorySegment memorySegment = sdrplay_api_EventParamsT.ofAddress(eventParametersPointer, arena.scope());
-                mDeviceEventListener.processGainChange(tunerSelect,
+            EventType eventType = EventType.fromValue(eventTypeId);
-                        EventParametersFactory.createGainCallbackParameters(memorySegment));
+            TunerSelect tunerSelect = TunerSelect.fromValue(tunerSelectId);
-            }
+
-            case POWER_OVERLOAD_CHANGE -> {
+            switch(eventType)
-                mDeviceEventListener.processPowerOverload(tunerSelect,
+            {
-                        EventParametersFactory.createPowerOverloadCallbackParameters(memorySegment));
+                case GAIN_CHANGE ->
-            }
+                {
-            case DEVICE_REMOVED -> {
+                    mDeviceEventListener.processGainChange(tunerSelect,
-                mDeviceEventListener.processDeviceRemoval(tunerSelect);
+                            EventParametersFactory.createGainCallbackParameters(memorySegment));
-            }
+                }
-            case RSP_DUO_MODE_CHANGE -> {
+                case POWER_OVERLOAD_CHANGE ->
-                mDeviceEventListener.processRspDuoModeChange(tunerSelect,
+                {
-                        EventParametersFactory.createRspDuoModeCallbackParameters(memorySegment));
+                    mDeviceEventListener.processPowerOverload(tunerSelect,
-            }
+                            EventParametersFactory.createPowerOverloadCallbackParameters(memorySegment));
-            case UNKNOWN -> {
+                }
-                mLog.warn("Unknown device event callback ignored.  Please contact the library developer as this may " +
+                case DEVICE_REMOVED ->
-                        "indicate a change to the SDRPlay API change. Tuner:" + tunerSelect + " Event Type ID:" +
+                {
-                        eventTypeId);
+                    mDeviceEventListener.processDeviceRemoval(tunerSelect);
-                mDeviceEventListener.processEvent(eventType, tunerSelect);
+                }
-            }
+                case RSP_DUO_MODE_CHANGE ->
-            default -> {
+                {
-                throw new IllegalStateException("DeviceEventAdapter must be updated handle EventType." + eventType);
+                    mDeviceEventListener.processRspDuoModeChange(tunerSelect,
+                            EventParametersFactory.createRspDuoModeCallbackParameters(memorySegment));
+                }
+                case UNKNOWN ->
+                {
+                    mLog.warn("Unknown device event callback ignored.  Please contact the library developer as this may " +
+                            "indicate a change to the SDRPlay API change. Tuner:" + tunerSelect + " Event Type ID:" +
+                            eventTypeId);
+                    mDeviceEventListener.processEvent(eventType, tunerSelect);
+                }
+                default ->
+                {
+                    throw new IllegalStateException("DeviceEventAdapter must be updated handle EventType." + eventType);
+                }
             }
         }
     }

src/main/java/io/github/dsheirer/source/tuner/sdrplay/api/callback/StreamCallbackAdapter.java

@@ -35,24 +35,16 @@ import org.slf4j.LoggerFactory;
 public class StreamCallbackAdapter implements sdrplay_api_StreamCallback_t
 {
     private static final Logger mLog = LoggerFactory.getLogger(StreamCallbackAdapter.class);
-    private Arena mArena;
     private IStreamListener mStreamListener;
     private IStreamCallbackListener mStreamCallbackListener;
 
     /**
      * Constructs an instance of the callback implementation
-     * @param arena for defining new foreign memory segments
      * @param streamListener to receive transferred I/Q samples and event details
      * @param listener to receive callback parameters
      */
-    public StreamCallbackAdapter(Arena arena, IStreamListener streamListener, IStreamCallbackListener listener)
+    public StreamCallbackAdapter(IStreamListener streamListener, IStreamCallbackListener listener)
     {
-        if(arena == null)
-        {
-            throw new IllegalArgumentException("Arena must be non-null");
-        }
-
-        mArena = arena;
         mStreamCallbackListener = listener;
         setListener(streamListener);
     }
@@ -81,24 +73,27 @@ public class StreamCallbackAdapter implements sdrplay_api_StreamCallback_t
     {
         if(mStreamListener != null || mStreamCallbackListener != null)
         {
-            //Translate the callback parameters pointer to a memory segment and re-construct the parameters as a Java object
+            try(Arena arena = Arena.openConfined())
-            StreamCallbackParameters parameters = new StreamCallbackParameters(sdrplay_api_StreamCbParamsT
-                    .ofAddress(parametersPointer, mArena.scope()));
-
-            if(mStreamCallbackListener != null)
             {
-                mStreamCallbackListener.process(mStreamListener.getTunerSelect(), parameters, reset);
+                //Translate the callback parameters pointer to a memory segment and re-construct the parameters as a Java object
-            }
+                StreamCallbackParameters parameters = new StreamCallbackParameters(sdrplay_api_StreamCbParamsT
+                        .ofAddress(parametersPointer, arena.scope()));
 
-            if(mStreamListener != null)
+                if(mStreamCallbackListener != null)
-            {
+                {
-                //Allocate memory segments from I/Q pointers, transfer from native to JVM array, and send to listener
+                    mStreamCallbackListener.process(mStreamListener.getTunerSelect(), parameters, reset);
-                long arrayByteSize = ValueLayout.JAVA_SHORT.byteSize() * sampleCount;
+                }
-                MemorySegment iSamples = MemorySegment.ofAddress(iSamplesPointer.address(), arrayByteSize, mArena.scope());
+
-                MemorySegment qSamples = MemorySegment.ofAddress(qSamplesPointer.address(), arrayByteSize, mArena.scope());
+                if(mStreamListener != null)
-                short[] i = iSamples.toArray(ValueLayout.JAVA_SHORT);
+                {
-                short[] q = qSamples.toArray(ValueLayout.JAVA_SHORT);
+                    //Allocate memory segments from I/Q pointers, transfer from native to JVM array, and send to listener
-                mStreamListener.processStream(i, q, parameters, Flag.evaluate(reset));
+                    long arrayByteSize = ValueLayout.JAVA_SHORT.byteSize() * sampleCount;
+                    MemorySegment iSamples = MemorySegment.ofAddress(iSamplesPointer.address(), arrayByteSize, arena.scope());
+                    MemorySegment qSamples = MemorySegment.ofAddress(qSamplesPointer.address(), arrayByteSize, arena.scope());
+                    short[] i = iSamples.toArray(ValueLayout.JAVA_SHORT);
+                    short[] q = qSamples.toArray(ValueLayout.JAVA_SHORT);
+                    mStreamListener.processStream(i, q, parameters, Flag.evaluate(reset));
+                }
             }
         }
     }

src/main/java/io/github/dsheirer/util/Dispatcher.java

@@ -41,46 +41,35 @@ import org.slf4j.LoggerFactory;
 public class Dispatcher<E> implements Listener<E>
 {
     private final static Logger mLog = LoggerFactory.getLogger(Dispatcher.class);
-    private static final long OVERFLOW_LOG_EVENT_WAIT_PERIOD = TimeUnit.SECONDS.toMillis(10);
+    private final LinkedTransferQueue<E> mQueue = new LinkedTransferQueue<>();
-    private LinkedTransferQueue<E> mQueue = new LinkedTransferQueue<>();
     private Listener<E> mListener;
-    private AtomicBoolean mRunning = new AtomicBoolean();
+    private final AtomicBoolean mRunning = new AtomicBoolean();
-    private String mThreadName;
+    private final String mThreadName;
     private ScheduledExecutorService mExecutorService;
     private ScheduledFuture<?> mScheduledFuture;
-    private int mBatchSize;
+    private final long mInterval;
-    private long mInterval;
     private HeartbeatManager mHeartbeatManager;
 
     /**
      * Constructs an instance of a Dispatcher with integrated heartbeat support.
      * @param threadName to name the dispatcher thread
-     * @param batchSize is maximum number of elements to process from the queue per interval.  This should be sized to
-     * the anticipated number of elements per second, divided by the number of intervals per second and increased by
-     * a factor of two.  This serves to constrain how many elements are processed per interval when the queue starts
-     * to back up, so that the thread doesn't dominate CPU resources.
      * @param interval for processing each batch in milliseconds.
      * @param heartbeatManager to receive a heartbeat command at each processing interval.
      */
-    public Dispatcher(String threadName, int batchSize, long interval, HeartbeatManager heartbeatManager)
+    public Dispatcher(String threadName, long interval, HeartbeatManager heartbeatManager)
     {
-        this(threadName, batchSize, interval);
+        this(threadName, interval);
         mHeartbeatManager = heartbeatManager;
     }
 
     /**
      * Constructs an instance
      * @param threadName to name the dispatcher thread
-     * @param batchSize is maximum number of elements to process from the queue per interval.  This should be sized to
-     * the anticipated number of elements per second, divided by the number of intervals per second and increased by
-     * a factor of two.  This serves to constrain how many elements are processed per interval when the queue starts
-     * to back up, so that the thread doesn't dominate CPU resources.
      * @param interval for processing each batch in milliseconds.
      */
-    public Dispatcher(String threadName, int batchSize, long interval)
+    public Dispatcher(String threadName, long interval)
     {
         mThreadName = threadName;
-        mBatchSize = batchSize;
         mInterval = interval;
     }
 
@@ -171,7 +160,7 @@ public class Dispatcher<E> implements Listener<E>
     {
         List<E> elements = new ArrayList<>();
 
-        mQueue.drainTo(elements, mBatchSize);
+        mQueue.drainTo(elements);
 
         for(E element: elements)
         {
@@ -195,7 +184,7 @@ public class Dispatcher<E> implements Listener<E>
      */
     class Processor implements Runnable
     {
-        private AtomicBoolean mRunning = new AtomicBoolean();
+        private final AtomicBoolean mRunning = new AtomicBoolean();
 
         @Override
         public void run()
@@ -214,7 +203,7 @@ public class Dispatcher<E> implements Listener<E>
      */
     class ProcessorWithHeartbeat implements Runnable
     {
-        private AtomicBoolean mRunning = new AtomicBoolean();
+        private final AtomicBoolean mRunning = new AtomicBoolean();
 
         @Override
         public void run()