diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index 67df2a4a2..256379566 100644 --- a/docs/CHANGELOG.md +++ b/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/ diff --git a/src/algorithms/libs/stream_batcher.cc b/src/algorithms/libs/stream_batcher.cc index 5c3a14c9f..bdbc3ba5d 100644 --- a/src/algorithms/libs/stream_batcher.cc +++ b/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(); + } } } diff --git a/src/core/receiver/gnss_flowgraph.cc b/src/core/receiver/gnss_flowgraph.cc index 9f3a095fe..f6cdc388a 100644 --- a/src/core/receiver/gnss_flowgraph.cc +++ b/src/core/receiver/gnss_flowgraph.cc @@ -956,6 +956,11 @@ void GNSSFlowgraph::plan_signal_conditioners() std::vector> 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(); diff --git a/tests/unit-tests/signal-processing-blocks/adapter/stream_batcher_test.cc b/tests/unit-tests/signal-processing-blocks/adapter/stream_batcher_test.cc index 8898aad3d..61cfcfb24 100644 --- a/tests/unit-tests/signal-processing-blocks/adapter/stream_batcher_test.cc +++ b/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& 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& sink) { return sink->received.load() == 17; }); top->stop(); top->wait(); + EXPECT_TRUE(published_before_limit); std::vector expected(17); std::iota(expected.begin(), expected.end(), 0); for (const auto& sink : sinks)