Merge pull request #14 from scalecube/fix-metrics-transmitter-race-condition

Fixed race-condition for MetricsRecorder/MetricsTransmitterAgent

Author: artem-v

metrics/src/main/java/io/scalecube/metrics/CountersReaderAgent.java

@@ -45,7 +45,7 @@ public enum State {
   private final UnsafeBuffer headerBuffer = new UnsafeBuffer();
   private long countersStartTimestamp = -1;
   private long countersPid = -1;
-  private long countersValuesBufferLength = -1;
+  private int countersValuesBufferLength = -1;
   private CountersReader countersReader;
   private State state = State.CLOSED;
 

metrics/src/main/java/io/scalecube/metrics/CountersRegistry.java

@@ -305,7 +305,7 @@ public static UnsafeBuffer createCountersMetaDataBuffer(
     public static UnsafeBuffer createCountersValuesBuffer(
         ByteBuffer buffer, DirectBuffer headerBuffer) {
       final var countersValuesBufferLength = countersValuesBufferLength(headerBuffer);
-      final int offset = HEADER_LENGTH + countersMetaDataBufferLength(countersValuesBufferLength);
+      final var offset = HEADER_LENGTH + countersMetaDataBufferLength(countersValuesBufferLength);
       final var length = countersValuesBufferLength;
       return new UnsafeBuffer(buffer, offset, length);
     }

metrics/src/main/java/io/scalecube/metrics/MetricsRecorder.java

@@ -277,20 +277,20 @@ private void concludeMetricsDirectory() {
       if (metricsDir.isDirectory()) {
         final var file = new File(metricsDir, METRICS_FILE);
         if (file.exists()) {
-          final var buffer = mapExistingFile(file, METRICS_FILE);
+          final var mappedByteBuffer = mapExistingFile(file, METRICS_FILE);
           try {
-            if (!LayoutDescriptor.isMetricsHeaderLengthSufficient(buffer.capacity())) {
+            if (!LayoutDescriptor.isMetricsHeaderLengthSufficient(mappedByteBuffer.capacity())) {
               delete(metricsDir, false);
             } else {
-              final var headerBuffer = LayoutDescriptor.createHeaderBuffer(buffer);
+              final var headerBuffer = LayoutDescriptor.createHeaderBuffer(mappedByteBuffer);
               final var startTimestamp = ManagementFactory.getRuntimeMXBean().getStartTime();
               final var pid = ManagementFactory.getRuntimeMXBean().getPid();
               if (!LayoutDescriptor.isMetricsActive(headerBuffer, startTimestamp, pid)) {
                 delete(metricsDir, false);
               }
             }
           } finally {
-            BufferUtil.free(buffer);
+            BufferUtil.free(mappedByteBuffer);
           }
         }
       }
@@ -317,14 +317,15 @@ private void concludeMetricsBuffer() {
       final var totalLength = headerLength + bufferLength;
 
       if (!file.exists()) {
-        final var buffer = mapNewFile(file, totalLength);
+        final var mappedByteBuffer = mapNewFile(file, totalLength);
         try {
-          final var headerBuffer = LayoutDescriptor.createHeaderBuffer(buffer);
+          final var headerBuffer = LayoutDescriptor.createHeaderBuffer(mappedByteBuffer);
           final var startTimestamp = ManagementFactory.getRuntimeMXBean().getStartTime();
           final var pid = ManagementFactory.getRuntimeMXBean().getPid();
           LayoutDescriptor.fillHeaderBuffer(headerBuffer, startTimestamp, pid, bufferLength);
+          mappedByteBuffer.force();
         } finally {
-          BufferUtil.free(buffer);
+          BufferUtil.free(mappedByteBuffer);
         }
       }
 
@@ -478,7 +479,7 @@ public static long pid(DirectBuffer headerBuffer) {
       return headerBuffer.getLong(PID_OFFSET);
     }
 
-    public static long metricsBufferLength(DirectBuffer headerBuffer) {
+    public static int metricsBufferLength(DirectBuffer headerBuffer) {
       return headerBuffer.getInt(METRICS_BUFFER_LENGTH_OFFSET);
     }
 

metrics/src/main/java/io/scalecube/metrics/MetricsTransmitterAgent.java

@@ -43,6 +43,7 @@ public enum State {
   private final UnsafeBuffer headerBuffer = new UnsafeBuffer();
   private long metricsStartTimestamp = -1;
   private long metricsPid = -1;
+  private int metricsBufferLength = -1;
   private State state = State.CLOSED;
 
   MetricsTransmitterAgent(
@@ -102,16 +103,20 @@ private int init() {
     }
 
     metricsByteBuffer = mapExistingFile(metricsFile, METRICS_FILE);
-    headerBuffer.wrap(metricsByteBuffer, 0, LayoutDescriptor.HEADER_LENGTH);
+    final var headerLength = LayoutDescriptor.HEADER_LENGTH;
+    headerBuffer.wrap(metricsByteBuffer, 0, headerLength);
     metricsStartTimestamp = LayoutDescriptor.startTimestamp(headerBuffer);
     metricsPid = LayoutDescriptor.pid(headerBuffer);
+    metricsBufferLength = LayoutDescriptor.metricsBufferLength(headerBuffer);
 
-    final var headerLength = LayoutDescriptor.HEADER_LENGTH;
-    final var totalLength = metricsByteBuffer.capacity();
-    final var length = totalLength - headerLength;
+    if (metricsBufferLength <= 0) {
+      state(State.CLEANUP);
+      return 0;
+    }
 
     metricsBuffer =
-        new ManyToOneRingBuffer(new UnsafeBuffer(metricsByteBuffer, headerLength, length));
+        new ManyToOneRingBuffer(
+            new UnsafeBuffer(metricsByteBuffer, headerLength, metricsBufferLength));
 
     state(State.RUNNING);
     LOGGER.info("[{}] Initialized, now running", roleName());
@@ -149,7 +154,7 @@ private boolean isActive(File metricsFile) {
         LOGGER.warn("[{}] {} has not sufficient length", roleName(), metricsFile);
         return false;
       }
-      if (metricsStartTimestamp != -1
+      if (metricsBufferLength != -1
           && !LayoutDescriptor.isMetricsActive(headerBuffer, metricsStartTimestamp, metricsPid)) {
         LOGGER.warn("[{}] {} is not active", roleName(), metricsFile);
         return false;
@@ -173,6 +178,7 @@ private int cleanup() {
     metricsFile = null;
     metricsStartTimestamp = -1;
     metricsPid = -1;
+    metricsBufferLength = -1;
 
     State previous = state;
     if (previous != State.CLOSED) { // when it comes from onClose()