Fix leak of Futures when unfetched Futures are repeatedly serialized (J…

…uliaLang#26143)

to the node with the reference store.

stdlib/Distributed/src/remotecall.jl

@@ -292,24 +292,29 @@ function serialize(s::ClusterSerializer, rr::AbstractRemoteRef, addclient)
 end
 
 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
+ f = invoke(deserialize, Tuple{ClusterSerializer, DataType}, s, t)
+ f2 = Future(f.where, RRID(f.whence, f.id), f.v) # ctor adds to client_refs table
+
+ # 1) send_add_client() is not executed when the ref is being serialized
+ # to where it exists, hence do it here.
+ # 2) If we have recieved a 'fetch'ed Future or if the Future ctor found an
+ # already 'fetch'ed instance in client_refs (Issue #25847), we should not
+ # track it in the backing RemoteValue store.
+ if f2.where == myid() && f2.v === nothing
+ add_client(remoteref_id(f2), myid())
+ end
+ f2
 end
 
 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{ClusterSerializer, DataType}, s, t)
  if rr.where == myid()
  # send_add_client() is not executed when the ref is being
  # serialized to where it exists
  add_client(remoteref_id(rr), myid())
  end
- rr
+ # 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
 
 # Future and RemoteChannel are serializable only in a running cluster.

stdlib/Distributed/test/distributed_exec.jl

@@ -180,6 +180,25 @@ testval = remotecall_fetch(wid2, fstore) do x
 end
 @test testval == 1
 
+# Issue number #25847
+@everywhere function f25847(ref)
+ fetch(ref)
+ return true
+end
+
+f = remotecall_wait(identity, id_other, ones(10))
+rrid = Distributed.RRID(f.whence, f.id)
+remotecall_fetch(f25847, id_other, f)
+@test BitSet([id_me]) == remotecall_fetch(()->Distributed.PGRP.refs[rrid].clientset, id_other)
+
+remotecall_fetch(f25847, id_other, f)
+@test BitSet([id_me]) == remotecall_fetch(()->Distributed.PGRP.refs[rrid].clientset, id_other)
+
+finalize(f)
+yield() # flush gc msgs
+@test false == remotecall_fetch(chk_rrid->haskey(Distributed.PGRP.refs, chk_rrid), id_other, rrid)
+
+
 # Distributed GC tests for RemoteChannels
 function test_remoteref_dgc(id)
  rr = RemoteChannel(id)