From a1298686d7022006adb1e685c9cf07b6f378a1b0 Mon Sep 17 00:00:00 2001 From: Tao Wang Date: Wed, 27 May 2020 16:51:51 +0800 Subject: [PATCH] [TEST]Use manager class to start/stop components instead of spreading duplicated codes everywhere (#8500) --- BUILD.bazel | 3 +- src/ray/common/test_util.cc | 165 ++++++++++++++--- src/ray/common/test_util.h | 69 ++++++-- src/ray/core_worker/test/core_worker_test.cc | 166 ++++-------------- .../test/global_state_accessor_test.cc | 17 +- .../test/service_based_gcs_client_test.cc | 19 +- .../gcs_server/test/gcs_server_rpc_test.cc | 14 +- .../test/redis_gcs_table_storage_test.cc | 13 +- src/ray/gcs/pubsub/test/gcs_pub_sub_test.cc | 17 +- .../test/redis_store_client_test.cc | 14 +- src/ray/gcs/test/accessor_test_base.h | 8 +- src/ray/gcs/test/asio_test.cc | 15 +- .../test/redis_actor_info_accessor_test.cc | 6 +- src/ray/gcs/test/redis_gcs_client_test.cc | 14 +- .../gcs/test/redis_job_info_accessor_test.cc | 6 +- .../gcs/test/redis_node_info_accessor_test.cc | 6 +- .../test/redis_object_info_accessor_test.cc | 6 +- .../gcs/test/subscription_executor_test.cc | 6 +- .../test/object_manager_stress_test.cc | 49 ++---- .../test/object_manager_test.cc | 47 ++--- .../raylet/object_manager_integration_test.cc | 19 +- streaming/src/test/queue_tests_base.h | 121 ++----------- .../src/test/run_streaming_queue_test.sh | 18 +- streaming/src/test/streaming_queue_tests.cc | 26 ++- 24 files changed, 386 insertions(+), 458 deletions(-) diff --git a/BUILD.bazel b/BUILD.bazel index 8b23694ea..69db22310 100644 --- a/BUILD.bazel +++ b/BUILD.bazel @@ -1205,9 +1205,10 @@ cc_library( cc_test( name = "redis_store_client_test", srcs = ["src/ray/gcs/store_client/test/redis_store_client_test.cc"], - args = ["$(location redis-server) $(location redis-cli)"], + args = ["$(location redis-server) $(location redis-cli) $(location libray_redis_module.so)"], copts = COPTS, data = [ + "//:libray_redis_module.so", "//:redis-cli", "//:redis-server", ], diff --git a/src/ray/common/test_util.cc b/src/ray/common/test_util.cc index f0f091c0f..497d9cee8 100644 --- a/src/ray/common/test_util.cc +++ b/src/ray/common/test_util.cc @@ -18,24 +18,24 @@ #include "ray/common/buffer.h" #include "ray/common/ray_object.h" +#include "ray/util/filesystem.h" #include "ray/util/logging.h" +#include "test_util.h" namespace ray { -void RedisServiceManagerForTest::SetUpTestCase() { - std::vector actual_redis_server_ports; - if (REDIS_SERVER_PORTS.empty()) { - actual_redis_server_ports.push_back(StartUpRedisServer(0)); +void TestSetupUtil::StartUpRedisServers(const std::vector &redis_server_ports) { + if (redis_server_ports.empty()) { + TEST_REDIS_SERVER_PORTS.push_back(StartUpRedisServer(0)); } else { - for (const auto &port : REDIS_SERVER_PORTS) { - actual_redis_server_ports.push_back(StartUpRedisServer(port)); + for (const auto &port : redis_server_ports) { + TEST_REDIS_SERVER_PORTS.push_back(StartUpRedisServer(port)); } } - REDIS_SERVER_PORTS = actual_redis_server_ports; } // start a redis server with specified port, use random one when 0 given -int RedisServiceManagerForTest::StartUpRedisServer(int port) { +int TestSetupUtil::StartUpRedisServer(const int &port) { int actual_port = port; if (port == 0) { // Use random port (in range [2000, 7000) to avoid port conflicts between UTs. @@ -43,12 +43,12 @@ int RedisServiceManagerForTest::StartUpRedisServer(int port) { } std::string load_module_command; - if (!REDIS_MODULE_LIBRARY_PATH.empty()) { + if (!TEST_REDIS_MODULE_LIBRARY_PATH.empty()) { // Fill load module command. - load_module_command = "--loadmodule " + REDIS_MODULE_LIBRARY_PATH; + load_module_command = "--loadmodule " + TEST_REDIS_MODULE_LIBRARY_PATH; } - std::string start_redis_command = REDIS_SERVER_EXEC_PATH + " --loglevel warning " + + std::string start_redis_command = TEST_REDIS_SERVER_EXEC_PATH + " --loglevel warning " + load_module_command + " --port " + std::to_string(actual_port) + " &"; RAY_LOG(INFO) << "Start redis command is: " << start_redis_command; @@ -57,15 +57,16 @@ int RedisServiceManagerForTest::StartUpRedisServer(int port) { return actual_port; } -void RedisServiceManagerForTest::TearDownTestCase() { - for (const auto &port : REDIS_SERVER_PORTS) { +void TestSetupUtil::ShutDownRedisServers() { + for (const auto &port : TEST_REDIS_SERVER_PORTS) { ShutDownRedisServer(port); } + TEST_REDIS_SERVER_PORTS = std::vector(); } -void RedisServiceManagerForTest::ShutDownRedisServer(int port) { +void TestSetupUtil::ShutDownRedisServer(const int &port) { std::string stop_redis_command = - REDIS_CLIENT_EXEC_PATH + " -p " + std::to_string(port) + " shutdown"; + TEST_REDIS_CLIENT_EXEC_PATH + " -p " + std::to_string(port) + " shutdown"; RAY_LOG(INFO) << "Stop redis command is: " << stop_redis_command; if (system(stop_redis_command.c_str()) != 0) { RAY_LOG(WARNING) << "Failed to stop redis. The redis process may no longer exist."; @@ -73,15 +74,15 @@ void RedisServiceManagerForTest::ShutDownRedisServer(int port) { usleep(100 * 1000); } -void RedisServiceManagerForTest::FlushAll() { - for (const auto &port : REDIS_SERVER_PORTS) { +void TestSetupUtil::FlushAllRedisServers() { + for (const auto &port : TEST_REDIS_SERVER_PORTS) { FlushRedisServer(port); } } -void RedisServiceManagerForTest::FlushRedisServer(int port) { +void TestSetupUtil::FlushRedisServer(const int &port) { std::string flush_all_redis_command = - REDIS_CLIENT_EXEC_PATH + " -p " + std::to_string(port) + " flushall"; + TEST_REDIS_CLIENT_EXEC_PATH + " -p " + std::to_string(port) + " flushall"; RAY_LOG(INFO) << "Cleaning up redis with command: " << flush_all_redis_command; if (system(flush_all_redis_command.c_str()) != 0) { RAY_LOG(WARNING) << "Failed to flush redis. The redis process may no longer exist."; @@ -89,6 +90,103 @@ void RedisServiceManagerForTest::FlushRedisServer(int port) { usleep(100 * 1000); } +std::string TestSetupUtil::StartObjectStore( + const boost::optional &socket_name) { + std::string socket_suffix; + if (socket_name) { + socket_suffix = *socket_name; + } else { + socket_suffix = ObjectID::FromRandom().Hex(); + } + std::string store_socket_name = + ray::JoinPaths(ray::GetUserTempDir(), "store" + socket_suffix); + std::string store_pid_file = store_socket_name + ".pid"; + std::string plasma_command = TEST_STORE_EXEC_PATH + " -m 10000000 -s " + + store_socket_name + + " 1> /dev/null 2> /dev/null & echo $! > " + store_pid_file; + RAY_LOG(DEBUG) << plasma_command; + RAY_CHECK(system(plasma_command.c_str()) == 0); + usleep(200 * 1000); + return store_socket_name; +} + +void TestSetupUtil::StopObjectStore(const std::string &store_socket_name) { + KillProcessBySocketName(store_socket_name); +} + +std::string TestSetupUtil::StartGcsServer(const std::string &redis_address) { + std::string gcs_server_socket_name = + ray::JoinPaths(ray::GetUserTempDir(), "gcs_server" + ObjectID::FromRandom().Hex()); + std::string gcs_server_start_cmd = TEST_GCS_SERVER_EXEC_PATH; + gcs_server_start_cmd.append(" --redis_address=" + redis_address) + .append(" --redis_port=6379") + .append(" --config_list=initial_reconstruction_timeout_milliseconds,2000") + .append(" & echo $! > " + gcs_server_socket_name + ".pid"); + + RAY_LOG(INFO) << "Start gcs server command: " << gcs_server_start_cmd; + RAY_CHECK(system(gcs_server_start_cmd.c_str()) == 0); + usleep(200 * 1000); + RAY_LOG(INFO) << "GCS server started."; + return gcs_server_socket_name; +} + +void TestSetupUtil::StopGcsServer(const std::string &gcs_server_socket_name) { + KillProcessBySocketName(gcs_server_socket_name); +} + +std::string TestSetupUtil::StartRaylet(const std::string &store_socket_name, + const std::string &node_ip_address, + const int &port, const std::string &redis_address, + const std::string &resource) { + std::string raylet_socket_name = + ray::JoinPaths(ray::GetUserTempDir(), "raylet" + ObjectID::FromRandom().Hex()); + std::string raylet_start_cmd = TEST_RAYLET_EXEC_PATH; + raylet_start_cmd.append(" --raylet_socket_name=" + raylet_socket_name) + .append(" --store_socket_name=" + store_socket_name) + .append(" --object_manager_port=0 --node_manager_port=" + std::to_string(port)) + .append(" --node_ip_address=" + node_ip_address) + .append(" --redis_address=" + redis_address) + .append(" --redis_port=6379") + .append(" --min-worker-port=0") + .append(" --max-worker-port=0") + .append(" --num_initial_workers=1") + .append(" --maximum_startup_concurrency=10") + .append(" --static_resource_list=" + resource) + .append(" --python_worker_command=\"" + TEST_MOCK_WORKER_EXEC_PATH + " " + + store_socket_name + " " + raylet_socket_name + " " + std::to_string(port) + + "\"") + .append(" --config_list=initial_reconstruction_timeout_milliseconds,2000") + .append(" & echo $! > " + raylet_socket_name + ".pid"); + + RAY_LOG(DEBUG) << "Raylet Start command: " << raylet_start_cmd; + RAY_CHECK(system(raylet_start_cmd.c_str()) == 0); + usleep(200 * 1000); + return raylet_socket_name; +} + +void TestSetupUtil::StopRaylet(const std::string &raylet_socket_name) { + KillProcessBySocketName(raylet_socket_name); +} + +std::string TestSetupUtil::StartRayletMonitor(const std::string &redis_address) { + std::string raylet_monitor_socket_name = ray::JoinPaths( + ray::GetUserTempDir(), "raylet_monitor" + ObjectID::FromRandom().Hex() + ".pid"); + std::string raylet_monitor_pid = raylet_monitor_socket_name + ".pid"; + std::string raylet_monitor_start_cmd = TEST_RAYLET_MONITOR_EXEC_PATH; + raylet_monitor_start_cmd.append(" --redis_address=" + redis_address) + .append(" --redis_port=6379") + .append(" & echo $! > " + raylet_monitor_pid); + + RAY_LOG(DEBUG) << "Raylet monitor Start command: " << raylet_monitor_start_cmd; + RAY_CHECK(system(raylet_monitor_start_cmd.c_str()) == 0); + usleep(200 * 1000); + return raylet_monitor_socket_name; +} + +void TestSetupUtil::StopRayletMonitor(const std::string &raylet_monitor_socket_name) { + KillProcessBySocketName(raylet_monitor_socket_name); +} + bool WaitForCondition(std::function condition, int timeout_ms) { int wait_time = 0; while (true) { @@ -107,6 +205,14 @@ bool WaitForCondition(std::function condition, int timeout_ms) { return false; } +void KillProcessBySocketName(std::string socket_name) { + std::string pid = socket_name + ".pid"; + std::string kill_9 = "kill -9 `cat " + pid + "`"; + RAY_LOG(DEBUG) << kill_9; + ASSERT_TRUE(system(kill_9.c_str()) == 0); + ASSERT_TRUE(system(("rm -f " + pid).c_str()) == 0); +} + TaskID RandomTaskId() { std::string data(TaskID::Size(), 0); FillRandom(&data); @@ -130,12 +236,25 @@ std::shared_ptr GenerateRandomObject( } /// Path to redis server executable binary. -std::string REDIS_SERVER_EXEC_PATH; +std::string TEST_REDIS_SERVER_EXEC_PATH; /// Path to redis client executable binary. -std::string REDIS_CLIENT_EXEC_PATH; +std::string TEST_REDIS_CLIENT_EXEC_PATH; /// Path to redis module library. -std::string REDIS_MODULE_LIBRARY_PATH; +std::string TEST_REDIS_MODULE_LIBRARY_PATH; /// Ports of redis server. -std::vector REDIS_SERVER_PORTS; +std::vector TEST_REDIS_SERVER_PORTS; + +/// Path to object store executable binary. +std::string TEST_STORE_EXEC_PATH; + +/// Path to gcs server executable binary. +std::string TEST_GCS_SERVER_EXEC_PATH; + +/// Path to raylet executable binary. +std::string TEST_RAYLET_EXEC_PATH; +/// Path to mock worker executable binary. Required by raylet. +std::string TEST_MOCK_WORKER_EXEC_PATH; +/// Path to raylet monitor executable binary. +std::string TEST_RAYLET_MONITOR_EXEC_PATH; } // namespace ray diff --git a/src/ray/common/test_util.h b/src/ray/common/test_util.h index faf707ea5..3551bd52b 100644 --- a/src/ray/common/test_util.h +++ b/src/ray/common/test_util.h @@ -20,6 +20,8 @@ #include #include +#include + #include "gtest/gtest.h" #include "ray/common/id.h" #include "ray/util/util.h" @@ -39,6 +41,9 @@ static const int64_t SHOULD_CHECK_MESSAGE_ORDER = 123450000; /// \return Whether the condition is met. bool WaitForCondition(std::function condition, int timeout_ms); +/// Used to kill process whose pid is stored in `socket_name.id` file. +void KillProcessBySocketName(std::string socket_name); + // A helper function to return a random task id. TaskID RandomTaskId(); @@ -48,24 +53,62 @@ std::shared_ptr GenerateRandomObject( const std::vector &inlined_ids = {}); /// Path to redis server executable binary. -extern std::string REDIS_SERVER_EXEC_PATH; +extern std::string TEST_REDIS_SERVER_EXEC_PATH; /// Path to redis client executable binary. -extern std::string REDIS_CLIENT_EXEC_PATH; +extern std::string TEST_REDIS_CLIENT_EXEC_PATH; /// Path to redis module library. -extern std::string REDIS_MODULE_LIBRARY_PATH; +extern std::string TEST_REDIS_MODULE_LIBRARY_PATH; /// Ports of redis server. -extern std::vector REDIS_SERVER_PORTS; +extern std::vector TEST_REDIS_SERVER_PORTS; -/// Test helper class, it will start redis server before the test runs, -/// and stop redis server after the test is completed. -class RedisServiceManagerForTest : public ::testing::Test { +/// Path to object store executable binary. +extern std::string TEST_STORE_EXEC_PATH; + +/// Path to gcs server executable binary. +extern std::string TEST_GCS_SERVER_EXEC_PATH; + +/// Path to raylet executable binary. +extern std::string TEST_RAYLET_EXEC_PATH; +/// Path to mock worker executable binary. Required by raylet. +extern std::string TEST_MOCK_WORKER_EXEC_PATH; +/// Path to raylet monitor executable binary. +extern std::string TEST_RAYLET_MONITOR_EXEC_PATH; + +//-------------------------------------------------------------------------------- +// COMPONENT MANAGEMENT CLASSES FOR TEST CASES +//-------------------------------------------------------------------------------- +/// Test cases can use it to +/// 1. start/stop/flush redis server(s) +/// 2. start/stop object store +/// 3. start/stop gcs server +/// 4. start/stop raylet +/// 5. start/stop raylet monitor +class TestSetupUtil { public: - static void SetUpTestCase(); - static int StartUpRedisServer(int port); - static void TearDownTestCase(); - static void ShutDownRedisServer(int port); - static void FlushAll(); - static void FlushRedisServer(int port); + static void StartUpRedisServers(const std::vector &redis_server_ports); + static void ShutDownRedisServers(); + static void FlushAllRedisServers(); + + static std::string StartObjectStore( + const boost::optional &socket_name = boost::none); + static void StopObjectStore(const std::string &store_socket_name); + + static std::string StartGcsServer(const std::string &redis_address); + static void StopGcsServer(const std::string &gcs_server_socket_name); + + static std::string StartRaylet(const std::string &store_socket_name, + const std::string &node_ip_address, const int &port, + const std::string &redis_address, + const std::string &resource); + static void StopRaylet(const std::string &raylet_socket_name); + + static std::string StartRayletMonitor(const std::string &redis_address); + static void StopRayletMonitor(const std::string &raylet_monitor_socket_name); + + private: + static int StartUpRedisServer(const int &port); + static void ShutDownRedisServer(const int &port); + static void FlushRedisServer(const int &port); }; } // namespace ray diff --git a/src/ray/core_worker/test/core_worker_test.cc b/src/ray/core_worker/test/core_worker_test.cc index f48059254..b73ddd15d 100644 --- a/src/ray/core_worker/test/core_worker_test.cc +++ b/src/ray/core_worker/test/core_worker_test.cc @@ -19,6 +19,7 @@ #include #include +#include "../../common/test_util.h" #include "absl/container/flat_hash_map.h" #include "absl/container/flat_hash_set.h" #include "gmock/gmock.h" @@ -38,12 +39,7 @@ namespace { -std::string store_executable; -std::string raylet_executable; int node_manager_port = 0; -std::string raylet_monitor_executable; -std::string mock_worker_executable; -std::string gcs_server_executable; } // namespace @@ -89,14 +85,15 @@ std::string MetadataToString(std::shared_ptr obj) { return std::string(reinterpret_cast(metadata->Data()), metadata->Size()); } -// inherit from RedisServiceManagerForTest for setting up redis server(s) -class CoreWorkerTest : public RedisServiceManagerForTest { +class CoreWorkerTest : public ::testing::Test { public: CoreWorkerTest(int num_nodes) : num_nodes_(num_nodes), gcs_options_("127.0.0.1", 6379, "") { #ifdef _WIN32 RAY_CHECK(false) << "port system() calls to Windows before running this test"; #endif + TestSetupUtil::StartUpRedisServers(std::vector{6379, 6380}); + // flush redis first. flushall_redis(); @@ -108,43 +105,45 @@ class CoreWorkerTest : public RedisServiceManagerForTest { // start plasma store. for (auto &store_socket : raylet_store_socket_names_) { - store_socket = StartStore(); + store_socket = TestSetupUtil::StartObjectStore(); } // start gcs server if (RayConfig::instance().gcs_service_enabled()) { - gcs_server_pid_ = StartGcsServer("127.0.0.1"); + gcs_server_socket_name_ = TestSetupUtil::StartGcsServer("127.0.0.1"); } else { // core worker test relies on node resources. It's important that one raylet can // receive the heartbeat from another. So starting raylet monitor is required here. - raylet_monitor_pid_ = StartRayletMonitor("127.0.0.1"); + raylet_monitor_socket_name_ = TestSetupUtil::StartRayletMonitor("127.0.0.1"); } // start raylet on each node. Assign each node with different resources so that // a task can be scheduled to the desired node. for (int i = 0; i < num_nodes; i++) { - raylet_socket_names_[i] = - StartRaylet(raylet_store_socket_names_[i], "127.0.0.1", node_manager_port + i, - "127.0.0.1", "\"CPU,4.0,resource" + std::to_string(i) + ",10\""); + raylet_socket_names_[i] = TestSetupUtil::StartRaylet( + raylet_store_socket_names_[i], "127.0.0.1", node_manager_port + i, "127.0.0.1", + "\"CPU,4.0,resource" + std::to_string(i) + ",10\""); } } ~CoreWorkerTest() { - for (const auto &raylet_socket : raylet_socket_names_) { - StopRaylet(raylet_socket); + for (const auto &raylet_socket_name : raylet_socket_names_) { + TestSetupUtil::StopRaylet(raylet_socket_name); } - for (const auto &store_socket : raylet_store_socket_names_) { - StopStore(store_socket); + for (const auto &store_socket_name : raylet_store_socket_names_) { + TestSetupUtil::StopObjectStore(store_socket_name); } - if (!raylet_monitor_pid_.empty()) { - StopRayletMonitor(raylet_monitor_pid_); + if (!raylet_monitor_socket_name_.empty()) { + TestSetupUtil::StopRayletMonitor(raylet_monitor_socket_name_); } - if (!gcs_server_pid_.empty()) { - StopGcsServer(gcs_server_pid_); + if (!gcs_server_socket_name_.empty()) { + TestSetupUtil::StopGcsServer(gcs_server_socket_name_); } + + TestSetupUtil::ShutDownRedisServers(); } JobID NextJobId() const { @@ -152,109 +151,6 @@ class CoreWorkerTest : public RedisServiceManagerForTest { return JobID::FromInt(job_counter++); } - std::string StartStore() { - std::string store_socket_name = - ray::JoinPaths(ray::GetUserTempDir(), "store" + ObjectID::FromRandom().Hex()); - std::string store_pid = store_socket_name + ".pid"; - std::string plasma_command = store_executable + " -m 10000000 -s " + - store_socket_name + - " 1> /dev/null 2> /dev/null & echo $! > " + store_pid; - RAY_LOG(DEBUG) << plasma_command; - RAY_CHECK(system(plasma_command.c_str()) == 0); - usleep(200 * 1000); - return store_socket_name; - } - - void StopStore(std::string store_socket_name) { - std::string store_pid = store_socket_name + ".pid"; - std::string kill_9 = "kill -9 `cat " + store_pid + "`"; - RAY_LOG(DEBUG) << kill_9; - ASSERT_EQ(system(kill_9.c_str()), 0); - ASSERT_EQ(system(("rm -rf " + store_socket_name).c_str()), 0); - ASSERT_EQ(system(("rm -rf " + store_socket_name + ".pid").c_str()), 0); - } - - std::string StartRaylet(std::string store_socket_name, std::string node_ip_address, - int port, std::string redis_address, std::string resource) { - std::string raylet_socket_name = - ray::JoinPaths(ray::GetUserTempDir(), "raylet" + ObjectID::FromRandom().Hex()); - std::string ray_start_cmd = raylet_executable; - ray_start_cmd.append(" --raylet_socket_name=" + raylet_socket_name) - .append(" --store_socket_name=" + store_socket_name) - .append(" --object_manager_port=0 --node_manager_port=" + std::to_string(port)) - .append(" --node_ip_address=" + node_ip_address) - .append(" --redis_address=" + redis_address) - .append(" --redis_port=6379") - .append(" --min-worker-port=0") - .append(" --max-worker-port=0") - .append(" --num_initial_workers=1") - .append(" --maximum_startup_concurrency=10") - .append(" --static_resource_list=" + resource) - .append(" --python_worker_command=\"" + mock_worker_executable + " " + - store_socket_name + " " + raylet_socket_name + " " + - std::to_string(port) + "\"") - .append(" --config_list=initial_reconstruction_timeout_milliseconds,2000") - .append(" & echo $! > " + raylet_socket_name + ".pid"); - - RAY_LOG(DEBUG) << "Ray Start command: " << ray_start_cmd; - RAY_CHECK(system(ray_start_cmd.c_str()) == 0); - usleep(200 * 1000); - return raylet_socket_name; - } - - void StopRaylet(std::string raylet_socket_name) { - std::string raylet_pid = raylet_socket_name + ".pid"; - std::string kill_9 = "kill -9 `cat " + raylet_pid + "`"; - RAY_LOG(DEBUG) << kill_9; - ASSERT_TRUE(system(kill_9.c_str()) == 0); - ASSERT_TRUE(system(("rm -rf " + raylet_socket_name).c_str()) == 0); - ASSERT_TRUE(system(("rm -rf " + raylet_socket_name + ".pid").c_str()) == 0); - } - - std::string StartRayletMonitor(std::string redis_address) { - std::string raylet_monitor_pid = ray::JoinPaths( - ray::GetUserTempDir(), "raylet_monitor" + ObjectID::FromRandom().Hex() + ".pid"); - std::string raylet_monitor_start_cmd = raylet_monitor_executable; - raylet_monitor_start_cmd.append(" --redis_address=" + redis_address) - .append(" --redis_port=6379") - .append(" & echo $! > " + raylet_monitor_pid); - - RAY_LOG(DEBUG) << "Raylet monitor Start command: " << raylet_monitor_start_cmd; - RAY_CHECK(system(raylet_monitor_start_cmd.c_str()) == 0); - usleep(200 * 1000); - return raylet_monitor_pid; - } - - void StopRayletMonitor(std::string raylet_monitor_pid) { - std::string kill_9 = "kill -9 `cat " + raylet_monitor_pid + "`"; - RAY_LOG(DEBUG) << kill_9; - ASSERT_TRUE(system(kill_9.c_str()) == 0); - ASSERT_TRUE(system(("rm -f " + raylet_monitor_pid).c_str()) == 0); - } - - std::string StartGcsServer(std::string redis_address) { - std::string gcs_server_pid = ray::JoinPaths( - ray::GetUserTempDir(), "gcs_server" + ObjectID::FromRandom().Hex() + ".pid"); - std::string gcs_server_start_cmd = gcs_server_executable; - gcs_server_start_cmd.append(" --redis_address=" + redis_address) - .append(" --redis_port=6379") - .append(" --config_list=initial_reconstruction_timeout_milliseconds,2000") - .append(" & echo $! > " + gcs_server_pid); - - RAY_LOG(DEBUG) << "Starting GCS server, command: " << gcs_server_start_cmd; - RAY_CHECK(system(gcs_server_start_cmd.c_str()) == 0); - usleep(200 * 1000); - RAY_LOG(INFO) << "GCS server started."; - return gcs_server_pid; - } - - void StopGcsServer(std::string gcs_server_pid) { - std::string kill_9 = "kill -9 `cat " + gcs_server_pid + "`"; - RAY_LOG(DEBUG) << kill_9; - ASSERT_TRUE(system(kill_9.c_str()) == 0); - ASSERT_TRUE(system(("rm -f " + gcs_server_pid).c_str()) == 0); - } - void SetUp() { if (num_nodes_ > 0) { CoreWorkerOptions options = { @@ -319,9 +215,9 @@ class CoreWorkerTest : public RedisServiceManagerForTest { int num_nodes_; std::vector raylet_socket_names_; std::vector raylet_store_socket_names_; - std::string raylet_monitor_pid_; + std::string raylet_monitor_socket_name_; gcs::GcsClientOptions gcs_options_; - std::string gcs_server_pid_; + std::string gcs_server_socket_name_; }; bool CoreWorkerTest::WaitForDirectCallActorState(const ActorID &actor_id, bool wait_alive, @@ -1016,22 +912,20 @@ TEST_F(TwoNodeTest, TestActorTaskCrossNodesFailure) { int main(int argc, char **argv) { ::testing::InitGoogleTest(&argc, argv); RAY_CHECK(argc == 9); - store_executable = std::string(argv[1]); - raylet_executable = std::string(argv[2]); + ray::TEST_STORE_EXEC_PATH = std::string(argv[1]); + ray::TEST_RAYLET_EXEC_PATH = std::string(argv[2]); auto seed = std::chrono::high_resolution_clock::now().time_since_epoch().count(); std::mt19937 gen(seed); std::uniform_int_distribution random_gen{2000, 2009}; // Use random port to avoid port conflicts between UTs. node_manager_port = random_gen(gen); - raylet_monitor_executable = std::string(argv[3]); - mock_worker_executable = std::string(argv[4]); - gcs_server_executable = std::string(argv[5]); + ray::TEST_RAYLET_MONITOR_EXEC_PATH = std::string(argv[3]); + ray::TEST_MOCK_WORKER_EXEC_PATH = std::string(argv[4]); + ray::TEST_GCS_SERVER_EXEC_PATH = std::string(argv[5]); - ray::REDIS_CLIENT_EXEC_PATH = std::string(argv[6]); - ray::REDIS_SERVER_EXEC_PATH = std::string(argv[7]); - ray::REDIS_MODULE_LIBRARY_PATH = std::string(argv[8]); - ray::REDIS_SERVER_PORTS.push_back(6379); - ray::REDIS_SERVER_PORTS.push_back(6380); + ray::TEST_REDIS_CLIENT_EXEC_PATH = std::string(argv[6]); + ray::TEST_REDIS_SERVER_EXEC_PATH = std::string(argv[7]); + ray::TEST_REDIS_MODULE_LIBRARY_PATH = std::string(argv[8]); return RUN_ALL_TESTS(); } diff --git a/src/ray/gcs/gcs_client/test/global_state_accessor_test.cc b/src/ray/gcs/gcs_client/test/global_state_accessor_test.cc index 5dacd564f..eb91b1cd1 100644 --- a/src/ray/gcs/gcs_client/test/global_state_accessor_test.cc +++ b/src/ray/gcs/gcs_client/test/global_state_accessor_test.cc @@ -21,7 +21,12 @@ namespace ray { -class GlobalStateAccessorTest : public RedisServiceManagerForTest { +class GlobalStateAccessorTest : public ::testing::Test { + public: + GlobalStateAccessorTest() { TestSetupUtil::StartUpRedisServers(std::vector()); } + + virtual ~GlobalStateAccessorTest() { TestSetupUtil::ShutDownRedisServers(); } + protected: void SetUp() override { config.grpc_server_port = 0; @@ -29,7 +34,7 @@ class GlobalStateAccessorTest : public RedisServiceManagerForTest { config.grpc_server_thread_num = 1; config.redis_address = "127.0.0.1"; config.is_test = true; - config.redis_port = REDIS_SERVER_PORTS.front(); + config.redis_port = TEST_REDIS_SERVER_PORTS.front(); gcs_server_.reset(new gcs::GcsServer(config)); io_service_.reset(new boost::asio::io_service()); @@ -67,7 +72,7 @@ class GlobalStateAccessorTest : public RedisServiceManagerForTest { gcs_client_->Disconnect(); global_state_->Disconnect(); global_state_.reset(); - FlushAll(); + TestSetupUtil::FlushAllRedisServers(); } bool WaitReady(std::future future, const std::chrono::milliseconds &timeout_ms) { @@ -147,8 +152,8 @@ TEST_F(GlobalStateAccessorTest, TestObjectTable) { int main(int argc, char **argv) { ::testing::InitGoogleTest(&argc, argv); RAY_CHECK(argc == 4); - ray::REDIS_SERVER_EXEC_PATH = argv[1]; - ray::REDIS_CLIENT_EXEC_PATH = argv[2]; - ray::REDIS_MODULE_LIBRARY_PATH = argv[3]; + ray::TEST_REDIS_SERVER_EXEC_PATH = argv[1]; + ray::TEST_REDIS_CLIENT_EXEC_PATH = argv[2]; + ray::TEST_REDIS_MODULE_LIBRARY_PATH = argv[3]; return RUN_ALL_TESTS(); } diff --git a/src/ray/gcs/gcs_client/test/service_based_gcs_client_test.cc b/src/ray/gcs/gcs_client/test/service_based_gcs_client_test.cc index 7ceb90eeb..ff0bad23b 100644 --- a/src/ray/gcs/gcs_client/test/service_based_gcs_client_test.cc +++ b/src/ray/gcs/gcs_client/test/service_based_gcs_client_test.cc @@ -22,7 +22,12 @@ namespace ray { -class ServiceBasedGcsClientTest : public RedisServiceManagerForTest { +class ServiceBasedGcsClientTest : public ::testing::Test { + public: + ServiceBasedGcsClientTest() { TestSetupUtil::StartUpRedisServers(std::vector()); } + + virtual ~ServiceBasedGcsClientTest() { TestSetupUtil::ShutDownRedisServers(); } + protected: void SetUp() override { config.grpc_server_port = 0; @@ -30,7 +35,7 @@ class ServiceBasedGcsClientTest : public RedisServiceManagerForTest { config.grpc_server_thread_num = 1; config.redis_address = "127.0.0.1"; config.is_test = true; - config.redis_port = REDIS_SERVER_PORTS.front(); + config.redis_port = TEST_REDIS_SERVER_PORTS.front(); gcs_server_.reset(new gcs::GcsServer(config)); io_service_.reset(new boost::asio::io_service()); @@ -60,7 +65,7 @@ class ServiceBasedGcsClientTest : public RedisServiceManagerForTest { thread_io_service_->join(); thread_gcs_server_->join(); gcs_client_->Disconnect(); - FlushAll(); + TestSetupUtil::FlushAllRedisServers(); } void RestartGcsServer() { @@ -958,7 +963,7 @@ TEST_F(ServiceBasedGcsClientTest, TestTaskTableReSubscribe) { TEST_F(ServiceBasedGcsClientTest, TestGcsRedisFailureDetector) { // Stop redis. - TearDownTestCase(); + TestSetupUtil::ShutDownRedisServers(); // Sleep 3 times of gcs_redis_heartbeat_interval_milliseconds to make sure gcs_server // detects that the redis is failure and then stop itself. @@ -974,8 +979,8 @@ TEST_F(ServiceBasedGcsClientTest, TestGcsRedisFailureDetector) { int main(int argc, char **argv) { ::testing::InitGoogleTest(&argc, argv); RAY_CHECK(argc == 4); - ray::REDIS_SERVER_EXEC_PATH = argv[1]; - ray::REDIS_CLIENT_EXEC_PATH = argv[2]; - ray::REDIS_MODULE_LIBRARY_PATH = argv[3]; + ray::TEST_REDIS_SERVER_EXEC_PATH = argv[1]; + ray::TEST_REDIS_CLIENT_EXEC_PATH = argv[2]; + ray::TEST_REDIS_MODULE_LIBRARY_PATH = argv[3]; return RUN_ALL_TESTS(); } diff --git a/src/ray/gcs/gcs_server/test/gcs_server_rpc_test.cc b/src/ray/gcs/gcs_server/test/gcs_server_rpc_test.cc index 478b7cb0c..196e3b328 100644 --- a/src/ray/gcs/gcs_server/test/gcs_server_rpc_test.cc +++ b/src/ray/gcs/gcs_server/test/gcs_server_rpc_test.cc @@ -20,8 +20,12 @@ namespace ray { -class GcsServerTest : public RedisServiceManagerForTest { +class GcsServerTest : public ::testing::Test { public: + GcsServerTest() { TestSetupUtil::StartUpRedisServers(std::vector()); } + + virtual ~GcsServerTest() { TestSetupUtil::ShutDownRedisServers(); } + void SetUp() override { gcs::GcsServerConfig config; config.grpc_server_port = 0; @@ -29,7 +33,7 @@ class GcsServerTest : public RedisServiceManagerForTest { config.grpc_server_thread_num = 1; config.redis_address = "127.0.0.1"; config.is_test = true; - config.redis_port = REDIS_SERVER_PORTS.front(); + config.redis_port = TEST_REDIS_SERVER_PORTS.front(); gcs_server_.reset(new gcs::GcsServer(config)); thread_io_service_.reset(new std::thread([this] { @@ -606,8 +610,8 @@ TEST_F(GcsServerTest, TestWorkerInfo) { int main(int argc, char **argv) { ::testing::InitGoogleTest(&argc, argv); RAY_CHECK(argc == 4); - ray::REDIS_SERVER_EXEC_PATH = argv[1]; - ray::REDIS_CLIENT_EXEC_PATH = argv[2]; - ray::REDIS_MODULE_LIBRARY_PATH = argv[3]; + ray::TEST_REDIS_SERVER_EXEC_PATH = argv[1]; + ray::TEST_REDIS_CLIENT_EXEC_PATH = argv[2]; + ray::TEST_REDIS_MODULE_LIBRARY_PATH = argv[3]; return RUN_ALL_TESTS(); } diff --git a/src/ray/gcs/gcs_server/test/redis_gcs_table_storage_test.cc b/src/ray/gcs/gcs_server/test/redis_gcs_table_storage_test.cc index befdfb65b..9a751cc97 100644 --- a/src/ray/gcs/gcs_server/test/redis_gcs_table_storage_test.cc +++ b/src/ray/gcs/gcs_server/test/redis_gcs_table_storage_test.cc @@ -22,12 +22,13 @@ namespace ray { class RedisGcsTableStorageTest : public gcs::GcsTableStorageTestBase { public: - static void SetUpTestCase() { RedisServiceManagerForTest::SetUpTestCase(); } + static void SetUpTestCase() { TestSetupUtil::StartUpRedisServers(std::vector()); } - static void TearDownTestCase() { RedisServiceManagerForTest::TearDownTestCase(); } + static void TearDownTestCase() { TestSetupUtil::ShutDownRedisServers(); } void SetUp() override { - gcs::RedisClientOptions options("127.0.0.1", REDIS_SERVER_PORTS.front(), "", true); + gcs::RedisClientOptions options("127.0.0.1", TEST_REDIS_SERVER_PORTS.front(), "", + true); redis_client_ = std::make_shared(options); RAY_CHECK_OK(redis_client_->Connect(io_service_pool_->GetAll())); @@ -49,8 +50,8 @@ TEST_F(RedisGcsTableStorageTest, TestGcsTableWithJobIdApi) { TestGcsTableWithJob int main(int argc, char **argv) { ::testing::InitGoogleTest(&argc, argv); RAY_CHECK(argc == 4); - ray::REDIS_SERVER_EXEC_PATH = argv[1]; - ray::REDIS_CLIENT_EXEC_PATH = argv[2]; - ray::REDIS_MODULE_LIBRARY_PATH = argv[3]; + ray::TEST_REDIS_SERVER_EXEC_PATH = argv[1]; + ray::TEST_REDIS_CLIENT_EXEC_PATH = argv[2]; + ray::TEST_REDIS_MODULE_LIBRARY_PATH = argv[3]; return RUN_ALL_TESTS(); } diff --git a/src/ray/gcs/pubsub/test/gcs_pub_sub_test.cc b/src/ray/gcs/pubsub/test/gcs_pub_sub_test.cc index 1ab753c7e..0e21d285a 100644 --- a/src/ray/gcs/pubsub/test/gcs_pub_sub_test.cc +++ b/src/ray/gcs/pubsub/test/gcs_pub_sub_test.cc @@ -20,7 +20,12 @@ namespace ray { -class GcsPubSubTest : public RedisServiceManagerForTest { +class GcsPubSubTest : public ::testing::Test { + public: + GcsPubSubTest() { TestSetupUtil::StartUpRedisServers(std::vector()); } + + virtual ~GcsPubSubTest() { TestSetupUtil::ShutDownRedisServers(); } + protected: virtual void SetUp() override { thread_io_service_.reset(new std::thread([this] { @@ -29,8 +34,8 @@ class GcsPubSubTest : public RedisServiceManagerForTest { io_service_.run(); })); - gcs::RedisClientOptions redis_client_options("127.0.0.1", REDIS_SERVER_PORTS.front(), - "", true); + gcs::RedisClientOptions redis_client_options( + "127.0.0.1", TEST_REDIS_SERVER_PORTS.front(), "", true); client_ = std::make_shared(redis_client_options); RAY_CHECK_OK(client_->Connect(io_service_)); pub_sub_ = std::make_shared(client_); @@ -205,8 +210,8 @@ TEST_F(GcsPubSubTest, TestPubSubWithTableData) { int main(int argc, char **argv) { ::testing::InitGoogleTest(&argc, argv); RAY_CHECK(argc == 4); - ray::REDIS_SERVER_EXEC_PATH = argv[1]; - ray::REDIS_CLIENT_EXEC_PATH = argv[2]; - ray::REDIS_MODULE_LIBRARY_PATH = argv[3]; + ray::TEST_REDIS_SERVER_EXEC_PATH = argv[1]; + ray::TEST_REDIS_CLIENT_EXEC_PATH = argv[2]; + ray::TEST_REDIS_MODULE_LIBRARY_PATH = argv[3]; return RUN_ALL_TESTS(); } diff --git a/src/ray/gcs/store_client/test/redis_store_client_test.cc b/src/ray/gcs/store_client/test/redis_store_client_test.cc index 04843a5a3..4e589b0b3 100644 --- a/src/ray/gcs/store_client/test/redis_store_client_test.cc +++ b/src/ray/gcs/store_client/test/redis_store_client_test.cc @@ -13,6 +13,7 @@ // limitations under the License. #include "ray/gcs/store_client/redis_store_client.h" +#include "ray/common/test_util.h" #include "ray/gcs/redis_client.h" #include "ray/gcs/store_client/test/store_client_test_base.h" @@ -26,12 +27,12 @@ class RedisStoreClientTest : public StoreClientTestBase { virtual ~RedisStoreClientTest() {} - static void SetUpTestCase() { RedisServiceManagerForTest::SetUpTestCase(); } + static void SetUpTestCase() { TestSetupUtil::StartUpRedisServers(std::vector()); } - static void TearDownTestCase() { RedisServiceManagerForTest::TearDownTestCase(); } + static void TearDownTestCase() { TestSetupUtil::ShutDownRedisServers(); } void InitStoreClient() override { - RedisClientOptions options("127.0.0.1", REDIS_SERVER_PORTS.front(), "", true); + RedisClientOptions options("127.0.0.1", TEST_REDIS_SERVER_PORTS.front(), "", true); redis_client_ = std::make_shared(options); RAY_CHECK_OK(redis_client_->Connect(io_service_pool_->GetAll())); @@ -60,8 +61,9 @@ TEST_F(RedisStoreClientTest, AsyncGetAllAndBatchDeleteTest) { int main(int argc, char **argv) { ::testing::InitGoogleTest(&argc, argv); - RAY_CHECK(argc == 3); - ray::REDIS_SERVER_EXEC_PATH = argv[1]; - ray::REDIS_CLIENT_EXEC_PATH = argv[2]; + RAY_CHECK(argc == 4); + ray::TEST_REDIS_SERVER_EXEC_PATH = argv[1]; + ray::TEST_REDIS_CLIENT_EXEC_PATH = argv[2]; + ray::TEST_REDIS_MODULE_LIBRARY_PATH = argv[3]; return RUN_ALL_TESTS(); } diff --git a/src/ray/gcs/test/accessor_test_base.h b/src/ray/gcs/test/accessor_test_base.h index 7d6ed71b0..798fb6432 100644 --- a/src/ray/gcs/test/accessor_test_base.h +++ b/src/ray/gcs/test/accessor_test_base.h @@ -31,17 +31,17 @@ namespace ray { namespace gcs { template -class AccessorTestBase : public RedisServiceManagerForTest { +class AccessorTestBase : public ::testing::Test { public: - AccessorTestBase() {} + AccessorTestBase() { TestSetupUtil::StartUpRedisServers(std::vector()); } - virtual ~AccessorTestBase() {} + virtual ~AccessorTestBase() { TestSetupUtil::ShutDownRedisServers(); } virtual void SetUp() { GenTestData(); GcsClientOptions options = - GcsClientOptions("127.0.0.1", REDIS_SERVER_PORTS.front(), "", true); + GcsClientOptions("127.0.0.1", TEST_REDIS_SERVER_PORTS.front(), "", true); gcs_client_.reset(new RedisGcsClient(options)); RAY_CHECK_OK(gcs_client_->Connect(io_service_)); diff --git a/src/ray/gcs/test/asio_test.cc b/src/ray/gcs/test/asio_test.cc index 54d9ef7d4..bde3fb409 100644 --- a/src/ray/gcs/test/asio_test.cc +++ b/src/ray/gcs/test/asio_test.cc @@ -46,10 +46,15 @@ void GetCallback(redisAsyncContext *c, void *r, void *privdata) { io_service.stop(); } -class RedisAsioTest : public RedisServiceManagerForTest {}; +class RedisAsioTest : public ::testing::Test { + public: + RedisAsioTest() { TestSetupUtil::StartUpRedisServers(std::vector()); } + + virtual ~RedisAsioTest() { TestSetupUtil::ShutDownRedisServers(); } +}; TEST_F(RedisAsioTest, TestRedisCommands) { - redisAsyncContext *ac = redisAsyncConnect("127.0.0.1", REDIS_SERVER_PORTS.front()); + redisAsyncContext *ac = redisAsyncConnect("127.0.0.1", TEST_REDIS_SERVER_PORTS.front()); ASSERT_TRUE(ac->err == 0); ray::gcs::RedisAsyncContext redis_async_context(ac); @@ -71,8 +76,8 @@ TEST_F(RedisAsioTest, TestRedisCommands) { int main(int argc, char **argv) { ::testing::InitGoogleTest(&argc, argv); RAY_CHECK(argc == 4); - ray::REDIS_SERVER_EXEC_PATH = argv[1]; - ray::REDIS_CLIENT_EXEC_PATH = argv[2]; - ray::REDIS_MODULE_LIBRARY_PATH = argv[3]; + ray::TEST_REDIS_SERVER_EXEC_PATH = argv[1]; + ray::TEST_REDIS_CLIENT_EXEC_PATH = argv[2]; + ray::TEST_REDIS_MODULE_LIBRARY_PATH = argv[3]; return RUN_ALL_TESTS(); } diff --git a/src/ray/gcs/test/redis_actor_info_accessor_test.cc b/src/ray/gcs/test/redis_actor_info_accessor_test.cc index aadc5cad4..605ca75ba 100644 --- a/src/ray/gcs/test/redis_actor_info_accessor_test.cc +++ b/src/ray/gcs/test/redis_actor_info_accessor_test.cc @@ -212,8 +212,8 @@ TEST_F(ActorInfoAccessorTest, GetActorCheckpointTest) { int main(int argc, char **argv) { ::testing::InitGoogleTest(&argc, argv); RAY_CHECK(argc == 4); - ray::REDIS_SERVER_EXEC_PATH = argv[1]; - ray::REDIS_CLIENT_EXEC_PATH = argv[2]; - ray::REDIS_MODULE_LIBRARY_PATH = argv[3]; + ray::TEST_REDIS_SERVER_EXEC_PATH = argv[1]; + ray::TEST_REDIS_CLIENT_EXEC_PATH = argv[2]; + ray::TEST_REDIS_MODULE_LIBRARY_PATH = argv[3]; return RUN_ALL_TESTS(); } diff --git a/src/ray/gcs/test/redis_gcs_client_test.cc b/src/ray/gcs/test/redis_gcs_client_test.cc index a8b6e30c8..4b62a33a7 100644 --- a/src/ray/gcs/test/redis_gcs_client_test.cc +++ b/src/ray/gcs/test/redis_gcs_client_test.cc @@ -31,7 +31,7 @@ namespace gcs { /* Flush redis. */ static inline void flushall_redis(void) { - redisContext *context = redisConnect("127.0.0.1", REDIS_SERVER_PORTS.front()); + redisContext *context = redisConnect("127.0.0.1", TEST_REDIS_SERVER_PORTS.front()); freeReplyObject(redisCommand(context, "FLUSHALL")); redisFree(context); } @@ -42,15 +42,17 @@ inline JobID NextJobID() { return JobID::FromInt(++counter); } -class TestGcs : public RedisServiceManagerForTest { +class TestGcs : public ::testing::Test { public: TestGcs(CommandType command_type) : num_callbacks_(0), command_type_(command_type) { + TestSetupUtil::StartUpRedisServers(std::vector()); job_id_ = NextJobID(); } virtual ~TestGcs() { // Clear all keys in the GCS. flushall_redis(); + TestSetupUtil::ShutDownRedisServers(); }; virtual void Start() = 0; @@ -85,7 +87,7 @@ class TestGcsWithAsio : public TestGcs { } void SetUp() override { - GcsClientOptions options("127.0.0.1", REDIS_SERVER_PORTS.front(), "", true); + GcsClientOptions options("127.0.0.1", TEST_REDIS_SERVER_PORTS.front(), "", true); client_ = std::make_shared(options, command_type_); RAY_CHECK_OK(client_->Connect(io_service_)); } @@ -1490,8 +1492,8 @@ TEST_F(TestGcsWithAsio, TestHashTable) { int main(int argc, char **argv) { ::testing::InitGoogleTest(&argc, argv); RAY_CHECK(argc == 4); - ray::REDIS_SERVER_EXEC_PATH = argv[1]; - ray::REDIS_CLIENT_EXEC_PATH = argv[2]; - ray::REDIS_MODULE_LIBRARY_PATH = argv[3]; + ray::TEST_REDIS_SERVER_EXEC_PATH = argv[1]; + ray::TEST_REDIS_CLIENT_EXEC_PATH = argv[2]; + ray::TEST_REDIS_MODULE_LIBRARY_PATH = argv[3]; return RUN_ALL_TESTS(); } diff --git a/src/ray/gcs/test/redis_job_info_accessor_test.cc b/src/ray/gcs/test/redis_job_info_accessor_test.cc index f1752f224..ddf1830dc 100644 --- a/src/ray/gcs/test/redis_job_info_accessor_test.cc +++ b/src/ray/gcs/test/redis_job_info_accessor_test.cc @@ -91,8 +91,8 @@ TEST_F(RedisJobInfoAccessorTest, AddAndSubscribe) { int main(int argc, char **argv) { ::testing::InitGoogleTest(&argc, argv); RAY_CHECK(argc == 4); - ray::REDIS_SERVER_EXEC_PATH = argv[1]; - ray::REDIS_CLIENT_EXEC_PATH = argv[2]; - ray::REDIS_MODULE_LIBRARY_PATH = argv[3]; + ray::TEST_REDIS_SERVER_EXEC_PATH = argv[1]; + ray::TEST_REDIS_CLIENT_EXEC_PATH = argv[2]; + ray::TEST_REDIS_MODULE_LIBRARY_PATH = argv[3]; return RUN_ALL_TESTS(); } diff --git a/src/ray/gcs/test/redis_node_info_accessor_test.cc b/src/ray/gcs/test/redis_node_info_accessor_test.cc index 0dd2d3dfe..2975f8df8 100644 --- a/src/ray/gcs/test/redis_node_info_accessor_test.cc +++ b/src/ray/gcs/test/redis_node_info_accessor_test.cc @@ -172,8 +172,8 @@ TEST_F(NodeDynamicResourceTest, Subscribe) { int main(int argc, char **argv) { ::testing::InitGoogleTest(&argc, argv); RAY_CHECK(argc == 4); - ray::REDIS_SERVER_EXEC_PATH = argv[1]; - ray::REDIS_CLIENT_EXEC_PATH = argv[2]; - ray::REDIS_MODULE_LIBRARY_PATH = argv[3]; + ray::TEST_REDIS_SERVER_EXEC_PATH = argv[1]; + ray::TEST_REDIS_CLIENT_EXEC_PATH = argv[2]; + ray::TEST_REDIS_MODULE_LIBRARY_PATH = argv[3]; return RUN_ALL_TESTS(); } diff --git a/src/ray/gcs/test/redis_object_info_accessor_test.cc b/src/ray/gcs/test/redis_object_info_accessor_test.cc index cac6c19d8..96cfcbd9a 100644 --- a/src/ray/gcs/test/redis_object_info_accessor_test.cc +++ b/src/ray/gcs/test/redis_object_info_accessor_test.cc @@ -146,8 +146,8 @@ TEST_F(RedisObjectInfoAccessorTest, TestGetAddRemove) { int main(int argc, char **argv) { ::testing::InitGoogleTest(&argc, argv); RAY_CHECK(argc == 4); - ray::REDIS_SERVER_EXEC_PATH = argv[1]; - ray::REDIS_CLIENT_EXEC_PATH = argv[2]; - ray::REDIS_MODULE_LIBRARY_PATH = argv[3]; + ray::TEST_REDIS_SERVER_EXEC_PATH = argv[1]; + ray::TEST_REDIS_CLIENT_EXEC_PATH = argv[2]; + ray::TEST_REDIS_MODULE_LIBRARY_PATH = argv[3]; return RUN_ALL_TESTS(); } diff --git a/src/ray/gcs/test/subscription_executor_test.cc b/src/ray/gcs/test/subscription_executor_test.cc index de6853bf6..fdd4d5afc 100644 --- a/src/ray/gcs/test/subscription_executor_test.cc +++ b/src/ray/gcs/test/subscription_executor_test.cc @@ -218,8 +218,8 @@ TEST_F(SubscriptionExecutorTest, UnsubscribeTest) { int main(int argc, char **argv) { ::testing::InitGoogleTest(&argc, argv); RAY_CHECK(argc == 4); - ray::REDIS_SERVER_EXEC_PATH = argv[1]; - ray::REDIS_CLIENT_EXEC_PATH = argv[2]; - ray::REDIS_MODULE_LIBRARY_PATH = argv[3]; + ray::TEST_REDIS_SERVER_EXEC_PATH = argv[1]; + ray::TEST_REDIS_CLIENT_EXEC_PATH = argv[2]; + ray::TEST_REDIS_MODULE_LIBRARY_PATH = argv[3]; return RUN_ALL_TESTS(); } diff --git a/src/ray/object_manager/test/object_manager_stress_test.cc b/src/ray/object_manager/test/object_manager_stress_test.cc index 8db2aa7d4..26a1783a1 100644 --- a/src/ray/object_manager/test/object_manager_stress_test.cc +++ b/src/ray/object_manager/test/object_manager_stress_test.cc @@ -21,6 +21,7 @@ #include "gtest/gtest.h" #include "ray/common/status.h" +#include "ray/common/test_util.h" #include "ray/object_manager/object_manager.h" #include "ray/util/filesystem.h" @@ -28,8 +29,6 @@ namespace ray { using rpc::GcsNodeInfo; -std::string store_executable; - static inline void flushall_redis(void) { redisContext *context = redisConnect("127.0.0.1", 6379); freeReplyObject(redisCommand(context, "FLUSHALL")); @@ -87,34 +86,12 @@ class TestObjectManagerBase : public ::testing::Test { #endif } - std::string StartStore(const std::string &id) { - std::string store_id = ray::JoinPaths(ray::GetUserTempDir(), "store"); - store_id = store_id + id; - std::string store_pid = store_id + ".pid"; - std::string plasma_command = store_executable + " -m 1000000000 -s " + store_id + - " 1> /dev/null 2> /dev/null &" + " echo $! > " + - store_pid; - - RAY_LOG(DEBUG) << plasma_command; - int ec = system(plasma_command.c_str()); - RAY_CHECK(ec == 0); - sleep(1); - return store_id; - } - - void StopStore(const std::string &store_id) { - std::string store_pid = store_id + ".pid"; - std::string kill_1 = "kill -9 `cat " + store_pid + "`"; - int s = system(kill_1.c_str()); - ASSERT_TRUE(!s); - } - void SetUp() { flushall_redis(); // start store - store_id_1 = StartStore(UniqueID::FromRandom().Hex()); - store_id_2 = StartStore(UniqueID::FromRandom().Hex()); + socket_name_1 = TestSetupUtil::StartObjectStore(); + socket_name_2 = TestSetupUtil::StartObjectStore(); unsigned int pull_timeout_ms = 1000; uint64_t object_chunk_size = static_cast(std::pow(10, 3)); @@ -126,7 +103,7 @@ class TestObjectManagerBase : public ::testing::Test { gcs_client_1 = std::make_shared(client_options); RAY_CHECK_OK(gcs_client_1->Connect(main_service)); ObjectManagerConfig om_config_1; - om_config_1.store_socket_name = store_id_1; + om_config_1.store_socket_name = socket_name_1; om_config_1.pull_timeout_ms = pull_timeout_ms; om_config_1.object_chunk_size = object_chunk_size; om_config_1.push_timeout_ms = push_timeout_ms; @@ -138,7 +115,7 @@ class TestObjectManagerBase : public ::testing::Test { gcs_client_2 = std::make_shared(client_options); RAY_CHECK_OK(gcs_client_2->Connect(main_service)); ObjectManagerConfig om_config_2; - om_config_2.store_socket_name = store_id_2; + om_config_2.store_socket_name = socket_name_2; om_config_2.pull_timeout_ms = pull_timeout_ms; om_config_2.object_chunk_size = object_chunk_size; om_config_2.push_timeout_ms = push_timeout_ms; @@ -147,8 +124,8 @@ class TestObjectManagerBase : public ::testing::Test { server2.reset(new MockServer(main_service, om_config_2, gcs_client_2)); // connect to stores. - RAY_ARROW_CHECK_OK(client1.Connect(store_id_1)); - RAY_ARROW_CHECK_OK(client2.Connect(store_id_2)); + RAY_ARROW_CHECK_OK(client1.Connect(socket_name_1)); + RAY_ARROW_CHECK_OK(client2.Connect(socket_name_2)); } void TearDown() { @@ -162,8 +139,8 @@ class TestObjectManagerBase : public ::testing::Test { this->server1.reset(); this->server2.reset(); - StopStore(store_id_1); - StopStore(store_id_2); + TestSetupUtil::StopObjectStore(socket_name_1); + TestSetupUtil::StopObjectStore(socket_name_2); } ObjectID WriteDataToClient(plasma::PlasmaClient &client, int64_t data_size) { @@ -171,7 +148,7 @@ class TestObjectManagerBase : public ::testing::Test { RAY_LOG(DEBUG) << "ObjectID Created: " << object_id; uint8_t metadata[] = {5}; int64_t metadata_size = sizeof(metadata); - std::shared_ptr data; + std::shared_ptr data; RAY_ARROW_CHECK_OK( client.Create(object_id.ToPlasmaId(), data_size, metadata, metadata_size, &data)); RAY_ARROW_CHECK_OK(client.Seal(object_id.ToPlasmaId())); @@ -195,8 +172,8 @@ class TestObjectManagerBase : public ::testing::Test { std::vector v1; std::vector v2; - std::string store_id_1; - std::string store_id_2; + std::string socket_name_1; + std::string socket_name_2; }; class StressTestObjectManager : public TestObjectManagerBase { @@ -461,6 +438,6 @@ TEST_F(StressTestObjectManager, StartStressTestObjectManager) { int main(int argc, char **argv) { ::testing::InitGoogleTest(&argc, argv); - ray::store_executable = std::string(argv[1]); + ray::TEST_STORE_EXEC_PATH = std::string(argv[1]); return RUN_ALL_TESTS(); } diff --git a/src/ray/object_manager/test/object_manager_test.cc b/src/ray/object_manager/test/object_manager_test.cc index 3b3b7c578..3c45a3902 100644 --- a/src/ray/object_manager/test/object_manager_test.cc +++ b/src/ray/object_manager/test/object_manager_test.cc @@ -21,10 +21,10 @@ #include "gtest/gtest.h" #include "ray/common/status.h" +#include "ray/common/test_util.h" #include "ray/util/filesystem.h" namespace { -std::string store_executable; int64_t wait_timeout_ms; } // namespace @@ -82,33 +82,12 @@ class TestObjectManagerBase : public ::testing::Test { #endif } - std::string StartStore(const std::string &id) { - std::string store_id = ray::JoinPaths(ray::GetUserTempDir(), "store"); - store_id = store_id + id; - std::string store_pid = store_id + ".pid"; - std::string plasma_command = store_executable + " -m 1000000000 -s " + store_id + - " 1> /dev/null 2> /dev/null &" + " echo $! > " + - store_pid; - - RAY_LOG(DEBUG) << plasma_command; - int ec = system(plasma_command.c_str()); - RAY_CHECK(ec == 0); - sleep(1); - return store_id; - } - - void StopStore(std::string store_id) { - std::string store_pid = store_id + ".pid"; - std::string kill_1 = "kill -9 `cat " + store_pid + "`"; - ASSERT_TRUE(!system(kill_1.c_str())); - } - void SetUp() { flushall_redis(); // start store - store_id_1 = StartStore(UniqueID::FromRandom().Hex()); - store_id_2 = StartStore(UniqueID::FromRandom().Hex()); + socket_name_1 = TestSetupUtil::StartObjectStore(); + socket_name_2 = TestSetupUtil::StartObjectStore(); unsigned int pull_timeout_ms = 1; push_timeout_ms = 1000; @@ -119,7 +98,7 @@ class TestObjectManagerBase : public ::testing::Test { gcs_client_1 = std::make_shared(client_options); RAY_CHECK_OK(gcs_client_1->Connect(main_service)); ObjectManagerConfig om_config_1; - om_config_1.store_socket_name = store_id_1; + om_config_1.store_socket_name = socket_name_1; om_config_1.pull_timeout_ms = pull_timeout_ms; om_config_1.object_chunk_size = object_chunk_size; om_config_1.push_timeout_ms = push_timeout_ms; @@ -131,7 +110,7 @@ class TestObjectManagerBase : public ::testing::Test { gcs_client_2 = std::make_shared(client_options); RAY_CHECK_OK(gcs_client_2->Connect(main_service)); ObjectManagerConfig om_config_2; - om_config_2.store_socket_name = store_id_2; + om_config_2.store_socket_name = socket_name_2; om_config_2.pull_timeout_ms = pull_timeout_ms; om_config_2.object_chunk_size = object_chunk_size; om_config_2.push_timeout_ms = push_timeout_ms; @@ -140,8 +119,8 @@ class TestObjectManagerBase : public ::testing::Test { server2.reset(new MockServer(main_service, om_config_2, gcs_client_2)); // connect to stores. - RAY_ARROW_CHECK_OK(client1.Connect(store_id_1)); - RAY_ARROW_CHECK_OK(client2.Connect(store_id_2)); + RAY_ARROW_CHECK_OK(client1.Connect(socket_name_1)); + RAY_ARROW_CHECK_OK(client2.Connect(socket_name_2)); } void TearDown() { @@ -155,8 +134,8 @@ class TestObjectManagerBase : public ::testing::Test { this->server1.reset(); this->server2.reset(); - StopStore(store_id_1); - StopStore(store_id_2); + TestSetupUtil::StopObjectStore(socket_name_1); + TestSetupUtil::StopObjectStore(socket_name_2); } ObjectID WriteDataToClient(plasma::PlasmaClient &client, int64_t data_size) { @@ -168,7 +147,7 @@ class TestObjectManagerBase : public ::testing::Test { RAY_LOG(DEBUG) << "ObjectID Created: " << object_id; uint8_t metadata[] = {5}; int64_t metadata_size = sizeof(metadata); - std::shared_ptr data; + std::shared_ptr data; RAY_ARROW_CHECK_OK( client.Create(object_id.ToPlasmaId(), data_size, metadata, metadata_size, &data)); RAY_ARROW_CHECK_OK(client.Seal(object_id.ToPlasmaId())); @@ -192,8 +171,8 @@ class TestObjectManagerBase : public ::testing::Test { std::vector v1; std::vector v2; - std::string store_id_1; - std::string store_id_2; + std::string socket_name_1; + std::string socket_name_2; unsigned int push_timeout_ms; @@ -476,7 +455,7 @@ TEST_F(TestObjectManager, StartTestObjectManager) { int main(int argc, char **argv) { ::testing::InitGoogleTest(&argc, argv); - store_executable = std::string(argv[1]); + ray::TEST_STORE_EXEC_PATH = std::string(argv[1]); wait_timeout_ms = std::stoi(std::string(argv[2])); return RUN_ALL_TESTS(); } diff --git a/src/ray/raylet/object_manager_integration_test.cc b/src/ray/raylet/object_manager_integration_test.cc index e839c9e81..707724fe2 100644 --- a/src/ray/raylet/object_manager_integration_test.cc +++ b/src/ray/raylet/object_manager_integration_test.cc @@ -17,6 +17,7 @@ #include "gtest/gtest.h" #include "ray/common/status.h" +#include "ray/common/test_util.h" #include "ray/raylet/raylet.h" #include "ray/util/filesystem.h" @@ -25,7 +26,6 @@ namespace ray { namespace raylet { std::string test_executable; -std::string store_executable; // TODO(hme): Get this working once the dust settles. class TestObjectManagerBase : public ::testing::Test { @@ -37,17 +37,6 @@ class TestObjectManagerBase : public ::testing::Test { #endif } - std::string StartStore(const std::string &id) { - std::string store_id = ray::JoinPaths(ray::GetUserTempDir(), "store"); - store_id = store_id + id; - std::string plasma_command = store_executable + " -m 1000000000 -s " + store_id + - " 1> /dev/null 2> /dev/null &"; - RAY_LOG(INFO) << plasma_command; - int ec = system(plasma_command.c_str()); - RAY_CHECK(ec == 0); - return store_id; - } - NodeManagerConfig GetNodeManagerConfig(std::string raylet_socket_name, std::string store_socket_name) { // Configuration for the node manager. @@ -69,8 +58,8 @@ class TestObjectManagerBase : public ::testing::Test { void SetUp() { // start store - std::string store_sock_1 = StartStore("1"); - std::string store_sock_2 = StartStore("2"); + std::string store_sock_1 = TestSetupUtil::StartObjectStore("1"); + std::string store_sock_2 = TestSetupUtil::StartObjectStore("2"); // start first server gcs::GcsClientOptions client_options("127.0.0.1", 6379, /*password*/ "", true); @@ -250,6 +239,6 @@ TEST_F(TestObjectManagerIntegration, StartTestObjectManagerPush) { int main(int argc, char **argv) { ::testing::InitGoogleTest(&argc, argv); ray::raylet::test_executable = std::string(argv[0]); - ray::raylet::store_executable = std::string(argv[1]); + ray::TEST_STORE_EXEC_PATH = std::string(argv[1]); return RUN_ALL_TESTS(); } diff --git a/streaming/src/test/queue_tests_base.h b/streaming/src/test/queue_tests_base.h index 0057c5e72..483728853 100644 --- a/streaming/src/test/queue_tests_base.h +++ b/streaming/src/test/queue_tests_base.h @@ -1,3 +1,4 @@ +#include "ray/common/test_util.h" #include "ray/util/filesystem.h" namespace ray { @@ -15,17 +16,12 @@ static void flushall_redis(void) { /// Base class for real-world tests with streaming queue class StreamingQueueTestBase : public ::testing::TestWithParam { public: - StreamingQueueTestBase(int num_nodes, std::string raylet_exe, std::string store_exe, - int port, std::string actor_exe, std::string gcs_server_exe) - : gcs_options_("127.0.0.1", 6379, ""), - raylet_executable_(raylet_exe), - gcs_server_executable_(gcs_server_exe), - store_executable_(store_exe), - actor_executable_(actor_exe), - node_manager_port_(port) { + StreamingQueueTestBase(int num_nodes, int port) + : gcs_options_("127.0.0.1", 6379, ""), node_manager_port_(port) { #ifdef _WIN32 RAY_CHECK(false) << "port system() calls to Windows before running this test"; #endif + TestSetupUtil::StartUpRedisServers(std::vector{6379, 6380}); // flush redis first. flushall_redis(); @@ -38,32 +34,33 @@ class StreamingQueueTestBase : public ::testing::TestWithParam { // start plasma store. for (auto &store_socket : raylet_store_socket_names_) { - store_socket = StartStore(); + store_socket = TestSetupUtil::StartObjectStore(); } // start gcs server - gcs_server_pid_ = StartGcsServer("127.0.0.1"); + gcs_server_socket_name_ = TestSetupUtil::StartGcsServer("127.0.0.1"); // start raylet on each node. Assign each node with different resources so that // a task can be scheduled to the desired node. for (int i = 0; i < num_nodes; i++) { - raylet_socket_names_[i] = - StartRaylet(raylet_store_socket_names_[i], "127.0.0.1", node_manager_port_ + i, - "127.0.0.1", "\"CPU,4.0,resource" + std::to_string(i) + ",10\""); + raylet_socket_names_[i] = TestSetupUtil::StartRaylet( + raylet_store_socket_names_[i], "127.0.0.1", node_manager_port_ + i, "127.0.0.1", + "\"CPU,4.0,resource" + std::to_string(i) + ",10\""); } } ~StreamingQueueTestBase() { STREAMING_LOG(INFO) << "Stop raylet store and actors"; - for (const auto &raylet_socket : raylet_socket_names_) { - StopRaylet(raylet_socket); + for (const auto &raylet_socket_name : raylet_socket_names_) { + TestSetupUtil::StopRaylet(raylet_socket_name); } - for (const auto &store_socket : raylet_store_socket_names_) { - StopStore(store_socket); + for (const auto &store_socket_name : raylet_store_socket_names_) { + TestSetupUtil::StopObjectStore(store_socket_name); } - StopGcsServer(gcs_server_pid_); + TestSetupUtil::StopGcsServer(gcs_server_socket_name_); + TestSetupUtil::ShutDownRedisServers(); } JobID NextJobId() const { @@ -71,88 +68,6 @@ class StreamingQueueTestBase : public ::testing::TestWithParam { return JobID::FromInt(job_counter++); } - std::string StartStore() { - std::string store_socket_name = - ray::JoinPaths(ray::GetUserTempDir(), "store" + RandomObjectID().Hex()); - std::string store_pid = store_socket_name + ".pid"; - std::string plasma_command = store_executable_ + " -m 10000000 -s " + - store_socket_name + - " 1> /dev/null 2> /dev/null & echo $! > " + store_pid; - RAY_LOG(DEBUG) << plasma_command; - RAY_CHECK(system(plasma_command.c_str()) == 0); - usleep(200 * 1000); - return store_socket_name; - } - - void StopStore(std::string store_socket_name) { - std::string store_pid = store_socket_name + ".pid"; - std::string kill_9 = "kill -9 `cat " + store_pid + "`"; - RAY_LOG(DEBUG) << kill_9; - ASSERT_EQ(system(kill_9.c_str()), 0); - ASSERT_EQ(system(("rm -rf " + store_socket_name).c_str()), 0); - ASSERT_EQ(system(("rm -rf " + store_socket_name + ".pid").c_str()), 0); - } - - std::string StartGcsServer(std::string redis_address) { - std::string gcs_server_socket_name = ray::JoinPaths( - ray::GetUserTempDir(), "gcs_server" + ObjectID::FromRandom().Hex()); - std::string ray_start_cmd = gcs_server_executable_; - ray_start_cmd.append(" --redis_address=" + redis_address) - .append(" --redis_port=6379") - .append(" --config_list=initial_reconstruction_timeout_milliseconds,2000") - .append(" & echo $! > " + gcs_server_socket_name + ".pid"); - - RAY_LOG(INFO) << "Start gcs server command: " << ray_start_cmd; - RAY_CHECK(system(ray_start_cmd.c_str()) == 0); - usleep(200 * 1000); - RAY_LOG(INFO) << "Finished start gcs server."; - return gcs_server_socket_name; - } - - void StopGcsServer(std::string gcs_server_socket_name) { - std::string gcs_server_pid = gcs_server_socket_name + ".pid"; - std::string kill_9 = "kill -9 `cat " + gcs_server_pid + "`"; - RAY_LOG(DEBUG) << kill_9; - ASSERT_TRUE(system(kill_9.c_str()) == 0); - ASSERT_TRUE(system(("rm -rf " + gcs_server_socket_name).c_str()) == 0); - ASSERT_TRUE(system(("rm -rf " + gcs_server_socket_name + ".pid").c_str()) == 0); - } - - std::string StartRaylet(std::string store_socket_name, std::string node_ip_address, - int port, std::string redis_address, std::string resource) { - std::string raylet_socket_name = - ray::JoinPaths(ray::GetUserTempDir(), "raylet" + RandomObjectID().Hex()); - std::string ray_start_cmd = raylet_executable_; - ray_start_cmd.append(" --raylet_socket_name=" + raylet_socket_name) - .append(" --store_socket_name=" + store_socket_name) - .append(" --object_manager_port=0 --node_manager_port=" + std::to_string(port)) - .append(" --node_ip_address=" + node_ip_address) - .append(" --redis_address=" + redis_address) - .append(" --redis_port=6379") - .append(" --num_initial_workers=1") - .append(" --maximum_startup_concurrency=10") - .append(" --static_resource_list=" + resource) - .append(" --python_worker_command=\"" + actor_executable_ + " " + - store_socket_name + " " + raylet_socket_name + " " + - std::to_string(port) + "\"") - .append(" --config_list=initial_reconstruction_timeout_milliseconds,2000") - .append(" & echo $! > " + raylet_socket_name + ".pid"); - - RAY_LOG(DEBUG) << "Ray Start command: " << ray_start_cmd; - RAY_CHECK(system(ray_start_cmd.c_str()) == 0); - usleep(200 * 1000); - return raylet_socket_name; - } - - void StopRaylet(std::string raylet_socket_name) { - std::string raylet_pid = raylet_socket_name + ".pid"; - std::string kill_9 = "kill -9 `cat " + raylet_pid + "`"; - RAY_LOG(DEBUG) << kill_9; - ASSERT_TRUE(system(kill_9.c_str()) == 0); - ASSERT_TRUE(system(("rm -rf " + raylet_socket_name).c_str()) == 0); - ASSERT_TRUE(system(("rm -rf " + raylet_socket_name + ".pid").c_str()) == 0); - } - void InitWorker(ActorID &self_actor_id, ActorID &peer_actor_id, const queue::protobuf::StreamingQueueTestRole role, const std::vector &queue_ids, @@ -377,12 +292,8 @@ class StreamingQueueTestBase : public ::testing::TestWithParam { std::vector raylet_socket_names_; std::vector raylet_store_socket_names_; gcs::GcsClientOptions gcs_options_; - std::string raylet_executable_; - std::string gcs_server_executable_; - std::string store_executable_; - std::string actor_executable_; int node_manager_port_; - std::string gcs_server_pid_; + std::string gcs_server_socket_name_; }; } // namespace streaming diff --git a/streaming/src/test/run_streaming_queue_test.sh b/streaming/src/test/run_streaming_queue_test.sh index 7bf4bd1dd..a8da1863a 100755 --- a/streaming/src/test/run_streaming_queue_test.sh +++ b/streaming/src/test/run_streaming_queue_test.sh @@ -35,7 +35,7 @@ if [ -z "$RAY_ROOT" ] ; then exit 1 fi -bazel build "//:core_worker_test" "//:mock_worker" "//:raylet" "//:gcs_server" "//:libray_redis_module.so" "@plasma//:plasma_store_server" +bazel build "//:core_worker_test" "//:mock_worker" "//:raylet" "//:gcs_server" "//:libray_redis_module.so" "@plasma//:plasma_store_server" "//:redis-server" "//:redis-cli" bazel build //streaming:streaming_test_worker bazel build //streaming:streaming_queue_tests @@ -46,24 +46,14 @@ if [ ! -d "$RAY_ROOT/python" ]; then fi REDIS_MODULE="./bazel-bin/libray_redis_module.so" -LOAD_MODULE_ARGS="--loadmodule ${REDIS_MODULE}" +REDIS_SERVER_EXEC="./bazel-bin/redis-server" STORE_EXEC="./bazel-bin/external/plasma/plasma_store_server" +REDIS_CLIENT_EXEC="./bazel-bin/redis-cli" RAYLET_EXEC="./bazel-bin/raylet" STREAMING_TEST_WORKER_EXEC="./bazel-bin/streaming/streaming_test_worker" GCS_SERVER_EXEC="./bazel-bin/gcs_server" # Allow cleanup commands to fail. -bazel run //:redis-cli -- -p 6379 shutdown || true -sleep 1s -bazel run //:redis-cli -- -p 6380 shutdown || true -sleep 1s -bazel run //:redis-server -- --loglevel warning ${LOAD_MODULE_ARGS} --port 6379 & -sleep 2s -bazel run //:redis-server -- --loglevel warning ${LOAD_MODULE_ARGS} --port 6380 & -sleep 2s # Run tests. -./bazel-bin/streaming/streaming_queue_tests $STORE_EXEC $RAYLET_EXEC $RAYLET_PORT $STREAMING_TEST_WORKER_EXEC $GCS_SERVER_EXEC -sleep 1s -bazel run //:redis-cli -- -p 6379 shutdown -bazel run //:redis-cli -- -p 6380 shutdown +./bazel-bin/streaming/streaming_queue_tests $STORE_EXEC $RAYLET_EXEC $RAYLET_PORT $STREAMING_TEST_WORKER_EXEC $GCS_SERVER_EXEC $REDIS_SERVER_EXEC $REDIS_MODULE $REDIS_CLIENT_EXEC sleep 1s diff --git a/streaming/src/test/streaming_queue_tests.cc b/streaming/src/test/streaming_queue_tests.cc index 4bb4d2c5a..75715acb7 100644 --- a/streaming/src/test/streaming_queue_tests.cc +++ b/streaming/src/test/streaming_queue_tests.cc @@ -2,6 +2,7 @@ #include #include "gtest/gtest.h" #include "queue/queue_client.h" +#include "ray/common/test_util.h" #include "ray/core_worker/core_worker.h" #include "data_reader.h" @@ -16,24 +17,16 @@ using namespace std::placeholders; namespace ray { namespace streaming { -static std::string store_executable; -static std::string raylet_executable; -static std::string gcs_server_executable; -static std::string actor_executable; static int node_manager_port; class StreamingWriterTest : public StreamingQueueTestBase { public: - StreamingWriterTest() - : StreamingQueueTestBase(1, raylet_executable, store_executable, node_manager_port, - actor_executable, gcs_server_executable) {} + StreamingWriterTest() : StreamingQueueTestBase(1, node_manager_port) {} }; class StreamingExactlySameTest : public StreamingQueueTestBase { public: - StreamingExactlySameTest() - : StreamingQueueTestBase(1, raylet_executable, store_executable, node_manager_port, - actor_executable, gcs_server_executable) {} + StreamingExactlySameTest() : StreamingQueueTestBase(1, node_manager_port) {} }; TEST_P(StreamingWriterTest, streaming_writer_exactly_once_test) { @@ -57,11 +50,14 @@ INSTANTIATE_TEST_CASE_P(StreamingTest, StreamingExactlySameTest, int main(int argc, char **argv) { // set_streaming_log_config("streaming_writer_test", StreamingLogLevel::INFO, 0); ::testing::InitGoogleTest(&argc, argv); - RAY_CHECK(argc == 6); - ray::streaming::store_executable = std::string(argv[1]); - ray::streaming::raylet_executable = std::string(argv[2]); + RAY_CHECK(argc == 9); + ray::TEST_STORE_EXEC_PATH = std::string(argv[1]); + ray::TEST_RAYLET_EXEC_PATH = std::string(argv[2]); ray::streaming::node_manager_port = std::stoi(std::string(argv[3])); - ray::streaming::actor_executable = std::string(argv[4]); - ray::streaming::gcs_server_executable = std::string(argv[5]); + ray::TEST_MOCK_WORKER_EXEC_PATH = std::string(argv[4]); + ray::TEST_GCS_SERVER_EXEC_PATH = std::string(argv[5]); + ray::TEST_REDIS_SERVER_EXEC_PATH = std::string(argv[6]); + ray::TEST_REDIS_MODULE_LIBRARY_PATH = std::string(argv[7]); + ray::TEST_REDIS_CLIENT_EXEC_PATH = std::string(argv[8]); return RUN_ALL_TESTS(); }