Make function argument the first in remotecallX and remote_do to ease…

… the use

of do block syntax.

NEWS.md

@@ -1,3 +1,27 @@
+Julia v0.5.0 Release Notes
+==========================
+
+New language features
+---------------------
+
+Language changes
+----------------
+
+Command line option changes
+---------------------------
+
+Compiler/Runtime improvements
+-----------------------------
+
+Library improvements
+--------------------
+
+Deprecated or removed
+---------------------
+
+ * The function `remotecall`, `remotecall_fetch`, and `remotecall_wait` now have the
+ the function argument as the first argument to allow for do-block syntax. [#13338]
+
 Julia v0.4.0 Release Notes
 ==========================
 

base/client.jl

@@ -243,7 +243,7 @@ function process_options(opts::JLOptions, args::Vector{UTF8String})
  # load file immediately on all processors
  if opts.load != C_NULL
  @sync for p in procs()
- @async remotecall_fetch(p, include, bytestring(opts.load))
+ @async remotecall_fetch(include, p, bytestring(opts.load))
  end
  end
  # eval expression

base/deprecated.jl

@@ -834,3 +834,11 @@ end
 # 12839
 const AsyncStream = IO
 deprecate(:AsyncStream)
+
+for f in (:remotecall, :remotecall_fetch, :remotecall_wait)
+ @eval begin
+ @deprecate ($f)(w::LocalProcess, f::Function, args...) ($f)(f, w::LocalProcess, args...)
+ @deprecate ($f)(w::Worker, f::Function, args...) ($f)(f, w::Worker, args...)
+ @deprecate ($f)(id::Integer, f::Function, args...) ($f)(f, id::Integer, args...)
+ end
+end

base/docs/helpdb.jl

@@ -3609,7 +3609,7 @@ Returns the index of the current worker into the `pids` vector, i.e., the list o
 indexpids
 
 doc"""
- remotecall_wait(id, func, args...)
+ remotecall_wait(func, id, args...)
 
 Perform `wait(remotecall(...))` in one message.
 """
@@ -6259,7 +6259,7 @@ Search for the first occurrence of the given characters within the given string.
 search
 
 doc"""
- remotecall_fetch(id, func, args...)
+ remotecall_fetch(func, id, args...)
 
 Perform `fetch(remotecall(...))` in one message. Any remote exceptions are captured in a `RemoteException` and thrown.
 """
@@ -7370,7 +7370,7 @@ Determine whether a `RemoteRef` has a value stored to it. Note that this functio
 If the argument `RemoteRef` is owned by a different node, this call will block to wait for the answer. It is recommended to wait for `r` in a separate task instead, or to use a local `RemoteRef` as a proxy:
 
  rr = RemoteRef()
- @async put!(rr, remotecall_fetch(p, long_computation))
+ @async put!(rr, remotecall_fetch(long_computation, p))
  isready(rr) # will not block
 """
 isready
@@ -9490,7 +9490,7 @@ Read all available data on the stream, blocking the task only if no data is avai
 readavailable
 
 doc"""
- remotecall(id, func, args...)
+ remotecall(func, id, args...)
 
 Call a function asynchronously on the given arguments on the specified process. Returns a `RemoteRef`.
 """

base/loading.jl

@@ -30,7 +30,7 @@ function find_in_node_path(name, srcpath, node::Int=1)
  if myid() == node
  find_in_path(name, srcpath)
  else
- remotecall_fetch(node, find_in_path, name, srcpath)
+ remotecall_fetch(find_in_path, node, name, srcpath)
  end
 end
 
@@ -67,7 +67,7 @@ function _require_from_serialized(node::Int, mod::Symbol, path_to_try::ByteStrin
  if node == myid()
  content = open(readbytes, path_to_try)
  else
- content = remotecall_fetch(node, open, readbytes, path_to_try)
+ content = remotecall_fetch(open, node, readbytes, path_to_try)
  end
  restored = _include_from_serialized(content)
  if restored !== nothing
@@ -83,7 +83,7 @@ function _require_from_serialized(node::Int, mod::Symbol, path_to_try::ByteStrin
  myid() == 1 && recompile_stale(mod, path_to_try)
  restored = ccall(:jl_restore_incremental, Any, (Ptr{UInt8},), path_to_try)
  else
- content = remotecall_fetch(node, open, readbytes, path_to_try)
+ content = remotecall_fetch(open, node, readbytes, path_to_try)
  restored = _include_from_serialized(content)
  end
  # otherwise, continue search
@@ -304,7 +304,7 @@ function include_from_node1(_path::AbstractString)
  result = Core.include(path)
  nprocs()>1 && sleep(0.005)
  else
- result = include_string(remotecall_fetch(1, readall, path), path)
+ result = include_string(remotecall_fetch(readall, 1, path), path)
  end
  finally
  if prev === nothing

base/managers.jl

@@ -357,7 +357,7 @@ function connect_to_worker(host::AbstractString, bind_addr::AbstractString, port
 end
 
 function kill(manager::ClusterManager, pid::Int, config::WorkerConfig)
- remote_do(pid, exit) # For TCP based transports this will result in a close of the socket
+ remote_do(exit, pid) # For TCP based transports this will result in a close of the socket
  # at our end, which will result in a cleanup of the worker.
  nothing
 end

base/multi.jl

@@ -185,14 +185,14 @@ function flush_gc_msgs(w::Worker)
  msgs = copy(w.add_msgs)
  if !isempty(msgs)
  empty!(w.add_msgs)
- remote_do(w, add_clients, msgs)
+ remote_do(add_clients, w, msgs)
  end
 
  msgs = copy(w.del_msgs)
  if !isempty(msgs)
  empty!(w.del_msgs)
  #print("sending delete of $msgs\n")
- remote_do(w, del_clients, msgs)
+ remote_do(del_clients, w, msgs)
  end
 end
 
@@ -293,7 +293,7 @@ get_bind_addr(w::LocalProcess) = LPROC.bind_addr
 function get_bind_addr(w::Worker)
  if isnull(w.config.bind_addr)
  if w.id != myid()
- w.config.bind_addr = remotecall_fetch(w.id, get_bind_addr, w.id)
+ w.config.bind_addr = remotecall_fetch(get_bind_addr, w.id, w.id)
  end
  end
  get(w.config.bind_addr)
@@ -317,7 +317,7 @@ function procs(pid::Integer)
  Int[x.id for x in filter(w -> get_bind_addr(w) == ipatpid, PGRP.workers)]
  end
  else
- remotecall_fetch(1, procs, pid)
+ remotecall_fetch(procs, 1, pid)
  end
 end
 
@@ -425,7 +425,7 @@ function deregister_worker(pg, pid)
  if PGRP.topology != :all_to_all
  for rpid in workers()
  try
- remote_do(rpid, deregister_worker, pid)
+ remote_do(deregister_worker, rpid, pid)
  catch
  end
  end
@@ -494,10 +494,10 @@ function RemoteRef(pid::Integer=myid())
 end
 
 function RemoteRef(f::Function, pid::Integer=myid())
- remotecall_fetch(pid, (f, rrid) -> begin
+ remotecall_fetch(pid, f, next_rrid_tuple()) do f, rrid
  rv=lookup_ref(rrid, f)
  RemoteRef{typeof(rv.c)}(myid(), rrid[1], rrid[2])
- end, f, next_rrid_tuple())
+ end
 end
 
 hash(r::RemoteRef, h::UInt) = hash(r.whence, hash(r.id, h))
@@ -522,7 +522,7 @@ function isready(rr::RemoteRef, args...)
  if rr.where == myid()
  isready(lookup_ref(rid).c, args...)
  else
- remotecall_fetch(rr.where, id->isready(lookup_ref(rid).c, args...), rid)
+ remotecall_fetch(id->isready(lookup_ref(rid).c, args...), rr.where, rid)
  end
 end
 
@@ -698,28 +698,28 @@ function local_remotecall_thunk(f, args)
  # f(map(localize_ref,args)...)
 end
 
-function remotecall(w::LocalProcess, f, args...)
+function remotecall(f, w::LocalProcess, args...)
  rr = RemoteRef(w)
  schedule_call(rr2id(rr), local_remotecall_thunk(f,args))
  rr
 end
 
-function remotecall(w::Worker, f, args...)
+function remotecall(f, w::Worker, args...)
  rr = RemoteRef(w)
  #println("$(myid()) asking for $rr")
  send_msg(w, CallMsg{:call}(f, args, rr2id(rr)))
  rr
 end
 
-remotecall(id::Integer, f, args...) = remotecall(worker_from_id(id), f, args...)
+remotecall(f, id::Integer, args...) = remotecall(f, worker_from_id(id), args...)
 
 # faster version of fetch(remotecall(...))
-function remotecall_fetch(w::LocalProcess, f, args...)
+function remotecall_fetch(f, w::LocalProcess, args...)
  v=run_work_thunk(local_remotecall_thunk(f,args), false)
  isa(v, RemoteException) ? throw(v) : v
 end
 
-function remotecall_fetch(w::Worker, f, args...)
+function remotecall_fetch(f, w::Worker, args...)
  # can be weak, because the program will have no way to refer to the Ref
  # itself, it only gets the result.
  oid = next_rrid_tuple()
@@ -731,13 +731,13 @@ function remotecall_fetch(w::Worker, f, args...)
  isa(v, RemoteException) ? throw(v) : v
 end
 
-remotecall_fetch(id::Integer, f, args...) =
- remotecall_fetch(worker_from_id(id), f, args...)
+remotecall_fetch(f, id::Integer, args...) =
+ remotecall_fetch(f, worker_from_id(id), args...)
 
 # faster version of wait(remotecall(...))
-remotecall_wait(w::LocalProcess, f, args...) = wait(remotecall(w,f,args...))
+remotecall_wait(f, w::LocalProcess, args...) = wait(remotecall(f, w, args...))
 
-function remotecall_wait(w::Worker, f, args...)
+function remotecall_wait(f, w::Worker, args...)
  prid = next_rrid_tuple()
  rv = lookup_ref(prid)
  rv.waitingfor = w.id
@@ -748,10 +748,10 @@ function remotecall_wait(w::Worker, f, args...)
  rr
 end
 
-remotecall_wait(id::Integer, f, args...) =
- remotecall_wait(worker_from_id(id), f, args...)
+remotecall_wait(f, id::Integer, args...) =
+ remotecall_wait(f, worker_from_id(id), args...)
 
-function remote_do(w::LocalProcess, f, args...)
+function remote_do(f, w::LocalProcess, args...)
  # the LocalProcess version just performs in local memory what a worker
  # does when it gets a :do message.
  # same for other messages on LocalProcess.
@@ -760,20 +760,20 @@ function remote_do(w::LocalProcess, f, args...)
  nothing
 end
 
-function remote_do(w::Worker, f, args...)
+function remote_do(f, w::Worker, args...)
  send_msg(w, RemoteDoMsg(f, args))
  nothing
 end
 
-remote_do(id::Integer, f, args...) = remote_do(worker_from_id(id), f, args...)
+remote_do(f, id::Integer, args...) = remote_do(f, worker_from_id(id), args...)
 
 # have the owner of rr call f on it
 function call_on_owner(f, rr::RemoteRef, args...)
  rid = rr2id(rr)
  if rr.where == myid()
  f(rid, args...)
  else
- remotecall_fetch(rr.where, f, rid, args...)
+ remotecall_fetch(f, rr.where, rid, args...)
  end
 end
 
@@ -818,7 +818,7 @@ function deliver_result(sock::IO, msg, oid, value)
  elseif wid == 1
  exit(1)
  else
- remote_do(1, rmprocs, wid)
+ remote_do(rmprocs, 1, wid)
  end
  end
 end
@@ -1125,7 +1125,7 @@ function addprocs(manager::ClusterManager; kwargs...)
  # function returns to the caller.
  all_w = workers()
  for pid in all_w
- remote_do(pid, set_valid_processes, all_w)
+ remote_do(set_valid_processes, pid, all_w)
  end
 
  sort!(launched_q)
@@ -1169,7 +1169,7 @@ function launch_n_additional_processes(manager, frompid, fromconfig, cnt, launch
  exeflags = get(fromconfig.exeflags, ``)
  cmd = `$exename $exeflags`
 
- new_addresses = remotecall_fetch(frompid, launch_additional, cnt, cmd)
+ new_addresses = remotecall_fetch(launch_additional, frompid, cnt, cmd)
  for address in new_addresses
  (bind_addr, port) = address
 
@@ -1183,7 +1183,7 @@ function launch_n_additional_processes(manager, frompid, fromconfig, cnt, launch
  let wconfig=wconfig
  @async begin
  pid = create_worker(manager, wconfig)
- remote_do(frompid, redirect_output_from_additional_worker, pid, port)
+ remote_do(redirect_output_from_additional_worker, frompid, pid, port)
  push!(launched_q, pid)
  end
  end
@@ -1317,7 +1317,7 @@ let nextidx = 0
  end
 end
 
-spawnat(p, thunk) = sync_add(remotecall(p, thunk))
+spawnat(p, thunk) = sync_add(remotecall(thunk, p))
 
 spawn_somewhere(thunk) = spawnat(chooseproc(thunk),thunk)
 
@@ -1335,21 +1335,21 @@ macro fetch(expr)
  expr = localize_vars(:(()->($expr)), false)
  quote
  thunk = $(esc(expr))
- remotecall_fetch(chooseproc(thunk), thunk)
+ remotecall_fetch(thunk, chooseproc(thunk))
  end
 end
 
 macro fetchfrom(p, expr)
  expr = localize_vars(:(()->($expr)), false)
- :(remotecall_fetch($(esc(p)), $(esc(expr))))
+ :(remotecall_fetch($(esc(expr)), $(esc(p))))
 end
 
 macro everywhere(ex)
  quote
  sync_begin()
  thunk = ()->(eval(Main,$(Expr(:quote,ex))); nothing)
  for pid in workers()
- async_run_thunk(()->remotecall_fetch(pid, thunk))
+ async_run_thunk(()->remotecall_fetch(thunk, pid))
  yield() # ensure that the remotecall_fetch has been started
  end
 
@@ -1365,7 +1365,7 @@ end
 function pmap_static(f, lsts...)
  np = nprocs()
  n = length(lsts[1])
- Any[ remotecall(PGRP.workers[(i-1)%np+1].id, f, map(L->L[i], lsts)...) for i = 1:n ]
+ Any[ remotecall(f, PGRP.workers[(i-1)%np+1].id, map(L->L[i], lsts)...) for i = 1:n ]
 end
 
 pmap(f) = f()
@@ -1427,7 +1427,7 @@ function pmap(f, lsts...; err_retry=true, err_stop=false, pids = workers())
  (idx, fvals) = tasklet
  busy_workers[pididx] = true
  try
- results[idx] = remotecall_fetch(wpid, f, fvals...)
+ results[idx] = remotecall_fetch(f, wpid, fvals...)
  catch ex
  if err_retry
  push!(retryqueue, (idx,fvals, ex))
@@ -1482,7 +1482,7 @@ function preduce(reducer, f, N::Int)
 
  w_exec = Task[]
  for (idx,pid) in enumerate(all_w)
- t = Task(()->remotecall_fetch(pid, f, first(chunks[idx]), last(chunks[idx])))
+ t = Task(()->remotecall_fetch(f, pid, first(chunks[idx]), last(chunks[idx])))
  schedule(t)
  push!(w_exec, t)
  end
@@ -1625,7 +1625,7 @@ end
 
 function check_same_host(pids)
  if myid() != 1
- return remotecall_fetch(1, check_same_host, pids)
+ return remotecall_fetch(check_same_host, 1, pids)
  else
  # We checkfirst if all test pids have been started using the local manager,
  # else we check for the same bind_to addr. This handles the special case
@@ -1663,5 +1663,5 @@ function getindex(r::RemoteRef, args...)
  if r.where == myid()
  return getindex(fetch(r), args...)
  end
- return remotecall_fetch(r.where, getindex, r, args...)
+ return remotecall_fetch(getindex, r.where, r, args...)
 end