wassname/ray
1
0
Fork 0
mirror of https://github.com/wassname/ray.git synced 2026-07-06 05:16:30 +08:00
Code Issues Packages Projects Releases Wiki Activity

Improve backend debug logging, refactor scheduling queues (#3819)

Browse Source
This commit is contained in:
Stephanie Wang
2019-01-26 00:15:48 -08:00
committed by Hao Chen
parent 066fa8abf3
commit eddd60e14e
19 changed files with 509 additions and 437 deletions
Download Patch File Download Diff File Expand all files Collapse all files
doc/source/development.rst
+112 -56
View File
@@ -1,85 +1,141 @@
Development Tips
================
If you are doing development on the Ray codebase, the following tips may be
helpful.
Compilation
-----------
1. **Speeding up compilation:** Be sure to install Ray with
To speed up compilation, be sure to install Ray with
.. code-block:: shell
.. code-block:: shell
cd ray/python
pip install -e . --verbose
cd ray/python
pip install -e . --verbose
The ``-e`` means "editable", so changes you make to files in the Ray
directory will take effect without reinstalling the package. In contrast, if
you do ``python setup.py install``, files will be copied from the Ray
directory to a directory of Python packages (often something like
``/home/ubuntu/anaconda3/lib/python3.6/site-packages/ray``). This means that
changes you make to files in the Ray directory will not have any effect.
The ``-e`` means "editable", so changes you make to files in the Ray
directory will take effect without reinstalling the package. In contrast, if
you do ``python setup.py install``, files will be copied from the Ray
directory to a directory of Python packages (often something like
``/home/ubuntu/anaconda3/lib/python3.6/site-packages/ray``). This means that
changes you make to files in the Ray directory will not have any effect.
If you run into **Permission Denied** errors when running ``pip install``,
you can try adding ``--user``. You may also need to run something like ``sudo
chown -R $USER /home/ubuntu/anaconda3`` (substituting in the appropriate
path).
If you run into **Permission Denied** errors when running ``pip install``,
you can try adding ``--user``. You may also need to run something like ``sudo
chown -R $USER /home/ubuntu/anaconda3`` (substituting in the appropriate
path).
If you make changes to the C++ files, you will need to recompile them.
However, you do not need to rerun ``pip install -e .``. Instead, you can
recompile much more quickly by doing
If you make changes to the C++ files, you will need to recompile them.
However, you do not need to rerun ``pip install -e .``. Instead, you can
recompile much more quickly by doing
.. code-block:: shell
.. code-block:: shell
cd ray/build
make -j8
cd ray/build
make -j8
2. **Starting processes in a debugger:** When processes are crashing, it is
often useful to start them in a debugger (``gdb`` on Linux or ``lldb`` on
MacOS). See the latest discussion about how to do this `here`_.
Debugging
---------
3. **Running tests locally:** Suppose that one of the tests (e.g.,
``runtest.py``) is failing. You can run that test locally by running
``python test/runtest.py``. However, doing so will run all of the tests which
can take a while. To run a specific test that is failing, you can do
Starting processes in a debugger
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
When processes are crashing, it is often useful to start them in a debugger
(``gdb`` on Linux or ``lldb`` on MacOS). See the latest discussion about how to
do this `here`_.
.. code-block:: shell
You can also get a core dump of the ``raylet`` process, which is especially
useful when filing `issues`_. The process to obtain a core dump is OS-specific,
but usually involves running ``ulimit -c unlimited`` before starting Ray to
allow core dump files to be written.
cd ray
python -m pytest -v test/runtest.py::test_keyword_args
Inspecting Redis shards
~~~~~~~~~~~~~~~~~~~~~~~
To inspect Redis, you can use the ``ray.experimental.state.GlobalState`` Python
API. The easiest way to do this is to start or connect to a Ray cluster with
``ray.init()``, then query the API like so:
When running tests, usually only the first test failure matters. A single
test failure often triggers the failure of subsequent tests in the same
script.
.. code-block:: python
4. **Running linter locally:** To run the Python linter on a specific file, run
something like ``flake8 ray/python/ray/worker.py``. You may need to first run
``pip install flake8``.
ray.init()
ray.worker.global_state.client_table()
# Returns current information about the nodes in the cluster, such as:
# [{'ClientID': '2a9d2b34ad24a37ed54e4fcd32bf19f915742f5b',
# 'IsInsertion': True,
# 'NodeManagerAddress': '1.2.3.4',
# 'NodeManagerPort': 43280,
# 'ObjectManagerPort': 38062,
# 'ObjectStoreSocketName': '/tmp/ray/session_2019-01-21_16-28-05_4216/sockets/plasma_store',
# 'RayletSocketName': '/tmp/ray/session_2019-01-21_16-28-05_4216/sockets/raylet',
# 'Resources': {'CPU': 8.0, 'GPU': 1.0}}]
5. **Autoformatting code**. We use ``yapf`` https://github.com/google/yapf for
linting, and the config file is located at ``.style.yapf``. We recommend
running ``scripts/yapf.sh`` prior to pushing to format changed files.
Note that some projects such as dataframes and rllib are currently excluded.
To inspect the primary Redis shard manually, you can also query with commands
like the following.
6. **Inspecting Redis shards by hand:** To inspect the primary Redis shard by
hand, you can query it with commands like the following.
.. code-block:: python
.. code-block:: python
r_primary = ray.worker.global_worker.redis_client
r_primary.keys("*")
r_primary = ray.worker.global_worker.redis_client
r_primary.keys("*")
To inspect other Redis shards, you will need to create a new Redis client.
For example (assuming the relevant IP address is ``127.0.0.1`` and the
relevant port is ``1234``), you can do this as follows.
To inspect other Redis shards, you will need to create a new Redis client.
For example (assuming the relevant IP address is ``127.0.0.1`` and the
relevant port is ``1234``), you can do this as follows.
.. code-block:: python
.. code-block:: python
import redis
r = redis.StrictRedis(host='127.0.0.1', port=1234)
import redis
r = redis.StrictRedis(host='127.0.0.1', port=1234)
You can find a list of the relevant IP addresses and ports by running
You can find a list of the relevant IP addresses and ports by running
.. code-block:: python
.. code-block:: python
r_primary.lrange('RedisShards', 0, -1)
r_primary.lrange('RedisShards', 0, -1)
.. _backend-logging:
Backend logging
~~~~~~~~~~~~~~~
The ``raylet`` process logs detailed information about events like task
execution and object transfers between nodes. To set the logging level at
runtime, you can set the ``RAY_BACKEND_LOG_LEVEL`` environment variable before
starting Ray. For example, you can do:
.. code-block:: shell
export RAY_BACKEND_LOG_LEVEL=debug
ray start
This will print any ``RAY_LOG(DEBUG)`` lines in the source code to the
``raylet.err`` file, which you can find in the `Temporary Files`_.
Testing locally
---------------
Suppose that one of the tests (e.g., ``runtest.py``) is failing. You can run
that test locally by running ``python test/runtest.py``. However, doing so will
run all of the tests which can take a while. To run a specific test that is
failing, you can do
.. code-block:: shell
cd ray
python -m pytest -v test/runtest.py::test_keyword_args
When running tests, usually only the first test failure matters. A single
test failure often triggers the failure of subsequent tests in the same
script.
Linting
-------
**Running linter locally:** To run the Python linter on a specific file, run
something like ``flake8 ray/python/ray/worker.py``. You may need to first run
``pip install flake8``.
**Autoformatting code**. We use ``yapf`` https://github.com/google/yapf for
linting, and the config file is located at ``.style.yapf``. We recommend
running ``scripts/yapf.sh`` prior to pushing to format changed files.
Note that some projects such as dataframes and rllib are currently excluded.
.. _`issues`: https://github.com/ray-project/ray/issues
.. _`here`: https://github.com/ray-project/ray/issues/108
.. _`Temporary Files`: http://ray.readthedocs.io/en/latest/tempfile.html
src/ray/common/client_connection.cc
+17 -9
View File
@@ -187,22 +187,24 @@ template <class T>
std::shared_ptr<ClientConnection<T>> ClientConnection<T>::Create(
ClientHandler<T> &client_handler, MessageHandler<T> &message_handler,
boost::asio::basic_stream_socket<T> &&socket, const std::string &debug_label,
int64_t error_message_type) {
std::shared_ptr<ClientConnection<T>> self(new ClientConnection(
message_handler, std::move(socket), debug_label, error_message_type));
const std::vector<std::string> &message_type_enum_names, int64_t error_message_type) {
std::shared_ptr<ClientConnection<T>> self(
new ClientConnection(message_handler, std::move(socket), debug_label,
message_type_enum_names, error_message_type));
// Let our manager process our new connection.
client_handler(*self);
return self;
}
template <class T>
ClientConnection<T>::ClientConnection(MessageHandler<T> &message_handler,
boost::asio::basic_stream_socket<T> &&socket,
const std::string &debug_label,
int64_t error_message_type)
ClientConnection<T>::ClientConnection(
MessageHandler<T> &message_handler, boost::asio::basic_stream_socket<T> &&socket,
const std::string &debug_label,
const std::vector<std::string> &message_type_enum_names, int64_t error_message_type)
: ServerConnection<T>(std::move(socket)),
message_handler_(message_handler),
debug_label_(debug_label),
message_type_enum_names_(message_type_enum_names),
error_message_type_(error_message_type) {}
template <class T>
@@ -261,8 +263,14 @@ void ClientConnection<T>::ProcessMessage(const boost::system::error_code &error)
message_handler_(shared_ClientConnection_from_this(), read_type_, read_message_.data());
int64_t interval = current_time_ms() - start_ms;
if (interval > RayConfig::instance().handler_warning_timeout_ms()) {
RAY_LOG(WARNING) << "[" << debug_label_ << "]ProcessMessage with type " << read_type_
<< " took " << interval << " ms.";
std::string message_type;
if (message_type_enum_names_.empty()) {
message_type = std::to_string(read_type_);
} else {
message_type = message_type_enum_names_[read_type_];
}
RAY_LOG(WARNING) << "[" << debug_label_ << "]ProcessMessage with type "
<< message_type << " took " << interval << " ms.";
}
}
src/ray/common/client_connection.h
+11 -1
View File
@@ -145,10 +145,15 @@ class ClientConnection : public ServerConnection<T> {
/// \param new_client_handler A reference to the client handler.
/// \param message_handler A reference to the message handler.
/// \param socket The client socket.
/// \param debug_label Label that is printed in debug messages, to identify
/// the type of client.
/// \param message_type_enum_names A table of printable enum names for the
/// message types received from this client, used for debug messages.
/// \return std::shared_ptr<ClientConnection>.
static std::shared_ptr<ClientConnection<T>> Create(
ClientHandler<T> &new_client_handler, MessageHandler<T> &message_handler,
boost::asio::basic_stream_socket<T> &&socket, const std::string &debug_label,
const std::vector<std::string> &message_type_enum_names,
int64_t error_message_type);
std::shared_ptr<ClientConnection<T>> shared_ClientConnection_from_this() {
@@ -170,7 +175,9 @@ class ClientConnection : public ServerConnection<T> {
/// A private constructor for a node client connection.
ClientConnection(MessageHandler<T> &message_handler,
boost::asio::basic_stream_socket<T> &&socket,
const std::string &debug_label, int64_t error_message_type);
const std::string &debug_label,
const std::vector<std::string> &message_type_enum_names,
int64_t error_message_type);
/// Process an error from the last operation, then process the message
/// header from the client.
void ProcessMessageHeader(const boost::system::error_code &error);
@@ -184,6 +191,9 @@ class ClientConnection : public ServerConnection<T> {
MessageHandler<T> message_handler_;
/// A label used for debug messages.
const std::string debug_label_;
/// A table of printable enum names for the message types, used for debug
/// messages.
const std::vector<std::string> message_type_enum_names_;
/// The value for disconnect client message.
int64_t error_message_type_;
/// Buffers for the current message being read from the client.
src/ray/gcs/tables.cc
+3 -13
View File
@@ -247,10 +247,7 @@ Status ErrorTable::PushErrorToDriver(const JobID &job_id, const std::string &typ
data->type = type;
data->error_message = error_message;
data->timestamp = timestamp;
return Append(job_id, job_id, data, [](ray::gcs::AsyncGcsClient *client,
const JobID &id, const ErrorTableDataT &data) {
RAY_LOG(DEBUG) << "Error message pushed callback";
});
return Append(job_id, job_id, data, /*done_callback=*/nullptr);
}
std::string ErrorTable::DebugString() const {
@@ -264,10 +261,7 @@ Status ProfileTable::AddProfileEventBatch(const ProfileTableData &profile_events
profile_events.UnPackTo(data.get());
return Append(JobID::nil(), UniqueID::from_random(), data,
[](ray::gcs::AsyncGcsClient *client, const JobID &id,
const ProfileTableDataT &data) {
RAY_LOG(DEBUG) << "Profile message pushed callback";
});
/*done_callback=*/nullptr);
}
std::string ProfileTable::DebugString() const {
@@ -278,11 +272,7 @@ Status DriverTable::AppendDriverData(const JobID &driver_id, bool is_dead) {
auto data = std::make_shared<DriverTableDataT>();
data->driver_id = driver_id.binary();
data->is_dead = is_dead;
return Append(driver_id, driver_id, data,
[](ray::gcs::AsyncGcsClient *client, const JobID &id,
const DriverTableDataT &data) {
RAY_LOG(DEBUG) << "Driver entry added callback";
});
return Append(driver_id, driver_id, data, /*done_callback=*/nullptr);
}
void ClientTable::RegisterClientAddedCallback(const ClientTableCallback &callback) {
src/ray/object_manager/connection_pool.cc
-2
View File
@@ -123,14 +123,12 @@ std::shared_ptr<SenderConnection> ConnectionPool::Borrow(SenderMapType &conn_map
const ClientID &client_id) {
std::shared_ptr<SenderConnection> conn = std::move(conn_map[client_id].back());
conn_map[client_id].pop_back();
RAY_LOG(DEBUG) << "Borrow " << client_id << " " << conn_map[client_id].size();
return conn;
}
void ConnectionPool::Return(SenderMapType &conn_map, const ClientID &client_id,
std::shared_ptr<SenderConnection> conn) {
conn_map[client_id].push_back(std::move(conn));
RAY_LOG(DEBUG) << "Return " << client_id << " " << conn_map[client_id].size();
}
std::string ConnectionPool::DebugString() const {
src/ray/object_manager/object_buffer_pool.cc
-6
View File
@@ -41,7 +41,6 @@ std::pair<const ObjectBufferPool::ChunkInfo &, ray::Status> ObjectBufferPool::Ge
const ObjectID &object_id, uint64_t data_size, uint64_t metadata_size,
uint64_t chunk_index) {
std::lock_guard<std::mutex> lock(pool_mutex_);
RAY_LOG(DEBUG) << "GetChunk " << object_id << " " << data_size << " " << metadata_size;
if (get_buffer_state_.count(object_id) == 0) {
plasma::ObjectBuffer object_buffer;
plasma::ObjectID plasma_id = object_id.to_plasma_id();
@@ -72,7 +71,6 @@ void ObjectBufferPool::ReleaseGetChunk(const ObjectID &object_id, uint64_t chunk
std::lock_guard<std::mutex> lock(pool_mutex_);
GetBufferState &buffer_state = get_buffer_state_[object_id];
buffer_state.references--;
RAY_LOG(DEBUG) << "ReleaseBuffer " << object_id << " " << buffer_state.references;
if (buffer_state.references == 0) {
RAY_ARROW_CHECK_OK(store_client_.Release(object_id.to_plasma_id()));
get_buffer_state_.erase(object_id);
@@ -89,8 +87,6 @@ std::pair<const ObjectBufferPool::ChunkInfo &, ray::Status> ObjectBufferPool::Cr
const ObjectID &object_id, uint64_t data_size, uint64_t metadata_size,
uint64_t chunk_index) {
std::lock_guard<std::mutex> lock(pool_mutex_);
RAY_LOG(DEBUG) << "CreateChunk " << object_id << " " << data_size << " "
<< metadata_size;
if (create_buffer_state_.count(object_id) == 0) {
const plasma::ObjectID plasma_id = object_id.to_plasma_id();
int64_t object_size = data_size - metadata_size;
@@ -153,8 +149,6 @@ void ObjectBufferPool::SealChunk(const ObjectID &object_id, const uint64_t chunk
CreateChunkState::REFERENCED);
create_buffer_state_[object_id].chunk_state[chunk_index] = CreateChunkState::SEALED;
create_buffer_state_[object_id].num_seals_remaining--;
RAY_LOG(DEBUG) << "SealChunk" << object_id << " "
<< create_buffer_state_[object_id].num_seals_remaining;
if (create_buffer_state_[object_id].num_seals_remaining == 0) {
const plasma::ObjectID plasma_id = object_id.to_plasma_id();
RAY_ARROW_CHECK_OK(store_client_.Seal(plasma_id));
src/ray/object_manager/object_manager.cc
+31 -11
View File
@@ -109,6 +109,7 @@ ray::Status ObjectManager::SubscribeObjDeleted(
}
ray::Status ObjectManager::Pull(const ObjectID &object_id) {
RAY_LOG(DEBUG) << "Pull on " << client_id_ << " of object " << object_id;
// Check if object is already local.
if (local_objects_.count(object_id) != 0) {
RAY_LOG(ERROR) << object_id << " attempted to pull an object that's already local.";
@@ -188,6 +189,8 @@ void ObjectManager::TryPull(const ObjectID &object_id) {
RAY_CHECK(client_id != client_id_);
}
RAY_LOG(DEBUG) << "Sending pull request from " << client_id_ << " to " << client_id
<< " of object " << object_id;
// Try pulling from the client.
PullEstablishConnection(object_id, client_id);
@@ -289,6 +292,9 @@ void ObjectManager::HandleSendFinished(const ObjectID &object_id,
const ClientID &client_id, uint64_t chunk_index,
double start_time, double end_time,
ray::Status status) {
RAY_LOG(DEBUG) << "HandleSendFinished on " << client_id_ << " to " << client_id
<< " of object " << object_id << " chunk " << chunk_index
<< ", status: " << status.ToString();
if (!status.ok()) {
// TODO(rkn): What do we want to do if the send failed?
}
@@ -326,6 +332,8 @@ void ObjectManager::HandleReceiveFinished(const ObjectID &object_id,
}
void ObjectManager::Push(const ObjectID &object_id, const ClientID &client_id) {
RAY_LOG(DEBUG) << "Push on " << client_id_ << " to " << client_id << " of object "
<< object_id;
if (local_objects_.count(object_id) == 0) {
// Avoid setting duplicated timer for the same object and client pair.
auto &clients = unfulfilled_push_requests_[object_id];
@@ -374,6 +382,7 @@ void ObjectManager::Push(const ObjectID &object_id, const ClientID &client_id) {
RayConfig::instance().object_manager_repeated_push_delay_ms()) {
// We pushed this object to the object manager recently, so don't do it
// again.
RAY_LOG(DEBUG) << "Object " << object_id << " recently pushed to " << client_id;
return;
} else {
it->second = current_time;
@@ -420,8 +429,8 @@ ray::Status ObjectManager::ExecuteSendObject(
const ClientID &client_id, const ObjectID &object_id, uint64_t data_size,
uint64_t metadata_size, uint64_t chunk_index,
const RemoteConnectionInfo &connection_info) {
RAY_LOG(DEBUG) << "ExecuteSendObject " << client_id << " " << object_id << " "
<< chunk_index;
RAY_LOG(DEBUG) << "ExecuteSendObject on " << client_id_ << " to " << client_id
<< " of object " << object_id << " chunk " << chunk_index;
ray::Status status;
std::shared_ptr<SenderConnection> conn;
connection_pool_.GetSender(ConnectionPool::ConnectionType::TRANSFER, client_id, &conn);
@@ -485,8 +494,6 @@ ray::Status ObjectManager::SendObjectData(const ObjectID &object_id,
if (status.ok()) {
connection_pool_.ReleaseSender(ConnectionPool::ConnectionType::TRANSFER, conn);
RAY_LOG(DEBUG) << "SendCompleted " << client_id_ << " " << object_id << " "
<< config_.max_sends;
}
return status;
}
@@ -506,6 +513,7 @@ ray::Status ObjectManager::Wait(const std::vector<ObjectID> &object_ids,
int64_t timeout_ms, uint64_t num_required_objects,
bool wait_local, const WaitCallback &callback) {
UniqueID wait_id = UniqueID::from_random();
RAY_LOG(DEBUG) << "Wait request " << wait_id << " on " << client_id_;
RAY_RETURN_NOT_OK(AddWaitRequest(wait_id, object_ids, timeout_ms, num_required_objects,
wait_local, callback));
RAY_RETURN_NOT_OK(LookupRemainingWaitObjects(wait_id));
@@ -570,6 +578,8 @@ ray::Status ObjectManager::LookupRemainingWaitObjects(const UniqueID &wait_id) {
wait_state.remaining.erase(lookup_object_id);
wait_state.found.insert(lookup_object_id);
}
RAY_LOG(DEBUG) << "Wait request " << wait_id << ": " << client_ids.size()
<< " locations found for object " << lookup_object_id;
wait_state.requested_objects.erase(lookup_object_id);
if (wait_state.requested_objects.empty()) {
SubscribeRemainingWaitObjects(wait_id);
@@ -593,6 +603,8 @@ void ObjectManager::SubscribeRemainingWaitObjects(const UniqueID &wait_id) {
// locations from the object directory.
for (const auto &object_id : wait_state.object_id_order) {
if (wait_state.remaining.count(object_id) > 0) {
RAY_LOG(DEBUG) << "Wait request " << wait_id << ": subscribing to object "
<< object_id;
wait_state.requested_objects.insert(object_id);
// Subscribe to object notifications.
RAY_CHECK_OK(object_directory_->SubscribeObjectLocations(
@@ -600,6 +612,9 @@ void ObjectManager::SubscribeRemainingWaitObjects(const UniqueID &wait_id) {
[this, wait_id](const ObjectID &subscribe_object_id,
const std::unordered_set<ClientID> &client_ids, bool created) {
if (!client_ids.empty()) {
RAY_LOG(DEBUG) << "Wait request " << wait_id
<< ": subscription notification received for object "
<< subscribe_object_id;
auto object_id_wait_state = active_wait_requests_.find(wait_id);
if (object_id_wait_state == active_wait_requests_.end()) {
// Depending on the timing of calls to the object directory, we
@@ -675,6 +690,8 @@ void ObjectManager::WaitComplete(const UniqueID &wait_id) {
}
wait_state.callback(found, remaining);
active_wait_requests_.erase(wait_id);
RAY_LOG(DEBUG) << "Wait request " << wait_id << " finished: found " << found.size()
<< " remaining " << remaining.size();
}
std::shared_ptr<SenderConnection> ObjectManager::CreateSenderConnection(
@@ -707,8 +724,11 @@ void ObjectManager::ProcessNewClient(TcpClientConnection &conn) {
void ObjectManager::ProcessClientMessage(std::shared_ptr<TcpClientConnection> &conn,
int64_t message_type, const uint8_t *message) {
auto message_type_value =
const auto message_type_value =
static_cast<object_manager_protocol::MessageType>(message_type);
RAY_LOG(DEBUG) << "[ObjectManager] Message "
<< object_manager_protocol::EnumNameMessageType(message_type_value)
<< "(" << message_type << ") from object manager";
switch (message_type_value) {
case object_manager_protocol::MessageType::PushRequest: {
ReceivePushRequest(conn, message);
@@ -806,8 +826,8 @@ void ObjectManager::ReceivePushRequest(std::shared_ptr<TcpClientConnection> &con
ray::Status ObjectManager::ExecuteReceiveObject(
const ClientID &client_id, const ObjectID &object_id, uint64_t data_size,
uint64_t metadata_size, uint64_t chunk_index, TcpClientConnection &conn) {
RAY_LOG(DEBUG) << "ExecuteReceiveObject " << client_id << " " << object_id << " "
<< chunk_index;
RAY_LOG(DEBUG) << "ExecuteReceiveObject on " << client_id_ << " from " << client_id
<< " of object " << object_id << " chunk " << chunk_index;
std::pair<const ObjectBufferPool::ChunkInfo &, ray::Status> chunk_status =
buffer_pool_.CreateChunk(object_id, data_size, metadata_size, chunk_index);
@@ -825,8 +845,7 @@ ray::Status ObjectManager::ExecuteReceiveObject(
// TODO(hme): This chunk failed, so create a pull request for this chunk.
}
} else {
RAY_LOG(DEBUG) << "Create Chunk Failed index = " << chunk_index << ": "
<< chunk_status.second.message();
RAY_LOG(DEBUG) << "ExecuteReceiveObject failed: " << chunk_status.second.message();
// Read object into empty buffer.
uint64_t buffer_length = buffer_pool_.GetBufferLength(chunk_index, data_size);
std::vector<uint8_t> mutable_vec;
@@ -841,8 +860,9 @@ ray::Status ObjectManager::ExecuteReceiveObject(
// TODO(hme): If the object isn't local, create a pull request for this chunk.
}
conn.ProcessMessages();
RAY_LOG(DEBUG) << "ReceiveCompleted " << client_id_ << " " << object_id << " "
<< "/" << config_.max_receives;
RAY_LOG(DEBUG) << "ExecuteReceiveObject completed on " << client_id_ << " from "
<< client_id << " of object " << object_id << " chunk " << chunk_index
<< " at " << current_sys_time_ms();
return chunk_status.second;
}
src/ray/object_manager/test/object_manager_stress_test.cc
+1 -1
View File
@@ -78,7 +78,7 @@ class MockServer {
// Accept a new local client and dispatch it to the node manager.
auto new_connection = TcpClientConnection::Create(
client_handler, message_handler, std::move(object_manager_socket_),
"object manager",
"object manager", {},
static_cast<int64_t>(object_manager::protocol::MessageType::DisconnectClient));
DoAcceptObjectManager();
}
src/ray/object_manager/test/object_manager_test.cc
+1 -1
View File
@@ -69,7 +69,7 @@ class MockServer {
// Accept a new local client and dispatch it to the node manager.
auto new_connection = TcpClientConnection::Create(
client_handler, message_handler, std::move(object_manager_socket_),
"object manager",
"object manager", {},
static_cast<int64_t>(object_manager::protocol::MessageType::DisconnectClient));
DoAcceptObjectManager();
}
src/ray/raylet/client_connection_test.cc
+6 -6
View File
@@ -40,10 +40,10 @@ TEST_F(ClientConnectionTest, SimpleSyncWrite) {
};
auto conn1 = LocalClientConnection::Create(
client_handler, message_handler, std::move(in_), "conn1", error_message_type_);
client_handler, message_handler, std::move(in_), "conn1", {}, error_message_type_);
auto conn2 = LocalClientConnection::Create(
client_handler, message_handler, std::move(out_), "conn2", error_message_type_);
client_handler, message_handler, std::move(out_), "conn2", {}, error_message_type_);
RAY_CHECK_OK(conn1->WriteMessage(0, 5, arr));
RAY_CHECK_OK(conn2->WriteMessage(0, 5, arr));
@@ -86,10 +86,10 @@ TEST_F(ClientConnectionTest, SimpleAsyncWrite) {
};
auto writer = LocalClientConnection::Create(
client_handler, noop_handler, std::move(in_), "writer", error_message_type_);
client_handler, noop_handler, std::move(in_), "writer", {}, error_message_type_);
reader = LocalClientConnection::Create(client_handler, message_handler, std::move(out_),
"reader", error_message_type_);
"reader", {}, error_message_type_);
std::function<void(const ray::Status &)> callback = [](const ray::Status &status) {
RAY_CHECK_OK(status);
@@ -114,7 +114,7 @@ TEST_F(ClientConnectionTest, SimpleAsyncError) {
const uint8_t *message) {};
auto writer = LocalClientConnection::Create(
client_handler, noop_handler, std::move(in_), "writer", error_message_type_);
client_handler, noop_handler, std::move(in_), "writer", {}, error_message_type_);
std::function<void(const ray::Status &)> callback = [](const ray::Status &status) {
ASSERT_TRUE(!status.ok());
@@ -136,7 +136,7 @@ TEST_F(ClientConnectionTest, CallbackWithSharedRefDoesNotLeakConnection) {
const uint8_t *message) {};
auto writer = LocalClientConnection::Create(
client_handler, noop_handler, std::move(in_), "writer", error_message_type_);
client_handler, noop_handler, std::move(in_), "writer", {}, error_message_type_);
std::function<void(const ray::Status &)> callback =
[writer](const ray::Status &status) {
src/ray/raylet/lineage_cache.cc
+5 -5
View File
@@ -222,7 +222,7 @@ void LineageCache::AddUncommittedLineage(const TaskID &task_id,
bool LineageCache::AddWaitingTask(const Task &task, const Lineage &uncommitted_lineage) {
auto task_id = task.GetTaskSpecification().TaskId();
RAY_LOG(DEBUG) << "add waiting task " << task_id << " on " << client_id_;
RAY_LOG(DEBUG) << "Add waiting task " << task_id << " on " << client_id_;
// Merge the uncommitted lineage into the lineage cache. Collect the IDs of
// tasks that we should subscribe to. These are all of the tasks that were
@@ -253,7 +253,7 @@ bool LineageCache::AddWaitingTask(const Task &task, const Lineage &uncommitted_l
bool LineageCache::AddReadyTask(const Task &task) {
const TaskID task_id = task.GetTaskSpecification().TaskId();
RAY_LOG(DEBUG) << "add ready task " << task_id << " on " << client_id_;
RAY_LOG(DEBUG) << "Add ready task " << task_id << " on " << client_id_;
// Set the task to READY.
if (lineage_.SetEntry(task, GcsStatus::UNCOMMITTED_READY)) {
@@ -268,7 +268,7 @@ bool LineageCache::AddReadyTask(const Task &task) {
}
bool LineageCache::RemoveWaitingTask(const TaskID &task_id) {
RAY_LOG(DEBUG) << "remove waiting task " << task_id << " on " << client_id_;
RAY_LOG(DEBUG) << "Remove waiting task " << task_id << " on " << client_id_;
auto entry = lineage_.GetEntryMutable(task_id);
if (!entry) {
// The task was already evicted.
@@ -417,7 +417,7 @@ void LineageCache::EvictTask(const TaskID &task_id) {
}
// Evict the task.
RAY_LOG(DEBUG) << "evicting task " << task_id << " on " << client_id_;
RAY_LOG(DEBUG) << "Evicting task " << task_id << " on " << client_id_;
lineage_.PopEntry(task_id);
committed_tasks_.erase(commit_it);
// Try to evict the children of the evict task. These are the tasks that have
@@ -431,7 +431,7 @@ void LineageCache::EvictTask(const TaskID &task_id) {
}
void LineageCache::HandleEntryCommitted(const TaskID &task_id) {
RAY_LOG(DEBUG) << "task committed: " << task_id;
RAY_LOG(DEBUG) << "Task committed: " << task_id;
auto entry = lineage_.GetEntry(task_id);
if (!entry) {
// The task has already been evicted due to a previous commit notification.
src/ray/raylet/node_manager.cc
+57 -55
View File
@@ -166,10 +166,9 @@ ray::Status NodeManager::RegisterGcs() {
HeartbeatBatchAdded(heartbeat_batch);
};
RAY_RETURN_NOT_OK(gcs_client_->heartbeat_batch_table().Subscribe(
UniqueID::nil(), UniqueID::nil(), heartbeat_batch_added, nullptr,
[](gcs::AsyncGcsClient *client) {
RAY_LOG(DEBUG) << "Heartbeat batch table subscription done.";
}));
UniqueID::nil(), UniqueID::nil(), heartbeat_batch_added,
/*subscribe_callback=*/nullptr,
/*done_callback=*/nullptr));
// Subscribe to driver table updates.
const auto driver_table_handler = [this](
@@ -246,7 +245,6 @@ void NodeManager::Heartbeat() {
}
last_heartbeat_at_ms_ = now_ms;
RAY_LOG(DEBUG) << "[Heartbeat] sending heartbeat.";
auto &heartbeat_table = gcs_client_->heartbeat_table();
auto heartbeat_data = std::make_shared<HeartbeatTableDataT>();
const auto &my_client_id = gcs_client_->client_table().GetLocalClientId();
@@ -254,8 +252,6 @@ void NodeManager::Heartbeat() {
heartbeat_data->client_id = my_client_id.binary();
// TODO(atumanov): modify the heartbeat table protocol to use the ResourceSet directly.
// TODO(atumanov): implement a ResourceSet const_iterator.
RAY_LOG(DEBUG) << "[Heartbeat] resources available: "
<< local_resources.GetAvailableResources().ToString();
for (const auto &resource_pair :
local_resources.GetAvailableResources().GetResourceMap()) {
heartbeat_data->resources_available_label.push_back(resource_pair.first);
@@ -274,16 +270,8 @@ void NodeManager::Heartbeat() {
ray::Status status = heartbeat_table.Add(
UniqueID::nil(), gcs_client_->client_table().GetLocalClientId(), heartbeat_data,
[](ray::gcs::AsyncGcsClient *client, const ClientID &id,
const HeartbeatTableDataT &data) {
RAY_LOG(DEBUG) << "[HEARTBEAT] heartbeat sent callback";
});
if (!status.ok()) {
RAY_LOG(INFO) << "heartbeat failed: string " << status.ToString() << status.message();
RAY_LOG(INFO) << "is redis error: " << status.IsRedisError();
}
RAY_CHECK_OK(status);
/*success_callback=*/nullptr);
RAY_CHECK_OK_PREPEND(status, "Heartbeat failed");
if (debug_dump_period_ > 0 &&
static_cast<int64_t>(now_ms - last_debug_dump_at_ms_) > debug_dump_period_) {
@@ -330,7 +318,7 @@ void NodeManager::GetObjectManagerProfileInfo() {
void NodeManager::ClientAdded(const ClientTableDataT &client_data) {
const ClientID client_id = ClientID::from_binary(client_data.client_id);
RAY_LOG(DEBUG) << "[ClientAdded] received callback from client id " << client_id;
RAY_LOG(DEBUG) << "[ClientAdded] Received callback from client id " << client_id;
if (client_id == gcs_client_->client_table().GetLocalClientId()) {
// We got a notification for ourselves, so we are connected to the GCS now.
// Save this NodeManager's resource information in the cluster resource map.
@@ -341,11 +329,10 @@ void NodeManager::ClientAdded(const ClientTableDataT &client_data) {
// TODO(atumanov): make remote client lookup O(1)
if (std::find(remote_clients_.begin(), remote_clients_.end(), client_id) ==
remote_clients_.end()) {
RAY_LOG(DEBUG) << "a new client: " << client_id;
remote_clients_.push_back(client_id);
} else {
// NodeManager connection to this client was already established.
RAY_LOG(DEBUG) << "received a new client connection that already exists: "
RAY_LOG(DEBUG) << "Received a new client connection that already exists: "
<< client_id;
return;
}
@@ -383,7 +370,7 @@ void NodeManager::ClientRemoved(const ClientTableDataT &client_data) {
// TODO(swang): If we receive a notification for our own death, clean up and
// exit immediately.
const ClientID client_id = ClientID::from_binary(client_data.client_id);
RAY_LOG(DEBUG) << "[ClientRemoved] received callback from client id " << client_id;
RAY_LOG(DEBUG) << "[ClientRemoved] Received callback from client id " << client_id;
RAY_CHECK(client_id != gcs_client_->client_table().GetLocalClientId())
<< "Exiting because this node manager has mistakenly been marked dead by the "
@@ -429,7 +416,6 @@ void NodeManager::ClientRemoved(const ClientTableDataT &client_data) {
void NodeManager::HeartbeatAdded(const ClientID &client_id,
const HeartbeatTableDataT &heartbeat_data) {
RAY_LOG(DEBUG) << "[HeartbeatAdded]: received heartbeat from client id " << client_id;
// Locate the client id in remote client table and update available resources based on
// the received heartbeat information.
auto it = cluster_resource_map_.find(client_id);
@@ -446,7 +432,6 @@ void NodeManager::HeartbeatAdded(const ClientID &client_id,
ResourceSet remote_load(heartbeat_data.resource_load_label,
heartbeat_data.resource_load_capacity);
// TODO(atumanov): assert that the load is a non-empty ResourceSet.
RAY_LOG(DEBUG) << "[HeartbeatAdded]: received load: " << remote_load.ToString();
remote_resources.SetAvailableResources(std::move(remote_available));
// Extract the load information and save it locally.
remote_resources.SetLoadResources(std::move(remote_load));
@@ -537,7 +522,7 @@ void NodeManager::HandleActorStateTransition(const ActorID &actor_id,
// The actor's location is now known. Dequeue any methods that were
// submitted before the actor's location was known.
// (See design_docs/task_states.rst for the state transition diagram.)
const auto &methods = local_queues_.GetMethodsWaitingForActorCreation();
const auto &methods = local_queues_.GetTasks(TaskState::WAITING_FOR_ACTOR_CREATION);
std::unordered_set<TaskID> created_actor_method_ids;
for (const auto &method : methods) {
if (method.GetTaskSpecification().ActorId() == actor_id) {
@@ -615,9 +600,9 @@ void NodeManager::DispatchTasks(
const std::unordered_map<ResourceSet, ordered_set<TaskID>> &tasks_with_resources) {
std::unordered_set<TaskID> removed_task_ids;
for (const auto &it : tasks_with_resources) {
const auto &task_resources = it.first;
for (const auto &task_id : it.second) {
const auto &task = local_queues_.GetReadyQueue().GetTask(task_id);
const auto &task_resources = task.GetTaskSpecification().GetRequiredResources();
const auto &task = local_queues_.GetTaskOfState(task_id, TaskState::READY);
if (!local_available_resources_.Contains(task_resources)) {
// All the tasks in it.second have the same resource shape, so
// once the first task is not feasible, we can break out of this loop
@@ -636,8 +621,11 @@ void NodeManager::ProcessClientMessage(
const uint8_t *message_data) {
auto registered_worker = worker_pool_.GetRegisteredWorker(client);
auto message_type_value = static_cast<protocol::MessageType>(message_type);
RAY_LOG(DEBUG) << "Message of " << protocol::EnumNameMessageType(message_type_value)
<< "(" << message_type << ")";
RAY_LOG(DEBUG) << "[Worker] Message "
<< protocol::EnumNameMessageType(message_type_value) << "("
<< message_type << ") from worker with PID "
<< (registered_worker ? std::to_string(registered_worker->Pid())
: "nil");
if (registered_worker && registered_worker->IsDead()) {
// For a worker that is marked as dead (because the driver has died already),
// all the messages are ignored except DisconnectClient.
@@ -711,7 +699,7 @@ void NodeManager::ProcessRegisterClientRequestMessage(
if (message->is_worker()) {
// Register the new worker.
worker_pool_.RegisterWorker(std::move(worker));
DispatchTasks(local_queues_.GetReadyQueue().GetTasksWithResources());
DispatchTasks(local_queues_.GetReadyTasksWithResources());
} else {
// Register the new driver. Note that here the driver_id in RegisterClientRequest
// message is actually the ID of the driver task, while client_id represents the
@@ -784,7 +772,7 @@ void NodeManager::ProcessGetTaskMessage(
cluster_resource_map_[local_client_id].SetLoadResources(
local_queues_.GetResourceLoad());
// Call task dispatch to assign work to the new worker.
DispatchTasks(local_queues_.GetReadyQueue().GetTasksWithResources());
DispatchTasks(local_queues_.GetReadyTasksWithResources());
}
void NodeManager::ProcessDisconnectClientMessage(
@@ -882,7 +870,7 @@ void NodeManager::ProcessDisconnectClientMessage(
<< "driver_id: " << worker->GetAssignedDriverId();
// Since some resources may have been released, we can try to dispatch more tasks.
DispatchTasks(local_queues_.GetReadyQueue().GetTasksWithResources());
DispatchTasks(local_queues_.GetReadyTasksWithResources());
} else if (is_driver) {
// The client is a driver.
RAY_CHECK_OK(gcs_client_->driver_table().AppendDriverData(client->GetClientId(),
@@ -1014,14 +1002,19 @@ void NodeManager::ProcessNewNodeManager(TcpClientConnection &node_manager_client
void NodeManager::ProcessNodeManagerMessage(TcpClientConnection &node_manager_client,
int64_t message_type,
const uint8_t *message_data) {
switch (static_cast<protocol::MessageType>(message_type)) {
const auto message_type_value = static_cast<protocol::MessageType>(message_type);
RAY_LOG(DEBUG) << "[NodeManager] Message "
<< protocol::EnumNameMessageType(message_type_value) << "("
<< message_type << ") from node manager";
switch (message_type_value) {
case protocol::MessageType::ForwardTaskRequest: {
auto message = flatbuffers::GetRoot<protocol::ForwardTaskRequest>(message_data);
TaskID task_id = from_flatbuf(*message->task_id());
Lineage uncommitted_lineage(*message);
const Task &task = uncommitted_lineage.GetEntry(task_id)->TaskData();
RAY_LOG(DEBUG) << "got task " << task.GetTaskSpecification().TaskId()
RAY_LOG(DEBUG) << "Received forwarded task " << task.GetTaskSpecification().TaskId()
<< " on node " << gcs_client_->client_table().GetLocalClientId()
<< " spillback=" << task.GetTaskExecutionSpec().NumForwards();
SubmitTask(task, uncommitted_lineage, /* forwarded = */ true);
} break;
@@ -1089,7 +1082,7 @@ void NodeManager::ScheduleTasks(
// TODO(atumanov): evaluate performance implications of registering all new tasks on
// submission vs. registering remaining queued placeable tasks here.
std::unordered_set<TaskID> move_task_set;
for (const auto &task : local_queues_.GetPlaceableTasks()) {
for (const auto &task : local_queues_.GetTasks(TaskState::PLACEABLE)) {
task_dependency_manager_.TaskPending(task);
move_task_set.insert(task.GetTaskSpecification().TaskId());
// Push a warning to the task's driver that this task is currently infeasible.
@@ -1119,7 +1112,7 @@ void NodeManager::ScheduleTasks(
// infeasible task queue. Infeasible task queue is checked when new nodes join.
local_queues_.MoveTasks(move_task_set, TaskState::PLACEABLE, TaskState::INFEASIBLE);
// Check the invariant that no placeable tasks remain after a call to the policy.
RAY_CHECK(local_queues_.GetPlaceableTasks().size() == 0);
RAY_CHECK(local_queues_.GetTasks(TaskState::PLACEABLE).size() == 0);
}
bool NodeManager::CheckDependencyManagerInvariant() const {
@@ -1176,7 +1169,8 @@ void NodeManager::TreatTaskAsFailed(const Task &task) {
void NodeManager::TreatTaskAsFailedIfLost(const Task &task) {
const TaskSpecification &spec = task.GetTaskSpecification();
RAY_LOG(DEBUG) << "Treating task " << spec.TaskId() << " as failed.";
RAY_LOG(DEBUG) << "Treating task " << spec.TaskId()
<< " as failed if return values lost.";
// Loop over the return IDs (except the dummy ID) and check whether a
// location for the return ID exists.
int64_t num_returns = spec.NumReturns();
@@ -1221,7 +1215,8 @@ void NodeManager::SubmitTask(const Task &task, const Lineage &uncommitted_lineag
<< ", actor_creation_id=" << spec.ActorCreationId()
<< ", actor_handle_id=" << spec.ActorHandleId()
<< ", actor_counter=" << spec.ActorCounter()
<< ", task_descriptor=" << spec.FunctionDescriptorString();
<< ", task_descriptor=" << spec.FunctionDescriptorString() << " on node "
<< gcs_client_->client_table().GetLocalClientId();
if (local_queues_.HasTask(task_id)) {
RAY_LOG(WARNING) << "Submitted task " << task_id
@@ -1307,7 +1302,7 @@ void NodeManager::SubmitTask(const Task &task, const Lineage &uncommitted_lineag
// Keep the task queued until we discover the actor's location.
// (See design_docs/task_states.rst for the state transition diagram.)
local_queues_.QueueMethodsWaitingForActorCreation({task});
local_queues_.QueueTasks({task}, TaskState::WAITING_FOR_ACTOR_CREATION);
// The actor has not yet been created and may have failed. To make sure
// that the actor is eventually recreated, we maintain the invariant that
// if a task is in the MethodsWaitingForActorCreation queue, then it is
@@ -1330,7 +1325,7 @@ void NodeManager::SubmitTask(const Task &task, const Lineage &uncommitted_lineag
EnqueuePlaceableTask(task);
} else {
// (See design_docs/task_states.rst for the state transition diagram.)
local_queues_.QueuePlaceableTasks({task});
local_queues_.QueueTasks({task}, TaskState::PLACEABLE);
ScheduleTasks(cluster_resource_map_);
// TODO(atumanov): assert that !placeable.isempty() => insufficient available
// resources locally.
@@ -1349,7 +1344,7 @@ void NodeManager::HandleTaskBlocked(const std::shared_ptr<LocalClientConnection>
// worker holds while it is blocked.
if (!worker->IsBlocked() && current_task_id == worker->GetAssignedTaskId()) {
const auto task = local_queues_.RemoveTask(current_task_id);
local_queues_.QueueRunningTasks({task});
local_queues_.QueueTasks({task}, TaskState::RUNNING);
// Get the CPU resources required by the running task.
const auto required_resources = task.GetTaskSpecification().GetRequiredResources();
double required_cpus = required_resources.GetNumCpus();
@@ -1365,7 +1360,7 @@ void NodeManager::HandleTaskBlocked(const std::shared_ptr<LocalClientConnection>
worker->MarkBlocked();
// Try dispatching tasks since we may have released some resources.
DispatchTasks(local_queues_.GetReadyQueue().GetTasksWithResources());
DispatchTasks(local_queues_.GetReadyTasksWithResources());
}
} else {
// The client is a driver. Drivers do not hold resources, so we simply mark
@@ -1401,7 +1396,7 @@ void NodeManager::HandleTaskUnblocked(
if (worker->IsBlocked() && current_task_id == worker->GetAssignedTaskId()) {
// (See design_docs/task_states.rst for the state transition diagram.)
const auto task = local_queues_.RemoveTask(current_task_id);
local_queues_.QueueRunningTasks({task});
local_queues_.QueueTasks({task}, TaskState::RUNNING);
// Get the CPU resources required by the running task.
const auto required_resources = task.GetTaskSpecification().GetRequiredResources();
double required_cpus = required_resources.GetNumCpus();
@@ -1458,10 +1453,10 @@ void NodeManager::EnqueuePlaceableTask(const Task &task) {
// in the READY state, else the WAITING state.
// (See design_docs/task_states.rst for the state transition diagram.)
if (args_ready) {
local_queues_.QueueReadyTasks({task});
local_queues_.QueueTasks({task}, TaskState::READY);
DispatchTasks(MakeTasksWithResources({task}));
} else {
local_queues_.QueueWaitingTasks({task});
local_queues_.QueueTasks({task}, TaskState::WAITING);
}
// Mark the task as pending. Once the task has finished execution, or once it
// has been forwarded to another node, the task must be marked as canceled in
@@ -1592,7 +1587,7 @@ bool NodeManager::AssignTask(const Task &task) {
}
// Mark the task as running.
// (See design_docs/task_states.rst for the state transition diagram.)
local_queues_.QueueRunningTasks(std::vector<Task>({assigned_task}));
local_queues_.QueueTasks({assigned_task}, TaskState::RUNNING);
// Notify the task dependency manager that we no longer need this task's
// object dependencies.
RAY_CHECK(task_dependency_manager_.UnsubscribeDependencies(spec.TaskId()));
@@ -1604,7 +1599,7 @@ bool NodeManager::AssignTask(const Task &task) {
// DispatchTasks() removed it from the ready queue. The task will be
// assigned to a worker once one becomes available.
// (See design_docs/task_states.rst for the state transition diagram.)
local_queues_.QueueReadyTasks({assigned_task});
local_queues_.QueueTasks({assigned_task}, TaskState::READY);
DispatchTasks(MakeTasksWithResources({assigned_task}));
}
});
@@ -1804,6 +1799,9 @@ void NodeManager::ResubmitTask(const Task &task) {
void NodeManager::HandleObjectLocal(const ObjectID &object_id) {
// Notify the task dependency manager that this object is local.
const auto ready_task_ids = task_dependency_manager_.HandleObjectLocal(object_id);
RAY_LOG(DEBUG) << "Object local " << object_id << ", "
<< " on " << gcs_client_->client_table().GetLocalClientId()
<< ready_task_ids.size() << " tasks ready";
// Transition the tasks whose dependencies are now fulfilled to the ready state.
if (ready_task_ids.size() > 0) {
std::unordered_set<TaskID> ready_task_id_set(ready_task_ids.begin(),
@@ -1821,7 +1819,7 @@ void NodeManager::HandleObjectLocal(const ObjectID &object_id) {
// Queue and dispatch the tasks that are ready to run (i.e., WAITING).
auto ready_tasks = local_queues_.RemoveTasks(ready_task_id_set);
local_queues_.QueueReadyTasks(ready_tasks);
local_queues_.QueueTasks(ready_tasks, TaskState::READY);
DispatchTasks(MakeTasksWithResources(ready_tasks));
}
}
@@ -1829,19 +1827,21 @@ void NodeManager::HandleObjectLocal(const ObjectID &object_id) {
void NodeManager::HandleObjectMissing(const ObjectID &object_id) {
// Notify the task dependency manager that this object is no longer local.
const auto waiting_task_ids = task_dependency_manager_.HandleObjectMissing(object_id);
RAY_LOG(DEBUG) << "Object missing " << object_id << ", "
<< " on " << gcs_client_->client_table().GetLocalClientId()
<< waiting_task_ids.size() << " tasks waiting";
// Transition any tasks that were in the runnable state and are dependent on
// this object to the waiting state.
if (!waiting_task_ids.empty()) {
// Transition the tasks back to the waiting state. They will be made
// runnable once the deleted object becomes available again.
std::unordered_set<TaskID> waiting_task_id_set(waiting_task_ids.begin(),
waiting_task_ids.end());
local_queues_.MoveTasks(waiting_task_id_set, TaskState::READY, TaskState::WAITING);
// Check that remaining tasks that could not be transitioned are running
// workers or drivers, now blocked in a get.
// First filter out any tasks that can't be transitioned to READY. These
// are running workers or drivers, now blocked in a get.
local_queues_.FilterState(waiting_task_id_set, TaskState::RUNNING);
local_queues_.FilterState(waiting_task_id_set, TaskState::DRIVER);
// Transition the tasks back to the waiting state. They will be made
// runnable once the deleted object becomes available again.
local_queues_.MoveTasks(waiting_task_id_set, TaskState::READY, TaskState::WAITING);
RAY_CHECK(waiting_task_id_set.empty());
// Moving ready tasks to waiting may have changed the load, making space for placing
// new tasks locally.
@@ -1891,7 +1891,7 @@ void NodeManager::ForwardTaskOrResubmit(const Task &task,
} else {
// The task is not for an actor and may therefore be placed on another
// node immediately. Send it to the scheduling policy to be placed again.
local_queues_.QueuePlaceableTasks({task});
local_queues_.QueueTasks({task}, TaskState::PLACEABLE);
ScheduleTasks(cluster_resource_map_);
}
});
@@ -1914,7 +1914,9 @@ void NodeManager::ForwardTask(const Task &task, const ClientID &node_id,
auto request = uncommitted_lineage.ToFlatbuffer(fbb, task_id);
fbb.Finish(request);
RAY_LOG(DEBUG) << "Forwarding task " << task_id << " to " << node_id << " spillback="
RAY_LOG(DEBUG) << "Forwarding task " << task_id << " from "
<< gcs_client_->client_table().GetLocalClientId() << " to " << node_id
<< " spillback="
<< lineage_cache_entry_task.GetTaskExecutionSpec().NumForwards();
// Lookup remote server connection for this node_id and use it to send the request.
src/ray/raylet/raylet.cc
+28 -3
View File
@@ -7,6 +7,28 @@
#include "ray/status.h"
namespace {
const std::vector<std::string> GenerateEnumNames(const char *const *enum_names_ptr) {
std::vector<std::string> enum_names;
size_t i = 0;
while (true) {
const char *name = enum_names_ptr[i];
if (name == nullptr) {
break;
}
enum_names.push_back(name);
i++;
}
return enum_names;
}
static const std::vector<std::string> node_manager_message_enum =
GenerateEnumNames(ray::protocol::EnumNamesMessageType());
static const std::vector<std::string> object_manager_message_enum =
GenerateEnumNames(ray::object_manager::protocol::EnumNamesMessageType());
}
namespace ray {
namespace raylet {
@@ -75,8 +97,9 @@ ray::Status Raylet::RegisterGcs(const std::string &node_ip_address,
client_info.resources_total_capacity.push_back(resource_pair.second);
}
RAY_LOG(DEBUG) << "Node manager listening on: IP " << client_info.node_manager_address
<< " port " << client_info.node_manager_port;
RAY_LOG(DEBUG) << "Node manager " << gcs_client_->client_table().GetLocalClientId()
<< " started on " << client_info.node_manager_address << ":"
<< client_info.node_manager_port;
RAY_RETURN_NOT_OK(gcs_client_->client_table().Connect(client_info));
RAY_RETURN_NOT_OK(node_manager_.RegisterGcs());
@@ -102,6 +125,7 @@ void Raylet::HandleAcceptNodeManager(const boost::system::error_code &error) {
// Accept a new TCP client and dispatch it to the node manager.
auto new_connection = TcpClientConnection::Create(
client_handler, message_handler, std::move(node_manager_socket_), "node manager",
node_manager_message_enum,
static_cast<int64_t>(protocol::MessageType::DisconnectClient));
}
// We're ready to accept another client.
@@ -125,7 +149,7 @@ void Raylet::HandleAcceptObjectManager(const boost::system::error_code &error) {
// Accept a new TCP client and dispatch it to the node manager.
auto new_connection = TcpClientConnection::Create(
client_handler, message_handler, std::move(object_manager_socket_),
"object manager",
"object manager", object_manager_message_enum,
static_cast<int64_t>(object_manager::protocol::MessageType::DisconnectClient));
DoAcceptObjectManager();
}
@@ -148,6 +172,7 @@ void Raylet::HandleAccept(const boost::system::error_code &error) {
// Accept a new local client and dispatch it to the node manager.
auto new_connection = LocalClientConnection::Create(
client_handler, message_handler, std::move(socket_), "worker",
node_manager_message_enum,
static_cast<int64_t>(protocol::MessageType::DisconnectClient));
}
// We're ready to accept another client.
src/ray/raylet/scheduling_policy.cc
+5 -7
View File
@@ -19,10 +19,8 @@ std::unordered_map<TaskID, ClientID> SchedulingPolicy::Schedule(
const ClientID &local_client_id) {
// The policy decision to be returned.
std::unordered_map<TaskID, ClientID> decision;
// TODO(atumanov): protect DEBUG code blocks with ifdef DEBUG
RAY_LOG(DEBUG) << "[Schedule] cluster resource map: ";
#ifndef NDEBUG
RAY_LOG(DEBUG) << "Cluster resource map: ";
for (const auto &client_resource_pair : cluster_resources) {
// pair = ClientID, SchedulingResources
const ClientID &client_id = client_resource_pair.first;
@@ -33,9 +31,9 @@ std::unordered_map<TaskID, ClientID> SchedulingPolicy::Schedule(
#endif
// We expect all placeable tasks to be placed on exit from this policy method.
RAY_CHECK(scheduling_queue_.GetPlaceableTasks().size() <= 1);
RAY_CHECK(scheduling_queue_.GetTasks(TaskState::PLACEABLE).size() <= 1);
// Iterate over running tasks, get their resource demand and try to schedule.
for (const auto &t : scheduling_queue_.GetPlaceableTasks()) {
for (const auto &t : scheduling_queue_.GetTasks(TaskState::PLACEABLE)) {
// Get task's resource demand
const auto &spec = t.GetTaskSpecification();
const auto &resource_demand = spec.GetRequiredPlacementResources();
@@ -126,7 +124,7 @@ std::vector<TaskID> SchedulingPolicy::SpillOver(
ResourceSet new_load(remote_scheduling_resources.GetLoadResources());
// Check if we can accommodate infeasible tasks.
for (const auto &task : scheduling_queue_.GetInfeasibleTasks()) {
for (const auto &task : scheduling_queue_.GetTasks(TaskState::INFEASIBLE)) {
const auto &spec = task.GetTaskSpecification();
const auto &placement_resources = spec.GetRequiredPlacementResources();
if (placement_resources.IsSubset(remote_scheduling_resources.GetTotalResources())) {
@@ -136,7 +134,7 @@ std::vector<TaskID> SchedulingPolicy::SpillOver(
}
// Try to accommodate up to a single ready task.
for (const auto &task : scheduling_queue_.GetReadyTasks()) {
for (const auto &task : scheduling_queue_.GetTasks(TaskState::READY)) {
const auto &spec = task.GetTaskSpecification();
if (!spec.IsActorTask()) {
// Make sure the node has enough available resources to prevent forwarding cycles.
src/ray/raylet/scheduling_queue.cc
+146 -151
View File
@@ -6,44 +6,15 @@
namespace {
// Helper function to remove tasks in the given set of task_ids from a
// queue, and append them to the given vector removed_tasks.
template <typename TaskQueue>
void RemoveTasksFromQueue(ray::raylet::TaskState task_state, TaskQueue &queue,
std::unordered_set<ray::TaskID> &task_ids,
std::vector<ray::raylet::Task> *removed_tasks,
std::vector<ray::raylet::TaskState> *task_states = nullptr) {
for (auto it = task_ids.begin(); it != task_ids.end();) {
if (queue.RemoveTask(*it, removed_tasks)) {
it = task_ids.erase(it);
if (task_states != nullptr) {
task_states->push_back(task_state);
}
} else {
it++;
}
}
}
static constexpr const char *task_state_strings[] = {
"placeable", "waiting", "ready",
"running", "infeasible", "waiting for actor creation"};
static_assert(sizeof(task_state_strings) / sizeof(const char *) ==
static_cast<int>(ray::raylet::TaskState::kNumTaskQueues),
"Must specify a TaskState name for every task queue");
// Helper function to queue the given tasks to the given queue.
template <typename TaskQueue>
inline void QueueTasks(TaskQueue &queue, const std::vector<ray::raylet::Task> &tasks) {
for (const auto &task : tasks) {
queue.AppendTask(task.GetTaskSpecification().TaskId(), task);
}
}
// Helper function to filter out tasks of a given state.
template <typename TaskQueue>
inline void FilterStateFromQueue(const TaskQueue &queue,
std::unordered_set<ray::TaskID> &task_ids) {
for (auto it = task_ids.begin(); it != task_ids.end();) {
if (queue.HasTask(*it)) {
it = task_ids.erase(it);
} else {
it++;
}
}
inline const char *GetTaskStateString(ray::raylet::TaskState task_state) {
return task_state_strings[static_cast<int>(task_state)];
}
// Helper function to get tasks for a driver from a given state.
@@ -112,6 +83,12 @@ bool TaskQueue::HasTask(const TaskID &task_id) const {
const std::list<Task> &TaskQueue::GetTasks() const { return task_list_; }
const Task &TaskQueue::GetTask(const TaskID &task_id) const {
auto it = task_map_.find(task_id);
RAY_CHECK(it != task_map_.end());
return *it->second;
}
const ResourceSet &TaskQueue::GetCurrentResourceLoad() const {
return current_resource_load_;
}
@@ -131,60 +108,68 @@ bool ReadyQueue::RemoveTask(const TaskID &task_id, std::vector<Task> *removed_ta
return TaskQueue::RemoveTask(task_id, removed_tasks);
}
const std::list<Task> &SchedulingQueue::GetMethodsWaitingForActorCreation() const {
return methods_waiting_for_actor_creation_.GetTasks();
const std::unordered_map<ResourceSet, ordered_set<TaskID>>
&ReadyQueue::GetTasksWithResources() const {
return tasks_with_resources_;
}
const std::list<Task> &SchedulingQueue::GetWaitingTasks() const {
return waiting_tasks_.GetTasks();
const std::list<Task> &SchedulingQueue::GetTasks(TaskState task_state) const {
const auto &queue = GetTaskQueue(task_state);
return queue->GetTasks();
}
const std::list<Task> &SchedulingQueue::GetPlaceableTasks() const {
return placeable_tasks_.GetTasks();
const std::unordered_map<ResourceSet, ordered_set<TaskID>>
&SchedulingQueue::GetReadyTasksWithResources() const {
return ready_queue_->GetTasksWithResources();
}
const std::list<Task> &SchedulingQueue::GetReadyTasks() const {
return ready_tasks_.GetTasks();
}
const std::list<Task> &SchedulingQueue::GetInfeasibleTasks() const {
return infeasible_tasks_.GetTasks();
}
ResourceSet SchedulingQueue::GetReadyQueueResources() const {
return ready_tasks_.GetCurrentResourceLoad();
const Task &SchedulingQueue::GetTaskOfState(const TaskID &task_id,
TaskState task_state) const {
const auto &queue = GetTaskQueue(task_state);
return queue->GetTask(task_id);
}
ResourceSet SchedulingQueue::GetResourceLoad() const {
// TODO(atumanov): consider other types of tasks as part of load.
return ready_tasks_.GetCurrentResourceLoad();
}
const std::list<Task> &SchedulingQueue::GetRunningTasks() const {
return running_tasks_.GetTasks();
return ready_queue_->GetCurrentResourceLoad();
}
const std::unordered_set<TaskID> &SchedulingQueue::GetBlockedTaskIds() const {
return blocked_task_ids_;
}
void SchedulingQueue::FilterStateFromQueue(std::unordered_set<ray::TaskID> &task_ids,
TaskState task_state) const {
auto &queue = GetTaskQueue(task_state);
for (auto it = task_ids.begin(); it != task_ids.end();) {
if (queue->HasTask(*it)) {
it = task_ids.erase(it);
} else {
it++;
}
}
}
void SchedulingQueue::FilterState(std::unordered_set<TaskID> &task_ids,
TaskState filter_state) const {
switch (filter_state) {
case TaskState::PLACEABLE:
FilterStateFromQueue(placeable_tasks_, task_ids);
FilterStateFromQueue(task_ids, TaskState::PLACEABLE);
break;
case TaskState::WAITING_FOR_ACTOR_CREATION:
FilterStateFromQueue(methods_waiting_for_actor_creation_, task_ids);
FilterStateFromQueue(task_ids, TaskState::WAITING_FOR_ACTOR_CREATION);
break;
case TaskState::WAITING:
FilterStateFromQueue(waiting_tasks_, task_ids);
FilterStateFromQueue(task_ids, TaskState::WAITING);
break;
case TaskState::READY:
FilterStateFromQueue(ready_tasks_, task_ids);
FilterStateFromQueue(task_ids, TaskState::READY);
break;
case TaskState::RUNNING:
FilterStateFromQueue(running_tasks_, task_ids);
FilterStateFromQueue(task_ids, TaskState::RUNNING);
break;
case TaskState::INFEASIBLE:
FilterStateFromQueue(task_ids, TaskState::INFEASIBLE);
break;
case TaskState::BLOCKED: {
const auto blocked_ids = GetBlockedTaskIds();
@@ -196,9 +181,6 @@ void SchedulingQueue::FilterState(std::unordered_set<TaskID> &task_ids,
}
}
} break;
case TaskState::INFEASIBLE:
FilterStateFromQueue(infeasible_tasks_, task_ids);
break;
case TaskState::DRIVER: {
const auto driver_ids = GetDriverTaskIds();
for (auto it = task_ids.begin(); it != task_ids.end();) {
@@ -215,91 +197,119 @@ void SchedulingQueue::FilterState(std::unordered_set<TaskID> &task_ids,
}
}
std::vector<Task> SchedulingQueue::RemoveTasks(std::unordered_set<TaskID> &task_ids,
std::vector<TaskState> *task_states) {
const std::shared_ptr<TaskQueue> &SchedulingQueue::GetTaskQueue(
TaskState task_state) const {
RAY_CHECK(task_state < TaskState::kNumTaskQueues)
<< static_cast<int>(task_state) << "Task state " << static_cast<int>(task_state)
<< " does not correspond to a task queue";
return task_queues_[static_cast<int>(task_state)];
}
// Helper function to remove tasks in the given set of task_ids from a
// queue, and append them to the given vector removed_tasks.
void SchedulingQueue::RemoveTasksFromQueue(
ray::raylet::TaskState task_state, std::unordered_set<ray::TaskID> &task_ids,
std::vector<ray::raylet::Task> *removed_tasks) {
auto &queue = GetTaskQueue(task_state);
for (auto it = task_ids.begin(); it != task_ids.end();) {
const auto &task_id = *it;
if (queue->RemoveTask(task_id, removed_tasks)) {
RAY_LOG(DEBUG) << "Removed task " << task_id << " from "
<< GetTaskStateString(task_state) << " queue";
it = task_ids.erase(it);
} else {
it++;
}
}
}
std::vector<Task> SchedulingQueue::RemoveTasks(std::unordered_set<TaskID> &task_ids) {
// List of removed tasks to be returned.
std::vector<Task> removed_tasks;
// Try to find the tasks to remove from the queues.
RemoveTasksFromQueue(TaskState::WAITING_FOR_ACTOR, methods_waiting_for_actor_creation_,
task_ids, &removed_tasks, task_states);
RemoveTasksFromQueue(TaskState::WAITING, waiting_tasks_, task_ids, &removed_tasks,
task_states);
RemoveTasksFromQueue(TaskState::PLACEABLE, placeable_tasks_, task_ids, &removed_tasks,
task_states);
RemoveTasksFromQueue(TaskState::READY, ready_tasks_, task_ids, &removed_tasks,
task_states);
RemoveTasksFromQueue(TaskState::RUNNING, running_tasks_, task_ids, &removed_tasks,
task_states);
RemoveTasksFromQueue(TaskState::INFEASIBLE, infeasible_tasks_, task_ids, &removed_tasks,
task_states);
for (const auto &task_state : {
TaskState::PLACEABLE, TaskState::WAITING, TaskState::READY, TaskState::RUNNING,
TaskState::INFEASIBLE, TaskState::WAITING_FOR_ACTOR_CREATION,
}) {
RemoveTasksFromQueue(task_state, task_ids, &removed_tasks);
}
RAY_CHECK(task_ids.size() == 0);
if (task_states != nullptr) {
RAY_CHECK(removed_tasks.size() == task_states->size());
}
return removed_tasks;
}
Task SchedulingQueue::RemoveTask(const TaskID &task_id, TaskState *task_state) {
Task SchedulingQueue::RemoveTask(const TaskID &task_id, TaskState *removed_task_state) {
std::vector<Task> removed_tasks;
std::unordered_set<TaskID> task_id_set = {task_id};
std::vector<TaskState> task_state_vector;
auto const task = RemoveTasks(task_id_set, &task_state_vector).front();
RAY_CHECK(task_state_vector.size() == 1);
if (task_state != nullptr) {
*task_state = task_state_vector[0];
// Try to find the task to remove in the queues.
for (const auto &task_state : {
TaskState::PLACEABLE, TaskState::WAITING, TaskState::READY, TaskState::RUNNING,
TaskState::INFEASIBLE, TaskState::WAITING_FOR_ACTOR_CREATION,
}) {
RemoveTasksFromQueue(task_state, task_id_set, &removed_tasks);
if (task_id_set.empty()) {
// The task was removed from the current queue.
if (removed_task_state != nullptr) {
// If the state of the removed task was requested, then set it with the
// current queue's state.
*removed_task_state = task_state;
}
break;
}
}
// Make sure we got the removed task.
RAY_CHECK(removed_tasks.size() == 1);
const auto &task = removed_tasks.front();
RAY_CHECK(task.GetTaskSpecification().TaskId() == task_id);
return task;
}
void SchedulingQueue::MoveTasks(std::unordered_set<TaskID> &task_ids, TaskState src_state,
TaskState dst_state) {
// TODO(atumanov): check the states first to ensure the move is transactional.
std::vector<Task> removed_tasks;
// Remove the tasks from the specified source queue.
switch (src_state) {
case TaskState::PLACEABLE:
RemoveTasksFromQueue(TaskState::PLACEABLE, placeable_tasks_, task_ids,
&removed_tasks);
RemoveTasksFromQueue(TaskState::PLACEABLE, task_ids, &removed_tasks);
break;
case TaskState::WAITING:
RemoveTasksFromQueue(TaskState::WAITING, waiting_tasks_, task_ids, &removed_tasks);
RemoveTasksFromQueue(TaskState::WAITING, task_ids, &removed_tasks);
break;
case TaskState::READY:
RemoveTasksFromQueue(TaskState::READY, ready_tasks_, task_ids, &removed_tasks);
RemoveTasksFromQueue(TaskState::READY, task_ids, &removed_tasks);
break;
case TaskState::RUNNING:
RemoveTasksFromQueue(TaskState::RUNNING, running_tasks_, task_ids, &removed_tasks);
RemoveTasksFromQueue(TaskState::RUNNING, task_ids, &removed_tasks);
break;
case TaskState::INFEASIBLE:
RemoveTasksFromQueue(TaskState::INFEASIBLE, infeasible_tasks_, task_ids,
&removed_tasks);
RemoveTasksFromQueue(TaskState::INFEASIBLE, task_ids, &removed_tasks);
break;
default:
RAY_LOG(FATAL) << "Attempting to move tasks from unrecognized state "
<< static_cast<std::underlying_type<TaskState>::type>(src_state);
}
// Make sure that all tasks were able to be moved.
RAY_CHECK(task_ids.empty());
// Add the tasks to the specified destination queue.
switch (dst_state) {
case TaskState::PLACEABLE:
QueueTasks(placeable_tasks_, removed_tasks);
QueueTasks(removed_tasks, TaskState::PLACEABLE);
break;
case TaskState::WAITING:
QueueTasks(waiting_tasks_, removed_tasks);
QueueTasks(removed_tasks, TaskState::WAITING);
break;
case TaskState::READY:
QueueTasks(ready_tasks_, removed_tasks);
QueueTasks(removed_tasks, TaskState::READY);
break;
case TaskState::RUNNING:
QueueTasks(running_tasks_, removed_tasks);
QueueTasks(removed_tasks, TaskState::RUNNING);
break;
case TaskState::INFEASIBLE:
QueueTasks(infeasible_tasks_, removed_tasks);
QueueTasks(removed_tasks, TaskState::INFEASIBLE);
break;
default:
RAY_LOG(FATAL) << "Attempting to move tasks to unrecognized state "
@@ -307,78 +317,62 @@ void SchedulingQueue::MoveTasks(std::unordered_set<TaskID> &task_ids, TaskState
}
}
void SchedulingQueue::QueueMethodsWaitingForActorCreation(
const std::vector<Task> &tasks) {
QueueTasks(methods_waiting_for_actor_creation_, tasks);
void SchedulingQueue::QueueTasks(const std::vector<Task> &tasks, TaskState task_state) {
auto &queue = GetTaskQueue(task_state);
for (const auto &task : tasks) {
RAY_LOG(DEBUG) << "Added task " << task.GetTaskSpecification().TaskId() << " to "
<< GetTaskStateString(task_state) << " queue";
queue->AppendTask(task.GetTaskSpecification().TaskId(), task);
}
}
bool SchedulingQueue::HasTask(const TaskID &task_id) const {
return (methods_waiting_for_actor_creation_.HasTask(task_id) ||
waiting_tasks_.HasTask(task_id) || placeable_tasks_.HasTask(task_id) ||
ready_tasks_.HasTask(task_id) || running_tasks_.HasTask(task_id) ||
infeasible_tasks_.HasTask(task_id));
}
void SchedulingQueue::QueueWaitingTasks(const std::vector<Task> &tasks) {
QueueTasks(waiting_tasks_, tasks);
}
void SchedulingQueue::QueuePlaceableTasks(const std::vector<Task> &tasks) {
QueueTasks(placeable_tasks_, tasks);
}
void SchedulingQueue::QueueReadyTasks(const std::vector<Task> &tasks) {
QueueTasks(ready_tasks_, tasks);
}
void SchedulingQueue::QueueRunningTasks(const std::vector<Task> &tasks) {
QueueTasks(running_tasks_, tasks);
for (const auto &task_queue : task_queues_) {
if (task_queue->HasTask(task_id)) {
return true;
}
}
return false;
}
std::unordered_set<TaskID> SchedulingQueue::GetTaskIdsForDriver(
const DriverID &driver_id) const {
std::unordered_set<TaskID> task_ids;
GetDriverTasksFromQueue(methods_waiting_for_actor_creation_, driver_id, task_ids);
GetDriverTasksFromQueue(waiting_tasks_, driver_id, task_ids);
GetDriverTasksFromQueue(placeable_tasks_, driver_id, task_ids);
GetDriverTasksFromQueue(ready_tasks_, driver_id, task_ids);
GetDriverTasksFromQueue(running_tasks_, driver_id, task_ids);
GetDriverTasksFromQueue(infeasible_tasks_, driver_id, task_ids);
for (const auto &task_queue : task_queues_) {
GetDriverTasksFromQueue(*task_queue, driver_id, task_ids);
}
return task_ids;
}
std::unordered_set<TaskID> SchedulingQueue::GetTaskIdsForActor(
const ActorID &actor_id) const {
std::unordered_set<TaskID> task_ids;
GetActorTasksFromQueue(methods_waiting_for_actor_creation_, actor_id, task_ids);
GetActorTasksFromQueue(waiting_tasks_, actor_id, task_ids);
GetActorTasksFromQueue(placeable_tasks_, actor_id, task_ids);
GetActorTasksFromQueue(ready_tasks_, actor_id, task_ids);
GetActorTasksFromQueue(running_tasks_, actor_id, task_ids);
GetActorTasksFromQueue(infeasible_tasks_, actor_id, task_ids);
for (const auto &task_queue : task_queues_) {
GetActorTasksFromQueue(*task_queue, actor_id, task_ids);
}
return task_ids;
}
void SchedulingQueue::AddBlockedTaskId(const TaskID &task_id) {
RAY_LOG(DEBUG) << "Added blocked task " << task_id;
auto inserted = blocked_task_ids_.insert(task_id);
RAY_CHECK(inserted.second);
}
void SchedulingQueue::RemoveBlockedTaskId(const TaskID &task_id) {
RAY_LOG(DEBUG) << "Removed blocked task " << task_id;
auto erased = blocked_task_ids_.erase(task_id);
RAY_CHECK(erased == 1);
}
void SchedulingQueue::AddDriverTaskId(const TaskID &driver_id) {
RAY_LOG(DEBUG) << "Added driver task " << driver_id;
auto inserted = driver_task_ids_.insert(driver_id);
RAY_CHECK(inserted.second);
}
void SchedulingQueue::RemoveDriverTaskId(const TaskID &driver_id) {
RAY_LOG(DEBUG) << "Removed driver task " << driver_id;
auto erased = driver_task_ids_.erase(driver_id);
RAY_CHECK(erased == 1);
}
@@ -390,13 +384,14 @@ const std::unordered_set<TaskID> &SchedulingQueue::GetDriverTaskIds() const {
std::string SchedulingQueue::DebugString() const {
std::stringstream result;
result << "SchedulingQueue:";
result << "\n- num placeable tasks: " << placeable_tasks_.GetTasks().size();
result << "\n- num waiting tasks: " << waiting_tasks_.GetTasks().size();
result << "\n- num ready tasks: " << ready_tasks_.GetTasks().size();
result << "\n- num running tasks: " << running_tasks_.GetTasks().size();
result << "\n- num infeasible tasks: " << infeasible_tasks_.GetTasks().size();
result << "\n- num methods waiting for actor creation: "
<< methods_waiting_for_actor_creation_.GetTasks().size();
for (const auto &task_state : {
TaskState::PLACEABLE, TaskState::WAITING, TaskState::READY, TaskState::RUNNING,
TaskState::INFEASIBLE, TaskState::WAITING_FOR_ACTOR_CREATION,
}) {
result << "\n- num " << GetTaskStateString(task_state)
<< " tasks: " << GetTaskQueue(task_state)->GetTasks().size();
}
result << "\n- num tasks blocked: " << blocked_task_ids_.size();
return result.str();
}
src/ray/raylet/scheduling_queue.h
+84 -105
View File
@@ -1,6 +1,7 @@
#ifndef RAY_RAYLET_SCHEDULING_QUEUE_H
#define RAY_RAYLET_SCHEDULING_QUEUE_H
#include <array>
#include <list>
#include <unordered_map>
#include <unordered_set>
@@ -15,11 +16,8 @@ namespace ray {
namespace raylet {
enum class TaskState {
INIT,
// The task may be placed on a node.
PLACEABLE,
// The task is for an actor whose location we do not know yet.
WAITING_FOR_ACTOR_CREATION,
// The task has been placed on a node and is waiting for some object
// dependencies to become local.
WAITING,
@@ -29,6 +27,17 @@ enum class TaskState {
// The task is running on a worker. The task may also be blocked in a ray.get
// or ray.wait call, in which case it also has state BLOCKED.
RUNNING,
// The task has resources that cannot be satisfied by any node, as far as we
// know.
INFEASIBLE,
// The task is an actor method and is waiting to learn where the actor was
// created.
WAITING_FOR_ACTOR_CREATION,
// The number of task queues. All states that precede this enum must have an
// associated TaskQueue in SchedulingQueue. All states that succeed
// this enum do not have an associated TaskQueue, since the tasks
// in those states may not have any associated task data.
kNumTaskQueues,
// The task is running but blocked in a ray.get or ray.wait call. Tasks that
// were explicitly assigned by us may be both BLOCKED and RUNNING, while
// tasks that were created out-of-band (e.g., the application created
@@ -36,12 +45,6 @@ enum class TaskState {
BLOCKED,
// The task is a driver task.
DRIVER,
// The task has resources that cannot be satisfied by any node, as far as we
// know.
INFEASIBLE,
// The task is an actor method and is waiting to learn where the actor was
// created.
WAITING_FOR_ACTOR,
};
class TaskQueue {
@@ -51,7 +54,7 @@ class TaskQueue {
/// \param task_id The task ID for the task to append.
/// \param task The task to append to the queue.
/// \return Whether the append operation succeeds.
bool AppendTask(const TaskID &task_id, const Task &task);
virtual bool AppendTask(const TaskID &task_id, const Task &task);
/// \brief Remove a task from queue.
///
@@ -60,7 +63,8 @@ class TaskQueue {
/// removed from the queue, the task data is appended to the vector. Can
/// be a nullptr, in which case nothing is appended.
/// \return Whether the removal succeeds.
bool RemoveTask(const TaskID &task_id, std::vector<Task> *removed_tasks = nullptr);
virtual bool RemoveTask(const TaskID &task_id,
std::vector<Task> *removed_tasks = nullptr);
/// \brief Check if the queue contains a specific task id.
///
@@ -69,10 +73,18 @@ class TaskQueue {
bool HasTask(const TaskID &task_id) const;
/// \brief Return the task list of the queue.
///
/// \return A list of tasks contained in this queue.
const std::list<Task> &GetTasks() const;
/// Get a task from the queue. The caller must ensure that the task is in
/// the queue.
///
/// \return The task.
const Task &GetTask(const TaskID &task_id) const;
/// \brief Get the total resources required by the tasks in the queue.
///
/// \return Total resources required by the tasks in the queue.
const ResourceSet &GetCurrentResourceLoad() const;
@@ -96,31 +108,19 @@ class ReadyQueue : public TaskQueue {
/// \param task_id The task ID for the task to append.
/// \param task The task to append to the queue.
/// \return Whether the append operation succeeds.
bool AppendTask(const TaskID &task_id, const Task &task);
bool AppendTask(const TaskID &task_id, const Task &task) override;
/// \brief Remove a task from queue.
///
/// \param task_id The task ID for the task to remove from the queue.
/// \return Whether the removal succeeds.
bool RemoveTask(const TaskID &task_id, std::vector<Task> *removed_tasks);
/// \brief Get task associated to task_id in this queue.
///
/// \param task_id The task ID for the task to get.
/// \return The task corresponding to task_id.
const Task &GetTask(const TaskID &task_id) const {
auto it = task_map_.find(task_id);
RAY_CHECK(it != task_map_.end());
return *it->second;
}
bool RemoveTask(const TaskID &task_id, std::vector<Task> *removed_tasks) override;
/// \brief Get a mapping from resource shape to tasks.
///
/// \return Mapping from resource set to task IDs with these resource requirements.
const std::unordered_map<ResourceSet, ordered_set<TaskID>> &GetTasksWithResources()
const {
return tasks_with_resources_;
}
const;
private:
/// Index from resource shape to tasks that require these resources.
@@ -134,7 +134,19 @@ class ReadyQueue : public TaskQueue {
class SchedulingQueue {
public:
/// Create a scheduling queue.
SchedulingQueue() {}
SchedulingQueue() : ready_queue_(std::make_shared<ReadyQueue>()) {
for (const auto &task_state : {
TaskState::PLACEABLE, TaskState::WAITING, TaskState::READY,
TaskState::RUNNING, TaskState::INFEASIBLE,
TaskState::WAITING_FOR_ACTOR_CREATION,
}) {
if (task_state == TaskState::READY) {
task_queues_[static_cast<int>(task_state)] = ready_queue_;
} else {
task_queues_[static_cast<int>(task_state)] = std::make_shared<TaskQueue>();
}
}
}
/// SchedulingQueue destructor.
virtual ~SchedulingQueue() {}
@@ -145,30 +157,25 @@ class SchedulingQueue {
/// \return Whether the task_id exists in the queue.
bool HasTask(const TaskID &task_id) const;
/// Get the queue of tasks that are destined for actors that have not yet
/// been created.
/// \brief Get all tasks in the given state.
///
/// \return A const reference to the queue of tasks that are destined for
/// actors that have not yet been created.
const std::list<Task> &GetMethodsWaitingForActorCreation() const;
/// \param task_state The requested task state. This must correspond to one
/// of the task queues (has value < TaskState::kNumTaskQueues).
const std::list<Task> &GetTasks(TaskState task_state) const;
/// Get the queue of tasks in the waiting state.
/// Get a reference to the queue of ready tasks.
///
/// \return A const reference to the queue of tasks that are waiting for
/// object dependencies to become available.
const std::list<Task> &GetWaitingTasks() const;
/// \return A reference to the queue of ready tasks.
const std::unordered_map<ResourceSet, ordered_set<TaskID>> &GetReadyTasksWithResources()
const;
/// Get the queue of tasks in the placeable state.
/// Get a task from the queue of a given state. The caller must ensure that
/// the task has the given state.
///
/// \return A const reference to the queue of tasks that have all
/// dependencies local and that are waiting to be scheduled.
const std::list<Task> &GetPlaceableTasks() const;
/// Get the queue of tasks in the infeasible state.
///
/// \return A const reference to the queue of tasks whose resource
/// requirements are not satisfied by any node in the cluster.
const std::list<Task> &GetInfeasibleTasks() const;
/// \param task_id The task to get.
/// \param task_state The state that the requested task should be in.
/// \return The task.
const Task &GetTaskOfState(const TaskID &task_id, TaskState task_state) const;
/// \brief Return an aggregate resource set for all tasks exerting load on this raylet.
///
@@ -176,23 +183,6 @@ class SchedulingQueue {
/// this raylet.
ResourceSet GetResourceLoad() const;
/// Get the queue of tasks in the ready state.
///
/// \return A const reference to the queue of tasks ready
/// to execute but that are waiting for a worker.
const std::list<Task> &GetReadyTasks() const;
/// Get a reference to the queue of ready tasks.
///
/// \return A reference to the queue of ready tasks.
const ReadyQueue &GetReadyQueue() const { return ready_tasks_; }
/// Get the queue of tasks in the running state.
///
/// \return A const reference to the queue of tasks that are currently
/// executing on a worker.
const std::list<Task> &GetRunningTasks() const;
/// Get the tasks in the blocked state.
///
/// \return A const reference to the tasks that are are blocked on a data
@@ -210,14 +200,11 @@ class SchedulingQueue {
/// Remove tasks from the task queue.
///
/// \param tasks The set of task IDs to remove from the queue. The
/// \param task_ids The set of task IDs to remove from the queue. The
/// corresponding tasks must be contained in the queue. The IDs of removed
/// tasks will be erased from the set.
/// \param task_states If this is not nullptr, then, the states of the removed
/// tasks will be appended to this vector.
/// \return A vector of the tasks that were removed.
std::vector<Task> RemoveTasks(std::unordered_set<TaskID> &task_ids,
std::vector<TaskState> *task_states = nullptr);
std::vector<Task> RemoveTasks(std::unordered_set<TaskID> &task_ids);
/// Remove a task from the task queue.
///
@@ -233,31 +220,13 @@ class SchedulingQueue {
/// \param The driver task ID to remove.
void RemoveDriverTaskId(const TaskID &task_id);
/// Queue tasks that are destined for actors that have not yet been created.
/// Add tasks to the given queue.
///
/// \param tasks The tasks to queue.
void QueueMethodsWaitingForActorCreation(const std::vector<Task> &tasks);
/// Queue tasks in the waiting state. These are tasks that cannot yet be
/// dispatched since they are blocked on a missing data dependency.
///
/// \param tasks The tasks to queue.
void QueueWaitingTasks(const std::vector<Task> &tasks);
/// Queue tasks in the placeable state.
///
/// \param tasks The tasks to queue.
void QueuePlaceableTasks(const std::vector<Task> &tasks);
/// Queue tasks in the ready state.
///
/// \param tasks The tasks to queue.
void QueueReadyTasks(const std::vector<Task> &tasks);
/// Queue tasks in the running state.
///
/// \param tasks The tasks to queue.
void QueueRunningTasks(const std::vector<Task> &tasks);
/// \param task_state The state of the tasks to queue. The requested task
/// state must correspond to one of the task queues (has value <
/// TaskState::kNumTaskQueues).
void QueueTasks(const std::vector<Task> &tasks, TaskState task_state);
/// Add a task ID in the blocked state. These are tasks that have been
/// dispatched to a worker but are blocked on a data dependency that was
@@ -320,23 +289,33 @@ class SchedulingQueue {
std::string DebugString() const;
private:
/// Tasks that are destined for actors that have not yet been created.
TaskQueue methods_waiting_for_actor_creation_;
/// Tasks that are waiting for an object dependency to appear locally.
TaskQueue waiting_tasks_;
/// Tasks whose object dependencies are locally available, but that are
/// waiting to be scheduled.
TaskQueue placeable_tasks_;
/// Tasks ready for dispatch, but that are waiting for a worker.
ReadyQueue ready_tasks_;
/// Tasks that are running on a worker.
TaskQueue running_tasks_;
/// Get the task queue in the given state. The requested task state must
/// correspond to one of the task queues (has value <
/// TaskState::kNumTaskQueues).
const std::shared_ptr<TaskQueue> &GetTaskQueue(TaskState task_state) const;
/// A helper function to remove tasks from a given queue. The requested task
/// state must correspond to one of the task queues (has value <
/// TaskState::kNumTaskQueues).
void RemoveTasksFromQueue(ray::raylet::TaskState task_state,
std::unordered_set<ray::TaskID> &task_ids,
std::vector<ray::raylet::Task> *removed_tasks);
/// A helper function to filter out tasks of a given state from the set of
/// task IDs. The requested task state must correspond to one of the task
/// queues (has value < TaskState::kNumTaskQueues).
void FilterStateFromQueue(std::unordered_set<ray::TaskID> &task_ids,
TaskState task_state) const;
// A pointer to the ready queue.
const std::shared_ptr<ReadyQueue> ready_queue_;
// A pointer to the task queues. These contain all tasks that have a task
// state < TaskState::kNumTaskQueues.
std::array<std::shared_ptr<TaskQueue>, static_cast<int>(TaskState::kNumTaskQueues)>
task_queues_;
/// Tasks that were dispatched to a worker but are blocked on a data
/// dependency that was missing at runtime.
std::unordered_set<TaskID> blocked_task_ids_;
/// Tasks that require resources that are not available on any of the nodes
/// in the cluster.
TaskQueue infeasible_tasks_;
/// The set of currently running driver tasks. These are empty tasks that are
/// started by a driver process on initialization.
std::unordered_set<TaskID> driver_task_ids_;
src/ray/raylet/task_dependency_manager.cc
-1
View File
@@ -74,7 +74,6 @@ void TaskDependencyManager::HandleRemoteDependencyCanceled(const ObjectID &objec
std::vector<TaskID> TaskDependencyManager::HandleObjectLocal(
const ray::ObjectID &object_id) {
RAY_LOG(DEBUG) << "object ready " << object_id.hex();
// Add the object to the table of locally available objects.
auto inserted = local_objects_.insert(object_id);
RAY_CHECK(inserted.second);
src/ray/raylet/worker_pool.cc
+1 -3
View File
@@ -102,7 +102,7 @@ void WorkerPool::StartWorkerProcess(const Language &language) {
if (static_cast<int>(starting_worker_processes_.size()) >=
maximum_startup_concurrency_) {
// Workers have been started, but not registered. Force start disabled -- returning.
RAY_LOG(DEBUG) << starting_worker_processes_.size()
RAY_LOG(DEBUG) << "Worker not started, " << starting_worker_processes_.size()
<< " worker processes pending registration";
return;
}
@@ -241,8 +241,6 @@ std::vector<std::shared_ptr<Worker>> WorkerPool::GetWorkersRunningTasksForDriver
for (const auto &entry : states_by_lang_) {
for (const auto &worker : entry.second.registered_workers) {
RAY_LOG(DEBUG) << "worker: pid : " << worker->Pid()
<< " driver_id: " << worker->GetAssignedDriverId();
if (worker->GetAssignedDriverId() == driver_id) {
workers.push_back(worker);
}
src/ray/raylet/worker_pool_test.cc
+1 -1
View File
@@ -48,7 +48,7 @@ class WorkerPoolTest : public ::testing::Test {
boost::asio::local::stream_protocol::socket socket(io_service_);
auto client =
LocalClientConnection::Create(client_handler, message_handler, std::move(socket),
"worker", error_message_type_);
"worker", {}, error_message_type_);
return std::shared_ptr<Worker>(new Worker(pid, language, client));
}