MPI IO Example
We have covered only a few of the many MPI IO procedures available, but enough to write a working example so we can see how they fit together. We will assign a single-character upper-case letter, starting with A, to each process. We will have each process write its character to a file for N repetitions. So in particular, if we have 4 processes and we ask for 3 repetitions, the file should contain
ABCDABCDABCD
Each character occupies a byte (so for Python we will use the MPI_BYTE type) which simplifies computing the offset. We will use MPI_File_write_at since this is a small write, and the overhead of a collective write might be excessive.
We will then have each process read the file created and print it to verify that it is as expected.
We can illustrate the file layout with offsets
The offset is 4 bytes for each processes and repetition. If we loop over a variable i, then some thought shows that the offset for each process is
offset=rank+(i-1)*nprocs
where i is the loop index over the number of repetitions.
Examples for each language
C++
Contents of mpi_io_example.cxx
#include <cstring>
#include <cstdio>
#include <iostream>
#include <fstream>
#include <string>
#include <mpi.h>
using namespace std;
int main (int argc, char *argv[]) {
// Declarations for MPI
int rank, nprocs;
int errcode;
MPI_Status mpi_stat;
MPI_Info info;
MPI_Offset offset;
MPI_File fh;
int root=0, tag=0;
int mpi_err;
// Check number of parameters and read in filename
if (argc < 2) {
printf ("USAGE: %s output-file\n", argv[0]);
exit(1);
}
const char *fname=argv[1];
//Initialize MPI
MPI_Init(&argc, &argv);
MPI_Comm_size(MPI_COMM_WORLD,&nprocs);
MPI_Comm_rank(MPI_COMM_WORLD,&rank);
int ord=int('A')+rank;
char my_char=(char)ord;
//Fun with pointers (sometimes fh is dereferenced, sometimes not)
int amode=MPI_MODE_CREATE | MPI_MODE_WRONLY;
mpi_err=MPI_File_open(MPI_COMM_WORLD,fname,amode,MPI_INFO_NULL,&fh);
if (mpi_err != MPI_SUCCESS) {
MPI_Finalize();
exit(2);
}
int nreps=20;
for (int i=0;i<nreps;i++) {
offset=rank+i*nprocs;
MPI_File_write_at(fh,offset,&my_char,1,MPI_CHAR,&mpi_stat);
}
MPI_File_close(&fh);
// Read back in as ordinary file
// We don't care about efficiency here
if (rank==root) {
ifstream fp(fname, ios::in);
string sstr((istreambuf_iterator<char>(fp)), istreambuf_iterator<char>());
cout<<sstr<<endl;
fp.close();
}
//All processes read entire file
//Blocking so will wait for root to finish above read
amode=MPI_MODE_RDONLY;
mpi_err=MPI_File_open(MPI_COMM_WORLD,fname,amode,MPI_INFO_NULL,&fh);
if ( mpi_err != MPI_SUCCESS) {
MPI_Finalize();
cout<<"Unable to open MPI file for reading\n";
exit(2);
}
MPI_Offset fsize;
int tsize;
MPI_File_get_size(fh,&fsize);
MPI_Type_size(MPI_CHAR,&tsize);
if (rank==root) {
cout<<"File size is "<<fsize<<" Type size is "<<tsize<<" bytes\n";
}
int nchar=fsize/tsize;
char fbuf[nchar];
MPI_File_read_all(fh, fbuf, nchar, MPI_CHAR, &mpi_stat);
MPI_File_close(&fh);
cout<<"Rank "<<rank<<" ";
for (int i=0; i<nchar; ++i) {
cout<<fbuf[i];
}
cout<<endl;
//Read back in as MPI file
amode=MPI_MODE_RDONLY;
mpi_err=MPI_File_open(MPI_COMM_WORLD,fname,amode,MPI_INFO_NULL,&fh);
if (mpi_err != MPI_SUCCESS) {
MPI_Finalize();
cout<<"Unable to open MPI file for reading\n";
exit(2);
}
char rbuf[nreps];
offset=0;
for (int i=0;i<nreps;i++) {
offset=rank+i*nprocs;
MPI_File_read_at(fh,offset,&rbuf[i],1,MPI_CHAR,&mpi_stat);
}
cout<<"Rank "<<rank<<" ";
for (int i=0; i<nreps; ++i) {
cout<<rbuf[i];
}
cout<<endl;
MPI_File_close(&fh);
MPI_Finalize();
}
Fortran
Contents of mpi_io_example.f90
program mpiwrite
use mpi_f08
implicit none
integer :: i
character(len=80) :: arg
integer :: numargs
integer :: rank, nprocs
integer :: mpi_err
integer, parameter :: root=0
type(MPI_Status) :: mpi_stat
type(MPI_File) :: fh
integer :: amode
integer :: tsize
INTEGER(KIND=MPI_OFFSET_KIND) :: fsize, offset
character(len=24) :: fname
character(len=1) :: my_char
integer :: nreps
character(len=:), allocatable :: gu,u
character(len=256) :: rbuf
! check number of parameters and read in filename
! all ranks do this, avoids broadcast
numargs=command_argument_count()
if (numargs .lt. 1) then
stop 'USAGE: output-file'
else
call get_command_argument(1,fname)
endif
!Initialize MPI
call MPI_INIT()
call MPI_COMM_SIZE(MPI_COMM_WORLD,nprocs)
call MPI_COMM_RANK(MPI_COMM_WORLD,rank)
my_char=char(iachar('A')+rank)
amode=ior(MPI_MODE_CREATE, MPI_MODE_WRONLY)
call MPI_FILE_OPEN(MPI_COMM_WORLD,trim(fname),amode,MPI_INFO_NULL,fh,mpi_err)
if ( mpi_err /= MPI_SUCCESS) then
call MPI_FINALIZE()
stop "Unable to open MPI file for reading"
endif
nreps=20
do i=1,nreps
! Explcit cast for offset not really necessary
offset=int(rank+(i-1)*nprocs,kind=MPI_OFFSET_KIND)
call MPI_FILE_WRITE_AT(fh, offset, my_char, 1, MPI_CHARACTER, mpi_stat)
enddo
call MPI_FILE_CLOSE(fh)
! Read as ordinary file
! Still can't read variable-length strings in Fortran
if (rank==root) then
rbuf=' '
open(unit=10,file=fname,status='unknown')
read(10,*) rbuf
gu=trim(rbuf)
write(*,'(a,a)') "Read at root ",gu
close(10)
endif
!All processes read entire file
!Blocking so will wait for root to finish above read
amode=MPI_MODE_RDONLY
call MPI_FILE_OPEN(MPI_COMM_WORLD,trim(fname),amode,MPI_INFO_NULL,fh,mpi_err)
if ( mpi_err /= MPI_SUCCESS) then
call MPI_FINALIZE()
stop "Unable to open MPI file for reading"
endif
!Reading into "big enough" buffer means we don't need to get the size,
!but this is the syntax. Note type of fsize in declarations
call MPI_FILE_GET_SIZE(fh,fsize)
call MPI_TYPE_SIZE(MPI_CHARACTER,tsize)
if (rank==root) then
write(*,*) "File size is ",fsize," Type size is ",tsize," bytes"
endif
rbuf=' '
call MPI_FILE_READ_ALL(fh, rbuf, len(rbuf), MPI_CHARACTER, mpi_stat)
gu=trim(rbuf)
write(*,'(a,i4.4,a,a)') 'Full file at rank ',rank,' ',gu
!Read back the MPI file portion for reach rank
rbuf=' '
amode=MPI_MODE_RDONLY
call MPI_FILE_OPEN(MPI_COMM_WORLD,trim(fname),amode,MPI_INFO_NULL,fh,mpi_err)
if ( mpi_err /= MPI_SUCCESS) then
call MPI_FINALIZE()
stop "Unable to open MPI file for reading"
else
do i=1,nreps
call MPI_FILE_READ_AT(fh, offset, rbuf(i:i), 1, MPI_CHARACTER, mpi_stat)
enddo
endif
write(*,'(a,i4.4,a,a)') 'Rank ',rank,' ',trim(rbuf)
call MPI_FILE_CLOSE(fh)
call MPI_Finalize()
end program
Python
Contents of mpi_io_example.py
import sys
import numpy as np
from mpi4py import MPI
if len(sys.argv)<2:
print("Usage: filename")
exit()
else:
filename=sys.argv[1]
comm=MPI.COMM_WORLD
rank=comm.Get_rank()
nprocs=comm.Get_size()
root=0
my_char=chr(ord('A') + rank).encode('ascii')
# Set up the buffer to contain a one-byte character (consistent with C char)
buf=np.array([my_char])
status=MPI.Status()
#Info is used for "hints" to various MPI routines. Usually we don't need it.
info=MPI.Info()
amode=MPI.MODE_CREATE | MPI.MODE_WRONLY
fh=MPI.File.Open(comm,filename,amode,info)
#Use MPI Byte type since Python doesn't really support characters
nreps=20
for i in range(nreps):
offset=rank+i*nprocs
fh.Write_at(offset, [buf, MPI.BYTE],status=status)
fh.Close()
if rank==0:
#First read back with ordinary IO
with open(filename,'rb') as fp:
array=np.fromfile(fp,dtype='byte')
array=[chr(array[i]) for i in range(array.size)]
print("Read at root "+"".join(array))
fp.close()
#All processes read entire file
#Blocking so will wait for root to finish above read
amode=MPI.MODE_RDONLY
fh=MPI.File.Open(comm,filename,amode)
fsize=fh.Get_size()
itembytes=MPI.BYTE.Get_size()
if rank==root:
print("File size is "+str(fsize)+" type size is "+str(itembytes)+" bytes")
nbytes=fsize//itembytes
rbuf=np.empty((nbytes,),dtype='byte')
fh.Read_all([rbuf,MPI.BYTE])
all_chars=[chr(rbuf[i]) for i in range(rbuf.size)]
print("Entire file at rank "+str(rank)+" "+''.join(all_chars))
fh.Close()
#Read back the MPI file portion for each rank
amode=MPI.MODE_RDONLY
fh=MPI.File.Open(comm,filename,amode)
my_vals=[]
offset=0
for i in range(nreps):
offset=rank+i*nprocs
fh.Read_at(offset, [buf, MPI.BYTE])
my_vals.append(buf[0].decode())
fh.Close()
my_string=''.join(my_vals)
print(str(rank)+' '+my_string)