Skip to content

Commit

Permalink
remove lock race between heartbeat and append log channels
Browse files Browse the repository at this point in the history
  • Loading branch information
SzyWilliam committed Jan 2, 2025
1 parent 320b207 commit c776882
Show file tree
Hide file tree
Showing 2 changed files with 16 additions and 1 deletion.
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@
*/
package org.apache.ratis.util;

import java.util.Objects;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.Lock;

Expand Down Expand Up @@ -45,6 +46,13 @@ public static AutoCloseableLock acquire(final Lock lock, Runnable preUnlock) {
return new AutoCloseableLock(lock, preUnlock);
}

public static AutoCloseableLock tryAcquire(final Lock lock, Runnable preUnlock, TimeDuration timeout)
throws InterruptedException {
Objects.requireNonNull(timeout, "timeout == null");
final boolean locked = lock.tryLock(timeout.getDuration(), timeout.getUnit());
return locked? new AutoCloseableLock(lock, preUnlock): null;
}

private final Lock underlying;
private final AtomicBoolean closed = new AtomicBoolean(false);
private final Runnable preUnlock;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -122,7 +122,7 @@ public boolean isOpened() {

@Override
public boolean updateCommitIndex(long majorityIndex, long currentTerm, boolean isLeader) {
try(AutoCloseableLock writeLock = writeLock()) {
try(AutoCloseableLock writeLock = tryWriteLock(TimeDuration.ONE_SECOND)) {
final long oldCommittedIndex = getLastCommittedIndex();
final long newCommitIndex = Math.min(majorityIndex, getFlushIndex());
if (oldCommittedIndex < newCommitIndex) {
Expand All @@ -136,6 +136,9 @@ public boolean updateCommitIndex(long majorityIndex, long currentTerm, boolean i
return commitIndex.updateIncreasingly(newCommitIndex, traceIndexChange);
}
}
} catch (InterruptedException e) {
LOG.warn("{}: Interrupted to updateCommitIndex: majorityIndex={}, currentTerm={}, isLeader={}",
getName(), majorityIndex, currentTerm, isLeader, e);
}
return false;
}
Expand Down Expand Up @@ -389,6 +392,10 @@ public AutoCloseableLock writeLock() {
return AutoCloseableLock.acquire(lock.writeLock());
}

public AutoCloseableLock tryWriteLock(TimeDuration timeout) throws InterruptedException {
return AutoCloseableLock.tryAcquire(lock.writeLock(), null, timeout);
}

public boolean hasWriteLock() {
return this.lock.isWriteLockedByCurrentThread();
}
Expand Down

0 comments on commit c776882

Please sign in to comment.