x

Signed-off-by: Stephanie Wang <[email protected]>

python/ray/_private/worker.py

@@ -2264,9 +2264,9 @@ def get(
  if is_individual_id:
  object_refs = [object_refs]
 
- if not (isinstance(object_refs, list) or isinstance(object_refs)):
+ if not isinstance(object_refs, list):
  raise ValueError(
- "'object_refs' must either be an ObjectRef or " "a list of ObjectRefs."
+ "'object_refs' must either be an ObjectRef or a list of ObjectRefs."
  )
 
  # TODO(ujvl): Consider how to allow user to retrieve the ready objects.

python/ray/_raylet.pyx

@@ -860,15 +860,10 @@ cdef execute_task(
  worker, errors,
  returns)
  if dynamic_returns != NULL:
- # We generated dynamic objects during the first execution and
- # we are now re-executing the task during object
- # reconstruction. Store the error for the dynamically generated
- # objects too.
+ # Store errors for any dynamically generated objects too.
  dynamic_errors = []
  for _ in range(dynamic_returns[0].size()):
  dynamic_errors.append(failure_object)
- # We pass is_dynamic=False because we have a fixed number of
- # return objects to populate.
  core_worker.store_task_outputs(
  worker, dynamic_errors,
  dynamic_returns)
@@ -2184,6 +2179,10 @@ cdef class CoreWorker:
  else:
  # This is the first execution of the task, so we don't know how
  # many return objects it should have yet.
+ # NOTE(swang): returns could also be empty if the task returned
+ # an empty generator and was re-executed. However, this should
+ # not happen because we never reconstruct empty
+ # ObjectRefGenerators (since these are not stored in plasma).
  num_returns = -1
  else:
  # The task specified how many return values it should have.

python/ray/tests/test_generators.py

@@ -316,6 +316,61 @@ def fetch(x):
  # ray.get(ref)
 
 
+def test_dynamic_empty_generator_reconstruction_nondeterministic(ray_start_cluster):
+ config = {
+ "num_heartbeats_timeout": 10,
+ "raylet_heartbeat_period_milliseconds": 100,
+ "max_direct_call_object_size": 100,
+ "task_retry_delay_ms": 100,
+ "object_timeout_milliseconds": 200,
+ "fetch_warn_timeout_milliseconds": 1000,
+ }
+ cluster = ray_start_cluster
+ # Head node with no resources.
+ cluster.add_node(
+ num_cpus=0,
+ _system_config=config,
+ enable_object_reconstruction=True,
+ resources={"head": 1},
+ )
+ ray.init(address=cluster.address)
+ # Node to place the initial object.
+ node_to_kill = cluster.add_node(num_cpus=1, object_store_memory=10 ** 8)
+ cluster.wait_for_nodes()
+
+ @ray.remote(num_cpus=0, resources={"head": 1})
+ class ExecutionCounter:
+ def __init__(self):
+ self.count = 0
+
+ def inc(self):
+ self.count += 1
+ return self.count
+
+ def get_count(self):
+ return self.count
+
+ @ray.remote(num_returns="dynamic")
+ def maybe_empty_generator(exec_counter):
+ if ray.get(exec_counter.inc.remote()) > 1:
+ for i in range(3):
+ yield np.ones(1_000_000, dtype=np.int8) * i
+
+ @ray.remote
+ def check(empty_generator):
+ return len(empty_generator) == 0
+
+ exec_counter = ExecutionCounter.remote()
+ gen = maybe_empty_generator.remote(exec_counter)
+ assert ray.get(check.remote(gen))
+ cluster.remove_node(node_to_kill, allow_graceful=False)
+ node_to_kill = cluster.add_node(num_cpus=1, object_store_memory=10 ** 8)
+ assert ray.get(check.remote(gen))
+
+ # We should never reconstruct an empty generator.
+ assert exec_counter.get_count.remote() == 1
+
+
 if __name__ == "__main__":
  import os