Yehor/FineWebUA-links-p1-500m

// gcc -O3 -pthread fast_counter.c -o fast_counter

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <pthread.h>

#define HASH_SIZE 30000023  // Bumped higher for 7GB files to avoid collisions
#define MAX_LINE 8192

typedef struct Node {
    char *str;
    struct Node *next;
} Node;

typedef struct {
    char *filename;
    Node **domains_hash;
    Node **sitemaps_hash;
    long long unique_domains;
    long long unique_sitemaps;
} ThreadData;

// DJB2 Hash Function
unsigned long hash(const char *str) {
    unsigned long hash = 5381;
    int c;
    while ((c = (unsigned char)*str++)) {
        hash = ((hash << 5) + hash) + c;
    }
    return hash % HASH_SIZE;
}

// Thread-safe isolation insertion
int insert(Node **table, const char *str) {
    unsigned long h = hash(str);
    Node *curr = table[h];
    while (curr != NULL) {
        if (strcmp(curr->str, str) == 0) {
            return 0;
        }
        curr = curr->next;
    }
    Node *new_node = malloc(sizeof(Node));
    if (!new_node) {
        perror("Allocation failed");
        exit(EXIT_FAILURE);
    }
    new_node->str = strdup(str);
    new_node->next = table[h];
    table[h] = new_node;
    return 1;
}

// Fixed: Safely extracts domain and returns 1 on success, 0 on failure
int extract_domain(const char *url, char *domain) {
    if (strncmp(url, "http://", 7) == 0) {
        url += 7;
    } else if (strncmp(url, "https://", 8) == 0) {
        url += 8;
    } else {
        return 0;
    }
    char *start = domain;
    while (*url && *url != '/' && *url != ':' && *url != '\t' && *url != ' ' && *url != '\n' && *url != '\r') {
        *domain++ = *url++;
    }
    *domain = '\0';
    return (domain > start); // Returns 1 if we actually wrote characters
}

void *process_file(void *arg) {
    ThreadData *data = (ThreadData *)arg;

    data->domains_hash = calloc(HASH_SIZE, sizeof(Node *));
    data->sitemaps_hash = calloc(HASH_SIZE, sizeof(Node *));
    if (!data->domains_hash || !data->sitemaps_hash) {
        perror("Thread table allocation failed");
        exit(EXIT_FAILURE);
    }

    FILE *file = fopen(data->filename, "r");
    if (!file) {
        fprintf(stderr, "Error opening file: %s\n", data->filename);
        return NULL;
    }

    char line[MAX_LINE];
    char domain[MAX_LINE];
    char sub_sitemap[MAX_LINE];

    while (fgets(line, sizeof(line), file)) {
        // Fast custom TSV scanning instead of destructive strtok
        char *col1 = line;
        char *tab1 = strchr(col1, '\t');
        if (!tab1) continue;

        char *col2 = tab1 + 1;
        char *tab2 = strchr(col2, '\t');
        if (!tab2) continue;

        // Extract sub_sitemap column safely without altering original line stream
        size_t sitemap_len = tab2 - col2;
        if (sitemap_len >= MAX_LINE) sitemap_len = MAX_LINE - 1;
        memcpy(sub_sitemap, col2, sitemap_len);
        sub_sitemap[sitemap_len] = '\0';

        // Add unique sitemap
        if (insert(data->sitemaps_hash, sub_sitemap)) {
            data->unique_sitemaps++;
        }

        // Extract and add unique domain from col1
        if (extract_domain(col1, domain)) {
            if (insert(data->domains_hash, domain)) {
                data->unique_domains++;
            }
        }
    }

    fclose(file);
    return NULL;
}

void free_table(Node **table) {
    for (int i = 0; i < HASH_SIZE; i++) {
        Node *curr = table[i];
        while (curr != NULL) {
            Node *temp = curr;
            curr = curr->next;
            free(temp->str);
            free(temp);
        }
    }
    free(table);
}

int main(int argc, char *argv[]) {
    if (argc < 2) {
        fprintf(stderr, "Usage: %s <file1.tsv> [file2.tsv ...]\n", argv);
        return EXIT_FAILURE;
    }

    int num_files = argc - 1;
    pthread_t *threads = malloc(num_files * sizeof(pthread_t));
    ThreadData *thread_data = malloc(num_files * sizeof(ThreadData));

    printf("Spawning %d threads for processing...\n", num_files);

    for (int i = 0; i < num_files; i++) {
        thread_data[i].filename = argv[i + 1];
        thread_data[i].unique_domains = 0;
        thread_data[i].unique_sitemaps = 0;
        pthread_create(&threads[i], NULL, process_file, &thread_data[i]);
    }

    for (int i = 0; i < num_files; i++) {
        pthread_join(threads[i], NULL);
    }

    printf("Merging results...\n");

    Node **final_domains = calloc(HASH_SIZE, sizeof(Node *));
    Node **final_sitemaps = calloc(HASH_SIZE, sizeof(Node *));
    long long total_unique_domains = 0;
    long long total_unique_sitemaps = 0;

    for (int i = 0; i < num_files; i++) {
        for (int h = 0; h < HASH_SIZE; h++) {
            Node *curr = thread_data[i].domains_hash[h];
            while (curr != NULL) {
                if (insert(final_domains, curr->str)) {
                    total_unique_domains++;
                }
                curr = curr->next;
            }
            curr = thread_data[i].sitemaps_hash[h];
            while (curr != NULL) {
                if (insert(final_sitemaps, curr->str)) {
                    total_unique_sitemaps++;
                }
                curr = curr->next;
            }
        }
        free_table(thread_data[i].domains_hash);
        free_table(thread_data[i].sitemaps_hash);
    }

    printf("\n=====================================\n");
    printf("FINAL VERIFIED REPORT\n");
    printf("=====================================\n");
    printf("Total Unique Domains:  %lld\n", total_unique_domains);
    printf("Total Unique Sitemaps: %lld\n", total_unique_sitemaps);
    printf("=====================================\n");

    free_table(final_domains);
    free_table(final_sitemaps);
    free(threads);
    free(thread_data);

    return EXIT_SUCCESS;
}