From d3dc1a8667a489b78c0ae5d53eaa878750e1afcd Mon Sep 17 00:00:00 2001 From: Dennis Sheirer Date: Sun, 5 Nov 2023 04:18:29 -0500 Subject: [PATCH] #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. --- .../ComplexPolyphaseChannelizerM2.java | 8 +- .../channelizer/PolyphaseChannelManager.java | 2 +- .../output/ChannelOutputProcessor.java | 3 +- .../record/binary/BinaryRecorder.java | 2 +- .../wave/ComplexSamplesWaveRecorder.java | 2 +- .../record/wave/NativeBufferWaveRecorder.java | 2 +- .../channel/HalfBandTunerChannelSource.java | 3 +- .../channel/PassThroughChannelSource.java | 2 +- .../source/tuner/sdrplay/api/SDRplay.java | 2 +- .../api/callback/CallbackFunctions.java | 20 ++--- .../api/callback/DeviceEventAdapter.java | 77 +++++++++---------- .../api/callback/StreamCallbackAdapter.java | 45 +++++------ .../io/github/dsheirer/util/Dispatcher.java | 31 +++----- 13 files changed, 90 insertions(+), 109 deletions(-) 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 fdf2e700..df51ecf4 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 @@ -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> { - 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 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 8c901377..63ed5963 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 @@ -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); } 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 e75cbf45..b07b6de5 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 @@ -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 { diff --git a/src/main/java/io/github/dsheirer/record/binary/BinaryRecorder.java b/src/main/java/io/github/dsheirer/record/binary/BinaryRecorder.java index e0dbf000..6464996d 100644 --- a/src/main/java/io/github/dsheirer/record/binary/BinaryRecorder.java +++ b/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 mBufferProcessor = new Dispatcher<>("sdrtrunk binary recorder", 200, 250); + private Dispatcher mBufferProcessor = new Dispatcher<>("sdrtrunk binary recorder", 250); private AtomicBoolean mRunning = new AtomicBoolean(); private Path mBaseRecordingPath; private String mRecordingIdentifier; diff --git a/src/main/java/io/github/dsheirer/record/wave/ComplexSamplesWaveRecorder.java b/src/main/java/io/github/dsheirer/record/wave/ComplexSamplesWaveRecorder.java index b3340442..151b4f03 100644 --- a/src/main/java/io/github/dsheirer/record/wave/ComplexSamplesWaveRecorder.java +++ b/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 mBufferProcessor = new Dispatcher<>("sdrtrunk complex wave recorder", 100, 250); + private Dispatcher mBufferProcessor = new Dispatcher<>("sdrtrunk complex wave recorder", 250); private AtomicBoolean mRunning = new AtomicBoolean(); private BufferWaveWriter mWriter; private String mFilePrefix; diff --git a/src/main/java/io/github/dsheirer/record/wave/NativeBufferWaveRecorder.java b/src/main/java/io/github/dsheirer/record/wave/NativeBufferWaveRecorder.java index 49e987f4..abb6d9d5 100644 --- a/src/main/java/io/github/dsheirer/record/wave/NativeBufferWaveRecorder.java +++ b/src/main/java/io/github/dsheirer/record/wave/NativeBufferWaveRecorder.java @@ -47,7 +47,7 @@ public class NativeBufferWaveRecorder extends Module implements Listener mBufferProcessor = new Dispatcher<>("sdrtrunk native buffer wave recorder", 100, 250); + private Dispatcher mBufferProcessor = new Dispatcher<>("sdrtrunk native buffer wave recorder", 250); private AtomicBoolean mRunning = new AtomicBoolean(); private NativeBufferWaveWriter mWriter; diff --git a/src/main/java/io/github/dsheirer/source/tuner/channel/HalfBandTunerChannelSource.java b/src/main/java/io/github/dsheirer/source/tuner/channel/HalfBandTunerChannelSource.java index 2e60df46..f3df9fb0 100644 --- a/src/main/java/io/github/dsheirer/source/tuner/channel/HalfBandTunerChannelSource.java +++ b/src/main/java/io/github/dsheirer/source/tuner/channel/HalfBandTunerChannelSource.java @@ -74,8 +74,7 @@ public class HalfBandTunerChannelSource 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 diff --git a/src/main/java/io/github/dsheirer/source/tuner/channel/PassThroughChannelSource.java b/src/main/java/io/github/dsheirer/source/tuner/channel/PassThroughChannelSource.java index 0979df02..ac3f37f6 100644 --- a/src/main/java/io/github/dsheirer/source/tuner/channel/PassThroughChannelSource.java +++ b/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()); } diff --git a/src/main/java/io/github/dsheirer/source/tuner/sdrplay/api/SDRplay.java b/src/main/java/io/github/dsheirer/source/tuner/sdrplay/api/SDRplay.java index b482d032..26788aa8 100644 --- a/src/main/java/io/github/dsheirer/source/tuner/sdrplay/api/SDRplay.java +++ b/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(); diff --git a/src/main/java/io/github/dsheirer/source/tuner/sdrplay/api/callback/CallbackFunctions.java b/src/main/java/io/github/dsheirer/source/tuner/sdrplay/api/callback/CallbackFunctions.java index d2645d77..89f38587 100644 --- a/src/main/java/io/github/dsheirer/source/tuner/sdrplay/api/callback/CallbackFunctions.java +++ b/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); - 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 diff --git a/src/main/java/io/github/dsheirer/source/tuner/sdrplay/api/callback/DeviceEventAdapter.java b/src/main/java/io/github/dsheirer/source/tuner/sdrplay/api/callback/DeviceEventAdapter.java index f82fd4c2..cbb68633 100644 --- a/src/main/java/io/github/dsheirer/source/tuner/sdrplay/api/callback/DeviceEventAdapter.java +++ b/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()); - 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); + } } } } diff --git a/src/main/java/io/github/dsheirer/source/tuner/sdrplay/api/callback/StreamCallbackAdapter.java b/src/main/java/io/github/dsheirer/source/tuner/sdrplay/api/callback/StreamCallbackAdapter.java index 8bab1fc2..8951f1be 100644 --- a/src/main/java/io/github/dsheirer/source/tuner/sdrplay/api/callback/StreamCallbackAdapter.java +++ b/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 - 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)); + } } } } diff --git a/src/main/java/io/github/dsheirer/util/Dispatcher.java b/src/main/java/io/github/dsheirer/util/Dispatcher.java index 428cbdc8..f26cb783 100644 --- a/src/main/java/io/github/dsheirer/util/Dispatcher.java +++ b/src/main/java/io/github/dsheirer/util/Dispatcher.java @@ -41,46 +41,35 @@ import org.slf4j.LoggerFactory; public class Dispatcher implements Listener { private final static Logger mLog = LoggerFactory.getLogger(Dispatcher.class); - private static final long OVERFLOW_LOG_EVENT_WAIT_PERIOD = TimeUnit.SECONDS.toMillis(10); - private LinkedTransferQueue mQueue = new LinkedTransferQueue<>(); + private final LinkedTransferQueue mQueue = new LinkedTransferQueue<>(); private Listener 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 implements Listener { List elements = new ArrayList<>(); - mQueue.drainTo(elements, mBatchSize); + mQueue.drainTo(elements); for(E element: elements) { @@ -195,7 +184,7 @@ public class Dispatcher implements Listener */ 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 implements Listener */ class ProcessorWithHeartbeat implements Runnable { - private AtomicBoolean mRunning = new AtomicBoolean(); + private final AtomicBoolean mRunning = new AtomicBoolean(); @Override public void run()