Clean up release tests (#11420)

release/1.0.0.env

@@ -0,0 +1,2 @@
+export ray_version="1.0.0rc1"
+export commit=fd5ddb661e659c2b0c968661d96d0405426912e5

release/README.md

@@ -0,0 +1,2 @@
+Running the kickoff script:
+

release/RELEASE_PROCESS.rst

@@ -0,0 +1 @@
+../doc/dev/RELEASE_PROCESS.rst

release/kickoff.sh

@@ -0,0 +1,29 @@
+#!/usr/bin/env bash
+
+
+source "$2"
+
+ray_version=${ray_version:-}
+commit=${commit:-}
+
+if [[ $ray_version == "" || $commit == "" || $1 == "" ]]
+then
+    echo "Provide --ray-version, --commit, and --ray-branch"
+    exit 1
+fi
+
+echo "version: $ray_version"
+echo "commit: $commit"
+echo "workload: $1"
+
+DATESTR=$(date +%Y%m%d-%H%M)
+SESSION="$1-$DATESTR"
+
+echo "session: $SESSION"
+
+chmod +x ./run.sh
+if [ -z "$NO_UP" ]; then
+  anyscale up "$SESSION"
+fi
+anyscale push "$SESSION"
+anyscale exec -n "$SESSION" "./run.sh $1 --ray-version=$ray_version --commit=$commit"

release/long_running_distributed_tests/README.rst

@@ -0,0 +1,45 @@
+Long Running Distributed Tests
+==============================
+
+This directory contains the long-running multi-node workloads which are intended to run
+forever until they fail. To set up the project you need to run
+
+.. code-block:: bash
+
+    $ pip install anyscale
+    $ anyscale init
+
+Running the Workloads
+---------------------
+Easiest approach is to use the `Anyscale UI <https://www.anyscale.dev/>`_. First run ``anyscale snapshot create`` from the command line to create a project snapshot. Then from the UI, you can launch an individual session and execute the test_workload command for each test. 
+
+You can also start the workloads using the CLI with:
+
+.. code-block:: bash
+
+    $ anyscale start --ray-wheel=<RAY_WHEEL_LINK>
+    $ anyscale run test_workload --workload=<WORKLOAD_NAME>
+
+
+Doing this for each workload will start one EC2 instance per workload and will start the workloads
+running (one per instance). A list of
+available workload options is available in the `ray_projects/project.yaml` file.
+
+
+Debugging
+---------
+The primary method to debug the test while it is running is to view the logs and the dashboard from the UI. After the test has failed, you can still view the stdout logs in the UI and also inspect
+the logs under ``/tmp/ray/session*/logs/`` and
+``/tmp/ray/session*/debug_state.txt``.
+
+Shut Down the Workloads
+-----------------------
+
+The instances running the workloads can all be killed by running
+``anyscale stop <SESSION_NAME>``.
+
+Adding a Workload
+-----------------
+
+To create a new workload, simply add a new Python file under ``workloads/`` and
+add the workload in the run command in `ray-project/project.yaml`.

release/long_running_distributed_tests/cluster.yaml

@@ -0,0 +1,73 @@
+# This file is generated by `ray project create`.
+
+# A unique identifier for the head node and workers of this cluster.
+cluster_name: long-running-distributed-tests
+
+# The minimum number of workers nodes to launch in addition to the head
+# node. This number should be >= 0.
+min_workers: 3
+# The maximum number of workers nodes to launch in addition to the head
+# node. This takes precedence over min_workers. min_workers defaults to 0.
+max_workers: 3
+
+# The autoscaler will scale up the cluster to this target fraction of resource
+# usage. For example, if a cluster of 10 nodes is 100% busy and
+# target_utilization is 0.8, it would resize the cluster to 13. This fraction
+# can be decreased to increase the aggressiveness of upscaling.
+# This value must be less than 1.0 for scaling to happen.
+target_utilization_fraction: 0.8
+
+# If a node is idle for this many minutes, it will be removed.
+idle_timeout_minutes: 5
+
+# Cloud-provider specific configuration.
+provider:
+    type: aws
+    region: us-west-2
+    availability_zone: us-west-2a
+    cache_stopped_nodes: False
+
+# How Ray will authenticate with newly launched nodes.
+auth:
+    ssh_user: ubuntu
+
+# By default Ray creates a new private keypair, but you can also use your own.
+# If you do so, make sure to also set "KeyName" in the head and worker node
+# configurations below.
+#    ssh_private_key: /path/to/your/key.pem
+
+# Provider-specific config for the head node, e.g. instance type. By default
+# Ray will auto-configure unspecified fields such as SubnetId and KeyName.
+# For more documentation on available fields, see:
+# http://boto3.readthedocs.io/en/latest/reference/services/ec2.html#EC2.ServiceResource.create_instances
+head_node:
+    InstanceType: g3.8xlarge
+    ImageId: ami-0888a3b5189309429  # DLAMI 7/1/19
+    BlockDeviceMappings:
+        - DeviceName: /dev/sda1
+          Ebs:
+              VolumeSize: 150
+
+worker_nodes:
+  InstanceType: g3.8xlarge
+  ImageId: ami-0888a3b5189309429  # DLAMI 7/1/19
+  BlockDeviceMappings:
+        - DeviceName: /dev/sda1
+          Ebs:
+              VolumeSize: 150
+  InstanceMarketOptions:
+    MarketType: spot
+
+setup_commands: []
+
+# Command to start ray on the head node. You don't need to change this.
+head_start_ray_commands:
+    - ray stop
+    - export RAY_BACKEND_LOG_LEVEL=debug
+    - ray start --head --port=6379 --object-manager-port=8076 --autoscaling-config=~/ray_bootstrap_config.yaml
+
+# Command to start ray on worker nodes. You don't need to change this.
+worker_start_ray_commands:
+    - ray stop
+    - export RAY_BACKEND_LOG_LEVEL=debug
+    - ray start --address=$RAY_HEAD_IP:6379 --object-manager-port=8076

release/long_running_distributed_tests/run.sh

@@ -0,0 +1,61 @@
+#!/usr/bin/env bash
+
+ray_version="" 
+commit=""
+ray_branch=""
+workload=""
+
+usage() {
+    echo "Start one microbenchmark trial."
+}
+
+for i in "$@"
+do
+echo "$i"
+case "$i" in
+    --ray-version=*)
+    ray_version="${i#*=}"
+
+    ;;
+    --commit=*)
+    commit="${i#*=}"
+    ;;
+    --ray-branch=*)
+    ray_branch="${i#*=}"
+    ;;
+    --workload=*)
+    workload="${i#*=}"
+    ;;
+    --help)
+    usage
+    exit
+    ;;
+    *)
+    echo "unknown arg, $i"
+    exit 1
+    ;;
+esac
+done
+
+echo "version: $ray_version"
+echo "commit: $commit"
+echo "branch: $ray_branch"
+echo "workload: $workload"
+
+wheel="https://s3-us-west-2.amazonaws.com/ray-wheels/$ray_branch/$commit/ray-$ray_version-cp36-cp36m-manylinux1_x86_64.whl"
+
+conda uninstall -y terminado || true
+pip install -U pip
+pip install terminado
+pip install -U "$wheel"
+pip install "ray[rllib]"
+pip install -U ipdb
+# There have been some recent problems with torch 1.5 and torchvision 0.6
+# not recognizing GPUs.
+# So, we force install torch 1.4 and torchvision 0.5. 
+# https://github.com/pytorch/pytorch/issues/37212#issuecomment-623198624.
+pip install torch==1.4.0 torchvision==0.5.0
+echo set-window-option -g mouse on > ~/.tmux.conf
+echo 'termcapinfo xterm* ti@:te@' > ~/.screenrc
+
+python "workloads/$workload.py"

release/long_running_distributed_tests/workloads/pytorch_pbt_failure.py

@@ -0,0 +1,155 @@
+import argparse
+import numpy as np
+import os
+import random
+import torch
+import torch.nn as nn
+from torch.utils.data import DataLoader, Subset
+from torchvision.datasets import CIFAR10
+import torchvision.transforms as transforms
+
+import ray
+from ray import tune
+from ray.autoscaler._private.commands import kill_node
+from ray.tune import CLIReporter
+from ray.tune.ray_trial_executor import RayTrialExecutor
+from ray.tune.schedulers import PopulationBasedTraining
+from ray.tune.utils.util import merge_dicts
+from ray.util.sgd.torch import TorchTrainer
+from ray.util.sgd.torch.resnet import ResNet18
+from ray.util.sgd.utils import BATCH_SIZE
+
+parser = argparse.ArgumentParser()
+parser.add_argument(
+    "--smoke-test",
+    action="store_true",
+    default=False,
+    help="Finish quickly for training.")
+args = parser.parse_args()
+
+
+class FailureInjectorExecutor(RayTrialExecutor):
+    """Adds random failure injection to the TrialExecutor."""
+
+    def on_step_begin(self, trial_runner):
+        """Before step(), update available resources and inject failure."""
+        self._update_avail_resources()
+        # With 10% probability inject failure to a worker.
+        if random.random() < 0.1 and not args.smoke_test:
+            # With 10% probability fully terminate the node.
+            should_terminate = random.random() < 0.1
+            kill_node(
+                "/home/ubuntu/ray_bootstrap_config.yaml",
+                yes=True,
+                hard=should_terminate,
+                override_cluster_name=None)
+
+
+def initialization_hook():
+    # Need this for avoiding a connection restart issue on AWS.
+    os.environ["NCCL_SOCKET_IFNAME"] = "^docker0,lo"
+    os.environ["NCCL_LL_THRESHOLD"] = "0"
+
+    # set the below if needed
+    # print("NCCL DEBUG SET")
+    # os.environ["NCCL_DEBUG"] = "INFO"
+
+
+def cifar_creator(config):
+    transform_train = transforms.Compose([
+        transforms.RandomCrop(32, padding=4),
+        transforms.RandomHorizontalFlip(),
+        transforms.ToTensor(),
+        transforms.Normalize((0.4914, 0.4822, 0.4465),
+                             (0.2023, 0.1994, 0.2010)),
+    ])  # meanstd transformation
+
+    transform_test = transforms.Compose([
+        transforms.ToTensor(),
+        transforms.Normalize((0.4914, 0.4822, 0.4465),
+                             (0.2023, 0.1994, 0.2010)),
+    ])
+    train_dataset = CIFAR10(
+        root="~/data", train=True, download=True, transform=transform_train)
+    validation_dataset = CIFAR10(
+        root="~/data", train=False, download=False, transform=transform_test)
+
+    if config.get("test_mode"):
+        train_dataset = Subset(train_dataset, list(range(64)))
+        validation_dataset = Subset(validation_dataset, list(range(64)))
+
+    train_loader = DataLoader(
+        train_dataset, batch_size=config[BATCH_SIZE], num_workers=2)
+    validation_loader = DataLoader(
+        validation_dataset, batch_size=config[BATCH_SIZE], num_workers=2)
+    return train_loader, validation_loader
+
+
+def optimizer_creator(model, config):
+    """Returns optimizer"""
+    return torch.optim.SGD(
+        model.parameters(),
+        lr=config.get("lr", 0.1),
+        momentum=config.get("momentum", 0.9))
+
+
+ray.init(address="auto" if not args.smoke_test else None, _log_to_driver=True)
+num_training_workers = 1 if args.smoke_test else 3
+
+executor = FailureInjectorExecutor(queue_trials=True)
+
+TorchTrainable = TorchTrainer.as_trainable(
+    model_creator=ResNet18,
+    data_creator=cifar_creator,
+    optimizer_creator=optimizer_creator,
+    loss_creator=nn.CrossEntropyLoss,
+    initialization_hook=initialization_hook,
+    num_workers=num_training_workers,
+    config={
+        "test_mode": args.smoke_test,
+        BATCH_SIZE: 128 * num_training_workers,
+    },
+    use_gpu=not args.smoke_test)
+
+
+class NoFaultToleranceTrainable(TorchTrainable):
+    def _train(self):
+        train_stats = self.trainer.train(max_retries=0, profile=True)
+        validation_stats = self.trainer.validate(profile=True)
+        stats = merge_dicts(train_stats, validation_stats)
+        return stats
+
+
+pbt_scheduler = PopulationBasedTraining(
+    time_attr="training_iteration",
+    metric="val_loss",
+    mode="min",
+    perturbation_interval=1,
+    hyperparam_mutations={
+        # distribution for resampling
+        "lr": lambda: np.random.uniform(0.001, 1),
+        # allow perturbations within this set of categorical values
+        "momentum": [0.8, 0.9, 0.99],
+    })
+
+reporter = CLIReporter()
+reporter.add_metric_column("val_loss", "loss")
+reporter.add_metric_column("val_accuracy", "acc")
+
+analysis = tune.run(
+    NoFaultToleranceTrainable,
+    num_samples=4,
+    config={
+        "lr": tune.choice([0.001, 0.01, 0.1]),
+        "momentum": 0.8,
+        "head_location": None,
+        "worker_locations": None
+    },
+    max_failures=-1,  # used for fault tolerance
+    checkpoint_freq=2,  # used for fault tolerance
+    progress_reporter=reporter,
+    scheduler=pbt_scheduler,
+    trial_executor=executor,
+    stop={"training_iteration": 1} if args.smoke_test else None)
+
+print(analysis.get_best_config(metric="val_loss", mode="min"))

release/long_running_tests/.gitignore

@@ -0,0 +1 @@
+config_temporary.yaml

release/long_running_tests/README.rst

@@ -0,0 +1,53 @@
+Long Running Tests
+==================
+
+This directory contains the long-running workloads which are intended to run
+forever until they fail. To set up the project you need to run
+
+.. code-block:: bash
+
+    $ pip install anyscale
+    $ anyscale init
+
+Note that all the long running test is running inside virtual environment, tensorflow_p36
+
+Running the Workloads
+---------------------
+Easiest approach is to use the `Anyscale UI <https://www.anyscale.dev/>`. First run ``anyscale snapshot create`` from the command line to create a project snapshot. Then from the UI, you can launch an individual session and execute the run command for each test. 
+
+You can also start the workloads using the CLI with:
+
+.. code-block:: bash
+
+    $ anyscale start
+    $ anyscale run test_workload --workload=<WORKLOAD_NAME> --wheel=<RAY_WHEEL_LINK>
+
+
+Doing this for each workload will start one EC2 instance per workload and will start the workloads
+running (one per instance). A list of
+available workload options is available in the `ray_projects/project.yaml` file.
+
+
+Debugging
+---------
+The primary method to debug the test while it is running is to view the logs and the dashboard from the UI. After the test has failed, you can still view the stdout logs in the UI and also inspect
+the logs under ``/tmp/ray/session*/logs/`` and
+``/tmp/ray/session*/debug_state.txt``.
+
+.. To check up on the workloads, run either
+.. ``anyscale session --name="*" execute check-load``, which
+.. will print the load on each machine, or
+.. ``anyscale session --name="*" execute show-output``, which
+.. will print the tail of the output for each workload.
+
+Shut Down the Workloads
+-----------------------
+
+The instances running the workloads can all be killed by running
+``anyscale stop <SESSION_NAME>``.
+
+Adding a Workload
+-----------------
+
+To create a new workload, simply add a new Python file under ``workloads/`` and
+add the workload in the run command in `ray-project/project.yaml`.

release/long_running_tests/cluster.yaml

@@ -0,0 +1,48 @@
+cluster_name: default
+min_workers: 0
+max_workers: 0
+target_utilization_fraction: 0.8
+idle_timeout_minutes: 5
+
+# Cloud-provider specific configuration.
+provider:
+    type: aws
+    region: us-west-2
+    availability_zone: us-west-2a
+auth:
+    ssh_user: ubuntu
+
+head_node:
+    InstanceType: m5.2xlarge
+    ImageId: ami-0888a3b5189309429  # DLAMI 7/1/19
+    BlockDeviceMappings:
+        - DeviceName: /dev/sda1
+          Ebs:
+              VolumeSize: 150
+
+worker_nodes:
+    InstanceType: m5.large
+    ImageId: ami-0888a3b5189309429  # DLAMI 7/1/19
+    BlockDeviceMappings:
+        - DeviceName: /dev/sda1
+          Ebs:
+              VolumeSize: 150
+
+    # Run workers on spot by default. Comment this out to use on-demand.
+    InstanceMarketOptions:
+        MarketType: spot
+
+# List of shell commands to run to set up nodes.
+setup_commands: []
+
+# Custom commands that will be run on the head node after common setup.
+head_setup_commands: []
+
+# Custom commands that will be run on worker nodes after common setup.
+worker_setup_commands: []
+
+# Command to start ray on the head node. You don't need to change this.
+head_start_ray_commands: []
+
+# Command to start ray on worker nodes. You don't need to change this.
+worker_start_ray_commands: []

release/long_running_tests/run.sh

@@ -0,0 +1,66 @@
+#!/usr/bin/env bash
+
+ray_version="" 
+commit=""
+ray_branch=""
+workload=""
+
+usage() {
+    echo "Start one microbenchmark trial."
+}
+
+for i in "$@"
+do
+echo "$i"
+case "$i" in
+    --ray-version=*)
+    ray_version="${i#*=}"
+
+    ;;
+    --commit=*)
+    commit="${i#*=}"
+    ;;
+    --ray-branch=*)
+    ray_branch="${i#*=}"
+    ;;
+    --workload=*)
+    workload="${i#*=}"
+    ;;
+    --help)
+    usage
+    exit
+    ;;
+    *)
+    echo "unknown arg, $i"
+    exit 1
+    ;;
+esac
+done
+
+if [[ $ray_version == "" || $commit == "" || $ray_branch == "" ]]
+then
+    echo "Provide --ray-version, --commit, and --ray-branch"
+    exit 1
+fi
+
+echo "version: $ray_version"
+echo "commit: $commit"
+echo "branch: $ray_branch"
+echo "workload: $workload"
+
+wheel="https://s3-us-west-2.amazonaws.com/ray-wheels/$ray_branch/$commit/ray-$ray_version-cp36-cp36m-manylinux1_x86_64.whl"
+
+echo set-window-option -g mouse on > ~/.tmux.conf
+echo 'termcapinfo xterm* ti@:te@' > ~/.screenrc
+# Serve load testing tool
+rm -r wrk || true && git clone https://github.com/wg/wrk.git wrk && cd wrk && make -j && sudo cp wrk /usr/local/bin
+pip install -U pip
+unset RAY_ADDRESS
+source activate tensorflow_p36 
+conda remove -y --force wrapt || true
+pip install --upgrade pip
+pip install -U tensorflow==1.14
+pip install -q -U "$wheel" Click
+pip install -q "ray[all]" "gym[atari]"
+python "workloads/$workload.py"
+

release/long_running_tests/workloads/actor_deaths.py

@@ -0,0 +1,104 @@
+# This workload tests repeatedly killing actors and submitting tasks to them.
+
+import numpy as np
+import sys
+import time
+
+import ray
+from ray.cluster_utils import Cluster
+
+num_redis_shards = 1
+redis_max_memory = 10**8
+object_store_memory = 10**8
+num_nodes = 2
+
+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), message
+
+# 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=8,
+        num_gpus=0,
+        resources={str(i): 2},
+        object_store_memory=object_store_memory,
+        redis_max_memory=redis_max_memory,
+        dashboard_host="0.0.0.0")
+ray.init(address=cluster.address)
+
+# Run the workload.
+
+num_parents = 5
+num_children = 5
+death_probability = 0.95
+
+
+@ray.remote
+class Child(object):
+    def __init__(self, death_probability):
+        self.death_probability = death_probability
+
+    def ping(self):
+        # Exit process with some probability.
+        exit_chance = np.random.rand()
+        if exit_chance > self.death_probability:
+            sys.exit(-1)
+
+
+@ray.remote
+class Parent(object):
+    def __init__(self, num_children, death_probability):
+        self.death_probability = death_probability
+        self.children = [
+            Child.remote(death_probability) for _ in range(num_children)
+        ]
+
+    def ping(self, num_pings):
+        children_outputs = []
+        for _ in range(num_pings):
+            children_outputs += [
+                child.ping.remote() for child in self.children
+            ]
+        try:
+            ray.get(children_outputs)
+        except Exception:
+            # Replace the children if one of them died.
+            self.__init__(len(self.children), self.death_probability)
+
+    def kill(self):
+        # Clean up children.
+        ray.get([child.__ray_terminate__.remote() for child in self.children])
+
+
+parents = [
+    Parent.remote(num_children, death_probability) for _ in range(num_parents)
+]
+
+iteration = 0
+start_time = time.time()
+previous_time = start_time
+while True:
+    ray.get([parent.ping.remote(10) for parent in parents])
+
+    # Kill a parent actor with some probability.
+    exit_chance = np.random.rand()
+    if exit_chance > death_probability:
+        parent_index = np.random.randint(len(parents))
+        parents[parent_index].kill.remote()
+        parents[parent_index] = Parent.remote(num_children, death_probability)
+
+    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

release/long_running_tests/workloads/apex.py

@@ -0,0 +1,49 @@
+# This workload tests running APEX
+
+import ray
+from ray.cluster_utils import Cluster
+from ray.tune import run_experiments
+
+num_redis_shards = 5
+redis_max_memory = 10**8
+object_store_memory = 10**9
+num_nodes = 3
+
+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), message
+
+# 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=20,
+        num_gpus=0,
+        resources={str(i): 2},
+        object_store_memory=object_store_memory,
+        redis_max_memory=redis_max_memory,
+        dashboard_host="0.0.0.0")
+ray.init(address=cluster.address)
+
+# Run the workload.
+
+run_experiments({
+    "apex": {
+        "run": "APEX",
+        "env": "Pong-v0",
+        "config": {
+            "num_workers": 8,
+            "num_gpus": 0,
+            "buffer_size": 10000,
+            "learning_starts": 0,
+            "rollout_fragment_length": 1,
+            "train_batch_size": 1,
+            "min_iter_time_s": 10,
+            "timesteps_per_iteration": 10,
+        },
+    }
+})

release/long_running_tests/workloads/impala.py

@@ -0,0 +1,48 @@
+# This workload tests running IMPALA with remote envs
+
+import ray
+from ray.tune import run_experiments
+from ray.cluster_utils import Cluster
+
+num_redis_shards = 5
+redis_max_memory = 10**8
+object_store_memory = 10**8
+num_nodes = 1
+
+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), message
+
+# 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=10,
+        num_gpus=0,
+        resources={str(i): 2},
+        object_store_memory=object_store_memory,
+        redis_max_memory=redis_max_memory,
+        dashboard_host="0.0.0.0")
+ray.init(address=cluster.address)
+
+# Run the workload.
+
+run_experiments({
+    "impala": {
+        "run": "IMPALA",
+        "env": "CartPole-v0",
+        "config": {
+            "num_workers": 8,
+            "num_gpus": 0,
+            "num_envs_per_worker": 5,
+            "remote_worker_envs": True,
+            "remote_env_batch_wait_ms": 99999999,
+            "rollout_fragment_length": 50,
+            "train_batch_size": 100,
+        },
+    },
+})

release/long_running_tests/workloads/many_actor_tasks.py

@@ -0,0 +1,70 @@
+# This workload tests submitting many actor methods.
+
+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), message
+
+# 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=5,
+        num_gpus=0,
+        resources={str(i): 2},
+        object_store_memory=object_store_memory,
+        redis_max_memory=redis_max_memory,
+        dashboard_host="0.0.0.0")
+ray.init(address=cluster.address)
+
+# Run the workload.
+
+
+@ray.remote
+class Actor(object):
+    def __init__(self):
+        self.value = 0
+
+    def method(self):
+        self.value += 1
+        return np.zeros(1024, dtype=np.uint8)
+
+
+actors = [
+    Actor._remote([], {}, num_cpus=0.1, resources={str(i % num_nodes): 0.1})
+    for i in range(num_nodes * 5)
+]
+
+iteration = 0
+start_time = time.time()
+previous_time = start_time
+while True:
+    for _ in range(100):
+        previous_ids = [a.method.remote() for a in actors]
+
+    ray.get(previous_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

release/long_running_tests/workloads/many_drivers.py

@@ -0,0 +1,102 @@
+# This workload tests many drivers using the same cluster.
+
+import time
+
+import ray
+from ray.cluster_utils import Cluster
+from ray.test_utils import run_string_as_driver
+
+num_redis_shards = 5
+redis_max_memory = 10**8
+object_store_memory = 10**8
+num_nodes = 4
+
+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), message
+
+# 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=4,
+        num_gpus=0,
+        resources={str(i): 5},
+        object_store_memory=object_store_memory,
+        redis_max_memory=redis_max_memory,
+        dashboard_host="0.0.0.0")
+ray.init(address=cluster.address)
+
+# Run the workload.
+
+# Define a driver script that runs a few tasks and actors on each node in the
+# cluster.
+driver_script = """
+import ray
+
+ray.init(address="{}")
+
+num_nodes = {}
+
+
+@ray.remote
+def f():
+    return 1
+
+
+@ray.remote
+class Actor(object):
+    def method(self):
+        return 1
+
+
+for _ in range(5):
+    for i in range(num_nodes):
+        assert (ray.get(
+            f._remote(args=[], kwargs={{}}, resources={{str(i): 1}})) == 1)
+        actor = Actor._remote(args=[], kwargs={{}}, resources={{str(i): 1}})
+        assert ray.get(actor.method.remote()) == 1
+
+print("success")
+""".format(cluster.address, num_nodes)
+
+
+@ray.remote
+def run_driver():
+    output = run_string_as_driver(driver_script)
+    assert "success" in output
+
+
+iteration = 0
+running_ids = [
+    run_driver._remote(
+        args=[], kwargs={}, num_cpus=0, resources={str(i): 0.01})
+    for i in range(num_nodes)
+]
+start_time = time.time()
+previous_time = start_time
+while True:
+    # Wait for a driver to finish and start a new driver.
+    [ready_id], running_ids = ray.wait(running_ids, num_returns=1)
+    ray.get(ready_id)
+
+    running_ids.append(
+        run_driver._remote(
+            args=[],
+            kwargs={},
+            num_cpus=0,
+            resources={str(iteration % num_nodes): 0.01}))
+
+    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

release/long_running_tests/workloads/many_tasks.py

@@ -0,0 +1,66 @@
+# This workload tests submitting and getting many tasks over and over.
+
+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), message
+
+# 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,
+        dashboard_host="0.0.0.0")
+ray.init(address=cluster.address)
+
+# Run the workload.
+
+
+@ray.remote
+def f(*xs):
+    return np.zeros(1024, dtype=np.uint8)
+
+
+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

release/long_running_tests/workloads/many_tasks_serialized_ids.py

@@ -0,0 +1,93 @@
+# This workload stresses distributed reference counting by passing and
+# returning serialized ObjectRefs.
+
+import time
+import random
+
+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), message
+
+# 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,
+        dashboard_host="0.0.0.0")
+ray.init(address=cluster.address)
+
+# Run the workload.
+
+
+@ray.remote(max_retries=0)
+def churn():
+    return ray.put(np.zeros(1024 * 1024, dtype=np.uint8))
+
+
+@ray.remote(max_retries=0)
+def child(*xs):
+    obj_ref = ray.put(np.zeros(1024 * 1024, dtype=np.uint8))
+    return obj_ref
+
+
+@ray.remote(max_retries=0)
+def f(*xs):
+    if xs:
+        return random.choice(xs)
+    else:
+        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
+
+    # Fill the object store while the tasks are running.
+    for i in range(num_nodes):
+        for _ in range(10):
+            [
+                churn._remote(args=[], resources={str(i % num_nodes): 1})
+                for _ in range(10)
+            ]
+
+    # Make sure that the objects are still available.
+    child_ids = ray.get(ids)
+    for child_id in child_ids:
+        ray.get(child_id)
+
+    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

release/long_running_tests/workloads/node_failures.py

@@ -0,0 +1,68 @@
+# This workload tests repeatedly killing a node and adding a new node.
+
+import time
+
+import ray
+from ray.cluster_utils import Cluster
+from ray.test_utils import get_other_nodes
+
+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), message
+
+# 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,
+        dashboard_host="0.0.0.0")
+ray.init(address=cluster.address)
+
+# Run the workload.
+
+
+@ray.remote
+def f(*xs):
+    return 1
+
+
+iteration = 0
+previous_ids = [1 for _ in range(100)]
+start_time = time.time()
+previous_time = start_time
+while True:
+    for _ in range(100):
+        previous_ids = [f.remote(previous_id) for previous_id in previous_ids]
+
+    ray.get(previous_ids)
+
+    for _ in range(100):
+        previous_ids = [f.remote(previous_id) for previous_id in previous_ids]
+    node_to_kill = get_other_nodes(cluster, exclude_head=True)[0]
+
+    # Remove the first non-head node.
+    cluster.remove_node(node_to_kill)
+    cluster.add_node()
+
+    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

release/long_running_tests/workloads/pbt.py

@@ -0,0 +1,56 @@
+# This workload tests running PBT
+
+import ray
+from ray.tune import run_experiments
+from ray.tune.schedulers import PopulationBasedTraining
+from ray.cluster_utils import Cluster
+
+num_redis_shards = 5
+redis_max_memory = 10**8
+object_store_memory = 10**8
+num_nodes = 3
+
+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), message
+
+# 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=10,
+        num_gpus=0,
+        resources={str(i): 2},
+        object_store_memory=object_store_memory,
+        redis_max_memory=redis_max_memory,
+        dashboard_host="0.0.0.0")
+ray.init(address=cluster.address)
+
+# Run the workload.
+
+pbt = PopulationBasedTraining(
+    time_attr="training_iteration",
+    metric="episode_reward_mean",
+    mode="max",
+    perturbation_interval=10,
+    hyperparam_mutations={
+        "lr": [0.1, 0.01, 0.001, 0.0001],
+    })
+
+run_experiments(
+    {
+        "pbt_test": {
+            "run": "PG",
+            "env": "CartPole-v0",
+            "num_samples": 8,
+            "config": {
+                "lr": 0.01,
+            },
+        }
+    },
+    scheduler=pbt,
+    verbose=False)

release/long_running_tests/workloads/serve.py

@@ -0,0 +1,66 @@
+import time
+import subprocess
+from subprocess import PIPE
+
+import requests
+
+import ray
+from ray import serve
+from ray.cluster_utils import Cluster
+
+num_redis_shards = 1
+redis_max_memory = 10**8
+object_store_memory = 10**8
+num_nodes = 5
+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=8,
+        num_gpus=0,
+        resources={str(i): 2},
+        object_store_memory=object_store_memory,
+        redis_max_memory=redis_max_memory,
+        dashboard_host="0.0.0.0")
+
+ray.init(address=cluster.address, dashboard_host="0.0.0.0")
+client = serve.start()
+
+
+@serve.accept_batch
+def echo(_):
+    time.sleep(0.01)  # Sleep for 10ms
+    ray.show_in_dashboard(
+        str(serve.context.batch_size), key="Current batch size")
+    return ["hi {}".format(i) for i in range(serve.context.batch_size)]
+
+
+config = {"num_replicas": 30, "max_batch_size": 16}
+client.create_backend("echo:v1", echo, config=config)
+client.create_endpoint("echo", backend="echo:v1", route="/echo")
+
+print("Warming up")
+for _ in range(5):
+    resp = requests.get("http://127.0.0.1:8000/echo").text
+    print(resp)
+    time.sleep(0.5)
+
+connections = int(config["num_replicas"] * config["max_batch_size"] * 0.75)
+num_threads = 2
+time_to_run = "60m"
+
+while True:
+    proc = subprocess.Popen(
+        [
+            "wrk", "-c",
+            str(connections), "-t",
+            str(num_threads), "-s", time_to_run, "http://127.0.0.1:8000/echo"
+        ],
+        stdout=PIPE,
+        stderr=PIPE)
+    print("started load testing")
+    proc.wait()
+    out, err = proc.communicate()
+    print(out.decode())
+    print(err.decode())

release/long_running_tests/workloads/serve_failure.py

@@ -0,0 +1,121 @@
+import random
+import string
+import time
+
+import requests
+
+import ray
+from ray import serve
+from ray.cluster_utils import Cluster
+
+num_redis_shards = 1
+redis_max_memory = 10**8
+object_store_memory = 10**8
+num_nodes = 5
+cpus_per_node = 2
+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,
+        dashboard_host="0.0.0.0")
+
+ray.init(
+    address=cluster.address, dashboard_host="0.0.0.0", log_to_driver=False)
+client = serve.start(detached=True)
+
+
+@ray.remote
+class RandomKiller:
+    def __init__(self, client, kill_period_s=1):
+        self.client = client
+        self.kill_period_s = kill_period_s
+
+    def _get_all_serve_actors(self):
+        controller = self.client._controller
+        routers = list(ray.get(controller.get_routers.remote()).values())
+        all_handles = routers + [controller]
+        worker_handle_dict = ray.get(
+            controller.get_all_worker_handles.remote())
+        for _, replica_dict in worker_handle_dict.items():
+            all_handles.extend(list(replica_dict.values()))
+
+        return all_handles
+
+    def run(self):
+        while True:
+            ray.kill(
+                random.choice(self._get_all_serve_actors()), no_restart=False)
+            time.sleep(self.kill_period_s)
+
+
+class RandomTest:
+    def __init__(self, client, max_endpoints=1):
+        self.client = client
+        self.max_endpoints = max_endpoints
+        self.weighted_actions = [
+            (self.create_endpoint, 1),
+            (self.verify_endpoint, 4),
+        ]
+        self.endpoints = []
+        for _ in range(max_endpoints):
+            self.create_endpoint()
+
+    def create_endpoint(self):
+        if len(self.endpoints) == self.max_endpoints:
+            endpoint_to_delete = self.endpoints.pop()
+            self.client.delete_endpoint(endpoint_to_delete)
+            self.client.delete_backend(endpoint_to_delete)
+
+        new_endpoint = "".join(
+            [random.choice(string.ascii_letters) for _ in range(10)])
+
+        def handler(self, *args):
+            return new_endpoint
+
+        self.client.create_backend(new_endpoint, handler)
+        self.client.create_endpoint(
+            new_endpoint, backend=new_endpoint, route="/" + new_endpoint)
+
+        self.endpoints.append(new_endpoint)
+
+    def verify_endpoint(self):
+        endpoint = random.choice(self.endpoints)
+        for _ in range(100):
+            try:
+                r = requests.get("http://127.0.0.1:8000/" + endpoint)
+                assert r.text == endpoint
+            except Exception:
+                print("Request to {} failed.".format(endpoint))
+                time.sleep(0.01)
+
+    def run(self):
+        iteration = 0
+        start_time = time.time()
+        previous_time = start_time
+        while True:
+            for _ in range(100):
+                actions, weights = zip(*self.weighted_actions)
+                random.choices(actions, weights=weights)[0]()
+
+            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
+
+
+random_killer = RandomKiller.remote(client)
+random_killer.run.remote()
+# Subtract 4 from the CPUs available for master, router, HTTP proxy,
+# and metric monitor actors.
+RandomTest(client, max_endpoints=(num_nodes * cpus_per_node) - 4).run()

release/microbenchmark/cluster.yaml

@@ -0,0 +1,57 @@
+cluster_name: ray-release-microbenchmark
+min_workers: 0
+max_workers: 0
+target_utilization_fraction: 0.8
+idle_timeout_minutes: 5
+
+# Cloud-provider specific configuration.
+provider:
+    type: aws
+    region: us-west-2
+    availability_zone: us-west-2a
+auth:
+    ssh_user: ubuntu
+
+head_node:
+    InstanceType: m4.16xlarge
+    ImageId: ami-06d51e91cea0dac8d  # Ubuntu 18.04
+    BlockDeviceMappings:
+        - DeviceName: /dev/sda1
+          Ebs:
+              VolumeSize: 150
+
+worker_nodes:
+    InstanceType: m5.large
+    ImageId: ami-06d51e91cea0dac8d  # Ubuntu 18.04
+    BlockDeviceMappings:
+        - DeviceName: /dev/sda1
+          Ebs:
+              VolumeSize: 150
+
+    # Run workers on spot by default. Comment this out to use on-demand.
+    InstanceMarketOptions:
+        MarketType: spot
+
+# List of shell commands to run to set up nodes.
+setup_commands:
+    # Install latest TensorFlow
+    - echo set-window-option -g mouse on > ~/.tmux.conf
+    - echo 'termcapinfo xterm* ti@:te@' > ~/.screenrc
+
+# Custom commands that will be run on the head node after common setup.
+head_setup_commands: 
+    # Install Anaconda.
+    - wget --quiet https://repo.continuum.io/archive/Anaconda3-5.0.1-Linux-x86_64.sh || true
+    - bash Anaconda3-5.0.1-Linux-x86_64.sh -b -p $HOME/anaconda3 || true
+    - echo 'export PATH="$HOME/anaconda3/bin:$PATH"' >> ~/.bashrc
+    - pip install -U pip 
+    - conda uninstall -y terminado
+
+# Custom commands that will be run on worker nodes after common setup.
+worker_setup_commands: []
+
+# Command to start ray on the head node. You don't need to change this.
+head_start_ray_commands: []
+
+# Command to start ray on worker nodes. You don't need to change this.
+worker_start_ray_commands: []

release/microbenchmark/run.sh

@@ -0,0 +1,60 @@
+#!/usr/bin/env bash
+
+ray_version=""
+commit=""
+ray_branch=""
+
+usage() {
+    echo "Start one microbenchmark trial."
+}
+
+for i in "$@"
+do
+case "$i" in
+    --ray-version=*)
+    ray_version="${i#*=}"
+
+    ;;
+    --commit=*)
+    commit="${i#*=}"
+    ;;
+    --ray-branch=*)
+    ray_branch="${i#*=}"
+    ;;
+    --workload=*)
+    workload="${i#*=}"
+    ;;
+    --help)
+    usage
+    exit
+    ;;
+    *)
+    echo "unknown arg, $2"
+    exit 1
+    ;;
+esac
+done
+
+if [ -z "$ray_version" ] || [ -z "$commit" ] || [ -z "$ray_branch" ]
+then
+    echo "Provide --ray-version, --commit, and --ray-branch"
+    exit 1
+fi
+
+
+echo "version: $ray_version"
+echo "commit: $commit"
+echo "branch: $ray_branch"
+echo "workload: $workload"
+
+wheel="https://s3-us-west-2.amazonaws.com/ray-wheels/$ray_branch/$commit/ray-$ray_version-cp38-cp38-manylinux1_x86_64.whl"
+
+
+echo set-window-option -g mouse on > ~/.tmux.conf
+echo 'termcapinfo xterm* ti@:te@' > ~/.screenrc
+pip uninstall -y -q ray
+pip install --upgrade pip
+pip install -U "$wheel"
+
+unset RAY_ADDRESS
+OMP_NUM_THREADS=64 ray microbenchmark

release/rllib_tests/regression_tests/cluster.yaml

@@ -0,0 +1,39 @@
+cluster_name: ray-rllib-regression-tests
+
+min_workers: 0
+max_workers: 0
+
+# Cloud-provider specific configuration.
+provider:
+    type: aws
+    region: us-west-2
+    availability_zone: us-west-2a
+    cache_stopped_nodes: False
+
+# How Ray will authenticate with newly launched nodes.
+auth:
+    ssh_user: ubuntu
+
+head_node:
+    InstanceType: p3.16xlarge
+    ImageId: ami-07728e9e2742b0662 # Deep Learning AMI (Ubuntu 16.04)
+
+    # Set primary volume to 25 GiB
+    BlockDeviceMappings:
+        - DeviceName: /dev/sda1
+          Ebs:
+              VolumeSize: 100
+
+
+# List of shell commands to run to set up nodes.
+setup_commands: []
+
+# Command to start ray on the head node. You don't need to change this.
+head_start_ray_commands:
+    - source activate tensorflow_p36 && ray stop
+    - ulimit -n 65536; source activate tensorflow_p36 && OMP_NUM_THREADS=1 ray start --head --port=6379 --object-manager-port=8076 --autoscaling-config=~/ray_bootstrap_config.yaml
+
+# Command to start ray on worker nodes. You don't need to change this.
+worker_start_ray_commands:
+    - source activate tensorflow_p36 && ray stop
+    - ulimit -n 65536; source activate tensorflow_p36 && OMP_NUM_THREADS=1 ray start --address=$RAY_HEAD_IP:6379 --object-manager-port=8076

release/rllib_tests/regression_tests/compact-regression-test.yaml

@@ -0,0 +1,145 @@
+# This file runs on a single g3.16xl or p3.16xl node. It is suggested
+# to run these in a DLAMI / tensorflow_p36 env. Note that RL runs are
+# inherently high variance, so you'll have to check to see if the
+# rewards reached seem reasonably in line with previous results.
+#
+# You can find the reference results here:
+# https://github.com/ray-project/ray/tree/master/doc/dev/release_logs
+atari-impala:
+    env: BreakoutNoFrameskip-v4
+    run: IMPALA
+    num_samples: 4
+    stop:
+        time_total_s: 3600
+    config:
+        rollout_fragment_length: 50
+        train_batch_size: 500
+        num_workers: 10
+        num_envs_per_worker: 5
+        clip_rewards: True
+        lr_schedule: [
+            [0, 0.0005],
+            [20000000, 0.000000000001],
+        ]
+        num_gpus: 1
+atari-ppo-tf:
+    env: BreakoutNoFrameskip-v4
+    run: PPO
+    num_samples: 4
+    stop:
+        time_total_s: 3600
+    config:
+        lambda: 0.95
+        kl_coeff: 0.5
+        clip_rewards: True
+        clip_param: 0.1
+        vf_clip_param: 10.0
+        entropy_coeff: 0.01
+        train_batch_size: 5000
+        rollout_fragment_length: 100
+        sgd_minibatch_size: 500
+        num_sgd_iter: 10
+        num_workers: 10
+        num_envs_per_worker: 5
+        batch_mode: truncate_episodes
+        observation_filter: NoFilter
+        vf_share_layers: true
+        num_gpus: 1
+atari-ppo-torch:
+    env: BreakoutNoFrameskip-v4
+    run: PPO
+    num_samples: 4
+    stop:
+        time_total_s: 3600
+    config:
+        framework: torch
+        lambda: 0.95
+        kl_coeff: 0.5
+        clip_rewards: True
+        clip_param: 0.1
+        vf_clip_param: 10.0
+        entropy_coeff: 0.01
+        train_batch_size: 5000
+        rollout_fragment_length: 100
+        sgd_minibatch_size: 500
+        num_sgd_iter: 10
+        num_workers: 10
+        num_envs_per_worker: 5
+        batch_mode: truncate_episodes
+        observation_filter: NoFilter
+        vf_share_layers: true
+        num_gpus: 1
+apex:
+    env: BreakoutNoFrameskip-v4
+    run: APEX
+    num_samples: 4
+    stop:
+        time_total_s: 3600
+    config:
+        double_q: false
+        dueling: false
+        num_atoms: 1
+        noisy: false
+        n_step: 3
+        lr: .0001
+        adam_epsilon: .00015
+        hiddens: [512]
+        buffer_size: 1000000
+        exploration_config:
+          epsilon_timesteps: 200000
+          final_epsilon: 0.01
+        prioritized_replay_alpha: 0.5
+        final_prioritized_replay_beta: 1.0
+        prioritized_replay_beta_annealing_timesteps: 2000000
+        num_gpus: 1
+        num_workers: 8
+        num_envs_per_worker: 8
+        rollout_fragment_length: 20
+        train_batch_size: 512
+        target_network_update_freq: 50000
+        timesteps_per_iteration: 25000
+atari-a2c:
+    env: BreakoutNoFrameskip-v4
+    run: A2C
+    num_samples: 4
+    stop:
+        time_total_s: 3600
+    config:
+        rollout_fragment_length: 20
+        clip_rewards: True
+        num_workers: 5
+        num_envs_per_worker: 5
+        num_gpus: 1
+        lr_schedule: [
+            [0, 0.0007],
+            [20000000, 0.000000000001],
+        ]
+atari-basic-dqn:
+    env: BreakoutNoFrameskip-v4
+    run: DQN
+    num_samples: 4
+    stop:
+        time_total_s: 3600
+    config:
+        double_q: false
+        dueling: false
+        num_atoms: 1
+        noisy: false
+        prioritized_replay: false
+        n_step: 1
+        target_network_update_freq: 8000
+        lr: .0000625
+        adam_epsilon: .00015
+        hiddens: [512]
+        learning_starts: 20000
+        buffer_size: 1000000
+        rollout_fragment_length: 4
+        train_batch_size: 32
+        exploration_config:
+          epsilon_timesteps: 200000
+          final_epsilon: 0.01
+        prioritized_replay_alpha: 0.5
+        final_prioritized_replay_beta: 1.0
+        prioritized_replay_beta_annealing_timesteps: 2000000
+        num_gpus: 0.2
+        timesteps_per_iteration: 10000

release/rllib_tests/regression_tests/requirements.txt

@@ -0,0 +1 @@
+ray[rllib]

release/rllib_tests/regression_tests/run.sh

@@ -0,0 +1,52 @@
+#!/usr/bin/env bash
+ray_version="" 
+commit=""
+ray_branch=""
+
+for i in "$@"
+do
+echo "$i"
+case "$i" in
+    --ray-version=*)
+    ray_version="${i#*=}"
+
+    ;;
+    --commit=*)
+    commit="${i#*=}"
+    ;;
+    --ray-branch=*)
+    ray_branch="${i#*=}"
+    ;;
+    --workload=*)
+    ;;
+    --help)
+    usage
+    exit
+    ;;
+    *)
+    echo "unknown arg, $i"
+    exit 1
+    ;;
+esac
+done
+
+if [[ $ray_version == "" || $commit == "" || $ray_branch == "" ]]
+then
+    echo "Provide --ray-version, --commit, and --ray-branch"
+    exit 1
+fi
+
+echo "version: $ray_version"
+echo "commit: $commit"
+echo "branch: $ray_branch"
+echo "workload: ignored"
+
+wheel="https://s3-us-west-2.amazonaws.com/ray-wheels/$ray_branch/$commit/ray-$ray_version-cp36-cp36m-manylinux1_x86_64.whl"
+
+conda uninstall -y terminado
+source activate tensorflow_p36 && pip install -U pip
+source activate tensorflow_p36 && pip install -U "$wheel"
+source activate tensorflow_p36 && pip install "ray[rllib]" "ray[debug]"
+source activate tensorflow_p36 && pip install torch==1.6 torchvision
+source activate tensorflow_p36 && pip install boto3==1.4.8 cython==0.29.0
+source activate tensorflow_p36 && rllib train -f compact-regression-test.yaml

release/rllib_tests/stress_tests/atari_impala_xlarge.yaml

@@ -0,0 +1,23 @@
+# Taken from rllib/tuned_examples/atari_impala_large.yaml
+# Runs on a g3.16xl node with 5 m5.24xl workers
+# Takes roughly 10 minutes. x10?
+atari-impala:
+    env:
+        grid_search:
+            - BreakoutNoFrameskip-v4
+            - BeamRiderNoFrameskip-v4
+            - QbertNoFrameskip-v4
+            - SpaceInvadersNoFrameskip-v4
+    run: IMPALA
+    stop:
+        timesteps_total: 30000000
+    config:
+        rollout_fragment_length: 50
+        train_batch_size: 500
+        num_workers: 128
+        num_envs_per_worker: 5
+        clip_rewards: True
+        lr_schedule: [
+            [0, 0.0005],
+            [20000000, 0.000000000001],
+        ]

release/rllib_tests/stress_tests/cluster.yaml

@@ -0,0 +1,105 @@
+####################################################################
+# All nodes in this cluster will auto-terminate in 1 hour
+####################################################################
+
+# An unique identifier for the head node and workers of this cluster.
+cluster_name: ray-rllib-stress-tests
+
+# The minimum number of workers nodes to launch in addition to the head
+# node. This number should be >= 0.
+min_workers: 9
+
+# The maximum number of workers nodes to launch in addition to the head
+# node. This takes precedence over min_workers.
+max_workers: 9
+
+# The autoscaler will scale up the cluster to this target fraction of resource
+# usage. For example, if a cluster of 10 nodes is 100% busy and
+# target_utilization is 0.8, it would resize the cluster to 13. This fraction
+# can be decreased to increase the aggressiveness of upscaling.
+# This value must be less than 1.0 for scaling to happen.
+target_utilization_fraction: 0.8
+
+# If a node is idle for this many minutes, it will be removed.
+idle_timeout_minutes: 5
+
+# Cloud-provider specific configuration.
+provider:
+    type: aws
+    region: us-west-2
+    availability_zone: us-west-2a
+    cache_stopped_nodes: False
+
+# How Ray will authenticate with newly launched nodes.
+auth:
+    ssh_user: ubuntu
+# By default Ray creates a new private keypair, but you can also use your own.
+# If you do so, make sure to also set "KeyName" in the head and worker node
+# configurations below.
+#    ssh_private_key: /path/to/your/key.pem
+
+# Provider-specific config for the head node, e.g. instance type. By default
+# Ray will auto-configure unspecified fields such as SubnetId and KeyName.
+# For more documentation on available fields, see:
+# http://boto3.readthedocs.io/en/latest/reference/services/ec2.html#EC2.ServiceResource.create_instances
+head_node:
+    InstanceType: p3.16xlarge
+    ImageId: ami-07728e9e2742b0662 # Deep Learning AMI (Ubuntu 16.04)
+
+    # Set primary volume to 25 GiB
+    BlockDeviceMappings:
+        - DeviceName: /dev/sda1
+          Ebs:
+              VolumeSize: 100
+
+    # Additional options in the boto docs.
+
+# Provider-specific config for worker nodes, e.g. instance type. By default
+# Ray will auto-configure unspecified fields such as SubnetId and KeyName.
+# For more documentation on available fields, see:
+# http://boto3.readthedocs.io/en/latest/reference/services/ec2.html#EC2.ServiceResource.create_instances
+worker_nodes:
+    InstanceType: m4.16xlarge
+    ImageId: ami-07728e9e2742b0662 # Deep Learning AMI (Ubuntu 16.04)
+
+
+    # Set primary volume to 25 GiB
+    BlockDeviceMappings:
+        - DeviceName: /dev/sda1
+          Ebs:
+              VolumeSize: 100
+
+    # Run workers on spot by default. Comment this out to use on-demand.
+    # InstanceMarketOptions:
+    #     MarketType: spot
+        # Additional options can be found in the boto docs, e.g.
+        #   SpotOptions:
+        #       MaxPrice: MAX_HOURLY_PRICE
+
+    # Additional options in the boto docs.
+
+# Files or directories to copy to the head and worker nodes. The format is a
+# dictionary from REMOTE_PATH: LOCAL_PATH, e.g.
+file_mounts: {
+#    "/path1/on/remote/machine": "/path1/on/local/machine",
+#    "/path2/on/remote/machine": "/path2/on/local/machine",
+}
+
+# List of shell commands to run to set up nodes.
+setup_commands: []
+
+# Custom commands that will be run on the head node after common setup.
+head_setup_commands: []
+
+# Custom commands that will be run on worker nodes after common setup.
+worker_setup_commands: []
+
+# Command to start ray on the head node. You don't need to change this.
+head_start_ray_commands:
+    - source activate tensorflow_p36 && ray stop
+    - ulimit -n 65536; source activate tensorflow_p36 && OMP_NUM_THREADS=1 ray start --head --port=6379 --object-manager-port=8076 --autoscaling-config=~/ray_bootstrap_config.yaml
+
+# Command to start ray on worker nodes. You don't need to change this.
+worker_start_ray_commands:
+    - source activate tensorflow_p36 && ray stop
+    - ulimit -n 65536; source activate tensorflow_p36 && OMP_NUM_THREADS=1 ray start --address=$RAY_HEAD_IP:6379 --object-manager-port=8076

release/rllib_tests/stress_tests/requirements.txt

@@ -0,0 +1 @@
+ray[rllib]

release/rllib_tests/stress_tests/run.sh

@@ -0,0 +1,56 @@
+#!/usr/bin/env bash
+
+ray_version="" 
+commit=""
+ray_branch=""
+
+for i in "$@"
+do
+echo "$i"
+case "$i" in
+    --ray-version=*)
+    ray_version="${i#*=}"
+
+    ;;
+    --commit=*)
+    commit="${i#*=}"
+    ;;
+    --ray-branch=*)
+    ray_branch="${i#*=}"
+    ;;
+    --workload=*)
+    ;;
+    --help)
+    usage
+    exit
+    ;;
+    *)
+    echo "unknown arg, $i"
+    exit 1
+    ;;
+esac
+done
+
+if [[ $ray_version == "" || $commit == "" || $ray_branch == "" ]]
+then
+    echo "Provide --ray-version, --commit, and --ray-branch"
+    exit 1
+fi
+
+echo "version: $ray_version"
+echo "commit: $commit"
+echo "branch: $ray_branch"
+echo "workload: ignored"
+
+wheel="https://s3-us-west-2.amazonaws.com/ray-wheels/$ray_branch/$commit/ray-$ray_version-cp36-cp36m-manylinux1_x86_64.whl"
+
+conda uninstall -y terminado
+source activate tensorflow_p36 && pip install -U pip
+source activate tensorflow_p36 && pip install -U "$wheel"
+source activate tensorflow_p36 && pip install "ray[rllib]" "ray[debug]"
+source activate tensorflow_p36 && pip install boto3==1.4.8 cython==0.29.0
+source activate tensorflow_p36
+
+python3 wait_cluster.py
+
+rllib train -f atari_impala_xlarge.yaml --ray-address=auto --queue-trials

release/rllib_tests/stress_tests/wait_cluster.py

@@ -0,0 +1,10 @@
+import ray
+import time
+
+ray.init(address="auto")
+
+curr_nodes = 0
+while not curr_nodes > 8:
+    print("Waiting for more nodes to come up: {}/{}".format(curr_nodes, 8))
+    curr_nodes = len(ray.nodes())
+    time.sleep(5)

release/rllib_tests/unit_gpu_tests/cluster.yaml

@@ -0,0 +1,39 @@
+cluster_name: ray-rllib-regression-tests
+
+min_workers: 0
+max_workers: 0
+
+# Cloud-provider specific configuration.
+provider:
+    type: aws
+    region: us-west-2
+    availability_zone: us-west-2a
+    cache_stopped_nodes: False
+
+# How Ray will authenticate with newly launched nodes.
+auth:
+    ssh_user: ubuntu
+
+head_node:
+    InstanceType: p2.xlarge  # Cheaper 1GPU K80 instance
+    ImageId: ami-07728e9e2742b0662 # Deep Learning AMI (Ubuntu 16.04)
+
+    # Set primary volume to 25 GiB
+    BlockDeviceMappings:
+        - DeviceName: /dev/sda1
+          Ebs:
+              VolumeSize: 100
+
+
+# List of shell commands to run to set up nodes.
+setup_commands: []
+
+# Command to start ray on the head node. You don't need to change this.
+head_start_ray_commands:
+    - source activate tensorflow_p36 && ray stop
+    - ulimit -n 65536; source activate tensorflow_p36 && OMP_NUM_THREADS=1 ray start --head --redis-port=6379 --object-manager-port=8076 --autoscaling-config=~/ray_bootstrap_config.yaml
+
+# Command to start ray on worker nodes. You don't need to change this.
+worker_start_ray_commands:
+    - source activate tensorflow_p36 && ray stop
+    - ulimit -n 65536; source activate tensorflow_p36 && OMP_NUM_THREADS=1 ray start --address=$RAY_HEAD_IP:6379 --object-manager-port=8076

release/rllib_tests/unit_gpu_tests/requirements.txt

@@ -0,0 +1,6 @@
+ray[rllib]
+ray[debug]
+torch==1.6+cu101
+torchvision==0.7.0+cu101
+boto3==1.4.8
+cython==0.29.0

release/rllib_tests/unit_gpu_tests/run.sh

@@ -0,0 +1,53 @@
+#!/usr/bin/env bash
+
+ray_version="" 
+commit=""
+ray_branch=""
+
+for i in "$@"
+do
+echo "$i"
+case "$i" in
+    --ray-version=*)
+    ray_version="${i#*=}"
+
+    ;;
+    --commit=*)
+    commit="${i#*=}"
+    ;;
+    --ray-branch=*)
+    ray_branch="${i#*=}"
+    ;;
+    --workload=*)
+    ;;
+    --help)
+    usage
+    exit
+    ;;
+    *)
+    echo "unknown arg, $i"
+    exit 1
+    ;;
+esac
+done
+
+if [[ $ray_version == "" || $commit == "" || $ray_branch == "" ]]
+then
+    echo "Provide --ray-version, --commit, and --ray-branch"
+    exit 1
+fi
+
+echo "version: $ray_version"
+echo "commit: $commit"
+echo "branch: $ray_branch"
+echo "workload: ignored"
+
+wheel="https://s3-us-west-2.amazonaws.com/ray-wheels/$ray_branch/$commit/ray-$ray_version-cp36-cp36m-manylinux1_x86_64.whl"
+
+conda uninstall -y terminado
+source activate tensorflow_p36 && pip install -U pip
+source activate tensorflow_p36 && pip install -U "$wheel"
+
+# Run all test cases, but with a forced num_gpus=1.
+# TODO: (sven) chose correct dir and run over all RLlib tests and example scripts!
+source activate tensorflow_p36 && export RAY_FORCE_NUM_GPUS=1 && cd ~ && python -m pytest test_attention_net_learning.py

release/stress_tests/cluster.yaml

@@ -0,0 +1,120 @@
+####################################################################
+# All nodes in this cluster will auto-terminate in 1 hour
+####################################################################
+
+# An unique identifier for the head node and workers of this cluster.
+cluster_name: ray-stress-tests
+
+# The minimum number of workers nodes to launch in addition to the head
+# node. This number should be >= 0.
+min_workers: 105
+
+# The maximum number of workers nodes to launch in addition to the head
+# node. This takes precedence over min_workers.
+max_workers: 105
+
+# The autoscaler will scale up the cluster to this target fraction of resource
+# usage. For example, if a cluster of 10 nodes is 100% busy and
+# target_utilization is 0.8, it would resize the cluster to 13. This fraction
+# can be decreased to increase the aggressiveness of upscaling.
+# This value must be less than 1.0 for scaling to happen.
+target_utilization_fraction: 0.8
+
+# If a node is idle for this many minutes, it will be removed.
+idle_timeout_minutes: 5
+
+# Cloud-provider specific configuration.
+provider:
+    type: aws
+    region: us-west-2
+    availability_zone: us-west-2a
+    cache_stopped_nodes: False
+
+# How Ray will authenticate with newly launched nodes.
+auth:
+    ssh_user: ubuntu
+# By default Ray creates a new private keypair, but you can also use your own.
+# If you do so, make sure to also set "KeyName" in the head and worker node
+# configurations below.
+#    ssh_private_key: /path/to/your/key.pem
+
+# Provider-specific config for the head node, e.g. instance type. By default
+# Ray will auto-configure unspecified fields such as SubnetId and KeyName.
+# For more documentation on available fields, see:
+# http://boto3.readthedocs.io/en/latest/reference/services/ec2.html#EC2.ServiceResource.create_instances
+head_node:
+    InstanceType: m4.16xlarge
+    ImageId: ami-06d51e91cea0dac8d # Ubuntu 18.04
+
+    # Set primary volume to 25 GiB
+    BlockDeviceMappings:
+        - DeviceName: /dev/sda1
+          Ebs:
+              VolumeSize: 100
+
+    # Additional options in the boto docs.
+
+# Provider-specific config for worker nodes, e.g. instance type. By default
+# Ray will auto-configure unspecified fields such as SubnetId and KeyName.
+# For more documentation on available fields, see:
+# http://boto3.readthedocs.io/en/latest/reference/services/ec2.html#EC2.ServiceResource.create_instances
+worker_nodes:
+    InstanceType: m4.large
+    ImageId: ami-06d51e91cea0dac8d # Ubuntu 18.04
+
+    # Set primary volume to 25 GiB
+    BlockDeviceMappings:
+        - DeviceName: /dev/sda1
+          Ebs:
+              VolumeSize: 100
+
+    # Run workers on spot by default. Comment this out to use on-demand.
+    InstanceMarketOptions:
+        MarketType: spot
+        # Additional options can be found in the boto docs, e.g.
+        #   SpotOptions:
+        #       MaxPrice: MAX_HOURLY_PRICE
+
+    # Additional options in the boto docs.
+
+# Files or directories to copy to the head and worker nodes. The format is a
+# dictionary from REMOTE_PATH: LOCAL_PATH, e.g.
+file_mounts: {
+#    "/path1/on/remote/machine": "/path1/on/local/machine",
+#    "/path2/on/remote/machine": "/path2/on/local/machine",
+}
+
+# List of shell commands to run to set up nodes.
+setup_commands: []
+    # Uncomment these if you want to build ray from source.
+    # - sudo apt-get -qq update
+    # - sudo apt-get install -y build-essential curl unzip
+    # Install Anaconda.
+    - wget --quiet https://repo.continuum.io/archive/Anaconda3-5.0.1-Linux-x86_64.sh || true
+    - bash Anaconda3-5.0.1-Linux-x86_64.sh -b -p $HOME/anaconda3 || true
+    - echo 'export PATH="$HOME/anaconda3/bin:$PATH"' >> ~/.bashrc
+    # # Build Ray.
+    # - git clone https://github.com/ray-project/ray || true
+    # - ray/ci/travis/install-bazel.sh
+    - pip install -U pip 
+    - conda uninstall -y terminado || true
+    - pip install terminado
+    - pip install boto3==1.4.8 cython==0.29.0
+    # - cd ray/python; git checkout master; git pull; pip install -e . --verbose
+    - "pip install https://s3-us-west-2.amazonaws.com/ray-wheels/{{ray_branch}}/{{commit}}/ray-{{ray_version}}-cp36-cp36m-manylinux1_x86_64.whl"
+
+# Custom commands that will be run on the head node after common setup.
+head_setup_commands: []
+
+# Custom commands that will be run on worker nodes after common setup.
+worker_setup_commands: []
+
+# Command to start ray on the head node. You don't need to change this.
+head_start_ray_commands:
+    - ray stop
+    - ulimit -n 65536; ray start --head --port=6379 --autoscaling-config=~/ray_bootstrap_config.yaml
+
+# Command to start ray on worker nodes. You don't need to change this.
+worker_start_ray_commands:
+    - ray stop
+    - ulimit -n 65536; ray start --address=$RAY_HEAD_IP:6379 --num-gpus=100

release/stress_tests/requirements.txt

@@ -0,0 +1 @@
+ray[debug]

release/stress_tests/run.sh

@@ -0,0 +1,57 @@
+#!/usr/bin/env bash
+
+ray_version="" 
+commit=""
+ray_branch=""
+workload=""
+
+for i in "$@"
+do
+echo "$i"
+case "$i" in
+    --ray-version=*)
+    ray_version="${i#*=}"
+
+    ;;
+    --commit=*)
+    commit="${i#*=}"
+    ;;
+    --ray-branch=*)
+    ray_branch="${i#*=}"
+    ;;
+    --workload=*)
+    workload="${i#*=}"
+    ;;
+    --help)
+    usage
+    exit
+    ;;
+    *)
+    echo "unknown arg, $i"
+    exit 1
+    ;;
+esac
+done
+
+echo "version: $ray_version"
+echo "commit: $commit"
+echo "branch: $ray_branch"
+echo "workload: $workload"
+
+wheel="https://s3-us-west-2.amazonaws.com/ray-wheels/$ray_branch/$commit/ray-$ray_version-cp36-cp36m-manylinux1_x86_64.whl"
+
+# Install Anaconda.
+wget --quiet https://repo.continuum.io/archive/Anaconda3-5.0.1-Linux-x86_64.sh || true
+bash Anaconda3-5.0.1-Linux-x86_64.sh -b -p "$HOME/anaconda3" || true
+# shellcheck disable=SC2016
+echo 'export PATH="$HOME/anaconda3/bin:$PATH"' >> ~/.bashrc
+
+conda uninstall -y terminado
+source activate tensorflow_p36 && pip install -U pip
+source activate tensorflow_p36 && pip install -U "$wheel"
+
+pip install -U pip 
+conda uninstall -y terminado || true
+pip install terminado
+pip install boto3==1.4.8 cython==0.29.0
+python "workloads/$workload.py"

release/stress_tests/workloads/test_dead_actors.py

@@ -0,0 +1,98 @@
+#!/usr/bin/env python
+
+import logging
+import numpy as np
+import sys
+import time
+
+import ray
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+ray.init(address="auto")
+
+# These numbers need to correspond with the autoscaler config file.
+# The number of remote nodes in the autoscaler should upper bound
+# these because sometimes nodes fail to update.
+num_remote_nodes = 100
+head_node_cpus = 2
+num_remote_cpus = num_remote_nodes * head_node_cpus
+
+# Wait until the expected number of nodes have joined the cluster.
+while True:
+    num_nodes = len(ray.nodes())
+    logger.info("Waiting for nodes {}/{}".format(num_nodes,
+                                                 num_remote_nodes + 1))
+    if num_nodes >= num_remote_nodes + 1:
+        break
+    time.sleep(5)
+logger.info("Nodes have all joined. There are %s resources.",
+            ray.cluster_resources())
+
+
+@ray.remote
+class Child(object):
+    def __init__(self, death_probability):
+        self.death_probability = death_probability
+
+    def ping(self):
+        # Exit process with some probability.
+        exit_chance = np.random.rand()
+        if exit_chance > self.death_probability:
+            sys.exit(-1)
+
+
+@ray.remote
+class Parent(object):
+    def __init__(self, num_children, death_probability):
+        self.death_probability = death_probability
+        self.children = [
+            Child.remote(death_probability) for _ in range(num_children)
+        ]
+
+    def ping(self, num_pings):
+        children_outputs = []
+        for _ in range(num_pings):
+            children_outputs += [
+                child.ping.remote() for child in self.children
+            ]
+        try:
+            ray.get(children_outputs)
+        except Exception:
+            # Replace the children if one of them died.
+            self.__init__(len(self.children), self.death_probability)
+
+    def kill(self):
+        # Clean up children.
+        ray.get([child.__ray_terminate__.remote() for child in self.children])
+
+
+num_parents = 10
+num_children = 10
+death_probability = 0.95
+
+parents = [
+    Parent.remote(num_children, death_probability) for _ in range(num_parents)
+]
+
+start = time.time()
+loop_times = []
+for i in range(100):
+    loop_start = time.time()
+    ray.get([parent.ping.remote(10) for parent in parents])
+
+    # Kill a parent actor with some probability.
+    exit_chance = np.random.rand()
+    if exit_chance > death_probability:
+        parent_index = np.random.randint(len(parents))
+        parents[parent_index].kill.remote()
+        parents[parent_index] = Parent.remote(num_children, death_probability)
+
+    logger.info("Finished trial %s", i)
+    loop_times.append(time.time() - loop_start)
+
+print("Finished in: {}s".format(time.time() - start))
+print("Average iteration time: {}s".format(sum(loop_times) / len(loop_times)))
+print("Max iteration time: {}s".format(max(loop_times)))
+print("Min iteration time: {}s".format(min(loop_times)))

release/stress_tests/workloads/test_many_tasks.py

@@ -0,0 +1,188 @@
+#!/usr/bin/env python
+
+from collections import defaultdict
+import numpy as np
+import logging
+import time
+
+import ray
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+ray.init(address="auto")
+
+# These numbers need to correspond with the autoscaler config file.
+# The number of remote nodes in the autoscaler should upper bound
+# these because sometimes nodes fail to update.
+num_remote_nodes = 100
+head_node_cpus = 2
+num_remote_cpus = num_remote_nodes * head_node_cpus
+
+# Wait until the expected number of nodes have joined the cluster.
+while True:
+    num_nodes = len(ray.nodes())
+    logger.info("Waiting for nodes {}/{}".format(num_nodes,
+                                                 num_remote_nodes + 1))
+    if num_nodes >= num_remote_nodes + 1:
+        break
+    time.sleep(5)
+logger.info("Nodes have all joined. There are %s resources.",
+            ray.cluster_resources())
+
+
+# Require 1 GPU to force the tasks to be on remote machines.
+@ray.remote(num_gpus=1)
+def f(size, *xs):
+    return np.ones(size, dtype=np.uint8)
+
+
+# Require 1 GPU to force the actors to be on remote machines.
+@ray.remote(num_cpus=1, num_gpus=1)
+class Actor(object):
+    def method(self, size, *xs):
+        return np.ones(size, dtype=np.uint8)
+
+
+# Stage 0: Submit a bunch of small tasks with large returns.
+stage_0_iterations = []
+start_time = time.time()
+logger.info("Submitting many tasks with large returns.")
+for i in range(10):
+    iteration_start = time.time()
+    logger.info("Iteration %s", i)
+    ray.get([f.remote(1000000) for _ in range(1000)])
+    stage_0_iterations.append(time.time() - iteration_start)
+
+stage_0_time = time.time() - start_time
+logger.info("Finished stage 0 after %s seconds.", stage_0_time)
+
+# Stage 1: Launch a bunch of tasks.
+stage_1_iterations = []
+start_time = time.time()
+logger.info("Submitting many tasks.")
+for i in range(10):
+    iteration_start = time.time()
+    logger.info("Iteration %s", i)
+    ray.get([f.remote(0) for _ in range(100000)])
+    stage_1_iterations.append(time.time() - iteration_start)
+
+stage_1_time = time.time() - start_time
+logger.info("Finished stage 1 after %s seconds.", stage_1_time)
+
+# Launch a bunch of tasks, each with a bunch of dependencies. TODO(rkn): This
+# test starts to fail if we increase the number of tasks in the inner loop from
+# 500 to 1000. (approximately 615 seconds)
+stage_2_iterations = []
+start_time = time.time()
+logger.info("Submitting tasks with many dependencies.")
+x_ids = []
+for _ in range(5):
+    iteration_start = time.time()
+    for i in range(20):
+        logger.info("Iteration %s. Cumulative time %s seconds", i,
+                    time.time() - start_time)
+        x_ids = [f.remote(0, *x_ids) for _ in range(500)]
+    ray.get(x_ids)
+    stage_2_iterations.append(time.time() - iteration_start)
+    logger.info("Finished after %s seconds.", time.time() - start_time)
+
+stage_2_time = time.time() - start_time
+logger.info("Finished stage 2 after %s seconds.", stage_2_time)
+
+# Create a bunch of actors.
+start_time = time.time()
+logger.info("Creating %s actors.", num_remote_cpus)
+actors = [Actor.remote() for _ in range(num_remote_cpus)]
+stage_3_creation_time = time.time() - start_time
+logger.info("Finished stage 3 actor creation in %s seconds.",
+            stage_3_creation_time)
+
+# Submit a bunch of small tasks to each actor. (approximately 1070 seconds)
+start_time = time.time()
+logger.info("Submitting many small actor tasks.")
+for N in [1000, 100000]:
+    x_ids = []
+    for i in range(N):
+        x_ids = [a.method.remote(0) for a in actors]
+        if i % 100 == 0:
+            logger.info("Submitted {}".format(i * len(actors)))
+    ray.get(x_ids)
+stage_3_time = time.time() - start_time
+logger.info("Finished stage 3 in %s seconds.", stage_3_time)
+
+# This tests https://github.com/ray-project/ray/issues/10150. The only way to
+# integration test this is via performance. The goal is to fill up the cluster
+# so that all tasks can be run, but spillback is required. Since the driver
+# submits all these tasks it should easily be able to schedule each task in
+# O(1) iterative spillback queries. If spillback behavior is incorrect, each
+# task will require O(N) queries. Since we limit the number of inflight
+# requests, we will run into head of line blocking and we should be able to
+# measure this timing.
+num_tasks = int(ray.cluster_resources()["GPU"])
+logger.info(f"Scheduling many tasks for spillback.")
+
+
+@ray.remote(num_gpus=1)
+def func(t):
+    if t % 100 == 0:
+        logger.info(f"[spillback test] {t}/{num_tasks}")
+    start = time.perf_counter()
+    time.sleep(1)
+    end = time.perf_counter()
+    return start, end, ray.worker.global_worker.node.unique_id
+
+
+results = ray.get([func.remote(i) for i in range(num_tasks)])
+
+host_to_start_times = defaultdict(list)
+for start, end, host in results:
+    host_to_start_times[host].append(start)
+
+spreads = []
+for host in host_to_start_times:
+    last = max(host_to_start_times[host])
+    first = min(host_to_start_times[host])
+    spread = last - first
+    spreads.append(spread)
+    logger.info(f"Spread: {last - first}\tLast: {last}\tFirst: {first}")
+
+# avg_spread ~ 115 with Ray 1.0 scheduler. ~695 with (buggy) 0.8.7 scheduler.
+avg_spread = sum(spreads) / len(spreads)
+logger.info(f"Avg spread: {sum(spreads)/len(spreads)}")
+
+print("Stage 0 results:")
+print("\tTotal time: {}".format(stage_0_time))
+
+print("Stage 1 results:")
+print("\tTotal time: {}".format(stage_1_time))
+print("\tAverage iteration time: {}".format(
+    sum(stage_1_iterations) / len(stage_1_iterations)))
+print("\tMax iteration time: {}".format(max(stage_1_iterations)))
+print("\tMin iteration time: {}".format(min(stage_1_iterations)))
+
+print("Stage 2 results:")
+print("\tTotal time: {}".format(stage_2_time))
+print("\tAverage iteration time: {}".format(
+    sum(stage_2_iterations) / len(stage_2_iterations)))
+print("\tMax iteration time: {}".format(max(stage_2_iterations)))
+print("\tMin iteration time: {}".format(min(stage_2_iterations)))
+
+print("Stage 3 results:")
+print("\tActor creation time: {}".format(stage_3_creation_time))
+print("\tTotal time: {}".format(stage_3_time))
+
+print("Stage 4 results:")
+print(f"\tScheduling spread: {avg_spread}.")
+
+# TODO(rkn): The test below is commented out because it currently does not
+# pass.
+# # Submit a bunch of actor tasks with all-to-all communication.
+# start_time = time.time()
+# logger.info("Submitting actor tasks with all-to-all communication.")
+# x_ids = []
+# for _ in range(50):
+#     for size_exponent in [0, 1, 2, 3, 4, 5, 6]:
+#         x_ids = [a.method.remote(10**size_exponent, *x_ids) for a in actors]
+# ray.get(x_ids)
+# logger.info("Finished after %s seconds.", time.time() - start_time)

release/stress_tests/workloads/test_placement_group.py

@@ -0,0 +1,155 @@
+# This is stress test to run placement group.
+# Please don't run it in the cluster
+# setup yet. This test uses the cluster util to simulate the
+# cluster environment.
+
+import time
+
+from time import perf_counter
+from random import random
+
+import ray
+
+from ray.cluster_utils import Cluster
+from ray.util.placement_group import (placement_group, remove_placement_group)
+
+# TODO(sang): Increase the number in the actual stress test.
+# This number should be divisible by 3.
+resource_quantity = 666
+num_nodes = 5
+custom_resources = {"pg_custom": resource_quantity}
+# Create pg that uses 1 resource of cpu & custom resource.
+num_pg = resource_quantity
+
+# TODO(sang): Cluster setup. Remove when running in real clusters.
+cluster = Cluster()
+nodes = []
+for _ in range(num_nodes):
+    nodes.append(
+        cluster.add_node(
+            num_cpus=3, num_gpus=resource_quantity,
+            resources=custom_resources))
+cluster.wait_for_nodes()
+
+ray.init(address=cluster.address)
+while not ray.is_initialized():
+    time.sleep(0.1)
+
+# Scenario 1: Create bunch of placement groups and measure how long it takes.
+total_creating_time = 0
+total_removing_time = 0
+repeat = 1
+total_trial = repeat * num_pg
+bundles = [{"GPU": 1, "pg_custom": 1}] * num_nodes
+
+# Create and remove placement groups.
+for _ in range(repeat):
+    pgs = []
+    for i in range(num_pg):
+        start = perf_counter()
+        pgs.append(placement_group(bundles, strategy="PACK", name=str(i)))
+        end = perf_counter()
+        total_creating_time += (end - start)
+
+    ray.get([pg.ready() for pg in pgs])
+
+    for pg in pgs:
+        start = perf_counter()
+        remove_placement_group(pg)
+        end = perf_counter()
+        total_removing_time += (end - start)
+
+# Validate the correctness.
+assert ray.cluster_resources()["GPU"] == num_nodes * resource_quantity
+assert ray.cluster_resources()["pg_custom"] == num_nodes * resource_quantity
+
+
+# Scenario 2:
+# - Launch 30% of placement group in the driver and pass them.
+# - Launch 70% of placement group at each remote tasks.
+# - Randomly remove placement groups and schedule tasks and actors.
+#
+# Goal:
+# - Make sure jobs are done without breaking GCS server.
+# - Make sure all the resources are recovered after the job is done.
+# - Measure the creation latency in the stressful environment.
+@ray.remote(num_cpus=0, num_gpus=1, max_calls=0)
+def mock_task():
+    time.sleep(0.1)
+    return True
+
+
+@ray.remote(num_cpus=0, num_gpus=1, max_restarts=0)
+class MockActor:
+    def __init__(self):
+        pass
+
+    def ping(self):
+        pass
+
+
+@ray.remote(num_cpus=0)
+def pg_launcher(pre_created_pgs, num_pgs_to_create):
+    pgs = []
+    pgs += pre_created_pgs
+    for i in range(num_pgs_to_create):
+        pgs.append(
+            placement_group(bundles, strategy="STRICT_SPREAD", name=str(i)))
+
+    pgs_removed = []
+    pgs_unremoved = []
+    # Randomly choose placement groups to remove.
+    for pg in pgs:
+        if random() < .5:
+            pgs_removed.append(pg)
+        else:
+            pgs_unremoved.append(pg)
+
+    tasks = []
+    max_actor_cnt = 5
+    actor_cnt = 0
+    actors = []
+    # Randomly schedule tasks or actors on placement groups that
+    # are not removed.
+    for pg in pgs_unremoved:
+        # TODO(sang): Comment in this line causes GCS actor management
+        # failure. We need to fix it.
+        if random() < .5:
+            tasks.append(mock_task.options(placement_group=pg).remote())
+        else:
+            if actor_cnt < max_actor_cnt:
+                actors.append(MockActor.options(placement_group=pg).remote())
+                actor_cnt += 1
+
+    # Remove the rest of placement groups.
+    for pg in pgs_removed:
+        remove_placement_group(pg)
+
+    ray.get([pg.ready() for pg in pgs_unremoved])
+    ray.get(tasks)
+    ray.get([actor.ping.remote() for actor in actors])
+    # Since placement groups are scheduled, remove them.
+    for pg in pgs_unremoved:
+        remove_placement_group(pg)
+
+
+pre_created_num_pgs = round(num_pg * 0.3)
+num_pgs_to_create = num_pg - pre_created_num_pgs
+pg_launchers = []
+for i in range(3):
+    pre_created_pgs = [
+        placement_group(bundles, strategy="STRICT_SPREAD")
+        for _ in range(pre_created_num_pgs // 3)
+    ]
+    pg_launchers.append(
+        pg_launcher.remote(pre_created_pgs, num_pgs_to_create // 3))
+
+ray.get(pg_launchers)
+assert ray.cluster_resources()["GPU"] == num_nodes * resource_quantity
+assert ray.cluster_resources()["pg_custom"] == num_nodes * resource_quantity
+ray.shutdown()
+print("Avg placement group creating time: "
+      f"{total_creating_time / total_trial * 1000} ms")
+print("Avg placement group removing time: "
+      f"{total_removing_time / total_trial* 1000} ms")
+print("Stress Test succeed.")