From 183eb2f7c78f37682d4ab98af9869449885d8f2c Mon Sep 17 00:00:00 2001 From: Denny Date: Fri, 10 Feb 2017 06:36:57 -0500 Subject: [PATCH] Resolves #165 - P25 decoding channel stops arbitrarily and never recovers. Issue identified with Patch Group management and circular dependencies that can develop because of stale patch groups. Refactored all patch group management into a new PatchGroupManager. Patch groups are now automatically deleted if we don't receive any update messages within 5 seconds. The PatchGroupAlias was also updated to not roll-up attributes from subordinate patched group aliases when those aliases are also patch group aliases -- this eliminates the circular dependency chaining. Added IOverflowListener interface and updated Sources to support overflow listeners. Updated ChannelState to register as an overflow listener so that it can visually depict to the user when the channel enters/exits source overflow state. Added new color pattern for OVERFLOW with yellow text on red. --- CHANGELOG | 4 +- src/alias/PatchGroupAlias.java | 32 ++- .../metadata/ChannelMetadataPanel.java | 4 + .../metadata/ChannelMetadataViewer.java | 2 +- src/channel/state/ChannelState.java | 26 +- src/channel/state/DecoderState.java | 8 + src/channel/state/State.java | 180 +++++++------ src/controller/NamingThreadFactory.java | 32 ++- src/module/ProcessingChain.java | 4 + src/module/decode/p25/P25DecoderState.java | 205 +-------------- src/module/decode/p25/PatchGroup.java | 102 ++++++++ src/module/decode/p25/PatchGroupManager.java | 243 ++++++++++++++++++ src/sample/OverflowableTransferQueue.java | 32 +-- src/sample/real/IOverflowListener.java | 28 ++ src/source/Source.java | 23 ++ src/source/tuner/TunerChannelSource.java | 37 +-- .../tuner/usb/USBTransferProcessor.java | 7 +- 17 files changed, 615 insertions(+), 354 deletions(-) create mode 100644 src/module/decode/p25/PatchGroup.java create mode 100644 src/module/decode/p25/PatchGroupManager.java create mode 100644 src/sample/real/IOverflowListener.java diff --git a/CHANGELOG b/CHANGELOG index b040d09a..5fee2ff5 100644 --- a/CHANGELOG +++ b/CHANGELOG @@ -1,10 +1,10 @@ 0.3.0-beta1 20170129 + - REQUIRES: Java 8 (or newer) for compatibility and eventual changeover to JavaFX - FEATURE: audio streaming - adds capability to stream to multiple remote broadcast servers (Broadcastify, Icecast v2.3 and v2.4+, and Shoutcast V1 and V2). Each stream can be independently configured with a delay. Audio calls are directed to one or more streams by attaching one or more stream identifiers to each alias. - - Requires Java 8 (or newer) for compatibility and eventual changeover to JavaFX - Extensive GUI updates for tuner, channel and alias management. Includes new 'Now Playing' channel view and 'Detail' view. - Updated alias list management of alias id values and wildcards. Weighting is now @@ -17,7 +17,7 @@ - Implemented bounded queues for both tuner and individual channel IQ sample buffers. Bounded queues have ability to enter/exit overflow state when queue size reaches overflow threshold and reset when queue size falls below the reset - threshold. + threshold. Channel state will show OVERFLOW in red/yellow. - Updated airspy controller for airspy-mini specific sample rate handling. - New developer remez filter designer tool and internal usage diff --git a/src/alias/PatchGroupAlias.java b/src/alias/PatchGroupAlias.java index b4788a86..b2c0d2fa 100644 --- a/src/alias/PatchGroupAlias.java +++ b/src/alias/PatchGroupAlias.java @@ -168,11 +168,14 @@ public class PatchGroupAlias extends Alias for(Alias alias: mPatchedAliases) { - for(AliasAction action: alias.getAction()) + if(!(alias instanceof PatchGroupAlias)) { - if(!aliasActions.contains(action)) + for(AliasAction action: alias.getAction()) { - aliasActions.add(action); + if(!aliasActions.contains(action)) + { + aliasActions.add(action); + } } } } @@ -185,7 +188,7 @@ public class PatchGroupAlias extends Alias { for(Alias alias: mPatchedAliases) { - if(alias.hasActions()) + if(!(alias instanceof PatchGroupAlias) && alias.hasActions()) { return true; } @@ -201,7 +204,9 @@ public class PatchGroupAlias extends Alias for(Alias alias: mPatchedAliases) { - if(alias.hasCallPriority() && alias.getCallPriority() < highestPriority) + if(!(alias instanceof PatchGroupAlias) && + alias.hasCallPriority() && + alias.getCallPriority() < highestPriority) { highestPriority = alias.getCallPriority(); } @@ -220,7 +225,7 @@ public class PatchGroupAlias extends Alias for(Alias alias: mPatchedAliases) { - if(alias.hasCallPriority()) + if(!(alias instanceof PatchGroupAlias) && alias.hasCallPriority()) { return true; } @@ -239,7 +244,7 @@ public class PatchGroupAlias extends Alias for(Alias alias: mPatchedAliases) { - if(alias.isRecordable()) + if(!(alias instanceof PatchGroupAlias) && alias.isRecordable()) { return true; } @@ -258,7 +263,7 @@ public class PatchGroupAlias extends Alias for(Alias alias: mPatchedAliases) { - if(alias.isStreamable()) + if(!(alias instanceof PatchGroupAlias) && alias.isStreamable()) { return true; } @@ -279,11 +284,14 @@ public class PatchGroupAlias extends Alias for(Alias alias: mPatchedAliases) { - for(BroadcastChannel broadcastChannel: alias.getBroadcastChannels()) + if(!(alias instanceof PatchGroupAlias)) { - if(!broadcastChannels.contains(broadcastChannel)) + for(BroadcastChannel broadcastChannel: alias.getBroadcastChannels()) { - broadcastChannels.add(broadcastChannel); + if(!broadcastChannels.contains(broadcastChannel)) + { + broadcastChannels.add(broadcastChannel); + } } } } @@ -301,7 +309,7 @@ public class PatchGroupAlias extends Alias for(Alias alias: mPatchedAliases) { - if(alias.hasBroadcastChannel(channel)) + if(!(alias instanceof PatchGroupAlias) && alias.hasBroadcastChannel(channel)) { return true; } diff --git a/src/channel/metadata/ChannelMetadataPanel.java b/src/channel/metadata/ChannelMetadataPanel.java index e9ecef00..5fa80231 100644 --- a/src/channel/metadata/ChannelMetadataPanel.java +++ b/src/channel/metadata/ChannelMetadataPanel.java @@ -210,6 +210,10 @@ public class ChannelMetadataPanel extends JPanel implements ListSelectionListene background = SystemProperties.getInstance().get(PROPERTY_PREFIX_BACKGROUND + "idle", Color.WHITE); foreground = SystemProperties.getInstance().get(PROPERTY_PREFIX_FOREGROUND + "idle", Color.BLUE); break; + case OVERFLOW: + background = SystemProperties.getInstance().get(PROPERTY_PREFIX_BACKGROUND + "overflow", Color.RED); + foreground = SystemProperties.getInstance().get(PROPERTY_PREFIX_FOREGROUND + "overflow", Color.YELLOW); + break; case RESET: background = SystemProperties.getInstance().get(PROPERTY_PREFIX_BACKGROUND + "reset", Color.PINK); foreground = SystemProperties.getInstance().get(PROPERTY_PREFIX_FOREGROUND + "reset", Color.YELLOW); diff --git a/src/channel/metadata/ChannelMetadataViewer.java b/src/channel/metadata/ChannelMetadataViewer.java index 53996650..e7efc2c2 100644 --- a/src/channel/metadata/ChannelMetadataViewer.java +++ b/src/channel/metadata/ChannelMetadataViewer.java @@ -51,7 +51,7 @@ public class ChannelMetadataViewer extends JPanel mChannelDetailPanel = new ChannelDetailPanel(channelProcessingManager); mCallEventPanel = new CallEventPanel(iconManager); mMessageActivityPanel = new MessageActivityPanel(channelProcessingManager); - mChannelSpectrumPanel = new ChannelSpectrumPanel(settingsManager, channelProcessingManager); +// mChannelSpectrumPanel = new ChannelSpectrumPanel(settingsManager, channelProcessingManager); mChannelMetadataPanel = new ChannelMetadataPanel(channelProcessingManager, iconManager); init(); diff --git a/src/channel/state/ChannelState.java b/src/channel/state/ChannelState.java index 66a22984..768c1062 100644 --- a/src/channel/state/ChannelState.java +++ b/src/channel/state/ChannelState.java @@ -35,6 +35,7 @@ import module.decode.event.ICallEventProvider; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import sample.Listener; +import sample.real.IOverflowListener; import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.ScheduledExecutorService; @@ -42,7 +43,7 @@ import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; public class ChannelState extends Module implements ICallEventProvider, IDecoderStateEventListener, - IDecoderStateEventProvider, ISquelchStateProvider, IAttributeChangeRequestListener + IDecoderStateEventProvider, ISquelchStateProvider, IAttributeChangeRequestListener, IOverflowListener { private final static Logger mLog = LoggerFactory.getLogger(ChannelState.class); @@ -63,6 +64,7 @@ public class ChannelState extends Module implements ICallEventProvider, IDecoder private boolean mSquelchLocked = false; private boolean mSelected = false; + private boolean mSourceOverflow = false; private long mStandardChannelFadeTimeout = FADE_TIMEOUT_DELAY; private long mTrafficChannelFadeTimeout = DecodeConfiguration.DEFAULT_CALL_TIMEOUT_SECONDS * 1000; @@ -123,6 +125,7 @@ public class ChannelState extends Module implements ICallEventProvider, IDecoder mState = State.IDLE; mMutableMetadata.receive(new AttributeChangeRequest(Attribute.CHANNEL_STATE, mState)); + mSourceOverflow = false; } @Override @@ -350,15 +353,26 @@ public class ChannelState extends Module implements ICallEventProvider, IDecoder break; } - mMutableMetadata.receive(new AttributeChangeRequest(Attribute.CHANNEL_STATE, mState)); - } - else - { -// mLog.debug( "Can't change from [" + mState + "] to [" + state + "]" ); + mMutableMetadata.receive(new AttributeChangeRequest(Attribute.CHANNEL_STATE, + mSourceOverflow ? State.OVERFLOW : mState)); } } } + /** + * This method is invoked if the source buffer provider goes into overflow state. Since this is an external state, + * we use the mSourceOverflow variable to override the internal state reported to external listeners. + * @param overflow true to indicate an overflow state + */ + @Override + public void sourceOverflow(boolean overflow) + { + mSourceOverflow = overflow; + + mMutableMetadata.receive(new AttributeChangeRequest(Attribute.CHANNEL_STATE, + mSourceOverflow ? State.OVERFLOW : mState)); + } + /** * Sets the state and processes related actions */ diff --git a/src/channel/state/DecoderState.java b/src/channel/state/DecoderState.java index a27dd7be..8f7ecbf8 100644 --- a/src/channel/state/DecoderState.java +++ b/src/channel/state/DecoderState.java @@ -51,6 +51,14 @@ public abstract class DecoderState extends Module implements ActivitySummaryProv return mAliasList != null; } + /** + * Provides subclass reference to the call event broadcaster + */ + protected Broadcaster getCallEventBroadcaster() + { + return mCallEventBroadcaster; + } + /** * Reset the decoder state to prepare for processing a different sample * source diff --git a/src/channel/state/State.java b/src/channel/state/State.java index 83336d2c..f2c75767 100644 --- a/src/channel/state/State.java +++ b/src/channel/state/State.java @@ -1,17 +1,17 @@ /******************************************************************************* * SDR Trunk * Copyright (C) 2014-2016 Dennis Sheirer - * + * * This program is free software: you can redistribute it and/or modify * it under the terms of the GNU General Public License as published by * the Free Software Foundation, either version 3 of the License, or * (at your option) any later version. - * + * * This program is distributed in the hope that it will be useful, * but WITHOUT ANY WARRANTY; without even the implied warranty of * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the * GNU General Public License for more details. - * + * * You should have received a copy of the GNU General Public License * along with this program. If not, see ******************************************************************************/ @@ -23,98 +23,94 @@ import java.util.EnumSet; * Details the set of states for a channel and the allowable transition states */ public enum State -{ - IDLE( "IDLE" ) - { - @Override - public boolean canChangeTo( State state ) +{ + IDLE("IDLE") { - return state != TEARDOWN && - state != RESET; - } - }, - CALL( "CALL" ) - { - @Override - public boolean canChangeTo( State state ) + @Override + public boolean canChangeTo(State state) + { + return state != TEARDOWN && state != RESET && state != OVERFLOW; + } + }, + CALL("CALL") { - return state == CONTROL || - state == DATA || - state == ENCRYPTED || - state == FADE || - state == TEARDOWN; - } - }, - DATA( "DATA" ) - { - @Override - public boolean canChangeTo( State state ) + @Override + public boolean canChangeTo(State state) + { + return state == CONTROL || state == DATA || state == ENCRYPTED || state == FADE || state == TEARDOWN; + } + }, + DATA("DATA") { - return state == CALL || - state == CONTROL || - state == ENCRYPTED || - state == FADE || - state == TEARDOWN; - } - }, - ENCRYPTED( "ENCRYPTED" ) + @Override + public boolean canChangeTo(State state) + { + return state == CALL || state == CONTROL || state == ENCRYPTED || state == FADE || state == TEARDOWN; + } + }, + ENCRYPTED("ENCRYPTED") + { + @Override + public boolean canChangeTo(State state) + { + return state == FADE || state == TEARDOWN; + } + }, + CONTROL("CONTROL") + { + @Override + public boolean canChangeTo(State state) + { + return state == IDLE || state == FADE; + } + }, + FADE("FADE") + { + @Override + public boolean canChangeTo(State state) + { + return state != FADE && state != RESET && state != OVERFLOW; + } + }, + OVERFLOW("OVERFLOW") + { + @Override + public boolean canChangeTo(State state) + { + return false; + } + }, + RESET("RESET") + { + @Override + public boolean canChangeTo(State state) + { + return state == IDLE; + } + }, + TEARDOWN("TEARDOWN") + { + @Override + public boolean canChangeTo(State state) + { + return state == RESET; + } + }; + + private String mDisplayValue; + + public static final EnumSet CALL_STATES = EnumSet.of(CALL, CONTROL, DATA, ENCRYPTED); + public static final EnumSet IDLE_STATES = EnumSet.of(IDLE, FADE); + + private State(String displayValue) { - @Override - public boolean canChangeTo( State state ) - { - return state == FADE || state == TEARDOWN; - } - }, - CONTROL( "CONTROL" ) + mDisplayValue = displayValue; + } + + public abstract boolean canChangeTo(State state); + + public String getDisplayValue() { - @Override - public boolean canChangeTo( State state ) - { - return state == IDLE || - state == FADE; - } - }, - FADE( "FADE" ) - { - @Override - public boolean canChangeTo( State state ) - { - return state != FADE && - state != RESET; - } - }, - RESET( "RESET" ) - { - @Override - public boolean canChangeTo(State state) - { - return state == IDLE; - } - }, - - TEARDOWN( "TEARDOWN" ) - { - @Override - public boolean canChangeTo(State state) - { - return state == RESET; - } - }; - - private String mDisplayValue; - - public static final EnumSet CALL_STATES = EnumSet.of( CALL, CONTROL, DATA, ENCRYPTED ); - public static final EnumSet IDLE_STATES = EnumSet.of( IDLE, FADE ); - - private State( String displayValue ) - { - mDisplayValue = displayValue; - } - - public abstract boolean canChangeTo( State state ); - - public String getDisplayValue() - { - return mDisplayValue; - } + return mDisplayValue; + } } \ No newline at end of file diff --git a/src/controller/NamingThreadFactory.java b/src/controller/NamingThreadFactory.java index ae3302b1..ed7bd0dc 100644 --- a/src/controller/NamingThreadFactory.java +++ b/src/controller/NamingThreadFactory.java @@ -1,5 +1,9 @@ package controller; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import util.ThreadPool; + import java.util.concurrent.ThreadFactory; import java.util.concurrent.atomic.AtomicInteger; @@ -8,24 +12,23 @@ import java.util.concurrent.atomic.AtomicInteger; */ public class NamingThreadFactory implements ThreadFactory { - static final AtomicInteger poolNumber = new AtomicInteger(1); + private final static Logger mLog = LoggerFactory.getLogger(ThreadPool.class); + + private static final AtomicInteger mPoolNumber = new AtomicInteger(1); - final ThreadGroup mThreadGroup; + private final ThreadGroup mThreadGroup; - final AtomicInteger mThreadNumber = new AtomicInteger(1); + private final AtomicInteger mThreadNumber = new AtomicInteger(1); - final String mNamePrefix; + private final String mNamePrefix; public NamingThreadFactory( String prefix ) { SecurityManager s = System.getSecurityManager(); - mThreadGroup = (s != null)? s.getThreadGroup() : - Thread.currentThread().getThreadGroup(); + mThreadGroup = (s != null)? s.getThreadGroup() : Thread.currentThread().getThreadGroup(); - mNamePrefix = prefix + " pool-" + - poolNumber.getAndIncrement() + - "-thread-"; + mNamePrefix = prefix + " pool-" + mPoolNumber.getAndIncrement() + "-thread-"; } public Thread newThread( Runnable runnable ) @@ -42,7 +45,16 @@ public class NamingThreadFactory implements ThreadFactory { thread.setPriority( Thread.NORM_PRIORITY ); } - + + thread.setUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() + { + @Override + public void uncaughtException(Thread t, Throwable e) + { + mLog.error("Error while executing runnable in scheduled thread pool [" + t.getName() + "]", e); + } + }); + return thread; } } \ No newline at end of file diff --git a/src/module/ProcessingChain.java b/src/module/ProcessingChain.java index d8a13e9a..fdf60e86 100644 --- a/src/module/ProcessingChain.java +++ b/src/module/ProcessingChain.java @@ -538,6 +538,9 @@ public class ProcessingChain implements IChannelEventListener module.reset(); } + //Setup the channel state to monitor source overflow conditions + mSource.setOverflowListener(mChannelState); + /* Register with the source to receive sample data. Setup a * timer task to process the buffer queues 50 times a second * (every 20 ms) */ @@ -606,6 +609,7 @@ public class ProcessingChain implements IChannelEventListener if(mSource != null) { mSource.stop(); + mSource.setOverflowListener(null); switch(mSource.getSampleType()) { diff --git a/src/module/decode/p25/P25DecoderState.java b/src/module/decode/p25/P25DecoderState.java index 156c4fdd..2b835de3 100644 --- a/src/module/decode/p25/P25DecoderState.java +++ b/src/module/decode/p25/P25DecoderState.java @@ -19,9 +19,7 @@ package module.decode.p25; import alias.Alias; import alias.AliasList; -import alias.PatchGroupAlias; import alias.id.AliasIDType; -import alias.id.talkgroup.TalkgroupID; import channel.metadata.AliasedStringAttributeMonitor; import channel.metadata.Attribute; import channel.metadata.AttributeChangeRequest; @@ -163,7 +161,6 @@ public class P25DecoderState extends DecoderState private Map mBands = new HashMap<>(); private Map mRegistrations = new HashMap<>(); private Map mNeighborMap = new HashMap<>(); - private Map> mPatchGroupMap = new HashMap<>(); private String mLastCommandEventID; private String mLastPageEventID; @@ -188,6 +185,7 @@ public class P25DecoderState extends DecoderState private P25CallEvent mCurrentCallEvent; private List mCallDetectTalkgroups = new ArrayList<>(); private Map mChannelCallMap = new HashMap<>(); + private PatchGroupManager mPatchGroupManager; public P25DecoderState(AliasList aliasList, ChannelType channelType, @@ -199,6 +197,7 @@ public class P25DecoderState extends DecoderState mModulation = modulation; mIgnoreDataCalls = ignoreDataCalls; + mPatchGroupManager = new PatchGroupManager(aliasList, getCallEventBroadcaster()); mSiteAttributeMonitor = new AliasedStringAttributeMonitor(Attribute.NETWORK_ID_2, getAttributeChangeRequestListener(), getAliasList(), AliasIDType.SITE); mFromTalkgroupMonitor = new AliasedStringAttributeMonitor(Attribute.PRIMARY_ADDRESS_FROM, @@ -2719,6 +2718,9 @@ public class P25DecoderState extends DecoderState switch(((MotorolaTSBKMessage)tsbk).getMotorolaOpcode()) { case PATCH_GROUP_CHANNEL_GRANT: + //Cleanup patch groups - auto-expire any patch groups before we allocate a channel + mPatchGroupManager.cleanupPatchGroups(); + PatchGroupVoiceChannelGrant pgvcg = (PatchGroupVoiceChannelGrant)tsbk; channel = pgvcg.getChannel(); @@ -2754,6 +2756,9 @@ public class P25DecoderState extends DecoderState mChannelCallMap.get(channel))); break; case PATCH_GROUP_CHANNEL_GRANT_UPDATE: + //Cleanup patch groups - auto-expire any patch groups before we allocate a channel + mPatchGroupManager.cleanupPatchGroups(); + PatchGroupVoiceChannelGrantUpdate gvcgu = (PatchGroupVoiceChannelGrantUpdate)tsbk; channel = gvcgu.getChannel1(); @@ -2784,35 +2789,11 @@ public class P25DecoderState extends DecoderState break; case PATCH_GROUP_ADD: PatchGroupAdd pga = (PatchGroupAdd)tsbk; - - if(addPatchGroup(pga.getPatchGroupAddress(), pga.getPatchedTalkgroups())) - { - StringBuilder sb = new StringBuilder(); - - sb.append("PATCH GROUP:").append(pga.getPatchGroupAddress()); - sb.append(" ").append(pga.getPatchedTalkgroups()); - - broadcast(new P25CallEvent.Builder(CallEventType.PATCH_GROUP_ADD) - .aliasList(getAliasList()) - .details(sb.toString()) - .build()); - } + mPatchGroupManager.updatePatchGroup(pga.getPatchGroupAddress(), pga.getPatchedTalkgroups()); break; case PATCH_GROUP_DELETE: PatchGroupDelete pgd = (PatchGroupDelete)tsbk; - - if(removePatchGroup(pgd.getPatchGroupAddress(), pgd.getPatchedTalkgroups())) - { - StringBuilder sbpgd = new StringBuilder(); - - sbpgd.append("PATCH GROUP:").append(pgd.getPatchGroupAddress()); - sbpgd.append(" ").append(pgd.getPatchedTalkgroups()); - - broadcast(new P25CallEvent.Builder(CallEventType.PATCH_GROUP_DELETE) - .aliasList(getAliasList()) - .details(sbpgd.toString()) - .build()); - } + mPatchGroupManager.removePatchGroup(pgd.getPatchGroupAddress()); break; case CCH_PLANNED_SHUTDOWN: if(!mControlChannelShutdownLogged) @@ -2827,174 +2808,14 @@ public class P25DecoderState extends DecoderState } } - /** - * Creates a new patch group or updates an existing patch group with the patched talkgroup entries. - * - * @param patchGroupID for the patch group - * @param talkgroupsToAdd to include in the patch group - * @return true if the patch group or the patch group map was updated - */ - private boolean addPatchGroup(String patchGroupID, List talkgroupsToAdd) - { - boolean updated = false; - - if(patchGroupID != null && !talkgroupsToAdd.isEmpty()) - { - List patchedTalkgroups = mPatchGroupMap.get(patchGroupID); - - if(patchedTalkgroups == null) - { - patchedTalkgroups = new ArrayList<>(); - } - - for(String talkgroupToAdd : talkgroupsToAdd) - { - //Exclude the patch group ID if it is included in the patched talkgroup list so that we don't have - //an infinite loop situation - if(talkgroupToAdd.equals(patchGroupID)) - { - continue; - } - - if(!patchedTalkgroups.contains(talkgroupToAdd)) - { - patchedTalkgroups.add(talkgroupToAdd); - updated = true; - } - } - - if(updated) - { - mPatchGroupMap.put(patchGroupID, patchedTalkgroups); - updatePatchAlias(patchGroupID, patchedTalkgroups); - } - } - - return updated; - } - - /** - * Removes the patched talkgroups from the patch group and removes the patch group from the patch group map if - * there are no more patched talkgroups in the patch group. - * - * @param patchGroupID for the patch group - * @param talkgroupsToRemove to remove from the patch group - * @return true if the patch group or the patch group map was updated - */ - private boolean removePatchGroup(String patchGroupID, List talkgroupsToRemove) - { - boolean updated = false; - - if(patchGroupID != null && !talkgroupsToRemove.isEmpty()) - { - List patchedTalkgroups = mPatchGroupMap.get(patchGroupID); - - if(patchedTalkgroups != null) - { - for(String talkgroupToRemove : talkgroupsToRemove) - { - if(patchedTalkgroups.contains(talkgroupToRemove)) - { - patchedTalkgroups.remove(talkgroupToRemove); - updated = true; - } - } - - if(patchedTalkgroups.isEmpty()) - { - mPatchGroupMap.remove(patchGroupID); - removePatchGroupAlias(patchGroupID); - updated = true; - } - else - { - updatePatchAlias(patchGroupID, patchedTalkgroups); - } - } - } - - return updated; - } - - /** - * Adds/updates a patch group alias to the alias list containing aliases for each of the patched talkgroups. If the - * patch group alias already exists, any patched talkgroups will be removed from the existing patch group alias and - * replaced with the aliases corresponding to the patched talkgroup aliases. - * - * @param patchGroupID for the patch group - * @param patchedTalkgroups containing the talkgroup IDs for each of the patched talkgroups - */ - private void updatePatchAlias(String patchGroupID, List patchedTalkgroups) - { - PatchGroupAlias patchGroupAlias = null; - - //Check for an existing alias for the patch talkgroup - do not include wildcard aliases - Alias existingAlias = getAliasList().getTalkgroupAlias(patchGroupID, false); - - if(existingAlias instanceof PatchGroupAlias) - { - patchGroupAlias = (PatchGroupAlias)existingAlias; - } - else - { - patchGroupAlias = new PatchGroupAlias(); - - patchGroupAlias.addAliasID(new TalkgroupID(patchGroupID)); - - if(existingAlias != null) - { - getAliasList().removeAlias(existingAlias); - - patchGroupAlias.setPatchGroupAlias(existingAlias); - } - - getAliasList().addAlias(patchGroupAlias); - } - - if(patchGroupAlias != null) - { - patchGroupAlias.setPatchedTalkgroupIDs(patchedTalkgroups); - - patchGroupAlias.clearPatchedAliases(); - - for(String patchedTalkgroup : patchedTalkgroups) - { - Alias patchedAlias = getAliasList().getTalkgroupAlias(patchedTalkgroup); - - if(patchedAlias != null) - { - patchGroupAlias.addPatchedAlias(patchedAlias); - } - } - } - } - - /** - * Removes the temporary patch group alias from the alias list - */ - private void removePatchGroupAlias(String patchGroupID) - { - Alias alias = getAliasList().getTalkgroupAlias(patchGroupID); - - if(alias instanceof PatchGroupAlias) - { - PatchGroupAlias patchGroupAlias = (PatchGroupAlias)alias; - - getAliasList().removeAlias(patchGroupAlias); - - //Replace our temporary patch group alias with the original alias for the patch group ID - if(patchGroupAlias.hasPatchGroupAlias()) - { - getAliasList().addAlias(patchGroupAlias.getPatchGroupAlias()); - } - } - } - /** * Process a traffic channel allocation message */ private void processTSBKChannelGrant(TSBKMessage message) { + //Cleanup patch groups - auto-expire any patch groups before we allocate a channel + mPatchGroupManager.cleanupPatchGroups(); + String channel = null; String from = null; String to = null; diff --git a/src/module/decode/p25/PatchGroup.java b/src/module/decode/p25/PatchGroup.java new file mode 100644 index 00000000..525602ad --- /dev/null +++ b/src/module/decode/p25/PatchGroup.java @@ -0,0 +1,102 @@ +/******************************************************************************* + * sdrtrunk + * Copyright (C) 2014-2017 Dennis Sheirer + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see + * + ******************************************************************************/ +package module.decode.p25; + +import java.util.ArrayList; +import java.util.List; + +public class PatchGroup +{ + private long mExpireThreshold = 5000; //5 seconds + private long mLastUpdated; + private String mPatchGroupID; + private List mPatchedGroups = new ArrayList<>(); + + /** + * Constructs a patch group with the specified patch group ID + * @param patchGroupID that identifies this patch group + */ + public PatchGroup(String patchGroupID) + { + mPatchGroupID = patchGroupID; + mLastUpdated = System.currentTimeMillis(); + } + + /** + * Patch group ID for this patch group + */ + public String getPatchGroupID() + { + return mPatchGroupID; + } + + /** + * Adds the patched group to this patch group. Duplicate patched groups are ignored. + * + * Note: the patch group ID will not be added as a patched group, to avoid circular dependencies. + * + * @return true if the patched group was added to this patch group + */ + public boolean addPatchedGroup(String patchedGroup) + { + if(patchedGroup != null && !patchedGroup.equals(mPatchGroupID) && !mPatchedGroups.contains(patchedGroup)) + { + mPatchedGroups.add(patchedGroup); + return true; + } + + return false; + } + + /** + * Patched talkgroups that are part of this patch group. + */ + public List getPatchedGroups() + { + return mPatchedGroups; + } + + /** + * Updates the timestamp for this patch group. + */ + public void updateTimestamp() + { + mLastUpdated = System.currentTimeMillis(); + } + + /** + * Indicates if this patch group is expired, meaning that it hasn't been updated in the past (expire + * threshold) seconds. + */ + public boolean isExpired() + { + return System.currentTimeMillis() - mLastUpdated > mExpireThreshold; + } + + /** + * Sets the patch group expiration threshold (default = 5000 ms). A patch group is considered expired if + * no new updates have been received within the past (expire threshold) milliseconds + * + * @param expireThreshold in milliseconds + */ + public void setExpireThreshold(long expireThreshold) + { + mExpireThreshold = expireThreshold; + } +} diff --git a/src/module/decode/p25/PatchGroupManager.java b/src/module/decode/p25/PatchGroupManager.java new file mode 100644 index 00000000..79955038 --- /dev/null +++ b/src/module/decode/p25/PatchGroupManager.java @@ -0,0 +1,243 @@ +/******************************************************************************* + * sdrtrunk + * Copyright (C) 2014-2017 Dennis Sheirer + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see + * + ******************************************************************************/ +package module.decode.p25; + +import alias.Alias; +import alias.AliasList; +import alias.PatchGroupAlias; +import alias.id.talkgroup.TalkgroupID; +import module.decode.event.CallEvent; +import sample.Broadcaster; + +import java.util.ArrayList; +import java.util.Iterator; +import java.util.List; + +public class PatchGroupManager +{ + private AliasList mAliasList; + private List mPatchGroups = new ArrayList<>(); + private Broadcaster mCallEventBroadcaster; + + /** + * Manages active P25 patch groups and auto-expires patch groups if not updated every 5 seconds. + * @param aliasList to keep updated with patch group changes + * @param broadcaster for rebroadcasting patch group related call events + */ + public PatchGroupManager(AliasList aliasList, Broadcaster broadcaster) + { + mAliasList = aliasList; + mCallEventBroadcaster = broadcaster; + } + + /** + * Removes the patch group. + * + * @param patchGroupID of the patch group to remove + */ + public void removePatchGroup(String patchGroupID) + { + removePatchGroup(patchGroupID, false); + } + + /** + * Removes the patch group. + * + * @param patchGroupID of the patch group to remove + * @param expired true if the patch group is being removed because it expired, false if we receive a + * message commanding commanding a patch group delete. + */ + private void removePatchGroup(String patchGroupID, boolean expired) + { + PatchGroup patchGroup = getPatchGroup(patchGroupID); + + if(patchGroup != null) + { + //Cleanup the alias list + if(mAliasList != null) + { + Alias alias = mAliasList.getTalkgroupAlias(patchGroupID); + + if(alias instanceof PatchGroupAlias) + { + PatchGroupAlias patchGroupAlias = (PatchGroupAlias)alias; + + mAliasList.removeAlias(patchGroupAlias); + + //Replace our temporary patch group alias with the original alias for the patch group ID + if(patchGroupAlias.hasPatchGroupAlias()) + { + mAliasList.addAlias(patchGroupAlias.getPatchGroupAlias()); + } + } + } + + mPatchGroups.remove(patchGroup); + + if(mCallEventBroadcaster != null) + { + StringBuilder sb = new StringBuilder(); + + sb.append((expired ? "EXPIRED " : "DELETED ")); + sb.append("PATCH GROUP:").append(patchGroup.getPatchGroupID()); + sb.append(" ").append(patchGroup.getPatchedGroups()); + + + mCallEventBroadcaster.broadcast(new P25CallEvent.Builder(CallEvent.CallEventType.PATCH_GROUP_DELETE) + .to(patchGroup.getPatchGroupID()) + .details(sb.toString()) + .build()); + } + } + } + + /** + * Adds/updates a patch group alias to the alias list containing aliases for each of the patched talkgroups. If the + * patch group alias already exists, any patched talkgroups will be removed from the existing patch group alias and + * replaced with the aliases corresponding to the patched talkgroup aliases. + * + * @param patchGroupID for the patch group + * @param patchedTalkgroups containing the talkgroup IDs for each of the patched talkgroups + */ + public void updatePatchGroup(String patchGroupID, List patchedTalkgroups) + { + boolean updated = false; + + PatchGroup patchGroup = getPatchGroup(patchGroupID); + + if(patchGroup == null) + { + //Remove any expired patch groups before we add a new one + cleanupPatchGroups(); + patchGroup = new PatchGroup(patchGroupID); + mPatchGroups.add(patchGroup); + } + else + { + patchGroup.updateTimestamp(); + } + + for(String patchedGroup: patchedTalkgroups) + { + if(patchGroup.addPatchedGroup(patchedGroup)) + { + updated = true; + } + } + + //If we make any changes to the patch group, update the patch group alias + if(updated) + { + PatchGroupAlias patchGroupAlias = null; + + //Check for an existing alias for the patch talkgroup - do not include wildcard aliases + if(mAliasList != null) + { + Alias existingAlias = mAliasList.getTalkgroupAlias(patchGroupID, false); + + if(existingAlias instanceof PatchGroupAlias) + { + patchGroupAlias = (PatchGroupAlias)existingAlias; + } + else + { + patchGroupAlias = new PatchGroupAlias(); + + patchGroupAlias.addAliasID(new TalkgroupID(patchGroupID)); + + if(existingAlias != null) + { + mAliasList.removeAlias(existingAlias); + + patchGroupAlias.setPatchGroupAlias(existingAlias); + } + + mAliasList.addAlias(patchGroupAlias); + } + + if(patchGroupAlias != null) + { + patchGroupAlias.setPatchedTalkgroupIDs(patchGroup.getPatchedGroups()); + + patchGroupAlias.clearPatchedAliases(); + + for(String patchedTalkgroup : patchGroup.getPatchedGroups()) + { + Alias patchedAlias = mAliasList.getTalkgroupAlias(patchedTalkgroup); + + if(patchedAlias != null) + { + patchGroupAlias.addPatchedAlias(patchedAlias); + } + } + } + } + + if(mCallEventBroadcaster != null) + { + StringBuilder sb = new StringBuilder(); + + sb.append("PATCH GROUP:").append(patchGroup.getPatchGroupID()); + sb.append(" ").append(patchGroup.getPatchedGroups()); + + mCallEventBroadcaster.broadcast(new P25CallEvent.Builder(CallEvent.CallEventType.PATCH_GROUP_ADD) + .aliasList(mAliasList) + .details(sb.toString()) + .build()); + } + } + } + + /** + * Returns the patch group matching the patch group ID or null + */ + private PatchGroup getPatchGroup(String patchGroupID) + { + for(PatchGroup patchGroup: mPatchGroups) + { + if(patchGroup.getPatchGroupID().equals(patchGroupID)) + { + return patchGroup; + } + } + + return null; + } + + /** + * Checks the last updated timestamp on all patch groups and removes any patch groups that have not been + * updated in the past 5 seconds + */ + public void cleanupPatchGroups() + { + Iterator it = mPatchGroups.iterator(); + + while(it.hasNext()) + { + PatchGroup patchGroup = it.next(); + + if(patchGroup.isExpired()) + { + it.remove(); + + removePatchGroup(patchGroup.getPatchGroupID(), true); + } + } + } +} diff --git a/src/sample/OverflowableTransferQueue.java b/src/sample/OverflowableTransferQueue.java index d70308ca..f9c9c190 100644 --- a/src/sample/OverflowableTransferQueue.java +++ b/src/sample/OverflowableTransferQueue.java @@ -20,6 +20,7 @@ package sample; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import sample.real.IOverflowListener; import source.tuner.TunerChannelSource; import java.util.Collection; @@ -32,7 +33,7 @@ public class OverflowableTransferQueue private final static Logger mLog = LoggerFactory.getLogger(OverflowableTransferQueue.class); public enum State {NORMAL, OVERFLOW}; - private Listener mStateListener; + private IOverflowListener mOverflowListener; private LinkedTransferQueue mQueue = new LinkedTransferQueue(); private AtomicInteger mCounter = new AtomicInteger(); @@ -82,11 +83,6 @@ public class OverflowableTransferQueue int size = mCounter.addAndGet(-drainCount); - if(mOverflow.get()) - { - mLog.debug("Overflow - current size:" + size + " threshold:" + mResetThreshold); - } - if(mOverflow.get() && size <= mResetThreshold) { setOverflow(false); @@ -96,32 +92,24 @@ public class OverflowableTransferQueue } /** - * Sets a listener to receive state change events + * Sets a listener to receive overflow state change events */ - public void setStateListener(Listener listener) + public void setOverflowListener(IOverflowListener listener) { - mStateListener = listener; + mOverflowListener = listener; } /** - * Current state of the queue, normal or overflow - */ - public State getState() - { - return mOverflow.get() ? State.OVERFLOW : State.NORMAL; - } - - /** - * Toggles the overflow state + * Toggles the overflow state and broadcast state change to listener */ private void setOverflow(boolean overflow) { if(mOverflow.compareAndSet(!overflow, overflow)) { - if(mStateListener != null) + if(mOverflowListener != null) { - mStateListener.receive(overflow ? State.OVERFLOW : State.NORMAL); + mOverflowListener.sourceOverflow(overflow); } } } @@ -136,6 +124,10 @@ public class OverflowableTransferQueue mQueue.clear(); mCounter.set(0); mOverflow.set(false); + if(mOverflowListener != null) + { + mOverflowListener.sourceOverflow(false); + } } } } diff --git a/src/sample/real/IOverflowListener.java b/src/sample/real/IOverflowListener.java new file mode 100644 index 00000000..61202ea3 --- /dev/null +++ b/src/sample/real/IOverflowListener.java @@ -0,0 +1,28 @@ +/******************************************************************************* + * sdrtrunk + * Copyright (C) 2014-2017 Dennis Sheirer + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see + * + ******************************************************************************/ +package sample.real; + +public interface IOverflowListener +{ + /** + * Indicates that the source is in overflow state (true) or out of overflow state (false) + * @param overflow true to indicate an overflow state + */ + void sourceOverflow(boolean overflow); +} diff --git a/src/source/Source.java b/src/source/Source.java index 217def00..2de310b8 100644 --- a/src/source/Source.java +++ b/src/source/Source.java @@ -19,6 +19,7 @@ package source; import module.Module; import sample.SampleType; +import sample.real.IOverflowListener; /** * Abstract class to define the minimum functionality of a sample data provider. @@ -26,6 +27,7 @@ import sample.SampleType; public abstract class Source extends Module { protected SampleType mSampleType; + protected IOverflowListener mOverflowListener; public Source( SampleType sampleType ) { @@ -47,4 +49,25 @@ public abstract class Source extends Module { mSampleType = sampleType; } + + /** + * Registers the listener to receive overflow state changes. Use null argument to clear the listener + */ + public void setOverflowListener(IOverflowListener listener) + { + mOverflowListener = listener; + } + + /** + * Broadcasts an overflow state + * + * @param overflow true if overlow, false if normal + */ + protected void broadcastOverflowState(boolean overflow) + { + if(mOverflowListener != null) + { + mOverflowListener.sourceOverflow(overflow); + } + } } diff --git a/src/source/tuner/TunerChannelSource.java b/src/source/tuner/TunerChannelSource.java index 369b6079..13be8ba9 100644 --- a/src/source/tuner/TunerChannelSource.java +++ b/src/source/tuner/TunerChannelSource.java @@ -29,6 +29,7 @@ import sample.Listener; import sample.OverflowableTransferQueue; import sample.complex.Complex; import sample.complex.ComplexBuffer; +import sample.real.IOverflowListener; import source.ComplexSource; import source.SourceException; import source.tuner.frequency.FrequencyChangeEvent; @@ -101,21 +102,6 @@ public class TunerChannelSource extends ComplexSource implements IFrequencyChang mTunerFrequency = mTuner.getTunerController().getFrequency(); mBuffer = new OverflowableTransferQueue<>(BUFFER_MAX_CAPACITY, BUFFER_OVERFLOW_RESET_THRESHOLD); - mBuffer.setStateListener(new Listener() - { - @Override - public void receive(OverflowableTransferQueue.State state) - { - if(state == OverflowableTransferQueue.State.NORMAL) - { - mLog.debug("Channel [" + mTunerChannel.getFrequency() + "] buffer overflow - temporary pause until processing catches up"); - } - else - { - mLog.debug("Channel [" + mTunerChannel.getFrequency() + "] buffer overflow cleared - resuming normal processing"); - } - } - }); /* Setup the frequency translator to the current source frequency */ long frequencyOffset = mTunerFrequency - mTunerChannel.getFrequency(); @@ -126,6 +112,14 @@ public class TunerChannelSource extends ComplexSource implements IFrequencyChang frequencyChanged(new FrequencyChangeEvent(Event.NOTIFICATION_SAMPLE_RATE_CHANGE, mTuner.getTunerController().getSampleRate())); } + /** + * Overrides the default source overflow listener management to delegate responsibility to the overflow buffer + */ + @Override + public void setOverflowListener(IOverflowListener listener) + { + mBuffer.setOverflowListener(listener); + } public void start(ScheduledExecutorService executor) { @@ -175,7 +169,7 @@ public class TunerChannelSource extends ComplexSource implements IFrequencyChang } else { - mLog.warn("Attempt to stop() and already stopped tuner channel source was ignored"); + mLog.warn("Attempt to stop() an already stopped tuner channel source was ignored"); } } @@ -316,6 +310,11 @@ public class TunerChannelSource extends ComplexSource implements IFrequencyChang public void shutdown() { + if(getTunerChannel().getFrequency() == 460500000) + { + mLog.debug("************** Shutdown has been invoked on the P25 primary control channel: 460500000"); + } + mProcessing = false; } @@ -373,12 +372,18 @@ public class TunerChannelSource extends ComplexSource implements IFrequencyChang } catch(Exception e) { + mLog.error("Error encountered during decimation process", e); + /* Only log the stack trace if we're still processing */ if(mProcessing) { mLog.error("Error encountered during decimation process", e); } } + catch(Throwable throwable) + { + mLog.error("Bad Error", throwable); + } /* Check to see if we've been shutdown */ if(!mProcessing) diff --git a/src/source/tuner/usb/USBTransferProcessor.java b/src/source/tuner/usb/USBTransferProcessor.java index 5039a0d4..aeb56bb9 100644 --- a/src/source/tuner/usb/USBTransferProcessor.java +++ b/src/source/tuner/usb/USBTransferProcessor.java @@ -30,6 +30,7 @@ import sample.Listener; import sample.OverflowableTransferQueue; import sample.adapter.ISampleAdapter; import sample.complex.ComplexBuffer; +import sample.real.IOverflowListener; import source.tuner.TunerManager; import util.ThreadPool; @@ -101,12 +102,12 @@ public class USBTransferProcessor implements TransferCallback mBufferSize = bufferSize; mFilledBuffers = new OverflowableTransferQueue<>(FILLED_BUFFER_MAX_CAPACITY, FILLED_BUFFER_OVERFLOW_RESET_THRESHOLD); - mFilledBuffers.setStateListener(new Listener() + mFilledBuffers.setOverflowListener(new IOverflowListener() { @Override - public void receive(OverflowableTransferQueue.State state) + public void sourceOverflow(boolean overflow) { - if(state == OverflowableTransferQueue.State.NORMAL) + if(overflow) { mLog.debug(mDeviceName + " - buffer overflow - temporary pause until processing catches up"); }