[ISSUE #118] Support multi-dirs storage

src/main/java/io/openmessaging/storage/dledger/DLedgerConfig.java

@@ -24,6 +24,7 @@ public class DLedgerConfig {
 
     public static final String MEMORY = "MEMORY";
     public static final String FILE = "FILE";
+    public static final String MULTI_PATH_SPLITTER = System.getProperty("dLedger.multiPath.Splitter", ",");
 
     @Parameter(names = {"--group", "-g"}, description = "Group of this server")
     private String group = "default";
@@ -37,6 +38,8 @@ public class DLedgerConfig {
     @Parameter(names = {"--store-base-dir", "-s"}, description = "The base store dir of this server")
     private String storeBaseDir = File.separator + "tmp" + File.separator + "dledgerstore";
 
+    @Parameter(names = {"--read-only-data-store-dirs"}, description = "The dirs of this server to be read only")
+    private String readOnlyDataStoreDirs = null;
 
     @Parameter(names = {"--peer-push-throttle-point"}, description = "When the follower is behind the leader more than this value, it will trigger the throttle")
     private int peerPushThrottlePoint = 300 * 1024 * 1024;
@@ -407,4 +410,12 @@ public long getLeadershipTransferWaitTimeout() {
     public void setLeadershipTransferWaitTimeout(long leadershipTransferWaitTimeout) {
         this.leadershipTransferWaitTimeout = leadershipTransferWaitTimeout;
     }
+
+    public String getReadOnlyDataStoreDirs() {
+        return readOnlyDataStoreDirs;
+    }
+
+    public void setReadOnlyDataStoreDirs(String readOnlyDataStoreDirs) {
+        this.readOnlyDataStoreDirs = readOnlyDataStoreDirs;
+    }
 }

src/main/java/io/openmessaging/storage/dledger/store/file/DLedgerMmapFileStore.java

@@ -30,8 +30,11 @@
 import java.io.File;
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Properties;
+import java.util.Set;
 import java.util.concurrent.atomic.AtomicBoolean;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -67,10 +70,17 @@ public class DLedgerMmapFileStore extends DLedgerStore {
     private AtomicBoolean hasLoaded = new AtomicBoolean(false);
     private AtomicBoolean hasRecovered = new AtomicBoolean(false);
 
+    private volatile Set<String> fullStorePaths = Collections.emptySet();
+
     public DLedgerMmapFileStore(DLedgerConfig dLedgerConfig, MemberState memberState) {
         this.dLedgerConfig = dLedgerConfig;
         this.memberState = memberState;
-        this.dataFileList = new MmapFileList(dLedgerConfig.getDataStorePath(), dLedgerConfig.getMappedFileSizeForEntryData());
+        if (dLedgerConfig.getDataStorePath().contains(DLedgerConfig.MULTI_PATH_SPLITTER)) {
+            this.dataFileList = new MultiPathMmapFileList(dLedgerConfig, dLedgerConfig.getMappedFileSizeForEntryData(),
+                    this::getFullStorePaths);
+        } else {
+            this.dataFileList = new MmapFileList(dLedgerConfig.getDataStorePath(), dLedgerConfig.getMappedFileSizeForEntryData());
+        }
         this.indexFileList = new MmapFileList(dLedgerConfig.getIndexStorePath(), dLedgerConfig.getMappedFileSizeForEntryIndex());
         localEntryBuffer = ThreadLocal.withInitial(() -> ByteBuffer.allocate(4 * 1024 * 1024));
         localIndexBuffer = ThreadLocal.withInitial(() -> ByteBuffer.allocate(INDEX_UNIT_SIZE * 2));
@@ -615,6 +625,14 @@ public MmapFileList getIndexFileList() {
         return indexFileList;
     }
 
+    public Set<String> getFullStorePaths() {
+        return fullStorePaths;
+    }
+
+    public void setFullStorePaths(Set<String> fullStorePaths) {
+        this.fullStorePaths = fullStorePaths;
+    }
+
     public interface AppendHook {
         void doHook(DLedgerEntry entry, ByteBuffer buffer, int bodyOffset);
     }
@@ -656,7 +674,7 @@ public FlushDataService(String name, Logger logger) {
     class CleanSpaceService extends ShutdownAbleThread {
 
         double storeBaseRatio = DLedgerUtils.getDiskPartitionSpaceUsedPercent(dLedgerConfig.getStoreBaseDir());
-        double dataRatio = DLedgerUtils.getDiskPartitionSpaceUsedPercent(dLedgerConfig.getDataStorePath());
+        double dataRatio = calcDataStorePathPhysicRatio();
 
         public CleanSpaceService(String name, Logger logger) {
             super(name, logger);
@@ -665,7 +683,7 @@ public CleanSpaceService(String name, Logger logger) {
         @Override public void doWork() {
             try {
                 storeBaseRatio = DLedgerUtils.getDiskPartitionSpaceUsedPercent(dLedgerConfig.getStoreBaseDir());
-                dataRatio = DLedgerUtils.getDiskPartitionSpaceUsedPercent(dLedgerConfig.getDataStorePath());
+                dataRatio = calcDataStorePathPhysicRatio();
                 long hourOfMs = 3600L * 1000L;
                 long fileReservedTimeMs = dLedgerConfig.getFileReservedHours() *  hourOfMs;
                 if (fileReservedTimeMs < hourOfMs) {
@@ -727,5 +745,21 @@ private boolean isNeedForbiddenWrite() {
             }
             return false;
         }
+
+        public double calcDataStorePathPhysicRatio() {
+            Set<String> fullStorePath = new HashSet<>();
+            String storePath = dLedgerConfig.getDataStorePath();
+            String[] paths = storePath.trim().split(DLedgerConfig.MULTI_PATH_SPLITTER);
+            double minPhysicRatio = 100;
+            for (String path : paths) {
+                double physicRatio = DLedgerUtils.isPathExists(path) ? DLedgerUtils.getDiskPartitionSpaceUsedPercent(path) : -1;
+                minPhysicRatio = Math.min(minPhysicRatio, physicRatio);
+                if (physicRatio > dLedgerConfig.getDiskSpaceRatioToForceClean()) {
+                    fullStorePath.add(path);
+                }
+            }
+            DLedgerMmapFileStore.this.setFullStorePaths(fullStorePath);
+            return minPhysicRatio;
+        }
     }
 }

src/main/java/io/openmessaging/storage/dledger/store/file/MmapFileList.java

@@ -282,30 +282,34 @@ public boolean load() {
         File dir = new File(this.storePath);
         File[] files = dir.listFiles();
         if (files != null) {
-            // ascending order
-            Arrays.sort(files);
-            for (File file : files) {
+            return doLoad(Arrays.asList(files));
+        }
+        return true;
+    }
+
+    public boolean doLoad(List<File> files) {
+        // ascending order
+        files.sort(Comparator.comparing(File::getName));
+        for (File file : files) {
 
-                if (file.length() != this.mappedFileSize) {
-                    logger.warn(file + "\t" + file.length()
+            if (file.length() != this.mappedFileSize) {
+                logger.warn(file + "\t" + file.length()
                         + " length not matched message store config value, please check it manually. You should delete old files before changing mapped file size");
-                    return false;
-                }
-                try {
-                    MmapFile mappedFile = new DefaultMmapFile(file.getPath(), mappedFileSize);
-
-                    mappedFile.setWrotePosition(this.mappedFileSize);
-                    mappedFile.setFlushedPosition(this.mappedFileSize);
-                    mappedFile.setCommittedPosition(this.mappedFileSize);
-                    this.mappedFiles.add(mappedFile);
-                    logger.info("load " + file.getPath() + " OK");
-                } catch (IOException e) {
-                    logger.error("load file " + file + " error", e);
-                    return false;
-                }
+                return false;
             }
-        }
+            try {
+                MmapFile mappedFile = new DefaultMmapFile(file.getPath(), mappedFileSize);
 
+                mappedFile.setWrotePosition(this.mappedFileSize);
+                mappedFile.setFlushedPosition(this.mappedFileSize);
+                mappedFile.setCommittedPosition(this.mappedFileSize);
+                this.mappedFiles.add(mappedFile);
+                logger.info("load " + file.getPath() + " OK");
+            } catch (IOException e) {
+                logger.error("load file " + file + " error", e);
+                return false;
+            }
+        }
         return true;
     }
 
@@ -320,25 +324,33 @@ public MmapFile getLastMappedFile(final long startOffset, boolean needCreate) {
         }
 
         if (createOffset != -1 && needCreate) {
-            String nextFilePath = this.storePath + File.separator + DLedgerUtils.offset2FileName(createOffset);
-            MmapFile mappedFile = null;
-            try {
-                mappedFile = new DefaultMmapFile(nextFilePath, this.mappedFileSize);
-            } catch (IOException e) {
-                logger.error("create mappedFile exception", e);
-            }
+            return tryCreateMappedFile(createOffset);
+        }
 
-            if (mappedFile != null) {
-                if (this.mappedFiles.isEmpty()) {
-                    mappedFile.setFirstCreateInQueue(true);
-                }
-                this.mappedFiles.add(mappedFile);
-            }
+        return mappedFileLast;
+    }
+
+    protected MmapFile tryCreateMappedFile(long createOffset) {
+        String nextFilePath = this.storePath + File.separator + DLedgerUtils.offset2FileName(createOffset);
+        return doCreateMappedFile(nextFilePath);
+    }
 
-            return mappedFile;
+    protected MmapFile doCreateMappedFile(String nextFilePath) {
+        MmapFile mappedFile = null;
+        try {
+            mappedFile = new DefaultMmapFile(nextFilePath, this.mappedFileSize);
+        } catch (IOException e) {
+            logger.error("create mappedFile exception", e);
         }
 
-        return mappedFileLast;
+        if (mappedFile != null) {
+            if (this.mappedFiles.isEmpty()) {
+                mappedFile.setFirstCreateInQueue(true);
+            }
+            this.mappedFiles.add(mappedFile);
+        }
+
+        return mappedFile;
     }
 
     public MmapFile getLastMappedFile(final long startOffset) {

src/main/java/io/openmessaging/storage/dledger/store/file/MultiPathMmapFileList.java

@@ -0,0 +1,121 @@
+/*
+ * Copyright 2017-2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.openmessaging.storage.dledger.store.file;
+
+import io.netty.util.internal.StringUtil;
+import io.openmessaging.storage.dledger.DLedgerConfig;
+import io.openmessaging.storage.dledger.utils.DLedgerUtils;
+
+import java.io.File;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.function.Supplier;
+
+public class MultiPathMmapFileList extends MmapFileList {
+
+    private final Supplier<Set<String>> fullStorePathsSupplier;
+    private final DLedgerConfig config;
+
+    public MultiPathMmapFileList(DLedgerConfig config, int mappedFileSize, Supplier<Set<String>> fullStorePathsSupplier) {
+        super(config.getDataStorePath(), mappedFileSize);
+        this.config = config;
+        this.fullStorePathsSupplier = fullStorePathsSupplier;
+    }
+
+    private Set<String> getPaths() {
+        String[] paths = this.config.getDataStorePath().trim().split(DLedgerConfig.MULTI_PATH_SPLITTER);
+        return new HashSet<>(Arrays.asList(paths));
+    }
+
+    private Set<String> getReadonlyPaths() {
+        String pathStr = config.getReadOnlyDataStoreDirs();
+        if (StringUtil.isNullOrEmpty(pathStr)) {
+            return Collections.emptySet();
+        }
+        String[] paths = pathStr.trim().split(DLedgerConfig.MULTI_PATH_SPLITTER);
+        return new HashSet<>(Arrays.asList(paths));
+    }
+
+    @Override
+    public boolean load() {
+        Set<String> storePathSet = getPaths();
+        storePathSet.addAll(getReadonlyPaths());
+
+        List<File> files = new ArrayList<>();
+        for (String path : storePathSet) {
+            File dir = new File(path);
+            File[] ls = dir.listFiles();
+            if (ls != null) {
+                Collections.addAll(files, ls);
+            }
+        }
+
+        return doLoad(files);
+    }
+
+    @Override
+    protected MmapFile tryCreateMappedFile(long createOffset) {
+        long fileIdx = createOffset / this.getMappedFileSize();
+        Set<String> storePath = getPaths();
+        Set<String> readonlyPathSet = getReadonlyPaths();
+        Set<String> fullStorePaths =
+                fullStorePathsSupplier == null ? Collections.emptySet() : fullStorePathsSupplier.get();
+
+
+        HashSet<String> availableStorePath = new HashSet<>(storePath);
+        //do not create file in readonly store path.
+        availableStorePath.removeAll(readonlyPathSet);
+
+        //do not create file is space is nearly full.
+        availableStorePath.removeAll(fullStorePaths);
+
+        //if no store path left, fall back to writable store path.
+        if (availableStorePath.isEmpty()) {
+            availableStorePath = new HashSet<>(storePath);
+            availableStorePath.removeAll(readonlyPathSet);
+        }
+
+        String[] paths = availableStorePath.toArray(new String[]{});
+        Arrays.sort(paths);
+        String nextFilePath = paths[(int) (fileIdx % paths.length)] + File.separator
+                + DLedgerUtils.offset2FileName(createOffset);
+        return doCreateMappedFile(nextFilePath);
+    }
+
+    @Override
+    public void destroy() {
+        for (MmapFile mf : this.getMappedFiles()) {
+            mf.destroy(1000 * 3);
+        }
+        this.getMappedFiles().clear();
+        this.setFlushedWhere(0);
+
+        Set<String> storePathSet = getPaths();
+        storePathSet.addAll(getReadonlyPaths());
+
+        for (String path : storePathSet) {
+            File file = new File(path);
+            if (file.isDirectory()) {
+                file.delete();
+            }
+        }
+    }
+}

src/main/java/io/openmessaging/storage/dledger/utils/DLedgerUtils.java

@@ -90,4 +90,9 @@ public static double getDiskPartitionSpaceUsedPercent(final String path) {
         }
         return -1;
     }
+
+    public static boolean isPathExists(final String path) {
+        File file = new File(path);
+        return file.exists();
+    }
 }