Skip to content

Commit 5cb223f

Browse files
committed
Update for Julia v0.7
1 parent aed7b03 commit 5cb223f

File tree

5 files changed

+37
-32
lines changed

5 files changed

+37
-32
lines changed

src/window-cman.jl

Lines changed: 12 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,6 @@
1-
import Base: launch, kill, manage, connect
1+
import Base: kill
2+
using Distributed
3+
import Distributed: launch, kill, manage, connect
24
export MPIWindowIOManager, launch, kill, manage, connect, @cluster
35

46
"""
@@ -12,8 +14,8 @@ mutable struct MPIWindowIOManager <: ClusterManager
1214

1315
function MPIWindowIOManager(comm::MPI.Comm, workers_wait::Bool)
1416
nb_procs = MPI.Comm_size(comm)
15-
connection_windows = Vector{WindowIO}(nb_procs)
16-
stdio_windows = Vector{WindowIO}(nb_procs)
17+
connection_windows = Vector{WindowIO}(undef,nb_procs)
18+
stdio_windows = Vector{WindowIO}(undef,nb_procs)
1719

1820
for i in 1:nb_procs
1921
connection_windows[i] = WindowIO(comm)
@@ -62,7 +64,7 @@ function connect(mgr::MPIWindowIOManager, pid::Int, config::WorkerConfig)
6264
myrank = MPI.Comm_rank(mgr.comm)
6365
if myrank == 0
6466
proc_stdio = mgr.stdio_windows[pid]
65-
@schedule while !eof(proc_stdio)
67+
@async while !eof(proc_stdio)
6668
try
6769
println("\tFrom worker $(pid):\t$(readline(proc_stdio))")
6870
catch e
@@ -74,7 +76,7 @@ end
7476

7577
function redirect_to_mpi(s::WindowWriter)
7678
(rd, wr) = redirect_stdout()
77-
@schedule while !eof(rd) && isopen(s.winio)
79+
@async while !eof(rd) && isopen(s.winio)
7880
av = readline(rd)
7981
if isopen(s.winio)
8082
println(s,av)
@@ -112,13 +114,13 @@ function start_window_worker(comm::Comm, workers_wait)
112114

113115
manager = MPIWindowIOManager(comm, workers_wait)
114116
cookie = string(comm)
115-
if length(cookie) > Base.Distributed.HDR_COOKIE_LEN
116-
cookie = cookie[1:Base.Distributed.HDR_COOKIE_LEN]
117+
if length(cookie) > Distributed.HDR_COOKIE_LEN
118+
cookie = cookie[1:Distributed.HDR_COOKIE_LEN]
117119
end
118120

119121
try
120122
if rank == 0
121-
Base.cluster_cookie(cookie)
123+
Distributed.cluster_cookie(cookie)
122124
MPI.Barrier(comm)
123125
addprocs(manager)
124126
@assert nprocs() == N
@@ -134,14 +136,14 @@ function start_window_worker(comm::Comm, workers_wait)
134136
redirect_to_mpi(WindowWriter(manager.stdio_windows[rank+1], 0))
135137
for i in vcat([1], (rank+2):N)
136138
# Receiving end of connections to all higher workers and master
137-
Base.process_messages(manager.connection_windows[i], WindowWriter(manager.connection_windows[rank+1], i-1))
139+
Distributed.process_messages(manager.connection_windows[i], WindowWriter(manager.connection_windows[rank+1], i-1))
138140
end
139141

140142
global _stop_requested = Condition()
141143
wait_for_events()
142144
end
143145
catch e
144-
Base.display_error(STDERR,"exception $e on rank $rank",backtrace())
146+
Base.display_error(stderr,"exception $e on rank $rank",backtrace())
145147
end
146148

147149
if workers_wait && rank != 0

src/window-io.jl

Lines changed: 8 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -38,7 +38,7 @@ mutable struct WindowIO <: IO
3838
waiter
3939

4040
function WindowIO(comm=MPI.COMM_WORLD, bufsize=1024^2)
41-
buffer = Array{UInt8,1}(bufsize)
41+
buffer = Array{UInt8,1}(undef, bufsize)
4242
header_win = MPI.Win()
4343
header = BufferHeader(0, MPI.Get_address(buffer), bufsize, bufsize)
4444
remote_header = BufferHeader(0, MPI.Get_address(buffer), bufsize, bufsize)
@@ -140,7 +140,7 @@ mutable struct WindowWriter <: IO
140140
nb_written::Int
141141

142142
function WindowWriter(w::WindowIO, target::Integer)
143-
return new(w, target, Vector{UInt8}(1024^2), ReentrantLock(), 0)
143+
return new(w, target, Vector{UInt8}(undef,1024^2), ReentrantLock(), 0)
144144
end
145145
end
146146

@@ -164,7 +164,7 @@ Base.isreadable(::WindowWriter) = false
164164
function Base.close(w::WindowIO)
165165
w.is_open = false
166166
notify(w.read_requested)
167-
wait(w.waiter) # Wait for the data notification loop to finish
167+
fetch(w.waiter) # Wait for the data notification loop to finish
168168
MPI.Win_lock(MPI.LOCK_EXCLUSIVE, w.myrank, 0, w.header_win)
169169
w.header.count = 0
170170
w.ptr = 0
@@ -211,18 +211,18 @@ function Base.readbytes!(w::WindowIO, b::AbstractVector{UInt8}, nb=length(b); al
211211
if nb_read == 0
212212
return 0
213213
end
214-
copy!(b, 1, w.buffer, w.ptr+1, nb_read)
214+
copyto!(b, 1, w.buffer, w.ptr+1, nb_read)
215215
w.ptr += nb_read
216216
complete_read(w)
217217
return nb_read
218218
end
219219

220-
Base.readavailable(w::WindowIO) = read!(w, Vector{UInt8}(nb_available(w)))
220+
Base.readavailable(w::WindowIO) = read!(w, Vector{UInt8}(undef,nb_available(w)))
221221

222222
@inline function Base.unsafe_read(w::WindowIO, p::Ptr{UInt8}, nb::UInt)
223223
nb_obtained = wait_nb_available(w,nb)
224224
nb_read = min(nb_obtained, nb)
225-
unsafe_copy!(p, pointer(w.buffer, w.ptr+1), nb_read)
225+
unsafe_copyto!(p, pointer(w.buffer, w.ptr+1), nb_read)
226226
w.ptr += nb_read
227227
complete_read(w)
228228
if nb_read != nb
@@ -232,7 +232,7 @@ Base.readavailable(w::WindowIO) = read!(w, Vector{UInt8}(nb_available(w)))
232232
end
233233

234234
function Base.read(w::WindowIO, nb::Integer; all::Bool=true)
235-
buf = Vector{UInt8}(nb)
235+
buf = Vector{UInt8}(undef,nb)
236236
readbytes!(w, buf, nb, all=all)
237237
return buf
238238
end
@@ -253,7 +253,7 @@ function Base.unsafe_write(w::WindowWriter, p::Ptr{UInt8}, nb::UInt)
253253
offset = w.nb_written+1
254254
w.nb_written += nb
255255
ensureroom(w)
256-
copy!(w.write_buffer, offset, unsafe_wrap(Array{UInt8}, p, nb), 1, nb)
256+
copyto!(w.write_buffer, offset, unsafe_wrap(Array{UInt8}, p, nb), 1, nb)
257257
return nb
258258
end
259259

test/test_windowcman.jl

Lines changed: 6 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,15 +1,17 @@
1+
using Distributed
2+
13
const using_mpi = (nprocs() == 1)
24

35
if using_mpi
46
using MPI
5-
using Base.Test
7+
using Test
68
transport_mode = MPI.MPI_WINDOW_IO # This test can run with MPPI_TRANSPORT_ALL and MPI_WINDOW_IO
79
mgr = MPI.start_main_loop(transport_mode)
810
@everywhere const comm = MPI.COMM_WORLD
911
rank = MPI.Comm_rank(comm)
1012
size = MPI.Comm_size(comm)
1113
else
12-
@everywhere using Base.Test
14+
@everywhere using Test
1315
end
1416

1517
@everywhere const N = nprocs()
@@ -23,7 +25,7 @@ if N > 1
2325
@test workers() == collect(2:N)
2426
end
2527

26-
results = Vector{Any}(N)
28+
results = Vector{Any}(undef,N)
2729
for i in 1:N
2830
results[i] = remotecall(myid, i)
2931
end
@@ -52,7 +54,7 @@ has_arrays = true
5254
try
5355
@everywhere using DistributedArrays
5456
catch e
55-
has_arrays = false
57+
global has_arrays = false
5658
end
5759

5860
if has_arrays

test/test_windowcman_nowait.jl

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
1+
using Distributed
12
using MPI
2-
using Base.Test
3+
using Test
34

45
mgr = MPI.start_main_loop(MPI_WINDOW_NOWAIT)
56

test/test_windowio.jl

Lines changed: 9 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
using Base.Test
1+
using Test
22
using MPI
33

44
MPI.Init()
@@ -33,7 +33,7 @@ if rank == 1
3333
flush(writer) # Must flush to trigger communication
3434
elseif rank == 0
3535
@test MPI.wait_nb_available(winio) == 5
36-
arr = Vector{UInt8}(5)
36+
arr = Vector{UInt8}(undef,5)
3737
readbytes!(winio, arr)
3838
@test String(arr) == "hello"
3939
@test nb_available(winio) == 0
@@ -46,7 +46,7 @@ if rank == 1
4646
flush(writer) # Must flush to trigger communication
4747
elseif rank == 0
4848
@test MPI.wait_nb_available(winio) == 5
49-
arr = Vector{UInt8}(5)
49+
arr = Vector{UInt8}(undef,5)
5050
unsafe_read(winio, pointer(arr), 2)
5151
@test nb_available(winio) == 3
5252
unsafe_read(winio, pointer(arr,3), 3)
@@ -60,10 +60,10 @@ if rank == 1
6060
write(writer, "hello")
6161
flush(writer) # Must flush to trigger communication
6262
elseif rank == 0
63-
arr = Vector{UInt8}(3)
63+
arr = Vector{UInt8}(undef,3)
6464
readbytes!(winio, arr) # waits for data
6565
@test nb_available(winio) == 2
66-
@test String(arr) == "hel"
66+
@test String(copy(arr)) == "hel"
6767
fill!(arr, UInt8('a'))
6868
readbytes!(winio, arr, all=false) # reads what's available
6969
@test String(arr) == "loa"
@@ -121,7 +121,7 @@ if rank == 1
121121
@time flush(writer)
122122
MPI.Barrier(comm)
123123
elseif rank == 0
124-
const recarr = Vector{UInt8}(100000)
124+
const recarr = Vector{UInt8}(undef,100000)
125125
MPI.Barrier(comm)
126126

127127
println("read timings:")
@@ -141,12 +141,12 @@ if rank != 0
141141
println(writer, message*string(rank))
142142
flush(writer) # Must flush to trigger communication
143143
else
144-
nb_received = 0
144+
global nb_received = 0
145145
while nb_received != N-1
146146
line = readline(winio)
147147
@test startswith(line, message)
148148
@test line[length(message)+1:end] string.(1:N)
149-
nb_received += 1
149+
global nb_received += 1
150150
end
151151
end
152152

@@ -162,7 +162,7 @@ if rank != 0
162162
write(writer2, rank*ones(BS))
163163
flush(writer2) # Must flush to trigger communication
164164
else
165-
result = Vector{Float64}(BS*(N-1))
165+
result = Vector{Float64}(undef,BS*(N-1))
166166
read!(winio2, result)
167167
header = winio2.header
168168
@test nb_available(winio2) == 0

0 commit comments

Comments
 (0)