Efficient Directory Polling Implementation in C++ for Mac OS X

Efficient Directory Polling Implementation in C++ for Mac OS X

In this sample code project I demonstrate the implementation of a straightforward directory poller in C++. Typically, on a Linux system, the inotify API can be used for this purpose. However, due to its incompatibility with Mac OS, I adopt an alternative approach, leveraging the kqueue and kevent API.

The DirectoryPoller Class

Let us first define the interface of our DirectoryPoller class:

#ifdef __APPLE__

#ifndef DIRECTORY_POLLER_MAC_OS_H
#define DIRECTORY_POLLER_MAC_OS_H
#include <memory>
#include <string>
#include <vector>
#include <sys/event.h> // for kqueue() etc.
#include <set>
#include "AbstractPoller.h"
#include "impl/PollResult.h"

class DirectoryPollerBuilder;

class DirectoryPoller : public AbstractPoller
{
public:
  DirectoryPoller(std::string name, std::string dir_to_watch, std::shared_ptr<SignalChannel> sig_channel);
  ~DirectoryPoller() override;
  AbstractPoller *clone() const override;

  friend class DirectoryPollerBuilder;
  static DirectoryPollerBuilder builder(std::string name);

private:
  int m_kq = -1;
  int m_fd = -1;
  struct kevent *m_ke;
  std::string m_dir_to_watch;
  std::vector<std::string> m_file_paths;
  std::set<std::string> m_last_files;
  bool init_dir_watch();
  std::set<std::string> list_files();
  PollResult poll() override;
  bool should_add_file(const std::string &f, bool starting_up);
  void clean() override;
};

#endif
#endif

And here is the associated implementation:

#ifdef __APPLE__
#include "DirectoryPollerMacOS.h"
#include "DirectoryPollerBuilder.h"
#include "logging/Logging.h"
#include "Util.h"
#include <stdlib.h> // for srand(), rand()
#include <thread>
#include <sstream>
#include <string> // for strerror()
#include <chrono>
#include <iostream>
#include <stdio.h>
#include <errno.h>  // for errno
#include <fcntl.h>  // for O_RDONLY
#include <stdio.h>  // for fprintf()
#include <stdlib.h> // for EXIT_SUCCESS
#include <string.h> // for strerror()
#include <unistd.h> // for close()
#include <filesystem>
#include <algorithm> // for set_difference()
#include <signal.h>

DirectoryPoller::DirectoryPoller(std::string name, std::string dir_to_watch, std::shared_ptr<SignalChannel> sig_channel) : AbstractPoller(name, sig_channel), m_dir_to_watch(dir_to_watch)
{
  m_last_files = list_files();
  for (const std::string &f : m_last_files)
  {
    if (should_add_file(f, true))
    {
      Logging::INFO("Adding file:'" + f + "'", m_name);
      m_file_paths.emplace_back(f);
    }
  }
}

bool DirectoryPoller::should_add_file(const std::string &f, bool starting_up)
{
  return (starting_up || !Util::str_ends_with(f.c_str(), "_inprogress")) && !Util::str_ends_with(f.c_str(), "_done");
}

std::set<std::string> DirectoryPoller::list_files()
{
  std::set<std::string> files;
  for (const auto &entry : std::filesystem::directory_iterator(m_dir_to_watch))
  {
    files.emplace(entry.path());
  }
  return files;
}

bool DirectoryPoller::init_dir_watch()
{
  Logging::INFO("Watching '" + m_dir_to_watch + "'", m_name);

  /* A single kqueue */
  // https://www.freebsd.org/cgi/man.cgi?kqueue
  m_kq = kqueue();

  /* Two kevent structs */
  m_ke = (struct kevent *)malloc(sizeof(struct kevent));

  /* Initialise the struct for the file descriptor */
  m_fd = open(m_dir_to_watch.c_str(), O_RDONLY);
  EV_SET(m_ke, m_fd, EVFILT_VNODE, EV_ADD | EV_CLEAR, NOTE_DELETE | NOTE_RENAME | NOTE_WRITE, 0, NULL);

  /* Register for the event */
  if (kevent(m_kq, m_ke, 1, NULL, 0, NULL) < 0)
  {
    perror("kevent");
    return false;
  }

  return true;
}

PollResult DirectoryPoller::poll()
{
  // Do not wait for events if we already have files
  if (!m_file_paths.empty())
  {
    std::string p = m_file_paths.back();
    m_file_paths.pop_back();
    return PollResult(p);
  }
  else
  {
    // Poll for events
    if (m_fd == -1 || m_kq == -1)
    {
      if (!init_dir_watch())
      {
        Logging::ERROR("Unable to init directory watch for '" + m_dir_to_watch + "'", m_name);
        kill(getpid(), SIGINT);
      }
    }

    memset(m_ke, 0x00, sizeof(struct kevent));

    // Camp here for event
    struct timespec timeout;
    timeout.tv_sec = 0;
    timeout.tv_nsec = 1;
    int return_value = kevent(m_kq, NULL, 0, m_ke, 1, &timeout);
    if (return_value < 0)
    {
      perror("kevent");
    }
    else if (return_value == 0)
    {
      // timeout
    }
    else
    {
      switch (m_ke->filter)
      {
      /* File descriptor event: let's examine what happened to the file */
      case EVFILT_VNODE:
        Logging::DEBUG("Events " + std::to_string(m_ke->fflags) + " on file descriptor " + std::to_string(m_ke->ident), m_name);

        if (m_ke->fflags & NOTE_DELETE)
        {
          Logging::DEBUG("The unlink() system call was called on the file referenced by the descriptor", m_name);
        }
        if (m_ke->fflags & NOTE_WRITE)
        {
          Logging::DEBUG("A write occurred on the file referenced by the descriptor", m_name);
          std::set<std::string> current = list_files();

          std::set<std::string> added;
          // In the end, the set 'added' will contain the current-m_last_files.
          std::set_difference(current.begin(), current.end(), m_last_files.begin(), m_last_files.end(), std::inserter(added, added.end()));
          for (const std::string &f : added)
          {
            if (should_add_file(f, false))
            {
              Logging::INFO("Adding file '" + f + "'", m_name);
              m_file_paths.emplace_back(f);
            }
          }

          std::set<std::string> removed;
          // In the end, the set 'removed' will contain the m_last_files-current.
          std::set_difference(m_last_files.begin(), m_last_files.end(), current.begin(), current.end(), std::inserter(removed, removed.end()));
          for (const std::string &f : removed)
          {
            Logging::INFO("File '" + f + "' removed from directory", m_name);
          }

          m_last_files = current;
        }
        if (m_ke->fflags & NOTE_EXTEND)
        {
          Logging::INFO("The file referenced by the descriptor was extended", m_name);
        }
        if (m_ke->fflags & NOTE_ATTRIB)
        {
          Logging::INFO("The file referenced by the descriptor had its attributes changed", m_name);
        }
        if (m_ke->fflags & NOTE_LINK)
        {
          Logging::INFO("The link count on the file changed", m_name);
        }
        if (m_ke->fflags & NOTE_RENAME)
        {
          Logging::INFO("The file referenced by the descriptor was renamed", m_name);
        }
        if (m_ke->fflags & NOTE_REVOKE)
        {
          Logging::INFO("Access to the file was revoked via revoke(2) or the underlying fileystem was unmounted", m_name);
        }
        break;

      /* This should never happen */
      default:
        Logging::ERROR("Unknown filter", m_name);
      }
    }

    if (m_file_paths.empty())
    {
      return PollResult("");
    }
    else
    {
      std::string p = m_file_paths.back();
      m_file_paths.pop_back();
      return PollResult(p);
    }
  }
}

AbstractPoller *DirectoryPoller::clone() const
{
  return new DirectoryPoller(*this);
}

DirectoryPollerBuilder DirectoryPoller::builder(std::string name)
{
  return DirectoryPollerBuilder(name);
}

void DirectoryPoller::clean()
{
  Logging::INFO("Cleaning up", m_name);
}

DirectoryPoller::~DirectoryPoller()
{
  if (fcntl(m_kq, F_GETFD) != -1)
  {
    close(m_kq);
  }
}
#endif

The primary concept is that a client has the capability to invoke the poll() function, obtaining a new file path encapsulated within a PollResult object, if one is available.

DirectoryPollerBuilder

If you’re curious about the implementation of the DirectoryPollerBuilder, here is the associated code:

#ifndef DIRECTORY_POLLER_BUILDER_H
#define DIRECTORY_POLLER_BUILDER_H

#ifdef __linux__
#include "DirectoryPoller.h"
#elif __APPLE__
#include "DirectoryPollerMacOS.h"
#endif

#include <string>

class DirectoryPollerBuilder
{
private:
    std::string m_name;
    std::string m_dir_to_watch;
    std::shared_ptr<SignalChannel> m_sig_channel;

public:
    DirectoryPollerBuilder(std::string name);
    DirectoryPollerBuilder &with_directory(std::string p);
    DirectoryPollerBuilder &with_sig_channel(std::shared_ptr<SignalChannel> sc);
    DirectoryPoller build();
};

#endif

#include "DirectoryPollerBuilder.h"
#include "DirectoryPoller.h"

DirectoryPollerBuilder::DirectoryPollerBuilder(std::string name) : m_name(name)
{
}

DirectoryPollerBuilder &DirectoryPollerBuilder::with_directory(std::string p)
{
    m_dir_to_watch = p;
    return *this;
}

DirectoryPollerBuilder &DirectoryPollerBuilder::with_sig_channel(std::shared_ptr<SignalChannel> sc)
{
    m_sig_channel = sc;
    return *this;
}

DirectoryPoller DirectoryPollerBuilder::build()
{

    if (m_dir_to_watch.empty())
    {
        throw std::runtime_error("No directory must be provided");
    }

    if (!m_sig_channel)
    {
        throw std::runtime_error("No signal channel provided");
    }

    DirectoryPoller poller = DirectoryPoller(m_name, m_dir_to_watch, m_sig_channel);

    return poller;
}

The poller can then be used as follows:

  DirectoryPoller poller = DirectoryPoller::builder("DirectoryPoller")
                               .with_directory(dir_to_watch)
                               .with_sig_channel(sig_channel)
                               .build();

...
std::unique_ptr<std::thread> m_t = std::make_unique<std::thread>(&AbstractPoller::run, &poller);
...
// Possibly join()

AbstractPoller & AbstractWorker

AbstractPoller essentially constitutes an abstract class delineating the foundational polling interface for the poller:

#ifndef ABSTRACT_POLLER_H
#define ABSTRACT_POLLER_H

#include "AbstractWorker.h"
#include "impl/PollResult.h"

class AbstractPoller : public AbstractWorker
{
private:
  void step() override;
  virtual PollResult poll() = 0;
  virtual void clean() = 0;

public:
  AbstractPoller(std::string name, std::shared_ptr<SignalChannel> sig_channel);
  virtual ~AbstractPoller(){};
  virtual AbstractPoller *clone() const = 0;
};

#endif

#include "AbstractPoller.h"

AbstractPoller::AbstractPoller(std::string name, std::shared_ptr<SignalChannel> sig_channel) : AbstractWorker(name, sig_channel)
{
}

void AbstractPoller::step()
{
  PollResult r = poll();
  if (!r.empty())
  {
    m_queue->enqueue(r);
  }
}

#ifndef ABSTRACT_WORKER_H
#define ABSTRACT_WORKER_H

#include "SafeQueue.h"
#include "impl/PollResult.h"
#include "SignalChannel.h"
#include <string>

class AbstractWorker
{
public:
  AbstractWorker(std::string name, std::shared_ptr<SignalChannel> sig_channel) : m_name(name), m_sig_channel(sig_channel){};
  void set_queue(SafeQueue<PollResult> *queue_);
  void run();
  ~AbstractWorker(){};

protected:
  SafeQueue<PollResult> *m_queue;
  const std::string m_name;

private:
  std::shared_ptr<SignalChannel> m_sig_channel;
  virtual void step() = 0;
  virtual void clean() = 0;
};

#endif

#include "AbstractWorker.h"
#include "logging/Logging.h"

void AbstractWorker::set_queue(SafeQueue<PollResult> *queue) { m_queue = queue; }

void AbstractWorker::run()
{
  while (!m_sig_channel->m_shutdown_requested.load())
  {
    {
      std::unique_lock shutdown_lock(m_sig_channel->m_cv_mutex);
      m_sig_channel->m_cv.wait_for(shutdown_lock, std::chrono::milliseconds(10), [this]()
                                   { bool should_shutdown = m_sig_channel->m_shutdown_requested.load();
                                     return should_shutdown; });
    }

    step();
  }
  Logging::INFO("Shutting down", m_name);
  clean();
}

SignalChannel

SignalChannel functions as a shared variable across threads, facilitating the communication of notifications to the poller, prompting it to halt the polling process:

#ifndef SIGNAL_CHANNEL_H
#define SIGNAL_CHANNEL_H

#include <atomic>
#include <condition_variable>

class SignalChannel
{
public:
    std::atomic<bool> m_shutdown_requested = false;
    std::mutex m_cv_mutex;
    std::condition_variable m_cv;
};

#endif

Logging

Finally, Logging represents a logging utility utilized in my personal projects. A detailed discussion on this utility will be presented in a subsequent post.