Merge branch 'calib-and-memory' into rc173 (calibration re-bins every image; clean out-of-host-memory errors; large-cell memory docs)

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01D1G8gJVAy6gp1K5Dz3NE5C

# Conflicts:
#	docs/CHANGELOG.md
#	rugnux/Rugnux.cpp
This commit is contained in:
leonarski_f committed 2026-09-28 23:26:49 +02:00
commit d1114b37de
12 files changed
+280 -58

No files matched your search

+18 -8
View File
@@ -60,16 +60,26 @@ namespace parallel_detail {
WorkerPool() {
const unsigned hw = std::max(1u, std::thread::hardware_concurrency());
workers.reserve(hw - 1);
for (unsigned i = 0; i + 1 < hw; i++) // the submitting thread takes a share too
workers.emplace_back([this] {
// It inherited whatever CPUs the thread that first used the pool was kept to.
RestoreThreadAffinity();
InWorkerFlag() = true;
Loop();
});
try {
for (unsigned i = 0; i + 1 < hw; i++) // the submitting thread takes a share too
workers.emplace_back([this] {
// It inherited whatever CPUs the thread that first used the pool was kept to.
RestoreThreadAffinity();
InWorkerFlag() = true;
Loop();
});
} catch (...) {
// A thread that cannot be started (no memory left for its stack) fails the caller with
// that error - after joining the ones already running, since destroying a joinable
// std::thread calls std::terminate instead.
StopAndJoin();
throw;
}
}
~WorkerPool() {
~WorkerPool() { StopAndJoin(); }
void StopAndJoin() {
{
std::lock_guard lock(m);
stop = true;
+2
View File
@@ -18,6 +18,8 @@
* Rugnux without a GPU no longer stops Bragg prediction at 20,000 reflections per frame, so a large unit cell integrates the same reflections on a CPU-only build as on a GPU.
* Rugnux keeps a reflection's background ring clear only of the neighbouring reflections that put at least 5% of their flux on the frame, so finely sliced dense rotation data no longer lose most of their observations for want of a background.
* Rugnux's `--model` rigid-body refinement runs on the GPU where one is available, the real fit and the null's replicates alike, and falls back to the CPU otherwise; placements can differ slightly between the two.
* `rugnux --mode calibration` re-bins every image of a multi-image powder run for its second fit, not only the last image each thread read, and gives the same result at any thread count.
* Rugnux reports running out of host memory as `out of host memory ... - this data set needs more RAM than is available` and exits non-zero, instead of `std::bad_alloc` or a crash.
### 1.0.0-rc.173
+2 -1
View File
@@ -110,7 +110,8 @@ A few things worth knowing before reaching for more flags:
- **A run wants memory in proportion to what it integrates**, not to the detector: 2.5-14 GB of
host RAM and 3-7 GB on the card over the datasets measured, both peaking in scaling and
merging. [Installing Rugnux ▸ Memory](RUGNUX_INSTALL.md#memory) has the table and the two
flags that lower it.
flags that lower it. A very large cell needs far more, and a CPU-only build most of all — tens of
GB of host RAM ([Very large unit cells](RUGNUX_INSTALL.md#very-large-unit-cells)).
- Everything else is in [Running Rugnux](RUGNUX_TUTORIAL.md#running-rugnux) and the full
[Command-line options](RUGNUX_ADVANCED.md#command-line-options).
+35
View File
@@ -169,3 +169,38 @@ non-zero exit status. That is deliberate: a CUDA out-of-memory says nothing abou
flight and everything about the machine, so skipping frames would leave a run that looks complete
with quietly different merged numbers. A run is never silently short; it either completes or fails.
On a GPU shared with another job, lower `-N` or wait for the card.
### Very large unit cells
The rule above holds for the cells in that table, and a very large cell sits far beyond it. A long
axis, a dense pattern and many frames multiply together: each frame predicts more reflections, each
reflection is split over more frames, and **tens of millions of partial observations are normal** for
a big cell. One sweep of a crystal with an axis of about 640 Å predicted 236 million reflections and
peaked at **17.5 GB of host memory on a GPU build and 72.5 GB on a CPU-only build**. A CPU-only build
keeps in host RAM what a GPU build keeps on the card, so it is the one that needs the large machine.
When such a set is too large for GPU scaling on the card, the run stops and says to run it on the
CPU — which then wants that much host memory.
### When host memory runs out
Nothing estimates the host memory a run will need before it starts, so a set that does not fit runs
until the memory is gone, and then ends in one of two ways.
Where the allocation is refused — under an address-space limit (`ulimit -v`, which some batch
systems set), with strict overcommit, or for one request larger than the machine could ever grant —
the run stops with
```
Processing failed: out of host memory (std::bad_alloc) - this data set needs more RAM than is available
```
and a non-zero exit status. The CUDA driver reserves a large address space of its own, so a GPU build
under a tight `ulimit -v` can stop before reading anything, with `CUDA (GPU) error (out of memory)`.
More often on Linux the allocation succeeds and the memory runs out as it is used. The kernel's
**OOM killer** then ends the process with SIGKILL, which no program can catch: the run stops with no
message of its own, the shell prints `Killed` (exit status 137), and `dmesg` or the batch system's
log has an `Out of memory: Killed process` line. A job capped by a cgroup memory limit ends the same
way. A run that disappears like this needs more RAM than the machine or the job was given, exactly as
if it had said so. `--no-merge` lowers the host peak, as above; otherwise the run needs a larger
machine.
+1 -1
View File
@@ -303,7 +303,7 @@ public:
CudaHostPtr() = default;
explicit CudaHostPtr(size_t count) {
if (cudaMallocHost(&ptr_, count * sizeof(T)) != cudaSuccess)
throw JFJochException(JFJochExceptionCategory::GPUCUDAError,
throw JFJochException(JFJochExceptionCategory::MemAllocFailed,
"Failed to allocate pinned host memory");
}
~CudaHostPtr() {
@@ -6,7 +6,7 @@
#include <iostream>
#include <limits>
#include <thread>
#include <future>
namespace {
struct config_ifssr final {
@@ -160,15 +160,17 @@ std::vector<CrystalLattice> Refine(const std::vector<Coord> &in_spots,
if (nblocks == 1) {
RefineCandidateCells(spots.topRows(nspots), oCell, scores, cifssr);
} else {
std::vector<std::thread> workers;
// Futures, not bare threads: an exception in a block (out of memory) then reaches the caller
// instead of calling std::terminate.
std::vector<std::future<void>> workers;
workers.reserve(nblocks - 1);
for (unsigned b = 1; b < nblocks; b++)
workers.emplace_back([&, b] {
workers.push_back(std::async(std::launch::async, [&, b] {
RefineCandidateCells(spots.topRows(nspots), oCell, scores, cifssr, b, nblocks);
});
}));
RefineCandidateCells(spots.topRows(nspots), oCell, scores, cifssr, 0, nblocks);
for (auto &w : workers)
w.join();
w.get();
}
std::vector<RefinedCandidate> candidates;
@@ -654,12 +654,14 @@ namespace {
size_t free_bytes = 0, total_bytes = 0;
cudaMemGetInfo(&free_bytes, &total_bytes);
cudaGetLastError();
char msg[640];
char msg[768];
snprintf(msg, sizeof(msg),
"Scaling %lld partial observations needs more GPU memory than this card has: %.1f GB for "
"their per-observation arrays alone (%zu bytes each), before the merge's other buffers, "
"on a %.1f GB card. This data set is too large for GPU scaling on this card - run it on "
"the CPU (CUDA_VISIBLE_DEVICES= rugnux ..., or a CPU-only build)",
"the CPU (CUDA_VISIBLE_DEVICES= rugnux ..., or a CPU-only build) on a machine with plenty of "
"host memory: the CPU path keeps all of it in RAM, which for a set this large can be "
"tens of GB",
static_cast<long long>(n_obs), double(n_obs) * OBS_BYTES / 1e9, OBS_BYTES,
double(total_bytes) / 1e9);
throw JFJochException(JFJochExceptionCategory::GPUCUDAError, msg);
+30 -2
View File
@@ -2,15 +2,31 @@
// SPDX-License-Identifier: GPL-3.0-only
#include <vector>
#include <csetjmp>
#include <cstdio>
#include <cstdint>
#include <cmath>
#include <new>
#include <jpeglib.h>
#include <jerror.h>
#include "../common/JFJochException.h"
#include "../common/CompressedImage.h"
#include "JFJochJPEG.h"
namespace {
// libjpeg's own error handler prints its message and calls exit(), from whichever thread was
// encoding - out of memory included. This one jumps back to WriteJPEGToMem, which throws instead.
struct JPEGErrorManager {
jpeg_error_mgr mgr;
std::jmp_buf jump;
};
void JPEGErrorExit(j_common_ptr cinfo) {
std::longjmp(reinterpret_cast<JPEGErrorManager *>(cinfo->err)->jump, 1);
}
}
std::string WriteJPEGToMem(const CompressedImage& image, int quality) {
unsigned char *buf = nullptr;
@@ -24,9 +40,21 @@ std::string WriteJPEGToMem(const CompressedImage& image, int quality) {
image.GetUncompressed(buffer);
struct jpeg_compress_struct cinfo;
struct jpeg_error_mgr jerr;
JPEGErrorManager jerr;
cinfo.err = jpeg_std_error(&jerr);
cinfo.err = jpeg_std_error(&jerr.mgr);
jerr.mgr.error_exit = JPEGErrorExit;
if (setjmp(jerr.jump)) {
const bool out_of_memory = (jerr.mgr.msg_code == JERR_OUT_OF_MEMORY);
char message[JMSG_LENGTH_MAX];
jerr.mgr.format_message(reinterpret_cast<j_common_ptr>(&cinfo), message);
jpeg_destroy_compress(&cinfo);
free(buf);
if (out_of_memory)
throw std::bad_alloc();
throw JFJochException(JFJochExceptionCategory::InputParameterInvalid,
std::string("JPEG encoding failed: ") + message);
}
jpeg_create_compress(&cinfo);
jpeg_mem_dest(&cinfo, &buf, &buf_size);
+30 -35
View File
@@ -147,7 +147,7 @@ namespace {
// the dataset and still exits 0, so the run looks complete while every merged number downstream
// has quietly changed. Fail the run instead, and let the operator free the card and repeat it.
bool IsFatalResourceError(const std::exception &e) {
if (dynamic_cast<const std::bad_alloc *>(&e)) return true;
if (dynamic_cast<const std::bad_alloc *>(&e) || dynamic_cast<const std::length_error *>(&e)) return true;
// Not a resource, but just as fatal: a memo verification (verify_first_pass_memo) that found a
// difference must end the run, not pass for a probe that scored nothing.
if (dynamic_cast<const MemoMismatch *>(&e)) return true;
@@ -750,42 +750,37 @@ std::optional<CalibrationResult> Rugnux::RebinAndRefit(const CalibrationResult &
.PoniRot2_rad(geometry.GetPoniRot2_rad());
const AzimuthalIntegrationMapping rebinned_mapping(rebinned_experiment, pixel_mask_);
std::atomic<int> next_ordinal = 0;
std::mutex merge_m;
AzimuthalIntegrationProfile summed(rebinned_mapping);
// The sum over every image. The profile is a float sum, so the order it is added up in sets its
// last bits: each fixed block of images is folded in image order into its own slot, and the
// slots are added in block order, which gives the same profile at any thread count.
std::vector<std::unique_ptr<AzimuthalIntegrationProfile>> block_sum(ReductionBlocks(images_to_process, 1));
ParallelBlocks(images_to_process, std::max(1, config_.nthreads), [&](int b, int lo, int hi) {
std::vector<uint8_t> decompression_buffer;
JFJochReaderRawImage img;
ImagePreprocessorCPU preprocessor(experiment_, pixel_mask_);
ImagePreprocessorBuffer buffer(experiment_.GetPixelsNum());
AzIntEngineCPU azint(rebinned_mapping);
AzimuthalIntegrationProfile image_profile(rebinned_mapping); // cleared by every azint.Run
auto block = std::make_unique<AzimuthalIntegrationProfile>(rebinned_mapping);
const int nthreads = std::max(1, config_.nthreads);
std::vector<std::future<void>> workers;
workers.reserve(nthreads);
for (int t = 0; t < nthreads; ++t) {
workers.emplace_back(std::async(std::launch::async, [&]() {
std::vector<uint8_t> decompression_buffer;
JFJochReaderRawImage img;
ImagePreprocessorCPU preprocessor(experiment_, pixel_mask_);
ImagePreprocessorBuffer buffer(experiment_.GetPixelsNum());
AzIntEngineCPU azint(rebinned_mapping);
AzimuthalIntegrationProfile local(rebinned_mapping);
while (!cancelled_) {
const int ordinal = next_ordinal.fetch_add(1);
const int image_idx = start_image + ordinal * config_.stride;
if (image_idx >= end_image) break;
try {
if (!reader_.ReadRawImage(image_idx, img)) continue;
const uint8_t *image_ptr = img.image.GetUncompressedPtr(decompression_buffer);
preprocessor.Analyze(buffer, image_ptr, img.image.GetMode());
azint.Run(buffer, local);
} catch (const std::exception &e) {
if (IsFatalResourceError(e)) throw;
logger.Error("Failed to re-integrate image {}: {}", image_idx, e.what());
}
for (int ordinal = lo; ordinal < hi && !cancelled_; ++ordinal) {
const int image_idx = start_image + ordinal * config_.stride;
try {
if (!reader_.ReadRawImage(image_idx, img)) continue;
const uint8_t *image_ptr = img.image.GetUncompressedPtr(decompression_buffer);
preprocessor.Analyze(buffer, image_ptr, img.image.GetMode());
azint.Run(buffer, image_profile);
*block += image_profile;
} catch (const std::exception &e) {
if (IsFatalResourceError(e)) throw;
logger.Error("Failed to re-integrate image {}: {}", image_idx, e.what());
}
std::lock_guard lock(merge_m);
summed += local;
}));
}
for (auto &w : workers)
w.get();
}
block_sum[b] = std::move(block);
}, 1);
AzimuthalIntegrationProfile summed(rebinned_mapping);
for (const auto &block : block_sum)
summed += *block;
if (cancelled_)
return std::nullopt;
+40 -4
View File
@@ -5,6 +5,9 @@
#include <atomic>
#include <chrono>
#include <cmath>
#include <cstdio>
#include <cstdlib>
#include <cstring>
#include <csignal>
#include <filesystem>
#include <future>
@@ -13,7 +16,9 @@
#include <iostream>
#include <limits>
#include <sstream>
#include <stdexcept>
#include <string>
#include <system_error>
#include <thread>
#include <type_traits>
#include <vector>
@@ -661,6 +666,37 @@ namespace {
}
return out;
}
// Running out of host memory ends a run with an exception that names only what failed in C++ terms -
// std::bad_alloc, a vector past its maximum size, a thread that could not get its stack - which
// does not tell the user what to do. The data set is not at fault; the machine is too small for it.
std::string FailureText(const std::exception &e) {
const auto *sys = dynamic_cast<const std::system_error *>(&e);
const auto *jf = dynamic_cast<const JFJochException *>(&e);
const bool out_of_memory = dynamic_cast<const std::bad_alloc *>(&e) != nullptr
|| dynamic_cast<const std::length_error *>(&e) != nullptr
|| (sys != nullptr && sys->code() == std::errc::resource_unavailable_try_again)
|| (jf != nullptr && jf->Category() == JFJochExceptionCategory::MemAllocFailed);
if (!out_of_memory)
return e.what();
return fmt::format("out of host memory ({}) - this data set needs more RAM than is available",
e.what());
}
}
// FFTW ends the program itself when its allocator comes back empty (CK(p) in kernel/alloc.c) - while
// planning, or inside a buffered transform - from whichever thread that happens in, so it cannot become
// an exception. Defining its handler here keeps FFTW's own (kernel/assert.c) out of the link, and says
// what happened in the same words as every other way of running out of memory.
extern "C" void fftwf_assertion_failed(const char *s, int line, const char *file) {
std::fflush(stdout);
if (std::strcmp(s, "p") == 0) {
std::fprintf(stderr, "Processing failed: out of host memory (FFTW could not allocate) - "
"this data set needs more RAM than is available\n");
std::_Exit(EXIT_FAILURE);
}
std::fprintf(stderr, "fftw: %s:%d: assertion failed: %s\n", file, line, s);
std::abort();
}
// The body of main. Settings setters and the pipeline itself throw JFJochException on input the parser
@@ -2151,7 +2187,7 @@ static int RunRugnux(int argc, char **argv) {
try {
result = process.Run();
} catch (const std::exception &e) {
logger.Error("Processing failed: {}", e.what());
logger.Error("Processing failed: {}", FailureText(e));
exit(EXIT_FAILURE);
}
g_active_process = nullptr;
@@ -2244,7 +2280,7 @@ static int RunRugnux(int argc, char **argv) {
try {
result = process.Run();
} catch (const std::exception &e) {
logger.Error("Calibration failed: {}", e.what());
logger.Error("Calibration failed: {}", FailureText(e));
return 1;
}
g_active_process = nullptr;
@@ -2812,7 +2848,7 @@ static int RunRugnux(int argc, char **argv) {
WriteResultReport(output_prefix, input_file, experiment, failed, logger, provenance);
exit(EXIT_FAILURE);
} catch (const std::exception &e) {
logger.Error("Processing failed: {}", e.what());
logger.Error("Processing failed: {}", FailureText(e));
exit(EXIT_FAILURE);
}
g_active_process = nullptr;
@@ -2944,7 +2980,7 @@ int main(int argc, char **argv) {
enable_gpu_numa_binding();
return RunRugnux(argc, argv);
} catch (const std::exception &e) {
Logger("rugnux").Error("{}", e.what());
Logger("rugnux").Error("{}", FailureText(e));
return EXIT_FAILURE;
}
}
+7
View File
@@ -33,3 +33,10 @@ TEST_CASE("JPEGTest","[JPEG]") {
std::ofstream f("x.jpeg", std::ios::binary);
f.write(s.data(), s.size());
}
// libjpeg's own error handler calls exit(); an image it refuses must come back as an exception.
TEST_CASE("JPEGTest_ErrorThrows","[JPEG]") {
const std::vector<rgb> val(70000); // wider than JPEG_MAX_DIMENSION
const CompressedImage image(val, 70000, 1);
CHECK_THROWS_AS(WriteJPEGToMem(image), JFJochException);
}
+104
View File
@@ -12,6 +12,7 @@
#include "../rugnux/Rugnux.h"
#include "../rugnux/RugnuxCommandLine.h"
#include "../rugnux/SpotWidth.h"
#include "../image_analysis/geom_refinement/Calibrants.h"
namespace {
// Write a small VDS dataset of `n` flat images and return nothing (prefix_master.h5 +
@@ -135,6 +136,109 @@ TEST_CASE("Rugnux_Cancel", "[HDF5][Full]") {
REQUIRE(H5Fget_obj_count(H5F_OBJ_ALL, H5F_OBJ_ALL) == 0);
}
// A calibration re-integrates its images binned about the geometry it fitted, and fits again. That
// second profile must be the sum over every image: here each of six frames carries a different sixth
// of the LaB6 rings, so a profile missing a frame is missing rings. The header's beam centre is 5 px
// out, far enough that the first pass's smeared profile loses to the re-binned one - so the geometry
// that comes out is the one fitted on the re-binned sum, and it must not depend on the thread count.
TEST_CASE("Rugnux_CalibrationRebinsEveryImage", "[HDF5][DetGeomCalib]") {
RegisterHDF5Filter();
constexpr int n = 6;
constexpr float true_beam_x = 1090.0f, true_beam_y = 1100.0f;
const std::vector<float> rings_q = CalibrantRings("lab6");
DiffractionExperiment x(DetJF4M());
x.FilePrefix("process_calib_in").ImagesPerTrigger(n).OverwriteExistingFiles(true);
x.BitDepthImage(16).ImagesPerFile(n).SetFileWriterFormat(FileWriterFormat::NXmxVDS).PixelSigned(true);
x.Compression(CompressionAlgorithm::NO_COMPRESSION);
x.BeamX_pxl(true_beam_x).BeamY_pxl(true_beam_y).DetectorDistance_mm(100).IncidentEnergy_keV(WVL_1A_IN_KEV)
.FrameTime(std::chrono::microseconds(500), std::chrono::microseconds(10));
const DiffractionGeometry geom_true = x.GetDiffractionGeometry();
x.BeamX_pxl(true_beam_x + 5.0f); // what the header says
// Radii of the rings the beam reaches, in pixels about the true centre.
std::vector<float> ring_radius;
for (const float q : rings_q) {
const float d = static_cast<float>(2.0 * PI) / q;
if (d > geom_true.GetWavelength_A() / 2.0f)
ring_radius.push_back(geom_true.ResToPxl(d));
}
{
StartMessage start_message;
x.FillMessage(start_message);
FileWriter file_set(start_message);
ScanResultGenerator generator(x);
std::vector<int16_t> image(x.GetPixelsNum());
for (int i = 0; i < n; i++) {
for (int64_t p = 0; p < x.GetPixelsNum(); p++) {
const float px = static_cast<float>(p % x.GetXPixelsNum());
const float py = static_cast<float>(p / x.GetXPixelsNum());
const float r = std::hypot(px - true_beam_x, py - true_beam_y);
float v = 5.0f;
for (size_t k = i; k < ring_radius.size(); k += n) {
const float dr = r - ring_radius[k];
v += 400.0f * std::exp(-0.5f * dr * dr / (1.5f * 1.5f));
}
image[p] = static_cast<int16_t>(std::lround(v));
}
DataMessage message{};
message.image = CompressedImage(image, x.GetXPixelsNum(), x.GetYPixelsNum());
message.number = i;
REQUIRE_NOTHROW(file_set.WriteHDF5(message));
generator.Add(message);
}
EndMessage end_message;
end_message.max_image_number = n;
generator.FillEndMessage(end_message);
file_set.WriteHDF5(end_message);
file_set.Finalize();
}
JFJochHDF5Reader reader;
REQUIRE_NOTHROW(reader.ReadFile("process_calib_in_master.h5"));
auto dataset = reader.GetDataset();
REQUIRE(dataset);
DiffractionExperiment experiment(dataset->experiment);
auto azint = experiment.GetAzimuthalIntegrationSettings();
azint.AzimuthalBinCount(32);
experiment.ImportAzimuthalIntegrationSettings(azint);
std::vector<CalibrationResult> results;
for (const int nthreads : {1, 4}) {
ProcessConfig config;
config.mode = ProcessMode::Calibration;
config.calibrant_ring_q = rings_q;
config.nthreads = nthreads;
config.write_process_h5 = false;
config.beam_center_check = false;
config.spot_finding = DiffractionExperiment::DefaultDataProcessingSettings();
config.spot_finding.indexing = false;
Rugnux process(reader, experiment, *dataset->pixel_mask, config);
ProcessResult result;
REQUIRE_NOTHROW(result = process.Run());
REQUIRE(result.calibration.has_value());
results.push_back(*result.calibration);
}
for (const auto &cal : results) {
CHECK(cal.converged);
CHECK(cal.geometry.GetBeamX_pxl() == Catch::Approx(true_beam_x).margin(0.5));
CHECK(cal.geometry.GetBeamY_pxl() == Catch::Approx(true_beam_y).margin(0.5));
}
CHECK(results[0].ring_points == results[1].ring_points);
CHECK(results[0].rms_radial_pxl == results[1].rms_radial_pxl);
CHECK(results[0].geometry.GetBeamX_pxl() == results[1].geometry.GetBeamX_pxl());
CHECK(results[0].geometry.GetBeamY_pxl() == results[1].geometry.GetBeamY_pxl());
CHECK(results[0].geometry.GetDetectorDistance_mm() == results[1].geometry.GetDetectorDistance_mm());
reader.Close();
remove("process_calib_in_master.h5");
remove("process_calib_in_data_000001.h5");
REQUIRE(H5Fget_obj_count(H5F_OBJ_ALL, H5F_OBJ_ALL) == 0);
}
TEST_CASE("RugnuxCommandLine_Full", "[process]") {
DiffractionExperiment x(DetJF(1));
IndexingSettings idx;