GitHub/sdrtrunk
1
0
Fork 0
mirror of https://github.com/DSheirer/sdrtrunk.git synced 2026-11-09 15:17:24 -05:00
Code Issues Packages Projects Releases 2 Wiki Activity

#2438 resolves broadcaster error that happens during tuner source swapout (#2550)

Browse Source 
Co-authored-by: denny <dennis.sheirer@gmail.com>
This commit is contained in:
Denny Sheirer
2026-11-01 06:04:57 -05:00
committed by GitHub
parent 49c3d3ff55
commit e385d2fdec
5 changed files with 300 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
src/main/java/io/github/dsheirer/dsp/filter/channelizer/PolyphaseChannelSource.java
View File

@@ -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<ComplexSamples> mSamplesListener;
//Output processor replacement can overlap the outgoing and incoming dispatcher threads, and consumers aren't thread safe
private final SerializedListener<ComplexSamples> mSamplesListener = new SerializedListener<>();
private double mChannelSampleRate;
private long mIndexCenterFrequency;
private List<Integer> mOutputProcessorIndexes = new ArrayList<>();
@@ -148,23 +150,20 @@ public class PolyphaseChannelSource extends TunerChannelSource implements Listen
@Override
public void setListener(final Listener<ComplexSamples> 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);
}
}

71
src/main/java/io/github/dsheirer/sample/SerializedListener.java Normal file
View File

@@ -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 <http://www.gnu.org/licenses/>
* ****************************************************************************
*/
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 <T> type of element delivered to the listener
*/
public class SerializedListener<T> implements Listener<T>
{
private final ReentrantLock mLock = new ReentrantLock();
private volatile Listener<T> 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<T> 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<T> listener = mListener;
if(listener != null)
{
listener.receive(t);
}
}
finally
{
mLock.unlock();
}
}
}

8
src/main/java/io/github/dsheirer/source/tuner/channel/MultiFrequencyTunerChannelSource.java
View File

@@ -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<Long> mLockedFrequencies = new ArrayList<>();
private int mFrequencyListPointer = 0;
private ChannelSpecification mChannelSpecification;
private Listener<ComplexSamples> mComplexSamplesListener;
//Rotating to the next source can overlap the outgoing and incoming dispatcher threads and consumers aren't thread safe
private final SerializedListener<ComplexSamples> mComplexSamplesListener = new SerializedListener<>();
private Listener<Heartbeat> mHeartbeatListener;
private String mPreferredTuner;
private AtomicBoolean mChangingChannels = new AtomicBoolean();
@@ -285,11 +287,11 @@ public class MultiFrequencyTunerChannelSource extends TunerChannelSource
@Override
public void setListener(Listener<ComplexSamples> complexSamplesListener)
{
mComplexSamplesListener = complexSamplesListener;
mComplexSamplesListener.setListener(complexSamplesListener);
if(mTunerChannelSource != null)
{
mTunerChannelSource.setListener(complexSamplesListener);
mTunerChannelSource.setListener(mComplexSamplesListener);
}
}

97
src/test/java/io/github/dsheirer/dsp/filter/channelizer/PolyphaseChannelSourceTest.java
View File

@@ -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<Thread> 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.
*/

117
src/test/java/io/github/dsheirer/sample/SerializedListenerTest.java Normal file
View File

@@ -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 <http://www.gnu.org/licenses/>
* ****************************************************************************
*/
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<Integer> 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<Integer> 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<Integer>().receive(1));
}
private static void await(CountDownLatch latch)
{
try
{
latch.await(10, TimeUnit.SECONDS);
}
catch(InterruptedException ie)
{
Thread.currentThread().interrupt();
}
}
}