Skip to content

Commit 13566a2

Browse files
Sol review fixes (#61)
* Make parallel_for_chunks exception-safe A throw from a worker chunk previously called std::terminate (unhandled exception in std::thread), and a throw from the thread-0 chunk destroyed joinable threads — either way aborting the whole process, e.g. for negative labels in the distributed block scanners with number_of_threads > 1. Capture the first exception, always join every worker, and rethrow on the calling thread. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * Compute standard deviations numerically stably The naive sum_of_squares/count - mean^2 formula suffers catastrophic cancellation for values with a large baseline and small spread (e.g. values [1e8, 1e8+1] gave std 0.0 instead of 0.5). - Distributed: switch the (n, 5) partial-stats layout from [count, sum, sum_of_squares, min, max] to the Welford/Chan representation [count, mean, M2, min, max]; merge via the Chan combine, finalize as variance = M2 / count. - In-core complex features: compute the variance two-pass from the values already stored for the percentiles (exact). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * Update the feature-merge accumulator in place merge_block_edge_stats copied the entire (E, 5) global accumulator on every call, making the documented per-block merge loop O(number_of_blocks x global_edges). Mutate the caller's accumulator in place instead (and return it, so existing loops keep working): one merge is now O(block edges) regardless of global graph size. The Python wrapper requires a C-contiguous writable float64 accumulator and no longer coerces it (a silent copy would discard the update). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * Deduplicate merge_edges via sort + unique The unordered_set added ~8x node/bucket overhead over the raw edge data (about 114 MB extra for one million edges). Canonicalize into a vector and sort + unique instead: same sorted-unique output, peak memory close to one copy of the input. Document that outputs are valid inputs, so callers can bound peak memory with a hierarchical merge. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * Reject non-integral and negative inputs in graph.distributed - own_begin / own_shape now require integer dtypes (floats such as [0.9, 0] were silently truncated by int()). - Affinity offsets go through operator.index, which rejects floats. - merge_edges rejects floating-dtype edge arrays and negative node ids in signed arrays (an int64 edge [-1, 2] silently wrapped to [2, 2**64 - 1] via the unconditional uint64 cast). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * Return the Python UndirectedGraph subclass from from_(unique_)edges The core statics return the raw _core.UndirectedGraph, so results lacked the subclass convenience wrappers (find_edges on lists, ...) and failed isinstance checks against bic.graph.UndirectedGraph. Add a constructor overload taking (number_of_nodes, uvs, unique) — an __init__ overload constructs the derived Python class, unlike a static — and build the subclass classmethods on it. from_edges also gains the single-pass C++ construction instead of insert_edges. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * Document the dense-label requirement of the distributed pipeline Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * Remove unused imports and variables (pyflakes) Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * Share one clipped grid sweep between in-core and distributed scans The 2D/3D offset-box sweeps in feature_accumulation.hxx and the owned-box sweeps in the distributed block extraction duplicated the same clamp + loop logic. Move a single sweep_clipped_box_{2d,3d} with a per-axis clip box into detail/grid.hxx; the in-core wrappers pass the thread slab with unclipped remaining axes, the distributed dispatch folds the owned box and slab into the clips. Inner loops are unchanged. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * Fix build issues on Mac by removing jthread --------- Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
1 parent fad254f commit 13566a2

16 files changed

Lines changed: 505 additions & 261 deletions

File tree

MIGRATION_GUIDE.md

Lines changed: 12 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -500,7 +500,11 @@ per-block subgraphs/features to zarr/N5/HDF5) is intentionally left to the
500500
caller, since I/O and block scheduling belong in Python.
501501

502502
The primitives assume **globally consistent labels** (a segment has the same id
503-
in every block — as after a stitched distributed watershed). A block owns the
503+
in every block — as after a stitched distributed watershed). Labels must also
504+
be reasonably **dense**: the global graph allocates memory proportional to the
505+
largest node id (`from_unique_edges(number_of_nodes, ...)` builds a dense CSR
506+
over ids `0 .. number_of_nodes - 1`), so sparse or very large globally unique
507+
id ranges need a relabeling pass before building the graph. A block owns the
504508
pixel-pairs whose reference pixel lies in its inner (non-halo) box, so the
505509
caller reads each block with a halo (≥1 on the forward faces for the region
506510
graph / an edge map; ≥ `max |offset|` per side for affinities) and passes the
@@ -519,19 +523,22 @@ block_edges, block_stats = d.block_affinity_stats(labels_block, aff_block, offse
519523
global_edges = d.merge_edges([edges_block_0, edges_block_1, ...])
520524
graph = bic.graph.UndirectedGraph.from_unique_edges(number_of_nodes, global_edges)
521525

522-
# fold per-block features onto the global edges, then finalize:
526+
# fold per-block features onto the global edges (in place), then finalize:
523527
acc = d.empty_edge_stats(graph.number_of_edges)
524528
for be, bs in per_block_stats:
525-
acc = d.merge_block_edge_stats(graph, acc, be, bs)
529+
d.merge_block_edge_stats(graph, acc, be, bs)
526530
features = d.finalize_edge_features(acc, compute_complex_features=True)
527531
```
528532

529533
Notes:
530534

531535
- Blocked results reproduce the whole-volume `region_adjacency_graph` /
532536
`features.*_features` exactly for `size`, `min` and `max`, and to
533-
floating-point tolerance for `mean` / `std` (the running sums depend on thread
534-
count and merge order).
537+
floating-point tolerance for `mean` / `std` (the running moments depend on
538+
thread count and merge order). The `(n, 5)` partial statistics use the
539+
numerically stable Welford/Chan representation `[count, mean, M2, min, max]`
540+
(`M2` = sum of squared deviations from the mean), so `std` stays accurate
541+
for values with a large baseline and small spread.
535542
- **Median and percentiles are not distributable.** The distributed complex
536543
output is the moment subset `[mean, std, min, max, size]` — the corresponding
537544
columns of the in-core 12-column complex features.

include/bioimage_cpp/detail/grid.hxx

Lines changed: 95 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
#pragma once
22

3+
#include <algorithm>
34
#include <cstddef>
45
#include <cstdint>
56
#include <vector>
@@ -111,6 +112,100 @@ inline void valid_axis_range(
111112
}
112113
}
113114

115+
// Sweep every (node, target) pair on a 2D row-major grid for which
116+
// `node + offset` stays inside the grid and the reference node lies in the
117+
// half-open clip box `[clip_y_lo, clip_y_hi) x [clip_x_lo, clip_x_hi)`. The
118+
// body receives flat C-order indices for both endpoints and is expected to
119+
// inline at -O2 since this is a header-only template with a fully-known
120+
// callable type at instantiation. Callers fold any additional restriction —
121+
// a per-thread axis-0 slab, an owned block box — into the clip box; passing
122+
// the full grid extent sweeps every valid reference node. Shared by the
123+
// in-core feature accumulation / lifted-edge sweeps and the distributed
124+
// block extraction.
125+
template <class Body>
126+
void sweep_clipped_box_2d(
127+
const std::ptrdiff_t dy,
128+
const std::ptrdiff_t dx,
129+
const std::size_t height,
130+
const std::size_t width,
131+
const std::size_t clip_y_lo,
132+
const std::size_t clip_y_hi,
133+
const std::size_t clip_x_lo,
134+
const std::size_t clip_x_hi,
135+
const Body &body
136+
) {
137+
std::size_t y_lo_v, y_hi_v, x_lo_v, x_hi_v;
138+
valid_axis_range(dy, height, y_lo_v, y_hi_v);
139+
valid_axis_range(dx, width, x_lo_v, x_hi_v);
140+
const auto y_lo = std::max(y_lo_v, clip_y_lo);
141+
const auto y_hi = std::min(y_hi_v, clip_y_hi);
142+
const auto x_lo = std::max(x_lo_v, clip_x_lo);
143+
const auto x_hi = std::min(x_hi_v, clip_x_hi);
144+
if (y_lo >= y_hi || x_lo >= x_hi) {
145+
return;
146+
}
147+
const auto offset_stride = dy * static_cast<std::ptrdiff_t>(width) + dx;
148+
for (std::size_t y = y_lo; y < y_hi; ++y) {
149+
const auto row_offset = y * width;
150+
for (std::size_t x = x_lo; x < x_hi; ++x) {
151+
const auto node = row_offset + x;
152+
const auto target = static_cast<std::uint64_t>(
153+
static_cast<std::ptrdiff_t>(node) + offset_stride
154+
);
155+
body(static_cast<std::uint64_t>(node), target);
156+
}
157+
}
158+
}
159+
160+
// 3D variant of `sweep_clipped_box_2d`.
161+
template <class Body>
162+
void sweep_clipped_box_3d(
163+
const std::ptrdiff_t dz,
164+
const std::ptrdiff_t dy,
165+
const std::ptrdiff_t dx,
166+
const std::size_t depth,
167+
const std::size_t height,
168+
const std::size_t width,
169+
const std::size_t clip_z_lo,
170+
const std::size_t clip_z_hi,
171+
const std::size_t clip_y_lo,
172+
const std::size_t clip_y_hi,
173+
const std::size_t clip_x_lo,
174+
const std::size_t clip_x_hi,
175+
const Body &body
176+
) {
177+
std::size_t z_lo_v, z_hi_v, y_lo_v, y_hi_v, x_lo_v, x_hi_v;
178+
valid_axis_range(dz, depth, z_lo_v, z_hi_v);
179+
valid_axis_range(dy, height, y_lo_v, y_hi_v);
180+
valid_axis_range(dx, width, x_lo_v, x_hi_v);
181+
const auto z_lo = std::max(z_lo_v, clip_z_lo);
182+
const auto z_hi = std::min(z_hi_v, clip_z_hi);
183+
const auto y_lo = std::max(y_lo_v, clip_y_lo);
184+
const auto y_hi = std::min(y_hi_v, clip_y_hi);
185+
const auto x_lo = std::max(x_lo_v, clip_x_lo);
186+
const auto x_hi = std::min(x_hi_v, clip_x_hi);
187+
if (z_lo >= z_hi || y_lo >= y_hi || x_lo >= x_hi) {
188+
return;
189+
}
190+
const auto slice_size = height * width;
191+
const auto offset_stride =
192+
dz * static_cast<std::ptrdiff_t>(slice_size) +
193+
dy * static_cast<std::ptrdiff_t>(width) + dx;
194+
for (std::size_t z = z_lo; z < z_hi; ++z) {
195+
const auto slice_offset = z * slice_size;
196+
for (std::size_t y = y_lo; y < y_hi; ++y) {
197+
const auto row_offset = slice_offset + y * width;
198+
for (std::size_t x = x_lo; x < x_hi; ++x) {
199+
const auto node = row_offset + x;
200+
const auto target = static_cast<std::uint64_t>(
201+
static_cast<std::ptrdiff_t>(node) + offset_stride
202+
);
203+
body(static_cast<std::uint64_t>(node), target);
204+
}
205+
}
206+
}
207+
}
208+
114209
// Number of leading positions to skip along an axis of length `length` for a
115210
// signed offset `delta` (i.e. `max(0, -delta)` clamped to `length`). Returns a
116211
// plain `std::ptrdiff_t` so the affinity kernels keep their inline

include/bioimage_cpp/detail/threading.hxx

Lines changed: 21 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -30,7 +30,7 @@ inline std::size_t normalize_thread_count(
3030

3131
// Split [0, number_of_work_items) into n_threads contiguous chunks and invoke
3232
// `chunk(thread_id, begin, end)` on each chunk. Thread 0 runs on the calling
33-
// thread; threads 1..n_threads-1 run in parallel std::jthreads. The caller is
33+
// thread; threads 1..n_threads-1 run in parallel std::threads. The caller is
3434
// responsible for picking n_threads via normalize_thread_count and for the
3535
// thread safety of `chunk`.
3636
//
@@ -70,17 +70,27 @@ void parallel_for_chunks(
7070
}
7171
};
7272

73-
// jthread's destructor joins, so a failure while creating a later worker
74-
// cannot destroy an earlier still-joinable thread and terminate the
75-
// process. We still join explicitly below to rethrow only after all work
76-
// has completed.
77-
std::vector<std::jthread> threads;
73+
// A joinable std::thread whose destructor runs without a prior join calls
74+
// std::terminate. If spawning a later worker throws (e.g. the OS refuses a
75+
// new thread), the threads vector would otherwise destroy the already
76+
// joinable workers and terminate the process, so join them first and then
77+
// rethrow. (std::jthread would join on destruction automatically, but it is
78+
// not available in AppleClang's libc++, and portable macOS wheels are a
79+
// hard requirement.)
80+
std::vector<std::thread> threads;
7881
threads.reserve(n_threads > 0 ? n_threads - 1 : 0);
79-
for (std::size_t thread_id = 1; thread_id < n_threads; ++thread_id) {
80-
const auto [begin, end] = bounds(thread_id);
81-
threads.emplace_back([thread_id, begin, end, &guarded_chunk]() {
82-
guarded_chunk(thread_id, begin, end);
83-
});
82+
try {
83+
for (std::size_t thread_id = 1; thread_id < n_threads; ++thread_id) {
84+
const auto [begin, end] = bounds(thread_id);
85+
threads.emplace_back([thread_id, begin, end, &guarded_chunk]() {
86+
guarded_chunk(thread_id, begin, end);
87+
});
88+
}
89+
} catch (...) {
90+
for (auto &thread : threads) {
91+
thread.join();
92+
}
93+
throw;
8494
}
8595
const auto [begin, end] = bounds(0);
8696
guarded_chunk(0, begin, end);

0 commit comments

Comments
 (0)