Great question! Handling non-contiguous, column-based parallel I/O for a table where you need to evaluate columns on-the-fly (and only keep the ones worth storing) is a super common pain point when moving from single-process C++ code to MPI parallelism. Let’s break down how to tackle this step by step.

1. First: Assign Columns to Processes #

Since you’re working with columns, start by splitting the column workload across your MPI processes. This avoids redundant work and keeps each process focused on a manageable subset:

• Use a cyclic distribution: Each process rank handles columns rank, rank+nprocs, rank+2*nprocs, .... This works even if the total number of columns n isn’t perfectly divisible by the number of processes nprocs.
• If you don’t know n upfront (total columns), have one process first read the table’s metadata/header to get this number, then broadcast it to all other processes using MPI_Bcast so everyone knows their assigned columns.

2. Use MPI Datatypes for Column-Based Reads #

MPI’s real power here comes from custom datatypes, which let you read non-contiguous data (like columns in a row-major stored table) as a single contiguous block from the file system. Here’s how to set this up:

• Define a datatype that represents an entire column. For an m×n table stored as row-major integers, each element in the column is n ints apart (since each row has n elements). Example code:

  MPI_Datatype mpi_column_type;
  // m = number of rows, n = total columns
  MPI_Type_vector(m, 1, n, MPI_INT, &mpi_column_type);
  MPI_Type_commit(&mpi_column_type);
  

  Let’s break this down:
  • count = m: Number of elements in the column (one per row)
  • blocklength = 1: Each "block" is a single integer
  • stride = n: Skip n-1 integers between blocks to jump to the next row’s column value
• Once the datatype is committed, each process can read its assigned columns directly. Here’s a loop example:

  MPI_File fh;
  MPI_Status status;
  int nprocs, rank;
  MPI_Comm_size(MPI_COMM_WORLD, &nprocs);
  MPI_Comm_rank(MPI_COMM_WORLD, &rank);
  
  // Assume n (total columns) is known via broadcast or metadata read
  int column_idx = rank;
  while (column_idx < n) {
      // Allocate temp buffer for the column (only needed during processing)
      int* col_buf = (int*)malloc(m * sizeof(int));
      
      // Seek to the start of the target column (offset in bytes)
      MPI_File_seek(fh, column_idx * sizeof(int), MPI_SEEK_SET);
      // Read the entire column using our custom datatype
      MPI_File_read(fh, col_buf, 1, mpi_column_type, &status);
      
      // --- Your column processing logic here ---
      // Calculate statistics, decide if this column needs to stay in memory
      
      // Free the temp buffer if we don't need to keep the column
      if (!should_keep_column(col_buf)) {
          free(col_buf);
      } else {
          // Add col_buf to a local list of kept columns for later use
      }
      
      // Move to the next column assigned to this process
      column_idx += nprocs;
  }
  
  // Clean up the custom datatype when done
  MPI_Type_free(&mpi_column_type);
  

3. Handling Dynamic Memory for Kept Columns #

Since each process manages its own columns, you don’t need cross-process coordination unless you need to aggregate results later:

• If you need to collect all kept columns into a single process (e.g., for final output), use MPI_Gatherv (since the number of kept columns per process may vary—this handles variable-sized data).
• For distributed workflows where each process keeps its own useful columns, just maintain a local vector or list of persistent buffers—no extra communication needed unless you need to share statistics or results.

4. Key Optimizations to Boost Performance #

• Batch Reads: If your file system supports it, create a datatype that reads multiple columns at once per process to reduce the number of I/O calls.
• MPI Info Hints: When opening the file with MPI_File_open, pass hints (like striping_factor for parallel file systems) via MPI_INFO to optimize how the file system handles parallel access.
• Avoid Tiny I/O: If m (rows) is small, consider reading larger chunks of the file (e.g., full rows) and extracting your assigned columns locally—this reduces the overhead of multiple small read operations.

5. Alternative: Use High-Level Libraries (If You Want to Avoid Low-Level MPI) #

If writing custom MPI datatypes feels too tedious, use libraries built on MPI that simplify columnar parallel I/O:

• HDF5: Natively supports columnar storage and parallel I/O. You can store each column as a separate dataset or use compound datatypes, then read columns in parallel with minimal code.
• NetCDF: Designed for scientific data, with great support for parallel I/O and non-contiguous access to tabular data.

Quick Note for Text Files (Like CSV):
If your table is in a text format (not binary), the datatype approach won’t work directly (rows are variable-length due to commas/newlines). In this case:

1. Convert the CSV to a binary format first (using a single process) for easier parallel access.
2. Or, have each process read entire rows via collective I/O, then extract their assigned columns locally. This is less efficient but works for text data.

内容的提问来源于stack exchange，提问作者Sean