Repository navigation
Pair a join's sources by capture time, and a test for the livelock it fixes - #189
leandrobmarinho wants to merge 3 commits into
Conversation
38a03ff to
9dbb868
Compare
|
Two updates. Rebased onto main, the only conflict was RELEASE.md where main gained the v1.5.2 heading. Both sides kept.
Evidence from the real pipeline, not the bench. The two camera Wingstop pipeline is now running this change with For contrast, on the same pipeline earlier the same day: the instance that froze ran 53 minutes 59 seconds clean before its first barrier warning, then logged 26229 of them and stalled four times in two hours, once for 17 minutes with the pod still reporting healthy. Worth stating plainly because it bears on the default: a one hour clean run proves nothing here, since the pipeline that froze was clean for 54 minutes first. The deterministic test is the evidence, the soak is only corroboration. |
|
Added the version bump and the release note, so this merges as a release rather than leaving the fix in Unreleased. VERSION is now v1.5.3 and RELEASE.md has a Fixed entry for the livelock alongside the Added entry for the flag. The reason it matters here: a deployed pipeline only picks this up once the filters are rebuilt against a released openfilter. Right now the correction exists only in a filter-rt-detr image built by hand against this branch, so any rebuild from main loses it. |
A receiver joining several sources keeps one `min_recv_id` barrier for all of
them and decides that two frames belong together when their message ids match.
Message ids are per-source counters. Two cameras nominally at the same frame
rate do not sample at exactly the same rate, so the counters drift apart and
keep drifting, and id equality stops implying time equality.
The counter gap is not itself a time gap. A source with a lower id has taken
fewer samples, which on its own says nothing about when its latest frame was
taken. What the drift costs is the only key the join has, plus a second-order
effect that does cost real time: the shared barrier is dragged forward by
whichever source leads, the lagging source's sender fast-forwards its own
counter to match ("downstream requested newer message id"), and the leading
source spends the wait publishing into a subscriber queue nobody drains.
ZMQ_SUB_HWM bounds how stale that gets, it does not stop it happening.
`sources_pair_by_ts` gives every source its own barrier, drains every source on
every poll, and holds the newest complete frame set from each in one slot. A
batch is the newest frame every source has. Depth one is not a shortcut:
timestamps rise within a source, so its newest held frame is always the closest
to the newest frame seen anywhere.
Capture time rides in the envelope the receiver already parses.
MQ.frames2topicmsgs lifts meta['ts'] into the per-message `xtra` dict, which
ZMQSender.send already copies into the envelope, so the pairing costs no extra
deserialization. Only image frames carry it, so a frame with no image keeps the
wire shape it had and an older receiver reads it unchanged.
Off by default and resolved to off below two synchronized sources, so a
single-source hop runs the original code path. Rejected together with
`sources_balance`, where the sources are one stream split across workers and
exactly one of them delivers per round. A frame with no timestamp pairs by
arrival time and warns once an hour, rather than stalling the join.
Measured on two file sources at 29.6 and 16.5 fps joined by one receiver, 60 s
per config, as the wall-clock distance between the two frames of a delivered
batch:
default 273 ms p50, 287 ms p95, over 4191 batches
sources_low_latency 3.2 ms p50, 3.5 ms p95, over 15089 batches
sources_pair_by_ts 1.0 ms p50, 2.8 ms p95, over 15044 batches
The flag reaches low_latency's skew while keeping the prefetch, so it does not
pay low_latency's throughput cost on the single-source hops of the same
pipeline. Per-source id gap over the same 60 s goes from 254 p50 under the
shared barrier, where the fast-forward keeps resyncing it, to 870 p50 under the
flag, where the counters are left alone. That gap is bookkeeping, not lag.
New tests drive two senders at 40 ms and 46 ms with the capture timestamp taken
before the send, which is the shape a blocking camera read produces. Pairing
skew p95 over 12 s: 519 ms with the default join, 45 ms with the flag, the
latter being half the slower source's sampling period, which is the floor for
"newest frame each source has". The single-source case is test_send_recv run
twice, once with the flag on, asserting the same results.
Signed-off-by: Leandro Bezerra Marinho <lmarinho@plainsight.ai>
A receiver joining two topic-mapped sources stops delivering for good when one source skips a single message id, which is what the production freeze looked like: last batch at id 28459, then 17 minutes of "received newer message id N than expected 28460" at 5 Hz with both cameras still publishing and process() never called. Three conditions have to hold together, and the fixture sets all three: - The sources are topic mapped. A mapped source gets recvd_new as a dict, so Sender.new_recv() returns a dict of Nones instead of None and process_msg never takes its "recvd is None" first-message branch. Every message whose id is not exactly the barrier then invalidates the other source's held frame. An unmapped fixture of the same shape does not reproduce this at all, which is why earlier attempts came back clean. - The two sources' published ids differ by one. The sender skips ids on its own through the "downstream requested newer message id" path, which moves min_send_id without publishing that id, and a frame dropped at ZMQ_PUB_HWM leaves the same hole in the sequence the receiver reads. - The receiver calls recv() with a timeout no longer than ZMQ_POLL_TIMEOUT, which is what Filter does. Such a call issues exactly one request, before any message of the round is read, so it carries a prev_id computed from the stale self.prev_id and the sender-side fast forward never fires to close the gap. test_one_skipped_id_does_not_stop_the_join fails on the default join, on main and on this branch, with a 5.2 s gap and 5 batches over a 6 s run, 20 runs out of 20. test_pair_by_ts_survives_one_skipped_id passes with the flag on, 20 out of 20, so the per-source barriers do fix it but only for receivers that set the flag. Signed-off-by: Leandro Bezerra Marinho <lmarinho@plainsight.ai>
Cuts the per-source join barrier and the livelock regression test into a version. The fix only reaches a deployed pipeline once the filters are rebuilt against a released openfilter, so leaving it in Unreleased would have made the merge a half step. Signed-off-by: Leandro Bezerra Marinho <lmarinho@plainsight.ai>
8184ec6 to
4a3ac47
Compare
lucasmundim
left a comment
lucasmundim
left a comment
There was a problem hiding this comment.
Thanks for the deterministic livelock reproduction. The shared-barrier analysis and the per-source barrier design look right to me. A few things before this can merge:
Blocking
- Under
pair_by_ts, the batch id goes backwards after a source restarts gracefully, and downstream output is then dropped silently (see the inline comment onzeromq.py). - The changelog lists the livelock under Fixed, but the default join still livelocks (see the inline comment on
RELEASE.md). You asked for a decision on the default. Either way, the entry should match what actually ships.
Non-blocking
- Test coverage misses the production path. The new tests put
{'ts': ...}straight intoxtra, so nothing coversMQ._frame_ts, theframes2topicmsgslift, or the wiring fromFILTER_SOURCES_PAIR_BY_TS/sources_ts_tolerancethrough toZMQReceiver. A small unit test of_frame_tsand the lift (dict and non-dictmeta, bool and strts, a frame with no image), plus one config round-trip, would cover it. - Releases have been cut in separate PRs (openfilter#173, openfilter#176, openfilter#186). I see why you bundled this one, but it means the release cannot be held back independently if the default-flip discussion takes longer.
- The PR description names commits
2e904c4/38a03ff, but after the sign-off rewrite the branch has9b45c4f/8c747cf.
| # counter and they are not meant to agree. The highest one is the only choice that stays | ||
| # monotonic, which is all `prev_id` and the handoff to the sender half (mq_msgid_sync) need | ||
| # from it. Downstream of a timestamp join an id no longer identifies a moment, only an order. | ||
| min_recv_id = max(s.min_recv_id for s in sync_sendervs) |
There was a problem hiding this comment.
This id is not monotonic when a source restarts gracefully. On MSG_ID_CLOSE, recv_once sets sender.min_recv_id = MSG_ID_INITIAL (L1173). Without the flag that only affected ephemeral senders. Under pair_ts it also resets a synchronized sender's barrier. Its next request then carries mid = -1, so the restarted sender never fast-forwards and starts again at 0.
Reproduced with two senders and pair_by_ts=True, source B ahead in id space, then B.destroy() and a new ZMQSender on the same address. Delivered state.msg_id went 5054, 5055, 5056 → 62, 63, 64. The same run with the flag off went 5050 → 5051, because the shared-barrier request makes the new sender fast-forward.
With MQ_MSGID_SYNC on (the default), that state goes to ZMQSender.send, which returns early without publishing when state.msg_id < self.min_send_id (L477). The filter keeps calling process(), and its output is dropped silently until the max id passes the old value. Per-source counters drift freely under this flag, so that can take a long time, and much longer if both upstreams restart.
Suggestions: keep min_recv_id across CLOSE for synchronized senders under pair_ts, so the restarted sender is fast-forwarded the same way the default path does it. Also clamp the batch id to max(self.prev_id + 1, ...). Clearing sender.ready / sender.recvd on CLOSE would also avoid pairing a frame from before the restart. A restart case in test_the_state_msg_id_stays_monotonic would cover it.
|
|
||
| ### Fixed | ||
|
|
||
| - **A join could stop delivering for good after one of its sources skipped a single message id.** The |
There was a problem hiding this comment.
This reads as if the livelock is fixed in v1.5.3. With sources_pair_by_ts off by default, every deployed join keeps the exact behavior that test_one_skipped_id_does_not_stop_the_join still marks as an expected failure. If the default stays off, please say plainly that the default join is still affected and that sources_pair_by_ts is the opt-in workaround. Otherwise operators will upgrade, see "Fixed", and leave the flag off.
| poller.unregister(sender.sub) | ||
|
|
||
| elif sender_eph: | ||
| pass # ephemeral senders are not paired and keep the behavior they have without the flag |
There was a problem hiding this comment.
The behavior does change. Without the flag, a complete sender is unregistered (poller.unregister(sender.sub)), so nothing newer gets merged in. Here an ephemeral sender stays registered. For a multi-topic ephemeral source, the first topic of its next frame goes through process_msg → new_recv(msg, ...), which replaces the complete set with a partial one. got == 'some' then holds back the whole batch until that frame completes. Either unregister it as before, or fix the comment.
| ZMQ_SUB_HWM = int(os.getenv('ZMQ_SUB_HWM') or 4 * 2) # messages a subscriber will queue on its side, 4 topics x 2 frames, one being worked on and one next, and it has to stay above the topic fan-out of any source because a sender publishes one message per topic per frame. Unit is whole multipart messages, same as ZMQ_PUB_HWM above. On a receiver that joins two sources this is the only cap that applies to the source nobody is draining: the receiver holds one barrier for all of them and the only discard path is sender-side and can only fire on the source BEHIND that barrier, so the source AHEAD publishes into a socket nobody reads and the frames it eventually delivers are as old as this queue is deep. libzmq defaults it to 1000, which at 5 fps is near a minute of stale frames. This caps the subscriber's share only: how stale an undrained source can get is this queue plus ZMQ_PUB_HWM plus whatever the kernel socket buffer holds, and that last one is bounded in bytes rather than in messages, so it hides more the smaller the frames are (SNDHWM 20 with RCVHWM 8, messages delivered before the first drop: 8 over IPC at 64 KB, 14 over TCP at 64 KB, 60 over TCP at 64 B, 120 over IPC at 64 B) | ||
| ZMQ_LOW_LATENCY = bool(json_getval((os.getenv('ZMQ_LOW_LATENCY') or 'false').lower())) | ||
| ZMQ_PAIR_BY_TS = bool(json_getval((os.getenv('ZMQ_PAIR_BY_TS') or 'false').lower())) # pair multi-source frames by capture timestamp instead of by message id | ||
| ZMQ_PAIR_TS_TOLERANCE = float(os.getenv('ZMQ_PAIR_TS_TOLERANCE') or 0) # seconds, 0 for no gate |
There was a problem hiding this comment.
A negative value (here, or through sources_ts_tolerance) is truthy, so pair_skew_ok evaluates max - min <= negative. That is never true, and the join never delivers. Please reject values below zero at construction.
| # Expected to fail while `sources_pair_by_ts` defaults off. That is the defect this class | ||
| # documents, not a flaky test. The day the default flips, unittest reports this as an | ||
| # unexpected success, which is the signal to delete the decorator. | ||
| @unittest.expectedFailure |
There was a problem hiding this comment.
expectedFailure counts any failure: the sends precondition assertion, an exception in setup, or a socket error. So this test does not prove the livelock still reproduces. Consider asserting the stall directly (assertGreater(max_gap, MAX_GAP) after the sends check), as test_pairing_skew_is_large_without_it does. That keeps it a real tripwire.
|
|
||
| Measured on two file sources at 29.6 and 16.5 fps joined by one receiver, 60 s per config, as the | ||
| wall-clock distance between the two frames of a delivered batch: 273 ms p50 / 287 ms p95 over | ||
| 4123 batches by default, against the numbers in the table of the PR with the flag on. Single |
There was a problem hiding this comment.
Nit: the PR and commit message say 4191 batches. Also, RELEASE.md readers don't have the PR table, so please put the flag-on numbers inline (1.0 ms p50 / 2.8 ms p95 over 15044 batches).
The defect
A receiver joining two or more synchronized sources keeps ONE shared barrier,
min_recv_id, reset atzeromq.py:950fromself.prev_id + 1. When any source delivers above the barrier, lines 1092-1098 destroy every other source's already-received frame withs.new_recv()and line 1100 moves the barrier.self.prev_idadvances only when a batch is delivered.So one skipped message id on one source is enough to stop the join forever: source A's frame takes the barrier, source B's frame one id higher destroys it, the round ends with nothing to deliver, and the next round restarts from the same stale barrier. No
process()call, no output, and the container stays healthy.This is not theoretical. A two camera pipeline froze on it four times in two hours, once for 17 minutes, with the pod reporting 9 of 9 Running and zero restarts the whole time. It logged 26229
received newer message id N than expected 28460warnings while both cameras kept publishing and the receiver's expectation never moved.Three conditions, all present in any two camera join
addr;main>camA), which is what two cameras both publishingmainmust do. A mapped source getsrecvd_newas a dict, soSender.new_recv()returns a dict ofNones rather thanNone(zeromq.py:790), soprocess_msgnever takes its first-message branch at 1065-1067 and every off-barrier message invalidates the other source instead.min_send_id = prev_id + 1without publishing that id, and a frame dropped atZMQ_PUB_HWMleaves the same hole.recv()with a timeout no longer thanZMQ_POLL_TIMEOUT, which is exactly whatFilter.loop_oncedoes. Such a call issues onerequest()before any message of the round is read, so it carriesmin_recv_id - 1from the staleprev_id, which is what stops the sender side fast forward from ever rescuing the join again.What is in this PR
Two commits.
2e904c4gives each source its own barrier and pairs frames on capture time, behindsources_pair_by_ts/ZMQ_PAIR_BY_TS, default off.38a03ffaddsTestJoinBarrierLivelock, which reproduces the freeze deterministically.Capture time rides in the envelope the receiver already parses:
MQ.frames2topicmsgsliftsmeta['ts']into the per-messagextradict. Only image frames carry it, so a frame with no image keeps its wire shape. A source with nometa['ts']falls back to arrival time with an hourly warning rather than stalling.Evidence
Livelock test, 20 runs each:
main, default joinpair_by_tsoffpair_by_tsonZero variance. The failing runs stall 5.2 to 5.4 s out of a 6 s run with both sources publishing 30 frames each.
Two sources on a bench, 60 s per config, measuring the wall clock distance between the two frames of a delivered batch:
sources_low_latencysources_pair_by_tsSingle source, 60 s: 15629 frames at 7.9 ms p50 on
main, 15650 at 7.9 ms with the flag on. The flag resolves toFalseat construction below two synchronized sources, so that path runs unchanged. Flag off on this branch reproduces unpatchedmainmeasured in the same session, 4191 against 4123 batches and 272.9 against 273.0 ms.tests/test_zeromq.py: 64 passed, 1 failed, the intended one.test_filter_env_vars.pyfails 24 tests identically on cleanmain, pre-existing.The decision this PR needs
pair_by_tsdefaults to false, so merging as it stands leaves every deployed join exposed. Two things we measured that bear on the default:main,sources_low_latencyalone,sources_timeoutalone, and both together all fail the livelock test 0/20.sources_low_latencyis inert here because the prefetch it removes sits inside the delivery branch, which the frozen state never reaches.sources_timeoutis actively worse than the hang. On expiry the filter emitsframes={}and keeps sending, so a health check watching the sender stays green while nothing flows. Over 30 s it produced 5 real batches, 5 empty ticks and 29.2 s with no frame delivered.If the default should flip, it needs to resolve to
Falsefor balanced receivers rather than raise, sincezeromq.py:953currently raisesValueErrorforpair_by_ts and balanceand auto-enabling would turn working balanced pipelines into startup crashes. Happy to make that change here if that is the direction.What is not proven
The timestamp path assumes
meta['ts']comes from one clock. Bothvideo_inrun in the same pod in the case that motivated this, so it holds there. Across nodes the pairing error silently becomes the NTP offset, and nothing detects or bounds that.sources_ts_toleranceis implemented and reasoned about but has no test. No metric is exported, so an operator cannot see the skew without a custom filter.Pairing "the newest frame each source holds" leaves an irreducible skew of about half the slower source's sampling period, roughly 100 ms at 5 fps. Beating that needs buffering, which trades latency back for alignment.
Need help on this PR? Tag
@codesmith-botwith what you need.