2537 Complex Vector Oscillator (256/512) Lanes Can Drift Apart Over Time (#2538)

* #2535 Heterodyne Channelizer enhancements and corrections.

* #2537 Complex vector oscillator SIMD lane drift.

Realigns the SIMD lanes every 262,144 samples. At frequencies where the
rotation for a set of lanes is nearly a whole or half turn, rounding affects
each lane differently and the lanes drift apart until the next frequency
change, producing spurs at multiples of the sample rate divided by the lane
count.  Lane phase error at 3125 Hz with a 50 kHz sample rate was about -38 dBc
after one hour and is now about -83 dBc with 16 lanes and -73 dBc with 8 lanes.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* #2537 test updates

* #2537 test updates

---------

Co-authored-by: denny <dennis.sheirer@gmail.com>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

src/main/java/io/github/dsheirer/source/tuner/channel/HalfBandTunerChannelSource.java

@@ -30,6 +30,7 @@ import io.github.dsheirer.source.SourceEvent;
 import io.github.dsheirer.source.tuner.frequency.TunerFrequencyErrorManager;
 import io.github.dsheirer.util.Dispatcher;
 import java.util.Iterator;
+import java.util.concurrent.atomic.AtomicBoolean;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -49,11 +50,15 @@ public class HalfBandTunerChannelSource<T extends INativeBuffer> extends TunerCh
     private final ComplexMixer mFrequencyCorrectionMixer;
     private final IRealDecimationFilter mIDecimationFilter;
     private final IRealDecimationFilter mQDecimationFilter;
+    private final boolean mReuseMixerOutput;
+    private float[] mIMixerOutput;
+    private float[] mQMixerOutput;
     private Listener<ComplexSamples> mSamplesListener;
     private final double mChannelSampleRate;
-    private long mChannelFrequency;
+    private volatile long mChannelFrequency;
     private final long mFrequencyOffset;
-    private long mFrequencyCorrection;
+    private volatile long mFrequencyCorrection;
+    private final AtomicBoolean mPendingMixerUpdate = new AtomicBoolean();
 
     /**
      * Constructs a frequency translating and CIC decimating channel source.
@@ -80,8 +85,11 @@ public class HalfBandTunerChannelSource<T extends INativeBuffer> extends TunerCh
         mIDecimationFilter = DecimationFilterFactory.getRealDecimationFilter(decimation);
         mQDecimationFilter = DecimationFilterFactory.getRealDecimationFilter(decimation);
 
+        //The decimation filters copy the mixed samples, unless there's no decimation and the mixed samples are the output
+        mReuseMixerOutput = decimation > 1;
+
         //Set dispatcher to process 1/10 of estimated sample arrival rate, 20 times per second (up to 200% per interval)
-        mBufferDispatcher = new Dispatcher(threadName, 50, getHeartbeatManager());
+        mBufferDispatcher = new Dispatcher(threadName, 50, getHeartbeatManager(), BUFFER_MAX_CAPACITY);
         mBufferDispatcher.setListener(new NativeBufferProcessor());
 
         //Setup the frequency mixer to the current source frequency
@@ -92,21 +100,22 @@ public class HalfBandTunerChannelSource<T extends INativeBuffer> extends TunerCh
     }
 
     /**
-     * Sets and applies the channel decoder requested frequency correction.
+     * Sets the channel decoder requested frequency correction, which is applied on the sample processing thread.
      * @param correction requested by the downstream processors.
      */
     @Override
     public void setFrequencyCorrection(long correction)
     {
         mFrequencyCorrection = correction;
-        updateMixerFrequencyOffset();
+        mPendingMixerUpdate.set(true);
     }
 
     @Override
     public void start()
     {
-        super.start();
+        //The dispatcher has to be running to accept delayed sample buffers preloaded in response to the start request
         mBufferDispatcher.start();
+        super.start();
     }
 
     @Override
@@ -131,7 +140,8 @@ public class HalfBandTunerChannelSource<T extends INativeBuffer> extends TunerCh
     }
 
     /**
-     * Sets/updates the center frequency for the sample streaming being sent from the producer.
+     * Sets/updates the center frequency for the sample streaming being sent from the producer, which is applied on
+     * the sample processing thread.
      *
      * @param frequency in hertz
      */
@@ -139,12 +149,14 @@ public class HalfBandTunerChannelSource<T extends INativeBuffer> extends TunerCh
     public void setFrequency(long frequency)
     {
         mChannelFrequency = frequency;
-        updateMixerFrequencyOffset();
+        mPendingMixerUpdate.set(true);
     }
 
     /**
      * Calculates the local mixer frequency offset from the tuned frequency,
-     * channel's requested frequency, and channel frequency correction.
+     * channel's requested frequency, and channel frequency correction.  This is only invoked on the sample processing
+     * thread so that the latest values from the tuner and channel frequency correction threads are applied together
+     * and the mixer is never updated while it is generating samples.
      */
     private void updateMixerFrequencyOffset()
     {
@@ -189,6 +201,27 @@ public class HalfBandTunerChannelSource<T extends INativeBuffer> extends TunerCh
         return mChannelSampleRate;
     }
 
+    /**
+     * Mixes the samples to baseband, reusing the mixer output arrays when they aren't passed to the consumer.
+     * @param samples to mix, which aren't modified since they can be shared with other channels.
+     * @return mixed samples
+     */
+    private ComplexSamples mix(ComplexSamples samples)
+    {
+        if(!mReuseMixerOutput)
+        {
+            return mFrequencyCorrectionMixer.mix(samples);
+        }
+
+        if(mIMixerOutput == null || mIMixerOutput.length != samples.i().length)
+        {
+            mIMixerOutput = new float[samples.i().length];
+            mQMixerOutput = new float[samples.q().length];
+        }
+
+        return mFrequencyCorrectionMixer.mix(samples, mIMixerOutput, mQMixerOutput);
+    }
+
     /**
      * Processes native buffers received from the dispatcher and sends to the registered listener
      */
@@ -203,7 +236,12 @@ public class HalfBandTunerChannelSource<T extends INativeBuffer> extends TunerCh
 
                 while(iterator.hasNext())
                 {
-                    ComplexSamples basebanded = mFrequencyCorrectionMixer.mix(iterator.next());
+                    if(mPendingMixerUpdate.compareAndSet(true, false))
+                    {
+                        updateMixerFrequencyOffset();
+                    }
+
+                    ComplexSamples basebanded = mix(iterator.next());
                     float[] i = mIDecimationFilter.decimateReal(basebanded.i());
                     float[] q = mQDecimationFilter.decimateReal(basebanded.q());
 

src/main/java/io/github/dsheirer/source/tuner/channel/TunerChannelSource.java

@@ -40,6 +40,7 @@ public abstract class TunerChannelSource extends ComplexSource implements ISourc
     private Listener<SourceEvent> mConsumerSourceEventListener;
     protected String mThreadName;
     private final ChannelFrequencyErrorManager mChannelFrequencyErrorManager;
+    private volatile long mStartTimestamp;
 
     /**
      * Tuner Channel Source is a Digital Drop Channel (DDC) abstract class that defines the minimum functionality
@@ -111,6 +112,16 @@ public abstract class TunerChannelSource extends ComplexSource implements ISourc
         return mTunerChannel;
     }
 
+    /**
+     * Sets the timestamp for the oldest samples to provide to this channel when it starts.  This only applies to
+     * sources where the producer incorporates a sample delay buffer.
+     * @param timestamp in milliseconds since epoch, or 0 to start with current samples.
+     */
+    public void setStartTimestamp(long timestamp)
+    {
+        mStartTimestamp = timestamp;
+    }
+
     /**
      * Starts this tuner channel source producing sample stream.
      */
@@ -120,7 +131,8 @@ public abstract class TunerChannelSource extends ComplexSource implements ISourc
 
         //Broadcast current frequency and sample rate so consumer can configure correctly
         broadcastConsumerSourceEvent(SourceEvent.frequencyChange(this, getFrequency(), "Startup"));
-        broadcastProducerSourceEvent(SourceEvent.startSampleStreamRequest(this));
+        broadcastProducerSourceEvent(SourceEvent.startSampleStreamRequest(this,
+            mStartTimestamp > 0 ? mStartTimestamp : System.currentTimeMillis()));
     }
 
     /**

src/main/java/io/github/dsheirer/source/tuner/manager/HeterodyneChannelSourceManager.java

@@ -32,7 +32,9 @@ import io.github.dsheirer.source.tuner.channel.TunerChannel;
 import io.github.dsheirer.source.tuner.channel.TunerChannelSource;
 import io.github.dsheirer.source.tuner.frequency.TunerFrequencyErrorManager;
 import java.util.ArrayList;
+import java.util.HashSet;
 import java.util.List;
+import java.util.Set;
 import java.util.SortedSet;
 import java.util.TreeSet;
 import java.util.concurrent.CopyOnWriteArrayList;
@@ -49,7 +51,8 @@ public class HeterodyneChannelSourceManager extends ChannelSourceManager
     private final static int DELAY_BUFFER_DURATION_MILLISECONDS = 2000;
 
     private List<HalfBandTunerChannelSource> mChannelSources = new CopyOnWriteArrayList<>();
-    private SortedSet<TunerChannel> mTunerChannels = new TreeSet<>();
+    //Channel sources that are registered with the delay buffer, guarded by the tuner controller lock
+    private Set<HalfBandTunerChannelSource> mStartedChannelSources = new HashSet<>();
     private TunerController mTunerController;
     private ChannelSourceEventProcessor mChannelSourceEventProcessor = new ChannelSourceEventProcessor();
     private NativeSampleDelayBuffer mSampleDelayBuffer;
@@ -75,7 +78,7 @@ public class HeterodyneChannelSourceManager extends ChannelSourceManager
     public String getStateDescription()
     {
         StringBuilder sb = new StringBuilder();
-        sb.append("Heterodyne Channel Source Manager Providing [").append(mTunerChannels.size()).append("] Channels");
+        sb.append("Heterodyne Channel Source Manager Providing [").append(getTunerChannelCount()).append("] Channels");
         sb.append("\n\tTuner Controller Frequency: ").append(mTunerController.getFrequency());
         for(HalfBandTunerChannelSource channelSource: mChannelSources)
         {
@@ -102,16 +105,28 @@ public class HeterodyneChannelSourceManager extends ChannelSourceManager
         }
     }
 
+    /**
+     * Tuner channels for the channel sources.  Note: tuner channels are ordered by frequency, so channel sources that
+     * share a frequency have a single entry.
+     * @return new set of tuner channels.
+     */
     @Override
     public SortedSet<TunerChannel> getTunerChannels()
     {
-        return mTunerChannels;
+        SortedSet<TunerChannel> tunerChannels = new TreeSet<>();
+
+        for(HalfBandTunerChannelSource channelSource: mChannelSources)
+        {
+            tunerChannels.add(channelSource.getTunerChannel());
+        }
+
+        return tunerChannels;
     }
 
     @Override
     public int getTunerChannelCount()
     {
-        return mTunerChannels.size();
+        return mChannelSources.size();
     }
 
     @Override
@@ -129,9 +144,17 @@ public class HeterodyneChannelSourceManager extends ChannelSourceManager
 
         try
         {
-            if(CenterFrequencyCalculator.canTune(tunerChannel, mTunerController, mTunerChannels))
+            if(CenterFrequencyCalculator.canTune(tunerChannel, mTunerController, getTunerChannels()))
             {
-                //Attempt to create the channel source first, in case we get a filter design exception
+                //Update the tuner center frequency first, so that a channel isn't provided if the tuner can't support it
+                SortedSet<TunerChannel> tunerChannels = getTunerChannels();
+                tunerChannels.add(tunerChannel);
+
+                if(!updateTunerFrequency(tunerChannels))
+                {
+                    return null;
+                }
+
                 HalfBandTunerChannelSource<?> tunerChannelSource = new HalfBandTunerChannelSource(mChannelSourceEventProcessor,
                         tunerChannel, mTunerController.getSampleRate(), channelSpecification, threadName,
                         getTunerFrequencyErrorManager());
@@ -142,10 +165,6 @@ public class HeterodyneChannelSourceManager extends ChannelSourceManager
                 //Set the current tuner frequency
                 tunerChannelSource.setFrequency(mTunerController.getFrequency());
 
-                //Add to the channel list and update the tuner center frequency as needed
-                mTunerChannels.add(tunerChannel);
-                updateTunerFrequency();
-
                 //Lock the tuner controller frequency and sample rate
                 mTunerController.setLockedSampleRate(true);
 
@@ -176,18 +195,21 @@ public class HeterodyneChannelSourceManager extends ChannelSourceManager
     }
 
     /**
-     * Calculates a new center frequency and updates the tuner center frequency
+     * Calculates a new center frequency and updates the tuner center frequency when the tuner isn't tuned for the
+     * tuner channels.
+     * @param tunerChannels that the tuner must provide
+     * @return true if the tuner is tuned for the tuner channels.
      */
-    private void updateTunerFrequency()
+    private boolean updateTunerFrequency(SortedSet<TunerChannel> tunerChannels)
     {
-        if(!mTunerController.isTunedFor(getTunerChannels()))
+        if(!mTunerController.isTunedFor(tunerChannels))
         {
-            long centerFrequency = CenterFrequencyCalculator.getCenterFrequency(mTunerController, getTunerChannels());
+            long centerFrequency = CenterFrequencyCalculator.getCenterFrequency(mTunerController, tunerChannels);
 
             if(centerFrequency == CenterFrequencyCalculator.INVALID_FREQUENCY)
             {
                 mLog.error("Couldn't calculate center frequency for tuner and tuner channels");
-                return;
+                return false;
             }
 
             if(centerFrequency != mTunerController.getFrequency())
@@ -199,9 +221,12 @@ public class HeterodyneChannelSourceManager extends ChannelSourceManager
                 catch(SourceException se)
                 {
                     mLog.error("Couldn't update tuner center frequency to " + centerFrequency, se);
+                    return false;
                 }
             }
         }
+
+        return true;
     }
 
     @Override
@@ -214,9 +239,18 @@ public class HeterodyneChannelSourceManager extends ChannelSourceManager
                 updateTunerFrequency(tunerSourceEvent.getValue().longValue());
 
                 //Clear the delay buffer since any delayed samples will be centered on the previous frequency
-                if(mSampleDelayBuffer != null)
+                mTunerController.getLock().lock();
+
+                try
                 {
-                    mSampleDelayBuffer.clear();
+                    if(mSampleDelayBuffer != null)
+                    {
+                        mSampleDelayBuffer.clear();
+                    }
+                }
+                finally
+                {
+                    mTunerController.getLock().unlock();
                 }
                 break;
             case NOTIFICATION_FREQUENCY_CORRECTION_CHANGE:
@@ -291,7 +325,7 @@ public class HeterodyneChannelSourceManager extends ChannelSourceManager
      */
     private void stopDelayBuffer()
     {
-        if(mSampleDelayBuffer != null && !mSampleDelayBuffer.hasListeners())
+        if(mSampleDelayBuffer != null)
         {
             mTunerController.removeBufferListener(mSampleDelayBuffer);
             mSampleDelayBuffer.dispose();
@@ -299,6 +333,72 @@ public class HeterodyneChannelSourceManager extends ChannelSourceManager
         }
     }
 
+    /**
+     * Registers the channel source with the delay buffer to start the flow of sample buffers to the channel.
+     * @param channelSource to start
+     * @param timestamp of the oldest delayed sample buffers to preload to the channel
+     */
+    private void startChannelSource(HalfBandTunerChannelSource channelSource, long timestamp)
+    {
+        mTunerController.getLock().lock();
+
+        try
+        {
+            startDelayBuffer();
+
+            //The start sample stream request contains a start timestamp and the delay buffer
+            //will preload the channel with delayed sample buffers that either contain the
+            //timestamp or occur later/newer than the timestamp.
+            mSampleDelayBuffer.addListener(channelSource, timestamp);
+            mStartedChannelSources.add(channelSource);
+        }
+        finally
+        {
+            mTunerController.getLock().unlock();
+        }
+    }
+
+    /**
+     * Stops the flow of sample buffers to the channel source and releases the tuner channel.  Stops the delay buffer
+     * when this is the last started channel.
+     *
+     * Note: the delay buffer applies listener removal on the next sample buffer, so the started channel sources set
+     * is used to detect the last channel instead of the delay buffer's listeners.
+     * @param channelSource to stop, which may not have been started.
+     */
+    private void stopChannelSource(HalfBandTunerChannelSource channelSource)
+    {
+        mTunerController.getLock().lock();
+
+        try
+        {
+            if(mStartedChannelSources.remove(channelSource))
+            {
+                mSampleDelayBuffer.removeListener(channelSource);
+
+                if(mStartedChannelSources.isEmpty())
+                {
+                    stopDelayBuffer();
+                }
+            }
+
+            mChannelSources.remove(channelSource);
+            channelSource.dispose();
+
+            //Unlock the tuner controller if there are no more channels
+            if(getTunerChannelCount() == 0)
+            {
+                mTunerController.setLockedSampleRate(false);
+            }
+
+            broadcast(SourceEvent.channelCountChange(getTunerChannelCount()));
+        }
+        finally
+        {
+            mTunerController.getLock().unlock();
+        }
+    }
+
     /**
      * Processes channel source events
      */
@@ -309,34 +409,16 @@ public class HeterodyneChannelSourceManager extends ChannelSourceManager
         {
             switch(sourceEvent.getEvent())
             {
-//TODO: protect start/stop processing with a reentrant lock
                 case REQUEST_START_SAMPLE_STREAM:
-                    if(sourceEvent.getSource() instanceof HalfBandTunerChannelSource)
+                    if(sourceEvent.getSource() instanceof HalfBandTunerChannelSource halfBandSource)
                     {
-                        startDelayBuffer();
-
-                        //The start sample stream request contains a start timestamp and the delay buffer
-                        //will preload the channel with delayed sample buffers that either contain the
-                        //timestamp or occur later/newer than the timestamp.
-                        mSampleDelayBuffer.addListener((HalfBandTunerChannelSource)sourceEvent.getSource(),
-                                sourceEvent.getValue().longValue());
+                        startChannelSource(halfBandSource, sourceEvent.getValue().longValue());
                     }
                     break;
                 case REQUEST_STOP_SAMPLE_STREAM:
                     if(sourceEvent.getSource() instanceof HalfBandTunerChannelSource halfBandSource)
                     {
-                        mSampleDelayBuffer.removeListener(halfBandSource);
-                        stopDelayBuffer();
-                        mChannelSources.remove(halfBandSource);
-                        mTunerChannels.remove(halfBandSource.getTunerChannel());
-                        halfBandSource.dispose();
-
-                        //Unlock the tuner controller if there are no more channels
-                        if(getTunerChannelCount() == 0)
-                        {
-                            mTunerController.setLockedSampleRate(false);
-                        }
-                        broadcast(SourceEvent.channelCountChange(getTunerChannelCount()));
+                        stopChannelSource(halfBandSource);
                     }
                     break;
                 case NOTIFICATION_CHANNEL_COUNT_CHANGE: