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

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

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

src/instrument/gui/DecoderViewFrame.java

@@ -99,7 +99,7 @@ public class DecoderViewFrame extends JInternalFrame
 
 		mDecodeConfig = decodeConfig;
 		
-		mProcessingChain = new InstrumentableProcessingChain();
+		mProcessingChain = new InstrumentableProcessingChain( "instrumented" );
 		
 		List<Module> modules = DecoderFactory
 			.getModules( ChannelType.STANDARD, 

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<RealBuffer> mRealBufferBroadcaster = new Broadcaster<>();
 	private Broadcaster<SquelchState> mSquelchStateBroadcaster = new Broadcaster<>();
 	
-//	private ThreadPoolManager mThreadPoolManager;
+	private ScheduledExecutorService mScheduledExecutorService;
 	private AtomicBoolean mRunning = new AtomicBoolean();
 	
 	protected Source mSource;
 	private List<Module> 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

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;

src/source/recording/Recording.java

@@ -63,9 +63,8 @@ public class Recording implements Comparable<Recording>,
 	}
 	
 	@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;

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
     		{

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

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

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<ComplexBuffer> 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<ComplexBuffer>)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 )
 					{

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 )
 			{

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 )
 				{

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

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;

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 );
     }
 
 	/**

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

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 )
 		{