Merges changes from master branch to resolve issue with CPU and memory

and constraints on scheduled thread pool saturation.  Adds a new
scheduled thread pool for each processing chain that the components in
the chain can share, versus the single shared pool exposed by the
threadpoolmanager class.
This commit is contained in:
denny
2016-02-15 15:47:14 -05:00
parent e2a0451bfb
commit f7f6c2fc3e
45 changed files with 170 additions and 152 deletions

View File

@@ -18,6 +18,8 @@
package module;
import java.util.concurrent.ScheduledExecutorService;
public abstract class Module
{
/**
@@ -57,9 +59,9 @@ public abstract class Module
public abstract void reset();
/**
* Start processing.
* Start processing. Provides an executor to use for scheduling tasks
*/
public abstract void start();
public abstract void start( ScheduledExecutorService executor );
/**
* Stop processing

View File

@@ -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;
@@ -67,6 +68,7 @@ import audio.metadata.Metadata;
import audio.squelch.ISquelchStateListener;
import audio.squelch.ISquelchStateProvider;
import audio.squelch.SquelchState;
import controller.NamingThreadFactory;
import controller.ThreadPoolManager;
import controller.channel.Channel.ChannelType;
import controller.channel.ChannelEvent;
@@ -108,8 +110,8 @@ public class ProcessingChain implements IChannelEventListener
private Broadcaster<RealBuffer> mUnFilteredRealBufferBroadcaster = new Broadcaster<>();
private Broadcaster<SquelchState> mSquelchStateBroadcaster = new Broadcaster<>();
private ThreadPoolManager mThreadPoolManager;
private ScheduledFuture<?> mBufferProcessingTask;
private ScheduledExecutorService mScheduledExecutorService;
private String mName;
private AtomicBoolean mRunning = new AtomicBoolean();
protected Source mSource;
@@ -117,19 +119,18 @@ public class ProcessingChain implements IChannelEventListener
private CallEventModel mCallEventModel;
private ChannelState mChannelState;
private MessageActivityModel mMessageActivityModel;
public ProcessingChain( ChannelType channelType )
/**
* Creates a processing chain for managing a set of modules
*
* @param name for thread pool naming each thread
* @param channelType
*/
public ProcessingChain( String name, ChannelType channelType )
{
this( new ThreadPoolManager(), channelType );
}
public ProcessingChain( ThreadPoolManager threadManager,
ChannelType channelType )
{
mThreadPoolManager = threadManager;
mChannelState = new ChannelState( mThreadPoolManager, channelType );
mName = name;
mChannelState = new ChannelState( channelType );
addModule( mChannelState );
mCallEventModel = new CallEventModel();
@@ -169,10 +170,6 @@ public class ProcessingChain implements IChannelEventListener
mModules.clear();
mThreadPoolManager = null;
mBufferProcessingTask = null;
mAudioPacketBroadcaster.dispose();
mCallEventBroadcaster.dispose();
mChannelEventBroadcaster.dispose();
@@ -445,19 +442,6 @@ public class ProcessingChain implements IChannelEventListener
module.reset();
}
/* Start each of the modules */
for( Module module: mModules )
{
try
{
module.start();
}
catch( Exception e )
{
mLog.error( "Error starting module", e );
}
}
/* Register with the source to receive sample data. Setup a
* timer task to process the buffer queues 50 times a second
* (every 20 ms) */
@@ -492,8 +476,22 @@ public class ProcessingChain implements IChannelEventListener
{
mLog.error( "Error getting frequency from tuner channel source", e );
}
tcs.start();
}
mScheduledExecutorService = Executors.newScheduledThreadPool(
1, new NamingThreadFactory( "channel " + mName ) );
/* Start each of the modules */
for( Module module: mModules )
{
try
{
module.start( mScheduledExecutorService );
}
catch( Exception e )
{
mLog.error( "Error starting module", e );
}
}
for( Listener<RealBuffer> listener: mUnFilteredRealBufferBroadcaster.getListeners() )
@@ -536,12 +534,6 @@ public class ProcessingChain implements IChannelEventListener
mSource = null;
}
if( mBufferProcessingTask != null )
{
mThreadPoolManager.cancel( mBufferProcessingTask );
mBufferProcessingTask = null;
}
/* Stop each of the modules */
for( Module module: mModules )
{

View File

@@ -18,6 +18,8 @@
package module.decode.am;
import java.util.concurrent.ScheduledExecutorService;
import module.decode.DecoderType;
import module.decode.PrimaryDecoder;
import module.decode.config.DecodeConfiguration;
@@ -41,7 +43,7 @@ public class AMDecoder extends PrimaryDecoder
}
@Override
public void start()
public void start( ScheduledExecutorService executor )
{
}

View File

@@ -25,6 +25,7 @@ import instrument.tap.stream.FloatBufferTap;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ScheduledExecutorService;
import module.decode.Decoder;
import module.decode.DecoderType;
@@ -180,7 +181,7 @@ public class Fleetsync2Decoder extends Decoder implements IFilteredRealBufferLis
}
@Override
public void start()
public void start( ScheduledExecutorService executor )
{
}

View File

@@ -19,6 +19,7 @@ package module.decode.fleetsync2;
import java.util.Iterator;
import java.util.TreeSet;
import java.util.concurrent.ScheduledExecutorService;
import message.Message;
import module.decode.DecoderType;
@@ -58,7 +59,7 @@ public class Fleetsync2DecoderState extends DecoderState
}
@Override
public void start()
public void start( ScheduledExecutorService executor )
{
}

View File

@@ -26,6 +26,7 @@ import instrument.tap.stream.FloatTap;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ScheduledExecutorService;
import module.decode.Decoder;
import module.decode.DecoderType;
@@ -198,7 +199,7 @@ public class LJ1200Decoder extends Decoder implements IFilteredRealBufferListene
}
@Override
public void start()
public void start( ScheduledExecutorService executor )
{
}

View File

@@ -22,6 +22,7 @@ import java.util.Collections;
import java.util.List;
import java.util.Set;
import java.util.TreeSet;
import java.util.concurrent.ScheduledExecutorService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -58,7 +59,7 @@ public class LJ1200DecoderState extends DecoderState
}
@Override
public void start()
public void start( ScheduledExecutorService executor )
{
}

View File

@@ -23,6 +23,7 @@ import instrument.tap.TapGroup;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ScheduledExecutorService;
import message.MessageDirection;
import module.decode.Decoder;
@@ -120,7 +121,7 @@ public class LTRNetDecoder extends Decoder
}
@Override
public void start()
public void start( ScheduledExecutorService executor )
{
}

View File

@@ -21,6 +21,7 @@ import java.text.DecimalFormat;
import java.util.HashMap;
import java.util.Iterator;
import java.util.TreeSet;
import java.util.concurrent.ScheduledExecutorService;
import message.Message;
import module.decode.DecoderType;
@@ -777,7 +778,7 @@ public class LTRNetDecoderState extends DecoderState
}
@Override
public void start()
public void start( ScheduledExecutorService executor )
{
}

View File

@@ -17,6 +17,8 @@
******************************************************************************/
package module.decode.ltrstandard;
import java.util.concurrent.ScheduledExecutorService;
import message.MessageDirection;
import module.decode.Decoder;
import module.decode.DecoderType;
@@ -83,7 +85,7 @@ public class LTRStandardDecoder extends Decoder implements IUnFilteredRealBuffer
}
@Override
public void start()
public void start( ScheduledExecutorService executor )
{
}

View File

@@ -22,6 +22,7 @@ import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.TreeSet;
import java.util.concurrent.ScheduledExecutorService;
import message.Message;
import module.decode.DecoderType;
@@ -61,7 +62,7 @@ public class LTRStandardDecoderState extends DecoderState
}
@Override
public void start()
public void start( ScheduledExecutorService executor )
{
}

View File

@@ -26,6 +26,7 @@ import instrument.tap.stream.FloatTap;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ScheduledExecutorService;
import module.decode.Decoder;
import module.decode.DecoderType;
@@ -210,7 +211,7 @@ public class MDCDecoder extends Decoder implements IFilteredRealBufferListener,
}
@Override
public void start()
public void start( ScheduledExecutorService executor )
{
// TODO Auto-generated method stub

View File

@@ -19,6 +19,7 @@ package module.decode.mdc1200;
import java.util.Iterator;
import java.util.TreeSet;
import java.util.concurrent.ScheduledExecutorService;
import message.Message;
import module.decode.DecoderType;
@@ -54,7 +55,7 @@ public class MDCDecoderState extends DecoderState
}
@Override
public void start()
public void start( ScheduledExecutorService executor )
{
}

View File

@@ -26,6 +26,7 @@ import instrument.tap.stream.FloatTap;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ScheduledExecutorService;
import module.decode.Decoder;
import module.decode.DecoderType;
@@ -306,7 +307,7 @@ public class MPT1327Decoder extends Decoder
}
@Override
public void start()
public void start( ScheduledExecutorService executor )
{
// TODO Auto-generated method stub

View File

@@ -23,6 +23,7 @@ import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.TreeSet;
import java.util.concurrent.ScheduledExecutorService;
import message.Message;
import module.decode.DecoderType;
@@ -85,7 +86,7 @@ public class MPT1327DecoderState extends DecoderState
}
@Override
public void start()
public void start( ScheduledExecutorService executor )
{
}

View File

@@ -17,6 +17,8 @@
******************************************************************************/
package module.decode.nbfm;
import java.util.concurrent.ScheduledExecutorService;
import module.decode.DecoderType;
import module.decode.PrimaryDecoder;
import module.decode.config.DecodeConfiguration;
@@ -40,7 +42,7 @@ public class NBFMDecoder extends PrimaryDecoder
}
@Override
public void start()
public void start( ScheduledExecutorService executor )
{
}

View File

@@ -25,6 +25,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TreeSet;
import java.util.concurrent.ScheduledExecutorService;
import message.Message;
import module.decode.DecoderType;
@@ -204,7 +205,7 @@ public class P25DecoderState extends DecoderState
}
@Override
public void start()
public void start( ScheduledExecutorService executor )
{
}

View File

@@ -25,6 +25,7 @@ import instrument.tap.stream.FloatTap;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ScheduledExecutorService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -229,7 +230,7 @@ public class P25_C4FMDecoder extends P25Decoder
}
@Override
public void start()
public void start( ScheduledExecutorService executor )
{
// TODO Auto-generated method stub

View File

@@ -26,6 +26,7 @@ import instrument.tap.stream.QPSKTap;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ScheduledExecutorService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -222,7 +223,7 @@ public class P25_LSMDecoder extends P25Decoder implements IComplexBufferListener
}
@Override
public void start()
public void start( ScheduledExecutorService executor )
{
// TODO Auto-generated method stub

View File

@@ -1,5 +1,7 @@
package module.decode.p25.audio;
import java.util.concurrent.ScheduledExecutorService;
import jmbe.iface.AudioConversionLibrary;
import jmbe.iface.AudioConverter;
import message.IMessageListener;
@@ -78,7 +80,7 @@ public class P25AudioModule extends Module implements Listener<Message>,
}
@Override
public void start()
public void start( ScheduledExecutorService executor )
{
}

View File

@@ -23,6 +23,7 @@ import instrument.tap.TapGroup;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ScheduledExecutorService;
import module.decode.Decoder;
import module.decode.DecoderType;
@@ -109,7 +110,7 @@ public class PassportDecoder extends Decoder
}
@Override
public void start()
public void start( ScheduledExecutorService executor )
{
}

View File

@@ -23,6 +23,7 @@ import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.TreeSet;
import java.util.concurrent.ScheduledExecutorService;
import message.Message;
import module.decode.DecoderType;
@@ -75,7 +76,7 @@ public class PassportDecoderState extends DecoderState
}
@Override
public void start()
public void start( ScheduledExecutorService executor )
{
}

View File

@@ -17,6 +17,8 @@
******************************************************************************/
package module.decode.state;
import java.util.concurrent.ScheduledExecutorService;
import message.Message;
import module.decode.DecoderType;
import module.decode.state.DecoderStateEvent.Event;
@@ -90,7 +92,7 @@ public class AlwaysUnsquelchedDecoderState extends DecoderState
}
@Override
public void start()
public void start( ScheduledExecutorService executor )
{
broadcast( new Metadata( MetadataType.TO, mChannelName, false ) );
broadcast( new DecoderStateEvent( this, Event.ALWAYS_UNSQUELCH, State.IDLE ) );

View File

@@ -18,6 +18,7 @@
package module.decode.state;
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;
@@ -74,7 +75,6 @@ public class ChannelState extends Module implements ICallEventProvider,
private AtomicBoolean mMonitoring = new AtomicBoolean();
private ScheduledFuture<?> mStateMonitorFuture;
private ThreadPoolManager mThreadPoolManager;
private ChannelType mChannelType;
private TrafficChannelStatusListener mTrafficChannelStatusListener;
@@ -106,9 +106,8 @@ public class ChannelState extends Module implements ICallEventProvider,
* ended, while continuing to display the call details for the user
* TEARDOWN Indicates a traffic channel that will be torn down for reuse.
*/
public ChannelState( ThreadPoolManager manager, ChannelType channelType )
public ChannelState( ChannelType channelType )
{
mThreadPoolManager = manager;
mChannelType = channelType;
}
@@ -126,7 +125,7 @@ public class ChannelState extends Module implements ICallEventProvider,
}
@Override
public void start()
public void start( ScheduledExecutorService executor )
{
//Start a monitoring task to check for FADE and END event timeouts
if( mMonitoring.compareAndSet( false, true ) )
@@ -135,8 +134,8 @@ public class ChannelState extends Module implements ICallEventProvider,
{
try
{
mStateMonitorFuture = mThreadPoolManager.scheduleFixedRate(
ThreadType.DECODER, mStateMonitor, 20, TimeUnit.MILLISECONDS );
mStateMonitorFuture = executor.scheduleAtFixedRate(
mStateMonitor, 0, 20, TimeUnit.MILLISECONDS );
}
catch( RejectedExecutionException ree )
{
@@ -157,7 +156,7 @@ public class ChannelState extends Module implements ICallEventProvider,
{
if( mStateMonitorFuture != null )
{
boolean success = mThreadPoolManager.cancel( mStateMonitorFuture );
boolean success = mStateMonitorFuture.cancel( true );
if( !success )
{
@@ -175,7 +174,6 @@ public class ChannelState extends Module implements ICallEventProvider,
mChangedAttributeListener = null;
mDecoderStateListener = null;
mSquelchStateListener = null;
mThreadPoolManager = null;
mStateMonitor = null;
}

View File

@@ -4,6 +4,7 @@ import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledExecutorService;
import module.Module;
import module.ProcessingChain;
@@ -86,13 +87,21 @@ public class TrafficChannelManager extends Module
mSystem = system;
mSite = site;
mAliasListName = aliasListName;
// mTrafficChannelTimeout = trafficChannelTimeout;
mTrafficChannelPoolMaximumSize = trafficChannelPoolSize;
}
@Override
public void dispose()
{
for( Channel trafficChannel: mTrafficChannelPool )
{
mChannelModel.broadcast( new ChannelEvent( trafficChannel, Event.REQUEST_DISABLE ) );
}
mTrafficChannelPool.clear();
mTrafficChannelsInUse.clear();
mCallEventListener = null;
mDecodeConfiguration = null;
mPreviousDoNotMonitorCallEvent = null;
@@ -350,7 +359,7 @@ public class TrafficChannelManager extends Module
}
@Override
public void start()
public void start( ScheduledExecutorService executor )
{
}

View File

@@ -26,6 +26,7 @@ import instrument.tap.stream.FloatTap;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ScheduledExecutorService;
import module.decode.Decoder;
import module.decode.DecoderType;
@@ -198,7 +199,7 @@ public class Tait1200Decoder extends Decoder implements IFilteredRealBufferListe
}
@Override
public void start()
public void start( ScheduledExecutorService executor )
{
// TODO Auto-generated method stub

View File

@@ -21,6 +21,7 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.TreeSet;
import java.util.concurrent.ScheduledExecutorService;
import org.jdesktop.swingx.mapviewer.GeoPosition;
@@ -60,7 +61,7 @@ public class Tait1200DecoderState extends DecoderState
}
@Override
public void start()
public void start( ScheduledExecutorService executor )
{
}

View File

@@ -17,6 +17,8 @@
******************************************************************************/
package module.demodulate.am;
import java.util.concurrent.ScheduledExecutorService;
import module.Module;
import sample.Listener;
import sample.complex.ComplexBuffer;
@@ -111,7 +113,7 @@ public class AMDemodulatorModule extends Module
}
@Override
public void start()
public void start( ScheduledExecutorService executor )
{
}

View File

@@ -17,6 +17,8 @@
******************************************************************************/
package module.demodulate.audio;
import java.util.concurrent.ScheduledExecutorService;
import module.Module;
import org.slf4j.Logger;
@@ -84,7 +86,7 @@ public class DemodulatedAudioFilterModule extends Module
}
@Override
public void start()
public void start( ScheduledExecutorService executor )
{
}

View File

@@ -17,6 +17,8 @@
******************************************************************************/
package module.demodulate.fm;
import java.util.concurrent.ScheduledExecutorService;
import module.Module;
import org.slf4j.Logger;
@@ -97,7 +99,7 @@ public class FMDemodulatorModule extends Module
}
@Override
public void start()
public void start( ScheduledExecutorService executor )
{
}

View File

@@ -58,6 +58,5 @@ public class CallEventLogger extends EventLogger implements ICallEventListener,
@Override
public void reset()
{
super.start();
}
}

View File

@@ -24,6 +24,7 @@ import java.io.IOException;
import java.io.OutputStreamWriter;
import java.io.Writer;
import java.nio.file.Path;
import java.util.concurrent.ScheduledExecutorService;
import module.Module;
@@ -67,7 +68,8 @@ public abstract class EventLogger extends Module
public abstract String getHeader();
public void start()
@Override
public void start( ScheduledExecutorService executor )
{
if( mLogFile == null )
{

View File

@@ -58,7 +58,6 @@ public class MessageEventLogger extends EventLogger implements
@Override
public void reset()
{
super.start();
}
@Override