diff --git a/rugnux/Rugnux.cpp b/rugnux/Rugnux.cpp index 1935e9a3a..179f7f311 100644 --- a/rugnux/Rugnux.cpp +++ b/rugnux/Rugnux.cpp @@ -388,11 +388,16 @@ ProcessResult Rugnux::Run(RugnuxObserver *observer) { // first pass being invisible in it. Time the whole thing here, where every pass is inside. const auto run_start = std::chrono::steady_clock::now(); DiffractionExperiment header_experiment = experiment_; + // A stored sweep is all there from the start, so its first passes decide their near ties on all of + // its frames at once. Live, only the frames in so far are: the first passes decide on those, and + // their near ties are re-checked once the collection is over (decide_late_). + decide_late_ = !config_.live; // The run as it starts, to be made again where a near tie decided early ends elsewhere on the rest - // of the sweep (decide_late_). + // of the sweep. const auto again = std::make_shared(*this); again->near_tie_rechecks_ = std::make_shared(); again->decide_late_ = true; + again->decide_all_late_ = true; Rugnux *run = this; ProcessResult result; bool lost = false; diff --git a/rugnux/Rugnux.h b/rugnux/Rugnux.h index 48f30d0d3..100d78f03 100644 --- a/rugnux/Rugnux.h +++ b/rugnux/Rugnux.h @@ -105,6 +105,11 @@ struct ProcessConfig { // keep the fixed budget they are configured with. bool measure_spot_budget = false; + // The images arrive while the run goes on, from a collection still being written (rugnux following + // a broker, or --simulate-live), rather than all being there from the start. Decides how near ties + // are decided (Rugnux::decide_late_). + bool live = false; + // Rotation indexing (FullAnalysis) bool rotation_indexing = false; bool two_pass_rotation = true; @@ -991,18 +996,24 @@ class Rugnux { // way integrates the same reflections and can only tie. std::string stop_at_first_pass_outcome_; - // --- Near ties: decided early, re-checked late (see FirstPassRun::DecideLate) --- + // --- Near ties: offline decided late, live decided early and re-checked late (see FirstPassRun::DecideLate) --- // - // A first pass acts on its validation frames at once. Where it decided a near tie that the rest - // of the sweep could re-check, the copy of the run taken as its pass began is kept, and once the - // run is over - every frame is in by then, live too - that pass is run again on it as a probe, - // deciding late. Only where a re-check's first pass ends elsewhere than the pass's did is the - // run made again, every first pass deciding late. Offline the answer is then the same as deciding - // late throughout; live, no near tie holds the run up while the collection goes on. + // Offline every frame is there from the start, and a first pass decides a near tie on the later + // frames as well as its validation frames. Live (ProcessConfig::live), a first pass acts on its + // validation frames at once. Where it decided a near tie that the rest of the sweep could re-check, + // the copy of the run taken as its pass began is kept, and once the run is over - every frame is in + // by then - that pass is run again on it as a probe, deciding late. Only where a re-check's first + // pass ends elsewhere than the pass's did is the run made again, deciding everything late; no near + // tie holds the run up while the collection goes on. // - // Whether the first passes decide their near ties on all the frames: on a re-check, and on the run - // made again. + // Whether the first passes decide their near ties on all the frames: offline, on a re-check, and on + // the run made again. bool decide_late_ = false; + // Whether what the run otherwise decides on the pre-pass frames - the rotation-scale walk's refits, + // the demoted metric-symmetry arm, the integration radius (RefinedPasses) - is decided on all the + // frames: only on the run made again. Offline too these are decided on the pre-pass frames, and a near + // tie among them is re-checked after the run. + bool decide_all_late_ = false; // Whether this copy of the run is a re-check (for the log). bool near_tie_recheck_ = false; // Set by the first pass now running where it decided a near tie the later frames can re-check. diff --git a/rugnux/RugnuxFirstPass.h b/rugnux/RugnuxFirstPass.h index 4d3ddfb35..f055a6e11 100644 --- a/rugnux/RugnuxFirstPass.h +++ b/rugnux/RugnuxFirstPass.h @@ -172,15 +172,15 @@ struct Rugnux::FirstPassRun { bool NearTiePooled(const std::string &decision, const std::string &a, const PooledEvidence &ea, const std::string &b, const PooledEvidence &eb) const; - // A near tie is decided early and checked late. The pass acts on the validation frames at once; - // the comparison made again, unchanged, on the validation frames and late_validation together - - // a gate set at a share of the validation frames at the same share of all of them - is the - // re-check, which Run() makes once the rest of the sweep is in (Rugnux::decide_late_). A + // A near tie is decided on the validation frames and late_validation together - the comparison made + // again, unchanged, a gate set at a share of the validation frames at the same share of all of + // them. Offline that is the decision; live the pass acts on the validation frames at once, and this + // is the re-check, which Run() makes once the rest of the sweep is in (Rugnux::decide_late_). A // decision that is not a near tie never reads later frames; nor does one on a sweep the first // part already covers, where there is nothing later to read (late_validation, below). // - // Whether this near tie is decided on all the frames now: on the re-check, and on the run Run() - // makes again where a re-check ended elsewhere. Otherwise it notes that the pass carried one. + // Whether this near tie is decided on all the frames now: offline, on the re-check, and on the run + // Run() makes again where a re-check ended elsewhere. Otherwise it notes that the pass carried one. bool DecideLate() const; // The validation frames followed by late_validation. std::vector AllValidationFrames() const; diff --git a/rugnux/RugnuxImagePass.cpp b/rugnux/RugnuxImagePass.cpp index 4c95d0593..3a29de032 100644 --- a/rugnux/RugnuxImagePass.cpp +++ b/rugnux/RugnuxImagePass.cpp @@ -147,9 +147,9 @@ void Rugnux::ImagePass(PipelineLocals &p) { // The crowding of the background rings at the widened radii, counted on frames spread over the // pass (widened_crowding_): on the geometry pre-pass, which integrates at the fixed radius while the // widened one is held back (bragg_adaptive_), and on a canonical pass a near tie left at the fixed - // radius (radius_recheck_). A run deciding late decides the radius on the canonical pass itself. + // radius (radius_recheck_). The run made again decides the radius on the canonical pass itself. const std::optional crowding_at = - geometry_prepass ? (decide_late_ ? std::nullopt : bragg_adaptive_) + geometry_prepass ? (decide_all_late_ ? std::nullopt : bragg_adaptive_) : prepass_frames_only_ ? std::nullopt : radius_recheck_; const int crowding_stride = std::max(1, (geometry_prepass ? prepass_images : images_to_process) / CROWDING_FRAMES); diff --git a/rugnux/RugnuxPasses.cpp b/rugnux/RugnuxPasses.cpp index 4464e65ce..986c273f6 100644 --- a/rugnux/RugnuxPasses.cpp +++ b/rugnux/RugnuxPasses.cpp @@ -668,7 +668,7 @@ RotationScaleWalk Rugnux::WalkRotationScaleBeside(const ProcessResult &pass1, Ro // A refit reads only the scale its post-refinement fits on top of k - a step for the walk // to take, which the validation frames of the whole sweep then judge - so it is fitted on // the frames pass 1 fitted the first one on. - r->prepass_frames_only_ = !index_only && !decide_late_; + r->prepass_frames_only_ = !index_only && !decide_all_late_; // An indexing probe runs beside the refined passes, on the same card, and gives its memory // back in seconds: what of theirs does not fit beside it waits for it to end. A refit // runs only once they have ended, so it is nobody's work beside. @@ -984,7 +984,7 @@ std::optional Rugnux::RefinedPasses(RugnuxObserver *observer, Pro demoted_run.lattice_angles_demoted_ = angle_arm; demoted_run.stop_at_first_pass_outcome_ = promoted_probe.first_pass_outcome_; // The arm is asked on the frames the pre-pass read, as the pre-pass asked its own fit, and - // the canonical pass compares its fit on the same frames. A run deciding late asks it on + // the canonical pass compares its fit on the same frames. The run made again asks it on // all of them; a near tie keeps this copy to do that once the run is over. demoted_recheck = std::make_shared(demoted_run); demoted_recheck->decide_late_ = true; @@ -992,7 +992,7 @@ std::optional Rugnux::RefinedPasses(RugnuxObserver *observer, Pro demoted_recheck->first_pass_memo_.clear(); demoted_recheck->speculative_probe_.reset(); demoted_recheck->rotation_scale_check_.reset(); - demoted_run.prepass_frames_only_ = !decide_late_; + demoted_run.prepass_frames_only_ = !decide_all_late_; demoted_arm = demoted_run.RunPipeline(observer, /*write_output=*/false, /*geometry_prepass=*/false); ++arm_passes; for (const auto &m : demoted_run.first_pass_memo_) diff --git a/rugnux/rugnux_cli.cpp b/rugnux/rugnux_cli.cpp index ab0f8e056..714bd73d1 100644 --- a/rugnux/rugnux_cli.cpp +++ b/rugnux/rugnux_cli.cpp @@ -106,6 +106,12 @@ void print_usage() { "collection delivered. Rotation data, --mode mx only. The collection's token is read from the " "environment variable JUNGFRAUJOCH_HTTP_TOKEN and sent only to the broker. With on disk, " "the URL is ignored" << std::endl; + std::cout << " --simulate-live Read the stored as the collection it was while being written, " + "the way a broker URL follows one: image i becomes readable (i+1)* ms after the run starts (0: all " + "at once). The run then takes the live path - the first passes decide their near ties on the images in " + "so far, re-checked once the last one is in, and the run is made again where one ends elsewhere - " + "where on a stored sweep they decide on all of its images at once. Same result; for testing the " + "live workflow without a broker. --mode mx only" << std::endl; std::cout << "Options:" << std::endl; std::cout << " -o, --output-prefix Output file prefix (default: output). mx and scale runs always write " "_report.txt, the results report; every run writes _log.txt, the full timestamped log " @@ -310,6 +316,7 @@ enum { OPT_NO_SCALE_FULLS, OPT_WRITE_PROCESS_H5, OPT_FINALIST_LEDGER, + OPT_SIMULATE_LIVE, OPT_DEVELOPER, OPT_FORCE_STILL, OPT_AZIM_MIN_Q, @@ -366,6 +373,7 @@ static option long_options[] = { {"no-scale-fulls", no_argument, nullptr, OPT_NO_SCALE_FULLS}, {"write-process-h5", no_argument, nullptr, OPT_WRITE_PROCESS_H5}, {"finalist-ledger", no_argument, nullptr, OPT_FINALIST_LEDGER}, + {"simulate-live", required_argument, nullptr, OPT_SIMULATE_LIVE}, {"developer", no_argument, nullptr, OPT_DEVELOPER}, {"smooth-g", optional_argument, nullptr, OPT_SMOOTH_G}, {"relative-b", optional_argument, nullptr, OPT_RELATIVE_B}, @@ -652,6 +660,20 @@ void configure_offline_output(DiffractionExperiment &experiment, const std::stri namespace { std::atomic g_active_process{nullptr}; + // --simulate-live: a stored sweep read as the collection it was while being written - image i is + // read no earlier than (i + 1) frame times after the run began. + std::chrono::microseconds simulated_frame_time{0}; + std::chrono::steady_clock::time_point simulated_start; + + template + class SimulatedLive : public Reader { + public: + bool ReadRawImage(int64_t image_number, JFJochReaderRawImage &image) override { + std::this_thread::sleep_until(simulated_start + simulated_frame_time * (image_number + 1)); + return Reader::ReadRawImage(image_number, image); + } + }; + // The terminal (see RugnuxConsole). Made first thing in RunRugnux; main uses it too, to close the // run on an error that escapes it. std::shared_ptr console; @@ -907,6 +929,7 @@ static int RunRugnux(int argc, char **argv) { bool run_scaling = true; // merge is on by default; --no-merge turns it off std::optional scale_fulls_arg; // --scale-fulls / --no-scale-fulls; default on for rot3d bool write_process_h5_flag = false; // --write-process-h5; also write _process.h5 when merging + bool simulate_live = false; // --simulate-live bool finalist_ledger_flag = false; // --finalist-ledger; report-only symmetry evidence table std::optional detect_ice_rings; // --detect-ice-rings[=on|off]; unset => use the dataset (file) value bool index_ice_rings = false; // --index-ice-rings[=on|off]; index on the ice-band spots too @@ -1330,6 +1353,11 @@ static int RunRugnux(int argc, char **argv) { case OPT_FINALIST_LEDGER: finalist_ledger_flag = true; break; + case OPT_SIMULATE_LIVE: + simulate_live = true; + simulated_frame_time = std::chrono::microseconds( + std::lround(1000.0 * parse_double_arg(optarg, "--simulate-live", logger))); + break; case OPT_DEVELOPER: provenance.developer = true; break; @@ -1632,10 +1660,10 @@ static int RunRugnux(int argc, char **argv) { // any frame of a sweep reads the whole sweep, which is how these are archived and how a user // thinks of them. console->OnStep("Reading input"); - JFJochHDF5Reader hdf5_reader; - JFJochCBFReader cbf_reader; - JFJochMarCCDReader marccd_reader; - JFJochSMVReader smv_reader; + SimulatedLive hdf5_reader; + SimulatedLive cbf_reader; + SimulatedLive marccd_reader; + SimulatedLive smv_reader; JFJochStartMessageReader live_reader; std::unique_ptr broker_feed; JFJochReader *reader_ptr = nullptr; @@ -3103,6 +3131,11 @@ static int RunRugnux(int argc, char **argv) { // _process.h5 unless explicitly requested. Without merging, the _process.h5 is the only output. config.write_process_h5 = run_scaling ? write_process_h5_flag : true; config.finalist_ledger = finalist_ledger_flag; + config.live = broker_feed || simulate_live; + if (simulate_live) + logger.Info("Simulated live collection: image i readable (i+1) x {:.1f} ms after the run starts", + simulated_frame_time.count() / 1000.0); + simulated_start = std::chrono::steady_clock::now(); Rugnux process(reader, experiment, *dataset->pixel_mask, config);