wassname/ray
1
0
Fork 0
mirror of https://github.com/wassname/ray.git synced 2026-09-12 12:51:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

[TEST]Use manager class to start/stop components instead of spreading duplicated codes everywhere (#8500)

Browse Source
This commit is contained in:
Tao Wang
2020-05-27 16:51:51 +08:00
committed by GitHub
parent c7a2e3f309
commit a1298686d7
24 changed files with 386 additions and 458 deletions
Download Patch File Download Diff File Expand all files Collapse all files
streaming/src/test/queue_tests_base.h
+16 -105
View File
@@ -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<uint64_t> {
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<int>{6379, 6380});
// flush redis first.
flushall_redis();
@@ -38,32 +34,33 @@ class StreamingQueueTestBase : public ::testing::TestWithParam<uint64_t> {
// 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<uint64_t> {
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<ObjectID> &queue_ids,
@@ -377,12 +292,8 @@ class StreamingQueueTestBase : public ::testing::TestWithParam<uint64_t> {
std::vector<std::string> raylet_socket_names_;
std::vector<std::string> 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