Fix for GNU Radio 3.7

docs/CHANGELOG.md

@@ -20,7 +20,15 @@ All notable changes to GNSS-SDR will be documented in this file.
   sample batching and redundant pass-through removal, reducing channel
   scheduling overhead without configuration changes. Preserves samples and
   stream tags, flushes partial batches on deadlines or EOF, and improves
-  flowgraph connection cleanup.
+  flowgraph connection cleanup. The new `SignalConditioner.batch_size_ms`,
+  `SignalConditioner.max_batch_latency_ms`,
+  `SignalConditioner.remove_pass_through` and
+  `SignalConditioner.output_buffer_ms` parameters override the automatic
+  choices, and the new
+  [`Bypass`](https://gnss-sdr.org/docs/sp-blocks/signal-conditioner/#implementation-bypass)
+  Signal Conditioner implementation connects the channels directly to the signal
+  source. See the
+  [Signal Conditioner documentation](https://gnss-sdr.org/docs/sp-blocks/signal-conditioner/#batching-and-buffering).
 
 See the definitions of concepts and metrics at
 https://gnss-sdr.org/design-forces/

src/algorithms/libs/stream_batcher.cc

@@ -159,9 +159,16 @@ void stream_batcher::deadline_expired(const pmt::pmt_t& message)
     deadlines_->acknowledge(deadline_id_);
     if (armed_ && pmt::is_uint64(message) && pmt::to_uint64(message) == generation_)
         {
-            // GNU Radio 3.7 handles messages without leaving the wait for
-            // input: there the partial batch goes out with the next arrival.
             flush_ = true;
+            // GNU Radio 3.7 dispatches messages inside its input wait. A
+            // notification alone does not make it reevaluate the forecast.
+            const auto state = detail();
+            if (state)
+                {
+                    gr::thread::scoped_lock lock(state->d_tpb.mutex);
+                    state->d_tpb.input_changed = true;
+                    state->d_tpb.input_cond.notify_one();
+                }
         }
 }
 

src/core/receiver/gnss_flowgraph.cc

@@ -956,6 +956,11 @@ void GNSSFlowgraph::plan_signal_conditioners()
     std::vector<std::set<std::string>> shared_resamplers(sig_conditioner_.size());
     for (size_t i = 0; i < channels_.size(); ++i)
         {
+            if (channels_[i] == nullptr)
+                {
+                    // connect_channels() reports construction errors and unwinds connections.
+                    continue;
+                }
             const auto id = conditioner_id_for_channel(i);
             readers[id].insert(channels_[i]->get_left_block_trk().get());
             const auto signal = channels_[i]->get_signal().get_signal_str();

tests/unit-tests/signal-processing-blocks/adapter/stream_batcher_test.cc

@@ -620,14 +620,17 @@ TEST_F(StreamBatcherTest, DeadlinePublishesWithoutFurtherInputOrEof)
             sinks.push_back(sink);
         }
     top->start();
-    const auto limit = std::chrono::steady_clock::now() + std::chrono::seconds(2);
+    // Allow scheduler jitter, but do not accept the legacy 250 ms input poll.
+    const auto limit = std::chrono::steady_clock::now() + std::chrono::milliseconds(200);
     while (std::chrono::steady_clock::now() < limit &&
            std::any_of(sinks.begin(), sinks.end(), [](const gnss_shared_ptr<ObservedTestSink>& sink) { return sink->received.load() != 17; }))
         {
             std::this_thread::sleep_for(std::chrono::milliseconds(2));
         }
+    const bool published_before_limit = std::all_of(sinks.begin(), sinks.end(), [](const gnss_shared_ptr<ObservedTestSink>& sink) { return sink->received.load() == 17; });
     top->stop();
     top->wait();
+    EXPECT_TRUE(published_before_limit);
     std::vector<int> expected(17);
     std::iota(expected.begin(), expected.end(), 0);
     for (const auto& sink : sinks)