diff --git a/doc/source/development.rst b/doc/source/development.rst index 9b67ee01f..58169aab9 100644 --- a/doc/source/development.rst +++ b/doc/source/development.rst @@ -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 diff --git a/src/ray/common/client_connection.cc b/src/ray/common/client_connection.cc index 21588177a..51d3859d5 100644 --- a/src/ray/common/client_connection.cc +++ b/src/ray/common/client_connection.cc @@ -187,22 +187,24 @@ template std::shared_ptr> ClientConnection::Create( ClientHandler &client_handler, MessageHandler &message_handler, boost::asio::basic_stream_socket &&socket, const std::string &debug_label, - int64_t error_message_type) { - std::shared_ptr> self(new ClientConnection( - message_handler, std::move(socket), debug_label, error_message_type)); + const std::vector &message_type_enum_names, int64_t error_message_type) { + std::shared_ptr> 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 -ClientConnection::ClientConnection(MessageHandler &message_handler, - boost::asio::basic_stream_socket &&socket, - const std::string &debug_label, - int64_t error_message_type) +ClientConnection::ClientConnection( + MessageHandler &message_handler, boost::asio::basic_stream_socket &&socket, + const std::string &debug_label, + const std::vector &message_type_enum_names, int64_t error_message_type) : ServerConnection(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 @@ -261,8 +263,14 @@ void ClientConnection::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."; } } diff --git a/src/ray/common/client_connection.h b/src/ray/common/client_connection.h index 28bdad5b5..22d4a8ba0 100644 --- a/src/ray/common/client_connection.h +++ b/src/ray/common/client_connection.h @@ -145,10 +145,15 @@ class ClientConnection : public ServerConnection { /// \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. static std::shared_ptr> Create( ClientHandler &new_client_handler, MessageHandler &message_handler, boost::asio::basic_stream_socket &&socket, const std::string &debug_label, + const std::vector &message_type_enum_names, int64_t error_message_type); std::shared_ptr> shared_ClientConnection_from_this() { @@ -170,7 +175,9 @@ class ClientConnection : public ServerConnection { /// A private constructor for a node client connection. ClientConnection(MessageHandler &message_handler, boost::asio::basic_stream_socket &&socket, - const std::string &debug_label, int64_t error_message_type); + const std::string &debug_label, + const std::vector &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 { MessageHandler 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 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. diff --git a/src/ray/gcs/tables.cc b/src/ray/gcs/tables.cc index 8e60c8e67..00ed64d7c 100644 --- a/src/ray/gcs/tables.cc +++ b/src/ray/gcs/tables.cc @@ -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(); 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) { diff --git a/src/ray/object_manager/connection_pool.cc b/src/ray/object_manager/connection_pool.cc index dcfb2a77b..d1261d7da 100644 --- a/src/ray/object_manager/connection_pool.cc +++ b/src/ray/object_manager/connection_pool.cc @@ -123,14 +123,12 @@ std::shared_ptr ConnectionPool::Borrow(SenderMapType &conn_map const ClientID &client_id) { std::shared_ptr 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 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 { diff --git a/src/ray/object_manager/object_buffer_pool.cc b/src/ray/object_manager/object_buffer_pool.cc index 8f9b8a461..11ae44239 100644 --- a/src/ray/object_manager/object_buffer_pool.cc +++ b/src/ray/object_manager/object_buffer_pool.cc @@ -41,7 +41,6 @@ std::pair ObjectBufferPool::Ge const ObjectID &object_id, uint64_t data_size, uint64_t metadata_size, uint64_t chunk_index) { std::lock_guard 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 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 ObjectBufferPool::Cr const ObjectID &object_id, uint64_t data_size, uint64_t metadata_size, uint64_t chunk_index) { std::lock_guard 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)); diff --git a/src/ray/object_manager/object_manager.cc b/src/ray/object_manager/object_manager.cc index 063601336..e9904d960 100644 --- a/src/ray/object_manager/object_manager.cc +++ b/src/ray/object_manager/object_manager.cc @@ -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 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 &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 &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 ObjectManager::CreateSenderConnection( @@ -707,8 +724,11 @@ void ObjectManager::ProcessNewClient(TcpClientConnection &conn) { void ObjectManager::ProcessClientMessage(std::shared_ptr &conn, int64_t message_type, const uint8_t *message) { - auto message_type_value = + const auto message_type_value = static_cast(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 &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 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 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; } 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 3f2d50ead..91b0ffc3d 100644 --- a/src/ray/object_manager/test/object_manager_stress_test.cc +++ b/src/ray/object_manager/test/object_manager_stress_test.cc @@ -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(object_manager::protocol::MessageType::DisconnectClient)); DoAcceptObjectManager(); } diff --git a/src/ray/object_manager/test/object_manager_test.cc b/src/ray/object_manager/test/object_manager_test.cc index a71d7636a..98ad9bbfb 100644 --- a/src/ray/object_manager/test/object_manager_test.cc +++ b/src/ray/object_manager/test/object_manager_test.cc @@ -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(object_manager::protocol::MessageType::DisconnectClient)); DoAcceptObjectManager(); } diff --git a/src/ray/raylet/client_connection_test.cc b/src/ray/raylet/client_connection_test.cc index 5623beb90..4ef236bc1 100644 --- a/src/ray/raylet/client_connection_test.cc +++ b/src/ray/raylet/client_connection_test.cc @@ -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 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 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 callback = [writer](const ray::Status &status) { diff --git a/src/ray/raylet/lineage_cache.cc b/src/ray/raylet/lineage_cache.cc index 8180319eb..af1f42167 100644 --- a/src/ray/raylet/lineage_cache.cc +++ b/src/ray/raylet/lineage_cache.cc @@ -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. diff --git a/src/ray/raylet/node_manager.cc b/src/ray/raylet/node_manager.cc index 952fadabe..a7f76edf3 100644 --- a/src/ray/raylet/node_manager.cc +++ b/src/ray/raylet/node_manager.cc @@ -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(); 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(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 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> &tasks_with_resources) { std::unordered_set 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(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(message_type)) { + const auto message_type_value = static_cast(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(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 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 // 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 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({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 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 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. diff --git a/src/ray/raylet/raylet.cc b/src/ray/raylet/raylet.cc index 23f0dba6c..288f0a80b 100644 --- a/src/ray/raylet/raylet.cc +++ b/src/ray/raylet/raylet.cc @@ -7,6 +7,28 @@ #include "ray/status.h" +namespace { + +const std::vector GenerateEnumNames(const char *const *enum_names_ptr) { + std::vector 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 node_manager_message_enum = + GenerateEnumNames(ray::protocol::EnumNamesMessageType()); +static const std::vector 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(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(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(protocol::MessageType::DisconnectClient)); } // We're ready to accept another client. diff --git a/src/ray/raylet/scheduling_policy.cc b/src/ray/raylet/scheduling_policy.cc index 177d2fd80..01a3cde57 100644 --- a/src/ray/raylet/scheduling_policy.cc +++ b/src/ray/raylet/scheduling_policy.cc @@ -19,10 +19,8 @@ std::unordered_map SchedulingPolicy::Schedule( const ClientID &local_client_id) { // The policy decision to be returned. std::unordered_map 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 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 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 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. diff --git a/src/ray/raylet/scheduling_queue.cc b/src/ray/raylet/scheduling_queue.cc index 47c353a9a..dc6316868 100644 --- a/src/ray/raylet/scheduling_queue.cc +++ b/src/ray/raylet/scheduling_queue.cc @@ -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 -void RemoveTasksFromQueue(ray::raylet::TaskState task_state, TaskQueue &queue, - std::unordered_set &task_ids, - std::vector *removed_tasks, - std::vector *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(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 -inline void QueueTasks(TaskQueue &queue, const std::vector &tasks) { - for (const auto &task : tasks) { - queue.AppendTask(task.GetTaskSpecification().TaskId(), task); - } -} - -// Helper function to filter out tasks of a given state. -template -inline void FilterStateFromQueue(const TaskQueue &queue, - std::unordered_set &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(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 &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 *removed_ta return TaskQueue::RemoveTask(task_id, removed_tasks); } -const std::list &SchedulingQueue::GetMethodsWaitingForActorCreation() const { - return methods_waiting_for_actor_creation_.GetTasks(); +const std::unordered_map> + &ReadyQueue::GetTasksWithResources() const { + return tasks_with_resources_; } -const std::list &SchedulingQueue::GetWaitingTasks() const { - return waiting_tasks_.GetTasks(); +const std::list &SchedulingQueue::GetTasks(TaskState task_state) const { + const auto &queue = GetTaskQueue(task_state); + return queue->GetTasks(); } -const std::list &SchedulingQueue::GetPlaceableTasks() const { - return placeable_tasks_.GetTasks(); +const std::unordered_map> + &SchedulingQueue::GetReadyTasksWithResources() const { + return ready_queue_->GetTasksWithResources(); } -const std::list &SchedulingQueue::GetReadyTasks() const { - return ready_tasks_.GetTasks(); -} - -const std::list &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 &SchedulingQueue::GetRunningTasks() const { - return running_tasks_.GetTasks(); + return ready_queue_->GetCurrentResourceLoad(); } const std::unordered_set &SchedulingQueue::GetBlockedTaskIds() const { return blocked_task_ids_; } +void SchedulingQueue::FilterStateFromQueue(std::unordered_set &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 &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 &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 &task_ids, } } -std::vector SchedulingQueue::RemoveTasks(std::unordered_set &task_ids, - std::vector *task_states) { +const std::shared_ptr &SchedulingQueue::GetTaskQueue( + TaskState task_state) const { + RAY_CHECK(task_state < TaskState::kNumTaskQueues) + << static_cast(task_state) << "Task state " << static_cast(task_state) + << " does not correspond to a task queue"; + return task_queues_[static_cast(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 &task_ids, + std::vector *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 SchedulingQueue::RemoveTasks(std::unordered_set &task_ids) { // List of removed tasks to be returned. std::vector 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 removed_tasks; std::unordered_set task_id_set = {task_id}; - std::vector 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 &task_ids, TaskState src_state, TaskState dst_state) { - // TODO(atumanov): check the states first to ensure the move is transactional. std::vector 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::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 &task_ids, TaskState } } -void SchedulingQueue::QueueMethodsWaitingForActorCreation( - const std::vector &tasks) { - QueueTasks(methods_waiting_for_actor_creation_, tasks); +void SchedulingQueue::QueueTasks(const std::vector &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 &tasks) { - QueueTasks(waiting_tasks_, tasks); -} - -void SchedulingQueue::QueuePlaceableTasks(const std::vector &tasks) { - QueueTasks(placeable_tasks_, tasks); -} - -void SchedulingQueue::QueueReadyTasks(const std::vector &tasks) { - QueueTasks(ready_tasks_, tasks); -} - -void SchedulingQueue::QueueRunningTasks(const std::vector &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 SchedulingQueue::GetTaskIdsForDriver( const DriverID &driver_id) const { std::unordered_set 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 SchedulingQueue::GetTaskIdsForActor( const ActorID &actor_id) const { std::unordered_set 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 &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(); } diff --git a/src/ray/raylet/scheduling_queue.h b/src/ray/raylet/scheduling_queue.h index dad94a6d3..f00e02ba3 100644 --- a/src/ray/raylet/scheduling_queue.h +++ b/src/ray/raylet/scheduling_queue.h @@ -1,6 +1,7 @@ #ifndef RAY_RAYLET_SCHEDULING_QUEUE_H #define RAY_RAYLET_SCHEDULING_QUEUE_H +#include #include #include #include @@ -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 *removed_tasks = nullptr); + virtual bool RemoveTask(const TaskID &task_id, + std::vector *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 &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 *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 *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> &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()) { + 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(task_state)] = ready_queue_; + } else { + task_queues_[static_cast(task_state)] = std::make_shared(); + } + } + } /// 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 &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 &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 &GetWaitingTasks() const; + /// \return A reference to the queue of ready tasks. + const std::unordered_map> &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 &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 &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 &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 &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 RemoveTasks(std::unordered_set &task_ids, - std::vector *task_states = nullptr); + std::vector RemoveTasks(std::unordered_set &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 &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 &tasks); - - /// Queue tasks in the placeable state. - /// - /// \param tasks The tasks to queue. - void QueuePlaceableTasks(const std::vector &tasks); - - /// Queue tasks in the ready state. - /// - /// \param tasks The tasks to queue. - void QueueReadyTasks(const std::vector &tasks); - - /// Queue tasks in the running state. - /// - /// \param tasks The tasks to queue. - void QueueRunningTasks(const std::vector &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 &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 &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 &task_ids, + std::vector *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 &task_ids, + TaskState task_state) const; + + // A pointer to the ready queue. + const std::shared_ptr ready_queue_; + // A pointer to the task queues. These contain all tasks that have a task + // state < TaskState::kNumTaskQueues. + std::array, static_cast(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 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 driver_task_ids_; diff --git a/src/ray/raylet/task_dependency_manager.cc b/src/ray/raylet/task_dependency_manager.cc index 1d491b75f..e0d447824 100644 --- a/src/ray/raylet/task_dependency_manager.cc +++ b/src/ray/raylet/task_dependency_manager.cc @@ -74,7 +74,6 @@ void TaskDependencyManager::HandleRemoteDependencyCanceled(const ObjectID &objec std::vector 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); diff --git a/src/ray/raylet/worker_pool.cc b/src/ray/raylet/worker_pool.cc index 41e056e51..8387656e0 100644 --- a/src/ray/raylet/worker_pool.cc +++ b/src/ray/raylet/worker_pool.cc @@ -102,7 +102,7 @@ void WorkerPool::StartWorkerProcess(const Language &language) { if (static_cast(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> 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); } diff --git a/src/ray/raylet/worker_pool_test.cc b/src/ray/raylet/worker_pool_test.cc index 4faa3e99b..e03c61bb5 100644 --- a/src/ray/raylet/worker_pool_test.cc +++ b/src/ray/raylet/worker_pool_test.cc @@ -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(new Worker(pid, language, client)); }