Transparent threads for cuFile

gds/tools/ioblitz/CMakeLists.txt

@@ -0,0 +1,64 @@
+cmake_minimum_required(VERSION 3.10 FATAL_ERROR)
+
+# set(CMAKE_BUILD_TYPE Release)
+#set(CMAKE_BUILD_TYPE Debug)
+set(CMAKE_BUILD_TYPE RelWithDebInfo)
+
+set(CMAKE_EXPORT_COMPILE_COMMANDS ON)
+set(CMAKE_COLOR_MAKEFILE ON)
+set_property(GLOBAL PROPERTY USE_FOLDERS ON)
+
+project(ioblitz LANGUAGES C CXX CUDA VERSION 0.3.0)
+set(CMAKE_MODULE_PATH ${PROJECT_SOURCE_DIR}/cmake/modules ${CMAKE_MODULE_PATH})
+
+find_package(CUDA REQUIRED)
+find_package(MPI REQUIRED)
+include_directories(${CMAKE_CUDA_TOOLKIT_INCLUDE_DIRECTORIES})
+include_directories("/usr/local/gds/lib")
+include_directories(${MPI_INCLUDE_PATH})
+
+# CUDA flags
+set(CMAKE_CXX_STANDARD 17)
+set(CMAKE_CUDA_STANDARD 17)
+set(CMAKE_CXX_STANDARD_REQUIRED ON)
+
+# treat warnings as errors
+# set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Wall -Wextra -Wshadow -pedantic -Werror")
+
+# do not treat warnings as errors
+set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Wall -Wextra -Wshadow")
+
+set(CMAKE_CXX_FLAGS_DEBUG "-g")
+set(CMAKE_CXX_FLAGS_RELEASE "-O3")
+
+set(CUDA_USE_STATIC_CUDA_RUNTIME ON)
+set(CUDA_VERBOSE_BUILD ON)
+
+##############################################################################
+# BUILD libioblitz.so
+
+set (LIB_IOBLITZ_SRCS
+  src/ioblitz.h src/ioblitz.c
+)
+
+include_directories(./src)
+
+SET(CMAKE_LIBRARY_OUTPUT_DIRECTORY "${PROJECT_SOURCE_DIR}/lib/")
+
+set(THREADS_PREFER_PTHREAD_FLAG ON)
+find_package(Threads REQUIRED)
+
+link_directories("/lib64")
+link_directories("/usr/lib64")
+link_directories("/usr/local/cuda/lib64")
+link_directories("/sw/summit/cuda/11.0.3/lib64")
+
+add_library(ioblitz SHARED ${LIB_IOBLITZ_SRCS})
+target_link_libraries(ioblitz Threads::Threads)
+target_link_libraries(ioblitz dl)
+target_link_libraries(ioblitz ${CUDA_LIBRARIES})
+target_link_libraries(ioblitz cuda)
+target_link_libraries(ioblitz nvToolsExt)
+target_link_libraries(ioblitz "-Wl,--no-undefined")
+
+##############################################################################

gds/tools/ioblitz/README.md

@@ -0,0 +1,5 @@
+# ioblitz
+ioblitz is a lightweight C library that services IO requests using threads. 
+
+see https://github.com/hpc-io/ioblitz for more information...
+

gds/tools/ioblitz/src/ioblitz.c

@@ -0,0 +1,147 @@
+/* -*- mode: c; c-basic-offset: 2; indent-tabs-mode: nil; -*-
+ * vim:expandtab:shiftwidth=2:tabstop=2:
+ */
+/******************************************************************************\
+*                                                                              *
+*        Copyright (c) 2020, John J. Ravi                                      *
+*      See the file COPYRIGHT for a complete copyright notice and license.     *
+*                                                                              *
+********************************************************************************
+*
+* Definitions and prototypes of abstract I/O interface
+*
+\******************************************************************************/
+
+#define _GNU_SOURCE
+
+#include "ioblitz.h"
+
+static void *copy_thread_fn(void *data) {
+  thread_data_t *td = (thread_data_t *)data;
+  // int ret =
+    real_memcpy((char *)td->dst+td->offset, (char *)td->src+td->offset, td->size);
+  return NULL;
+}
+
+void * memcpy(void *destination, const void *source, size_t num) {
+  void *return_value;
+
+  // printf("john memcpy\n");
+  char *io_blitz_size_threshold;
+  if( (io_blitz_size_threshold = getenv("IO_BLITZ_SIZE_THRESHOLD")) &&
+      (num >= atoi(io_blitz_size_threshold))) {
+    char *io_blitz_threads;
+
+    uint8_t num_worker = 1;
+    if(io_blitz_threads = getenv("IO_BLITZ_THREADS")) {
+      num_worker = atoi(io_blitz_threads);
+    }
+
+    pthread_t threads[num_worker];
+    thread_data_t td[num_worker];
+
+    size_t io_chunk = num / num_worker;
+    size_t io_chunk_rem = num % num_worker;
+
+    for (uint8_t ii = 0; ii < num_worker; ii++) {
+      td[ii].dst = destination;
+      td[ii].src = source;
+      td[ii].size = io_chunk;
+      td[ii].offset = (size_t)ii*io_chunk;
+
+      if(ii == num_worker-1) {
+        td[ii].size = (size_t)(io_chunk + io_chunk_rem);
+      }
+    }
+
+    for (int ii = 0; ii < num_worker; ii++) {
+      // ret =
+        pthread_create(&threads[ii], NULL, &copy_thread_fn, &td[ii]);
+    }
+
+    for (int ii = 0; ii < num_worker; ii++) {
+      pthread_join(threads[ii], NULL);
+    }
+  }
+  else {
+    return_value = real_memcpy(destination, source, num);
+  }
+
+  return return_value;
+}
+
+static void *cufile_write_thread_fn(void *data) {
+  cufile_thread_data_t *td = (cufile_thread_data_t *)data;
+
+  ssize_t ret =
+   real_cuFileWrite(
+     td->cfr_handle, 
+     td->wr_devPtr, 
+     td->size, 
+     td->offset, 
+     td->devPtr_offset
+   );
+
+  if (ret != td->size) {
+    fprintf(stderr, "thread write failed!\n");
+  }
+
+  // printf("ret code: %ld\n", ret);
+
+  return NULL;
+}
+
+ssize_t cuFileWrite(CUfileHandle_t fh, const void *devPtr_base, size_t size, off_t file_offset, off_t devPtr_offset) {
+  // TODO: return size only if completed successfully
+  ssize_t return_value = size;
+
+  // printf("john cuFileWrite\n");
+
+  char *io_blitz_size_threshold;
+  if( (io_blitz_size_threshold = getenv("IO_BLITZ_SIZE_THRESHOLD")) &&
+      (size >= atoi(io_blitz_size_threshold))) {
+    char *io_blitz_threads;
+
+    uint8_t num_worker = 1;
+    if(io_blitz_threads = getenv("IO_BLITZ_THREADS")) {
+      num_worker = atoi(io_blitz_threads);
+    }
+
+    pthread_t threads[num_worker];
+    cufile_thread_data_t td[num_worker];
+
+    size_t io_chunk = size / num_worker;
+    size_t io_chunk_rem = size % num_worker;
+
+    for (uint8_t ii = 0; ii < num_worker; ii++) {
+      td[ii].cfr_handle = fh;
+      td[ii].wr_devPtr = devPtr_base;
+      td[ii].size = io_chunk;
+      td[ii].offset = (size_t)ii*file_offset;
+      td[ii].devPtr_offset = (size_t)ii*devPtr_offset;
+
+      td[ii].offset = (size_t)(file_offset + ii*io_chunk);
+      td[ii].devPtr_offset = (size_t)(devPtr_offset + ii*io_chunk);
+
+      if(ii == num_worker-1) {
+        td[ii].size = (size_t)(io_chunk + io_chunk_rem);
+      }
+    }
+
+    for (int ii = 0; ii < num_worker; ii++) {
+      // ret =
+        pthread_create(&threads[ii], NULL, &cufile_write_thread_fn, &td[ii]);
+    }
+
+    for (int ii = 0; ii < num_worker; ii++) {
+      pthread_join(threads[ii], NULL);
+    }
+  }
+  else {
+   return_value = real_cuFileWrite(fh, devPtr_base, size, file_offset, devPtr_offset);
+  }
+
+  return return_value;
+
+}
+

gds/tools/ioblitz/src/ioblitz.h

@@ -0,0 +1,76 @@
+/* -*- mode: c; c-basic-offset: 2; indent-tabs-mode: nil; -*-
+ * vim:expandtab:shiftwidth=2:tabstop=2:
+ */
+/******************************************************************************\
+*                                                                              *
+*        Copyright (c) 2020, John J. Ravi                                      *
+*      See the file LICENSE for a complete copyright notice and license.       *
+*                                                                              *
+\******************************************************************************/
+
+#ifndef _IOBLITZ_H
+#define _IOBLITZ_H
+
+#include <dlfcn.h>
+#include <pthread.h>
+#include <stdint.h>
+#include <stdlib.h>
+#include <stdio.h>
+
+#include <cuda.h>
+#include <cufile.h>
+
+typedef struct thread_data_t {
+  void *dst;
+  const void *src;
+  size_t size;
+  size_t offset;
+} thread_data_t;
+
+typedef struct cufile_thread_data_t {
+  union {
+    void *rd_devPtr;            /* read device address */
+    const void *wr_devPtr;      /* write device address */
+  };
+  CUfileHandle_t cfr_handle; /* cuFile Handle */
+  off_t offset;              /* File offset */
+  off_t devPtr_offset;       /* device address offset */
+  size_t size;               /* Read/Write size */
+} cufile_thread_data_t;
+
+typedef void * (*real_memcpy_t)(void *, const void *, size_t);
+
+typedef ssize_t (*real_open_t)(const char *, int, mode_t);
+typedef ssize_t (*real_close_t)(int);
+
+typedef ssize_t (*real_pread_t)(int, void *, size_t, off_t);
+typedef ssize_t (*real_pwrite_t)(int, const void *, size_t, off_t);
+
+typedef ssize_t (*real_cuFileWrite_t)(CUfileHandle_t, const void *, size_t, off_t, off_t);
+
+void * real_memcpy(void *destination, const void *source, size_t num) {
+  return ((real_memcpy_t)dlsym(RTLD_NEXT, "memcpy"))(destination, source, num);
+}
+
+int real_open(const char *pathname, int flags, mode_t mode) {
+  return ((real_open_t)dlsym(RTLD_NEXT, "open"))(pathname, flags, mode);
+}
+
+int real_close(int fd) {
+  return ((real_close_t)dlsym(RTLD_NEXT, "close"))(fd);
+}
+
+ssize_t real_pread(int fd, void *data, size_t size, off_t offset) {
+  return ((real_pread_t)dlsym(RTLD_NEXT, "pread"))(fd, data, size, offset);
+}
+
+ssize_t real_pwrite(int fd, const void *data, size_t size, off_t offset) {
+  return ((real_pwrite_t)dlsym(RTLD_NEXT, "pwrite"))(fd, data, size, offset);
+}
+
+ssize_t real_cuFileWrite(CUfileHandle_t fh, const void *devPtr_base, size_t size, off_t file_offset, off_t devPtr_offset) {
+  return ((real_cuFileWrite_t)dlsym(RTLD_NEXT, "cuFileWrite"))(fh, devPtr_base, size, file_offset, devPtr_offset);
+}
+
+
+#endif /* not _IOBLITZ_H */