Program Listing for File fifo_writer.cpp

↰ Return to documentation for file (monitoring/fifo_writer.cpp)

#include "fifo_writer.hpp"

#include <format>
#include <sys/stat.h>
#include <ers/Severity.h>

namespace {

class ScopedFd {

public:
    explicit ScopedFd(int fd) : m_fd(fd) {}
    ~ScopedFd() { if (m_fd >= 0) ::close(m_fd); }

    ScopedFd(const ScopedFd&)            = delete;
    ScopedFd& operator=(const ScopedFd&) = delete;
    ScopedFd(const ScopedFd&&)            = delete;
    ScopedFd&& operator=(const ScopedFd&&) = delete;

    [[nodiscard]] bool valid()   const { return m_fd >= 0; }
    [[nodiscard]] int  get()     const { return m_fd; }
    [[nodiscard]] int  release() { int tmp = m_fd; m_fd = -1; return tmp; }

private:
    int m_fd {-1};
};

} // namespace


FIFOWriter::FIFOWriter(const std::string& fifoname) {
    try {
        validate_fifo(fifoname);
    } catch (const monitoring_log::fifo_issue& err) {
        ers::error(err);
        return;
    }

    // Suppressing SIGPIPE so that a broken reader yields to EPIPE on write()
    // instead of killing the process
    signal(SIGPIPE, SIG_IGN);

    // A write-only open on a FIFO blocks until a reader exists.
    // holding a temporary read-only open just to unblock the write-only one.
    ScopedFd tmp_fd{open(fifoname.c_str(), O_RDONLY | O_NDELAY)};
    if (not tmp_fd.valid()) {
        ers::error(monitoring_log::fifo_issue(
            std::format("Cannot open read end of FIFO {}, error {}: {}",
                        fifoname, errno, strerror(errno))));
        return;
    }

    ScopedFd write_fd{open(fifoname.c_str(), O_WRONLY | O_NDELAY | O_NONBLOCK)};
    if (not write_fd.valid()) {
        ers::error(monitoring_log::fifo_issue(
            std::format("Cannot open write end of FIFO {}, error {}: {}",
                        fifoname, errno, strerror(errno))));
        return;
    }

    constexpr int buffer_resize_1MB = 1048576;
    if (fcntl(write_fd.get(), F_SETPIPE_SZ, buffer_resize_1MB ) == -1) {
        ers::warning(monitoring_log::fifo_issue(
            std::format("Could not resize FIFO {} pipe buffer, error {}: {}",
                        fifoname, errno, strerror(errno))));
    }

    m_fifo_fd = write_fd.release();
    ERS_INFO(std::format("Monitoring FIFO {} opened, fd={}", fifoname, m_fifo_fd));
}

FIFOWriter::~FIFOWriter() {
    if (m_fifo_fd >= 0) {
        close(m_fifo_fd);
    }
}

void FIFOWriter::write_message(const nlohmann::json& message) {
    if (m_fifo_fd < 0) {
        return;
    }

    const std::string msg = std::format("{}\n", message.dump());
    std::string_view remaining{msg};

    while (!remaining.empty()) {
        const ssize_t result = write(m_fifo_fd, remaining.data(), remaining.size());
        if (result < 0) {
            if (errno == EINTR) continue; // interrupted by signal — retry
            if (errno != EPIPE) {         // EPIPE means the FIFO reader is gone
                ers::warning(monitoring_log::fifo_issue(
                    std::format("Error writing to FIFO. Code {}: {}", errno, strerror(errno))));
            }
            break;
        }
        remaining.remove_prefix(static_cast<size_t>(result));
    }
}

void FIFOWriter::validate_fifo(const std::string& fifoname) {
    if (fifoname.empty()) {
        throw monitoring_log::fifo_issue("FIFO path is empty");
    }

    // Check if file exists and is a FIFO
    struct stat file_stat{};
    if (stat(fifoname.c_str(), &file_stat) == -1) {
        throw monitoring_log::fifo_issue(
            std::format("Cannot stat FIFO {}, error {}: {}", fifoname, errno, strerror(errno)));
    }

    if (not S_ISFIFO(file_stat.st_mode)) {
        throw monitoring_log::fifo_issue(
            std::format("{} exists but is not a UNIX FIFO", fifoname));
    }
}