feat(core): add accumulator-based fold routing and edge-case hardening - #540
feat(core): add accumulator-based fold routing and edge-case hardening#540knoepfel wants to merge 5 commits into
Conversation
Codecov Report❌ Patch coverage is @@ Coverage Diff @@
## main #540 +/- ##
==========================================
+ Coverage 83.89% 84.10% +0.20%
==========================================
Files 171 172 +1
Lines 7234 7392 +158
Branches 850 868 +18
==========================================
+ Hits 6069 6217 +148
- Misses 881 887 +6
- Partials 284 288 +4
Flags with carried forward coverage won't be shown. Click here to find out more.
... and 4 files with indirect coverage changes Continue to review full report in Codecov by Harness.
🚀 New features to boost your workflow:
|
7a865dc to
3d19c63
Compare
3d19c63 to
530eb15
Compare
530eb15 to
c3caf2f
Compare
c3caf2f to
197de5e
Compare
197de5e to
10aac8e
Compare
322927f to
decf3ae
Compare
decf3ae to
e7eb445
Compare
e564fb3 to
5471faa
Compare
34e1b5e to
e106e73
Compare
📝 WalkthroughWalkthroughThe PR replaces store-counter fold caching with ChangesFold pipeline refactor
Estimated code review effort: 4 (Complex) | ~60 minutes Possibly related PRs
Suggested reviewers: 🚥 Pre-merge checks | ✅ 3 | ❌ 2❌ Failed checks (2 warnings)
✅ Passed checks (3 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 6
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@phlex/core/concepts.hpp`:
- Around line 20-21: The `sendable` concept's constraint `sendable_type<T>{t}`
is ill-formed for `std::atomic<T>` types because direct brace-initialization
from an atomic reference is not allowed (e.g., `int{atomic_const_ref}` fails).
Fix this by reverting to the previous constraint approach that used `{ send(t) }
-> std::move_constructible`, which properly handles type conversion through the
`send()` function, or alternatively verify that `sendable_type<T>` exists
without attempting direct initialization.
In `@phlex/core/declared_fold.hpp`:
- Line 149: Remove the unused member variable `initializer_` from the class
definition in declared_fold.hpp. This member variable is default-constructed and
never referenced in the class implementation, making it leftover code. The
`initializer` constructor parameter is already being passed directly to
`make_initializer` and `join_`, so the member variable is not needed. Simply
delete the `InitTuple initializer_;` declaration line.
- Line 157: The member variable `product_count_` in the class is declared with
an accessor but never incremented, unlike the similar implementations in
declared_unfold.hpp and declared_transform.hpp. Either remove the
`product_count_` member variable and its corresponding accessor method if this
metric is not applicable to the fold pipeline design, or implement the logic to
increment `product_count_` at the appropriate point in the fold operation
processing to ensure it actually tracks the product count as intended,
maintaining consistency with the other declared pipeline classes.
In `@phlex/core/detail/accumulator_node.hpp`:
- Around line 295-304: The cleanup_cache_entry method dereferences
entry->accumulator_msg without checking if it is null. When flush_received is
true and counter is zero, but the partition message has not yet arrived,
accumulator_msg remains null, causing a null pointer dereference. Add a null
pointer check for entry->accumulator_msg before calling release_as_message on
it, and only proceed with the try_put and cached_results_.erase operations if
accumulator_msg is not null.
In `@phlex/core/fold_join_node.hpp`:
- Line 67: Fix the typo in the comment on line 67 of the fold_join_node.hpp
file. Change "qacross" to "across" in the comment that describes message
matching behavior. The corrected comment should read "Messages with the same ID
across" instead of "Messages with the same ID qacross".
In `@phlex/core/fold/send.hpp`:
- Around line 27-38: The two specializations of sendable_type_impl can become
ambiguous if a type is both std::move_constructible and has a valid send()
overload. To make these specializations mutually exclusive, add a requires
clause to the first sendable_type_impl specialization (the one that matches
std::move_constructible T) that explicitly excludes types satisfying the send()
requirement. This requires clause should check that the type does not satisfy
the condition in the second specialization, ensuring only one specialization can
match for any given type.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: 26a2d1c9-a41c-4558-90d8-009f33711ab4
📒 Files selected for processing (25)
.amazonq/rules/memory-bank/structure.mdphlex/core/CMakeLists.txtphlex/core/concepts.hppphlex/core/declared_fold.cppphlex/core/declared_fold.hppphlex/core/declared_observer.hppphlex/core/declared_unfold.cppphlex/core/declared_unfold.hppphlex/core/detail/accumulator_node.hppphlex/core/fold/send.hppphlex/core/fold_join_node.hppphlex/core/framework_graph.cppphlex/core/framework_graph.hppphlex/core/fwd.hppphlex/core/index_router.cppphlex/core/index_router.hppphlex/core/message.hppphlex/core/multilayer_join_node.hppphlex/core/store_counters.cppphlex/core/store_counters.hppphlex/model/flush_gate.cppphlex/model/flush_gate.hpptest/CMakeLists.txttest/accumulator_test.cpptest/flush_gate_test.cpp
💤 Files with no reviewable changes (6)
- phlex/core/declared_observer.hpp
- phlex/core/store_counters.hpp
- phlex/core/CMakeLists.txt
- phlex/core/store_counters.cpp
- phlex/core/framework_graph.hpp
- phlex/core/fwd.hpp
📜 Review details
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (4)
- GitHub Check: build (gcc, none)
- GitHub Check: Analyze cpp with CodeQL
- GitHub Check: clang-tidy-check
- GitHub Check: coverage
🧰 Additional context used
📓 Path-based instructions (2)
**/*.{cpp,cc,cxx,h,hpp}
📄 CodeRabbit inference engine (.github/copilot-instructions.md)
**/*.{cpp,cc,cxx,h,hpp}: Use clang-format tool for all C++ code formatting (VS Code auto-formats on save); configuration defined in.clang-formatwith 100-character line limit and 2-space indentation
Follow clang-tidy recommendations defined in.clang-tidy
Files:
phlex/core/fold/send.hppphlex/core/declared_unfold.cppphlex/core/declared_fold.cppphlex/core/concepts.hppphlex/core/multilayer_join_node.hppphlex/model/flush_gate.hppphlex/core/declared_unfold.hpptest/flush_gate_test.cppphlex/core/fold_join_node.hppphlex/core/detail/accumulator_node.hpptest/accumulator_test.cppphlex/model/flush_gate.cppphlex/core/framework_graph.cppphlex/core/message.hppphlex/core/index_router.hppphlex/core/declared_fold.hppphlex/core/index_router.cpp
**/*.{hpp,cpp}
📄 CodeRabbit inference engine (AGENTS.md)
**/*.{hpp,cpp}: Use.hppfor C++ header files and.cppfor implementation files; test files should be named*_test.cpp
All C++ identifiers (namespaces, classes, structs, enums, functions, variables, parameters, members, constants, type aliases) must uselower_casenaming
Template parameters in C++ must useCamelCasenaming
C++ macros must useUPPER_CASEnaming
Private, protected, and constant C++ members must have a trailing underscore (_)
Useeast-conststyle in C++:int const xnotconst int x
Preferenum classover plainenumin C++
Avoid boolean parameters in C++ function interfaces; prefer enumerations instead
Usestd::shared_ptrfor shared ownership,std::unique_ptrfor exclusive ownership, and raw pointers only for non-owning references in C++
C++ clang-format enforces 100-character line limit and 2-space indentation withQualifierAlignment: RightandPointerAlignment: Left
Use namespacesphlex::for core code andphlex::experimental::for experimental features in C++
Use functors with agent noun naming (e.g.,ModelEvaluator evaluate_model(...)) in C++
Files:
phlex/core/fold/send.hppphlex/core/declared_unfold.cppphlex/core/declared_fold.cppphlex/core/concepts.hppphlex/core/multilayer_join_node.hppphlex/model/flush_gate.hppphlex/core/declared_unfold.hpptest/flush_gate_test.cppphlex/core/fold_join_node.hppphlex/core/detail/accumulator_node.hpptest/accumulator_test.cppphlex/model/flush_gate.cppphlex/core/framework_graph.cppphlex/core/message.hppphlex/core/index_router.hppphlex/core/declared_fold.hppphlex/core/index_router.cpp
🪛 Cppcheck (2.20.0)
test/accumulator_test.cpp
[style] 24-24: The function 'convert' is never used.
(unusedFunction)
🔇 Additional comments (29)
phlex/core/detail/accumulator_node.hpp (1)
1-186: LGTM!Also applies to: 206-237, 239-293
phlex/core/fold_join_node.hpp (1)
1-66: LGTM!Also applies to: 68-171, 173-233
phlex/core/declared_fold.hpp (1)
38-54: LGTM!Also applies to: 59-70, 74-148, 150-156, 158-158
phlex/core/declared_fold.cpp (1)
1-15: LGTM!phlex/core/message.hpp (1)
52-77: LGTM!phlex/core/multilayer_join_node.hpp (1)
114-117: LGTM!phlex/core/index_router.hpp (1)
27-65: LGTM!Also applies to: 85-98, 103-125, 148-184, 190-226
phlex/core/index_router.cpp (8)
45-106: LGTM!
108-126: LGTM!
128-143: LGTM!
199-258: LGTM!
362-450: LGTM!
452-513: LGTM!
577-611: LGTM!
175-175: This is not a concern—the project explicitly targets C++23 as its baseline standard (per AGENTS.md and CLANG_TIDY_CONFIGURATION.md), andstd::ranges::containsis already used elsewhere in the codebase without issue (e.g., phlex/model/fixed_hierarchy.cpp:24). No action needed.phlex/core/framework_graph.cpp (4)
17-41: LGTM!
43-75: LGTM!
218-223: LGTM!
122-128: LGTM!phlex/core/declared_unfold.hpp (1)
45-55: LGTM!test/accumulator_test.cpp (3)
20-80: LGTM!
81-117: LGTM!
120-268: LGTM!phlex/core/declared_unfold.cpp (1)
23-23: LGTM!test/CMakeLists.txt (1)
14-14: LGTM!test/flush_gate_test.cpp (1)
29-29: LGTM!Also applies to: 225-231
.amazonq/rules/memory-bank/structure.md (1)
19-19: LGTM!phlex/model/flush_gate.hpp (1)
43-43: LGTM!Also applies to: 61-61, 70-70, 88-88
phlex/model/flush_gate.cpp (1)
11-11: LGTM!Also applies to: 22-22, 28-28, 38-42, 86-86
e106e73 to
e47a8b2
Compare
dd941a9 to
c7374d2
Compare
0ddf5ee to
4abdd36
Compare
4abdd36 to
a23f06f
Compare
a23f06f to
5f5bb15
Compare
5574fad to
1f1e0f2
Compare
1f1e0f2 to
a5c2559
Compare
knoepfel
changed the title
|
@CodeRabbit review |
✅ Action performedReview finished.
|
|
Note GitHub couldn't provide a complete incremental comparison for this pull request, so CodeRabbit is performing a full review instead. This review may take a little longer. |
There was a problem hiding this comment.
Actionable comments posted: 13
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
phlex/core/index_router.cpp (1)
371-405: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winAvoid sending flush tokens from join nodes that do not route this index.
end_token_entriesare populated in the per-slot loop, butmessage_slotsonly commits inside thehas_exact_match and matched_count == slots.size()gate. A join node where one slot is neither exact nor a parent therefore skips the index message while still sendingindexed_end_tokenentries at flush time. These tokens create cache entries with negative counters; bothrepeater_nodeandaccumulator_nodeerase entries only whencounter == 0, so an early unmatched flush can keep the cache entry alive and mask later decrements.Buffer end-token entries in the node-local
matching_slotsloop and append them only inside the same gate that appendsmessage_slots.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@phlex/core/index_router.cpp` around lines 371 - 405, Buffer the flush-derived end-token entries alongside the matched slots in the per-slot loop, and only append them to the node’s committed end-token collection inside the existing has_exact_match && matched_count == slots.size() gate. Ensure slots that are neither exact matches nor parents produce no committed tokens, while preserving the current counting-layer hash generation for valid matches.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@phlex/core/declared_fold.hpp`:
- Around line 106-118: Update the fold_ node declaration and its lambda to use a
TBB function_node returning continue_msg, removing the unused output-port
parameter and the dead message_tuple<1> output declaration. Preserve the
existing apply_fold, calls_, and notify_result_repeater_port behavior; add a
brief comment only if the output port is intentionally retained for future use.
- Around line 62-68: Wrap the long explanatory comment above the constexpr
branch to keep each line within 100 columns. In the non-empty branch of the
lambda returned by the surrounding fold-building function, move-capture the
by-value args tuple instead of copy-capturing it, while preserving the existing
std::get<Is>(args)... expansion behavior.
- Around line 136-147: The apply_fold path currently allows concurrent calls on
the same shared partial_result; update fold_node/fold_ handling to prevent this.
Add per-partition synchronization around partial_result->call (covering the
shared accumulator state), or reject configurations where the requested
concurrency is unsafe for the fold type, while preserving parallelism across
independent partitions.
In `@phlex/core/detail/accumulator_node.hpp`:
- Around line 39-42: Update accumulator<Result>::release_as_product() to make
phlex::experimental::send visible before the compile-time branch and gate the
branch with phlex::experimental::has_send<Result>. Preserve unqualified
send(*result) invocation so custom sendable overloads participate, while atomic
results use the load-based send path instead of moving the value.
In `@phlex/core/fold_join_node.hpp`:
- Around line 141-150: Update fold_join_node::index_ports so it never selects
layers_[0] as the counting layer for multi-input folds; resolve and pass the
actual deepest input layer into fold_join_node before wiring the partition slot,
or explicitly reject multi-input folds until resolution is available. Preserve
partition_layer for empty layers and ensure the router counter uses the resolved
deepest layer.
In `@phlex/core/framework_graph.cpp`:
- Around line 64-74: Update the unfold-processing loop over n->input() so each
unique input_layer is counted and added to result.layer_pairs only once per
unfold, even when multiple selectors share that layer. Deduplicate within each
unfold before incrementing result.count_per_input_layer or appending the pair,
and add coverage for an unfold with two selectors from the same layer.
In `@phlex/core/index_router.cpp`:
- Around line 257-263: Update indexed_end_token::count and all corresponding
flush-token handling, including repeater_node::handle_flush_token, to use
std::ptrdiff_t so the committed count from committed_count_for_layer is not
narrowed. Remove the static_cast<int> in the set_flush_callback lambda and
preserve the existing count propagation behavior.
- Around line 180-189: Clean up wire_fold_partition_index_sets by iterating
fold_partition_ports values without binding the unused fold_node_name, and use
partition_port.layer directly when emplacing into index_set_nodes_ instead of
the redundant static_cast. Preserve the existing node creation and make_edge
behavior.
- Around line 126-152: Bound expansion in establish_layer_hierarchy using a
maximum depth derived from the initial deepest layer path plus
layer_pairs.size(), and reject candidates exceeding that bound. Throw a clear
diagnostic naming the offending unfold pair(s) instead of continuing the
fixed-point loop; preserve deduplication and normal hierarchy expansion within
the bound.
In `@phlex/core/index_router.hpp`:
- Around line 120-125: Change finalize’s layer_paths_from_driver parameter to
pass by value, matching establish_layer_hierarchy’s ownership-transfer usage and
allowing the existing std::move call to move the vector instead of copying it.
In `@phlex/core/message.hpp`:
- Around line 72-77: Update named_index_port::counting_layer in
phlex/core/message.hpp:72-77 to std::optional<phlex::experimental::identifier>,
documenting that an empty optional selects the node’s deepest layer and a
populated value selects the specified layer. In
phlex/core/multilayer_join_node.hpp:116-119, pass an empty optional instead of
layer and revise the comment to describe balancing against the node’s deepest
layer.
In `@phlex/model/layer_path.cpp`:
- Line 104: Update the declaration of layer_path::depth() in layer_path.hpp with
a one-sentence comment documenting that it counts stored path segments only and
does not include the implicit job root, unlike hash() and hashes(); leave the
implementation unchanged.
In `@test/accumulator_test.cpp`:
- Around line 62-67: Update use_ostream_logger to save spdlog::default_logger()
before installing the ostream logger, then add scope-bound cleanup that restores
the saved logger before the local ostringstream is destroyed.
---
Outside diff comments:
In `@phlex/core/index_router.cpp`:
- Around line 371-405: Buffer the flush-derived end-token entries alongside the
matched slots in the per-slot loop, and only append them to the node’s committed
end-token collection inside the existing has_exact_match && matched_count ==
slots.size() gate. Ensure slots that are neither exact matches nor parents
produce no committed tokens, while preserving the current counting-layer hash
generation for valid matches.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: d9207be0-30aa-482b-bc06-418cc7f17952
📒 Files selected for processing (24)
phlex/core/CMakeLists.txtphlex/core/concepts.hppphlex/core/declared_fold.cppphlex/core/declared_fold.hppphlex/core/declared_observer.hppphlex/core/declared_unfold.cppphlex/core/declared_unfold.hppphlex/core/detail/accumulator_node.hppphlex/core/detail/repeater_node.cppphlex/core/fold/send.hppphlex/core/fold_join_node.hppphlex/core/framework_graph.cppphlex/core/framework_graph.hppphlex/core/fwd.hppphlex/core/index_router.cppphlex/core/index_router.hppphlex/core/message.hppphlex/core/multilayer_join_node.hppphlex/core/store_counters.cppphlex/core/store_counters.hppphlex/model/layer_path.cppphlex/model/layer_path.hpptest/CMakeLists.txttest/accumulator_test.cpp
💤 Files with no reviewable changes (5)
- phlex/core/CMakeLists.txt
- phlex/core/declared_observer.hpp
- phlex/core/store_counters.hpp
- phlex/core/store_counters.cpp
- phlex/core/fwd.hpp
📜 Review details
⏰ Context from checks skipped due to timeout. (4)
- GitHub Check: Analyze cpp with CodeQL
- GitHub Check: build (gcc, none)
- GitHub Check: clang-tidy-check
- GitHub Check: coverage
🧰 Additional context used
📓 Path-based instructions (3)
**/*.{cpp,cc,cxx,h,hpp}
📄 CodeRabbit inference engine (.github/copilot-instructions.md)
**/*.{cpp,cc,cxx,h,hpp}: Use clang-format tool for all C++ code formatting (VS Code auto-formats on save); configuration defined in.clang-formatwith 100-character line limit and 2-space indentation
Follow clang-tidy recommendations defined in.clang-tidy
Files:
phlex/model/layer_path.hppphlex/core/detail/repeater_node.cppphlex/core/declared_unfold.cppphlex/core/declared_fold.cppphlex/core/multilayer_join_node.hppphlex/core/concepts.hppphlex/model/layer_path.cppphlex/core/declared_unfold.hppphlex/core/framework_graph.cppphlex/core/message.hpptest/accumulator_test.cppphlex/core/fold/send.hppphlex/core/framework_graph.hppphlex/core/detail/accumulator_node.hppphlex/core/fold_join_node.hppphlex/core/declared_fold.hppphlex/core/index_router.hppphlex/core/index_router.cpp
**/*.{hpp,cpp}
📄 CodeRabbit inference engine (AGENTS.md)
**/*.{hpp,cpp}: Use.hppfor header files,.cppfor implementation, and*_test.cppfor test files in C++
Enforce 100-character line limit and 2-space indentation in C++ code via.clang-format
UseQualifierAlignment: Right(east-const) style:int const xnotconst int xin C++
UsePointerAlignment: Leftin C++ (pointer*attached to type, not variable name)
All C++ identifiers must uselower_casenaming: namespaces, classes, structs, enums, functions, variables, parameters, members, and constants
Exception to C++ naming: template parameters useCamelCase
Exception to C++ naming: macros useUPPER_CASE
Private, protected, and constant members in C++ must have a trailing underscore (_), no trailing underscore on anything else
Useenum classpreferred over plainenumin C++
Usestd::shared_ptrfor shared ownership,std::unique_ptrfor exclusive ownership, raw pointers for non-owning references only in C++
Use functors with agent-noun pattern:ModelEvaluator evaluate_model(...)in C++
Apply.clang-tidychecks for bugprone, cert, clang-analyzer, concurrency, cppcoreguidelines, misc, modernize, performance, portability, and readability as defined in the.clang-tidyconfiguration file
Usephlex::namespace for core code,phlex::experimental::for experimental features in C++
Files:
phlex/model/layer_path.hppphlex/core/detail/repeater_node.cppphlex/core/declared_unfold.cppphlex/core/declared_fold.cppphlex/core/multilayer_join_node.hppphlex/core/concepts.hppphlex/model/layer_path.cppphlex/core/declared_unfold.hppphlex/core/framework_graph.cppphlex/core/message.hpptest/accumulator_test.cppphlex/core/fold/send.hppphlex/core/framework_graph.hppphlex/core/detail/accumulator_node.hppphlex/core/fold_join_node.hppphlex/core/declared_fold.hppphlex/core/index_router.hppphlex/core/index_router.cpp
**/*.hpp
📄 CodeRabbit inference engine (AGENTS.md)
Avoid boolean parameters in C++ interfaces; prefer enumerations instead
Files:
phlex/model/layer_path.hppphlex/core/multilayer_join_node.hppphlex/core/concepts.hppphlex/core/declared_unfold.hppphlex/core/message.hppphlex/core/fold/send.hppphlex/core/framework_graph.hppphlex/core/detail/accumulator_node.hppphlex/core/fold_join_node.hppphlex/core/declared_fold.hppphlex/core/index_router.hpp
🪛 Cppcheck (2.21.0)
phlex/model/layer_path.cpp
[style] 104-104: The function 'depth' is never used.
(unusedFunction)
test/accumulator_test.cpp
[style] 24-24: The function 'convert' is never used.
(unusedFunction)
🔇 Additional comments (22)
phlex/core/fold/send.hpp (1)
12-51: LGTM!phlex/core/concepts.hpp (1)
21-21: LGTM!phlex/core/detail/accumulator_node.hpp (1)
50-337: LGTM!phlex/core/detail/repeater_node.cpp (1)
155-170: LGTM!phlex/core/fold_join_node.hpp (1)
18-140: LGTM!Also applies to: 162-234
phlex/core/framework_graph.cpp (1)
18-51: LGTM!Also applies to: 145-145, 225-252
phlex/core/framework_graph.hpp (1)
79-81: LGTM!Also applies to: 184-186
phlex/core/declared_unfold.cpp (1)
25-25: LGTM!phlex/core/declared_unfold.hpp (1)
45-55: LGTM!test/CMakeLists.txt (1)
14-14: LGTM!test/accumulator_test.cpp (1)
1-60: LGTM!Also applies to: 69-209, 254-293
phlex/core/declared_fold.hpp (3)
42-57: LGTM!
121-134: LGTM!
149-155: LGTM!phlex/core/declared_fold.cpp (1)
6-9: LGTM!phlex/core/message.hpp (1)
52-71: LGTM!phlex/model/layer_path.hpp (1)
47-49: LGTM!phlex/core/index_router.hpp (1)
28-65: LGTM!Also applies to: 85-99, 148-160, 169-195, 210-229
phlex/core/index_router.cpp (4)
154-167: LGTM!
195-228: LGTM!Also applies to: 433-463
92-100: LGTM!Also applies to: 272-285, 294-310, 415-431, 481-490, 516-518, 533-542
246-257: 🩺 Stability & AvailabilityNo change needed.
all_children_accounted()returns false whileflush_callback_is absent, and there is no code path that callssend_flush()without a callback at this point.
| // The compiler emits a warning if an empty args tuple is captured by reference and then not expanded inside the lambda expression. We therefore only capture args in the lambda expression if sizeof...(Is) > 0. | ||
| if constexpr (sizeof...(Is) == 0) { | ||
| return [](data_cell_index const&) { return std::make_unique<FoldResult>(); }; | ||
| } else { | ||
| return [args](data_cell_index const&) { | ||
| return std::make_unique<FoldResult>(std::get<Is>(args)...); | ||
| }; |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟠 Major | ⚡ Quick win
Wrap the comment to 100 columns, and move args into the lambda.
Line 62 is a single comment line of roughly 200 characters. The project enforces a 100-character limit through .clang-format. Wrap it.
Also, args arrives by value and is then copy-captured. A std::move capture avoids one tuple copy of the initializer arguments.
♻️ Proposed reformat and move-capture
- // The compiler emits a warning if an empty args tuple is captured by reference and then not expanded inside the lambda expression. We therefore only capture args in the lambda expression if sizeof...(Is) > 0.
+ // The compiler emits a warning if an empty args tuple is captured and then not expanded
+ // inside the lambda expression. We therefore only capture args when sizeof...(Is) > 0.
if constexpr (sizeof...(Is) == 0) {
return [](data_cell_index const&) { return std::make_unique<FoldResult>(); };
} else {
- return [args](data_cell_index const&) {
+ return [args = std::move(args)](data_cell_index const&) {
return std::make_unique<FoldResult>(std::get<Is>(args)...);
};
}As per coding guidelines: "Enforce 100-character line limit and 2-space indentation in C++ code via .clang-format".
📝 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.
| // The compiler emits a warning if an empty args tuple is captured by reference and then not expanded inside the lambda expression. We therefore only capture args in the lambda expression if sizeof...(Is) > 0. | |
| if constexpr (sizeof...(Is) == 0) { | |
| return [](data_cell_index const&) { return std::make_unique<FoldResult>(); }; | |
| } else { | |
| return [args](data_cell_index const&) { | |
| return std::make_unique<FoldResult>(std::get<Is>(args)...); | |
| }; | |
| // The compiler emits a warning if an empty args tuple is captured and then not expanded | |
| // inside the lambda expression. We therefore only capture args when sizeof...(Is) > 0. | |
| if constexpr (sizeof...(Is) == 0) { | |
| return [](data_cell_index const&) { return std::make_unique<FoldResult>(); }; | |
| } else { | |
| return [args = std::move(args)](data_cell_index const&) { | |
| return std::make_unique<FoldResult>(std::get<Is>(args)...); | |
| }; |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@phlex/core/declared_fold.hpp` around lines 62 - 68, Wrap the long explanatory
comment above the constexpr branch to keep each line within 100 columns. In the
non-empty branch of the lambda returned by the surrounding fold-building
function, move-capture the by-value args tuple instead of copy-capturing it,
while preserving the existing std::get<Is>(args)... expansion behavior.
Source: Coding guidelines
| fold_{g, | ||
| concurrency, | ||
| [this, ft = alg.release_algorithm()](messages_t<num_inputs> const& messages, auto&) { | ||
| // N.B. The assumption is that a fold will *never* need to cache | ||
| // the product store it creates. Any flush messages *do not* need | ||
| // to be propagated to downstream nodes. | ||
| auto const& msg = most_derived(messages); | ||
| auto const& index = msg.store->index(); | ||
|
|
||
| auto fold_index = index->parent(partition_); | ||
| if (not fold_index) { | ||
| return; | ||
| } | ||
|
|
||
| auto index_hash_for_counter = fold_index->hash(); | ||
| [this, ft = alg.release_algorithm()]( | ||
| accumulator_with_messages<result_type, num_inputs> const& accum_with_msgs, auto&) { | ||
| std::size_t const partition_hash = apply_fold(ft, accum_with_msgs); | ||
|
|
||
| call(ft, messages, std::make_index_sequence<num_inputs>{}); | ||
| ++calls_; | ||
|
|
||
| counter_for(index_hash_for_counter).increment(index->layer_hash()); | ||
|
|
||
| emit_and_evict_if_done(fold_index); | ||
| join_.notify_result_repeater_port().try_put(partition_hash); | ||
| }} | ||
| { | ||
| if constexpr (num_inputs > 1ull) { | ||
| make_edge(join_, fold_); | ||
| } | ||
| make_edge(join_, fold_); | ||
| } |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value
The fold_ node declares an output port that nothing uses.
The lambda takes auto& for the output ports and never writes to them. Result emission now happens through join_.output_port(), and completion is signalled through notify_result_repeater_port(). So message_tuple<1> on line 152-154 is dead surface area.
A tbb::flow::function_node<accumulator_with_messages<result_type, num_inputs>, tbb::flow::continue_msg> expresses the same behavior with one less moving part. If the port is a placeholder for upcoming work, a short comment saying so would save the next reader a trip through TBB docs.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@phlex/core/declared_fold.hpp` around lines 106 - 118, Update the fold_ node
declaration and its lambda to use a TBB function_node returning continue_msg,
removing the unused output-port parameter and the dead message_tuple<1> output
declaration. Preserve the existing apply_fold, calls_, and
notify_result_repeater_port behavior; add a brief comment only if the output
port is intentionally retained for future use.
| std::size_t apply_fold( | ||
| function_t const& ft, | ||
| accumulator_with_messages<result_type, num_inputs> const& accum_with_msgs) | ||
| { | ||
| auto& result = results_.at(store->index()->hash()); | ||
| using phlex::experimental::send; | ||
| if constexpr (requires { send(*result); }) { | ||
| store->add_product(output()[0], send(*result)); | ||
| } else { | ||
| store->add_product(output()[0], std::move(*result)); | ||
| } | ||
| // Reclaim some memory; it would be better to erase the entire entry from the map, | ||
| // but that is not thread-safe. | ||
| result.reset(); | ||
| // We have to do awkward index management until we can use structured bindings with packs. | ||
| auto& accumulator = std::get<0>(accum_with_msgs); | ||
| [&]<std::size_t... Is>(std::index_sequence<Is...>) { | ||
| accumulator.partial_result->call( | ||
| ft, std::get<Is>(input_).retrieve(std::get<Is + 1>(accum_with_msgs))...); | ||
| }(std::make_index_sequence<num_inputs>{}); | ||
| return accumulator.index->hash(); | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy lift
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Inspect accumulator_node/fold_join_node for per-partition serialization of partial_result.
fd -t f 'accumulator_node.hpp|fold_join_node.hpp' | while IFS= read -r f; do
echo "=== $f ==="
rg -n -C 6 'partial_result|mutex|serial|tbb::flow::serial|concurrency|accumulator_with_messages' "$f"
done
# Find how fold concurrency is supplied at registration sites.
rg -nP -C 4 '\bfold_node<' --type cppRepository: Framework-R-D/phlex
Length of output: 3404
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "=== declared_fold.hpp outline ==="
ast-grep outline phlex/core/declared_fold.hpp --view compact || true
echo "=== declared_fold.hpp relevant lines ==="
sed -n '1,220p' phlex/core/declared_fold.hpp
echo "=== fold_join_node.hpp outline ==="
ast-grep outline phlex/core/fold_join_node.hpp --view compact || true
echo "=== fold_join_node.hpp relevant lines ==="
sed -n '1,260p' phlex/core/fold_join_node.hpp
echo "=== all accumulator/partial_result/fold declarations in C++ files ==="
rg -n -C 5 'partial_result|call<|struct accumulator|class .*accumulator|struct .*fold' --type cpp --type hpp .Repository: Framework-R-D/phlex
Length of output: 17980
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "=== accumulator_node.hpp relevant lines ==="
sed -n '1,380p' phlex/core/detail/accumulator_node.hpp
echo "=== repeater_node.hpp relevant lines ==="
fd -t f 'repeater_node.hpp' | while IFS= read -r f; do
echo "=== $f ==="
sed -n '1,260p' "$f"
done
echo "=== fold creation call sites and concurrency usages ==="
rg -n -C 5 'fold_node<|declared_folds|make_fold|fold_node|set_unlimited|concurrency' --glob '*.cpp' --glob '*.hpp' .Repository: Framework-R-D/phlex
Length of output: 50376
Ensure per-partition fold calls do not share partial_result.
fold_node passes user concurrency into fold_, while accumulator_node emits one std::shared_ptr<accumulator<T>> partial_result per partition and TBB multifunction_node does not serialize concurrent body executions. If concurrency is greater than 1, multiple fold jobs for the same partition can call std::invoke(f, *accumulator_, ...) on the same fold object at the same time. Use per-partition synchronization or issue an error when concurrency is not safe for the fold type.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@phlex/core/declared_fold.hpp` around lines 136 - 147, The apply_fold path
currently allows concurrent calls on the same shared partial_result; update
fold_node/fold_ handling to prevent this. Add per-partition synchronization
around partial_result->call (covering the shared accumulator state), or reject
configurations where the requested concurrency is unsafe for the fold type,
while preserving parallelism across independent partitions.
| if constexpr (requires { send(*result); }) { | ||
| return std::make_unique<product<sendable_t>>(send(*result)); | ||
| } else { | ||
| return std::make_unique<product<sendable_t>>(std::move(*result)); |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Inspect the declaration namespace and all send call sites. Do not run repository code.
rg -n -C 3 'namespace phlex::experimental|send\(std::atomic|concept has_send' \
phlex/core/fold/send.hpp
rg -n -C 4 'requires \{ send\(\*result\); \}|send\(\*result\)|using phlex::experimental::send' \
phlex/core/detail/accumulator_node.hppRepository: Framework-R-D/phlex
Length of output: 814
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== accumulator_node outlines/slices =="
wc -l phlex/core/detail/accumulator_node.hpp
ast-grep outline phlex/core/detail/accumulator_node.hpp --view expanded || true
sed -n '1,140p' phlex/core/detail/accumulator_node.hpp
echo "== send.hpp slices =="
sed -n '1,80p' phlex/core/fold/send.hpp
echo "== namespace and include references =="
rg -n 'accumulator_node|include .*send|using phlex::experimental::send|has_send|std::atomic|struct accumulator|class accumulator' phlex/core/detail/accumulator_node.hpp phlex/core/fold phlex/core 2>/dev/null | head -200
echo "== behavioral lookup probe if tooling is present =="
python3 - <<'PY' from pathlib import Path
# Read-only static checks derived from C++ ADL/unqualified lookup semantics.
from pathlib import Path
import re
p=Path('phlex/core/detail/accumulator_node.hpp')
s=p.read_text()
print('contains_unqualified_send_call=', 'send(*result)' in s)
print('contains_using_experimental_send=', 'using phlex::experimental::send' in s)
print('send_declared_namespace=', re.search(r'namespace\s+([^;{]+)\s*\{[\s\S]*?send\(std::atomic', Path('phlex/core/fold/send.hpp').read_text()))
PYRepository: Framework-R-D/phlex
Length of output: 17589
Make send visible before selecting the send branch.
accumulator<Result>::release_as_product() calls unqualified send(*result) without using phlex::experimental::send;. That overload is declared in phlex::experimental, and ADL for std::atomic<T> searches std, so atomic fold results fall through to the std::move(*result) path instead of invoking load()-based send. Gate with phlex::experimental::has_send<Result> and import the experimental overload so custom sendable overloads still participate.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@phlex/core/detail/accumulator_node.hpp` around lines 39 - 42, Update
accumulator<Result>::release_as_product() to make phlex::experimental::send
visible before the compile-time branch and gate the branch with
phlex::experimental::has_send<Result>. Preserve unqualified send(*result)
invocation so custom sendable overloads participate, while atomic results use
the load-based send path instead of moving the value.
| // FIXME: For multi-input folds, the "most-derived" input layer is the deepest of | ||
| // `layers_` in the data hierarchy. Until the router exposes a depth comparison, we | ||
| // pick `layers_[0]`; all current tests have single-input folds so this is correct in | ||
| // practice. | ||
| std::vector<named_index_port> index_ports() | ||
| { | ||
| std::vector<named_index_port> result; | ||
| result.reserve(1 + repeaters_.size()); // +1 for the result repeater | ||
| phlex::experimental::identifier const counting_layer_for_partition = | ||
| layers_.empty() ? partition_layer_ : layers_[0]; |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | 🏗️ Heavy lift
Resolve the actual deepest counting layer before wiring the partition slot.
layers_[0] is not necessarily the most-derived input layer. For a multi-input fold, this makes the router produce the partition flush count at the wrong layer. The accumulator can then retain results indefinitely or release a result before all fold calls complete.
Pass the resolved deepest layer into fold_join_node, or reject multi-input folds until layer-depth resolution is available. Do not use an arbitrary input layer for this counter.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@phlex/core/fold_join_node.hpp` around lines 141 - 150, Update
fold_join_node::index_ports so it never selects layers_[0] as the counting layer
for multi-input folds; resolve and pass the actual deepest input layer into
fold_join_node before wiring the partition slot, or explicitly reject
multi-input folds until resolution is available. Preserve partition_layer for
empty layers and ensure the router counter uses the resolved deepest layer.
| gate_for(index)->set_flush_callback( | ||
| [this, end_token_slots = std::move(end_token_slots), index, message_id]( | ||
| flush_gate const& fc) { | ||
| for (auto const& slot : *end_token_slots) { | ||
| slot->put_end_token(index, fc); | ||
| [end_token_entries = std::move(end_token_entries)](flush_gate const& fc) { | ||
| for (auto const& entry : *end_token_entries) { | ||
| auto const count = fc.committed_count_for_layer(entry.counting_layer_hash); | ||
| entry.flush_port->try_put({.index = fc.index(), .count = static_cast<int>(count)}); | ||
| } | ||
|
|
||
| // Used only for folds, until folds use the slot infrastructure above. | ||
| flusher_.try_put({index, fc.committed_counts(), message_id}); | ||
| }); |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
static_cast<int>(count) narrows a std::size_t committed count.
committed_count_for_layer returns an unsigned count, and indexed_end_token::count is int. A partition that accumulates more than INT_MAX cells produces an implementation-defined (and almost certainly negative) value, which repeater_node::handle_flush_token then subtracts. For a long-running job with a job-level partition this is not purely theoretical.
Widening indexed_end_token::count to std::ptrdiff_t would remove the cliff. If you prefer to keep int, an assertion that the count fits turns a silent corruption into a loud failure.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@phlex/core/index_router.cpp` around lines 257 - 263, Update
indexed_end_token::count and all corresponding flush-token handling, including
repeater_node::handle_flush_token, to use std::ptrdiff_t so the committed count
from committed_count_for_layer is not narrowed. Remove the static_cast<int> in
the set_flush_callback lambda and preserve the existing count propagation
behavior.
| void finalize(tbb::flow::graph& g, | ||
| std::vector<phlex::experimental::layer_path> const& layer_paths_from_driver, | ||
| unfold_data unfolds, | ||
| provider_input_ports_t provider_input_ports, | ||
| fold_partition_ports_t fold_partition_ports, | ||
| std::map<std::string, named_index_ports> multilayer_join_ports); |
There was a problem hiding this comment.
🚀 Performance & Scalability | 🟠 Major | ⚡ Quick win
finalize takes the paths by const&, but establish_layer_hierarchy takes them by value.
phlex/core/index_router.cpp line 112 calls establish_layer_hierarchy(std::move(layer_paths_from_driver), ...). Because layer_paths_from_driver is a const&, std::move produces a const&&, which binds to the by-value parameter through the copy constructor. The vector of layer_path is copied, and the std::move reads as an optimization that never happens.
Take the parameter by value in finalize so the move is real, or drop the std::move at the call site so the copy is honest.
♻️ Proposed signature change
void finalize(tbb::flow::graph& g,
- std::vector<phlex::experimental::layer_path> const& layer_paths_from_driver,
+ std::vector<phlex::experimental::layer_path> layer_paths_from_driver,
unfold_data unfolds,And in phlex/core/index_router.cpp:
void index_router::finalize(tbb::flow::graph& g,
- std::vector<layer_path> const& layer_paths_from_driver,
+ std::vector<layer_path> layer_paths_from_driver,
unfold_data unfolds,📝 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.
| void finalize(tbb::flow::graph& g, | |
| std::vector<phlex::experimental::layer_path> const& layer_paths_from_driver, | |
| unfold_data unfolds, | |
| provider_input_ports_t provider_input_ports, | |
| fold_partition_ports_t fold_partition_ports, | |
| std::map<std::string, named_index_ports> multilayer_join_ports); | |
| void finalize(tbb::flow::graph& g, | |
| std::vector<phlex::experimental::layer_path> layer_paths_from_driver, | |
| unfold_data unfolds, | |
| provider_input_ports_t provider_input_ports, | |
| fold_partition_ports_t fold_partition_ports, | |
| std::map<std::string, named_index_ports> multilayer_join_ports); |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@phlex/core/index_router.hpp` around lines 120 - 125, Change finalize’s
layer_paths_from_driver parameter to pass by value, matching
establish_layer_hierarchy’s ownership-transfer usage and allowing the existing
std::move call to move the vector instead of copying it.
| struct named_index_port { | ||
| phlex::experimental::identifier layer; | ||
| phlex::experimental::identifier counting_layer; | ||
| tbb::flow::receiver<indexed_end_token>* token_port; | ||
| tbb::flow::receiver<index_message>* index_port; | ||
| }; |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟠 Major | ⚡ Quick win
counting_layer has no explicit "unset" representation, so its meaning is encoded by equality. named_index_port::counting_layer carries three different meanings depending on its value: equal to layer means "use the node's deepest layer" (per index_router.cpp lines 220-226), a different non-empty value means "use this layer", and a default-constructed value means "resolve to no counting hash at all", which silently stops end tokens. One representation choice fixes all of it.
phlex/core/message.hpp#L72-L77: change the field tostd::optional<phlex::experimental::identifier> counting_layer, where an empty optional means "use the node's deepest layer", and update the doc comment to describe that contract. This also removes the silent-empty-identifier failure mode for any producer that forgets the field.phlex/core/multilayer_join_node.hpp#L116-L119: pass the empty optional instead of repeatinglayer, and correct the comment, which currently states the count balances againstlayer's own committed-counts entries rather than the node's deepest layer.
📍 Affects 2 files
phlex/core/message.hpp#L72-L77(this comment)phlex/core/multilayer_join_node.hpp#L116-L119
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@phlex/core/message.hpp` around lines 72 - 77, Update
named_index_port::counting_layer in phlex/core/message.hpp:72-77 to
std::optional<phlex::experimental::identifier>, documenting that an empty
optional selects the node’s deepest layer and a populated value selects the
specified layer. In phlex/core/multilayer_join_node.hpp:116-119, pass an empty
optional instead of layer and revise the comment to describe balancing against
the node’s deepest layer.
| return hashes; | ||
| } | ||
|
|
||
| std::size_t layer_path::depth() const noexcept { return layer_path_.size(); } |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value
Document that depth() ignores the implicit job root.
hash() and hashes() both treat an incomplete path as having an implicit job root. depth() does not: /job/event reports 2 while the incomplete event reports 1. index_router::deepest_layer_name compares depth() values across sorted_layer_paths_, so mixing complete and incomplete paths in that vector would skew the comparison.
One sentence on the declaration in phlex/model/layer_path.hpp stating that depth counts stored segments only, unlike hash(), keeps the two conventions straight.
Side note: the Cppcheck unusedFunction hint on this line is a false positive. depth() is called from phlex/core/index_router.cpp line 444, which Cppcheck cannot see in a single-translation-unit pass.
🧰 Tools
🪛 Cppcheck (2.21.0)
[style] 104-104: The function 'depth' is never used.
(unusedFunction)
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@phlex/model/layer_path.cpp` at line 104, Update the declaration of
layer_path::depth() in layer_path.hpp with a one-sentence comment documenting
that it counts stored path segments only and does not include the implicit job
root, unlike hash() and hashes(); leave the implementation unchanged.
Source: Linters/SAST tools
| void use_ostream_logger(std::ostringstream& oss) | ||
| { | ||
| auto ostream_sink = std::make_shared<spdlog::sinks::ostream_sink_mt>(oss); | ||
| auto ostream_logger = std::make_shared<spdlog::logger>("my_logger", ostream_sink); | ||
| spdlog::set_default_logger(ostream_logger); | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== files =="
git ls-files | rg '(^|/)accumulator_test\.cpp$|cpp\.cmake|CMakeLists|README|package' || true
echo "== accumulator_test outline =="
if [ -f test/accumulator_test.cpp ]; then
wc -l test/accumulator_test.cpp
sed -n '1,290p' test/accumulator_test.cpp
fi
echo "== spdlog usages in test =="
rg -n "spdlog::|default_logger|set_default_logger|use_ostream_logger|oss" test/accumulator_test.cpp test ctest || trueRepository: Framework-R-D/phlex
Length of output: 19978
🌐 Web query:
spdlog documentation set_default_logger global default logger sink ostream mt
💡 Result:
In spdlog, you can set a global default logger using spdlog::set_default_logger [1][2]. When you want this logger to output to an ostream (such as std::ostringstream or std::cout) in a thread-safe manner, you can use the spdlog::sinks::ostream_sink_mt sink [3][4]. To implement this, you must include the ostream_sink header, create the sink pointing to your stream, create a logger that uses this sink, and then set it as the global default [3][5]. Example Implementation: #include "spdlog/spdlog.h" #include "spdlog/sinks/ostream_sink.h" #include #include int main { // 1. Create your stream std::ostringstream oss; // 2. Create the ostream_sink_mt (thread-safe) auto ostream_sink = std::make_sharedspdlog::sinks::ostream_sink_mt(oss); // 3. Create a logger using that sink auto logger = std::make_sharedspdlog::logger("my_logger", ostream_sink); // 4. Set it as the global default logger spdlog::set_default_logger(logger); // Now all spdlog::info, spdlog::debug, etc. calls will use this logger spdlog::info("This will go to the ostream"); } Key Details: - Thread Safety: Use spdlog::sinks::ostream_sink_mt for thread-safe operations [3][4]. If thread safety is not required, you can use spdlog::sinks::ostream_sink_st [6][4]. - Header: You must include "spdlog/sinks/ostream_sink.h" [3]. - Default Logger: spdlog::set_default_logger accepts a std::shared_ptrspdlog::logger and replaces the existing global default logger [1][2]. Note that the initial default logger is typically already configured to output to stdout [1].
Citations:
- 1: https://github.com/gabime/spdlog/wiki/Default-logger
- 2: https://github.com/gabime/spdlog
- 3: https://github.com/gabime/spdlog/wiki/Sinks
- 4: https://fossies.org/dox/spdlog-1.17.0/ostream__sink_8h_source.html
- 5: https://stackoverflow.com/questions/66473052/how-can-i-read-spdlog-output-in-a-google-test
- 6: https://github.com/gabime/spdlog/blob/486b5555/include/spdlog/sinks/ostream_sink.h
🌐 Web query:
"spdlog::stderr_logger_mt" "spdlog::set_default_logger"
💡 Result:
In spdlog, spdlog::stderr_logger_mt is a factory function used to create a thread-safe logger that outputs to standard error (stderr) [1][2][3]. The spdlog::set_default_logger function is used to replace the library's global default logger with a logger of your choice [4][5]. Commonly, these are used together to initialize a custom global logger that directs output to stderr [6]. Example Usage: #include "spdlog/spdlog.h" #include "spdlog/sinks/stderr_color_sinks.h" // Optional: for color support void init_logging { // 1. Create a thread-safe stderr logger with the name "my_logger" auto logger = spdlog::stderr_logger_mt("my_logger"); // 2. Set it as the default global logger spdlog::set_default_logger(logger); } int main { init_logging; // Now you can use global macros like SPDLOG_INFO, which will use the new default logger SPDLOG_INFO("This message goes to stderr."); return 0; } Key Details: - stderr_logger_mt: The _mt suffix indicates a multi-threaded, thread-safe logger [7]. If you do not require thread safety, you can use the _st (single-threaded) variant [1]. - set_default_logger: Once set, global logging macros (such as SPDLOG_INFO, SPDLOG_ERROR, etc.) will automatically route their output through this logger instance [8][5]. This is useful for configuring application-wide logging behavior in one place [6].
Citations:
- 1: https://fossies.org/dox/spdlog-1.17.0/namespacespdlog.html
- 2: http://www.sherikov.net/px4ws/spdlog/namespacespdlog.html
- 3: https://livox-sdk.github.io/Livox-SDK-Doc/namespacespdlog.html
- 4: https://internal.dunescience.org/doxygen/namespacespdlog.html
- 5: https://code.usgs.gov/astrogeology/usgscsm/-/blob/1.4.1/include/spdlog/spdlog.h
- 6: https://github.com/OpenNMT/CTranslate2/blob/0d8bcd36/src/logging.cc
- 7: https://github.com/gabime/spdlog
- 8: http://www.sherikov.net/px4ws/spdlog/spdlog_8h.html
🌐 Web query:
"spdlog::ostream_sink_mt" "std::ostream*" source
💡 Result:
In spdlog, spdlog::sinks::ostream_sink_mt is a thread-safe sink that writes log messages to a provided std::ostream object [1][2][3]. The constructor takes a reference to an std::ostream as its source [1][3]. Usage Example: To use this sink, you typically pass a std::ostringstream or another std::ostream to the constructor, wrap it in a std::shared_ptr, and attach it to a logger [4][5][6]: #include <spdlog/sinks/ostream_sink.h> #include #include std::ostringstream oss; // Create the sink, providing the output stream as the source auto ostream_sink = std::make_sharedspdlog::sinks::ostream_sink_mt(oss); // Create a logger using the sink auto logger = std::make_sharedspdlog::logger("my_logger", ostream_sink); Key Considerations: 1. Lifetime Management: The developer is responsible for ensuring the lifetime of the provided std::ostream object (e.g., the oss variable in the example) exceeds the lifetime of the sink [7]. 2. Thread Safety: While ostream_sink_mt uses a mutex to synchronize writes to the underlying stream [1][3], it does not automatically make the underlying stream object thread-safe for external access [7][8]. If you need to read from the stream (e.g., retrieving the string from a std::ostringstream) while the logger is active, you must ensure that your access is synchronized with the sink's internal mutex to prevent memory corruption [8]. 3. Sink Variants: The suffix _mt indicates a multi-threaded (thread-safe) sink [2][9]. A single-threaded version, spdlog::sinks::ostream_sink_st, is also available for performance-critical scenarios where only one thread will access the logger [2].
Citations:
- 1: https://fossies.org/dox/spdlog-1.17.0/ostream__sink_8h_source.html
- 2: https://repo.radio/TeaSpeak/spdlog/src/commit/52d02af950633455736a86fad4dde74476d9cfe8/include/spdlog/sinks/ostream_sink.h
- 3: https://internal.dunescience.org/doxygen/classspdlog_1_1sinks_1_1ostream__sink.html
- 4: https://www.programmersought.com/article/63131261602/
- 5: Add logging to libcudf rapidsai/cudf#12637
- 6: https://repo.radio/TeaSpeak/spdlog/commit/8387d779b44c1228fb0ef66188f92531e12f65dd.diff
- 7: get data from ostream_sink_mt gabime/spdlog#2564
- 8: Getting results from ostream_sink_mt with std::ostringstream gabime/spdlog#1316
- 9: https://gitlab.au.dk/ausa/simx/-/blob/compile_updates/external/include/spdlog/sinks/ostream_sink.h?ref_type=heads
Restore the process default logger.
use_ostream_logger replaces the global spdlog default with a local std::ostringstream sink. As long as that logger remains installed, later spdlog calls can write to oss after it leaves scope. Save the previous spdlog::default_logger() before this override and restore it with scope-bound cleanup before oss is destroyed.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@test/accumulator_test.cpp` around lines 62 - 67, Update use_ostream_logger to
save spdlog::default_logger() before installing the ostream logger, then add
scope-bound cleanup that restores the saved logger before the local
ostringstream is destroyed.
1 participant
This PR replaces the caching mechanism for fold results with a similar mechanism as used in #118.
Resolves #359
Core
accumulator_nodeandfold_join_nodeto aggregate partition results and synchronize fold inputs.multilayer_join_node-style layer and counting metadata.index_routerfinalization and end-token routing for explicit fold, unfold, provider, and layer-path metadata.declared_foldand replaceflush_port()withpartition_port().flush_message,flusher_t,store_counters, and related APIs.sendable_typetraits and update thesendableconcept.layer_path::depth()and correct the unfold child counter.Build
Tests
accumulatorCTest target.Maintenance