From f0dc80437ca828983eab2a5a9d193608bbcefcb9 Mon Sep 17 00:00:00 2001 From: Denny Sheirer Date: Sat, 11 Mar 2023 16:04:55 -0500 Subject: [PATCH] #1487 Updates hot code classes to remove usage of reentrant locks for thread synchronization. (#1488) Co-authored-by: Dennis Sheirer --- .../channelizer/PolyphaseChannelManager.java | 33 +-- .../channelizer/PolyphaseChannelSource.java | 220 +++++++++++------- .../github/dsheirer/sample/Broadcaster.java | 60 +---- .../channel/StreamProcessorWithHeartbeat.java | 30 +-- 4 files changed, 164 insertions(+), 179 deletions(-) diff --git a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/PolyphaseChannelManager.java b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/PolyphaseChannelManager.java index e8a8c1d0..83a72ea2 100644 --- a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/PolyphaseChannelManager.java +++ b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/PolyphaseChannelManager.java @@ -44,7 +44,6 @@ import java.util.Map; import java.util.SortedSet; import java.util.TreeSet; import java.util.concurrent.CopyOnWriteArrayList; -import java.util.concurrent.locks.ReentrantLock; import org.apache.commons.math3.util.FastMath; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -481,7 +480,6 @@ public class PolyphaseChannelManager implements ISourceEventProcessor public class NativeBufferReceiver implements Listener { private boolean mOutputProcessorUpdateRequired = false; - private ReentrantLock mUpdateFlagLock = new ReentrantLock(); /** * Processes tuner center frequency change source events to flag when output processors need updating. @@ -493,18 +491,9 @@ public class PolyphaseChannelManager implements ISourceEventProcessor if(mChannelCalculator.getCenterFrequency() != frequency) { - mUpdateFlagLock.lock(); - - try - { - //Update the channel calculator frequency so that it's ready when the output processor update occurs - mChannelCalculator.setCenterFrequency(frequency); - mOutputProcessorUpdateRequired = true; - } - finally - { - mUpdateFlagLock.unlock(); - } + //Update the channel calculator frequency so that it's ready when the output processor update occurs + mChannelCalculator.setCenterFrequency(frequency); + mOutputProcessorUpdateRequired = true; } } @@ -516,19 +505,17 @@ public class PolyphaseChannelManager implements ISourceEventProcessor @Override public void receive(INativeBuffer nativeBuffer) { - mUpdateFlagLock.lock(); - - try + if(mOutputProcessorUpdateRequired) { - if(mOutputProcessorUpdateRequired) + try { updateOutputProcessors(); - mOutputProcessorUpdateRequired = false; } - } - finally - { - mUpdateFlagLock.unlock(); + catch(Exception e) + { + mLog.error("Error updating polyphase channel output processors"); + } + mOutputProcessorUpdateRequired = false; } if(mPolyphaseChannelizer != null) diff --git a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/PolyphaseChannelSource.java b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/PolyphaseChannelSource.java index f6796d5e..3ba96a25 100644 --- a/src/main/java/io/github/dsheirer/dsp/filter/channelizer/PolyphaseChannelSource.java +++ b/src/main/java/io/github/dsheirer/dsp/filter/channelizer/PolyphaseChannelSource.java @@ -28,12 +28,10 @@ import io.github.dsheirer.source.SourceEvent; import io.github.dsheirer.source.tuner.channel.StreamProcessorWithHeartbeat; import io.github.dsheirer.source.tuner.channel.TunerChannel; import io.github.dsheirer.source.tuner.channel.TunerChannelSource; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - import java.util.ArrayList; import java.util.List; -import java.util.concurrent.locks.ReentrantLock; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; /** * Polyphase Channelizer's Tuner Channel Source implementation. Wraps a ChannelOutputProcessor instance and @@ -47,12 +45,10 @@ public class PolyphaseChannelSource extends TunerChannelSource implements Listen private StreamProcessorWithHeartbeat mStreamHeartbeatProcessor; private double mChannelSampleRate; private long mIndexCenterFrequency; - private ReentrantLock mOutputProcessorLock = new ReentrantLock(); - private List mOutputProcessorIndexes = new ArrayList<>(); private double mTunerSampleRate; private double mTunerCenterFrequency; - + private PendingOutputProcessorUpdate mPendingOutputProcessorUpdate; /** * Constructs an instance @@ -69,7 +65,7 @@ public class PolyphaseChannelSource extends TunerChannelSource implements Listen super(producerSourceEventListener, tunerChannel); mChannelSampleRate = channelCalculator.getChannelSampleRate(); mStreamHeartbeatProcessor = new StreamProcessorWithHeartbeat<>(getHeartbeatManager(), HEARTBEAT_INTERVAL_MS); - updateOutputProcessor(channelCalculator, filterManager); + doUpdateOutputProcessor(channelCalculator, filterManager); } /** @@ -150,102 +146,108 @@ public class PolyphaseChannelSource extends TunerChannelSource implements Listen } /** - * Updates the output processor whenever the source tuner's center frequency changes. + * Queues a request to update the output processor whenever the source tuner's center frequency changes. * @param channelCalculator providing access to updated tuner center frequency, sample rate, etc. * @param filterManager for designing and caching synthesis filters */ - public void updateOutputProcessor(ChannelCalculator channelCalculator, SynthesisFilterManager filterManager) throws IllegalArgumentException + public void updateOutputProcessor(ChannelCalculator channelCalculator, SynthesisFilterManager filterManager) + throws IllegalArgumentException + { + mPendingOutputProcessorUpdate = new PendingOutputProcessorUpdate(channelCalculator, filterManager); + } + + /** + * Performs the output processor update. This method must be invoked on the channel results calling thread to + * ensure we don't have thread contention on the output processor. + * @param channelCalculator for updates + * @param filterManager for updates + * @throws IllegalArgumentException if there's an issue. + */ + public void doUpdateOutputProcessor(ChannelCalculator channelCalculator, SynthesisFilterManager filterManager) + throws IllegalArgumentException { String errorMessage = null; - mOutputProcessorLock.lock(); + //If a change in sample rate or center frequency makes this channel no longer viable, then the channel + //calculator will throw an IllegalArgException ... handled below + List indexes = channelCalculator.getChannelIndexes(getTunerChannel()); - try + mOutputProcessorIndexes.clear(); + mOutputProcessorIndexes.addAll(indexes); + mTunerCenterFrequency = channelCalculator.getCenterFrequency(); + mTunerSampleRate = channelCalculator.getSampleRate(); + + //The provided channels are necessarily aligned to the center frequency that this source is providing and an + //oscillator will mix the provided channels to bring the desired center frequency to baseband. + setFrequency(channelCalculator.getCenterFrequencyForIndexes(indexes)); + + if(mPolyphaseChannelOutputProcessor != null && mPolyphaseChannelOutputProcessor.getInputChannelCount() == indexes.size()) { - //If a change in sample rate or center frequency makes this channel no longer viable, then the channel - //calculator will throw an IllegalArgException ... handled below - List indexes = channelCalculator.getChannelIndexes(getTunerChannel()); + mPolyphaseChannelOutputProcessor.setPolyphaseChannelIndices(indexes); - mOutputProcessorIndexes.clear(); - mOutputProcessorIndexes.addAll(indexes); - mTunerCenterFrequency = channelCalculator.getCenterFrequency(); - mTunerSampleRate = channelCalculator.getSampleRate(); - - //The provided channels are necessarily aligned to the center frequency that this source is providing and an - //oscillator will mix the provided channels to bring the desired center frequency to baseband. - setFrequency(channelCalculator.getCenterFrequencyForIndexes(indexes)); - - if(mPolyphaseChannelOutputProcessor != null && mPolyphaseChannelOutputProcessor.getInputChannelCount() == indexes.size()) + if(indexes.size() > 1) { - mPolyphaseChannelOutputProcessor.setPolyphaseChannelIndices(indexes); - - if(indexes.size() > 1) + try { + float[] filter = filterManager.getFilter(channelCalculator.getChannelSampleRate(), + channelCalculator.getChannelBandwidth(), indexes.size()); + mPolyphaseChannelOutputProcessor.setSynthesisFilter(filter); + } + catch(FilterDesignException fde) + { + mLog.error("Error creating an updated synthesis filter for the channel output processor"); + errorMessage ="Cannot update output processor - unable to design synthesis filter for [" + + indexes.size() + "] channel indices - channel sample rate [" + channelCalculator.getChannelSampleRate() + + "] channel bandwidth [" + channelCalculator.getChannelBandwidth() + "]"; + } + } + + mPolyphaseChannelOutputProcessor.setFrequencyOffset(getFrequencyOffset()); + } + else //Create a new output processor. + { + if(mPolyphaseChannelOutputProcessor != null) + { + mPolyphaseChannelOutputProcessor.setListener(null); + mPolyphaseChannelOutputProcessor.stop(); + } + + mPolyphaseChannelOutputProcessor = null; + + switch(indexes.size()) + { + case 1: + mPolyphaseChannelOutputProcessor = new OneChannelOutputProcessor(channelCalculator.getChannelSampleRate(), + indexes, channelCalculator.getChannelCount()); + mPolyphaseChannelOutputProcessor.setListener(this); + mPolyphaseChannelOutputProcessor.setFrequencyOffset(getFrequencyOffset()); + mPolyphaseChannelOutputProcessor.start(); + break; + case 2: try { float[] filter = filterManager.getFilter(channelCalculator.getChannelSampleRate(), - channelCalculator.getChannelBandwidth(), indexes.size()); - mPolyphaseChannelOutputProcessor.setSynthesisFilter(filter); - } - catch(FilterDesignException fde) - { - mLog.error("Error creating an updated synthesis filter for the channel output processor"); - errorMessage ="Cannot update output processor - unable to design synthesis filter for [" + - indexes.size() + "] channel indices"; - } - } - - mPolyphaseChannelOutputProcessor.setFrequencyOffset(getFrequencyOffset()); - } - else //Create a new output processor. - { - if(mPolyphaseChannelOutputProcessor != null) - { - mPolyphaseChannelOutputProcessor.setListener(null); - mPolyphaseChannelOutputProcessor.stop(); - } - - mPolyphaseChannelOutputProcessor = null; - - switch(indexes.size()) - { - case 1: - mPolyphaseChannelOutputProcessor = new OneChannelOutputProcessor(channelCalculator.getChannelSampleRate(), - indexes, channelCalculator.getChannelCount()); + channelCalculator.getChannelBandwidth(), 2); + mPolyphaseChannelOutputProcessor = new TwoChannelOutputProcessor(channelCalculator.getChannelSampleRate(), + indexes, filter, channelCalculator.getChannelCount()); mPolyphaseChannelOutputProcessor.setListener(this); mPolyphaseChannelOutputProcessor.setFrequencyOffset(getFrequencyOffset()); mPolyphaseChannelOutputProcessor.start(); - break; - case 2: - try - { - float[] filter = filterManager.getFilter(channelCalculator.getChannelSampleRate(), - channelCalculator.getChannelBandwidth(), 2); - mPolyphaseChannelOutputProcessor = new TwoChannelOutputProcessor(channelCalculator.getChannelSampleRate(), - indexes, filter, channelCalculator.getChannelCount()); - mPolyphaseChannelOutputProcessor.setListener(this); - mPolyphaseChannelOutputProcessor.setFrequencyOffset(getFrequencyOffset()); - mPolyphaseChannelOutputProcessor.start(); - } - catch(FilterDesignException fde) - { - errorMessage = "Cannot create new output processor - unable to design synthesis filter for [" + - indexes.size() + "] channel indices"; - mLog.error("Error creating a synthesis filter for a new channel output processor"); - } - break; - default: - mLog.error("Request to create an output processor for unexpected channel index size:" + indexes.size()); - mLog.info(channelCalculator.toString()); - errorMessage = "Unable to create new channel output processor - unexpected channel index size: " + - indexes.size(); - } + } + catch(FilterDesignException fde) + { + errorMessage = "Cannot create new output processor - unable to design synthesis filter for [" + + indexes.size() + "] channel indices"; + mLog.error("Error creating a synthesis filter for a new channel output processor"); + } + break; + default: + mLog.error("Request to create an output processor for unexpected channel index size:" + indexes.size()); + mLog.info(channelCalculator.toString()); + errorMessage = "Unable to create new channel output processor - unexpected channel index size: " + + indexes.size(); } } - finally - { - mOutputProcessorLock.unlock(); - } //Unlikely, but if we had an error designing a synthesis filter, throw an exception if(errorMessage != null) @@ -264,7 +266,13 @@ public class PolyphaseChannelSource extends TunerChannelSource implements Listen */ public void receiveChannelResults(List channelResultsList, long currentSamplesTimestamp) { - mOutputProcessorLock.lock(); + if(mPendingOutputProcessorUpdate != null) + { + ChannelCalculator channelCalculator = mPendingOutputProcessorUpdate.getChannelCalculator(); + SynthesisFilterManager filterManager = mPendingOutputProcessorUpdate.getSynthesisFilterManager(); + mPendingOutputProcessorUpdate = null; + doUpdateOutputProcessor(channelCalculator, filterManager); + } try { @@ -273,9 +281,9 @@ public class PolyphaseChannelSource extends TunerChannelSource implements Listen mPolyphaseChannelOutputProcessor.receiveChannelResults(channelResultsList, currentSamplesTimestamp); } } - finally + catch(Exception e) { - mOutputProcessorLock.unlock(); + mLog.error("Error processing channel results", e); } } @@ -344,4 +352,40 @@ public class PolyphaseChannelSource extends TunerChannelSource implements Listen return "POLYPHASE [" + mPolyphaseChannelOutputProcessor.getInputChannelCount() + "] " + getTunerChannel().getFrequency(); } + + /** + * Support classes needed to update the output processor. + */ + public class PendingOutputProcessorUpdate + { + private final ChannelCalculator mChannelCalculator; + private final SynthesisFilterManager mSynthesisFilterManager; + + /** + * Constructs an instance + * @param channelCalculator for support + * @param synthesisFilterManager for support + */ + public PendingOutputProcessorUpdate(ChannelCalculator channelCalculator, SynthesisFilterManager synthesisFilterManager) + { + mChannelCalculator = channelCalculator; + mSynthesisFilterManager = synthesisFilterManager; + } + + /** + * Channel calculator + */ + public ChannelCalculator getChannelCalculator() + { + return mChannelCalculator; + } + + /** + * Channel calculator + */ + public SynthesisFilterManager getSynthesisFilterManager() + { + return mSynthesisFilterManager; + } + } } diff --git a/src/main/java/io/github/dsheirer/sample/Broadcaster.java b/src/main/java/io/github/dsheirer/sample/Broadcaster.java index 703e9cf3..3ba9e0f8 100644 --- a/src/main/java/io/github/dsheirer/sample/Broadcaster.java +++ b/src/main/java/io/github/dsheirer/sample/Broadcaster.java @@ -1,6 +1,6 @@ /* * ***************************************************************************** - * Copyright (C) 2014-2022 Dennis Sheirer + * Copyright (C) 2014-2023 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 @@ -18,13 +18,11 @@ */ package io.github.dsheirer.sample; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.util.ArrayList; import java.util.Collections; import java.util.List; -import java.util.concurrent.locks.ReentrantLock; +import java.util.concurrent.CopyOnWriteArrayList; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; /** * Broadcasts an item to multiple listeners @@ -32,8 +30,7 @@ import java.util.concurrent.locks.ReentrantLock; public class Broadcaster implements Listener { private final static Logger mLog = LoggerFactory.getLogger(Broadcaster.class); - private ReentrantLock mLock = new ReentrantLock(); - private List> mListeners = new ArrayList<>(); + private List> mListeners = new CopyOnWriteArrayList<>(); public Broadcaster() { @@ -90,25 +87,9 @@ public class Broadcaster implements Listener */ public void addListener(Listener listener) { - if(listener != null) + if(listener != null && !mListeners.contains(listener)) { - mLock.lock(); - - try - { - if(!mListeners.contains(listener)) - { - mListeners.add(listener); - } - } - catch(Exception e) - { - mLog.error("Unexpected error while adding listener", e); - } - finally - { - mLock.unlock(); - } + mListeners.add(listener); } } @@ -119,20 +100,7 @@ public class Broadcaster implements Listener { if(listener != null) { - mLock.lock(); - - try - { - mListeners.remove(listener); - } - catch(Exception e) - { - mLog.error("Unexpected error while removing listener", e); - } - finally - { - mLock.unlock(); - } + mListeners.remove(listener); } } @@ -149,24 +117,16 @@ public class Broadcaster implements Listener */ public void broadcast(T t) { - mLock.lock(); - try { - List> listeners = new ArrayList<>(mListeners); - - for(Listener listener: listeners) + for(Listener listener: mListeners) { listener.receive(t); } } catch(Exception e) { - mLog.error("Unexpected error while broadcasting to listeners", e); - } - finally - { - mLock.unlock(); + mLog.error("Error while broadcasting [" + t.getClass() + "] to listeners"); } } } diff --git a/src/main/java/io/github/dsheirer/source/tuner/channel/StreamProcessorWithHeartbeat.java b/src/main/java/io/github/dsheirer/source/tuner/channel/StreamProcessorWithHeartbeat.java index 262402bb..2f15836f 100644 --- a/src/main/java/io/github/dsheirer/source/tuner/channel/StreamProcessorWithHeartbeat.java +++ b/src/main/java/io/github/dsheirer/source/tuner/channel/StreamProcessorWithHeartbeat.java @@ -1,6 +1,6 @@ /* * ***************************************************************************** - * Copyright (C) 2014-2022 Dennis Sheirer + * Copyright (C) 2014-2023 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 @@ -22,10 +22,10 @@ package io.github.dsheirer.source.tuner.channel; import io.github.dsheirer.sample.Listener; import io.github.dsheirer.source.heartbeat.HeartbeatManager; import io.github.dsheirer.util.ThreadPool; - import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; -import java.util.concurrent.locks.ReentrantLock; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; /** * Thread-safe implementation that coordinates the flow of data/buffer stream processing with a parallel, periodic @@ -42,7 +42,7 @@ import java.util.concurrent.locks.ReentrantLock; */ public class StreamProcessorWithHeartbeat implements Listener { - private ReentrantLock mLock = new ReentrantLock(); + private static final Logger mLog = LoggerFactory.getLogger(StreamProcessorWithHeartbeat.class); private HeartbeatManager mHeartbeatManager; private long mHeartbeatInterval; private Listener mListener; @@ -88,25 +88,19 @@ public class StreamProcessorWithHeartbeat implements Listener */ private void fireOptionalHeartbeat() { - //Acquire lock only if it's not already locked, otherwise ignore heartbeat for this timer interval. - if(mLock.tryLock()) + try { - try - { - mHeartbeatManager.broadcast(); - } - finally - { - mLock.unlock(); - } + mHeartbeatManager.broadcast(); + } + catch(Exception e) + { + mLog.error("Error firing optional heartbeat", e); } } @Override public void receive(T data) { - mLock.lock(); - try { mHeartbeatManager.broadcast(); @@ -118,9 +112,9 @@ public class StreamProcessorWithHeartbeat implements Listener listener.receive(data); } } - finally + catch(Exception e) { - mLock.unlock(); + mLog.error("Error distributing data to listeners following heartbeat", e); } }