Update CMakeLists for newer subdirectory names

CMakeLists.txt

@@ -22,6 +22,7 @@ if (CMAKE_CXX_COMPILER_ID MATCHES "GNU|Clang")
         -Wpedantic
         -Wconversion
         -Wshadow
+        -fno-rtti
     )
 endif()
 
@@ -35,11 +36,12 @@ set_property(GLOBAL PROPERTY USE_FOLDERS ON)
 # Add Chapters
 # ------------------------------------------------------------
 
+add_subdirectory(SetupChapter)
 add_subdirectory(Common)
-add_subdirectory(Chapter4-TypeErasure)
-add_subdirectory(Chapter8-SensorPipelines)
-add_subdirectory(Chapter9-SystemController)
-add_subdirectory(Chapter10-Reactor)
-add_subdirectory(Chapter11-FusionTiming)
-add_subdirectory(Chapter12-DeterministicFusionAndOutput)
+add_subdirectory(Chapter3-TypeErasure)
+add_subdirectory(Chapter7-SensorPipelines)
+add_subdirectory(Chapter8-SystemCompositionAndControl)
+add_subdirectory(Chapter9-IntegratingExternalInputs)
+add_subdirectory(Chapter10-FusionTiming)
+add_subdirectory(Chapter11-DeterministicFusionAndOutput)
 add_subdirectory(ExternalDataSource)

Chapter7-SensorPipelines/CMakeLists.txt

@@ -10,6 +10,8 @@ add_executable(SensorPipelineDemo
     include/PositionProducer.h
     include/PositionConsumer.h
     include/SensorPipeline.h
+    include/BurstySensorPipelineIntervals.h
+    include/DefaultSensorPipelineIntervals.h
     src/SensorPipeline.cpp
     src/PositionConsumer.cpp
     src/PositionProducer.cpp

Chapter7-SensorPipelines/README.md

@@ -1 +1 @@
-This repository accompanies Chapter 8 of _Designing Deterministic Systems with Modern C++_.
+This repository accompanies Chapter 7 of _Designing Deterministic Systems with Modern C++_.

Chapter7-SensorPipelines/include/BurstySensorPipelineIntervals.h

@@ -0,0 +1,32 @@
+#pragma once
+
+#include <chrono>
+
+#include "SensorPipeline.h"
+
+// A test scaffold for the producer loop that allows for bursts of measurements,
+// the idea is to _force_, from time to time, the producer to block on emplacing
+// a measurement to a full queue
+static
+SensorPipelineIntervals& getBurstyIntervals()
+{
+    static constexpr long burstInterval = 1000;
+    static constexpr long burstOf = 300;
+    static long count {0};
+
+    struct Bursty : public SensorPipelineIntervals
+    {
+        std::chrono::nanoseconds pollingInterval() const
+        {
+            return std::chrono::microseconds(count++ % burstInterval <= burstOf);
+        }
+
+        std::chrono::nanoseconds blockedWaitingInterval() const
+        {
+            return std::chrono::microseconds(50);
+        }
+    };
+
+    static Bursty bursty {};
+    return bursty;
+}

Chapter7-SensorPipelines/include/DefaultSensorPipelineIntervals.h

@@ -0,0 +1,19 @@
+#pragma once
+
+#include <chrono>
+
+#include "SensorPipeline.h"
+
+
+static
+SensorPipelineIntervals& getDefaultIntervals()
+{
+    struct Defaults : public SensorPipelineIntervals
+    {
+        std::chrono::nanoseconds pollingInterval() const { return std::chrono::microseconds(1000); }
+        std::chrono::nanoseconds blockedWaitingInterval() const { return std::chrono::microseconds(50); }
+    };
+
+    static Defaults defaults {};
+    return defaults;
+}

Chapter7-SensorPipelines/include/SensorPipeline.h

@@ -1,29 +1,39 @@
 #pragma once
 
 #include <atomic>
+#include <chrono>
 #include <cstddef>
 #include <thread>
 
 #include "readerwriterqueue.h"
+#include "Counters.h"
 
 #include "AnyMeasurement.h"
 
+class SensorPipelineIntervals;
+
 //
 // A simple, fixed-topology pipeline:
 // Producer thread -> SPSC queue -> Consumer thread
 //
 class SensorPipeline
 {
 public:
-    explicit SensorPipeline(std::size_t capacity);
+    explicit SensorPipeline(std::size_t capacity); // creates this "holder" but doesn't start work
+    explicit SensorPipeline(std::size_t capacity,
+                            const SensorPipelineIntervals& producerIntervals,
+                            const SensorPipelineIntervals& consumerIntervals);
     ~SensorPipeline();
 
     SensorPipeline(const SensorPipeline&) = delete;
     SensorPipeline& operator=(const SensorPipeline&) = delete;
 
-    void start();
-    void stop();
-    void join();
+    void start(); // create pipeline (including its threads) and start flow
+    void stop();  // _request_ work to end (does not block)
+    void join();  // wait for completion of pipeline (blocks) and end threads (and resources)
+    // ...now SensorPipeline is safe to destruct
+
+    void status(std::ostream& os);
 
 private:
     void producerLoop();
@@ -34,4 +44,20 @@ class SensorPipeline
     std::thread m_producerThread;
     std::thread m_consumerThread;
     std::atomic<bool> m_running { false };
+
+    const SensorPipelineIntervals& m_producerIntervals;
+    const SensorPipelineIntervals& m_consumerIntervals;
+
+    Counter producer_enqueued;
+    Counter producer_blocked;
+    Counter consumer_blocked;
+    Counter consumer_dequeued;
+};
+
+struct SensorPipelineIntervals
+{
+    virtual std::chrono::nanoseconds pollingInterval() const = 0;
+    virtual std::chrono::nanoseconds blockedWaitingInterval() const = 0;
+
+    virtual ~SensorPipelineIntervals() = default;
 };

Chapter7-SensorPipelines/src/SensorPipeline.cpp

@@ -12,6 +12,7 @@
 #include "MeasurementTypes.h"   // weather::Position, MeasurementHeaderV1, SourceId
 #include "PositionProducer.h"
 #include "PositionConsumer.h"
+#include "DefaultSensorPipelineIntervals.h"
 
 namespace
 {
@@ -23,13 +24,24 @@ std::uint64_t nowNs() noexcept
         std::chrono::duration_cast<std::chrono::nanoseconds>(
             Clock::now().time_since_epoch()).count());
 }
+
 } // namespace
 
 SensorPipeline::SensorPipeline(std::size_t capacity)
-    : m_queue(capacity)
+  : SensorPipeline(capacity, getDefaultIntervals(), getDefaultIntervals())
+{
+}
+
+SensorPipeline::SensorPipeline(std::size_t capacity,
+                               const SensorPipelineIntervals& producerIntervals,
+                               const SensorPipelineIntervals& consumerIntervals)
+  : m_queue(capacity)
+  , m_producerIntervals(producerIntervals)
+  , m_consumerIntervals(consumerIntervals)
 {
 }
 
+
 SensorPipeline::~SensorPipeline()
 {
     stop();
@@ -64,10 +76,19 @@ void SensorPipeline::join()
     }
 }
 
+void SensorPipeline::status(std::ostream& os)
+{
+  os << "Producer enqueued: " << producer_enqueued << "\n"
+     << "         blocked:  " << producer_blocked  << "\n"
+     << "Consumer blocked:  " << consumer_blocked  << "\n"
+     << "         dequeued: " << consumer_dequeued << "\n";
+}
+
 void SensorPipeline::producerLoop()
 {
     PositionProducer producer;
 
+    // Loop getting measurements until done (stopped)
     while (m_running.load())
     {
         const weather::Position pos = producer.nextPosition();
@@ -81,12 +102,20 @@ void SensorPipeline::producerLoop()
 
         weather::AnyMeasurement msg(header, pos);
 
-        while (m_running.load() && !m_queue.try_enqueue(msg))
+        // busy loop until measurement successfully enqueued
+        while (m_running.load())
         {
-            std::this_thread::sleep_for(std::chrono::microseconds(50));
+          if (m_queue.try_enqueue(msg))
+          {
+              producer_enqueued++;
+              break;
+          } else {
+              producer_blocked++;
+              std::this_thread::sleep_for(m_producerIntervals.blockedWaitingInterval());
+          }
         }
 
-        std::this_thread::sleep_for(std::chrono::milliseconds(100));
+        std::this_thread::sleep_for(m_producerIntervals.pollingInterval());
     }
 }
 
@@ -107,9 +136,11 @@ void SensorPipeline::consumerLoop()
     {
         if (!m_queue.try_dequeue(msg))
         {
-            std::this_thread::sleep_for(std::chrono::microseconds(50));
+            consumer_blocked++;
+            std::this_thread::sleep_for(m_consumerIntervals.blockedWaitingInterval());
             continue;
         }
+        consumer_dequeued++;
 
         const weather::Position* pos = msg.try_get<weather::Position>();
         if (pos != nullptr)
@@ -122,13 +153,20 @@ void SensorPipeline::consumerLoop()
         }
     }
 
-    // Optional drain after stop (keeps output tidy)
-    while (m_queue.try_dequeue(msg))
+    // Drain after stop to ensure all measurements are used and
+    // pipeline is emptied)
+    while (true)
     {
-        const weather::Position* pos = msg.try_get<weather::Position>();
-        if (pos != nullptr)
+        if (m_queue.try_dequeue(msg))
         {
-            consumer.consume(*pos);
+            consumer_dequeued++;
+            const weather::Position* pos = msg.try_get<weather::Position>();
+            if (pos != nullptr)
+            {
+                consumer.consume(*pos);
+            }
+        } else {
+            break;
         }
     }
 }

Chapter7-SensorPipelines/src/main.cpp

@@ -1,13 +1,19 @@
+#include <chrono>
+#include <iostream>
 #include <thread>
 
 #include "SensorPipeline.h"
+#include "BurstySensorPipelineIntervals.h"
+#include "DefaultSensorPipelineIntervals.h"
 
 using namespace weather;
 
 int main()
 {
     // "Capacity" is an initial sizing parameter for the SPSC queue.
-    SensorPipeline pipeline(/*capacity*/ 128);
+    SensorPipeline pipeline(128 /* capacity */,
+                            getBurstyIntervals() /* producer intervals */,
+                            getDefaultIntervals() /* consumer intervals */);
 
     pipeline.start();
 
@@ -17,6 +23,7 @@ int main()
     pipeline.stop();
     pipeline.join();
 
+    pipeline.status(std::cerr);
+
     return 0;
 }
-

Chapter8-SytemCompositionAndControl/CMakeLists.txt → Chapter8-SystemCompositionAndControl/CMakeLists.txt

@@ -1,21 +1,20 @@
 cmake_minimum_required(VERSION 3.16)
-project(chapter9_system_demo LANGUAGES CXX)
+project(chapter8_system_demo LANGUAGES CXX)
 
 set(CMAKE_CXX_STANDARD 17)
 set(CMAKE_CXX_STANDARD_REQUIRED ON)
 
-add_executable(chapter9_demo
+add_executable(chapter8_demo
     src/main.cpp
     src/WeatherSystem.cpp
     src/WeatherSystemBuilder.cpp
-    src/Chapter9RunLoops.cpp
+    src/Chapter8RunLoops.cpp
     include/WeatherSystem.h
     include/WeatherSystemBuilder.h
-    include/Chapter9MeasurementFactory.h
-    include/Chapter9RunLoops.h
+    include/Chapter8RunLoops.h
 )
 
-target_include_directories(chapter9_demo PRIVATE
+target_include_directories(chapter8_demo PRIVATE
     include
     ../Common/include
     ../ThirdParty/readerwriterqueue

Chapter8-SytemCompositionAndControl/README.md → Chapter8-SystemCompositionAndControl/README.md

@@ -1,11 +1,11 @@
 // SPDX-License-Identifier: MIT
 // Copyright (c) 2025 Autumnal Software
 
-# Chapter 9 - System Controller (Snapshot)
+# Chapter 8 - System Composition And Control (Snapshot)
 
-This repository accompanies Chapter 9 of _Designing Deterministic Systems with Modern C++_.
+This repository accompanies Chapter 8 of _Designing Deterministic Systems with Modern C++_.
 
-This Chapter 9 demo is about system composition and control.
+This Chapter 8 demo is about system composition and control.
 
 - Uses `std::thread`
 - No Boost.Asio
@@ -18,7 +18,7 @@ This Chapter 9 demo is about system composition and control.
 This snapshot assumes the repository already provides:
 
 - `ThirdParty/readerwriterqueue`
-- `ThirdParty/boost_asio_1_36_0` (not used in Chapter 9, but present for later chapters)
+- `ThirdParty/boost_asio_1_36_0` (not used in Chapter 8, but present for later chapters)
 - `../Chapter7-ImprovedAnyMeasurement/include` containing:
   - `MeasurementTypes.h`
   - `AnyMeasurement.h`

Chapter8-SytemCompositionAndControl/include/Chapter9RunLoops.h → Chapter8-SystemCompositionAndControl/include/Chapter8RunLoops.h

@@ -9,7 +9,7 @@
 #include "LogEvent.h"
 #include "readerwriterqueue.h"
 
-struct Chapter9Stats
+struct Chapter8Stats
 {
     std::atomic<std::uint64_t> enqTempOk{0};
     std::atomic<std::uint64_t> enqPosOk{0};
@@ -27,10 +27,10 @@ struct Chapter9Stats
 
 
 
-class Chapter9RunLoops
+class Chapter8RunLoops
 {
 public:
-    Chapter9RunLoops(moodycamel::ReaderWriterQueue<weather::AnyMeasurement>& inQ,
+    Chapter8RunLoops(moodycamel::ReaderWriterQueue<weather::AnyMeasurement>& inQ,
                      moodycamel::ReaderWriterQueue<LogEvent>& logQ);
 
     void producer(const std::atomic<bool>& stop);
@@ -41,5 +41,5 @@ class Chapter9RunLoops
     moodycamel::ReaderWriterQueue<weather::AnyMeasurement>& m_inQ;
     moodycamel::ReaderWriterQueue<LogEvent>& m_logQ;
 
-    Chapter9Stats m_stats;
+    Chapter8Stats m_stats;
 };