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

Improve FileChannel read/write for persisted ChronicleMaps

Browse Source
This commit is contained in:
leventov
2015-11-26 18:49:54 +03:00
parent e29a473e05
commit 4faefe8ca9
2 changed files with 25 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/hash/impl/VanillaChronicleHash.java
+10 -5
View File
@@ -376,15 +376,20 @@ public abstract class VanillaChronicleHash<K,
}
}
public final void initBeforeMapping(FileChannel ch) throws IOException {
headerSize = roundUpMapHeaderSize(ch.position());
public final void initBeforeMapping(FileChannel ch, long headerSize) throws IOException {
this.headerSize = roundUpMapHeaderSize(headerSize);
if (!createdOrInMemory) {
// This block is for reading segmentHeadersOffset before main mapping
// After the mapping globalMutableState value's bytes are reassigned
ch.position(headerSize + GLOBAL_MUTABLE_STATE_VALUE_OFFSET);
ByteBuffer globalMutableStateBuffer =
ByteBuffer.allocate((int) globalMutableState.maxSize());
ch.read(globalMutableStateBuffer);
while (globalMutableStateBuffer.remaining() > 0) {
if (ch.read(globalMutableStateBuffer,
this.headerSize + GLOBAL_MUTABLE_STATE_VALUE_OFFSET +
globalMutableStateBuffer.position()) == -1) {
throw new RuntimeException();
}
}
globalMutableStateBuffer.flip();
globalMutableState.bytesStore(BytesStore.wrap(globalMutableStateBuffer), 0,
globalMutableState.maxSize());
@@ -814,7 +819,7 @@ public abstract class VanillaChronicleHash<K,
FileChannel fileChannel = raf.getChannel();
if (fileChannel.size() < minFileSize) {
// handle a possible race condition between processes.
try (FileLock lock = fileChannel.lock()) {
try (FileLock ignored = fileChannel.lock()) {
if (fileChannel.size() < minFileSize) {
raf.setLength(minFileSize);
}
src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java
+15 -6
View File
@@ -42,6 +42,7 @@ import org.slf4j.LoggerFactory;
import java.io.*;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.channels.FileLock;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@@ -1375,7 +1376,11 @@ public final class ChronicleMapBuilder<K, V> implements
try (RandomAccessFile raf = new RandomAccessFile(file, "rw")) {
FileChannel fileChannel = raf.getChannel();
ByteBuffer headerBuffer = ByteBuffer.allocate(MAX_BOOTSTRAPPING_HEADER_SIZE);
fileChannel.read(headerBuffer);
while (headerBuffer.remaining() > 0 &&
headerBuffer.position() < fileChannel.size()) {
if (fileChannel.read(headerBuffer, headerBuffer.position()) == -1)
break;
}
headerBuffer.flip();
Bytes<ByteBuffer> headerBytes = Bytes.wrapForRead(headerBuffer);
headerBytes.readLimit(headerBuffer.limit());
@@ -1384,8 +1389,7 @@ public final class ChronicleMapBuilder<K, V> implements
map.initTransientsFromBuilder(this);
initTransientsFromReplication(map, singleHashReplication, channel);
fileChannel.position(headerBytes.readPosition());
map.initBeforeMapping(fileChannel);
map.initBeforeMapping(fileChannel, headerBytes.readPosition());
long expectedFileLength = map.expectedFileSize();
if (expectedFileLength != fileLength) {
throw new IOException("The file " + file + " the map is serialized from " +
@@ -1422,9 +1426,14 @@ public final class ChronicleMapBuilder<K, V> implements
Wire wire = new TextWire(headerBytes);
wire.getValueOut().typedMarshallable(map);
headerBuffer.limit((int) headerBytes.writePosition());
fileChannel.write(headerBuffer);
map.initBeforeMapping(fileChannel);
int headerSize = (int) headerBytes.writePosition();
headerBuffer.limit(headerSize);
try (FileLock ignored = fileChannel.lock(0, headerSize, false)) {
while (headerBuffer.remaining() > 0) {
fileChannel.write(headerBuffer, headerBuffer.position());
}
map.initBeforeMapping(fileChannel, headerSize);
}
map.createMappedStoreAndSegments(file, raf);
}