[core] Fix check fail when task buffer periodical runner runs before RayEvent is initialized (#55249)

Signed-off-by: dayshah <[email protected]>

Author: dayshah

src/ray/core_worker/core_worker_process.cc

@@ -722,6 +722,18 @@ CoreWorkerProcessImpl::CoreWorkerProcessImpl(const CoreWorkerOptions &options)
   // We need init stats before using it/spawning threads.
   stats::Init(global_tags, options_.metrics_agent_port, worker_id_);
 
+  // Initialize event framework before starting up worker.
+  if (RayConfig::instance().event_log_reporter_enabled() && !options_.log_dir.empty()) {
+    const std::vector<SourceTypeVariant> source_types = {
+        ray::rpc::Event_SourceType::Event_SourceType_CORE_WORKER,
+        ray::rpc::ExportEvent_SourceType::ExportEvent_SourceType_EXPORT_TASK};
+    RayEventInit(source_types,
+                 /*custom_fields=*/{},
+                 options_.log_dir,
+                 RayConfig::instance().event_level(),
+                 RayConfig::instance().emit_event_to_log_file());
+  }
+
   {
     // Notify that core worker is initialized.
     absl::Cleanup initialzed_scope_guard = [this] {
@@ -732,18 +744,6 @@ CoreWorkerProcessImpl::CoreWorkerProcessImpl(const CoreWorkerOptions &options)
     auto write_locked = core_worker_.LockForWrite();
     write_locked.Get() = worker;
   }
-
-  // Initialize event framework.
-  if (RayConfig::instance().event_log_reporter_enabled() && !options_.log_dir.empty()) {
-    const std::vector<SourceTypeVariant> source_types = {
-        ray::rpc::Event_SourceType::Event_SourceType_CORE_WORKER,
-        ray::rpc::ExportEvent_SourceType::ExportEvent_SourceType_EXPORT_TASK};
-    RayEventInit(source_types,
-                 absl::flat_hash_map<std::string, std::string>(),
-                 options_.log_dir,
-                 RayConfig::instance().event_level(),
-                 RayConfig::instance().emit_event_to_log_file());
-  }
 }
 
 CoreWorkerProcessImpl::~CoreWorkerProcessImpl() {

src/ray/util/event.cc

@@ -23,8 +23,6 @@
 #include <string>
 #include <vector>
 
-#include "absl/base/call_once.h"
-#include "absl/time/time.h"
 #include "ray/util/random.h"
 #include "ray/util/string_utils.h"
 #include "ray/util/timestamp_utils.h"
@@ -52,7 +50,7 @@ LogEventReporter::LogEventReporter(SourceTypeVariant source_type,
   // generate file name, if the soucrce type is RAYLET or GCS, the file name would like
   // event_GCS.log, event_RAYLET.log other condition would like
   // event_CORE_WOREKER_{pid}.log
-  std::string source_type_name = "";
+  std::string source_type_name;
   bool add_pid_to_file = false;
   if (auto event_source_type_ptr = std::get_if<rpc::Event_SourceType>(&source_type)) {
     rpc::Event_SourceType event_source_type = *event_source_type_ptr;
@@ -215,14 +213,11 @@ void EventManager::Publish(const rpc::Event &event, const json &custom_fields) {
 
 void EventManager::PublishExportEvent(const rpc::ExportEvent &export_event) {
   auto element = export_log_reporter_map_.find(export_event.source_type());
-  if (element != export_log_reporter_map_.end()) {
-    (element->second)->ReportExportEvent(export_event);
-  } else {
-    RAY_LOG(FATAL)
-        << "RayEventInit wasn't called with the necessary source type "
-        << ExportEvent_SourceType_Name(export_event.source_type())
-        << ". This indicates a bug in the code, and the event will be dropped.";
-  }
+  RAY_CHECK(element != export_log_reporter_map_.end())
+      << "RayEventInit wasn't called with the necessary source type "
+      << ExportEvent_SourceType_Name(export_event.source_type())
+      << ". This indicates a bug in the code, and the event will be dropped.";
+  element->second->ReportExportEvent(export_event);
 }
 
 void EventManager::AddReporter(std::shared_ptr<BaseEventReporter> reporter) {
@@ -270,7 +265,7 @@ void RayEventContext::SetEventContext(
   SetSourceType(source_type);
   UpdateCustomFields(custom_fields);
 
-  if (!global_context_started_setting_.fetch_or(1)) {
+  if (global_context_started_setting_.fetch_or(1) == 0) {
     global_context_ = std::make_unique<RayEventContext>();
     global_context_->SetSourceType(source_type);
     global_context_->UpdateCustomFields(custom_fields);
@@ -471,15 +466,13 @@ void RayExportEvent::SendEvent() {
   EventManager::Instance().PublishExportEvent(export_event);
 }
 
-static absl::once_flag init_once_;
-
-void RayEventInit_(const std::vector<SourceTypeVariant> source_types,
+void RayEventInit_(const std::vector<SourceTypeVariant> &source_types,
                    const absl::flat_hash_map<std::string, std::string> &custom_fields,
                    const std::string &log_dir,
                    const std::string &event_level,
                    bool emit_event_to_log_file) {
   for (const auto &source_type : source_types) {
-    std::string source_type_name = "";
+    std::string source_type_name;
     auto event_dir = std::filesystem::path(log_dir) / std::filesystem::path("events");
     if (auto event_source_type_ptr = std::get_if<rpc::Event_SourceType>(&source_type)) {
       // Set custom fields for non export events
@@ -503,17 +496,16 @@ void RayEventInit_(const std::vector<SourceTypeVariant> source_types,
   SetEmitEventToLogFile(emit_event_to_log_file);
 }
 
-void RayEventInit(const std::vector<SourceTypeVariant> source_types,
+void RayEventInit(const std::vector<SourceTypeVariant> &source_types,
                   const absl::flat_hash_map<std::string, std::string> &custom_fields,
                   const std::string &log_dir,
                   const std::string &event_level,
                   bool emit_event_to_log_file) {
-  absl::call_once(
-      init_once_,
-      [&source_types, &custom_fields, &log_dir, &event_level, emit_event_to_log_file]() {
-        RayEventInit_(
-            source_types, custom_fields, log_dir, event_level, emit_event_to_log_file);
-      });
+  static std::once_flag init_once_;
+  std::call_once(init_once_, [&]() {
+    RayEventInit_(
+        source_types, custom_fields, log_dir, event_level, emit_event_to_log_file);
+  });
 }
 
 bool IsExportAPIEnabledSourceType(

src/ray/util/event.h

@@ -25,6 +25,7 @@
 #include <memory>
 #include <sstream>
 #include <string>
+#include <utility>
 #include <variant>
 #include <vector>
 
@@ -143,6 +144,12 @@ class EventManager final {
  public:
   static EventManager &Instance();
 
+  EventManager(const EventManager &manager) = delete;
+
+  const EventManager &operator=(const EventManager &manager) = delete;
+
+  ~EventManager() = default;
+
   bool IsEmpty();
 
   // We added `const json &custom_fields` here because we need to support typed custom
@@ -167,11 +174,6 @@ class EventManager final {
  private:
   EventManager();
 
-  EventManager(const EventManager &manager) = delete;
-
-  const EventManager &operator=(const EventManager &manager) = delete;
-
- private:
   absl::flat_hash_map<std::string, std::shared_ptr<BaseEventReporter>> reporter_map_;
   absl::flat_hash_map<rpc::ExportEvent_SourceType, std::shared_ptr<LogEventReporter>>
       export_log_reporter_map_;
@@ -183,7 +185,7 @@ class RayEventContext final {
  public:
   static RayEventContext &Instance();
 
-  RayEventContext() {}
+  RayEventContext() = default;
 
   void SetEventContext(
       rpc::Event_SourceType source_type,
@@ -201,31 +203,31 @@ class RayEventContext final {
   void UpdateCustomFields(
       const absl::flat_hash_map<std::string, std::string> &custom_fields);
 
-  inline void SetSourceType(rpc::Event_SourceType source_type) {
-    source_type_ = source_type;
-  }
+  void SetSourceType(rpc::Event_SourceType source_type) { source_type_ = source_type; }
 
-  inline const rpc::Event_SourceType &GetSourceType() const { return source_type_; }
+  const rpc::Event_SourceType &GetSourceType() const { return source_type_; }
 
-  inline const std::string &GetSourceHostname() const { return source_hostname_; }
+  const std::string &GetSourceHostname() const { return source_hostname_; }
 
-  inline int32_t GetSourcePid() const { return source_pid_; }
+  int32_t GetSourcePid() const { return source_pid_; }
 
-  inline const absl::flat_hash_map<std::string, std::string> &GetCustomFields() const {
+  const absl::flat_hash_map<std::string, std::string> &GetCustomFields() const {
     return custom_fields_;
   }
 
-  inline bool GetInitialzed() const {
+  bool GetInitialzed() const {
     return source_type_ != rpc::Event_SourceType::Event_SourceType_COMMON;
   }
 
- private:
-  static RayEventContext &GlobalInstance();
-
   RayEventContext(const RayEventContext &event_context) = delete;
 
   const RayEventContext &operator=(const RayEventContext &event_context) = delete;
 
+  ~RayEventContext() = default;
+
+ private:
+  static RayEventContext &GlobalInstance();
+
   rpc::Event_SourceType source_type_ = rpc::Event_SourceType::Event_SourceType_COMMON;
   std::string source_hostname_ = boost::asio::ip::host_name();
   int32_t source_pid_ = getpid();
@@ -251,12 +253,12 @@ class RayEvent {
   // deconstructed. Otherwise we might have memory issues.
   RayEvent(rpc::Event_Severity severity,
            RayLogLevel log_severity,
-           const std::string &label,
+           std::string label,
            const char *file_name,
            int line_number)
       : severity_(severity),
         log_severity_(log_severity),
-        label_(label),
+        label_(std::move(label)),
         file_name_(file_name),
         line_number_(line_number) {}
 
@@ -296,15 +298,15 @@ class RayEvent {
 
   ~RayEvent();
 
+  RayEvent(const RayEvent &event) = delete;
+
+  const RayEvent &operator=(const RayEvent &event) = delete;
+
  private:
   RayEvent() = default;
 
   void SendMessage(const std::string &message);
 
-  RayEvent(const RayEvent &event) = delete;
-
-  const RayEvent &operator=(const RayEvent &event) = delete;
-
   // Only for test
   static void SetLevel(const std::string &event_level);
   // Only for test
@@ -331,13 +333,12 @@ using ExportEventDataPtr = std::variant<std::shared_ptr<rpc::ExportTaskEventData
 class RayExportEvent {
  public:
   explicit RayExportEvent(ExportEventDataPtr event_data_ptr)
-      : event_data_ptr_(event_data_ptr) {}
+      : event_data_ptr_(std::move(event_data_ptr)) {}
 
   ~RayExportEvent();
 
   void SendEvent();
 
- private:
   RayExportEvent(const RayExportEvent &event) = delete;
 
   const RayExportEvent &operator=(const RayExportEvent &event) = delete;
@@ -366,7 +367,7 @@ bool IsExportAPIEnabledSourceType(
 /// \param emit_event_to_log_file if True, it will emit the event to the process log file
 /// (e.g., gcs_server.out). Otherwise, event will only be recorded to the event log file.
 /// \return void.
-void RayEventInit(const std::vector<SourceTypeVariant> source_types,
+void RayEventInit(const std::vector<SourceTypeVariant> &source_types,
                   const absl::flat_hash_map<std::string, std::string> &custom_fields,
                   const std::string &log_dir,
                   const std::string &event_level = "warning",
@@ -376,7 +377,7 @@ void RayEventInit(const std::vector<SourceTypeVariant> source_types,
 /// and has been separated out so RayEventInit can be called multiple times in
 /// tests.
 /// **Note**: This should only be called from tests.
-void RayEventInit_(const std::vector<SourceTypeVariant> source_types,
+void RayEventInit_(const std::vector<SourceTypeVariant> &source_types,
                    const absl::flat_hash_map<std::string, std::string> &custom_fields,
                    const std::string &log_dir,
                    const std::string &event_level,