tlbmc/utils/inotify_watcher.cc

#include "tlbmc/utils/inotify_watcher.h"

#include <sys/inotify.h>
#include <unistd.h>

#include <cstddef>
#include <cstdint>
#include <memory>
#include <string>
#include <utility>

#include "absl/status/status.h"
#include "absl/strings/str_cat.h"
#include "absl/synchronization/mutex.h"

namespace milotic_tlbmc {

constexpr size_t kEventSize = sizeof(struct inotify_event);
constexpr size_t kEventBufferLength = 4096;

absl::StatusOr<std::shared_ptr<InotifyWatcher>> InotifyWatcher::Create(
    std::shared_ptr<boost::asio::io_context> io_context,
    EventCallback callback) {
  int fd = inotify_init1(IN_CLOEXEC | IN_NONBLOCK);
  if (fd < 0) {
    return absl::InternalError("Failed to initialize inotify");
  }
  auto watcher = std::shared_ptr<InotifyWatcher>(
      new InotifyWatcher(std::move(io_context), std::move(callback), fd));
  watcher->StartAsyncRead();
  return watcher;
}

InotifyWatcher::InotifyWatcher(
    std::shared_ptr<boost::asio::io_context> io_context, EventCallback callback,
    int fd)
    : inotify_fd_(fd),
      io_context_(std::move(io_context)),
      stream_descriptor_(*io_context_, fd),
      read_buffer_(kEventBufferLength),
      callback_(std::move(callback)) {}

InotifyWatcher::~InotifyWatcher() { Stop(); }

absl::Status InotifyWatcher::WatchDirectory(std::string dir_path,
                                            uint32_t event_mask) {
  absl::MutexLock lock(mutex_);
  if (stopped_) {
    return absl::FailedPreconditionError("InotifyWatcher is stopped");
  }
  int wd = inotify_add_watch(inotify_fd_, dir_path.c_str(), event_mask);
  if (wd < 0) {
    return absl::InternalError(
        absl::StrCat("Failed to add watch for ", dir_path));
  }
  watch_descriptors_[wd] = std::move(dir_path);
  return absl::OkStatus();
}

void InotifyWatcher::Stop() {
  if (stopped_.exchange(true)) {
    return;
  }
  boost::system::error_code ec;
  (void)stream_descriptor_.cancel(ec);  // NOLINT(bugprone-unused-return-value)
  absl::MutexLock lock(mutex_);
  (void)stream_descriptor_.close(ec);  // NOLINT(bugprone-unused-return-value)
  inotify_fd_ = -1;
}

void InotifyWatcher::StartAsyncRead() {
  if (stopped_) {
    return;
  }
  auto self = weak_from_this();
  stream_descriptor_.async_read_some(
      boost::asio::buffer(read_buffer_),
      [self](const boost::system::error_code& error,
             std::size_t bytes_transferred) {
        if (auto shared_self = self.lock()) {
          shared_self->HandleRead(error, bytes_transferred);
        }
      });
}

void InotifyWatcher::HandleRead(const boost::system::error_code& error,
                                std::size_t bytes_transferred) {
  if (error || stopped_) {
    return;
  }

  if (bytes_transferred > read_buffer_.size()) {
    return;
  }

  size_t i = 0;
  while (i + kEventSize <= bytes_transferred) {
    if (i >= read_buffer_.size()) {
      break;
    }
    struct inotify_event* event =
        reinterpret_cast<struct inotify_event*>(&read_buffer_[i]);
    if (i + kEventSize + event->len > bytes_transferred) {
      break;
    }
    std::string dir_path;
    {
      absl::MutexLock lock(mutex_);
      auto it = watch_descriptors_.find(event->wd);
      if (it != watch_descriptors_.end()) {
        dir_path = it->second;
        if ((event->mask & IN_IGNORED) != 0) {
          watch_descriptors_.erase(it);
        }
      }
    }
    // event->len == 0 indicates an event on the watched directory itself (e.g.,
    // directory deletion or watch eviction), so we pass an empty filename.
    absl::string_view filename = (event->len > 0) ? event->name : "";
    callback_(dir_path, filename, event->mask);
    i += kEventSize + event->len;
  }

  StartAsyncRead();
}

}  // namespace milotic_tlbmc