# This workload stresses distributed reference counting by passing and
# returning serialized ObjectIDs.

import time

import numpy as np

import ray
from ray.cluster_utils import Cluster

num_redis_shards = 5
redis_max_memory = 10**8
object_store_memory = 10**8
num_nodes = 10

message = ("Make sure there is enough memory on this machine to run this "
           "workload. We divide the system memory by 2 to provide a buffer.")
assert (num_nodes * object_store_memory + num_redis_shards * redis_max_memory <
        ray.utils.get_system_memory() / 2)

# Simulate a cluster on one machine.

cluster = Cluster()
for i in range(num_nodes):
    cluster.add_node(
        redis_port=6379 if i == 0 else None,
        num_redis_shards=num_redis_shards if i == 0 else None,
        num_cpus=2,
        num_gpus=0,
        resources={str(i): 2},
        object_store_memory=object_store_memory,
        redis_max_memory=redis_max_memory,
        webui_host="0.0.0.0")
ray.init(address=cluster.address)

# Run the workload.


@ray.remote(max_retries=0)
def child(*xs):
    return np.zeros(1024, dtype=np.uint8)


@ray.remote(max_retries=0)
def f(xs):
    return child.remote(*xs)


iteration = 0
ids = []
start_time = time.time()
previous_time = start_time
while True:
    for _ in range(50):
        new_constrained_ids = [
            f._remote(args=[[ids]], resources={str(i % num_nodes): 1})
            for i in range(25)
        ]
        new_unconstrained_ids = [f.remote([ids]) for _ in range(25)]
        ids = new_constrained_ids + new_unconstrained_ids

    ray.get(ids)

    new_time = time.time()
    print("Iteration {}:\n"
          "  - Iteration time: {}.\n"
          "  - Absolute time: {}.\n"
          "  - Total elapsed time: {}.".format(
              iteration, new_time - previous_time, new_time,
              new_time - start_time))
    previous_time = new_time
    iteration += 1

@@ -110,16 +110,42 @@
 ort time
+%0Aimport json%0Aimport random
 %0A%0Aimport
@@ -613,16 +613,84 @@
 chine.%0A%0A
+config = json.dumps(%7B%0A    %22distributed_ref_counting_enabled%22: 1,%0A%7D)%0A
 cluster 
@@ -1049,16 +1049,55 @@
 0.0.0.0%22
+,%0A        _internal_config=config,%0A    
 )%0Aray.in
@@ -1184,29 +1184,136 @@
 f ch
-ild(*xs):%0A    return 
+urn():%0A    return ray.put(np.zeros(1024 * 1024, dtype=np.uint8))%0A%0A%0A@ray.remote(max_retries=0)%0Adef child(*xs):%0A    oid = ray.put(
 np.z
@@ -1317,24 +1317,31 @@
 p.zeros(1024
+ * 1024
 , dtype=np.u
@@ -1345,16 +1345,32 @@
 p.uint8)
+)%0A    return oid
 %0A%0A%0A@ray.
@@ -1397,21 +1397,80 @@
 )%0Adef f(
+*
 xs):%0A
+    if xs:%0A        return random.choice(xs)%0A    else:%0A    
     retu
@@ -1666,15 +1666,11 @@
 rgs=
-%5B%5B
 ids
-%5D%5D
 , re
@@ -1789,13 +1789,12 @@
 ote(
-%5B
+*
 ids
-%5D
 ) fo
@@ -1878,19 +1878,396 @@
     
-ray.get(ids
+# Fill the object store while the tasks are running.%0A    for i in range(num_nodes):%0A        for _ in range(10):%0A            %5B%0A                churn._remote(args=%5B%5D, resources=%7Bstr(i %25 num_nodes): 1%7D)%0A                for _ in range(10)%0A            %5D%0A%0A    # Make sure that the objects are still available.%0A    child_ids = ray.get(ids)%0A    for child_id in child_ids:%0A        ray.get(child_id
 )%0A%0A 

Make the ref counting test more stressful (#7473)