Feature/tracing poc

conanfile.py

@@ -17,6 +17,7 @@ def build_requirements(self):
     def requirements(self):
         self.requires("hdf5/1.10.6")
         self.requires("protobuf/3.17.1")
+        self.requires("nlohmann_json/3.11.2")
         self.requires("libcurl/7.78.0")
         self.requires("qt/5.15.2")
         self.requires("spdlog/1.9.2")
@@ -51,6 +52,6 @@ def generate(self):
         else:
             tc.variables["Protobuf_PROTOC_EXECUTABLE"] = os.path.join(self.deps_cpp_info["protobuf"].rootpath, "bin", "protoc")
         tc.generate()
-        
-        
-       
+
+
+

ecal/core/CMakeLists.txt

@@ -43,6 +43,7 @@ find_package(asio             REQUIRED)
 find_package(Threads          REQUIRED)
 find_package(ecaludp          REQUIRED)
 find_package(protozero        REQUIRED)
+find_package(nlohmann_json    REQUIRED)
 
 if (ECAL_CORE_CONFIGURATION)
   find_package(yaml-cpp       REQUIRED)
@@ -459,6 +460,14 @@ if(ECAL_CORE_TIMEPLUGIN)
   )
 endif()
 
+######################################
+# tracing
+######################################
+set(ecal_tracing_src
+    src/tracing/tracing.cpp
+    src/tracing/tracing.h
+)
+
 ######################################
 # util
 ######################################
@@ -635,6 +644,7 @@ set(ecal_sources
     ${ecal_serialization_src}
     ${ecal_service_src}
     ${ecal_time_src}
+    ${ecal_tracing_src}
     ${ecal_util_src}
     ${ecal_cmn_src}
     ${ecal_builder_src}
@@ -717,6 +727,7 @@ target_link_libraries(ecal_core_private
     $<$<BOOL:${ECAL_CORE_TRANSPORT_TCP}>:tcp_pubsub::tcp_pubsub>
     $<$<BOOL:${ECAL_CORE_TIMEPLUGIN}>:ecaltime>
     ecaludp::ecaludp
+    nlohmann_json::nlohmann_json
     $<$<AND:$<BOOL:${UNIX}>,$<NOT:$<BOOL:${QNXNTO}>>>:dl>
     $<$<AND:$<BOOL:${UNIX}>,$<NOT:$<BOOL:${APPLE}>>,$<NOT:$<BOOL:${QNXNTO}>>>:rt>
     $<$<STREQUAL:${CMAKE_SYSTEM_NAME},FreeBSD>:util>

ecal/core/src/io/shm/ecal_memfile_sync.cpp

@@ -28,6 +28,7 @@
 #include "ecal_memfile_header.h"
 #include "ecal_memfile_naming.h"
 #include "ecal_memfile_sync.h"
+#include "tracing/tracing.h"
 
 #include <chrono>
 #include <mutex>
@@ -161,6 +162,9 @@ namespace eCAL
     m_attr.timeout_ack_ms = data_.acknowledge_timeout_ms;
     if (m_attr.timeout_ack_ms < 0) m_attr.timeout_ack_ms = 0;
 
+    // store clock for tracing (does not modify SHM protocol)
+    m_current_clock_for_tracing = data_.clock;
+
     // write header and payload into the memory file
 #ifndef NDEBUG
     Logging::Log(Logging::log_level_debug4, m_base_name + "::CSyncMemoryFile::Write");
@@ -250,6 +254,12 @@ namespace eCAL
     return m_attr.min_size;
   }
 
+  void CSyncMemoryFile::SetTracingInfo(uint64_t entity_id_, int32_t process_id_)
+  {
+    m_entity_id_for_tracing = entity_id_;
+    m_process_id_for_tracing = process_id_;
+  }
+
   bool CSyncMemoryFile::Create(const std::string& base_name_, size_t size_)
   {
     if (m_created) return false;
@@ -346,6 +356,13 @@ namespace eCAL
   {
     if (!m_created) return;
 
+    // Trace the SHM handshake operation (does not modify SHM protocol)
+    eCAL::tracing::CShmHandshakeSpan handshake_span(
+      m_entity_id_for_tracing,
+      m_process_id_for_tracing,
+      m_current_clock_for_tracing
+    );
+
     // fire the publisher events
     // connected subscribers will read the content from the memory file
 

ecal/core/src/io/shm/ecal_memfile_sync.h

@@ -58,6 +58,8 @@ namespace eCAL
     bool CheckSize(size_t size_);
     bool Write(CPayloadWriter& payload_, const SWriterAttr& data_, bool force_full_write_ = false);
 
+    void SetTracingInfo(uint64_t entity_id_, int32_t process_id_);
+
     std::string GetName() const;
     size_t GetSize() const;
     bool IsCreated() const { return m_created; };
@@ -76,6 +78,11 @@ namespace eCAL
     CMemoryFile         m_memfile;
     bool                m_created;
 
+    // Tracing context (not part of core SHM protocol)
+    uint64_t            m_entity_id_for_tracing{0};
+    int32_t             m_process_id_for_tracing{0};
+    long long           m_current_clock_for_tracing{0};
+
     struct SEventHandlePair
     {
       EventHandleT event_snd;

ecal/core/src/pubsub/ecal_publisher.cpp

@@ -30,6 +30,7 @@
 
 #include "config/builder/writer_attribute_builder.h"
 #include "ecal_config_internal.h"
+#include "tracing/tracing.h"
 
 #include <iostream>
 #include <memory>
@@ -99,9 +100,21 @@ namespace eCAL
   }
 
   bool CPublisher::Send(CPayloadWriter& payload_, long long time_)
-  {
+  {    
     auto publisher_impl = m_publisher_impl.lock();
     if (!publisher_impl) return false;
+
+    Registration::Sample sample;
+    publisher_impl->GetRegistration(sample);
+
+    eCAL::tracing::CSendSpan send_span(
+      publisher_impl->GetTopicId(), 
+      sample.topic.data_clock, 
+      eCAL::eTLayerType::tl_all, 
+      payload_.GetSize(),
+      eCAL::tracing::operation_type::send
+    );
+
     // in an optimization case the
      // publisher can send an empty package
      // or we do not have any subscription at all

ecal/core/src/pubsub/ecal_publisher_impl.cpp

@@ -45,6 +45,7 @@
 
 #include "registration/ecal_registration_provider.h"
 #include "logging/ecal_log_provider.h"
+#include "tracing/tracing.h"
 
 #include <algorithm>
 #include <chrono>
@@ -124,6 +125,19 @@ namespace eCAL
     m_topic_id.topic_id.host_name = m_attributes.host_name;
     m_topic_id.topic_id.process_id = m_attributes.process_id;
 
+    // record topic metadata for tracing
+    {
+      eCAL::tracing::STopicMetadata meta;
+      meta.entity_id  = m_publisher_id;
+      meta.process_id = m_attributes.process_id;
+      meta.host_name  = m_attributes.host_name;
+      meta.topic_name = m_attributes.topic_name;
+      meta.encoding   = m_topic_info.encoding;
+      meta.type_name  = m_topic_info.name;
+      meta.direction  = eCAL::tracing::topic_direction::publisher;
+      eCAL::tracing::CTraceProvider::getInstance().addTopicMetadata(meta);
+    }
+
     // mark as created
     m_created = true;
   }
@@ -758,6 +772,9 @@ namespace eCAL
     // create writer
     m_writer_shm = std::make_unique<CDataWriterSHM>(eCAL::eCALWriter::BuildSHMAttributes(m_attributes), g_memfile_map());
 
+    // set tracing info for handshake tracing (does not modify SHM protocol)
+    m_writer_shm->SetTracingInfo(m_publisher_id, m_attributes.process_id);
+
     // register activated layer
     Register();
 

ecal/core/src/pubsub/ecal_publisher_impl.h

@@ -119,6 +119,7 @@ namespace eCAL
 
     void GetRegistrationSample(Registration::Sample& sample);
     void GetUnregistrationSample(Registration::Sample& sample);
+    long long GetClock() const { return m_clock; }
 
     bool StartUdpLayer();
     bool StartShmLayer();

ecal/core/src/pubsub/ecal_subscriber_impl.cpp

@@ -23,6 +23,7 @@
 **/
 
 #include "ecal_subscriber_impl.h"
+#include "tracing/tracing.h"
 #include <ecal/config.h>
 #include <ecal/log.h>
 #include <ecal/process.h>
@@ -95,6 +96,19 @@ namespace eCAL
     m_topic_id.topic_id.host_name = m_attributes.host_name;
     m_topic_id.topic_id.process_id = m_attributes.process_id;
 
+    // record topic metadata for tracing
+    {
+      eCAL::tracing::STopicMetadata meta;
+      meta.entity_id  = m_subscriber_id;
+      meta.process_id = m_attributes.process_id;
+      meta.host_name  = m_attributes.host_name;
+      meta.topic_name = m_attributes.topic_name;
+      meta.encoding   = m_topic_info.encoding;
+      meta.type_name  = m_topic_info.name;
+      meta.direction  = eCAL::tracing::topic_direction::subscriber;
+      eCAL::tracing::CTraceProvider::getInstance().addTopicMetadata(meta);
+    }
+
     // start transport layers
     InitializeLayers();
     StartTransportLayer();
@@ -375,6 +389,15 @@ namespace eCAL
 
   size_t CSubscriberImpl::ApplySample(const Payload::TopicInfo& topic_info_, const char* payload_, size_t size_, long long id_, long long clock_, long long time_, size_t /*hash_*/, eTLayerType layer_)
   {
+
+    eCAL::tracing::CReceiveSpan receive_span(
+      m_subscriber_id,
+      topic_info_,
+      clock_,
+      layer_,
+      eCAL::tracing::operation_type::receive
+    );
+
     // ensure thread safety
     const std::lock_guard<std::mutex> lock(m_receive_callback_mutex);
     if (!m_created) return(0);
@@ -452,7 +475,17 @@ namespace eCAL
 
         // execute it
         const std::lock_guard<std::mutex> exec_lock(m_connection_map_mtx);
-        (m_receive_callback)(topic_id, m_connection_map[pub_info].data_type_info, cb_data);
+        {
+          eCAL::tracing::CReceiveSpan receive_span(
+            m_subscriber_id,
+            topic_info_,
+            clock_,
+            layer_,
+            eCAL::tracing::operation_type::callback_execution
+          );
+
+
+        (m_receive_callback)(topic_id, m_connection_map[pub_info].data_type_info, cb_data); }
         processed = true;
       }
     }

ecal/core/src/readwrite/shm/ecal_writer_shm.cpp

@@ -151,6 +151,14 @@ namespace eCAL
     }
   }
 
+  void CDataWriterSHM::SetTracingInfo(uint64_t entity_id_, int32_t process_id_)
+  {
+    for (auto& memory_file : m_memory_file_vec)
+    {
+      memory_file->SetTracingInfo(entity_id_, process_id_);
+    }
+  }
+
   Registration::LayerParShm CDataWriterSHM::GetConnectionParameter()
   {
     Registration::LayerParShm layer_par_shm;
@@ -207,6 +215,16 @@ namespace eCAL
         return false;
       }
     }
+    // Propagate tracing info to newly created memory files if it has been set
+    if (m_tracing_entity_id != 0)
+    {
+      SetTracingInfo(m_tracing_entity_id, m_tracing_process_id);
+    }
+    // Propagate tracing info to newly created memory files if it has been set
+    if (m_tracing_entity_id != 0)
+    {
+      SetTracingInfo(m_tracing_entity_id, m_tracing_process_id);
+    }
 
     return true;
   }

ecal/core/src/readwrite/shm/ecal_writer_shm.h

@@ -53,6 +53,8 @@ namespace eCAL
     void ApplySubscription(const std::string& host_name_, int32_t process_id_, const EntityIdT& topic_id_, const std::string& conn_par_) override;
     void RemoveSubscription(const std::string& host_name_, int32_t process_id_, const EntityIdT& topic_id_) override;
 
+    void SetTracingInfo(uint64_t entity_id_, int32_t process_id_);
+
     Registration::LayerParShm GetConnectionParameter() override;
 
   protected:
@@ -69,5 +71,9 @@ namespace eCAL
     ProcessIDTopicIDSetT                          m_process_id_topic_id_set_map;
 
     std::shared_ptr<CMemFileMap>                  m_memfile_map;
+
+    // Store tracing info to propagate to memory files
+    uint64_t                                      m_tracing_entity_id{0};
+    int32_t                                       m_tracing_process_id{0};
   };
 }