perf: read each package once per track, not once per attempt

Vision took a crop from 5 ms to 25 ms, so re-reading the same package on every
attempt was the measured cause of the live coverage collapse: 789 -> 60 crops/s
at 8 workers, and 12/8/2/6 labels on the same clip at four pacings.

The reading is now cached on the track and later attempts re-run only the local
lane. Measured through the real server on add/image6.jpg:

  saturated pool   15.9 -> 49.4 crops/s (8 workers, 64 crops)
  per crop         142 -> 92 ms on a 300x400 crop
  3 s clip, send   19.7 mean labels (15-24, unstable) -> 24.0 (23-25, stable)
  12 s clip, send  48/131 -> 53/131; result pacing 53/134 -> 50/135

The 12 s numbers are within run-to-run noise; the short clip is where the
pacing sensitivity lived and it is now flat across pacings.

An empty read is not cached - a crop too small or too blurry to read says
nothing about the next one - and the cache is cleared with the rest of the
evidence on a method change or a drift relabel, so a reading never outlives the
product it described. The native crop lane skips OCR on the same rule.

Test: OCR runs once per track while the local lane keeps running. Disabling the
skip makes it fail 2 != 1. Suite 280 -> 281.
This commit is contained in:
tngo 2026-09-28 23:25:09 +09:00
parent fd72251160
commit 5f6de4d48a
3 changed files with 66 additions and 7 deletions

View file

@ -523,7 +523,7 @@ class RecognitionEngine:
return DecisionPolicy().decide(lane.rank(crop), text_rank, text_read=read) return DecisionPolicy().decide(lane.rank(crop), text_rank, text_read=read)
def score_crop_lanes( def score_crop_lanes(
self, crop: Any, method: Optional[str] = None self, crop: Any, method: Optional[str] = None, skip_ocr: bool = False
) -> Tuple[Dict[str, float], List[Tuple[str, float]], str]: ) -> Tuple[Dict[str, float], List[Tuple[str, float]], str]:
"""Both lanes' evidence for one crop: local scores, text ranking, raw read. """Both lanes' evidence for one crop: local scores, text ranking, raw read.
@ -538,6 +538,8 @@ class RecognitionEngine:
`decide_crop`, which fuses a single crop. `decide_crop`, which fuses a single crop.
""" """
scores = self.score_crop(crop, method) scores = self.score_crop(crop, method)
if skip_ocr:
return scores, [], ""
text_lane = self.identity_text_lane() text_lane = self.identity_text_lane()
if text_lane is None: if text_lane is None:
return scores, [], "" return scores, [], ""

View file

@ -147,6 +147,13 @@ class TrackState:
crop_attempts: int = 0 crop_attempts: int = 0
# Which evidence produced the newest observation: frame or crop. # Which evidence produced the newest observation: frame or crop.
last_source: str = SOURCE_FRAME last_source: str = SOURCE_FRAME
# The newest text reading for this object, cached so OCR runs once per
# track rather than once per attempt. Vision takes a crop from 5 ms to
# 25 ms, which is what made streamed coverage pacing-dependent: at 8
# workers throughput fell from 789 to 60 crops/s. Both fields are cleared
# with the rest of the evidence when the object generation changes.
text_rank: List[Tuple[str, float]] = field(default_factory=list)
text_read: str = ""
def area(self) -> int: def area(self) -> int:
if not self.raw_box: if not self.raw_box:
@ -421,6 +428,8 @@ class LiveSession:
track.candidates = [] track.candidates = []
track.score_sums = {} track.score_sums = {}
track.observations = [] track.observations = []
track.text_rank = []
track.text_read = ""
track.attempts = 0 track.attempts = 0
track.in_flight = 0 track.in_flight = 0
track.last_attempt_seq = 0 track.last_attempt_seq = 0
@ -741,6 +750,10 @@ class LiveSession:
track.candidates = [] track.candidates = []
track.score_sums = {} track.score_sums = {}
track.observations = [] track.observations = []
# The reading described the product the box has left, so it goes with
# the label rather than naming whatever moved into the box.
track.text_rank = []
track.text_read = ""
track.attempts = 0 track.attempts = 0
track.in_flight = 0 track.in_flight = 0
track.last_attempt_seq = 0 track.last_attempt_seq = 0
@ -813,8 +826,15 @@ class LiveSession:
# drift check later compares against. # drift check later compares against.
submitted_box = list(track.box) submitted_box = list(track.box)
track_id = track.track_id track_id = track.track_id
# OCR once per object: a track that already has a reading re-runs
# only the local lane, which is the difference between 5 ms and
# 25 ms per crop. A native crop is preferred for the one reading
# because it is the sharpest view the phone will send.
skip_ocr = bool(track.text_read)
try: try:
future = self._executor.submit(self.engine.score_crop_lanes, crop, method) future = self._executor.submit(
self.engine.score_crop_lanes, crop, method, skip_ocr
)
except RuntimeError: # executor already shut down except RuntimeError: # executor already shut down
with self._lock: with self._lock:
track.in_flight = max(0, track.in_flight - 1) track.in_flight = max(0, track.in_flight - 1)
@ -988,8 +1008,14 @@ class LiveSession:
self._release_crop_slot(record) self._release_crop_slot(record)
self._crops_rejected += 1 self._crops_rejected += 1
return _crop_error("CROP_STALE", stale) return _crop_error("CROP_STALE", stale)
# A native crop is the sharpest view the phone sends, so it is the
# one worth reading - but only once per object, like the frame path.
track = self._tracks.get(track_id_out)
skip_ocr = bool(track is not None and track.text_read)
try: try:
future = self._executor.submit(self.engine.score_crop_lanes, crop, method) future = self._executor.submit(
self.engine.score_crop_lanes, crop, method, skip_ocr
)
except RuntimeError: # executor already shut down except RuntimeError: # executor already shut down
self._release_crop_slot(record) self._release_crop_slot(record)
self._crops_rejected += 1 self._crops_rejected += 1
@ -1134,6 +1160,15 @@ class LiveSession:
track.score_n += 1 track.score_n += 1
self._classify_times.append(finished_at - submitted_at) self._classify_times.append(finished_at - submitted_at)
# The reading is cached on the track so later attempts skip OCR.
# An empty read is not cached: a crop that was too small or too
# blurry to read says nothing about the next one.
if text_read:
track.text_rank = list(text_rank or [])
track.text_read = text_read
elif track.text_read:
text_rank, text_read = track.text_rank, track.text_read
if not any(track.observations) and not text_rank: if not any(track.observations) and not text_rank:
if track.attempts >= self.settings.max_attempts: if track.attempts >= self.settings.max_attempts:
track.state = STATE_UNRECOGNIZED track.state = STATE_UNRECOGNIZED

View file

@ -107,15 +107,20 @@ class LocalScaleEngine(RecognitionEngine):
def __init__(self, *args, **kwargs): def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs) super().__init__(*args, **kwargs)
self.lane_answers = [] self.lane_answers = []
self.ocr_calls = 0
@property @property
def identity_scale(self): def identity_scale(self):
return "local" return "local"
def score_crop_lanes(self, crop, method=None): def score_crop_lanes(self, crop, method=None, skip_ocr=False):
if self.lane_answers: answer = self.lane_answers.pop(0) if self.lane_answers else ({}, [], "")
return self.lane_answers.pop(0) if skip_ocr:
return {}, [], "" # The real engine re-runs only the local lane and returns no text;
# the live path reuses the reading it cached on the track.
return answer[0], [], ""
self.ocr_calls += 1
return answer
class FakeInventory: class FakeInventory:
@ -381,6 +386,23 @@ class LocalScaleTests(unittest.TestCase):
) )
engine.shutdown() engine.shutdown()
def test_ocr_runs_once_per_track(self):
"""Vision on every attempt is what made coverage pacing-dependent.
A crop costs 5 ms without OCR and 25 ms with it, so re-reading the same
package on every attempt cut throughput from 789 to 60 crops/s at 8
workers. The reading is cached on the track instead, and the local lane
keeps running on every attempt.
"""
session, engine, _, _ = make_session(engine_cls=LocalScaleEngine)
# Below the text accept line, so the track stays undecided and retries:
# the retry is what would re-run OCR without the cache.
engine.lane_answers = [({}, [("sku_a", 0.3)], "コクうま 65%")] * 6
for frame_id in range(1, 14):
process(session, frame_id)
self.assertEqual(engine.ocr_calls, 1, "OCR must run once per track")
engine.shutdown()
class TagPassIntegrationTests(unittest.TestCase): class TagPassIntegrationTests(unittest.TestCase):
"""The tag pass must reach the live result without blocking frames.""" """The tag pass must reach the live result without blocking frames."""