You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@geode.apache.org by lg...@apache.org on 2017/07/26 19:12:17 UTC
[2/2] geode git commit: GEODE-2654: Backups can capture different
members from different points in time Online backup now takes a backup from a
single point in time. This is done by making all write operations get a
backup lock which causes them to wait
GEODE-2654: Backups can capture different members from different points in time
Online backup now takes a backup from a single point in time.
This is done by making all write operations get a backup lock which causes them to
wait until the oplogs are rolled by backup.
Project: http://git-wip-us.apache.org/repos/asf/geode/repo
Commit: http://git-wip-us.apache.org/repos/asf/geode/commit/f1326be5
Tree: http://git-wip-us.apache.org/repos/asf/geode/tree/f1326be5
Diff: http://git-wip-us.apache.org/repos/asf/geode/diff/f1326be5
Branch: refs/heads/develop
Commit: f1326be55c54e96d75fe8ec6391aa4581f312f86
Parents: bf8971c
Author: Lynn Gallinat <lg...@pivotal.io>
Authored: Thu Jun 22 11:34:37 2017 -0700
Committer: Lynn Gallinat <lg...@pivotal.io>
Committed: Wed Jul 26 12:02:09 2017 -0700
----------------------------------------------------------------------
.../apache/geode/internal/cache/BackupLock.java | 66 +-
.../geode/internal/cache/DiskInitFile.java | 287 +++----
.../geode/internal/cache/DiskStoreImpl.java | 16 +-
.../org/apache/geode/internal/cache/Oplog.java | 764 ++++++++++---------
.../BackupPrepareAndFinishMsgDUnitTest.java | 571 ++++++++++++++
.../geode/codeAnalysis/excludedClasses.txt | 1 +
.../codeAnalysis/sanctionedSerializables.txt | 1 -
7 files changed, 1187 insertions(+), 519 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/geode/blob/f1326be5/geode-core/src/main/java/org/apache/geode/internal/cache/BackupLock.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/BackupLock.java b/geode-core/src/main/java/org/apache/geode/internal/cache/BackupLock.java
index 4b4fb10..f531860 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/BackupLock.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/BackupLock.java
@@ -28,54 +28,72 @@ import java.util.concurrent.locks.ReentrantLock;
*/
public class BackupLock extends ReentrantLock {
- private Thread backupThread;
+ private final ThreadLocal<Boolean> isBackupThread = new ThreadLocal<Boolean>();
boolean isBackingUp;
Condition backupDone = super.newCondition();
+ // test hook
+ private BackupLockTestHook hook = null;
+
+ public interface BackupLockTestHook {
+ /**
+ * Test hook called before the wait for backup to complete
+ */
+ public void beforeWaitForBackupCompletion();
+ }
+
+ public void setBackupLockTestHook(BackupLockTestHook testHook) {
+ hook = testHook;
+ }
+
public void lockForBackup() {
super.lock();
isBackingUp = true;
super.unlock();
}
- public void setBackupThread(Thread thread) {
- super.lock();
- backupThread = thread;
- super.unlock();
+ public void setBackupThread() {
+ isBackupThread.set(true);
}
public void unlockForBackup() {
super.lock();
isBackingUp = false;
- backupThread = null;
+ isBackupThread.set(false);
backupDone.signalAll();
super.unlock();
}
- /**
- * Acquire this lock, waiting for an in progress backup if one is in progress.
- */
+ public boolean isCurrentThreadDoingBackup() {
+ Boolean result = isBackupThread.get();
+ return (result != null) && result;
+ }
+
@Override
- public void lock() {
- lock(true);
+ public void unlock() {
+ // The backup thread does not need to unlock this lock since it never gets the lock. It is the
+ // only thread that has permission to modify disk files during backup.
+ if (!isCurrentThreadDoingBackup()) {
+ super.unlock();
+ }
}
/**
- * Acquire this lock, Optionally waiting for a backup to finish the first phase. Any operations
- * that update metadata related to the distributed system state should pass true for this flag,
- * because we need to make sure we get a point in time snapshot of the init files across members
- * to for metadata consistentency.
+ * Acquire this lock, waiting for a backup to finish the first phase.
*
- * Updates which update only record changes to the local state on this member(eg, switching
- * oplogs), do not need to wait for the backup.
- *
- * @param waitForBackup if true, we will wait for an in progress backup before acquiring this
- * lock.
*/
- public void lock(boolean waitForBackup) {
- super.lock();
- while (isBackingUp && waitForBackup && !(Thread.currentThread() == backupThread)) {
- backupDone.awaitUninterruptibly();
+ @Override
+ public void lock() {
+ // The backup thread is a noop; it does not need to get the lock since it is the only thread
+ // with permission to modify disk files during backup
+ if (!isCurrentThreadDoingBackup()) {
+ super.lock();
+ while (isBackingUp) {
+ if (hook != null) {
+ hook.beforeWaitForBackupCompletion();
+ }
+ backupDone.awaitUninterruptibly();
+ }
}
}
}
http://git-wip-us.apache.org/repos/asf/geode/blob/f1326be5/geode-core/src/main/java/org/apache/geode/internal/cache/DiskInitFile.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/DiskInitFile.java b/geode-core/src/main/java/org/apache/geode/internal/cache/DiskInitFile.java
index 0925d28..33aa0b5 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/DiskInitFile.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/DiskInitFile.java
@@ -14,6 +14,34 @@
*/
package org.apache.geode.internal.cache;
+import java.io.BufferedInputStream;
+import java.io.ByteArrayInputStream;
+import java.io.DataInput;
+import java.io.DataInputStream;
+import java.io.DataOutput;
+import java.io.EOFException;
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.io.PrintStream;
+import java.io.RandomAccessFile;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.EnumSet;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.locks.ReentrantLock;
+
import it.unimi.dsi.fastutil.ints.Int2ObjectMap;
import it.unimi.dsi.fastutil.ints.Int2ObjectOpenHashMap;
import it.unimi.dsi.fastutil.ints.IntOpenHashSet;
@@ -21,10 +49,13 @@ import it.unimi.dsi.fastutil.longs.LongIterator;
import it.unimi.dsi.fastutil.longs.LongOpenHashSet;
import it.unimi.dsi.fastutil.objects.ObjectIterator;
import org.apache.commons.io.FileUtils;
+import org.apache.logging.log4j.Logger;
+
import org.apache.geode.CancelCriterion;
import org.apache.geode.CancelException;
import org.apache.geode.DataSerializer;
import org.apache.geode.Instantiator;
+import org.apache.geode.InternalGemFireException;
import org.apache.geode.cache.DiskAccessException;
import org.apache.geode.cache.EvictionAction;
import org.apache.geode.cache.EvictionAlgorithm;
@@ -52,34 +83,6 @@ import org.apache.geode.internal.concurrent.ConcurrentHashSet;
import org.apache.geode.internal.i18n.LocalizedStrings;
import org.apache.geode.internal.logging.LogService;
import org.apache.geode.internal.logging.log4j.LogMarker;
-import org.apache.logging.log4j.Logger;
-
-import java.io.BufferedInputStream;
-import java.io.ByteArrayInputStream;
-import java.io.DataInput;
-import java.io.DataInputStream;
-import java.io.DataOutput;
-import java.io.EOFException;
-import java.io.File;
-import java.io.FileInputStream;
-import java.io.IOException;
-import java.io.PrintStream;
-import java.io.RandomAccessFile;
-import java.nio.ByteBuffer;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.EnumSet;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Iterator;
-import java.util.LinkedHashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.UUID;
-import java.util.concurrent.ConcurrentHashMap;
/**
* Does all the IF file work for a DiskStoreImpl.
@@ -96,6 +99,8 @@ public class DiskInitFile implements DiskInitFileInterpreter {
public static final byte END_OF_RECORD_ID = 21;
static final int OPLOG_FILE_ID_REC_SIZE = 1 + 8 + 1;
+ private final ReentrantLock lock = new ReentrantLock();
+
/**
* Written to IF Byte Format: 8: leastSigBits of UUID 8: mostSigBits of UUID 1: EndOfRecordMarker
*/
@@ -381,11 +386,19 @@ public class DiskInitFile implements DiskInitFileInterpreter {
transient private boolean gotEOF;
- /**
- * Lock used to synchronize access to the init file. This is a lock rather than a synchronized
- * block because the backup tool needs to acquire this lock.
- */
- private final BackupLock lock = new BackupLock();
+ private void lock(boolean useBackupLock) {
+ if (useBackupLock) {
+ getDiskStore().getBackupLock().lock();
+ }
+ this.lock.lock();
+ }
+
+ private void unlock(boolean useBackupLock) {
+ if (useBackupLock) {
+ getDiskStore().getBackupLock().unlock();
+ }
+ this.lock.unlock();
+ }
private void recoverFromFailedCompaction() {
File tmpFile = getTempFile();
@@ -530,11 +543,11 @@ public class DiskInitFile implements DiskInitFileInterpreter {
private int liveRegions = 0; // added for bug 41618
public boolean hasLiveRegions() {
- lock.lock(false);
+ lock(false);
try {
return this.liveRegions > 0;
} finally {
- lock.unlock();
+ unlock(false);
}
}
@@ -817,7 +830,7 @@ public class DiskInitFile implements DiskInitFileInterpreter {
}
DiskRegionView takeDiskRegionByName(String name) {
- lock.lock(false);
+ lock(false);
try {
DiskRegionView result = this.drMapByName.remove(name);
if (result != null) {
@@ -825,16 +838,16 @@ public class DiskInitFile implements DiskInitFileInterpreter {
}
return result;
} finally {
- lock.unlock();
+ unlock(false);
}
}
Map<Long, PlaceHolderDiskRegion> getDRMap() {
- lock.lock(false);
+ lock(false);
try {
return new HashMap<Long, PlaceHolderDiskRegion>(drMap);
} finally {
- lock.unlock();
+ unlock(false);
}
}
@@ -843,7 +856,7 @@ public class DiskInitFile implements DiskInitFileInterpreter {
DiskRegionStats stats, CancelCriterion cancel, DiskExceptionHandler exceptionHandler,
RegionAttributes ra, EnumSet<DiskRegionFlag> flags, String partitionName,
int startingBucketId, Compressor compressor, boolean offHeap) {
- lock.lock(false);
+ lock(true);
try {
// need to call the constructor and addDiskRegion while synced
DiskRegion result = new DiskRegion(dsi, name, isBucket, isPersistBackup, overflowEnabled,
@@ -852,21 +865,21 @@ public class DiskInitFile implements DiskInitFileInterpreter {
dsi.addDiskRegion(result);
return result;
} finally {
- lock.unlock();
+ unlock(true);
}
}
DiskRegionView getDiskRegionByName(String name) {
- lock.lock(false);
+ lock(false);
try {
return this.drMapByName.get(name);
} finally {
- lock.unlock();
+ unlock(false);
}
}
DiskRegionView getDiskRegionByPrName(String name) {
- lock.lock(false);
+ lock(false);
try {
for (PlaceHolderDiskRegion dr : this.drMapByName.values()) {
if (dr.isBucket()) {
@@ -877,7 +890,7 @@ public class DiskInitFile implements DiskInitFileInterpreter {
}
return null;
} finally {
- lock.unlock();
+ unlock(false);
}
}
@@ -1249,7 +1262,7 @@ public class DiskInitFile implements DiskInitFileInterpreter {
private void saveInstantiator(int id, String instantiatorClassName,
String instantiatedClassName) {
- lock.lock();
+ lock(true);
try {
if (!this.compactInProgress && this.instIds.contains(id)) {
// instantiator already written to disk so just return
@@ -1273,7 +1286,7 @@ public class DiskInitFile implements DiskInitFileInterpreter {
.toLocalizedString(ex),
this.parent);
} finally {
- lock.unlock();
+ unlock(true);
}
}
@@ -1308,7 +1321,7 @@ public class DiskInitFile implements DiskInitFileInterpreter {
* Write the specified DataSerializer to the file.
*/
private void saveDataSerializer(DataSerializer ds) {
- lock.lock();
+ lock(true);
try {
if (!this.compactInProgress && this.dsIds.contains(ds.getId())) {
// dataSerializer already written to disk so just return
@@ -1327,7 +1340,7 @@ public class DiskInitFile implements DiskInitFileInterpreter {
.toLocalizedString(ex),
this.parent);
} finally {
- lock.unlock();
+ unlock(true);
}
}
@@ -1412,7 +1425,7 @@ public class DiskInitFile implements DiskInitFileInterpreter {
private static final double COMPACT_RATIO = 0.5;
private void compactIfNeeded() {
- lock.lock(false);
+ lock(true);
try {
if (this.compactInProgress)
return;
@@ -1428,7 +1441,7 @@ public class DiskInitFile implements DiskInitFileInterpreter {
} catch (IOException ignore) {
return;
} finally {
- lock.unlock();
+ unlock(true);
}
}
@@ -1441,7 +1454,7 @@ public class DiskInitFile implements DiskInitFileInterpreter {
}
private void compact() {
- lock.lock(false);
+ lock(true);
this.compactInProgress = true;
try {
try {
@@ -1501,16 +1514,16 @@ public class DiskInitFile implements DiskInitFileInterpreter {
}
} finally {
this.compactInProgress = false;
- lock.unlock();
+ unlock(true);
}
}
public void copyTo(File targetDir) throws IOException {
- lock.lock(false);
+ lock(false);
try {
FileUtils.copyFileToDirectory(this.ifFile, targetDir);
} finally {
- lock.unlock();
+ unlock(false);
}
}
@@ -1569,7 +1582,7 @@ public class DiskInitFile implements DiskInitFileInterpreter {
* Write all live data to the init file
*/
private void writeLiveData() {
- lock.lock(false);
+ lock(true);
try {
this.ifLiveRecordCount = 0;
this.ifTotalRecordCount = 0;
@@ -1594,7 +1607,7 @@ public class DiskInitFile implements DiskInitFileInterpreter {
this.ifTotalRecordCount);
}
} finally {
- lock.unlock();
+ unlock(true);
}
}
@@ -1646,7 +1659,7 @@ public class DiskInitFile implements DiskInitFileInterpreter {
}
private void writeDiskStoreId() {
- lock.lock();
+ lock(true);
try {
ByteBuffer bb = getIFWriteBuffer(1 + 6 + 1);
bb.put(OPLOG_MAGIC_SEQ_ID);
@@ -1669,7 +1682,7 @@ public class DiskInitFile implements DiskInitFileInterpreter {
}
throw dae;
} finally {
- lock.unlock();
+ unlock(true);
}
}
@@ -1916,7 +1929,7 @@ public class DiskInitFile implements DiskInitFileInterpreter {
}
void closeRegion(DiskRegionView dr) {
- lock.lock();
+ lock(true);
try {
this.parent.rmById(dr.getId()); // fix for bug 41334
PlaceHolderDiskRegion phdr = new PlaceHolderDiskRegion(dr);
@@ -1924,12 +1937,12 @@ public class DiskInitFile implements DiskInitFileInterpreter {
this.drMapByName.put(dr.getName(), phdr);
// @todo make sure we only have one instance of the region for this name
} finally {
- lock.unlock();
+ unlock(true);
}
}
void clearRegion(DiskRegionView dr, long clearOplogEntryId) {
- lock.lock();
+ lock(true);
try {
if (clearOplogEntryId != DiskStoreImpl.INVALID_ID) {
this.ifTotalRecordCount++;
@@ -1945,7 +1958,7 @@ public class DiskInitFile implements DiskInitFileInterpreter {
writeIFRecord(IFREC_CLEAR_REGION_ID, dr, clearOplogEntryId);
}
} finally {
- lock.unlock();
+ unlock(true);
}
}
@@ -1953,7 +1966,7 @@ public class DiskInitFile implements DiskInitFileInterpreter {
* Clear the region using an RVV.
*/
void clearRegion(DiskRegion dr, RegionVersionVector rvv) {
- lock.lock();
+ lock(true);
try {
this.ifTotalRecordCount++;
if (dr.getClearRVV() == null) {
@@ -1964,7 +1977,7 @@ public class DiskInitFile implements DiskInitFileInterpreter {
dr.setClearRVV(rvv);
writeClearRecord(dr, rvv);
} finally {
- lock.unlock();
+ unlock(true);
}
}
@@ -2001,7 +2014,7 @@ public class DiskInitFile implements DiskInitFileInterpreter {
}
void createRegion(DiskRegionView drv) {
- lock.lock();
+ lock(true);
try {
if (!drv.isRecreated()) {
writeIFRecord(IFREC_CREATE_REGION_ID, drv, drv.getName());
@@ -2015,24 +2028,24 @@ public class DiskInitFile implements DiskInitFileInterpreter {
}
}
} finally {
- lock.unlock();
+ unlock(true);
}
}
void beginDestroyRegion(DiskRegionView dr) {
- lock.lock();
+ lock(true);
try {
if (regionStillCreated(dr)) {
cmnBeginDestroyRegion(dr);
writeIFRecord(IFREC_BEGIN_DESTROY_REGION_ID, dr);
}
} finally {
- lock.unlock();
+ unlock(true);
}
}
void endDestroyRegion(DiskRegionView dr) {
- lock.lock();
+ lock(true);
try {
if (regionStillCreated(dr)) {
cmnEndDestroyRegion(dr);
@@ -2043,127 +2056,127 @@ public class DiskInitFile implements DiskInitFileInterpreter {
}
}
} finally {
- lock.unlock();
+ unlock(true);
}
}
void beginDestroyDataStorage(DiskRegionView dr) {
- lock.lock();
+ lock(true);
try {
assert regionStillCreated(dr);
cmnBeginPartialDestroyRegion(dr);
writeIFRecord(IFREC_BEGIN_PARTIAL_DESTROY_REGION_ID, dr);
} finally {
- lock.unlock();
+ unlock(true);
}
}
void endDestroyDataStorage(DiskRegionView dr) {
- lock.lock();
+ lock(true);
try {
assert regionStillCreated(dr);
cmnEndPartialDestroyRegion(dr);
writeIFRecord(IFREC_END_PARTIAL_DESTROY_REGION_ID, dr);
} finally {
- lock.unlock();
+ unlock(true);
}
}
public void createPersistentPR(String name, PRPersistentConfig config) {
- lock.lock();
+ lock(true);
try {
if (cmnPRCreate(name, config)) {
writePRCreate(name, config);
}
} finally {
- lock.unlock();
+ unlock(true);
}
}
public void destroyPersistentPR(String name) {
- lock.lock();
+ lock(true);
try {
if (cmnPRDestroy(name)) {
writePRDestroy(name);
}
} finally {
- lock.unlock();
+ unlock(true);
}
}
public PRPersistentConfig getPersistentPR(String name) {
- lock.lock(false);
+ lock(false);
try {
return prMap.get(name);
} finally {
- lock.unlock();
+ unlock(false);
}
}
public Map<String, PRPersistentConfig> getAllPRs() {
- lock.lock(false);
+ lock(false);
try {
return new HashMap<String, PRPersistentConfig>(prMap);
} finally {
- lock.unlock();
+ unlock(false);
}
}
void crfCreate(long oplogId) {
- lock.lock(false);
+ lock(true);
try {
cmnCrfCreate(oplogId);
writeIFRecord(IFREC_CRF_CREATE, oplogId);
} finally {
- lock.unlock();
+ unlock(true);
}
}
void drfCreate(long oplogId) {
- lock.lock(false);
+ lock(true);
try {
cmnDrfCreate(oplogId);
writeIFRecord(IFREC_DRF_CREATE, oplogId);
} finally {
- lock.unlock();
+ unlock(true);
}
}
void krfCreate(long oplogId) {
- lock.lock(false);
+ lock(true);
try {
cmnKrfCreate(oplogId);
writeIFRecord(IFREC_KRF_CREATE, oplogId);
} finally {
- lock.unlock();
+ unlock(true);
}
}
void crfDelete(long oplogId) {
- lock.lock(false);
+ lock(true);
try {
if (cmnCrfDelete(oplogId)) {
// call writeIFRecord AFTER cmnCrfDelete to fix bug 41505
writeIFRecord(IFREC_CRF_DELETE, oplogId);
}
} finally {
- lock.unlock();
+ unlock(true);
}
}
void drfDelete(long oplogId) {
- lock.lock(false);
+ lock(true);
try {
if (cmnDrfDelete(oplogId)) {
writeIFRecord(IFREC_DRF_DELETE, oplogId);
}
} finally {
- lock.unlock();
+ unlock(true);
}
}
int getOrCreateCanonicalId(Object object) {
- lock.lock(false);
+ lock(true);
try {
int id = canonicalIdHolder.getId(object);
if (id <= 0) {
@@ -2172,21 +2185,21 @@ public class DiskInitFile implements DiskInitFileInterpreter {
}
return id;
} finally {
- lock.unlock();
+ unlock(true);
}
}
Object getCanonicalObject(int id) {
- lock.lock(false);
+ lock(false);
try {
return canonicalIdHolder.getObject(id);
} finally {
- lock.unlock();
+ unlock(false);
}
}
void close() {
- lock.lock();
+ lock(true);
try {
if (this.closed)
return;
@@ -2203,17 +2216,17 @@ public class DiskInitFile implements DiskInitFileInterpreter {
basicDestroy();
}
} finally {
- lock.unlock();
+ unlock(true);
}
}
void destroy() {
- lock.lock();
+ lock(true);
try {
close();
basicDestroy();
} finally {
- lock.unlock();
+ unlock(true);
}
}
@@ -2229,31 +2242,31 @@ public class DiskInitFile implements DiskInitFileInterpreter {
void addMyInitializingPMID(DiskRegionView dr, PersistentMemberID pmid) {
- lock.lock();
+ lock(true);
try {
if (regionStillCreated(dr)) {
cmnAddMyInitializingPMID(dr, pmid);
writePMIDRecord(IFREC_MY_MEMBER_INITIALIZING_ID, dr, pmid, false);
}
} finally {
- lock.unlock();
+ unlock(true);
}
}
void markInitialized(DiskRegionView dr) {
- lock.lock();
+ lock(true);
try {
if (regionStillCreated(dr)) {
writeIFRecord(IFREC_MY_MEMBER_INITIALIZED_ID, dr);
cmnMarkInitialized(dr);
}
} finally {
- lock.unlock();
+ unlock(true);
}
}
void addOnlinePMID(DiskRegionView dr, PersistentMemberID pmid) {
- lock.lock();
+ lock(true);
try {
if (regionStillCreated(dr)) {
if (dr.addOnlineMember(pmid)) {
@@ -2264,12 +2277,12 @@ public class DiskInitFile implements DiskInitFileInterpreter {
}
}
} finally {
- lock.unlock();
+ unlock(true);
}
}
void addOfflinePMID(DiskRegionView dr, PersistentMemberID pmid) {
- lock.lock();
+ lock(true);
try {
if (regionStillCreated(dr)) {
if (dr.addOfflineMember(pmid)) {
@@ -2280,12 +2293,12 @@ public class DiskInitFile implements DiskInitFileInterpreter {
}
}
} finally {
- lock.unlock();
+ unlock(true);
}
}
void addOfflineAndEqualPMID(DiskRegionView dr, PersistentMemberID pmid) {
- lock.lock();
+ lock(true);
try {
if (regionStillCreated(dr)) {
if (dr.addOfflineAndEqualMember(pmid)) {
@@ -2296,12 +2309,12 @@ public class DiskInitFile implements DiskInitFileInterpreter {
}
}
} finally {
- lock.unlock();
+ unlock(true);
}
}
void rmPMID(DiskRegionView dr, PersistentMemberID pmid) {
- lock.lock();
+ lock(true);
try {
if (regionStillCreated(dr)) {
if (dr.rmOnlineMember(pmid) || dr.rmOfflineMember(pmid) || dr.rmEqualMember(pmid)) {
@@ -2312,7 +2325,7 @@ public class DiskInitFile implements DiskInitFileInterpreter {
}
}
} finally {
- lock.unlock();
+ unlock(true);
}
}
@@ -2323,7 +2336,7 @@ public class DiskInitFile implements DiskInitFileInterpreter {
return;
}
- lock.lock();
+ lock(true);
try {
if (cmnRevokeDiskStoreId(revokedPattern)) {
// we now have two records to gc (this one and the live one we removed).
@@ -2332,7 +2345,7 @@ public class DiskInitFile implements DiskInitFileInterpreter {
writeRevokedMember(revokedPattern);
}
} finally {
- lock.unlock();
+ unlock(true);
}
}
@@ -2340,12 +2353,12 @@ public class DiskInitFile implements DiskInitFileInterpreter {
* Get the set of members known to be revoked
*/
public Set<PersistentMemberPattern> getRevokedIDs() {
- lock.lock(false);
+ lock(false);
try {
// Return a copy of the set, because we modify it in place.
return new HashSet<PersistentMemberPattern>(this.revokedMembers);
} finally {
- lock.unlock();
+ unlock(false);
}
}
@@ -2354,21 +2367,21 @@ public class DiskInitFile implements DiskInitFileInterpreter {
* Return true if the given dr is still created in this IF.
*/
boolean regionStillCreated(DiskRegionView dr) {
- lock.lock(false);
+ lock(false);
try {
return getDiskRegionById(dr.getId()) != null;
} finally {
- lock.unlock();
+ unlock(false);
}
}
boolean regionExists(long drId) {
- lock.lock(false);
+ lock(false);
try {
// @todo make drMap concurrent so this call can be fast
return this.drMap.containsKey(drId);
} finally {
- lock.unlock();
+ unlock(false);
}
}
@@ -2386,11 +2399,11 @@ public class DiskInitFile implements DiskInitFileInterpreter {
}
Collection<DiskRegionView> getKnown() {
- lock.lock(false);
+ lock(false);
try {
return new ArrayList<DiskRegionView>(this.drMap.values());
} finally {
- lock.unlock();
+ unlock(false);
}
}
@@ -2585,7 +2598,7 @@ public class DiskInitFile implements DiskInitFileInterpreter {
public void destroyPRRegion(String prName) {
ArrayList<PlaceHolderDiskRegion> buckets = new ArrayList<PlaceHolderDiskRegion>();
- lock.lock();
+ lock(true);
try {
for (PlaceHolderDiskRegion dr : this.drMapByName.values()) {
if (dr.isBucket()) {
@@ -2595,7 +2608,7 @@ public class DiskInitFile implements DiskInitFileInterpreter {
}
}
} finally {
- lock.unlock();
+ unlock(true);
}
for (PlaceHolderDiskRegion dr : buckets) {
endDestroyRegion(dr);
@@ -2612,7 +2625,7 @@ public class DiskInitFile implements DiskInitFileInterpreter {
String offHeapOption, boolean printToConsole) {
StringBuffer sb = new StringBuffer();
ArrayList<PlaceHolderDiskRegion> buckets = new ArrayList<PlaceHolderDiskRegion>();
- lock.lock();
+ lock(true);
try {
for (PlaceHolderDiskRegion dr : this.drMapByName.values()) {
if (dr.isBucket()) {
@@ -2633,7 +2646,7 @@ public class DiskInitFile implements DiskInitFileInterpreter {
printInfo = false;
}
} finally {
- lock.unlock();
+ unlock(true);
}
return sb.toString();
}
@@ -2642,13 +2655,13 @@ public class DiskInitFile implements DiskInitFileInterpreter {
String lruLimitOption, String concurrencyLevelOption, String initialCapacityOption,
String loadFactorOption, String compressorClassNameOption, String statisticsEnabledOption,
String offHeapOption, boolean printToConsole) {
- lock.lock();
+ lock(true);
try {
return basicModifyRegion(false, drv, lruOption, lruActionOption, lruLimitOption,
concurrencyLevelOption, initialCapacityOption, loadFactorOption,
compressorClassNameOption, statisticsEnabledOption, offHeapOption, printToConsole);
} finally {
- lock.unlock();
+ unlock(true);
}
}
@@ -2770,20 +2783,8 @@ public class DiskInitFile implements DiskInitFileInterpreter {
return message;
}
- public void lockForBackup() {
- lock.lockForBackup();
- }
-
- public void unlockForBackup() {
- lock.unlockForBackup();
- }
-
- public void setBackupThread(Thread thread) {
- lock.setBackupThread(thread);
- }
-
private void writeGemfireVersion(Version version) {
- lock.lock();
+ lock(true);
try {
ByteBuffer bb = getIFWriteBuffer(1 + 3 + 1);
bb.put(IFREC_GEMFIRE_VERSION);
@@ -2799,7 +2800,7 @@ public class DiskInitFile implements DiskInitFileInterpreter {
}
throw dae;
} finally {
- lock.unlock();
+ unlock(true);
}
}
}
http://git-wip-us.apache.org/repos/asf/geode/blob/f1326be5/geode-core/src/main/java/org/apache/geode/internal/cache/DiskStoreImpl.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/DiskStoreImpl.java b/geode-core/src/main/java/org/apache/geode/internal/cache/DiskStoreImpl.java
index 3e97d0e..94d1253 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/DiskStoreImpl.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/DiskStoreImpl.java
@@ -4002,6 +4002,16 @@ public class DiskStoreImpl implements DiskStore {
private final HashMap<String, LRUStatistics> prlruStatMap = new HashMap<String, LRUStatistics>();
+ /**
+ * Lock used to synchronize access to the init file. This is a lock rather than a synchronized
+ * block because the backup tool needs to acquire this lock.
+ */
+ private final BackupLock backupLock = new BackupLock();
+
+ public BackupLock getBackupLock() {
+ return backupLock;
+ }
+
LRUStatistics getOrCreatePRLRUStats(PlaceHolderDiskRegion dr) {
String prName = dr.getPrName();
LRUStatistics result = null;
@@ -4046,14 +4056,14 @@ public class DiskStoreImpl implements DiskStore {
// level operations, we will need to be careful
// to block them *before* they are put in the async
// queue
- getDiskInitFile().lockForBackup();
+ getBackupLock().lockForBackup();
}
/**
* Release the lock that is preventing operations on this disk store during the backup process.
*/
public void releaseBackupLock() {
- getDiskInitFile().unlockForBackup();
+ getBackupLock().unlockForBackup();
}
/**
@@ -4063,7 +4073,7 @@ public class DiskStoreImpl implements DiskStore {
*/
public void startBackup(File targetDir, BackupInspector baselineInspector,
RestoreScript restoreScript) throws IOException {
- getDiskInitFile().setBackupThread(Thread.currentThread());
+ getBackupLock().setBackupThread();
boolean done = false;
try {
for (;;) {
http://git-wip-us.apache.org/repos/asf/geode/blob/f1326be5/geode-core/src/main/java/org/apache/geode/internal/cache/Oplog.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/Oplog.java b/geode-core/src/main/java/org/apache/geode/internal/cache/Oplog.java
index 5399d5a..80f19b5 100755
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/Oplog.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/Oplog.java
@@ -688,6 +688,8 @@ public class Oplog implements CompactableOplog, Flushable {
public void replaceIncompatibleEntry(DiskRegionView dr, DiskEntry old, DiskEntry repl) {
boolean useNextOplog = false;
+ // No need to get the backup lock prior to synchronizing (correct lock order) since the
+ // synchronized block does not attempt to get the backup lock (incorrect lock order)
synchronized (this.lock) {
if (getOplogSet().getChild() != this) {
// make sure to only call replaceIncompatibleEntry for child, because
@@ -1159,6 +1161,8 @@ public class Oplog implements CompactableOplog, Flushable {
*/
File getOplogFile() throws SyncFailedException, IOException {
// @todo check callers for drf
+ // No need to get the backup lock prior to synchronizing (correct lock order) since the
+ // synchronized block does not attempt to get the backup lock (incorrect lock order)
synchronized (this.lock/* crf */) {
if (!this.crf.RAFClosed) {
this.crf.raf.getFD().sync();
@@ -3296,6 +3300,8 @@ public class Oplog implements CompactableOplog, Flushable {
private void basicClose(boolean forceDelete) {
flushAll();
+ // No need to get the backup lock prior to synchronizing (correct lock order) since the
+ // synchronized block does not attempt to get the backup lock (incorrect lock order)
synchronized (this.lock/* crf */) {
unpreblow(this.crf, getMaxCrfSize());
if (!this.crf.RAFClosed) {
@@ -3312,6 +3318,8 @@ public class Oplog implements CompactableOplog, Flushable {
}
this.closed = true;
}
+ // No need to get the backup lock prior to synchronizing (correct lock order) since the
+ // synchronized block does not attempt to get the backup lock (incorrect lock order)
synchronized (this.lock/* drf */) {
unpreblow(this.drf, getMaxDrfSize());
if (!this.drf.RAFClosed) {
@@ -3560,118 +3568,123 @@ public class Oplog implements CompactableOplog, Flushable {
System.out.println("basicCreate KRF_DEBUG");
Thread.sleep(1000);
}
- synchronized (this.lock) { // TODO soplog perf analysis shows this as a
- // contention point
- // synchronized (this.crf) {
- initOpState(OPLOG_NEW_ENTRY_0ID, dr, entry, value, userBits, false);
- // Check if the current data in ByteBuffer will cause a
- // potential increase in the size greater than the max allowed
- long temp = (getOpStateSize() + this.crf.currSize);
- if (!this.wroteNewEntryBase) {
- temp += OPLOG_NEW_ENTRY_BASE_REC_SIZE;
- }
- if (this != getOplogSet().getChild()) {
- useNextOplog = true;
- } else if (temp > getMaxCrfSize() && !isFirstRecord()) {
- switchOpLog(dr, getOpStateSize(), entry);
- useNextOplog = true;
- } else {
- if (this.lockedForKRFcreate) {
- CacheClosedException cce = new CacheClosedException("The disk store is closed.");
- dr.getCancelCriterion().checkCancelInProgress(cce);
- throw cce;
- }
- this.firstRecord = false;
- writeNewEntryBaseRecord(async);
- // Now we can finally call newOplogEntryId.
- // We need to make sure the create records
- // are written in the same order as they are created.
- // This allows us to not encode the oplogEntryId explicitly in the
- // record
- long createOplogEntryId = getOplogSet().newOplogEntryId();
- id.setKeyId(createOplogEntryId);
-
- // startPosForSynchOp = this.crf.currSize;
- // Allow it to be added to the OpLOg so increase the
- // size of currenstartPosForSynchOpt oplog
- int dataLength = getOpStateSize();
- // It is necessary that we set the
- // Oplog ID here without releasing the lock on object as we are
- // writing to the file after releasing the lock. This can cause
- // a situation where the
- // switching thread has added Oplog for compaction while the previous
- // thread has still not started writing. Thus compactor can
- // miss an entry as the oplog Id was not set till then.
- // This is because a compactor thread will iterate over the entries &
- // use only those which have OplogID equal to that of Oplog being
- // compacted without taking any lock. A lock is taken only if the
- // entry is a potential candidate.
- // Further the compactor may delete the file as a compactor thread does
- // not require to take any shared/exclusive lock at DiskStoreImpl
- // or Oplog level.
- // It is also assumed that compactor thread will take a lock on both
- // entry as well as DiskID while compacting. In case of synch
- // mode we can
- // safely set OplogID without taking lock on DiskId. But
- // for asynch mode
- // we have to take additional precaution as the asynch
- // writer of previous
- // oplog can interfere with the current oplog.
- id.setOplogId(getOplogId());
- // do the io while holding lock so that switch can set doneAppending
- // Write the data to the opLog for the synch mode
- startPosForSynchOp = writeOpLogBytes(this.crf, async, true);
- // if (this.crf.currSize != startPosForSynchOp) {
- // assert false;
- // }
- this.crf.currSize = temp;
- if (EntryBits.isNeedsValue(userBits)) {
- id.setValueLength(value.getLength());
+ getParent().getBackupLock().lock();
+ try {
+ synchronized (this.lock) { // TODO soplog perf analysis shows this as a
+ // contention point
+ // synchronized (this.crf) {
+ initOpState(OPLOG_NEW_ENTRY_0ID, dr, entry, value, userBits, false);
+ // Check if the current data in ByteBuffer will cause a
+ // potential increase in the size greater than the max allowed
+ long temp = (getOpStateSize() + this.crf.currSize);
+ if (!this.wroteNewEntryBase) {
+ temp += OPLOG_NEW_ENTRY_BASE_REC_SIZE;
+ }
+ if (this != getOplogSet().getChild()) {
+ useNextOplog = true;
+ } else if (temp > getMaxCrfSize() && !isFirstRecord()) {
+ switchOpLog(dr, getOpStateSize(), entry);
+ useNextOplog = true;
} else {
- id.setValueLength(0);
- }
- id.setUserBits(userBits);
+ if (this.lockedForKRFcreate) {
+ CacheClosedException cce = new CacheClosedException("The disk store is closed.");
+ dr.getCancelCriterion().checkCancelInProgress(cce);
+ throw cce;
+ }
+ this.firstRecord = false;
+ writeNewEntryBaseRecord(async);
+ // Now we can finally call newOplogEntryId.
+ // We need to make sure the create records
+ // are written in the same order as they are created.
+ // This allows us to not encode the oplogEntryId explicitly in the
+ // record
+ long createOplogEntryId = getOplogSet().newOplogEntryId();
+ id.setKeyId(createOplogEntryId);
+
+ // startPosForSynchOp = this.crf.currSize;
+ // Allow it to be added to the OpLOg so increase the
+ // size of currenstartPosForSynchOpt oplog
+ int dataLength = getOpStateSize();
+ // It is necessary that we set the
+ // Oplog ID here without releasing the lock on object as we are
+ // writing to the file after releasing the lock. This can cause
+ // a situation where the
+ // switching thread has added Oplog for compaction while the previous
+ // thread has still not started writing. Thus compactor can
+ // miss an entry as the oplog Id was not set till then.
+ // This is because a compactor thread will iterate over the entries &
+ // use only those which have OplogID equal to that of Oplog being
+ // compacted without taking any lock. A lock is taken only if the
+ // entry is a potential candidate.
+ // Further the compactor may delete the file as a compactor thread does
+ // not require to take any shared/exclusive lock at DiskStoreImpl
+ // or Oplog level.
+ // It is also assumed that compactor thread will take a lock on both
+ // entry as well as DiskID while compacting. In case of synch
+ // mode we can
+ // safely set OplogID without taking lock on DiskId. But
+ // for asynch mode
+ // we have to take additional precaution as the asynch
+ // writer of previous
+ // oplog can interfere with the current oplog.
+ id.setOplogId(getOplogId());
+ // do the io while holding lock so that switch can set doneAppending
+ // Write the data to the opLog for the synch mode
+ startPosForSynchOp = writeOpLogBytes(this.crf, async, true);
+ // if (this.crf.currSize != startPosForSynchOp) {
+ // assert false;
+ // }
+ this.crf.currSize = temp;
+ if (EntryBits.isNeedsValue(userBits)) {
+ id.setValueLength(value.getLength());
+ } else {
+ id.setValueLength(0);
+ }
+ id.setUserBits(userBits);
- if (logger.isTraceEnabled()) {
- logger.trace("Oplog::basicCreate:Release dByteBuffer with data for Disk ID = {}", id);
- }
- // As such for any put or get operation , a synch is taken
- // on the Entry object in the DiskEntry's Helper functions.
- // Compactor thread will also take a lock on entry object. Therefore
- // we do not require a lock on DiskID, as concurrent access for
- // value will not occur.
- startPosForSynchOp += getOpStateValueOffset();
- if (logger.isTraceEnabled(LogMarker.PERSIST_WRITES)) {
- VersionTag tag = null;
- if (entry.getVersionStamp() != null) {
- tag = entry.getVersionStamp().asVersionTag();
+ if (logger.isTraceEnabled()) {
+ logger.trace("Oplog::basicCreate:Release dByteBuffer with data for Disk ID = {}", id);
}
- logger.trace(LogMarker.PERSIST_WRITES,
- "basicCreate: id=<{}> key=<{}> valueOffset={} userBits={} valueLen={} valueBytes={} drId={} versionTag={} oplog#{}",
- abs(id.getKeyId()), entry.getKey(), startPosForSynchOp, userBits,
- (value != null ? value.getLength() : 0), value.getBytesAsString(), dr.getId(), tag,
- getOplogId());
- }
- id.setOffsetInOplog(startPosForSynchOp);
- addLive(dr, entry);
- // Size of the current oplog being increased
- // due to 'create' operation. Set the change in stats.
- this.dirHolder.incrementTotalOplogSize(dataLength);
- incTotalCount();
-
- // Update the region version vector for the disk store.
- // This needs to be done under lock so that we don't switch oplogs
- // unit the version vector accurately represents what is in this oplog
- RegionVersionVector rvv = dr.getRegionVersionVector();
- if (rvv != null && entry.getVersionStamp() != null) {
- rvv.recordVersion(entry.getVersionStamp().getMemberID(),
- entry.getVersionStamp().getRegionVersion());
- }
-
- EntryLogger.logPersistPut(dr.getName(), entry.getKey(), dr.getDiskStoreID());
+ // As such for any put or get operation , a synch is taken
+ // on the Entry object in the DiskEntry's Helper functions.
+ // Compactor thread will also take a lock on entry object. Therefore
+ // we do not require a lock on DiskID, as concurrent access for
+ // value will not occur.
+ startPosForSynchOp += getOpStateValueOffset();
+ if (logger.isTraceEnabled(LogMarker.PERSIST_WRITES)) {
+ VersionTag tag = null;
+ if (entry.getVersionStamp() != null) {
+ tag = entry.getVersionStamp().asVersionTag();
+ }
+ logger.trace(LogMarker.PERSIST_WRITES,
+ "basicCreate: id=<{}> key=<{}> valueOffset={} userBits={} valueLen={} valueBytes={} drId={} versionTag={} oplog#{}",
+ abs(id.getKeyId()), entry.getKey(), startPosForSynchOp, userBits,
+ (value != null ? value.getLength() : 0), value.getBytesAsString(), dr.getId(), tag,
+ getOplogId());
+ }
+ id.setOffsetInOplog(startPosForSynchOp);
+ addLive(dr, entry);
+ // Size of the current oplog being increased
+ // due to 'create' operation. Set the change in stats.
+ this.dirHolder.incrementTotalOplogSize(dataLength);
+ incTotalCount();
+
+ // Update the region version vector for the disk store.
+ // This needs to be done under lock so that we don't switch oplogs
+ // unit the version vector accurately represents what is in this oplog
+ RegionVersionVector rvv = dr.getRegionVersionVector();
+ if (rvv != null && entry.getVersionStamp() != null) {
+ rvv.recordVersion(entry.getVersionStamp().getMemberID(),
+ entry.getVersionStamp().getRegionVersion());
+ }
+
+ EntryLogger.logPersistPut(dr.getName(), entry.getKey(), dr.getDiskStoreID());
+ }
+ clearOpState();
+ // }
}
- clearOpState();
- // }
+ } finally {
+ getParent().getBackupLock().unlock();
}
if (useNextOplog) {
if (LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER) {
@@ -3700,10 +3713,15 @@ public class Oplog implements CompactableOplog, Flushable {
*/
void forceRolling(DiskRegion dr) {
if (getOplogSet().getChild() == this) {
- synchronized (this.lock) {
- if (getOplogSet().getChild() == this) {
- switchOpLog(dr, 0, null);
+ getParent().getBackupLock().lock();
+ try {
+ synchronized (this.lock) {
+ if (getOplogSet().getChild() == this) {
+ switchOpLog(dr, 0, null);
+ }
}
+ } finally {
+ getParent().getBackupLock().unlock();
}
if (LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER) {
CacheObserverHolder.getInstance().afterSwitchingOplog();
@@ -4129,6 +4147,8 @@ public class Oplog implements CompactableOplog, Flushable {
lockCompactor();
try {
+ // No need to get the backup lock prior to synchronizing (correct lock order) since the
+ // synchronized block does not attempt to get the backup lock (incorrect lock order)
synchronized (this.lock) {
// 42733: after set it to true, we will not reset it, since this oplog
// will be
@@ -4579,96 +4599,101 @@ public class Oplog implements CompactableOplog, Flushable {
System.out.println("basicModify KRF_DEBUG");
Thread.sleep(1000);
}
- synchronized (this.lock) {
- // synchronized (this.crf) {
- if (getOplogSet().getChild() != this) {
- useNextOplog = true;
- } else {
- initOpState(OPLOG_MOD_ENTRY_1ID, dr, entry, value, userBits, false);
- adjustment = getOpStateSize();
- assert adjustment > 0;
- long temp = (this.crf.currSize + adjustment);
- if (temp > getMaxCrfSize() && !isFirstRecord()) {
- switchOpLog(dr, adjustment, entry);
- // we can't reuse it since it contains variable length data
+ getParent().getBackupLock().lock();
+ try {
+ synchronized (this.lock) {
+ // synchronized (this.crf) {
+ if (getOplogSet().getChild() != this) {
useNextOplog = true;
} else {
- if (this.lockedForKRFcreate) {
- CacheClosedException cce = new CacheClosedException("The disk store is closed.");
- dr.getCancelCriterion().checkCancelInProgress(cce);
- throw cce;
- }
- this.firstRecord = false;
- long oldOplogId;
- // do the io while holding lock so that switch can set doneAppending
- // Write the data to the opLog for the synch mode
- startPosForSynchOp = writeOpLogBytes(this.crf, async, true);
- this.crf.currSize = temp;
- startPosForSynchOp += getOpStateValueOffset();
- if (logger.isTraceEnabled(LogMarker.PERSIST_WRITES)) {
- VersionTag tag = null;
- if (entry.getVersionStamp() != null) {
- tag = entry.getVersionStamp().asVersionTag();
- }
- logger.trace(LogMarker.PERSIST_WRITES,
- "basicModify: id=<{}> key=<{}> valueOffset={} userBits={} valueLen={} valueBytes=<{}> drId={} versionStamp={} oplog#{}",
- abs(id.getKeyId()), entry.getKey(), startPosForSynchOp, userBits, value.getLength(),
- value.getBytesAsString(), dr.getId(), tag, getOplogId());
- }
- if (EntryBits.isNeedsValue(userBits)) {
- id.setValueLength(value.getLength());
+ initOpState(OPLOG_MOD_ENTRY_1ID, dr, entry, value, userBits, false);
+ adjustment = getOpStateSize();
+ assert adjustment > 0;
+ long temp = (this.crf.currSize + adjustment);
+ if (temp > getMaxCrfSize() && !isFirstRecord()) {
+ switchOpLog(dr, adjustment, entry);
+ // we can't reuse it since it contains variable length data
+ useNextOplog = true;
} else {
- id.setValueLength(0);
- }
- id.setUserBits(userBits);
- if (logger.isTraceEnabled()) {
- logger.trace("Oplog::basicModify:Released ByteBuffer with data for Disk ID = {}", id);
- }
- synchronized (id) {
- // Need to do this while synced on id
- // now that we compact forward to most recent oplog.
- // @todo darrel: The sync logic in the disk code is so complex
- // a really doubt is is correct.
- // I think we need to do a fresh rewrite of it.
- oldOplogId = id.setOplogId(getOplogId());
- if (EntryBits.isAnyInvalid(userBits) || EntryBits.isTombstone(userBits)) {
- id.setOffsetInOplog(-1);
+ if (this.lockedForKRFcreate) {
+ CacheClosedException cce = new CacheClosedException("The disk store is closed.");
+ dr.getCancelCriterion().checkCancelInProgress(cce);
+ throw cce;
+ }
+ this.firstRecord = false;
+ long oldOplogId;
+ // do the io while holding lock so that switch can set doneAppending
+ // Write the data to the opLog for the synch mode
+ startPosForSynchOp = writeOpLogBytes(this.crf, async, true);
+ this.crf.currSize = temp;
+ startPosForSynchOp += getOpStateValueOffset();
+ if (logger.isTraceEnabled(LogMarker.PERSIST_WRITES)) {
+ VersionTag tag = null;
+ if (entry.getVersionStamp() != null) {
+ tag = entry.getVersionStamp().asVersionTag();
+ }
+ logger.trace(LogMarker.PERSIST_WRITES,
+ "basicModify: id=<{}> key=<{}> valueOffset={} userBits={} valueLen={} valueBytes=<{}> drId={} versionStamp={} oplog#{}",
+ abs(id.getKeyId()), entry.getKey(), startPosForSynchOp, userBits,
+ value.getLength(), value.getBytesAsString(), dr.getId(), tag, getOplogId());
+ }
+ if (EntryBits.isNeedsValue(userBits)) {
+ id.setValueLength(value.getLength());
} else {
- id.setOffsetInOplog(startPosForSynchOp);
+ id.setValueLength(0);
}
- }
- // Set the oplog size change for stats
- this.dirHolder.incrementTotalOplogSize(adjustment);
- this.incTotalCount();
-
- EntryLogger.logPersistPut(dr.getName(), entry.getKey(), dr.getDiskStoreID());
- if (oldOplogId != getOplogId()) {
- Oplog oldOplog = getOplogSet().getChild(oldOplogId);
- if (oldOplog != null) {
- oldOplog.rmLive(dr, entry);
- emptyOplog = oldOplog;
+ id.setUserBits(userBits);
+ if (logger.isTraceEnabled()) {
+ logger.trace("Oplog::basicModify:Released ByteBuffer with data for Disk ID = {}", id);
+ }
+ synchronized (id) {
+ // Need to do this while synced on id
+ // now that we compact forward to most recent oplog.
+ // @todo darrel: The sync logic in the disk code is so complex
+ // a really doubt is is correct.
+ // I think we need to do a fresh rewrite of it.
+ oldOplogId = id.setOplogId(getOplogId());
+ if (EntryBits.isAnyInvalid(userBits) || EntryBits.isTombstone(userBits)) {
+ id.setOffsetInOplog(-1);
+ } else {
+ id.setOffsetInOplog(startPosForSynchOp);
+ }
+ }
+ // Set the oplog size change for stats
+ this.dirHolder.incrementTotalOplogSize(adjustment);
+ this.incTotalCount();
+
+ EntryLogger.logPersistPut(dr.getName(), entry.getKey(), dr.getDiskStoreID());
+ if (oldOplogId != getOplogId()) {
+ Oplog oldOplog = getOplogSet().getChild(oldOplogId);
+ if (oldOplog != null) {
+ oldOplog.rmLive(dr, entry);
+ emptyOplog = oldOplog;
+ }
+ addLive(dr, entry);
+ // Note if this mod was done to oldOplog then this entry is already
+ // in
+ // the linked list. All we needed to do in this case is call
+ // incTotalCount
+ } else {
+ getOrCreateDRI(dr).update(entry);
}
- addLive(dr, entry);
- // Note if this mod was done to oldOplog then this entry is already
- // in
- // the linked list. All we needed to do in this case is call
- // incTotalCount
- } else {
- getOrCreateDRI(dr).update(entry);
- }
- // Update the region version vector for the disk store.
- // This needs to be done under lock so that we don't switch oplogs
- // unit the version vector accurately represents what is in this oplog
- RegionVersionVector rvv = dr.getRegionVersionVector();
- if (rvv != null && entry.getVersionStamp() != null) {
- rvv.recordVersion(entry.getVersionStamp().getMemberID(),
- entry.getVersionStamp().getRegionVersion());
+ // Update the region version vector for the disk store.
+ // This needs to be done under lock so that we don't switch oplogs
+ // unit the version vector accurately represents what is in this oplog
+ RegionVersionVector rvv = dr.getRegionVersionVector();
+ if (rvv != null && entry.getVersionStamp() != null) {
+ rvv.recordVersion(entry.getVersionStamp().getMemberID(),
+ entry.getVersionStamp().getRegionVersion());
+ }
}
+ clearOpState();
}
- clearOpState();
+ // }
}
- // }
+ } finally {
+ getParent().getBackupLock().unlock();
}
if (useNextOplog) {
if (LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER) {
@@ -4702,43 +4727,48 @@ public class Oplog implements CompactableOplog, Flushable {
throws IOException, InterruptedException {
boolean useNextOplog = false;
int adjustment = 0;
- synchronized (this.lock) {
- if (getOplogSet().getChild() != this) {
- useNextOplog = true;
- } else {
- this.opState.initialize(OPLOG_CONFLICT_VERSION, dr.getId(), tag);
- adjustment = getOpStateSize();
- assert adjustment > 0;
- long temp = (this.crf.currSize + adjustment);
- if (temp > getMaxCrfSize() && !isFirstRecord()) {
- switchOpLog(dr, adjustment, null);
- // we can't reuse it since it contains variable length data
+ getParent().getBackupLock().lock();
+ try {
+ synchronized (this.lock) {
+ if (getOplogSet().getChild() != this) {
useNextOplog = true;
} else {
- if (this.lockedForKRFcreate) {
- CacheClosedException cce = new CacheClosedException("The disk store is closed.");
- dr.getCancelCriterion().checkCancelInProgress(cce);
- throw cce;
- }
- this.firstRecord = false;
- writeOpLogBytes(this.crf, async, true);
- this.crf.currSize = temp;
- if (logger.isTraceEnabled(LogMarker.PERSIST_WRITES)) {
- logger.trace(LogMarker.PERSIST_WRITES,
- "basicSaveConflictVersionTag: drId={} versionStamp={} oplog#{}", dr.getId(), tag,
- getOplogId());
- }
- this.dirHolder.incrementTotalOplogSize(adjustment);
- // Update the region version vector for the disk store.
- // This needs to be done under lock so that we don't switch oplogs
- // unit the version vector accurately represents what is in this oplog
- RegionVersionVector rvv = dr.getRegionVersionVector();
- if (rvv != null && dr.getFlags().contains(DiskRegionFlag.IS_WITH_VERSIONING)) {
- rvv.recordVersion(tag.getMemberID(), tag.getRegionVersion());
+ this.opState.initialize(OPLOG_CONFLICT_VERSION, dr.getId(), tag);
+ adjustment = getOpStateSize();
+ assert adjustment > 0;
+ long temp = (this.crf.currSize + adjustment);
+ if (temp > getMaxCrfSize() && !isFirstRecord()) {
+ switchOpLog(dr, adjustment, null);
+ // we can't reuse it since it contains variable length data
+ useNextOplog = true;
+ } else {
+ if (this.lockedForKRFcreate) {
+ CacheClosedException cce = new CacheClosedException("The disk store is closed.");
+ dr.getCancelCriterion().checkCancelInProgress(cce);
+ throw cce;
+ }
+ this.firstRecord = false;
+ writeOpLogBytes(this.crf, async, true);
+ this.crf.currSize = temp;
+ if (logger.isTraceEnabled(LogMarker.PERSIST_WRITES)) {
+ logger.trace(LogMarker.PERSIST_WRITES,
+ "basicSaveConflictVersionTag: drId={} versionStamp={} oplog#{}", dr.getId(), tag,
+ getOplogId());
+ }
+ this.dirHolder.incrementTotalOplogSize(adjustment);
+ // Update the region version vector for the disk store.
+ // This needs to be done under lock so that we don't switch oplogs
+ // unit the version vector accurately represents what is in this oplog
+ RegionVersionVector rvv = dr.getRegionVersionVector();
+ if (rvv != null && dr.getFlags().contains(DiskRegionFlag.IS_WITH_VERSIONING)) {
+ rvv.recordVersion(tag.getMemberID(), tag.getRegionVersion());
+ }
}
+ clearOpState();
}
- clearOpState();
}
+ } finally {
+ getParent().getBackupLock().unlock();
}
if (useNextOplog) {
if (LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER) {
@@ -4755,6 +4785,7 @@ public class Oplog implements CompactableOplog, Flushable {
boolean useNextOplog = false;
long startPosForSynchOp = -1L;
int adjustment = 0;
+ // No need to get the backup lock since this is only for offline compaction
synchronized (this.lock) {
// synchronized (this.crf) {
if (getOplogSet().getChild() != this) {
@@ -4907,19 +4938,24 @@ public class Oplog implements CompactableOplog, Flushable {
*/
public void writeGCRVV(DiskRegion dr) {
boolean useNextOplog = false;
- synchronized (this.lock) {
- if (getOplogSet().getChild() != this) {
- useNextOplog = true;
- } else {
- try {
- writeRVVRecord(this.drf,
- Collections.<Long, AbstractDiskRegion>singletonMap(dr.getId(), dr), true);
- } catch (IOException ex) {
- dr.getCancelCriterion().checkCancelInProgress(ex);
- throw new DiskAccessException(LocalizedStrings.Oplog_FAILED_RECORDING_RVV_BECAUSE_OF_0
- .toLocalizedString(this.diskFile.getPath()), ex, dr.getName());
+ getParent().getBackupLock().lock();
+ try {
+ synchronized (this.lock) {
+ if (getOplogSet().getChild() != this) {
+ useNextOplog = true;
+ } else {
+ try {
+ writeRVVRecord(this.drf,
+ Collections.<Long, AbstractDiskRegion>singletonMap(dr.getId(), dr), true);
+ } catch (IOException ex) {
+ dr.getCancelCriterion().checkCancelInProgress(ex);
+ throw new DiskAccessException(LocalizedStrings.Oplog_FAILED_RECORDING_RVV_BECAUSE_OF_0
+ .toLocalizedString(this.diskFile.getPath()), ex, dr.getName());
+ }
}
}
+ } finally {
+ getParent().getBackupLock().unlock();
}
if (useNextOplog) {
getOplogSet().getChild().writeGCRVV(dr);
@@ -4936,36 +4972,41 @@ public class Oplog implements CompactableOplog, Flushable {
*/
public void writeRVV(DiskRegion dr, RegionVersionVector sourceRVV, Boolean isRVVTrusted) {
boolean useNextOplog = false;
- synchronized (this.lock) {
- if (getOplogSet().getChild() != this) {
- useNextOplog = true;
- } else {
+ getParent().getBackupLock().lock();
+ try {
+ synchronized (this.lock) {
+ if (getOplogSet().getChild() != this) {
+ useNextOplog = true;
+ } else {
- try {
- // We'll update the RVV of the disk region while holding the lock on
- // the oplog,
- // to make sure we don't switch oplogs while we're in the middle of
- // this.
- if (sourceRVV != null) {
- dr.getRegionVersionVector().recordVersions(sourceRVV);
- } else {
- // it's original EndGII, not to write duplicate rvv if its trusted
- if (dr.getRVVTrusted()) {
- return;
+ try {
+ // We'll update the RVV of the disk region while holding the lock on
+ // the oplog,
+ // to make sure we don't switch oplogs while we're in the middle of
+ // this.
+ if (sourceRVV != null) {
+ dr.getRegionVersionVector().recordVersions(sourceRVV);
+ } else {
+ // it's original EndGII, not to write duplicate rvv if its trusted
+ if (dr.getRVVTrusted()) {
+ return;
+ }
}
+ if (isRVVTrusted != null) {
+ // isRVVTrusted == null means "as is"
+ dr.setRVVTrusted(isRVVTrusted);
+ }
+ writeRVVRecord(this.crf,
+ Collections.<Long, AbstractDiskRegion>singletonMap(dr.getId(), dr), false);
+ } catch (IOException ex) {
+ dr.getCancelCriterion().checkCancelInProgress(ex);
+ throw new DiskAccessException(LocalizedStrings.Oplog_FAILED_RECORDING_RVV_BECAUSE_OF_0
+ .toLocalizedString(this.diskFile.getPath()), ex, dr.getName());
}
- if (isRVVTrusted != null) {
- // isRVVTrusted == null means "as is"
- dr.setRVVTrusted(isRVVTrusted);
- }
- writeRVVRecord(this.crf,
- Collections.<Long, AbstractDiskRegion>singletonMap(dr.getId(), dr), false);
- } catch (IOException ex) {
- dr.getCancelCriterion().checkCancelInProgress(ex);
- throw new DiskAccessException(LocalizedStrings.Oplog_FAILED_RECORDING_RVV_BECAUSE_OF_0
- .toLocalizedString(this.diskFile.getPath()), ex, dr.getName());
}
}
+ } finally {
+ getParent().getBackupLock().unlock();
}
if (useNextOplog) {
getOplogSet().getChild().writeRVV(dr, sourceRVV, isRVVTrusted);
@@ -5009,74 +5050,79 @@ public class Oplog implements CompactableOplog, Flushable {
System.out.println("basicRemove KRF_DEBUG");
Thread.sleep(1000);
}
- synchronized (this.lock) {
- if (getOplogSet().getChild() != this) {
- useNextOplog = true;
- } else if ((this.drf.currSize + MAX_DELETE_ENTRY_RECORD_BYTES) > getMaxDrfSize()
- && !isFirstRecord()) {
- switchOpLog(dr, MAX_DELETE_ENTRY_RECORD_BYTES, entry);
- useNextOplog = true;
- } else {
- if (this.lockedForKRFcreate) {
- CacheClosedException cce = new CacheClosedException("The disk store is closed.");
- dr.getCancelCriterion().checkCancelInProgress(cce);
- throw cce;
- }
- long oldOplogId = id.setOplogId(getOplogId());
- if (!isClear) {
- this.firstRecord = false;
- // Ok now we can go ahead and find out its actual size
- // This is the only place to set notToUseUserBits=true
- initOpState(OPLOG_DEL_ENTRY_1ID, dr, entry, null, (byte) 0, true);
- int adjustment = getOpStateSize();
-
- this.drf.currSize += adjustment;
- // do the io while holding lock so that switch can set doneAppending
- if (logger.isTraceEnabled()) {
- logger.trace(
- "Oplog::basicRemove: Recording the Deletion of entry in the Oplog with id = {} The Oplog Disk ID for the entry being deleted = {} Mode is Synch",
- getOplogId(), id);
+ getParent().getBackupLock().lock();
+ try {
+ synchronized (this.lock) {
+ if (getOplogSet().getChild() != this) {
+ useNextOplog = true;
+ } else if ((this.drf.currSize + MAX_DELETE_ENTRY_RECORD_BYTES) > getMaxDrfSize()
+ && !isFirstRecord()) {
+ switchOpLog(dr, MAX_DELETE_ENTRY_RECORD_BYTES, entry);
+ useNextOplog = true;
+ } else {
+ if (this.lockedForKRFcreate) {
+ CacheClosedException cce = new CacheClosedException("The disk store is closed.");
+ dr.getCancelCriterion().checkCancelInProgress(cce);
+ throw cce;
}
+ long oldOplogId = id.setOplogId(getOplogId());
+ if (!isClear) {
+ this.firstRecord = false;
+ // Ok now we can go ahead and find out its actual size
+ // This is the only place to set notToUseUserBits=true
+ initOpState(OPLOG_DEL_ENTRY_1ID, dr, entry, null, (byte) 0, true);
+ int adjustment = getOpStateSize();
+
+ this.drf.currSize += adjustment;
+ // do the io while holding lock so that switch can set doneAppending
+ if (logger.isTraceEnabled()) {
+ logger.trace(
+ "Oplog::basicRemove: Recording the Deletion of entry in the Oplog with id = {} The Oplog Disk ID for the entry being deleted = {} Mode is Synch",
+ getOplogId(), id);
+ }
- // Write the data to the opLog for the synch mode
- // TODO: if we don't sync write destroys what will happen if
- // we do 1. create k1 2. destroy k1 3. create k1?
- // It would be possible for the crf to be flushed but not the drf.
- // Then during recovery we will find identical keys with different
- // entryIds.
- // I think we can safely have drf writes be async as long as we flush
- // the drf
- // before we flush the crf.
- // However we can't have removes by async if we are doing a sync write
- // because we might be killed right after we do this write.
- startPosForSynchOp = writeOpLogBytes(this.drf, async, true);
- setHasDeletes(true);
- if (logger.isDebugEnabled(LogMarker.PERSIST_WRITES)) {
- logger.debug("basicRemove: id=<{}> key=<{}> drId={} oplog#{}", abs(id.getKeyId()),
- entry.getKey(), dr.getId(), getOplogId());
- }
+ // Write the data to the opLog for the synch mode
+ // TODO: if we don't sync write destroys what will happen if
+ // we do 1. create k1 2. destroy k1 3. create k1?
+ // It would be possible for the crf to be flushed but not the drf.
+ // Then during recovery we will find identical keys with different
+ // entryIds.
+ // I think we can safely have drf writes be async as long as we flush
+ // the drf
+ // before we flush the crf.
+ // However we can't have removes by async if we are doing a sync write
+ // because we might be killed right after we do this write.
+ startPosForSynchOp = writeOpLogBytes(this.drf, async, true);
+ setHasDeletes(true);
+ if (logger.isDebugEnabled(LogMarker.PERSIST_WRITES)) {
+ logger.debug("basicRemove: id=<{}> key=<{}> drId={} oplog#{}", abs(id.getKeyId()),
+ entry.getKey(), dr.getId(), getOplogId());
+ }
- if (logger.isTraceEnabled()) {
- logger.trace("Oplog::basicRemove:Released ByteBuffer for Disk ID = {}", id);
+ if (logger.isTraceEnabled()) {
+ logger.trace("Oplog::basicRemove:Released ByteBuffer for Disk ID = {}", id);
+ }
+ this.dirHolder.incrementTotalOplogSize(adjustment);
}
- this.dirHolder.incrementTotalOplogSize(adjustment);
- }
- // Set the oplog size change for stats
- id.setOffsetInOplog(-1);
+ // Set the oplog size change for stats
+ id.setOffsetInOplog(-1);
- EntryLogger.logPersistDestroy(dr.getName(), entry.getKey(), dr.getDiskStoreID());
- Oplog rmOplog = null;
- if (oldOplogId == getOplogId()) {
- rmOplog = this;
- } else {
- rmOplog = getOplogSet().getChild(oldOplogId);
- }
- if (rmOplog != null) {
- rmOplog.rmLive(dr, entry);
- emptyOplog = rmOplog;
+ EntryLogger.logPersistDestroy(dr.getName(), entry.getKey(), dr.getDiskStoreID());
+ Oplog rmOplog = null;
+ if (oldOplogId == getOplogId()) {
+ rmOplog = this;
+ } else {
+ rmOplog = getOplogSet().getChild(oldOplogId);
+ }
+ if (rmOplog != null) {
+ rmOplog.rmLive(dr, entry);
+ emptyOplog = rmOplog;
+ }
+ clearOpState();
}
- clearOpState();
}
+ } finally {
+ getParent().getBackupLock().unlock();
}
if (useNextOplog) {
if (LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER) {
@@ -5130,6 +5176,8 @@ public class Oplog implements CompactableOplog, Flushable {
private void flush(OplogFile olf, boolean doSync) throws IOException {
try {
+ // No need to get the backup lock prior to synchronizing (correct lock order) since the
+ // synchronized block does not attempt to get the backup lock (incorrect lock order)
synchronized (this.lock/* olf */) {
if (olf.RAFClosed) {
return;
@@ -5187,6 +5235,8 @@ public class Oplog implements CompactableOplog, Flushable {
private void flush(OplogFile olf, ByteBuffer b1, ByteBuffer b2) throws IOException {
try {
+ // No need to get the backup lock prior to synchronizing (correct lock order) since the
+ // synchronized block does not attempt to get the backup lock (incorrect lock order)
synchronized (this.lock/* olf */) {
if (olf.RAFClosed) {
return;
@@ -5255,34 +5305,40 @@ public class Oplog implements CompactableOplog, Flushable {
private long writeOpLogBytes(OplogFile olf, boolean async, boolean doFlushIfSync)
throws IOException {
long startPos = -1L;
- synchronized (this.lock/* olf */) {
- Assert.assertTrue(!this.doneAppending);
- if (this.closed) {
- Assert.assertTrue(false, "The Oplog " + this.oplogId + " for store " + getParent().getName()
- + " has been closed for synch mode while writing is going on. This should not happen");
- }
- // It is assumed that the file pointer is already at the
- // appropriate position in the file so as to allow writing at the end.
- // Any fault in operations will set the pointer back to the write
- // location.
- // Also it is only in case of synch writing, we are writing more
- // than what is actually needed, we will have to reset the pointer.
- // Also need to add in offset in writeBuf in case we are not flushing
- // writeBuf
- startPos = olf.channel.position() + olf.writeBuf.position();
- // Assert.assertTrue(startPos > lastWritePos,
- // "startPos=" + startPos +
- // " was not > lastWritePos=" + lastWritePos);
- long bytesWritten = this.opState.write(olf);
- if (!async && doFlushIfSync) {
- flushAndSync(olf);
- }
- getStats().incWrittenBytes(bytesWritten, async);
-
- // // Moved the set of lastWritePos to after write
- // // so if write throws an exception it will not be updated.
- // // This fixes bug 40449.
- // this.lastWritePos = startPos;
+ getParent().getBackupLock().lock();
+ try {
+ synchronized (this.lock/* olf */) {
+ Assert.assertTrue(!this.doneAppending);
+ if (this.closed) {
+ Assert.assertTrue(false, "The Oplog " + this.oplogId + " for store "
+ + getParent().getName()
+ + " has been closed for synch mode while writing is going on. This should not happen");
+ }
+ // It is assumed that the file pointer is already at the
+ // appropriate position in the file so as to allow writing at the end.
+ // Any fault in operations will set the pointer back to the write
+ // location.
+ // Also it is only in case of synch writing, we are writing more
+ // than what is actually needed, we will have to reset the pointer.
+ // Also need to add in offset in writeBuf in case we are not flushing
+ // writeBuf
+ startPos = olf.channel.position() + olf.writeBuf.position();
+ // Assert.assertTrue(startPos > lastWritePos,
+ // "startPos=" + startPos +
+ // " was not > lastWritePos=" + lastWritePos);
+ long bytesWritten = this.opState.write(olf);
+ if (!async && doFlushIfSync) {
+ flushAndSync(olf);
+ }
+ getStats().incWrittenBytes(bytesWritten, async);
+
+ // // Moved the set of lastWritePos to after write
+ // // so if write throws an exception it will not be updated.
+ // // This fixes bug 40449.
+ // this.lastWritePos = startPos;
+ }
+ } finally {
+ getParent().getBackupLock().unlock();
}
return startPos;
}
@@ -5296,6 +5352,8 @@ public class Oplog implements CompactableOplog, Flushable {
boolean closeRAF() {
if (this.beingRead)
return false;
+ // No need to get the backup lock prior to synchronizing (correct lock order) since the
+ // synchronized block does not attempt to get the backup lock (incorrect lock order)
synchronized (this.lock/* crf */) {
if (this.beingRead)
return false;
@@ -5325,6 +5383,8 @@ public class Oplog implements CompactableOplog, Flushable {
* @return true if oplog file is open and can be read from; false if not
*/
private boolean reopenFileIfClosed() throws IOException {
+ // No need to get the backup lock prior to synchronizing (correct lock order) since the
+ // synchronized block does not attempt to get the backup lock (incorrect lock order)
synchronized (this.lock/* crf */) {
boolean result = !this.crf.RAFClosed;
if (!result && this.okToReopen) {
@@ -5343,6 +5403,8 @@ public class Oplog implements CompactableOplog, Flushable {
boolean didReopen = false;
boolean accessedInactive = false;
try {
+ // No need to get the backup lock prior to synchronizing (correct lock order) since the
+ // synchronized block does not attempt to get the backup lock (incorrect lock order)
synchronized (this.lock/* crf */) {
// if (this.closed || this.deleted.get()) {
// throw new DiskAccessException("attempting get on "
@@ -5523,6 +5585,8 @@ public class Oplog implements CompactableOplog, Flushable {
}
} else {
try {
+ // No need to get the backup lock prior to synchronizing (correct lock order) since the
+ // synchronized block does not attempt to get the backup lock (incorrect lock order)
synchronized (this.lock/* crf */) {
if (/*
* !getParent().isSync() since compactor groups writes &&
@@ -5708,6 +5772,8 @@ public class Oplog implements CompactableOplog, Flushable {
}
private void deleteFile(final OplogFile olf) {
+ // No need to get the backup lock prior to synchronizing (correct lock order) since the
+ // synchronized block does not attempt to get the backup lock (incorrect lock order)
synchronized (this.lock) {
if (olf.currSize != 0) {
this.dirHolder.decrementTotalOplogSize(olf.currSize);
@@ -5799,6 +5865,8 @@ public class Oplog implements CompactableOplog, Flushable {
}
private void finishedAppending() {
+ // No need to get the backup lock prior to synchronizing (correct lock order) since the
+ // synchronized block does not attempt to get the backup lock (incorrect lock order)
synchronized (this.lock/* crf */) {
this.doneAppending = true;
}