I/O: add shared file pointer support (#397)

Author: simonbyrne

deps/consts_microsoftmpi.jl

@@ -96,17 +96,21 @@ const MPI_MODE_UNIQUE_OPEN = reinterpret(Cint, 0x00000020)
 const MPI_MODE_SEQUENTIAL = reinterpret(Cint, 0x00000100)
 const MPI_MODE_APPEND = reinterpret(Cint, 0x00000080)
 
-const MPI_BOTTOM = reinterpret(SentinelPtr, 0)
-const MPI_IN_PLACE = reinterpret(SentinelPtr, -1)
-
-const MPI_STATUS_IGNORE = reinterpret(SentinelPtr, 1)
-const MPI_STATUSES_IGNORE = reinterpret(SentinelPtr, 1)
+const MPI_SEEK_SET = Cint(600)
+const MPI_SEEK_CUR = Cint(602)
+const MPI_SEEK_END = Cint(604)
 
 const MPI_IDENT = Cint(0)
 const MPI_CONGRUENT = Cint(1)
 const MPI_SIMILAR = Cint(2)
 const MPI_UNEQUAL = Cint(3)
 
+const MPI_BOTTOM = reinterpret(SentinelPtr, 0)
+const MPI_IN_PLACE = reinterpret(SentinelPtr, -1)
+
+const MPI_STATUS_IGNORE = reinterpret(SentinelPtr, 1)
+const MPI_STATUSES_IGNORE = reinterpret(SentinelPtr, 1)
+
 struct Status
     _pad1::Cint
     _pad2::Cint

deps/consts_mpich.jl

@@ -101,12 +101,16 @@ const MPI_MODE_UNIQUE_OPEN = Cint(32)
 const MPI_MODE_SEQUENTIAL = Cint(256)
 const MPI_MODE_APPEND = Cint(128)
 
-const MPI_BOTTOM = reinterpret(SentinelPtr, 0)
-const MPI_IN_PLACE = reinterpret(SentinelPtr, -1)
-const MPI_STATUS_IGNORE = reinterpret(SentinelPtr, 1)
-const MPI_STATUSES_IGNORE = reinterpret(SentinelPtr, 1)
+const MPI_SEEK_SET = Cint(600)
+const MPI_SEEK_CUR = Cint(602)
+const MPI_SEEK_END = Cint(604)
 
 const MPI_IDENT = Cint(0)
 const MPI_CONGRUENT = Cint(1)
 const MPI_SIMILAR = Cint(2)
 const MPI_UNEQUAL = Cint(3)
+
+const MPI_BOTTOM = reinterpret(SentinelPtr, 0)
+const MPI_IN_PLACE = reinterpret(SentinelPtr, -1)
+const MPI_STATUS_IGNORE = reinterpret(SentinelPtr, 1)
+const MPI_STATUSES_IGNORE = reinterpret(SentinelPtr, 1)

deps/consts_openmpi.jl

@@ -107,12 +107,17 @@ const MPI_MODE_UNIQUE_OPEN = Cint(32)
 const MPI_MODE_SEQUENTIAL = Cint(256)
 const MPI_MODE_APPEND = Cint(128)
 
-const MPI_BOTTOM = reinterpret(SentinelPtr, 0)
-const MPI_IN_PLACE = reinterpret(SentinelPtr, 1)
-const MPI_STATUS_IGNORE = reinterpret(SentinelPtr, 0)
-const MPI_STATUSES_IGNORE = reinterpret(SentinelPtr, 0)
+const MPI_SEEK_SET = Cint(600)
+const MPI_SEEK_CUR = Cint(602)
+const MPI_SEEK_END = Cint(604)
 
 const MPI_IDENT = Cint(0)
 const MPI_CONGRUENT = Cint(1)
 const MPI_SIMILAR = Cint(2)
 const MPI_UNEQUAL = Cint(3)
+
+const MPI_BOTTOM = reinterpret(SentinelPtr, 0)
+const MPI_IN_PLACE = reinterpret(SentinelPtr, 1)
+const MPI_STATUS_IGNORE = reinterpret(SentinelPtr, 0)
+const MPI_STATUSES_IGNORE = reinterpret(SentinelPtr, 0)
+

deps/gen_consts.jl

@@ -135,6 +135,9 @@ MPI_Cints = [
     :MPI_MODE_UNIQUE_OPEN,
     :MPI_MODE_SEQUENTIAL,
     :MPI_MODE_APPEND,
+    :MPI_SEEK_SET,
+    :MPI_SEEK_CUR,
+    :MPI_SEEK_END,
     :MPI_IDENT,
     :MPI_CONGRUENT,
     :MPI_SIMILAR,

docs/src/io.md

@@ -10,6 +10,7 @@ MPI.File.open
 
 ```@docs
 MPI.File.set_view!
+MPI.File.get_byte_offset
 ```
 
 ## Consistency
@@ -28,3 +29,14 @@ MPI.File.read_at_all!
 MPI.File.write_at
 MPI.File.write_at_all
 ```
+
+### Shared pointer
+
+```@docs
+MPI.File.read_shared!
+MPI.File.write_shared
+MPI.File.read_ordered!
+MPI.File.write_ordered
+MPI.File.seek_shared
+MPI.File.get_position_shared
+```

src/buffers.jl

@@ -18,6 +18,15 @@ function Base.unsafe_convert(::Type{MPIPtr}, x::MPIBuffertype{T}) where T
     ptr = Base.unsafe_convert(Ptr{T}, x)
     reinterpret(MPIPtr, ptr)
 end
+
+
+function Base.cconvert(::Type{MPIPtr}, x::String)
+    x
+end
+function Base.unsafe_convert(::Type{MPIPtr}, x::String)
+    reinterpret(MPIPtr, pointer(x))
+end
+
 function Base.cconvert(::Type{MPIPtr}, ::Nothing)
     reinterpret(MPIPtr, C_NULL)
 end
@@ -123,5 +132,8 @@ Construct a [`Buffer`](@ref) object for a send operation from `data`, allowing c
 `isbits(data)`.
 """
 Buffer_send(data) = isbits(data) ? Buffer(Ref(data)) : Buffer(data)
+Buffer_send(str::String) = Buffer(str, sizeof(str), MPI.CHAR)
+
+
 
 const BUFFER_NULL = Buffer(C_NULL, 0, DATATYPE_NULL)

src/io.jl

@@ -97,6 +97,26 @@ function set_view!(file::FileHandle, disp::Integer, etype::Datatype, filetype::D
     set_view!(file, disp, etype, filetype, datarep, Info(infokwargs...))
 end
 
+"""
+    MPI.File.get_byte_offset(file::FileHandle, offset::Integer)
+
+Converts a view-relative offset into an absolute byte position. Returns the absolute byte
+position (from the beginning of the file) of `offset` relative to the current view of
+`file`.
+
+# External links
+$(_doc_external("MPI_File_get_byte_offset"))
+"""
+function get_byte_offset(file::FileHandle, offset::Integer)
+    r_displ = Ref{MPI_Offset}()
+    # int MPI_File_get_byte_offset(MPI_File fh, MPI_Offset offset,
+    #              MPI_Offset *disp)
+    @mpichk ccall((:MPI_File_get_byte_offset, libmpi), Cint,
+                  (MPI_File, MPI_Offset,  Ptr{MPI_Offset}),
+                  file, offset, r_displ)
+    r_displ[]
+end
+
 """
     MPI.File.sync(fh::FileHandle)
 
@@ -212,6 +232,148 @@ end
 write_at_all(file::FileHandle, offset::Integer, data) = write_at_all(file, offset, Buffer_send(data))
 
 
+# Shared file pointer
+"""
+    MPI.File.read_shared!(file::FileHandle, data)
+
+Reads from `file` using the shared file pointer into `data`.  `data` can be a
+[`Buffer`](@ref), or any object for which `Buffer(data)` is defined.
+
+# See also
+- [`MPI.File.read_ordered!`](@ref) for the collective operation
+
+# External links
+$(_doc_external("MPI_File_read_shared"))
+"""
+function read_shared!(file::FileHandle, buf::Buffer)
+    stat_ref = Ref{Status}(MPI.STATUS_EMPTY)
+    # int MPI_File_read_shared(MPI_File fh, void *buf, int count,
+    #              MPI_Datatype datatype, MPI_Status *status)
+    @mpichk ccall((:MPI_File_read_shared, libmpi), Cint,
+                  (MPI_File, MPIPtr, Cint, MPI_Datatype, Ptr{Status}),
+                  file, buf.data, buf.count, buf.datatype, stat_ref)
+    return stat_ref[]
+end
+read_shared!(file::FileHandle, data) = read_shared!(file, Buffer(data))
+
+"""
+    MPI.File.write_shared(file::FileHandle, data)
+
+Writes to `file` using the shared file pointer from `data`. `data` can be a
+[`Buffer`](@ref), or any object for which `Buffer(data)` is defined.
+
+# See also
+- [`MPI.File.write_ordered`](@ref) for the noncollective operation
+
+# External links
+$(_doc_external("MPI_File_write_shared"))
+"""
+function write_shared(file::FileHandle, buf::Buffer)
+    stat_ref = Ref{Status}(MPI.STATUS_EMPTY)
+    # int MPI_File_write_shared(MPI_File fh, const void *buf, int count,
+    #          MPI_Datatype datatype, MPI_Status *status)
+    @mpichk ccall((:MPI_File_write_shared, libmpi), Cint,
+                  (MPI_File, MPIPtr, Cint, MPI_Datatype, Ptr{Status}),
+                  file, buf.data, buf.count, buf.datatype, stat_ref)
+    return stat_ref[]
+end
+write_shared(file::FileHandle, buf) = write_shared(file, Buffer_send(buf))
+
+# Shared collective operations
+"""
+    MPI.File.read_ordered!(file::FileHandle, data)
+
+Collectively reads in rank order from `file` using the shared file pointer into `data`.
+`data` can be a [`Buffer`](@ref), or any object for which `Buffer(data)` is defined. This
+is a collective operation, so must be called on all ranks in the communicator on which
+`file` was opened.
+
+# See also
+- [`MPI.File.read_shared!`](@ref) for the non-collective operation
+
+# External links
+$(_doc_external("MPI_File_read_ordered"))
+"""
+function read_ordered!(file::FileHandle, buf::Buffer)
+    stat_ref = Ref{Status}(MPI.STATUS_EMPTY)
+    # int MPI_File_read_ordered(MPI_File fh, void *buf, int count,
+    #              MPI_Datatype datatype, MPI_Status *status)
+    @mpichk ccall((:MPI_File_read_ordered, libmpi), Cint,
+                  (MPI_File, MPIPtr, Cint, MPI_Datatype, Ptr{Status}),
+                  file, buf.data, buf.count, buf.datatype, stat_ref)
+    return stat_ref[]
+end
+read_ordered!(file::FileHandle, data) = read_ordered!(file, Buffer(data))
+
+"""
+    MPI.File.write_ordered(file::FileHandle, data)
+
+Collectively writes in rank order to `file` using the shared file pointer from `data`.
+`data` can be a [`Buffer`](@ref), or any object for which `Buffer(data)` is defined. This
+is a collective operation, so must be called on all ranks in the communicator on which
+`file` was opened.
+
+# See also
+- [`MPI.File.write_shared`](@ref) for the noncollective operation
+
+# External links
+$(_doc_external("MPI_File_write_ordered"))
+"""
+function write_ordered(file::FileHandle, buf::Buffer)
+    stat_ref = Ref{Status}(MPI.STATUS_EMPTY)
+    # int MPI_File_write_ordered(MPI_File fh, const void *buf, int count,
+    #              MPI_Datatype datatype, MPI_Status *status)    
+    @mpichk ccall((:MPI_File_write_ordered, libmpi), Cint,
+                  (MPI_File, MPIPtr, Cint, MPI_Datatype, Ptr{Status}),
+                  file, buf.data, buf.count, buf.datatype, stat_ref)
+    return stat_ref[]
+end
+write_ordered(file::FileHandle, buf) = write_ordered(file, Buffer_send(buf))
 
+# seek
+@enum Seek begin
+    SEEK_SET = MPI.MPI_SEEK_SET
+    SEEK_CUR = MPI.MPI_SEEK_CUR
+    SEEK_END = MPI.MPI_SEEK_END
+end    
+
+"""
+    MPI.File.seek_shared(file::FileHandle, offset::Integer, whence::Seek=SEEK_SET)
+
+Updates the shared file pointer according to `whence`, which has the following possible
+values:
+
+- `MPI.File.SEEK_SET` (default): the pointer is set to `offset`
+- `MPI.File.SEEK_CUR`: the pointer is set to the current pointer position plus `offset`
+- `MPI.File.SEEK_END`: the pointer is set to the end of file plus `offset`
+
+This is a collective operation, and must be called with the same value on all processes in
+the communicator.
+
+# External links
+$(_doc_external("MPI_File_seek_shared"))
+"""
+function seek_shared(file::FileHandle, offset::Integer, whence::Seek=SEEK_SET)
+    # int MPI_File_seek_shared(MPI_File fh, MPI_Offset offset, int whence)
+    @mpichk ccall((:MPI_File_seek_shared, libmpi), Cint,
+                  (MPI_File, MPI_Offset, Cint),
+                  file, offset, whence)
+end
+
+"""
+    MPI.File.get_position_shared(file::FileHandle)
+
+The current position of the shared file pointer (in `etype` units) relative to the current view.
+
+# External links
+$(_doc_external("MPI_File_get_position_shared"))
+"""
+function get_position_shared(file::FileHandle)
+    r = Ref{MPI_Offset}()
+    # int MPI_File_get_position_shared(MPI_File fh, MPI_Offset *offset)
+    @mpichk ccall((:MPI_File_get_position_shared, libmpi), Cint,
+                  (MPI_File, Ptr{MPI_Offset}), file, r)
+    return r[]
+end
 
 end # module

test/test_io_shared.jl

@@ -0,0 +1,68 @@
+using Test
+using MPI
+using Random
+
+if get(ENV,"JULIA_MPI_TEST_ARRAYTYPE","") == "CuArray"
+    using CuArrays
+    ArrayType = CuArray
+else
+    ArrayType = Array
+end
+
+MPI.Init()
+comm = MPI.COMM_WORLD
+rank = MPI.Comm_rank(comm)
+sz = MPI.Comm_size(comm)
+filename = MPI.bcast(tempname(), 0, comm)
+
+MPI.Barrier(comm)
+
+# Collective write
+fh = MPI.File.open(comm, filename, read=true, write=true, create=true)
+@test MPI.File.get_position_shared(fh) == 0
+
+header = "my header"
+
+if rank == 0
+    MPI.File.write_shared(fh, header)
+end
+
+# TODO: is there a better way to synchronise shared pointers?
+MPI.Barrier(comm)
+
+offset = MPI.File.get_position_shared(fh)
+@test offset == sizeof(header)
+byte_offset = MPI.File.get_byte_offset(fh, offset)
+@test byte_offset == offset
+
+MPI.File.set_view!(fh, byte_offset, MPI.Datatype(Int64), MPI.Datatype(Int64))
+@test MPI.File.get_position_shared(fh) == 0
+
+MPI.File.write_ordered(fh, fill(Int64(rank), rank+1))
+@test MPI.File.get_position_shared(fh) == sum(1:sz)
+
+MPI.File.seek_shared(fh, 0)
+@test MPI.File.get_position_shared(fh) == 0
+
+buf = zeros(Int64, rank+1)
+MPI.File.read_ordered!(fh, buf)
+@test buf == fill(Int64(rank), rank+1)
+
+MPI.Barrier(comm)
+@test MPI.File.get_position_shared(fh) == sum(1:sz)
+
+MPI.File.set_view!(fh, 0, MPI.Datatype(UInt8), MPI.Datatype(UInt8))
+MPI.File.seek_shared(fh, 0)
+@test MPI.File.get_position_shared(fh) == 0
+
+if rank == sz-1
+    buf = Array{UInt8}(undef, sizeof(header))
+    MPI.File.read_shared!(fh, buf)
+    @test String(buf) == header
+end
+
+MPI.Barrier(comm)
+
+@test MPI.File.get_position_shared(fh) == sizeof(header)
+
+close(fh)