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

HCOLL-397 msync before pointing to newly created tier bulk in the header

Browse Source
This commit is contained in:
leventov
2016-01-31 21:16:42 +07:00
parent 8891626bb2
commit 299cd8341c
4 changed files with 170 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/hash/impl/PosixMsync.java
+42
View File
@@ -0,0 +1,42 @@
/*
* 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;
import com.sun.jna.Native;
import com.sun.jna.Platform;
import java.io.IOException;
final class PosixMsync {
private static final int MS_SYNC = 4;
public static void msync(long addr, long length) throws IOException {
if (msync(addr, length, MS_SYNC) == -1)
throw new IOException(strerror_r(Native.getLastError(), null, 0));
}
private static native String strerror_r(int errnum, char[] buf, int buflen);
private static native int msync(long addr, long length, int flags);
static {
Native.register(PosixMsync.class, Platform.C_LIBRARY_NAME);
}
private PosixMsync() {}
}
src/main/java/net/openhft/chronicle/hash/impl/VanillaChronicleHash.java
+55 -13
View File
@@ -731,10 +731,24 @@ 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 {
int allocatedExtraTierBulks = globalMutableState.getAllocatedExtraTierBulks();
mapTierBulks(allocatedExtraTierBulks);
mapTierBulks(raf, allocatedExtraTierBulks);
// integer overflow aware
long firstTierIndex = extraTierIndexToTierIndex(allocatedExtraTierBulks * tiersInBulk);
BytesStore tierBytesStore = tierBytesStore(firstTierIndex);
long firstTierOffset = tierBytesOffset(firstTierIndex);
@@ -746,13 +760,30 @@ public abstract class VanillaChronicleHash<K,
long lastTierIndex = firstTierIndex + tiersInBulk - 1;
linkAndZeroOutFreeTiers(firstTierIndex, lastTierIndex);
// TODO HCOLL-397 insert msync here!
if (persisted()) {
long address = tierBytesStore.address(firstTierOffset - tierBulkInnerOffsetToTiers);
// address should be a multiple of page size
if (OS.pageAlign(address) != address) {
address = OS.pageAlign(address) - OS.pageSize();
}
long endAddress = tierBytesStore.address(tierBytesOffset(lastTierIndex)) + tierSize;
long length = endAddress - address;
msync(raf, address, length);
}
// after we are sure the new bulk is initialized, update the global mutable state
globalMutableState.setAllocatedExtraTierBulks(allocatedExtraTierBulks + 1);
globalMutableState.setFirstFreeTierIndex(firstTierIndex);
}
private static void msync(RandomAccessFile raf, long address, long length) throws IOException {
if (OS.isWindows()) {
WindowsMsync.msync(raf, address, length);
} else {
PosixMsync.msync(address, length);
}
}
public void linkAndZeroOutFreeTiers(long firstTierIndex, long lastTierIndex) {
for (long tierIndex = firstTierIndex; tierIndex <= lastTierIndex; tierIndex++) {
long tierOffset = tierBytesOffset(tierIndex);
@@ -822,9 +853,23 @@ 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(upToBulkIndex);
mapTierBulksMapped(raf, upToBulkIndex);
} catch (IOException e) {
throw new RuntimeException(e);
}
@@ -834,7 +879,7 @@ public abstract class VanillaChronicleHash<K,
}
}
private void mapTierBulksMapped(int upToBulkIndex) throws IOException {
private void mapTierBulksMapped(RandomAccessFile raf, int upToBulkIndex) throws IOException {
int firstBulkToMapIndex = tierBulkOffsets.size();
int bulksToMap = upToBulkIndex + 1 - firstBulkToMapIndex;
long mapSize = bulksToMap * tierBulkSizeInBytes;
@@ -852,14 +897,11 @@ public abstract class VanillaChronicleHash<K,
// Now might need to have bigger mapSize
mapSize += firstBulkToMapOffsetWithinMapping;
}
// TODO optimize -- create a file channel not on each tier bulk creation
try (RandomAccessFile raf = new RandomAccessFile(file, "rw")) {
// 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);
appendBulkData(firstBulkToMapIndex, upToBulkIndex, extraStore,
firstBulkToMapOffsetWithinMapping);
}
// 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);
appendBulkData(firstBulkToMapIndex, upToBulkIndex, extraStore,
firstBulkToMapOffsetWithinMapping);
}
/**
src/main/java/net/openhft/chronicle/hash/impl/WindowsMsync.java
+61
View File
@@ -0,0 +1,61 @@
/*
* 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;
import com.sun.jna.Native;
import com.sun.jna.platform.win32.Kernel32Util;
import java.io.IOException;
import java.io.RandomAccessFile;
import static com.sun.jna.platform.win32.WinError.ERROR_LOCK_VIOLATION;
final class WindowsMsync {
public static void msync(RandomAccessFile raf, long addr, long length)
throws IOException {
int retry = 0;
boolean success;
int lastError = 0;
// FlushViewOfFile can fail with ERROR_LOCK_VIOLATION if the memory system is writing dirty
// pages to disk. As there is no way to synchronize the flushing then we retry a limited
// number of times.
do {
if ((success = FlushViewOfFile(addr, length)) ||
(lastError = GetLastError()) != ERROR_LOCK_VIOLATION)
break;
retry++;
} while (retry < 3);
if (success) {
// Finally calls FlushFileBuffers
raf.getChannel().force(false);
} else {
throw new IOException(Kernel32Util.formatMessage(lastError));
}
}
private static native boolean FlushViewOfFile(long addr, long length);
private static native int GetLastError();
static {
Native.register(WindowsMsync.class, "kernel32");
}
private WindowsMsync() {}
}