vkpipeline-db: add support for multi-threading

Reviewed-by: Samuel Pitoiset <samuel.pitoiset@gmail.com>

2 files changed, 121 insertions, 49 deletions

diff --git a/CMakeLists.txt b/CMakeLists.txt
index 3e1546c..c1d86c8 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -4,6 +4,7 @@ project(vkpipeline-db)
 
 find_package(Vulkan REQUIRED)
 
+SET(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -fopenmp")
 set(CMAKE_CXX_FLAGS_DEBUG "${CMAKE_CXX_FLAGS_DEBUG} -Wall")
 set(CMAKE_C_FLAGS_DEBUG "${CMAKE_CXX_FLAGS_DEBUG} -Wall")
 if(CMAKE_COMPILER_IS_GNUCXX)
diff --git a/run.c b/run.c
index 1d641a0..eafbd7d 100644
--- a/run.c
+++ b/run.c
@@ -31,16 +31,47 @@
 #include <sys/stat.h>
 #include <fcntl.h>
 #include <unistd.h>
+#include <pthread.h>
+#include <signal.h>
 #include <string.h>
 #include <sys/mman.h>
 #include <sys/stat.h>
 #include <ftw.h>
 #include <time.h>
+#include <omp.h>
 
 #include "serialize.h"
 
 #define unlikely(x) __builtin_expect(!!(x), 0)
 
+int max_threads;
+const char **current_pipeline_names;
+static pthread_mutex_t printf_mutex;
+
+#define sigputs(str) write(STDERR_FILENO, str, strlen(str))
+
+#pragma GCC diagnostic push
+#pragma GCC diagnostic ignored "-Wunused-result"
+static void
+abort_handler(int signo)
+{
+    if (current_pipeline_names) {
+        sigputs("\n => CRASHED <= while processing these pipelines:\n\n");
+        for (int i = 0; i < max_threads; i++) {
+            if (current_pipeline_names[i]) {
+                sigputs("    ");
+                sigputs(current_pipeline_names[i]);
+                sigputs("\n");
+            }
+        }
+    } else {
+       sigputs("\n => CRASHED <= during final teardown.\n");
+    }
+    sigputs("\n");
+    _exit(-1);
+}
+#pragma GCC diagnostic pop
+
 /* Pipeline tests. */
 static unsigned pipeline_test_size = 1 << 15; /* next-pow-2(num pipelines in db) */
 static unsigned pipeline_test_length;
@@ -94,7 +125,7 @@ struct shader_stats
 };
 
 #define PARSE_STAT(key, value) \
-    line = strtok(NULL, "\n"); \
+    line = strtok_r(NULL, "\n", &saveptr); \
     if (sscanf(line, key, value) != 1) \
         return -1;
 
@@ -102,12 +133,13 @@ static int
 parse_shader_stats(char *buf, struct shader_stats *stats)
 {
     char *line;
+    char *saveptr;
 
-    line = strtok(buf, "\n");
+    line = strtok_r(buf, "\n", &saveptr);
     while(line) {
         if (!strcmp(line, "*** SHADER STATS ***"))
             break;
-        line = strtok(NULL, "\n");
+        line = strtok_r(NULL, "\n", &saveptr);
     }
 
     if (unlikely(!line))
@@ -200,6 +232,7 @@ static void
 print_shader_stats(const char *pipeline_name, VkShaderStageFlagBits stage,
                    const struct shader_stats *stats)
 {
+    pthread_mutex_lock(&printf_mutex);
     printf("%s (%s) - ", pipeline_name, get_shader_stage_name(stage));
     printf("Shader Stats: ");
     printf("SGPRS: %d ", stats->num_sgprs);
@@ -212,6 +245,7 @@ print_shader_stats(const char *pipeline_name, VkShaderStageFlagBits stage,
     printf("Spilled VGPRs: %d ", stats->num_spilled_vgprs);
     printf("PrivMem VGPRs: %d ", stats->priv_mem_vgprs);
     printf("\n");
+    pthread_mutex_unlock(&printf_mutex);
 }
 
 static VkResult
@@ -401,7 +435,8 @@ static void
 print_usage(const char *prog_name)
 {
     fprintf(stderr,
-            "Usage: %s <directories and *.pipeline_test files>\n",
+            "Usage: %s [-j <max_threads>] "
+            "<directories and *.pipeline_test files>\n",
             prog_name);
 }
 
@@ -410,16 +445,31 @@ int main(int argc, char **argv)
     const char *extensionNames[] = { "VK_AMD_shader_info" };
     VkQueueFamilyProperties queue_family;
     VkPhysicalDevice *physical_devices;
-    struct timespec start, end;
     uint32_t device_count;
     uint32_t queue_count = 1;
     VkInstance instance;
-    VkDevice device;
     VkResult result;
     int ret = 0;
+    int opt;
+
+    pthread_mutex_init(&printf_mutex, NULL);
+
+    max_threads = omp_get_max_threads();
+
+    while ((opt = getopt(argc, argv, "j:")) != -1) {
+        switch(opt) {
+        case 'j':
+            max_threads = atoi(optarg);
+            break;
+        default:
+            fprintf(stderr, "Unknown option: %x\n", opt);
+            print_usage(argv[0]);
+            return -1;
+        }
+    }
 
-    if (argc < 2) {
-        fprintf(stderr, "No directories specified!\n");
+    if (unlikely(optind >= argc)) {
+        fprintf(stderr, "No directories specified\n");
         print_usage(argv[0]);
         return -1;
     }
@@ -464,6 +514,7 @@ int main(int argc, char **argv)
     assert(queue_family.queueFlags & VK_QUEUE_GRAPHICS_BIT);
 
     /* Create logical device. */
+    VkDevice device;
     VkDeviceQueueCreateInfo queueCreateInfo = {};
     queueCreateInfo.sType = VK_STRUCTURE_TYPE_DEVICE_QUEUE_CREATE_INFO;
     queueCreateInfo.queueFamilyIndex = 0;
@@ -494,54 +545,74 @@ int main(int argc, char **argv)
      */
     /* Gather all pipeline tests. */
     pipeline_test = malloc(pipeline_test_size * sizeof(struct pipeline_test));
-    for (int i = 1; i < argc; i++) {
+    for (int i = optind; i < argc; i++) {
         ftw(argv[i], gather_pipeline_test, 100);
     }
 
-    clock_gettime(CLOCK_PROCESS_CPUTIME_ID, &start);
-
-    /* Process each pipeline tests. */
-    for (unsigned i = 0; i < pipeline_test_length; i++) {
-        const char *current_pipeline_name = pipeline_test[i].filename;
-        off_t filesize = pipeline_test[i].filesize;
-        char *data;
-        int fd;
-
-        fprintf(stderr, "--> %s\n", current_pipeline_name);
-
-        fd = open(current_pipeline_name, O_RDONLY);
-        if (unlikely(fd == -1)) {
-            perror("open");
-            continue;
+    unsigned *pipline_counts = calloc(max_threads, sizeof(unsigned));
+    current_pipeline_names = calloc(max_threads, sizeof(const char *));
+    omp_set_num_threads(max_threads);
+
+    if (signal(SIGABRT, abort_handler) == SIG_ERR)
+        fprintf(stderr, "WARNING: could not install SIGABRT handler.\n");
+    if (signal(SIGSEGV, abort_handler) == SIG_ERR)
+        fprintf(stderr, "WARNING: could not install SIGSEGV handler.\n");
+
+    #pragma omp parallel if(max_threads > 1 && pipeline_test_length > max_threads)
+    {
+        struct timespec start, end;
+        clock_gettime(CLOCK_THREAD_CPUTIME_ID, &start);
+
+        /* Process each pipeline tests. */
+        #pragma omp for schedule(dynamic)
+        for (unsigned i = 0; i < pipeline_test_length; i++) {
+            const char *current_pipeline_name = pipeline_test[i].filename;
+            off_t filesize = pipeline_test[i].filesize;
+            char *data;
+            int fd;
+
+            current_pipeline_names[omp_get_thread_num()] = current_pipeline_name;
+
+            fprintf(stderr, "--> %s\n", current_pipeline_name);
+
+            fd = open(current_pipeline_name, O_RDONLY);
+            if (unlikely(fd == -1)) {
+                perror("open");
+                continue;
+            }
+
+            data = mmap(NULL, filesize, PROT_READ, MAP_PRIVATE, fd, 0);
+            if (unlikely(data == MAP_FAILED)) {
+                perror("mmap");
+                continue;
+            }
+
+            if (unlikely(close(fd) == -1)) {
+                perror("close");
+                continue;
+            }
+
+            if (unlikely(run(device, current_pipeline_name, data, filesize) < 0))
+                continue;
+
+            pipline_counts[omp_get_thread_num()]++;
+
+            if (unlikely(munmap(data, filesize) == -1)) {
+                perror("munmap");
+                continue;
+            }
+
+            current_pipeline_names[omp_get_thread_num()] = NULL;
+            free(pipeline_test[i].filename);
         }
 
-        data = mmap(NULL, filesize, PROT_READ, MAP_PRIVATE, fd, 0);
-        if (unlikely(data == MAP_FAILED)) {
-            perror("mmap");
-            continue;
-        }
-
-        if (unlikely(close(fd) == -1)) {
-            perror("close");
-            continue;
-        }
-
-        if (unlikely(run(device, current_pipeline_name, data, filesize) < 0))
-            continue;
-
-        if (unlikely(munmap(data, filesize) == -1)) {
-            perror("munmap");
-            continue;
-        }
-
-        free(pipeline_test[i].filename);
+        clock_gettime(CLOCK_THREAD_CPUTIME_ID, &end);
+        printf("Thread %d took %.2lf seconds and compiled %u pipelines\n",
+               omp_get_thread_num(),
+               (end.tv_sec - start.tv_sec) + 10e-9 * (end.tv_nsec - start.tv_nsec),
+               pipline_counts[omp_get_thread_num()]);
     }
 
-    clock_gettime(CLOCK_PROCESS_CPUTIME_ID, &end);
-    printf("Process took %.2lf seconds and compiled %u pipelines\n",
-           (end.tv_sec - start.tv_sec) + 10e-9 * (end.tv_nsec - start.tv_nsec),
-           pipeline_test_length);
-
     free(pipeline_test);
 
     vkDestroyDevice(device, NULL);