From 0adaaa935dfc0a3ad77b2c78f431e05faee0273e Mon Sep 17 00:00:00 2001
From: leventov <leventov@ya.ru>
Date: Mon, 5 Oct 2015 10:15:31 +0300
Subject: [PATCH] Store segmentHeadersOffset in global Mutable state (according
 to spec; to make ChronicleMap file independent from current mapping page
 size)

---
 .../hash/VanillaGlobalMutableState.java       |  3 ++
 .../hash/impl/VanillaChronicleHash.java       | 52 ++++++++++++++-----
 .../chronicle/map/ChronicleMapBuilder.java    | 19 +++----
 3 files changed, 51 insertions(+), 23 deletions(-)

diff --git a/src/main/java/net/openhft/chronicle/hash/VanillaGlobalMutableState.java b/src/main/java/net/openhft/chronicle/hash/VanillaGlobalMutableState.java
index e658efee3..b85990dff 100644
--- a/src/main/java/net/openhft/chronicle/hash/VanillaGlobalMutableState.java
+++ b/src/main/java/net/openhft/chronicle/hash/VanillaGlobalMutableState.java
@@ -35,4 +35,7 @@ public interface VanillaGlobalMutableState extends Byteable {
 
     long getExtraTiersInUse();
     void setExtraTiersInUse(long extraTiersInUse);
+
+    long getSegmentHeadersOffset();
+    void setSegmentHeadersOffset(long segmentHeadersOffset);
 }
diff --git a/src/main/java/net/openhft/chronicle/hash/impl/VanillaChronicleHash.java b/src/main/java/net/openhft/chronicle/hash/impl/VanillaChronicleHash.java
index 9d4e11480..ae44dc1e4 100644
--- a/src/main/java/net/openhft/chronicle/hash/impl/VanillaChronicleHash.java
+++ b/src/main/java/net/openhft/chronicle/hash/impl/VanillaChronicleHash.java
@@ -40,7 +40,10 @@ import java.io.File;
 import java.io.IOException;
 import java.io.ObjectInputStream;
 import java.io.Serializable;
+import java.nio.ByteBuffer;
 import java.nio.channels.FileChannel;
+import java.nio.file.OpenOption;
+import java.nio.file.StandardOpenOption;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.TimeUnit;
@@ -61,6 +64,7 @@ public abstract class VanillaChronicleHash<K, KI, MKI extends MetaBytesInterop<K
     private static final long serialVersionUID = 0L;
 
     public static final long TIER_COUNTERS_AREA_SIZE = 64;
+    public static final long RESERVED_GLOBAL_MUTABLE_STATE_BYTES = 1024;
 
     /////////////////////////////////////////////////
     // Version
@@ -289,6 +293,26 @@ public abstract class VanillaChronicleHash<K, KI, MKI extends MetaBytesInterop<K
         }
     }
 
+    public final void initBeforeMapping(FileChannel ch) throws IOException {
+        headerSize = roundUpMapHeaderSize(ch.position());
+        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(globalMutableState.maxSize());
+            ch.read(globalMutableStateBuffer);
+            globalMutableStateBuffer.flip();
+            globalMutableState.bytes(new ByteBufferBytes(globalMutableStateBuffer), 0);
+        }
+    }
+
+    private static long roundUpMapHeaderSize(long headerSize) {
+        long roundUp = (headerSize + 127L) & ~127L;
+        if (roundUp - headerSize < 64)
+            roundUp += 128;
+        return roundUp;
+    }
+
     public final void createMappedStoreAndSegments(BytesStore bytesStore) throws IOException {
         this.ms = bytesStore;
         bytes = ms.bytes();
@@ -296,12 +320,16 @@ public abstract class VanillaChronicleHash<K, KI, MKI extends MetaBytesInterop<K
 
         onHeaderCreated();
 
-        segmentHeadersOffset = mapHeaderOuterSize();
+        segmentHeadersOffset = segmentHeadersOffset();
+
         long segmentHeadersSize = actualSegments * segmentHeaderSize;
         segmentsOffset = segmentHeadersOffset + segmentHeadersSize;
 
-        if (createdOrInMemory)
+        if (createdOrInMemory) {
             zeroOutNewlyMappedChronicleMapBytes();
+            // write the segment headers offset after zeroing out
+            globalMutableState.setSegmentHeadersOffset(segmentHeadersOffset);
+        }
 
         initTierBulks(bytesStore.size());
     }
@@ -389,15 +417,15 @@ public abstract class VanillaChronicleHash<K, KI, MKI extends MetaBytesInterop<K
         return BuildVersion.version();
     }
 
-    private long mapHeaderOuterSize() {
-        // Align segment headers on page boundary to minimize number of pages that
-        // segment headers span
-        return roundUpRuntimePageSize(mapHeaderInnerSize());
-    }
-
-    private static long roundUpRuntimePageSize(long size) {
-        long pageMask = UNSAFE.pageSize() - 1;
-        return (size + pageMask) & ~pageMask;
+    private long segmentHeadersOffset() {
+        if (createdOrInMemory) {
+            // Align segment headers on page boundary to minimize number of pages that
+            // segment headers span
+            long reserved = RESERVED_GLOBAL_MUTABLE_STATE_BYTES - globalMutableStateTotalUsedSize();
+            return OS.pageAlign(mapHeaderInnerSize() + reserved);
+        } else {
+            return globalMutableState.getSegmentHeadersOffset();
+        }
     }
 
     public long mapHeaderInnerSize() {
@@ -410,7 +438,7 @@ public abstract class VanillaChronicleHash<K, KI, MKI extends MetaBytesInterop<K
     }
 
     public final long sizeInBytesWithoutTiers() {
-        return mapHeaderOuterSize() + actualSegments * (segmentHeaderSize + segmentSize);
+        return segmentHeadersOffset() + actualSegments * (segmentHeaderSize + segmentSize);
     }
 
     public final long expectedFileSize() {
diff --git a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java
index 7289f24f9..ceb2a12f1 100755
--- a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java
+++ b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java
@@ -236,13 +236,6 @@ public final class ChronicleMapBuilder<K, V> implements
         return new ChronicleMapBuilder<>(keyClass, valueClass);
     }
 
-    private static long roundUpMapHeaderSize(long headerSize) {
-        long roundUp = (headerSize + 127L) & ~127L;
-        if (roundUp - headerSize < 64)
-            roundUp += 128;
-        return roundUp;
-    }
-
     private static void checkSegments(long segments) {
         if (segments <= 0) {
             throw new IllegalArgumentException("segments should be positive, " +
@@ -1533,18 +1526,22 @@ public final class ChronicleMapBuilder<K, V> implements
                     VanillaChronicleMap<K, ?, ?, V, ?, ?, ?> map =
                             (VanillaChronicleMap<K, ?, ?, V, ?, ?, ?>) m;
                     map.initTransientsFromBuilder(this);
-                    map.headerSize = roundUpMapHeaderSize(fis.getChannel().position());
+                    map.initBeforeMapping(fis.getChannel());
                     map.createMappedStoreAndSegments(file);
                     long expectedFileLength = map.expectedFileSize();
                     if (expectedFileLength != fileLength) {
-                        throw new IOException("The file " + file + "the map is serialized from " +
+                        throw new IOException("The file " + file + " the map is serialized from " +
                                 "has unexpected length " + fileLength + ", probably corrupted. " +
                                 "Expected length is " + expectedFileLength);
                     }
                     // This is needed to property initialize key and value serialization builders,
                     // which are later used in replication
                     preMapConstruction();
-                    return establishReplication(map, singleHashReplication, channel);
+                    establishReplication(map, singleHashReplication, channel);
+                    fis.getChannel().force(true);
+                    // TODO according to Self Boostrapping Data spec, should write "init complete"
+                    // byte *after* the above force instruction, see HCOLL-396
+                    return map;
                 }
             }
             if (file.createNewFile() || fileLength == 0) {
@@ -1569,7 +1566,7 @@ public final class ChronicleMapBuilder<K, V> implements
                 oos.writeObject(map);
             }
             oos.flush();
-            map.headerSize = roundUpMapHeaderSize(fos.getChannel().position());
+            map.initBeforeMapping(fos.getChannel());
             map.createMappedStoreAndSegments(file);
         }