nx1.info | Fixing Threading Bugs (Kafka, Qt, and Concurrency)

Action Prompts [^]

On the smartbench we have the ability to indicate to the user that a specific action should occur. We call this an "action_prompt" and currently it supports two actions: 1. Take a photo 2. Take a spectrometer scan The way that this information is transmitted to the user is that the relevant section on the GUI will blink. The action prompts are triggered by logic in the decision tree. The decision tree works by iterating over a list of "rules" sorted by priority known as a "rule set". Each rule defines a final outcome (such as a recycling grade) and one or more conditions. If all conditions for a rule evaluate to true, that rule is selected. A rule may also have an optional "requires" field that specifies which data must be captured for the rule to be finalised and evaluate as true. The requires field can be one of four possible states:

requires: []                     # No requirements are specified for the rule
requires: ['photos']             # A deliberate (macro) photo taken by the user is required
requires: ['material']           # A deliberate spectrometer scan is required
requires: ['photos', 'material'] # Both a photo and a spectrometer scan are required

When the decision tree encounters a rule that has a non-empty "requires" field, it checks the incoming data for the boolean flags "has_photo" and "has_material". If any are missing, they are stored as "missing_requires".

Deferred Sensor Field Mapping [^]

The data fields that are dependent on either photos or spectrometer scan (material) are mapped in the following way:

REQUIREMENT_FIELDS = {
    'material': {'material', 'material_percent', 'structure'},
    'photo'   : {'colour'},
}

Any conditions that check these fields are deferred (skipped) while the sensor measurement is missing. When a rule is evaluated, if all the non-deferred conditions are true but the rule has missing requirements, it will treat that rule as being "deferred" or "pending" and store the appropriate action prompt. NOTE: When both a photo and a spectrometer scan are required, the tree prompts for them sequentially in list order: it first requests the photo, and once that is captured, it advances to prompt for the spectrometer scan. Only after both requirements are satisfied are all deferred conditions evaluated to finalize the rule. The decision tree will then continue iterating through the rule set and attach the action prompt to whichever rule is eventually returned.

The Observed Bug [^]

During the operation of the smartbench we observed the following issue: A pair of jeans with holes and metal hardware was put through the smartbench. Both of these disruptors were identified by live detection and correctly stored:

ml_live_detection: ["metal", "damage"]

These disruptors correctly appeared on the GUI. The decision tree evaluated the presence of these disruptors and hit the following rule:

{"name"       : "Cotton Recycle",
 "conditions" : [{"field": "material",         "oper": "in", "value": ["1"]},
                 {"field": "material_percent", "oper": "==", "value": 100},
                 {"field": "disruptors",       "oper": "in", "value": ["damage", "stain"]}],
 "requires"   : ["material"]}

Because the disruptors condition was true but a spectrometer scan had not yet been taken, the tree returned the action prompt "material", causing the spectrometer section of the GUI to blink. Next, the garment was removed from the scale. This auto-submitted the report, created a fresh SampleReport for the next garment, and reset the GUI blocks. However, intermittently, the spectrometer section on the GUI would immediately start blinking again. The disruptors section on the GUI showed no disruptors ("Disruptors (0):"), yet looking under the hood at the logs during the transition between the two sample IDs (6caHeCQoZX -> TKUEtYsWhE), we can see the exact moment the disruptor leaked:

Reproduction Log Analysis [^]

# 1. Old sample (6caHeCQoZX) completes with live disruptors:
2026-10-29 16:32:46 [INFO] [KafkaMessageHandler] {'sample_id': '6caHeCQoZX', ..., 'ml_live_detection': ['metal', 'damage']}

# 2. Garment removed -> New sample created clean:
2026-10-29 16:32:51 [INFO] [KafkaMessageHandler] {'sample_id': 'TKUEtYsWhE', ..., 'ml_live_detection': []}
2026-10-29 16:32:51 [INFO] [KafkaMessageHandler] Received outcome message: {'name': 'Waste', ..., 'action_prompt': None}

# 3. Same second (16:32:51), 'damage' suddenly leaks into the new sample:
2026-10-29 16:32:51 [INFO] [KafkaMessageHandler] {'sample_id': 'TKUEtYsWhE', ..., 'ml_live_detection': ['damage']}

# 4. Decision tree re-evaluates and immediately fires the action prompt:
2026-10-29 16:32:51 [INFO] [KafkaMessageHandler] Received outcome message: {'name': 'Waste', ..., 'action_prompt': 'material'}

Because the leaked disruptor satisfied the condition on the "Cotton Recycle" rule, the rule was once again treated as deferred, causing the spectrometer action prompt to fire on a clean, empty bench while the GUI showed "Disruptors (0):".

Why This Happened [^]

This bug was caused by a race condition between two separate threads communicating across the Qt event loop, combined with an asymmetrical state-clearing mechanism. To understand it, we need to look at three moving parts: 1. DeepStream Video Thread: A native background GStreamer pipeline thread that runs object detection (YOLO) at 20–30 FPS and maintains a 30-detection rolling buffer of detected disruptors. When a disruptor appears in 7 or more detections, it emits a Qt signal: "disruptor_keys_signal". 2. Qt Main GUI Thread: Runs the user interface and event loop. It receives "disruptor_keys_signal" via a "Qt.QueuedConnection", renders disruptor widgets on the screen, and broadcasts newly seen disruptor keys to Kafka on the "ml_live_detection" topic. 3. DataManager (Kafka Service): An external process that tracks the current SampleReport. Whenever it receives a message on "ml_live_detection", it updates the sample's disruptor list and re-evaluates the decision tree.

How the Rolling Buffer Actually Works (Detections vs Frames) [^]

A critical detail about DeepStream's implementation: "self.detections = deque(maxlen=30)" is a rolling buffer of individual detections, not video frames! Let's look at the actual code in smartbench/app_gui/deepstream_processor.py:

with self.detections_lock:
    for det_name, _ in dets_data:
        self.detections.append(det_name)
    type_counts = Counter(self.detections)
    new_keys = {k for k, v in type_counts.items() if v >= self.disruptor_threshold}
    self.disruptor_keys_signal.emit(new_keys, self._reset_epoch)

Notice what happens on each video frame: 1. Multi-detection frames: If a single frame contains three detections (e.g. two stains and a hole), the loop appends three items to the deque in that single frame. 2. Empty frames (Empty Bench): If a frame contains zero detections ("dets_data = []"), the "for" loop does not execute. Nothing is appended, and crucially, NOTHING is evicted! Empty frames do NOT age out or dilute old detections. For example:

# Frame 1: Three detections
Frame 1: ["stain", "stain", "damage"] --> buffer: ["stain", "stain", "damage"]          (len = 3)

# Frame 2: Empty frame (no objects detected in view)
Frame 2: []                           --> buffer: ["stain", "stain", "damage"]          (len = 3, untouched!)

# Frame 3: One detection
Frame 3: ["damage"]                   --> buffer: ["stain", "stain", "damage", "damage"]  (len = 4)

This means a detection stays in the buffer indefinitely until one of two things happens: - NEW detections arrive and push it out past the 30-item capacity. - An explicit sample reset runs: "self.detections.clear()".

The Event-Loop Race Condition Sequence [^]

Here is the exact chain of events that produced the bug: Step 1: The old garment triggers a detection While the operator is lifting the damaged jeans off the scale, DeepStream processes the final frames where the garment is still in camera view and emits:

self.disruptor_keys_signal.emit({'damage'})

Because this is a cross-thread signal ("Qt.QueuedConnection"), Qt places this event at the back of the Main GUI thread's FIFO event queue. Step 2: The weight removal triggers a reset At the exact same moment, the weighing scale detects weight removal and broadcasts to Kafka. DataManager auto-submits the old garment and sends a new "sample_id". The GUI thread handles the new "sample_id" immediately and calls "reset_blocks()":

self.results_block.reset_block()         # Clears GUI rows & resets last_sent_disruptor_types = set()
self.deepstream.reset_garment_lifetime() # Clears DeepStream's rolling window deque

Step 3: The stale queued signal fires on the new sample Crucially, "reset_blocks()" runs synchronously on the GUI thread. When it finishes, the GUI thread returns to its event queue and finally pops the pending signal from Step 1:

on_disruptor_keys({'damage'})

Because "reset_block()" had just wiped "last_sent_disruptor_types" back to an empty set, the GUI thought this was a brand-new detection for the *new* garment! It immediately published to Kafka:

producer.send('ml_live_detection', {'live_detection': ['damage']})

DataManager received this and permanently attached "damage" to the new sample. Step 4: The GUI wipes the disruptor from sight A fraction of a second later, the camera is pointing at an empty pan. DeepStream's sliding window now has zero detections and emits:

self.disruptor_keys_signal.emit(set())

When the GUI received this empty set, its handler previously executed:

if not disruptor_types_set:
    self.clear_disruptors()
    return

This wiped all disruptor rows from the screen and updated the title to "Disruptors (0):". However, it did NOT unsend or clear the disruptor that was already published to Kafka! Note on why only "damage" leaked (and not "metal"): Although the previous sample (6caHeCQoZX) had both "metal" and "damage" listed under "ml_live_detection", metal detection does not come from live video. It originates from a separate hardware metal detector sensor, but is published over the "ml_live_detection" Kafka topic and treated as a disruptor because it simplifies the decision tree logic. Because the hardware metal detector state resets synchronously on sample rollover, only the asynchronous DeepStream video detection ("damage") remained in-flight in the event queue when the reset occurred. The Result: The GUI showed zero disruptors, but DataManager still held "damage" in the sample report. The decision tree evaluated the sample, matched the deferred "Cotton Recycle" rule, and sent back "action_prompt: material". The operator was left looking at an empty bench with no disruptors displayed, wondering why on earth the spectrometer prompt was blinking.

The Fix [^]

To fix this bug properly, we had to address two distinct problems: 1. Prevent stale in-flight messages from leaking across the thread boundary on a reset. 2. Synchronize the GUI display lifecycle with the Kafka / SampleReport lifecycle.

Part 1: The In-Tray Analogy and the "Epoch" Guard

To understand why the message leaked, think of the two threads as two people working in an office: - Worker A (DeepStream Thread): Sits at a window monitoring the camera feed. - Worker B (Main GUI Thread): Sits at a desk managing the customer file and whiteboard. Because Worker B is often busy drawing buttons or handling network packets, Worker A cannot interrupt them directly. Instead, Worker A writes notes and drops them into an in-tray on Worker B's desk:

Worker A (DeepStream) drops note: ["Damage detected!"] --> IN-TRAY (Qt FIFO Queue) --> Worker B (Main GUI)

This in-tray is a FIFO queue (First In, First Out)—exactly how Qt's "QueuedConnection" works. Here is where the concurrency bug occurred: 1. Worker A notices damage on Garment #1 and drops a note into the in-tray: [ "Damage detected!" ] 2. At that exact millisecond, the weighing scale bell rings: Garment #1 is removed! 3. Worker B immediately jumps up, wipes the whiteboard clean, and opens a blank file for Garment #2 ("reset_blocks()"). 4. Worker B sits back down, looks at their in-tray, and picks up the note: [ "Damage detected!" ] 5. Worker B has no way of knowing *when* that note was written! They assume it belongs to the current garment, write "Damage" onto Garment #2's file, and send it to Kafka. This is a classic concurrency problem: a message was completely valid when it was created, but by the time the receiving thread got around to reading it, the state of the world had changed.

The Solution: Generation / Epoch Versioning

To fix this, we introduce an "Epoch" counter (like a batch number or timestamp). Every time a garment arrives, we increment the epoch: - Garment #1 is Epoch 0 - Garment #2 is Epoch 1 Worker A now stamps every note with the current epoch number before dropping it in the tray: [ "Damage detected!", epoch=0 ] When Worker B picks up the note, they compare the note's epoch with the current garment's epoch: - Is note.epoch (0) == current_epoch (1)? - No! This note belongs to the past. Worker B throws it straight into the wastebasket.

Code Implementation (DeepStream & Main GUI) [^]

Here are the actual code changes: In smartbench/app_gui/deepstream_processor.py:

# 1. Initialize the epoch counter in __init__:
self._reset_epoch = 0

# 2. Advance the epoch inside the thread lock whenever a sample resets:
def reset_garment_lifetime(self):
    with self.detections_lock:
        self._reset_epoch += 1
        self.detections.clear()

# 3. Stamp every emitted signal with the current epoch:
def count_dets_class(self, dets_data):
    with self.detections_lock:
        # ... count detections over rolling window ...
        self.disruptor_keys_signal.emit(new_keys, self._reset_epoch)

In smartbench/app_gui/main_gui.py:

# 4. In the GUI slot, discard any signal whose epoch doesn't match:
@QtCore.Slot(object, int)
def on_disruptor_keys(self, new_keys, epoch=None):
    if epoch is not None and self.deepstream and epoch != self.deepstream._reset_epoch:
        return  # Stale signal from an older garment! Discard immediately!

    # Process valid signal...

Part 2: Making Live Disruptors "Sticky" [^]

The second part of the bug was that the GUI erased disruptors from the screen while DataManager remembered them forever. Why did the GUI erase them? Whenever a sample was reset, "reset_garment_lifetime()" cleared the deque ("self.detections.clear()"). On the subsequent frames, the buffer had fewer than 7 detections of any class, so "new_keys" evaluated to "set()". DeepStream then emitted "disruptor_keys_signal.emit(set())". Furthermore, during the lifetime of a garment, if other detections later arrived and displaced a disruptor below the 7-count threshold, "new_keys" would no longer include it. The old GUI code did this:

# OLD CODE in smartbench/app_gui/blocks/results_block.py:
if not disruptor_types_set:
    self.clear_disruptors()  # <-- Wiped the GUI back to 0!
    return

This was a fundamental design flaw: a tear does not magically vanish from a garment just because it rotated away from the camera lens! Kafka had already recorded the tear, but the GUI was now lying to the operator by displaying "Disruptors (0):". The fix: Live disruptors must be "sticky" for the lifetime of that garment:

# NEW CODE in smartbench/app_gui/blocks/results_block.py:
# Keep existing disruptors sticky for the garment's lifetime when rolling window is empty
if not disruptor_types_set:
    return  # Do NOT clear the GUI!

Now, once a disruptor is identified on a garment, it remains visible on screen until the entire garment is finished and "reset_blocks()" clears the sample.

Part 3: Prioritizing Critical Disruptors on the GUI [^]

Because of physical screen space, the GUI layout can only display a maximum of 3 disruptors. Previously, if 3 harmless/neutral disruptors were detected first (e.g. "label", "logo", "button"), and a critical defect like "damage" or "stain" was detected later, the layout was already full and ignored the critical defect. We updated the display logic to sort disruptors by severity: - Tier 1 (Critical): "damage", "stain", "metal" - Tier 2 (Neutral): "label", "logo", "button", "zip", etc. Even if 3 neutral disruptors are already showing, the moment "damage" is detected, it automatically displaces a neutral disruptor to ensure the operator can always see the exact defect driving the decision tree's action prompt.

---