feat: Cache ready tasks query results in scheduler - #58
Conversation
WalkthroughThe pull request introduces enhancements to the task scheduling and management system in the Spider project. Key changes include the addition of a new Changes
Sequence DiagramsequenceDiagram
participant Worker
participant WorkerClient
participant SchedulerServer
participant FifoPolicy
participant SchedulerTaskCache
participant MetadataStorage
Worker->>WorkerClient: Request next task
WorkerClient->>SchedulerServer: Get next task
SchedulerServer->>FifoPolicy: schedule_next()
FifoPolicy->>SchedulerTaskCache: get_ready_task()
SchedulerTaskCache->>MetadataStorage: fetch_ready_tasks()
MetadataStorage-->>SchedulerTaskCache: Return tasks
SchedulerTaskCache-->>FifoPolicy: Return task ID
FifoPolicy-->>SchedulerServer: Return task ID
SchedulerServer-->>WorkerClient: Return task and task instance IDs
WorkerClient-->>Worker: Provide task details
Possibly related PRs
Finishing Touches
Thank you for using CodeRabbit. We offer it for free to the OSS community and would appreciate your support in helping us grow. If you find it useful, would you consider giving us a shout-out on your favorite social media? 🪧 TipsChatThere are 3 ways to chat with CodeRabbit:
Note: Be mindful of the bot's finite context window. It's strongly recommended to break down tasks such as reading entire modules into smaller chunks. For a focused discussion, use review comments to chat about specific files and their changes, instead of using the PR comments. CodeRabbit Commands (Invoked using PR comments)
Other keywords and placeholders
CodeRabbit Configuration File (
|
sitaowang1998
changed the title
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
🧹 Nitpick comments (12)
src/spider/scheduler/SchedulerMessage.hpp (2)
58-79: Enhance readability by replacingstd::tuplewith a struct for task IDsUsing
std::tuple<boost::uuids::uuid, boost::uuids::uuid>form_task_idsmay reduce code clarity. Consider defining a struct with named members, such asTaskIdentifiers, to improve readability and maintainability.Apply this change to define a struct and update the code:
// Define the struct struct TaskIdentifiers { boost::uuids::uuid task_id; boost::uuids::uuid task_instance_id; }; // Update the member variable std::optional<TaskIdentifiers> m_task_ids = std::nullopt; // Update constructors ScheduleTaskResponse( boost::uuids::uuid const task_id, boost::uuids::uuid const task_instance_id ) : m_task_ids{TaskIdentifiers{task_id, task_instance_id}} {} explicit ScheduleTaskResponse(TaskIdentifiers const& task_ids) : m_task_ids{task_ids} {} // Update methods [[nodiscard]] auto get_task_ids() const -> TaskIdentifiers const& { return m_task_ids.value(); } // Update MSGPACK definition MSGPACK_DEFINE_ARRAY(m_task_ids);
68-68: Rename method to reflect plural task IDsThe method
has_task_id()checks for multiple task IDs. For clarity and consistency, consider renaming it tohas_task_ids().Apply this diff to rename the method:
-[[nodiscard]] auto has_task_id() const -> bool { return m_task_ids.has_value(); } +[[nodiscard]] auto has_task_ids() const -> bool { return m_task_ids.has_value(); }src/spider/scheduler/SchedulerTaskCache.cpp (1)
54-58: Optimize task retrieval to avoid unnecessary copyingIn
pop_next_task, copyingm_tasksinto a vectortaskscan be inefficient for large datasets. Consider iterating overm_tasksdirectly or modifyingm_get_next_task_functionto accept the map or its iterators to improve performance.Apply this change to pass
m_tasksdirectly:// Modify the function signature if possible auto pop_next_task( boost::uuids::uuid const& worker_id, std::string const& worker_addr ) -> std::optional<core::Task> { // Pass m_tasks directly std::optional<boost::uuids::uuid> const task_id = std::invoke( m_get_next_task_function, std::ref(m_tasks), std::cref(worker_id), std::cref(worker_addr) ); // ... rest of the code }src/spider/scheduler/FifoPolicy.cpp (4)
62-76: Use initialization list for member variablesIn the constructor, member variables are being assigned within the constructor body. It's more efficient and idiomatic in C++ to initialize them using an initialization list.
Apply this diff to use an initialization list:
-FifoPolicy::FifoPolicy( - std::shared_ptr<core::MetadataStorage> const& metadata_store, - std::shared_ptr<core::DataStorage> const& data_store -) - : m_metadata_store{metadata_store}, - m_data_store{data_store}, - m_task_cache{ - metadata_store, - data_store, - [&](std::vector<core::Task>& tasks, - boost::uuids::uuid const& worker_id, - std::string const& worker_addr) -> std::optional<boost::uuids::uuid> { - return get_next_task(tasks, worker_id, worker_addr); - } - } {} +FifoPolicy::FifoPolicy( + std::shared_ptr<core::MetadataStorage> const& metadata_store, + std::shared_ptr<core::DataStorage> const& data_store +) : + m_metadata_store{metadata_store}, + m_data_store{data_store}, + m_task_cache{ + metadata_store, + data_store, + [&](std::vector<core::Task>& tasks, + boost::uuids::uuid const& worker_id, + std::string const& worker_addr) -> std::optional<boost::uuids::uuid> { + return get_next_task(tasks, worker_id, worker_addr); + } + } {}
83-84: Handle potential exceptions in lambda functionThe lambda function used in
std::erase_ifmay throw exceptions iftask_locality_satisfiedfails. Consider wrapping the lambda's body with exception handling to ensure robustness.
102-102: Correct grammatical error in exception messageThe error message "Task with id {} not exists." should be "Task with id {} does not exist." to be grammatically correct.
Apply this diff to fix the message:
- throw std::runtime_error(fmt::format( - "Task with id {} not exists.", - boost::uuids::to_string(task_id) - )); + throw std::runtime_error(fmt::format( + "Task with id {} does not exist.", + boost::uuids::to_string(task_id) + ));
118-118: Correct grammatical error in exception messageSimilarly, the error message "Job with id {} not exists." should be "Job with id {} does not exist."
Apply this diff to fix the message:
- throw std::runtime_error(fmt::format( - "Job with id {} not exists.", - boost::uuids::to_string(job_id) - )); + throw std::runtime_error(fmt::format( + "Job with id {} does not exist.", - boost::uuids::to_string(job_id) + ));src/spider/scheduler/SchedulerTaskCache.hpp (2)
50-51: Document the reason for NOLINT directive.Add a brief comment explaining why the include-cleaner check is disabled here.
52-53: Document the cache update strategy.Add documentation explaining:
- The purpose of
m_last_update- The significance of
m_update_count- The cache invalidation strategy
src/spider/scheduler/SchedulerServer.cpp (1)
159-169: Consider retrying task instance creation.The error handling is good, but for transient storage errors, consider implementing a retry mechanism with exponential backoff.
Example implementation:
+constexpr int MAX_RETRIES = 3; +constexpr std::chrono::milliseconds INITIAL_DELAY{100}; + if (task_id.has_value()) { core::TaskInstance const instance{task_id.value()}; - core::StorageErr const err = m_metadata_store->create_task_instance(instance); - if (err.success()) { - response = ScheduleTaskResponse{task_id.value(), instance.id}; - } else { + core::StorageErr err; + for (int retry = 0; retry < MAX_RETRIES; ++retry) { + err = m_metadata_store->create_task_instance(instance); + if (err.success()) { + response = ScheduleTaskResponse{task_id.value(), instance.id}; + break; + } + if (retry < MAX_RETRIES - 1) { + std::this_thread::sleep_for(INITIAL_DELAY * (1 << retry)); + continue; + } spdlog::error( "Cannot create task instance {}: {}", boost::uuids::to_string(task_id.value()), err.description ); - } + } }src/spider/scheduler/scheduler.cpp (1)
192-192: LGTM! Consider adding cache performance monitoring.The FifoPolicy is correctly initialized with the required storage components. Consider adding metrics to monitor cache hit rates and performance improvements.
src/spider/storage/MetadataStorage.hpp (1)
57-58: Add documentation for newcreate_task_instancemethodConsider adding comments or documentation to explain the purpose and usage of the
create_task_instancemethod for clarity and maintainability.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (17)
src/spider/CMakeLists.txt(1 hunks)src/spider/scheduler/FifoPolicy.cpp(4 hunks)src/spider/scheduler/FifoPolicy.hpp(1 hunks)src/spider/scheduler/SchedulerMessage.hpp(2 hunks)src/spider/scheduler/SchedulerPolicy.hpp(1 hunks)src/spider/scheduler/SchedulerServer.cpp(2 hunks)src/spider/scheduler/SchedulerTaskCache.cpp(1 hunks)src/spider/scheduler/SchedulerTaskCache.hpp(1 hunks)src/spider/scheduler/scheduler.cpp(1 hunks)src/spider/storage/MetadataStorage.hpp(2 hunks)src/spider/storage/MySqlStorage.cpp(4 hunks)src/spider/storage/MySqlStorage.hpp(2 hunks)src/spider/worker/WorkerClient.cpp(3 hunks)src/spider/worker/WorkerClient.hpp(2 hunks)src/spider/worker/worker.cpp(3 hunks)tests/scheduler/test-SchedulerPolicy.cpp(3 hunks)tests/scheduler/test-SchedulerServer.cpp(2 hunks)
🧰 Additional context used
🪛 cppcheck (2.10-2)
src/spider/scheduler/SchedulerTaskCache.cpp
[performance] 97-97: Variable 'id' is assigned in constructor body. Consider performing initialization in initialization list.
(useInitializationList)
src/spider/scheduler/FifoPolicy.cpp
[performance] 97-97: Variable 'id' is assigned in constructor body. Consider performing initialization in initialization list.
(useInitializationList)
⏰ Context from checks skipped due to timeout of 90000ms (3)
- GitHub Check: non-storage-unit-tests (ubuntu-24.04)
- GitHub Check: non-storage-unit-tests (ubuntu-22.04)
- GitHub Check: lint
🔇 Additional comments (29)
src/spider/scheduler/FifoPolicy.hpp (5)
23-26: Constructor initializes member variables appropriatelyThe new constructor effectively initializes
FifoPolicywith references toMetadataStorageandDataStorage, enhancing encapsulation and simplifying method interfaces.
28-29: Simplifiedschedule_nextmethod improves interface clarityBy removing the metadata and data store parameters from
schedule_next, the method interface is cleaner and less error-prone, enhancing usability.
33-37: Addition ofget_next_taskenhances modularityThe new private method
get_next_taskencapsulates task selection logic, promoting code modularity and readability within theFifoPolicyclass.
39-40: Member variables for storage references improve state managementIntroducing
m_metadata_storeandm_data_storeas member variables centralizes access to storage components, streamlining state management within the scheduler policy.
42-42: Inclusion ofSchedulerTaskCachesupports efficient task cachingAdding
m_task_cacheas a member variable integrates task caching into the scheduling policy, potentially enhancing performance by reducing redundant storage queries.src/spider/scheduler/SchedulerMessage.hpp (1)
76-76: Verify MSGPACK serialization of optional tupleEnsure that
MSGPACK_DEFINE_ARRAY(m_task_ids);correctly serializes and deserializes thestd::optional<std::tuple<boost::uuids::uuid, boost::uuids::uuid>>member. It's important to confirm that the optionality and tuple are properly handled by MSGPACK to prevent runtime serialization errors.src/spider/worker/WorkerClient.cpp (3)
10-10: Include<tuple>header for tuple usageIncluding
<tuple>is necessary for the use ofstd::tuplein the updated return type.
98-98: Update response handling to match new return typeReturning
response.get_task_ids()aligns with the updated return type. Ensure thatresponse.get_task_ids()returns the expected tuple and that downstream code correctly handles both task IDs.
38-38: Ensure all usages ofget_next_taskhandle the new return typeThe return type of
get_next_taskhas changed tostd::optional<std::tuple<boost::uuids::uuid, boost::uuids::uuid>>. Please verify that all calls to this function and any code that processes its return value are updated to handle the tuple appropriately.Run the following script to find all references to
get_next_taskand check their usage:✅ Verification successful
All usages of
get_next_taskproperly handle the new return typeThe only usage of
WorkerClient::get_next_taskis inworker.cpp, and it correctly handles the newstd::optional<std::tuple<boost::uuids::uuid, boost::uuids::uuid>>return type with proper optional and tuple handling.🏁 Scripts executed
The following scripts were executed for the analysis:
Script:
#!/bin/bash # Description: Find all usages of `get_next_task` and display surrounding context. rg -A 3 -B 3 'get_next_task\('Length of output: 2946
src/spider/scheduler/FifoPolicy.cpp (1)
132-135: Ensure consistent use of member functionsThe
schedule_nextmethod now relies onm_task_cache.get_ready_task. Verify thatget_ready_taskis implemented correctly and that it interacts properly with theFifoPolicy.src/spider/scheduler/SchedulerPolicy.hpp (1)
19-20: Update derived classes to match new method signatureThe
schedule_nextmethod signature has changed by removing themetadata_storeanddata_storeparameters. Ensure all classes deriving fromSchedulerPolicyare updated to reflect this change.Run the following script to identify all implementations of
schedule_next:src/spider/worker/WorkerClient.hpp (2)
7-7: LGTM! Include directive added for tuple support.The include directive is correctly placed and necessary for the updated return type.
33-33: LGTM! Method signature updated to return task and instance IDs.The return type change to
std::optional<std::tuple<boost::uuids::uuid, boost::uuids::uuid>>is well-structured and supports the new task instance tracking feature.src/spider/scheduler/SchedulerTaskCache.hpp (1)
23-34: LGTM! Well-structured constructor with dependency injection.The constructor properly takes shared pointers to storage and task function dependencies, using initialization list for efficient member initialization.
tests/scheduler/test-SchedulerServer.cpp (1)
49-49: LGTM! Updated FifoPolicy initialization with required dependencies.The initialization correctly passes metadata and data store dependencies to support task caching.
src/spider/scheduler/SchedulerServer.cpp (1)
155-156: LGTM! Simplified policy interface.The scheduling call is now more focused, with storage dependencies handled by the policy itself.
tests/scheduler/test-SchedulerPolicy.cpp (3)
105-111: LGTM! Hard locality constraints are properly tested with the cache.The test effectively validates that the task cache respects hard locality constraints when scheduling tasks.
152-155: LGTM! Soft locality preferences are properly tested with the cache.The test effectively validates that the task cache handles soft locality preferences correctly while maintaining FIFO ordering.
59-65: LGTM! Test coverage validates FIFO ordering with the new cache.The test correctly validates that tasks are scheduled in FIFO order using the updated FifoPolicy constructor and schedule_next signatures.
Let's verify the test coverage for the new cache implementation:
✅ Verification successful
Cache implementation is properly tested through store interfaces
The test suite comprehensively validates the cache functionality through:
- FIFO scheduling tests with metadata and data stores
- Task scheduling with both hard and soft locality constraints
- Server integration tests with store operations
The cache implementation is effectively tested through these store interfaces, even though not explicitly named as "SchedulerTaskCache" in the tests.
🏁 Scripts executed
The following scripts were executed for the analysis:
Script:
#!/bin/bash # Description: Check test coverage for the new cache implementation # Test: Search for cache-related test cases rg -l "SchedulerTaskCache" "tests/"Length of output: 33
Script:
#!/bin/bash # Search for scheduler policy tests and store usage rg -l "Policy.*Test|test.*Policy" tests/ rg -C 3 "metadata_store|data_store" "tests/scheduler/"Length of output: 10990
src/spider/worker/worker.cpp (2)
128-134: LGTM! Task ID handling is properly implemented.The function correctly handles the new tuple-based task identification system, maintaining proper error handling and retry logic.
219-220: LGTM! Task instance creation is properly implemented.The code correctly destructures the task IDs and creates the task instance with both IDs.
src/spider/CMakeLists.txt (1)
113-114: LGTM! Build system properly includes the new cache implementation.The SchedulerTaskCache components are correctly added to the scheduler sources.
src/spider/storage/MySqlStorage.cpp (2)
75-75: Appropriate addition of index oncreation_timecolumnAdding an index on the
creation_timecolumn in thejobstable enhances query performance when filtering or sorting by this field.
Line range hint
1494-1536: Fix SQL syntax errors inget_task_timeoutmethodIn the
get_task_timeoutmethod, the SQL query at lines 1509-1514 has syntax issues:
Missing columns in SELECT statement: The query starts with
SELECT FROM, which is invalid. Specify the columns to select or useSELECT *.Incorrect backticks in WHERE clause: In the condition
WHERE \t1.task_id` = ?, backticks are misapplied. The correct syntax isWHERE `t1`.`task_id` = ?`.Apply this diff to fix the SQL query:
- conn->prepareStatement("SELECT FROM `task_instances` as `t1` JOIN `tasks` ON" + conn->prepareStatement("SELECT * FROM `task_instances` as `t1` JOIN `tasks` ON " "`t1`.`task_id` = `tasks`.`id` WHERE `t1.task_id` = ? AND " - "TIMESTAMPDIFF(MICROSECOND, `t1`.`start_time`, " + "TIMESTAMPDIFF(MICROSECOND, `t1`.`start_time`, " "CURRENT_TIMESTAMP()) < `tasks`.`timeout` * 1000")Please ensure the corrected SQL query retrieves the intended data and functions as expected.
✅ Verification successful
SQL syntax errors in
get_task_timeoutmethod are correctly identifiedThe review comment accurately identifies both SQL syntax issues and provides the correct fix that aligns with the codebase's SQL query patterns. The suggested changes:
- Fix the invalid "SELECT FROM" by using "SELECT *"
- Correct the backtick usage to properly quote table and column identifiers
🏁 Scripts executed
The following scripts were executed for the analysis:
Error: No shell scripts found
Script:
#!/bin/bash # Search for SQL SELECT statements in the codebase rg -U "SELECT.*FROM.*task_instances.*JOIN.*tasks" --type cpp -A 2 -B 2 # Search for backtick usage patterns in SQL queries rg -U "WHERE.*\`.*\`.*=" --type cpp -A 2 -B 2 # Look for other prepared statements rg "prepareStatement\(" --type cpp -A 3 -B 3Length of output: 73845
src/spider/storage/MetadataStorage.hpp (2)
5-5: Proper inclusion of<tuple>headerIncluding the
<tuple>header is necessary due to the usage ofstd::tuplein method signatures.
63-64: Update method signature forget_task_timeoutThe method signature for
get_task_timeouthas been updated to accept a vector ofstd::tuple<TaskInstance, Task>, which appropriately reflects the need to return both task instances and their associated tasks.src/spider/storage/MySqlStorage.hpp (3)
6-6: Duplicate inclusion of<tuple>headerThe inclusion of the
<tuple>header is appropriate here due to the use ofstd::tuplein method signatures.
61-61: Ensure correct implementation ofcreate_task_instancemethodThe
create_task_instancemethod is declared and correctly overrides the base class method. Ensure that its implementation aligns with the intended functionality.
65-66: Update method signature forget_task_timeoutThe updated method signature for
get_task_timeoutreflects the changes made to return tuples ofTaskInstanceandTask, which provides additional context for timeout handling.
|
|
||
| namespace { | ||
| constexpr int cUpdateCount = 100; | ||
| constexpr int cUpdateInterval = 5; // 10 ms |
There was a problem hiding this comment.
Correct the inconsistency between cUpdateInterval value and comment
The value of cUpdateInterval is defined as 5, but the comment states 10 ms. Please update the value or the comment for consistency.
Apply this diff to fix the inconsistency:
Option 1 (if the intended interval is 10 ms):
-constexpr int cUpdateInterval = 5; // 10 ms
+constexpr int cUpdateInterval = 10; // 10 msOption 2 (if the interval should be 5 ms):
-constexpr int cUpdateInterval = 5; // 10 ms
+constexpr int cUpdateInterval = 5; // 5 msCommittable suggestion skipped: line range outside the PR's diff.
| = object.as<spider::scheduler::ScheduleTaskResponse>(); | ||
| REQUIRE(res.has_task_id()); | ||
| REQUIRE(res.get_task_id() == parent_task.get_id()); | ||
| REQUIRE(std::get<0>(res.get_task_ids()) == parent_task.get_id()); |
There was a problem hiding this comment.
💡 Codebase verification
Missing validation for task instance ID in tuple response
The test only validates the task ID (first element) but not the task instance ID (second element) from the tuple returned by get_task_ids(). Both IDs are essential parts of the response structure.
- Add validation for the second tuple element using
std::get<1>(res.get_task_ids())to ensure the task instance ID is correct.
🔗 Analysis chain
LGTM! Updated task ID retrieval for tuple support.
The test correctly validates the first element (task ID) from the response tuple.
Let's verify that we're testing both tuple elements:
🏁 Scripts executed
The following scripts were executed for the analysis:
Script:
#!/bin/bash
# Search for other test cases that might be validating the instance ID
rg "get_task_ids\(\)" -A 5 "tests/"
Length of output: 418
Script:
#!/bin/bash
# Find the definition of get_task_ids()
ast-grep --pattern 'get_task_ids()'
# Look for any usage of std::get<1> in test files
rg "std::get<1>" tests/
# Find any files related to task IDs
rg "task_ids" --type cpp
Length of output: 4391
| auto MySqlMetadataStorage::create_task_instance(TaskInstance const& instance) -> StorageErr { | ||
| std::variant<MySqlConnection, StorageErr> conn_result = MySqlConnection::create(m_url); | ||
| if (std::holds_alternative<StorageErr>(conn_result)) { | ||
| return std::get<StorageErr>(conn_result); | ||
| } | ||
| auto& conn = std::get<MySqlConnection>(conn_result); | ||
| try { | ||
| // Check the state of the task | ||
| std::unique_ptr<sql::PreparedStatement> ready_statement(conn->prepareStatement( | ||
| "SELECT `state` FROM `tasks` WHERE `id` = ? AND `state` = 'ready'" | ||
| )); | ||
| sql::bytes id_bytes = uuid_get_bytes(instance.task_id); | ||
| ready_statement->setBytes(1, &id_bytes); | ||
| std::unique_ptr<sql::ResultSet> const ready_res(ready_statement->executeQuery()); | ||
| bool const task_ready = ready_res->rowsCount() > 0; | ||
| // Check all task instances have timed out | ||
| std::unique_ptr<sql::PreparedStatement> not_timeout_statement(conn->prepareStatement( | ||
| "SELECT `t1`.`id` FROM `task_instances` as `t1` JOIN `tasks` ON `t1`.`task_id` = " | ||
| "`tasks`.`id` WHERE `t1`.`task_id` = ? AND `tasks`.`timeout` < 0.0001 AND " | ||
| "TIMESTAMPDIFF(MICROSECOND, `t1`.`start_time`, CURRENT_TIMESTAMP()) < " | ||
| "`tasks`.`timeout` * 1000" | ||
| )); | ||
| not_timeout_statement->setBytes(1, &id_bytes); | ||
| std::unique_ptr<sql::ResultSet> const not_timeout_res(not_timeout_statement->executeQuery() | ||
| ); | ||
| bool const all_timeout = not_timeout_res->rowsCount() == 0; | ||
| if (!task_ready && !all_timeout) { | ||
| conn->rollback(); | ||
| return StorageErr{StorageErrType::OtherErr, "Task not ready or timed out"}; | ||
| } | ||
| // Set the state to running | ||
| std::unique_ptr<sql::PreparedStatement> const running_statement( | ||
| conn->prepareStatement("UPDATE `tasks` SET `state` = 'running' WHERE `id` = ?") | ||
| ); | ||
| running_statement->setBytes(1, &id_bytes); | ||
| running_statement->executeUpdate(); | ||
| // Insert task instance | ||
| std::unique_ptr<sql::PreparedStatement> const instance_statement(conn->prepareStatement( | ||
| "INSERT INTO `task_instances` (`id`, `task_id`, `start_time`) VALUES(?, ?, " | ||
| "CURRENT_TIMESTAMP())" | ||
| )); | ||
| sql::bytes instance_id_bytes = uuid_get_bytes(instance.id); | ||
| instance_statement->setBytes(1, &instance_id_bytes); | ||
| instance_statement->setBytes(2, &id_bytes); | ||
| instance_statement->executeUpdate(); | ||
| } catch (sql::SQLException& e) { | ||
| conn->rollback(); | ||
| return StorageErr{StorageErrType::OtherErr, e.what()}; | ||
| } | ||
|
|
||
| conn->commit(); | ||
| return StorageErr{}; | ||
| } | ||
|
|
There was a problem hiding this comment.
Address logical errors in create_task_instance method
There are potential logical issues in the create_task_instance method:
-
Inaccurate condition for task timeout check: In lines 1319-1321, the SQL query uses
AND \tasks`.`timeout` < 0.0001, which filters tasks with timeouts less than0.0001. This seems counterintuitive, as it effectively ignores tasks with valid timeouts. Consider changing the condition to> 0.0001` to include tasks with meaningful timeouts. -
Incorrect logical operator in condition: At line 1327, the condition
if (!task_ready && !all_timeout)may not correctly handle scenarios where either the task is not ready or not all instances have timed out. It might be more appropriate to use the logical OR operator||to ensure that the method only proceeds when the task is ready and all instances have timed out.
Apply this diff to correct the condition:
- if (!task_ready && !all_timeout) {
+ if (!task_ready || !all_timeout) {Please review and adjust the conditions to ensure correct task instance creation logic.
📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| auto MySqlMetadataStorage::create_task_instance(TaskInstance const& instance) -> StorageErr { | |
| std::variant<MySqlConnection, StorageErr> conn_result = MySqlConnection::create(m_url); | |
| if (std::holds_alternative<StorageErr>(conn_result)) { | |
| return std::get<StorageErr>(conn_result); | |
| } | |
| auto& conn = std::get<MySqlConnection>(conn_result); | |
| try { | |
| // Check the state of the task | |
| std::unique_ptr<sql::PreparedStatement> ready_statement(conn->prepareStatement( | |
| "SELECT `state` FROM `tasks` WHERE `id` = ? AND `state` = 'ready'" | |
| )); | |
| sql::bytes id_bytes = uuid_get_bytes(instance.task_id); | |
| ready_statement->setBytes(1, &id_bytes); | |
| std::unique_ptr<sql::ResultSet> const ready_res(ready_statement->executeQuery()); | |
| bool const task_ready = ready_res->rowsCount() > 0; | |
| // Check all task instances have timed out | |
| std::unique_ptr<sql::PreparedStatement> not_timeout_statement(conn->prepareStatement( | |
| "SELECT `t1`.`id` FROM `task_instances` as `t1` JOIN `tasks` ON `t1`.`task_id` = " | |
| "`tasks`.`id` WHERE `t1`.`task_id` = ? AND `tasks`.`timeout` < 0.0001 AND " | |
| "TIMESTAMPDIFF(MICROSECOND, `t1`.`start_time`, CURRENT_TIMESTAMP()) < " | |
| "`tasks`.`timeout` * 1000" | |
| )); | |
| not_timeout_statement->setBytes(1, &id_bytes); | |
| std::unique_ptr<sql::ResultSet> const not_timeout_res(not_timeout_statement->executeQuery() | |
| ); | |
| bool const all_timeout = not_timeout_res->rowsCount() == 0; | |
| if (!task_ready && !all_timeout) { | |
| conn->rollback(); | |
| return StorageErr{StorageErrType::OtherErr, "Task not ready or timed out"}; | |
| } | |
| // Set the state to running | |
| std::unique_ptr<sql::PreparedStatement> const running_statement( | |
| conn->prepareStatement("UPDATE `tasks` SET `state` = 'running' WHERE `id` = ?") | |
| ); | |
| running_statement->setBytes(1, &id_bytes); | |
| running_statement->executeUpdate(); | |
| // Insert task instance | |
| std::unique_ptr<sql::PreparedStatement> const instance_statement(conn->prepareStatement( | |
| "INSERT INTO `task_instances` (`id`, `task_id`, `start_time`) VALUES(?, ?, " | |
| "CURRENT_TIMESTAMP())" | |
| )); | |
| sql::bytes instance_id_bytes = uuid_get_bytes(instance.id); | |
| instance_statement->setBytes(1, &instance_id_bytes); | |
| instance_statement->setBytes(2, &id_bytes); | |
| instance_statement->executeUpdate(); | |
| } catch (sql::SQLException& e) { | |
| conn->rollback(); | |
| return StorageErr{StorageErrType::OtherErr, e.what()}; | |
| } | |
| conn->commit(); | |
| return StorageErr{}; | |
| } | |
| auto MySqlMetadataStorage::create_task_instance(TaskInstance const& instance) -> StorageErr { | |
| std::variant<MySqlConnection, StorageErr> conn_result = MySqlConnection::create(m_url); | |
| if (std::holds_alternative<StorageErr>(conn_result)) { | |
| return std::get<StorageErr>(conn_result); | |
| } | |
| auto& conn = std::get<MySqlConnection>(conn_result); | |
| try { | |
| // Check the state of the task | |
| std::unique_ptr<sql::PreparedStatement> ready_statement(conn->prepareStatement( | |
| "SELECT `state` FROM `tasks` WHERE `id` = ? AND `state` = 'ready'" | |
| )); | |
| sql::bytes id_bytes = uuid_get_bytes(instance.task_id); | |
| ready_statement->setBytes(1, &id_bytes); | |
| std::unique_ptr<sql::ResultSet> const ready_res(ready_statement->executeQuery()); | |
| bool const task_ready = ready_res->rowsCount() > 0; | |
| // Check all task instances have timed out | |
| std::unique_ptr<sql::PreparedStatement> not_timeout_statement(conn->prepareStatement( | |
| "SELECT `t1`.`id` FROM `task_instances` as `t1` JOIN `tasks` ON `t1`.`task_id` = " | |
| "`tasks`.`id` WHERE `t1`.`task_id` = ? AND `tasks`.`timeout` < 0.0001 AND " | |
| "TIMESTAMPDIFF(MICROSECOND, `t1`.`start_time`, CURRENT_TIMESTAMP()) < " | |
| "`tasks`.`timeout` * 1000" | |
| )); | |
| not_timeout_statement->setBytes(1, &id_bytes); | |
| std::unique_ptr<sql::ResultSet> const not_timeout_res(not_timeout_statement->executeQuery() | |
| ); | |
| bool const all_timeout = not_timeout_res->rowsCount() == 0; | |
| if (!task_ready || !all_timeout) { | |
| conn->rollback(); | |
| return StorageErr{StorageErrType::OtherErr, "Task not ready or timed out"}; | |
| } | |
| // Set the state to running | |
| std::unique_ptr<sql::PreparedStatement> const running_statement( | |
| conn->prepareStatement("UPDATE `tasks` SET `state` = 'running' WHERE `id` = ?") | |
| ); | |
| running_statement->setBytes(1, &id_bytes); | |
| running_statement->executeUpdate(); | |
| // Insert task instance | |
| std::unique_ptr<sql::PreparedStatement> const instance_statement(conn->prepareStatement( | |
| "INSERT INTO `task_instances` (`id`, `task_id`, `start_time`) VALUES(?, ?, " | |
| "CURRENT_TIMESTAMP())" | |
| )); | |
| sql::bytes instance_id_bytes = uuid_get_bytes(instance.id); | |
| instance_statement->setBytes(1, &instance_id_bytes); | |
| instance_statement->setBytes(2, &id_bytes); | |
| instance_statement->executeUpdate(); | |
| } catch (sql::SQLException& e) { | |
| conn->rollback(); | |
| return StorageErr{StorageErrType::OtherErr, e.what()}; | |
| } | |
| conn->commit(); | |
| return StorageErr{}; | |
| } |
There was a problem hiding this comment.
Actionable comments posted: 0
🧹 Nitpick comments (2)
src/spider/scheduler/SchedulerTaskCache.cpp (2)
54-58: Optimize task collection to improve performance.In
pop_next_task, you create a newstd::vector<core::Task>and copy tasks fromm_tasks, which may be inefficient for large task sets. Consider iterating overm_tasksdirectly or modifyingm_get_next_task_functionto accept an iterator or a different data structure to avoid unnecessary copying.
76-79: Simplify theshould_fetch_taskscondition.You can streamline the logic in
should_fetch_tasksby combining the conditions into a single return statement:-return m_update_count > cUpdateCount; +return (m_last_update + std::chrono::milliseconds(cUpdateInterval) < now) || (m_update_count > cUpdateCount);This enhances readability by reducing the number of return points.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (1)
src/spider/scheduler/SchedulerTaskCache.cpp(1 hunks)
🧰 Additional context used
🪛 cppcheck (2.10-2)
src/spider/scheduler/SchedulerTaskCache.cpp
[performance] 97-97: Variable 'id' is assigned in constructor body. Consider performing initialization in initialization list.
(useInitializationList)
⏰ Context from checks skipped due to timeout of 90000ms (3)
- GitHub Check: lint
- GitHub Check: non-storage-unit-tests (ubuntu-24.04)
- GitHub Check: non-storage-unit-tests (ubuntu-22.04)
🔇 Additional comments (3)
src/spider/scheduler/SchedulerTaskCache.cpp (3)
26-43: 🛠️ Refactor suggestionReview the multiple increments of
m_update_countinget_ready_task.The
m_update_countvariable is incremented multiple times withinget_ready_task, specifically at lines 33, 37, and 42. This could lead tom_update_countincreasing faster than expected, potentially causing the cache to refresh more frequently than intended. Please verify if this behaviour is intentional or adjust the increments accordingly.
86-88:⚠️ Potential issueEnsure uniqueness of task IDs when populating
m_tasks.When inserting tasks into
m_tasksfromget_ready_tasksandget_task_timeout, there is a possibility of duplicate task IDs, which could overwrite existing entries. Verify that this behaviour is intended, or implement a check to handle duplicates appropriately.Also applies to: 91-93
69-71:⚠️ Potential issueHandle potential exceptions when accessing
m_tasks.At line 69,
m_tasks.at(task_id.value())is used, which can throw an exception if thetask_idis not found. Although you've checked thattask_idexists, it's safer to usefindto prevent any unexpected exceptions:-core::Task task = m_tasks.at(task_id.value()); +m_tasks.erase(task_id.value()); +core::Task task = std::move(it->second); +m_tasks.erase(it);Likely invalid or redundant comment.
1 participant
Description
Scheduler fetches all ready tasks from storage on each schedule task request, which is inefficient. Now scheduler caches the tasks queried from storage for a short period or number of schedule requests before another storage query.
Validation performed
Summary by CodeRabbit
Release Notes
New Features
Improvements
Changes
Bug Fixes