Merge rc175 into review-fixes-rc175
This commit is contained in:
+6
-1
@@ -47,11 +47,16 @@ ENDIF()
|
||||
# came out differently from the same source and data. Off, every x86-64 level computes the same bits
|
||||
# (measured: identical p.mtz on nine battery sets, no measurable time cost), and aarch64, where FMA is
|
||||
# always there and GCC contracts by default, follows the same arithmetic. MSVC does not contract under
|
||||
# its default /fp:precise. Host code of .cu sources gets it through nvcc; device code is not affected.
|
||||
# its default /fp:precise. Host code of .cu sources gets it through nvcc.
|
||||
IF (NOT MSVC)
|
||||
ADD_COMPILE_OPTIONS($<$<COMPILE_LANGUAGE:C,CXX>:-ffp-contract=off>)
|
||||
ADD_COMPILE_OPTIONS($<$<COMPILE_LANGUAGE:CUDA>:-Xcompiler=-ffp-contract=off>)
|
||||
ENDIF()
|
||||
# The same for device code. nvcc contracts by default and does it separately for every architecture
|
||||
# it compiles for, so a T4 (sm_75) and an RTX 5080 (sm_120) ran the same kernels to different bits
|
||||
# and a run that hit a knife edge came out differently on the two machines. Off, the architectures
|
||||
# compute the same bits (measured: identical p.hkl from sm_75 and sm_120 builds, no measurable time cost).
|
||||
ADD_COMPILE_OPTIONS($<$<COMPILE_LANGUAGE:CUDA>:--fmad=false>)
|
||||
|
||||
SET(JFJOCH_WRITER_ONLY OFF CACHE BOOL "Compile HDF5 writer only")
|
||||
SET(JFJOCH_INSTALL_DRIVER_SOURCE OFF CACHE BOOL "Install kernel driver source (ignored if building writer only; necessary for RPM building)")
|
||||
|
||||
+29
-1
@@ -21,7 +21,8 @@ Options:
|
||||
-d<path> | --root_dir=<path> Root directory for file writing (-R is a deprecated alias)
|
||||
-r<int> | --zmq_repub_port=<int> ZeroMQ port for PUSH socket to republish images
|
||||
-f<int> | --zmq_file_port=<int> ZeroMQ port for PUB socket for notifications on finalized files
|
||||
-w<int> | --rcv_watermark=<int> Receiving ZeroMQ socket watermark (default = 100)
|
||||
-w<int> | --rcv_watermark=<int> Receiving ZeroMQ socket watermark in messages (default = 10)
|
||||
-m<int> | --rcv_ring_mb=<int> Receive ring buffer in MB; holds images until written and republished (default = 2048)
|
||||
-W<int> | --repub_watermark=<int> Republish ZeroMQ socket watermark (default = 1000)
|
||||
-b<int> | --rcv_buffer=<int> Receiving TCP socket buffer in bytes (only with -T; disables autotuning, capped by net.core.rmem_max)
|
||||
-v | --verbose Verbose output
|
||||
@@ -32,6 +33,33 @@ for example:
|
||||
jfjoch_writer -d /data tcp://dcu-address:5400
|
||||
```
|
||||
|
||||
## Memory
|
||||
Every message the writer receives is placed in a ring buffer of a fixed size (`-m`, default 2048 MB,
|
||||
per writer process - `-j 8` is eight rings) and stays there until it has been written to HDF5 and, if
|
||||
republish is on, republished. The ring is the writer's memory budget for buffered images, and the
|
||||
peak use is printed at the end of every run. Size it as the number of images you can afford to have
|
||||
in flight times their compressed size: at 2048 MB that is some 250-500 JUNGFRAU 4M images but only
|
||||
50-100 JUNGFRAU 9M images, so a 9M site will want more.
|
||||
|
||||
When the ring is full the writer stops reading from the socket and the sender has to wait. How long
|
||||
it is prepared to wait depends on the stream:
|
||||
|
||||
- **ZeroMQ.** The socket queue fills up to its watermark (`-w`, a count of messages, kept small for
|
||||
that reason) and the broker's send blocks for one second. After that the broker drops the image,
|
||||
logs the fact, and sends without waiting for the rest of the run, so every later image that finds
|
||||
the queue full is dropped too. A writer that stalls for longer than its ring covers therefore
|
||||
loses images on ZeroMQ; the ring has to be large enough for the stalls of the filesystem it writes
|
||||
to. (Before the ring the same stall was absorbed by queues of about 1500 messages, whatever their
|
||||
size - a budget nobody had chosen.)
|
||||
- **TCP/IP.** The broker sees the writer's BUSY heartbeats and waits, giving up only after a minute
|
||||
in which not a single image was written.
|
||||
|
||||
Republish holds its own images in the same ring. Because the ring is reclaimed in arrival order, an
|
||||
image kept for a slow republish consumer would hold back everything received after it, so republish
|
||||
may hold at most a quarter of the ring; beyond that images are simply not republished, exactly as
|
||||
when its queue is full. A single message larger than the whole ring - a START or END message of a
|
||||
very long run - is held outside it.
|
||||
|
||||
## Status and cancellation
|
||||
When a data collection is finalized, each writer reports its outcome back to `jfjoch_broker` over
|
||||
the writer notification socket — a ZeroMQ address the broker passes in the START message
|
||||
|
||||
@@ -192,6 +192,8 @@ One page per job, so the answer needed is near the top of a short page:
|
||||
anisotropy section.
|
||||
- [Advanced usage](RUGNUX_ADVANCED.md) — reference data and the indexing ambiguity, model
|
||||
validation, re-merging, and the full command-line option tables.
|
||||
- [Models and reference data](RUGNUX_MODEL_AND_REFERENCE.md) — every model and reference option,
|
||||
and decision by decision what a model can and cannot change.
|
||||
- [Detector calibration](RUGNUX_CALIBRATION.md) — the geometry from a calibrant's powder rings
|
||||
(`--mode calibration`).
|
||||
- [CPU/GPU data analysis](CPU_DATA_ANALYSIS.md) — the algorithms behind all of it.
|
||||
|
||||
@@ -147,6 +147,9 @@ determined on, CCP4 style (`k,l,h` is h'=k, k'=l, l'=h); `h,k,l` where nothing m
|
||||
|
||||
## Validating against a model (`rugnux --model`)
|
||||
|
||||
Which decisions a model can influence, and which it cannot, is set out decision by decision in
|
||||
[Models and reference data](RUGNUX_MODEL_AND_REFERENCE.md).
|
||||
|
||||
Given an atomic model of the same structure, `--model model.pdb` scales the model structure
|
||||
factors to the merged amplitudes — fitting a flat bulk-solvent contribution and an overall
|
||||
anisotropic *B* — and reports **R-work / R-free** and the mean 2mFo−DFc density at the atom centres.
|
||||
|
||||
@@ -0,0 +1,333 @@
|
||||
# Models and reference data
|
||||
|
||||
What Rugnux does with outside knowledge — an atomic model, a predicted model, a reference dataset, a
|
||||
cell or a space group you already know — and, decision by decision, what that knowledge is and is not
|
||||
allowed to change. The first section is the short version; the tables after it are the precise one;
|
||||
the rest walks through a run step by step.
|
||||
|
||||
```{contents} On this page
|
||||
:local:
|
||||
:depth: 2
|
||||
```
|
||||
|
||||
## In short
|
||||
|
||||
- **Rugnux needs nothing but the images.** Indexing, the cell, the space group, the resolution cut,
|
||||
scaling, merging and the error model all come from the data. Everything on this page is optional.
|
||||
- **A model (`--model`, `--mr`) does not change how your data are processed.** The images are
|
||||
indexed, integrated, scaled and merged the same way with or without it, and the space group is
|
||||
determined before the model is used. The merged intensities and their sigmas in
|
||||
`<prefix>.mtz` are the same numbers either way.
|
||||
- **The model can only relabel, and only if the data accept it.** It is first tested against
|
||||
copies of itself placed in random orientations. A model that fits no better than those changes
|
||||
nothing at all: the reflection files are byte for byte what a run without a model writes. A model
|
||||
that does fit may change three *labels* in `<prefix>.mtz`: the **hand** of the space group
|
||||
(P4₁2₁2 against P4₃2₁2, which the intensities cannot tell apart), the **indexing** where the
|
||||
lattice allows two equally valid ones, and the **axes** the files are written on. No
|
||||
intensity is changed, no reflection is added or removed, and no other space group is ever taken.
|
||||
- **A different space group chosen by the model goes to a separate file.** If the model prefers
|
||||
another space group, or a twinned subgroup, that merge is written *beside* the answer as
|
||||
`<prefix>_mr.mtz`/`.cif`, with the model as placed there. `<prefix>.mtz` stays the data's own
|
||||
answer.
|
||||
- **A reference dataset (`-z`) and a space group or cell you give (`-S`, `-C`) are different:** they
|
||||
are taken as facts about the crystal. They fix the space group and the cell before processing,
|
||||
so with them the space group is yours, not the data's.
|
||||
- **One exception to "processed the same":** on **serial stills** given `--model` together with
|
||||
`-C` and `-S`, the model is used during integration to choose, crystal by crystal, between the two
|
||||
equally valid indexings — the same job a reference dataset does there (see
|
||||
[Serial stills](#serial-stills-the-one-place-a-model-acts-before-the-merge)).
|
||||
|
||||
## Every option that brings in a model or reference data
|
||||
|
||||
| Option | What you give | What Rugnux uses it for | Can it change `<prefix>.mtz`? |
|
||||
|---|---|---|---|
|
||||
| *(none)* | — | Everything from the data | — |
|
||||
| `--model <file>` | An atomic model, PDB or mmCIF, gzipped or not, recognised by content: an experimental structure or a prediction | Validation after the merge (R-work/R-free, CC(model, data), maps, the placed model); a rigid-body placement checked against a null, with molecular replacement where the rigid body does not place it, and straight away for a prediction | Labels only (hand, indexing, axes), and only where the model fits. Other space groups → `<prefix>_mr.*` |
|
||||
| `--mr <file>` | The same kinds of model | As `--model`, but molecular replacement straight away, without the rigid body. `--mode mx` only, not together with `--model` | Hand label only. Other space groups → `<prefix>_mr.*` |
|
||||
| PAE file beside a prediction | `AF-<id>-F1-predicted_aligned_error_v<n>.json` next to `AF-<id>-F1-model_v<n>.cif` (read automatically, no option) | Cuts the prediction into rigid domains for molecular replacement | No (a domain alone can only confirm the answer's group) |
|
||||
| `-z, --reference <file>` (`--reference-mtz`) | Merged data of the same crystal form: an MTZ, or a PDB structure-factor mmCIF (`-sf.cif`) | Fixes the space group (in its own setting) and the cell (softly), puts the merge on its axes, resolves the indexing ambiguity, hands over its R-free set, reports CC<sub>ref</sub> on stills. Never a scale anchor | **Yes**: the space group, the axes, the indexing and the free set are the reference's |
|
||||
| `--reference-column <label>` | An MTZ column label | Which column of the reference is read | Through `-z` only |
|
||||
| `-S, --space-group` | A space group | Fixes it: no space-group search | **Yes**: the group is yours (a fitting model may still relabel its hand) |
|
||||
| `-C, --unit-cell` | A cell | Indexing looks for this cell (within tolerance); the FFT search range is widened to cover it; on stills it turns on the shared-geometry refinement and, with `-S` and `--model`, the per-image use of the model | **Yes**: the cell, and the axis order the files are written in |
|
||||
| `--force-rotation-lattice` | Nine numbers: a lattice | Skips the first-pass indexing and uses this lattice | **Yes**: the lattice |
|
||||
| `--report-resolution <dmin>[,<dmax>]` | Another program's resolution range | A second statistics table at that range (`REFRES_*`) | No — report only |
|
||||
|
||||
`-S`, `-C`, `-z` and `--force-rotation-lattice` are *assertions*: Rugnux takes them as true and
|
||||
does not test them against a null. Use them when you know; leave them out to let the data decide
|
||||
(a de-novo run is the recommended default — see [Running Rugnux](RUGNUX_TUTORIAL.md#rotation-data)).
|
||||
`--model` and `--mr` are *hypotheses*: Rugnux tests them, and what it lets them change is limited
|
||||
and stated below. In `--mode calibration` a cell given with `-C` names the calibrant instead
|
||||
([Detector calibration](RUGNUX_CALIBRATION.md)); nothing on this page applies there.
|
||||
|
||||
## Four kinds of influence
|
||||
|
||||
Every decision and every output of a run falls in one of four classes. The decision table below uses
|
||||
these letters.
|
||||
|
||||
- **(a) Data only.** Never influenced by any model. The same with and without `--model`/`--mr`.
|
||||
- **(b) Model-informed label in `<prefix>.mtz`.** A choice between alternatives the data *cannot*
|
||||
separate and that are all equally valid descriptions of the same measurement: the hand of an
|
||||
enantiomorphic pair, an indexing ambiguity, the setting (axes). The model may make it only after it
|
||||
has passed its null; the intensities do not change. Reported as `SPACE_GROUP_ENANTIOMORPH=
|
||||
ASSUMED_FROM_MODEL`, `MODEL_DECISIONS_TAKEN=`, `MODEL_ENANTIOMORPH_ADOPTED=`,
|
||||
`MODEL_INDEXING_OPERATOR=`, `SETTING_SOURCE= MODEL`, and the withdrawal of
|
||||
`INDEXING_AMBIGUITY_OPERATORS=`.
|
||||
- **(c) Model-driven, separate files only.** A choice the data *could* in principle make but did not
|
||||
make the model's way: another space group with the same rotations (other screw axes), or the twin
|
||||
hypothesis' subgroup. Written only as `<prefix>_mr.mtz`/`.cif` with `<prefix>_mr_model.pdb`/`.cif`
|
||||
(and named by `MODEL_CHOICE_FILE_PREFIX=`, `MODEL_CHOICE_SPACE_GROUP=`, `MR_DECISION=`). Never in
|
||||
`<prefix>.mtz`.
|
||||
- **(d) Report-only model statistics.** Numbers about the model against the data, which nothing in
|
||||
the pipeline acts on: `R_WORK`/`R_FREE`, `R_MODEL_SHELL_SCALED`, `CC_MODEL_*`, `MODEL_FIT*`,
|
||||
`MODEL_PLACEMENT`, the `MR_*` keys, the anomalous sites, the maps.
|
||||
|
||||
## Decision by decision
|
||||
|
||||
"Reference" means `-z`. A dash means the option plays no part.
|
||||
|
||||
| Decision | Without anything | With `-S` / `-C` | With `-z` | With `--model` / `--mr` | Class |
|
||||
|---|---|---|---|---|---|
|
||||
| Spot finding, beam centre, detector geometry | data | — | — | — | (a) |
|
||||
| Indexing, lattice, cell | data (de novo) | your cell / lattice | the reference's cell, softly | — | (a) |
|
||||
| Integration | data | — | — | — (stills: see below) | (a) |
|
||||
| Scaling, error model, ISa | data | — | — (never a scale anchor) | — | (a) |
|
||||
| Outlier rejection, frame disposition | data | — | — | — | (a) |
|
||||
| Resolution cut | data (CC1/2 fall-off) | — | — | — (`CC_MODEL_CONFIRMED_TO_D_MIN` is reported, never used) | (a) |
|
||||
| Point group, Laue class, centring | data (space-group search) | `-S` fixes it | the reference's | — | (a) |
|
||||
| Screw axes (space group within the point group) | data | `-S` | the reference's | a different choice → `<prefix>_mr.*` only | (a) in `p.mtz`, (c) beside it |
|
||||
| Hand (enantiomorph) | not decidable: `SPACE_GROUP_ENANTIOMORPH= UNDETERMINED` | `GIVEN` | `GIVEN` | relabelled to the model's where it fits, past the anomalous veto: `ASSUMED_FROM_MODEL` | (b) |
|
||||
| Indexing ambiguity (merohedral reindex) | arbitrary on rotation (`INDEXING_AMBIGUITY_OPERATORS=`); mixed on stills (warning) | — | resolved against the reference | resolved by the rigid-body model where its lead beats its null, rotation; per image on stills with `-C`/`-S`. Never by molecular replacement | (b) |
|
||||
| Setting (which axis is a, b, c) | standard setting | `-C`'s axis order; `-S`'s symbol | the reference's | the model's, where it fits and there is no `-z` (`SETTING_SOURCE= MODEL`) | (b) |
|
||||
| Twinning check (L-test, twin laws) | data | — | — | — | (a) |
|
||||
| Twin hypothesis `<prefix>_twin_<SG>.*` | data (when the group may have absorbed a twin law) | — | — | the model may also *ask* for this merge; written then only where the model chooses it | (a) / (c) |
|
||||
| Anomalous signal, Friedel handling | data (`-A`) | — | — | — (the anomalous map is report-only, and the veto on the hand) | (a) |
|
||||
| R-free test set | hash of the index | — | the reference's flags, where it matches | — (drawn again on the model's axes if the setting moved; never chosen by the model) | (a) |
|
||||
| Merged intensities and sigmas in `p.mtz` | data | — | — | — | (a) |
|
||||
| `p.mtz` space-group label | data | yours | the reference's | hand relabel only | (b) |
|
||||
| `<prefix>_mr.mtz`/`.cif`, `<prefix>_mr_model.*` | not written | — | — | another group or the twin subgroup the model chose | (c) |
|
||||
| Maps, `<prefix>_model.*`, section 5 of the report | not written | — | — | always, whatever the verdict | (d) |
|
||||
|
||||
## A run with `--model`, step by step
|
||||
|
||||
1. **The file is read and classified.** PDB or mmCIF, by content. A **prediction** — ModelCIF's
|
||||
quality metrics, or no cell and one B value per residue between 0 and 100 (an AlphaFold file) —
|
||||
is told apart from an **experimental structure** (`MR_MODEL_KIND=`). A model that cannot be read
|
||||
does not fail the run; the report says `MODEL_VALIDATION= NOT_PERFORMED` and carries a warning.
|
||||
2. **The data are processed without the model.** Indexing, integration, scaling, the space-group
|
||||
search, the resolution cut, the error model, outlier rejection: none of them reads the model
|
||||
(on serial stills with `-C`/`-S`, see the next section). The only thing the model asks for here is
|
||||
one extra merge: the merge in a subgroup (the twin hypothesis) where the indexed lattice's group
|
||||
was promoted by a metric re-ask, or where the model's own group has fewer rotations than the
|
||||
adopted one — so that the model can be scored there. That merge is written only if the model
|
||||
chooses it.
|
||||
3. **Rigid-body placement** (experimental structures, not with `--mr`). The model is put into the
|
||||
data's cell and refined as one rigid body on the working reflections; the step is kept only if
|
||||
it lowers R-free. A model written in another description of the lattice (other axes, other
|
||||
centring) is first put into the data's (`MODEL_CHANGE_OF_BASIS=`).
|
||||
4. **The null.** The same model is turned to nine random orientations about its own centroid and
|
||||
placed the same way. The real fit has to beat them by 4.51 of their standard deviations on
|
||||
R-work (3σ allowing for the null being nine draws): `MODEL_FIT= ACCEPTED`, `MODEL_FIT_SIGMA=`.
|
||||
Otherwise `REJECTED`, and the model decides nothing. There is no fixed threshold on R
|
||||
([§14.5](CPU_DATA_ANALYSIS_DECISIONS.md#145-does-the-model-fit-the-null-it-is-scored-against)).
|
||||
5. **What an accepted model may do** (class b), applied to the reflections before they are written:
|
||||
- **hand** — where the data's group is the enantiomorph of the model's, the files take the
|
||||
model's group as their label. No reflection moves and I(+)/I(-) stay as measured. Vetoed where
|
||||
the anomalous difference map read at the model's atoms is inverted (warning `MODEL_HAND`);
|
||||
- **indexing** — where the lattice admits alternative indexings and no `-z` already settled it,
|
||||
the candidate with the lowest R-free, *if* its lead over the runner-up beats the lead a random
|
||||
placement of the same model takes (`MODEL_INDEXING_MARGIN_SIGMA=` ≥ 3). The data are
|
||||
reindexed into the model's frame, never the model into the data's;
|
||||
- **setting** — a model written on other axes takes the files to its setting
|
||||
(`SETTING_SOURCE= MODEL`), so that reflections and coordinates can be used together.
|
||||
6. **The twin hypothesis.** Where a subgroup merge exists, the model is scored against it plainly
|
||||
and as twinned by its law. The maps come from whichever merge the model fits better on R-free
|
||||
(`MAPS_FROM=`). Where the model also fits there against its null, it has chosen the subgroup,
|
||||
and that merge is written again as `<prefix>_mr.mtz`/`.cif` (class c).
|
||||
7. **Molecular replacement**, for a prediction, with `--mr`, or for an experimental structure the
|
||||
rigid body did not place (in the answer's merge or in the subgroup's). As many copies as the cell
|
||||
most probably holds are searched for in every space group the merge cannot tell apart and in the
|
||||
twin hypothesis' subgroup, and the choice has its own null. `MR_DECISION=` is one of:
|
||||
`AGREES` (the answer's group — nothing changes); `ENANTIOMORPH` (the answer's files take the hand
|
||||
label, past the same anomalous veto, else `ENANTIOMORPH_VETOED`); `SPACE_GROUP` or
|
||||
`TWIN_SUBGROUP` (written as `<prefix>_mr.*` only); `UNDECIDED` or `NOT_PLACED` (nothing is
|
||||
written). Molecular replacement never reindexes and never sets `MODEL_FIT`. Not placing a model
|
||||
is a result, not a warning ([Molecular replacement](RUGNUX_REPORT.md#molecular-replacement)).
|
||||
8. **Validation and files.** R-factors, CC(model, data), maps and the placed model are computed
|
||||
for the placement that was kept, against the merge it was chosen on, and written:
|
||||
`<prefix>_2fofc.ccp4`, `_fofc.ccp4`, `_anom.ccp4`, `_maps.mtz`, `_model.cif`/`.pdb` — or the
|
||||
`<prefix>_mr_*` equivalents where the choice was `<prefix>_mr`. A rejected model gets its maps
|
||||
too: that is the case where looking at it in the density is most useful.
|
||||
9. **The reflection files** — `<prefix>.mtz`, `.cif`, `.hkl`, `_unmerged.mtz` — are written last,
|
||||
carrying only the class-(b) labels from step 5 and the hand from step 7.
|
||||
|
||||
`MODEL_PLACEMENT=` summarises steps 3–7: `RIGID_BODY`, `MR` or `NOT_PLACED`.
|
||||
|
||||
In `--mode scale`, `--model` is steps 3–5 only: the rigid body, its null, and the hand and indexing
|
||||
it may settle. There is no molecular replacement, no setting change and no twin hypothesis, and
|
||||
`--mr` is refused.
|
||||
|
||||
### Serial stills: the one place a model acts before the merge
|
||||
|
||||
On serial stills each crystal is indexed on its own, so where the lattice admits two indexings, each
|
||||
crystal lands in one of them at random and a merge averages reflections that are not equivalent.
|
||||
That has to be settled per crystal, at integration time — by the time there is a merge, the hands
|
||||
are already mixed. A reference dataset (`-z`) is the tool for it. Without one, `--model` given
|
||||
together with `-C` and `-S` supplies the reference intensities instead: structure factors computed
|
||||
from the model, against which each crystal's intensities are correlated to choose its indexing.
|
||||
|
||||
This is the only point at which a model affects which observations are merged together, and it is
|
||||
not gated by the null — there is no merge yet to test the model on. It chooses only between
|
||||
indexings that are equally valid for each crystal; a model that does not belong to the crystal
|
||||
carries no information to choose with, so the crystals stay mixed, as they would without a model.
|
||||
After the merge, the model is validated as in the steps above. Without `-C` and `-S`, or on rotation
|
||||
data, the model is not used before the merge at all.
|
||||
|
||||
## Where a model can touch the space group
|
||||
|
||||
Spelled out, because this is the question that matters most:
|
||||
|
||||
1. **The space group in `<prefix>.mtz` is decided without the model** — by the space-group search,
|
||||
or by you (`-S`, `-z`). The model is read after that decision.
|
||||
2. **The model can change one thing about that group: its hand,** between the two members of an
|
||||
enantiomorphic pair (P3₁/P3₂, P4₁2₁2/P4₃2₁2, P6₁22/P6₅22, …). The intensities are identical for
|
||||
both, so this is the one group choice no data can make; a model that fits carries chemical
|
||||
knowledge (the hand of its amino acids) the intensities do not. It is a label: no reflection
|
||||
moves. It applies to a group you gave with `-S` or `-z` too. The report says
|
||||
`SPACE_GROUP_ENANTIOMORPH= ASSUMED_FROM_MODEL` — *assumed*, not determined — also where the model
|
||||
simply agrees with the hand the run wrote.
|
||||
3. **Everything else the model prefers goes beside the answer.** Other screw axes with the same
|
||||
rotations (e.g. P2 against P2₁ where the absences did not decide), or the twin hypothesis'
|
||||
subgroup: `<prefix>_mr.mtz`/`.cif`. The model never changes the point group, the lattice or the
|
||||
centring of `<prefix>.mtz`.
|
||||
|
||||
The safeguards behind 2 and 3:
|
||||
|
||||
- **The null.** The model must beat its own random placements by 4.51 sd, rigid body and molecular
|
||||
replacement alike, before it may relabel anything or have a `<prefix>_mr` file written.
|
||||
- **The anomalous veto.** The hand is taken only where the anomalous difference map, read at the
|
||||
model's atoms, is not inverted. The veto can act only where the merge kept the Bijvoet split —
|
||||
every rotation merge does; serial stills only with `-A`. R-free cannot see the hand at all.
|
||||
- **The indexing margin.** An indexing is taken only where the model's preference beats what a
|
||||
randomly placed copy of it would show.
|
||||
- **Molecular replacement's own rules.** Its choice of group has its own null (`NULL_SIGMA`); a
|
||||
placement must pack (`CLASH`, `MATTHEWS`); a choice that translational NCS could explain, or one
|
||||
where the merge's own group was searched to fewer copies, is left `UNDECIDED`; a single domain of
|
||||
a prediction cut by its PAE can only confirm the answer's group, never change it; in P1 the first
|
||||
copy's origin is fixed and it counts only on its rotation function. The details are in
|
||||
[Molecular replacement](RUGNUX_REPORT.md#molecular-replacement).
|
||||
- **The wrong-model test.** Rugnux's validation battery runs a set of datasets once without a model
|
||||
and once with each of five models of *unrelated* proteins (similar size, unrelated sequence and
|
||||
fold, and for the enantiomorphic sets a model in the other hand), and requires the reflection data
|
||||
and every report key outside section 5 to be identical, `MODEL_FIT` not accepted, molecular
|
||||
replacement not placing the model and no model decision taken
|
||||
("The wrong-model null" in `tools/battery/README.md`; see also [Tests](TESTS.md)).
|
||||
|
||||
## Free set and model bias
|
||||
|
||||
**How the free set is made.** Rugnux draws its own test set (5 %, raised towards about 500 reflections on small data but never above 10 %):
|
||||
each reflection's flag is a hash of its index, taken over its orbit under the lattice's metric
|
||||
symmetry, so symmetry mates, Friedel mates and twin-law mates share a flag, every dataset of one
|
||||
crystal form written on the same axes gets the same set, and the set does not depend on the space
|
||||
group a file is merged in
|
||||
([R-free test-set flags](CPU_DATA_ANALYSIS_INTEGRATION.md#107-r-free-test-set-flags)). The model
|
||||
plays no part in it. Where a fitting model moves the files to its setting, the set is drawn again on
|
||||
the new axes by the same rule; an indexing-ambiguity reindex leaves it unchanged, since twin-law
|
||||
mates share a flag. With `-z`, the reference's flags are inherited where the merge matches it
|
||||
(`REFERENCE_FREE_FLAGS_INHERITED=`).
|
||||
|
||||
**What is fitted on which reflections.** Everything fitted to place and scale the model — the
|
||||
overall scale, the anisotropic *B*, the bulk solvent, the rigid body — is fitted on the working
|
||||
reflections only. The free set is then used for a few discrete *selections*: whether the rigid-body
|
||||
step is kept, which indexing wins, and in molecular replacement which placement and which group.
|
||||
Each of those leaves R-free very slightly optimistic where it fires. The coordinates, B-factors and
|
||||
occupancies of the model are never refined.
|
||||
|
||||
**What `R_FREE` means here — and what it does not.** It is the model as placed, scored on Rugnux's
|
||||
free set: a check that the model fits these data, and a trend number between datasets of one
|
||||
campaign. It is **not** a refinement R-free, and it is not cross-validated against the model's own
|
||||
history:
|
||||
|
||||
- A **deposited model** was refined by its depositor against *their* working set. Rugnux's free set
|
||||
is a different 5 %, so about 95 % of Rugnux's free reflections were in the working set the model
|
||||
was refined on. R-free here therefore reads close to R-work (the two differ by a median 0.001
|
||||
over our corpus), and it is optimistic in the usual model-bias sense.
|
||||
- To score a deposited model on its own test set, give its structure factors as the reference
|
||||
(`-z 1abc-sf.cif`): the depositor's free flags are then inherited. But `-z` is an assertion — it
|
||||
also fixes the space group and the cell — so that is no longer a de-novo run.
|
||||
- A **predicted model** has not been refined against any diffraction data, so it carries no such
|
||||
memory; what remains is the selection effect above.
|
||||
- For comparing two runs against one model, read `R_MODEL_SHELL_SCALED`, not `R_FREE`
|
||||
([Comparing two runs](RUGNUX_REPORT.md#comparing-two-runs-r_model_shell_scaled)).
|
||||
|
||||
**What cannot leak, and what can.** The model never enters the merged intensities, their sigmas, the
|
||||
choice of reflections, the resolution cut or the weights, so `<prefix>.mtz` carries no model bias in
|
||||
its numbers — only the class-(b) labels. The **maps** are model-phased, and carry the model bias any
|
||||
model-phased map carries: read the mFo−DFc map, and treat density that only the 2mFo−DFc map shows
|
||||
with the usual care. CC(model, data) is read in one direction only: a confirmed shell is evidence for
|
||||
the data, a weak shell says nothing about them, and nothing acts on it.
|
||||
|
||||
## Practical guidance
|
||||
|
||||
**When to give a model.**
|
||||
|
||||
- **You have a structure of this crystal form** (a fragment campaign, a ligand soak, a reference
|
||||
structure): give it with `--model`. You get maps and a placed model ready for refinement, the
|
||||
indexing of every dataset brought into the model's, and the hand fixed — the things that make a
|
||||
campaign's files directly comparable. A wrong or rejected model costs a few seconds and changes
|
||||
nothing.
|
||||
- **You have a structure of the same protein in another crystal form**: `--model` too. The rigid
|
||||
body will not place it, and molecular replacement is tried automatically.
|
||||
- **You have only a prediction** (AlphaFold, ModelCIF): `--model` (or `--mr`). Download the PAE JSON
|
||||
with it and keep its AlphaFold DB file name, so it is cut into domains when the whole model does
|
||||
not place. A prediction is good for: confirming the space group, naming the hand where the data
|
||||
cannot, offering a twinned or lower-symmetry alternative where the data are ambiguous, and a
|
||||
first set of maps. It is not a refined model, and its R-factors read accordingly.
|
||||
- **You have a previous dataset of the same crystal form** and want the same indexing and free set:
|
||||
`-z`, not `--model`. It is the stronger tool for the indexing ambiguity and the only one that
|
||||
carries a free set over, but it pins the space group and cell.
|
||||
- **You want to compare with another program's statistics**: `--report-resolution`, which changes
|
||||
nothing else.
|
||||
|
||||
**Which file to use for what.**
|
||||
|
||||
| File | What it is | Use it for |
|
||||
|---|---|---|
|
||||
| `<prefix>.mtz` / `.cif` / `.hkl` | The answer: the data's space group, with only the class-(b) labels from the model | Refinement, deposition, everything by default |
|
||||
| `<prefix>_model.cif` / `.pdb`, `<prefix>_2fofc.ccp4`, `_fofc.ccp4`, `_anom.ccp4`, `_maps.mtz` | The model as placed, in the frame of `<prefix>.mtz`, and its maps | Starting refinement (REFMAC5, phenix.refine), PanDDA, Coot |
|
||||
| `<prefix>_twin_<SG>.mtz` / `.cif` | The same data merged in a subgroup with a twin law — written from the data, model or not | Twin refinement, where you suspect the higher group absorbed a twin law |
|
||||
| `<prefix>_mr.mtz` / `.cif`, `<prefix>_mr_model.pdb` / `.cif`, `<prefix>_mr_*` maps | The merge in the space group the model chose (`MODEL_CHOICE_SPACE_GROUP=`), and the model placed there | Refinement *if* you accept the model's choice; compare R-free against the answer's |
|
||||
|
||||
`<prefix>_mr` and `<prefix>_twin_<SG>` can describe the same subgroup; the `_mr` file then also
|
||||
carries the model placed in it.
|
||||
|
||||
**Reading section 5 of the report**, in order:
|
||||
|
||||
1. `MODEL_VALIDATION=` — did it run? If `NOT_PERFORMED`, `MODEL_VALIDATION_REASON=` says why.
|
||||
2. `MODEL_PLACEMENT=` — how the model was placed: `RIGID_BODY`, `MR` or `NOT_PLACED`.
|
||||
3. `MODEL_FIT=`, `MODEL_FIT_SIGMA=` — the rigid body's verdict against its null. Absent where the
|
||||
rigid body was not tried (a prediction, `--mr`).
|
||||
4. `MODEL_DECISIONS_TAKEN=` — `NONE` means the reflection files are what a run without a model
|
||||
writes. Otherwise `ENANTIOMORPH`, `INDEXING`, both, or `MR_ENANTIOMORPH`, with
|
||||
`MODEL_ENANTIOMORPH_ADOPTED=` and `MODEL_INDEXING_OPERATOR=`.
|
||||
5. `MODEL_CHOICE_FILE_PREFIX=`, `MODEL_CHOICE_SPACE_GROUP=`, `MR_DECISION=` — whether a
|
||||
`<prefix>_mr` file was written, and in which group.
|
||||
6. `R_WORK=`, `R_FREE=`, `CC_MODEL_CONFIRMED_TO_D_MIN=`, the anomalous sites — how well the placed
|
||||
model describes the data (class d).
|
||||
|
||||
Outside section 5, `SPACE_GROUP_ENANTIOMORPH=` (`UNDETERMINED`, `GIVEN` or `ASSUMED_FROM_MODEL`),
|
||||
`SETTING_SOURCE=` and the presence or absence of `INDEXING_AMBIGUITY_OPERATORS=` say what the model
|
||||
did to the answer's labels. Every key is described in [The results report](RUGNUX_REPORT.md#model-validation).
|
||||
|
||||
## Related pages
|
||||
|
||||
- [Advanced Rugnux](RUGNUX_ADVANCED.md#reference-data-and-the-indexing-ambiguity) — what `-z` does,
|
||||
the indexing ambiguity, the setting the files are written in, and
|
||||
[validating against a model](RUGNUX_ADVANCED.md#validating-against-a-model-rugnux---model).
|
||||
- [The results report](RUGNUX_REPORT.md) — [model validation](RUGNUX_REPORT.md#model-validation),
|
||||
[molecular replacement](RUGNUX_REPORT.md#molecular-replacement), the
|
||||
[twin hypothesis](RUGNUX_REPORT.md#the-twin-hypothesis-twin_subgroup_) and the
|
||||
[reference-range table](RUGNUX_REPORT.md#the-reference-range-table).
|
||||
- [Data analysis: space group and validation](CPU_DATA_ANALYSIS_DECISIONS.md) — the space-group
|
||||
search (§13) and the model-validation algorithms (§14).
|
||||
- [Reference data](CPU_DATA_ANALYSIS_INTEGRATION.md#109-reference-data-fixing-the-space-group-and-resolving-the-indexing-ambiguity)
|
||||
and [R-free test-set flags](CPU_DATA_ANALYSIS_INTEGRATION.md#107-r-free-test-set-flags).
|
||||
@@ -26,6 +26,7 @@ Jungfraujoch is distributed under the GPLv3 license.
|
||||
RUGNUX_INTEGRATION
|
||||
RUGNUX_REPORT
|
||||
RUGNUX_ADVANCED
|
||||
RUGNUX_MODEL_AND_REFERENCE
|
||||
RUGNUX_CALIBRATION
|
||||
CPU_DATA_ANALYSIS
|
||||
CPU_DATA_ANALYSIS_IMAGE
|
||||
|
||||
@@ -4,5 +4,7 @@ ADD_LIBRARY(JFJochImagePuller ZMQImagePuller.cpp ZMQImagePuller.h
|
||||
TestImagePuller.cpp
|
||||
TestImagePuller.h
|
||||
TCPImagePuller.cpp
|
||||
TCPImagePuller.h)
|
||||
TCPImagePuller.h
|
||||
ReceiveRingBuffer.cpp
|
||||
ReceiveRingBuffer.h)
|
||||
TARGET_LINK_LIBRARIES(JFJochImagePuller JFJochCommon JFJochZMQ JFJochLogger CBORStream2FrameSerialize)
|
||||
|
||||
@@ -3,7 +3,15 @@
|
||||
|
||||
#include "ImagePuller.h"
|
||||
|
||||
ImagePuller::ImagePuller(size_t fifo_size) : outside_fifo(fifo_size) {}
|
||||
ImagePuller::ImagePuller(size_t fifo_size, size_t ring_size) : outside_fifo(fifo_size), ring(ring_size) {}
|
||||
|
||||
size_t ImagePuller::GetRingBufferCapacity() const {
|
||||
return ring.GetCapacity();
|
||||
}
|
||||
|
||||
size_t ImagePuller::GetMaxRingBufferUtilization() const {
|
||||
return ring.GetMaxUtilization();
|
||||
}
|
||||
|
||||
size_t ImagePuller::GetCurrentFifoUtilization() const {
|
||||
return outside_fifo.GetCurrentUtilization();
|
||||
|
||||
@@ -3,6 +3,7 @@
|
||||
|
||||
#pragma once
|
||||
|
||||
#include <atomic>
|
||||
#include <string>
|
||||
#include <memory>
|
||||
|
||||
@@ -10,6 +11,7 @@
|
||||
#include "../frame_serialize/CBORStream2Deserializer.h"
|
||||
#include "../common/ThreadSafeFIFO.h"
|
||||
#include "../common/JfjochTCP.h"
|
||||
#include "ReceiveRingBuffer.h"
|
||||
|
||||
struct PullerAckMessage {
|
||||
TCPFrameType ack_for = TCPFrameType::DATA;
|
||||
@@ -25,12 +27,14 @@ struct PullerAckMessage {
|
||||
|
||||
struct RawFrame {
|
||||
TcpFrameHeader header{};
|
||||
std::vector<uint8_t> payload;
|
||||
bool end = false;
|
||||
};
|
||||
|
||||
struct ImagePullerOutput {
|
||||
std::shared_ptr<ZMQMessage> zmq_msg;
|
||||
// The message as it came off the socket, in the puller's ring buffer. It stays there for as long as
|
||||
// any copy of this struct is alive - the deserialised image in cbor points into it, and republish
|
||||
// sends it as is.
|
||||
std::shared_ptr<ReceiveBufferRegion> raw;
|
||||
std::shared_ptr<RawFrame> tcp_msg;
|
||||
std::shared_ptr<CBORStream2DeserializerOutput> cbor;
|
||||
};
|
||||
@@ -42,16 +46,32 @@ struct ImagePullerStatistics {
|
||||
class ImagePuller {
|
||||
protected:
|
||||
ThreadSafeFIFO<ImagePullerOutput> outside_fifo; // assuming 1 file is <1000 files, this should be enough to buffer
|
||||
// Every received message is held here until consumed; the queues count messages, this bounds bytes.
|
||||
// Size 0 means no ring: messages are held on the heap, bounded by the queues alone, as they always
|
||||
// were - the broker's own DECTRIS puller runs that way, jfjoch_writer sets a size.
|
||||
ReceiveRingBuffer ring;
|
||||
|
||||
// Bytes republish holds right now. Republish keeps the OLDEST not yet sent images, and the ring is
|
||||
// reclaimed in arrival order, so an image waiting for a slow republish consumer pins everything
|
||||
// received after it - the writer would then run at the republish rate. What republish may hold is
|
||||
// therefore capped at a share of the ring; an image over the cap is not republished, exactly as one
|
||||
// that finds the republish queue full. Without a ring there is no cap.
|
||||
std::atomic<size_t> repub_bytes{0};
|
||||
bool RepubCanHold(size_t bytes) const {
|
||||
return ring.GetCapacity() == 0 || repub_bytes.load() + bytes <= ring.GetCapacity() / 4;
|
||||
}
|
||||
public:
|
||||
constexpr static size_t DefaultQueueSize = 1000;
|
||||
constexpr static auto PollTimeout = std::chrono::milliseconds(100);
|
||||
|
||||
explicit ImagePuller(size_t fifo_size = DefaultQueueSize);
|
||||
explicit ImagePuller(size_t fifo_size = DefaultQueueSize, size_t ring_size = 0);
|
||||
virtual ~ImagePuller() = default;
|
||||
|
||||
[[nodiscard]] std::optional<ImagePullerOutput> PollImage(std::chrono::microseconds timeout = PollTimeout);
|
||||
[[nodiscard]] size_t GetMaxFifoUtilization() const;
|
||||
[[nodiscard]] size_t GetCurrentFifoUtilization() const;
|
||||
[[nodiscard]] size_t GetRingBufferCapacity() const;
|
||||
[[nodiscard]] size_t GetMaxRingBufferUtilization() const;
|
||||
|
||||
virtual bool SupportsAck() const { return false; }
|
||||
virtual bool SendAck(const PullerAckMessage &) { return false; }
|
||||
|
||||
@@ -0,0 +1,146 @@
|
||||
// SPDX-FileCopyrightText: 2026 Filip Leonarski, Paul Scherrer Institute <filip.leonarski@psi.ch>
|
||||
// SPDX-License-Identifier: GPL-3.0-only
|
||||
|
||||
#include "ReceiveRingBuffer.h"
|
||||
|
||||
#include <algorithm>
|
||||
#include <optional>
|
||||
|
||||
struct ReceiveRingBuffer::State {
|
||||
// Uninitialised on purpose: a ring that is never filled should not cost its capacity in
|
||||
// resident memory.
|
||||
std::unique_ptr<uint8_t[]> memory;
|
||||
const size_t capacity;
|
||||
|
||||
std::mutex m;
|
||||
std::condition_variable cv;
|
||||
|
||||
// Live regions in arrival order. Only the oldest can give back space, so a region released out of
|
||||
// turn is marked and waits.
|
||||
struct Slot {
|
||||
size_t offset;
|
||||
size_t size;
|
||||
bool released = false;
|
||||
};
|
||||
std::deque<Slot> slots;
|
||||
// Next byte to hand out. The ring is empty when slots is empty; the oldest live byte is then
|
||||
// slots.front().offset.
|
||||
size_t tail = 0;
|
||||
|
||||
size_t utilization = 0;
|
||||
size_t max_utilization = 0;
|
||||
bool stopped = false;
|
||||
|
||||
explicit State(size_t in_capacity)
|
||||
: memory(in_capacity > 0 ? std::make_unique_for_overwrite<uint8_t[]>(in_capacity) : nullptr),
|
||||
capacity(in_capacity) {}
|
||||
|
||||
// Where a region of `size` bytes would start, or nothing if it does not fit right now. Called
|
||||
// under the mutex.
|
||||
std::optional<size_t> FindSpace(size_t size) const {
|
||||
if (slots.empty())
|
||||
return 0;
|
||||
const size_t head = slots.front().offset;
|
||||
// tail has come round to head: treated as full. (After a wrap the piece between the old tail
|
||||
// and the end is free, but it is behind the live data and not worth tracking.)
|
||||
if (tail == head)
|
||||
return std::nullopt;
|
||||
if (tail > head) {
|
||||
// Live data is [head, tail): free space is behind it to the end, or before it at the start.
|
||||
if (capacity - tail >= size)
|
||||
return tail;
|
||||
if (head >= size)
|
||||
return 0;
|
||||
return std::nullopt;
|
||||
}
|
||||
// Wrapped: live data is [head, capacity) + [0, tail), free space is [tail, head).
|
||||
if (head - tail >= size)
|
||||
return tail;
|
||||
return std::nullopt;
|
||||
}
|
||||
|
||||
void Release(size_t offset) {
|
||||
std::unique_lock ul(m);
|
||||
for (auto &s: slots) {
|
||||
if (s.offset == offset) {
|
||||
s.released = true;
|
||||
break;
|
||||
}
|
||||
}
|
||||
while (!slots.empty() && slots.front().released) {
|
||||
utilization -= slots.front().size;
|
||||
slots.pop_front();
|
||||
}
|
||||
if (slots.empty())
|
||||
tail = 0;
|
||||
cv.notify_all();
|
||||
}
|
||||
|
||||
void ReleaseHeap(size_t size) {
|
||||
std::unique_lock ul(m);
|
||||
utilization -= size;
|
||||
}
|
||||
};
|
||||
|
||||
ReceiveRingBuffer::ReceiveRingBuffer(size_t capacity) : state(std::make_shared<State>(capacity)) {}
|
||||
|
||||
std::shared_ptr<ReceiveBufferRegion> ReceiveRingBuffer::Allocate(size_t size) {
|
||||
// Nothing to hold, and a zero-length slot would share its offset with the next real one.
|
||||
if (size == 0)
|
||||
return std::make_shared<ReceiveBufferRegion>();
|
||||
|
||||
std::unique_lock ul(state->m);
|
||||
|
||||
if (size > state->capacity) {
|
||||
// Heap fallback: counted, not bounded.
|
||||
state->utilization += size;
|
||||
state->max_utilization = std::max(state->max_utilization, state->utilization);
|
||||
auto *region = new ReceiveBufferRegion{new uint8_t[size], size};
|
||||
return std::shared_ptr<ReceiveBufferRegion>(region, [s = state](ReceiveBufferRegion *r) {
|
||||
delete[] r->data;
|
||||
s->ReleaseHeap(r->size);
|
||||
delete r;
|
||||
});
|
||||
}
|
||||
|
||||
std::optional<size_t> offset;
|
||||
state->cv.wait(ul, [&] {
|
||||
if (state->stopped)
|
||||
return true;
|
||||
offset = state->FindSpace(size);
|
||||
return offset.has_value();
|
||||
});
|
||||
if (!offset)
|
||||
return nullptr;
|
||||
|
||||
state->slots.push_back({*offset, size});
|
||||
state->tail = *offset + size;
|
||||
state->utilization += size;
|
||||
state->max_utilization = std::max(state->max_utilization, state->utilization);
|
||||
|
||||
auto *region = new ReceiveBufferRegion{state->memory.get() + *offset, size};
|
||||
return std::shared_ptr<ReceiveBufferRegion>(region, [s = state, off = *offset](ReceiveBufferRegion *r) {
|
||||
s->Release(off);
|
||||
delete r;
|
||||
});
|
||||
}
|
||||
|
||||
void ReceiveRingBuffer::Stop() {
|
||||
std::unique_lock ul(state->m);
|
||||
state->stopped = true;
|
||||
state->cv.notify_all();
|
||||
}
|
||||
|
||||
size_t ReceiveRingBuffer::GetCapacity() const {
|
||||
return state->capacity;
|
||||
}
|
||||
|
||||
size_t ReceiveRingBuffer::GetUtilization() const {
|
||||
std::unique_lock ul(state->m);
|
||||
return state->utilization;
|
||||
}
|
||||
|
||||
size_t ReceiveRingBuffer::GetMaxUtilization() const {
|
||||
std::unique_lock ul(state->m);
|
||||
return state->max_utilization;
|
||||
}
|
||||
@@ -0,0 +1,52 @@
|
||||
// SPDX-FileCopyrightText: 2026 Filip Leonarski, Paul Scherrer Institute <filip.leonarski@psi.ch>
|
||||
// SPDX-License-Identifier: GPL-3.0-only
|
||||
|
||||
#pragma once
|
||||
|
||||
#include <condition_variable>
|
||||
#include <cstdint>
|
||||
#include <deque>
|
||||
#include <memory>
|
||||
#include <mutex>
|
||||
|
||||
// A received message lives here from the moment it comes off the socket until the last consumer is
|
||||
// done with it - the writer has stored the image and the republish thread has sent it. Both consumers
|
||||
// hold a shared_ptr to the region, and the region's destructor is what gives the bytes back.
|
||||
struct ReceiveBufferRegion {
|
||||
uint8_t *data = nullptr;
|
||||
size_t size = 0;
|
||||
};
|
||||
|
||||
// Bounds the memory a puller holds in received, not yet consumed, messages - by bytes, which is the
|
||||
// quantity that matters, where the queues in front of and behind it count messages. One image of a
|
||||
// large detector is tens of megabytes, so a queue of a thousand of them is a quantity of memory nobody
|
||||
// chose.
|
||||
//
|
||||
// Regions are handed out in arrival order and the space is taken back in arrival order too, which is
|
||||
// what makes it a ring: the writer and the republish thread may finish with messages out of order, so
|
||||
// a region released early is only remembered, and the space is reclaimed once everything received
|
||||
// before it is released as well. Republish holds at most its queue's worth of messages, so that is as
|
||||
// far as the reclaim can lag.
|
||||
//
|
||||
// A message larger than the whole ring goes on the heap instead, so an oversized START or END message
|
||||
// cannot wedge the receiver; everything else waits for space, which is the back-pressure that keeps the
|
||||
// sender from running ahead.
|
||||
class ReceiveRingBuffer {
|
||||
struct State;
|
||||
std::shared_ptr<State> state;
|
||||
public:
|
||||
explicit ReceiveRingBuffer(size_t capacity);
|
||||
|
||||
// Blocks until the bytes are available; returns nullptr once Stop() has been called. A request
|
||||
// larger than the capacity (so every request of a ring of capacity 0) is served from the heap,
|
||||
// at once and whether stopped or not.
|
||||
std::shared_ptr<ReceiveBufferRegion> Allocate(size_t size);
|
||||
|
||||
// Releases every waiter in Allocate; the regions already handed out stay valid.
|
||||
void Stop();
|
||||
|
||||
size_t GetCapacity() const;
|
||||
// Bytes held in live regions right now / at most since construction (heap fallbacks included).
|
||||
size_t GetUtilization() const;
|
||||
size_t GetMaxUtilization() const;
|
||||
};
|
||||
@@ -46,8 +46,10 @@ static std::pair<std::string, uint16_t> ParseTcpAddressPull(const std::string &a
|
||||
TCPImagePuller::TCPImagePuller(const std::string &tcp_addr,
|
||||
std::optional<int32_t> rcv_buffer_size,
|
||||
const std::string &repub_address,
|
||||
const std::optional<int32_t> &repub_watermark)
|
||||
: addr(tcp_addr),
|
||||
const std::optional<int32_t> &repub_watermark,
|
||||
const std::optional<size_t> &ring_size)
|
||||
: ImagePuller(DefaultQueueSize, ring_size.value_or(0)),
|
||||
addr(tcp_addr),
|
||||
receive_buffer_size(rcv_buffer_size) {
|
||||
auto parsed = ParseTcpAddressPull(tcp_addr);
|
||||
host = parsed.first;
|
||||
@@ -130,7 +132,7 @@ void TCPImagePuller::CBORThread() {
|
||||
if (type == TCPFrameType::CANCEL) {
|
||||
outside_fifo.PutBlocking(ret);
|
||||
} else {
|
||||
ret.cbor = CBORStream2Deserialize(ret.tcp_msg->payload.data(), ret.tcp_msg->payload.size());
|
||||
ret.cbor = CBORStream2Deserialize(ret.raw ? ret.raw->data : nullptr, ret.raw ? ret.raw->size : 0);
|
||||
outside_fifo.PutBlocking(ret);
|
||||
// A PREFLIGHT carries a StartMessage payload but starts nothing, so republishing it
|
||||
// would announce a run to downstream consumers that is not going to happen.
|
||||
@@ -138,11 +140,13 @@ void TCPImagePuller::CBORThread() {
|
||||
if ((ret.cbor->msg_type == CBORImageType::START)
|
||||
|| (ret.cbor->msg_type == CBORImageType::END))
|
||||
repub_fifo.PutBlocking(ret);
|
||||
else
|
||||
repub_fifo.Put(ret);
|
||||
else if (RepubCanHold(ret.raw->size) && repub_fifo.Put(ret))
|
||||
repub_bytes += ret.raw->size;
|
||||
}
|
||||
}
|
||||
} catch (const JFJochException &e) {
|
||||
} catch (const std::exception &e) {
|
||||
// std::exception: the deserialiser allocates a vector per per-image array of an END message,
|
||||
// and a bad_alloc leaving this thread would end the process.
|
||||
logger.ErrorException(e);
|
||||
}
|
||||
ret = cbor_fifo.GetBlocking();
|
||||
@@ -170,11 +174,13 @@ void TCPImagePuller::RepubThread() {
|
||||
logger.Info("Republish active");
|
||||
} else {
|
||||
if (repub_active)
|
||||
repub_socket->Send(ret.tcp_msg->payload.data(), ret.tcp_msg->payload.size(), true);
|
||||
repub_socket->Send(ret.raw->data, ret.raw->size, true);
|
||||
}
|
||||
} catch (const JFJochException &e) {
|
||||
} catch (const std::exception &e) {
|
||||
logger.ErrorException(e);
|
||||
}
|
||||
if (ret.cbor && (ret.cbor->msg_type != CBORImageType::START) && (ret.cbor->msg_type != CBORImageType::END))
|
||||
repub_bytes -= ret.raw->size;
|
||||
ret = repub_fifo.GetBlocking();
|
||||
}
|
||||
if (repub_active)
|
||||
@@ -372,14 +378,21 @@ void TCPImagePuller::ReceiverThread() {
|
||||
ImagePullerOutput out;
|
||||
out.tcp_msg = std::make_shared<RawFrame>();
|
||||
out.tcp_msg->header = frame.header;
|
||||
out.tcp_msg->payload.resize(frame.header.payload_size);
|
||||
|
||||
if (frame.header.payload_size > 0
|
||||
&& !ReadExact(out.tcp_msg->payload.data(), out.tcp_msg->payload.size())) {
|
||||
logger.Info("TCP payload read failed, reconnecting to " + addr);
|
||||
CloseSocket();
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(20));
|
||||
continue;
|
||||
if (frame.header.payload_size > 0) {
|
||||
// Read straight into the ring (or, with no ring, onto the heap). A full ring holds the
|
||||
// read here, the socket buffer fills, and the pusher's send blocks - its BUSY heartbeats
|
||||
// tell it apart from a dead writer, and it gives up only after a minute in which not one
|
||||
// image was written.
|
||||
out.raw = ring.Allocate(frame.header.payload_size);
|
||||
if (!out.raw)
|
||||
break; // Stop() - the puller is being disconnected.
|
||||
if (!ReadExact(out.raw->data, out.raw->size)) {
|
||||
logger.Info("TCP payload read failed, reconnecting to " + addr);
|
||||
CloseSocket();
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(20));
|
||||
continue;
|
||||
}
|
||||
}
|
||||
|
||||
cbor_fifo.PutBlocking(out);
|
||||
@@ -438,6 +451,8 @@ void TCPImagePuller::Disconnect() {
|
||||
cbor_fifo.Stop();
|
||||
repub_fifo.Stop();
|
||||
outside_fifo.Stop();
|
||||
// And ReceiverThread may be waiting for ring space that the stopped consumers will never free.
|
||||
ring.Stop();
|
||||
|
||||
if (receiver_thread.joinable())
|
||||
receiver_thread.join();
|
||||
|
||||
@@ -53,7 +53,8 @@ public:
|
||||
explicit TCPImagePuller(const std::string &tcp_addr,
|
||||
std::optional<int32_t> rcv_buffer_size = {},
|
||||
const std::string &repub_address = "",
|
||||
const std::optional<int32_t> &repub_watermark = {});
|
||||
const std::optional<int32_t> &repub_watermark = {},
|
||||
const std::optional<size_t> &ring_size = {});
|
||||
|
||||
~TCPImagePuller() override;
|
||||
bool SupportsAck() const override { return true; }
|
||||
|
||||
@@ -4,10 +4,14 @@
|
||||
#include "ZMQImagePuller.h"
|
||||
#include "../frame_serialize/CBORStream2Serializer.h"
|
||||
|
||||
#include <cstring>
|
||||
|
||||
ZMQImagePuller::ZMQImagePuller(const std::string &in_addr,
|
||||
const std::string &repub_address,
|
||||
const std::optional<int32_t> &rcv_watermark,
|
||||
const std::optional<int32_t> &repub_watermark) :
|
||||
const std::optional<int32_t> &repub_watermark,
|
||||
const std::optional<size_t> &ring_size) :
|
||||
ImagePuller(DefaultQueueSize, ring_size.value_or(0)),
|
||||
socket (ZMQSocketType::Pull), addr(in_addr) {
|
||||
auto start_time = std::chrono::steady_clock::now();
|
||||
puller_thread = std::thread(&ZMQImagePuller::PullerThread, this);
|
||||
@@ -43,6 +47,8 @@ void ZMQImagePuller::Disconnect() {
|
||||
cbor_fifo.Stop();
|
||||
repub_fifo.Stop();
|
||||
outside_fifo.Stop();
|
||||
// And PullerThread may be waiting for ring space that the stopped consumers will never free.
|
||||
ring.Stop();
|
||||
|
||||
if (puller_thread.joinable())
|
||||
puller_thread.join();
|
||||
@@ -59,8 +65,7 @@ void ZMQImagePuller::Disconnect() {
|
||||
|
||||
void ZMQImagePuller::PullerThread() {
|
||||
while (true) {
|
||||
ImagePullerOutput ret;
|
||||
ret.zmq_msg = std::make_shared<ZMQMessage>();
|
||||
auto msg = std::make_shared<ZMQMessage>();
|
||||
bool received = false;
|
||||
while (!received) {
|
||||
if (disconnect) {
|
||||
@@ -68,13 +73,33 @@ void ZMQImagePuller::PullerThread() {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
received = socket.Receive(*ret.zmq_msg, false);
|
||||
received = socket.Receive(*msg, false);
|
||||
if (!received)
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(1));
|
||||
} catch (const JFJochException &e) {
|
||||
logger.ErrorException(e);
|
||||
}
|
||||
}
|
||||
ImagePullerOutput ret;
|
||||
if (ring.GetCapacity() == 0) {
|
||||
// No ring: the ZeroMQ message itself is the buffer, kept alive by the region. No copy.
|
||||
ret.raw = std::shared_ptr<ReceiveBufferRegion>(new ReceiveBufferRegion{const_cast<uint8_t *>(msg->data()), msg->size()},
|
||||
[msg](ReceiveBufferRegion *r) { delete r; });
|
||||
} else {
|
||||
// Copied into the ring, and the ZeroMQ message is released with the next receive. When the
|
||||
// ring is full this waits and the socket is not read: its queue fills to the watermark and
|
||||
// the sender's send blocks - for as long as the sender is prepared to wait. The broker's
|
||||
// ZeroMQ pusher waits one second, then drops the image and sends without waiting for the
|
||||
// rest of the run, so a ring that is too small for the writer's stalls costs images.
|
||||
ret.raw = ring.Allocate(msg->size());
|
||||
if (!ret.raw) {
|
||||
// Stop() - the puller is being disconnected.
|
||||
cbor_fifo.PutBlocking(ImagePullerOutput{});
|
||||
return;
|
||||
}
|
||||
if (msg->size() > 0)
|
||||
memcpy(ret.raw->data, msg->data(), msg->size());
|
||||
}
|
||||
cbor_fifo.PutBlocking(ret);
|
||||
}
|
||||
}
|
||||
@@ -82,9 +107,9 @@ void ZMQImagePuller::PullerThread() {
|
||||
void ZMQImagePuller::CBORThread() {
|
||||
auto ret = cbor_fifo.GetBlocking();
|
||||
|
||||
while (ret.zmq_msg) {
|
||||
while (ret.raw) {
|
||||
try {
|
||||
ret.cbor = CBORStream2Deserialize(ret.zmq_msg->data(), ret.zmq_msg->size());
|
||||
ret.cbor = CBORStream2Deserialize(ret.raw->data, ret.raw->size);
|
||||
// Even if we suspend consuming of the messages by the receiver,
|
||||
// it is still reasonable to have them republished
|
||||
// so republish functionality is not affected by Suspend()
|
||||
@@ -94,10 +119,12 @@ void ZMQImagePuller::CBORThread() {
|
||||
if ((ret.cbor->msg_type == CBORImageType::START)
|
||||
|| (ret.cbor->msg_type == CBORImageType::END))
|
||||
repub_fifo.PutBlocking(ret);
|
||||
else
|
||||
repub_fifo.Put(ret);
|
||||
else if (RepubCanHold(ret.raw->size) && repub_fifo.Put(ret))
|
||||
repub_bytes += ret.raw->size;
|
||||
}
|
||||
} catch (const JFJochException &e) {
|
||||
} catch (const std::exception &e) {
|
||||
// std::exception: the deserialiser allocates a vector per per-image array of an END message,
|
||||
// and a bad_alloc leaving this thread would end the process.
|
||||
logger.ErrorException(e);
|
||||
}
|
||||
ret = cbor_fifo.GetBlocking();
|
||||
@@ -111,7 +138,7 @@ void ZMQImagePuller::RepubThread() {
|
||||
auto ret = repub_fifo.GetBlocking();
|
||||
bool repub_active = false;
|
||||
|
||||
while (ret.zmq_msg) {
|
||||
while (ret.raw) {
|
||||
try {
|
||||
if (ret.cbor->msg_type == CBORImageType::START) {
|
||||
// Start message needs to be cleaned when running republish
|
||||
@@ -125,11 +152,13 @@ void ZMQImagePuller::RepubThread() {
|
||||
logger.Info("Republish active");
|
||||
} else {
|
||||
if (repub_active)
|
||||
repub_socket->Send(ret.zmq_msg->data(), ret.zmq_msg->size(), true);
|
||||
repub_socket->Send(ret.raw->data, ret.raw->size, true);
|
||||
}
|
||||
} catch (const JFJochException &e) {
|
||||
} catch (const std::exception &e) {
|
||||
logger.ErrorException(e);
|
||||
}
|
||||
if (ret.cbor && (ret.cbor->msg_type != CBORImageType::START) && (ret.cbor->msg_type != CBORImageType::END))
|
||||
repub_bytes -= ret.raw->size;
|
||||
ret = repub_fifo.GetBlocking();
|
||||
}
|
||||
if (repub_active)
|
||||
|
||||
@@ -15,7 +15,9 @@
|
||||
|
||||
|
||||
class ZMQImagePuller : public ImagePuller {
|
||||
const uint32_t default_receive_watermark = 100;
|
||||
// Messages, not bytes: what the socket holds beyond the ring buffer. Kept small for that reason -
|
||||
// the ring is where the memory is accounted for.
|
||||
const uint32_t default_receive_watermark = 10;
|
||||
const uint32_t default_repub_watermark = 220;
|
||||
|
||||
// ZeroMQ receive timeout allows to check for abort value from time to time
|
||||
@@ -47,7 +49,8 @@ public:
|
||||
const std::string &zmq_addr,
|
||||
const std::string &repub_address = "",
|
||||
const std::optional<int32_t> &rcv_watermark = {},
|
||||
const std::optional<int32_t> &repub_watermark = {});
|
||||
const std::optional<int32_t> &repub_watermark = {},
|
||||
const std::optional<size_t> &ring_size = {});
|
||||
~ZMQImagePuller();
|
||||
|
||||
void Disconnect() override;
|
||||
|
||||
@@ -3,6 +3,8 @@
|
||||
|
||||
#include "ZMQStream2PusherSocket.h"
|
||||
|
||||
#include <spdlog/spdlog.h>
|
||||
|
||||
void zmq_socket_free(void *data, void *hint) {
|
||||
auto z = (ZeroCopyReturnValue *) hint;
|
||||
z->release();
|
||||
@@ -56,6 +58,10 @@ void ZMQStream2PusherSocket::WriterThread() {
|
||||
// When hitting timeout on SendZeroCopy consider it a transmission error and switch to non-blocking communication
|
||||
bool blocking_send = !transmission_error;
|
||||
if (!s.SendZeroCopy(e.z->GetImage(), e.z->GetImageSize(), zmq_socket_free, e.z, blocking_send)) {
|
||||
if (!transmission_error)
|
||||
spdlog::warn("ZeroMQ image socket {}: send blocked for over a second - image {} dropped, and "
|
||||
"images that find the queue full are dropped for the rest of the run",
|
||||
s.GetEndpointName(), e.z->GetImageNumber());
|
||||
transmission_error = true;
|
||||
// The slot is already released: on a failed send SendZeroCopy closes the message, and
|
||||
// closing a zmq_msg_init_data message runs its free function (zmq_socket_free).
|
||||
|
||||
@@ -57,6 +57,7 @@ ADD_EXECUTABLE(jfjoch_test
|
||||
AcquisitionCountersTest.cpp
|
||||
IndexingUnitTest.cpp
|
||||
ThreadSafeFIFOSetTest.cpp
|
||||
ReceiveRingBufferTest.cpp
|
||||
FPGANetworkTest.cpp
|
||||
JFPedestalTest.cpp
|
||||
JFCalibrationTest.cpp
|
||||
|
||||
@@ -0,0 +1,129 @@
|
||||
// SPDX-FileCopyrightText: 2026 Filip Leonarski, Paul Scherrer Institute <filip.leonarski@psi.ch>
|
||||
// SPDX-License-Identifier: GPL-3.0-only
|
||||
|
||||
#include <catch2/catch_all.hpp>
|
||||
#include <future>
|
||||
#include <thread>
|
||||
|
||||
#include "../image_puller/ReceiveRingBuffer.h"
|
||||
|
||||
using namespace std::chrono_literals;
|
||||
|
||||
TEST_CASE("ReceiveRingBuffer_sequential_reuse", "[ImagePuller][Unit]") {
|
||||
ReceiveRingBuffer ring(1024);
|
||||
// Far more messages than fit at once: each one is released before the next is taken.
|
||||
for (int i = 0; i < 100; i++) {
|
||||
auto r = ring.Allocate(300);
|
||||
REQUIRE(r);
|
||||
REQUIRE(r->size == 300);
|
||||
r->data[0] = 1;
|
||||
r->data[299] = 2;
|
||||
}
|
||||
CHECK(ring.GetUtilization() == 0);
|
||||
CHECK(ring.GetMaxUtilization() == 300);
|
||||
CHECK(ring.GetCapacity() == 1024);
|
||||
}
|
||||
|
||||
TEST_CASE("ReceiveRingBuffer_wraps_to_the_start", "[ImagePuller][Unit]") {
|
||||
ReceiveRingBuffer ring(1000);
|
||||
auto a = ring.Allocate(400);
|
||||
auto b = ring.Allocate(400);
|
||||
REQUIRE(a);
|
||||
REQUIRE(b);
|
||||
uint8_t *a_data = a->data;
|
||||
|
||||
// 200 bytes left at the end and nothing free at the start: a third 400 has to wait.
|
||||
auto f = std::async(std::launch::async, [&] { return ring.Allocate(400); });
|
||||
CHECK(f.wait_for(50ms) == std::future_status::timeout);
|
||||
|
||||
a.reset(); // frees [0, 400): the waiter wraps round to the start
|
||||
auto c = f.get();
|
||||
REQUIRE(c);
|
||||
CHECK(c->data == a_data);
|
||||
CHECK(ring.GetUtilization() == 800);
|
||||
// Never two live regions on the same bytes
|
||||
CHECK((b->data + 400 <= c->data || c->data + 400 <= b->data));
|
||||
}
|
||||
|
||||
TEST_CASE("ReceiveRingBuffer_full_ring_blocks", "[ImagePuller][Unit]") {
|
||||
ReceiveRingBuffer ring(1000);
|
||||
auto a = ring.Allocate(500);
|
||||
auto b = ring.Allocate(500); // tail has come round to head
|
||||
REQUIRE(a);
|
||||
REQUIRE(b);
|
||||
uint8_t *a_data = a->data;
|
||||
auto f = std::async(std::launch::async, [&] { return ring.Allocate(1); });
|
||||
CHECK(f.wait_for(50ms) == std::future_status::timeout);
|
||||
a.reset();
|
||||
auto c = f.get();
|
||||
REQUIRE(c);
|
||||
CHECK(c->data == a_data);
|
||||
}
|
||||
|
||||
TEST_CASE("ReceiveRingBuffer_out_of_order_release_reclaims_in_order", "[ImagePuller][Unit]") {
|
||||
ReceiveRingBuffer ring(1000);
|
||||
auto a = ring.Allocate(500);
|
||||
auto b = ring.Allocate(500);
|
||||
REQUIRE(a);
|
||||
REQUIRE(b);
|
||||
|
||||
b.reset(); // the newer region goes first - as republish may do
|
||||
CHECK(ring.GetUtilization() == 1000); // its bytes are behind a's and cannot be reused yet
|
||||
|
||||
auto f = std::async(std::launch::async, [&] { return ring.Allocate(600); });
|
||||
CHECK(f.wait_for(50ms) == std::future_status::timeout);
|
||||
|
||||
a.reset(); // now both are reclaimed
|
||||
auto c = f.get();
|
||||
REQUIRE(c);
|
||||
CHECK(ring.GetUtilization() == 600);
|
||||
}
|
||||
|
||||
TEST_CASE("ReceiveRingBuffer_oversized_message_goes_to_heap", "[ImagePuller][Unit]") {
|
||||
ReceiveRingBuffer ring(100);
|
||||
auto small = ring.Allocate(100);
|
||||
REQUIRE(small);
|
||||
// Does not wait for the full ring - it never fits in it
|
||||
auto big = ring.Allocate(1000);
|
||||
REQUIRE(big);
|
||||
REQUIRE(big->size == 1000);
|
||||
big->data[999] = 1;
|
||||
CHECK(ring.GetUtilization() == 1100);
|
||||
CHECK(ring.GetMaxUtilization() == 1100);
|
||||
big.reset();
|
||||
CHECK(ring.GetUtilization() == 100);
|
||||
}
|
||||
|
||||
TEST_CASE("ReceiveRingBuffer_zero_size", "[ImagePuller][Unit]") {
|
||||
ReceiveRingBuffer ring(100);
|
||||
auto a = ring.Allocate(0);
|
||||
auto b = ring.Allocate(100);
|
||||
REQUIRE(a);
|
||||
REQUIRE(b);
|
||||
CHECK(a->size == 0);
|
||||
a.reset();
|
||||
CHECK(ring.GetUtilization() == 100);
|
||||
b.reset();
|
||||
CHECK(ring.GetUtilization() == 0);
|
||||
}
|
||||
|
||||
TEST_CASE("ReceiveRingBuffer_stop_releases_waiter", "[ImagePuller][Unit]") {
|
||||
ReceiveRingBuffer ring(100);
|
||||
auto a = ring.Allocate(100);
|
||||
REQUIRE(a);
|
||||
auto f = std::async(std::launch::async, [&] { return ring.Allocate(50); });
|
||||
CHECK(f.wait_for(50ms) == std::future_status::timeout);
|
||||
ring.Stop();
|
||||
CHECK(f.get() == nullptr);
|
||||
}
|
||||
|
||||
TEST_CASE("ReceiveRingBuffer_region_outlives_ring", "[ImagePuller][Unit]") {
|
||||
std::shared_ptr<ReceiveBufferRegion> r;
|
||||
{
|
||||
ReceiveRingBuffer ring(100);
|
||||
r = ring.Allocate(50);
|
||||
}
|
||||
REQUIRE(r);
|
||||
r->data[49] = 1;
|
||||
r.reset();
|
||||
}
|
||||
@@ -11,6 +11,7 @@
|
||||
#include "../image_pusher/ZMQStream2Pusher.h"
|
||||
#include "../receiver/JFJochReceiverService.h"
|
||||
#include "../image_pusher/ZMQWriterNotificationPuller.h"
|
||||
#include "../image_puller/TestImagePuller.h"
|
||||
|
||||
TEST_CASE("StreamWriterTest_ZMQ", "[StreamWriter]") {
|
||||
RegisterHDF5Filter();
|
||||
@@ -246,3 +247,52 @@ TEST_CASE("StreamWriterTest_ZMQ_MasterAfterAllDataFiles", "[StreamWriter]") {
|
||||
|
||||
std::filesystem::remove_all("subdir4");
|
||||
}
|
||||
|
||||
// A failure that is not a JFJochException must end the run the same way as one that is: error state,
|
||||
// END processed, state Finalized. Before this test the writer caught only its own exception type at the
|
||||
// per-image and END sites, and anything else - here the JSON library refusing a sample name that is not
|
||||
// UTF-8 while publishing a file-done notification - left Run() with the state stuck in Receiving, no END
|
||||
// acknowledgement and no notification, and on TCP a broker waiting for that acknowledgement for ever.
|
||||
TEST_CASE("StreamWriterTest_NonJFJochException_EndsRunInErrorState", "[StreamWriter]") {
|
||||
RegisterHDF5Filter();
|
||||
Logger logger("StreamWriterTest_NonJFJochException");
|
||||
|
||||
DiffractionExperiment x(DetJF(2));
|
||||
x.FilePrefix("subdir5/StreamWriterTest5").NumTriggers(1).ImagesPerTrigger(2).ImagesPerFile(1)
|
||||
.Compression(CompressionAlgorithm::NO_COMPRESSION).OverwriteExistingFiles(true)
|
||||
.SampleName("\xff\xfe not utf-8");
|
||||
|
||||
StartMessage start_message;
|
||||
x.FillMessage(start_message);
|
||||
start_message.write_master_file = true;
|
||||
|
||||
TestImagePuller puller;
|
||||
// The file-done PUB socket is what makes the writer serialise the sample name to JSON.
|
||||
StreamWriter writer(logger, puller, "ipc://*");
|
||||
|
||||
puller.Put({.cbor = std::make_shared<CBORStream2DeserializerOutput>(start_message)});
|
||||
std::vector<int16_t> image(x.GetPixelsNum());
|
||||
for (int i = 0; i < 2; i++) {
|
||||
DataMessage message{};
|
||||
message.image = CompressedImage(image, x.GetXPixelsNum(), x.GetYPixelsNum());
|
||||
message.number = i;
|
||||
puller.Put({.cbor = std::make_shared<CBORStream2DeserializerOutput>(message)});
|
||||
}
|
||||
EndMessage end_message{};
|
||||
end_message.max_image_number = 2;
|
||||
puller.Put({.cbor = std::make_shared<CBORStream2DeserializerOutput>(end_message)});
|
||||
|
||||
StreamWriterOutput out;
|
||||
REQUIRE_NOTHROW(out = writer.Run());
|
||||
|
||||
// Image 0 filled its one-image data file; closing it published the notification, which threw.
|
||||
CHECK(out.stats.state == StreamWriterState::Finalized);
|
||||
CHECK(out.stats.processed_images == 0);
|
||||
CHECK(out.data_file_stats.empty());
|
||||
// The master is never written for a run that ended in error; the data file was renamed into
|
||||
// place before the notification was attempted.
|
||||
CHECK(!std::filesystem::exists("subdir5/StreamWriterTest5_master.h5"));
|
||||
CHECK(std::filesystem::exists("subdir5/StreamWriterTest5_data_000001.h5"));
|
||||
|
||||
std::filesystem::remove_all("subdir5");
|
||||
}
|
||||
|
||||
@@ -41,8 +41,13 @@ TEST_CASE("TCPImageCommTest_2Writers_WithAck", "[TCP]") {
|
||||
|
||||
TCPStreamPusher pusher("tcp://127.0.0.1:*", npullers);
|
||||
std::vector<std::unique_ptr<TCPImagePuller> > puller;
|
||||
// The second writer receives into a ring that holds three images (with their CBOR framing): a
|
||||
// full ring stalls its read, the pusher's send blocks, and nothing may be lost or reordered.
|
||||
const size_t small_ring = 3 * x.GetPixelsNum() * sizeof(uint16_t) + 3 * 65536;
|
||||
for (int i = 0; i < npullers; i++)
|
||||
puller.push_back(std::make_unique<TCPImagePuller>(pusher.GetAddress()[0], 64 * 1024 * 1024));
|
||||
puller.push_back(std::make_unique<TCPImagePuller>(pusher.GetAddress()[0], 64 * 1024 * 1024, "", std::nullopt,
|
||||
(i == 1) ? std::optional<size_t>(small_ring) : std::nullopt));
|
||||
REQUIRE(puller[1]->GetRingBufferCapacity() == small_ring);
|
||||
|
||||
// Wait for all pullers to connect before starting data collection
|
||||
for (int attempt = 0; attempt < 100 && pusher.GetConnectedWriters() < static_cast<size_t>(npullers); ++attempt)
|
||||
|
||||
@@ -420,3 +420,75 @@ TEST_CASE("ZMQImagePuller_RecoversFromFullQueue","[ZeroMQ]") {
|
||||
|
||||
puller.Disconnect();
|
||||
}
|
||||
|
||||
// The whole run goes through a ring that holds three images: the puller has to wait for space, the
|
||||
// socket queue fills, the pusher blocks, and nothing is lost or reordered. The peak ring use proves
|
||||
// the bound held.
|
||||
TEST_CASE("ZMQImageCommTest_SmallRingBuffer","[ZeroMQ]") {
|
||||
const size_t nframes = 64;
|
||||
|
||||
Logger logger(Catch::getResultCapture().getCurrentTestName());
|
||||
|
||||
DiffractionExperiment x(DetJF(1));
|
||||
x.Raw();
|
||||
x.PedestalG0Frames(0).NumTriggers(1).UseInternalPacketGenerator(false).IncidentEnergy_keV(12.4)
|
||||
.ImagesPerTrigger(nframes).Compression(CompressionAlgorithm::NO_COMPRESSION);
|
||||
REQUIRE(x.GetImageNum() == nframes);
|
||||
|
||||
std::mt19937 g1(1387);
|
||||
std::uniform_int_distribution<uint16_t> dist;
|
||||
std::vector<uint16_t> image1(x.GetPixelsNum()*nframes);
|
||||
for (auto &i: image1) i = dist(g1);
|
||||
|
||||
const size_t image_bytes = x.GetPixelsNum() * sizeof(uint16_t);
|
||||
const size_t ring_size = 3 * image_bytes + 3 * 65536; // three images with their CBOR framing
|
||||
|
||||
ZMQStream2Pusher pusher({"ipc://*"});
|
||||
std::vector<size_t> diff_size(1), diff_content(1), diff_split(1), nimages(1);
|
||||
|
||||
auto pusher_addr = pusher.GetAddress();
|
||||
ZMQImagePuller puller(pusher_addr[0], "", {}, {}, ring_size);
|
||||
REQUIRE(puller.GetRingBufferCapacity() == ring_size);
|
||||
|
||||
std::thread sender_thread = std::thread([&] {
|
||||
std::vector<uint8_t> serialization_buffer(16*1024*1024);
|
||||
CBORStream2Serializer serializer(serialization_buffer.data(), serialization_buffer.size());
|
||||
|
||||
StartMessage message {
|
||||
.images_per_file = 16,
|
||||
.write_master_file = true
|
||||
};
|
||||
EndMessage end_message{};
|
||||
|
||||
pusher.StartDataCollection(message);
|
||||
for (int i = 0; i < nframes; i++) {
|
||||
DataMessage data_message;
|
||||
data_message.number = i;
|
||||
data_message.image = CompressedImage(image1.data() + i * x.GetPixelsNum(),
|
||||
image_bytes,
|
||||
x.GetXPixelsNum(),
|
||||
x.GetYPixelsNum(),
|
||||
x.GetImageMode(),
|
||||
x.GetCompressionAlgorithm());
|
||||
serializer.SerializeImage(data_message);
|
||||
pusher.SendImage(serialization_buffer.data(), serializer.GetBufferSize(), i);
|
||||
}
|
||||
|
||||
pusher.EndDataCollection(end_message);
|
||||
});
|
||||
|
||||
std::thread puller_thread(test_puller, &puller, std::cref(x), std::cref(image1), 1, 0,
|
||||
std::ref(diff_split), std::ref(diff_size), std::ref(diff_content),
|
||||
std::ref(nimages));
|
||||
|
||||
sender_thread.join();
|
||||
puller_thread.join();
|
||||
|
||||
puller.Disconnect();
|
||||
|
||||
REQUIRE(nimages[0] == nframes);
|
||||
REQUIRE(diff_size[0] == 0);
|
||||
REQUIRE(diff_content[0] == 0);
|
||||
CHECK(puller.GetMaxRingBufferUtilization() <= ring_size);
|
||||
CHECK(puller.GetMaxRingBufferUtilization() >= image_bytes);
|
||||
}
|
||||
|
||||
@@ -27,9 +27,9 @@
|
||||
{"id": "insu_I_x06da_5keV", "input": "insu_I_x06da_5keV/Ins_I_6_006_master.h5", "wavelength": 2.47923, "ref": {"sgno": 197, "cell": [77.785, 77.785, 77.785, 90.0, 90.0, 90.0], "anomalous": true, "isa": 17.54, "completeness": 91.7, "r_meas": 0.073, "cc_half": 0.999, "multiplicity": 11.17, "dmin_low": 7.21, "r_meas_low": 0.048, "cc_half_high": 0.855, "dmin": 2.45, "dmin_rule": "xds_range", "dmin_xds": 2.45, "dmax": 50.0}, "tags": ["h5", "insulin", "long-wavelength"], "model": "2bn3"},
|
||||
{"id": "insu_I_x06da_6keV", "input": "insu_I_x06da_6keV/Ins_I_6_004_master.h5", "wavelength": 2.06648, "ref": {"sgno": 197, "cell": [77.71, 77.71, 77.71, 90.0, 90.0, 90.0], "anomalous": true, "isa": 17.95, "completeness": 92.4, "r_meas": 0.065, "cc_half": 0.999, "multiplicity": 12.42, "dmin_low": 6.03, "r_meas_low": 0.047, "cc_half_high": 0.905, "dmin": 2.04, "dmin_rule": "xds_range", "dmin_xds": 2.04, "dmax": 50.0}, "tags": ["h5", "insulin", "long-wavelength"], "model": "2bn3"},
|
||||
{"id": "insu_I_x06da_5keV_2", "input": "insu_I_x06da_5keV_2/Ins_I_7_001_master.h5", "wavelength": 2.47923, "ref": {"sgno": 197, "cell": [77.694, 77.694, 77.694, 90.0, 90.0, 90.0], "anomalous": true, "isa": 20.01, "completeness": 91.8, "r_meas": 0.076, "cc_half": 0.999, "multiplicity": 12.73, "dmin_low": 7.21, "r_meas_low": 0.048, "cc_half_high": 0.888, "dmin": 2.45, "dmin_rule": "xds_range", "dmin_xds": 2.45, "dmax": 50.0}, "tags": ["h5", "insulin", "long-wavelength"], "tiers": {"smoke": "cubic insulin at 5 keV (long wavelength), fast"}, "model": "2bn3"},
|
||||
{"id": "cytc_x10sa", "input": "cytc_x10sa/cytC_10_002_master.h5", "wavelength": 0.99989, "ref": {"sgno": 152, "cell": [83.659, 83.659, 86.011, 90.0, 90.0, 120.0], "anomalous": true, "isa": 31.75, "completeness": 99.8, "r_meas": 0.232, "cc_half": 0.999, "multiplicity": 10.65, "dmin_low": 6.05, "r_meas_low": 0.036, "cc_half_high": 0.046, "dmin": 2.267, "dmin_rule": "cc_half_0.30", "dmin_xds": 2.039, "dmax": 50.0}, "tags": ["h5", "cytochrome-c"], "model": "reference_model.cif", "tiers": {"ci": "trigonal P3121 with tNCS; the hand comes from the reference model"}},
|
||||
{"id": "cytc_x06da_1", "input": "cytc_x06da_1/cytC_2_0770b5_master.h5", "wavelength": 0.95375, "ref": {"sgno": 152, "cell": [83.836, 83.836, 86.733, 90.0, 90.0, 120.0], "anomalous": true, "isa": 24.49, "completeness": 99.9, "r_meas": 0.084, "cc_half": 0.999, "multiplicity": 7.67, "dmin_low": 5.58, "r_meas_low": 0.033, "cc_half_high": 0.366, "dmin": 1.875, "dmin_rule": "xds_range", "dmin_xds": 1.875, "dmax": 50.0}, "tags": ["h5", "cytochrome-c"], "model": "reference_model.cif"},
|
||||
{"id": "cytc_x06da_2", "input": "cytc_x06da_2/cytC_3_874285_master.h5", "wavelength": 0.95375, "ref": {"sgno": 152, "cell": [83.719, 83.719, 88.617, 90.0, 90.0, 120.0], "anomalous": true, "isa": 26.98, "completeness": 99.9, "r_meas": 0.097, "cc_half": 0.999, "multiplicity": 7.82, "dmin_low": 5.03, "r_meas_low": 0.032, "cc_half_high": 0.236, "dmin": 1.69, "dmin_rule": "xds_range", "dmin_xds": 1.69, "dmax": 50.0}, "tags": ["h5", "cytochrome-c"], "model": "reference_model.cif"},
|
||||
{"id": "cytc_x10sa", "input": "cytc_x10sa/cytC_10_002_master.h5", "wavelength": 0.99989, "ref": {"sgno": 152, "cell": [83.659, 83.659, 86.011, 90.0, 90.0, 120.0], "anomalous": true, "isa": 31.75, "completeness": 99.8, "r_meas": 0.232, "cc_half": 0.999, "multiplicity": 10.65, "dmin_low": 6.05, "r_meas_low": 0.036, "cc_half_high": 0.046, "dmin": 2.267, "dmin_rule": "cc_half_0.30", "dmin_xds": 2.039, "dmax": 50.0}, "tags": ["h5", "cytochrome-c"], "model": "reference_model.cif", "tiers": {"ci": "trigonal P3121 with tNCS; the hand comes from the reference model"}, "ref_override": {"sg": "P 32 2 1", "sgno": 154}, "ref_override_why": "The reference model (P 32 2 1, three copies) fixes the hand that XDS left arbitrary: the 2-fold of P 32 2 1 holds to measurement noise in every shell and a P 32 six-copy model, twinned or not, fits no better on the same data (in-house analysis 2026-10-13)"},
|
||||
{"id": "cytc_x06da_1", "input": "cytc_x06da_1/cytC_2_0770b5_master.h5", "wavelength": 0.95375, "ref": {"sgno": 152, "cell": [83.836, 83.836, 86.733, 90.0, 90.0, 120.0], "anomalous": true, "isa": 24.49, "completeness": 99.9, "r_meas": 0.084, "cc_half": 0.999, "multiplicity": 7.67, "dmin_low": 5.58, "r_meas_low": 0.033, "cc_half_high": 0.366, "dmin": 1.875, "dmin_rule": "xds_range", "dmin_xds": 1.875, "dmax": 50.0}, "tags": ["h5", "cytochrome-c"], "model": "reference_model.cif", "ref_override": {"sg": "P 32 2 1", "sgno": 154}, "ref_override_why": "The reference model (P 32 2 1, three copies) fixes the hand that XDS left arbitrary: the 2-fold of P 32 2 1 holds to measurement noise in every shell and a P 32 six-copy model, twinned or not, fits no better on the same data (in-house analysis 2026-10-13)"},
|
||||
{"id": "cytc_x06da_2", "input": "cytc_x06da_2/cytC_3_874285_master.h5", "wavelength": 0.95375, "ref": {"sgno": 152, "cell": [83.719, 83.719, 88.617, 90.0, 90.0, 120.0], "anomalous": true, "isa": 26.98, "completeness": 99.9, "r_meas": 0.097, "cc_half": 0.999, "multiplicity": 7.82, "dmin_low": 5.03, "r_meas_low": 0.032, "cc_half_high": 0.236, "dmin": 1.69, "dmin_rule": "xds_range", "dmin_xds": 1.69, "dmax": 50.0}, "tags": ["h5", "cytochrome-c"], "model": "reference_model.cif", "ref_override": {"sg": "P 32 2 1", "sgno": 154}, "ref_override_why": "The reference model (P 32 2 1, three copies) fixes the hand that XDS left arbitrary: the 2-fold of P 32 2 1 holds to measurement noise in every shell and a P 32 six-copy model, twinned or not, fits no better on the same data (in-house analysis 2026-10-13)"},
|
||||
{"id": "myob_x10sa", "input": "myob_x10sa/MyoB_13_110ee0_master.h5", "wavelength": 0.99988, "ref": {"sgno": 4, "cell": [35.229, 28.461, 63.076, 90.0, 106.504, 90.0], "anomalous": true, "isa": 5.21, "completeness": 97.7, "r_meas": 0.247, "cc_half": 0.965, "multiplicity": 3.29, "dmin_low": 5.16, "r_meas_low": 0.186, "cc_half_high": 0.359, "dmin": 1.742, "dmin_rule": "xds_range", "dmin_xds": 1.742, "dmax": 50.0}, "tags": ["h5", "myoglobin"], "model": "4dc8", "tiers": {"ci": "monoclinic P21, Eiger"}},
|
||||
{"id": "myob_x06da", "input": "myob_x06da/MyoB_9_446b70_master.h5", "wavelength": 0.95365, "ref": {"sgno": 3, "cell": [35.182, 28.544, 62.83, 90.0, 106.138, 90.0], "anomalous": true, "isa": 7.59, "completeness": 99.4, "r_meas": 0.232, "cc_half": 0.987, "multiplicity": 3.47, "dmin_low": 4.23, "r_meas_low": 0.105, "cc_half_high": 0.267, "dmin": 1.422, "dmin_rule": "xds_range", "dmin_xds": 1.422, "dmax": 50.0}, "tags": ["h5", "myoglobin"], "tiers": {"smoke": "monoclinic myoglobin, screw not arbitrable by XDS"}, "ref_override": {"sg": "P 1 21 1", "sgno": 4}, "ref_override_why": "myoglobin, P 1 21 1. XDS does not judge screw axes (it reports P 2)", "model": "4dc8"},
|
||||
{"id": "myob_x06da_powder_1", "input": "myob_x06da_powder_1/MyoB2-1_75979a_master.h5", "wavelength": 0.95373, "ref": {"sgno": 1, "cell": [28.705, 34.796, 64.373, 74.314, 89.879, 90.057], "anomalous": false, "isa": 5.54, "completeness": 74.2, "r_meas": 0.493, "cc_half": 0.966, "multiplicity": 2.87, "dmin_low": 4.44, "r_meas_low": 0.125, "cc_half_high": 0.953, "dmin": 1.495, "dmin_rule": "xds_range", "dmin_xds": 1.495, "dmax": 50.0}, "tags": ["h5", "myoglobin", "twin"], "tiers": {"smoke": "known hard: triclinic twin"}, "ref_override": {"sg": "P 1 21 1", "sgno": 4, "cell": [35.14, 28.88, 64.64, 90.0, 105.9, 90.0]}, "ref_override_why": "myoglobin, P 1 21 1. XDS indexed this sweep (a powder-contaminated crystal) in P 1 on a reduced cell (28.7 34.8 64.4 74.3 89.9 90.1); the monoclinic cell is that cell's conventional setting", "model": "4dc8"},
|
||||
|
||||
@@ -41,11 +41,11 @@
|
||||
{"id": "6oel", "input": "6oel/1449-8_3_001.img", "wavelength": 0.99994, "ref": {"sg": "F 41 3 2", "sgno": 210, "cell": [328.1, 328.1, 328.1, 90.0, 90.0, 90.0], "dmin": 3.1}, "tags": ["SMV", "cubic"], "tiers": {"smoke": "SMV reader (ADSC), cubic F4132"}},
|
||||
{"id": "6p8p", "input": "6p8p/14_12_1_0001.cbf", "wavelength": 0.9792, "ref": {"sg": "P 4", "sgno": 75, "cell": [97.525, 97.525, 60.134, 90.0, 90.0, 90.0], "dmin": 1.635}, "tags": ["cbf", "tetragonal"]},
|
||||
{"id": "6pb3", "input": "6pb3/14_16_1_000001.cbf", "wavelength": 0.97918, "ref": {"sg": "P 6", "sgno": 168, "cell": [100.436, 100.436, 48.86, 90.0, 90.0, 120.0], "dmin": 2.048}, "tags": ["cbf", "hexagonal"]},
|
||||
{"id": "6pxb", "input": "6pxb/TJB1_3_rachel_1_0001.cbf", "wavelength": 0.9792, "pinned": true, "ref": {"sg": "P 32", "sgno": 145, "cell": [64.021, 64.021, 119.447, 90.0, 90.0, 120.0], "dmin": 1.747}, "ref_alternatives": [{"sg": "P 32 1 2", "sgno": 153, "why": "The deposition merged in point group 3 (its 55502 unique reflections to 1.747 A are what Laue class -3 holds, twice what -3 1 m would) and refined six chains in P 32. The intensities read point group 312 instead: the three added two-folds correlate at 0.98-0.99, at or above the three-folds nobody disputes (0.98), against 0.37-0.40 for the 321 and 622 operators, POINTLESS on our P1 merge picks P -3 1 m (likelihood 1.000), and the reflections centric in 312 but not in 3 are distributed as centric (+435 nats), which a twin law cannot produce. Against that, the deposited chains pair under the added two-fold at 0.3-0.7 A, more than coordinate error at 1.75 A, and the model tells the two indexings apart (R 0.22 against 0.25), so the two-fold may be a near-exact non-crystallographic one; ZANUDA settles on P 32 2 1, whose operators these data do not support. Neither answer is established, so both are accepted; P 31 1 2, which the data cannot separate from P 32 1 2, is accepted as its hand"}], "symmetry_evidence": "symmetry/6pxb", "tags": ["cbf", "trigonal"]},
|
||||
{"id": "6pxb", "input": "6pxb/TJB1_3_rachel_1_0001.cbf", "wavelength": 0.9792, "pinned": true, "ref": {"sg": "P 32", "sgno": 145, "cell": [64.021, 64.021, 119.447, 90.0, 90.0, 120.0], "dmin": 1.747}, "ref_alternatives": [{"sg": "P 32 1 2", "sgno": 153, "why": "The deposition merged in point group 3 (its 55502 unique reflections to 1.747 A are what Laue class -3 holds, twice what -3 1 m would) and refined six chains in P 32. The intensities read point group 312 instead: the three added two-folds correlate at 0.98-0.99, at or above the three-folds nobody disputes (0.98), against 0.37-0.40 for the 321 and 622 operators, POINTLESS on our P1 merge picks P -3 1 m (likelihood 1.000), and the reflections centric in 312 but not in 3 are distributed as centric (+435 nats), which a twin law cannot produce. Against that, the deposited chains pair under the added two-fold at 0.3-0.7 A, more than coordinate error at 1.75 A, and the model tells the two indexings apart (R 0.22 against 0.25), so the two-fold may be a near-exact non-crystallographic one; ZANUDA 1.097 with the deposited model keeps P 32 on our data (three free sets) and the depositor's; the model refined in P 32 1 2 reaches R-free 0.49-0.53 against 0.33-0.37. Neither answer is established, so both are accepted; P 31 1 2, which the data cannot separate from P 32 1 2, is accepted as its hand"}], "symmetry_evidence": "symmetry/6pxb", "tags": ["cbf", "trigonal"]},
|
||||
{"id": "6pxc", "input": "6pxc/TJB6_1_Rachel_1_0001.cbf", "wavelength": 0.9791, "ref": {"sg": "I 2 2 2", "sgno": 23, "cell": [44.19, 64.827, 87.239, 90.0, 90.0, 90.0], "dmin": 1.6}, "tags": ["cbf", "orthorhombic"], "tiers": {"ci": "body-centred orthorhombic I222"}},
|
||||
{"id": "6r72", "input": "6r72/V-CK63-8-ld_1_master.h5", "wavelength": 0.98011, "ref": {"sg": "P 1 21 1", "sgno": 4, "cell": [117.805, 110.754, 155.615, 90.0, 93.23, 90.0], "dmin": 3.95}, "pinned": true, "tags": ["h5", "monoclinic"]},
|
||||
{"id": "6rlr", "input": "6rlr/_b4_1_master.h5", "wavelength": 0.97949, "ref": {"sg": "P 1", "sgno": 1, "cell": [39.986, 39.998, 63.643, 80.39, 76.29, 68.15], "dmin": 2.0}, "tags": ["h5", "triclinic", "twin"]},
|
||||
{"id": "6toc", "input": "6toc/zenodo/dts_1_00001.cbf", "wavelength": 1.00004, "ref": {"sg": "P 42", "sgno": 77, "cell": [31.523, 31.523, 81.599, 90.0, 90.0, 90.0], "dmin": 1.853}, "ref_alternatives": [{"sg": "P 42 2 2", "sgno": 93, "why": "Our reduction and an independent POINTLESS run on our own P1 merge both read point group 422, and the deposited asymmetric unit's two chains are related by the very two-fold the higher group adds, to 0.16 A CA RMSD over 43 residues - coordinate error at 1.85 A. Merging in P 42 2 2 costs 0.0006 in R_meas for 1.75x the multiplicity and correlates better with the deposited model (0.9802 vs 0.9764). The refinement test is NOT unanimous: ZANUDA 1.097 refines P 42 2 2 to R-free 0.2561 against P 42's 0.2636 at half the parameters and reports the deposited assignment incorrect, while an independent Refmac 5.8.0431 comparison on a symmetry-consistent free set puts P 42 ahead by 0.004-0.020 depending on cycle count - less than the spread between refinement protocols. Neither answer is established: a pseudo-symmetry too exact for any test we have remains a live explanation, and so does the deposited assignment. Either is accepted"}], "symmetry_evidence": "symmetry/6toc", "tags": ["cbf", "tetragonal", "twin"]},
|
||||
{"id": "6toc", "input": "6toc/zenodo/dts_1_00001.cbf", "wavelength": 1.00004, "ref": {"sg": "P 42", "sgno": 77, "cell": [31.523, 31.523, 81.599, 90.0, 90.0, 90.0], "dmin": 1.853}, "ref_alternatives": [{"sg": "P 42 2 2", "sgno": 93, "why": "Our reduction and an independent POINTLESS run on our own P1 merge both read point group 422, and the deposited asymmetric unit's two chains are related by the very two-fold the higher group adds, to 0.16 A CA RMSD over 43 residues - coordinate error at 1.85 A. Merging in P 42 2 2 costs 0.0006 in R_meas for 1.75x the multiplicity and correlates better with the deposited model (0.9802 vs 0.9764). The refinement test is NOT unanimous: ZANUDA 1.097 prefers P 42 2 2 on all four data sets tried (ours with three free sets, the depositor's), but R-free favours P 42 2 2 once by 0.008 and P 42 three times by 0.015-0.018, while an independent Refmac 5.8.0431 comparison on a symmetry-consistent free set puts P 42 ahead by 0.004-0.020 depending on cycle count - less than the spread between refinement protocols. Neither answer is established: a pseudo-symmetry too exact for any test we have remains a live explanation, and so does the deposited assignment. Either is accepted"}], "symmetry_evidence": "symmetry/6toc", "tags": ["cbf", "tetragonal", "twin"]},
|
||||
{"id": "6ttn", "input": "6ttn/6ttn/data/H6H_33_01_hyo_full_1_0001.cbf", "wavelength": 0.9184, "ref": {"sg": "P 21 21 21", "sgno": 19, "cell": [39.894, 79.841, 104.701, 90.0, 90.0, 90.0], "dmin": 1.12}, "tags": ["cbf", "orthorhombic"]},
|
||||
{"id": "6u7g", "input": "6u7g/6u7g/data/SNB02_11_8_1_master.h5", "wavelength": 1.03318, "ref": {"sg": "P 1 21 1", "sgno": 4, "cell": [99.555, 98.682, 147.525, 90.0, 104.6, 90.0], "dmin": 2.35}, "pinned": true, "tags": ["h5", "monoclinic"]},
|
||||
{"id": "6ukf", "input": "6ukf/6ukf/data/XDC-7_Pn6_000001.cbf", "wavelength": 0.8, "ref": {"sg": "P 1 21 1", "sgno": 4, "cell": [61.018, 37.317, 69.027, 90.0, 109.768, 90.0], "dmin": 1.0}, "tags": ["cbf", "monoclinic"]},
|
||||
@@ -106,9 +106,9 @@
|
||||
{"id": "8u0i", "input": "8u0i/8u0i/data/dI-2-05a_4_00001.cbf", "wavelength": 1.00004, "ref": {"sg": "P 43 21 2", "sgno": 96, "cell": [50.317, 50.317, 90.596, 90.0, 90.0, 90.0], "dmin": 1.541}, "tags": ["cbf", "tetragonal"], "tiers": {"ci": "miniCBF, tetragonal P43212"}},
|
||||
{"id": "8v4o", "input": "8v4o/CaalA_00629_a_FS11-AMP_8v4o/data/PSL-1602_2180_master.h5", "wavelength": 0.97949, "ref": {"sg": "P 61 2 2", "sgno": 178, "cell": [139.46, 139.46, 544.99, 90.0, 90.0, 120.0], "dmin": 2.7}, "tags": ["h5", "hexagonal"]},
|
||||
{"id": "8xbp", "input": "8xbp/8xbp/data/puck2872_3_1_master.h5", "wavelength": 0.97856, "ref": {"sg": "C 1 2 1", "sgno": 5, "cell": [148.29, 50.78, 60.21, 90.0, 92.33, 90.0], "dmin": 1.99}, "ref_override": {"cell": [148.736, 51.777, 60.181, 90.0, 92.36, 90.0]}, "ref_override_why": "the uploaded sweep is a different crystal from the one the deposited cell describes: master data_collection_date 2023-06-21, deposited _diffrn_detector.pdbx_collection_date 2023-06-23, and the deposited b = 50.78 is 2.0% from the b these images give. The override is the cell DIALS 3.29 indexes de novo on this master, b = 51.777(11); the deposited space group and d_min are kept", "tags": ["h5", "monoclinic"]},
|
||||
{"id": "8xte", "input": "8xte/1101/G2_1_00001.cbf", "wavelength": 0.97853, "ref": {"sg": "P 32", "sgno": 145, "cell": [208.77, 208.77, 67.22, 90.0, 90.0, 120.0], "dmin": 1.99}, "ref_alternatives": [{"sg": "P 31 2 1", "sgno": 152, "why": "The evidence favours the higher group here. The twin-immune centric zone of the added two-folds - reflections the higher group makes centric, which are their own twin mates and so cannot be made to read centric by a merohedral twin law - gives <|E^2-1|> = 0.946 +/- 0.012 against a centric expectation of 0.968 and an acentric 0.736, at +707 nats. Re-refinement on a shared free set with the twin law removed from both sides favours P 3_2 2 1 (0.2177/0.2322) over P 3_2 (0.2460/0.2612), and the deposited entry's published R values reproduce only with an undeclared twin law h,-h-k,-l at alpha = 0.50, which is itself a 321-symmetric target. No refinement R can close the question in principle, because a merohedral twin at exactly alpha = 0.5 and true 321 predict identical intensities; the case rests on the centric zone. Both answers are accepted, ours being the better supported"}], "symmetry_evidence": "symmetry/8xte", "tags": ["cbf", "trigonal"]},
|
||||
{"id": "8xte", "input": "8xte/1101/G2_1_00001.cbf", "wavelength": 0.97853, "ref": {"sg": "P 32", "sgno": 145, "cell": [208.77, 208.77, 67.22, 90.0, 90.0, 120.0], "dmin": 1.99}, "ref_alternatives": [{"sg": "P 31 2 1", "sgno": 152, "why": "The evidence favours the higher group here. The twin-immune centric zone of the added two-folds - reflections the higher group makes centric, which are their own twin mates and so cannot be made to read centric by a merohedral twin law - gives <|E^2-1|> = 0.946 +/- 0.012 against a centric expectation of 0.968 and an acentric 0.736, at +707 nats. Re-refinement on a shared free set with the twin law removed from both sides favours P 3_2 2 1 (0.2177/0.2322) over P 3_2 (0.2460/0.2612), and the deposited entry's published R values reproduce only with an undeclared twin law h,-h-k,-l at alpha = 0.50, which is itself a 321-symmetric target. No refinement R can close the question in principle, because a merohedral twin at exactly alpha = 0.5 and true 321 predict identical intensities; the case rests on the centric zone. ZANUDA 1.097 keeps P 32: the deposited chains are 6.1 A from the 321 symmetry and refine to R-free 0.54-0.57 in P 32 2 1; the deposited model gives the higher group no support. Both answers are accepted, neither established"}], "symmetry_evidence": "symmetry/8xte", "tags": ["cbf", "trigonal"]},
|
||||
{"id": "8xtf", "input": "8xtf/1102/puck03_12_1_master.h5", "wavelength": 0.97918, "ref": {"sg": "H 3 2", "sgno": 155, "cell": [211.75, 211.75, 67.42, 90.0, 90.0, 120.0], "dmin": 2.13}, "tags": ["h5", "trigonal"]},
|
||||
{"id": "8xtg", "input": "8xtg/1100/mpag15_1_00001.cbf", "wavelength": 0.97853, "ref": {"sg": "P 32", "sgno": 145, "cell": [199.54, 199.54, 67.15, 90.0, 90.0, 120.0], "dmin": 2.0}, "ref_alternatives": [{"sg": "P 31 2 1", "sgno": 152, "why": "The evidence favours the DEPOSITION here, and this row must not be read as the 8xte one. Every correlation-based instrument we have - our own operator correlations, POINTLESS (0.85 on our P1 merge) - reads point group 321, but our own twin-immune centric-zone test, the only one that separates real symmetry from pseudo-symmetry, read <|E^2-1|> = 0.869 at -44.9 nats against the promotion when it assumed an untwinned zone; read at the fraction of the crystal's other twin laws (rc174) it now favours the promotion, with an L-test twin fraction of 0.20-0.26. Whether this crystal is partially twinned or purely pseudo-symmetric is not established. Both answers are accepted, the deposition being the better supported"}], "symmetry_evidence": "symmetry/8xtg", "tags": ["cbf", "trigonal"]},
|
||||
{"id": "8xtg", "input": "8xtg/1100/mpag15_1_00001.cbf", "wavelength": 0.97853, "ref": {"sg": "P 32", "sgno": 145, "cell": [199.54, 199.54, 67.15, 90.0, 90.0, 120.0], "dmin": 2.0}, "ref_alternatives": [{"sg": "P 31 2 1", "sgno": 152, "why": "The evidence favours the DEPOSITION here, and this row must not be read as the 8xte one. Every correlation-based instrument we have - our own operator correlations, POINTLESS (0.85 on our P1 merge) - reads point group 321, but our own twin-immune centric-zone test, the only one that separates real symmetry from pseudo-symmetry, read <|E^2-1|> = 0.869 at -44.9 nats against the promotion when it assumed an untwinned zone; read at the fraction of the crystal's other twin laws (rc174) it now favours the promotion, with an L-test twin fraction of 0.20-0.26. Whether this crystal is partially twinned or purely pseudo-symmetric is not established. ZANUDA 1.097 prefers P 32 2 1 on all four data sets (model within 1.2 A of it), while R-free favours P 32 by 0.006-0.019 on each. Both answers are accepted, the deposition being the better supported"}], "symmetry_evidence": "symmetry/8xtg", "tags": ["cbf", "trigonal"]},
|
||||
{"id": "8y74", "input": "8y74/1_1/1_1_master.h5", "wavelength": 0.97918, "ref": {"sg": "C 1 2 1", "sgno": 5, "cell": [125.782, 76.553, 87.143, 90.0, 92.449, 90.0], "dmin": 1.9}, "tags": ["h5", "monoclinic"]},
|
||||
{"id": "8ys9", "input": "8ys9/8YS9_xray-data_8ys9/data/SDS_09_18112_master.h5", "wavelength": 0.97957, "ref": {"sg": "P 21 21 21", "sgno": 19, "cell": [71.04, 77.68, 83.16, 90.0, 90.0, 90.0], "dmin": 1.46}, "tags": ["h5", "orthorhombic"]},
|
||||
{"id": "9b22", "input": "9b22/KlpnC_20447_a_B1-AR6-AMP_9b22/data/PSL-1011_1580_master.h5", "wavelength": 0.97856, "ref": {"sg": "P 1 21 1", "sgno": 4, "cell": [39.826, 92.659, 57.716, 90.0, 91.68, 90.0], "dmin": 1.3}, "tags": ["h5", "monoclinic"], "tiers": {"ci": "HDF5, monoclinic P21 at 1.1 A"}},
|
||||
@@ -186,7 +186,7 @@
|
||||
{"id": "2xfw", "input": "2xfw/E192N_alone_MS_1_001.img", "wavelength": 0.9796, "ref": {"sg": "P 1 21 1", "sgno": 4, "cell": [78.11, 116.53, 83.74, 90, 117.95, 90], "dmin": 1.65}, "tags": ["smv", "monoclinic", "twin", "pseudo-merohedral"]},
|
||||
{"id": "2wnn", "input": "2wnn/wt_pyr_pos3_MS_3_001.img", "wavelength": 0.92, "ref": {"sg": "P 1 21 1", "sgno": 4, "cell": [54.693, 142.456, 83.626, 90, 109.16, 90], "dmin": 1.65}, "tags": ["smv", "monoclinic", "twin", "pseudo-merohedral"]},
|
||||
{"id": "2wnz", "input": "2wnz/E192_pyr_MS_1_001.img", "wavelength": 0.9796, "ref": {"sg": "P 1 21 1", "sgno": 4, "cell": [56.921, 143.041, 83.918, 90, 109.79, 90], "dmin": 1.85}, "tags": ["smv", "monoclinic", "pseudo-symmetry"]},
|
||||
{"id": "6oww", "input": "6oww/data/t1_1_00001.cbf", "wavelength": 1.3857, "ref": {"sg": "P 1 21 1", "sgno": 4, "cell": [85.832, 86.015, 214.34, 90.0, 89.91, 90.0], "dmin": 3.84}, "ref_alternatives": [{"sg": "P 41 21 2", "sgno": 92, "why": "OPEN QUESTION, both accepted. The deposited 8-chain P 1 21 1 model itself obeys a 4-fold (0.13-0.16 A) and a 2-fold (0.10 A) with consistent translations, and the model's |Fc|^2 correlate across the added 422 operators at 1.00/0.98/0.97 - the deposition is likely under-called, and our data support point group 422. Not established: no refinement in P 41 21 2 has been compared with the deposited P 1 21 1, so a pseudo-symmetry too exact for these tests remains a live explanation, as does the deposited assignment (cf. 6toc)"}], "symmetry_evidence": "symmetry/6oww", "tags": ["cbf", "monoclinic", "pseudo-symmetry"]},
|
||||
{"id": "6oww", "input": "6oww/data/t1_1_00001.cbf", "wavelength": 1.3857, "ref": {"sg": "P 1 21 1", "sgno": 4, "cell": [85.832, 86.015, 214.34, 90.0, 89.91, 90.0], "dmin": 3.84}, "ref_alternatives": [{"sg": "P 41 21 2", "sgno": 92, "why": "OPEN QUESTION, both accepted. The deposited 8-chain P 1 21 1 model obeys a 4-fold (0.13-0.16 A) and a 2-fold (0.10 A), and its |Fc|^2 correlate across the added 422 operators at 1.00/0.98/0.97. POINTLESS reads 4/mmm on our observations (0.98) and on the depositor's amplitudes (0.999). ZANUDA 1.097 with the deposited model prefers P 43 21 2 on our data (three free sets, R-free 0.414-0.430 against 0.435-0.447) and on the depositor's (0.333 against 0.359). A higher symmetry is consistent with our data and the model; the model's hand is P 43 21 2. An exact pseudo-symmetry may explain this (evidence: tools/battery/symmetry/6oww)"}], "symmetry_evidence": "symmetry/6oww", "tags": ["cbf", "monoclinic", "pseudo-symmetry"]},
|
||||
{"id": "6p8j", "input": "6p8j/13_12_2_000001.cbf", "wavelength": 0.9792, "ref": {"sg": "P 1 21 1", "sgno": 4, "cell": [46.484, 152.166, 88.255, 90.0, 90.12, 90.0], "dmin": 1.47}, "symmetry_evidence": "symmetry/6p8j", "tags": ["cbf", "monoclinic", "pseudo-symmetry"]},
|
||||
{"id": "9qvv", "input": "9qvv/CHRD_HEP_M/CHRD_404_7_1.nxs", "wavelength": 0.9762, "ref": {"sg": "I 2 2 2", "sgno": 23, "cell": [120.142, 123.8, 123.787, 90.0, 90.0, 90.0], "dmin": 2.72}, "tags": ["nxs", "orthorhombic", "pseudo-symmetry"]},
|
||||
{"id": "5ojv", "input": "5ojv/bg3047_w1_1_0001.cbf", "wavelength": 0.97625, "ref": {"sg": "P 21 21 2", "sgno": 18, "cell": [87.792, 158.242, 87.589, 90.0, 90.0, 90.0], "dmin": 2.062}, "tags": ["cbf", "orthorhombic", "pseudo-symmetry"]},
|
||||
|
||||
+11
-7
@@ -87,9 +87,11 @@ void FileWriter::WriteHDF5(const DataMessage& msg) {
|
||||
if (files.size() <= file_number)
|
||||
files.resize(file_number + 1);
|
||||
|
||||
if (!files[file_number])
|
||||
if (!files[file_number]) {
|
||||
files[file_number] = std::make_unique<HDF5DataFile>(start_message, file_number,
|
||||
HDF5Metadata::DataFileName(start_message, file_number));
|
||||
open_files.insert(file_number);
|
||||
}
|
||||
files[file_number]->Write(msg, image_number);
|
||||
|
||||
if (files[file_number]->GetNumImages() == start_message.images_per_file) {
|
||||
@@ -111,6 +113,7 @@ void FileWriter::CloseFile(uint64_t file_number) {
|
||||
auto file_stats = files[file_number]->Close();
|
||||
files[file_number].reset();
|
||||
closed_files.insert(file_number);
|
||||
open_files.erase(file_number);
|
||||
|
||||
AddStats(file_stats);
|
||||
}
|
||||
@@ -119,14 +122,15 @@ void FileWriter::CloseOldFiles(uint64_t current_image_number) {
|
||||
if (start_message.images_per_file == 0)
|
||||
return;
|
||||
|
||||
for (uint64_t f = 0; f < files.size(); ++f) {
|
||||
if (!files[f] || closed_files.contains(f))
|
||||
continue;
|
||||
|
||||
// Ascending, so the first file that is not old enough ends the walk; CloseFile erases from the set,
|
||||
// hence the iterator is advanced first.
|
||||
for (auto it = open_files.begin(); it != open_files.end();) {
|
||||
const uint64_t f = *it++;
|
||||
const uint64_t file_end_image = (f + 1) * start_message.images_per_file - 1;
|
||||
if (current_image_number > file_end_image + close_file_lag_images) {
|
||||
if (current_image_number > file_end_image + close_file_lag_images)
|
||||
CloseFile(f);
|
||||
}
|
||||
else
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -10,6 +10,7 @@
|
||||
#include "../common/ZMQWrappers.h"
|
||||
#include "HDF5NXmx.h"
|
||||
#include <functional>
|
||||
#include <set>
|
||||
#include <unordered_set>
|
||||
|
||||
class FileWriter {
|
||||
@@ -23,6 +24,10 @@ class FileWriter {
|
||||
std::function<void(const HDF5DataFileStatistics &)> file_closed_callback;
|
||||
|
||||
std::unordered_set<uint64_t> closed_files;
|
||||
// File numbers with a live HDF5DataFile, in order. CloseOldFiles walks this and not `files`, whose
|
||||
// slots stay (empty) for every file of the run - a thousand of them for a long run, visited on every
|
||||
// image.
|
||||
std::set<uint64_t> open_files;
|
||||
constexpr static uint64_t close_file_lag_images = 1000;
|
||||
constexpr static uint64_t default_images_per_file = 1000;
|
||||
void CreateHDF5MasterFile(const StartMessage& msg);
|
||||
|
||||
+15
-5
@@ -178,7 +178,12 @@ void StreamWriter::ProcessDataImage() {
|
||||
if (verbose)
|
||||
logger.Info("Written");
|
||||
NotifyTcpAck(TCPFrameType::DATA, true, false, TCPAckCode::None);
|
||||
} catch (const JFJochException &e) {
|
||||
} catch (const std::exception &e) {
|
||||
// std::exception, not JFJochException, for the reason given in ProcessStartMessage: an
|
||||
// exception that leaves this function skips the error state and the fatal ACK, and the
|
||||
// broker then waits for an END ACK that never comes. Not every thrower below is ours -
|
||||
// std::bad_alloc from a per-image array, filesystem_error from a path check, the JSON
|
||||
// library when a file-done notification carries a string that is not UTF-8.
|
||||
logger.ErrorException(e);
|
||||
logger.Warning("Error writing image - switching to error state");
|
||||
state = StreamWriterState::Error;
|
||||
@@ -206,7 +211,11 @@ void StreamWriter::ProcessEndMessage() {
|
||||
if ((image_puller_output.cbor->end_message->max_image_number == 0) && (max_image_number > 0))
|
||||
image_puller_output.cbor->end_message->max_image_number = max_image_number;
|
||||
file_writer->WriteHDF5(*image_puller_output.cbor->end_message);
|
||||
} catch (const JFJochException &e) {
|
||||
} catch (const std::exception &e) {
|
||||
// std::exception for the same reason as in ProcessDataImage. Here the cost of a leak is the
|
||||
// highest: the END ACK and the notification below are never sent, the state never reaches
|
||||
// Finalized, and on TCP the broker waits for the ACK for as long as the writer keeps
|
||||
// heartbeating - which it does, whatever state it is in.
|
||||
logger.ErrorException(e);
|
||||
logger.Error("Error writing end message - switching to error state");
|
||||
state = StreamWriterState::Error;
|
||||
@@ -333,9 +342,9 @@ StreamWriterOutput StreamWriter::Run() {
|
||||
StreamWriterOutput ret;
|
||||
ret.data_file_stats = hdf5_data_file_statistics;
|
||||
ret.stats = GetStatistics();
|
||||
logger.Info("Write task done. Images = {} Throughput = {:.0f} MB/s Frame rate = {:.0f} Hz max occupation of FIFO {}",
|
||||
logger.Info("Write task done. Images = {} Throughput = {:.0f} MB/s Frame rate = {:.0f} Hz max occupation of FIFO {} ring buffer peak {} MB",
|
||||
ret.stats.processed_images, ret.stats.performance_MBs, ret.stats.performance_Hz,
|
||||
ret.stats.max_puller_fifo_utilization);
|
||||
ret.stats.max_puller_fifo_utilization, image_puller.GetMaxRingBufferUtilization() >> 20);
|
||||
return ret;
|
||||
}
|
||||
|
||||
@@ -411,7 +420,8 @@ void StreamWriter::NotifyReceiverOnFinalizedWrite(const std::string &detector_up
|
||||
s.SendTimeout(std::chrono::seconds(1));
|
||||
s.Connect(detector_update_zmq_addr);
|
||||
s.Send(j.dump());
|
||||
} catch (const JFJochException &e) {
|
||||
} catch (const std::exception &e) {
|
||||
// std::exception: json::dump throws its own type when run_name is not valid UTF-8.
|
||||
logger.ErrorException(e);
|
||||
logger.Error("Error sending notification to detector update socket");
|
||||
}
|
||||
|
||||
@@ -25,7 +25,8 @@ void print_usage() {
|
||||
logger.Info("-d<path>| --root_dir=<path> Root directory for file writing");
|
||||
logger.Info("-r<int> | --zmq_repub_port=<int> ZeroMQ port for PUSH socket to republish images");
|
||||
logger.Info("-f<int> | --zmq_file_port=<int> ZeroMQ port for PUB socket to inform about finalized files");
|
||||
logger.Info("-w<int> | --rcv_watermark=<int> Receiving ZeroMQ socket watermark (default = 100)");
|
||||
logger.Info("-w<int> | --rcv_watermark=<int> Receiving ZeroMQ socket watermark in messages (default = 10)");
|
||||
logger.Info("-m<int> | --rcv_ring_mb=<int> Receive ring buffer in MB; holds images until written and republished (default = 2048)");
|
||||
logger.Info("-W<int> | --repub_watermark=<int> Republish ZeroMQ socket watermark (default = 1000)");
|
||||
logger.Info("-b<int> | --rcv_buffer=<int> Receiving TCP socket buffer in bytes (only with -T; disables autotuning, capped by net.core.rmem_max)");
|
||||
logger.Info("-v | --verbose Verbose output");
|
||||
@@ -74,6 +75,8 @@ int main(int argc, char **argv) {
|
||||
std::optional<int32_t> rcv_watermark;
|
||||
std::optional<int32_t> repub_watermark;
|
||||
std::optional<int32_t> rcv_buffer;
|
||||
// jfjoch_writer always runs with a ring; the puller's own default (no ring) is for the broker.
|
||||
size_t ring_size = static_cast<size_t>(2048) << 20;
|
||||
bool verbose = false;
|
||||
|
||||
static struct option long_options[] = {
|
||||
@@ -84,6 +87,7 @@ int main(int argc, char **argv) {
|
||||
{"rcv_watermark", required_argument, 0, 'w'},
|
||||
{"repub_watermark", required_argument, 0, 'W'},
|
||||
{"rcv_buffer", required_argument, 0, 'b'},
|
||||
{"rcv_ring_mb", required_argument, 0, 'm'},
|
||||
{"verbose", no_argument, 0, 'v'},
|
||||
{"tcp", no_argument, 0, 'T'},
|
||||
{"nproc", required_argument, 0, 'j'},
|
||||
@@ -93,7 +97,7 @@ int main(int argc, char **argv) {
|
||||
int option_index = 0;
|
||||
int opt;
|
||||
bool raw_tcp = false;
|
||||
while ((opt = getopt_long(argc, argv, "?hH:r:f:R:d:W:w:b:vTj:",long_options, &option_index)) != -1 ) {
|
||||
while ((opt = getopt_long(argc, argv, "?hH:r:f:R:d:W:w:b:m:vTj:",long_options, &option_index)) != -1 ) {
|
||||
switch (opt) {
|
||||
case 'j':
|
||||
nforks = atoi(optarg);
|
||||
@@ -129,6 +133,13 @@ int main(int argc, char **argv) {
|
||||
exit(EXIT_FAILURE);
|
||||
}
|
||||
break;
|
||||
case 'm':
|
||||
if (atoi(optarg) <= 0) {
|
||||
std::cerr << "Ring buffer size must be positive number of MB" << std::endl;
|
||||
exit(EXIT_FAILURE);
|
||||
}
|
||||
ring_size = static_cast<size_t>(atoi(optarg)) << 20;
|
||||
break;
|
||||
case 'W':
|
||||
repub_watermark = atoi(optarg);
|
||||
if (repub_watermark.value() <= 0) {
|
||||
@@ -240,9 +251,12 @@ int main(int argc, char **argv) {
|
||||
|
||||
std::unique_ptr<ImagePuller> puller;
|
||||
if (raw_tcp)
|
||||
puller = std::make_unique<TCPImagePuller>(argv[first_argc], rcv_buffer, repub_zmq_addr, repub_watermark);
|
||||
puller = std::make_unique<TCPImagePuller>(argv[first_argc], rcv_buffer, repub_zmq_addr, repub_watermark,
|
||||
ring_size);
|
||||
else
|
||||
puller = std::make_unique<ZMQImagePuller>(argv[first_argc], repub_zmq_addr, rcv_watermark, repub_watermark);
|
||||
puller = std::make_unique<ZMQImagePuller>(argv[first_argc], repub_zmq_addr, rcv_watermark, repub_watermark,
|
||||
ring_size);
|
||||
logger.Info("Receive ring buffer {} MB", puller->GetRingBufferCapacity() >> 20);
|
||||
|
||||
|
||||
writer = new StreamWriter(logger,*puller,file_done_zmq_addr, verbose);
|
||||
|
||||
Reference in New Issue
Block a user