feat: Add worker that request tasks from scheduler and execute tasks #38
Conversation
WalkthroughThis pull request introduces several enhancements to the Spider project, focusing on task management, error handling, and serialization across multiple components. Key modifications include updates to the metadata storage interface, MySQL storage implementation, function management, task execution, and worker client functionality. The changes improve error handling, logging, and flexibility in processing tasks and results, with a particular emphasis on supporting more complex serialization and result retrieval mechanisms. Changes
Possibly Related PRs
📜 Recent review detailsConfiguration used: CodeRabbit UI 📒 Files selected for processing (1)
🚧 Files skipped from review as they are similar to previous changes (1)
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 (
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
🔭 Outside diff range comments (2)
src/spider/worker/TaskExecutor.hpp (1)
Line range hint
38-74: Consider reducing code duplication and ensuring exception safety
- The constructor has complex logic (spawning processes, sending messages). If an exception arises (e.g., environment variables missing, etc.), ensuring the partial construction state is properly handled is crucial.
- The portion that builds arguments and calls create_args_request closely mirrors logic in the second constructor. Extracting a common helper method would reduce code repetition.
src/spider/CMakeLists.txt (1)
Line range hint
98-98: Remove duplicate StopToken.hpp inclusionOnce moved to core headers, this duplicate inclusion can be removed.
- utils/StopToken.hpp
🧹 Nitpick comments (17)
src/spider/worker/task_executor.cpp (3)
60-63: Consider externalizing logging configurationWhile the logging setup is comprehensive, consider these improvements:
- Move the logging pattern to a configuration file for easier maintenance
- Standardize the component name format (e.g., "spider.executor" instead of "spider][executor")
- spdlog::set_pattern("[%Y-%m-%d %H:%M:%S.%e] [%^%l%$] [spider][executor] %v"); + spdlog::set_pattern("[%Y-%m-%d %H:%M:%S.%e] [%^%l%$] [spider.executor] %v");
89-90: Enhance debug logging with task contextConsider adding more context to help with debugging, such as the libraries loaded and any relevant task metadata.
- spdlog::debug("Function to run: {}", func_name); + spdlog::debug("Executing task - function: {}, loaded libraries: {}", func_name, fmt::join(libs, ", "));
Line range hint
92-133: Add task execution safeguardsThe task execution could benefit from additional safety measures:
- Implement a timeout mechanism for long-running tasks
- Add memory limits for result buffers
- Consider adding task cancellation support
Would you like me to provide an example implementation of these safeguards?
src/spider/worker/TaskExecutor.hpp (2)
76-109: Duplicate code for process initialization
The constructor replicates the same overall logic as the variadic template constructor, differing mainly in the way arguments are prepared. Consider consolidating the duplicated process setup (m_process, co_spawn, etc.) into a shared function.
133-133: Returning potentially large result buffers
get_result_buffers() returns a copy of the underlying vector. If these sbuffers can be large, consider returning a reference or a smart pointer to avoid extra copies. However, if performance usage is minimal, this approach is acceptable.src/spider/worker/FunctionManager.hpp (1)
Line range hint
199-218: create_result_response enhancements
The extension to handle tuples is cleanly implemented. Similarly to the tuple-based get_result approach, consider the readability and test coverage for multi-field returns.src/spider/worker/worker.cpp (3)
2-75: parse_args and environment setup
The parse_args function is straightforward. If the application can handle optional parameters or defaults, consider descriptive usage text in the --help option for better clarity.
121-132: fetch_task infinite loop
fetch_task uses a while(true) with a sleep of cFetchTaskTimeout. Validate that an exceptionally long or permanent scheduler outage will not cause side effects. You may consider a backoff or a maximum wait limit.
194-289: task_loop concurrency & error handling
- The approach is robust in marking tasks as running/fail/success.
- The while(!stop_requested()) loop could lead to indefinite waits if the scheduler stops sending tasks but does not trigger a stop_token. Consider a graceful shutdown mechanism.
- The metadata_store->task_fail call transitions tasks to 'fail' only after an instance. Ensure that repeated partial failures are covered if the system restarts.
src/spider/storage/MysqlStorage.cpp (2)
11-11: Potential large includes
Consider whether is the most efficient approach for URL parsing. If performance is an issue, a manual parse or a simpler library might be more efficient; if not, this is fine.
246-260: Regex-based JDBC URL parsing
This block checks user/password from the URL. Good approach. Consider adding more robust checking for edge cases, e.g., URL with multiple parameters.tests/worker/worker-test.cpp (1)
8-8: Consider using a more test-appropriate logging approach.Writing to standard error during tests could interfere with test runners and continuous integration systems. Consider these alternatives:
- Use Catch2's built-in
INFOmacro for test-specific logging- Create a dedicated test logger that can be controlled via test configuration
- std::cerr << x << " + " << y << " = " << x + y << "\n"; + INFO("Computing sum: " << x << " + " << y << " = " << x + y);src/spider/worker/WorkerClient.cpp (1)
73-74: Consider connection pooling for better performance.Creating a new
io_contextand socket for each task request could impact performance, especially under high load. Consider implementing a connection pool to reuse connections to schedulers.tests/worker/test-FunctionManager.cpp (1)
34-38: Improve test readability with constant and explicit serializationThe introduction of
cExpectedand explicit msgpack serialization enhances test clarity and maintainability. However, consider extracting the msgpack serialization into a test helper function for reuse across test cases.+namespace { +template<typename T> +auto serialize_expected(T const& value) -> msgpack::sbuffer { + msgpack::sbuffer buffer{}; + msgpack::pack(buffer, value); + return buffer; +} +} // namespace constexpr int cExpected = 2 + 3; msgpack::sbuffer const result = (*function)(args_buffers); -msgpack::sbuffer buffer{}; -msgpack::pack(buffer, cExpected); +auto const expected_buffer = serialize_expected(cExpected);tests/scheduler/test-SchedulerServer.cpp (1)
28-29: Consider documenting the rationale for the warmup time valueThe 5ms warmup time constant helps address server connection failures, but it would be beneficial to document why this specific duration was chosen.
constexpr int cServerWarmupTime = 5; +// Allow 5ms for server to stabilize after state changes to prevent connection failuressrc/spider/worker/TaskExecutor.cpp (2)
55-60: Consider adding timeout to condition variable waitWhile the synchronization is correct, consider adding a timeout to prevent indefinite waiting.
- m_complete_cv.wait(lock, [this] { + if (!m_complete_cv.wait_for(lock, std::chrono::seconds(30), [this] { return TaskExecutorState::Succeed == m_state || TaskExecutorState::Error == m_state || TaskExecutorState::Cancelled == m_state; - }); + })) { + m_state = TaskExecutorState::Error; + core::create_error_buffer( + core::FunctionInvokeError::FunctionExecutionError, + "Task execution timed out", + m_result_buffer + ); + }
91-96: Ensure consistent state transition patternsThe state transitions and notifications follow a consistent pattern across different response types. Consider extracting the common pattern into a helper method to reduce code duplication.
+ private: + void transition_state(TaskExecutorState new_state, const msgpack::sbuffer& response) { + { + std::lock_guard const lock(m_state_mutex); + m_state = new_state; + m_result_buffer.write(response.data(), response.size()); + } + m_complete_cv.notify_all(); + }Also applies to: 102-107, 111-116
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (15)
src/spider/CMakeLists.txt(1 hunks)src/spider/storage/MetadataStorage.hpp(1 hunks)src/spider/storage/MysqlStorage.cpp(5 hunks)src/spider/storage/MysqlStorage.hpp(1 hunks)src/spider/worker/FunctionManager.hpp(5 hunks)src/spider/worker/TaskExecutor.cpp(5 hunks)src/spider/worker/TaskExecutor.hpp(4 hunks)src/spider/worker/WorkerClient.cpp(2 hunks)src/spider/worker/WorkerClient.hpp(0 hunks)src/spider/worker/task_executor.cpp(4 hunks)src/spider/worker/worker.cpp(1 hunks)tests/scheduler/test-SchedulerServer.cpp(3 hunks)tests/worker/test-FunctionManager.cpp(2 hunks)tests/worker/test-MessagePipe.cpp(1 hunks)tests/worker/worker-test.cpp(1 hunks)
💤 Files with no reviewable changes (1)
- src/spider/worker/WorkerClient.hpp
🧰 Additional context used
🪛 cppcheck (2.10-2)
src/spider/worker/worker.cpp
[performance] 88-88: Variable 'id' is assigned in constructor body. Consider performing initialization in initialization list.
(useInitializationList)
[performance] 107-107: Variable 'm_id' is assigned in constructor body. Consider performing initialization in initialization list.
(useInitializationList)
🔇 Additional comments (22)
src/spider/worker/TaskExecutor.hpp (3)
4-4: Include looks fine
No concerns here; the inclusion of <condition_variable> is appropriate for the synchronization need.
129-129: Ensure thread-safe read of result buffer
Marking get_result() as const is a good addition, but confirm that m_result_buffer access is safe if called concurrently while the task is still running. If concurrency could occur, consider locking or clarifying usage expectations for calling this method.
141-141: m_complete_cv declared but usage not shown
Ensure that m_complete_cv is properly used—paired with a unique_lock on m_state_mutex and wait/notify calls. Otherwise, it may be superfluous.
src/spider/worker/FunctionManager.hpp (4)
Line range hint 108-125: Check exception handling
The function response_get_result() appropriately handles msgpack::type_error exceptions, returning nullopt. Verify that your calling code logs these issues or otherwise handles unexpected null returns gracefully.
132-159: Flexible tuple deserialization
The multiple-type tuple support is a solid addition. The for_n helper is well utilized to unpack the data. Keep in mind that large tuples can degrade readability; ensure call sites handle these results carefully.
161-197: Use of spdlog::error is helpful
Logging errors for incorrect response types or sizes is excellent for debugging. Double-check that these logs do not risk exposing sensitive information in production.
240-251: Concatenate argument buffers carefully
Creating a single msgpack array from many buffers is powerful. However, ensure all call sites provide valid, non-empty buffers, as any partial or empty buffer would be directly appended. Safeguard or clarify usage.
src/spider/worker/worker.cpp (4)
77-119: heartbeat_loop's retry mechanism
The fail_count logic is clear. Just ensure that failing the heartbeat due to transient conditions won't repeatedly kill the worker. Currently, cRetryCount is 5, which seems reasonable.
🧰 Tools
🪛 cppcheck (2.10-2)
[performance] 88-88: Variable 'id' is assigned in constructor body. Consider performing initialization in initialization list.
(useInitializationList)
[performance] 107-107: Variable 'm_id' is assigned in constructor body. Consider performing initialization in initialization list.
(useInitializationList)
134-160: Preparation of argument buffers
The logic in get_args_buffers is direct but be cautious with raw string writes for large or binary data. If inputs can be large or embedded with null bytes, consider a more specialized encoding rather than .write(value.data(), value.size()).
162-192: parse_outputs handles multiple returns
Good approach for processing each result buffer. For Data type parsing, watch for potential mismatch between expected and actual binary content in result_buffers[i].
293-384: main function
The registration of the worker, creation of threads for heartbeat and tasks, and final join is a standard approach. Make sure the worker receives a graceful cancellation signal so that threads exit cleanly.
src/spider/storage/MysqlStorage.cpp (2)
1088-1123: task_fail method
- Removing the task instance first, then conditionally marking the task as fail is a good design for concurrency.
- Logging with spdlog::error is helpful. If a partial DB rollback is needed, confirm the rollback state if an exception occurs mid-logic.
1314-1329: MySqlDataStorage::connect
Again, your approach matches the MySqlMetadataStorage connect method. Ensure that connecting without credentials is safe and that the pointer is always set if the connection is successful.
tests/worker/test-MessagePipe.cpp (1)
47-47: LGTM! Improved type safety in response handling.
The change from tuple-based to individual type parameters provides better type safety and clearer intent.
src/spider/worker/WorkerClient.cpp (2)
96-98: LGTM! Robust task ID validation.
The added null check for task ID improves the robustness of the response handling.
73-74: Verify existing connection management patterns.
Let's check if there are similar connection management patterns in the codebase that we could leverage.
✅ Verification successful
Let me gather more specific information about the connection management patterns.
Let me check the actual implementation of SchedulerServer and WorkerClient to understand the connection management patterns better.
The current socket management approach is appropriate for this use case.
The local context and socket creation in WorkerClient is correct because:
- The WorkerClient establishes short-lived connections to schedulers for task scheduling
- Each connection is used for a single request-response cycle and then closed
- The SchedulerServer also follows this pattern by accepting connections and processing them independently
The code follows a consistent pattern across the codebase where transient connections are used for individual message exchanges, rather than maintaining long-lived connections or connection pools.
🏁 Scripts executed
The following scripts were executed for the analysis:
Script:
#!/bin/bash
# Search for connection pooling or socket management patterns
echo "Searching for connection pooling patterns..."
rg -l "io_context|tcp::socket" --type cpp
echo "Searching for socket reuse patterns..."
ast-grep --pattern 'class $_ {
$$$
boost::asio::io_context
$$$
}'
Length of output: 759
Script:
#!/bin/bash
# Check BoostAsio.hpp for any connection management patterns
echo "Checking BoostAsio.hpp..."
cat src/spider/io/BoostAsio.hpp
# Check TaskExecutor implementation for connection handling
echo -e "\nChecking TaskExecutor implementation..."
cat src/spider/worker/TaskExecutor.hpp
# Check SchedulerServer for connection management
echo -e "\nChecking SchedulerServer implementation..."
cat src/spider/scheduler/SchedulerServer.hpp
Length of output: 9488
Script:
#!/bin/bash
# Check SchedulerServer and WorkerClient implementations
echo "Checking SchedulerServer.cpp..."
cat src/spider/scheduler/SchedulerServer.cpp
echo -e "\nChecking WorkerClient.cpp..."
cat src/spider/worker/WorkerClient.cpp
Length of output: 8827
tests/worker/test-FunctionManager.cpp (1)
70-73: LGTM: Improved type safety in tuple result handling
The change to use separate template parameters for response_get_result improves type safety by making the expected types explicit.
src/spider/storage/MetadataStorage.hpp (1)
53-54: LGTM: Well-designed task failure handling interface
The new task_fail virtual method provides a clean interface for handling task failures, maintaining consistency with existing error handling patterns. The method appropriately takes both the task instance and error message as parameters.
tests/scheduler/test-SchedulerServer.cpp (1)
57-58: Verify if the warmup time is sufficient across different environments
While the warmup time helps prevent connection failures, consider implementing a more robust solution:
- Add logging when connections fail
- Consider implementing a retry mechanism with exponential backoff
Also applies to: 86-87
✅ Verification successful
The previous git log command wasn't suitable as it's resource-intensive. Let's examine the current codebase state to understand the connection handling and warmup time implementation.
The current warmup time implementation appears adequate for test scenarios
The codebase shows:
- A constant warmup time of 5ms is used consistently across test cases
- No historical connection failures or errors are evident in the test file
- The connection is made using boost::asio, which has built-in error handling
- This is a test-only concern and doesn't affect production code
Since this is specifically for test scenarios and there's no evidence of connection failures, the current implementation is sufficient. Adding retry mechanisms would unnecessarily complicate the test code.
🏁 Scripts executed
The following scripts were executed for the analysis:
Script:
#!/bin/bash
# Check for any existing connection-related test failures
git log -p | rg -B 5 -A 5 "connection.*fail|connect.*error" tests/scheduler/test-SchedulerServer.cpp
Length of output: 0
Script:
#!/bin/bash
# Search for the warmup time constant and its usage
rg -A 2 "cServerWarmupTime" --type cpp
# Look for any test failures or connection issues in test logs
rg "fail|error|connect" tests/scheduler/test-SchedulerServer.cpp
# Find any retry mechanisms or connection handling patterns
ast-grep --pattern 'connect($$$)' --lang cpp
Length of output: 889
src/spider/worker/TaskExecutor.cpp (2)
22-23: LGTM: Proper handling of cancelled state
The addition of TaskExecutorState::Cancelled to completion check ensures proper task lifecycle management.
127-129: LGTM: Clean implementation of result buffer retrieval
The new method properly delegates to the core functionality while maintaining const correctness.
src/spider/storage/MysqlStorage.hpp (1)
57-57: LGTM: Clean interface extension for task failure handling
The addition of task_fail method properly extends the storage interface to handle task failure scenarios, which is essential for robust task execution in the worker implementation.
✅ Verification successful
Verified: Task failure handling is properly implemented across storage layers
The task_fail method is correctly declared as a pure virtual function in the MetadataStorage base class and properly overridden in MysqlStorage. The method signatures match exactly, ensuring consistent task failure handling across the storage implementation.
🏁 Scripts executed
The following scripts were executed for the analysis:
Script:
#!/bin/bash
# Verify consistent implementation across storage types
ast-grep --pattern 'class $_ : public MetadataStorage {
$$$
auto task_fail($_, $_) -> StorageErr override;
$$$
}'
Length of output: 121
Script:
#!/bin/bash
# First, let's find all storage implementation files
fd "Storage\.(hpp|cpp)$"
# Then, let's check for the base class definition
rg "class\s+MetadataStorage" -A 5
# Finally, let's search for task_fail method declarations
rg "task_fail.*StorageErr" -A 1
Length of output: 1315
| @@ -105,6 +111,7 @@ auto main(int const argc, char** argv) -> int { | |||
| msgpack::sbuffer args_buffer; | |||
| msgpack::packer packer{args_buffer}; | |||
| packer.pack(args_object); | |||
| spdlog::debug("Args buffer parsed"); | |||
There was a problem hiding this comment.
🛠️ Refactor suggestion
Enhance execution logging for better observability
The current execution logging could be more comprehensive to aid in debugging and monitoring:
- Add execution timing
- Log result buffer size
- Add more detailed error logging
+ auto start_time = std::chrono::steady_clock::now();
spdlog::debug("Args buffer parsed");
// ... execution ...
- spdlog::debug("Function executed");
+ auto end_time = std::chrono::steady_clock::now();
+ auto duration = std::chrono::duration_cast<std::chrono::milliseconds>(end_time - start_time);
+ spdlog::debug("Function executed - duration: {}ms, result size: {} bytes",
+ duration.count(), result_buffer.size());Also applies to: 130-130
| @@ -51,6 +51,7 @@ set(SPIDER_WORKER_SOURCES | |||
| worker/message_pipe.hpp | |||
| worker/WorkerClient.hpp | |||
| worker/WorkerClient.cpp | |||
| utils/StopToken.hpp | |||
There was a problem hiding this comment.
🛠️ Refactor suggestion
Consider moving shared utilities to core component
StopToken.hpp is used by both worker and scheduler components, suggesting it's a core utility. Consider moving it to the SPIDER_CORE_HEADERS list to avoid duplication and better reflect its shared nature.
- utils/StopToken.hppAdd to SPIDER_CORE_HEADERS:
set(SPIDER_CORE_HEADERS
core/Error.hpp
core/Data.hpp
+ utils/StopToken.hpp
...
)Committable suggestion skipped: line range outside the PR's diff.
There was a problem hiding this comment.
Actionable comments posted: 3
🧹 Nitpick comments (2)
src/spider/worker/worker.cpp (2)
50-50: Consider making retry count configurableThe hardcoded retry count might not be suitable for all deployment environments. Consider making this configurable through command-line arguments or environment variables.
-constexpr int cRetryCount = 5; +// Read from environment variable with a default value +constexpr int cRetryCount = []() { + if (const char* env = std::getenv("SPIDER_RETRY_COUNT")) { + return std::stoi(env); + } + return 5; +}();
362-362: Consider periodic refresh of environment variablesEnvironment variables are only fetched once at startup. If they change during runtime (e.g., PATH updates), these changes won't be reflected.
Consider implementing a periodic refresh mechanism or watching for environment changes if this is a requirement for your deployment environment.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (3)
src/spider/scheduler/scheduler.cpp(1 hunks)src/spider/worker/task_executor.cpp(4 hunks)src/spider/worker/worker.cpp(1 hunks)
✅ Files skipped from review due to trivial changes (1)
- src/spider/scheduler/scheduler.cpp
🚧 Files skipped from review as they are similar to previous changes (1)
- src/spider/worker/task_executor.cpp
🧰 Additional context used
🪛 cppcheck (2.10-2)
src/spider/worker/worker.cpp
[performance] 88-88: Variable 'id' is assigned in constructor body. Consider performing initialization in initialization list.
(useInitializationList)
[performance] 107-107: Variable 'm_id' is assigned in constructor body. Consider performing initialization in initialization list.
(useInitializationList)
| auto fetch_task(spider::worker::WorkerClient& client) -> boost::uuids::uuid { | ||
| spdlog::debug("Fetching task"); | ||
| while (true) { | ||
| std::optional<boost::uuids::uuid> const optional_task_id = client.get_next_task(); | ||
| if (optional_task_id.has_value()) { | ||
| return optional_task_id.value(); | ||
| } | ||
| std::this_thread::sleep_for(std::chrono::milliseconds(cFetchTaskTimeout)); | ||
| } | ||
| } |
There was a problem hiding this comment.
Add timeout handling to fetch_task
The infinite loop in fetch_task could potentially block indefinitely. Consider adding a maximum retry count or timeout mechanism.
-auto fetch_task(spider::worker::WorkerClient& client) -> boost::uuids::uuid {
+auto fetch_task(
+ spider::worker::WorkerClient& client,
+ std::chrono::seconds timeout = std::chrono::seconds(300)
+) -> std::optional<boost::uuids::uuid> {
spdlog::debug("Fetching task");
+ auto start_time = std::chrono::steady_clock::now();
while (true) {
std::optional<boost::uuids::uuid> const optional_task_id = client.get_next_task();
if (optional_task_id.has_value()) {
return optional_task_id.value();
}
+ if (std::chrono::steady_clock::now() - start_time > timeout) {
+ spdlog::warn("Timeout while fetching task");
+ return std::nullopt;
+ }
std::this_thread::sleep_for(std::chrono::milliseconds(cFetchTaskTimeout));
}
}📝 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 fetch_task(spider::worker::WorkerClient& client) -> boost::uuids::uuid { | |
| spdlog::debug("Fetching task"); | |
| while (true) { | |
| std::optional<boost::uuids::uuid> const optional_task_id = client.get_next_task(); | |
| if (optional_task_id.has_value()) { | |
| return optional_task_id.value(); | |
| } | |
| std::this_thread::sleep_for(std::chrono::milliseconds(cFetchTaskTimeout)); | |
| } | |
| } | |
| auto fetch_task( | |
| spider::worker::WorkerClient& client, | |
| std::chrono::seconds timeout = std::chrono::seconds(300) | |
| ) -> std::optional<boost::uuids::uuid> { | |
| spdlog::debug("Fetching task"); | |
| auto start_time = std::chrono::steady_clock::now(); | |
| while (true) { | |
| std::optional<boost::uuids::uuid> const optional_task_id = client.get_next_task(); | |
| if (optional_task_id.has_value()) { | |
| return optional_task_id.value(); | |
| } | |
| if (std::chrono::steady_clock::now() - start_time > timeout) { | |
| spdlog::warn("Timeout while fetching task"); | |
| return std::nullopt; | |
| } | |
| std::this_thread::sleep_for(std::chrono::milliseconds(cFetchTaskTimeout)); | |
| } | |
| } |
| return cStorageErr; | ||
| } | ||
|
|
||
| spider::core::StopToken stop_token; |
There was a problem hiding this comment.
Add signal handling for graceful shutdown
The application should handle SIGTERM/SIGINT signals to ensure graceful shutdown of running tasks.
+ std::signal(SIGTERM, [](int) {
+ spdlog::info("Received SIGTERM, initiating graceful shutdown");
+ stop_token.request_stop();
+ });
+ std::signal(SIGINT, [](int) {
+ spdlog::info("Received SIGINT, initiating graceful shutdown");
+ stop_token.request_stop();
+ });Committable suggestion skipped: line range outside the PR's diff.
| context.run(); | ||
| executor.wait(); |
There was a problem hiding this comment.
Add timeout handling to task execution
The blocking wait on task execution could potentially hang indefinitely. Consider adding a timeout mechanism.
- context.run();
- executor.wait();
+ auto future = std::async(std::launch::async, [&]() {
+ context.run();
+ executor.wait();
+ });
+
+ if (future.wait_for(std::chrono::minutes(30)) == std::future_status::timeout) {
+ spdlog::error("Task execution timed out");
+ metadata_store->task_fail(
+ instance,
+ fmt::format("Task {} timed out", task.get_function_name())
+ );
+ continue;
+ }Committable suggestion skipped: line range outside the PR's diff.
Description
As title. Fix an infrequent bug in scheduler server unit test that connection fails by sleeping after server restart and add more server restart.
Validation performed
Summary by CodeRabbit
New Features
Bug Fixes
Documentation
Tests