Alawn/Chronicle-Map
1
0
Fork 0
mirror of https://github.com/OpenHFT/Chronicle-Map.git synced 2021-09-13 01:26:12 +00:00
Code Releases Activity

Use canonical RandomAccessFiles (=> FileChannels, too) to guarantee fileLocks are not released during critical operation, because some concurrent thread closes fileChannel on different instance of a map

Browse Source
This commit is contained in:
leventov
2016-02-07 19:59:47 +07:00
parent 64cf485b92
commit 00819bff66
3 changed files with 134 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/hash/impl/VanillaChronicleHash.java
+50 -44
View File
@@ -24,6 +24,7 @@ import net.openhft.chronicle.core.Maths;
import net.openhft.chronicle.core.OS;
import net.openhft.chronicle.hash.*;
import net.openhft.chronicle.hash.impl.util.BuildVersion;
import net.openhft.chronicle.hash.impl.util.CanonicalRandomAccessFiles;
import net.openhft.chronicle.hash.impl.util.jna.PosixMsync;
import net.openhft.chronicle.hash.impl.util.jna.WindowsMsync;
import net.openhft.chronicle.hash.serialization.DataAccess;
@@ -38,6 +39,7 @@ import net.openhft.chronicle.wire.WireOut;
import org.jetbrains.annotations.NotNull;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import sun.misc.Cleaner;
import java.io.File;
import java.io.IOException;
@@ -126,6 +128,8 @@ public abstract class VanillaChronicleHash<K,
/////////////////////////////////////////////////
// Bytes Store (essentially, the base address) and serialization-dependent offsets
private transient File file;
private transient RandomAccessFile raf;
private transient Cleaner releaseRaf;
protected transient BytesStore bs;
public static class TierBulkData {
@@ -394,6 +398,7 @@ public abstract class VanillaChronicleHash<K,
}
}
globalMutableStateBuffer.flip();
//noinspection unchecked
globalMutableState.bytesStore(BytesStore.wrap(globalMutableStateBuffer), 0,
globalMutableState.maxSize());
}
@@ -436,30 +441,54 @@ public abstract class VanillaChronicleHash<K,
bytesStore.start() + ", 0 expected");
}
this.bs = bytesStore;
//noinspection unchecked
globalMutableState.bytesStore(bs, headerSize + GLOBAL_MUTABLE_STATE_VALUE_OFFSET,
globalMutableState.maxSize());
onHeaderCreated();
}
private void setFile(File file, RandomAccessFile raf) {
this.file = file;
this.raf = raf;
this.releaseRaf = Cleaner.create(this, new RafRelease(file));
}
private static class RafRelease implements Runnable {
final File file;
private RafRelease(File file) {
this.file = file;
}
@Override
public void run() {
try {
CanonicalRandomAccessFiles.release(file);
} catch (IOException e) {
LOG.error("error on releasing RAF for file {}: {}", file, e);
}
}
}
public final void createMappedStoreAndSegments(File file, RandomAccessFile raf)
throws IOException {
// TODO this method had been moved -- not clear where
//OS.warnOnWindows(sizeInBytesWithoutTiers());
this.file = file;
setFile(file, raf);
long mapSize = expectedFileSize();
createMappedStoreAndSegments(map(raf, mapSize, 0));
createMappedStoreAndSegments(map(mapSize, 0));
}
public final void basicRecover(File file, RandomAccessFile raf) throws IOException {
this.file = file;
setFile(file, raf);
long segmentHeadersOffset = computeSegmentHeadersOffset();
long sizeInBytesWithoutTiers = computeSizeInBytesWithoutTiers(segmentHeadersOffset);
long sizeBeyondSegments = Math.max(raf.length() - sizeInBytesWithoutTiers, 0);
int allocatedExtraTierBulks = (int) (sizeBeyondSegments / tierBulkSizeInBytes);
long mapSize = pageAlign(sizeInBytesWithoutTiers +
allocatedExtraTierBulks * tierBulkSizeInBytes);
initBytesStoreAndHeadersViews(map(raf, mapSize, 0));
initBytesStoreAndHeadersViews(map(mapSize, 0));
resetGlobalMutableStateLock(file);
recoverAllocatedExtraTierBulks(file, allocatedExtraTierBulks);
@@ -594,6 +623,8 @@ public abstract class VanillaChronicleHash<K,
assert bulkData.bytesStore.refCount() == 0;
}
}
if (releaseRaf != null)
releaseRaf.clean();
}
@Override
@@ -724,7 +755,11 @@ public abstract class VanillaChronicleHash<K,
globalMutableState.setFirstFreeTierIndex(nextFreeTierIndex);
return firstFreeTierIndex;
} else {
allocateTierBulk();
try {
allocateTierBulk();
} catch (IOException e) {
throw new RuntimeException(e);
}
}
}
} finally {
@@ -732,24 +767,10 @@ public abstract class VanillaChronicleHash<K,
}
}
private void allocateTierBulk() {
try {
RandomAccessFile raf = persisted() ? new RandomAccessFile(file, "rw") : null;
try {
allocateTierBulk0(raf);
} finally {
if (raf != null)
raf.close();
}
} catch (IOException e) {
throw new RuntimeException(e);
}
}
private void allocateTierBulk0(RandomAccessFile raf) throws IOException {
private void allocateTierBulk() throws IOException {
int allocatedExtraTierBulks = globalMutableState.getAllocatedExtraTierBulks();
mapTierBulks(raf, allocatedExtraTierBulks);
mapTierBulks(allocatedExtraTierBulks);
long firstTierIndex = extraTierIndexToTierIndex(allocatedExtraTierBulks * tiersInBulk);
BytesStore tierBytesStore = tierBytesStore(firstTierIndex);
@@ -767,7 +788,7 @@ public abstract class VanillaChronicleHash<K,
long address = tierBytesStore.address(firstTierOffset - tierBulkInnerOffsetToTiers);
long endAddress = tierBytesStore.address(tierBytesOffset(lastTierIndex)) + tierSize;
long length = endAddress - address;
msync(raf, address, length);
msync(address, length);
}
// after we are sure the new bulk is initialized, update the global mutable state
@@ -775,11 +796,11 @@ public abstract class VanillaChronicleHash<K,
globalMutableState.setFirstFreeTierIndex(firstTierIndex);
}
public void msync(RandomAccessFile raf) throws IOException {
msync(raf, bsAddress(), bs.capacity());
public void msync() throws IOException {
msync(bsAddress(), bs.capacity());
}
private static void msync(RandomAccessFile raf, long address, long length) throws IOException {
private void msync(long address, long length) throws IOException {
// address should be a multiple of page size
if (OS.pageAlign(address) != address) {
long oldAddress = address;
@@ -862,23 +883,9 @@ public abstract class VanillaChronicleHash<K,
}
private void mapTierBulks(int upToBulkIndex) {
try {
RandomAccessFile raf = persisted() ? new RandomAccessFile(file, "rw") : null;
try {
mapTierBulks(raf, upToBulkIndex);
} finally {
if (raf != null)
raf.close();
}
} catch (IOException e) {
throw new RuntimeException(e);
}
}
private void mapTierBulks(RandomAccessFile raf, int upToBulkIndex) {
if (persisted()) {
try {
mapTierBulksMapped(raf, upToBulkIndex);
mapTierBulksMapped(upToBulkIndex);
} catch (IOException e) {
throw new RuntimeException(e);
}
@@ -888,7 +895,7 @@ public abstract class VanillaChronicleHash<K,
}
}
private void mapTierBulksMapped(RandomAccessFile raf, int upToBulkIndex) throws IOException {
private void mapTierBulksMapped(int upToBulkIndex) throws IOException {
int firstBulkToMapIndex = tierBulkOffsets.size();
int bulksToMap = upToBulkIndex + 1 - firstBulkToMapIndex;
long mapSize = bulksToMap * tierBulkSizeInBytes;
@@ -908,7 +915,7 @@ public abstract class VanillaChronicleHash<K,
}
// mapping by hand, because MappedFile/MappedBytesStore doesn't allow to create a BS
// which starts not from the beginning of the file, but has start() of 0
NativeBytesStore extraStore = map(raf, mapSize, mappingOffsetInFile);
NativeBytesStore extraStore = map(mapSize, mappingOffsetInFile);
appendBulkData(firstBulkToMapIndex, upToBulkIndex, extraStore,
firstBulkToMapOffsetWithinMapping);
}
@@ -916,8 +923,7 @@ public abstract class VanillaChronicleHash<K,
/**
* @see net.openhft.chronicle.bytes.MappedFile#acquireByteStore(long, MappedBytesStoreFactory)
*/
private static NativeBytesStore map(
RandomAccessFile raf, long mapSize, long mappingOffsetInFile) throws IOException {
private NativeBytesStore map(long mapSize, long mappingOffsetInFile) throws IOException {
long minFileSize = mappingOffsetInFile + mapSize;
FileChannel fileChannel = raf.getChannel();
if (fileChannel.size() < minFileSize) {
src/main/java/net/openhft/chronicle/hash/impl/util/CanonicalRandomAccessFiles.java
+73
View File
@@ -0,0 +1,73 @@
/*
* Copyright (C) 2015 higherfrequencytrading.com
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU Lesser General Public License as published by
* the Free Software Foundation, either version 3 of the License.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU Lesser General Public License for more details.
*
* You should have received a copy of the GNU Lesser General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
package net.openhft.chronicle.hash.impl.util;
import net.openhft.chronicle.core.Jvm;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.util.concurrent.ConcurrentHashMap;
public final class CanonicalRandomAccessFiles {
private static class RafReference {
RandomAccessFile raf;
int refCount;
RafReference(RandomAccessFile raf) {
this.raf = raf;
refCount = 1;
}
}
private static final ConcurrentHashMap<File, RafReference> canonicalRafs =
new ConcurrentHashMap<>();
public static RandomAccessFile acquire(File file) throws FileNotFoundException {
return canonicalRafs.compute(file, (f, ref) -> {
if (ref == null) {
try {
return new RafReference(new RandomAccessFile(f, "rw"));
} catch (FileNotFoundException e) {
throw Jvm.rethrow(e);
}
} else {
ref.refCount++;
return ref;
}
}).raf;
}
public static void release(File file) throws IOException {
canonicalRafs.compute(file, (f, ref) -> {
if (ref == null)
throw new IllegalStateException("releasing not referenced RAF of file " + f);
if (--ref.refCount == 0) {
try {
ref.raf.close();
} catch (IOException e) {
throw Jvm.rethrow(e);
}
return null;
} else {
return ref;
}
});
}
}
src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java
+11 -5
View File
@@ -29,6 +29,7 @@ import net.openhft.chronicle.hash.impl.CompactOffHeapLinearHashTable;
import net.openhft.chronicle.hash.impl.SizePrefixedBlob;
import net.openhft.chronicle.hash.impl.VanillaChronicleHash;
import net.openhft.chronicle.hash.impl.stage.entry.ChecksumStrategy;
import net.openhft.chronicle.hash.impl.util.CanonicalRandomAccessFiles;
import net.openhft.chronicle.hash.impl.util.math.PoissonDistribution;
import net.openhft.chronicle.hash.serialization.*;
import net.openhft.chronicle.hash.serialization.impl.SerializationBuilder;
@@ -1430,9 +1431,11 @@ public final class ChronicleMapBuilder<K, V> implements
//noinspection ResultOfMethodCallIgnored
file.createNewFile();
}
try (RandomAccessFile raf = new RandomAccessFile(file, "rw")) {
ChronicleMap<K, V> result = null;
RandomAccessFile raf = CanonicalRandomAccessFiles.acquire(file);
try {
if (raf.length() > 0)
return openWithExistingFile(file, raf, recover, overrideBuilderConfig);
return result = openWithExistingFile(file, raf, recover, overrideBuilderConfig);
@SuppressWarnings("unchecked")
VanillaChronicleMap<K, V, ?>[] map = new VanillaChronicleMap[1];
@@ -1452,10 +1455,13 @@ public final class ChronicleMapBuilder<K, V> implements
if (newFile[0]) {
int headerSize = headerBuffer[0].remaining();
return createWithNewFile(map[0], file, raf, headerBuffer[0], headerSize);
return result = createWithNewFile(map[0], file, raf, headerBuffer[0], headerSize);
} else {
return openWithExistingFile(file, raf, recover, overrideBuilderConfig);
return result = openWithExistingFile(file, raf, recover, overrideBuilderConfig);
}
} finally {
if (result == null)
CanonicalRandomAccessFiles.release(file);
}
}
@@ -1668,7 +1674,7 @@ public final class ChronicleMapBuilder<K, V> implements
throws IOException {
FileChannel fileChannel = raf.getChannel();
// see HCOLL-396
map.msync(raf);
map.msync();
//noinspection PointlessBitwiseExpression
headerBuffer.putInt(SIZE_WORD_OFFSET, READY | DATA | headerSize);