Refactor and simplify inotify watches

commit_hash:df3b57704dbfda17915e4aa68619491e48d96937

Author: babenko

yt/yt/core/logging/log_manager.cpp

@@ -609,27 +609,61 @@ class TLogManager::TImpl
         return EmplaceOrCrash(KeyToCachedWriter_, cacheKey, writers)->second;
     }
 
-    std::unique_ptr<TInotifyWatch> CreateNotificationWatch(
+    TInotifyHandle* TryGetNotificationHandle()
+    {
+        if (!NotificationHandle_ && !NotificationHandleCreationFailed_) {
+            try {
+                NotificationHandle_ = std::make_unique<TInotifyHandle>();
+            } catch (const std::exception& ex) {
+                YT_LOG_ERROR(ex, "Error creating inotify handle, watching disabled");
+                NotificationHandleCreationFailed_ = true;
+            }
+        }
+        return NotificationHandle_.get();
+    }
+
+    std::unique_ptr<TInotifyWatch> TryCreateNotificationWatch(
         const TLogManagerConfigPtr& config,
         const IFileLogWriterPtr& writer)
     {
 #ifdef _linux_
         if (config->WatchPeriod) {
-            if (!NotificationHandle_) {
-                NotificationHandle_ = std::make_unique<TInotifyHandle>();
+            auto* notifcationHandle = TryGetNotificationHandle();
+            if (!notifcationHandle) {
+                return nullptr;
+            }
+
+            try {
+                return std::make_unique<TInotifyWatch>(
+                    notifcationHandle,
+                    writer->GetFileName(),
+                    EInotifyWatchEvents::DeleteSelf | EInotifyWatchEvents::MoveSelf);
+            } catch (const std::exception& ex) {
+                // Watch can fail to initialize if the writer is disabled
+                // e.g. due to the lack of space.
+                YT_LOG_ERROR(ex, "Error creating inotify watch (Path: %v)",
+                    writer->GetFileName());
+                return nullptr;
             }
-            return std::make_unique<TInotifyWatch>(
-                NotificationHandle_.get(),
-                writer->GetFileName(),
-                EInotifyWatchEvents::DeleteSelf | EInotifyWatchEvents::MoveSelf,
-                BIND(&ILogWriter::Reload, writer));
         }
 #else
         Y_UNUSED(config, writer);
 #endif
         return nullptr;
     }
 
+    void CreateNotificationWatchForWriter(
+        const TLogManagerConfigPtr& config,
+        const IFileLogWriterPtr& writer)
+    {
+        if (auto watch = TryCreateNotificationWatch(config, writer)) {
+            EmplaceOrCrash(NotificationWatchWDToWriter_, watch->GetWD(), writer);
+            EmplaceOrCrash(WriterToNotificationWatch_, writer, std::move(watch));
+        } else {
+            InsertOrCrash(WritersWithFailedNotificationWatches_, writer);
+        }
+    }
+
     void UpdateConfig(const TConfigEvent& event)
     {
         YT_ASSERT_THREAD_AFFINITY(LoggingThread);
@@ -709,9 +743,10 @@ class TLogManager::TImpl
 
         NameToWriter_.clear();
         KeyToCachedWriter_.clear();
-        WDToNotificationWatch_.clear();
-        NotificationWatches_.clear();
-        InvalidNotificationWatches_.clear();
+
+        WriterToNotificationWatch_.clear();
+        NotificationWatchWDToWriter_.clear();
+        WritersWithFailedNotificationWatches_.clear();
 
         for (const auto& [name, writerConfig] : config->Writers) {
             auto typedWriterConfig = ConvertTo<TLogWriterConfigPtr>(writerConfig);
@@ -729,11 +764,7 @@ class TLogManager::TImpl
             EmplaceOrCrash(NameToWriter_, name, writer);
 
             if (auto fileWriter = DynamicPointerCast<IFileLogWriter>(writer)) {
-                auto watch = CreateNotificationWatch(config, fileWriter);
-                if (watch) {
-                    RegisterNotificatonWatch(watch.get());
-                    NotificationWatches_.push_back(std::move(watch));
-                }
+                CreateNotificationWatchForWriter(config, fileWriter);
             }
         }
 
@@ -827,56 +858,31 @@ class TLogManager::TImpl
         }
     }
 
-    void RegisterNotificatonWatch(TInotifyWatch* watch)
-    {
-        YT_ASSERT_THREAD_AFFINITY(LoggingThread);
-
-        if (watch->IsValid()) {
-            // Watch can fail to initialize if the writer is disabled
-            // e.g. due to the lack of space.
-            EmplaceOrCrash(WDToNotificationWatch_, watch->GetWD(), watch);
-        } else {
-            InvalidNotificationWatches_.push_back(watch);
-        }
-    }
-
     void WatchWriters()
     {
         YT_ASSERT_THREAD_AFFINITY(LoggingThread);
 
-        if (!NotificationHandle_) {
+        auto* notificationHandle = TryGetNotificationHandle();
+        if (!notificationHandle) {
             return;
         }
 
-        int previousWD = -1;
-        while (auto pollResult = NotificationHandle_->Poll()) {
-            if (pollResult->WD == previousWD) {
-                continue;
-            }
-            auto it = WDToNotificationWatch_.find(pollResult->WD);
-            auto jt = WDToNotificationWatch_.end();
-            if (it == jt) {
-                continue;
-            }
+        auto config = Config_.Acquire();
 
-            auto* watch = it->second;
-            watch->Run();
+        // Always reload writers and retry registration for invalid watches.
+        auto writersToReconsider = std::exchange(WritersWithFailedNotificationWatches_, {});
 
-            if (watch->GetWD() != pollResult->WD) {
-                WDToNotificationWatch_.erase(it);
-                RegisterNotificatonWatch(watch);
+        while (auto pollResult = notificationHandle->Poll()) {
+            if (auto writer = GetOrDefault(NotificationWatchWDToWriter_, pollResult->WD)) {
+                EraseOrCrash(NotificationWatchWDToWriter_, pollResult->WD);
+                EraseOrCrash(WriterToNotificationWatch_, writer);
+                InsertOrCrash(writersToReconsider, writer);
             }
-
-            previousWD = pollResult->WD;
         }
-        // Handle invalid watches, try to register they again.
-        {
-            std::vector<TInotifyWatch*> invalidNotificationWatches;
-            invalidNotificationWatches.swap(InvalidNotificationWatches_);
-            for (auto* watch : invalidNotificationWatches) {
-                watch->Run();
-                RegisterNotificatonWatch(watch);
-            }
+
+        for (const auto& writer : writersToReconsider) {
+            writer->Reload();
+            CreateNotificationWatchForWriter(config, writer);
         }
     }
 
@@ -1358,9 +1364,11 @@ class TLogManager::TImpl
     const IThreadPoolPtr CompressionThreadPool_;
 
     std::unique_ptr<TInotifyHandle> NotificationHandle_;
-    std::vector<std::unique_ptr<TInotifyWatch>> NotificationWatches_;
-    THashMap<int, TInotifyWatch*> WDToNotificationWatch_;
-    std::vector<TInotifyWatch*> InvalidNotificationWatches_;
+    bool NotificationHandleCreationFailed_ = false;
+
+    THashMap<IFileLogWriterPtr, std::unique_ptr<TInotifyWatch>> WriterToNotificationWatch_;
+    THashMap<int, IFileLogWriterPtr> NotificationWatchWDToWriter_;
+    THashSet<IFileLogWriterPtr> WritersWithFailedNotificationWatches_;
 
     THashMap<TString, TLoggingAnchor*> AnchorMap_;
     std::atomic<TLoggingAnchor*> FirstAnchor_ = nullptr;

yt/yt/core/misc/inotify.cpp

@@ -77,64 +77,30 @@ std::optional<TInotifyHandle::TPollResult> TInotifyHandle::Poll()
 TInotifyWatch::TInotifyWatch(
     TInotifyHandle* handle,
     std::string path,
-    EInotifyWatchEvents mask,
-    TClosure callback)
+    EInotifyWatchEvents mask)
     : FD_(handle->GetFD())
     , Path_(std::move(path))
     , Mask_(mask)
-    , Callback_(std::move(callback))
 {
-    YT_VERIFY(FD_ >= 0);
-    CreateWatch();
-}
-
-TInotifyWatch::~TInotifyWatch()
-{
-    DropWatch();
-}
-
-bool TInotifyWatch::IsValid() const
-{
-    return WD_ >= 0;
-}
-
-void TInotifyWatch::Run()
-{
-    // Unregister before create a new file.
-    DropWatch();
-    Callback_();
-    // Register the newly created file.
-    CreateWatch();
-}
-
-void TInotifyWatch::CreateWatch()
-{
-    YT_VERIFY(WD_ < 0);
 #ifdef _linux_
+    YT_VERIFY(FD_ >= 0);
     WD_ = inotify_add_watch(
         FD_,
         Path_.c_str(),
         ToUnderlying(Mask_));
-
     if (WD_ < 0) {
-        YT_LOG_ERROR(
-            TError::FromSystem(errno),
-            "Error registering watch (Path: %v)",
-            Path_);
         WD_ = -1;
-    } else if (WD_ > 0) {
-        YT_LOG_DEBUG("Watch registered (WD: %v, Path: %v)",
-            WD_,
-            Path_);
-    } else {
-        YT_ABORT();
+        THROW_ERROR_EXCEPTION("Error registering watch for %v",
+            Path_)
+            << TError::FromSystem();
     }
-#else
-    WD_ = -1;
+    YT_LOG_DEBUG("Watch registered (WD: %v, Path: %v)",
+        WD_,
+        Path_);
 #endif
 }
 
-void TInotifyWatch::DropWatch()
+TInotifyWatch::~TInotifyWatch()
 {
 #ifdef _linux_
     if (WD_ > 0) {
@@ -144,7 +110,6 @@ void TInotifyWatch::DropWatch()
         inotify_rm_watch(FD_, WD_);
     }
 #endif
-    WD_ = -1;
 }
 
 ////////////////////////////////////////////////////////////////////////////////

yt/yt/core/misc/inotify.h

@@ -58,24 +58,15 @@ class TInotifyWatch
     TInotifyWatch(
         TInotifyHandle* handle,
         std::string path,
-        EInotifyWatchEvents mask,
-        TClosure callback);
+        EInotifyWatchEvents mask);
     ~TInotifyWatch();
 
-    bool IsValid() const;
-
-    void Run();
-
     DEFINE_BYVAL_RO_PROPERTY(int, FD, -1);
     DEFINE_BYVAL_RO_PROPERTY(int, WD, -1);
 
 private:
     const std::string Path_;
     const EInotifyWatchEvents Mask_;
-    const TClosure Callback_;
-
-    void CreateWatch();
-    void DropWatch();
 };
 
 ////////////////////////////////////////////////////////////////////////////////