You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@crail.apache.org by ps...@apache.org on 2018/01/22 14:21:01 UTC
[17/29] incubator-crail git commit: Apache package renaming
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/client/src/main/java/com/ibm/crail/core/CoreStream.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/com/ibm/crail/core/CoreStream.java b/client/src/main/java/com/ibm/crail/core/CoreStream.java
deleted file mode 100644
index 1e9107f..0000000
--- a/client/src/main/java/com/ibm/crail/core/CoreStream.java
+++ /dev/null
@@ -1,263 +0,0 @@
-/*
- * Crail: A Multi-tiered Distributed Direct Access File System
- *
- * Author: Patrick Stuedi <st...@zurich.ibm.com>
- *
- * Copyright (C) 2016, IBM Corporation
- *
- * 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
- *
- * 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 com.ibm.crail.core;
-
-import java.io.IOException;
-import java.util.HashMap;
-import java.util.LinkedList;
-import java.util.concurrent.Future;
-import java.util.concurrent.TimeUnit;
-
-import org.slf4j.Logger;
-
-import com.ibm.crail.CrailBuffer;
-import com.ibm.crail.conf.CrailConstants;
-import com.ibm.crail.metadata.BlockInfo;
-import com.ibm.crail.metadata.FileInfo;
-import com.ibm.crail.rpc.RpcErrors;
-import com.ibm.crail.rpc.RpcConnection;
-import com.ibm.crail.rpc.RpcGetBlock;
-import com.ibm.crail.rpc.RpcFuture;
-import com.ibm.crail.storage.StorageEndpoint;
-import com.ibm.crail.storage.StorageFuture;
-import com.ibm.crail.utils.BufferCheckpoint;
-import com.ibm.crail.utils.EndpointCache;
-import com.ibm.crail.utils.CrailUtils;
-import com.ibm.crail.utils.BlockCache.FileBlockCache;
-import com.ibm.crail.utils.NextBlockCache.FileNextBlockCache;
-
-public abstract class CoreStream {
- private static final Logger LOG = CrailUtils.getLogger();
-
- protected CoreFileSystem fs;
- protected CoreNode node;
-
- private EndpointCache endpointCache;
- private RpcConnection namenodeClientRpc;
- private FileBlockCache blockCache;
- private FileNextBlockCache nextBlockCache;
- private BufferCheckpoint bufferCheckpoint;
- private FileInfo fileInfo;
- private long position;
- private long syncedCapacity;
- private long streamId;
- private CoreIOStatistics ioStats;
- private HashMap<Integer, CoreSubOperation> blockMap;
- private LinkedList<RpcFuture<RpcGetBlock>> pendingBlocks;
-
- abstract StorageFuture trigger(StorageEndpoint endpoint, CoreSubOperation opDesc, CrailBuffer buffer, BlockInfo block) throws Exception;
- abstract void update(long newCapacity);
-
- CoreStream(CoreNode node, long streamId, long fileOffset) throws Exception {
- this.node = node;
- this.fs = node.getFileSystem();
- this.fileInfo = node.getFileInfo();
- this.endpointCache = fs.getDatanodeEndpointCache();
- this.namenodeClientRpc = fs.getNamenodeClientRpc();
- this.blockCache = fs.getBlockCache(fileInfo.getFd());
- this.nextBlockCache = fs.getNextBlockCache(fileInfo.getFd());
- this.bufferCheckpoint = fs.getBufferCheckpoint();
-
- this.position = fileOffset;
- this.syncedCapacity = fileInfo.getCapacity();
- this.streamId = streamId;
- this.ioStats = new CoreIOStatistics("core");
-
- this.blockMap = new HashMap<Integer, CoreSubOperation>();
- this.pendingBlocks = new LinkedList<RpcFuture<RpcGetBlock>>();
- }
-
- final CoreDataOperation dataOperation(CrailBuffer dataBuf) throws Exception {
- blockMap.clear();
- pendingBlocks.clear();
- CoreDataOperation multiOperation = new CoreDataOperation(this, dataBuf);
-
- //compute off, len for the fragments, start transfer or start RPC if block info is missing
- while(multiOperation.remaining() > 0){
- long blockRemaining = blockRemaining();
- int opLen = CrailUtils.minFileBuf(blockRemaining, multiOperation.remaining());
- CoreSubOperation subOperation = new CoreSubOperation(fileInfo.getFd(), position, multiOperation.getCurrentBufferPosition(), opLen);
-// LOG.info("OpDesc: " + opDesc.toString());
- ioStats.incTotalOps((long) opLen);
-
- if (blockCache.containsKey(subOperation.key())){
- BlockInfo block = blockCache.get(subOperation.key());
- StorageFuture subFuture = this.prepareAndTrigger(subOperation, dataBuf, block);
- multiOperation.add(subFuture);
- this.ioStats.incCachedOps();
- } else if (nextBlockCache.containsKey(subOperation.key())){
- RpcFuture<RpcGetBlock> rpcFuture = nextBlockCache.get(subOperation.key());
- blockMap.put(rpcFuture.getTicket(), subOperation);
- pendingBlocks.add(rpcFuture);
- } else {
- this.syncedCapacity = fileInfo.getCapacity();
- RpcFuture<RpcGetBlock> rpcFuture = namenodeClientRpc.getBlock(fileInfo.getFd(), fileInfo.getToken(), position, syncedCapacity);
- blockMap.put(rpcFuture.getTicket(), subOperation);
- pendingBlocks.add(rpcFuture);
- }
-
- position += opLen;
- multiOperation.incProcessedLen(opLen);
- }
-
- //wait for RPC results and start reads for those blocks as well
- for (RpcFuture<RpcGetBlock> rpcFuture = pendingBlocks.poll(); rpcFuture != null; rpcFuture = pendingBlocks.poll()){
- if (!rpcFuture.isDone()){
- this.ioStats.incBlockingOps();
- if (rpcFuture.isPrefetched()){
- this.ioStats.incPrefetchedBlockingOps();
- }
- } else {
- this.ioStats.incNonblockingOps();
- if (rpcFuture.isPrefetched()){
- this.ioStats.incPrefetchedNonblockingOps();
- }
- }
-
- RpcGetBlock getBlockRes = rpcFuture.get(CrailConstants.RPC_TIMEOUT, TimeUnit.MILLISECONDS);
- if (!rpcFuture.isDone()){
- throw new IOException("rpc timeout ");
- }
- if (getBlockRes.getError() != RpcErrors.ERR_OK) {
- LOG.info("inputStream: " + RpcErrors.messages[getBlockRes.getError()]);
- throw new IOException(RpcErrors.messages[getBlockRes.getError()]);
- }
- BlockInfo block = getBlockRes.getBlockInfo();
- CoreSubOperation subOperation = blockMap.get(rpcFuture.getTicket());
- StorageFuture subFuture = prepareAndTrigger(subOperation, dataBuf, block);
- multiOperation.add(subFuture);
- blockCache.put(subOperation.key(), block);
- }
-
- if (!multiOperation.isProcessed()){
- throw new IOException("Internal error, processed data != operation length");
- }
-
- dataBuf.limit(multiOperation.getBufferLimit());
- dataBuf.position(multiOperation.getCurrentBufferPosition());
- return multiOperation;
- }
-
- final void prefetchMetadata() throws Exception {
- long key = CoreSubOperation.createKey(fileInfo.getFd(), position);
- if (blockCache.containsKey(key)){
- return;
- }
- if (nextBlockCache.containsKey(key)){
- return;
- }
- this.syncedCapacity = fileInfo.getCapacity();
- RpcFuture<RpcGetBlock> nextBlock = namenodeClientRpc.getBlock(fileInfo.getFd(), fileInfo.getToken(), position, syncedCapacity);
- nextBlock.setPrefetched(true);
- nextBlockCache.put(key, nextBlock);
- this.ioStats.incPrefetchedOps();
- }
-
- void seek(long pos) throws IOException {
- long newOffset = Math.min(fileInfo.getCapacity(), Math.max(0, pos));
- if (newOffset == pos){
- this.position = newOffset;
- } else {
- throw new IOException("seek position out of range, pos " + pos + ", fileCapacity " + fileInfo.getCapacity());
- }
- }
-
- Future<Void> sync() throws IOException {
- Future<Void> future = null;
- if (fileInfo.getToken() > 0 && syncedCapacity < fileInfo.getCapacity()){
- syncedCapacity = fileInfo.getCapacity();
- future = new SyncNodeFuture(namenodeClientRpc.setFile(fileInfo, false));
- } else {
- future = new NoOperation();
- }
-
- return future;
- }
-
- void updateIOStats() {
- ioStats.setCapacity(fileInfo.getCapacity());
- }
-
- long getStreamId() {
- return streamId;
- }
-
- public long position() {
- return position;
- }
-
- CoreIOStatistics getCoreStatistics(){
- return ioStats;
- }
-
- public CoreNode getFile(){
- return node;
- }
-
- BufferCheckpoint getBufferCheckpoint(){
- return bufferCheckpoint;
- }
-
- //-----------------
-
- void setCapacity(long currentCapacity) {
- fileInfo.setCapacity(currentCapacity);
- }
-
- private long blockRemaining(){
- long blockOffset = position % CrailConstants.BLOCK_SIZE;
- long blockRemaining = CrailConstants.BLOCK_SIZE - blockOffset;
- return blockRemaining;
- }
-
- private StorageFuture prepareAndTrigger(CoreSubOperation opDesc, CrailBuffer dataBuf, BlockInfo block) throws Exception {
- try {
- StorageEndpoint endpoint = endpointCache.getDataEndpoint(block.getDnInfo());
- dataBuf.clear();
- dataBuf.position(opDesc.getBufferPosition());
- dataBuf.limit(dataBuf.position() + opDesc.getLen());
- StorageFuture subFuture = trigger(endpoint, opDesc, dataBuf, block);
- incStats(endpoint.isLocal());
- return subFuture;
- } catch(IOException e){
- LOG.info("ERROR: failed data operation");
- e.printStackTrace();
- throw e;
- }
- }
-
- private void incStats(boolean isLocal){
- if (CrailConstants.STATISTICS){
- if (isLocal){
- ioStats.incLocalOps();
- if (fileInfo.getType().isDirectory()){
- ioStats.incLocalDirOps();
- }
- } else {
- ioStats.incRemoteOps();
- if (fileInfo.getType().isDirectory()){
- ioStats.incRemoteDirOps();
- }
- }
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/client/src/main/java/com/ibm/crail/core/CoreStreamStatistics.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/com/ibm/crail/core/CoreStreamStatistics.java b/client/src/main/java/com/ibm/crail/core/CoreStreamStatistics.java
deleted file mode 100644
index d56ffa3..0000000
--- a/client/src/main/java/com/ibm/crail/core/CoreStreamStatistics.java
+++ /dev/null
@@ -1,213 +0,0 @@
-/*
- * Crail: A Multi-tiered Distributed Direct Access File System
- *
- * Author: Patrick Stuedi <st...@zurich.ibm.com>
- *
- * Copyright (C) 2016, IBM Corporation
- *
- * 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
- *
- * 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 com.ibm.crail.core;
-
-import java.util.concurrent.atomic.AtomicLong;
-
-import com.ibm.crail.*;
-import com.ibm.crail.CrailStatistics.StatisticsProvider;
-
-public class CoreStreamStatistics implements CrailStatistics.StatisticsProvider {
- private AtomicLong open;
- private AtomicLong openInput;
- private AtomicLong openOutput;
- private AtomicLong openInputDir;
- private AtomicLong openOutputDir;
- private AtomicLong close;
- private AtomicLong closeInput;
- private AtomicLong closeOutput;
- private AtomicLong closeInputDir;
- private AtomicLong closeOutputDir;
- private AtomicLong currentInput;
- private AtomicLong currentOutput;
- private AtomicLong maxInput;
- private AtomicLong maxOutput;
-
- public CoreStreamStatistics(){
- this.open = new AtomicLong(0);
- this.openInput = new AtomicLong(0);
- this.openOutput = new AtomicLong(0);
- this.openInputDir = new AtomicLong(0);
- this.openOutputDir = new AtomicLong(0);
- this.close = new AtomicLong(0);
- this.closeInput = new AtomicLong(0);
- this.closeOutput = new AtomicLong(0);
- this.closeInputDir = new AtomicLong(0);
- this.closeOutputDir = new AtomicLong(0);
- this.currentInput = new AtomicLong(0);
- this.currentOutput = new AtomicLong(0);
- this.maxInput = new AtomicLong(0);
- this.maxOutput = new AtomicLong(0);
- }
-
- @Override
- public String providerName() {
- return "core/streams";
- }
-
- @Override
- public String printStatistics() {
- return "open " + getOpen() + ", openInput " + getOpenInput() + ", openOutput " + getOpenOutput() + ", openInputDir " + getOpenInputDir() + ", openOutputDir " + getOpenOutputDir() +
- ", close " + getClose() + ", closeInput " + getCloseInput() + ", closeOutput " + getCloseOutput() + ", closeInputDir " + getCloseInputDir() + ", closeOutputDir " + getCloseOutputDir() +
- ", maxInput " + getMaxInput() + ", maxOutput " + getMaxOutput();
- }
-
- public void resetStatistics(){
- this.open.set(0);
- this.openInput.set(0);
- this.openOutput.set(0);
- this.openInputDir.set(0);
- this.openOutputDir.set(0);
- this.close.set(0);
- this.closeInput.set(0);
- this.closeOutput.set(0);
- this.closeInputDir.set(0);
- this.closeOutputDir.set(0);
- this.currentInput.set(0);
- this.currentOutput.set(0);
- this.maxInput.set(0);
- this.maxOutput.set(0);
- }
-
- public void mergeStatistics(StatisticsProvider provider) {
- }
-
- public void incOpen() {
- this.open.incrementAndGet();
- }
-
- public void incOpenInput() {
- this.openInput.incrementAndGet();
- }
-
- public void incOpenOutput() {
- this.openOutput.incrementAndGet();
- }
-
- public void incOpenInputDir() {
- this.openInputDir.incrementAndGet();
- }
-
- public void incOpenOutputDir() {
- this.openOutputDir.incrementAndGet();
- }
-
- public void incClose() {
- this.close.incrementAndGet();
- }
-
- public void incCloseInput() {
- this.closeInput.incrementAndGet();
- }
-
- public void incCloseOutput() {
- this.closeOutput.incrementAndGet();
- }
-
- public void incCloseInputDir() {
- this.closeInputDir.incrementAndGet();
- }
-
- public void incCloseOutputDir() {
- this.closeOutputDir.incrementAndGet();
- }
-
- public void incCurrentInput() {
- this.currentInput.incrementAndGet();
- }
-
- public void incCurrentOutput() {
- this.currentOutput.incrementAndGet();
- }
-
- public void decCurrentInput() {
- this.currentInput.decrementAndGet();
- }
-
- public void decCurrentOutput() {
- this.currentOutput.decrementAndGet();
- }
-
- public void incMaxInput() {
- maxInput.updateAndGet(x -> Math.max(x, currentInput.get()));
- }
-
- public void incMaxOutput() {
- maxOutput.updateAndGet(x -> Math.max(x, currentOutput.get()));
- }
-
- public long getOpen() {
- return open.get();
- }
-
- public long getOpenInput() {
- return openInput.get();
- }
-
- public long getOpenOutput() {
- return openOutput.get();
- }
-
- public long getOpenInputDir() {
- return openInputDir.get();
- }
-
- public long getOpenOutputDir() {
- return openOutputDir.get();
- }
-
- public long getClose() {
- return close.get();
- }
-
- public long getCloseInput() {
- return closeInput.get();
- }
-
- public long getCloseOutput() {
- return closeOutput.get();
- }
-
- public long getCloseInputDir() {
- return closeInputDir.get();
- }
-
- public long getCloseOutputDir() {
- return closeOutputDir.get();
- }
-
- public long getCurrentInput() {
- return currentInput.get();
- }
-
- public long getCurrentOutput() {
- return currentOutput.get();
- }
-
- public long getMaxInput() {
- return maxInput.get();
- }
-
- public long getMaxOutput() {
- return maxOutput.get();
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/client/src/main/java/com/ibm/crail/core/CoreSubOperation.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/com/ibm/crail/core/CoreSubOperation.java b/client/src/main/java/com/ibm/crail/core/CoreSubOperation.java
deleted file mode 100644
index a7050b5..0000000
--- a/client/src/main/java/com/ibm/crail/core/CoreSubOperation.java
+++ /dev/null
@@ -1,85 +0,0 @@
-/*
- * Crail: A Multi-tiered Distributed Direct Access File System
- *
- * Author: Patrick Stuedi <st...@zurich.ibm.com>
- *
- * Copyright (C) 2016, IBM Corporation
- *
- * 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
- *
- * 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 com.ibm.crail.core;
-
-import java.io.IOException;
-
-import com.ibm.crail.conf.CrailConstants;
-import com.ibm.crail.utils.CrailUtils;
-
-public class CoreSubOperation {
- private long fd;
- private long fileOffset;
- private int bufferPosition;
- private int len;
-
- private long blockOffset;
- private long blockStart;
- private long key;
-
- public CoreSubOperation(long fd, long fileOffset, int bufferPosition, int writeLen) throws IOException {
- this.fd = fd;
- this.fileOffset = fileOffset;
- this.bufferPosition = bufferPosition;
- this.len = writeLen;
-
- this.blockOffset = fileOffset % CrailConstants.BLOCK_SIZE;
- this.blockStart = CrailUtils.blockStartAddress(fileOffset);
-// this.key = CoreSubOperation.createKey(fd, blockStart);
- this.key = blockStart;
- }
-
- public long getFd() {
- return fd;
- }
-
- public long getBlockOffset() {
- return blockOffset;
- }
-
- public long getBlockStart() {
- return blockStart;
- }
-
- public int getBufferPosition(){
- return bufferPosition;
- }
-
- public int getLen() {
- return len;
- }
-
- @Override
- public String toString() {
- return "fd " + fd + ", fileOffset " + fileOffset + ", blockOffset " + blockOffset + ", len " + len + ", blockStart " + blockStart;
- }
-
- public long key(){
- return this.key;
- }
-
- public static long createKey(long fd, long fileOffset){
- long offset = CrailUtils.blockStartAddress(fileOffset);
- return offset;
-// return fd + ":" + offset;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/client/src/main/java/com/ibm/crail/core/CoreSyncOperation.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/com/ibm/crail/core/CoreSyncOperation.java b/client/src/main/java/com/ibm/crail/core/CoreSyncOperation.java
deleted file mode 100644
index 6af1ea0..0000000
--- a/client/src/main/java/com/ibm/crail/core/CoreSyncOperation.java
+++ /dev/null
@@ -1,40 +0,0 @@
-/*
- * Crail: A Multi-tiered Distributed Direct Access File System
- *
- * Author: Patrick Stuedi <st...@zurich.ibm.com>
- *
- * Copyright (C) 2016, IBM Corporation
- *
- * 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
- *
- * 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 com.ibm.crail.core;
-
-import java.util.concurrent.Future;
-import com.ibm.crail.CrailResult;
-
-public class CoreSyncOperation {
- private DirectoryOutputStream stream;
- private Future<CrailResult> future;
-
- public CoreSyncOperation(DirectoryOutputStream stream, Future<CrailResult> future) {
- this.stream = stream;
- this.future = future;
- }
-
- public void close() throws Exception {
- future.get();
- stream.close();
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/client/src/main/java/com/ibm/crail/core/DirectoryInputStream.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/com/ibm/crail/core/DirectoryInputStream.java b/client/src/main/java/com/ibm/crail/core/DirectoryInputStream.java
deleted file mode 100644
index 343e339..0000000
--- a/client/src/main/java/com/ibm/crail/core/DirectoryInputStream.java
+++ /dev/null
@@ -1,177 +0,0 @@
-/*
- * Crail: A Multi-tiered Distributed Direct Access File System
- *
- * Author: Patrick Stuedi <st...@zurich.ibm.com>
- *
- * Copyright (C) 2016, IBM Corporation
- *
- * 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
- *
- * 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 com.ibm.crail.core;
-
-import java.io.IOException;
-import java.util.Iterator;
-import java.util.Random;
-import java.util.concurrent.Future;
-import java.util.concurrent.TimeUnit;
-
-import org.slf4j.Logger;
-
-import com.ibm.crail.CrailBuffer;
-import com.ibm.crail.CrailResult;
-import com.ibm.crail.conf.CrailConstants;
-import com.ibm.crail.utils.CrailUtils;
-
-
-public class DirectoryInputStream implements Iterator<String> {
- private static final Logger LOG = CrailUtils.getLogger();
-
- private CoreInputStream stream;
- private CrailBuffer internalBuf;
- private CoreFileSystem fs;
- private String parent;
- private String currentFile;
- private int consumedRecords;
- private int availableRecords;
- private int[] blockTickets;
- private Random random;
- private boolean randomize;
- private boolean open;
-
- public DirectoryInputStream(CoreInputStream stream, boolean randomize) throws Exception {
- this.stream = stream;
- this.randomize = randomize;
- this.fs = stream.getFile().getFileSystem();
- this.parent = stream.getFile().getPath();
- this.internalBuf = fs.allocateBuffer();
- this.internalBuf.clear();
- this.internalBuf.position(this.internalBuf.capacity());
- this.currentFile = null;
- this.availableRecords = 0;
- this.consumedRecords = 0;
- this.random = new Random();
- this.blockTickets = new int[CrailConstants.BUFFER_SIZE/CrailConstants.DIRECTORY_RECORD];
- for (int i = 0; i < blockTickets.length; i++){
- blockTickets[i] = i;
- }
- this.open = true;
- }
-
- public boolean hasNext() {
- if (!open) {
- return false;
- }
- if (currentFile != null){
- return true;
- }
- while(hasRecord()){
- DirectoryRecord record = nextRecord();
- if (record.isValid()){
- currentFile = CrailUtils.combinePath(record.getParent(), record.getFile());
- break;
- }
- }
- return currentFile != null;
- }
-
- public String next(){
- if (!open) {
- return null;
- }
-
- String ret = currentFile;
- currentFile = null;
- return ret;
- }
-
- public boolean hasRecord() {
- if (!open) {
- return false;
- }
- if (fetchIfEmpty() > 0){
- return true;
- }
- try {
- close();
- } catch(Exception e){
- LOG.info("error when closing directory stream " + e.getMessage());
- }
- return false;
- }
-
- public DirectoryRecord nextRecord() {
- if (!open) {
- return null;
- }
-
- DirectoryRecord record = new DirectoryRecord(parent);
- int offset = blockTickets[consumedRecords]*CrailConstants.DIRECTORY_RECORD;
- internalBuf.position(offset);
- record.update(internalBuf);
- consumedRecords++;
- return record;
- }
-
- private int fetchIfEmpty() {
- try {
- if (consumedRecords == availableRecords){
- internalBuf.clear();
- Future<CrailResult> future = stream.read(internalBuf);
- if (future != null){
- long ret = future.get(CrailConstants.DATA_TIMEOUT, TimeUnit.MILLISECONDS).getLen();
- if (ret > 0){
- internalBuf.flip();
- availableRecords = internalBuf.remaining() / CrailConstants.DIRECTORY_RECORD;
- consumedRecords = 0;
- if (randomize){
- shuffleTickets(blockTickets, availableRecords);
- }
- }
- }
- }
-
- return availableRecords - consumedRecords;
- } catch(Exception e){
- return 0;
- }
- }
-
- public void close() throws IOException {
- try {
- if (!open) {
- return;
- }
-
- stream.close();
- fs.freeBuffer(internalBuf);
- internalBuf = null;
- open = false;
- } catch (Exception e) {
- throw new IOException(e);
- }
- }
-
- void shuffleTickets(int[] tickets, int length) {
- for (int i = 0; i < availableRecords; i++){
- blockTickets[i] = i;
- }
- for (int i = length - 1; i > 0; i--) {
- int index = random.nextInt(i + 1);
- int tmp = tickets[index];
- tickets[index] = tickets[i];
- tickets[i] = tmp;
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/client/src/main/java/com/ibm/crail/core/DirectoryOutputStream.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/com/ibm/crail/core/DirectoryOutputStream.java b/client/src/main/java/com/ibm/crail/core/DirectoryOutputStream.java
deleted file mode 100644
index ff72c14..0000000
--- a/client/src/main/java/com/ibm/crail/core/DirectoryOutputStream.java
+++ /dev/null
@@ -1,86 +0,0 @@
-/*
- * Crail: A Multi-tiered Distributed Direct Access File System
- *
- * Author: Patrick Stuedi <st...@zurich.ibm.com>
- *
- * Copyright (C) 2016, IBM Corporation
- *
- * 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
- *
- * 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 com.ibm.crail.core;
-
-import java.io.IOException;
-import java.util.concurrent.Future;
-
-import com.ibm.crail.CrailBuffer;
-import com.ibm.crail.CrailResult;
-
-public class DirectoryOutputStream {
- private CoreOutputStream stream;
- private CrailBuffer internalBuf;
- private CoreFileSystem fs;
- private boolean open;
-
- public DirectoryOutputStream(CoreOutputStream stream)
- throws Exception {
- this.fs = stream.getFile().getFileSystem();
- this.stream = stream;
- this.internalBuf = fs.allocateBuffer();
- this.open = true;
- }
-
- Future<CrailResult> writeRecord(DirectoryRecord record, long offset) throws Exception {
- if (!open) {
- throw new IOException("stream closed");
- }
-
- internalBuf.clear();
- record.write(internalBuf);
- internalBuf.flip();
- stream.seek(offset);
- Future<CrailResult> future = stream.write(internalBuf);
- return future;
- }
-
- public void close() throws IOException {
- try {
- if (!open){
- return;
- }
-
- stream.close();
- fs.freeBuffer(internalBuf);
- internalBuf = null;
- open = false;
- } catch (Exception e) {
- throw new IOException(e);
- }
- }
-
- //debug
-
- public int getBufCapacity(){
- return internalBuf.capacity();
- }
-
- public int getBufPosition(){
- return internalBuf.position();
- }
-
- public int getBufLimit(){
- return internalBuf.limit();
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/client/src/main/java/com/ibm/crail/core/DirectoryRecord.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/com/ibm/crail/core/DirectoryRecord.java b/client/src/main/java/com/ibm/crail/core/DirectoryRecord.java
deleted file mode 100644
index 0b4a1b2..0000000
--- a/client/src/main/java/com/ibm/crail/core/DirectoryRecord.java
+++ /dev/null
@@ -1,83 +0,0 @@
-/*
- * Crail: A Multi-tiered Distributed Direct Access File System
- *
- * Author: Patrick Stuedi <st...@zurich.ibm.com>
- *
- * Copyright (C) 2016, IBM Corporation
- *
- * 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
- *
- * 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 com.ibm.crail.core;
-
-import com.ibm.crail.CrailBuffer;
-import com.ibm.crail.conf.CrailConstants;
-import com.ibm.crail.utils.CrailUtils;
-
-public class DirectoryRecord {
- private int valid;
- private String parent;
- private String filename;
-
- public DirectoryRecord(boolean valid, String path){
- this.valid = valid == true ? 1 : 0;
- this.parent = CrailUtils.getParent(path);
- this.filename = CrailUtils.getName(path);
- }
-
- public DirectoryRecord(String parent) {
- this.valid = 0;
- this.parent = parent;
- this.filename = null;
-
- }
-
- public void write(CrailBuffer buffer) throws Exception {
- int oldposition = buffer.position();
- buffer.putInt(valid);
- byte barray[] = filename.getBytes();
- buffer.putInt(barray.length);
- buffer.put(barray);
- buffer.position(oldposition + CrailConstants.DIRECTORY_RECORD);
- }
-
- public void update(CrailBuffer buffer) {
- int oldlimit = buffer.limit();
- int tmplimit = buffer.position() + CrailConstants.DIRECTORY_RECORD;
- buffer.limit(tmplimit);
- valid = buffer.getInt();
- int length = buffer.getInt();
- byte barray[] = new byte[length];
- buffer.get(barray);
- filename = new String(barray);
- buffer.position(tmplimit);
- buffer.limit(oldlimit);
- }
-
- public boolean isValid() {
- return valid == 1;
- }
-
- public String toString() {
- return valid + "\t\t" + filename;
- }
-
- public String getFile() {
- return filename;
- }
-
- public String getParent() {
- return parent;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/client/src/main/java/com/ibm/crail/memory/BufferCache.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/com/ibm/crail/memory/BufferCache.java b/client/src/main/java/com/ibm/crail/memory/BufferCache.java
deleted file mode 100644
index b465c23..0000000
--- a/client/src/main/java/com/ibm/crail/memory/BufferCache.java
+++ /dev/null
@@ -1,145 +0,0 @@
-/*
- * Crail: A Multi-tiered Distributed Direct Access File System
- *
- * Author: Patrick Stuedi <st...@zurich.ibm.com>
- *
- * Copyright (C) 2016, IBM Corporation
- *
- * 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
- *
- * 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 com.ibm.crail.memory;
-
-import java.io.IOException;
-import java.nio.ByteBuffer;
-
-import com.ibm.crail.CrailStatistics.StatisticsProvider;
-import com.ibm.crail.conf.CrailConstants;
-import com.ibm.crail.utils.CrailUtils;
-
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.atomic.AtomicLong;
-
-import org.slf4j.Logger;
-
-import com.ibm.crail.*;
-
-public abstract class BufferCache implements CrailStatistics.StatisticsProvider {
- 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{
- 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);
- }
-
- @Override
- public String providerName() {
- return "cache/buffer";
- }
-
- @Override
- 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);
- this.cacheMisses.set(0);
- this.cacheOut.set(0);
- this.cacheMax.set(0);
- this.cacheMissesMap.set(0);
- this.cacheMissesHeap.set(0);
- }
-
- public void mergeStatistics(StatisticsProvider provider){
-
- }
-
- public CrailBuffer getBuffer() 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();
- if (buffer == null){
- buffer = OffHeapBuffer.wrap(ByteBuffer.allocateDirect(CrailConstants.BUFFER_SIZE));
- cacheMissesHeap.incrementAndGet();
- } else {
- cacheMissesMap.incrementAndGet();
- }
- }
- }
- }
-
- buffer.clear();
- return buffer;
- }
-
- public void putBuffer(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 {
- Class<?> nodeClass = Class.forName(name);
- if (BufferCache.class.isAssignableFrom(nodeClass)){
- Class<? extends BufferCache> bufferCacheClass = (Class<? extends BufferCache>) nodeClass;
- BufferCache bufferCache = bufferCacheClass.newInstance();
- return bufferCache;
- } else {
- throw new Exception("Cannot instantiate storage client of type " + name);
- }
-
- }
-}
-
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/client/src/main/java/com/ibm/crail/memory/MappedBufferCache.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/com/ibm/crail/memory/MappedBufferCache.java b/client/src/main/java/com/ibm/crail/memory/MappedBufferCache.java
deleted file mode 100644
index 134cb63..0000000
--- a/client/src/main/java/com/ibm/crail/memory/MappedBufferCache.java
+++ /dev/null
@@ -1,118 +0,0 @@
-/*
- * Crail: A Multi-tiered Distributed Direct Access File System
- *
- * Author: Patrick Stuedi <st...@zurich.ibm.com>
- *
- * Copyright (C) 2016, IBM Corporation
- *
- * 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
- *
- * 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 com.ibm.crail.memory;
-
-import java.io.File;
-import java.io.IOException;
-import java.io.RandomAccessFile;
-import java.nio.MappedByteBuffer;
-import java.nio.channels.FileChannel;
-import java.nio.channels.FileChannel.MapMode;
-import org.slf4j.Logger;
-
-import com.ibm.crail.conf.CrailConstants;
-import com.ibm.crail.utils.CrailUtils;
-import com.ibm.crail.CrailBuffer;
-
-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;
-
- public MappedBufferCache() throws IOException {
- super();
-
- id = "" + System.currentTimeMillis();
- directory = CrailUtils.getCacheDirectory(id);
- dir = new File(directory);
- if (!dir.exists()){
- dir.mkdirs();
- }
- for (File child : dir.listFiles()) {
- child.delete();
- }
-
- 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);
- }
-
- @Override
- public void close() {
- super.close();
-
- if (dir.exists()){
- for (File child : dir.listFiles()) {
- child.delete();
- }
- dir.delete();
- }
- LOG.info("mapped client cache closed");
- }
-
- public CrailBuffer allocateBuffer() throws IOException{
- return allocateRegion();
- }
-
- private CrailBuffer allocateRegion() throws IOException {
- if (currentRegion >= allocationCount){
- return null;
- }
-
- String path = directory + "/" + currentRegion++;
- RandomAccessFile randomFile = new RandomAccessFile(path, "rw");
- randomFile.setLength(CrailConstants.REGION_SIZE);
- FileChannel channel = randomFile.getChannel();
- MappedByteBuffer _mappedBuffer = channel.map(MapMode.READ_WRITE, 0,
- CrailConstants.REGION_SIZE);
- CrailBuffer mappedBuffer = OffHeapBuffer.wrap(_mappedBuffer);
- randomFile.close();
- 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();
- return buffer;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/client/src/main/java/com/ibm/crail/memory/OffHeapBuffer.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/com/ibm/crail/memory/OffHeapBuffer.java b/client/src/main/java/com/ibm/crail/memory/OffHeapBuffer.java
deleted file mode 100644
index 4c60546..0000000
--- a/client/src/main/java/com/ibm/crail/memory/OffHeapBuffer.java
+++ /dev/null
@@ -1,237 +0,0 @@
-/*
- * Crail: A Multi-tiered Distributed Direct Access File System
- *
- * Author: Patrick Stuedi <st...@zurich.ibm.com>
- *
- * Copyright (C) 2016, IBM Corporation
- *
- * 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
- *
- * 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 com.ibm.crail.memory;
-
-import java.io.IOException;
-import java.nio.ByteBuffer;
-
-import com.ibm.crail.CrailBuffer;
-import com.ibm.crail.utils.CrailUtils;
-
-import sun.nio.ch.DirectBuffer;
-
-public class OffHeapBuffer implements CrailBuffer {
- private CrailBuffer region;
- private ByteBuffer buffer;
- private long address;
-
- OffHeapBuffer(ByteBuffer buffer) {
- this.region = this;
- this.buffer = buffer;
- this.address = CrailUtils.getAddress(buffer);
- }
-
- OffHeapBuffer(CrailBuffer region, ByteBuffer buffer) {
- this.region = region;
- this.buffer = buffer;
- this.address = CrailUtils.getAddress(buffer);
- }
-
- public static OffHeapBuffer wrap(ByteBuffer buffer) throws IOException {
- if (!(buffer instanceof DirectBuffer)) {
- throw new IOException("buffer not offheap");
- }
- return new OffHeapBuffer(buffer);
- }
-
- @Override
- public int capacity() {
- return buffer.capacity();
- }
-
- @Override
- public int position() {
- return buffer.position();
- }
-
- @Override
- public CrailBuffer position(int newPosition) {
- buffer.position(newPosition);
- return this;
- }
-
- @Override
- public int limit() {
- return buffer.limit();
- }
-
- @Override
- public CrailBuffer limit(int newLimit) {
- buffer.limit(newLimit);
- return this;
- }
-
- @Override
- public CrailBuffer clear() {
- buffer.clear();
- return this;
- }
-
- @Override
- public CrailBuffer flip() {
- buffer.flip();
- return this;
- }
-
- @Override
- public int remaining() {
- return buffer.remaining();
- }
-
- @Override
- public boolean hasRemaining() {
- return buffer.hasRemaining();
- }
-
- @Override
- public CrailBuffer slice() {
- try {
- ByteBuffer slice = buffer.slice();
- return new OffHeapBuffer(region, slice);
- } catch(Exception e){
- return null;
- }
- }
-
- //Crail metadata
-
- public long address(){
- return address;
- }
-
- public CrailBuffer getRegion(){
- return region;
- }
-
- @Override
- public ByteBuffer getByteBuffer() {
- return buffer;
- }
-
- @Override
- public byte get() {
- return buffer.get();
- }
-
- //ByteBuffer data
-
- @Override
- public ByteBuffer put(byte b) {
- return buffer.put(b);
- }
-
- @Override
- public CrailBuffer get(byte[] dst, int offset, int length) {
- buffer.get(dst, offset, length);
- return this;
- }
-
- @Override
- public CrailBuffer put(byte[] src, int offset, int length) {
- buffer.put(src, offset, length);
- return this;
- }
-
- @Override
- public CrailBuffer get(byte[] bytes) {
- buffer.get(bytes);
- return this;
- }
-
- @Override
- public CrailBuffer put(byte[] bytes) {
- buffer.put(bytes);
- return this;
- }
-
- @Override
- public short getShort() {
- return buffer.getShort();
- }
-
- @Override
- public CrailBuffer putShort(short value) {
- buffer.putShort(value);
- return this;
- }
-
- @Override
- public int getInt() {
- return buffer.getInt();
- }
-
- @Override
- public CrailBuffer putInt(int value) {
- buffer.putInt(value);
- return this;
- }
-
- @Override
- public long getLong() {
- return buffer.getLong();
- }
-
- @Override
- public CrailBuffer putLong(long value) {
- buffer.putLong(value);
- return this;
- }
-
- @Override
- public float getFloat() {
- return buffer.getFloat();
- }
-
- @Override
- public CrailBuffer putFloat(float value) {
- buffer.putFloat(value);
- return this;
- }
-
- @Override
- public double getDouble() {
- return buffer.getDouble();
- }
-
- @Override
- public CrailBuffer putDouble(double value) {
- buffer.putDouble(value);
- return this;
- }
-
- @Override
- public CrailBuffer put(ByteBuffer src) {
- buffer.put(src);
- return this;
- }
-
- @Override
- public CrailBuffer get(ByteBuffer dst) {
- dst.put(buffer);
- return this;
- }
-
- @Override
- public String toString() {
- return buffer.toString();
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/client/src/main/java/com/ibm/crail/metadata/BlockInfo.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/com/ibm/crail/metadata/BlockInfo.java b/client/src/main/java/com/ibm/crail/metadata/BlockInfo.java
deleted file mode 100644
index 9fead62..0000000
--- a/client/src/main/java/com/ibm/crail/metadata/BlockInfo.java
+++ /dev/null
@@ -1,103 +0,0 @@
-/*
- * Crail: A Multi-tiered Distributed Direct Access File System
- *
- * Author: Patrick Stuedi <st...@zurich.ibm.com>
- *
- * Copyright (C) 2016, IBM Corporation
- *
- * 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
- *
- * 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 com.ibm.crail.metadata;
-
-import java.net.UnknownHostException;
-import java.nio.ByteBuffer;
-
-public class BlockInfo {
- public static int CSIZE = DataNodeInfo.CSIZE + 24;
-
- protected DataNodeInfo dnInfo;
- protected long lba;
- protected long addr;
- protected int length;
- protected int lkey;
-
- public BlockInfo(){
- this.dnInfo = new DataNodeInfo();
- this.lba = 0;
- this.addr = 0;
- this.length = 0;
- this.lkey = 0;
- }
-
- public BlockInfo(DataNodeInfo dnInfo, long lba, long addr, int length, int lkey){
- this.dnInfo = dnInfo;
- this.lba = lba;
- this.addr = addr;
- this.length = length;
- this.lkey = lkey;
- }
-
- public void setBlockInfo(BlockInfo blockInfo) {
- this.dnInfo.setDataNodeInfo(blockInfo.getDnInfo());
- this.lba = blockInfo.getLba();
- this.addr = blockInfo.getAddr();
- this.length = blockInfo.getLength();
- this.lkey = blockInfo.getLkey();
-
- }
-
- public int write(ByteBuffer buffer){
- this.getDnInfo().write(buffer);
- buffer.putLong(this.getLba());
- buffer.putLong(this.getAddr());
- buffer.putInt(this.getLength());
- buffer.putInt(this.getLkey());
- return CSIZE;
- }
-
- public void update(ByteBuffer buffer) throws UnknownHostException{
- this.dnInfo.update(buffer);
- this.lba = buffer.getLong();
- this.addr = buffer.getLong();
- this.length = buffer.getInt();
- this.lkey = buffer.getInt();
- }
-
- public long getLba() {
- return lba;
- }
-
- public long getAddr() {
- return addr;
- }
-
- public int getLength() {
- return length;
- }
-
- public int getLkey() {
- return lkey;
- }
-
- public DataNodeInfo getDnInfo() {
- return dnInfo;
- }
-
- @Override
- public String toString() {
- return "BlockInfo [dnInfo=" + dnInfo + ", lba=" + lba + ", addr="
- + addr + ", length=" + length + ", lkey=" + lkey + "]";
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/client/src/main/java/com/ibm/crail/metadata/DataNodeInfo.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/com/ibm/crail/metadata/DataNodeInfo.java b/client/src/main/java/com/ibm/crail/metadata/DataNodeInfo.java
deleted file mode 100644
index 48ae183..0000000
--- a/client/src/main/java/com/ibm/crail/metadata/DataNodeInfo.java
+++ /dev/null
@@ -1,127 +0,0 @@
-/*
- * Crail: A Multi-tiered Distributed Direct Access File System
- *
- * Author: Patrick Stuedi <st...@zurich.ibm.com>
- *
- * Copyright (C) 2016, IBM Corporation
- *
- * 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
- *
- * 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 com.ibm.crail.metadata;
-
-import java.net.UnknownHostException;
-import java.nio.ByteBuffer;
-import java.util.Arrays;
-
-import org.slf4j.Logger;
-
-import com.ibm.crail.utils.CrailUtils;
-
-public class DataNodeInfo {
- private static final Logger LOG = CrailUtils.getLogger();
- public static final int CSIZE = 20;
-
- private int storageType;
- private int storageClass;
- private int locationClass;
- private byte[] ipAddress;
- private int port;
-
- private long key;
-
- public DataNodeInfo(){
- this.storageType = 0;
- this.storageClass = 0;
- this.locationClass = 0;
- this.ipAddress = new byte[4];
- this.port = 0;
- this.key = 0;
- }
-
- public DataNodeInfo(int storageType, int storageClass, int locationClass, byte[] ipAddress, int port){
- this();
-
- this.storageType = storageType;
- this.storageClass = storageClass;
- this.locationClass = locationClass;
- for (int i = 0; i < ipAddress.length; i++){
- this.ipAddress[i] = ipAddress[i];
- }
- this.port = port;
- }
-
- void setDataNodeInfo(DataNodeInfo info) {
- this.storageType = info.getStorageType();
- this.storageClass = info.getStorageClass();
- this.locationClass = info.getLocationClass();
- for (int i = 0; i < ipAddress.length; i++){
- this.ipAddress[i] = info.getIpAddress()[i];
- }
- this.port = info.getPort();
- }
-
- public int write(ByteBuffer buffer){
- buffer.putInt(storageType);
- buffer.putInt(storageClass);
- buffer.putInt(locationClass);
- buffer.put(ipAddress);
- buffer.putInt(port);
- return CSIZE;
- }
-
- public void update(ByteBuffer buffer) throws UnknownHostException {
- this.storageType = buffer.getInt();
- this.storageClass = buffer.getInt();
- this.locationClass = buffer.getInt();
- buffer.get(ipAddress);
- this.port = buffer.getInt();
- }
-
- public byte[] getIpAddress() {
- return ipAddress;
- }
-
- public int getPort() {
- return port;
- }
-
- public int getLocationClass() {
- return locationClass;
- }
-
- public int getStorageType() {
- return storageType;
- }
-
- public int getStorageClass() {
- return storageClass;
- }
-
- public long key(){
- if (key == 0){
- int a = java.util.Arrays.hashCode(ipAddress);
- key = (((long)a) << 32) | (port & 0xffffffffL);
- }
- return key;
- }
-
- @Override
- public String toString() {
- return "DataNodeInfo [storageType=" + storageType + ", storageClass="
- + storageClass + ", locationClass=" + locationClass
- + ", ipAddress=" + Arrays.toString(ipAddress) + ", port="
- + port + ", key=" + key + "]";
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/client/src/main/java/com/ibm/crail/metadata/DataNodeStatistics.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/com/ibm/crail/metadata/DataNodeStatistics.java b/client/src/main/java/com/ibm/crail/metadata/DataNodeStatistics.java
deleted file mode 100644
index 8662cb0..0000000
--- a/client/src/main/java/com/ibm/crail/metadata/DataNodeStatistics.java
+++ /dev/null
@@ -1,69 +0,0 @@
-/*
- * Crail: A Multi-tiered Distributed Direct Access File System
- *
- * Author: Patrick Stuedi <st...@zurich.ibm.com>
- *
- * Copyright (C) 2016, IBM Corporation
- *
- * 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
- *
- * 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 com.ibm.crail.metadata;
-
-import java.net.UnknownHostException;
-import java.nio.ByteBuffer;
-
-public class DataNodeStatistics {
- public static final int CSIZE = 12;
-
- private long serviceId;
- private int freeBlockCount;
-
- public DataNodeStatistics(){
- this.serviceId = 0;
- this.freeBlockCount = 0;
- }
-
- public int write(ByteBuffer buffer){
- buffer.putLong(serviceId);
- buffer.putInt(freeBlockCount);
- return CSIZE;
- }
-
- public void update(ByteBuffer buffer) throws UnknownHostException {
- this.serviceId = buffer.getLong();
- this.freeBlockCount = buffer.getInt();
- }
-
- public int getFreeBlockCount() {
- return freeBlockCount;
- }
-
- public void setFreeBlockCount(int blockCount) {
- this.freeBlockCount = blockCount;
- }
-
- public void setStatistics(DataNodeStatistics statistics) {
- this.serviceId = statistics.getServiceId();
- this.freeBlockCount = statistics.getFreeBlockCount();
- }
-
- public void setServiceId(long serviceId) {
- this.serviceId = serviceId;
- }
-
- public long getServiceId(){
- return serviceId;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/client/src/main/java/com/ibm/crail/metadata/FileInfo.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/com/ibm/crail/metadata/FileInfo.java b/client/src/main/java/com/ibm/crail/metadata/FileInfo.java
deleted file mode 100644
index 5fe5b03..0000000
--- a/client/src/main/java/com/ibm/crail/metadata/FileInfo.java
+++ /dev/null
@@ -1,161 +0,0 @@
-/*
- * Crail: A Multi-tiered Distributed Direct Access File System
- *
- * Author: Patrick Stuedi <st...@zurich.ibm.com>
- *
- * Copyright (C) 2016, IBM Corporation
- *
- * 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
- *
- * 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 com.ibm.crail.metadata;
-
-import java.net.UnknownHostException;
-import java.nio.ByteBuffer;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicLong;
-
-import com.ibm.crail.CrailNodeType;
-import com.ibm.crail.conf.CrailConstants;
-
-public class FileInfo {
- public static final int CSIZE = 44;
-
- private long fd;
- protected AtomicLong capacity;
- private CrailNodeType type;
- private long dirOffset;
- private long token;
- private long modificationTime;
-
- public FileInfo(){
- this(-1, CrailNodeType.DATAFILE);
- }
-
- protected FileInfo(long fd, CrailNodeType type){
- this.fd = fd;
- this.type = type;
- this.dirOffset = -1;
- this.capacity = new AtomicLong(0);
- this.token = 0;
- this.modificationTime = 0;
- }
-
- public void setFileInfo(FileInfo fileInfo){
- this.fd = fileInfo.getFd();
- this.type = fileInfo.getType();
- this.dirOffset = fileInfo.getDirOffset();
-
- this.capacity.set(fileInfo.getCapacity());
- this.token = fileInfo.getToken();
- this.modificationTime = fileInfo.getModificationTime();
- }
-
- public int write(ByteBuffer buffer, boolean shipToken){
- buffer.putLong(fd);
- buffer.putLong(capacity.get());
- buffer.putInt(type.getLabel());
- buffer.putLong(dirOffset);
- if (shipToken){
- buffer.putLong(token);
- } else {
- buffer.putLong(0);
- }
- buffer.putLong(modificationTime);
-
- return CSIZE;
- }
-
- public void update(ByteBuffer buffer) throws UnknownHostException{
- fd = buffer.getLong();
- capacity.set(buffer.getLong());
- type = CrailNodeType.parse(buffer.getInt());
- dirOffset = buffer.getLong();
- token = buffer.getLong();
- modificationTime = buffer.getLong();
- }
-
- public long getCapacity() {
- return capacity.get();
- }
-
- public long setCapacity(long newcapacity) {
- long oldcapacity = capacity.get();
- if (newcapacity > oldcapacity){
- capacity.compareAndSet(oldcapacity, newcapacity);
- this.setModificationTime(System.currentTimeMillis());
- }
- return capacity.get();
- }
-
- public long incCapacity(int delta) {
- long newcapacity = capacity.addAndGet(delta);
- this.setModificationTime(System.currentTimeMillis());
- return newcapacity;
- }
-
- public void resetCapacity(){
- capacity.set(0);
- }
-
- public long getToken() {
- return token;
- }
-
- public void resetToken(){
- token = 0;
- }
-
- public void updateToken(){
- if (!type.isDirectory()){
- this.token = System.nanoTime() + TimeUnit.SECONDS.toNanos(CrailConstants.TOKEN_EXPIRATION);
- }
- }
-
- public long getModificationTime() {
- return modificationTime;
- }
-
- public void setModificationTime(long modificationTime) {
- this.modificationTime = modificationTime;
- }
-
- public long getDirOffset() {
- return dirOffset;
- }
-
- protected void setDirOffset(long dirOffset) {
- this.dirOffset = dirOffset;
- }
-
- public long getFd() {
- return fd;
- }
-
- public String toString() {
- return "fd " + fd + ", capacity " + capacity + ", type " + type.getLabel() + ", dirOffset " + dirOffset + ", token " + token;
- }
-
- public CrailNodeType getType(){
- return type;
- }
-
- public boolean tokenFree(){
- return System.nanoTime() > token;
- }
-
- public void setToken(long value) {
- this.token = value;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/client/src/main/java/com/ibm/crail/metadata/FileName.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/com/ibm/crail/metadata/FileName.java b/client/src/main/java/com/ibm/crail/metadata/FileName.java
deleted file mode 100644
index c74ca6e..0000000
--- a/client/src/main/java/com/ibm/crail/metadata/FileName.java
+++ /dev/null
@@ -1,129 +0,0 @@
-/*
- * Crail: A Multi-tiered Distributed Direct Access File System
- *
- * Author: Patrick Stuedi <st...@zurich.ibm.com>
- *
- * Copyright (C) 2016, IBM Corporation
- *
- * 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
- *
- * 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 com.ibm.crail.metadata;
-
-import java.io.IOException;
-import java.nio.ByteBuffer;
-import java.util.StringTokenizer;
-
-import com.ibm.crail.conf.CrailConstants;
-
-public class FileName {
- public static int CSIZE = 4 + CrailConstants.DIRECTORY_DEPTH*4;
-
- private int length;
- private int[] components;
-
- public FileName(){
- this.length = 0;
- this.components = new int[CrailConstants.DIRECTORY_DEPTH];
- for (int i = 0; i < components.length; i++){
- components[i] = 0;
- }
- }
-
- public FileName(String name) throws IOException {
- this();
- StringTokenizer tokenizer = new StringTokenizer(name, "/");
- if (tokenizer.countTokens() > components.length){
- throw new IOException("filename with too many tokens, filename " + name + ", tokens " + components.length);
- }
-
- this.length = tokenizer.countTokens();
- int i = 0;
- while (tokenizer.hasMoreElements()) {
- String token = tokenizer.nextToken();
- components[i] = token.hashCode();
- i++;
- }
- }
-
- public FileName(FileName name){
- this();
- this.length = name.length;
- for (int i = 0; i < components.length; i++){
- this.components[i] = name.components[i];
- }
- }
-
- public int write(ByteBuffer buffer) {
- buffer.putInt(length);
- int written = 4;
- for (int i = 0; i < components.length; i++){
- buffer.putInt(components[i]);
- written += 4;
- }
- return written;
- }
-
- public void update(ByteBuffer buffer) {
- this.length = buffer.getInt();
- for (int i = 0; i < components.length; i++){
- components[i] = buffer.getInt();
- }
- }
-
- public int getFileComponent(){
- return getComponent(length - 1);
- }
-
- public int getComponent(int index){
- if (index >= 0 && index < components.length){
- return components[index];
- } else {
- return 0;
- }
- }
-
- public int getLength(){
- return length;
- }
-
-
- public int size(){
- return CSIZE;
- }
-
- @Override
- public boolean equals(Object obj) {
- if (this == obj)
- return true;
- if (obj == null)
- return false;
- if (getClass() != obj.getClass())
- return false;
- FileName other = (FileName) obj;
- if (hashCode() != other.hashCode())
- return false;
- return true;
- }
-
- @Override
- public String toString() {
- String tmp = "components=/";
- for (int i = 0; i < components.length; i++){
- tmp = tmp + components[i] + "/";
- }
- tmp = tmp + ", length=" + length;
- return tmp;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/client/src/main/java/com/ibm/crail/rpc/RpcClient.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/com/ibm/crail/rpc/RpcClient.java b/client/src/main/java/com/ibm/crail/rpc/RpcClient.java
deleted file mode 100644
index d7e269a..0000000
--- a/client/src/main/java/com/ibm/crail/rpc/RpcClient.java
+++ /dev/null
@@ -1,43 +0,0 @@
-/*
- * Crail: A Multi-tiered Distributed Direct Access File System
- *
- * Author: Patrick Stuedi <st...@zurich.ibm.com>
- *
- * Copyright (C) 2016, IBM Corporation
- *
- * 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
- *
- * 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 com.ibm.crail.rpc;
-
-import java.net.InetSocketAddress;
-
-import com.ibm.crail.conf.Configurable;
-
-public interface RpcClient extends Configurable {
- public RpcConnection connect(InetSocketAddress address) throws Exception ;
- public void close();
-
- @SuppressWarnings("unchecked")
- public static RpcClient createInstance(String name) throws Exception {
- Class<?> nodeClass = Class.forName(name);
- if (RpcClient.class.isAssignableFrom(nodeClass)){
- Class<? extends RpcClient> clientClass = (Class<? extends RpcClient>) nodeClass;
- RpcClient clientInstance = clientClass.newInstance();
- return clientInstance;
- } else {
- throw new Exception("Cannot instantiate rpc client of type " + name);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/client/src/main/java/com/ibm/crail/rpc/RpcConnection.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/com/ibm/crail/rpc/RpcConnection.java b/client/src/main/java/com/ibm/crail/rpc/RpcConnection.java
deleted file mode 100644
index 3a4c619..0000000
--- a/client/src/main/java/com/ibm/crail/rpc/RpcConnection.java
+++ /dev/null
@@ -1,80 +0,0 @@
-/*
- * Crail: A Multi-tiered Distributed Direct Access File System
- *
- * Author: Patrick Stuedi <st...@zurich.ibm.com>
- *
- * Copyright (C) 2016, IBM Corporation
- *
- * 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
- *
- * 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 com.ibm.crail.rpc;
-
-import java.io.IOException;
-
-import com.ibm.crail.CrailNodeType;
-import com.ibm.crail.metadata.BlockInfo;
-import com.ibm.crail.metadata.DataNodeInfo;
-import com.ibm.crail.metadata.FileInfo;
-import com.ibm.crail.metadata.FileName;
-
-public interface RpcConnection {
- public abstract RpcFuture<RpcCreateFile> createFile(
- FileName filename, CrailNodeType type, int storageClass, int locationClass) throws IOException;
-
- public abstract RpcFuture<RpcGetFile> getFile(FileName filename,
- boolean writeable) throws IOException;
-
- public abstract RpcFuture<RpcVoid> setFile(FileInfo fileInfo,
- boolean close) throws IOException;
-
- public abstract RpcFuture<RpcDeleteFile> removeFile(
- FileName filename, boolean recursive) throws IOException;
-
- public abstract RpcFuture<RpcRenameFile> renameFile(
- FileName srcHash, FileName dstHash) throws IOException;
-
- public abstract RpcFuture<RpcGetBlock> getBlock(long fd,
- long token, long position, long capacity) throws IOException;
-
- public abstract RpcFuture<RpcGetLocation> getLocation(
- FileName fileName, long position) throws IOException;
-
- public abstract RpcFuture<RpcVoid> setBlock(BlockInfo blockInfo)
- throws Exception;
-
- public abstract RpcFuture<RpcGetDataNode> getDataNode(
- DataNodeInfo dnInfo) throws Exception;
-
- public abstract RpcFuture<RpcVoid> dumpNameNode()
- throws Exception;
-
- public abstract RpcFuture<RpcPing> pingNameNode()
- throws Exception;
-
- public abstract void close() throws Exception;
-
- @SuppressWarnings("unchecked")
- public static RpcConnection createInstance(String name) throws Exception {
- Class<?> nodeClass = Class.forName(name);
- if (RpcConnection.class.isAssignableFrom(nodeClass)){
- Class<? extends RpcConnection> rpcClass = (Class<? extends RpcConnection>) nodeClass;
- RpcConnection rpcInstance = rpcClass.newInstance();
- return rpcInstance;
- } else {
- throw new Exception("Cannot instantiate RPC client of type " + name);
- }
- }
-}
-
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/client/src/main/java/com/ibm/crail/rpc/RpcCreateFile.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/com/ibm/crail/rpc/RpcCreateFile.java b/client/src/main/java/com/ibm/crail/rpc/RpcCreateFile.java
deleted file mode 100644
index 813f52e..0000000
--- a/client/src/main/java/com/ibm/crail/rpc/RpcCreateFile.java
+++ /dev/null
@@ -1,32 +0,0 @@
-/*
- * Crail: A Multi-tiered Distributed Direct Access File System
- *
- * Author: Patrick Stuedi <st...@zurich.ibm.com>
- *
- * Copyright (C) 2016, IBM Corporation
- *
- * 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
- *
- * 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 com.ibm.crail.rpc;
-
-import com.ibm.crail.metadata.BlockInfo;
-import com.ibm.crail.metadata.FileInfo;
-
-public interface RpcCreateFile extends RpcResponse {
- public FileInfo getFile();
- public FileInfo getParent();
- public BlockInfo getFileBlock();
- public BlockInfo getDirBlock();
-}
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/client/src/main/java/com/ibm/crail/rpc/RpcDeleteFile.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/com/ibm/crail/rpc/RpcDeleteFile.java b/client/src/main/java/com/ibm/crail/rpc/RpcDeleteFile.java
deleted file mode 100644
index c596f12..0000000
--- a/client/src/main/java/com/ibm/crail/rpc/RpcDeleteFile.java
+++ /dev/null
@@ -1,29 +0,0 @@
-/*
- * Crail: A Multi-tiered Distributed Direct Access File System
- *
- * Author: Patrick Stuedi <st...@zurich.ibm.com>
- *
- * Copyright (C) 2016, IBM Corporation
- *
- * 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
- *
- * 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 com.ibm.crail.rpc;
-
-import com.ibm.crail.metadata.FileInfo;
-
-public interface RpcDeleteFile extends RpcResponse {
- public FileInfo getFile();
- public FileInfo getParent();
-}
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/client/src/main/java/com/ibm/crail/rpc/RpcDispatcher.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/com/ibm/crail/rpc/RpcDispatcher.java b/client/src/main/java/com/ibm/crail/rpc/RpcDispatcher.java
deleted file mode 100644
index 3b5b61b..0000000
--- a/client/src/main/java/com/ibm/crail/rpc/RpcDispatcher.java
+++ /dev/null
@@ -1,148 +0,0 @@
-package com.ibm.crail.rpc;
-
-import java.io.IOException;
-import java.util.concurrent.ConcurrentLinkedQueue;
-
-import org.slf4j.Logger;
-
-import com.ibm.crail.CrailNodeType;
-import com.ibm.crail.metadata.BlockInfo;
-import com.ibm.crail.metadata.DataNodeInfo;
-import com.ibm.crail.metadata.FileInfo;
-import com.ibm.crail.metadata.FileName;
-import com.ibm.crail.utils.CrailUtils;
-
-public class RpcDispatcher implements RpcConnection {
- private static final Logger LOG = CrailUtils.getLogger();
-
- private RpcConnection[] connections;
- private int setBlockIndex;
- private int getDataNodeIndex;
-
- public RpcDispatcher(ConcurrentLinkedQueue<RpcConnection> connectionList) {
- connections = new RpcConnection[connectionList.size()];
- for (int i = 0; i < connections.length; i++){
- connections[i] = connectionList.poll();
- }
- this.setBlockIndex = 0;
- this.getDataNodeIndex = 0;
- }
-
- @Override
- public RpcFuture<RpcCreateFile> createFile(FileName filename,
- CrailNodeType type, int storageClass, int locationClass)
- throws IOException {
- int index = computeIndex(filename.getComponent(0));
-// LOG.info("issuing create file for filename [" + filename.toString() + "], on index " + index);
- return connections[index].createFile(filename, type, storageClass, locationClass);
- }
-
- @Override
- public RpcFuture<RpcGetFile> getFile(FileName filename, boolean writeable)
- throws IOException {
- int index = computeIndex(filename.getComponent(0));
-// LOG.info("issuing get file for filename [" + filename.toString() + "], on index " + index);
- return connections[index].getFile(filename, writeable);
- }
-
- @Override
- public RpcFuture<RpcVoid> setFile(FileInfo fileInfo, boolean close)
- throws IOException {
- int index = computeIndex(fileInfo.getFd());
-// LOG.info("issuing set file for fd [" + fileInfo.getFd() + "], on index " + index);
- return connections[index].setFile(fileInfo, close);
- }
-
- @Override
- public RpcFuture<RpcDeleteFile> removeFile(FileName filename,
- boolean recursive) throws IOException {
- int index = computeIndex(filename.getComponent(0));
-// LOG.info("issuing remove file for filename [" + filename.toString() + "], on index " + index);
- return connections[index].removeFile(filename, recursive);
- }
-
- @Override
- public RpcFuture<RpcRenameFile> renameFile(FileName srcHash,
- FileName dstHash) throws IOException {
- int srcIndex = computeIndex(srcHash.getComponent(0));
- int dstIndex = computeIndex(srcHash.getComponent(0));
-// LOG.info("issuing remove file for src [" + srcHash.toString() + "," + dstHash.toString() + "], on index " + srcIndex);
- if (srcIndex != dstIndex){
- throw new IOException("Rename not supported across namenode domains");
- } else {
- return connections[srcIndex].renameFile(srcHash, dstHash);
- }
- }
-
- @Override
- public RpcFuture<RpcGetBlock> getBlock(long fd, long token, long position,
- long capacity) throws IOException {
- int index = computeIndex(fd);
-// LOG.info("issuing get block for fd [" + fd + "], on index " + index);
- return connections[index].getBlock(fd, token, position, capacity);
- }
-
- @Override
- public RpcFuture<RpcGetLocation> getLocation(FileName fileName,
- long position) throws IOException {
- int index = computeIndex(fileName.getComponent(0));
-// LOG.info("issuing get location for filename [" + fileName.toString() + "], on index " + index);
- return connections[index].getLocation(fileName, position);
- }
-
- @Override
- public RpcFuture<RpcVoid> setBlock(BlockInfo blockInfo) throws Exception {
-// LOG.info("issuing set block on index " + setBlockIndex);
- RpcFuture<RpcVoid> res = connections[setBlockIndex].setBlock(blockInfo);
- setBlockIndex = (setBlockIndex + 1) % connections.length;
- return res;
- }
-
- @Override
- public RpcFuture<RpcGetDataNode> getDataNode(DataNodeInfo dnInfo)
- throws Exception {
-// LOG.info("issuing get datanode on index " + getDataNodeIndex);
- RpcFuture<RpcGetDataNode> res = connections[getDataNodeIndex].getDataNode(dnInfo);
- getDataNodeIndex = (getDataNodeIndex + 1) % connections.length;
- return res;
- }
-
- @Override
- public RpcFuture<RpcVoid> dumpNameNode() throws Exception {
- return connections[0].dumpNameNode();
- }
-
- @Override
- public RpcFuture<RpcPing> pingNameNode() throws Exception {
- return connections[0].pingNameNode();
- }
-
- @Override
- public void close() throws Exception {
- for (RpcConnection connection : connections){
- connection.close();
- }
- }
-
- @Override
- public String toString() {
- String address = "";
- for (RpcConnection connection : connections){
- address = address + ", " + connection.toString();
- }
-
- return address;
- }
-
- private int computeIndex(int component) {
- int index = ((component % connections.length) + connections.length) % connections.length;
- return index;
- }
-
- private int computeIndex(long component) {
- long connectionsLength = (long) connections.length;
- long _index = ((component % connectionsLength) + connectionsLength) % connectionsLength;
- int index = (int) _index;
- return index;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/client/src/main/java/com/ibm/crail/rpc/RpcErrors.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/com/ibm/crail/rpc/RpcErrors.java b/client/src/main/java/com/ibm/crail/rpc/RpcErrors.java
deleted file mode 100644
index 44d9d1a..0000000
--- a/client/src/main/java/com/ibm/crail/rpc/RpcErrors.java
+++ /dev/null
@@ -1,94 +0,0 @@
-/*
- * Crail: A Multi-tiered Distributed Direct Access File System
- *
- * Author: Patrick Stuedi <st...@zurich.ibm.com>
- *
- * Copyright (C) 2016, IBM Corporation
- *
- * 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
- *
- * 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 com.ibm.crail.rpc;
-
-import org.slf4j.Logger;
-
-import com.ibm.crail.utils.CrailUtils;
-
-public class RpcErrors {
- private static final Logger LOG = CrailUtils.getLogger();
- public static String[] messages = new String[32];
-
- //errors
- public static short ERR_OK = 0;
- public static short ERR_UNKNOWN = 1;
- public static short ERR_PROTOCOL_MISMATCH = 2;
- public static short ERR_CREATE_FILE_FAILED = 3;
- public static short ERR_GET_FILE_FAILED = 4;
- public static short ERR_TOKEN_TAKEN = 5;
- public static short ERR_FILE_NOT_OPEN = 6;
- public static short ERR_TOKEN_MISMATCH = 7;
- public static short ERR_CAPACITY_EXCEEDED = 8;
- public static short ERR_POSITION_NEGATIV = 9;
- public static short ERR_OFFSET_TOO_LARGE = 10;
- public static short ERR_NO_FREE_BLOCKS = 11;
- public static short ERR_NO_DATANODES = 12;
- public static short ERR_DATANODE_NOT_REGISTERED = 13;
- public static short ERR_INVALID_RPC_CMD = 14;
- public static short ERR_PARENT_MISSING = 15;
- public static short ERR_PARENT_NOT_DIR = 16;
- public static short ERR_FILE_EXISTS = 17;
- public static short ERR_FILE_NOT_FOUND = 18;
- public static short ERR_SRC_FILE_NOT_FOUND = 19;
- public static short ERR_DST_PARENT_NOT_FOUND = 20;
- public static short ERR_HAS_CHILDREN = 21;
- public static short ERR_FILE_COMPONENTS_EXCEEDED = 22;
- public static short ERR_DIRECTORY_EXISTS = 23;
- public static short ERR_TREE_CORRUPT = 24;
- public static short ERR_FILE_IS_NOT_DIR = 25;
- public static short ERR_DIR_LOCATION_AFFINITY_MISMATCH = 26;
- public static short ERR_ADD_BLOCK_FAILED = 27;
- public static short ERR_CREATE_FILE_BUG = 28;
-
- static {
- messages[ERR_OK] = "ERROR: No error, all fine";
- messages[ERR_UNKNOWN] = "ERROR: Unknown error";
- messages[ERR_PROTOCOL_MISMATCH] = "ERROR: Protocol mismatch";
- messages[ERR_CREATE_FILE_FAILED] = "ERROR: Create file failed";
- messages[ERR_GET_FILE_FAILED] = "ERROR: Get file failed";
- messages[ERR_TOKEN_TAKEN] = "ERROR: Token already taken";
- messages[ERR_FILE_NOT_OPEN] = "ERROR: File not open";
- messages[ERR_TOKEN_MISMATCH] = "ERROR: Token not maching";
- messages[ERR_CAPACITY_EXCEEDED] = "ERROR: Capacity exceeded";
- messages[ERR_POSITION_NEGATIV] = "ERROR: Position < 0";
- messages[ERR_OFFSET_TOO_LARGE] = "ERROR: Offset too large";
- messages[ERR_NO_FREE_BLOCKS] = "ERROR: No free blocks";
- messages[ERR_NO_DATANODES] = "ERROR: No data node";
- messages[ERR_DATANODE_NOT_REGISTERED] = "ERROR: Data node not registered";
- messages[ERR_INVALID_RPC_CMD] = "ERROR: Invalid RPC command";
- messages[ERR_PARENT_MISSING] = "ERROR: Parent node missing";
- messages[ERR_PARENT_NOT_DIR] = "ERROR: Parent node is not a dirctory";
- messages[ERR_FILE_EXISTS] = "ERROR: File already exists";
- messages[ERR_FILE_NOT_FOUND] = "ERROR: File not found";
- messages[ERR_SRC_FILE_NOT_FOUND] = "ERROR: Source file not found";
- messages[ERR_DST_PARENT_NOT_FOUND] = "ERROR: Destination parent folder not found";
- messages[ERR_HAS_CHILDREN] = "ERROR: Cannot remove folder with subfolders";
- messages[ERR_FILE_COMPONENTS_EXCEEDED] = "ERROR: Too many subfolders in directory name";
- messages[ERR_DIRECTORY_EXISTS] = "Directory exists";
- messages[ERR_TREE_CORRUPT] = "ERROR: File tree corrupt";
- messages[ERR_FILE_IS_NOT_DIR] = "File is not a directory";
- messages[ERR_DIR_LOCATION_AFFINITY_MISMATCH] = "Directories cannot have local affinity";
- messages[ERR_ADD_BLOCK_FAILED] = "Could not add block";
- messages[ERR_CREATE_FILE_BUG] = "Could not retrieve parent block";
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/client/src/main/java/com/ibm/crail/rpc/RpcFuture.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/com/ibm/crail/rpc/RpcFuture.java b/client/src/main/java/com/ibm/crail/rpc/RpcFuture.java
deleted file mode 100644
index 0c863fa..0000000
--- a/client/src/main/java/com/ibm/crail/rpc/RpcFuture.java
+++ /dev/null
@@ -1,34 +0,0 @@
-/*
- * Crail: A Multi-tiered Distributed Direct Access File System
- *
- * Author: Patrick Stuedi <st...@zurich.ibm.com>
- *
- * Copyright (C) 2016, IBM Corporation
- *
- * 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
- *
- * 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 com.ibm.crail.rpc;
-
-import java.util.concurrent.Future;
-
-public interface RpcFuture<T> extends Future<T> {
-
- public abstract int getTicket();
-
- public abstract boolean isPrefetched();
-
- public abstract void setPrefetched(boolean prefetched);
-
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/client/src/main/java/com/ibm/crail/rpc/RpcGetBlock.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/com/ibm/crail/rpc/RpcGetBlock.java b/client/src/main/java/com/ibm/crail/rpc/RpcGetBlock.java
deleted file mode 100644
index 73ad191..0000000
--- a/client/src/main/java/com/ibm/crail/rpc/RpcGetBlock.java
+++ /dev/null
@@ -1,29 +0,0 @@
-/*
- * Crail: A Multi-tiered Distributed Direct Access File System
- *
- * Author: Patrick Stuedi <st...@zurich.ibm.com>
- *
- * Copyright (C) 2016, IBM Corporation
- *
- * 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
- *
- * 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 com.ibm.crail.rpc;
-
-import com.ibm.crail.metadata.BlockInfo;
-
-public interface RpcGetBlock extends RpcResponse {
- public BlockInfo getBlockInfo();
- public void setBlockInfo(BlockInfo blockInfo);
-}
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/client/src/main/java/com/ibm/crail/rpc/RpcGetDataNode.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/com/ibm/crail/rpc/RpcGetDataNode.java b/client/src/main/java/com/ibm/crail/rpc/RpcGetDataNode.java
deleted file mode 100644
index 9123234..0000000
--- a/client/src/main/java/com/ibm/crail/rpc/RpcGetDataNode.java
+++ /dev/null
@@ -1,28 +0,0 @@
-/*
- * Crail: A Multi-tiered Distributed Direct Access File System
- *
- * Author: Patrick Stuedi <st...@zurich.ibm.com>
- *
- * Copyright (C) 2016, IBM Corporation
- *
- * 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
- *
- * 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 com.ibm.crail.rpc;
-
-import com.ibm.crail.metadata.DataNodeStatistics;
-
-public interface RpcGetDataNode extends RpcResponse {
- public DataNodeStatistics getStatistics();
-}
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/client/src/main/java/com/ibm/crail/rpc/RpcGetFile.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/com/ibm/crail/rpc/RpcGetFile.java b/client/src/main/java/com/ibm/crail/rpc/RpcGetFile.java
deleted file mode 100644
index af7b906..0000000
--- a/client/src/main/java/com/ibm/crail/rpc/RpcGetFile.java
+++ /dev/null
@@ -1,30 +0,0 @@
-/*
- * Crail: A Multi-tiered Distributed Direct Access File System
- *
- * Author: Patrick Stuedi <st...@zurich.ibm.com>
- *
- * Copyright (C) 2016, IBM Corporation
- *
- * 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
- *
- * 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 com.ibm.crail.rpc;
-
-import com.ibm.crail.metadata.BlockInfo;
-import com.ibm.crail.metadata.FileInfo;
-
-public interface RpcGetFile extends RpcResponse {
- public FileInfo getFile();
- public BlockInfo getFileBlock();
-}
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/client/src/main/java/com/ibm/crail/rpc/RpcGetLocation.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/com/ibm/crail/rpc/RpcGetLocation.java b/client/src/main/java/com/ibm/crail/rpc/RpcGetLocation.java
deleted file mode 100644
index 317d943..0000000
--- a/client/src/main/java/com/ibm/crail/rpc/RpcGetLocation.java
+++ /dev/null
@@ -1,31 +0,0 @@
-/*
- * Crail: A Multi-tiered Distributed Direct Access File System
- *
- * Author: Patrick Stuedi <st...@zurich.ibm.com>
- *
- * Copyright (C) 2016, IBM Corporation
- *
- * 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
- *
- * 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 com.ibm.crail.rpc;
-
-import java.io.IOException;
-
-import com.ibm.crail.metadata.BlockInfo;
-
-public interface RpcGetLocation extends RpcResponse {
- public BlockInfo getBlockInfo() throws IOException;
- public long getFd();
-}
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/client/src/main/java/com/ibm/crail/rpc/RpcPing.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/com/ibm/crail/rpc/RpcPing.java b/client/src/main/java/com/ibm/crail/rpc/RpcPing.java
deleted file mode 100644
index 596add2..0000000
--- a/client/src/main/java/com/ibm/crail/rpc/RpcPing.java
+++ /dev/null
@@ -1,26 +0,0 @@
-/*
- * Crail: A Multi-tiered Distributed Direct Access File System
- *
- * Author: Patrick Stuedi <st...@zurich.ibm.com>
- *
- * Copyright (C) 2016, IBM Corporation
- *
- * 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
- *
- * 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 com.ibm.crail.rpc;
-
-public interface RpcPing extends RpcResponse {
- public int getData();
-}