mirror of
https://github.com/OpenHFT/Chronicle-Map.git
synced 2021-09-13 01:26:12 +00:00
tidy.
This commit is contained in:
@@ -39,6 +39,8 @@ import net.openhft.chronicle.hash.serialization.impl.TypedMarshallableReaderWrit
|
||||
import net.openhft.chronicle.map.internal.AnalyticsHolder;
|
||||
import net.openhft.chronicle.map.replication.MapRemoteOperations;
|
||||
import net.openhft.chronicle.set.ChronicleSetBuilder;
|
||||
import net.openhft.chronicle.threads.Pauser;
|
||||
import net.openhft.chronicle.threads.TimingPauser;
|
||||
import net.openhft.chronicle.values.ValueModel;
|
||||
import net.openhft.chronicle.values.Values;
|
||||
import net.openhft.chronicle.wire.Marshallable;
|
||||
@@ -58,6 +60,7 @@ import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.TimeoutException;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
import java.util.concurrent.atomic.AtomicReference;
|
||||
|
||||
@@ -1703,25 +1706,31 @@ public final class ChronicleMapBuilder<K, V> implements
|
||||
final AtomicBoolean newFile = new AtomicBoolean();
|
||||
final FileChannel fileChannel = raf.getChannel();
|
||||
|
||||
while (true) {
|
||||
if (raf.length() == 0) {
|
||||
final boolean locked = FileLockUtil.tryRunExclusively(canonicalFile, fileChannel, () -> {
|
||||
if (raf.length() == 0) {
|
||||
map.set(newMap());
|
||||
headerBuffer.set(writeHeader(fileChannel, map.get()));
|
||||
newFile.set(true);
|
||||
} else {
|
||||
newFile.set(false);
|
||||
}
|
||||
});
|
||||
TimingPauser pauser = Pauser.balanced();
|
||||
|
||||
if (locked)
|
||||
break;
|
||||
else
|
||||
Jvm.pause(10);
|
||||
}
|
||||
else
|
||||
while (raf.length() == 0) {
|
||||
final boolean locked = FileLockUtil.tryRunExclusively(canonicalFile, fileChannel, () -> {
|
||||
if (raf.length() == 0) {
|
||||
map.set(newMap());
|
||||
headerBuffer.set(writeHeader(fileChannel, map.get()));
|
||||
newFile.set(true);
|
||||
} else {
|
||||
newFile.set(false);
|
||||
}
|
||||
});
|
||||
|
||||
if (locked)
|
||||
break;
|
||||
else {
|
||||
try {
|
||||
pauser.pause(10, TimeUnit.SECONDS);
|
||||
}
|
||||
catch (TimeoutException e) {
|
||||
LOG.warn("Failed to write header: can't acquire exclusive file lock on empty file for 10 seconds", e);
|
||||
|
||||
Jvm.rethrow(e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if (newFile.get()) {
|
||||
|
||||
@@ -10,6 +10,7 @@ import java.io.File;
|
||||
import java.io.IOException;
|
||||
import java.nio.channels.FileChannel;
|
||||
import java.nio.channels.FileLock;
|
||||
import java.nio.channels.OverlappingFileLockException;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
|
||||
@@ -91,10 +92,11 @@ public final class FileLockUtil {
|
||||
}
|
||||
|
||||
/**
|
||||
* Executes a closure under exclusive file lock.
|
||||
* Tries to execute a closure under exclusive file lock.
|
||||
* If USE_LOCKING is false, provides synchronization only within local JVM.
|
||||
*
|
||||
* @param fileIOAction Closure to run, can throw {@link IOException}s.
|
||||
* @return <code>true</code> if the lock was successfully acquired and IO action was executed, <code>false</code> otherwise.
|
||||
*/
|
||||
public static boolean tryRunExclusively(@NotNull final File canonicalFile,
|
||||
@NotNull final FileChannel fileChannel,
|
||||
@@ -115,6 +117,9 @@ public final class FileLockUtil {
|
||||
|
||||
locked.set(true);
|
||||
}
|
||||
catch (OverlappingFileLockException ignored) {
|
||||
// File lock is being held by this JVM, unsuccessful attempt.
|
||||
}
|
||||
} else {
|
||||
fileIOAction.fileIOAction();
|
||||
|
||||
|
||||
Reference in New Issue
Block a user