You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@crail.apache.org by pe...@apache.org on 2018/04/04 09:21:03 UTC
[1/3] incubator-crail git commit: StorageClient: provide statistics
and buffer cache
Repository: incubator-crail
Updated Branches:
refs/heads/master 56db1c776 -> c053477a7
StorageClient: provide statistics and buffer cache
Initialize the storage client with statistics and the buffer cache
to be used in the client.
https://issues.apache.org/jira/projects/CRAIL/issues/CRAIL-20
Signed-off-by: Jonas Pfefferle <pe...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/incubator-crail/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-crail/commit/efd44e75
Tree: http://git-wip-us.apache.org/repos/asf/incubator-crail/tree/efd44e75
Diff: http://git-wip-us.apache.org/repos/asf/incubator-crail/diff/efd44e75
Branch: refs/heads/master
Commit: efd44e75dfbdfc3dc0c606dead4f44c812f359a5
Parents: 56db1c7
Author: Jonas Pfefferle <pe...@apache.org>
Authored: Thu Mar 29 14:27:37 2018 +0200
Committer: Jonas Pfefferle <pe...@apache.org>
Committed: Thu Mar 29 14:27:37 2018 +0200
----------------------------------------------------------------------
.../java/org/apache/crail/CrailBufferCache.java | 6 +
.../main/java/org/apache/crail/CrailStore.java | 18 +-
.../org/apache/crail/core/CoreDataStore.java | 230 +++++++++----------
.../org/apache/crail/storage/StorageClient.java | 20 +-
.../crail/storage/tcp/TcpStorageClient.java | 9 +-
.../crail/storage/nvmf/NvmfStorageClient.java | 11 +-
.../crail/storage/rdma/RdmaConstants.java | 48 ++--
.../crail/storage/rdma/RdmaStorageClient.java | 23 +-
8 files changed, 193 insertions(+), 172 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/efd44e75/client/src/main/java/org/apache/crail/CrailBufferCache.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/crail/CrailBufferCache.java b/client/src/main/java/org/apache/crail/CrailBufferCache.java
new file mode 100644
index 0000000..ca9d6d9
--- /dev/null
+++ b/client/src/main/java/org/apache/crail/CrailBufferCache.java
@@ -0,0 +1,6 @@
+package org.apache.crail;
+
+public interface CrailBufferCache {
+ CrailBuffer allocateBuffer() throws Exception;
+ void freeBuffer(CrailBuffer buffer) throws Exception;
+}
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/efd44e75/client/src/main/java/org/apache/crail/CrailStore.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/crail/CrailStore.java b/client/src/main/java/org/apache/crail/CrailStore.java
index 256c530..3abc894 100644
--- a/client/src/main/java/org/apache/crail/CrailStore.java
+++ b/client/src/main/java/org/apache/crail/CrailStore.java
@@ -28,21 +28,19 @@ import org.apache.crail.core.CoreDataStore;
import org.apache.crail.utils.CrailUtils;
import org.slf4j.Logger;
-public abstract class CrailStore {
+public abstract class CrailStore implements CrailBufferCache {
private static final Logger LOG = CrailUtils.getLogger();
private static AtomicLong referenceCounter = new AtomicLong(0);
private static CrailStore instance = null;
-
+
public abstract Upcoming<CrailNode> create(String path, CrailNodeType type, CrailStorageClass storageClass, CrailLocationClass locationClass, boolean enumerable) throws Exception;
public abstract Upcoming<CrailNode> lookup(String path) throws Exception;
public abstract Upcoming<CrailNode> rename(String srcPath, String dstPath) throws Exception;
public abstract Upcoming<CrailNode> delete(String path, boolean recursive) throws Exception;
- public abstract CrailBuffer allocateBuffer() throws Exception;
- public abstract void freeBuffer(CrailBuffer buffer) throws Exception;
public abstract CrailStatistics getStatistics();
public abstract CrailLocationClass getLocationClass();
protected abstract void closeFileSystem() throws Exception;
-
+
public void close() throws Exception {
synchronized(referenceCounter){
if (CrailConstants.SINGLETON){
@@ -62,12 +60,12 @@ public abstract class CrailStore {
closeFileSystem();
} catch (Exception e){
throw new IOException(e);
- }
+ }
}
-
+
}
}
-
+
public static CrailStore newInstance(CrailConfiguration conf) throws Exception {
synchronized(referenceCounter){
boolean isSingleton = conf.getBoolean(CrailConstants.SINGLETON_KEY, false);
@@ -82,10 +80,10 @@ public abstract class CrailStore {
return instance;
}
} else {
- LOG.info("creating non-singleton crail file system");
+ LOG.info("creating non-singleton crail file system");
return new CoreDataStore(conf);
}
}
- }
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/efd44e75/client/src/main/java/org/apache/crail/core/CoreDataStore.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/crail/core/CoreDataStore.java b/client/src/main/java/org/apache/crail/core/CoreDataStore.java
index ce68633..59c7da9 100644
--- a/client/src/main/java/org/apache/crail/core/CoreDataStore.java
+++ b/client/src/main/java/org/apache/crail/core/CoreDataStore.java
@@ -74,55 +74,55 @@ import org.slf4j.Logger;
public class CoreDataStore extends CrailStore {
private static final Logger LOG = CrailUtils.getLogger();
private static AtomicInteger fsCount = new AtomicInteger(0);
-
+
//namenode operations
private RpcClient rpcClient;
private RpcConnection rpcConnection;
-
+
//datanode operations
private EndpointCache datanodeEndpointCache;
-
+
private AtomicLong streamCounter;
private ConcurrentHashMap<Long, CoreInputStream> openInputStreams;
private ConcurrentHashMap<Long, CoreOutputStream> openOutputStreams;
-
+
private BlockCache blockCache;
private NextBlockCache nextBlockCache;
private BufferCache bufferCache;
private BufferCheckpoint bufferCheckpoint;
private ConcurrentHashMap<String, String> locationMap;
-
+
private boolean isOpen;
private int fsId;
private CrailLocationClass localClass;
-
+
private CoreIOStatistics ioStatsIn;
private CoreIOStatistics ioStatsOut;
private CoreStreamStatistics streamStats;
private CrailStatistics statistics;
-
+
public CoreDataStore(CrailConfiguration conf) throws Exception {
CrailConstants.updateConstants(conf);
CrailConstants.printConf();
- CrailConstants.verify();
-
+ CrailConstants.verify();
+
//Datanodes
StringTokenizer tokenizer = new StringTokenizer(CrailConstants.STORAGE_TYPES, ",");
- LinkedList<StorageClient> dataNodeClients = new LinkedList<StorageClient>();
+ LinkedList<StorageClient> dataNodeClients = new LinkedList<StorageClient>();
while (tokenizer.hasMoreTokens()){
String name = tokenizer.nextToken();
StorageClient dataNode = StorageClient.createInstance(name);
- dataNode.init(conf, null);
+ dataNode.init(getStatistics(), this, conf, null);
dataNode.printConf(LOG);
dataNodeClients.add(dataNode);
}
this.datanodeEndpointCache = new EndpointCache(fsId, dataNodeClients);
-
+
//Namenode
InetSocketAddress nnAddr = CrailUtils.getNameNodeAddress();
this.rpcClient = RpcClient.createInstance(CrailConstants.NAMENODE_RPC_TYPE);
rpcClient.init(conf, null);
- rpcClient.printConf(LOG);
+ rpcClient.printConf(LOG);
ConcurrentLinkedQueue<InetSocketAddress> namenodeList = CrailUtils.getNameNodeList();
ConcurrentLinkedQueue<RpcConnection> connectionList = new ConcurrentLinkedQueue<RpcConnection>();
while(!namenodeList.isEmpty()){
@@ -135,8 +135,8 @@ public class CoreDataStore extends CrailStore {
} else {
this.rpcConnection = new RpcDispatcher(connectionList);
}
- LOG.info("connected to namenode(s) " + rpcConnection);
-
+ LOG.info("connected to namenode(s) " + rpcConnection);
+
//Client
this.fsId = fsCount.getAndIncrement();
this.localClass = CrailUtils.getLocationClass();
@@ -150,7 +150,7 @@ public class CoreDataStore extends CrailStore {
this.bufferCheckpoint = new BufferCheckpoint();
this.locationMap = new ConcurrentHashMap<String, String>();
CrailUtils.parseMap(CrailConstants.LOCATION_MAP, locationMap);
-
+
this.statistics = new CrailStatistics();
this.ioStatsIn = new CoreIOStatistics("core/input");
statistics.addProvider(ioStatsIn);
@@ -161,18 +161,18 @@ public class CoreDataStore extends CrailStore {
statistics.addProvider(bufferCache);
statistics.addProvider(datanodeEndpointCache);
}
-
+
public Upcoming<CrailNode> create(String path, CrailNodeType type, CrailStorageClass storageClass, CrailLocationClass locationClass, boolean enumerable) throws Exception {
FileName name = new FileName(path);
-
+
if (CrailConstants.DEBUG){
LOG.info("createNode: name " + path + ", type " + type + ", storageAffinity " + storageClass + ", locationAffinity " + locationClass);
}
RpcFuture<RpcCreateFile> fileRes = rpcConnection.createFile(name, type, storageClass.value(), locationClass.value(), enumerable);
return new CreateNodeFuture(this, path, type, fileRes);
- }
-
+ }
+
CoreNode _createNode(String path, CrailNodeType type, RpcCreateFile fileRes) throws Exception {
if (fileRes.getError() == RpcErrors.ERR_PARENT_MISSING){
throw new IOException("createNode: " + RpcErrors.messages[fileRes.getError()] + ", name " + path);
@@ -181,8 +181,8 @@ public class CoreDataStore extends CrailStore {
} else if (fileRes.getError() != RpcErrors.ERR_OK){
LOG.info("createNode: " + RpcErrors.messages[fileRes.getError()] + ", name " + path);
throw new IOException("createNode: " + RpcErrors.messages[fileRes.getError()] + ", error " + fileRes.getError());
- }
-
+ }
+
FileInfo fileInfo = fileRes.getFile();
FileInfo dirInfo = fileRes.getParent();
if (fileInfo == null || dirInfo == null){
@@ -195,7 +195,7 @@ public class CoreDataStore extends CrailStore {
blockCache.remove(fileInfo.getFd());
nextBlockCache.remove(fileInfo.getFd());
CoreNode node = CoreNode.create(this, fileInfo, path);
-
+
BlockInfo fileBlock = fileRes.getFileBlock();
getBlockCache(fileInfo.getFd()).put(CoreSubOperation.createKey(fileInfo.getFd(), 0), fileBlock);
@@ -206,25 +206,25 @@ public class CoreDataStore extends CrailStore {
CoreSyncOperation syncOperation = getSyncOperation(dirInfo, fileInfo, path, true);
node.addSyncOperation(syncOperation);
}
-
+
if (CrailConstants.DEBUG){
LOG.info("createFile: name " + path + ", success, fd " + fileInfo.getFd() + ", token " + fileInfo.getToken());
}
-
+
return node;
- }
-
+ }
+
public Upcoming<CrailNode> lookup(String path) throws Exception {
FileName name = new FileName(path);
-
+
if (CrailConstants.DEBUG){
LOG.info("lookupDirectory: path " + path);
}
-
+
RpcFuture<RpcGetFile> fileRes = rpcConnection.getFile(name, false);
return new LookupNodeFuture(this, path, fileRes);
- }
-
+ }
+
CoreNode _lookupNode(RpcGetFile fileRes, String path) throws Exception {
if (fileRes.getError() == RpcErrors.ERR_GET_FILE_FAILED){
return null;
@@ -232,10 +232,10 @@ public class CoreDataStore extends CrailStore {
else if (fileRes.getError() != RpcErrors.ERR_OK){
LOG.info("lookupDirectory: " + RpcErrors.messages[fileRes.getError()]);
return null;
- }
-
+ }
+
FileInfo fileInfo = fileRes.getFile();
-
+
CoreNode node = null;
if (fileInfo != null){
if (CrailConstants.DEBUG){
@@ -243,25 +243,25 @@ public class CoreDataStore extends CrailStore {
}
BlockInfo fileBlock = fileRes.getFileBlock();
getBlockCache(fileInfo.getFd()).put(CoreSubOperation.createKey(fileInfo.getFd(), 0), fileBlock);
-
+
node = CoreNode.create(this, fileInfo, path);
- }
+ }
return node;
- }
-
+ }
+
public Upcoming<CrailNode> rename(String src, String dst) throws Exception {
FileName srcPath = new FileName(src);
FileName dstPath = new FileName(dst);
-
+
if (CrailConstants.DEBUG){
LOG.info("rename: srcname " + src + ", dstname " + dst);
}
-
+
RpcFuture<RpcRenameFile> renameRes = rpcConnection.renameFile(srcPath, dstPath);
return new RenameNodeFuture(this, src, dst, renameRes);
}
-
+
CrailNode _rename(RpcRenameFile renameRes, String src, String dst) throws Exception {
if (renameRes.getError() == RpcErrors.ERR_SRC_FILE_NOT_FOUND){
LOG.info("rename: " + RpcErrors.messages[renameRes.getError()]);
@@ -277,48 +277,48 @@ public class CoreDataStore extends CrailStore {
}
if (renameRes.getError() != RpcErrors.ERR_OK){
LOG.info("rename: " + RpcErrors.messages[renameRes.getError()]);
- throw new IOException(RpcErrors.messages[renameRes.getError()]);
- }
+ throw new IOException(RpcErrors.messages[renameRes.getError()]);
+ }
if (renameRes.getDstParent().getCapacity() < renameRes.getDstFile().getDirOffset() + CrailConstants.DIRECTORY_RECORD){
LOG.info("rename: parent capacity does not match dst file offset, capacity " + renameRes.getDstParent().getCapacity() + ", offset " + renameRes.getDstFile().getDirOffset());
- }
-
+ }
+
FileInfo srcParent = renameRes.getSrcParent();
FileInfo srcFile = renameRes.getSrcFile();
FileInfo dstDir = renameRes.getDstParent();
FileInfo dstFile = renameRes.getDstFile();
-
+
BlockInfo srcBlock = renameRes.getSrcBlock();
getBlockCache(srcParent.getFd()).put(CoreSubOperation.createKey(srcParent.getFd(), srcFile.getDirOffset()), srcBlock);
BlockInfo dirBlock = renameRes.getDstBlock();
- getBlockCache(dstDir.getFd()).put(CoreSubOperation.createKey(dstDir.getFd(), dstFile.getDirOffset()), dirBlock);
-
+ getBlockCache(dstDir.getFd()).put(CoreSubOperation.createKey(dstDir.getFd(), dstFile.getDirOffset()), dirBlock);
+
CoreSyncOperation syncOperationSrc = getSyncOperation(srcParent, srcFile, src, false);
CoreSyncOperation syncOperationDst = getSyncOperation(dstDir, dstFile, dst, true);
-
+
blockCache.remove(srcFile.getFd());
-
+
if (CrailConstants.DEBUG){
LOG.info("rename: srcname " + src + ", dstname " + dst + ", success");
}
-
+
CoreNode node = CoreNode.create(this, dstFile, dst);
node.addSyncOperation(syncOperationSrc);
node.addSyncOperation(syncOperationDst);
return node;
}
-
+
public Upcoming<CrailNode> delete(String path, boolean recursive) throws Exception {
FileName name = new FileName(path);
-
+
if (CrailConstants.DEBUG){
LOG.info("delete: name " + path + ", recursive " + recursive);
}
RpcFuture<RpcDeleteFile> fileRes = rpcConnection.removeFile(name, recursive);
return new DeleteNodeFuture(this, path, recursive, fileRes);
- }
-
+ }
+
CrailNode _delete(RpcDeleteFile fileRes, String path, boolean recursive) throws Exception {
if (fileRes.getError() == RpcErrors.ERR_HAS_CHILDREN) {
LOG.info("delete: " + RpcErrors.messages[fileRes.getError()]);
@@ -327,31 +327,31 @@ public class CoreDataStore extends CrailStore {
if (fileRes.getError() != RpcErrors.ERR_OK) {
LOG.info("delete: " + RpcErrors.messages[fileRes.getError()]);
return null;
- }
-
+ }
+
FileInfo fileInfo = fileRes.getFile();
FileInfo dirInfo = fileRes.getParent();
-
+
CoreSyncOperation syncOperation = getSyncOperation(dirInfo, fileInfo, path, false);
-
+
blockCache.remove(fileInfo.getFd());
-
+
if (CrailConstants.DEBUG){
LOG.info("delete: name " + path + ", recursive " + recursive + ", success");
}
-
+
CoreNode node = CoreNode.create(this, fileInfo, path);
node.addSyncOperation(syncOperation);
return node;
- }
-
+ }
+
public DirectoryInputStream listEntries(String name) throws Exception {
return _listEntries(name, CrailConstants.DIRECTORY_RANDOMIZE);
}
-
+
public DirectoryInputStream _listEntries(String name, boolean randomize) throws Exception {
FileName directory = new FileName(name);
-
+
if (CrailConstants.DEBUG){
LOG.info("getDirectoryList: " + name);
}
@@ -361,23 +361,23 @@ public class CoreDataStore extends CrailStore {
LOG.info("getDirectoryList: " + RpcErrors.messages[fileRes.getError()]);
throw new FileNotFoundException(RpcErrors.messages[fileRes.getError()]);
}
-
+
FileInfo dirInfo = fileRes.getFile();
if (!dirInfo.getType().isContainer()){
LOG.info("getDirectoryList: " + RpcErrors.messages[RpcErrors.ERR_FILE_IS_NOT_DIR]);
throw new FileNotFoundException(RpcErrors.messages[RpcErrors.ERR_FILE_IS_NOT_DIR]);
}
-
+
CoreDirectory dirFile = new CoreDirectory(this, dirInfo, name);
DirectoryInputStream inputStream = dirFile.getDirectoryInputStream(randomize);
return inputStream;
- }
-
+ }
+
public CrailBlockLocation[] getBlockLocations(String path, long start, long len) throws Exception {
if (CrailConstants.DEBUG){
LOG.info("location: path " + path + ", start " + start + ", len " + len);
- }
-
+ }
+
if (path == null) {
LOG.info("Path null");
return null;
@@ -386,8 +386,8 @@ public class CoreDataStore extends CrailStore {
LOG.info("Start or len invalid");
throw new IOException("Invalid start or len parameter");
}
-
-
+
+
FileName name = new FileName(path);
long rangeStart = CrailUtils.blockStartAddress(start);
long range = start + len - CrailUtils.blockStartAddress(start);
@@ -398,14 +398,14 @@ public class CoreDataStore extends CrailStore {
CoreBlockLocation[] blockLocations = new CoreBlockLocation[(int) blockCount];
HashMap<Long, DataNodeInfo> dataNodeSet = new HashMap<Long, DataNodeInfo>();
HashMap<Long, DataNodeInfo> offset2DataNode = new HashMap<Long, DataNodeInfo>();
-
+
for (long current = CrailUtils.blockStartAddress(start); current < start + len; current += CrailConstants.BLOCK_SIZE){
RpcGetLocation getLocationRes = rpcConnection.getLocation(name, current).get(CrailConstants.RPC_TIMEOUT, TimeUnit.MILLISECONDS);
if (getLocationRes.getError() != RpcErrors.ERR_OK) {
LOG.info("location: " + RpcErrors.messages[getLocationRes.getError()]);
throw new IOException(RpcErrors.messages[getLocationRes.getError()]);
}
-
+
DataNodeInfo dataNodeInfo = getLocationRes.getBlockInfo().getDnInfo();
dataNodeSet.put(dataNodeInfo.key(), dataNodeInfo);
CoreBlockLocation location = new CoreBlockLocation();
@@ -415,7 +415,7 @@ public class CoreDataStore extends CrailStore {
blockLocations[(int) index] = location;
offset2DataNode.put(current, dataNodeInfo);
}
-
+
//asign an identifier to each data node
ArrayList<DataNodeInfo> dataNodeArray = new ArrayList<DataNodeInfo>(dataNodeSet.size());
int index = 0;
@@ -423,7 +423,7 @@ public class CoreDataStore extends CrailStore {
dataNodeArray.add(index, dn);
index++;
}
-
+
int locationSize = Math.min(CrailConstants.SHADOW_REPLICATION, dataNodeSet.size());
int blockIndex = 0;
for (int i = 0; i < blockLocations.length; i++){
@@ -434,11 +434,11 @@ public class CoreDataStore extends CrailStore {
int[] storageType = new int[locationSize];
int[] storageClass = new int[locationSize];
int[] locationTiers = new int[locationSize];
-
+
DataNodeInfo dnInfo = offset2DataNode.get(location.getOffset());
DataNodeInfo mainDataNode = dataNodeSet.get(dnInfo.key());
InetSocketAddress address = CrailUtils.datanodeInfo2SocketAddr(mainDataNode);
- names[0] = getMappedLocation(address.getAddress().getCanonicalHostName()) + ":" + address.getPort();
+ names[0] = getMappedLocation(address.getAddress().getCanonicalHostName()) + ":" + address.getPort();
hosts[0] = getMappedLocation(address.getAddress().getCanonicalHostName());
topology[0] = "/default-rack/" + names[0];
storageType[0] = mainDataNode.getStorageType();
@@ -447,12 +447,12 @@ public class CoreDataStore extends CrailStore {
for (int j = 1; j < locationSize; j++){
DataNodeInfo replicaDataNode = dataNodeArray.get(blockIndex);
address = CrailUtils.datanodeInfo2SocketAddr(replicaDataNode);
- names[j] = getMappedLocation(address.getAddress().getCanonicalHostName()) + ":" + address.getPort();
+ names[j] = getMappedLocation(address.getAddress().getCanonicalHostName()) + ":" + address.getPort();
hosts[j] = getMappedLocation(address.getAddress().getCanonicalHostName());
topology[j] = "/default-rack/" + names[j];
storageType[j] = replicaDataNode.getStorageType();
storageClass[j] = replicaDataNode.getStorageClass();
- locationTiers[j] = replicaDataNode.getLocationClass();
+ locationTiers[j] = replicaDataNode.getLocationClass();
blockIndex = (blockIndex + 1) % dataNodeArray.size();
}
location.setNames(names);
@@ -462,51 +462,51 @@ public class CoreDataStore extends CrailStore {
location.setStorageClasses(storageClass);
location.setLocationAffinities(locationTiers);
}
-
+
return blockLocations;
}
-
+
public void dumpNameNode() throws Exception {
rpcConnection.dumpNameNode().get(CrailConstants.RPC_TIMEOUT, TimeUnit.MILLISECONDS);
- }
-
+ }
+
public void ping() throws Exception {
RpcPing pingRes = rpcConnection.pingNameNode().get(CrailConstants.RPC_TIMEOUT, TimeUnit.MILLISECONDS);
if (pingRes.getError() != RpcErrors.ERR_OK) {
LOG.info("Ping: " + RpcErrors.messages[pingRes.getError()]);
throw new IOException(RpcErrors.messages[pingRes.getError()]);
- }
+ }
}
public CrailBuffer allocateBuffer() throws IOException {
return this.bufferCache.getBuffer();
}
-
+
public void freeBuffer(CrailBuffer buffer) throws IOException {
this.bufferCache.putBuffer(buffer);
- }
-
+ }
+
public int getFsId() {
return fsId;
- }
-
+ }
+
public CrailLocationClass getLocationClass() {
return localClass;
}
-
+
public BufferCheckpoint getBufferCheckpoint() {
return bufferCheckpoint;
}
-
+
public CrailStatistics getStatistics(){
return statistics;
}
-
+
public void closeFileSystem() throws Exception {
if (!isOpen) {
return;
}
-
+
LinkedList<CoreInputStream> tmpIn = new LinkedList<CoreInputStream>();
for (CoreInputStream stream : openInputStreams.values()) {
tmpIn.add(stream);
@@ -514,7 +514,7 @@ public class CoreDataStore extends CrailStore {
for (CoreInputStream stream : tmpIn) {
stream.close();
}
-
+
LinkedList<CoreOutputStream> tmpOut = new LinkedList<CoreOutputStream>();
LinkedList<CoreOutputStream> tmpOutDir = new LinkedList<CoreOutputStream>();
for (CoreOutputStream stream : openOutputStreams.values()) {
@@ -526,11 +526,11 @@ public class CoreDataStore extends CrailStore {
}
for (CoreOutputStream stream : tmpOut) {
stream.close();
- }
+ }
for (CoreOutputStream stream : tmpOutDir) {
stream.close();
- }
-
+ }
+
bufferCache.close();
datanodeEndpointCache.close();
rpcConnection.close();
@@ -540,7 +540,7 @@ public class CoreDataStore extends CrailStore {
public void closeFile(FileInfo fileInfo) throws Exception {
if (fileInfo.getToken() > 0){
- rpcConnection.setFile(fileInfo, true).get(CrailConstants.RPC_TIMEOUT, TimeUnit.MILLISECONDS);
+ rpcConnection.setFile(fileInfo, true).get(CrailConstants.RPC_TIMEOUT, TimeUnit.MILLISECONDS);
}
}
@@ -554,7 +554,7 @@ public class CoreDataStore extends CrailStore {
}
//-------------------------------------------------------------
-
+
CoreOutputStream getOutputStream(CoreNode file, long writeHint) throws Exception {
CoreOutputStream outputStream = new CoreOutputStream(file, streamCounter.incrementAndGet(), writeHint);
openOutputStreams.put(outputStream.getStreamId(), outputStream);
@@ -569,12 +569,12 @@ public class CoreDataStore extends CrailStore {
}
}
return outputStream;
- }
-
+ }
+
CoreInputStream getInputStream(CoreNode file, long readHint) throws Exception {
CoreInputStream inputStream = new CoreInputStream(file, streamCounter.incrementAndGet(), readHint);
openInputStreams.put(inputStream.getStreamId(), inputStream);
-
+
if (CrailConstants.STATISTICS){
streamStats.incOpen();
streamStats.incOpenInput();
@@ -586,7 +586,7 @@ public class CoreDataStore extends CrailStore {
}
return inputStream;
}
-
+
CoreStream unregisterInputStream(CoreInputStream coreStream) {
CoreStream stream = this.openInputStreams.remove(coreStream.getStreamId());
if (stream != null && CrailConstants.STATISTICS){
@@ -598,10 +598,10 @@ public class CoreDataStore extends CrailStore {
streamStats.incCloseInputDir();
}
}
-
+
return stream;
- }
-
+ }
+
CoreStream unregisterOutputStream(CoreOutputStream coreStream) {
CoreStream stream = this.openOutputStreams.remove(coreStream.getStreamId());
if (stream != null && CrailConstants.STATISTICS){
@@ -613,17 +613,17 @@ public class CoreDataStore extends CrailStore {
streamStats.incCloseOutputDir();
}
}
-
+
return stream;
}
-
+
FileBlockCache getBlockCache(long fd){
return blockCache.getFileBlockCache(fd);
}
-
+
FileNextBlockCache getNextBlockCache(long fd){
return nextBlockCache.getFileBlockCache(fd);
- }
+ }
RpcConnection getNamenodeClientRpc() {
return rpcConnection;
@@ -632,12 +632,12 @@ public class CoreDataStore extends CrailStore {
EndpointCache getDatanodeEndpointCache() {
return datanodeEndpointCache;
}
-
+
String getMappedLocation(String hostname){
String mappedValue = locationMap.get(hostname);
return mappedValue != null ? mappedValue : hostname;
}
-
+
CoreSyncOperation getSyncOperation(FileInfo dirInfo, FileInfo fileInfo, String path, boolean valid) throws Exception{
long adjustedCapacity = fileInfo.getDirOffset()*CrailConstants.DIRECTORY_RECORD + CrailConstants.DIRECTORY_RECORD;
dirInfo.setCapacity(Math.max(dirInfo.getCapacity(), adjustedCapacity));
@@ -645,7 +645,7 @@ public class CoreDataStore extends CrailStore {
DirectoryOutputStream stream = dirFile.getDirectoryOutputStream();
DirectoryRecord record = new DirectoryRecord(valid, path);
Future<CrailResult> future = stream.writeRecord(record, fileInfo.getDirOffset());
- CoreSyncOperation syncOperation = new CoreSyncOperation(stream, future);
+ CoreSyncOperation syncOperation = new CoreSyncOperation(stream, future);
return syncOperation;
}
}
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/efd44e75/client/src/main/java/org/apache/crail/storage/StorageClient.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/crail/storage/StorageClient.java b/client/src/main/java/org/apache/crail/storage/StorageClient.java
index d66cb60..cded48f 100644
--- a/client/src/main/java/org/apache/crail/storage/StorageClient.java
+++ b/client/src/main/java/org/apache/crail/storage/StorageClient.java
@@ -21,13 +21,21 @@ package org.apache.crail.storage;
import java.io.IOException;
+import org.apache.crail.CrailBufferCache;
+import org.apache.crail.CrailStatistics;
import org.apache.crail.conf.Configurable;
+import org.apache.crail.conf.CrailConfiguration;
import org.apache.crail.metadata.DataNodeInfo;
+import org.slf4j.Logger;
+
+public interface StorageClient {
+ StorageEndpoint createEndpoint(DataNodeInfo info) throws IOException;
+ void close() throws Exception;
+ void init(CrailStatistics statistics, CrailBufferCache bufferCache, CrailConfiguration configuration,
+ String[] args) throws IOException;
+ void printConf(Logger log);
+
-public interface StorageClient extends Configurable {
- public abstract StorageEndpoint createEndpoint(DataNodeInfo info) throws IOException;
- public abstract void close() throws Exception;
-
@SuppressWarnings("unchecked")
public static StorageClient createInstance(String name) throws Exception {
Class<?> nodeClass = Class.forName(name);
@@ -38,6 +46,6 @@ public interface StorageClient extends Configurable {
} else {
throw new Exception("Cannot instantiate storage client of type " + name);
}
-
- }
+
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/efd44e75/storage-narpc/src/main/java/org/apache/crail/storage/tcp/TcpStorageClient.java
----------------------------------------------------------------------
diff --git a/storage-narpc/src/main/java/org/apache/crail/storage/tcp/TcpStorageClient.java b/storage-narpc/src/main/java/org/apache/crail/storage/tcp/TcpStorageClient.java
index e637f02..2a526fb 100644
--- a/storage-narpc/src/main/java/org/apache/crail/storage/tcp/TcpStorageClient.java
+++ b/storage-narpc/src/main/java/org/apache/crail/storage/tcp/TcpStorageClient.java
@@ -21,6 +21,8 @@ package org.apache.crail.storage.tcp;
import java.io.IOException;
+import org.apache.crail.CrailBufferCache;
+import org.apache.crail.CrailStatistics;
import org.apache.crail.conf.CrailConfiguration;
import org.apache.crail.conf.CrailConstants;
import org.apache.crail.metadata.DataNodeInfo;
@@ -36,9 +38,10 @@ public class TcpStorageClient implements StorageClient {
private NaRPCClientGroup<TcpStorageRequest, TcpStorageResponse> clientGroup;
@Override
- public void init(CrailConfiguration conf, String[] args) throws Exception {
+ public void init(CrailStatistics statistics, CrailBufferCache bufferCache, CrailConfiguration conf, String[] args)
+ throws IOException {
TcpStorageConstants.updateConstants(conf);
-
+
this.clientGroup = new NaRPCClientGroup<TcpStorageRequest, TcpStorageResponse>(TcpStorageConstants.STORAGE_TCP_QUEUE_DEPTH, (int) CrailConstants.BLOCK_SIZE*2, false);
}
@@ -62,5 +65,5 @@ public class TcpStorageClient implements StorageClient {
throw new IOException(e);
}
}
-
+
}
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/efd44e75/storage-nvmf/src/main/java/org/apache/crail/storage/nvmf/NvmfStorageClient.java
----------------------------------------------------------------------
diff --git a/storage-nvmf/src/main/java/org/apache/crail/storage/nvmf/NvmfStorageClient.java b/storage-nvmf/src/main/java/org/apache/crail/storage/nvmf/NvmfStorageClient.java
index 69c8f0d..d270f5d 100644
--- a/storage-nvmf/src/main/java/org/apache/crail/storage/nvmf/NvmfStorageClient.java
+++ b/storage-nvmf/src/main/java/org/apache/crail/storage/nvmf/NvmfStorageClient.java
@@ -21,6 +21,8 @@ package org.apache.crail.storage.nvmf;
import java.io.IOException;
+import org.apache.crail.CrailBufferCache;
+import org.apache.crail.CrailStatistics;
import org.apache.crail.conf.CrailConfiguration;
import org.apache.crail.metadata.DataNodeInfo;
import org.apache.crail.storage.StorageClient;
@@ -37,12 +39,13 @@ public class NvmfStorageClient implements StorageClient {
private static NvmeEndpointGroup clientGroup;
private boolean initialized = false;
- public void init(CrailConfiguration crailConfiguration, String[] args) throws IOException {
+ public void init(CrailStatistics statistics, CrailBufferCache bufferCache, CrailConfiguration crailConfiguration,
+ String[] args) throws IOException {
if (initialized) {
throw new IOException("NvmfStorageTier already initialized");
}
initialized = true;
-
+
NvmfStorageConstants.parseCmdLine(crailConfiguration, args);
}
@@ -61,8 +64,8 @@ public class NvmfStorageClient implements StorageClient {
public synchronized StorageEndpoint createEndpoint(DataNodeInfo info) throws IOException {
return new NvmfStorageEndpoint(getEndpointGroup(), CrailUtils.datanodeInfo2SocketAddr(info));
}
-
+
public void close() throws Exception {
- }
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/efd44e75/storage-rdma/src/main/java/org/apache/crail/storage/rdma/RdmaConstants.java
----------------------------------------------------------------------
diff --git a/storage-rdma/src/main/java/org/apache/crail/storage/rdma/RdmaConstants.java b/storage-rdma/src/main/java/org/apache/crail/storage/rdma/RdmaConstants.java
index 70687d4..25cea6c 100644
--- a/storage-rdma/src/main/java/org/apache/crail/storage/rdma/RdmaConstants.java
+++ b/storage-rdma/src/main/java/org/apache/crail/storage/rdma/RdmaConstants.java
@@ -36,13 +36,13 @@ import org.slf4j.Logger;
public class RdmaConstants {
private static final Logger LOG = CrailUtils.getLogger();
-
+
public static final String STORAGE_RDMA_INTERFACE_KEY = "crail.storage.rdma.interface";
public static String STORAGE_RDMA_INTERFACE = "eth5";
public static final String STORAGE_RDMA_PORT_KEY = "crail.storage.rdma.port";
- public static int STORAGE_RDMA_PORT = 50020;
-
+ public static int STORAGE_RDMA_PORT = 50020;
+
public static final String STORAGE_RDMA_STORAGE_LIMIT_KEY = "crail.storage.rdma.storagelimit";
public static long STORAGE_RDMA_STORAGE_LIMIT = 1073741824;
@@ -54,46 +54,46 @@ public class RdmaConstants {
public static final String STORAGE_RDMA_LOCAL_MAP_KEY = "crail.storage.rdma.localmap";
public static boolean STORAGE_RDMA_LOCAL_MAP = true;
-
+
public static final String STORAGE_RDMA_QUEUESIZE_KEY = "crail.storage.rdma.queuesize";
public static int STORAGE_RDMA_QUEUESIZE = 32;
-
+
public static final String STORAGE_RDMA_TYPE_KEY = "crail.storage.rdma.type";
- public static String STORAGE_RDMA_TYPE = "passive";
-
+ public static String STORAGE_RDMA_TYPE = "passive";
+
public static final String STORAGE_RDMA_PERSISTENT_KEY = "crail.storage.rdma.persistent";
- public static boolean STORAGE_RDMA_PERSISTENT = false;
-
+ public static boolean STORAGE_RDMA_PERSISTENT = false;
+
public static void updateConstants(CrailConfiguration conf){
if (conf.get(STORAGE_RDMA_INTERFACE_KEY) != null) {
STORAGE_RDMA_INTERFACE = conf.get(STORAGE_RDMA_INTERFACE_KEY);
}
if (conf.get(STORAGE_RDMA_PORT_KEY) != null) {
STORAGE_RDMA_PORT = Integer.parseInt(conf.get(STORAGE_RDMA_PORT_KEY));
- }
+ }
if (conf.get(STORAGE_RDMA_STORAGE_LIMIT_KEY) != null) {
STORAGE_RDMA_STORAGE_LIMIT = Long.parseLong(conf.get(STORAGE_RDMA_STORAGE_LIMIT_KEY));
- }
+ }
if (conf.get(STORAGE_RDMA_ALLOCATION_SIZE_KEY) != null) {
STORAGE_RDMA_ALLOCATION_SIZE = Long.parseLong(conf.get(STORAGE_RDMA_ALLOCATION_SIZE_KEY));
- }
+ }
if (conf.get(STORAGE_RDMA_DATA_PATH_KEY) != null) {
STORAGE_RDMA_DATA_PATH = conf.get(STORAGE_RDMA_DATA_PATH_KEY);
- }
+ }
if (conf.get(STORAGE_RDMA_LOCAL_MAP_KEY) != null) {
STORAGE_RDMA_LOCAL_MAP = conf.getBoolean(STORAGE_RDMA_LOCAL_MAP_KEY, false);
- }
+ }
if (conf.get(STORAGE_RDMA_QUEUESIZE_KEY) != null) {
STORAGE_RDMA_QUEUESIZE = Integer.parseInt(conf.get(STORAGE_RDMA_QUEUESIZE_KEY));
- }
+ }
if (conf.get(STORAGE_RDMA_TYPE_KEY) != null) {
STORAGE_RDMA_TYPE = conf.get(STORAGE_RDMA_TYPE_KEY);
- }
+ }
if (conf.get(STORAGE_RDMA_PERSISTENT_KEY) != null) {
STORAGE_RDMA_PERSISTENT = conf.getBoolean(STORAGE_RDMA_PERSISTENT_KEY, false);
- }
+ }
}
-
+
public static void verify() throws IOException {
if (STORAGE_RDMA_ALLOCATION_SIZE % CrailConstants.BLOCK_SIZE != 0){
throw new IOException("crail.storage.rdma.allocationsize must be multiple of crail.blocksize");
@@ -103,12 +103,12 @@ public class RdmaConstants {
}
if (!STORAGE_RDMA_TYPE.equalsIgnoreCase("passive") && !STORAGE_RDMA_TYPE.equalsIgnoreCase("active")){
throw new IOException("crail.storage.rdma.type must be either <active> or <passive>, found " + STORAGE_RDMA_TYPE);
- }
+ }
}
public static void printConf(Logger logger) {
logger.info(STORAGE_RDMA_INTERFACE_KEY + " " + STORAGE_RDMA_INTERFACE);
- logger.info(STORAGE_RDMA_PORT_KEY + " " + STORAGE_RDMA_PORT);
+ logger.info(STORAGE_RDMA_PORT_KEY + " " + STORAGE_RDMA_PORT);
logger.info(STORAGE_RDMA_STORAGE_LIMIT_KEY + " " + STORAGE_RDMA_STORAGE_LIMIT);
logger.info(STORAGE_RDMA_ALLOCATION_SIZE_KEY + " " + STORAGE_RDMA_ALLOCATION_SIZE);
logger.info(STORAGE_RDMA_DATA_PATH_KEY + " " + STORAGE_RDMA_DATA_PATH);
@@ -116,8 +116,8 @@ public class RdmaConstants {
logger.info(STORAGE_RDMA_QUEUESIZE_KEY + " " + STORAGE_RDMA_QUEUESIZE);
logger.info(STORAGE_RDMA_TYPE_KEY + " " + STORAGE_RDMA_TYPE);
}
-
- public static void init(CrailConfiguration conf, String[] args) throws Exception {
+
+ public static void init(CrailConfiguration conf, String[] args) throws IOException {
if (args != null) {
Option interfaceOption = Option.builder("i").desc("interface to start server on").hasArg().build();
Option portOption = Option.builder("p").desc("port to start server on").hasArg().build();
@@ -142,7 +142,7 @@ public class RdmaConstants {
}
if (line.hasOption(persistencyOption.getOpt())) {
conf.set(RdmaConstants.STORAGE_RDMA_PERSISTENT_KEY, "true");
- }
+ }
} catch (ParseException e) {
HelpFormatter formatter = new HelpFormatter();
formatter.printHelp("RDMA storage tier", options);
@@ -152,5 +152,5 @@ public class RdmaConstants {
RdmaConstants.updateConstants(conf);
RdmaConstants.verify();
- }
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/efd44e75/storage-rdma/src/main/java/org/apache/crail/storage/rdma/RdmaStorageClient.java
----------------------------------------------------------------------
diff --git a/storage-rdma/src/main/java/org/apache/crail/storage/rdma/RdmaStorageClient.java b/storage-rdma/src/main/java/org/apache/crail/storage/rdma/RdmaStorageClient.java
index 0c7f8cf..72b749f 100644
--- a/storage-rdma/src/main/java/org/apache/crail/storage/rdma/RdmaStorageClient.java
+++ b/storage-rdma/src/main/java/org/apache/crail/storage/rdma/RdmaStorageClient.java
@@ -21,6 +21,8 @@ package org.apache.crail.storage.rdma;
import java.io.IOException;
+import org.apache.crail.CrailBufferCache;
+import org.apache.crail.CrailStatistics;
import org.apache.crail.conf.CrailConfiguration;
import org.apache.crail.metadata.DataNodeInfo;
import org.apache.crail.storage.StorageClient;
@@ -34,23 +36,24 @@ import org.slf4j.Logger;
public class RdmaStorageClient implements StorageClient {
private static final Logger LOG = CrailUtils.getLogger();
-
+
private MrCache clientMrCache = null;
private RdmaStorageGroup clientGroup = null;
-
+
public RdmaStorageClient(){
this.clientGroup = null;
this.clientMrCache = null;
}
-
- public void init(CrailConfiguration conf, String[] args) throws Exception {
+
+ public void init(CrailStatistics statistics, CrailBufferCache bufferCache, CrailConfiguration conf, String[] args)
+ throws IOException {
RdmaConstants.init(conf, args);
}
-
+
public void printConf(Logger logger){
RdmaConstants.printConf(logger);
- }
-
+ }
+
@Override
public StorageEndpoint createEndpoint(DataNodeInfo info) throws IOException {
if (clientMrCache == null){
@@ -73,11 +76,11 @@ public class RdmaStorageClient implements StorageClient {
RdmaStorageActiveGroup _endpointGroup = new RdmaStorageActiveGroup(100, false, RdmaConstants.STORAGE_RDMA_QUEUESIZE, 4, RdmaConstants.STORAGE_RDMA_QUEUESIZE*2, clientMrCache);
_endpointGroup.init(new RdmaStorageActiveEndpointFactory(_endpointGroup));
this.clientGroup = _endpointGroup;
- }
+ }
}
}
}
-
+
return clientGroup.createEndpoint(info);
}
@@ -86,5 +89,5 @@ public class RdmaStorageClient implements StorageClient {
if (clientGroup != null){
this.clientGroup.close();
}
- }
+ }
}
[3/3] incubator-crail git commit: CrailStore: refactor alloc/free
buffer interface
Posted by pe...@apache.org.
CrailStore: refactor alloc/free buffer interface
Do not implement CrailBufferCache but instead use
abstract method to define allocate/free buffer interface.
Pass BufferCache instead of CrailStore to StorageClient init.
Close #14
Signed-off-by: Jonas Pfefferle <pe...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/incubator-crail/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-crail/commit/c053477a
Tree: http://git-wip-us.apache.org/repos/asf/incubator-crail/tree/c053477a
Diff: http://git-wip-us.apache.org/repos/asf/incubator-crail/diff/c053477a
Branch: refs/heads/master
Commit: c053477a7f0d8a530bd996c8937ffde7f024366d
Parents: fa9c6af
Author: Jonas Pfefferle <pe...@apache.org>
Authored: Wed Apr 4 10:52:46 2018 +0200
Committer: Jonas Pfefferle <pe...@apache.org>
Committed: Wed Apr 4 11:20:00 2018 +0200
----------------------------------------------------------------------
client/src/main/java/org/apache/crail/CrailStore.java | 4 +++-
client/src/main/java/org/apache/crail/core/CoreDataStore.java | 7 ++++---
2 files changed, 7 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/c053477a/client/src/main/java/org/apache/crail/CrailStore.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/crail/CrailStore.java b/client/src/main/java/org/apache/crail/CrailStore.java
index 3abc894..6c3aae6 100644
--- a/client/src/main/java/org/apache/crail/CrailStore.java
+++ b/client/src/main/java/org/apache/crail/CrailStore.java
@@ -28,7 +28,7 @@ import org.apache.crail.core.CoreDataStore;
import org.apache.crail.utils.CrailUtils;
import org.slf4j.Logger;
-public abstract class CrailStore implements CrailBufferCache {
+public abstract class CrailStore {
private static final Logger LOG = CrailUtils.getLogger();
private static AtomicLong referenceCounter = new AtomicLong(0);
private static CrailStore instance = null;
@@ -37,6 +37,8 @@ public abstract class CrailStore implements CrailBufferCache {
public abstract Upcoming<CrailNode> lookup(String path) throws Exception;
public abstract Upcoming<CrailNode> rename(String srcPath, String dstPath) throws Exception;
public abstract Upcoming<CrailNode> delete(String path, boolean recursive) throws Exception;
+ public abstract CrailBuffer allocateBuffer() throws Exception;
+ public abstract void freeBuffer(CrailBuffer buffer) throws Exception;
public abstract CrailStatistics getStatistics();
public abstract CrailLocationClass getLocationClass();
protected abstract void closeFileSystem() throws Exception;
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/c053477a/client/src/main/java/org/apache/crail/core/CoreDataStore.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/crail/core/CoreDataStore.java b/client/src/main/java/org/apache/crail/core/CoreDataStore.java
index 8b5e54f..92d5bec 100644
--- a/client/src/main/java/org/apache/crail/core/CoreDataStore.java
+++ b/client/src/main/java/org/apache/crail/core/CoreDataStore.java
@@ -106,13 +106,16 @@ public class CoreDataStore extends CrailStore {
CrailConstants.printConf();
CrailConstants.verify();
+ this.bufferCache = BufferCache.createInstance(CrailConstants.CACHE_IMPL);
+ this.statistics = new CrailStatistics();
+
//Datanodes
StringTokenizer tokenizer = new StringTokenizer(CrailConstants.STORAGE_TYPES, ",");
LinkedList<StorageClient> dataNodeClients = new LinkedList<StorageClient>();
while (tokenizer.hasMoreTokens()){
String name = tokenizer.nextToken();
StorageClient dataNode = StorageClient.createInstance(name);
- dataNode.init(getStatistics(), this, conf, null);
+ dataNode.init(statistics, bufferCache, conf, null);
dataNode.printConf(LOG);
dataNodeClients.add(dataNode);
}
@@ -140,7 +143,6 @@ public class CoreDataStore extends CrailStore {
//Client
this.fsId = fsCount.getAndIncrement();
this.localClass = CrailUtils.getLocationClass();
- this.bufferCache = BufferCache.createInstance(CrailConstants.CACHE_IMPL);
this.blockCache = new BlockCache();
this.nextBlockCache = new NextBlockCache();
this.openInputStreams = new ConcurrentHashMap<Long, CoreInputStream>();
@@ -151,7 +153,6 @@ public class CoreDataStore extends CrailStore {
this.locationMap = new ConcurrentHashMap<String, String>();
CrailUtils.parseMap(CrailConstants.LOCATION_MAP, locationMap);
- this.statistics = new CrailStatistics();
this.ioStatsIn = new CoreIOStatistics("core/input");
statistics.addProvider(ioStatsIn);
this.ioStatsOut = new CoreIOStatistics("core/output");
[2/3] incubator-crail git commit: BufferCache: use new interface in
buffer cache
Posted by pe...@apache.org.
BufferCache: use new interface in buffer cache
Use newly introduced buffer cache interface in buffer cache
implementation.
Signed-off-by: Jonas Pfefferle <pe...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/incubator-crail/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-crail/commit/fa9c6af1
Tree: http://git-wip-us.apache.org/repos/asf/incubator-crail/tree/fa9c6af1
Diff: http://git-wip-us.apache.org/repos/asf/incubator-crail/diff/fa9c6af1
Branch: refs/heads/master
Commit: fa9c6af1acd47934c54c5ad47014c376066e68a7
Parents: efd44e7
Author: Jonas Pfefferle <pe...@apache.org>
Authored: Tue Apr 3 15:50:59 2018 +0200
Committer: Jonas Pfefferle <pe...@apache.org>
Committed: Tue Apr 3 15:59:11 2018 +0200
----------------------------------------------------------------------
.../org/apache/crail/core/CoreDataStore.java | 4 +-
.../org/apache/crail/memory/BufferCache.java | 58 ++++++++++----------
.../apache/crail/memory/MappedBufferCache.java | 30 +++++-----
.../crail/storage/nvmf/NvmfBufferCache.java | 2 +-
.../nvmf/client/NvmfStorageEndpoint.java | 6 +-
5 files changed, 48 insertions(+), 52 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/fa9c6af1/client/src/main/java/org/apache/crail/core/CoreDataStore.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/crail/core/CoreDataStore.java b/client/src/main/java/org/apache/crail/core/CoreDataStore.java
index 59c7da9..8b5e54f 100644
--- a/client/src/main/java/org/apache/crail/core/CoreDataStore.java
+++ b/client/src/main/java/org/apache/crail/core/CoreDataStore.java
@@ -479,11 +479,11 @@ public class CoreDataStore extends CrailStore {
}
public CrailBuffer allocateBuffer() throws IOException {
- return this.bufferCache.getBuffer();
+ return this.bufferCache.allocateBuffer();
}
public void freeBuffer(CrailBuffer buffer) throws IOException {
- this.bufferCache.putBuffer(buffer);
+ this.bufferCache.freeBuffer(buffer);
}
public int getFsId() {
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/fa9c6af1/client/src/main/java/org/apache/crail/memory/BufferCache.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/crail/memory/BufferCache.java b/client/src/main/java/org/apache/crail/memory/BufferCache.java
index 59f3dc9..108d75f 100644
--- a/client/src/main/java/org/apache/crail/memory/BufferCache.java
+++ b/client/src/main/java/org/apache/crail/memory/BufferCache.java
@@ -30,32 +30,34 @@ import org.apache.crail.conf.CrailConstants;
import org.apache.crail.utils.CrailUtils;
import org.slf4j.Logger;
-public abstract class BufferCache implements CrailStatistics.StatisticsProvider {
+public abstract class BufferCache implements CrailStatistics.StatisticsProvider, CrailBufferCache {
private static final Logger LOG = CrailUtils.getLogger();
private LinkedBlockingQueue<CrailBuffer> cache;
-
+
private AtomicLong cacheGet;
private AtomicLong cachePut;
private AtomicLong cacheMisses;
private AtomicLong cacheOut;
private AtomicLong cacheMax;
-
+
private AtomicLong cacheMissesMap;
- private AtomicLong cacheMissesHeap;
-
- public BufferCache() throws IOException{
+ private AtomicLong cacheMissesHeap;
+
+ public abstract CrailBuffer allocateRegion() throws IOException;
+
+ public BufferCache() throws IOException {
this.cache = new LinkedBlockingQueue<CrailBuffer>();
-
+
this.cacheGet = new AtomicLong(0);
this.cachePut = new AtomicLong(0);
this.cacheMisses = new AtomicLong(0);
this.cacheOut = new AtomicLong(0);
this.cacheMax = new AtomicLong(0);
-
+
this.cacheMissesMap = new AtomicLong(0);
- this.cacheMissesHeap = new AtomicLong(0);
+ this.cacheMissesHeap = new AtomicLong(0);
}
-
+
@Override
public String providerName() {
return "cache/buffer";
@@ -65,7 +67,7 @@ public abstract class BufferCache implements CrailStatistics.StatisticsProvider
public String printStatistics() {
return "cacheGet " + cacheGet.get() + ", cachePut " + cachePut.get() + ", cacheMiss " + cacheMisses.get() + ", cacheSize " + cache.size() + ", cacheMax " + cacheMax.get() + ", mapMiss " + cacheMissesMap.get() + ", mapHeap " + cacheMissesHeap.get();
}
-
+
public void resetStatistics(){
this.cacheGet.set(0);
this.cachePut.set(0);
@@ -74,24 +76,24 @@ public abstract class BufferCache implements CrailStatistics.StatisticsProvider
this.cacheMax.set(0);
this.cacheMissesMap.set(0);
this.cacheMissesHeap.set(0);
- }
-
+ }
+
public void mergeStatistics(StatisticsProvider provider){
-
+
}
-
- public CrailBuffer getBuffer() throws IOException {
+
+ public CrailBuffer allocateBuffer() throws IOException {
cacheGet.incrementAndGet();
cacheOut.incrementAndGet();
cacheMax.updateAndGet(x -> Math.max(x, cacheOut.get()));
-
+
CrailBuffer buffer = cache.poll();
if (buffer == null){
synchronized(this){
buffer = cache.poll();
if (buffer == null){
cacheMisses.incrementAndGet();
- buffer = allocateBuffer();
+ buffer = allocateRegion();
if (buffer == null){
buffer = OffHeapBuffer.wrap(ByteBuffer.allocateDirect(CrailConstants.BUFFER_SIZE));
cacheMissesHeap.incrementAndGet();
@@ -100,29 +102,27 @@ public abstract class BufferCache implements CrailStatistics.StatisticsProvider
}
}
}
- }
-
+ }
+
buffer.clear();
return buffer;
}
-
- public void putBuffer(CrailBuffer buffer) throws IOException{
+
+ public void freeBuffer(CrailBuffer buffer) throws IOException{
if (buffer != null){
cachePut.incrementAndGet();
cacheOut.decrementAndGet();
putBufferInternal(buffer);
}
}
-
+
public void putBufferInternal(CrailBuffer buffer) throws IOException{
cache.add(buffer);
- }
-
+ }
+
public void close(){
cache.clear();
}
-
- public abstract CrailBuffer allocateBuffer() throws IOException;
@SuppressWarnings("unchecked")
public static BufferCache createInstance(String name) throws Exception {
@@ -134,7 +134,7 @@ public abstract class BufferCache implements CrailStatistics.StatisticsProvider
} else {
throw new Exception("Cannot instantiate storage client of type " + name);
}
-
- }
+
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/fa9c6af1/client/src/main/java/org/apache/crail/memory/MappedBufferCache.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/crail/memory/MappedBufferCache.java b/client/src/main/java/org/apache/crail/memory/MappedBufferCache.java
index 567e300..9e37ca2 100644
--- a/client/src/main/java/org/apache/crail/memory/MappedBufferCache.java
+++ b/client/src/main/java/org/apache/crail/memory/MappedBufferCache.java
@@ -33,23 +33,23 @@ import org.slf4j.Logger;
public class MappedBufferCache extends BufferCache {
private static final Logger LOG = CrailUtils.getLogger();
-
+
private String id;
private String directory;
private File dir;
private long allocationCount;
private long bufferCount;
- private long currentRegion;
-
+ private long currentRegion;
+
public MappedBufferCache() throws IOException {
super();
-
+
this.allocationCount = CrailConstants.CACHE_LIMIT / CrailConstants.REGION_SIZE;
long _bufferSize = (long) CrailConstants.BUFFER_SIZE;
this.bufferCount = CrailConstants.REGION_SIZE / _bufferSize;
this.currentRegion = 0;
LOG.info("buffer cache, allocationCount " + allocationCount + ", bufferCount " + bufferCount);
-
+
if (allocationCount > 0){
id = "" + System.currentTimeMillis();
directory = CrailUtils.getCacheDirectory(id);
@@ -62,11 +62,11 @@ public class MappedBufferCache extends BufferCache {
}
}
}
-
+
@Override
public void close() {
super.close();
-
+
if (allocationCount > 0 && dir.exists()){
for (File child : dir.listFiles()) {
child.delete();
@@ -75,16 +75,12 @@ public class MappedBufferCache extends BufferCache {
}
LOG.info("mapped client cache closed");
}
-
- public CrailBuffer allocateBuffer() throws IOException{
- return allocateRegion();
- }
- private CrailBuffer allocateRegion() throws IOException {
+ public CrailBuffer allocateRegion() throws IOException {
if (currentRegion >= allocationCount){
return null;
}
-
+
String path = directory + "/" + currentRegion++;
RandomAccessFile randomFile = new RandomAccessFile(path, "rw");
randomFile.setLength(CrailConstants.REGION_SIZE);
@@ -96,23 +92,23 @@ public class MappedBufferCache extends BufferCache {
channel.close();
CrailBuffer firstBuffer = slice(mappedBuffer, 0);
-
+
for (int j = 1; j < bufferCount; j++) {
int position = j * CrailConstants.BUFFER_SIZE;
CrailBuffer sliceBuffer = slice(mappedBuffer, position);
this.putBufferInternal(sliceBuffer);
}
mappedBuffer.clear();
-
+
return firstBuffer;
}
-
+
private CrailBuffer slice(CrailBuffer mappedBuffer, int position){
int limit = position + CrailConstants.BUFFER_SIZE;
mappedBuffer.clear();
mappedBuffer.position(position);
mappedBuffer.limit(limit);
- CrailBuffer buffer = mappedBuffer.slice();
+ CrailBuffer buffer = mappedBuffer.slice();
return buffer;
}
}
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/fa9c6af1/storage-nvmf/src/main/java/org/apache/crail/storage/nvmf/NvmfBufferCache.java
----------------------------------------------------------------------
diff --git a/storage-nvmf/src/main/java/org/apache/crail/storage/nvmf/NvmfBufferCache.java b/storage-nvmf/src/main/java/org/apache/crail/storage/nvmf/NvmfBufferCache.java
index f7b19d9..19142ec 100644
--- a/storage-nvmf/src/main/java/org/apache/crail/storage/nvmf/NvmfBufferCache.java
+++ b/storage-nvmf/src/main/java/org/apache/crail/storage/nvmf/NvmfBufferCache.java
@@ -52,7 +52,7 @@ public class NvmfBufferCache extends BufferCache {
}
@Override
- public CrailBuffer allocateBuffer() throws IOException {
+ public CrailBuffer allocateRegion() throws IOException {
ByteBuffer buffer = endpointGroup.allocateBuffer(CrailConstants.BUFFER_SIZE, ALIGNMENT);
bufferPool.add(buffer);
return OffHeapBuffer.wrap(buffer);
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/fa9c6af1/storage-nvmf/src/main/java/org/apache/crail/storage/nvmf/client/NvmfStorageEndpoint.java
----------------------------------------------------------------------
diff --git a/storage-nvmf/src/main/java/org/apache/crail/storage/nvmf/client/NvmfStorageEndpoint.java b/storage-nvmf/src/main/java/org/apache/crail/storage/nvmf/client/NvmfStorageEndpoint.java
index 0e04edf..b4b1054 100644
--- a/storage-nvmf/src/main/java/org/apache/crail/storage/nvmf/client/NvmfStorageEndpoint.java
+++ b/storage-nvmf/src/main/java/org/apache/crail/storage/nvmf/client/NvmfStorageEndpoint.java
@@ -41,7 +41,7 @@ import java.net.URI;
import java.net.URISyntaxException;
import java.util.concurrent.*;
-public class NvmfStorageEndpoint implements StorageEndpoint {
+public class NvmfStorageEndpoint implements StorageEndpoint {
private static final Logger LOG = CrailUtils.getLogger();
private final InetSocketAddress inetSocketAddress;
@@ -150,7 +150,7 @@ public class NvmfStorageEndpoint implements StorageEndpoint {
// LOG.info("unaligned");
long alignedLength = NvmfStorageUtils.alignLength(sectorSize, remoteOffset, length);
- CrailBuffer stagingBuffer = cache.getBuffer();
+ CrailBuffer stagingBuffer = cache.allocateBuffer();
stagingBuffer.limit((int)alignedLength);
try {
switch(op) {
@@ -213,7 +213,7 @@ public class NvmfStorageEndpoint implements StorageEndpoint {
}
void putBuffer(CrailBuffer buffer) throws IOException {
- cache.putBuffer(buffer);
+ cache.freeBuffer(buffer);
}
public void close() throws IOException, InterruptedException {