Resolves issue with scheduled thread pool overtasking. Removes tuner

channel source decimation processing over to its own thread pool.
This commit is contained in:
denny
2016-02-15 08:42:29 -05:00
parent e7d3713791
commit e2a0451bfb
16 changed files with 62 additions and 58 deletions

View File

@@ -243,7 +243,6 @@ public class NameConfigurationEditor extends ChannelConfigurationEditor
mChannelName.setText( DEFAULT_NAME ); mChannelName.setText( DEFAULT_NAME );
mSystemNameCombo.setModel( EMPTY_MODEL ); mSystemNameCombo.setModel( EMPTY_MODEL );
mSiteNameCombo.setModel( EMPTY_MODEL ); mSiteNameCombo.setModel( EMPTY_MODEL );
mAliasListCombo.setSelectedIndex( 0 );
} }
} }
} }

View File

@@ -495,6 +495,11 @@ public class ProcessingChain implements IChannelEventListener
tcs.start(); tcs.start();
} }
for( Listener<RealBuffer> listener: mUnFilteredRealBufferBroadcaster.getListeners() )
{
mLog.debug( "*** Listener:" + listener.getClass() );
}
} }
else else
{ {

View File

@@ -17,10 +17,9 @@
******************************************************************************/ ******************************************************************************/
package sample; package sample;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.CopyOnWriteArrayList;
import module.decode.state.ChannelState;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
@@ -29,11 +28,7 @@ import org.slf4j.LoggerFactory;
*/ */
public class Broadcaster<T> implements Listener<T> public class Broadcaster<T> implements Listener<T>
{ {
private final static Logger mLog = private List<Listener<T>> mListeners = new CopyOnWriteArrayList<>();
LoggerFactory.getLogger( Broadcaster.class );
private CopyOnWriteArrayList<Listener<T>> mListeners =
new CopyOnWriteArrayList<Listener<T>>();
@Override @Override
public void receive( T t ) public void receive( T t )
@@ -59,6 +54,11 @@ public class Broadcaster<T> implements Listener<T>
return mListeners.size(); return mListeners.size();
} }
public List<Listener<T>> getListeners()
{
return mListeners;
}
public void addListener( Listener<T> listener ) public void addListener( Listener<T> listener )
{ {
mListeners.add( listener ); mListeners.add( listener );

View File

@@ -37,7 +37,7 @@ public class SourceManager
ThreadPoolManager threadPoolManager ) ThreadPoolManager threadPoolManager )
{ {
mMixerManager = new MixerManager(); mMixerManager = new MixerManager();
mRecordingSourceManager = new RecordingSourceManager( settingsManager, threadPoolManager ); mRecordingSourceManager = new RecordingSourceManager( settingsManager );
mTunerManager = new TunerManager( mMixerManager, settingsManager, threadPoolManager ); mTunerManager = new TunerManager( mMixerManager, settingsManager, threadPoolManager );
} }

View File

@@ -64,9 +64,8 @@ public class Recording implements Comparable<Recording>,
} }
@Override @Override
public TunerChannelSource getChannel( ThreadPoolManager threadPoolManager, public TunerChannelSource getChannel( TunerChannel channel )
TunerChannel channel ) throws RejectedExecutionException, throws RejectedExecutionException, SourceException
SourceException
{ {
// TODO Auto-generated method stub // TODO Auto-generated method stub
return null; return null;

View File

@@ -41,13 +41,10 @@ public class RecordingSourceManager
private ArrayList<Recording> mRecordings = new ArrayList<Recording>(); private ArrayList<Recording> mRecordings = new ArrayList<Recording>();
private SettingsManager mSettingsManager; private SettingsManager mSettingsManager;
private ThreadPoolManager mThreadPoolManager;
public RecordingSourceManager( SettingsManager settingsManager, public RecordingSourceManager( SettingsManager settingsManager )
ThreadPoolManager threadPoolManager )
{ {
mSettingsManager = settingsManager; mSettingsManager = settingsManager;
mThreadPoolManager = threadPoolManager;
loadRecordings(); loadRecordings();
} }
@@ -73,7 +70,7 @@ public class RecordingSourceManager
if( recording != null ) if( recording != null )
{ {
retVal = recording.getChannel( mThreadPoolManager, tunerChannel ); retVal = recording.getChannel( tunerChannel );
} }
else else
{ {

View File

@@ -20,7 +20,6 @@ package source.tuner;
import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.RejectedExecutionException;
import source.SourceException; import source.SourceException;
import controller.ThreadPoolManager;
public interface ITunerChannelProvider public interface ITunerChannelProvider
{ {
@@ -31,8 +30,8 @@ public interface ITunerChannelProvider
* *
* @return - source for 48k sample rate * @return - source for 48k sample rate
*/ */
public abstract TunerChannelSource getChannel( ThreadPoolManager threadPoolManager, public abstract TunerChannelSource getChannel( TunerChannel channel )
TunerChannel channel ) throws RejectedExecutionException, SourceException; throws RejectedExecutionException, SourceException;
/** /**
* Releases the tuned channel resources * Releases the tuned channel resources

View File

@@ -145,8 +145,8 @@ public abstract class Tuner implements IFrequencyChangeProcessor,
* *
* @return - source for 48k sample rate * @return - source for 48k sample rate
*/ */
public abstract TunerChannelSource getChannel( ThreadPoolManager threadPoolManager, public abstract TunerChannelSource getChannel( TunerChannel channel )
TunerChannel channel ) throws RejectedExecutionException, SourceException; throws RejectedExecutionException, SourceException;
/** /**
* Releases the tuned channel resources * Releases the tuned channel resources

View File

@@ -19,8 +19,10 @@ package source.tuner;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.List; import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedTransferQueue; import java.util.concurrent.LinkedTransferQueue;
import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture; import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicBoolean;
@@ -37,7 +39,7 @@ import source.SourceException;
import source.tuner.frequency.FrequencyChangeEvent; import source.tuner.frequency.FrequencyChangeEvent;
import source.tuner.frequency.FrequencyChangeEvent.Event; import source.tuner.frequency.FrequencyChangeEvent.Event;
import source.tuner.frequency.IFrequencyChangeProcessor; import source.tuner.frequency.IFrequencyChangeProcessor;
import controller.ThreadPoolManager; import controller.NamingThreadFactory;
import controller.ThreadPoolManager.ThreadType; import controller.ThreadPoolManager.ThreadType;
import dsp.filter.FilterFactory; import dsp.filter.FilterFactory;
import dsp.filter.Window.WindowType; import dsp.filter.Window.WindowType;
@@ -61,7 +63,7 @@ public class TunerChannelSource extends ComplexSource
private ComplexPrimeCICDecimate mDecimationFilter; private ComplexPrimeCICDecimate mDecimationFilter;
private Listener<ComplexBuffer> mListener; private Listener<ComplexBuffer> mListener;
private IFrequencyChangeProcessor mFrequencyChangeProcessor; private IFrequencyChangeProcessor mFrequencyChangeProcessor;
private ThreadPoolManager mThreadPoolManager; private ScheduledExecutorService mScheduledExecutorService;
private ScheduledFuture<?> mTaskHandle; private ScheduledFuture<?> mTaskHandle;
private long mTunerFrequency = 0; private long mTunerFrequency = 0;
@@ -91,12 +93,9 @@ public class TunerChannelSource extends ComplexSource
* accept the decimation processing task * accept the decimation processing task
* @throws SourceException if the tuner has an issue providing IQ samples * @throws SourceException if the tuner has an issue providing IQ samples
*/ */
public TunerChannelSource( ThreadPoolManager threadPoolManager, Tuner tuner, public TunerChannelSource( Tuner tuner, TunerChannel tunerChannel )
TunerChannel tunerChannel )
throws RejectedExecutionException, SourceException throws RejectedExecutionException, SourceException
{ {
mThreadPoolManager = threadPoolManager;
mTuner = tuner; mTuner = tuner;
mTuner.addListener( this ); mTuner.addListener( this );
@@ -133,9 +132,12 @@ public class TunerChannelSource extends ComplexSource
if( mRunning.compareAndSet( false, true )) if( mRunning.compareAndSet( false, true ))
{ {
/* Schedule the decimation task to run every 20 ms (50 iterations/second) */ mScheduledExecutorService = Executors.newScheduledThreadPool( 1,
mTaskHandle = mThreadPoolManager.scheduleFixedRate( ThreadType.DECIMATION, new NamingThreadFactory( "tuner channel source" ) );
mDecimationProcessor, 20, TimeUnit.MILLISECONDS );
/* Schedule the decimation task to run every 10 ms (100 iterations/second) */
mTaskHandle = mScheduledExecutorService.scheduleAtFixedRate(
mDecimationProcessor, 0, 10, TimeUnit.MILLISECONDS );
/* Finally, register to receive samples from the tuner */ /* Finally, register to receive samples from the tuner */
mTuner.addListener( (Listener<ComplexBuffer>)this ); mTuner.addListener( (Listener<ComplexBuffer>)this );
@@ -156,10 +158,15 @@ public class TunerChannelSource extends ComplexSource
mTuner.releaseChannel( this ); mTuner.releaseChannel( this );
mDecimationProcessor.shutdown(); mDecimationProcessor.shutdown();
if( mThreadPoolManager != null && mTaskHandle != null ) if( mScheduledExecutorService != null && mTaskHandle != null )
{ {
mThreadPoolManager.cancel( mTaskHandle ); if( mTaskHandle != null )
mTaskHandle = null; {
mTaskHandle.cancel( true );
mTaskHandle = null;
}
mScheduledExecutorService.shutdownNow();
} }
mBuffer.clear(); mBuffer.clear();
@@ -340,7 +347,9 @@ public class TunerChannelSource extends ComplexSource
{ {
if( mBuffer != null ) if( mBuffer != null )
{ {
mBuffer.drainTo( mSampleBuffers, 4 ); long start = System.currentTimeMillis();
mBuffer.drainTo( mSampleBuffers );
for( Buffer buffer: mSampleBuffers ) for( Buffer buffer: mSampleBuffers )
{ {
@@ -378,6 +387,8 @@ public class TunerChannelSource extends ComplexSource
} }
} }
mLog.debug( "Processed [" + mSampleBuffers.size() + "] in " + ( System.currentTimeMillis() - start ) );
mSampleBuffers.clear(); mSampleBuffers.clear();
} }
} }

View File

@@ -194,8 +194,8 @@ public abstract class TunerController implements Tunable
* *
* @throws RejectedExecutionException if the decimation processor has an error * @throws RejectedExecutionException if the decimation processor has an error
*/ */
public TunerChannelSource getChannel( ThreadPoolManager threadPoolManager, public TunerChannelSource getChannel( Tuner tuner, TunerChannel channel )
Tuner tuner, TunerChannel channel ) throws RejectedExecutionException throws RejectedExecutionException
{ {
TunerChannelSource source = null; TunerChannelSource source = null;
@@ -210,8 +210,7 @@ public abstract class TunerController implements Tunable
updateLOFrequency(); updateLOFrequency();
} }
source = new TunerChannelSource( threadPoolManager, source = new TunerChannelSource( tuner, channel );
tuner, channel );
} }
catch( SourceException se ) catch( SourceException se )
{ {

View File

@@ -104,7 +104,7 @@ public class TunerManager
try try
{ {
retVal = tuner.getChannel( mThreadPoolManager, tunerChannel ); retVal = tuner.getChannel( tunerChannel );
} }
catch ( RejectedExecutionException ree ) catch ( RejectedExecutionException ree )
{ {

View File

@@ -35,7 +35,6 @@ import source.tuner.TunerClass;
import source.tuner.TunerConfiguration; import source.tuner.TunerConfiguration;
import source.tuner.TunerType; import source.tuner.TunerType;
import source.tuner.airspy.AirspyTunerController.BoardID; import source.tuner.airspy.AirspyTunerController.BoardID;
import controller.ThreadPoolManager;
public class AirspyTuner extends Tuner public class AirspyTuner extends Tuner
{ {
@@ -128,8 +127,8 @@ public class AirspyTuner extends Tuner
} }
@Override @Override
public TunerChannelSource getChannel( ThreadPoolManager threadPoolManager, public TunerChannelSource getChannel( TunerChannel channel )
TunerChannel channel ) throws RejectedExecutionException, SourceException throws RejectedExecutionException, SourceException
{ {
//TODO: this channel has a decimated sample rate of: //TODO: this channel has a decimated sample rate of:
// 10.0 MSps = 10,000,000 / 208 = 48076.923 // 10.0 MSps = 10,000,000 / 208 = 48076.923
@@ -138,7 +137,7 @@ public class AirspyTuner extends Tuner
//Consider implementing a fractional resampler to get a correct 48 kHz //Consider implementing a fractional resampler to get a correct 48 kHz
//output sample rate //output sample rate
return mController.getChannel( threadPoolManager, this, channel ); return mController.getChannel( this, channel );
} }
@Override @Override

View File

@@ -99,8 +99,8 @@ public class B100Tuner extends Tuner
} }
@Override @Override
public TunerChannelSource getChannel( ThreadPoolManager threadPoolManager, public TunerChannelSource getChannel( TunerChannel channel )
TunerChannel channel ) throws RejectedExecutionException, SourceException throws RejectedExecutionException, SourceException
{ {
//TODO: pass this call to the tuner controller //TODO: pass this call to the tuner controller
return null; return null;

View File

@@ -33,7 +33,6 @@ import source.tuner.TunerChannelSource;
import source.tuner.TunerClass; import source.tuner.TunerClass;
import source.tuner.TunerConfiguration; import source.tuner.TunerConfiguration;
import source.tuner.TunerType; import source.tuner.TunerType;
import controller.ThreadPoolManager;
public class FCDTuner extends MixerTuner public class FCDTuner extends MixerTuner
{ {
@@ -111,11 +110,10 @@ public class FCDTuner extends MixerTuner
} }
@Override @Override
public TunerChannelSource getChannel( ThreadPoolManager threadPoolManager, public TunerChannelSource getChannel( TunerChannel tunerChannel )
TunerChannel tunerChannel ) throws RejectedExecutionException, throws RejectedExecutionException, SourceException
SourceException
{ {
return mController.getChannel( threadPoolManager, this, tunerChannel ); return mController.getChannel( this, tunerChannel );
} }
/** /**

View File

@@ -35,7 +35,6 @@ import source.tuner.TunerClass;
import source.tuner.TunerConfiguration; import source.tuner.TunerConfiguration;
import source.tuner.TunerType; import source.tuner.TunerType;
import source.tuner.hackrf.HackRFTunerController.BoardID; import source.tuner.hackrf.HackRFTunerController.BoardID;
import controller.ThreadPoolManager;
public class HackRFTuner extends Tuner public class HackRFTuner extends Tuner
{ {
@@ -106,10 +105,10 @@ public class HackRFTuner extends Tuner
} }
@Override @Override
public TunerChannelSource getChannel( ThreadPoolManager threadPoolManager, public TunerChannelSource getChannel( TunerChannel channel )
TunerChannel channel ) throws RejectedExecutionException, SourceException throws RejectedExecutionException, SourceException
{ {
return mController.getChannel( threadPoolManager, this, channel ); return mController.getChannel( this, channel );
} }
@Override @Override

View File

@@ -148,8 +148,7 @@ public class RTL2832Tuner extends Tuner
} }
@Override @Override
public TunerChannelSource getChannel( ThreadPoolManager threadPoolManager, public TunerChannelSource getChannel( TunerChannel channel )
TunerChannel channel )
throws RejectedExecutionException, throws RejectedExecutionException,
SourceException SourceException
{ {
@@ -157,7 +156,7 @@ public class RTL2832Tuner extends Tuner
try try
{ {
source = mController.getChannel( threadPoolManager, this, channel ); source = mController.getChannel( this, channel );
} }
catch( Exception e ) catch( Exception e )
{ {