From e385d2fdec72503fc086d8994eb335c55d710d09 Mon Sep 17 00:00:00 2001 From: Denny Sheirer Date: Sun, 1 Nov 2026 06:04:57 -0500 Subject: [PATCH] #2438 resolves broadcaster error that happens during tuner source swapout (#2550) Co-authored-by: denny --- .../channelizer/PolyphaseChannelSource.java | 21 ++-- .../dsheirer/sample/SerializedListener.java | 71 +++++++++++ .../MultiFrequencyTunerChannelSource.java | 8 +- .../PolyphaseChannelSourceTest.java | 97 +++++++++++++++ .../sample/SerializedListenerTest.java | 117 ++++++++++++++++++ 5 files changed, 300 insertions(+), 14 deletions(-) create mode 100644 src/main/java/io/github/dsheirer/sample/SerializedListener.java create mode 100644 src/test/java/io/github/dsheirer/sample/SerializedListenerTest.java 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 1834f66b..7bc08ce6 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 @@ -23,6 +23,7 @@ import io.github.dsheirer.dsp.filter.channelizer.output.OneChannelOutputProcesso import io.github.dsheirer.dsp.filter.channelizer.output.TwoChannelOutputProcessor; import io.github.dsheirer.dsp.filter.design.FilterDesignException; import io.github.dsheirer.sample.Listener; +import io.github.dsheirer.sample.SerializedListener; import io.github.dsheirer.sample.complex.ComplexSamples; import io.github.dsheirer.source.SourceEvent; import io.github.dsheirer.source.tuner.channel.TunerChannel; @@ -45,7 +46,8 @@ public class PolyphaseChannelSource extends TunerChannelSource implements Listen { private Logger mLog = LoggerFactory.getLogger(PolyphaseChannelSource.class); private IPolyphaseChannelOutputProcessor mPolyphaseChannelOutputProcessor; - private Listener mSamplesListener; + //Output processor replacement can overlap the outgoing and incoming dispatcher threads, and consumers aren't thread safe + private final SerializedListener mSamplesListener = new SerializedListener<>(); private double mChannelSampleRate; private long mIndexCenterFrequency; private List mOutputProcessorIndexes = new ArrayList<>(); @@ -148,23 +150,20 @@ public class PolyphaseChannelSource extends TunerChannelSource implements Listen @Override public void setListener(final Listener listener) { - mSamplesListener = listener; + mSamplesListener.setListener(listener); mPolyphaseChannelOutputProcessor.setListener(this); } @Override public void receive(ComplexSamples complexSamples) { - if(mSamplesListener != null) + try { - try - { - mSamplesListener.receive(complexSamples); - } - catch(Throwable t) - { - mLog.error("Error dispatching complex samples to listener [" + mSamplesListener.getClass() + "]", t); - } + mSamplesListener.receive(complexSamples); + } + catch(Throwable t) + { + mLog.error("Error dispatching complex samples to listener", t); } } diff --git a/src/main/java/io/github/dsheirer/sample/SerializedListener.java b/src/main/java/io/github/dsheirer/sample/SerializedListener.java new file mode 100644 index 00000000..1500ea3a --- /dev/null +++ b/src/main/java/io/github/dsheirer/sample/SerializedListener.java @@ -0,0 +1,71 @@ +/* + * ***************************************************************************** + * Copyright (C) 2014-2026 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 io.github.dsheirer.sample; + +import java.util.concurrent.locks.ReentrantLock; + +/** + * Delivers to a listener that isn't thread safe from producers that can overlap. Dispatcher.stop() doesn't wait for + * an in-progress buffer, so when a producer is replaced, the outgoing dispatcher thread can still be inside the listener + * when the incoming dispatcher thread delivers its first buffer. This class makes the incoming thread wait for the + * outgoing thread to leave the listener, which means that the thread that replaces the producer doesn't have to wait. + * + * Producers should acquire no other locks before calling receive(), so that this is always the outermost lock. + * + * @param type of element delivered to the listener + */ +public class SerializedListener implements Listener +{ + private final ReentrantLock mLock = new ReentrantLock(); + private volatile Listener mListener; + + /** + * Sets or changes the listener. This does not wait for a delivery that is in progress to the previous listener. + * @param listener to receive elements, or null to ignore elements. + */ + public void setListener(Listener listener) + { + mListener = listener; + } + + /** + * Delivers the element to the listener once no other thread is delivering to the listener. + * @param t element to deliver + */ + @Override + public void receive(T t) + { + mLock.lock(); + + try + { + Listener listener = mListener; + + if(listener != null) + { + listener.receive(t); + } + } + finally + { + mLock.unlock(); + } + } +} diff --git a/src/main/java/io/github/dsheirer/source/tuner/channel/MultiFrequencyTunerChannelSource.java b/src/main/java/io/github/dsheirer/source/tuner/channel/MultiFrequencyTunerChannelSource.java index fb8426e7..e2e99753 100644 --- a/src/main/java/io/github/dsheirer/source/tuner/channel/MultiFrequencyTunerChannelSource.java +++ b/src/main/java/io/github/dsheirer/source/tuner/channel/MultiFrequencyTunerChannelSource.java @@ -21,6 +21,7 @@ package io.github.dsheirer.source.tuner.channel; import com.google.common.eventbus.Subscribe; import io.github.dsheirer.sample.Listener; +import io.github.dsheirer.sample.SerializedListener; import io.github.dsheirer.sample.complex.ComplexSamples; import io.github.dsheirer.source.Source; import io.github.dsheirer.source.SourceEvent; @@ -51,7 +52,8 @@ public class MultiFrequencyTunerChannelSource extends TunerChannelSource private List mLockedFrequencies = new ArrayList<>(); private int mFrequencyListPointer = 0; private ChannelSpecification mChannelSpecification; - private Listener mComplexSamplesListener; + //Rotating to the next source can overlap the outgoing and incoming dispatcher threads and consumers aren't thread safe + private final SerializedListener mComplexSamplesListener = new SerializedListener<>(); private Listener mHeartbeatListener; private String mPreferredTuner; private AtomicBoolean mChangingChannels = new AtomicBoolean(); @@ -285,11 +287,11 @@ public class MultiFrequencyTunerChannelSource extends TunerChannelSource @Override public void setListener(Listener complexSamplesListener) { - mComplexSamplesListener = complexSamplesListener; + mComplexSamplesListener.setListener(complexSamplesListener); if(mTunerChannelSource != null) { - mTunerChannelSource.setListener(complexSamplesListener); + mTunerChannelSource.setListener(mComplexSamplesListener); } } diff --git a/src/test/java/io/github/dsheirer/dsp/filter/channelizer/PolyphaseChannelSourceTest.java b/src/test/java/io/github/dsheirer/dsp/filter/channelizer/PolyphaseChannelSourceTest.java index 9ebcc14f..73613d28 100644 --- a/src/test/java/io/github/dsheirer/dsp/filter/channelizer/PolyphaseChannelSourceTest.java +++ b/src/test/java/io/github/dsheirer/dsp/filter/channelizer/PolyphaseChannelSourceTest.java @@ -25,6 +25,7 @@ import java.util.ArrayDeque; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; import org.junit.jupiter.api.Test; import static org.junit.jupiter.api.Assertions.assertEquals; @@ -125,6 +126,102 @@ public class PolyphaseChannelSourceTest assertEquals(1, received.get()); } + /** + * Test: replacing the output processor while the outgoing dispatcher thread is still inside the consumer doesn't + * let the incoming dispatcher thread into the consumer until the outgoing thread leaves. + */ + @Test + void outputProcessorReplacementDoesNotOverlapConsumerThreads() throws Exception + { + CountDownLatch firstEntered = new CountDownLatch(1); + CountDownLatch releaseFirst = new CountDownLatch(1); + CountDownLatch secondEntered = new CountDownLatch(1); + AtomicInteger active = new AtomicInteger(); + AtomicInteger maximumActive = new AtomicInteger(); + AtomicReference firstThread = new AtomicReference<>(); + + PolyphaseChannelSource source = source(); + source.setListener(samples -> { + maximumActive.accumulateAndGet(active.incrementAndGet(), Math::max); + + if(firstThread.compareAndSet(null, Thread.currentThread())) + { + firstEntered.countDown(); + + try + { + releaseFirst.await(10, TimeUnit.SECONDS); + } + catch(InterruptedException ie) + { + Thread.currentThread().interrupt(); + } + } + else if(Thread.currentThread() != firstThread.get()) + { + secondEntered.countDown(); + } + + active.decrementAndGet(); + }); + + source.start(); + + try + { + source.receiveChannelResults(largeBatch()); + assertTrue(firstEntered.await(5, TimeUnit.SECONDS), "Consumer did not receive samples"); + + source.updateOutputProcessor(calculatorForTwoChannels(source), new SynthesisFilterManager()); + source.applyPendingOutputProcessorUpdate(); + assertEquals(2, source.getOutputProcessorIndexes().size()); + + //Give the replacement processor enough samples and time to attempt delivery to the consumer + for(int x = 0; x < 20; x++) + { + source.receiveChannelResults(largeBatch()); + Thread.sleep(25); + } + + assertEquals(1, maximumActive.get(), "Dispatcher threads overlapped in the consumer"); + + releaseFirst.countDown(); + assertTrue(secondEntered.await(5, TimeUnit.SECONDS), "Replacement processor did not deliver samples"); + assertEquals(1, maximumActive.get(), "Dispatcher threads overlapped in the consumer"); + } + finally + { + releaseFirst.countDown(); + source.stop(); + } + } + + /** + * Finds a tuner center frequency that positions the source's channel across two polyphase channels. + */ + private static ChannelCalculator calculatorForTwoChannels(PolyphaseChannelSource source) + { + for(long offset = 0; offset < 25000; offset += 500) + { + ChannelCalculator calculator = calculator(CENTER + offset); + + if(calculator.getChannelIndexes(source.getTunerChannel()).size() == 2) + { + return calculator; + } + } + + throw new IllegalStateException("Unable to position the channel across two polyphase channels"); + } + + /** + * Batch with enough blocks to complete an output processor sample buffer. + */ + private static ChannelResultsBatch largeBatch() + { + return new ChannelResultsBatch(new float[CHANNEL_COUNT * 2 * 4096], CHANNEL_COUNT * 2, 0, new ArrayDeque<>()); + } + /** * Polyphase channel source for overriding the channel results processing. */ diff --git a/src/test/java/io/github/dsheirer/sample/SerializedListenerTest.java b/src/test/java/io/github/dsheirer/sample/SerializedListenerTest.java new file mode 100644 index 00000000..0ccccbac --- /dev/null +++ b/src/test/java/io/github/dsheirer/sample/SerializedListenerTest.java @@ -0,0 +1,117 @@ +/* + * ***************************************************************************** + * Copyright (C) 2014-2026 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 io.github.dsheirer.sample; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +public class SerializedListenerTest +{ + /** + * Test: a second thread waits until the first thread leaves the listener. + */ + @Test + void secondThreadWaitsForFirstThreadToLeaveListener() throws Exception + { + CountDownLatch firstEntered = new CountDownLatch(1); + CountDownLatch releaseFirst = new CountDownLatch(1); + CountDownLatch secondEntered = new CountDownLatch(1); + + SerializedListener listener = new SerializedListener<>(); + listener.setListener(value -> { + if(value == 1) + { + firstEntered.countDown(); + await(releaseFirst); + } + else + { + secondEntered.countDown(); + } + }); + + Thread first = new Thread(() -> listener.receive(1)); + first.start(); + assertTrue(firstEntered.await(5, TimeUnit.SECONDS)); + + Thread second = new Thread(() -> listener.receive(2)); + second.start(); + assertFalse(secondEntered.await(200, TimeUnit.MILLISECONDS), "Threads overlapped in the listener"); + + releaseFirst.countDown(); + assertTrue(secondEntered.await(5, TimeUnit.SECONDS), "Second thread was not released"); + first.join(5000); + second.join(5000); + } + + /** + * Test: a listener exception releases the lock for the next delivery. + */ + @Test + void listenerExceptionReleasesLock() throws Exception + { + AtomicInteger received = new AtomicInteger(); + SerializedListener listener = new SerializedListener<>(); + listener.setListener(value -> { + if(value == 1) + { + throw new IllegalStateException("Test failure"); + } + received.incrementAndGet(); + }); + + assertThrows(IllegalStateException.class, () -> listener.receive(1)); + + Thread other = new Thread(() -> listener.receive(2)); + other.start(); + other.join(5000); + + assertEquals(1, received.get()); + } + + /** + * Test: elements are ignored when there isn't a listener. + */ + @Test + void nullListenerIgnoresElements() + { + assertDoesNotThrow(() -> new SerializedListener().receive(1)); + } + + private static void await(CountDownLatch latch) + { + try + { + latch.await(10, TimeUnit.SECONDS); + } + catch(InterruptedException ie) + { + Thread.currentThread().interrupt(); + } + } +}