Skip to content

Commit b2d6853

Browse files
author
dengzhiwen1
committed
support multi dirs storage
1 parent 3e210b6 commit b2d6853

File tree

5 files changed

+221
-38
lines changed

5 files changed

+221
-38
lines changed

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

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@ public class DLedgerConfig {
2424

2525
public static final String MEMORY = "MEMORY";
2626
public static final String FILE = "FILE";
27+
public static final String MULTI_PATH_SPLITTER = System.getProperty("dLedger.multiPath.Splitter", ",");
2728

2829
@Parameter(names = {"--group", "-g"}, description = "Group of this server")
2930
private String group = "default";
@@ -37,6 +38,8 @@ public class DLedgerConfig {
3738
@Parameter(names = {"--store-base-dir", "-s"}, description = "The base store dir of this server")
3839
private String storeBaseDir = File.separator + "tmp" + File.separator + "dledgerstore";
3940

41+
@Parameter(names = {"--read-only-data-store-dirs"}, description = "The dirs of this server to be read only")
42+
private String readOnlyDataStoreDirs = null;
4043

4144
@Parameter(names = {"--peer-push-throttle-point"}, description = "When the follower is behind the leader more than this value, it will trigger the throttle")
4245
private int peerPushThrottlePoint = 300 * 1024 * 1024;
@@ -398,4 +401,12 @@ public int getMaxBatchPushSize() {
398401
public void setMaxBatchPushSize(int maxBatchPushSize) {
399402
this.maxBatchPushSize = maxBatchPushSize;
400403
}
404+
405+
public String getReadOnlyDataStoreDirs() {
406+
return readOnlyDataStoreDirs;
407+
}
408+
409+
public void setReadOnlyDataStoreDirs(String readOnlyDataStoreDirs) {
410+
this.readOnlyDataStoreDirs = readOnlyDataStoreDirs;
411+
}
401412
}

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

Lines changed: 37 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -30,8 +30,11 @@
3030
import java.io.File;
3131
import java.nio.ByteBuffer;
3232
import java.util.ArrayList;
33+
import java.util.Collections;
34+
import java.util.HashSet;
3335
import java.util.List;
3436
import java.util.Properties;
37+
import java.util.Set;
3538
import java.util.concurrent.atomic.AtomicBoolean;
3639
import org.slf4j.Logger;
3740
import org.slf4j.LoggerFactory;
@@ -67,10 +70,17 @@ public class DLedgerMmapFileStore extends DLedgerStore {
6770
private AtomicBoolean hasLoaded = new AtomicBoolean(false);
6871
private AtomicBoolean hasRecovered = new AtomicBoolean(false);
6972

73+
private volatile Set<String> fullStorePaths = Collections.emptySet();
74+
7075
public DLedgerMmapFileStore(DLedgerConfig dLedgerConfig, MemberState memberState) {
7176
this.dLedgerConfig = dLedgerConfig;
7277
this.memberState = memberState;
73-
this.dataFileList = new MmapFileList(dLedgerConfig.getDataStorePath(), dLedgerConfig.getMappedFileSizeForEntryData());
78+
if (dLedgerConfig.getDataStorePath().contains(DLedgerConfig.MULTI_PATH_SPLITTER)) {
79+
this.dataFileList = new MultiPathMmapFileList(dLedgerConfig, dLedgerConfig.getMappedFileSizeForEntryData(),
80+
this::getFullStorePaths);
81+
} else {
82+
this.dataFileList = new MmapFileList(dLedgerConfig.getDataStorePath(), dLedgerConfig.getMappedFileSizeForEntryData());
83+
}
7484
this.indexFileList = new MmapFileList(dLedgerConfig.getIndexStorePath(), dLedgerConfig.getMappedFileSizeForEntryIndex());
7585
localEntryBuffer = ThreadLocal.withInitial(() -> ByteBuffer.allocate(4 * 1024 * 1024));
7686
localIndexBuffer = ThreadLocal.withInitial(() -> ByteBuffer.allocate(INDEX_UNIT_SIZE * 2));
@@ -615,6 +625,14 @@ public MmapFileList getIndexFileList() {
615625
return indexFileList;
616626
}
617627

628+
public Set<String> getFullStorePaths() {
629+
return fullStorePaths;
630+
}
631+
632+
public void setFullStorePaths(Set<String> fullStorePaths) {
633+
this.fullStorePaths = fullStorePaths;
634+
}
635+
618636
public interface AppendHook {
619637
void doHook(DLedgerEntry entry, ByteBuffer buffer, int bodyOffset);
620638
}
@@ -656,7 +674,7 @@ public FlushDataService(String name, Logger logger) {
656674
class CleanSpaceService extends ShutdownAbleThread {
657675

658676
double storeBaseRatio = DLedgerUtils.getDiskPartitionSpaceUsedPercent(dLedgerConfig.getStoreBaseDir());
659-
double dataRatio = DLedgerUtils.getDiskPartitionSpaceUsedPercent(dLedgerConfig.getDataStorePath());
677+
double dataRatio = calcDataStorePathPhysicRatio();
660678

661679
public CleanSpaceService(String name, Logger logger) {
662680
super(name, logger);
@@ -665,7 +683,7 @@ public CleanSpaceService(String name, Logger logger) {
665683
@Override public void doWork() {
666684
try {
667685
storeBaseRatio = DLedgerUtils.getDiskPartitionSpaceUsedPercent(dLedgerConfig.getStoreBaseDir());
668-
dataRatio = DLedgerUtils.getDiskPartitionSpaceUsedPercent(dLedgerConfig.getDataStorePath());
686+
dataRatio = calcDataStorePathPhysicRatio();
669687
long hourOfMs = 3600L * 1000L;
670688
long fileReservedTimeMs = dLedgerConfig.getFileReservedHours() * hourOfMs;
671689
if (fileReservedTimeMs < hourOfMs) {
@@ -727,5 +745,21 @@ private boolean isNeedForbiddenWrite() {
727745
}
728746
return false;
729747
}
748+
749+
public double calcDataStorePathPhysicRatio() {
750+
Set<String> fullStorePath = new HashSet<>();
751+
String storePath = dLedgerConfig.getDataStorePath();
752+
String[] paths = storePath.trim().split(DLedgerConfig.MULTI_PATH_SPLITTER);
753+
double minPhysicRatio = 100;
754+
for (String path : paths) {
755+
double physicRatio = DLedgerUtils.isPathExists(path) ? DLedgerUtils.getDiskPartitionSpaceUsedPercent(path) : -1;
756+
minPhysicRatio = Math.min(minPhysicRatio, physicRatio);
757+
if (physicRatio > dLedgerConfig.getDiskSpaceRatioToForceClean()) {
758+
fullStorePath.add(path);
759+
}
760+
}
761+
DLedgerMmapFileStore.this.setFullStorePaths(fullStorePath);
762+
return minPhysicRatio;
763+
}
730764
}
731765
}

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

Lines changed: 47 additions & 35 deletions
Original file line numberDiff line numberDiff line change
@@ -283,30 +283,34 @@ public boolean load() {
283283
File dir = new File(this.storePath);
284284
File[] files = dir.listFiles();
285285
if (files != null) {
286-
// ascending order
287-
Arrays.sort(files);
288-
for (File file : files) {
286+
return doLoad(Arrays.asList(files));
287+
}
288+
return true;
289+
}
290+
291+
public boolean doLoad(List<File> files) {
292+
// ascending order
293+
files.sort(Comparator.comparing(File::getName));
294+
for (File file : files) {
289295

290-
if (file.length() != this.mappedFileSize) {
291-
logger.warn(file + "\t" + file.length()
296+
if (file.length() != this.mappedFileSize) {
297+
logger.warn(file + "\t" + file.length()
292298
+ " length not matched message store config value, please check it manually. You should delete old files before changing mapped file size");
293-
return false;
294-
}
295-
try {
296-
MmapFile mappedFile = new DefaultMmapFile(file.getPath(), mappedFileSize);
297-
298-
mappedFile.setWrotePosition(this.mappedFileSize);
299-
mappedFile.setFlushedPosition(this.mappedFileSize);
300-
mappedFile.setCommittedPosition(this.mappedFileSize);
301-
this.mappedFiles.add(mappedFile);
302-
logger.info("load " + file.getPath() + " OK");
303-
} catch (IOException e) {
304-
logger.error("load file " + file + " error", e);
305-
return false;
306-
}
299+
return false;
307300
}
308-
}
301+
try {
302+
MmapFile mappedFile = new DefaultMmapFile(file.getPath(), mappedFileSize);
309303

304+
mappedFile.setWrotePosition(this.mappedFileSize);
305+
mappedFile.setFlushedPosition(this.mappedFileSize);
306+
mappedFile.setCommittedPosition(this.mappedFileSize);
307+
this.mappedFiles.add(mappedFile);
308+
logger.info("load " + file.getPath() + " OK");
309+
} catch (IOException e) {
310+
logger.error("load file " + file + " error", e);
311+
return false;
312+
}
313+
}
310314
return true;
311315
}
312316

@@ -321,25 +325,33 @@ public MmapFile getLastMappedFile(final long startOffset, boolean needCreate) {
321325
}
322326

323327
if (createOffset != -1 && needCreate) {
324-
String nextFilePath = this.storePath + File.separator + DLedgerUtils.offset2FileName(createOffset);
325-
MmapFile mappedFile = null;
326-
try {
327-
mappedFile = new DefaultMmapFile(nextFilePath, this.mappedFileSize);
328-
} catch (IOException e) {
329-
logger.error("create mappedFile exception", e);
330-
}
328+
return tryCreateMappedFile(createOffset);
329+
}
331330

332-
if (mappedFile != null) {
333-
if (this.mappedFiles.isEmpty()) {
334-
mappedFile.setFirstCreateInQueue(true);
335-
}
336-
this.mappedFiles.add(mappedFile);
337-
}
331+
return mappedFileLast;
332+
}
333+
334+
protected MmapFile tryCreateMappedFile(long createOffset) {
335+
String nextFilePath = this.storePath + File.separator + DLedgerUtils.offset2FileName(createOffset);
336+
return doCreateMappedFile(nextFilePath);
337+
}
338338

339-
return mappedFile;
339+
protected MmapFile doCreateMappedFile(String nextFilePath) {
340+
MmapFile mappedFile = null;
341+
try {
342+
mappedFile = new DefaultMmapFile(nextFilePath, this.mappedFileSize);
343+
} catch (IOException e) {
344+
logger.error("create mappedFile exception", e);
340345
}
341346

342-
return mappedFileLast;
347+
if (mappedFile != null) {
348+
if (this.mappedFiles.isEmpty()) {
349+
mappedFile.setFirstCreateInQueue(true);
350+
}
351+
this.mappedFiles.add(mappedFile);
352+
}
353+
354+
return mappedFile;
343355
}
344356

345357
public MmapFile getLastMappedFile(final long startOffset) {

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

Lines changed: 121 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,121 @@
1+
/*
2+
* Copyright 2017-2020 the original author or authors.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* https://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package io.openmessaging.storage.dledger.store.file;
18+
19+
import io.netty.util.internal.StringUtil;
20+
import io.openmessaging.storage.dledger.DLedgerConfig;
21+
import io.openmessaging.storage.dledger.utils.DLedgerUtils;
22+
23+
import java.io.File;
24+
import java.util.ArrayList;
25+
import java.util.Arrays;
26+
import java.util.Collections;
27+
import java.util.HashSet;
28+
import java.util.List;
29+
import java.util.Set;
30+
import java.util.function.Supplier;
31+
32+
public class MultiPathMmapFileList extends MmapFileList {
33+
34+
private final Supplier<Set<String>> fullStorePathsSupplier;
35+
private final DLedgerConfig config;
36+
37+
public MultiPathMmapFileList(DLedgerConfig config, int mappedFileSize, Supplier<Set<String>> fullStorePathsSupplier) {
38+
super(config.getDataStorePath(), mappedFileSize);
39+
this.config = config;
40+
this.fullStorePathsSupplier = fullStorePathsSupplier;
41+
}
42+
43+
private Set<String> getPaths() {
44+
String[] paths = this.config.getDataStorePath().trim().split(DLedgerConfig.MULTI_PATH_SPLITTER);
45+
return new HashSet<>(Arrays.asList(paths));
46+
}
47+
48+
private Set<String> getReadonlyPaths() {
49+
String pathStr = config.getReadOnlyDataStoreDirs();
50+
if (StringUtil.isNullOrEmpty(pathStr)) {
51+
return Collections.emptySet();
52+
}
53+
String[] paths = pathStr.trim().split(DLedgerConfig.MULTI_PATH_SPLITTER);
54+
return new HashSet<>(Arrays.asList(paths));
55+
}
56+
57+
@Override
58+
public boolean load() {
59+
Set<String> storePathSet = getPaths();
60+
storePathSet.addAll(getReadonlyPaths());
61+
62+
List<File> files = new ArrayList<>();
63+
for (String path : storePathSet) {
64+
File dir = new File(path);
65+
File[] ls = dir.listFiles();
66+
if (ls != null) {
67+
Collections.addAll(files, ls);
68+
}
69+
}
70+
71+
return doLoad(files);
72+
}
73+
74+
@Override
75+
protected MmapFile tryCreateMappedFile(long createOffset) {
76+
long fileIdx = createOffset / this.getMappedFileSize();
77+
Set<String> storePath = getPaths();
78+
Set<String> readonlyPathSet = getReadonlyPaths();
79+
Set<String> fullStorePaths =
80+
fullStorePathsSupplier == null ? Collections.emptySet() : fullStorePathsSupplier.get();
81+
82+
83+
HashSet<String> availableStorePath = new HashSet<>(storePath);
84+
//do not create file in readonly store path.
85+
availableStorePath.removeAll(readonlyPathSet);
86+
87+
//do not create file is space is nearly full.
88+
availableStorePath.removeAll(fullStorePaths);
89+
90+
//if no store path left, fall back to writable store path.
91+
if (availableStorePath.isEmpty()) {
92+
availableStorePath = new HashSet<>(storePath);
93+
availableStorePath.removeAll(readonlyPathSet);
94+
}
95+
96+
String[] paths = availableStorePath.toArray(new String[]{});
97+
Arrays.sort(paths);
98+
String nextFilePath = paths[(int) (fileIdx % paths.length)] + File.separator
99+
+ DLedgerUtils.offset2FileName(createOffset);
100+
return doCreateMappedFile(nextFilePath);
101+
}
102+
103+
@Override
104+
public void destroy() {
105+
for (MmapFile mf : this.getMappedFiles()) {
106+
mf.destroy(1000 * 3);
107+
}
108+
this.getMappedFiles().clear();
109+
this.setFlushedWhere(0);
110+
111+
Set<String> storePathSet = getPaths();
112+
storePathSet.addAll(getReadonlyPaths());
113+
114+
for (String path : storePathSet) {
115+
File file = new File(path);
116+
if (file.isDirectory()) {
117+
file.delete();
118+
}
119+
}
120+
}
121+
}

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

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -90,4 +90,9 @@ public static double getDiskPartitionSpaceUsedPercent(final String path) {
9090
}
9191
return -1;
9292
}
93+
94+
public static boolean isPathExists(final String path) {
95+
File file = new File(path);
96+
return file.exists();
97+
}
9398
}

0 commit comments

Comments
 (0)