From dd22397f75b69997cc21d12e5043c60fef4ccea5 Mon Sep 17 00:00:00 2001 From: denny Date: Mon, 15 Feb 2016 09:46:33 -0500 Subject: [PATCH] Resolves issue with application memory growth and thread pool saturation. Each tunerchannelsource was sharing a thread pool and as the processing load increased, the thread pool was not expanding and the processing tasks were backing up, causing the buffers to grow with unprocessed data. Changed so that each tuner channel source is handed a thread pool to use for processing incoming sample data. Processing chain maintains the thread pool and starts the channel source. This allows traffic channels to reuse the thread pool across multiple iterations. --- src/controller/channel/Channel.java | 5 ++- .../InstrumentableProcessingChain.java | 3 +- src/instrument/gui/DecoderViewFrame.java | 2 +- src/module/ProcessingChain.java | 20 +++++++++--- .../decode/state/TrafficChannelManager.java | 9 ++++++ src/source/recording/Recording.java | 5 ++- .../recording/RecordingSourceManager.java | 4 +-- src/source/tuner/Tuner.java | 7 ++--- src/source/tuner/TunerChannelProvider.java | 4 +-- src/source/tuner/TunerChannelSource.java | 31 ++++++++----------- src/source/tuner/TunerController.java | 7 ++--- src/source/tuner/TunerManager.java | 3 +- src/source/tuner/airspy/AirspyTuner.java | 6 ++-- src/source/tuner/ettus/B100Tuner.java | 4 +-- src/source/tuner/fcd/FCDTuner.java | 7 ++--- src/source/tuner/hackrf/HackRFTuner.java | 6 ++-- src/source/tuner/rtl/RTL2832Tuner.java | 9 ++---- 17 files changed, 70 insertions(+), 62 deletions(-) diff --git a/src/controller/channel/Channel.java b/src/controller/channel/Channel.java index e23e5d30..2b150c4f 100644 --- a/src/controller/channel/Channel.java +++ b/src/controller/channel/Channel.java @@ -297,6 +297,9 @@ public class Channel extends Configuration /* Broadcast channel deleted event */ fireChannelEvent( Event.CHANNEL_DELETED ); + + mProcessingChain.dispose(); + mProcessingChain = null; } /** @@ -809,7 +812,7 @@ public class Channel extends Configuration { if( mProcessingChain == null ) { - mProcessingChain = new ProcessingChain(); + mProcessingChain = new ProcessingChain( getName() ); /* Get the optional alias list for the decode modules to use */ AliasList aliasList = mResourceManager.getPlaylistManager() diff --git a/src/instrument/InstrumentableProcessingChain.java b/src/instrument/InstrumentableProcessingChain.java index 3efef0ae..43ca9942 100644 --- a/src/instrument/InstrumentableProcessingChain.java +++ b/src/instrument/InstrumentableProcessingChain.java @@ -12,8 +12,9 @@ import source.Source; public class InstrumentableProcessingChain extends ProcessingChain implements Instrumentable { - public InstrumentableProcessingChain() + public InstrumentableProcessingChain( String name ) { + super( name ); } public void setSource( Source source ) throws IllegalStateException diff --git a/src/instrument/gui/DecoderViewFrame.java b/src/instrument/gui/DecoderViewFrame.java index 08738bdb..7ac54e4f 100644 --- a/src/instrument/gui/DecoderViewFrame.java +++ b/src/instrument/gui/DecoderViewFrame.java @@ -99,7 +99,7 @@ public class DecoderViewFrame extends JInternalFrame mDecodeConfig = decodeConfig; - mProcessingChain = new InstrumentableProcessingChain(); + mProcessingChain = new InstrumentableProcessingChain( "instrumented" ); List modules = DecoderFactory .getModules( ChannelType.STANDARD, diff --git a/src/module/ProcessingChain.java b/src/module/ProcessingChain.java index c8e3da4f..75fe805e 100644 --- a/src/module/ProcessingChain.java +++ b/src/module/ProcessingChain.java @@ -19,7 +19,8 @@ package module; import java.util.ArrayList; import java.util.List; -import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.atomic.AtomicBoolean; import message.IMessageListener; @@ -61,7 +62,7 @@ import audio.metadata.Metadata; import audio.squelch.ISquelchStateListener; import audio.squelch.ISquelchStateProvider; import audio.squelch.SquelchState; -import controller.ThreadPoolManager; +import controller.NamingThreadFactory; import controller.channel.ChannelEvent; import controller.channel.IChannelEventListener; import controller.channel.IChannelEventProvider; @@ -99,15 +100,17 @@ public class ProcessingChain implements IChannelEventListener private Broadcaster mRealBufferBroadcaster = new Broadcaster<>(); private Broadcaster mSquelchStateBroadcaster = new Broadcaster<>(); -// private ThreadPoolManager mThreadPoolManager; + private ScheduledExecutorService mScheduledExecutorService; private AtomicBoolean mRunning = new AtomicBoolean(); protected Source mSource; private List mModules = new ArrayList<>(); private IFrequencyCorrectionController mFrequencyCorrectionController; - public ProcessingChain() + public ProcessingChain( String name ) { + mScheduledExecutorService = Executors.newScheduledThreadPool( 1, + new NamingThreadFactory( "channel " + name ) ); } public void dispose() @@ -121,6 +124,9 @@ public class ProcessingChain implements IChannelEventListener mModules.clear(); + mScheduledExecutorService.shutdownNow(); + mScheduledExecutorService = null; + mFrequencyCorrectionController = null; mAudioPacketBroadcaster.dispose(); @@ -432,9 +438,11 @@ public class ProcessingChain implements IChannelEventListener * of the decoder state's */ if( mSource instanceof TunerChannelSource ) { + TunerChannelSource tunerChannelSource = (TunerChannelSource)mSource; + try { - long frequency = ((TunerChannelSource)mSource).getFrequency(); + long frequency = tunerChannelSource.getFrequency(); mDecoderStateEventBroadcaster.broadcast( new DecoderStateEvent( this, Event.SOURCE_FREQUENCY, @@ -444,6 +452,8 @@ public class ProcessingChain implements IChannelEventListener { mLog.error( "Error getting frequency from tuner channel source", e ); } + + tunerChannelSource.start( mScheduledExecutorService ); } } else diff --git a/src/module/decode/state/TrafficChannelManager.java b/src/module/decode/state/TrafficChannelManager.java index 34bb2ecc..46374e70 100644 --- a/src/module/decode/state/TrafficChannelManager.java +++ b/src/module/decode/state/TrafficChannelManager.java @@ -91,6 +91,15 @@ public class TrafficChannelManager extends Module @Override public void dispose() { + mLog.debug( "dispose() on traffic channel manager" ); + for( Channel trafficChannel: mTrafficChannelPool ) + { + trafficChannel.setEnabled( false ); + trafficChannel.dispose(); + } + + mTrafficChannelPool.clear(); + mCallEventListener = null; mResourceManager = null; mDecodeConfiguration = null; diff --git a/src/source/recording/Recording.java b/src/source/recording/Recording.java index 90fad405..492efec7 100644 --- a/src/source/recording/Recording.java +++ b/src/source/recording/Recording.java @@ -63,9 +63,8 @@ public class Recording implements Comparable, } @Override - public TunerChannelSource getChannel( ThreadPoolManager threadPoolManager, - TunerChannel channel ) throws RejectedExecutionException, - SourceException + public TunerChannelSource getChannel( TunerChannel channel ) + throws RejectedExecutionException, SourceException { // TODO Auto-generated method stub return null; diff --git a/src/source/recording/RecordingSourceManager.java b/src/source/recording/RecordingSourceManager.java index 0fcabaef..260c33b7 100644 --- a/src/source/recording/RecordingSourceManager.java +++ b/src/source/recording/RecordingSourceManager.java @@ -71,9 +71,7 @@ public class RecordingSourceManager if( recording != null ) { - retVal = recording.getChannel( - mResourceManager.getThreadPoolManager(), - tunerChannel ); + retVal = recording.getChannel( tunerChannel ); } else { diff --git a/src/source/tuner/Tuner.java b/src/source/tuner/Tuner.java index e274c97f..2697ac9d 100644 --- a/src/source/tuner/Tuner.java +++ b/src/source/tuner/Tuner.java @@ -30,10 +30,9 @@ import sample.complex.ComplexBuffer; import source.SourceException; import source.tuner.frequency.FrequencyChangeBroadcaster; import source.tuner.frequency.FrequencyChangeEvent; -import source.tuner.frequency.FrequencyChangeListener; import source.tuner.frequency.FrequencyChangeEvent.Event; +import source.tuner.frequency.FrequencyChangeListener; import controller.ResourceManager; -import controller.ThreadPoolManager; /** * Tuner - provides tuner channel sources, representing a channel frequency @@ -147,8 +146,8 @@ public abstract class Tuner implements FrequencyChangeBroadcaster, * * @return - source for 48k sample rate */ - public abstract TunerChannelSource getChannel( ThreadPoolManager threadPoolManager, - TunerChannel channel ) throws RejectedExecutionException, SourceException; + public abstract TunerChannelSource getChannel( TunerChannel channel ) + throws RejectedExecutionException, SourceException; /** * Releases the tuned channel resources diff --git a/src/source/tuner/TunerChannelProvider.java b/src/source/tuner/TunerChannelProvider.java index f8fabe48..f8b7a49b 100644 --- a/src/source/tuner/TunerChannelProvider.java +++ b/src/source/tuner/TunerChannelProvider.java @@ -14,8 +14,8 @@ public interface TunerChannelProvider * * @return - source for 48k sample rate */ - public abstract TunerChannelSource getChannel( ThreadPoolManager threadPoolManager, - TunerChannel channel ) throws RejectedExecutionException, SourceException; + public abstract TunerChannelSource getChannel( TunerChannel channel ) + throws RejectedExecutionException, SourceException; /** * Releases the tuned channel resources diff --git a/src/source/tuner/TunerChannelSource.java b/src/source/tuner/TunerChannelSource.java index 0c61dc60..243c81e1 100644 --- a/src/source/tuner/TunerChannelSource.java +++ b/src/source/tuner/TunerChannelSource.java @@ -21,6 +21,7 @@ import java.util.ArrayList; import java.util.List; import java.util.concurrent.LinkedTransferQueue; import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; @@ -37,8 +38,6 @@ import source.SourceException; import source.tuner.frequency.FrequencyChangeEvent; import source.tuner.frequency.FrequencyChangeEvent.Event; import source.tuner.frequency.FrequencyChangeListener; -import controller.ThreadPoolManager; -import controller.ThreadPoolManager.ThreadType; import dsp.filter.FilterFactory; import dsp.filter.Window.WindowType; import dsp.filter.cic.ComplexPrimeCICDecimate; @@ -62,7 +61,6 @@ public class TunerChannelSource extends ComplexSource private ComplexPrimeCICDecimate mDecimationFilter; private Listener mListener; private FrequencyChangeListener mFrequencyChangeListener; - private ThreadPoolManager mThreadPoolManager; private ScheduledFuture mTaskHandle; private long mTunerFrequency = 0; @@ -71,15 +69,11 @@ public class TunerChannelSource extends ComplexSource private DecimationProcessor mDecimationProcessor = new DecimationProcessor(); - public TunerChannelSource( ThreadPoolManager threadPoolManager, - Tuner tuner, - TunerChannel tunerChannel ) - throws RejectedExecutionException, SourceException + public TunerChannelSource( Tuner tuner, TunerChannel tunerChannel ) + throws RejectedExecutionException, SourceException { super( "Tuner Channel Source" ); - mThreadPoolManager = threadPoolManager; - mTuner = tuner; mTuner.addListener( (FrequencyChangeListener)this ); @@ -95,14 +89,17 @@ public class TunerChannelSource extends ComplexSource /* Fire a sample rate change event to setup the decimation chain */ frequencyChanged( new FrequencyChangeEvent( Event.SAMPLE_RATE_CHANGE_NOTIFICATION, mTuner.getSampleRate() ) ); - - /* Schedule the decimation task to run every 20 ms (50 iterations/second) */ - mTaskHandle = mThreadPoolManager.scheduleFixedRate( ThreadType.DECIMATION, - mDecimationProcessor, 20, TimeUnit.MILLISECONDS ); + } + + public void start( ScheduledExecutorService scheduledExecutorService ) + { + /* Schedule the decimation task to run every 10 ms (100 iterations/second) */ + mTaskHandle = scheduledExecutorService.scheduleAtFixedRate( + mDecimationProcessor, 0, 10, TimeUnit.MILLISECONDS ); /* Finally, register to receive samples from the tuner */ mTuner.addListener( (Listener)this ); - } + } @Override public void dispose() @@ -129,12 +126,10 @@ public class TunerChannelSource extends ComplexSource { if( mTaskHandle != null ) { - mThreadPoolManager.cancel( mTaskHandle ); - mThreadPoolManager = null; + mTaskHandle.cancel( true ); mTaskHandle = null; } - mBuffer.clear(); mBuffer = null; @@ -290,7 +285,7 @@ public class TunerChannelSource extends ComplexSource { if( mBuffer != null ) { - mBuffer.drainTo( mSampleBuffers, 4 ); + mBuffer.drainTo( mSampleBuffers, 8 ); for( Buffer buffer: mSampleBuffers ) { diff --git a/src/source/tuner/TunerController.java b/src/source/tuner/TunerController.java index aacb5f6b..46e6b0a2 100644 --- a/src/source/tuner/TunerController.java +++ b/src/source/tuner/TunerController.java @@ -194,8 +194,8 @@ public abstract class TunerController implements Tunable * * @throws RejectedExecutionException if the decimation processor has an error */ - public TunerChannelSource getChannel( ThreadPoolManager threadPoolManager, - Tuner tuner, TunerChannel channel ) throws RejectedExecutionException + public TunerChannelSource getChannel( Tuner tuner, TunerChannel channel ) + throws RejectedExecutionException { TunerChannelSource source = null; @@ -210,8 +210,7 @@ public abstract class TunerController implements Tunable updateLOFrequency(); } - source = new TunerChannelSource( threadPoolManager, - tuner, channel ); + source = new TunerChannelSource( tuner, channel ); } catch( SourceException se ) { diff --git a/src/source/tuner/TunerManager.java b/src/source/tuner/TunerManager.java index 345757f3..0668da18 100644 --- a/src/source/tuner/TunerManager.java +++ b/src/source/tuner/TunerManager.java @@ -100,8 +100,7 @@ public class TunerManager try { - retVal = tuner.getChannel( - mResourceManager.getThreadPoolManager(), tunerChannel ); + retVal = tuner.getChannel( tunerChannel ); } catch ( RejectedExecutionException ree ) { diff --git a/src/source/tuner/airspy/AirspyTuner.java b/src/source/tuner/airspy/AirspyTuner.java index 36df0315..9c85b201 100644 --- a/src/source/tuner/airspy/AirspyTuner.java +++ b/src/source/tuner/airspy/AirspyTuner.java @@ -128,8 +128,8 @@ public class AirspyTuner extends Tuner } @Override - public TunerChannelSource getChannel( ThreadPoolManager threadPoolManager, - TunerChannel channel ) throws RejectedExecutionException, SourceException + public TunerChannelSource getChannel( TunerChannel channel ) + throws RejectedExecutionException, SourceException { //TODO: this channel has a decimated sample rate of: // 10.0 MSps = 10,000,000 / 208 = 48076.923 @@ -138,7 +138,7 @@ public class AirspyTuner extends Tuner //Consider implementing a fractional resampler to get a correct 48 kHz //output sample rate - return mController.getChannel( threadPoolManager, this, channel ); + return mController.getChannel( this, channel ); } @Override diff --git a/src/source/tuner/ettus/B100Tuner.java b/src/source/tuner/ettus/B100Tuner.java index 9ec27054..c8cddf02 100644 --- a/src/source/tuner/ettus/B100Tuner.java +++ b/src/source/tuner/ettus/B100Tuner.java @@ -99,8 +99,8 @@ public class B100Tuner extends Tuner } @Override - public TunerChannelSource getChannel( ThreadPoolManager threadPoolManager, - TunerChannel channel ) throws RejectedExecutionException, SourceException + public TunerChannelSource getChannel( TunerChannel channel ) + throws RejectedExecutionException, SourceException { //TODO: pass this call to the tuner controller return null; diff --git a/src/source/tuner/fcd/FCDTuner.java b/src/source/tuner/fcd/FCDTuner.java index ea84063c..6c6324f4 100644 --- a/src/source/tuner/fcd/FCDTuner.java +++ b/src/source/tuner/fcd/FCDTuner.java @@ -112,11 +112,10 @@ public class FCDTuner extends MixerTuner } @Override - public TunerChannelSource getChannel( ThreadPoolManager threadPoolManager, - TunerChannel tunerChannel ) throws RejectedExecutionException, - SourceException + public TunerChannelSource getChannel( TunerChannel tunerChannel ) + throws RejectedExecutionException, SourceException { - return mController.getChannel( threadPoolManager, this, tunerChannel ); + return mController.getChannel( this, tunerChannel ); } /** diff --git a/src/source/tuner/hackrf/HackRFTuner.java b/src/source/tuner/hackrf/HackRFTuner.java index 4b38c4bd..e847b58e 100644 --- a/src/source/tuner/hackrf/HackRFTuner.java +++ b/src/source/tuner/hackrf/HackRFTuner.java @@ -106,10 +106,10 @@ public class HackRFTuner extends Tuner } @Override - public TunerChannelSource getChannel( ThreadPoolManager threadPoolManager, - TunerChannel channel ) throws RejectedExecutionException, SourceException + public TunerChannelSource getChannel( TunerChannel channel ) + throws RejectedExecutionException, SourceException { - return mController.getChannel( threadPoolManager, this, channel ); + return mController.getChannel( this, channel ); } @Override diff --git a/src/source/tuner/rtl/RTL2832Tuner.java b/src/source/tuner/rtl/RTL2832Tuner.java index 02634918..e7f4a3ac 100644 --- a/src/source/tuner/rtl/RTL2832Tuner.java +++ b/src/source/tuner/rtl/RTL2832Tuner.java @@ -37,7 +37,6 @@ import source.tuner.TunerType; import source.tuner.frequency.FrequencyChangeListener; import source.tuner.rtl.RTL2832TunerController.SampleRate; import controller.ResourceManager; -import controller.ThreadPoolManager; public class RTL2832Tuner extends Tuner { @@ -148,16 +147,14 @@ public class RTL2832Tuner extends Tuner } @Override - public TunerChannelSource getChannel( ThreadPoolManager threadPoolManager, - TunerChannel channel ) - throws RejectedExecutionException, - SourceException + public TunerChannelSource getChannel( TunerChannel channel ) + throws RejectedExecutionException, SourceException { TunerChannelSource source = null; try { - source = mController.getChannel( threadPoolManager, this, channel ); + source = mController.getChannel( this, channel ); } catch( Exception e ) {