bamaer commented on PR #9142:
URL: https://github.com/apache/hop/pull/9142#issuecomment-6506689277
@hansva thanks, this looks good. I built core and engine, ran the new and
existing row set, BaseTransform and Pipeline tests, and pushed 5M rows through
7 Dummy transforms (including a 4-copy transform and a copy-to-two-targets
split that joins again). Every row arrived, lock-free was about 30% faster
(14.5 s vs 21 s), and stopping the pipeline mid-run ended in 5 ms.
**Should fix**
- **Idle transforms use a lot more CPU.** A transform waiting for rows wakes
up every 1 ms (`MAX_PARK_NANOS`, also the cap in `awaitAny`). With 30 idle
Dummy transforms that was 25% of a core, against 1.5% with the current row
sets. That matters for long-running pipelines that are mostly idle. A plain
backoff makes slow trickles of rows much later, because a writer only wakes a
parked reader at `wakeBatch`. What worked for me was all three of these
together:
1. Back off in `getRowLockFree` from 1 ms up to 20 ms.
2. While the reader is backed off, have the writer wake it on its next row
(a `wakeOnAnyRow` flag next to `parkedConsumer`).
3. Back off only when every input is a `LockFreeRowSet`, since hops that
spill to disk cannot wake the reader.
On the same pipelines: throughput unchanged, idle CPU 1.7%, and a single
row every 100 ms got through in 1.3 ms median instead of 4.1 ms. One cost: the
writer then reads `parkedConsumer` for every row, not every 64th. I saw no
difference, but it is worth checking on your own benchmarks. Diff below.
Minor: the buffer size is rounded up to a power of two, so the default of
10,000 rows becomes 16,384. A line about that in the docs would help anyone who
sized their buffers to save memory.
<details><summary>Diff</summary>
```diff
--- a/core/src/main/java/org/apache/hop/core/LockFreeRowSet.java
+++ b/core/src/main/java/org/apache/hop/core/LockFreeRowSet.java
@@ -133,14 +133,15 @@
* that many rows.
*/
private void rowAdded(long index) {
- if ((index & wakeCheckMask) != 0) {
+ Thread consumer = parkedConsumer;
+ if (consumer == null) {
return;
}
- Thread consumer = parkedConsumer;
- // claim the parked thread first, so it is unparked once and not again
for every row added
- // before it gets to run
- if (consumer != null
- && producerIndex - consumerIndex >= wakeBatch
+ boolean anyRow = wakeOnAnyRow; // read after parkedConsumer, which the
consumer writes last
+ if ((index & wakeCheckMask) != 0 && !anyRow) {
+ return;
+ }
+ if ((anyRow || producerIndex - consumerIndex >= wakeBatch)
&& PARKED_CONSUMER.compareAndSet(this, consumer, null)) {
LockSupport.unpark(consumer);
}
@@ -278,18 +279,33 @@
Thread.onSpinWait();
}
Thread current = Thread.currentThread();
+ // a long park only when every input can wake us; other row sets are
polled every 1 ms
+ boolean deep = maxNanos > MAX_PARK_NANOS;
+ for (IRowSet rowSet : rowSets) {
+ if (!(rowSet instanceof LockFreeRowSet)) {
+ deep = false;
+ maxNanos = MAX_PARK_NANOS;
+ break;
+ }
+ }
for (IRowSet rowSet : rowSets) {
if (rowSet instanceof LockFreeRowSet lockFree) {
+ if (deep) {
+ lockFree.wakeOnAnyRow = true;
+ }
lockFree.parkedConsumer = current;
}
}
// check again after announcing ourselves on every input, so a wake-up
is not lost
if (!anyReady(rowSets)) {
- LockSupport.parkNanos(rowSets, Math.min(maxNanos, MAX_PARK_NANOS));
+ LockSupport.parkNanos(rowSets, maxNanos);
}
for (IRowSet rowSet : rowSets) {
if (rowSet instanceof LockFreeRowSet lockFree) {
PARKED_CONSUMER.compareAndSet(lockFree, current, null);
+ if (deep) {
+ lockFree.wakeOnAnyRow = false;
+ }
}
}
}
@@ -411,6 +427,7 @@
/** Written by the consumer when it parks, read by producers for every row.
*/
abstract class LockFreeRowSetParkedConsumer extends LockFreeRowSetPad3 {
volatile Thread parkedConsumer;
+ volatile boolean wakeOnAnyRow;
LockFreeRowSetParkedConsumer(int maxSize) {
super(maxSize);
---
a/engine/src/main/java/org/apache/hop/pipeline/transform/BaseTransform.java
+++
b/engine/src/main/java/org/apache/hop/pipeline/transform/BaseTransform.java
@@ -168,6 +168,9 @@
/** The longest a transform parks waiting on its lock-free inputs before
it checks them again. */
private static final long LOCK_FREE_MAX_WAIT_NANOS =
TimeUnit.MILLISECONDS.toNanos(1);
+ /** The longest a transform parks once its lock-free inputs have been
idle for a while. */
+ private static final long LOCK_FREE_MAX_IDLE_NANOS =
TimeUnit.MILLISECONDS.toNanos(20);
+
private static final VarHandle LINES_READ;
private static final VarHandle LINES_WRITTEN;
private static final VarHandle LINES_INPUT;
@@ -2029,6 +2032,7 @@
*/
private Object[] getRowLockFree(IRowSet inputRowSet) {
int emptyInputs = 0;
+ long idleWait = LOCK_FREE_MAX_WAIT_NANOS;
while (!isStopped()) {
Object[] row = inputRowSet.getRowImmediate();
if (row == null && inputRowSet.isDone()) {
@@ -2057,7 +2061,9 @@
return row;
}
if (++emptyInputs >= inputRowSets.size()) {
- LockFreeRowSet.awaitAny(inputRowSets, LOCK_FREE_MAX_WAIT_NANOS);
+ LockFreeRowSet.awaitAny(inputRowSets, idleWait);
+ // back off while nothing arrives: lock-free inputs then wake us on
their next row
+ idleWait = Math.min(idleWait * 2, LOCK_FREE_MAX_IDLE_NANOS);
emptyInputs = 0;
}
nextInputStream();
```
</details>
--
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]
