Introduce zombie list for stopping blocks

This allows to remove blocks without having to stop the scheduler.
Running blocks can't transition directly to stopped state, so we
move them into a "zombie list". There's a maintenance period where
we delete blocks that have stopped.

Furthermore, deleted blocks are now removed from each thread's
local block list.

Signed-off-by: Sergio Martins <[email protected]>

Author: iamsergio

core/include/gnuradio-4.0/Block.hpp

@@ -498,6 +498,7 @@ class Block : public lifecycle::StateMachine<Derived> {
 
     ~Block() { // NOSONAR -- need to request the (potentially) running ioThread to stop
         if (lifecycle::isActive(this->state())) {
+            // Only happens in artificial cases likes qa_Block test. In practice blocks stay in zombie list if active
             emitErrorMessageIfAny("~Block()", this->changeStateTo(lifecycle::State::REQUESTED_STOP));
         }
         if constexpr (blockingIO) {

core/include/gnuradio-4.0/Graph.hpp

@@ -319,6 +319,9 @@ class Graph : public gr::Block<Graph> {
 
     constexpr static block::Category blockCategory = block::Category::TransparentBlockGroup;
 
+    std::mutex                               _zombieBlocksMutex;
+    std::vector<std::unique_ptr<BlockModel>> _zombieBlocks;
+
     Graph(property_map settings = {}) : gr::Block<Graph>(std::move(settings)) {
         _blocks.reserve(100); // TODO: remove
 
@@ -504,7 +507,7 @@ class Graph : public gr::Block<Graph> {
             return std::addressof(edge.sourceBlock()) == it->get() || std::addressof(edge.destinationBlock()) == it->get();
         });
 
-        _blocks.erase(it);
+        makeZombie(it);
     }
 
     gr::BlockModel* replaceBlock(const std::string& uniqueName, const std::string& type, const property_map& properties) {
@@ -531,11 +534,25 @@ class Graph : public gr::Block<Graph> {
                 edge._destinationBlock = newBlockRaw;
             }
         }
-        _blocks.erase(it);
+
+        makeZombie(it);
 
         return newBlockRaw;
     }
 
+    void makeZombie(auto it) {
+        if ((*it)->state() == lifecycle::State::PAUSED || (*it)->state() == lifecycle::State::RUNNING) {
+            emitErrorMessageIfAny("makeZombie", (*it)->changeStateTo(lifecycle::State::REQUESTED_STOP));
+        }
+
+        {
+            std::lock_guard guard(_zombieBlocksMutex);
+            _zombieBlocks.push_back(std::move(*it));
+        }
+
+        _blocks.erase(it);
+    }
+
     void emplaceEdge(std::string_view sourceBlock, std::string sourcePort, std::string_view destinationBlock, //
         std::string destinationPort, [[maybe_unused]] const std::size_t minBufferSize, [[maybe_unused]] const std::int32_t weight, std::string_view edgeName) {
         auto sourceBlockIt = std::ranges::find_if(_blocks, [&sourceBlock](const auto& block) { return block->uniqueName() == sourceBlock; });

core/include/gnuradio-4.0/Scheduler.hpp

@@ -356,6 +356,11 @@ class SchedulerBase : public Block<Derived> {
                 if (runnerID == 0UZ || _nRunningJobs.load(std::memory_order_acquire) == 0UZ) {
                     this->processScheduledMessages(); // execute the scheduler- and Graph-specific message handler only once globally
                 }
+
+                // Zombies are cleaned per-thread, as we remove from the localBlockList as well.
+                // Cleaning zombies has low priority, so uses process_stream_to_message_ratio (a different ratio could be introduced)
+                cleanupZombieBlocks(localBlockList);
+
                 std::ranges::for_each(localBlockList, [](auto& block) { block->processScheduledMessages(); });
                 activeState = this->state();
                 msgToCount++;
@@ -376,10 +381,6 @@ class SchedulerBase : public Block<Derived> {
                     break;
                 }
             } else if (activeState == lifecycle::State::PAUSED) {
-                if (_graph.hasTopologyChanged()) {
-                    // TODO: update localBlockList topology if needed
-                    _graph.ackTopologyChange();
-                }
                 std::this_thread::sleep_for(std::chrono::milliseconds(timeout_ms));
                 msgToCount = 0UZ;
             } else { // other states
@@ -416,6 +417,7 @@ class SchedulerBase : public Block<Derived> {
                 this->emitErrorMessageIfAny("forEachBlock -> stop() -> LifecycleState", block->changeStateTo(lifecycle::State::STOPPED));
             }
         });
+
         this->emitErrorMessageIfAny("stop() -> LifecycleState ->STOPPED", this->changeStateTo(lifecycle::State::STOPPED));
         this->emitErrorMessageIfAny("stop() -> LifecycleState ->IDLE", this->changeStateTo(lifecycle::State::IDLE));
     }
@@ -502,6 +504,93 @@ class SchedulerBase : public Block<Derived> {
         return message;
     }
 
+    /*
+      Zombie Tutorial:
+
+      Blocks can't be deleted unless stopped, but since stopping can take time (async) we move such blocks
+      to the "zombie list" and disconnect them immediately from the graph. This allows them to stop and be deleted
+      safely.
+
+      Periodically, we call cleanupZombieBlocks(), which iterates the zombie list and deletes the blocks that are now stopped.
+
+      cleanupZombieBlocks() is called *per-thread*, since we also need to update the localBlockList, i.e.: removing dangling block pointers
+      from the localBlockList.
+
+      We also update the _jobLists member variable, but probably that member can be removed, seems unneeded and only used so unit-tests can
+      query it.
+     */
+    void cleanupZombieBlocks(std::vector<BlockModel*>& localBlockList) {
+        if (localBlockList.empty()) {
+            return;
+        }
+
+        std::lock_guard guard(_graph._zombieBlocksMutex);
+
+        auto& zombieBlocks = _graph._zombieBlocks;
+        auto  it           = zombieBlocks.begin();
+
+        while (it != zombieBlocks.end()) {
+            auto localBlockIt = std::find(localBlockList.begin(), localBlockList.end(), it->get());
+            if (localBlockIt == localBlockList.end()) {
+                // we only care about the blocks local to our thread.
+                ++it;
+                continue;
+            }
+
+            bool shouldDelete = false;
+
+            switch ((*it)->state()) {
+            case lifecycle::State::IDLE:
+            case lifecycle::State::STOPPED:
+            case lifecycle::State::INITIALISED:
+                // This block can be deleted immediately
+                shouldDelete = true;
+                break;
+            case lifecycle::State::ERROR:
+                // Delete as well. (Separate case block, as better ideas welcome)
+                shouldDelete = true;
+                break;
+            case lifecycle::State::REQUESTED_STOP:
+                // This block will be deleted later
+                break;
+            case lifecycle::State::REQUESTED_PAUSE:
+                // This block will be deleted later
+                // There's no transition from REQUESTED_PAUSE to REQUESTED_STOP
+                // Will be moved to REQUESTED_STOP as soon as it's possible
+                break;
+            case lifecycle::State::PAUSED:
+                // This zombie was in REQUESTED_PAUSE and now finally in PAUSED. Can be stopped now.
+                // Will be deleted in a next zombie maintenance period
+                this->emitErrorMessageIfAny("cleanupZombieBlocks", (*it)->changeStateTo(lifecycle::State::REQUESTED_STOP));
+                break;
+            case lifecycle::State::RUNNING: assert(false && "Doesn't happen: zombie blocks are never running"); break;
+            }
+
+            if (shouldDelete) {
+                localBlockList.erase(localBlockIt);
+
+                BlockModel* zombieRaw = it->get();
+                it                    = zombieBlocks.erase(it); // ~Block() runs here
+
+                // We need to remove zombieRaw from jobLists as well, in case Scheduler ever goes to INITIALIZED
+                // again.
+                // TODO: I'd argue we should remove _jobLists to minimize having to maintain state. Instead, a job list can be
+                // calculated in start().
+                std::lock_guard lock(_jobListsMutex);
+                for (auto& jobList : *this->_jobLists) {
+                    auto job_it = std::remove(jobList.begin(), jobList.end(), zombieRaw);
+                    if (job_it != jobList.end()) {
+                        jobList.erase(job_it, jobList.end());
+                        break;
+                    }
+                }
+
+            } else {
+                ++it;
+            }
+        }
+    }
+
     std::optional<Message> propertyCallbackReplaceBlock([[maybe_unused]] std::string_view propertyName, Message message) {
         assert(propertyName == scheduler::property::kReplaceBlock);
         using namespace std::string_literals;