monitor.h

#ifndef CAPIO_CL_MONITOR_H
#define CAPIO_CL_MONITOR_H
 
#include <atomic>
#include <cstddef>
#include <cstdint>
#include <filesystem>
#include <mutex>
#include <optional>
#include <set>
#include <string>
#include <thread>
#include <unordered_map>
#include <vector>
 
#include "configuration.h"

namespace capiocl::monitor {

inline constexpr std::size_t HOSTNAME_BUFFER_SIZE = 1024;

inline constexpr std::size_t PATH_BUFFER_SIZE = 4096;

static const std::string NO_HOME_NODE = "<NONE>";

class MonitorException final : public std::exception {
    std::string message{};
 
  public:
    explicit MonitorException(const std::string &msg);

    [[nodiscard]] const char *what() const noexcept override { return message.c_str(); }
};

class MonitorInterface {
  protected:
    mutable std::mutex committed_lock;

    mutable std::mutex home_node_lock;

    mutable std::vector<std::string> _committed_files;

    mutable std::unordered_map<std::string, std::string> _home_nodes;

    mutable char _hostname[HOSTNAME_BUFFER_SIZE] = {0};
 
  public:
    virtual ~MonitorInterface() = default;

    virtual bool isCommitted(const std::filesystem::path &path) const;

    virtual void setCommitted(const std::filesystem::path &path) const;

    virtual std::optional<bool> increaseCloseCount(const std::filesystem::path &path,
                                                   long threshold) const;

    virtual void setHomeNode(const std::filesystem::path &path) const;

    virtual std::string getHomeNode(const std::filesystem::path &path) const;
};

class MulticastMonitor final : public MonitorInterface {
 
    static constexpr int MESSAGE_SIZE = 2 + (2 * PATH_BUFFER_SIZE); 

    std::thread commit_thread, home_node_thread;

    std::string MULTICAST_COMMIT_ADDR;
 
    std::string MULTICAST_HOME_NODE_ADDR;

    std::atomic<bool> terminate = false;

    mutable std::unordered_map<std::string, std::unordered_map<std::string, std::uint64_t>>
        close_counts;
    mutable std::mutex close_count_lock;
    std::string close_count_origin;

    static constexpr int MULTICAST_THREAD_POLL_INTERVAL = 250;

    int MULTICAST_COMMIT_PORT{};
 
    int MULTICAST_HOME_NODE_PORT{};

    int MULTICAST_DELAY_MILLIS{};

    typedef enum { SET = '!', GET = '?', COUNT_SET = 'C', COUNT_GET = 'Q' } MESSAGE_COMMANDS;

    static void _send_message(const std::string &ip_addr, int ip_port, const std::string &payload,
                              MESSAGE_COMMANDS action);

    static void
    commit_listener(std::vector<std::string> &committed_files, std::mutex &lock,
                    std::unordered_map<std::string, std::unordered_map<std::string, std::uint64_t>>
                        &close_counts,
                    std::mutex &close_count_lock, const std::string &ip_addr, int ip_port,
                    const std::atomic<bool> *terminate);

    static void home_node_listener(std::unordered_map<std::string, std::string> &home_nodes,
                                   std::mutex &lock, const std::string &ip_addr, int ip_port,
                                   const std::atomic<bool> *terminate);
 
  public:
    MulticastMonitor(const capiocl::configuration::CapioClConfiguration &config);

    ~MulticastMonitor() override;
 
    bool isCommitted(const std::filesystem::path &path) const override;
    void setCommitted(const std::filesystem::path &path) const override;
    std::optional<bool> increaseCloseCount(const std::filesystem::path &path,
                                           long threshold) const override;
    void setHomeNode(const std::filesystem::path &path) const override;
    std::string getHomeNode(const std::filesystem::path &path) const override;
};

class FileSystemMonitor final : public MonitorInterface {
 
    typedef enum { COMMIT, HOME_NODE } CAPIO_CL_COMMIT_TOKEN_TYPES;
    std::filesystem::path metadata_root;

    static std::filesystem::path
    compute_capiocl_token_name(const std::filesystem::path &path,
                               CAPIO_CL_COMMIT_TOKEN_TYPES type = COMMIT);

    static void generate_commit_token(const std::filesystem::path &path);

    static void generate_home_node_token(const std::filesystem::path &path,
                                         const std::string &home_node);
 
  public:
    FileSystemMonitor();

    explicit FileSystemMonitor(const configuration::CapioClConfiguration &config);

    ~FileSystemMonitor() override = default;
 
    bool isCommitted(const std::filesystem::path &path) const override;
    void setCommitted(const std::filesystem::path &path) const override;
    std::optional<bool> increaseCloseCount(const std::filesystem::path &path,
                                           long threshold) const override;
    void setHomeNode(const std::filesystem::path &path) const override;
    std::string getHomeNode(const std::filesystem::path &path) const override;
};

class Monitor {
    friend class Engine;
    friend class MonitorInterface;
 
    std::vector<const MonitorInterface *> interfaces;
 
  public:
    [[nodiscard]] bool isCommitted(const std::filesystem::path &path) const;

    void setCommitted(std::filesystem::path path) const;

    [[nodiscard]] bool increaseCloseCount(const std::filesystem::path &path, long threshold) const;

    void registerMonitorBackend(const MonitorInterface *interface);

    void setHomeNode(const std::filesystem::path &path) const;

    [[nodiscard]] std::set<std::string> getHomeNode(const std::filesystem::path &path) const;
 
    ~Monitor();
};
} // namespace capiocl::monitor
 
#endif // CAPIO_CL_MONITOR_H