#1487 Updates hot code classes to remove usage of reentrant locks for thread synchronization. (#1488)

Co-authored-by: Dennis Sheirer <dsheirer@github.com>

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<INativeBuffer>
     {
         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();
+                //Update the channel calculator frequency so that it's ready when the output processor update occurs
-
+                mChannelCalculator.setCenterFrequency(frequency);
-                try
+                mOutputProcessorUpdateRequired = true;
-                {
-                    //Update the channel calculator frequency so that it's ready when the output processor update occurs
-                    mChannelCalculator.setCenterFrequency(frequency);
-                    mOutputProcessorUpdateRequired = true;
-                }
-                finally
-                {
-                    mUpdateFlagLock.unlock();
-                }
             }
         }
 
@@ -516,19 +505,17 @@ public class PolyphaseChannelManager implements ISourceEventProcessor
         @Override
         public void receive(INativeBuffer nativeBuffer)
         {
-            mUpdateFlagLock.lock();
+            if(mOutputProcessorUpdateRequired)
-
-            try
             {
-                if(mOutputProcessorUpdateRequired)
+                try
                 {
                     updateOutputProcessors();
-                    mOutputProcessorUpdateRequired = false;
                 }
-            }
+                catch(Exception e)
-            finally
+                {
-            {
+                    mLog.error("Error updating polyphase channel output processors");
-                mUpdateFlagLock.unlock();
+                }
+                mOutputProcessorUpdateRequired = false;
             }
 
             if(mPolyphaseChannelizer != null)

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<ComplexSamples> mStreamHeartbeatProcessor;
     private double mChannelSampleRate;
     private long mIndexCenterFrequency;
-    private ReentrantLock mOutputProcessorLock = new ReentrantLock();
-
     private List<Integer> 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<Integer> 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
+            mPolyphaseChannelOutputProcessor.setPolyphaseChannelIndices(indexes);
-            //calculator will throw an IllegalArgException ... handled below
-            List<Integer> indexes = channelCalculator.getChannelIndexes(getTunerChannel());
 
-            mOutputProcessorIndexes.clear();
+            if(indexes.size() > 1)
-            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())
             {
-                mPolyphaseChannelOutputProcessor.setPolyphaseChannelIndices(indexes);
+                try
-
-                if(indexes.size() > 1)
                 {
+                    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());
+                                channelCalculator.getChannelBandwidth(), 2);
-                        mPolyphaseChannelOutputProcessor.setSynthesisFilter(filter);
+                        mPolyphaseChannelOutputProcessor = new TwoChannelOutputProcessor(channelCalculator.getChannelSampleRate(),
-                    }
+                                indexes, filter, channelCalculator.getChannelCount());
-                    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());
                         mPolyphaseChannelOutputProcessor.setListener(this);
                         mPolyphaseChannelOutputProcessor.setFrequencyOffset(getFrequencyOffset());
                         mPolyphaseChannelOutputProcessor.start();
-                        break;
+                    }
-                    case 2:
+                    catch(FilterDesignException fde)
-                        try
+                    {
-                        {
+                        errorMessage = "Cannot create new output processor - unable to design synthesis filter for [" +
-                            float[] filter = filterManager.getFilter(channelCalculator.getChannelSampleRate(),
+                                indexes.size() + "] channel indices";
-                                    channelCalculator.getChannelBandwidth(), 2);
+                        mLog.error("Error creating a synthesis filter for a new channel output processor");
-                            mPolyphaseChannelOutputProcessor = new TwoChannelOutputProcessor(channelCalculator.getChannelSampleRate(),
+                    }
-                                    indexes, filter, channelCalculator.getChannelCount());
+                    break;
-                            mPolyphaseChannelOutputProcessor.setListener(this);
+                default:
-                            mPolyphaseChannelOutputProcessor.setFrequencyOffset(getFrequencyOffset());
+                    mLog.error("Request to create an output processor for unexpected channel index size:" + indexes.size());
-                            mPolyphaseChannelOutputProcessor.start();
+                    mLog.info(channelCalculator.toString());
-                        }
+                    errorMessage = "Unable to create new channel output processor - unexpected channel index size: " +
-                        catch(FilterDesignException fde)
+                            indexes.size();
-                        {
-                            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<float[]> 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;
+        }
+    }
 }

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<T> implements Listener<T>
 {
     private final static Logger mLog = LoggerFactory.getLogger(Broadcaster.class);
-    private ReentrantLock mLock = new ReentrantLock();
+    private List<Listener<T>> mListeners = new CopyOnWriteArrayList<>();
-    private List<Listener<T>> mListeners = new ArrayList<>();
 
     public Broadcaster()
     {
@@ -90,25 +87,9 @@ public class Broadcaster<T> implements Listener<T>
      */
     public void addListener(Listener<T> listener)
     {
-        if(listener != null)
+        if(listener != null && !mListeners.contains(listener))
         {
-            mLock.lock();
+            mListeners.add(listener);
-
-            try
-            {
-                if(!mListeners.contains(listener))
-                {
-                    mListeners.add(listener);
-                }
-            }
-            catch(Exception e)
-            {
-                mLog.error("Unexpected error while adding listener", e);
-            }
-            finally
-            {
-                mLock.unlock();
-            }
         }
     }
 
@@ -119,20 +100,7 @@ public class Broadcaster<T> implements Listener<T>
     {
         if(listener != null)
         {
-            mLock.lock();
+            mListeners.remove(listener);
-
-            try
-            {
-                mListeners.remove(listener);
-            }
-            catch(Exception e)
-            {
-                mLog.error("Unexpected error while removing listener", e);
-            }
-            finally
-            {
-                mLock.unlock();
-            }
         }
     }
 
@@ -149,24 +117,16 @@ public class Broadcaster<T> implements Listener<T>
      */
     public void broadcast(T t)
     {
-        mLock.lock();
-
         try
         {
-            List<Listener<T>> listeners = new ArrayList<>(mListeners);
+            for(Listener<T> listener: mListeners)
-
-            for(Listener listener: listeners)
             {
                 listener.receive(t);
             }
         }
         catch(Exception e)
         {
-            mLog.error("Unexpected error while broadcasting to listeners", e);
+            mLog.error("Error while broadcasting [" + t.getClass() + "] to listeners");
-        }
-        finally
-        {
-            mLock.unlock();
         }
     }
 }

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<T> implements Listener<T>
 {
-    private ReentrantLock mLock = new ReentrantLock();
+    private static final Logger mLog = LoggerFactory.getLogger(StreamProcessorWithHeartbeat.class);
     private HeartbeatManager mHeartbeatManager;
     private long mHeartbeatInterval;
     private Listener<T> mListener;
@@ -88,25 +88,19 @@ public class StreamProcessorWithHeartbeat<T> implements Listener<T>
      */
     private void fireOptionalHeartbeat()
     {
-        //Acquire lock only if it's not already locked, otherwise ignore heartbeat for this timer interval.
+        try
-        if(mLock.tryLock())
         {
-            try
+            mHeartbeatManager.broadcast();
-            {
+        }
-                mHeartbeatManager.broadcast();
+        catch(Exception e)
-            }
+        {
-            finally
+            mLog.error("Error firing optional heartbeat", e);
-            {
-                mLock.unlock();
-            }
         }
     }
 
     @Override
     public void receive(T data)
     {
-        mLock.lock();
-
         try
         {
             mHeartbeatManager.broadcast();
@@ -118,9 +112,9 @@ public class StreamProcessorWithHeartbeat<T> implements Listener<T>
                 listener.receive(data);
             }
         }
-        finally
+        catch(Exception e)
         {
-            mLock.unlock();
+            mLog.error("Error distributing data to listeners following heartbeat", e);
         }
     }