hansva commented on code in PR #9142: URL: https://github.com/apache/hop/pull/9142#discussion_r4443846383
########## core/src/main/java/org/apache/hop/core/LockFreeRowSet.java: ########## @@ -0,0 +1,595 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.hop.core; + +import java.lang.invoke.MethodHandles; +import java.lang.invoke.VarHandle; +import java.util.List; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.locks.LockSupport; +import org.apache.hop.core.row.IRowMeta; + +/** + * A row buffer without locks on the hot path: a bounded multi-producer / single-consumer ring (the + * design of JCTools' MpscArrayQueue). + * + * <ul> + * <li>A slot that is not null holds a row, so there is no separate sequence array. + * <li>The slot array is rounded up to a power of two, so an index maps to a slot with a mask. The + * row set still holds at most the configured number of rows. + * <li>Producers claim a slot with one compare-and-set on the producer index. They keep a cached + * limit (consumer index + size) and only read the consumer index again when they reach it, so + * a producer does not touch the consumer's cache line for every row. + * <li>A row set is read by exactly one transform thread, so the consumer moves its index with a + * plain release store. (Several producers are allowed: some transforms call putRow from + * worker threads, e.g. LanguageModelChat.) + * <li>The producer fields, the consumer fields and the parked-thread fields each sit on their own + * cache line (padding through the superclasses below, which the JVM lays out first). + * </ul> + * + * <p>When the ring is full (or empty) the thread spins briefly, then parks. The other side only + * unparks it once a batch of rows (or free slots) is ready, a quarter of the size, or when the row + * set is done, claiming the parked thread so it is woken once. That lets a fast consumer process + * runs of rows instead of waking up for every row, which is what the per-row size() check and + * Thread.sleep in BaseTransform approximate for the lock-based row sets. + * + * <p>A thread parks for {@link LockFreeRowSetFields#MAX_PARK_NANOS} at first. While nothing changes + * it parks longer, up to {@link LockFreeRowSetFields#MAX_IDLE_PARK_NANOS}, so an idle transform + * costs little CPU. A reader in such a long park is woken by the next row instead of a batch, so a + * trickle of rows is not delayed either. + */ +public class LockFreeRowSet extends LockFreeRowSetPad5 implements Comparable<IRowSet>, IRowSet { + private static final VarHandle SLOT = MethodHandles.arrayElementVarHandle(Object[][].class); + private static final VarHandle PRODUCER_INDEX; + private static final VarHandle PRODUCER_LIMIT; + private static final VarHandle CONSUMER_INDEX; + private static final VarHandle PARKED_CONSUMER; + private static final VarHandle PARKED_PRODUCER; + + static { + try { + MethodHandles.Lookup lookup = MethodHandles.lookup(); + PRODUCER_INDEX = + lookup.findVarHandle(LockFreeRowSetProducerFields.class, "producerIndex", long.class); + PRODUCER_LIMIT = + lookup.findVarHandle(LockFreeRowSetProducerFields.class, "producerLimit", long.class); + CONSUMER_INDEX = + lookup.findVarHandle(LockFreeRowSetConsumerFields.class, "consumerIndex", long.class); + PARKED_CONSUMER = + lookup.findVarHandle(LockFreeRowSetParkedConsumer.class, "parkedConsumer", Thread.class); + PARKED_PRODUCER = + lookup.findVarHandle(LockFreeRowSetParkedProducer.class, "parkedProducer", Thread.class); + } catch (ReflectiveOperationException e) { + throw new ExceptionInInitializerError(e); + } + } + + /** + * @param maxSize the number of rows the buffer holds + */ + public LockFreeRowSet(int maxSize) { + super(maxSize); + } + + // ---- ring ---------------------------------------------------------------------------------- + + /** + * @return the claimed index, or -1 when the ring is full + */ + private long offer(Object[] row) { + long limit = (long) PRODUCER_LIMIT.getOpaque(this); + long index; + do { + index = producerIndex; + if (index >= limit) { + limit = consumerIndex + maxRows; + if (index >= limit) { + return -1; // full + } + PRODUCER_LIMIT.setOpaque(this, limit); Review Comment: fixed -- 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]
