1 Billion Row Challenge

The 1 Billion Row Challenge was introduced to the world at the beginning of 2024 as a way to learn how quickly we can process data. The premise is simple, you are given a file with 1 billion lines, each containing a weather station name, and a temperature recording, separated by a ; and finished with a \n.

Hamburg;12.0
Bulawayo;8.9
Palembang;38.8
St. John's;15.2
Cracow;12.6
Bridgetown;26.9
Istanbul;6.2
Roseau;34.4
Conakry;31.2
Istanbul;23.0

Your program then takes this data, organizes it alphabetically by station name, and finally outputs to STDOUT in the following format: <station_name>=<min>/<mean>/<max>, ....

For me, the most appealing part of this challenge is that the naive solution is extremely simple, but simple doesn’t cut it when we are dealing with an input file around 15GB in size. Nonetheless, we will start with a simple solution, and gradually evolve it as we go along.

The original 1BR challenge required solutions to be written in Java, but since then participants from many other languages have joined, though not officially. I thought that since the solution can be simple I would use this as an opportunity to improve my skills in C. The basic process will be to read the file line-by-line and enter each station into a hashmap. If the station already exists we will simply update its record. At the end we will sort the backing array alphabetically and print it all out.

I have previously written about a quadratic probing hashmap in C and that will make a great starting point for our solution.

With the setup work out of the way, we can finally look at the complete naive solution.

#include "hashmap/hashmap.h"
#include <assert.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

// Comparison function for qsort
int compare_pairs(const void* a, const void* b) {
    // Covered above
}

int main() {
    hashmap h = hashmap_init(&fnv_string_hash, &str_equals, 0.5, 17720);

    FILE* fin;
    char* line = NULL;
    size_t len = 0;
    ssize_t nread;

    fin = fopen("./data/measurements.txt", "r");

    while ((nread = getline(&line, &len, fin)) != -1) {
        char* copy = line;
        strtok(copy, ";");
        char* temp_s = strtok(NULL, "\n");
        float temp_f = strtof(temp_s, NULL);
        int temp_i = temp_f * 10;
        pair* p = hashmap_find(&h, copy);
        if (p) {
            p->value.count++;
            p->value.total += temp_i;
            if (temp_i > p->value.max) {
                p->value.max = temp_i;
            } else if (temp_i < p->value.min) {
                p->value.min = temp_i;
            }
        } else {
            hashmap_insert(&h, strdup(copy),
                           (Record){.name = copy,
                                    .total = temp_i,
                                    .count = 1,
                                    .max = temp_i,
                                    .min = temp_i});
        }
    }

    qsort(h.array, h.capacity, sizeof(pair), compare_pairs);

    for (size_t i = 0; i < h.capacity && h.array[i].key; i++) {
        printf("%s=%.1f/%.1f/%.1f, ", (char*)h.array[i].key,
               (float)h.array[i].value.min / 10,
               (float)h.array[i].value.total / 10 /
                   (float)h.array[i].value.count,
               (float)h.array[i].value.max / 10);
        free(h.array[i].key);
    }

    free(line);
    fclose(fin);
    hashmap_free(&h);
}

Let’s go over each part of the code in a bit more detail before we begin our optimizations. We start by initializing our hashmap with some sane default values. A load factor of 0.5 was determined to be optimal for a quadratic probing hashmap, and we know the size of our input so we can pre-allocate the array to 17720 elements (this may be considered cheating, but the constraints say nothing about pre-allocation).

From there we setup a simple getline loop that reads the file one line at a time. Within the loop we can split the line using strtok on the ; and \n characters respectively. Floating point numbers are parsed using strtof before being converted to an integer to save time in future computations.

The last step in this loop is to check if the station exists in the hashmap, updating it if it does, and creating it if it does not.

The loop ends when we hit EOF, and we can then sort the array. As stated before, we pass in the array itself, the size, the size of each element, and the sorting predicate.

Finally, we can rip through the array printing out the required data.

For iteration 2 were are going to change from a buffered file stream through fopen, to mmapping the entire file into our address space. This allows us to treat the file as one giant array, while the operating system manages the actual fetching of pages from disk. This change may seem simple, but it causes us to have to rewrite our parsing logic just slightly.

#include "hashmap/hashmap.h"
#include <assert.h>
#include <fcntl.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/mman.h>
#include <sys/stat.h>
#include <unistd.h>
// ....

int main() {
    hashmap h = hashmap_init(&fnv_string_hash, &str_equals, 0.5, 17720);

    char* file;
    int fd;
    struct stat sb;

    fd = open("./data/measurements.txt", O_RDONLY);

    fstat(fd, &sb);

    file = mmap(NULL, sb.st_size, PROT_READ, MAP_PRIVATE, fd, 0);

    size_t cursor = 0;
    char name[100];
    char temp[5];
    while (cursor < sb.st_size) {
        int name_pos = 0;
        int temp_pos = 0;
        while (file[cursor] != ';') {
            name[name_pos++] = file[cursor++];
        }
        name[name_pos] = '\0';
        cursor++;
        while (file[cursor] != '\n') {
            temp[temp_pos++] = file[cursor++];
        }
        temp[temp_pos] = '\0';
        cursor++;
        float temp_f = strtof(temp, NULL);
        int temp_i = temp_f * 10;
        pair* p = hashmap_find(&h, name);
        if (p) {
            p->value.count++;
            p->value.total += temp_i;
            if (temp_i > p->value.max) {
                p->value.max = temp_i;
            } else if (temp_i < p->value.min) {
                p->value.min = temp_i;
            }
        } else {
            hashmap_insert(&h, strdup(name),
                           (Record){.name = name,
                                    .total = temp_i,
                                    .count = 1,
                                    .max = temp_i,
                                    .min = temp_i});
        }
    }

    qsort(h.array, h.capacity, sizeof(pair), compare_pairs);
    // ....

Previously we used getline, which is a great function for small program, but can drastically slow down more complex ones. One reason why is that we end up doing twice the amount of work to read over a string, once to find the newline, and a second time to split the string up. Instead of relying on getline we can tune the parsing to be specific to our use case.

We create two buffers, one for the station name, and another for the temperature. According to the original problem statement we know the station names can be at most 100 characters, and the temperatures will be a most 5 characters (-55.5). From there we read along one character at a time adding it to the name array, which gets null terminated once we the hit ;. We then move on to the temperature repeating the exact same process until we come across a newline.

Do these changes net us any benefit? Let’s find out.

So far we have been converting parsed floats directly into integers to keep future computations off the FPU, which is much slower than the ALU. Because of the strict input format we can actually skip the floating-point conversion entirely.

// ....
        while (file[cursor] != '\n') {
            if (file[cursor] == '.') {
                cursor++;
            }
            temp[temp_pos++] = file[cursor++];
        }
        temp[temp_pos] = '\0';
        cursor++;
        int temp_i = atoi(temp);
// ....

We made one optimization that saves us quite a bit time. Instead of parsing 55.5 as a float and then multiplying by 10 we can simply drop the decimal and parse the number as an int. This results in the same value as if we multiplied the float by 10 before converting to an integer.

A change like that is easy to read and would probably net us a sizeable improvement for such little effort, but that’s not what this challenge is about. We want absolute speed at any cost, so we will write our own integer parsing function.

int parse_int(const char* c) {
    bool negative = (*c == '-');
    c += negative;

    char num[3] = {0};
    int num_pos = 0;
    while (*c != '\0') {
        num[num_pos] = (*c != '.')
                           ? *c
                           : num[num_pos]; // If c is '.', keep num[num_pos] the
                                           // same, otherwise assign *c to it
        num_pos += (*c != '.'); // Increment num_pos only if c is not '.'
        c++;
    }

    int accum = 0;
    char* cursor = num;
    while (*cursor) {
        accum = (accum * 10) + (*cursor - '0');
        cursor++;
    }

    // 2's complement by hand
    return (accum ^ -negative) + negative;
}


// ....
        while (file[cursor] != '\n') {
            temp[temp_pos++] = file[cursor++];
        }
// ....

A bit more complex, but we wrote it ourselves so surely it must be faster 🤓. The majority of the optimizations here are to remove branches from our code as much as possible.

The first line is a little trick that takes advantage of the fact the true/false are just stored as 1/0. If we see a negative sign first it will advance our cursor so that we look past it. From there we create a fixed size buffer and parse our integer as before. Our code scans along until the null byte, adding each character unless it happens to be a '.'.

We then create the actual integer by first multiplying the accumulator by 10, and adding the digit value of the character. The last step is to XOR the accumulator with -negative, and add back in negative. This is possibly the most confusing part so let’s step through both cases.

If the number is to remain positive we would XOR accum with 0, and finally add back in 0, which would do nothing to our number. Easy enough to understand. However if the number is meant to be negative things get more complex.

If you’ve previously learned two’s complement this may look familiar. Negative numbers can be represented using their positive counterpart with some addition bit manipulation. To convert between the two forms we can follow two simple steps.

1. Flip every bit, (0 => 1) and (1 => 0)
2. Add 1

That might seem like it has nothing to do with the last line of our parse_int function, but in fact that is exactly what we’re doing. If negative is true it will hold a value of 1. XORing any value with all 1’s will flip its bits, step 1 accomplished. The only remaining step is to add 1, which just so happens to be the value held in negative!

Does this slightly unreadable code give us significant gains?

In day to day programming it is pretty rare that you get to manipulate the underlying data structures being used in a program, but luckily in C we have to write our own data structures! I happen to know that we can make some significant optimizations around our find and insert logic to hopefully speed up our program. This optimization is actually going to be inspired my the hashmap behavior in C++. When you do a simple lookup on a key (using []) in C++ it will not only return you the value there, but it will also insert it if it doesn’t exist. This would simplify our logic as we insert rows, so let’s see what we can do to achieve that.

// In hashmap.c

// Returns a pointer to the pair if found,
// otherwise returns NULL
pair* hashmap_find(hashmap* h, void* key) {
    uint64_t index = h->hash_func(key) % h->capacity;
    for (int i = 0; i < h->capacity / 2 + 1; i++) {
        uint64_t new_index = (index + (i * i)) % h->capacity;
        pair* p = &h->array[new_index];
        switch ((uint64_t)p->key) {
        case 0: {
            p->key = strdup(key);
            p->value = (Record){.name = p->key,
                                .count = 0,
                                .total = 0,
                                .min = INT64_MAX,
                                .max = INT64_MIN};
        }
            return p;
        default:
            if (h->equals_func(p->key, key)) {
                return p;
            }
        }
    }
    return NULL;
}

// ....
        int temp_i = parse_int(temp);

        pair* p = hashmap_find(&h, name);
        p->value.count++;
        p->value.total += temp_i;
        if (temp_i > p->value.max) {
            p->value.max = temp_i;
        } else if (temp_i < p->value.min) {
            p->value.min = temp_i;
        }
// ....

This saves us from having to branch on every iteration of the loop, now we perform the exact same algorithm every time. This won’t give us the biggest time save as the branch predictor was probably able to figure out that the record exist after the first ~8000 or so iterations.

I want to experiment with a few different hash functions to see what different they make in our overall runtime.

Now we can finally get to the big performance improvements. I’m sure there is more we can squeeze out of a single threaded solution, but I am pretty happy with where we’ve ended up. For now, we can rewrite out code to start taking advantage of all these CPU cores.

There are many different approaches to concurrent and parallel programming, but we will try to choose the simplest of them all. With a file this large (~15GB) there is a lot of room on the table to work in parallel. I won’t say everything in this task can be done entirely independently, but a majority of the work before sorting the weather stations can.

For our first pass we will simply create 12 independent workers that will start at 12 different places in the file. They will each parse records independently, adding each to the hashmap. You may already see a problem with this approach, how do we know where to start each worker? If we split the file up evenly into 12 chunks we have no guarantee that each worker will start on the beginning of a line. We can solve this by having each worker walk backwards until they either hit the beginning of the file, or they find a newline.

Simple enough, but there is still an issue. Unless we give each worker their own hashmap (which would be a pain to reconcile at the end) we will have to synchronize their access to the memory. Shared resources are generally the hardest part of implementing a multithreaded program, with many valid solutions available. The simplest solution is a mutex surrounding their entire hashmap, which is simple enough to implement. Let’s take a look at how that would work.

#include <threads.h>

// ....

hashmap h;
mtx_t hashmap_mutex;

typedef struct thread_args {
  char *file;
  size_t size;
  ssize_t cursor;
  size_t end;
} thread_args;

We start off by including a new header file threads.h which is part of the C11 standard. You can choose to use either threads.h or pthread.h here, the choice is up to you. A mutex is defined to protect access to the hashmap, which are both defined as a global so they are easily accessible to all threads. Threads can take in an argument in the form of a pointer to some piece of memory, which we will define in a struct. Each thread will need a pointer to the file, the size of the file, plus a start and end point.

Our thread function will first walk back to find an appropriate place to start, and then process data until it reaches its endpoint.

int worker(void* arg) {
    thread_args ta = *(thread_args*)arg;
    if (ta.cursor != 0) {
        while (ta.cursor > 0 && ta.file[ta.cursor] != '\n') {
            ta.cursor--;
        }
        ta.cursor++;
    }

    // Cursor is now pointing to the beginning of a line
    char name[100];
    char temp[6];
    while (ta.cursor < ta.end) {
        int name_pos = 0;
        int temp_pos = 0;

        while (ta.file[ta.cursor] != ';') {
            name[name_pos++] = ta.file[ta.cursor++];
        }
        name[name_pos] = '\0';
        ta.cursor++;
        while (ta.file[ta.cursor] != '\n') {
            temp[temp_pos++] = ta.file[ta.cursor++];
        }
        temp[temp_pos] = '\0';
        ta.cursor++;
        int temp_i = parse_int(temp);

        mtx_lock(&hashmap_mutex);
        pair* p = hashmap_find(&h, strdup(name));
        p->value.count++;
        p->value.total += temp_i;
        if (temp_i > p->value.max) {
            p->value.max = temp_i;
        } else if (temp_i < p->value.min) {
            p->value.min = temp_i;
        }
        mtx_unlock(&hashmap_mutex);
    }

    return 0;
}

A fairly straightforward solution which follows the outline described above. The actual parsing logic is copied over from the previous main function, producing the same name and temp variables. We then have to lock the entire table, perform our insert/update, and then release the lock. Can you see the problem with the solution already? If not, don’t worry, we will check both the runtime and flamegraph for more hints.

The final step is to set up our thread pool in main so that we can split out the work.

int main() {
    size_t thread_count = 12;
    thrd_t threads[thread_count];
// ....

    size_t work_size = sb.st_size / thread_count;

    for (int i = 0; i < thread_count; i++) {
        thread_args* ta = malloc(sizeof(thread_args));
        *ta = (thread_args){.file = file,
                            .size = sb.st_size,
                            .cursor = i * work_size,
                            .end = (i * work_size) + work_size};
        thrd_create(&threads[i], worker, ta);
    }

    for (int i = 0; i < thread_count; i++) {
        thrd_join(threads[i], NULL);
    }
// ....
}

What if instead of locking the entire table we were able to just lock a single entry. That would allow for all 12 threads to do work independently, only rarely having to wait when they happen to try to insert a record for the same station. That sounds like a reasonable approach, and one made possible by the fact that we can preallocate our table, preventing it from ever having to grow.

A general purpose hashtable needs to be able to dynamically resize as more elements are added, but we know exactly how large our table needs to be at compile time. Because of this we are able to move closer to what is known as a “lock free” hashmap. Our solution will still require the use of a lock, but locking the entire table for updates will no longer be required. We will still require locking the entire table for insert calls, as we need to ensure proper selection of a position in the hashmap.

We can start by slightly modifying the pair struct which our hashmap is based on.

#include <stdatomic.h>

// ....

typedef struct {
  void *key;
  Record value;
  atomic_bool locked;
} pair;

Atomic operations are the building blocks on top of which concurrency control mechanisms like muticies are built on top of. They are specific hardware instructions that are able to execute atomically, meaning they are guarantee to do exactly what they say they will. These instructions are able to provide this guarantee because they block access by any other thread of execution while the operation is performed. This means the operations are slower than your average ADD or LSL, but their value is worth the runtime cost.

For our implementation each pair can either be “locked” or “unlocked”. If a thread goes to update a locked record it will simply wait its turn. This “waiting” behavior can be costly if the pair is locked for a long period of time, but in our program we know it will only be locked as long as it takes to update 4 fields of the struct. Our point of comparison here will be time spent in pthread_mutex_unlock vs time spent in spin_lock. Speaking of spin_lock let’s quickly implement one.

void spin_lock(pair* p) {
    while (atomic_exchange(&p->locked, true)) {
        // Spin until the lock is released
        while (atomic_load(&p->locked)) {
            // Do nothing, just spin
        }
    }
}

If you’ve heard of a spin lock before this will probably make perfect sense, but the concept is new to me. Let’s break down the atomic operations first, and then we’ll see how those factor in to the spin_lock.

atomic_exchange attempts to put the value true in locked, in either case it returns the previous value in locked. So if the lock was already acquired it will return true causing us to enter the inner loop. Once inside we continuously call atomic_load, which returns the current value of the lock. We will loop as long as the lock is held, exiting as soon as the value returned is false, at which point the outer look will acquire the lock and we are free to move on.

This concept is called a spin lock as is very common when synchronizing memory access across multiple threads. Just as we acquire a lock we will also need to release the lock.

void spin_unlock(pair* p) {
    atomic_store(&p->locked, false);
}

Unlock will only ever be called while we hold the lock, so it is safe to perform this directly with a store operation.

We can update the worker thread to take advantage of the new locks.

// ....
        pair* p = hashmap_find(&h, strdup(name));
        spin_lock(p);
        p->value.count++;
        p->value.total += temp_i;
        if (temp_i > p->value.max) {
            p->value.max = temp_i;
        } else if (temp_i < p->value.min) {
            p->value.min = temp_i;
        }
        spin_unlock(p);
// ....

Now all 12 threads should be able to seamlessly update records without having to fight over locks, and we have an elegant solution in place for when there is resource contention. But you may have noticed something that will make our solution invalid. We only lock down a record as we go to update it, which leaves the case of inserting a new record unprotected. Sadly, this means we will have to walk back out hashmap_find optimization and return to also calling hashmap_insert.

void hashmap_insert(hashmap* h, void* key, Record value) {
    pair new = {key, value, false};

    // Locking for the shortest possible time
    mtx_lock(&hm_mutex);
    pair* first_avail = hashmap_first_avail(h, key);
    *first_avail = new;
    mtx_unlock(&hm_mutex);

    h->size++;
}

I’m only showing the relevant code here, but we essentially will call insert and block all other inserts until our changes have settled. The nice thing about this change is that hopefully inserts will only happen in the first few moments of execution, and everything from then on will be updates.

// ....
        pair* p = hashmap_find(&h, name);
        if (p) {
            spin_lock(p);
            p->value.count++;
            p->value.total += temp_i;
            if (temp_i > p->value.max) {
                p->value.max = temp_i;
            } else if (temp_i < p->value.min) {
                p->value.min = temp_i;
            }
            spin_unlock(p);
        } else {
            char* name_copy = strdup(name);
            hashmap_insert(&h, name_copy,
                           (Record){.name = name_copy,
                                    .total = temp_i,
                                    .count = 1,
                                    .min = temp_i,
                                    .max = temp_i});
        }
// ....

Our insert/update logic now has to branch on if the record exists or not. Sadly for the best performance the locking on inserts is hidden from us in the main function, but I think we can ignore that for now. Let’s see if that helps our runtime.

Before we start rewriting our workers and the main function lets rationalize these decisions. Currently our program called hashmap_find 1 billion times, with around 8800 calls to insert when a new record is discovered. Giving each worker its own hashmap would keep the 1 billion called to hashmap_find, the same ~8800 calls to insert (per thread), and adding in 11 * ~8800 calls to hashmap_find to reconcile at the end.

That comes in right around 100,000 calls to hashmap_find after the threads have finished parsing. A drop in the bucket compared to other 1 billion calls, just 0.01% of the total operations.

Just judging on numbers alone this should be a worthwhile improvement, but as always we are going to implement it and check the performance.

To start off each thread will now need to take in a pointer to a hashmap.

typedef struct thread_args {
    char* file;
    size_t size;
    ssize_t cursor;
    size_t end;
    hashmap* internal_hm;
} thread_args;

Which we will create in main, and initialize per thread.

int main() {
    size_t thread_count = 12;
    thrd_t threads[thread_count];
    hashmap hashmaps[thread_count];
// ....

int worker(void* arg) {
    thread_args ta = *(thread_args*)arg;
    *ta.internal_hm = hashmap_init(&fnv_string_hash, &str_equals, 0.5, 17729);
    if (ta.cursor != 0) {
        while (ta.cursor > 0 && ta.file[ta.cursor] != '\n') {
// ....

The other nice part is that we can drop all locks and special insert/update logic. Our code will look much more similar to previous implementations.

// ....
        int temp_i = parse_int(temp);
        if (ta.cursor >= ta.end) {
            return 0;
        }
        pair* p = hashmap_find(ta.internal_hm, name);
        p->value.count++;
        p->value.total += temp_i;
        if (temp_i > p->value.max) {
            p->value.max = temp_i;
        } else if (temp_i < p->value.min) {
            p->value.min = temp_i;
        }
// ....

Meaning we will need to revert our changes to hashmap_find.

pair *hashmap_find(hashmap *h, void *key) {
  uint64_t index = h->hash_func(key) % h->capacity;
  for (int i = 0; i < h->capacity / 2 + 1; i++) {
    uint64_t new_index = (index + (i * i)) % h->capacity;
    pair *p = &h->array[new_index];
    if (p->key == 0) {
      char *copy_key = strdup(key);
      *p = (pair){.key = copy_key,
                  .value = (Record){
                      .name = key,
                      .total = 0,
                      .count = 0,
                      .min = INT64_MAX,
                      .max = INT64_MIN,
                  }};
      return p;
    } else if (!unsafe_strcmp(p->key, key)) {
      return p;
    }
  }
  return NULL;
}

Great! We are able to remove all locks and muticies from our code. As we observed previously our program spend around 15-20% of its time waiting to acquire a lock, or releasing it. Hopefully by making this change we will see a ~15% reduction in run time.

For now this is where I am going to call it. Of course the time could be dropped further, but that would likely require an architectural rewrite, which I don’t have the time for currently. I was able to bring the time down to 6.12 seconds by increasing the initial capacity of the map, subsequently dropping the collisions to ~1700, just around half of the previous value.

Another easy change to the hashmap is changing up the probing strategy. The original implementation uses quadratic probing, which can produce suboptimal results, especially with lots of non unqiue data. I had a suggestion to try linear probing, but that actually made both the number of collisions and the runtime worse. I could also try separate chaining, but something tells me that would also be slower. What I ultimately ended up settling on is chain hashing. Instead of calculating a hash once and then following a fixed probing strategy, we can calculate a new hash for each collision.

At that point we are no longer hashing an entire string, so we may see potential performance improvement. In fact, both the overall collisions and the runtime dropped, leaving us with a runtime of 6.08 seconds. Here’s one final flamegraph to show how far we’ve come!

The small blocks are once again mostly related to page faults, which I don’t really want to explore right now. So that’s it! A sizeable improvement over our original naive implementation, but it’s always good to know that both solutions solved the problem, and that original solution took less than 10 minutes to write, whereas the final solution is the effort of days of work!