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;
}