From c9334779153f906059ed57c11165f1c271ee074f Mon Sep 17 00:00:00 2001 From: Eric Liang Date: Wed, 28 Oct 2020 11:02:43 -0700 Subject: [PATCH] [new scheduler] Pass test_basic and add CI builds with flag on (#11635) --- .bazelrc | 1 + .travis.yml | 44 +++++++++ python/ray/memory_monitor.py | 4 +- python/ray/tests/BUILD | 92 +++++++++++++------ src/ray/common/ray_config_def.h | 6 +- src/ray/raylet/node_manager.cc | 10 +- .../raylet/scheduling/cluster_task_manager.cc | 2 - 7 files changed, 118 insertions(+), 41 deletions(-) diff --git a/.bazelrc b/.bazelrc index 3684885a2..f342bf475 100644 --- a/.bazelrc +++ b/.bazelrc @@ -21,6 +21,7 @@ test --action_env=VIRTUAL_ENV test --action_env=PYENV_VIRTUAL_ENV test --action_env=PYENV_VERSION test --action_env=PYENV_SHELL +test --action_env=RAY_ENABLE_NEW_SCHEDULER # This is needed for some core tests to run correctly test:windows --enable_runfiles # TODO(mehrdadn): Revert the "-\\.(asm|S)$" exclusion when this Bazel bug diff --git a/.travis.yml b/.travis.yml index 6603dc581..20eac886c 100644 --- a/.travis.yml +++ b/.travis.yml @@ -20,6 +20,50 @@ before_install: matrix: include: + - os: linux + env: + - PYTHON=3.6 SMALL_AND_LARGE_TESTS=1 RAY_ENABLE_NEW_SCHEDULER=1 + - PYTHONWARNINGS=ignore + - RAY_DEFAULT_BUILD=1 + - RAY_CYTHON_EXAMPLES=1 + - RAY_USE_RANDOM_PORTS=1 + install: + - . ./ci/travis/ci.sh init RAY_CI_SERVE_AFFECTED,RAY_CI_TUNE_AFFECTED,RAY_CI_PYTHON_AFFECTED,RAY_CI_DASHBOARD_AFFECTED + before_script: + - . ./ci/travis/ci.sh build + script: + # bazel python tests. This should be run last to keep its logs at the end of travis logs. + - if [ $RAY_CI_PYTHON_AFFECTED == "1" ]; then ./ci/keep_alive bazel test --config=ci $(./scripts/bazel_export_options) --test_tag_filters=-jenkins_only,-medium_size_python_tests_a_to_j,-medium_size_python_tests_k_to_z,-new_scheduler_broken python/ray/tests/...; fi + + - os: linux + env: + - PYTHON=3.6 MEDIUM_TESTS_A_TO_J=1 RAY_ENABLE_NEW_SCHEDULER=1 + - PYTHONWARNINGS=ignore + - RAY_DEFAULT_BUILD=1 + - RAY_CYTHON_EXAMPLES=1 + - RAY_USE_RANDOM_PORTS=1 + install: + - . ./ci/travis/ci.sh init RAY_CI_SERVE_AFFECTED,RAY_CI_TUNE_AFFECTED,RAY_CI_PYTHON_AFFECTED,RAY_CI_DASHBOARD_AFFECTED + before_script: + - . ./ci/travis/ci.sh build + script: + # bazel python tests for medium size tests. Used for parallelization. + - if [ $RAY_CI_PYTHON_AFFECTED == "1" ]; then ./ci/keep_alive bazel test --config=ci $(./scripts/bazel_export_options) --test_tag_filters=-jenkins_only,medium_size_python_tests_a_to_j,-new_scheduler_broken python/ray/tests/...; fi + + - os: linux + env: + - PYTHON=3.6 MEDIUM_TESTS_K_TO_Z=1 RAY_ENABLE_NEW_SCHEDULER=1 + - PYTHONWARNINGS=ignore + - RAY_DEFAULT_BUILD=1 + - RAY_CYTHON_EXAMPLES=1 + - RAY_USE_RANDOM_PORTS=1 + install: + - . ./ci/travis/ci.sh init RAY_CI_SERVE_AFFECTED,RAY_CI_TUNE_AFFECTED,RAY_CI_PYTHON_AFFECTED,RAY_CI_DASHBOARD_AFFECTED + before_script: + - . ./ci/travis/ci.sh build + script: + # bazel python tests for medium size tests. Used for parallelization. + - if [ $RAY_CI_PYTHON_AFFECTED == "1" ]; then ./ci/keep_alive bazel test --config=ci $(./scripts/bazel_export_options) --test_tag_filters=-jenkins_only,medium_size_python_tests_k_to_z,-new_scheduler_broken python/ray/tests/...; fi - os: linux env: - PYTHON=3.6 SMALL_AND_LARGE_TESTS=1 diff --git a/python/ray/memory_monitor.py b/python/ray/memory_monitor.py index c65f46d79..74bfee681 100644 --- a/python/ray/memory_monitor.py +++ b/python/ray/memory_monitor.py @@ -52,9 +52,7 @@ class RayOutOfMemoryError(Exception): f"The top 10 memory consumers are:\n\n{proc_str}" + "\n\nIn addition, up to {} GiB of shared memory is ".format( round(get_shared(psutil.virtual_memory()) / (1024**3), 2)) - + "currently being used by the Ray object store. You can set " - "the object store size with the `object_store_memory` " - "parameter when starting Ray.\n---\n" + + "currently being used by the Ray object store.\n---\n" "--- Tip: Use the `ray memory` command to list active " "objects in the cluster.\n---\n") diff --git a/python/ray/tests/BUILD b/python/ray/tests/BUILD index 84be3aa28..26d6d4b7f 100644 --- a/python/ray/tests/BUILD +++ b/python/ray/tests/BUILD @@ -10,28 +10,13 @@ SRCS = [] + select({ py_test_module_list( files = [ - "test_actor_advanced.py", - "test_actor_failures.py", - "test_actor.py", - "test_actor_resources.py", - "test_advanced_2.py", - "test_advanced_3.py", - "test_advanced.py", - "test_array.py", "test_async.py", - "test_basic_2.py", "test_basic.py", - "test_cancel.py", "test_cli.py", - "test_component_failures_2.py", "test_component_failures_3.py", - "test_dynres.py", "test_error_ray_not_initialized.py", "test_gcs_fault_tolerance.py", - "test_global_gc.py", - "test_global_state.py", "test_iter.py", - "test_joblib.py", "test_resource_demand_scheduler.py", ], size = "medium", @@ -40,25 +25,40 @@ py_test_module_list( deps = ["//:ray_lib"], ) +py_test_module_list( + files = [ + "test_actor_advanced.py", + "test_actor_failures.py", + "test_actor.py", + "test_actor_resources.py", + "test_advanced_2.py", + "test_advanced_3.py", + "test_advanced.py", + "test_array.py", + "test_basic_2.py", + "test_cancel.py", + "test_component_failures_2.py", + "test_dynres.py", + "test_global_gc.py", + "test_global_state.py", + "test_joblib.py", + ], + size = "medium", + extra_srcs = SRCS, + tags = ["exclusive", "medium_size_python_tests_a_to_j", "new_scheduler_broken"], + deps = ["//:ray_lib"], +) + py_test_module_list( files = [ "test_memory_limits.py", "test_memory_scheduling.py", "test_metrics.py", "test_multi_node_2.py", - "test_multi_tenancy.py", "test_multinode_failures_2.py", - "test_multinode_failures.py", - "test_multi_node.py", "test_multiprocessing.py", - "test_object_manager.py", "test_output.py", - "test_reconstruction.py", "test_reference_counting_2.py", - "test_reference_counting.py", - "test_serialization.py", - "test_stress.py", - "test_stress_sharded.py", "test_unreconstructable_errors.py", "test_tensorflow.py", "test_object_spilling.py", @@ -71,7 +71,24 @@ py_test_module_list( py_test_module_list( files = [ - "test_actor_pool.py", + "test_multinode_failures.py", + "test_multi_node.py", + "test_object_manager.py", + "test_reconstruction.py", + "test_reference_counting.py", + "test_serialization.py", + "test_stress.py", + "test_stress_sharded.py", + "test_multi_tenancy.py", + ], + size = "medium", + extra_srcs = SRCS, + tags = ["exclusive", "medium_size_python_tests_k_to_z", "new_scheduler_broken"], + deps = ["//:ray_lib"], +) + +py_test_module_list( + files = [ "test_args.py", "test_asyncio.py", "test_autoscaler.py", @@ -83,9 +100,7 @@ py_test_module_list( "test_dask_callback.py", "test_debug_tools.py", "test_job.py", - "test_memstat.py", "test_metrics_agent.py", - "test_microbenchmarks.py", "test_mini.py", "test_monitor.py", "test_node_manager.py", @@ -100,15 +115,36 @@ py_test_module_list( deps = ["//:ray_lib"], ) +py_test_module_list( + files = [ + "test_actor_pool.py", + "test_memstat.py", + "test_microbenchmarks.py", + ], + size = "small", + extra_srcs = SRCS, + tags = ["exclusive", "new_scheduler_broken"], + deps = ["//:ray_lib"], +) + py_test_module_list( files = [ "test_stress_failure.py", + ], + size = "large", + extra_srcs = SRCS, + tags = ["exclusive"], + deps = ["//:ray_lib"], +) + +py_test_module_list( + files = [ "test_failure.py", "test_placement_group.py", ], size = "large", extra_srcs = SRCS, - tags = ["exclusive"], + tags = ["exclusive", "new_scheduler_broken"], deps = ["//:ray_lib"], ) diff --git a/src/ray/common/ray_config_def.h b/src/ray/common/ray_config_def.h index 60e1642e5..bc1b9418b 100644 --- a/src/ray/common/ray_config_def.h +++ b/src/ray/common/ray_config_def.h @@ -108,9 +108,11 @@ RAY_CONFIG(size_t, free_objects_batch_size, 100) RAY_CONFIG(bool, lineage_pinning_enabled, false) /// Whether to enable the new scheduler. The new scheduler is designed -/// only to work with direct calls. Once direct calls afre becoming +/// only to work with direct calls. Once direct calls are becoming /// the default, this scheduler will also become the default. -RAY_CONFIG(bool, new_scheduler_enabled, false) +RAY_CONFIG(bool, new_scheduler_enabled, + getenv("RAY_ENABLE_NEW_SCHEDULER") != nullptr && + getenv("RAY_ENABLE_NEW_SCHEDULER") == std::string("1")) // The max allowed size in bytes of a return object from direct actor calls. // Objects larger than this size will be spilled/promoted to plasma. diff --git a/src/ray/raylet/node_manager.cc b/src/ray/raylet/node_manager.cc index 44226fc22..72041cacb 100644 --- a/src/ray/raylet/node_manager.cc +++ b/src/ray/raylet/node_manager.cc @@ -1357,8 +1357,6 @@ void NodeManager::ProcessDisconnectClientMessage( // Return the resources that were being used by this worker. if (new_scheduler_enabled_) { - new_resource_scheduler_->SubtractCPUResourceInstances( - worker->GetBorrowedCPUInstances()); new_resource_scheduler_->FreeLocalTaskResources(worker->GetAllocatedInstances()); worker->ClearAllocatedInstances(); new_resource_scheduler_->FreeLocalTaskResources( @@ -2206,7 +2204,7 @@ void NodeManager::SubmitTask(const Task &task) { void NodeManager::HandleDirectCallTaskBlocked( const std::shared_ptr &worker) { if (new_scheduler_enabled_) { - if (!worker) { + if (!worker || worker->IsBlocked()) { return; } std::vector cpu_instances; @@ -2236,7 +2234,8 @@ void NodeManager::HandleDirectCallTaskBlocked( void NodeManager::HandleDirectCallTaskUnblocked( const std::shared_ptr &worker) { if (new_scheduler_enabled_) { - if (!worker) { + // Important: avoid double unblocking if the unblock RPC finishes after task end. + if (!worker || !worker->IsBlocked()) { return; } std::vector cpu_instances; @@ -2246,6 +2245,7 @@ void NodeManager::HandleDirectCallTaskUnblocked( if (cpu_instances.size() > 0) { new_resource_scheduler_->SubtractCPUResourceInstances(cpu_instances); new_resource_scheduler_->AddCPUResourceInstances(worker->GetBorrowedCPUInstances()); + worker->ClearBorrowedCPUInstances(); worker->MarkUnblocked(); } ScheduleAndDispatch(); @@ -2495,8 +2495,6 @@ bool NodeManager::FinishAssignedTask(WorkerInterface &worker) { task = worker.GetAssignedTask(); // leased_workers_.erase(worker.WorkerId()); // Maybe RAY_CHECK ? if (worker.GetAllocatedInstances() != nullptr) { - new_resource_scheduler_->SubtractCPUResourceInstances( - worker.GetBorrowedCPUInstances()); new_resource_scheduler_->FreeLocalTaskResources(worker.GetAllocatedInstances()); worker.ClearAllocatedInstances(); } diff --git a/src/ray/raylet/scheduling/cluster_task_manager.cc b/src/ray/raylet/scheduling/cluster_task_manager.cc index bb0954672..5d1e4e419 100644 --- a/src/ray/raylet/scheduling/cluster_task_manager.cc +++ b/src/ray/raylet/scheduling/cluster_task_manager.cc @@ -174,8 +174,6 @@ void ClusterTaskManager::TasksUnblocked(const std::vector ready_ids) { } void ClusterTaskManager::HandleTaskFinished(std::shared_ptr worker) { - cluster_resource_scheduler_->SubtractCPUResourceInstances( - worker->GetBorrowedCPUInstances()); cluster_resource_scheduler_->FreeLocalTaskResources(worker->GetAllocatedInstances()); worker->ClearAllocatedInstances(); }