Serialize zeroed-out remote refs to non-ClusterSerializer objects (Ju…

…liaLang#22836)

base/distributed/remotecall.jl

@@ -20,6 +20,8 @@ mutable struct Future <: AbstractRemoteRef
 
  Future(w::Int, rrid::RRID) = Future(w, rrid, Nullable{Any}())
  Future(w::Int, rrid::RRID, v) = (r = new(w,rrid.whence,rrid.id,v); return test_existing_ref(r))
+
+ Future(t::Tuple) = new(t[1],t[2],t[3],t[4]) # Useful for creating dummy, zeroed-out instances
 end
 
 mutable struct RemoteChannel{T<:AbstractChannel} <: AbstractRemoteRef
@@ -31,6 +33,10 @@ mutable struct RemoteChannel{T<:AbstractChannel} <: AbstractRemoteRef
  r = new(w, rrid.whence, rrid.id)
  return test_existing_ref(r)
  end
+
+ function RemoteChannel{T}(t::Tuple) where T<:AbstractChannel
+ return new(t[1],t[2],t[3])
+ end
 end
 
 function test_existing_ref(r::AbstractRemoteRef)
@@ -273,29 +279,29 @@ end
 
 channel_type(rr::RemoteChannel{T}) where {T} = T
 
-serialize(s::AbstractSerializer, f::Future) = serialize(s, f, isnull(f.v))
-serialize(s::AbstractSerializer, rr::RemoteChannel) = serialize(s, rr, true)
-function serialize(s::AbstractSerializer, rr::AbstractRemoteRef, addclient)
+serialize(s::ClusterSerializer, f::Future) = serialize(s, f, isnull(f.v))
+serialize(s::ClusterSerializer, rr::RemoteChannel) = serialize(s, rr, true)
+function serialize(s::ClusterSerializer, rr::AbstractRemoteRef, addclient)
  if addclient
  p = worker_id_from_socket(s.io)
  (p !== rr.where) && send_add_client(rr, p)
  end
- invoke(serialize, Tuple{AbstractSerializer, Any}, s, rr)
+ invoke(serialize, Tuple{ClusterSerializer, Any}, s, rr)
 end
 
-function deserialize(s::AbstractSerializer, t::Type{<:Future})
+function deserialize(s::ClusterSerializer, t::Type{<:Future})
  f = deserialize_rr(s,t)
  Future(f.where, RRID(f.whence, f.id), f.v) # ctor adds to client_refs table
 end
 
-function deserialize(s::AbstractSerializer, t::Type{<:RemoteChannel})
+function deserialize(s::ClusterSerializer, t::Type{<:RemoteChannel})
  rr = deserialize_rr(s,t)
  # call ctor to make sure this rr gets added to the client_refs table
  RemoteChannel{channel_type(rr)}(rr.where, RRID(rr.whence, rr.id))
 end
 
 function deserialize_rr(s, t)
- rr = invoke(deserialize, Tuple{AbstractSerializer, DataType}, s, t)
+ rr = invoke(deserialize, Tuple{ClusterSerializer, DataType}, s, t)
  if rr.where == myid()
  # send_add_client() is not executed when the ref is being
  # serialized to where it exists
@@ -304,6 +310,18 @@ function deserialize_rr(s, t)
  rr
 end
 
+# Future and RemoteChannel are serializable only in a running cluster.
+# Serialize zeroed-out values to non ClusterSerializer objects
+function serialize(s::AbstractSerializer, ::Future)
+ zero_fut = Future((0,0,0,Nullable{Any}()))
+ invoke(serialize, Tuple{AbstractSerializer, Any}, s, zero_fut)
+end
+
+function serialize(s::AbstractSerializer, ::RemoteChannel)
+ zero_rc = RemoteChannel{Channel{Any}}((0,0,0))
+ invoke(serialize, Tuple{AbstractSerializer, Any}, s, zero_rc)
+end
+
 
 # make a thunk to call f on args in a way that simulates what would happen if
 # the function were sent elsewhere

doc/src/manual/parallel-computing.md

@@ -781,14 +781,16 @@ julia> @elapsed while n > 0 # print out results
 
 ## Remote References and Distributed Garbage Collection
 
-Objects referred to by remote references can be freed only when *all* held references in the cluster
-are deleted.
+Objects referred to by remote references can be freed only when *all* held references
+in the cluster are deleted.
 
 The node where the value is stored keeps track of which of the workers have a reference to it.
 Every time a [`RemoteChannel`](@ref) or a (unfetched) [`Future`](@ref) is serialized to a worker,
 the node pointed to by the reference is notified. And every time a [`RemoteChannel`](@ref) or
 a (unfetched) [`Future`](@ref) is garbage collected locally, the node owning the value is again
-notified.
+notified. This is implemented in an internal cluster aware serializer. Remote references are only
+valid in the context of a running cluster. Serializing and deserializing references to and from
+regular `IO` objects is not supported.
 
 The notifications are done via sending of "tracking" messages--an "add reference" message when
 a reference is serialized to a different process and a "delete reference" message when a reference

test/distributed_exec.jl

@@ -268,6 +268,27 @@ test_indexing(Future(id_other))
 test_indexing(RemoteChannel())
 test_indexing(RemoteChannel(id_other))
 
+# Test ser/deser to non-ClusterSerializer objects.
+function test_regular_io_ser(ref::Base.Distributed.AbstractRemoteRef)
+ io = IOBuffer()
+ serialize(io, ref)
+ seekstart(io)
+ ref2 = deserialize(io)
+ for fld in fieldnames(typeof(ref))
+ v = getfield(ref2, fld)
+ if isa(v, Number)
+ @test v === zero(typeof(v))
+ elseif isa(v, Nullable)
+ @test v === Nullable{Any}()
+ else
+ error(string("Add test for field ", fld))
+ end
+ end
+end
+
+test_regular_io_ser(Future())
+test_regular_io_ser(RemoteChannel())
+
 dims = (20,20,20)
 
 if Sys.islinux()