Merge branch 'vladisslav2011-monitor_decimation_fix' into next

AUTHORS

@@ -77,6 +77,7 @@ Piyush Gupta             piyush04111999@gmail.com               Contributor
 Rodrigo Muñoz            rodrigo.munoz@proteinlab.cl            Contributor
 Stefan van der Linden    spvdlinden@gmail.com                   Contributor
 Víctor Castillo-Agüero   victorcastilloaguero@gmail.com         Contributor
+Vladislav P              vladisslav2011@gmail.com               Contributor
 Wenhao Ou                ouwh@mail2.sysu.edu.cn                 Contributor
 Will Silberman           wsilberm@google.com                    Contributor
 Carlos Paniego           carpanie@hotmail.com                   Artwork

CITATION.cff

@@ -108,6 +108,10 @@ authors:
     email: cillian.odriscoll@gmail.com
     family-names: "O'Driscoll"
     given-names: Cillian
+  - alias: vladisslav2011
+    email: vladisslav2011@gmail.com
+    family-names: P
+    given-names: Vladislav
   - affiliation: "Finnish Geospatial Research Institute"
     alias: IntoPa
     email: into.paakkonen@aalto.fi

docs/CHANGELOG.md

@@ -329,6 +329,15 @@ All notable changes to GNSS-SDR will be documented in this file.
 
 ### Improvements in Reliability:
 
+- Fixed the decimation logic of the `Monitor`, `AcquisitionMonitor` and
+  `TrackingMonitor` blocks: `decimation_factor` now selects every N-th epoch and
+  always consumes all the input items, instead of grouping `Gnss_Synchro`
+  objects into bursts and skipping others, and empty datagrams are no longer
+  sent. Fixed a use-after-free memory corruption caused by
+  `google::protobuf::ShutdownProtobufLibrary()` being called from the destructor
+  of `Serdes_Gnss_Synchro`, before the protobuf library was actually used; the
+  library is now shut down only once, at program exit. Added a unit test for the
+  monitor decimation. Contributed by @vladisslav2011.
 - Hardened the Galileo OSNMA protocol implementation, adding support for Chain
   Renewal, Chain Revocation, Public Key Renewal, Public Key Revocation, Merkle
   Tree Renewal, and OSNMA Alert Message events. Improved the management of OSNMA

src/algorithms/PVT/libs/CMakeLists.txt

@@ -120,6 +120,7 @@ get_filename_component(PROTO_INCLUDE_HEADERS_DIR ${PROTO_HDRS} DIRECTORY)
 target_include_directories(pvt_libs
     PUBLIC
         ${GNSSSDR_SOURCE_DIR}/src/core/receiver
+        ${GNSSSDR_SOURCE_DIR}/src/core/libs
 )
 
 # Do not apply clang-tidy fixes to protobuf generated headers

src/algorithms/PVT/libs/serdes_galileo_eph.h

@@ -20,6 +20,7 @@
 
 #include "galileo_ephemeris.h"
 #include "galileo_ephemeris.pb.h"  // file created by Protocol Buffers at compile time
+#include "protobuf_cleanup_manager.h"
 #include <memory>
 #include <string>
 #include <utility>
@@ -42,11 +43,10 @@ public:
         // Verify that the version of the library that we linked against is
         // compatible with the version of the headers we compiled against.
         GOOGLE_PROTOBUF_VERIFY_VERSION;
-    }
 
-    ~Serdes_Galileo_Eph()
-    {
-        // google::protobuf::ShutdownProtobufLibrary();
+        // Make sure google::protobuf::ShutdownProtobufLibrary() is called only once,
+        // at the end of the program execution (see Protobuf_Cleanup_Manager)
+        Protobuf_Cleanup_Manager::get();
     }
 
     inline Serdes_Galileo_Eph(const Serdes_Galileo_Eph& other) noexcept : monitor_(other.monitor_)  //!< Copy constructor

src/algorithms/PVT/libs/serdes_gps_eph.h

@@ -20,6 +20,7 @@
 
 #include "gps_ephemeris.h"
 #include "gps_ephemeris.pb.h"  // file created by Protocol Buffers at compile time
+#include "protobuf_cleanup_manager.h"
 #include <memory>
 #include <string>
 #include <utility>
@@ -41,11 +42,10 @@ public:
         // Verify that the version of the library that we linked against is
         // compatible with the version of the headers we compiled against.
         GOOGLE_PROTOBUF_VERIFY_VERSION;
-    }
 
-    ~Serdes_Gps_Eph()
-    {
-        // google::protobuf::ShutdownProtobufLibrary();
+        // Make sure google::protobuf::ShutdownProtobufLibrary() is called only once,
+        // at the end of the program execution (see Protobuf_Cleanup_Manager)
+        Protobuf_Cleanup_Manager::get();
     }
 
     inline Serdes_Gps_Eph(const Serdes_Gps_Eph& other) noexcept : monitor_(other.monitor_)  //!< Copy constructor

src/algorithms/PVT/libs/serdes_monitor_pvt.h

@@ -20,6 +20,7 @@
 
 #include "monitor_pvt.h"
 #include "monitor_pvt.pb.h"  // file created by Protocol Buffers at compile time
+#include "protobuf_cleanup_manager.h"
 #include <memory>
 #include <string>
 #include <utility>
@@ -42,11 +43,10 @@ public:
         // Verify that the version of the library that we linked against is
         // compatible with the version of the headers we compiled against.
         GOOGLE_PROTOBUF_VERIFY_VERSION;
-    }
 
-    ~Serdes_Monitor_Pvt()
-    {
-        // google::protobuf::ShutdownProtobufLibrary();
+        // Make sure google::protobuf::ShutdownProtobufLibrary() is called only once,
+        // at the end of the program execution (see Protobuf_Cleanup_Manager)
+        Protobuf_Cleanup_Manager::get();
     }
 
     inline Serdes_Monitor_Pvt(const Serdes_Monitor_Pvt& other) noexcept : monitor_(other.monitor_)  //!< Copy constructor

src/core/libs/CMakeLists.txt

@@ -45,6 +45,7 @@ set(CORE_LIBS_HEADERS
     osnma_helper.h
     osnma_msg_receiver.h
     osnma_nav_data_manager.h
+    protobuf_cleanup_manager.h
     serdes_nav_message.h
     string_converter.h
 )

src/core/libs/protobuf_cleanup_manager.h

@@ -0,0 +1,63 @@
+/*!
+ * \file protobuf_cleanup_manager.h
+ * \brief Singleton class that implements the shared resource cleaner pattern
+ * for the Protocol Buffers library
+ * \author Vladislav P, 2026. vladisslav2011(at)gmail.com
+ *
+ * -----------------------------------------------------------------------------
+ *
+ * GNSS-SDR is a Global Navigation Satellite System software-defined receiver.
+ * This file is part of GNSS-SDR.
+ *
+ * Copyright (C) 2010-2026  (see AUTHORS file for a list of contributors)
+ * SPDX-License-Identifier: GPL-3.0-or-later
+ *
+ * -----------------------------------------------------------------------------
+ */
+
+#ifndef GNSS_SDR_PROTOBUF_CLEANUP_MANAGER_H
+#define GNSS_SDR_PROTOBUF_CLEANUP_MANAGER_H
+
+#include <google/protobuf/message_lite.h>
+
+/** \addtogroup Core
+ * \{ */
+/** \addtogroup Core_Receiver_Library
+ * \{ */
+
+/*!
+ * \brief This class implements the Protocol Buffers library shared memory cleanup manager.
+ *
+ * google::protobuf::ShutdownProtobufLibrary() must be called only once, at the end
+ * of the program execution: it is not safe to use any other part of the library
+ * after that call. Calling it earlier (e.g., from the destructor of a class that
+ * uses protobuf) results in use-after-free memory corruption and random crashes.
+ * Call Protobuf_Cleanup_Manager::get() in the constructor of every class that
+ * uses the protobuf library, so that google::protobuf::ShutdownProtobufLibrary()
+ * is called at program shutdown, when the singleton is destroyed.
+ */
+class Protobuf_Cleanup_Manager
+{
+public:
+    static Protobuf_Cleanup_Manager& get()
+    {
+        static Protobuf_Cleanup_Manager instance{};
+        return instance;
+    }
+
+    Protobuf_Cleanup_Manager(const Protobuf_Cleanup_Manager&) = delete;
+    Protobuf_Cleanup_Manager& operator=(const Protobuf_Cleanup_Manager&) = delete;
+    Protobuf_Cleanup_Manager(Protobuf_Cleanup_Manager&&) = delete;
+    Protobuf_Cleanup_Manager& operator=(Protobuf_Cleanup_Manager&&) = delete;
+
+private:
+    Protobuf_Cleanup_Manager() = default;
+    ~Protobuf_Cleanup_Manager()
+    {
+        google::protobuf::ShutdownProtobufLibrary();
+    }
+};
+
+/** \} */
+/** \} */
+#endif  // GNSS_SDR_PROTOBUF_CLEANUP_MANAGER_H

src/core/libs/serdes_nav_message.h

@@ -20,6 +20,7 @@
 
 #include "nav_message.pb.h"  // file created by Protocol Buffers at compile time
 #include "nav_message_packet.h"
+#include "protobuf_cleanup_manager.h"
 #include <cstdint>
 #include <memory>
 #include <string>
@@ -43,11 +44,10 @@ public:
         // Verify that the version of the library that we linked against is
         // compatible with the version of the headers we compiled against.
         GOOGLE_PROTOBUF_VERIFY_VERSION;
-    }
 
-    ~Serdes_Nav_Message()
-    {
-        // google::protobuf::ShutdownProtobufLibrary();
+        // Make sure google::protobuf::ShutdownProtobufLibrary() is called only once,
+        // at the end of the program execution (see Protobuf_Cleanup_Manager)
+        Protobuf_Cleanup_Manager::get();
     }
 
     inline Serdes_Nav_Message(const Serdes_Nav_Message& other) noexcept : navmsg_(other.navmsg_)  //!< Copy constructor

src/core/monitor/CMakeLists.txt

@@ -48,6 +48,7 @@ target_link_libraries(core_monitor
         Gnuradio::runtime
         protobuf::libprotobuf
         core_system_parameters
+        core_libs
     PRIVATE
         Boost::serialization
 )

src/core/monitor/gnss_synchro_monitor.cc

@@ -68,25 +68,42 @@ int gnss_synchro_monitor::general_work(int noutput_items __attribute__((unused))
 {
     // Get the input buffer pointer
     const auto** in = reinterpret_cast<const Gnss_Synchro**>(&input_items[0]);
-
-    // Loop through each input stream channel
+    // Find maximum number of items across of all inputs
+    int n_items_max = 0;
     for (int channel_index = 0; channel_index < d_nchannels; channel_index++)
         {
-            // Loop through each item in each input stream channel
-            for (int item_index = 0; item_index < ninput_items[channel_index]; item_index++)
+            n_items_max = std::max(n_items_max, ninput_items[channel_index]);
+        }
+
+    // Loop through each item in each input stream channel
+    for (int item_index = 0; item_index < n_items_max; item_index++)
+        {
+            // Use the count variable to limit how many items are sent per channel
+            count++;
+            // Gnss_Synchro objects selected in this round, sent in a single datagram
+            std::vector<Gnss_Synchro> stocks;
+            // Loop through each input stream channel
+            for (int channel_index = 0; channel_index < d_nchannels; channel_index++)
                 {
-                    // Use the count variable to limit how many items are sent per channel
-                    count++;
+                    if (item_index >= ninput_items[channel_index])
+                        {
+                            continue;
+                        }
                     if (count >= d_decimation_factor)
                         {
-                            // Convert to a vector and write to the UDP sink
-                            std::vector<Gnss_Synchro> stocks;
                             stocks.push_back(in[channel_index][item_index]);
+                        }
+                    // Consume the number of items for the input stream channel
+                    consume(channel_index, 1);
+                }
+            if (count >= d_decimation_factor)
+                {
+                    // Reset count variable
+                    count = 0;
+                    // Do not send empty datagrams
+                    if (!stocks.empty())
+                        {
                             udp_sink_ptr->write_gnss_synchro(stocks);
-                            // Reset count variable
-                            count = 0;
-                            // Consume the number of items for the input stream channel
-                            consume(channel_index, ninput_items[channel_index]);
                         }
                 }
         }

src/core/monitor/serdes_gnss_synchro.h

@@ -20,6 +20,7 @@
 
 #include "gnss_synchro.h"
 #include "gnss_synchro.pb.h"  // file created by Protocol Buffers at compile time
+#include "protobuf_cleanup_manager.h"
 #include <array>
 #include <string>
 #include <utility>
@@ -37,11 +38,10 @@ public:
         // Verify that the version of the library that we linked against is
         // compatible with the version of the headers we compiled against.
         GOOGLE_PROTOBUF_VERIFY_VERSION;
-    }
 
-    ~Serdes_Gnss_Synchro()
-    {
-        google::protobuf::ShutdownProtobufLibrary();
+        // Make sure google::protobuf::ShutdownProtobufLibrary() is called only once,
+        // at the end of the program execution (see Protobuf_Cleanup_Manager)
+        Protobuf_Cleanup_Manager::get();
     }
 
     inline Serdes_Gnss_Synchro(const Serdes_Gnss_Synchro& other) noexcept  //!< Copy constructor

tests/test_main.cc

@@ -144,6 +144,7 @@ private:
 #include "unit-tests/arithmetic/magnitude_squared_test.cc"
 #include "unit-tests/arithmetic/multiply_test.cc"
 #include "unit-tests/arithmetic/preamble_correlator_test.cc"
+#include "unit-tests/control-plane/gnss_synchro_monitor_test.cc"
 #include "unit-tests/control-plane/in_memory_configuration_test.cc"
 #include "unit-tests/control-plane/protobuf_test.cc"
 #include "unit-tests/control-plane/string_converter_test.cc"

tests/unit-tests/control-plane/gnss_synchro_monitor_test.cc

@@ -0,0 +1,169 @@
+/*!
+ * \file gnss_synchro_monitor_test.cc
+ * \brief This file implements tests for gnss_synchro_monitor
+ * \author Vladislav P, 2026. vladisslav2011(at)gmail.com
+ *
+ * -----------------------------------------------------------------------------
+ *
+ * GNSS-SDR is a Global Navigation Satellite System software-defined receiver.
+ * This file is part of GNSS-SDR.
+ *
+ * Copyright (C) 2010-2026  (see AUTHORS file for a list of contributors)
+ * SPDX-License-Identifier: GPL-3.0-or-later
+ *
+ * -----------------------------------------------------------------------------
+ */
+
+
+#include "gnss_block_interface.h"
+#include "gnss_synchro_monitor.h"
+#include <boost/asio.hpp>
+#include <boost/system/error_code.hpp>
+#include <gnuradio/runtime_types.h>
+#include <gnuradio/sync_block.h>
+#include <gnuradio/top_block.h>
+#include <chrono>
+#include <cstddef>
+#include <cstdint>
+#include <iostream>
+#include <set>
+#include <string>
+#include <thread>
+#include <vector>
+
+class gnss_synchro_emitter;
+
+using gnss_synchro_emitter_sptr = gnss_shared_ptr<gnss_synchro_emitter>;
+
+class gnss_synchro_emitter : public gr::sync_block
+{
+public:
+    virtual ~gnss_synchro_emitter() = default;
+
+    // Emit synchros one by one with a small pause to match the actual "Monitor.enable_monitor=true" GNSS-SDR configuration behavior
+    int work(int /*noutput_items*/, gr_vector_const_void_star& /*input_items*/, gr_vector_void_star& output_items)
+    {
+        auto** out = reinterpret_cast<Gnss_Synchro**>(&output_items[0]);
+        if (counter >= d_nitems)
+            {
+                return WORK_DONE;
+            }
+        for (int i = 0; i < d_nchannels; i++)
+            {
+                out[i][0] = vgs[i];
+                vgs[i].interp_TOW_ms++;
+            }
+        counter++;
+        std::this_thread::sleep_for(std::chrono::microseconds(d_interval_us));
+        return 1;
+    }
+
+private:
+    friend gnss_synchro_emitter_sptr gnss_synchro_make_emitter(int n_channels, int nitems, int interval_us);
+
+    gnss_synchro_emitter(int n_channels, int nitems, int interval_us) : sync_block(
+                                                                            "gnss_synchro_emitter", gr::io_signature::make(0, 0, 0), gr::io_signature::make(n_channels, n_channels, sizeof(Gnss_Synchro))),
+                                                                        d_nitems(nitems),
+                                                                        d_nchannels(n_channels),
+                                                                        d_interval_us(interval_us)
+    {
+        for (int i = 0; i < d_nchannels; i++)
+            {
+                Gnss_Synchro gs{};
+                // unique key: Channel_ID, interp_TOW_ms
+                gs.Channel_ID = i;
+                gs.interp_TOW_ms = 0;
+                // PRN is filled with a known value to validate deserialization result
+                gs.PRN = i + 1;
+                vgs.push_back(gs);
+            }
+    }
+
+    const int d_nitems;
+    const int d_nchannels;
+    const int d_interval_us;
+    std::vector<Gnss_Synchro> vgs;
+    int counter{0};
+};
+
+gnss_synchro_emitter_sptr gnss_synchro_make_emitter(int n_channels, int nitems, int interval_us)
+{
+    return gnss_synchro_emitter_sptr(new gnss_synchro_emitter(n_channels, nitems, interval_us));
+}
+
+
+TEST(Monitor, decimation)
+{
+    // Minimum number of channels to confirm that decimation by 2 is not performed just by dropping all synchros from one channel
+    constexpr uint32_t NCHANNELS = 2;
+    constexpr uint32_t NSYNCHRO = 16;
+    constexpr uint32_t NDEC_MAX = 8;
+    // Limit the test execution time in case something goes wrong
+    constexpr auto RX_TIMEOUT = std::chrono::seconds(5);
+    constexpr uint32_t GNSS_SYNCHRO_INTERVAL_US = 1000;
+    b_io_context io_context;
+    // Bind to an ephemeral port, so the test does not collide with a running receiver
+    boost::asio::ip::udp::socket socket(io_context, boost::asio::ip::udp::endpoint(boost::asio::ip::address_v4::loopback(), 0));
+    const std::string udp_port = std::to_string(socket.local_endpoint().port());
+    std::vector<char> recv_buf(8192);
+    boost::asio::ip::udp::endpoint remote_endpoint;
+    for (uint32_t ndec = 1; ndec <= NDEC_MAX; ndec <<= 1)
+        {
+            std::cout << "Testing decimation factor = " << ndec << "\n";
+            const uint32_t nexpected = NSYNCHRO / ndec;
+            std::set<uint32_t> expected{};
+            uint32_t counter{0};
+            for (uint32_t i = 0; i < NSYNCHRO; i++)
+                {
+                    counter++;
+                    if (counter >= ndec)
+                        {
+                            expected.insert(i);
+                            counter = 0;
+                        }
+                }
+            std::vector<std::set<uint32_t>> received(NCHANNELS);
+            gnss_sdr::Observables obs{};
+            Serdes_Gnss_Synchro serdes = Serdes_Gnss_Synchro();
+            auto top_block = gr::make_top_block("monitor_test");
+            auto emitter = gnss_synchro_make_emitter(NCHANNELS, NSYNCHRO, GNSS_SYNCHRO_INTERVAL_US);
+            auto monitor = gnss_synchro_make_monitor(NCHANNELS, ndec, {udp_port}, {"127.0.0.1"}, true);
+            for (uint32_t i = 0; i < NCHANNELS; i++)
+                {
+                    top_block->connect(emitter, i, monitor, i);
+                }
+            top_block->start();
+            const auto deadline = std::chrono::steady_clock::now() + RX_TIMEOUT;
+            while (std::chrono::steady_clock::now() < deadline)
+                {
+                    std::this_thread::sleep_for(std::chrono::microseconds(GNSS_SYNCHRO_INTERVAL_US));
+                    if (socket.available() == 0)
+                        {
+                            continue;
+                        }
+                    const std::size_t nbytes = socket.receive_from(boost::asio::buffer(recv_buf), remote_endpoint);
+                    EXPECT_GT(nbytes, 0U);
+                    EXPECT_TRUE(obs.ParseFromArray(recv_buf.data(), static_cast<int>(nbytes)));
+                    const std::vector<Gnss_Synchro> vgs_read = serdes.readProtobuffer(obs);
+                    // validate deserialization result and add its unique index to the set
+                    for (const auto& gs : vgs_read)
+                        {
+                            EXPECT_EQ(gs.PRN, gs.Channel_ID + 1);
+                            received[gs.Channel_ID].insert(gs.interp_TOW_ms);
+                        }
+                    // Everything received, exit loop
+                    if (received[0].size() >= nexpected)
+                        {
+                            break;
+                        }
+                }
+            top_block->stop();
+            top_block->wait();
+            // Check the number of unique synchros per channel against the expected number
+            for (const auto& gs_set : received)
+                {
+                    EXPECT_EQ(gs_set.size(), nexpected);
+                    EXPECT_EQ(gs_set, expected);
+                }
+        }
+}