feat: Add scheduler that schedules tasks and garbage collect periodically - #37
Conversation
…e cleanup and send heartbeat periodically
WalkthroughThe pull request introduces several significant changes to the Spider project, focusing on enhancing the scheduler's functionality and control flow. A new Changes
Sequence DiagramsequenceDiagram
participant Main as Main Application
participant Server as SchedulerServer
participant Heartbeat as Heartbeat Loop
participant Cleanup as Cleanup Loop
participant Storage as Metadata/Data Storage
Main->>Server: Initialize with StopToken
Main->>Heartbeat: Start Heartbeat Loop
Main->>Cleanup: Start Cleanup Loop
loop Periodic Heartbeat
Heartbeat->>Storage: Update Heartbeat
alt Update Fails
Heartbeat->>Main: Request Stop
end
end
loop Periodic Cleanup
Cleanup->>Server: Pause Server
Cleanup->>Storage: Execute Cleanup Policy
Cleanup->>Server: Resume Server
alt Cleanup Fails
Cleanup->>Main: Request Stop
end
end
Main->>Server: Manage Server State
This sequence diagram illustrates the new control flow introduced in the scheduler, highlighting the interaction between the main application, server, heartbeat loop, cleanup loop, and storage components, with the Tip CodeRabbit's docstrings feature is now available as part of our Early Access Program! Simply use the command 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: 0
🧹 Nitpick comments (11)
src/spider/scheduler/SchedulerServer.cpp (2)
38-41: Safeguard concurrency in constructor thread creation
Spawning a new thread in the constructor can make debugging more difficult if exceptions are thrown outside. Ensure that any operations that might throw are completed prior to constructing a new thread. Otherwise, consider using a dedicated method to start the thread after construction is complete.
75-97: Handle persistent failure in acceptor
If repeated acceptance errors occur (e.g. ephemeral failure) or frequent exceptions are raised, consider implementing reconnection/backoff logic rather than immediately stopping the entire server.🧰 Tools
🪛 cppcheck (2.10-2)
[performance] 88-88: Variable 'id' is assigned in constructor body. Consider performing initialization in initialization list.
(useInitializationList)
src/spider/scheduler/scheduler.cpp (4)
42-64: Validate command-line arguments
The parse_args function extracts port and storage_url. Confirm that the help message or usage instructions are displayed if these parameters are absent, as indicated in the code. You might consider providing default values or more robust error messages for a better user experience.
66-87: Heartbeat loop stoppage threshold
fail_count triggers a stop request after cRetryCount - 1. This offers limited attempts to recover from transitory errors. Adjust cRetryCount or add an exponential backoff if transient network issues may inadvertently stop the scheduler too frequently.
89-124: Cleanup loop logic
Pausing and resuming the server is a neat approach. However, ensure that no new tasks are accepted during the pause that might ultimately require resources that are being cleaned. Also, consider logging the cleanup outcome (number of cleaned items/number of dangling data removed) for diagnostic purposes.🧰 Tools
🪛 cppcheck (2.10-2)
[performance] 107-107: Variable 'm_id' is assigned in constructor body. Consider performing initialization in initialization list.
(useInitializationList)
127-237: Main function thread management
Both the heartbeat and cleanup loops join at lines 229 and 230, then the server stops. If an exception is thrown before the joins, the threads might be left running. Consider wrapping these threads in RAII structures or adding more robust exception handling to guarantee threads are joined and the server is stopped.src/spider/utils/StopToken.hpp (1)
9-15: Consider additional methods or conditions
stop_requested returns only a single boolean. If you foresee needing multiple reasons to stop or partial shutdown modes, you might generalize in the future (e.g., store an integer state or error code).src/spider/scheduler/SchedulerServer.hpp (2)
29-30: Constructor parameter clarity
Adding the stop token parameter helps unify the lifecycle management. Update any inline documentation or comments to reflect why the constructor now requires a reference to core::StopToken.
51-53: Avoid referencing StopToken directly as a non-const reference
While it is valid, be aware that other code might inadvertently reset or mutate StopToken. If you prefer an immutable contract, consider storing std::shared_ptrcore::StopToken instead, so that ownership and references are more explicit.src/spider/io/BoostAsio.hpp (1)
48-71: Consider enhancing the get_address() implementationWhile the implementation is generally sound, consider these improvements:
- Add a timeout to the resolver to prevent hanging
- Consider adding IPv6 support for future-proofing
- Extract the warning message to a constant
Here's a suggested improvement:
inline auto get_address() -> std::optional<std::string> { + static constexpr auto RESOLVER_TIMEOUT = std::chrono::seconds(5); + static constexpr auto WARNING_MSG = "No non-loopback address found, using loopback address"; try { boost::asio::io_context io_context; boost::asio::ip::tcp::resolver resolver(io_context); + auto timer = boost::asio::steady_timer(io_context, RESOLVER_TIMEOUT); + timer.async_wait([&io_context](auto const& ec) { + if (!ec) { io_context.stop(); } + }); auto const endpoints = resolver.resolve(boost::asio::ip::host_name(), ""); for (auto const& endpoint : endpoints) { if (endpoint.endpoint().address().is_v4() && !endpoint.endpoint().address().is_loopback()) { return endpoint.endpoint().address().to_string(); } } - spdlog::warn("No non-loopback address found, using loopback address"); + spdlog::warn(WARNING_MSG); for (auto const& endpoint : endpoints) { if (endpoint.endpoint().address().is_v4()) { return endpoint.endpoint().address().to_string(); } } return std::nullopt; } catch (boost::system::system_error const& e) { + spdlog::error("Failed to resolve address: {}", e.what()); return std::nullopt; } }tests/scheduler/test-SchedulerServer.cpp (1)
51-53: Consider enhancing lifecycle test coverageWhile the basic pause/resume operations are tested, consider adding test cases for:
- Multiple pause/resume cycles
- Verifying operations are blocked during paused state
- Error handling during state transitions
Example test cases to add:
// Test multiple pause/resume cycles server.pause(); server.resume(); server.pause(); server.resume(); // Test operations during paused state server.pause(); // Attempt operations, verify they're blocked server.resume(); // Test error handling server.pause(); server.pause(); // Should handle double pause gracefully server.resume();
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (8)
src/spider/CMakeLists.txt(2 hunks)src/spider/io/BoostAsio.hpp(3 hunks)src/spider/io/Serializer.hpp(1 hunks)src/spider/scheduler/SchedulerServer.cpp(3 hunks)src/spider/scheduler/SchedulerServer.hpp(3 hunks)src/spider/scheduler/scheduler.cpp(1 hunks)src/spider/utils/StopToken.hpp(1 hunks)tests/scheduler/test-SchedulerServer.cpp(2 hunks)
🧰 Additional context used
🪛 cppcheck (2.10-2)
src/spider/scheduler/SchedulerServer.cpp
[performance] 88-88: Variable 'id' is assigned in constructor body. Consider performing initialization in initialization list.
(useInitializationList)
src/spider/scheduler/scheduler.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 (10)
src/spider/scheduler/SchedulerServer.cpp (2)
43-50: Pause method concurrency
The pause method correctly stops the io_context and joins the thread. However, confirm that no new asynchronous operations are posted after the context is stopped, as that might cause race conditions or logic errors. Consider documenting the concurrency assumptions explicitly.
30-31: Ensure consistent handling of the new stop token parameter
The addition of the stop token parameter is good for clean shutdown logic. However, ensure that all instantiations of SchedulerServer have been updated accordingly elsewhere in the codebase. If any older constructor calls remain, they may cause a compile-time error.
✅ Verification successful
Let me search for actual instantiations of SchedulerServer to verify if there are any constructor calls that need updating.
Let me try one more search strategy to find any instantiations of SchedulerServer.
Let me check the test file and scheduler.cpp which might contain instantiations.
No issues found with stop_token parameter usage
Based on the code examination, all instantiations of SchedulerServer properly include the stop_token parameter. The only two instances found are in the test file and main scheduler implementation, both correctly using the new constructor signature with all required parameters including the stop_token.
🏁 Scripts executed
The following scripts were executed for the analysis:
Script:
#!/bin/bash
# Searching for all instantiations of SchedulerServer to verify constructor calls
rg -A 5 $'SchedulerServer\\('
Length of output: 1603
Script:
#!/bin/bash
# Search for actual instantiations of SchedulerServer using AST pattern matching
ast-grep --pattern 'new SchedulerServer($$$)'
# Also search for variable declarations that might instantiate SchedulerServer
ast-grep --pattern 'SchedulerServer $_ = $$$'
ast-grep --pattern 'auto $_ = SchedulerServer($$$)'
# Backup with ripgrep to catch any other potential instantiations
rg -B 2 -A 2 'SchedulerServer\s+\w+\s*\('
rg -B 2 -A 2 'new\s+SchedulerServer\s*\('
Length of output: 237
Script:
#!/bin/bash
# Search for files that might contain SchedulerServer instantiations
fd -e cpp -e hpp -e cc -e h | xargs rg -l "SchedulerServer"
# Then examine the content of these files
fd -e cpp -e hpp -e cc -e h | xargs rg -B 5 -A 5 "std::make_unique<SchedulerServer>|std::make_shared<SchedulerServer>|SchedulerServer\s*\w+\s*{"
Length of output: 614
Script:
#!/bin/bash
# Check the test file and scheduler.cpp for instantiations
cat tests/scheduler/test-SchedulerServer.cpp
cat src/spider/scheduler/scheduler.cpp
Length of output: 12683
src/spider/utils/StopToken.hpp (1)
7-20: Thread-safe stop mechanism
This new class offers a straightforward approach to stopping threads. Good use of std::atomic ensures there is no data race on stop token checks.
src/spider/scheduler/SchedulerServer.hpp (1)
33-34: New pause/resume methods
These methods simplify controlling io_context without externally manipulating threads. This approach improves readability by clearly separating the lifecycle states of the server.
src/spider/io/Serializer.hpp (2)
44-48: Well-designed template concept for flexible serialization
The new Packable concept provides a more flexible approach by parameterizing both the buffer type and the serializable type. This design enables better type safety and extensibility.
51-53: Verify backward compatibility of SerializableImpl changes
The redefinition of SerializableImpl to use Packable with msgpack::sbuffer is a good abstraction. However, we should verify existing implementations aren't affected.
✅ Verification successful
SerializableImpl changes are safe to proceed
The redefinition of SerializableImpl using Packable<msgpack::sbuffer, T> is backward compatible based on the following evidence:
SerializableImplis only used internally withinSerializer.hppto define the higher-levelSerializableconcept- The only direct usage of
Serializableis inTaskIoconcept insrc/spider/client/task.hpp - All existing implementations already use
msgpack::packwithmsgpack::sbufferas shown in the codebase scan - The change formalizes the existing pattern that's already in use throughout the codebase
🏁 Scripts executed
The following scripts were executed for the analysis:
Script:
#!/bin/bash
# Search for existing SerializableImpl implementations
rg -l "SerializableImpl" --type cpp --type hpp
Length of output: 74
Script:
#!/bin/bash
# Search for SerializableImpl usage with ripgrep, including all text files
rg -l "SerializableImpl"
# Also search for any Packable usage to understand impact
rg -l "Packable"
Length of output: 99
Script:
#!/bin/bash
# Get full context around SerializableImpl and Packable usage
rg "SerializableImpl|Packable" -C 5 src/spider/io/Serializer.hpp
Length of output: 555
Script:
#!/bin/bash
# Search for any usage of Serializable concept to understand impact on implementations
rg "Serializable\s*<" -C 3
# Also look for any msgpack::pack usage to find actual implementations
rg "msgpack::pack" -C 3
Length of output: 7756
src/spider/CMakeLists.txt (2)
102-102: LGTM: StopToken.hpp addition
The addition of StopToken.hpp aligns with the scheduler enhancement objectives.
114-114: Verify program_options usage
The addition of Boost::program_options is appropriate. Let's verify its usage in the scheduler component.
✅ Verification successful
Boost::program_options is correctly used in the scheduler component
The scheduler component properly includes and utilizes Boost.Program_options through:
- Complete set of necessary header inclusions
- Proper command-line argument parsing implementation in
parse_args - Appropriate error handling for program options parsing
- Consistent usage of program_options types and functions
🏁 Scripts executed
The following scripts were executed for the analysis:
Script:
#!/bin/bash
# Search for program_options usage in scheduler files
rg "program_options" src/spider/scheduler/
Length of output: 1414
tests/scheduler/test-SchedulerServer.cpp (2)
24-24: LGTM: Header inclusion is properly placed
The StopToken header is correctly included with other project headers.
47-49: LGTM: Proper initialization order and RAII compliance
The StopToken is correctly instantiated before server creation and passed to the constructor.
Let's verify the constructor signature consistency:
✅ Verification successful
Constructor signature matches the test initialization
The constructor signature in src/spider/scheduler/SchedulerServer.cpp perfectly aligns with the test initialization, confirming proper parameter ordering:
- port
- policy
- metadata_store
- data_store
- stop_token (passed by reference)
🏁 Scripts executed
The following scripts were executed for the analysis:
Script:
#!/bin/bash
# Description: Verify SchedulerServer constructor signature across codebase
ast-grep --pattern 'SchedulerServer($$$, stop_token)'
Length of output: 55
Script:
#!/bin/bash
# Search for SchedulerServer constructor definition
ast-grep --pattern 'SchedulerServer::SchedulerServer($$$)'
# Also search for the class definition to understand the constructor declaration
ast-grep --pattern 'class SchedulerServer {
$$$
}'
# Backup search using ripgrep for constructor
rg "SchedulerServer::SchedulerServer" -A 5
Length of output: 676
1 participant
Description
As title. Scheduler also send heartbeat message to storage periodcially.
Validation performed
Summary by CodeRabbit
New Features
StopTokenclass for managing stop signals in a thread-safe manner.SchedulerServer.Bug Fixes
SchedulerServerand heartbeat processes.Documentation
Tests
SchedulerServerto directly control server state without threading complexity.