You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by he...@apache.org on 2023/05/16 01:19:13 UTC
[iotdb] branch tiered_storage updated: add cache write
This is an automated email from the ASF dual-hosted git repository.
heiming pushed a commit to branch tiered_storage
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/tiered_storage by this push:
new 9d47e76916b add cache write
9d47e76916b is described below
commit 9d47e76916b15ac87ecc26d1d17acaa424a2f3a3
Author: HeimingZ <zh...@qq.com>
AuthorDate: Tue May 16 09:18:39 2023 +0800
add cache write
---
.../apache/iotdb/os/cache/CacheFileManager.java | 79 ++++++++++++++++++++++
.../{OSFileCacheKey.java => IOSFileCache.java} | 13 ++--
.../org/apache/iotdb/os/cache/OSFileCache.java | 58 +++++++++++++---
.../org/apache/iotdb/os/cache/OSFileCacheKey.java | 53 ++++++++++++++-
.../apache/iotdb/os/cache/OSFileCacheValue.java | 22 +++---
.../iotdb/os/cache/RemoteFileCacheLoader.java | 52 --------------
.../org/apache/iotdb/os/fileSystem/OSFile.java | 9 ++-
.../apache/iotdb/os/fileSystem/OSTsFileInput.java | 4 +-
.../java/org/apache/iotdb/os/fileSystem/OSURI.java | 5 +-
.../apache/iotdb/os/io/ObjectStorageConnector.java | 2 +-
.../iotdb/os/io/aws/S3ObjectStorageConnector.java | 22 ++++--
11 files changed, 235 insertions(+), 84 deletions(-)
diff --git a/object-storage/src/main/java/org/apache/iotdb/os/cache/CacheFileManager.java b/object-storage/src/main/java/org/apache/iotdb/os/cache/CacheFileManager.java
new file mode 100644
index 00000000000..6dd208ac174
--- /dev/null
+++ b/object-storage/src/main/java/org/apache/iotdb/os/cache/CacheFileManager.java
@@ -0,0 +1,79 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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
+ *
+ * http://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 org.apache.iotdb.os.cache;
+
+import org.apache.iotdb.os.conf.ObjectStorageConfig;
+import org.apache.iotdb.os.conf.ObjectStorageDescriptor;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
+import java.nio.file.StandardOpenOption;
+import java.util.concurrent.atomic.AtomicLong;
+
+/** This class manages all io operations to the cache files */
+public class CacheFileManager {
+ private static final Logger logger = LoggerFactory.getLogger(CacheFileManager.class);
+ private static final String CACHE_FILE_SUFFIX = ".cache";
+ private static final String TMP_CACHE_FILE_SUFFIX = ".cache.tmp";
+ private static final ObjectStorageConfig config =
+ ObjectStorageDescriptor.getInstance().getConfig();
+ private String[] cacheDirs;
+ private AtomicLong logFileId = new AtomicLong(0);
+
+ public CacheFileManager(String[] cacheDirs) {
+ this.cacheDirs = cacheDirs;
+ }
+
+ private long getNextCacheFileId() {
+ return logFileId.getAndIncrement();
+ }
+
+ private File getTmpCacheFile(long id) {
+ long dirId = id % cacheDirs.length;
+ return new File(cacheDirs[(int) dirId], id + TMP_CACHE_FILE_SUFFIX);
+ }
+
+ private File getCacheFile(long id) {
+ long dirId = id % cacheDirs.length;
+ return new File(cacheDirs[(int) dirId], id + CACHE_FILE_SUFFIX);
+ }
+
+ /** Persist data, return null when failing to persist data */
+ public OSFileCacheValue persist(OSFileCacheKey key, byte[] data) {
+ OSFileCacheValue res = null;
+ long cacheFileId = getNextCacheFileId();
+ File tmpCacheFile = getTmpCacheFile(cacheFileId);
+ try (FileChannel channel =
+ FileChannel.open(tmpCacheFile.toPath(), StandardOpenOption.CREATE_NEW)) {
+ ByteBuffer meta = key.serializeToByteBuffer();
+ channel.write(meta);
+ channel.write(ByteBuffer.wrap(data));
+ res = new OSFileCacheValue(tmpCacheFile, meta.capacity(), data.length);
+ } catch (IOException e) {
+ logger.error("Fail to persist data to cache file {}", tmpCacheFile, e);
+ tmpCacheFile.delete();
+ }
+ return tmpCacheFile.renameTo(getCacheFile(cacheFileId)) ? res : null;
+ }
+}
diff --git a/object-storage/src/main/java/org/apache/iotdb/os/cache/OSFileCacheKey.java b/object-storage/src/main/java/org/apache/iotdb/os/cache/IOSFileCache.java
similarity index 71%
copy from object-storage/src/main/java/org/apache/iotdb/os/cache/OSFileCacheKey.java
copy to object-storage/src/main/java/org/apache/iotdb/os/cache/IOSFileCache.java
index 8757483432c..d780083ff66 100644
--- a/object-storage/src/main/java/org/apache/iotdb/os/cache/OSFileCacheKey.java
+++ b/object-storage/src/main/java/org/apache/iotdb/os/cache/IOSFileCache.java
@@ -18,9 +18,14 @@
*/
package org.apache.iotdb.os.cache;
-import org.apache.iotdb.os.fileSystem.OSFile;
+import org.apache.iotdb.os.fileSystem.OSURI;
-public class OSFileCacheKey {
- private OSFile file;
- private long position;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.channels.FileChannel;
+
+public interface IOSFileCache {
+ InputStream getAsInputSteam(OSURI file, long startPosition) throws IOException;
+
+ FileChannel getLocalCacheFileChannel(OSURI file, long startPosition) throws IOException;
}
diff --git a/object-storage/src/main/java/org/apache/iotdb/os/cache/OSFileCache.java b/object-storage/src/main/java/org/apache/iotdb/os/cache/OSFileCache.java
index a2447758463..1525f382838 100644
--- a/object-storage/src/main/java/org/apache/iotdb/os/cache/OSFileCache.java
+++ b/object-storage/src/main/java/org/apache/iotdb/os/cache/OSFileCache.java
@@ -21,24 +21,46 @@ package org.apache.iotdb.os.cache;
import org.apache.iotdb.os.conf.ObjectStorageConfig;
import org.apache.iotdb.os.conf.ObjectStorageDescriptor;
+import org.apache.iotdb.os.fileSystem.OSURI;
+import org.apache.iotdb.os.io.ObjectStorageConnector;
+import org.apache.iotdb.os.io.aws.S3ObjectStorageConnector;
+import com.github.benmanes.caffeine.cache.CacheLoader;
import com.github.benmanes.caffeine.cache.Caffeine;
import com.github.benmanes.caffeine.cache.LoadingCache;
import com.github.benmanes.caffeine.cache.Weigher;
+import org.checkerframework.checker.nullness.qual.NonNull;
+import org.checkerframework.checker.nullness.qual.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.io.InputStream;
-import java.nio.channels.Channels;
import java.nio.channels.FileChannel;
-public class OSFileCache {
+public class OSFileCache implements IOSFileCache {
private static final Logger logger = LoggerFactory.getLogger(OSFileCache.class);
private static final ObjectStorageConfig config =
ObjectStorageDescriptor.getInstance().getConfig();
+ private static final ObjectStorageConnector connector;
+ static {
+ switch (config.getOsType()) {
+ case AWS_S3:
+ connector = new S3ObjectStorageConnector();
+ break;
+ default:
+ connector = null;
+ }
+ }
+
+ /**
+ * persistent LRU cache for remote TsFile, value is loaded successfully when it has been stored on
+ * the disk
+ */
private LoadingCache<OSFileCacheKey, OSFileCacheValue> remotePos2LocalCacheFile;
+ /** manage all io operations to the cache files */
+ private CacheFileManager cacheFileManager = new CacheFileManager(config.getCacheDirs());
private OSFileCache() {
remotePos2LocalCacheFile =
@@ -46,21 +68,41 @@ public class OSFileCache {
.maximumWeight(config.getCacheMaxDiskUsage())
.weigher(
(Weigher<OSFileCacheKey, OSFileCacheValue>)
- (key, value) -> Math.toIntExact(value.getOccupiedLength()))
- .build(new RemoteFileCacheLoader());
+ (key, value) -> value.getOccupiedLength())
+ .removalListener(
+ (key, value, cuase) -> {
+ if (value != null) {
+ value.clear();
+ }
+ })
+ .build(new OSFileCacheLoader());
}
- public InputStream getAsInputSteam(String fileName, long startPosition) throws IOException {
- FileChannel cacheFileChannel = getLocalCacheFileChannel(fileName, startPosition);
- return Channels.newInputStream(cacheFileChannel);
+ @Override
+ public InputStream getAsInputSteam(OSURI file, long startPosition) throws IOException {
+ return null;
}
- private FileChannel getLocalCacheFileChannel(String fileName, long startPosition) {
+ @Override
+ public FileChannel getLocalCacheFileChannel(OSURI file, long startPosition) {
// 根据 fileName 和 startPosition 计算出对应的本地文件路径，并返回对应的 FileChannel
// 如果是使用一个 CacheFile, 则寻找到对应的位置，可能需要封装一个自己的 FileChannel 防止读多
return null;
}
+ void put(OSFileCacheKey key, OSFileCacheValue value) {
+ remotePos2LocalCacheFile.put(key, value);
+ }
+
+ class OSFileCacheLoader implements CacheLoader<OSFileCacheKey, OSFileCacheValue> {
+ @Override
+ public @Nullable OSFileCacheValue load(@NonNull OSFileCacheKey key) throws Exception {
+ byte[] data =
+ connector.getRemoteFile(key.getFile().toOSURI(), key.getStartPosition(), key.getLength());
+ return cacheFileManager.persist(key, data);
+ }
+ }
+
public static OSFileCache getInstance() {
return OSFileCache.InstanceHolder.INSTANCE;
}
diff --git a/object-storage/src/main/java/org/apache/iotdb/os/cache/OSFileCacheKey.java b/object-storage/src/main/java/org/apache/iotdb/os/cache/OSFileCacheKey.java
index 8757483432c..ff157872635 100644
--- a/object-storage/src/main/java/org/apache/iotdb/os/cache/OSFileCacheKey.java
+++ b/object-storage/src/main/java/org/apache/iotdb/os/cache/OSFileCacheKey.java
@@ -20,7 +20,56 @@ package org.apache.iotdb.os.cache;
import org.apache.iotdb.os.fileSystem.OSFile;
-public class OSFileCacheKey {
+import java.io.Serializable;
+import java.nio.ByteBuffer;
+import java.util.Objects;
+
+public class OSFileCacheKey implements Serializable {
private OSFile file;
- private long position;
+ private long startPosition;
+ private int length;
+
+ public OSFileCacheKey(OSFile file, long startPosition, int length) {
+ this.file = file;
+ this.startPosition = startPosition;
+ this.length = length;
+ }
+
+ public ByteBuffer serializeToByteBuffer() {
+ byte[] pathBytes = file.toString().getBytes();
+ ByteBuffer buffer = ByteBuffer.allocate(Integer.BYTES + pathBytes.length + Integer.BYTES);
+ buffer.putInt(pathBytes.length);
+ buffer.put(pathBytes);
+ buffer.putInt(length);
+ return buffer;
+ }
+
+ public OSFile getFile() {
+ return file;
+ }
+
+ public long getStartPosition() {
+ return startPosition;
+ }
+
+ public int getLength() {
+ return length;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(file, startPosition, length);
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (this == obj) {
+ return true;
+ }
+ if (obj == null || getClass() != obj.getClass()) {
+ return false;
+ }
+ OSFileCacheKey that = (OSFileCacheKey) obj;
+ return file.equals(that.file) && startPosition == that.startPosition && length == that.length;
+ }
}
diff --git a/object-storage/src/main/java/org/apache/iotdb/os/cache/OSFileCacheValue.java b/object-storage/src/main/java/org/apache/iotdb/os/cache/OSFileCacheValue.java
index 70df017d0a2..e9afbacacd8 100644
--- a/object-storage/src/main/java/org/apache/iotdb/os/cache/OSFileCacheValue.java
+++ b/object-storage/src/main/java/org/apache/iotdb/os/cache/OSFileCacheValue.java
@@ -19,36 +19,42 @@
package org.apache.iotdb.os.cache;
+import java.io.File;
+
public class OSFileCacheValue {
- private String fileName;
+ private File cacheFile;
// 如果每个块用一个文件来存储，则该值一直为 0
// 如果使用一个大文件存储所有块，则该值为大文件中的起点
private long startPosition;
// 如果每个块用一个文件来存储，则该值一直为该文件的大小
// 如果使用一个大文件存储所有块，则该值为该块的实际长度
- private long length;
+ private int length;
- public OSFileCacheValue(String fileName, long startPosition, long length) {
- this.fileName = fileName;
+ public OSFileCacheValue(File cacheFile, long startPosition, int length) {
+ this.cacheFile = cacheFile;
this.startPosition = startPosition;
this.length = length;
}
- public String getFileName() {
- return fileName;
+ public File getCacheFile() {
+ return cacheFile;
}
public long getStartPosition() {
return startPosition;
}
- public long getLength() {
+ public int getLength() {
return length;
}
- public long getOccupiedLength() {
+ public int getOccupiedLength() {
// 如果使用多个文件，则返回该文件的大小
// 如果使用一个文件，则返回每个槽的大小
return length;
}
+
+ public void clear() {
+ cacheFile.delete();
+ }
}
diff --git a/object-storage/src/main/java/org/apache/iotdb/os/cache/RemoteFileCacheLoader.java b/object-storage/src/main/java/org/apache/iotdb/os/cache/RemoteFileCacheLoader.java
deleted file mode 100644
index 9d6702f7440..00000000000
--- a/object-storage/src/main/java/org/apache/iotdb/os/cache/RemoteFileCacheLoader.java
+++ /dev/null
@@ -1,52 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you 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
- *
- * http://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 org.apache.iotdb.os.cache;
-
-import org.apache.iotdb.os.conf.ObjectStorageConfig;
-import org.apache.iotdb.os.conf.ObjectStorageDescriptor;
-import org.apache.iotdb.os.io.ObjectStorageConnector;
-import org.apache.iotdb.os.io.aws.S3ObjectStorageConnector;
-
-import com.github.benmanes.caffeine.cache.CacheLoader;
-import org.checkerframework.checker.nullness.qual.NonNull;
-import org.checkerframework.checker.nullness.qual.Nullable;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class RemoteFileCacheLoader implements CacheLoader<OSFileCacheKey, OSFileCacheValue> {
- private static final Logger logger = LoggerFactory.getLogger(RemoteFileCacheLoader.class);
- private static final ObjectStorageConfig config =
- ObjectStorageDescriptor.getInstance().getConfig();
- private static final ObjectStorageConnector connector;
-
- static {
- switch (config.getOsType()) {
- case AWS_S3:
- connector = new S3ObjectStorageConnector();
- break;
- default:
- connector = null;
- }
- }
-
- @Override
- public @Nullable OSFileCacheValue load(@NonNull OSFileCacheKey key) throws Exception {
- return null;
- }
-}
diff --git a/object-storage/src/main/java/org/apache/iotdb/os/fileSystem/OSFile.java b/object-storage/src/main/java/org/apache/iotdb/os/fileSystem/OSFile.java
index 64bff215538..9f61ef431d7 100644
--- a/object-storage/src/main/java/org/apache/iotdb/os/fileSystem/OSFile.java
+++ b/object-storage/src/main/java/org/apache/iotdb/os/fileSystem/OSFile.java
@@ -355,7 +355,10 @@ public class OSFile extends File {
@Override
public boolean equals(Object obj) {
- if (!(obj instanceof OSFile)) {
+ if (this == obj) {
+ return true;
+ }
+ if (obj == null || getClass() != obj.getClass()) {
return false;
}
OSFile other = (OSFile) obj;
@@ -377,6 +380,10 @@ public class OSFile extends File {
throw new UnsupportedOperationException(UNSUPPORT_OPERATION);
}
+ public OSURI toOSURI() {
+ return osUri;
+ }
+
public BufferedReader getBufferedReader() {
try {
return new BufferedReader(new InputStreamReader(connector.getInputStream(osUri)));
diff --git a/object-storage/src/main/java/org/apache/iotdb/os/fileSystem/OSTsFileInput.java b/object-storage/src/main/java/org/apache/iotdb/os/fileSystem/OSTsFileInput.java
index 18df7c4e270..c42bc104651 100644
--- a/object-storage/src/main/java/org/apache/iotdb/os/fileSystem/OSTsFileInput.java
+++ b/object-storage/src/main/java/org/apache/iotdb/os/fileSystem/OSTsFileInput.java
@@ -33,7 +33,7 @@ import java.nio.channels.FileChannel;
public class OSTsFileInput implements TsFileInput {
private static final Logger logger = LoggerFactory.getLogger(OSTsFileInput.class);
- private String osFileName;
+ private OSFile file;
private OSFileCache cache = OSFileCache.getInstance();
@Override
@@ -101,6 +101,6 @@ public class OSTsFileInput implements TsFileInput {
}
public InputStream getNextInputStream(long position) throws IOException {
- return cache.getAsInputSteam(osFileName, position);
+ return cache.getAsInputSteam(file.toOSURI(), position);
}
}
diff --git a/object-storage/src/main/java/org/apache/iotdb/os/fileSystem/OSURI.java b/object-storage/src/main/java/org/apache/iotdb/os/fileSystem/OSURI.java
index 28e6b03a5cb..66a5b10424e 100644
--- a/object-storage/src/main/java/org/apache/iotdb/os/fileSystem/OSURI.java
+++ b/object-storage/src/main/java/org/apache/iotdb/os/fileSystem/OSURI.java
@@ -62,7 +62,10 @@ public class OSURI {
}
public boolean equals(Object obj) {
- if (!(obj instanceof OSURI)) {
+ if (this == obj) {
+ return true;
+ }
+ if (obj == null || getClass() != obj.getClass()) {
return false;
}
OSURI other = (OSURI) obj;
diff --git a/object-storage/src/main/java/org/apache/iotdb/os/io/ObjectStorageConnector.java b/object-storage/src/main/java/org/apache/iotdb/os/io/ObjectStorageConnector.java
index cbf004329e6..c396eaeb816 100644
--- a/object-storage/src/main/java/org/apache/iotdb/os/io/ObjectStorageConnector.java
+++ b/object-storage/src/main/java/org/apache/iotdb/os/io/ObjectStorageConnector.java
@@ -42,5 +42,5 @@ public interface ObjectStorageConnector {
void putLocalFile(OSURI osUri, File lcoalFile) throws ObjectStorageException;
- byte[] getRemoteFile(OSURI osUri, long position, int size) throws ObjectStorageException;
+ byte[] getRemoteFile(OSURI osUri, long position, int len) throws ObjectStorageException;
}
diff --git a/object-storage/src/main/java/org/apache/iotdb/os/io/aws/S3ObjectStorageConnector.java b/object-storage/src/main/java/org/apache/iotdb/os/io/aws/S3ObjectStorageConnector.java
index 92cfa90d767..699ffda13d2 100644
--- a/object-storage/src/main/java/org/apache/iotdb/os/io/aws/S3ObjectStorageConnector.java
+++ b/object-storage/src/main/java/org/apache/iotdb/os/io/aws/S3ObjectStorageConnector.java
@@ -24,6 +24,7 @@ import org.apache.iotdb.os.fileSystem.OSURI;
import org.apache.iotdb.os.io.IMetaData;
import org.apache.iotdb.os.io.ObjectStorageConnector;
+import software.amazon.awssdk.core.ResponseBytes;
import software.amazon.awssdk.core.sync.RequestBody;
import software.amazon.awssdk.regions.Region;
import software.amazon.awssdk.services.s3.S3Client;
@@ -31,6 +32,7 @@ import software.amazon.awssdk.services.s3.model.CopyObjectRequest;
import software.amazon.awssdk.services.s3.model.DeleteObjectRequest;
import software.amazon.awssdk.services.s3.model.DeleteObjectResponse;
import software.amazon.awssdk.services.s3.model.GetObjectRequest;
+import software.amazon.awssdk.services.s3.model.GetObjectResponse;
import software.amazon.awssdk.services.s3.model.HeadObjectRequest;
import software.amazon.awssdk.services.s3.model.HeadObjectResponse;
import software.amazon.awssdk.services.s3.model.ListObjectsRequest;
@@ -42,6 +44,7 @@ import java.io.File;
import java.io.InputStream;
public class S3ObjectStorageConnector implements ObjectStorageConnector {
+ private static final String RANGE_FORMAT = "%d-%d";
private S3Client s3Client;
public S3ObjectStorageConnector() {
@@ -161,11 +164,20 @@ public class S3ObjectStorageConnector implements ObjectStorageConnector {
}
@Override
- public byte[] getRemoteFile(OSURI osUri, long position, int size) throws ObjectStorageException {
- GetObjectRequest req =
- GetObjectRequest.builder().bucket(osUri.getBucket()).key(osUri.getKey()).build();
- s3Client.getObject(req);
- return new byte[0];
+ public byte[] getRemoteFile(OSURI osUri, long position, int len) throws ObjectStorageException {
+ String rangeStr = String.format(RANGE_FORMAT, position, position + len - 1);
+ try {
+ GetObjectRequest req =
+ GetObjectRequest.builder()
+ .bucket(osUri.getBucket())
+ .key(osUri.getKey())
+ .range(rangeStr)
+ .build();
+ ResponseBytes<GetObjectResponse> resp = s3Client.getObjectAsBytes(req);
+ return resp.asByteArray();
+ } catch (S3Exception e) {
+ throw new ObjectStorageException(e);
+ }
}
public void close() {