WIP

include/cactus_rt/app.h

@@ -42,7 +42,7 @@ class App {
   //
   // TODO: investigate into a weak pointer.
   std::list<std::shared_ptr<tracing::ThreadTracer>> thread_tracers_;
-  std::unique_ptr<tracing::TraceAggregator>         trace_aggregator_ = nullptr;
+  std::shared_ptr<tracing::TraceAggregator>         trace_aggregator_ = nullptr;
   std::mutex                                        aggregator_mutex_;
 
   void SetDefaultLogFormat(quill::Config& cfg) {
@@ -148,16 +148,6 @@ class App {
   void StartQuill();
 
  private:
-  /**
-   * @brief Register a thread tracer. Should only be called from Thread::RunThread.
-   */
-  void RegisterThreadTracer(std::shared_ptr<tracing::ThreadTracer> thread_tracer) noexcept;
-
-  /**
-   * @brief Remove a thread tracer. Should only be called from Thread::~Thread().
-   */
-  void DeregisterThreadTracer(const std::shared_ptr<tracing::ThreadTracer>& thread_tracer) noexcept;
-
   void CreateAndStartTraceAggregator(std::shared_ptr<tracing::Sink> sink) noexcept;
 
   void StopTraceAggregator() noexcept;

include/cactus_rt/thread.h

@@ -12,6 +12,7 @@
 #include "config.h"
 #include "quill/Quill.h"
 #include "tracing/thread_tracer.h"
+#include "tracing/trace_aggregator.h"
 
 namespace cactus_rt {
 
@@ -43,18 +44,15 @@ class Thread {
 
   friend class App;
 
-  // Non-owning App pointer. Used only for notifying that the thread has
-  // started/stopped for tracing purposes. Set by Thread::Start and read at
-  // the beginning of Thread::RunThread.
-  App* app_ = nullptr;
+  std::weak_ptr<tracing::TraceAggregator> trace_aggregator_;
 
   /**
    * Starts the thread in the background.
    *
    * @param start_monotonic_time_ns should be the start time in nanoseconds for the monotonic clock.
    * @param app The application that started this thread.
    */
-  void Start(int64_t start_monotonic_time_ns, App* app);
+  void Start(int64_t start_monotonic_time_ns, std::weak_ptr<tracing::TraceAggregator> trace_aggregator);
 
  public:
   /**

src/cactus_rt/app.cc

@@ -43,7 +43,7 @@ void App::Start() {
 
   auto start_monotonic_time_ns = NowNs();
   for (auto& thread : threads_) {
-    thread->Start(start_monotonic_time_ns, this);
+    thread->Start(start_monotonic_time_ns, trace_aggregator_);
   }
 }
 
@@ -100,28 +100,6 @@ bool App::StopTraceSession() noexcept {
   return true;
 }
 
-void App::RegisterThreadTracer(std::shared_ptr<tracing::ThreadTracer> thread_tracer) noexcept {
-  const std::scoped_lock lock(aggregator_mutex_);
-
-  thread_tracers_.push_back(thread_tracer);
-
-  if (trace_aggregator_ != nullptr) {
-    trace_aggregator_->RegisterThreadTracer(thread_tracer);
-  }
-}
-
-void App::DeregisterThreadTracer(const std::shared_ptr<tracing::ThreadTracer>& thread_tracer) noexcept {
-  const std::scoped_lock lock(aggregator_mutex_);
-
-  thread_tracers_.remove_if([thread_tracer](const std::shared_ptr<tracing::ThreadTracer>& t) {
-    return t == thread_tracer;
-  });
-
-  if (trace_aggregator_ != nullptr) {
-    trace_aggregator_->DeregisterThreadTracer(thread_tracer);
-  }
-}
-
 void App::LockMemory() const {
   // See https://lwn.net/Articles/837019/
 
@@ -196,7 +174,7 @@ void App::CreateAndStartTraceAggregator(std::shared_ptr<tracing::Sink> sink) noe
     return;
   }
 
-  trace_aggregator_ = std::make_unique<tracing::TraceAggregator>(name_, tracer_config_.trace_aggregator_cpu_affinity);
+  trace_aggregator_ = std::make_shared<tracing::TraceAggregator>(name_, tracer_config_.trace_aggregator_cpu_affinity);
   for (auto tracer : thread_tracers_) {
     trace_aggregator_->RegisterThreadTracer(tracer);
   }

src/cactus_rt/thread.cc

@@ -5,10 +5,11 @@
 #include <cerrno>
 #include <cstring>
 #include <ctime>
+#include <memory>
 #include <stdexcept>
 
-#include "cactus_rt/app.h"
 #include "cactus_rt/config.h"
+#include "cactus_rt/tracing/thread_tracer.h"
 
 namespace cactus_rt {
 
@@ -17,10 +18,11 @@ void* Thread::RunThread(void* data) {
   thread->config_.scheduler->SetSchedAttr();
 
   thread->tracer_->SetTid();
-  if (thread->app_ != nullptr) {
-    thread->app_->RegisterThreadTracer(thread->tracer_);
+  auto trace_aggregator = thread->trace_aggregator_.lock();
+  if (trace_aggregator) {
+    trace_aggregator->RegisterThreadTracer(thread->tracer_);
   } else {
-    LOG_WARNING(thread->Logger(), "thread {} does not have trace_aggregator_ and tracing is disabled. Did you all App::RegisterThread?", thread->name_);
+    LOG_WARNING(thread->Logger(), "thread {} does not have trace_aggregator_ and tracing is enabled. Did you all App::RegisterThread?", thread->name_);
   }
   quill::preallocate();  // Pre-allocates thread-local data to avoid the need to allocate on the first log message
 
@@ -30,9 +32,9 @@ void* Thread::RunThread(void* data) {
   return nullptr;
 }
 
-void Thread::Start(int64_t start_monotonic_time_ns, App* app) {
+void Thread::Start(int64_t start_monotonic_time_ns, std::weak_ptr<tracing::TraceAggregator> trace_aggregator) {
   start_monotonic_time_ns_ = start_monotonic_time_ns;
-  app_ = app;
+  trace_aggregator_ = std::move(trace_aggregator);
 
   pthread_attr_t attr;
 
@@ -80,8 +82,9 @@ int Thread::Join() const {
 }
 
 Thread::~Thread() {
-  if (app_ != nullptr) {
-    app_->DeregisterThreadTracer(tracer_);
+  auto trace_aggregator = trace_aggregator_.lock();
+  if (trace_aggregator) {
+    trace_aggregator->DeregisterThreadTracer(tracer_);
   }
 }
 

src/cactus_rt/tracing/trace_aggregator.cc

@@ -87,7 +87,7 @@ void TraceAggregator::DeregisterThreadTracer(const std::shared_ptr<ThreadTracer>
   const std::scoped_lock lock(mutex_);
 
   tracers_.remove_if([tracer](const std::shared_ptr<tracing::ThreadTracer>& t) {
-    return t == tracer;
+    return t.get() == tracer.get();
   });
 }
 

tests/tracing/helpers/assert_helpers.cc

@@ -3,8 +3,6 @@
 #include <google/protobuf/util/json_util.h>
 #include <gtest/gtest.h>
 
-#include <utility>
-
 std::string ProtoToJson(const google::protobuf::Message& proto) {
   std::string json;
   google::protobuf::util::MessageToJsonString(proto, &json);

tests/tracing/single_threaded_test.cc

@@ -1,4 +1,5 @@
 #include <cactus_rt/rt.h>
+#include <google/protobuf/text_format.h>
 #include <gtest/gtest.h>
 #include <quill/detail/LogManager.h>
 
@@ -266,6 +267,7 @@ TEST_F(SingleThreadTracingTest, StopTracingAndNoEventsAreRecorded) {
 
   auto traces = sink_->LoggedTraces();
   auto packets = GetPacketsFromTraces(traces);
+
   ASSERT_EQ(packets.size(), 1);
 
   AssertIsProcessTrackDescriptor(*packets[0], kAppName);
@@ -310,8 +312,6 @@ TEST_F(SingleThreadTracingTest, RestartTracingStartsNewSession) {
   AssertIsThreadTrackDescriptor(*packets2[1], kRegularThreadName, process_track_uuid);
   auto thread_track_uuid = packets2[1]->track_descriptor().uuid();
 
-  std::cout << "packets2: " << packets2[2]->ShortDebugString() << "\n";
-
   // Event1 is emitted as interned data because that thread is still active and the event name got interned previously.
   auto event_names = GetInternedEventNames(*packets2[2]);
   ASSERT_EQ(event_names.size(), 1);