mattcasters opened a new pull request, #7854: URL: https://github.com/apache/hop/pull/7854
## Summary Addresses [#7752](https://github.com/apache/hop/issues/7752) for the **classic local multi-threaded pipeline engine** only (Beam/Spark are out of scope: they materialize between stages). The issue proposed extra loader threads + `CountDownLatch` in Stream Lookup. That is transform-specific, risks loading whole streams into memory, and does not help Merge Join / Append / similar multi-input cases. This PR instead: 1. **Detects** bounded-buffer deadlock risk (split–rejoin with a shared ancestor). 2. **Mitigates** by spilling only the **minimal set of recommended hops** to disk when the in-memory rowset would block. Goal: prefer **correctness and progress** over silent hangs for data engineers, with clear logging when risk or spill engages. ## Problem (what we are actually fixing) Pipeline hops are already a DAG (`hasLoop` / GUI loop checks). Stream Lookup / Merge Join hangs are **not hop-graph cycles**. They are **bounded-buffer wait-for cycles**: - Rowsets (`BlockingRowSet`) have fixed capacity (rowset size). - A shared source that **copies** to multiple paths blocks if **any** output buffer is full. - Transforms that drain one input fully before another (e.g. Stream Lookup info-first) leave the other path full → source never finishes → never `setDone()` → hang. Raising rowset size, redesigning streams, or inserting Blocking are valid workarounds; they are not a generic engine fix. ## Decisions and rationale ### Rejected: Stream Lookup-only loader threads - Not generic (Merge Join and friends still hang). - Eager full-stream load increases OOM risk (why rowsets are bounded in the first place). - Extra transform lifecycle / concurrency complexity. ### Rejected: spill “info hops only” Info-only spill **does not fix** classic Stream Lookup: the **main** hop is the one left unread while the lookup cache loads, so the shared source still blocks on main. Merge Join happens to list both inputs as INFO, so info-only would accidentally help there — wrong rule for the headline case. ### Chosen: detect split–rejoin, then spill recommended inbound hops **Detection** (`PipelineBufferDeadlockAnalyzer`): - Multi-input transform `T` (≥2 predecessors via enabled hops, including info). - At least two predecessors share a **common ancestor** (walk `findPreviousTransforms`). - Risk record includes reconvergence, ancestor, inbound preds, and recommended spill hops. **Spill hop set (v1, safe minimal):** all inbound hops `predecessor → T` involved in the risk — typically **two hops per site**, not the whole pipeline. That includes main+info for Stream Lookup without encoding every transform’s read order. **Not in v1:** mid-path spill (e.g. before Sort on a long branch); auto-inserting Blocking transforms; unbounded pure-memory queues. ### Spilling rowset (`SpillingRowSet`) - Same contract as `BlockingRowSet` while under capacity. - When full (or FIFO requires append behind existing spill): serialize **row data only** via `IRowMeta.writeData` (not metadata per row; same pattern as Sort / Blocking / Join Rows). - `putRow` succeeds without waiting on the consumer — that is what breaks the wait-for cycle. - Temp files via HopVfs under configurable directory (default system temp). - Cleanup on `clear()` / done path. ### Critical performance fix: `size()` and BaseTransform sleeps First mitigation builds were pathologically slow (~38k rows/s, near-zero CPU/disk). Cause: - `BaseTransform` does `Thread.sleep(0, 1)` when `size() >= 99%` of rowset size (and a low-water sleep when nearly empty). - On many Linux/JVMs that “1 ns” sleep rounds to **~1 ms**. - While spilling, memory stays full, so every put hit that sleep. **Fix:** `SpillingRowSet.size()` is a **flow-control signal**: with pending work it reports a mid-level occupancy so producers/consumers are not artificially throttled; idle still reports `0`. After the fix, ~**825k rows/s** on a large test pipeline. ### Defaults (local run configuration GUI) | Option | Default | Why | | --- | --- | --- | | Detect buffer deadlocks | **On** | Always surface risk in prepare log + pipeline verify | | Mitigate (spill to disk) | **On** | Prefer progress over hanging; only analyzer-flagged hops spill | | Spill directory | empty → system temp | Standard Hop temp behavior | Existing saved run configs keep their stored flags. Only **new** local configs get these defaults. Product rationale for default-on mitigate: detection already logs risks and spill usage; locking up for hours is worse for data engineers than occasional disk I/O on known-bad topologies. Users who want strict bounded memory only can turn mitigation off. ### Scope boundaries - **Local multi-threaded engine only** (`LocalPipelineEngine` + `Pipeline` rowset allocation). - Single-threaded engine keeps `QueueRowSet`. - No transform plugin code changes (Stream Lookup unchanged). - Beam / Spark unchanged. ## Surfaces 1. **Pipeline verify** — WARNING per risk site. 2. **Prepare (local)** — log risks; if mitigate on, log hop count using spilling rowsets; if mitigate off, warn that hang is possible. 3. **User guide** — `avoiding-deadlocks.adoc` section 5 + short design notes. ## Test plan - [x] Unit: `SpillingRowSetTest` (FIFO past capacity, clear, size() not advertising full while spilling) - [x] Unit: `PipelineBufferDeadlockAnalyzerTest` (split–rejoin risk, independent sources no risk, linear chain no risk) - [x] Manual: large split–rejoin / Stream Lookup-style pipeline with mitigate on — completes; throughput after size() fix - [ ] CI: full build / spotless - [ ] Review: existing local run configs without new keys still load; new configs show both checkboxes on in the GUI ## Related - Closes / fixes discussion in https://github.com/apache/hop/issues/7752 - Related docs history: avoiding-deadlocks guide, #3740 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
