filesystem: refactor win32 watcher_impl to use absl mutex and fix buffer alignment

source/common/filesystem/win32/watcher_impl.cc

@@ -1,5 +1,7 @@
 #include "source/common/filesystem/watcher_impl.h"
 
+#include "envoy/common/exception.h"
+
 #include "source/common/api/os_sys_calls_impl.h"
 #include "source/common/common/assert.h"
 #include "source/common/common/fmt.h"
@@ -107,7 +109,10 @@ absl::Status WatcherImpl::addWatch(absl::string_view path, uint32_t events, OnCh
     ENVOY_LOG(debug, "created watch for directory: '{}' handle: {}", result.directory_, dir_handle);
   }
 
-  callback_map_[fii_key]->watches_.push_back({file, events, cb});
+  {
+    absl::WriterMutexLock lock(&callback_map_[fii_key]->watches_mutex_);
+    callback_map_[fii_key]->watches_.push_back({file, events, cb});
+  }
   ENVOY_LOG(debug, "added watch for file '{}' in directory '{}'", result.file_, result.directory_);
   return absl::OkStatus();
 }
@@ -142,10 +147,13 @@ void WatcherImpl::issueFirstRead(ULONG_PTR param) {
   // a pointer to DirectoryWatch as the OVERLAPPED for ReadDirectoryChangesW. Then, the
   // completion routine can use its OVERLAPPED* parameter to access the DirectoryWatch see:
   // https://docs.microsoft.com/en-us/windows/desktop/ipc/named-pipe-server-using-completion-routines
-  ReadDirectoryChangesW(dir_watch->dir_handle_, &(dir_watch->buffer_[0]),
-                        dir_watch->buffer_.capacity(), false,
-                        FILE_NOTIFY_CHANGE_FILE_NAME | FILE_NOTIFY_CHANGE_LAST_WRITE, nullptr,
-                        reinterpret_cast<LPOVERLAPPED>(param), &directoryChangeCompletion);
+  dir_watch->overlapped_.Internal = 0;
+  dir_watch->overlapped_.InternalHigh = 0;
+  const BOOL success = ReadDirectoryChangesW(
+      dir_watch->dir_handle_, dir_watch->buffer_.data(), dir_watch->buffer_.size() * sizeof(DWORD),
+      false, FILE_NOTIFY_CHANGE_FILE_NAME | FILE_NOTIFY_CHANGE_LAST_WRITE, nullptr,
+      reinterpret_cast<LPOVERLAPPED>(param), &directoryChangeCompletion);
+  RELEASE_ASSERT(success, fmt::format("ReadDirectoryChangesW failed: {}", GetLastError()));
 
   const BOOL rc = ::SetEvent(dir_watch->overlapped_.hEvent);
   ASSERT(rc);
@@ -200,20 +208,31 @@ void WatcherImpl::directoryChangeCompletion(DWORD err, DWORD num_bytes, LPOVERLA
     }
 
     constexpr absl::string_view data{"a"};
-    for (FileWatch& watch : dir_watch->watches_) {
-      if (watch.file_ == file && (watch.events_ & events)) {
-        ENVOY_LOG(debug, "matched callback: file: {}", watcher->wstring_converter_.to_bytes(file));
-        const auto cb = watch.cb_;
-        const auto cb_closure = [cb, events]() -> void { cb(events); };
-        watcher->active_callbacks_.push(cb_closure);
-        // write a byte to the other end of the socket that libevent is watching
-        // this tells the libevent callback to pull this callback off the active_callbacks_
-        // queue. We do this so that the callbacks are executed in the main libevent loop,
-        // not in this completion routine
-        Buffer::RawSlice buffer{(void*)data.data(), 1};
-        auto result = watcher->write_handle_->writev(&buffer, 1);
-        RELEASE_ASSERT(result.return_value_ == 1,
-                       fmt::format("failed to write 1 byte: {}", result.err_->getErrorDetails()));
+    // Protect watches_ access with ReaderMutexLock
+    {
+      absl::ReaderMutexLock lock(&dir_watch->watches_mutex_);
+      for (FileWatch& watch : dir_watch->watches_) {
+        // Windows file systems are case-insensitive.
+        // An empty watch.file_ matches any file change in the watched directory.
+        if ((watch.file_.empty() || _wcsicmp(watch.file_.c_str(), file.c_str()) == 0) &&
+            (watch.events_ & events)) {
+          ENVOY_LOG(debug, "matched callback: file: {}",
+                    watcher->wstring_converter_.to_bytes(file));
+          const auto cb = watch.cb_;
+          const std::string file_name = watcher->wstring_converter_.to_bytes(file);
+          const auto cb_closure = [watcher, cb, events, file_name]() -> void {
+            watcher->callAndLogOnError(cb, events, file_name);
+          };
+          watcher->active_callbacks_.push(cb_closure);
+          // write a byte to the other end of the socket that libevent is watching
+          // this tells the libevent callback to pull this callback off the active_callbacks_
+          // queue. We do this so that the callbacks are executed in the main libevent loop,
+          // not in this completion routine
+          Buffer::RawSlice buffer{(void*)data.data(), 1};
+          auto result = watcher->write_handle_->writev(&buffer, 1);
+          RELEASE_ASSERT(result.return_value_ == 1,
+                         fmt::format("failed to write 1 byte: {}", result.err_->getErrorDetails()));
+        }
       }
     }
 
@@ -226,10 +245,13 @@ void WatcherImpl::directoryChangeCompletion(DWORD err, DWORD num_bytes, LPOVERLA
     return;
   }
 
-  ReadDirectoryChangesW(dir_watch->dir_handle_, &(dir_watch->buffer_[0]),
-                        dir_watch->buffer_.capacity(), false,
-                        FILE_NOTIFY_CHANGE_FILE_NAME | FILE_NOTIFY_CHANGE_LAST_WRITE, nullptr,
-                        overlapped, directoryChangeCompletion);
+  overlapped->Internal = 0;
+  overlapped->InternalHigh = 0;
+  const BOOL success = ReadDirectoryChangesW(
+      dir_watch->dir_handle_, dir_watch->buffer_.data(), dir_watch->buffer_.size() * sizeof(DWORD),
+      false, FILE_NOTIFY_CHANGE_FILE_NAME | FILE_NOTIFY_CHANGE_LAST_WRITE, nullptr, overlapped,
+      directoryChangeCompletion);
+  RELEASE_ASSERT(success, fmt::format("ReadDirectoryChangesW failed: {}", GetLastError()));
 }
 
 void WatcherImpl::watchLoop() {
@@ -275,5 +297,28 @@ void WatcherImpl::watchLoop() {
   }
 }
 
+void WatcherImpl::callAndLogOnError(const OnChangedCb& cb, uint32_t events,
+                                    const std::string& file) {
+  TRY_ASSERT_MAIN_THREAD {
+    const absl::Status status = cb(events);
+    if (!status.ok()) {
+      // Use ENVOY_LOG_EVERY_POW_2 to avoid log spam if a callback keeps failing.
+      ENVOY_LOG_EVERY_POW_2(warn, "Filesystem watch callback for '{}' returned error: {}", file,
+                            status.message());
+    }
+  }
+  END_TRY
+  MULTI_CATCH(
+      const std::exception& e,
+      {
+        ENVOY_LOG_EVERY_POW_2(warn, "Filesystem watch callback for '{}' threw exception: {}", file,
+                              e.what());
+      },
+      {
+        ENVOY_LOG_EVERY_POW_2(warn, "Filesystem watch callback for '{}' threw unknown exception",
+                              file);
+      });
+}
+
 } // namespace Filesystem
 } // namespace Envoy

source/common/filesystem/win32/watcher_impl.h

@@ -1,11 +1,10 @@
 #pragma once
 
-#include <concurrent_queue.h>
-
 #include <codecvt>
 #include <cstdint>
 #include <list>
 #include <locale>
+#include <queue>
 #include <string>
 
 #include "envoy/api/api.h"
@@ -21,10 +20,33 @@
 #include "source/common/network/io_socket_handle_impl.h"
 
 #include "absl/container/node_hash_map.h"
+#include "absl/synchronization/mutex.h"
 
 namespace Envoy {
 namespace Filesystem {
 
+template <typename T> class ThreadSafeQueue {
+public:
+  void push(const T& value) {
+    absl::WriterMutexLock lock(&mutex_);
+    queue_.push(value);
+  }
+
+  bool try_pop(T& value) {
+    absl::WriterMutexLock lock(&mutex_);
+    if (queue_.empty()) {
+      return false;
+    }
+    value = std::move(queue_.front());
+    queue_.pop();
+    return true;
+  }
+
+private:
+  absl::Mutex mutex_;
+  std::queue<T> queue_;
+};
+
 class WatcherImpl : public Watcher, Logger::Loggable<Logger::Id::file> {
 public:
   WatcherImpl(Event::Dispatcher& dispatcher, Filesystem::Instance& file_system);
@@ -39,6 +61,7 @@ class WatcherImpl : public Watcher, Logger::Loggable<Logger::Id::file> {
   static void endDirectoryWatch(Network::IoHandle& io_handle, HANDLE hEvent);
   void watchLoop();
   void onDirectoryEvent();
+  void callAndLogOnError(const OnChangedCb& cb, uint32_t events, const std::string& file);
 
   struct FileWatch {
     // store the wide character string for ReadDirectoryChangesW
@@ -52,8 +75,9 @@ class WatcherImpl : public Watcher, Logger::Loggable<Logger::Id::file> {
   struct DirectoryWatch {
     OVERLAPPED overlapped_;
     std::list<FileWatch> watches_;
+    absl::Mutex watches_mutex_;
     HANDLE dir_handle_;
-    std::vector<uint8_t> buffer_;
+    std::vector<DWORD> buffer_;
     WatcherImpl* watcher_;
   };
 
@@ -68,7 +92,7 @@ class WatcherImpl : public Watcher, Logger::Loggable<Logger::Id::file> {
   HANDLE thread_exit_event_;
   std::vector<HANDLE> dir_watch_complete_events_;
   std::atomic<bool> keep_watching_;
-  concurrency::concurrent_queue<CbClosure> active_callbacks_;
+  ThreadSafeQueue<CbClosure> active_callbacks_;
   Api::OsSysCallsImpl& os_sys_calls_;
   std::wstring_convert<std::codecvt_utf8_utf16<wchar_t>> wstring_converter_;
 };

test/common/filesystem/watcher_impl_test.cc

@@ -1,5 +1,6 @@
 #include <cstdint>
 #include <fstream>
+#include <thread>
 
 #include "envoy/common/exception.h"
 
@@ -262,7 +263,9 @@ TEST_F(WatcherImplTest, MultipleCallbacksWithErrors) {
   Filesystem::WatcherPtr watcher = dispatcher_->createFilesystemWatcher();
 
   TestEnvironment::createPath(TestEnvironment::temporaryPath("envoy_test"));
-  std::ofstream file(TestEnvironment::temporaryPath("envoy_test/watcher_target"));
+  {
+    std::ofstream file(TestEnvironment::temporaryPath("envoy_test/watcher_target"));
+  }
 
   int callback_count = 0;
   ASSERT_OK(watcher->addWatch(TestEnvironment::temporaryPath("envoy_test/watcher_target"),
@@ -280,12 +283,19 @@ TEST_F(WatcherImplTest, MultipleCallbacksWithErrors) {
   dispatcher_->run(Event::Dispatcher::RunType::NonBlock);
 
   // Trigger first modification. The first callback returns error, but watcher continues.
-  file << "text1" << std::flush;
+  {
+    std::ofstream file(TestEnvironment::temporaryPath("envoy_test/watcher_target"), std::ios::app);
+    file << "text1";
+  }
+  std::this_thread::sleep_for(std::chrono::milliseconds(100)); // NO_CHECK_FORMAT(real_time)
   dispatcher_->run(Event::Dispatcher::RunType::NonBlock);
 
+  std::this_thread::sleep_for(std::chrono::milliseconds(100)); // NO_CHECK_FORMAT(real_time)
   // Trigger second modification. It should still work.
-  file << "text2" << std::flush;
-  file.close();
+  {
+    std::ofstream file(TestEnvironment::temporaryPath("envoy_test/watcher_target"), std::ios::app);
+    file << "text2";
+  }
   dispatcher_->run(Event::Dispatcher::RunType::Block);
 
   EXPECT_EQ(2, callback_count);