Parallel sfind per server

#include <unistd.h>

#include <vector>

#include <cstdlib>   // For getenv

#include <pthread.h> // Include Pthreads header

using namespace std;

// (struct dirent_extended, pfind_options_t, pfind_runtime_options_t are the

// same)

// (structs and parsing functions are identical to your provided code)

#pragma region structs

/* Minimal struct needed for io500 find */

struct dirent_extended {

#pragma endregion

static pfind_runtime_options_t runtime;

int pfind_rank = 0;

int pfind_size = 1;

static pfind_options_t* opt;

[[noreturn]] void

pfind_abort(const string& str) {

    cerr << "ERROR [Rank " << pfind_rank << "]: " << str << endl;

    cerr << "ERROR: " << str << endl;

    exit(1);

}

// (pfind_print_help and pfind_parse_args are identical to the original)

// (pfind_print_help and pfind_parse_args are identical to your provided code)

#pragma region parsing

static void

pfind_print_help(const pfind_options_t* res) {

                break;

        }

    }

    if(pfind_rank == 0 && print_help) {

        pfind_print_help(res);

    }

    if(print_help) {

        pfind_print_help(res);

        exit(0);

    }

    if(res->workdir.empty()) {

}

#pragma endregion

void

dirProcess(const string& path, unsigned long long& checked,

           unsigned long long& found, const pfind_options_t* opt) {

// Structure to pass data to and from each worker thread

struct ThreadData {

    int thread_id;

    const pfind_options_t* opt;

    const string* workdir;

    queue<int>* server_queue;

    pthread_mutex_t* queue_mutex;

    // Thread-local results to avoid race conditions

    unsigned long long local_found = 0;

    unsigned long long local_checked = 0;

};

// The function each worker thread will execute

void*

worker_routine(void* arg) {

    ThreadData* data = static_cast<ThreadData*>(arg);

    while(true) {

        int server_id = -1;

        // Safely get a server ID from the shared queue

        pthread_mutex_lock(data->queue_mutex);

        if(!data->server_queue->empty()) {

            server_id = data->server_queue->front();

            data->server_queue->pop();

        }

        pthread_mutex_unlock(data->queue_mutex);

        // If the queue was empty, our work is done

        if(server_id == -1) {

            break;

        }

        struct dirent_extended* entries = nullptr;

        long n = gkfs_getsingleserverdir(data->workdir->c_str(), &entries,

                                         server_id);

    // --- PARALLELIZATION LOGIC ---

    // Each process calculates its own range of servers to query.

    int servers_per_proc = opt->num_servers / pfind_size;

    int remainder = opt->num_servers % pfind_size;

    int start_server =

            pfind_rank * servers_per_proc + min(pfind_rank, remainder);

    int end_server =

            start_server + servers_per_proc + (pfind_rank < remainder ? 1 : 0);

    // Each process loops ONLY over its assigned servers

    for(int server = start_server; server < end_server; server++) {

        long unsigned int n = gkfs_getsingleserverdir(

                path.c_str(),

               &entries, server);

        if(n <= 0)

        if(n <= 0) { // Handle error or empty directory from server

            if(n < 0) {

                cerr << "Warning: Thread " << data->thread_id

                     << " received error from server " << server_id << endl;

            }

            if(entries)

                free(entries); // Ensure we free even on error if allocated

            continue;

        }

        // --- Success! We can now iterate through the results ---

        // Iterate through the results from this server

        char* ptr = reinterpret_cast<char*>(entries);

        int bytes_processed = 0;

        while(bytes_processed < n) {

            struct dirent_extended* temp = reinterpret_cast<struct dirent_extended*>(ptr);

            if(strlen(temp->d_name) == 0 || temp->d_reclen == 0)

                break;

            if(temp->d_type != 1) {

                bool timeOK = opt->timestamp_file.empty() ||

                              ((uint64_t) temp->ctime >= runtime.ctime_min);

                bool sizeOK =

                        (opt->size == std::numeric_limits<uint64_t>::max() ||

                         temp->size == opt->size);

                bool nameOK = opt->name_pattern.empty() ||

                              regex_search(temp->d_name, opt->name_regex);

            struct dirent_extended* temp =

                    reinterpret_cast<struct dirent_extended*>(ptr);

            if(temp->d_reclen == 0)

                break; // Safety break

                if(timeOK && sizeOK && nameOK)

                    found++;

                checked++;

            if(temp->d_type != 1) { // Process files, not directories

                bool timeOK = data->opt->timestamp_file.empty() ||

                              ((uint64_t) temp->ctime >= runtime.ctime_min);

                bool sizeOK = (data->opt->size ==

                                       std::numeric_limits<uint64_t>::max() ||

                               temp->size == data->opt->size);

                bool nameOK = data->opt->name_pattern.empty() ||

                              regex_search(temp->d_name, data->opt->name_regex);

                if(timeOK && sizeOK && nameOK) {

                    data->local_found++;

                }

                data->local_checked++;

            }

            bytes_processed += temp->d_reclen;

            ptr += temp->d_reclen;

        }

        free(entries); // Free the memory allocated by gkfs_getsingleserverdir

    }

    free(entries);

    return nullptr;

}

int

process_sequential_parallel(const pfind_options_t* opt) {

    unsigned long long local_found = 0;

    unsigned long long local_checked = 0;

process_parallel_pthreads(const pfind_options_t* opt) {

    unsigned long long global_found = 0;

    unsigned long long global_checked = 0;

    runtime = {};

    if(!opt->timestamp_file.empty()) {

    if(workdir.empty())

        workdir = "/";

    dirProcess(workdir, local_checked, local_found, opt);

    // --- Pthreads Setup ---

    int num_threads = 20; // Default number of parallel requests

    const char* env_threads = getenv("GFIND_NUM_THREADS");

    if(env_threads) {

        try {

            num_threads = stoi(env_threads);

        } catch(...) { /* ignore, use default */

        }

    }

    // Don't create more threads than there are servers to query

    num_threads = min(num_threads, opt->num_servers);

    cout << "Using " << num_threads << " threads to query " << opt->num_servers

         << " servers." << endl;

    // --- WRITE LOCAL RESULTS TO A UNIQUE FILE ---

    string output_filename =

            "gfind_results.rank-" + to_string(pfind_rank) + ".txt";

    ofstream output_file(output_filename);

    if(!output_file.is_open()) {

        pfind_abort("Failed to open output file: " + output_filename);

    // 1. Create the shared work queue

    queue<int> server_queue;

    for(int i = 0; i < opt->num_servers; ++i) {

        server_queue.push(i);

    }

    output_file << "MATCHED " << local_found << "/" << local_checked << endl;

    output_file.close();

    cout << "[Rank " << pfind_rank << "] Finished. Wrote results to "

         << output_filename << endl;

    cout << "MATCHED " << local_found << "/" << local_checked << endl;

    return 0;

    // 2. Initialize the mutex

    pthread_mutex_t queue_mutex;

    pthread_mutex_init(&queue_mutex, nullptr);

    // 3. Create thread data structures

    vector<pthread_t> threads(num_threads);

    vector<ThreadData> thread_args(num_threads);

    // 4. Launch worker threads

    for(int i = 0; i < num_threads; ++i) {

        thread_args[i] = {.thread_id = i,

                          .opt = opt,

                          .workdir = &workdir,

                          .server_queue = &server_queue,

                          .queue_mutex = &queue_mutex};

        pthread_create(&threads[i], nullptr, worker_routine, &thread_args[i]);

    }

int

main(int argc, char** argv) {

    // --- Get Rank/Size from Environment Variables ---

    const char* env_rank = getenv("SLURM_PROCID");

    const char* env_size = getenv("SLURM_NPROCS");

    // 5. Wait for all threads to complete

    for(int i = 0; i < num_threads; ++i) {

        pthread_join(threads[i], nullptr);

    }

    if(env_rank && env_size) {

        try {

            pfind_rank = stoi(env_rank);

            pfind_size = stoi(env_size);

        } catch(const std::exception& e) {

            cerr << "Could not parse SLURM environment variables: " << e.what()

                 << endl;

            // Fallback to sequential

            pfind_rank = 0;

            pfind_size = 1;

    // 6. Aggregate results from all threads

    for(int i = 0; i < num_threads; ++i) {

        global_found += thread_args[i].local_found;

        global_checked += thread_args[i].local_checked;

    }

    } else {

        // Fallback for running outside of srun

        cout << "SLURM variables not found. Running in sequential mode (rank 0 of 1)."

             << endl;

        pfind_rank = 0;

        pfind_size = 1;

    // 7. Clean up

    pthread_mutex_destroy(&queue_mutex);

    cout << "MATCHED " << global_found << "/" << global_checked << endl;

    return 0;

}

    // Each process parses its own arguments. This is simple and correct.

int

main(int argc, char** argv) {

    opt = pfind_parse_args(argc, argv, false);

    // Check if GekkoFS function is available

    if(gkfs_getsingleserverdir == nullptr) {

        pfind_abort(

                "GekkoFS functions not available. Is the library preloaded?");

    }

    int result = process_sequential_parallel(opt);

    int result = process_parallel_pthreads(opt);

    delete opt;

    return result;