MPI IO Fileviews
Any MPI file, like all files, is a linear sequence of bytes. We could store any data in any order we wished by careful computation of offsets. However, this rapidly becomes complicated and error-prone for more complex data structures. Fortunately, MPI provides the fileview to help with this task. The fileview is the “window” into the file that each process sees. In this way, we can define the global structure without having to allocate space for it on any process.
One of the most common data structures for numerically-intensive programming is the multidimensional array. We have used them for many examples. To create a fileview, we assign a portion of the array to each process, generally by the usual topology computation.
To accomplish this, we use a subarray. When we examined subarrays previously, we created subarrays as subsets of the local array on each process. For a fileview, we will create a subarray relative to the global array. We use the topology to compute the start and stop positions for row and column (Fortran programmers, remember we count from 0). The global dimensions make up the size and the local dimensions are the subsizes. Remember that the global array does not need to exist on the process; we are telling MPI how to subdivide it.
After defining the subarray we pass it to MPI_File_set_view, with the file handle of a previously-opened MPI file. We can then call MPI_File_write<_all> or MPI_file_read<_all> as we have done before, passing it the local data array.
MPI_File_set_view(fh, offset, elem_type, subtype, repr, info);
MPI_File_write_all(fh, loc_dat, count , elem_type, status);
call MPI_FILE_SET_VIEW(fh, offset, elem_type, subtype, repr, info)
call MPI_FILE_WRITE_ALL(fh,loc_data, count, elem_type, status)
```python
fh.Set_view(offset, elem_type, subtype, repr, info)
fh.Write_all(local_dat)
In this syntax, elem_type is the elementary type and subtype is the subarray we defined.
The repr is a string indicating the data representation format. It can take values:
«{ table >}}
| Value | Representation |
|---|---|
| ’native' | bytes are dumped directly from memory. |
| ‘internal’ | the representation used internally by the specific MPI implementation. |
| ’external` | The XDR portable data formats are used. |
XDR (External Data Representation) is a standard intended to make data easily convertible from one representation to another. It was especially important before computer systems largely settled on one standard; however, data representation can still vary based on platform and XDR is still used in a variety of applications, including network transmissions.
MPI guarantees full interoperability within an environment, so native is often used for the data representation if the file will be written and read on the same platform. Otherwise the representation external32 for 32-bit XDR data is widely used.
A full example for each language is below:
C++
Contents of mpi_io_fileview.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;
int nrows, ncols;
// Check number of parameters and read in filename
if (argc < 2) {
printf ("USAGE: %s output-file <nrows> <ncols>\n", argv[0]);
exit(1);
}
const char *fname=argv[1];
if (argc == 2) {
nrows=4;
ncols=4;
}
else if (argc == 3) {
nrows=atoi(argv[2]);
ncols=nrows;
}
else if (argc == 4) {
nrows=atoi(argv[2]);
ncols=atoi(argv[3]);
}
//Initialize MPI
MPI_Init(&argc, &argv);
MPI_Comm_size(MPI_COMM_WORLD,&nprocs);
MPI_Comm_rank(MPI_COMM_WORLD,&rank);
if (rows*cols != nprocs) {
cout<<"Number of rows times columns does not equal nprocs\n";
MPI_Finalize();
return 1;
}
// Hard-code sizes so we can see what we're doing
int nrl = 4;
int ncl = 4;
int N=nrl*nrows;
int M=ncl*ncols;
//Set up the topology
int lrow=rank/ncols;
int lcol=rank%ncols;
int **loc_u=new int*[nrl];
int *luptr=new int[(nrl)*(ncl)];
for (int i=0;i<nrl;++i,luptr+=ncl) {
loc_u[i] = luptr;
}
for ( int i = 0; i < nrl; i++ ) {
for (int j = 0; j < ncl; j++ ) {
loc_u[i][j] = rank+1;
}
}
int lusize=nrl*ncl;
//Write each segment to conventional file (the "old way")
ofstream fp;
string myfile=argv[1]+to_string(rank);
fp.open(myfile,ios::out);
for (int i = 0; i < nrl; i++) {
for (int j = 0; j < ncl; j++) {
fp<<loc_u[i][j]<<" ";
}
fp<<"\n";
}
fp.close();
int gdims[]={N,M};
int ldims[]={nrl,ncl};
int starts[]={ncl*lrow,nrl*lcol};
cout<<"Starts for rank "<<rank<<" are "<<starts[0]<<" "<<starts[1]<<endl;
MPI_Datatype locarray;
MPI_Type_create_subarray(2,gdims,ldims,starts,MPI_ORDER_C,MPI_INT,&locarray);
MPI_Type_commit(&locarray);
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) {
cout<<"Unable to open file for writing\n";
MPI_Finalize();
exit(2);
}
offset=0;
MPI_File_set_view(fh, offset, MPI_INT, locarray, "native", MPI_INFO_NULL);
MPI_File_write_all(fh,&loc_u[0][0], lusize , MPI_INT, &mpi_stat);
MPI_File_close(&fh);
//All processes read 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);
}
int **rbuf=new int*[nrl];
int *rptr=new int[(nrl)*(ncl)];
for (int i=0;i<nrl;++i,rptr+=ncl) {
rbuf[i] = rptr;
}
for ( int i = 0; i < nrl; i++ ) {
for (int j = 0; j < ncl; j++ ) {
rbuf[i][j] = 0;
}
}
MPI_File_set_view(fh, offset, MPI_INT, locarray, "native", MPI_INFO_NULL);
MPI_File_read_all(fh, &rbuf[0][0], lusize, MPI_INT, &mpi_stat);
MPI_File_close(&fh);
//Usual trick to print one rank at a time
int position;
int *u = new int[lusize];
if ( rank == 0 ) {
cout<<"Result for rank 0\n";
for (int i=0;i<nrl;i++) {
for (int j=0;j<ncl;j++) {
cout<<rbuf[i][j]<<" ";
}
cout<<endl;
}
for (int n=1;n<nprocs;++n) {
memset(u,0.,lusize);
MPI_Recv(u,lusize,MPI_INT,MPI_ANY_SOURCE,MPI_ANY_TAG,MPI_COMM_WORLD,&mpi_stat);
cout<<"Result for rank "<<mpi_stat.MPI_SOURCE<<endl;
position=0;
for (int i=0;i<nrl;i++) {
for (int j=0;j<ncl;j++) {
cout<<u[position++]<<" ";
}
cout<<endl;
}
}
}
else {
// Pack the 2D array into the buffer
position=0;
for (int i=0; i<nrl; i++)
for (int j=0;j<ncl; j++)
u[position++]=rbuf[i][j];
MPI_Send(u,lusize, MPI_INT, 0, rank, MPI_COMM_WORLD);
}
MPI_Type_free(&locarray);
MPI_Finalize();
}
Fortran
Contents of mpi_io_fileview.f90
program mpiwrite
use mpi_f08
implicit none
integer :: i,j
character(len=80) :: arg
integer :: numargs
integer :: nrows, ncols
integer :: nrl, ncl, lrow, lcol
integer :: N, M
integer :: rank, nprocs
integer :: mpi_err
integer, parameter :: root=0
type(MPI_Status) :: mpi_stat
type(MPI_File) :: fh
type(MPI_Datatype) :: locarray
integer :: amode
integer :: lusize
integer, dimension(2) :: gdims, ldims, starts
integer(kind=MPI_OFFSET_KIND) :: offset=0
integer, allocatable, dimension(:,:) :: loc_u, u, rbuf
character(len=24) :: fname
character(len=80) :: myfile
! 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 <nrows> <ncols>'
else
call get_command_argument(1,fname)
nrows=4
ncols=4
endif
if (numargs .eq. 2) then
call get_command_argument(2,arg)
read(arg,'(i4)') nrows
ncols=nrows
else if (numargs .eq. 3) then
call get_command_argument(2,arg)
read(arg,'(i4)') nrows
call get_command_argument(3,arg)
read(arg,'(i4)') ncols
endif
!Initialize MPI
call MPI_INIT()
call MPI_COMM_SIZE(MPI_COMM_WORLD,nprocs)
call MPI_COMM_RANK(MPI_COMM_WORLD,rank)
if (nrows*ncols /= nprocs) then
call MPI_Finalize()
stop "Number of rows times columns does not equal nprocs"
endif
! Hard-code sizes so we can see what we're doing
nrl = 4
ncl = 4
N=nrl*nrows
M=ncl*ncols
!Set up the topology
lrow=rank/ncols
lcol=mod(rank,ncols)
allocate(loc_u(nrl,ncl))
loc_u=(rank+1)
!Write each segment to conventional file (the "old way")
write(myfile,'(a,i2.2)') fname(1:len_trim(fname)),rank
open(unit=10,file=myfile,status='unknown')
do i=1,nrl
write (10,'(*(g0,1x))') loc_u(i,:)
enddo
close(10)
gdims=[N,M]
ldims=[nrl,ncl]
starts=[ncl*lrow,nrl*lcol]
write(*,*) rank,starts
call MPI_TYPE_CREATE_SUBARRAY(2, gdims, ldims, starts, MPI_ORDER_FORTRAN, &
MPI_INTEGER, locarray)
call MPI_TYPE_COMMIT(locarray)
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 writing"
endif
offset=0
call MPI_FILE_SET_VIEW(fh, offset, MPI_INTEGER, locarray, &
'native', MPI_INFO_NULL)
call MPI_FILE_WRITE_ALL(fh,loc_u, size(loc_u), MPI_INTEGER, mpi_stat)
call MPI_FILE_CLOSE(fh)
!All processes read file
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
allocate(u(nrl,ncl),rbuf(nrl,ncl))
rbuf=0
lusize=size(u)
call MPI_FILE_SET_VIEW(fh, offset, MPI_INTEGER, locarray, &
'native', MPI_INFO_NULL)
call MPI_FILE_READ_ALL(fh, rbuf, lusize, MPI_INTEGER, mpi_stat)
call MPI_FILE_CLOSE(fh)
!Usual trick to print one rank at a time
if (rank==0) then
write(*,*) 'Result for rank 0'
do j=1,nrl
write(*,'(*(i10))') rbuf(j,:)
enddo
do i=1,nprocs-1
u=0
call MPI_Recv(u,lusize, MPI_INTEGER,MPI_ANY_SOURCE, &
MPI_ANY_TAG, MPI_COMM_WORLD,mpi_stat)
write(*,*) 'Result for rank',mpi_stat%MPI_SOURCE
do j=1,nrl
write(*,'(*(i10))') u(j,:)
enddo
enddo
else
call MPI_Send(rbuf,lusize,MPI_INTEGER,0,rank,MPI_COMM_WORLD)
endif
call MPI_TYPE_FREE(locarray)
call MPI_Finalize()
end program
Python
Contents of mpi_io_fileview.py
import sys
import numpy as np
from mpi4py import MPI
if len(sys.argv)<2:
print("Usage: filename <opt> nrows <opt> ncols")
exit()
else:
filename=sys.argv[1]
if len(sys.argv)==2:
nrows=4
ncols=4
elif len(sys.argv)==3:
nrows=int(float(sys.argv[2]))
ncols=nrows
elif len(sys.argv)==4:
nrows=int(float(sys.argv[2]))
ncols=int(float(sys.argv[3]))
comm = MPI.COMM_WORLD
rank = comm.Get_rank()
nprocs = comm.Get_size()
if nrows*ncols != nprocs:
print("Number of rows times columns does not equal nprocs")
sys.exit()
# Hard-code local array size so we can see what we're doing
nrl = 4
ncl = 4
N=nrl*nrows
M=ncl*ncols
#Set up the topology
lrow=rank//ncols
lcol=rank%ncols
ndims=2
gdims=np.array([N,M],dtype='int')
ldims=np.array([nrl,ncl],dtype='int')
starts=np.array([ncl*lrow,nrl*lcol],dtype='int')
print(rank,starts)
#Generate arbitrary values.
loc_u=np.ones((nrl,ncl),dtype="int")*(rank+1)
#Write each segment to conventional file (the "old way")
myfile=f'{filename:}{rank:02}'
myfh=open(myfile,'w')
for i in range(nrl):
print(loc_u[i,:],file=myfh)
myfh.close()
locarr=MPI.INT.Create_subarray(gdims,ldims, starts, order=MPI.ORDER_C)
locarr.Commit()
amode=MPI.MODE_CREATE | MPI.MODE_WRONLY
fh=MPI.File.Open(comm,filename,amode)
#No header
disp=0
fh.Set_view(disp,MPI.INT,locarr,"native",MPI.INFO_NULL)
fh.Write_all(loc_u)
fh.Close()
amode=MPI.MODE_RDONLY
fh=MPI.File.Open(comm,filename,amode)
buf=np.zeros_like(loc_u,dtype='int')
fh.Set_view(disp,MPI.INT,locarr,"native",MPI.INFO_NULL)
fh.Read_all([buf, MPI.INT])
fh.Close()
#Usual trick to show one rank at a time
status=MPI.Status()
if rank==0:
u=np.zeros_like(buf)
print("Read for rank 0")
print(buf)
for i in range(1,nprocs):
comm.Recv([u,MPI.INT],source=MPI.ANY_SOURCE,tag=MPI.ANY_TAG,status=status)
print("Read for rank ",status.Get_source())
print(u)
else:
comm.Send([buf,MPI.INT],dest=0,tag=rank)
comm.Barrier()