Processing all files in directory

Question

DO NOT RUN THIS EXAMPLE BLINDLY, IT REMOVES ALL FILES UNDER argv[1]

Hello,

This little programm should read all files from a given directory and do some work with them. New files are constantly written into the directory by an outside process. For the duration of the processing, the file should stay in the input directory as it will be processed line by line and not getting loaded into memory all at once. This is also the reason for the process_list member so files don't get processed twice. num_threads will be a runtime variable. For simplicity I hardcoded it now, but that would be the reason for the dynamic allocation. There are some more simplifications, like omitting error checks, checking if the path points to a directory or file, etc, but I think it illustrates all potential issues I could think of.

I'd like feedback on the overall approach, if this is a good way to solve the task or if I could do better. Have I forgotten some potential issues (e.g. while writing this I came across iterator invalidation)? Also, any other best practices that I could apply here are also very welcome. Thanks.

#include <chrono>
#include <csignal>
#include <cstdio>
#include <mutex>
#include <filesystem>
#include <thread>
#include <vector>

volatile std::sig_atomic_t keep_running = 1;

static void signal_handler(int sig)
{
    std::fprintf(stderr, "Received signal [%d]. Terminating...\n", sig);
    keep_running = false;
}

namespace fs = std::filesystem;

class DirectoryView {
  private:
    fs::path dir;
    fs::directory_iterator it;
    std::mutex mtx;
    fs::path *process_list;
    static inline fs::directory_iterator end_it;
    int process_list_size;

    void reload(void) { it = fs::directory_iterator(dir); }

    bool is_in_processing(const fs::path& ret)
    {
        for (int i = 0; i != process_list_size; ++i) {
            if (process_list[i] == ret)
                return true;
        }
        return false;
    }

  public:
    DirectoryView(const char *p, const int num_threads) : 
            dir(p), it(p), process_list_size(num_threads)
    {
        process_list = new fs::path[num_threads];
    }

    ~DirectoryView(void)
    {
        delete[] process_list;
    }


    const fs::path next(const int i)
    {
        std::scoped_lock<std::mutex> lock(mtx);

        for (; it != end_it; ++it) {
            const fs::path ret(it->path());
            if (is_in_processing(ret))
                continue;
            else {
                process_list[i] = ret;
                ++it; // not nice, but prepare for next call to next() so it points to a new directory entry (loop increment is not executed here)
                return ret;
            }
        }

        reload();
        return fs::path(); // Possible improvement: try running next() once more, it is likely there is a new file in the directory for this thread to process
    }

    void finished_processing(int i)
    {
        process_list[i].clear();
    }
};

static void process_file(const fs::path& file, const int id)
{
    std::printf("Thread [%d]: Processing file [%s]\n", id, file.string().c_str());
    //std::this_thread::sleep_for(std::chrono::seconds(1)); // to illustrate more complex (= longer) processing
    fs::remove(file);
}

static void worker(const int id, DirectoryView& dv)
{
    while (keep_running) {
        const fs::path next = dv.next(id);
        if (next.empty()) {
            // FIXME wake up when new files in directory?
            std::this_thread::sleep_for(std::chrono::seconds(10));
        } else {
            process_file(next, id);
            dv.finished_processing(id);
        }
    }
}

int main(int argc, char **argv)
{
    if (argc < 2)
        return 1;

    std::signal(SIGINT, signal_handler);

    const int num_threads = 4;

    DirectoryView dv(argv[1], num_threads);

    std::vector<std::thread> workers;
    for (int i = 0; i < num_threads; ++i) {
        workers.push_back(std::thread(worker, i, std::ref(dv)));
    }

    for (auto& w : workers)
        w.join();

    return 0;
}

Answer 1

In all, it's a good effort. Here are some things that may help you improve your program.

Choose appropriate variable types

The keep_running flag is a std::sig_atomic_t but since you're using C++17 anyway (for std::filesystem), I'd recommend making that an atomic_bool which more clearly reflects its usage and intent.

Use const where practical

The is_in_processing function doesn't (and shouldn't) alter the underlying DirectoryView object, so it should be declared as a const function like this:

bool is_in_processing(const fs::path& ret) const;

Understand differences between C and C++

I don't know if you come from C or just wanted to be explicit, but this function:

void reload(void) { it = fs::directory_iterator(dir); }

could also be declared as

void reload() { it = fs::directory_iterator(dir); }

Unlike C, when the argument list is empty in C++, it means the same as void.

Use the standard library

The code currently contains this function:

bool is_in_processing(const fs::path& ret) 
{
    for (int i = 0; i != process_list_size; ++i) {
        if (process_list[i] == ret)
            return true;
    }
    return false;
}

Adding const as mentioned above, this could be very succinctly rewritten using std::any_of

bool is_in_processing(const fs::path& ret) const {
    return std::any_of(&process_list[0], &process_list[process_list_size], [ret](const fs::path &p){ 
            return p == ret; });
}

Prefer lock_guard to scoped_lock for single mutex

The essential difference between scoped_lock and lock_guard is that scoped_lock handles the coordination of multiple mutexes. Since you've only got one, it makes sense to use the somewhat simpler lock_guard instead.

Rethink the architecture

The current architecture uses multiple threads to process the same directory, but the descriptions seems to imply a desire to process a single directory using multiple threads. The difference is that the current design has multiple threads all processing every file with the result that if we have four threads, we could have the same file processed four or more times. A different design could split up a single directory of files among multiple threads and have only one thread process each file so that each file is only processed once. One way to do that is to have a DirWatcher looking for changes in the directory and then dispatching DirWorkers (possibly in multiple threads) to do the work. Alternatively, have them communicate via a thread-safe work queue and let each DirWorker independently grab an item and process it.