inotify_watch.cpp

Go to the documentation of this file.

/*
  ewmscp: copy program with extended functionality
    Copyright (C) 2018  Juergen Hannappel
 
    This program is free software: you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
    the Free Software Foundation, either version 3 of the License, or
    (at your option) any later version.
 
    This program is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    GNU General Public License for more details.
 
    You should have received a copy of the GNU General Public License
    along with this program.  If not, see <https://www.gnu.org/licenses/>.
*/
 
#include <chrono>
#include <csignal>
#include <dirent.h>
#include <fstream>
#include <iostream>
#include <list>
#include <map>
#include <poll.h>
#include <regex>
#include <atomic>
#include <scoped.h>
#include <string.h>
#include <sys/inotify.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <syslogstream.h>
#include <throwcall.h>
#include <unistd.h>
 
#include <Options.h>
#include <OptionsRegex.h>
typedef std::chrono::system_clock clock_type;
 
std::ostream* errStream(nullptr);
options::single<std::string> errPrefix('\0', "errPrefix", "prefix for error messages");
 
 
class maskName {
    const uint32_t bits;
    const std::string name;
    maskName(uint32_t aBits,const std::string& aName): bits(aBits), name(aName){};
    static const std::array<maskName,16> maskNames() {
        static const std::array<maskName,16> maskNamesArray {{{IN_ACCESS, "IN_ACCESS"},
                                                              {IN_ATTRIB, "IN_ATTRIB"},
                                                              {IN_CLOSE_WRITE, "IN_CLOSE_WRITE"},
                                                              {IN_CLOSE_NOWRITE, "IN_CLOSE_NOWRITE"},
                                                              {IN_CREATE, "IN_CREATE"},
                                                              {IN_DELETE, "IN_DELETE"},
                                                              {IN_DELETE_SELF, "IN_DELETE_SELF"},
                                                              {IN_MODIFY, "IN_MODIFY"},
                                                              {IN_MOVE_SELF, "IN_MOVE_SELF"},
                                                              {IN_MOVED_FROM, "IN_MOVED_FROM"},
                                                              {IN_MOVED_TO, "IN_MOVED_TO"},
                                                              {IN_OPEN, "IN_OPEN"},
                                                              {IN_IGNORED, "IN_IGNORED"},
                                                              {IN_ISDIR, "IN_ISDIR"},
                                                              {IN_Q_OVERFLOW, "IN_Q_OVERFLOW"},
                                                              {IN_UNMOUNT, "IN_UNMOUNT"}
            }};
        return(maskNamesArray);
    }
public:
    static void print(std::ostream& out, uint32_t mask) {
        bool needComma = false;
 
        for (const auto &item : maskNames()) {
            if (mask & item.bits) {
                if (needComma) {
                    out << ",";
                }
                out << item.name;
                needComma = true;
            }
        }
    };
    static uint32_t maskFromList(const options::container<std::string>& list) {
        uint32_t mask = 0;
        for (const auto& listItem : list) {
            for (const auto& item : maskNames()) {
                if (item.name == listItem) {
                    mask |= item.bits;
                }
            }
        }
        return mask;
    };
    static void addToRange(options::container<std::string>& opt) {
        for (const auto& item : maskNames()) {
            opt.fAddToRange(item.name);
        }
    };
};
 
 
class inotify_event_wrapper {
    inotify_event data;
  public:
    inotify_event_wrapper() = delete; // we only get "objects" of this type by reinterpret_casts
    ~inotify_event_wrapper() = delete;
    inline int wd() const {
        return data.wd;
    }
    inline uint32_t mask() const {
        return data.mask;
    }
    inline uint32_t cookie() const {
        return data.cookie;
    }
    inline uint32_t len() const {
        return data.len;
    }
    inline const char *name() const {
        return data.name;
    }
    const inotify_event_wrapper* next() const {
        return reinterpret_cast<const inotify_event_wrapper*>(&data.name[0] + len());
    }
    size_t size() const {
        return sizeof(inotify_event) + len();
    }
    static const inotify_event_wrapper* fromChar(char *pointer) {
        return reinterpret_cast<const inotify_event_wrapper*>(pointer);
    }
    friend std::ostream& operator<<(std::ostream& out, const inotify_event_wrapper& aEvent) {
        out << " mask: ";
        maskName::print(out, aEvent.mask());
        out << "(" << aEvent.mask() << ")";
        return out;
    }
};
 
class inotify_event_copy {
  protected:
    static std::map<uint32_t, inotify_event_copy>& getRenameEvents() {
        static std::remove_reference<decltype(getRenameEvents())>::type renameEvents;
        return renameEvents;
    }
    static bool cmp(const inotify_event_copy* lhs, const inotify_event_copy* rhs) {
        return lhs->then < rhs->then;
    }
    static std::multiset<inotify_event_copy*, decltype(cmp)*>& getSingleEvents() {
        static std::remove_reference<decltype(getSingleEvents())>::type singleEvents(&cmp);
        return singleEvents;
    }
    std::remove_reference<decltype(getSingleEvents())>::type::iterator singleEventIterator; 
    const inotify_event_wrapper* event; 
    clock_type::time_point then; 
    bool isCopy; 
 
    void doCopyIfNeeded() {
        if (!isCopy) {
            auto theEvent = getEvent();
            void* newEvent = malloc(theEvent->size());
            memcpy(newEvent, theEvent, theEvent->size());
            event = reinterpret_cast<inotify_event_wrapper*>(newEvent);
            isCopy = true;
            singleEventIterator = getSingleEvents().emplace(this);
        }
    }
 
  public:
    inotify_event_copy(const inotify_event_wrapper* aEvent):
        singleEventIterator(getSingleEvents().end()), event(aEvent), isCopy(false) {
    }
    ~inotify_event_copy() {
        if (isCopy) {
            free(reinterpret_cast<void*>(const_cast<inotify_event_wrapper*>(event)));
        }
    }
    clock_type::time_point timeStamp() const {
        return then;   
    }
    const inotify_event_wrapper* getEvent() const {
        return event;
    }
 
    static void doCopyLeftovers() {
        for (auto& it : getRenameEvents()) {
            it.second.doCopyIfNeeded();
        }
    }
 
    static inotify_event_copy* getStaleSingles() {
        auto it = getSingleEvents().begin();
 
        if (it != getSingleEvents().end() &&
                (*it)->then < clock_type::now() - std::chrono::seconds(5)) {
            auto retval = *it;
            return retval;
        }
 
        return nullptr;
    }
    static void forget(inotify_event_copy* what) {
        if (what->singleEventIterator != getSingleEvents().end()) {
            getSingleEvents().erase(what->singleEventIterator);
        }
 
        getRenameEvents().erase(what->getEvent()->cookie());
    }
 
    static inotify_event_copy* getPartner(const inotify_event_wrapper* event,
                                          clock_type::time_point then) {
        auto result = getRenameEvents().emplace(event->cookie(), event);
 
        if (result.second) {
            result.first->second.then = then;
            return nullptr;
        } else {
            return &(result.first->second);
        }
    }
};
 
class watchedDirType {
    static std::map<int, watchedDirType*>& getWatchedDirs() {
        static std::remove_reference<decltype(getWatchedDirs())>::type watchedDirs;
        return watchedDirs;
    }
    static int extraEventMask;
 
    std::string name; 
    watchedDirType *parent;
    int fd; 
    int id; 
    std::map<std::string, watchedDirType*> children;
    decltype(getWatchedDirs().begin()) watchIterator; 
    decltype(children.begin()) childIterator; 
 
 
    void becomeChildOfMyParent(watchedDirType* aParent) {
        parent = aParent;
 
        if (parent) {
            auto result = parent->children.emplace(name, this);
 
            if (result.second == false) {
                result.first->second->parent = nullptr; // orphanize old child
                result.first->second = this; // overwrite old child
            }
 
            childIterator = result.first;
        }
    }
 
  public:
    static void dump(std::ostream& out) {
        out << errPrefix << "begin list of watched directories (" << getWatchedDirs().size() << "):\n";
        for (const auto& item: getWatchedDirs()) {
            std::string path;
            item.second->buildPath(path);
            out << errPrefix << item.first << ": " << path << "\n";
        }
        out << errPrefix << "end list of watched directories.\n";
    }
 
    
    void buildPath(std::string& path) {
        if (parent) {
            parent->buildPath(path);
            path += "/";
        }
 
        path += name;
    }
    void buildPath(std::string& path,
                   const std::string& aName) {
        buildPath(path);
        path += "/";
        path += aName;
    }
    std::string getPath() {
        std::string path;
        buildPath(path);
        return path;
    }
 
    static watchedDirType* get(int aId) {
        return getWatchedDirs()[aId];
    }
    static bool empty() {
        return getWatchedDirs().empty();
    }
    static size_t size() {
        return getWatchedDirs().size();
    }
 
    static void setExtraEventMask(int extraMask) {
        extraEventMask = extraMask;
    }
 
    watchedDirType(const std::string &aName,
                   int aFd,
                   watchedDirType* aParent = nullptr):
        name(aName),
        parent(aParent),
        fd(aFd) {
        std::string path;
        buildPath(path);
        watchIterator = getWatchedDirs().end();
        id = inotify_add_watch(fd, path.c_str(),
                               IN_CLOSE_WRITE
                               | IN_DELETE
                               | IN_MOVED_TO | IN_MOVED_FROM
                               | IN_CREATE
                               | IN_DELETE_SELF
                               | IN_EXCL_UNLINK
                               | IN_DONT_FOLLOW
                               | IN_ONLYDIR
                               | extraEventMask);
 
        if (id == -1 && (errno == ENOENT || errno == ENOTDIR)) { // we fell prey to a race
            parent = nullptr;
            return;
        }
 
        throwcall::badval(id, -1, "inotify_add_watch failed on ", path);
 
        auto result = getWatchedDirs().emplace(id, this);
 
        if (result.second == false) { // watch for this object already exists, declare this as invalid
            parent = nullptr;
            id = -1;
            return;
        } else {
            watchIterator = result.first;
        }
 
        becomeChildOfMyParent(aParent);
    }
 
    ~watchedDirType() {
        if (id != -1) {
            auto retval = inotify_rm_watch(fd, id);
 
            if (retval && errno == EINVAL) {
                *errStream << errPrefix << "oops " << getPath() << " id " << id << "could not be unwatched" << std::endl;
            } else {
                throwcall::good0(retval, "can't unwatch ", getPath());
            }
        }
 
        if (watchIterator != getWatchedDirs().end()) {
            getWatchedDirs().erase(watchIterator);
        }
 
        for (auto& child : children) {
            *errStream << errPrefix << "oops " << getPath() << " still has child " << child.second->getPath() << ", deleting it" << std::endl;
            delete child.second;
        }
 
        if (parent) {
            parent->children.erase(childIterator);
        }
    }
    void invalidateId() {
        id = -1;
    }
    int getId() const {
        return id;
    }
    watchedDirType* getChild(const std::string& aName) {
        auto it = children.find(aName);
 
        if (it == children.end()) {
            return nullptr;
        } else {
            return it->second;
        }
    }
    void getAdoptedBy(const std::string& newName,
                      watchedDirType *newParent) {
        if (parent) {
            parent->children.erase(childIterator);
        }
 
        name = newName;
        becomeChildOfMyParent(newParent);
    }
};
 
 
enum class watchDirectoryReturnType {
    vanished,
    islink,
    ignore,
    iswatched
};
watchDirectoryReturnType watchDirectory(int fd,
                                        const std::string& name,
                                        watchedDirType* parent = nullptr) {
    std::string path;
 
    if (parent) {
        parent->buildPath(path, name);
    } else {
        path = name;
    }
 
    {
        struct stat statbuf;
        auto retval = lstat(path.c_str(), &statbuf);
 
        if (retval && (errno == ENOENT || errno == ENOTDIR)) {
            return watchDirectoryReturnType::vanished; // happens for very short lived files
        }
 
        throwcall::good0(retval, "can't lstat ", path);
 
        if (!S_ISDIR(statbuf.st_mode)) {
            return S_ISLNK(statbuf.st_mode) ? watchDirectoryReturnType::islink : watchDirectoryReturnType::ignore;
        }
    }
 
    auto watchedDir = new watchedDirType(name, fd, parent);
 
    if (watchedDir->getId() == -1) { // vanmished or already watched
        delete watchedDir;
        return  watchDirectoryReturnType::vanished;
    }
 
    // with the subdir list we have only one open dir at a time
    std::list<std::string> subdirs;
    {
        auto result = opendir(path.c_str());
 
        if (result == nullptr && (errno == ENOENT || errno == ENOTDIR)) {
            delete watchedDir;
            return  watchDirectoryReturnType::vanished; // dir vanished before we could check the content
        }
 
        scoped::Dir dir(path, result);
 
        while (auto entry = readdir(dir)) {
            if (entry->d_name[entry->d_name[0] != '.' ? 0 : entry->d_name[1] != '.' ? 1 : 2] == '\0') {
                continue; // skip . .. and empty strings
            }
            if (entry->d_type == DT_DIR) {
                subdirs.emplace_back(entry->d_name);
            }
        }
    }
 
    for (auto& subname : subdirs) {
        watchDirectory(fd, subname, watchedDir);
    }
 
    return watchDirectoryReturnType::iswatched;
}
int watchedDirType::extraEventMask = 0;
 
 
std::ostream& operator<< (std::ostream& stream,
                          std::chrono::system_clock::time_point t) {
    stream << std::chrono::duration_cast<std::chrono::duration<double>>(t.time_since_epoch()).count();
    return stream;
}
 
 
 
 
static options::single<std::string> pidFileName('p', "pidFile",
        "name of PID file");
 
static void pidFileRemover() {
    if (pidFileName != "") {
        throwcall::good0(unlink(pidFileName.c_str()), "can't unlink pid file ", pidFileName);
    }
}
static void sigTermHandler(int) {
    std::exit(EXIT_SUCCESS);
}
static void sigPipeHandler(int) {
    std::exit(EXIT_FAILURE);
}
static std::atomic<bool> dumpRequested(false);
static void sigUsrHandler(int) {
    dumpRequested = true;
}
 
class commandSender {
    static options::single<bool> nullDelimiter;
    static options::single<unsigned> keepaliveTimeout;
    char delimiter;
    clock_type::time_point lastSendTime;
    bool needFlush;
  public:
    commandSender() {
        delimiter = nullDelimiter ? '\0' : '\n';
        std::cout << std::fixed; // proper format for time stamps
    }
    void sendCommand(const char *command, std::chrono::system_clock::time_point when,
                     const std::string& path) {
        std::cout << command << ' ' << when << ' ' << path << delimiter;
        lastSendTime = when;
        needFlush = true;
    }
    void sendCommand(const char *command, std::chrono::system_clock::time_point when,
                     const std::string& path1, const std::string& path2) {
        sendCommand(command, when, path1);
        std::cout << path2 << delimiter;
    }
    void flush() {
        std::cout.flush();
    }
    void keepalive() {
        auto now = clock_type::now();
        if (now - lastSendTime > std::chrono::seconds(keepaliveTimeout)) {
            sendCommand("ig", now, "keepalive");
            flush();
        }
    }
    class flushDomain {
        commandSender& sender;
      public:
        flushDomain(commandSender& aSender): sender(aSender) {
            sender.needFlush = false;
        };
        ~flushDomain() {
            if (sender.needFlush) {
                sender.flush();
            }
        }
    };
};
decltype(commandSender::nullDelimiter) commandSender::nullDelimiter('0', "null",
        "use NUL as delimiter between requests", false);
decltype(commandSender::keepaliveTimeout) commandSender::keepaliveTimeout('\0', "keepalive",
        "keepalive timeout (seconds)", 120);
 
 
int main(int argc, const char* argv[]) {
    std::ios::sync_with_stdio(false);
    options::parser parser("inotify_watch", "", {});
    options::positional<options::container<std::string>>directories(10, "directory",
            "directories to be watched");
    options::single<std::string> workDir('\0', "workDir",
                                         "use DIRECTORY as work dir");
    options::single<std::regex> ignore('i', "ignore",
                                       "regexp for files to be ignored for IN_CLOSE_WRITE");
    options::single<bool> verbose('v', "verbose",
                                  "be verbose", false);
    options::single<std::string> errorStream('\0', "errors",
            "file to write errors to");
 
    options::container<std::string> extraEvents('e', "extraEvents",
            "extra events to listen to");
    maskName::addToRange(extraEvents);
    options::container<std::string> copyTriggers('c', "copyTrigger",
            "extra events beside IN_CLOSE_NOWRITE hat trigger a copy");
    maskName::addToRange(copyTriggers);
    parser.fParse(argc, argv);
 
    logstream::provider errStreamBase(errorStream, std::cerr);
    errStream = &(errStreamBase.getStream());
 
    watchedDirType::setExtraEventMask(maskName::maskFromList(extraEvents));
 
    uint32_t copyTriggerBits = IN_CLOSE_WRITE | maskName::maskFromList(copyTriggers);
    if (verbose) {
        std::cerr << "events of type ";
        maskName::print(std::cerr, copyTriggerBits);
        std::cerr << " (" << std::hex << copyTriggerBits << std::dec << ") will trigger a copy event\n";
    }
    try {
 
        if (pidFileName != "") {
            std::ofstream pidFile(pidFileName);
            pidFile << getpid() << "\n";
            throwcall::good0(std::atexit(pidFileRemover), "can't register pidFileRemover");
            std::signal(SIGTERM, sigTermHandler);
        }
 
        std::signal(SIGPIPE, sigPipeHandler);
        std::signal(SIGUSR1, sigUsrHandler);
 
        auto fd = throwcall::badval(inotify_init(), -1, "inotify_init");
 
        if (workDir.fIsSet()) {
            throwcall::good0(chdir(workDir.c_str()), "can't change work dir to ", workDir);
        }
 
        for (const auto& directory : directories) {
            watchDirectory(fd, directory);
        }
 
        *errStream << errPrefix << "watching " << watchedDirType::size() << " directories" << std::endl;
 
        *errStream << std::fixed;
        char buffer[4096 * 256] __attribute__ ((aligned(__alignof__(struct inotify_event))));
        struct pollfd pfd;
        pfd.fd = fd;
        pfd.events = POLLIN;
 
        commandSender sender;
 
        while (true) {
            if (dumpRequested) {
                dumpRequested = false;
                watchedDirType::dump(*errStream);
            }
            {
                {
                    // process single events that took forever to get a partner
                    commandSender::flushDomain flusher(sender);
                    while (auto copy = inotify_event_copy::getStaleSingles()) {
                        auto event = copy->getEvent();
                        std::string path;
                        watchedDirType::get(event->wd())->buildPath(path, event->name());
 
                        if (event->mask() & IN_MOVED_TO) {
                            auto result = watchDirectory(fd, event->name(), watchedDirType::get(event->wd()));
 
                            if (result == watchDirectoryReturnType::islink) {
                                sender.sendCommand("ln", copy->timeStamp(), path);
                            } else if (result == watchDirectoryReturnType::ignore) {
                                sender.sendCommand("cp", copy->timeStamp(), path);
                            }
 
                            if (verbose && result == watchDirectoryReturnType::iswatched) {
                                *errStream << errPrefix << "watching " << watchedDirType::size() << " directories (added subtree)" << std::endl;
                            }
                        } else {
                            auto parent = watchedDirType::get(event->wd());
                            auto meAsDirChild = parent->getChild(event->name());
 
                            if (meAsDirChild) {
                                delete meAsDirChild;
                            }
 
                            sender.sendCommand("rm", copy->timeStamp(), path);
                        }
 
                        inotify_event_copy::forget(copy);
                    }
                }
                auto result = poll(&pfd, 1, 1000);
 
                if (result == -1) {
                    if (errno == EINTR) {
                        continue;
                    }
 
                    throwcall::badval(result, -1, "poll failed");
                }
 
                if (result == 0) { // timeout, check continued presence of initial dirs
                    for (const auto& directory : directories) {
                        struct stat statbuf;
                        auto retval = stat(directory.c_str(), &statbuf);
 
                        if (retval != 0 && errno == ENOENT) {
                            return 0;
                        } else {
                            throwcall::good0(retval, "can't stat ", directory);
                        }
                    }
                    sender.keepalive();
                    continue;
                }
            }
            auto bytes = throwcall::badval(read(fd, buffer, sizeof(buffer)),
                                           -1, "read inotification");
            auto now = clock_type::now();
 
            if (verbose) {
                *errStream << errPrefix << now << " got " << bytes << " bytes of notification" << std::endl;
            }
 
            commandSender::flushDomain flusher(sender);
 
            for (auto event = inotify_event_wrapper::fromChar(buffer);
                    event < inotify_event_wrapper::fromChar(buffer + bytes);
                    event = event->next()) {
                if (verbose) {
                    *errStream << errPrefix << "watch id " << event->wd() << " @ " << (void*)event << " with name " << event->name() << *event << std::endl;
                }
 
                if (event->mask() & IN_IGNORED) {
                    continue;
                }
 
                if (event->mask() & IN_Q_OVERFLOW) {
                    *errStream << errPrefix << "watch queue overflow" << std::endl;
                    continue; // we cannot do more than lopg this
                }
 
                std::string path;
                auto watchedDir = watchedDirType::get(event->wd());
 
                if (watchedDir == nullptr) {
                    *errStream << errPrefix << "unknown watch id " << event->wd() << " @ " << (void*)event << " with name " << event->name() << *event << std::endl;
                    continue;
                }
 
                watchedDir->buildPath(path, event->name());
 
                if (event->mask() & (IN_MOVED_TO | IN_MOVED_FROM)) {
                    auto partner = inotify_event_copy::getPartner(event, now);
 
                    if (partner) { // we have a partner event
                        std::string partnerPath;
                        watchedDirType::get(partner->getEvent()->wd())->buildPath(partnerPath, partner->getEvent()->name());
                        const inotify_event_wrapper* from;
                        const inotify_event_wrapper* to;
                        const std::string* fromPath;
                        const std::string* toPath;
 
                        if (event->mask() & IN_MOVED_TO) {
                            from = partner->getEvent();
                            to = event;
                            fromPath = &partnerPath;
                            toPath = &path;
                        } else {
                            from = event;
                            to = partner->getEvent();
                            fromPath = &path;
                            toPath = &partnerPath;
                        }
 
                        auto fromAsDirChild = watchedDirType::get(from->wd())->getChild(from->name());
 
                        if (fromAsDirChild) {
                            fromAsDirChild->getAdoptedBy(to->name(), watchedDirType::get(to->wd()));
                        }
 
                        sender.sendCommand("mv", now, *fromPath, *toPath);
                        inotify_event_copy::forget(partner);
                    }
                }
 
                if (event->mask() & copyTriggerBits) {
                    if (!(ignore.fIsSet() && regex_match(path, ignore))) {
                        sender.sendCommand("cp", now, path);
                    }
                }
 
                if (event->mask() & IN_CREATE) { // watch new directories also
                    auto result = watchDirectory(fd, event->name(), watchedDir);
 
                    if (result == watchDirectoryReturnType::islink) {
                        sender.sendCommand("ln", now, path);
                    }
 
                    if (verbose && result == watchDirectoryReturnType::iswatched) {
                        *errStream << errPrefix << "watching " << watchedDirType::size() << " directories (added subtree)" << std::endl;
                    }
                }
 
                if (event->mask() & IN_DELETE_SELF) { // a watched directory is deleted
                    watchedDir->invalidateId();
                    delete watchedDir;
 
                    if (verbose) {
                        *errStream << errPrefix << "watching " << watchedDirType::size() << " directories (removed subtree)" << std::endl;
                    }
                }
 
                if (event->mask() & IN_DELETE) {
                    if (!(ignore.fIsSet() && regex_match(path, ignore))) {
                        sender.sendCommand("rm", now, path);
                    }
                }
            }
            // make copies of all leftover half move events befor re-using buffer
            inotify_event_copy::doCopyLeftovers();
 
            if (watchedDirType::empty()) {
                throwcall::good0(close(fd), "can't close inotify fd");
 
                *errStream << errPrefix << "no more stuff to watch, finishing...\n";
 
                return EXIT_SUCCESS;
            }
        }
    } catch (const std::exception& e) {
        *errStream << logstream::level::crit << errPrefix << " ended badly: " << e.what() << std::endl;
        return EXIT_FAILURE;
    }
 
    return EXIT_SUCCESS;
}