GitHub/sdrtrunk
1
0
Fork 0
mirror of https://github.com/DSheirer/sdrtrunk.git synced 2026-11-12 08:37:25 -05:00
Code Issues Packages Projects Releases 2 Wiki Activity

#1694 Resolves memory leak in SDRPlay RSP tuner implementation. Sample stream and event callback functions were using a global arena for memory allocation where memory was not being de-allocated. Changed to on-demand arena creation and deterministic memory de-allocation. Also, when using a reduced sample rate from the RSP tuner, the dispatcher was not keeping up with the inbound sample buffer rate, resulting in a buildup of RSP native sample buffers over time and causing growing audio delays with broken/stilted audio during playback.

Browse Source
This commit is contained in:
Dennis Sheirer
2023-11-05 04:18:29 -05:00
parent bf278120de
commit d3dc1a8667
13 changed files with 90 additions and 109 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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
private IFFTProcessorDispatcher mIFFTProcessorDispatcher = new IFFTProcessorDispatcher(152 / 20, 25);
//Sized to process 40 times per second
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

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

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

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

@@ -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),
50, mHeartbeatManager);
mChannelResultsDispatcher = new Dispatcher("sdrtrunk polyphase channel",50, mHeartbeatManager);
mChannelResultsDispatcher.setListener(floats -> {
try
{

2
src/main/java/io/github/dsheirer/record/binary/BinaryRecorder.java
View File

@@ -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;

2
src/main/java/io/github/dsheirer/record/wave/ComplexSamplesWaveRecorder.java
View File

@@ -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;

2
src/main/java/io/github/dsheirer/record/wave/NativeBufferWaveRecorder.java
View File

@@ -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;

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

@@ -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(),
(int)(sampleRate / 10), 50, getHeartbeatManager());
mBufferDispatcher = new Dispatcher("sdrtrunk heterodyne channel " + tunerChannel.getFrequency(), 50, getHeartbeatManager());
mBufferDispatcher.setListener(new NativeBufferProcessor());
//Setup the frequency mixer to the current source frequency

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

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

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

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

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

@@ -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);
MemorySegment eventFunction = sdrplay_api_EventCallback_t.allocate(mDeviceEventAdapter, arena.scope());
mDeviceEventAdapter = new DeviceEventAdapter(deviceEventListener);
MemorySegment eventFunction = sdrplay_api_EventCallback_t.allocate(mDeviceEventAdapter, longLifecycleArena.scope());
//Create the stream A callback function
mStreamACallbackAdapter = new StreamCallbackAdapter(arena, streamListener, streamCallbackListener);
MemorySegment streamAFunction = sdrplay_api_StreamCallback_t.allocate(mStreamACallbackAdapter, arena.scope());
mStreamACallbackAdapter = new StreamCallbackAdapter(streamListener, streamCallbackListener);
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

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

@@ -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());
EventType eventType = EventType.fromValue(eventTypeId);
TunerSelect tunerSelect = TunerSelect.fromValue(tunerSelectId);
switch(eventType)
try(Arena arena = Arena.openConfined())
{
case GAIN_CHANGE -> {
mDeviceEventListener.processGainChange(tunerSelect,
EventParametersFactory.createGainCallbackParameters(memorySegment));
}
case POWER_OVERLOAD_CHANGE -> {
mDeviceEventListener.processPowerOverload(tunerSelect,
EventParametersFactory.createPowerOverloadCallbackParameters(memorySegment));
}
case DEVICE_REMOVED -> {
mDeviceEventListener.processDeviceRemoval(tunerSelect);
}
case RSP_DUO_MODE_CHANGE -> {
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);
MemorySegment memorySegment = sdrplay_api_EventParamsT.ofAddress(eventParametersPointer, arena.scope());
EventType eventType = EventType.fromValue(eventTypeId);
TunerSelect tunerSelect = TunerSelect.fromValue(tunerSelectId);
switch(eventType)
{
case GAIN_CHANGE ->
{
mDeviceEventListener.processGainChange(tunerSelect,
EventParametersFactory.createGainCallbackParameters(memorySegment));
}
case POWER_OVERLOAD_CHANGE ->
{
mDeviceEventListener.processPowerOverload(tunerSelect,
EventParametersFactory.createPowerOverloadCallbackParameters(memorySegment));
}
case DEVICE_REMOVED ->
{
mDeviceEventListener.processDeviceRemoval(tunerSelect);
}
case RSP_DUO_MODE_CHANGE ->
{
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);
}
}
}
}

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

@@ -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
StreamCallbackParameters parameters = new StreamCallbackParameters(sdrplay_api_StreamCbParamsT
.ofAddress(parametersPointer, mArena.scope()));
if(mStreamCallbackListener != null)
try(Arena arena = Arena.openConfined())
{
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)
{
//Allocate memory segments from I/Q pointers, transfer from native to JVM array, and send to listener
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());
short[] i = iSamples.toArray(ValueLayout.JAVA_SHORT);
short[] q = qSamples.toArray(ValueLayout.JAVA_SHORT);
mStreamListener.processStream(i, q, parameters, Flag.evaluate(reset));
if(mStreamCallbackListener != null)
{
mStreamCallbackListener.process(mStreamListener.getTunerSelect(), parameters, reset);
}
if(mStreamListener != null)
{
//Allocate memory segments from I/Q pointers, transfer from native to JVM array, and send to listener
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));
}
}
}
}

31
src/main/java/io/github/dsheirer/util/Dispatcher.java
View File

@@ -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 LinkedTransferQueue<E> mQueue = new LinkedTransferQueue<>();
private final LinkedTransferQueue<E> mQueue = new LinkedTransferQueue<>();
private Listener<E> mListener;
private AtomicBoolean mRunning = new AtomicBoolean();
private String mThreadName;
private final AtomicBoolean mRunning = new AtomicBoolean();
private final String mThreadName;
private ScheduledExecutorService mExecutorService;
private ScheduledFuture<?> mScheduledFuture;
private int mBatchSize;
private long mInterval;
private final 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()