Juggling Arenas in Multithreaded Queues

Converting JSON to SQLite.

Introduction

I have collected a large amount of JSON-files and they take up a large amount of space. I had the idea of inserting the data into a database in the hope that it would be reduced in size. In a previous article I introduced Offsetof: Parsing structured data into structured data in C. In this article I will show how I used Offsetof, memory arenas, and multithreaded queues, to parse JSON files and store the result in an SQLite database.

Producer-Consumer-Main

This project was created to run on my old potato-computer that uses a hard drive for its main storage. An assumption I had when creating this was that it is probably a bad idea for each thread to hammer on the hard drive all at once. There is only one bus between the CPU and hard drive and all threads would have to contend over it. Therefore, it would be a better idea to use a producer-consumer model and have one thread – the producer – read everything from disk and have multiple consumer threads do the actual parsing of the JSON.

This still requires a thread to do the writes to the database. I have the main thread collect the parsed data and write it into the database in batched transactions. I was never happy with this solution because now there are two threads accessing the hard drive at the same time.

The Queues

Queues are used in order to transfer data between the threads. Since the queues are used by multiple threads they must be synchronized and this is done using mutexes. They are implemented as a simple ring buffer with an underlying array that is statically allocated. This means that if the queue is full then a writer must wait for a reader to remove a job before the writer can write to the queue. This is actually a good thing as we will see later.

Likewise, if the queue is empty a reader must wait for items to be written. Both of these waits are implemented using pthread’s conditional variables. After a writer has written it will signal the condition which will start at most one reader that is waiting.

/**
 * Enqueue `num_jobs` jobs from `jobs` to `jq`.
 */
i64 jobs_enqueue(JobQueue *jq, Job *jobs, i64 num_jobs) {

    // Acquire mutex for head
    pthread_mutex_lock(&jq->mutex_head);

    for (int i=0; i<num_jobs; ++i) {
        if (jobs_full(jq)) {
            // If the queue is full then the thread has to wait for it to
            // release a spot. See Example in pthread_cond_init(3).
            pthread_mutex_lock(&jq->full_condition_mutex);

            while (jobs_full(jq)) {
                pthread_cond_wait(
                    &jq->full_condition,
                    &jq->full_condition_mutex
                );
            }

            // In the Example they operate here but the mutex for head is
            // already held. The condition-mutex can be unlocked here and
            // the condition will still be true below.
            pthread_mutex_unlock(&jq->full_condition_mutex);
        }

        // Add job to queue
        assert(!jobs_full(jq));
        jobs[i].id = ++jq->id;
        jq->jobs[jq->head] = jobs[i];
        jq->head = (jq->head + 1) % jq->size;

        // Notify a waiting consumer that there is a new job. There is
        // at most one consumer waiting* so use pthread_cond_signal.
        //
        // *There can be multiple consumers waiting but at most one of
        // them is waiting on the empty condition -- the remainder are
        // waiting for the tail mutex.
        pthread_mutex_lock(&jq->empty_condition_mutex);
        pthread_cond_signal(&jq->empty_condition);
        pthread_mutex_unlock(&jq->empty_condition_mutex);
    }

    pthread_mutex_unlock(&jq->mutex_head);

    // Returns id for the last job
    return jobs[num_jobs-1].id;
}

Each queue is synchronized using four different mutexes. There is a mutex each for the head and the tail. These are acquired by the writers and the readers, respectively. The two condition variables, used for waiting if the queue is empty or full, also uses a mutex each.

The queues supports both multiple writers and multiple readers but no queue actually uses this. The todo-queue has a single writer and multiple readers, the done-queue has multiple writers (the consumer-threads) and one reader (the main thread), and the arenas queue has a single writer and a single reader.

The Memory Arenas

A lot has been written on memory arenas and how great they are – and they are correct.¹ Instead of managing millions of allocations and deallocations they enable you to manage a much smaller set of lifetimes instead. With arenas you don’t have to care about every single allocated object as the JSON is parsed, you only have to care about the parsed data as a whole. When you are done with the data everything can be (soft-)deallocated at the same time by setting job.arena = job.reset. In particular, I like this implementation of arenas by Chris Wellons which I have adapted to use in my own programs.

Since the arenas are never used by more than one thread at a time it is not necessary to synchronize them. This is how you are supposed to use arenas. I often see people asking whether the arenas are thread-safe (or how to make them so). They are not thread-safe and it is by design. You should architect your program so that the arenas are not used by multiple threads at the same time. The arenas are not even the important part! It is just a location to store things in. If the thing that you store is important and needs to be synchronized then the thing should be synchronized! You don’t have to synchronize the entire arena for that. If you have multiple threads that needs to to store things then let each thread have its own arena.

The Threads

The producer-thread pops a Job from the arena-queue which contains an Arena. It then reads the entire JSON-file into it and pushes the job to the todo-queue for the consumers to read.

struct Job {

    JobType type;

    // The main arena that the job should write to.
    Arena arena;

    // A copy of the original arena that is used to reset it.
    Arena reset;

    // The file to be read.
    String filename;

    // The producer reads the file and stores it in file_contents.
    String file_contents;

    // A consumer reads the file_contents and stores it in pd.
    ParsedData pd;
};

Each consumer-thread pops a Job and parses the JSON in file_contents into structs which are allocated in the same arena. It then pushes the Job into the done-queue for the main thread to process.

The main thread begins by starting a database-transaction and then starts waiting for parsed data from the consumers. It reads the parsed data and binds the values in prepared statements. After the main thread has processed a set limit of files the transaction is committed and the arenas recycled into the arenas-queue. If there are more files to read a new transaction is started.

Juggling Arenas

One of the problems I had was how to limit the producer-thread so that it does not overproduce files for the consumers. It has to be limited by something or it is going to read files until the memory runs out. In the end I managed to come up with an elegant self-limiting solution: have the main thread reset and recycle the arenas into a third queue – the arenas-queue – and have the producer-thread use arenas from that queue.

The main thread creates a fixed set of empty arenas and inserts them in the arenas-queue in the setup-phase before the other threads are started. The producer pops one of these arenas and reads a file into it. The Job is then passed to a consumer which parses the data into structs and writes the parsed data into the same arena. The Job is passed to the main thread which writes the data to the database. After the data is written the main thread will reset the arena (a simple 16-byte copy of job.reset) and insert the arena back into the arenas-queue.

Juggling the arenas like this has several positive effects. First, as has already been mentioned, it limits the producer so that it does not overproduce files. If the arenas-queue runs dry it has to wait for arenas to be reset and recycled by the main thread. Secondly, it limits how much memory can be used by the program. The arenas I use do not resize themselves and the program will terminate if an arena runs out of memory. Want to use more memory? Create more arenas (or larger ones)! Thirdly, it simplifies the use of arenas and every file can easily have its own arena.

Raylib GUI

In order to try to make sense of the program I had to implement some kind of visualization. It is so hard – with print statements or debugger – to try to find out what is actually happening in a multithreaded program. Where are the bottlenecks? Are the queues empty or full? Are the threads working or waiting? When I adjust one of the many knobs, is it performing better or worse?

Inspired by Dennis Gustafsson - Parallelizing the physics solver I implemented a simple GUI in raylib that displays the state of each thread to see whether it is working or waiting. It uses elapsed real time from clock_gettime(2) and this works surprisingly well and was easy to implement.²

Evaluation

Performance

Table 1: Performance data. The single-threaded type uses a single-threaded program to read, parse, and insert into a SQLite database. This method has not been discussed in this article. The multi-threaded type is juggling arenas as described in this article. The multi-threaded* type is juggling arenas as described in this article but the producer- and consumer-threads have been modified so that the reading of each file is done by the consumer-thread.

The program is analyzed using a small test case containing 299 files with a total size of 371 MiB. The performance-data are displayed in Table 1. With a cleared cache the test case is completed in 29.1 seconds for an average rate of 12.7 MiB³. This sounds bad even for my old hard drive which is probably capable of more than 30 MiB/s. In Figure 1 the visualizer is used to analyse the run. The consumers spend most their time waiting and are almost executing serially as the producer reads files. There is almost no parallelism among the consumers and they are all bottlenecked by the producer which cannot read the files fast enough. It is however an improvement from the single-threaded version as there is some parallelism between the producer-consumers-main.

Figure 2 shows a run of the same test case but with a hot cache. It finishes in 3.8 seconds for an average rate of 96.8 MiB/s. In this case we can see a lot of parallelism among the consumers and the program is now bottlenecked by the main thread which is not able to insert the data into the database and recycle the arenas fast enough, causing the producer to wait. This test is obviously less important than the cleared cache test as the program is meant to read every file just once.

An additional test run was made where the consumers read the file instead of the producer to test the starting assumption whether it is better for the consumer-threads hammer on the hard drive all at the same time. This completed the test case in 20.1 seconds (18.5 MiB/s). This is faster than the multi-threaded test run which suggests that the starting assumption was wrong.

Size

The stated reason for this project was to compress the JSON. The size-results are shown in Table 2. After the data was converted to an SQLite database the database file was 17.8 MiB large. This is a lot bigger than I was expecting. Alternatively, the JSON can be compressed down to 4 MiB using xz. xz can also be used to compress the SQLite database file which brings it down to 3.3 MiB. Technically this is a successful method to reduce the size of the JSON. I also tried exporting each table in the database as CSV and the size came out as 10.9 MiB (1.0 MiB compressed). It is however not trivial to generate CSV directly (skipping the SQLite step) since SQLite handles the auto-incrementing IDs and deduplication of data.

Conclusion

The speed of the program is not as high as I would like it and it looks like the starting assumption was wrong. I think my biggest mistake was that I read the entire file before parsing it. This is probably the biggest bottleneck. In Figure 1 the producer-thread is mostly green because it is working but in reality this work is mostly waiting on the hard disk drive. I should have streamed the data in smaller chunks and from multiple files. The consumers would then have to stop and wait if they run out of data. This would require a big architectural change which is why I have not tested it yet. Perhaps using a publish-subscribe model instead of producer-consumer would be a better fit for this. I have also been testing io_uring instead of fread.

A worry I had when writing the queues was that the mutexes would be slow and hold up the threads a lot. The visualizer shows that this is not a big problem. The mutexes are fast enough and the cost of synchronization is much lower than the actual work being done. The conditional variables work well and it is impressive how fast a waiting consumer is started after the producer has finished reading a file. However, I wonder if I am required to use memory barriers when passing around data between threads or whether pthreads takes care of this automatically?⁴ Perhaps there are other multi-threaded footguns that I am not aware of. But this works on my machine!

Writing the visualizer was not planned but I am glad I did. Without it it was very hard to understand what the program is doing. There are so many different knobs to turn and it is hard to determine whether it works better or worse.