[client] RecordAccumulator's ready method will be blocked if no batch is ready. #322
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -45,6 +45,8 @@ | |
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
|
|
||
| import javax.annotation.concurrent.GuardedBy; | ||
|
|
||
| import java.io.IOException; | ||
| import java.util.ArrayDeque; | ||
| import java.util.ArrayList; | ||
|
|
@@ -56,9 +58,13 @@ | |
| import java.util.Map; | ||
| import java.util.Set; | ||
| import java.util.concurrent.ConcurrentMap; | ||
| import java.util.concurrent.TimeUnit; | ||
| import java.util.concurrent.atomic.AtomicInteger; | ||
| import java.util.concurrent.locks.Condition; | ||
| import java.util.concurrent.locks.ReentrantLock; | ||
|
|
||
| import static com.alibaba.fluss.record.LogRecordBatch.NO_WRITER_ID; | ||
| import static com.alibaba.fluss.utils.concurrent.LockUtils.inLock; | ||
|
|
||
| /* This file is based on source code of Apache Kafka Project (https://kafka.apache.org/), licensed by the Apache | ||
| * Software Foundation (ASF) under the Apache License, Version 2.0. See the NOTICE file distributed with this work for | ||
|
|
@@ -107,6 +113,11 @@ public final class RecordAccumulator { | |
|
|
||
| private final IdempotenceManager idempotenceManager; | ||
|
|
||
| private final ReentrantLock batchReadyLock = new ReentrantLock(); | ||
|
|
||
| @GuardedBy("lock") | ||
| private final Condition batchIsReady = batchReadyLock.newCondition(); | ||
|
|
||
| // TODO add retryBackoffMs to retry the produce request upon receiving an error. | ||
| // TODO add deliveryTimeoutMs to report success or failure on record delivery. | ||
| // TODO add nextBatchExpiryTimeMs | ||
|
|
@@ -232,6 +243,10 @@ public RecordAppendResult append( | |
| * </pre> | ||
| */ | ||
| public ReadyCheckResult ready(Cluster cluster) { | ||
| return ready(cluster, batchTimeoutMs); | ||
| } | ||
|
|
||
| public ReadyCheckResult ready(Cluster cluster, long timeoutMs) { | ||
| Set<ServerNode> readyNodes = new HashSet<>(); | ||
| Set<PhysicalTablePath> unknownLeaderTables = new HashSet<>(); | ||
| // Go table by table so that we can get queue sizes for buckets in a table and calculate | ||
|
|
@@ -244,6 +259,9 @@ public ReadyCheckResult ready(Cluster cluster) { | |
| readyNodes, | ||
| unknownLeaderTables, | ||
| cluster)); | ||
| if (readyNodes.isEmpty()) { | ||
| waitBatchReady(timeoutMs); | ||
| } | ||
|
|
||
| // TODO and the earliest time at which any non-send-able bucket will be ready; | ||
|
|
||
|
|
@@ -402,59 +420,59 @@ private void bucketReady( | |
| Map<Integer, Deque<WriteBatch>> batches = bucketAndWriteBatches.batches; | ||
| // Collect the queue sizes for available buckets to be used in adaptive bucket allocate. | ||
|
|
||
| batches.forEach( | ||
| (bucketId, deque) -> { | ||
| TableBucket tableBucket = cluster.getTableBucket(physicalTablePath, bucketId); | ||
| ServerNode leader = cluster.leaderFor(tableBucket); | ||
|
|
||
| if (leader == null) { | ||
| // This is a bucket for which leader is not known, but messages are | ||
| // available to send. Note that entries are currently not removed from | ||
| // batches when deque is empty. | ||
| unknownLeaderTables.add(physicalTablePath); | ||
| } else { | ||
| batchReady(deque, leader, readyNodes); | ||
| } | ||
| }); | ||
| } | ||
|
|
||
| private void batchReady( | ||
| Deque<WriteBatch> deque, ServerNode leader, Set<ServerNode> readyNodes) { | ||
| if (!readyNodes.contains(leader)) { | ||
| // if the wait time larger than lingerMs, we can send this batch even if it is not full. | ||
| if (batchReady(deque)) { | ||
| readyNodes.add(leader); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| private boolean batchReady(Deque<WriteBatch> deque) { | ||
| final long waitedTimeMs; | ||
| final int dequeSize; | ||
| final boolean full; | ||
|
|
||
| boolean exhausted = writerBufferPool.queued() > 0; | ||
|
There was a problem hiding this comment. this should be moved out of batches foreach, as it requires to obtain a lock. |
||
| // Note: this loop is especially hot with large bucket counts. | ||
| // We are careful to only perform the minimum required inside the synchronized | ||
| // block, as this lock is also used to synchronize writer threads | ||
| // attempting to append() to a bucket/batch. | ||
| synchronized (deque) { | ||
| // Deque are often empty in this path, esp with large bucket counts, | ||
| // so we exit early if we can. | ||
| WriteBatch batch = deque.peekFirst(); | ||
| if (batch == null) { | ||
| return false; | ||
| } | ||
|
|
||
| waitedTimeMs = batch.waitedTimeMs(System.currentTimeMillis()); | ||
| dequeSize = deque.size(); | ||
| full = dequeSize > 1 || batch.isClosed(); | ||
| } | ||
| // if the wait time larger than lingerMs, we can send this batch even if it is not full. | ||
| boolean expired = waitedTimeMs >= (long) batchTimeoutMs; | ||
| return full || expired || exhausted || closed || flushInProgress(); | ||
| } | ||
|
|
||
| /** | ||
| * Are there any threads currently waiting on a flush? | ||
| * | ||
|
|
@@ -532,6 +550,7 @@ private RecordAppendResult tryAppend( | |
| WriteBatch last = deque.peekLast(); | ||
| if (last != null) { | ||
| boolean success = last.tryAppend(writeRecord, callback); | ||
| notifyBatchReady(deque); | ||
|
There was a problem hiding this comment. why here need to notify batch ready? There was a problem hiding this comment. @wuchong If not try to notify batch after each insert, the RecordAccumulator#ready will be blocked until timeout. I use use a producer-consumer-mode in RecordAccumulator, which block in consumer and signal by producer. Maybe I do same thing as kafka, accumulator.ready return the min wait time of each bucket, and then sender suspends with There was a problem hiding this comment. I mean why notify for each record? Not when a batch is ready? It will result in busy loop that the sender doesn't wait at all. |
||
| if (!success) { | ||
| last.close(); | ||
| } else { | ||
|
|
@@ -760,6 +779,30 @@ private void insertInSequenceOrder(Deque<WriteBatch> deque, WriteBatch batch) { | |
| } | ||
| } | ||
|
|
||
| private void notifyBatchReady(Deque<WriteBatch> deque) { | ||
| inLock( | ||
| batchReadyLock, | ||
| () -> { | ||
| if (batchReady(deque)) { | ||
| batchIsReady.signal(); | ||
| } | ||
| }); | ||
| } | ||
|
|
||
| private void waitBatchReady(long timeoutMs) { | ||
| try { | ||
| inLock( | ||
| batchReadyLock, | ||
| () -> { | ||
| if (timeoutMs > 0) { | ||
|
There was a problem hiding this comment. should be checked before lock . |
||
| batchIsReady.await(timeoutMs, TimeUnit.MILLISECONDS); | ||
| } | ||
| }); | ||
| } catch (InterruptedException e) { | ||
| throw new RuntimeException(e); | ||
| } | ||
| } | ||
|
|
||
| /** Metadata about a record just appended to the record accumulator. */ | ||
| public static final class RecordAppendResult { | ||
| public final boolean batchIsFull; | ||
|
|
@@ -790,7 +833,7 @@ public ReadyCheckResult( | |
| /** Close this accumulator and force all the record buffers to be drained. */ | ||
| public void close() { | ||
| closed = true; | ||
| inLock(batchReadyLock, batchIsReady::signalAll); | ||
| writerBufferPool.close(); | ||
| arrowWriterPool.close(); | ||
| bufferAllocator.close(); | ||
|
|
||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
why do we need to provide this method? only testing purpose? If yes, why not use the
batchTimeoutMsparameter in tests?