Parallelize File Processing

CMakeLists.txt

@@ -50,6 +50,8 @@ include(cmake/TestUtil.cmake)
 include(cmake/TestCoverageUtil.cmake)
 include(cmake/SanitizerUtil.cmake)
 
+find_package(Threads REQUIRED)
+
 if(NOT DEFINED RECKON_DO_NOT_OVERWRITE_OUTPUT_DIRECTORY)
     set_output_directories()
 endif()

VERSION

@@ -1 +1 @@
-1.8.1-dev
+1.9.0-dev

cmake/TestCoverageUtil.cmake

@@ -81,7 +81,10 @@ function(add_code_coverage target_name)
     target_compile_options(
         ${target_name}
         PRIVATE
-        -ftest-coverage -fprofile-arcs -fno-default-inline
+        -ftest-coverage
+        -fprofile-arcs
+        -fprofile-update=atomic
+        -fno-default-inline
         $<$<COMPILE_LANGUAGE:CXX>:-fno-elide-constructors>
     )
     target_link_libraries(

src/lib/CMakeLists.txt

@@ -31,8 +31,11 @@ target_sources(
     "c/encoding.c"
     "c/factories.c"
     "c/fileio.c"
+    "c/threading.c"
     "$<$<PLATFORM_ID:Linux>:${CMAKE_CURRENT_SOURCE_DIR}/c/linux/fileio.c>"
+    "$<$<PLATFORM_ID:Linux>:${CMAKE_CURRENT_SOURCE_DIR}/c/linux/threading.c>"
     "$<$<PLATFORM_ID:Windows>:${CMAKE_CURRENT_SOURCE_DIR}/c/win32/fileio.c>"
+    "$<$<PLATFORM_ID:Windows>:${CMAKE_CURRENT_SOURCE_DIR}/c/win32/threading.c>"
     "c/lang_c.c"
     "c/lang_java.c"
     "c/lang_python.c"
@@ -61,6 +64,7 @@ target_include_directories(
 target_link_libraries(
     ${RECKON_TARGET_LIB_OBJ}
     PUBLIC
+    Threads::Threads
     ${RECKON_DEPENDENCIES_LINK_TARGETS}
 )
 
@@ -88,6 +92,14 @@ if("${RECKON_MAIN_LIB_TYPE}" STREQUAL "SHARED")
     )
 endif()
 
+if(MSVC)
+    target_compile_options(
+        ${RECKON_TARGET_LIB_OBJ}
+        PRIVATE
+        /experimental:c11atomics
+    )
+endif()
+
 enable_compiler_warnings(
     ${RECKON_TARGET_LIB_OBJ}
     ${RECKON_IGNORE_WARNINGS}
@@ -124,6 +136,7 @@ target_include_directories(
 target_link_libraries(
     ${RECKON_TARGET_LIB}
     PRIVATE
+    Threads::Threads
     ${RECKON_DEPENDENCIES_LINK_TARGETS}
 )
 

src/lib/c/evaluation.h

@@ -59,6 +59,15 @@ typedef struct Span {
     size_t length;
 } Span;
 
+/**
+ * An indicator for a slice of an array.
+ * Has a start (inclusive) and an end (exclusive) index.
+ */
+typedef struct Slice {
+    size_t start;
+    size_t end;
+} Slice;
+
 /**
  * Opaque context type used during node evaluation.
  * Can be used by concrete `NodeVisitor` implementations to store

src/lib/c/linux/threading.c

@@ -0,0 +1,127 @@
+/*
+ * Copyright (C) 2026 Raven Computing
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#ifdef __linux__
+
+#include <stdlib.h>
+#include <stddef.h>
+#include <stdbool.h>
+#include <unistd.h>
+#include <assert.h>
+#include <pthread.h>
+
+#include "threading.h"
+
+/**
+ * The argument passed to pthread_create()
+ */
+typedef struct NativeThreadArg {
+    ThreadRoutine routine;
+    ThreadWork* arg;
+} NativeThreadArg;
+
+/**
+ * The start routine passed to pthread_create()
+ */
+static void* rcnRun(void* arg) {
+    NativeThreadArg* feeder = (NativeThreadArg*) arg;
+    ThreadRoutine routine = feeder->routine;
+    ThreadWork* threadArg = feeder->arg;
+    free(feeder);
+    routine(threadArg);
+    return NULL;
+}
+
+bool initThreadMutex(RCN_NATIVE_HANDLE* mutex) {
+    pthread_mutex_t* nativeMutex = malloc(sizeof(pthread_mutex_t));
+    if (!nativeMutex) {
+        return false;
+    }
+    if (pthread_mutex_init(nativeMutex, NULL) != 0) {
+        free(nativeMutex);
+        return false;
+    }
+    *mutex = nativeMutex;
+    return true;
+}
+
+void deinitThreadMutex(RCN_NATIVE_HANDLE mutex) {
+    pthread_mutex_t* nativeMutex = mutex;
+    if (!nativeMutex) {
+        return;
+    }
+    int status = pthread_mutex_destroy(nativeMutex);
+    assert(status == 0);
+    (void) status;
+    free(nativeMutex);
+}
+
+void lockThread(RCN_NATIVE_HANDLE mutex) {
+    int status = pthread_mutex_lock((pthread_mutex_t*) mutex);
+    assert(status == 0);
+    (void) status;
+}
+
+void unlockThread(RCN_NATIVE_HANDLE mutex) {
+    int status = pthread_mutex_unlock((pthread_mutex_t*) mutex);
+    assert(status == 0);
+    (void) status;
+}
+
+size_t getSystemConcurrency(void) {
+    long processorCount = sysconf(_SC_NPROCESSORS_ONLN);
+    if (processorCount <= 0) {
+        return 1;
+    }
+    return (size_t) processorCount;
+}
+
+bool createThread(
+    ThreadHandle* handle,
+    ThreadRoutine routine,
+    ThreadWork* arg
+) {
+    pthread_t* nativeThread = malloc(sizeof(pthread_t));
+    NativeThreadArg* feeder = malloc(sizeof(NativeThreadArg));
+    if (!nativeThread || !feeder) {
+        free(nativeThread);
+        free(feeder);
+        return false;
+    }
+    feeder->routine = routine;
+    feeder->arg = arg;
+    if (pthread_create(nativeThread, NULL, rcnRun, feeder) != 0) {
+        free(feeder);
+        free(nativeThread);
+        return false;
+    }
+    handle->instance = nativeThread;
+    return true;
+}
+
+void waitForThread(ThreadHandle* handle) {
+    pthread_t* nativeThread = (pthread_t*) handle->instance;
+    if (!nativeThread) {
+        return;
+    }
+    int status = pthread_join(*nativeThread, NULL);
+    assert(status == 0);
+    (void) status;
+    free(nativeThread);
+    handle->instance = NULL;
+}
+
+#endif // __linux__