Buffer library performance test and functional test successfully. (#1612)

* Buffer stream.

* Buffer file reader.

* Buffer library performance test and functional test successfully.

* Fixed the code merge mistake.

oap-server/server-library/library-buffer/src/main/java/org/apache/skywalking/oap/server/library/buffer/BufferFileUtils.java

@@ -18,8 +18,7 @@
 
 package org.apache.skywalking.oap.server.library.buffer;
 
-import java.text.*;
-import java.util.*;
+import java.util.Arrays;
 
 /**
  * @author peng-yongsheng
@@ -34,19 +33,17 @@ class BufferFileUtils {
     static final String OFFSET_FILE_PREFIX = "offset";
     private static final String SEPARATOR = "-";
     private static final String SUFFIX = ".sw";
-    private static final String DATA_FORMAT_STR = "yyyyMMddHHmmss";
 
     static void sort(String[] fileList) {
         Arrays.sort(fileList, (f1, f2) -> {
-            int fileId1 = Integer.parseInt(f1.split("_")[1]);
-            int fileId2 = Integer.parseInt(f2.split("_")[1]);
+            long t1 = Long.parseLong(f1.substring(0, f1.length() - 3).split(SEPARATOR)[1]);
+            long t2 = Long.parseLong(f2.substring(0, f2.length() - 3).split(SEPARATOR)[1]);
 
-            return fileId1 - fileId2;
+            return (int)(t1 - t2);
         });
     }
 
     static String buildFileName(String prefix) {
-        DateFormat dateFormat = new SimpleDateFormat(DATA_FORMAT_STR);
-        return prefix + SEPARATOR + dateFormat.format(new Date()) + SUFFIX;
+        return prefix + SEPARATOR + System.currentTimeMillis() + SUFFIX;
     }
-}
+}
\ No newline at end of file

oap-server/server-library/library-buffer/src/main/java/org/apache/skywalking/oap/server/library/buffer/DataStream.java

@@ -38,7 +38,7 @@ class DataStream<MESSAGE_TYPE extends GeneratedMessageV3> {
     @Getter private final DataStreamWriter<MESSAGE_TYPE> writer;
     private boolean initialized = false;
 
-    DataStream(File directory, int offsetFileMaxSize, int dataFileMaxSize, Parser<MESSAGE_TYPE> parser,
+    DataStream(File directory, int dataFileMaxSize, int offsetFileMaxSize, Parser<MESSAGE_TYPE> parser,
         DataStreamReader.CallBack<MESSAGE_TYPE> callBack) {
         this.directory = directory;
         this.offsetStream = new OffsetStream(directory, offsetFileMaxSize);

oap-server/server-library/library-buffer/src/main/java/org/apache/skywalking/oap/server/library/buffer/DataStreamReader.java

@@ -21,6 +21,7 @@ package org.apache.skywalking.oap.server.library.buffer;
 import com.google.protobuf.*;
 import java.io.*;
 import java.util.concurrent.*;
+import org.apache.commons.io.FileUtils;
 import org.apache.commons.io.filefilter.PrefixFileFilter;
 import org.apache.skywalking.apm.util.*;
 import org.slf4j.*;
@@ -36,6 +37,7 @@ class DataStreamReader<MESSAGE_TYPE extends GeneratedMessageV3> {
     private final Offset.ReadOffset readOffset;
     private final Parser<MESSAGE_TYPE> parser;
     private final CallBack<MESSAGE_TYPE> callBack;
+    private File readingFile;
     private InputStream inputStream;
 
     DataStreamReader(File directory, Offset.ReadOffset readOffset, Parser<MESSAGE_TYPE> parser,
@@ -49,35 +51,40 @@ class DataStreamReader<MESSAGE_TYPE extends GeneratedMessageV3> {
     void initialize() {
         preRead();
 
-        Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate(
+        Executors.newSingleThreadScheduledExecutor().scheduleWithFixedDelay(
             new RunnableWithExceptionProtection(this::read,
-                t -> logger.error("Segment buffer pre read failure.", t)), 3, 3, TimeUnit.SECONDS);
+                t -> logger.error("Buffer data pre read failure.", t)), 3, 1, TimeUnit.SECONDS);
     }
 
     private void preRead() {
         String fileName = readOffset.getFileName();
         if (StringUtil.isEmpty(fileName)) {
-            openInputStream(readEarliestCreateDataFile());
+            openInputStream(readEarliestDataFile());
         } else {
-            File dataFile = new File(directory, fileName);
-            if (dataFile.exists()) {
-                openInputStream(dataFile);
-                read();
+            File readingFile = new File(directory, fileName);
+            if (readingFile.exists()) {
+                openInputStream(readingFile);
+                try {
+                    inputStream.skip(readOffset.getOffset());
+                } catch (IOException e) {
+                    logger.error(e.getMessage(), e);
+                }
             } else {
-                openInputStream(readEarliestCreateDataFile());
+                openInputStream(readEarliestDataFile());
             }
         }
     }
 
-    private void openInputStream(File readFile) {
+    private void openInputStream(File readingFile) {
         try {
-            inputStream = new FileInputStream(readFile);
+            this.readingFile = readingFile;
+            inputStream = new FileInputStream(readingFile);
         } catch (FileNotFoundException e) {
             logger.error(e.getMessage(), e);
         }
     }
 
-    private File readEarliestCreateDataFile() {
+    private File readEarliestDataFile() {
         String[] fileNames = directory.list(new PrefixFileFilter(BufferFileUtils.DATA_FILE_PREFIX));
 
         if (fileNames != null && fileNames.length > 0) {
@@ -92,12 +99,20 @@ class DataStreamReader<MESSAGE_TYPE extends GeneratedMessageV3> {
 
     private void read() {
         try {
-            MESSAGE_TYPE messageType = parser.parseDelimitedFrom(inputStream);
-            if (messageType != null) {
-                callBack.call(messageType);
-                final int serialized = messageType.getSerializedSize();
-                final int offset = CodedOutputStream.computeUInt32SizeNoTag(serialized) + serialized;
-                readOffset.setOffset(readOffset.getOffset() + offset);
+            if (readOffset.getOffset() == readingFile.length() && !readOffset.isCurrentWriteFile()) {
+                FileUtils.forceDelete(readingFile);
+                openInputStream(readEarliestDataFile());
+            }
+
+            while (readOffset.getOffset() < readingFile.length()) {
+
+                MESSAGE_TYPE messageType = parser.parseDelimitedFrom(inputStream);
+                if (messageType != null) {
+                    callBack.call(messageType);
+                    final int serialized = messageType.getSerializedSize();
+                    final int offset = CodedOutputStream.computeUInt32SizeNoTag(serialized) + serialized;
+                    readOffset.setOffset(readOffset.getOffset() + offset);
+                }
             }
         } catch (IOException e) {
             logger.error(e.getMessage(), e);

oap-server/server-library/library-buffer/src/main/java/org/apache/skywalking/oap/server/library/buffer/DataStreamWriter.java

@@ -49,46 +49,46 @@ class DataStreamWriter<MESSAGE_TYPE extends GeneratedMessageV3> {
         if (!initialized) {
             String writeFileName = writeOffset.getFileName();
 
-            File dataFile;
+            File writingFile;
             if (StringUtil.isEmpty(writeFileName)) {
-                dataFile = createNewFile();
+                writingFile = createNewFile();
             } else {
-                dataFile = new File(directory, writeFileName);
-                if (!dataFile.exists()) {
-                    dataFile = createNewFile();
+                writingFile = new File(directory, writeFileName);
+                if (!writingFile.exists()) {
+                    writingFile = createNewFile();
                 }
             }
 
-            outputStream = FileUtils.openOutputStream(dataFile, true);
+            outputStream = FileUtils.openOutputStream(writingFile, true);
             initialized = true;
         }
     }
 
     private File createNewFile() throws IOException {
         String fileName = BufferFileUtils.buildFileName(BufferFileUtils.DATA_FILE_PREFIX);
-        File dataFile = new File(directory, fileName);
+        File writingFile = new File(directory, fileName);
 
-        boolean created = dataFile.createNewFile();
+        boolean created = writingFile.createNewFile();
         if (!created) {
-            logger.info("The file named {} already exists.", dataFile.getAbsolutePath());
+            logger.info("The file named {} already exists.", writingFile.getAbsolutePath());
         } else {
-            logger.info("Create a new buffer data file: {}", dataFile.getAbsolutePath());
+            logger.info("Create a new buffer data file: {}", writingFile.getAbsolutePath());
         }
 
         writeOffset.setOffset(0);
-        writeOffset.setFileName(dataFile.getName());
+        writeOffset.setFileName(writingFile.getName());
 
-        return dataFile;
+        return writingFile;
     }
 
-    void write(AbstractMessageLite messageLite) {
+    synchronized void write(AbstractMessageLite messageLite) {
         try {
             messageLite.writeDelimitedTo(outputStream);
             long position = outputStream.getChannel().position();
             writeOffset.setOffset(position);
-            if (position > (FileUtils.ONE_MB * dataFileMaxSize)) {
-                File dataFile = createNewFile();
-                outputStream = FileUtils.openOutputStream(dataFile, true);
+            if (position >= (FileUtils.ONE_MB * dataFileMaxSize)) {
+                File writingFile = createNewFile();
+                outputStream = FileUtils.openOutputStream(writingFile, true);
             }
         } catch (IOException e) {
             logger.error(e.getMessage(), e);

oap-server/server-library/library-buffer/src/main/java/org/apache/skywalking/oap/server/library/buffer/Offset.java

@@ -31,8 +31,8 @@ class Offset {
     @Getter private final WriteOffset writeOffset;
 
     Offset() {
-        readOffset = new ReadOffset();
         writeOffset = new WriteOffset();
+        readOffset = new ReadOffset(writeOffset);
     }
 
     String serialize() {
@@ -55,6 +55,15 @@ class Offset {
     static class ReadOffset {
         @Getter @Setter private String fileName;
         @Getter @Setter private long offset = 0;
+        private final WriteOffset writeOffset;
+
+        private ReadOffset(WriteOffset writeOffset) {
+            this.writeOffset = writeOffset;
+        }
+
+        boolean isCurrentWriteFile() {
+            return fileName.equals(writeOffset.fileName);
+        }
     }
 
     static class WriteOffset {

oap-server/server-library/library-buffer/src/main/java/org/apache/skywalking/oap/server/library/buffer/OffsetStream.java

@@ -66,8 +66,8 @@ class OffsetStream {
         if (!initialized) {
             String[] fileNames = directory.list(new PrefixFileFilter(BufferFileUtils.OFFSET_FILE_PREFIX));
             if (fileNames != null && fileNames.length > 0) {
-                for (int i = 0; i < fileNames.length; i++) {
-                }
+                BufferFileUtils.sort(fileNames);
+                offsetFile = new File(directory, fileNames[0]);
             } else {
                 offsetFile = newFile();
             }

oap-server/server-library/library-buffer/src/test/java/org/apache/skywalking/oap/server/library/buffer/BufferStreamTestCase.java

@@ -33,9 +33,9 @@ public class BufferStreamTestCase {
     public static void main(String[] args) throws IOException, InterruptedException {
         String directory = "/Users/pengys5/code/sky-walking/buffer-test";
         BufferStream.Builder<TraceSegmentObject> builder = new BufferStream.Builder<>(directory);
-        builder.cleanWhenRestart(true);
-        builder.dataFileMaxSize(1);
-        builder.offsetFileMaxSize(1);
+//        builder.cleanWhenRestart(true);
+        builder.dataFileMaxSize(50);
+        builder.offsetFileMaxSize(10);
         builder.parser(TraceSegmentObject.parser());
         builder.callBack(new SegmentParse());
 
@@ -44,18 +44,23 @@ public class BufferStreamTestCase {
 
         TimeUnit.SECONDS.sleep(5);
 
-        String str = "2018-08-27 11:59:45,261 main DEBUG Registering MBean org.apache.logging.log4j2:type=6d6f6e28" +
-            "main DEBUG Registering MBean org.apache.logging.log4j2:type=6d6f6e28 main DEBUG Registering MBean org.apache.logging.log4j2:type=6d6f6e28 main DEBUG Registering MBean org.apache.logging.log4j2:type=6d6f6e28" +
-            "main DEBUG Registering MBean org.apache.logging.log4j2:type=6d6f6e28 main DEBUG Registering MBean org.apache.logging.log4j2:type=6d6f6e28 main DEBUG Registering MBean org.apache.logging.log4j2:type=6d6f6e28" +
-            "main DEBUG Registering MBean org.apache.logging.log4j2:type=6d6f6e28 main DEBUG Registering MBean org.apache.logging.log4j2:type=6d6f6e28 main DEBUG Registering MBean org.apache.logging.log4j2:type=6d6f6e28";
+        StringBuilder str = new StringBuilder("2018-08-27 11:59:45,261 main DEBUG Registering MBean org.apache.logging.log4j2:type=6d6f6e28");
+        for (int i = 0; i < 1000; i++) {
+            str.append("main DEBUG Registering MBean org.apache.logging.log4j2:type=6d6f6e28 main DEBUG Registering MBean org.apache.logging.log4j2:type=6d6f6e28 main DEBUG Registering MBean org.apache.logging.log4j2:type=6d6f6e28");
+        }
 
-        for (int i = 0; i < 100; i++) {
+        for (int i = 0; i < 20000; i++) {
             TraceSegmentObject.Builder segment = TraceSegmentObject.newBuilder();
             SpanObject.Builder span = SpanObject.newBuilder();
 
-            span.setOperationName(String.valueOf(i) + "  " + str);
+            span.setSpanId(i);
+            span.setOperationName(str.toString());
             segment.addSpans(span);
             stream.write(segment.build());
+
+            if (i % 1000 == 0) {
+                TimeUnit.MILLISECONDS.sleep(50);
+            }
         }
 
     }
@@ -63,7 +68,7 @@ public class BufferStreamTestCase {
     private static class SegmentParse implements DataStreamReader.CallBack<TraceSegmentObject> {
 
         @Override public void call(TraceSegmentObject message) {
-            logger.info("segment parse: {}", message.getSpans(0).getOperationName());
+            logger.info("segment parse: {}", message.getSpans(0).getSpanId());
         }
     }
 }

skywalking-ui @ f9c60293

-Subproject commit ad3ee45dbadfae35d77238bdd7a1df593158f109
+Subproject commit f9c602936ab4f386576bf16f203efac61962e424