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:00 UTC
[16/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/rpc/RpcRenameFile.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/com/ibm/crail/rpc/RpcRenameFile.java b/client/src/main/java/com/ibm/crail/rpc/RpcRenameFile.java
deleted file mode 100644
index 32d8779..0000000
--- a/client/src/main/java/com/ibm/crail/rpc/RpcRenameFile.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 com.ibm.crail.metadata.BlockInfo;
-import com.ibm.crail.metadata.FileInfo;
-
-public interface RpcRenameFile extends RpcResponse {
- public FileInfo getSrcParent();
- public FileInfo getSrcFile();
- public FileInfo getDstParent();
- public FileInfo getDstFile();
- public BlockInfo getSrcBlock();
- public BlockInfo getDstBlock();
-}
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/client/src/main/java/com/ibm/crail/rpc/RpcResponse.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/com/ibm/crail/rpc/RpcResponse.java b/client/src/main/java/com/ibm/crail/rpc/RpcResponse.java
deleted file mode 100644
index 03a4f0d..0000000
--- a/client/src/main/java/com/ibm/crail/rpc/RpcResponse.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 RpcResponse {
- public short getError();
-}
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/client/src/main/java/com/ibm/crail/rpc/RpcVoid.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/com/ibm/crail/rpc/RpcVoid.java b/client/src/main/java/com/ibm/crail/rpc/RpcVoid.java
deleted file mode 100644
index 9e7f162..0000000
--- a/client/src/main/java/com/ibm/crail/rpc/RpcVoid.java
+++ /dev/null
@@ -1,25 +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 RpcVoid extends RpcResponse {
-}
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/client/src/main/java/com/ibm/crail/storage/StorageClient.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/com/ibm/crail/storage/StorageClient.java b/client/src/main/java/com/ibm/crail/storage/StorageClient.java
deleted file mode 100644
index f11b7c2..0000000
--- a/client/src/main/java/com/ibm/crail/storage/StorageClient.java
+++ /dev/null
@@ -1,44 +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.storage;
-
-import java.io.IOException;
-import com.ibm.crail.conf.Configurable;
-import com.ibm.crail.metadata.DataNodeInfo;
-
-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);
- if (StorageClient.class.isAssignableFrom(nodeClass)){
- Class<? extends StorageClient> storageClientClass = (Class<? extends StorageClient>) nodeClass;
- StorageClient client = storageClientClass.newInstance();
- return client;
- } 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/storage/StorageEndpoint.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/com/ibm/crail/storage/StorageEndpoint.java b/client/src/main/java/com/ibm/crail/storage/StorageEndpoint.java
deleted file mode 100644
index fb815eb..0000000
--- a/client/src/main/java/com/ibm/crail/storage/StorageEndpoint.java
+++ /dev/null
@@ -1,39 +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.storage;
-
-import java.io.IOException;
-import com.ibm.crail.CrailBuffer;
-import com.ibm.crail.metadata.BlockInfo;
-
-public interface StorageEndpoint{
-
- public abstract StorageFuture write(CrailBuffer buffer, BlockInfo remoteMr, long remoteOffset) throws IOException,
- InterruptedException;
-
- public abstract StorageFuture read(CrailBuffer buffer, BlockInfo remoteMr, long remoteOffset) throws IOException,
- InterruptedException;
-
- public abstract void close() throws IOException, InterruptedException;
-
- public abstract boolean isLocal();
-}
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/client/src/main/java/com/ibm/crail/storage/StorageFuture.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/com/ibm/crail/storage/StorageFuture.java b/client/src/main/java/com/ibm/crail/storage/StorageFuture.java
deleted file mode 100644
index 525c0be..0000000
--- a/client/src/main/java/com/ibm/crail/storage/StorageFuture.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.storage;
-
-import java.util.concurrent.Future;
-
-public interface StorageFuture extends Future<StorageResult> {
- public boolean isSynchronous();
-}
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/client/src/main/java/com/ibm/crail/storage/StorageResult.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/com/ibm/crail/storage/StorageResult.java b/client/src/main/java/com/ibm/crail/storage/StorageResult.java
deleted file mode 100644
index ae54994..0000000
--- a/client/src/main/java/com/ibm/crail/storage/StorageResult.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.storage;
-
-public interface StorageResult {
- public int getLen();
-}
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/client/src/main/java/com/ibm/crail/tools/CrailBenchmark.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/com/ibm/crail/tools/CrailBenchmark.java b/client/src/main/java/com/ibm/crail/tools/CrailBenchmark.java
deleted file mode 100644
index 436ba4e..0000000
--- a/client/src/main/java/com/ibm/crail/tools/CrailBenchmark.java
+++ /dev/null
@@ -1,1180 +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.tools;
-
-import java.nio.ByteBuffer;
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.Iterator;
-import java.util.Random;
-import java.util.concurrent.ArrayBlockingQueue;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentLinkedQueue;
-import java.util.concurrent.Future;
-import java.util.concurrent.LinkedBlockingQueue;
-
-import org.apache.commons.cli.CommandLine;
-import org.apache.commons.cli.CommandLineParser;
-import org.apache.commons.cli.DefaultParser;
-import org.apache.commons.cli.HelpFormatter;
-import org.apache.commons.cli.Option;
-import org.apache.commons.cli.Options;
-
-import com.ibm.crail.CrailBuffer;
-import com.ibm.crail.CrailBufferedInputStream;
-import com.ibm.crail.CrailBufferedOutputStream;
-import com.ibm.crail.CrailFile;
-import com.ibm.crail.CrailFS;
-import com.ibm.crail.CrailInputStream;
-import com.ibm.crail.CrailLocationClass;
-import com.ibm.crail.CrailNode;
-import com.ibm.crail.CrailOutputStream;
-import com.ibm.crail.CrailResult;
-import com.ibm.crail.CrailNodeType;
-import com.ibm.crail.CrailStorageClass;
-import com.ibm.crail.conf.CrailConfiguration;
-import com.ibm.crail.conf.CrailConstants;
-import com.ibm.crail.memory.OffHeapBuffer;
-import com.ibm.crail.utils.CrailUtils;
-import com.ibm.crail.utils.RingBuffer;
-
-public class CrailBenchmark {
- private int warmup;
- private CrailConfiguration conf;
- private CrailFS fs;
-
- public CrailBenchmark(int warmup) throws Exception {
- this.warmup = warmup;
- this.conf = new CrailConfiguration();
- this.fs = null;
- }
-
- private void open() throws Exception{
- if (fs == null){
- this.fs = CrailFS.newInstance(conf);
- }
- }
-
- private void close() throws Exception{
- if (fs != null){
- fs.close();
- fs = null;
- }
- }
-
- void write(String filename, int size, int loop, int storageClass, int locationClass, boolean buffered) throws Exception {
- System.out.println("write, filename " + filename + ", size " + size + ", loop " + loop + ", storageClass " + storageClass + ", locationClass " + locationClass + ", buffered " + buffered);
-
- CrailBuffer buf = null;
- if (size == CrailConstants.BUFFER_SIZE){
- buf = fs.allocateBuffer();
- } else if (size < CrailConstants.BUFFER_SIZE){
- CrailBuffer _buf = fs.allocateBuffer();
- _buf.clear().limit(size);
- buf = _buf.slice();
- } else {
- buf = OffHeapBuffer.wrap(ByteBuffer.allocateDirect(size));
- }
-
- //warmup
- ConcurrentLinkedQueue<CrailBuffer> bufferQueue = new ConcurrentLinkedQueue<CrailBuffer>();
- bufferQueue.add(buf);
- warmUp(filename, warmup, bufferQueue);
-
- //benchmark
- System.out.println("starting benchmark...");
- fs.getStatistics().reset();
- long _loop = (long) loop;
- long _bufsize = (long) CrailConstants.BUFFER_SIZE;
- long _capacity = _loop*_bufsize;
- double sumbytes = 0;
- double ops = 0;
- CrailFile file = fs.create(filename, CrailNodeType.DATAFILE, CrailStorageClass.get(storageClass), CrailLocationClass.get(locationClass)).get().asFile();
- CrailBufferedOutputStream bufferedStream = buffered ? file.getBufferedOutputStream(_capacity) : null;
- CrailOutputStream directStream = !buffered ? file.getDirectOutputStream(_capacity) : null;
- long start = System.currentTimeMillis();
- while (ops < loop) {
- buf.clear();
- if (buffered){
- bufferedStream.write(buf.getByteBuffer());
- } else {
- directStream.write(buf).get();
- }
- sumbytes = sumbytes + buf.capacity();
- ops = ops + 1.0;
- }
- if (buffered){
- bufferedStream.close();
- }
- long end = System.currentTimeMillis();
- double executionTime = ((double) (end - start)) / 1000.0;
- double throughput = 0.0;
- double latency = 0.0;
- double sumbits = sumbytes * 8.0;
- if (executionTime > 0) {
- throughput = sumbits / executionTime / 1000.0 / 1000.0;
- latency = 1000000.0 * executionTime / ops;
- }
-
- System.out.println("execution time " + executionTime);
- System.out.println("ops " + ops);
- System.out.println("sumbytes " + sumbytes);
- System.out.println("throughput " + throughput);
- System.out.println("latency " + latency);
-
- fs.getStatistics().print("close");
- }
-
- void writeAsync(String filename, int size, int loop, int batch, int storageClass, int locationClass) throws Exception {
- System.out.println("writeAsync, filename " + filename + ", size " + size + ", loop " + loop + ", batch " + batch + ", storageClass " + storageClass + ", locationClass " + locationClass);
-
- ConcurrentLinkedQueue<CrailBuffer> bufferQueue = new ConcurrentLinkedQueue<CrailBuffer>();
- for (int i = 0; i < batch; i++){
- CrailBuffer buf = null;
- if (size == CrailConstants.BUFFER_SIZE){
- buf = fs.allocateBuffer();
- } else if (size < CrailConstants.BUFFER_SIZE){
- CrailBuffer _buf = fs.allocateBuffer();
- _buf.clear().limit(size);
- buf = _buf.slice();
- } else {
- buf = OffHeapBuffer.wrap(ByteBuffer.allocateDirect(size));
- }
- bufferQueue.add(buf);
- }
-
- //warmup
- warmUp(filename, warmup, bufferQueue);
-
- //benchmark
- System.out.println("starting benchmark...");
- LinkedBlockingQueue<Future<CrailResult>> futureQueue = new LinkedBlockingQueue<Future<CrailResult>>();
- HashMap<Integer, CrailBuffer> futureMap = new HashMap<Integer, CrailBuffer>();
- fs.getStatistics().reset();
- long _loop = (long) loop;
- long _bufsize = (long) CrailConstants.BUFFER_SIZE;
- long _capacity = _loop*_bufsize;
- double sumbytes = 0;
- double ops = 0;
- CrailFile file = fs.create(filename, CrailNodeType.DATAFILE, CrailStorageClass.get(storageClass), CrailLocationClass.get(locationClass)).get().asFile();
- CrailOutputStream directStream = file.getDirectOutputStream(_capacity);
- long start = System.currentTimeMillis();
- for (int i = 0; i < batch - 1 && ops < loop; i++){
- CrailBuffer buf = bufferQueue.poll();
- buf.clear();
- Future<CrailResult> future = directStream.write(buf);
- futureQueue.add(future);
- futureMap.put(future.hashCode(), buf);
- ops = ops + 1.0;
- }
- while (ops < loop) {
- CrailBuffer buf = bufferQueue.poll();
- buf.clear();
- Future<CrailResult> future = directStream.write(buf);
- futureQueue.add(future);
- futureMap.put(future.hashCode(), buf);
-
- future = futureQueue.poll();
- future.get();
- buf = futureMap.get(future.hashCode());
- bufferQueue.add(buf);
-
- sumbytes = sumbytes + buf.capacity();
- ops = ops + 1.0;
- }
- while (!futureQueue.isEmpty()){
- Future<CrailResult> future = futureQueue.poll();
- future.get();
- CrailBuffer buf = futureMap.get(future.hashCode());
- sumbytes = sumbytes + buf.capacity();
- ops = ops + 1.0;
- }
- long end = System.currentTimeMillis();
- double executionTime = ((double) (end - start)) / 1000.0;
- double throughput = 0.0;
- double latency = 0.0;
- double sumbits = sumbytes * 8.0;
- if (executionTime > 0) {
- throughput = sumbits / executionTime / 1000.0 / 1000.0;
- latency = 1000000.0 * executionTime / ops;
- }
- directStream.close();
-
- System.out.println("execution time " + executionTime);
- System.out.println("ops " + ops);
- System.out.println("sumbytes " + sumbytes);
- System.out.println("throughput " + throughput);
- System.out.println("latency " + latency);
-
- fs.getStatistics().print("close");
- }
-
- void readSequential(String filename, int size, int loop, boolean buffered) throws Exception {
- System.out.println("readSequential, filename " + filename + ", size " + size + ", loop " + loop + ", buffered " + buffered);
-
- CrailBuffer buf = null;
- if (size == CrailConstants.BUFFER_SIZE){
- buf = fs.allocateBuffer();
- } else if (size < CrailConstants.BUFFER_SIZE){
- CrailBuffer _buf = fs.allocateBuffer();
- _buf.clear().limit(size);
- buf = _buf.slice();
- } else {
- buf = OffHeapBuffer.wrap(ByteBuffer.allocateDirect(size));
- }
-
- //warmup
- ConcurrentLinkedQueue<CrailBuffer> bufferQueue = new ConcurrentLinkedQueue<CrailBuffer>();
- bufferQueue.add(buf);
- warmUp(filename, warmup, bufferQueue);
-
- CrailFile file = fs.lookup(filename).get().asFile();
- CrailBufferedInputStream bufferedStream = file.getBufferedInputStream(file.getCapacity());
- CrailInputStream directStream = file.getDirectInputStream(file.getCapacity());
-
- //benchmark
- System.out.println("starting benchmark...");
- fs.getStatistics().reset();
- double sumbytes = 0;
- double ops = 0;
- long start = System.currentTimeMillis();
- while (ops < loop) {
- if (buffered){
- buf.clear();
- double ret = (double) bufferedStream.read(buf.getByteBuffer());
- if (ret > 0) {
- sumbytes = sumbytes + ret;
- ops = ops + 1.0;
- } else {
- ops = ops + 1.0;
- if (bufferedStream.position() == 0){
- break;
- } else {
- bufferedStream.seek(0);
- }
- }
- } else {
- buf.clear();
- double ret = (double) directStream.read(buf).get().getLen();
- if (ret > 0) {
- sumbytes = sumbytes + ret;
- ops = ops + 1.0;
- } else {
- ops = ops + 1.0;
- if (directStream.position() == 0){
- break;
- } else {
- directStream.seek(0);
- }
- }
- }
- }
- long end = System.currentTimeMillis();
- double executionTime = ((double) (end - start)) / 1000.0;
- double throughput = 0.0;
- double latency = 0.0;
- double sumbits = sumbytes * 8.0;
- if (executionTime > 0) {
- throughput = sumbits / executionTime / 1000.0 / 1000.0;
- latency = 1000000.0 * executionTime / ops;
- }
- bufferedStream.close();
- directStream.close();
-
- System.out.println("execution time " + executionTime);
- System.out.println("ops " + ops);
- System.out.println("sumbytes " + sumbytes);
- System.out.println("throughput " + throughput);
- System.out.println("latency " + latency);
-
- fs.getStatistics().print("close");
- }
-
- void readRandom(String filename, int size, int loop, boolean buffered) throws Exception{
- System.out.println("readRandom, filename " + filename + ", size " + size + ", loop " + loop + ", buffered " + buffered);
-
- CrailBuffer buf = null;
- if (size == CrailConstants.BUFFER_SIZE){
- buf = fs.allocateBuffer();
- } else if (size < CrailConstants.BUFFER_SIZE){
- CrailBuffer _buf = fs.allocateBuffer();
- _buf.clear().limit(size);
- buf = _buf.slice();
- } else {
- buf = OffHeapBuffer.wrap(ByteBuffer.allocateDirect(size));
- }
-
- //warmup
- ConcurrentLinkedQueue<CrailBuffer> bufferQueue = new ConcurrentLinkedQueue<CrailBuffer>();
- bufferQueue.add(buf);
- warmUp(filename, warmup, bufferQueue);
-
- //benchmark
- System.out.println("starting benchmark...");
- fs.getStatistics().reset();
- CrailFile file = fs.lookup(filename).get().asFile();
- CrailBufferedInputStream bufferedStream = file.getBufferedInputStream(file.getCapacity());
- CrailInputStream directStream = file.getDirectInputStream(file.getCapacity());
-
- double sumbytes = 0;
- double ops = 0;
- long _range = file.getCapacity() - ((long)buf.capacity());
- _range = _range / size;
- double range = (double) _range;
- Random random = new Random();
-
- long start = System.currentTimeMillis();
- while (ops < loop) {
- if (buffered){
- buf.clear();
- double _offset = range*random.nextDouble();
- long offset = (long) _offset*size;
- bufferedStream.seek(offset);
- double ret = (double) bufferedStream.read(buf.getByteBuffer());
- if (ret > 0) {
- sumbytes = sumbytes + ret;
- ops = ops + 1.0;
- } else {
- break;
- }
-
- } else {
- buf.clear();
- double _offset = range*random.nextDouble();
- long offset = (long) _offset*size;
- directStream.seek(offset);
- double ret = (double) directStream.read(buf).get().getLen();
- if (ret > 0) {
- sumbytes = sumbytes + ret;
- ops = ops + 1.0;
- } else {
- break;
- }
- }
- }
- long end = System.currentTimeMillis();
- double executionTime = ((double) (end - start)) / 1000.0;
- double throughput = 0.0;
- double latency = 0.0;
- double sumbits = sumbytes * 8.0;
- if (executionTime > 0) {
- throughput = sumbits / executionTime / 1000.0 / 1000.0;
- latency = 1000000.0 * executionTime / ops;
- }
- bufferedStream.close();
- directStream.close();
-
- System.out.println("execution time " + executionTime);
- System.out.println("ops " + ops);
- System.out.println("sumbytes " + sumbytes);
- System.out.println("throughput " + throughput);
- System.out.println("latency " + latency);
-
- fs.getStatistics().print("close");
- }
-
- void readSequentialAsync(String filename, int size, int loop, int batch) throws Exception {
- System.out.println("readSequentialAsync, filename " + filename + ", size " + size + ", loop " + loop + ", batch " + batch);
-
- ConcurrentLinkedQueue<CrailBuffer> bufferQueue = new ConcurrentLinkedQueue<CrailBuffer>();
- for (int i = 0; i < batch; i++){
- CrailBuffer buf = null;
- if (size == CrailConstants.BUFFER_SIZE){
- buf = fs.allocateBuffer();
- } else if (size < CrailConstants.BUFFER_SIZE){
- CrailBuffer _buf = fs.allocateBuffer();
- _buf.clear().limit(size);
- buf = _buf.slice();
- } else {
- buf = OffHeapBuffer.wrap(ByteBuffer.allocateDirect(size));
- }
- bufferQueue.add(buf);
- }
-
- //warmup
- warmUp(filename, warmup, bufferQueue);
-
- //benchmark
- System.out.println("starting benchmark...");
- double sumbytes = 0;
- double ops = 0;
- fs.getStatistics().reset();
- CrailFile file = fs.lookup(filename).get().asFile();
- CrailInputStream directStream = file.getDirectInputStream(file.getCapacity());
- HashMap<Integer, CrailBuffer> futureMap = new HashMap<Integer, CrailBuffer>();
- LinkedBlockingQueue<Future<CrailResult>> futureQueue = new LinkedBlockingQueue<Future<CrailResult>>();
- long start = System.currentTimeMillis();
- for (int i = 0; i < batch - 1 && ops < loop; i++){
- CrailBuffer buf = bufferQueue.poll();
- buf.clear();
- Future<CrailResult> future = directStream.read(buf);
- futureQueue.add(future);
- futureMap.put(future.hashCode(), buf);
- ops = ops + 1.0;
- }
- while (ops < loop) {
- CrailBuffer buf = bufferQueue.poll();
- buf.clear();
- Future<CrailResult> future = directStream.read(buf);
- futureQueue.add(future);
- futureMap.put(future.hashCode(), buf);
-
- future = futureQueue.poll();
- CrailResult result = future.get();
- buf = futureMap.get(future.hashCode());
- bufferQueue.add(buf);
-
- sumbytes = sumbytes + result.getLen();
- ops = ops + 1.0;
- }
- while (!futureQueue.isEmpty()){
- Future<CrailResult> future = futureQueue.poll();
- CrailResult result = future.get();
- futureMap.get(future.hashCode());
- sumbytes = sumbytes + result.getLen();
- ops = ops + 1.0;
- }
- long end = System.currentTimeMillis();
- double executionTime = ((double) (end - start)) / 1000.0;
- double throughput = 0.0;
- double latency = 0.0;
- double sumbits = sumbytes * 8.0;
- if (executionTime > 0) {
- throughput = sumbits / executionTime / 1000.0 / 1000.0;
- latency = 1000000.0 * executionTime / ops;
- }
- directStream.close();
-
- System.out.println("execution time " + executionTime);
- System.out.println("ops " + ops);
- System.out.println("sumbytes " + sumbytes);
- System.out.println("throughput " + throughput);
- System.out.println("latency " + latency);
-
- fs.getStatistics().print("close");
- }
-
- void readMultiStream(String filename, int size, int loop, int batch) throws Exception {
- System.out.println("readMultiStream, filename " + filename + ", size " + size + ", loop " + loop + ", batch " + batch);
-
- //warmup
- ConcurrentLinkedQueue<CrailBuffer> bufferQueue = new ConcurrentLinkedQueue<CrailBuffer>();
- for (int i = 0; i < warmup; i++){
- CrailBuffer buf = fs.allocateBuffer().limit(size).slice();
- bufferQueue.add(buf);
- }
- warmUp(filename, warmup, bufferQueue);
- while(!bufferQueue.isEmpty()){
- CrailBuffer buf = bufferQueue.poll();
- fs.freeBuffer(buf);
- }
-
- //benchmark
- System.out.println("starting benchmark...");
- fs.getStatistics().reset();
- CrailBuffer _buf = null;
- if (size == CrailConstants.BUFFER_SIZE){
- _buf = fs.allocateBuffer();
- } else if (size < CrailConstants.BUFFER_SIZE){
- CrailBuffer __buf = fs.allocateBuffer();
- __buf.clear().limit(size);
- _buf = __buf.slice();
- } else {
- _buf = OffHeapBuffer.wrap(ByteBuffer.allocateDirect(size));
- }
- ByteBuffer buf = _buf.getByteBuffer();
- for (int i = 0; i < loop; i++){
- CrailBufferedInputStream multiStream = fs.lookup(filename).get().asMultiFile().getMultiStream(batch);
- double sumbytes = 0;
- long _sumbytes = 0;
- double ops = 0;
- buf.clear();
- long start = System.currentTimeMillis();
- int ret = multiStream.read(buf);
- while(ret >= 0){
- sumbytes = sumbytes + ret;
- long _ret = (long) ret;
- _sumbytes += _ret;
- ops = ops + 1.0;
- buf.clear();
- ret = multiStream.read(buf);
- }
- long end = System.currentTimeMillis();
- multiStream.close();
-
- double executionTime = ((double) (end - start)) / 1000.0;
- double throughput = 0.0;
- double latency = 0.0;
- double sumbits = sumbytes * 8.0;
- if (executionTime > 0) {
- throughput = sumbits / executionTime / 1000.0 / 1000.0;
- latency = 1000000.0 * executionTime / ops;
- }
-
- System.out.println("round " + i + ":");
- System.out.println("bytes read " + _sumbytes);
- System.out.println("execution time " + executionTime);
- System.out.println("ops " + ops);
- System.out.println("throughput " + throughput);
- System.out.println("latency " + latency);
- }
-
- fs.getStatistics().print("close");
- }
-
- void createFile(String filename, int loop) throws Exception, InterruptedException {
- System.out.println("createFile, filename " + filename + ", loop " + loop);
-
- //warmup
- ConcurrentLinkedQueue<CrailBuffer> bufferQueue = new ConcurrentLinkedQueue<CrailBuffer>();
- CrailBuffer buf = fs.allocateBuffer();
- bufferQueue.add(buf);
- warmUp(filename, warmup, bufferQueue);
- fs.freeBuffer(buf);
-
- //benchmark
- System.out.println("starting benchmark...");
- fs.getStatistics().reset();
- LinkedBlockingQueue<String> pathQueue = new LinkedBlockingQueue<String>();
- fs.create(filename, CrailNodeType.DIRECTORY, CrailStorageClass.DEFAULT, CrailLocationClass.DEFAULT).get().syncDir();
- int filecounter = 0;
- for (int i = 0; i < loop; i++){
- String name = "" + filecounter++;
- String f = filename + "/" + name;
- pathQueue.add(f);
- }
-
- double ops = 0;
- long start = System.currentTimeMillis();
- while(!pathQueue.isEmpty()){
- String path = pathQueue.poll();
- fs.create(path, CrailNodeType.DATAFILE, CrailStorageClass.DEFAULT, CrailLocationClass.DEFAULT).get().syncDir();
- }
- long end = System.currentTimeMillis();
- double executionTime = ((double) (end - start)) / 1000.0;
- double latency = 0.0;
- if (executionTime > 0) {
- latency = 1000000.0 * executionTime / ops;
- }
-
- System.out.println("execution time " + executionTime);
- System.out.println("ops " + ops);
- System.out.println("latency " + latency);
-
- fs.getStatistics().print("close");
- }
-
- void createFileAsync(String filename, int loop, int batch) throws Exception, InterruptedException {
- System.out.println("createFileAsync, filename " + filename + ", loop " + loop + ", batch " + batch);
-
- //warmup
- ConcurrentLinkedQueue<CrailBuffer> bufferQueue = new ConcurrentLinkedQueue<CrailBuffer>();
- CrailBuffer buf = fs.allocateBuffer();
- bufferQueue.add(buf);
- warmUp(filename, warmup, bufferQueue);
- fs.freeBuffer(buf);
-
- //benchmark
- System.out.println("starting benchmark...");
- fs.getStatistics().reset();
- LinkedBlockingQueue<Future<CrailNode>> futureQueue = new LinkedBlockingQueue<Future<CrailNode>>();
- LinkedBlockingQueue<CrailFile> fileQueue = new LinkedBlockingQueue<CrailFile>();
- LinkedBlockingQueue<String> pathQueue = new LinkedBlockingQueue<String>();
- fs.create(filename, CrailNodeType.DIRECTORY, CrailStorageClass.DEFAULT, CrailLocationClass.DEFAULT).get().syncDir();
-
- for (int i = 0; i < loop; i++){
- String name = "/" + i;
- String f = filename + name;
- pathQueue.add(f);
- }
-
- long start = System.currentTimeMillis();
- for (int i = 0; i < loop; i += batch){
- //single operation == loop
- for (int j = 0; j < batch; j++) {
- String path = pathQueue.poll();
- Future<CrailNode> future = fs.create(path, CrailNodeType.DATAFILE, CrailStorageClass.DEFAULT, CrailLocationClass.DEFAULT);
- futureQueue.add(future);
- }
- for (int j = 0; j < batch; j++){
- Future<CrailNode> future = futureQueue.poll();
- CrailFile file = future.get().asFile();
- fileQueue.add(file);
- }
- for (int j = 0; j < batch; j++){
- CrailFile file = fileQueue.poll();
- file.syncDir();
- }
- }
- long end = System.currentTimeMillis();
- double executionTime = ((double) (end - start));
- double latency = executionTime*1000.0 / ((double) loop);
- System.out.println("execution time [ms] " + executionTime);
- System.out.println("latency [us] " + latency);
-
- fs.delete(filename, true).get().syncDir();
-
- fs.getStatistics().print("close");
-
- }
-
- void createMultiFile(String filename, int storageClass) throws Exception, InterruptedException {
- System.out.println("createMultiFile, filename " + filename);
- fs.create(filename, CrailNodeType.MULTIFILE, CrailStorageClass.get(storageClass), CrailLocationClass.DEFAULT).get().syncDir();
- }
-
- void getKey(String filename, int size, int loop) throws Exception {
- System.out.println("getKey, path " + filename + ", size " + size + ", loop " + loop);
-
- CrailBuffer buf = fs.allocateBuffer().clear().limit(size).slice();
- CrailFile file = fs.create(filename, CrailNodeType.DATAFILE, CrailStorageClass.DEFAULT, CrailLocationClass.DEFAULT).get().asFile();
- file.syncDir();
- CrailOutputStream directOutputStream = file.getDirectOutputStream(0);
- directOutputStream.write(buf).get();
- directOutputStream.close();
-
- //benchmark
- System.out.println("starting benchmark...");
- fs.getStatistics().reset();
- long start = System.currentTimeMillis();
- for (int i = 0; i < loop; i++){
- CrailInputStream directInputStream = fs.lookup(filename).get().asFile().getDirectInputStream(0);
- buf.clear();
- directInputStream.read(buf).get();
- directInputStream.close();
- }
- long end = System.currentTimeMillis();
- double executionTime = ((double) (end - start));
- double latency = executionTime * 1000.0 / ((double) loop);
- System.out.println("execution time [ms] " + executionTime);
- System.out.println("latency [us] " + latency);
-
- fs.getStatistics().print("close");
- }
-
- void getFile(String filename, int loop) throws Exception, InterruptedException {
- System.out.println("getFile, filename " + filename + ", loop " + loop);
-
- //warmup
- ConcurrentLinkedQueue<CrailBuffer> bufferQueue = new ConcurrentLinkedQueue<CrailBuffer>();
- CrailBuffer buf = fs.allocateBuffer();
- bufferQueue.add(buf);
- warmUp(filename, warmup, bufferQueue);
- fs.freeBuffer(buf);
-
- //benchmark
- System.out.println("starting benchmark...");
- fs.getStatistics().reset();
- double ops = 0;
- long start = System.currentTimeMillis();
- while (ops < loop) {
- ops = ops + 1.0;
- fs.lookup(filename).get().asFile();
- }
- long end = System.currentTimeMillis();
- double executionTime = ((double) (end - start)) / 1000.0;
- double latency = 0.0;
- if (executionTime > 0) {
- latency = 1000000.0 * executionTime / ops;
- }
- System.out.println("execution time " + executionTime);
- System.out.println("ops " + ops);
- System.out.println("latency " + latency);
-
- fs.getStatistics().print("close");
- fs.close();
- }
-
- void getFileAsync(String filename, int loop, int batch) throws Exception, InterruptedException {
- System.out.println("getFileAsync, filename " + filename + ", loop " + loop + ", batch " + batch);
-
- //warmup
- ConcurrentLinkedQueue<CrailBuffer> bufferQueue = new ConcurrentLinkedQueue<CrailBuffer>();
- CrailBuffer buf = fs.allocateBuffer();
- bufferQueue.add(buf);
- warmUp(filename, warmup, bufferQueue);
- fs.freeBuffer(buf);
-
- //benchmark
- System.out.println("starting benchmark...");
- fs.getStatistics().reset();
- LinkedBlockingQueue<Future<CrailNode>> fileQueue = new LinkedBlockingQueue<Future<CrailNode>>();
- long start = System.currentTimeMillis();
- for (int i = 0; i < loop; i++){
- //single operation == loop
- for (int j = 0; j < batch; j++){
- Future<CrailNode> future = fs.lookup(filename);
- fileQueue.add(future);
- }
- for (int j = 0; j < batch; j++){
- Future<CrailNode> future = fileQueue.poll();
- future.get();
- }
- }
- long end = System.currentTimeMillis();
- double executionTime = ((double) (end - start));
- double latency = executionTime*1000.0 / ((double) batch);
- System.out.println("execution time [ms] " + executionTime);
- System.out.println("latency [us] " + latency);
-
- fs.getStatistics().print("close");
- }
-
- void enumerateDir(String filename, int loop) throws Exception {
- System.out.println("reading enumarate dir, path " + filename);
-
- //warmup
- ConcurrentLinkedQueue<CrailBuffer> bufferQueue = new ConcurrentLinkedQueue<CrailBuffer>();
- CrailBuffer buf = fs.allocateBuffer();
- bufferQueue.add(buf);
- warmUp(filename, warmup, bufferQueue);
- fs.freeBuffer(buf);
-
- //benchmark
- System.out.println("starting benchmark...");
- fs.getStatistics().reset();
- long start = System.currentTimeMillis();
- for (int i = 0; i < loop; i++) {
- // single operation == loop
- Iterator<String> iter = fs.lookup(filename).get().asDirectory().listEntries();
- while (iter.hasNext()) {
- iter.next();
- }
- }
- long end = System.currentTimeMillis();
- double executionTime = ((double) (end - start));
- double latency = executionTime * 1000.0 / ((double) loop);
- System.out.println("execution time [ms] " + executionTime);
- System.out.println("latency [us] " + latency);
-
- fs.getStatistics().print("close");
- }
-
- void browseDir(String filename) throws Exception {
- System.out.println("reading enumarate dir, path " + filename);
-
- //benchmark
- System.out.println("starting benchmark...");
- fs.getStatistics().reset();
- CrailNode node = fs.lookup(filename).get();
- System.out.println("node type is " + node.getType());
-
- Iterator<String> iter = node.getType() == CrailNodeType.DIRECTORY ? node.asDirectory().listEntries() : node.asMultiFile().listEntries();
- while (iter.hasNext()) {
- String name = iter.next();
- System.out.println(name);
- }
- fs.getStatistics().print("close");
- }
-
- void early(String filename) throws Exception {
- ByteBuffer buf = ByteBuffer.allocateDirect(32);
- CrailFile file = fs.create(filename, CrailNodeType.DATAFILE, CrailStorageClass.DEFAULT, CrailLocationClass.DEFAULT).early().asFile();
- CrailBufferedOutputStream stream = file.getBufferedOutputStream(0);
- System.out.println("buffered stream initialized");
-
- Thread.sleep(1000);
- stream.write(buf);
- System.out.println("buffered stream written");
-
- Thread.sleep(1000);
- stream.write(buf);
- System.out.println("buffered stream written");
-
- stream.purge();
- stream.close();
-
- System.out.println("buffered stream closed");
-
- fs.getStatistics().print("close");
- }
-
- void writeInt(String filename, int loop) throws Exception {
- System.out.println("writeInt, filename " + filename + ", loop " + loop);
-
- //benchmark
- System.out.println("starting benchmark...");
- double ops = 0;
- CrailFile file = fs.create(filename, CrailNodeType.DATAFILE, CrailStorageClass.DEFAULT, CrailLocationClass.DEFAULT).get().asFile();
- CrailBufferedOutputStream outputStream = file.getBufferedOutputStream(loop*4);
- int intValue = 0;
- System.out.println("starting write at position " + outputStream.position());
- while (ops < loop) {
- System.out.println("writing position " + outputStream.position() + ", value " + intValue);
- outputStream.writeInt(intValue);
- intValue++;
- ops++;
- }
- outputStream.purge().get();
- outputStream.sync().get();
-
- fs.getStatistics().print("close");
- }
-
- void readInt(String filename, int loop) throws Exception {
- System.out.println("seek, filename " + filename + ", loop " + loop);
-
- //benchmark
- System.out.println("starting benchmark...");
- double ops = 0;
- CrailFile file = fs.lookup(filename).get().asFile();
- CrailBufferedInputStream inputStream = file.getBufferedInputStream(loop*4);
- System.out.println("starting read at position " + inputStream.position());
- while (ops < loop) {
- System.out.print("reading position " + inputStream.position() + ", expected " + inputStream.position()/4 + " ");
- int intValue = inputStream.readInt();
- System.out.println(", value " + intValue);
- ops++;
- }
- inputStream.close();
-
- fs.getStatistics().print("close");
- }
-
- void seekInt(String filename, int loop) throws Exception {
- System.out.println("seek, filename " + filename + ", loop " + loop);
-
- //benchmark
- System.out.println("starting benchmark...");
- double ops = 0;
- CrailFile file = fs.lookup(filename).get().asFile();
- Random random = new Random();
- long nbrOfInts = file.getCapacity() / 4;
- CrailBufferedInputStream seekStream = file.getBufferedInputStream(loop*4);
- System.out.println("starting seek phase, nbrOfInts " + nbrOfInts + ", position " + seekStream.position());
- long falseMatches = 0;
- while (ops < loop) {
- int intIndex = random.nextInt((int) nbrOfInts);
- int pos = intIndex*4;
- seekStream.seek((long) pos);
- int intValue = seekStream.readInt();
- if (intIndex != intValue){
- falseMatches++;
- System.out.println("reading, position " + pos + ", expected " + pos/4 + ", ########## value " + intValue);
- } else {
- System.out.println("reading, position " + pos + ", expected " + pos/4 + ", value " + intValue);
- }
- ops++;
- }
- seekStream.close();
- long end = System.currentTimeMillis();
-
- System.out.println("falseMatches " + falseMatches);
- fs.getStatistics().print("close");
- }
-
- void readMultiStreamInt(String filename, int loop, int batch) throws Exception {
- System.out.println("readMultiStreamInt, filename " + filename + ", loop " + loop + ", batch " + batch);
-
- System.out.println("starting benchmark...");
- fs.getStatistics().reset();
- CrailBufferedInputStream multiStream = fs.lookup(filename).get().asMultiFile().getMultiStream(batch);
- double ops = 0;
- long falseMatches = 0;
- while (ops < loop) {
- System.out.print("reading position " + multiStream.position() + ", expected " + multiStream.position()/4 + " ");
- long expected = multiStream.position()/4;
- int intValue = multiStream.readInt();
- if (expected != intValue){
- falseMatches++;
- }
- System.out.println(", value " + intValue);
- ops++;
- }
- multiStream.close();
-
- System.out.println("falseMatches " + falseMatches);
-
- fs.getStatistics().print("close");
- }
-
- void printLocationClass() throws Exception {
- System.out.println("locationClass " + fs.getLocationClass());
- }
-
- void locationMap() throws Exception {
- ConcurrentHashMap<String, String> locationMap = new ConcurrentHashMap<String, String>();
- CrailUtils.parseMap(CrailConstants.LOCATION_MAP, locationMap);
-
- System.out.println("Parsing locationMap " + CrailConstants.LOCATION_MAP);
- for (String key : locationMap.keySet()) {
- System.out.println("key " + key + ", value " + locationMap.get(key));
- }
- }
-
- void collectionTest(int size, int loop) throws Exception {
- System.out.println("collectionTest, size " + size + ", loop " + loop);
-
- RingBuffer<Object> ringBuffer = new RingBuffer<Object>(10);
- ArrayBlockingQueue<Object> arrayQueue = new ArrayBlockingQueue<Object>(10);
- LinkedBlockingQueue<Object> listQueue = new LinkedBlockingQueue<Object>();
-
- Object obj = new Object();
- long start = System.currentTimeMillis();
- for (int i = 0; i < loop; i++){
- for (int j = 0; j < size; j++){
- ringBuffer.add(obj);
- Object tmp = ringBuffer.peek();
- tmp = ringBuffer.poll();
- }
- }
- long end = System.currentTimeMillis();
- double executionTime = ((double) (end - start));
- System.out.println("ringbuffer, execution time [ms] " + executionTime);
-
- start = System.currentTimeMillis();
- for (int i = 0; i < loop; i++){
- for (int j = 0; j < size; j++){
- arrayQueue.add(obj);
- Object tmp = arrayQueue.peek();
- tmp = arrayQueue.poll();
- }
- }
- end = System.currentTimeMillis();
- executionTime = ((double) (end - start));
- System.out.println("arrayQueue, execution time [ms] " + executionTime);
-
- start = System.currentTimeMillis();
- for (int i = 0; i < loop; i++){
- for (int j = 0; j < size; j++){
- listQueue.add(obj);
- Object tmp = listQueue.peek();
- tmp = listQueue.poll();
- }
- }
- end = System.currentTimeMillis();
- executionTime = ((double) (end - start));
- System.out.println("arrayQueue, execution time [ms] " + executionTime);
- }
-
- private void warmUp(String filename, int operations, ConcurrentLinkedQueue<CrailBuffer> bufferList) throws Exception {
- Random random = new Random();
- String warmupFilename = filename + random.nextInt();
- System.out.println("warmUp, warmupFile " + warmupFilename + ", operations " + operations);
- if (operations > 0){
- CrailFile warmupFile = fs.create(warmupFilename, CrailNodeType.DATAFILE, CrailStorageClass.DEFAULT, CrailLocationClass.DEFAULT).get().asFile();
- CrailBufferedOutputStream warmupStream = warmupFile.getBufferedOutputStream(0);
- for (int i = 0; i < operations; i++){
- CrailBuffer buf = bufferList.poll();
- buf.clear();
- warmupStream.write(buf.getByteBuffer());
- bufferList.add(buf);
- }
- warmupStream.purge().get();
- warmupStream.close();
- fs.delete(warmupFilename, false).get().syncDir();
- }
- }
-
- public static void main(String[] args) throws Exception {
- String type = "";
- String filename = "/tmp.dat";
- int size = CrailConstants.BUFFER_SIZE;
- int loop = 1;
- int batch = 1;
- int warmup = 32;
- int experiments = 1;
- boolean keepOpen = false;
- int storageClass = 0;
- int locationClass = 0;
- boolean useBuffered = true;
-
- String benchmarkTypes = "write|writeAsync|readSequential|readRandom|readSequentialAsync|readMultiStream|"
- + "createFile|createFileAsync|createMultiFile|getKey|getFile|getFileAsync|enumerateDir|browseDir|"
- + "writeInt|readInt|seekInt|readMultiStreamInt|printLocationclass";
- Option typeOption = Option.builder("t").desc("type of experiment [" + benchmarkTypes + "]").hasArg().build();
- Option fileOption = Option.builder("f").desc("filename").hasArg().build();
- Option sizeOption = Option.builder("s").desc("buffer size [bytes]").hasArg().build();
- Option loopOption = Option.builder("k").desc("loop [1..n]").hasArg().build();
- Option batchOption = Option.builder("b").desc("batch size [1..n]").hasArg().build();
- Option storageOption = Option.builder("c").desc("storageClass for file [1..n]").hasArg().build();
- Option locationOption = Option.builder("p").desc("locationClass for file [1..n]").hasArg().build();
- Option warmupOption = Option.builder("w").desc("number of warmup operations [1..n]").hasArg().build();
- Option experimentOption = Option.builder("e").desc("number of experiments [1..n]").hasArg().build();
- Option openOption = Option.builder("o").desc("whether to keep the file system open [true|false]").hasArg().build();
- Option bufferedOption = Option.builder("m").desc("use buffer streams [true|false]").hasArg().build();
-
- Options options = new Options();
- options.addOption(typeOption);
- options.addOption(fileOption);
- options.addOption(sizeOption);
- options.addOption(loopOption);
- options.addOption(batchOption);
- options.addOption(storageOption);
- options.addOption(locationOption);
- options.addOption(warmupOption);
- options.addOption(experimentOption);
- options.addOption(openOption);
- options.addOption(bufferedOption);
-
- CommandLineParser parser = new DefaultParser();
- CommandLine line = parser.parse(options, Arrays.copyOfRange(args, 0, args.length));
- if (line.hasOption(typeOption.getOpt())) {
- type = line.getOptionValue(typeOption.getOpt());
- }
- if (line.hasOption(fileOption.getOpt())) {
- filename = line.getOptionValue(fileOption.getOpt());
- }
- if (line.hasOption(sizeOption.getOpt())) {
- size = Integer.parseInt(line.getOptionValue(sizeOption.getOpt()));
- }
- if (line.hasOption(loopOption.getOpt())) {
- loop = Integer.parseInt(line.getOptionValue(loopOption.getOpt()));
- }
- if (line.hasOption(batchOption.getOpt())) {
- batch = Integer.parseInt(line.getOptionValue(batchOption.getOpt()));
- }
- if (line.hasOption(storageOption.getOpt())) {
- storageClass = Integer.parseInt(line.getOptionValue(storageOption.getOpt()));
- }
- if (line.hasOption(locationOption.getOpt())) {
- locationClass = Integer.parseInt(line.getOptionValue(locationOption.getOpt()));
- }
- if (line.hasOption(warmupOption.getOpt())) {
- warmup = Integer.parseInt(line.getOptionValue(warmupOption.getOpt()));
- }
- if (line.hasOption(experimentOption.getOpt())) {
- experiments = Integer.parseInt(line.getOptionValue(experimentOption.getOpt()));
- }
- if (line.hasOption(openOption.getOpt())) {
- keepOpen = Boolean.parseBoolean(line.getOptionValue(openOption.getOpt()));
- }
- if (line.hasOption(bufferedOption.getOpt())) {
- useBuffered = Boolean.parseBoolean(line.getOptionValue(bufferedOption.getOpt()));
- }
-
- CrailBenchmark benchmark = new CrailBenchmark(warmup);
- if (type.equals("write")){
- benchmark.open();
- benchmark.write(filename, size, loop, storageClass, locationClass, useBuffered);
- benchmark.close();
- } else if (type.equalsIgnoreCase("writeAsync")) {
- benchmark.open();
- benchmark.writeAsync(filename, size, loop, batch, storageClass, locationClass);
- benchmark.close();
- } else if (type.equalsIgnoreCase("readSequential")){
- if (keepOpen) benchmark.open();
- for (int i = 0; i < experiments; i++){
- System.out.println("experiment " + i);
- if (!keepOpen) benchmark.open();
- benchmark.readSequential(filename, size, loop, useBuffered);
- if (!keepOpen) benchmark.close();
- }
- if (keepOpen) benchmark.close();
- } else if (type.equals("readRandom")){
- if (keepOpen) benchmark.open();
- for (int i = 0; i < experiments; i++){
- System.out.println("experiment " + i);
- if (!keepOpen) benchmark.open();
- benchmark.readRandom(filename, size, loop, useBuffered);
- if (!keepOpen) benchmark.close();
- }
- if (keepOpen) benchmark.close();
- } else if (type.equalsIgnoreCase("readSequentialAsync")) {
- if (keepOpen) benchmark.open();
- for (int i = 0; i < experiments; i++){
- System.out.println("experiment " + i);
- if (!keepOpen) benchmark.open();
- benchmark.readSequentialAsync(filename, size, loop, batch);
- if (!keepOpen) benchmark.close();
- }
- if (keepOpen) benchmark.close();
- } else if (type.equalsIgnoreCase("readMultiStream")) {
- if (keepOpen) benchmark.open();
- for (int i = 0; i < experiments; i++){
- System.out.println("experiment " + i);
- if (!keepOpen) benchmark.open();
- benchmark.readMultiStream(filename, size, loop, batch);
- if (!keepOpen) benchmark.close();
- }
- if (keepOpen) benchmark.close();
- } else if (type.equals("createFile")){
- benchmark.open();
- benchmark.createFile(filename, loop);
- benchmark.close();
- } else if (type.equals("createFileAsync")){
- benchmark.open();
- benchmark.createFileAsync(filename, loop, batch);
- benchmark.close();
- } else if (type.equalsIgnoreCase("createMultiFile")) {
- benchmark.open();
- benchmark.createMultiFile(filename, storageClass);
- benchmark.close();
- } else if (type.equalsIgnoreCase("getKey")) {
- benchmark.open();
- benchmark.getKey(filename, size, loop);
- benchmark.close();
- } else if (type.equals("getFile")){
- benchmark.open();
- benchmark.getFile(filename, loop);
- benchmark.close();
- } else if (type.equals("getFileAsync")){
- benchmark.open();
- benchmark.getFileAsync(filename, loop, batch);
- benchmark.close();
- } else if (type.equalsIgnoreCase("enumerateDir")) {
- benchmark.open();
- benchmark.enumerateDir(filename, batch);
- benchmark.close();
- } else if (type.equalsIgnoreCase("browseDir")) {
- benchmark.open();
- benchmark.browseDir(filename);
- benchmark.close();
- } else if (type.equalsIgnoreCase("early")) {
- benchmark.open();
- benchmark.early(filename);
- benchmark.close();
- } else if (type.equalsIgnoreCase("writeInt")) {
- benchmark.open();
- benchmark.writeInt(filename, loop);
- benchmark.close();
- } else if (type.equalsIgnoreCase("readInt")) {
- benchmark.open();
- benchmark.readInt(filename, loop);
- benchmark.close();
- } else if (type.equalsIgnoreCase("seekInt")) {
- benchmark.open();
- benchmark.seekInt(filename, loop);
- benchmark.close();
- } else if (type.equalsIgnoreCase("readMultiStreamInt")) {
- benchmark.open();
- benchmark.readMultiStreamInt(filename, loop, batch);
- benchmark.close();
- } else if (type.equalsIgnoreCase("printLocationClass")) {
- benchmark.open();
- benchmark.printLocationClass();
- benchmark.close();
- } else if (type.equalsIgnoreCase("collection")) {
- for (int i = 0; i < experiments; i++){
- benchmark.collectionTest(size, loop);
- }
- } else if (type.equalsIgnoreCase("locationMap")) {
- benchmark.locationMap();
- } else {
- HelpFormatter formatter = new HelpFormatter();
- formatter.printHelp("crail iobench", options);
- System.exit(-1);
- }
- }
-
-}
-
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/client/src/main/java/com/ibm/crail/tools/CrailFsck.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/com/ibm/crail/tools/CrailFsck.java b/client/src/main/java/com/ibm/crail/tools/CrailFsck.java
deleted file mode 100644
index cc1d273..0000000
--- a/client/src/main/java/com/ibm/crail/tools/CrailFsck.java
+++ /dev/null
@@ -1,241 +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.tools;
-
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.Iterator;
-import java.util.concurrent.atomic.AtomicInteger;
-
-import org.apache.commons.cli.CommandLine;
-import org.apache.commons.cli.CommandLineParser;
-import org.apache.commons.cli.DefaultParser;
-import org.apache.commons.cli.HelpFormatter;
-import org.apache.commons.cli.Option;
-import org.apache.commons.cli.Options;
-
-import com.ibm.crail.CrailBlockLocation;
-import com.ibm.crail.CrailDirectory;
-import com.ibm.crail.CrailFS;
-import com.ibm.crail.CrailFile;
-import com.ibm.crail.CrailLocationClass;
-import com.ibm.crail.CrailMultiFile;
-import com.ibm.crail.CrailNode;
-import com.ibm.crail.CrailNodeType;
-import com.ibm.crail.CrailStorageClass;
-import com.ibm.crail.conf.CrailConfiguration;
-import com.ibm.crail.conf.CrailConstants;
-import com.ibm.crail.core.CoreFileSystem;
-import com.ibm.crail.core.DirectoryInputStream;
-import com.ibm.crail.core.DirectoryRecord;
-import com.ibm.crail.metadata.FileName;
-import com.ibm.crail.utils.CrailUtils;
-
-public class CrailFsck {
-
- public CrailFsck(){
-
- }
-
- public void getLocations(String filename, long offset, long length) throws Exception {
- System.out.println("getLocations, filename " + filename + ", offset " + offset + ", len " + length);
- CrailConfiguration conf = new CrailConfiguration();
- CrailFS fs = CrailFS.newInstance(conf);
-
- CrailBlockLocation locations[] = fs.lookup(filename).get().getBlockLocations(offset, length);
- for (int i = 0; i < locations.length; i++){
- System.out.println("location " + i + " : " + locations[i].toString());
- }
- fs.close();
- }
-
- public void blockStatistics(String filename) throws Exception {
- HashMap<String, AtomicInteger> stats = new HashMap<String, AtomicInteger>();
- CrailConfiguration conf = new CrailConfiguration();
- CrailFS fs = CrailFS.newInstance(conf);
- CrailNode node = fs.lookup(filename).get();
-
- if (node.getType() == CrailNodeType.DIRECTORY){
- CrailDirectory directory = node.asDirectory();
- Iterator<String> iter = directory.listEntries();
- while (iter.hasNext()) {
- String path = iter.next();
- CrailFile child = fs.lookup(path).get().asFile();
- walkBlocks(stats, fs, child.getPath(), 0, child.getCapacity());
- }
- } else if (node.getType() == CrailNodeType.DATAFILE){
- CrailFile file = node.asFile();
- walkBlocks(stats, fs, file.getPath(), 0, file.getCapacity());
- } else if (node.getType() == CrailNodeType.MULTIFILE){
- CrailMultiFile directory = node.asMultiFile();
- Iterator<String> iter = directory.listEntries();
- while (iter.hasNext()) {
- String path = iter.next();
- CrailFile child = fs.lookup(path).get().asFile();
- walkBlocks(stats, fs, child.getPath(), 0, child.getCapacity());
- }
- }
-
- printStats(stats);
- fs.close();
- }
-
- public void namenodeDump() throws Exception {
- CrailConfiguration conf = new CrailConfiguration();
- CoreFileSystem fs = new CoreFileSystem(conf);
- fs.dumpNameNode();
- fs.close();
- }
-
- public void directoryDump(String filename, boolean randomize) throws Exception {
- CrailConfiguration conf = new CrailConfiguration();
- CrailConstants.updateConstants(conf);
- CoreFileSystem fs = new CoreFileSystem(conf);
- DirectoryInputStream iter = fs._listEntries(filename, randomize);
- System.out.println("#hash \t\tname\t\tfilecomponent");
- int i = 0;
- while(iter.hasRecord()){
- DirectoryRecord record = iter.nextRecord();
- String path = CrailUtils.combinePath(record.getParent(), record.getFile());
- FileName hash = new FileName(path);
- System.out.format(i + ": " + "%08d\t\t%s\t%d\n", record.isValid() ? 1 : 0, padRight(record.getFile(), 8), hash.getFileComponent());
- i++;
- }
- iter.close();
- fs.closeFileSystem();
- }
-
- public void ping() throws Exception {
- CrailConfiguration conf = new CrailConfiguration();
- CrailConstants.updateConstants(conf);
- CoreFileSystem fs = new CoreFileSystem(conf);
- fs.ping();
- fs.closeFileSystem();
- }
-
- public void createDirectory(String filename, int storageClass, int locationClass) throws Exception {
- System.out.println("createDirectory, filename " + filename + ", storageClass " + storageClass + ", locationClass " + locationClass);
- CrailConfiguration conf = new CrailConfiguration();
- CrailFS fs = CrailFS.newInstance(conf);
- fs.create(filename, CrailNodeType.DIRECTORY, CrailStorageClass.get(storageClass), CrailLocationClass.get(locationClass)).get().syncDir();
- fs.close();
- }
-
- //-----------------
-
- private String padRight(String s, int n) {
- return String.format("%1$-" + n + "s", s);
- }
-
- private void printStats(HashMap<String, AtomicInteger> stats) {
- for (Iterator<String> iter = stats.keySet().iterator(); iter.hasNext(); ){
- String key = iter.next();
- System.out.println(key + "\t" + stats.get(key));
- }
- }
-
- private void walkBlocks(HashMap<String, AtomicInteger> stats, CrailFS fs, String filePath, long offset, long len) throws Exception {
-// System.out.println("printing locations for path " + filePath);
- CrailBlockLocation locations[] = fs.lookup(filePath).get().asFile().getBlockLocations(offset, len);
- for (int i = 0; i < locations.length; i++){
- for (int j = 0; j < locations[i].getNames().length; j++){
- String name = locations[i].getNames()[j];
- String host = name.split(":")[0];
-// System.out.println("..........names " + host);
- incStats(stats, host);
- }
- }
- }
-
- private void incStats(HashMap<String, AtomicInteger> stats, String host) {
- if (!stats.containsKey(host)){
- stats.put(host, new AtomicInteger(0));
- }
- stats.get(host).incrementAndGet();
- }
-
-
- public static void main(String[] args) throws Exception {
- String type = "";
- String filename = "/tmp.dat";
- long offset = 0;
- long length = 1;
- boolean randomize = false;
- int storageClass = 0;
- int locationClass = 0;
-
- Option typeOption = Option.builder("t").desc("type of experiment [getLocations|directoryDump|namenodeDump|blockStatistics|ping|createDirectory]").hasArg().build();
- Option fileOption = Option.builder("f").desc("filename").hasArg().build();
- Option offsetOption = Option.builder("y").desc("offset into the file").hasArg().build();
- Option lengthOption = Option.builder("l").desc("length of the file [bytes]").hasArg().build();
- Option storageOption = Option.builder("c").desc("storageClass for file [1..n]").hasArg().build();
- Option locationOption = Option.builder("p").desc("locationClass for file [1..n]").hasArg().build();
-
- Options options = new Options();
- options.addOption(typeOption);
- options.addOption(fileOption);
- options.addOption(offsetOption);
- options.addOption(lengthOption);
- options.addOption(storageOption);
- options.addOption(locationOption);
-
- CommandLineParser parser = new DefaultParser();
- CommandLine line = parser.parse(options, Arrays.copyOfRange(args, 0, args.length));
- if (line.hasOption(typeOption.getOpt())) {
- type = line.getOptionValue(typeOption.getOpt());
- }
- if (line.hasOption(fileOption.getOpt())) {
- filename = line.getOptionValue(fileOption.getOpt());
- }
- if (line.hasOption(offsetOption.getOpt())) {
- offset = Long.parseLong(line.getOptionValue(offsetOption.getOpt()));
- }
- if (line.hasOption(lengthOption.getOpt())) {
- length = Long.parseLong(line.getOptionValue(lengthOption.getOpt()));
- }
- if (line.hasOption(storageOption.getOpt())) {
- storageClass = Integer.parseInt(line.getOptionValue(storageOption.getOpt()));
- }
- if (line.hasOption(locationOption.getOpt())) {
- locationClass = Integer.parseInt(line.getOptionValue(locationOption.getOpt()));
- }
-
- CrailFsck fsck = new CrailFsck();
- if (type.equals("getLocations")){
- fsck.getLocations(filename, offset, length);
- } else if (type.equals("directoryDump")){
- fsck.directoryDump(filename, randomize);
- } else if (type.equals("namenodeDump")){
- fsck.namenodeDump();
- } else if (type.equals("blockStatistics")){
- fsck.blockStatistics(filename);
- } else if (type.equals("ping")){
- fsck.ping();
- } else if (type.equals("createDirectory")){
- fsck.createDirectory(filename, storageClass, locationClass);
- } else {
- HelpFormatter formatter = new HelpFormatter();
- formatter.printHelp("crail fsck", options);
- System.exit(-1);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/client/src/main/java/com/ibm/crail/utils/AtomicIntegerModulo.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/com/ibm/crail/utils/AtomicIntegerModulo.java b/client/src/main/java/com/ibm/crail/utils/AtomicIntegerModulo.java
deleted file mode 100644
index fe9d20c..0000000
--- a/client/src/main/java/com/ibm/crail/utils/AtomicIntegerModulo.java
+++ /dev/null
@@ -1,36 +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.utils;
-
-import java.util.concurrent.atomic.AtomicInteger;
-
-public class AtomicIntegerModulo {
- private AtomicInteger counter;
-
- public AtomicIntegerModulo(){
- this.counter = new AtomicInteger(0);
- }
-
- public int getAndIncrement() {
- return counter.getAndIncrement() & Integer.MAX_VALUE;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/client/src/main/java/com/ibm/crail/utils/BlockCache.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/com/ibm/crail/utils/BlockCache.java b/client/src/main/java/com/ibm/crail/utils/BlockCache.java
deleted file mode 100644
index 9a76f2c..0000000
--- a/client/src/main/java/com/ibm/crail/utils/BlockCache.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.utils;
-
-import java.util.concurrent.ConcurrentHashMap;
-
-import com.ibm.crail.metadata.BlockInfo;
-
-public class BlockCache {
- private ConcurrentHashMap<Long, FileBlockCache> blockCache;
-
- public BlockCache(){
- this.blockCache = new ConcurrentHashMap<Long, FileBlockCache>();
- }
-
- public FileBlockCache getFileBlockCache(long fd){
- FileBlockCache fileBlockCache = blockCache.get(fd);
- if (fileBlockCache == null){
- fileBlockCache = new FileBlockCache(fd);
- FileBlockCache oldFileBlockCache = blockCache.putIfAbsent(fd, fileBlockCache);
- if (oldFileBlockCache != null){
- fileBlockCache = oldFileBlockCache;
- }
- }
- return fileBlockCache;
- }
-
- public void remove(long fd) {
- blockCache.remove(fd);
- }
-
- public void purge() {
- blockCache.clear();
- }
-
- public static class FileBlockCache {
- private long fd;
- private ConcurrentHashMap<Long, BlockInfo> fileBlockCache;
-
- public FileBlockCache(long fd){
- this.fd = fd;
- this.fileBlockCache = new ConcurrentHashMap<Long, BlockInfo>();
- }
-
- public void put(long blockstart, BlockInfo block){
- this.fileBlockCache.put(blockstart, block);
- }
-
- public BlockInfo get(long blockstart){
- return this.fileBlockCache.get(blockstart);
- }
-
- public boolean containsKey(long blockstart) {
- return this.fileBlockCache.containsKey(blockstart);
- }
-
- public long getFd() {
- return fd;
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/client/src/main/java/com/ibm/crail/utils/BufferCheckpoint.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/com/ibm/crail/utils/BufferCheckpoint.java b/client/src/main/java/com/ibm/crail/utils/BufferCheckpoint.java
deleted file mode 100644
index 4be142e..0000000
--- a/client/src/main/java/com/ibm/crail/utils/BufferCheckpoint.java
+++ /dev/null
@@ -1,55 +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.utils;
-
-import java.util.concurrent.ConcurrentHashMap;
-
-import org.slf4j.Logger;
-
-import com.ibm.crail.CrailBuffer;
-
-public class BufferCheckpoint {
- private static final Logger LOG = CrailUtils.getLogger();
- private ConcurrentHashMap<Long, CrailBuffer> checkpoint;
-
- public BufferCheckpoint(){
- this.checkpoint = new ConcurrentHashMap<Long, CrailBuffer>();
- }
-
- public void checkIn(CrailBuffer buffer) throws Exception {
- long address = buffer.address();
- CrailBuffer oldBuf = checkpoint.putIfAbsent(address, buffer);
- if (oldBuf != null){
- LOG.info("ERROR Buffer already in use!");
- StackTraceElement[] stack = Thread.currentThread().getStackTrace();
- for (StackTraceElement element : stack){
- LOG.info(element.toString());
- }
- throw new Exception("ERROR Buffer already in use!");
- }
- }
-
- public void checkOut(CrailBuffer buffer){
- long address = buffer.address();
- checkpoint.remove(address);
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/client/src/main/java/com/ibm/crail/utils/CrailImmediateOperation.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/com/ibm/crail/utils/CrailImmediateOperation.java b/client/src/main/java/com/ibm/crail/utils/CrailImmediateOperation.java
deleted file mode 100644
index 16dcbcd..0000000
--- a/client/src/main/java/com/ibm/crail/utils/CrailImmediateOperation.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.utils;
-
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.Future;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
-
-import com.ibm.crail.CrailResult;
-
-public class CrailImmediateOperation implements Future<CrailResult>, CrailResult{
- private long len;
-
- public CrailImmediateOperation(int len){
- this.len = len;
- }
-
- @Override
- public boolean cancel(boolean mayInterruptIfRunning) {
- return false;
- }
-
- @Override
- public boolean isCancelled() {
- return false;
- }
-
- @Override
- public boolean isDone() {
- return true;
- }
-
- @Override
- public CrailResult get() throws InterruptedException, ExecutionException {
- return this;
- }
-
- @Override
- public CrailResult get(long timeout, TimeUnit unit)
- throws InterruptedException, ExecutionException, TimeoutException {
- return this;
- }
-
- @Override
- public long getLen() {
- return len;
- }
-}
-
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/client/src/main/java/com/ibm/crail/utils/CrailUtils.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/com/ibm/crail/utils/CrailUtils.java b/client/src/main/java/com/ibm/crail/utils/CrailUtils.java
deleted file mode 100644
index bf77510..0000000
--- a/client/src/main/java/com/ibm/crail/utils/CrailUtils.java
+++ /dev/null
@@ -1,281 +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.utils;
-
-import java.net.InetAddress;
-import java.net.InetSocketAddress;
-import java.net.NetworkInterface;
-import java.net.SocketException;
-import java.net.URI;
-import java.net.UnknownHostException;
-import java.nio.ByteBuffer;
-import java.util.StringTokenizer;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentLinkedQueue;
-import java.util.concurrent.LinkedBlockingQueue;
-
-import org.slf4j.LoggerFactory;
-import org.slf4j.Logger;
-
-import com.ibm.crail.CrailLocationClass;
-import com.ibm.crail.conf.CrailConstants;
-import com.ibm.crail.metadata.DataNodeInfo;
-
-
-public class CrailUtils {
- private static Logger LOG = CrailUtils.getLogger();
-
- public static synchronized Logger getLogger(){
- if (LOG == null){
- LOG = LoggerFactory.getLogger("com.ibm.crail");
- }
- return LOG;
- }
-
- public static long getAddress(ByteBuffer buffer) {
- return ((sun.nio.ch.DirectBuffer) buffer).address();
- }
-
- public static InetSocketAddress getNameNodeAddress() {
- StringTokenizer tupleTokenizer = new StringTokenizer(CrailConstants.NAMENODE_ADDRESS, ",");
- LinkedBlockingQueue<URI> namenodes = new LinkedBlockingQueue<URI>();
- while(tupleTokenizer.hasMoreTokens()){
- String address = tupleTokenizer.nextToken();
- URI uri = URI.create(address);
- namenodes.add(uri);
- }
-
- URI master = namenodes.poll();
- InetSocketAddress nnAddr = createSocketAddrForHost(master.getHost(), master.getPort());
- return nnAddr;
- }
-
- public static URI getPrimaryNameNode() {
- StringTokenizer tupleTokenizer = new StringTokenizer(CrailConstants.NAMENODE_ADDRESS, ",");
- LinkedBlockingQueue<URI> namenodes = new LinkedBlockingQueue<URI>();
- while(tupleTokenizer.hasMoreTokens()){
- String address = tupleTokenizer.nextToken();
- URI uri = URI.create(address);
- namenodes.add(uri);
- }
-
- URI master = namenodes.poll();
- return master;
- }
-
- public static boolean verifyNamenode(String namenode) {
- StringTokenizer tupleTokenizer = new StringTokenizer(CrailConstants.NAMENODE_ADDRESS, ",");
- ConcurrentHashMap<String, Object> namenodes = new ConcurrentHashMap<String, Object>();
- while(tupleTokenizer.hasMoreTokens()){
- String address = tupleTokenizer.nextToken();
- URI uri = URI.create(address);
- String node = uri.getHost() + ":" + uri.getPort();
- namenodes.put(node, node);
- }
-
- URI uri = URI.create(namenode);
- String node = uri.getHost() + ":" + uri.getPort();
- return namenodes.containsKey(node);
- }
-
- public static ConcurrentLinkedQueue<InetSocketAddress> getNameNodeList() {
- StringTokenizer tupleTokenizer = new StringTokenizer(CrailConstants.NAMENODE_ADDRESS, ",");
- ConcurrentLinkedQueue<InetSocketAddress> namenodes = new ConcurrentLinkedQueue<InetSocketAddress>();
- while(tupleTokenizer.hasMoreTokens()){
- String token = tupleTokenizer.nextToken();
- URI uri = URI.create(token);
- InetSocketAddress address = createSocketAddrForHost(uri.getHost(), uri.getPort());
- namenodes.add(address);
- }
- return namenodes;
- }
-
- public static long getServiceId(String namenode) {
- StringTokenizer tupleTokenizer = new StringTokenizer(CrailConstants.NAMENODE_ADDRESS, ",");
- ConcurrentHashMap<String, Long> namenodes = new ConcurrentHashMap<String, Long>();
- long serviceId = 0;
- while(tupleTokenizer.hasMoreTokens()){
- String address = tupleTokenizer.nextToken();
- URI uri = URI.create(address);
- String node = uri.getHost() + ":" + uri.getPort();
- namenodes.put(node, serviceId++);
- }
-
- URI uri = URI.create(namenode);
- String node = uri.getHost() + ":" + uri.getPort();
- long id = namenodes.get(node);
- return id;
- }
-
- public static long getServiceSize() {
- StringTokenizer tupleTokenizer = new StringTokenizer(CrailConstants.NAMENODE_ADDRESS, ",");
- return tupleTokenizer.countTokens();
- }
-
- public static final long blockStartAddress(long offset) {
- long blockCount = offset / CrailConstants.BLOCK_SIZE;
- return blockCount*CrailConstants.BLOCK_SIZE;
- }
-
- public static final long bufferStartAddress(long position, long sliceSize) {
- long blockCount = position / sliceSize;
- return blockCount*sliceSize;
- }
-
- public static final long nextBlockAddress(long offset){
- if ((offset % CrailConstants.BLOCK_SIZE) == 0){
- return offset;
- } else {
- return blockStartAddress(offset) + CrailConstants.BLOCK_SIZE;
- }
- }
-
- public static final int minFileBuf(long fileSize, int bufSize) {
- int fileLeftOver = Integer.MAX_VALUE;
-
- long _maxInt = (long) Integer.MAX_VALUE;
- if (fileSize < _maxInt){
- fileLeftOver = (int) fileSize;
- }
-
- return Math.min(bufSize, fileLeftOver);
- }
-
- public static String getCacheDirectory(String id){
- return CrailConstants.CACHE_PATH + "/" + id;
- }
-
- public static void printStackTrace(){
- StackTraceElement[] elements = Thread.currentThread().getStackTrace();
- for (StackTraceElement e : elements) {
- LOG.info(e.toString());
- }
- }
-
- public static String getName(String path){
- if (path.equalsIgnoreCase("/")){
- return "";
- }
- if (path.endsWith("/")){
- int termSlash = path.lastIndexOf('/');
- path = path.substring(0, termSlash);
- }
-
- int lastSlash = path.lastIndexOf('/');
- if (lastSlash == -1) {
- return path;
- } else if (lastSlash == 0) {
- String name = path.substring(1);
- return name;
- } else {
- String name = path.substring(lastSlash+1);
- return name;
- }
- }
-
- public static String getParent(String path){
- if (path.equalsIgnoreCase("/")){
- return null;
- }
- if (path.endsWith("/")){
- int termSlash = path.lastIndexOf('/');
- path = path.substring(0, termSlash);
- }
-
- int lastSlash = path.lastIndexOf('/');
- if (lastSlash == -1) {
- return path;
- } else if (lastSlash == 0) {
- return "/";
- } else {
- String parent = path.substring(0, lastSlash);
- return parent;
- }
- }
-
- public static String combinePath(String parent, String name){
- if (parent.endsWith("/")){
- return parent + name;
- } else {
- return parent + "/" + name;
- }
- }
-
- public static int computeIndex(long offset) {
- long index = offset / CrailConstants.BLOCK_SIZE;
- return (int) index;
- }
-
- public static boolean isLocalAddress(InetAddress addr) {
- // Check if the address is a valid special local or loop back
- if (addr.isAnyLocalAddress() || addr.isLoopbackAddress())
- return true;
-
- // Check if the address is defined on any interface
- try {
- return NetworkInterface.getByInetAddress(addr) != null;
- } catch (SocketException e) {
- return false;
- }
- }
-
- public static InetSocketAddress createSocketAddrForHost(String host, int port) {
- return new InetSocketAddress(host, port);
- }
-
- public static InetSocketAddress datanodeInfo2SocketAddr(DataNodeInfo dnInfo) throws UnknownHostException{
- return new InetSocketAddress(InetAddress.getByAddress(dnInfo.getIpAddress()), dnInfo.getPort());
- }
-
- public static CrailLocationClass getLocationClass() throws UnknownHostException{
- return CrailLocationClass.get(InetAddress.getLocalHost().getCanonicalHostName().hashCode());
- }
-
- public static void parseMap(String config, ConcurrentHashMap<String, String> map) throws Exception {
- StringTokenizer tupleTokenizer = new StringTokenizer(config, "/");
- while(tupleTokenizer.hasMoreTokens()){
- String tuple = tupleTokenizer.nextToken();
- StringTokenizer commaTokenizer = new StringTokenizer(tuple, ",");
- if (commaTokenizer.countTokens() != 2){
- throw new Exception("parsing Map, wrong format!");
- }
- String key = commaTokenizer.nextToken();
- String value = commaTokenizer.nextToken();
- map.put(key, value);
- }
- }
-
- public static int getStorageClasses(String storageTypes) {
- StringTokenizer tokenizer = new StringTokenizer(storageTypes, ",");
- return tokenizer.countTokens();
- }
-
- public static String getIPAddressFromBytes(byte[] bytes){
- String address = "/unresolved";
- try {
- address = InetAddress.getByAddress(bytes).toString();
- } catch(Exception e){
- }
- return address;
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/client/src/main/java/com/ibm/crail/utils/EndpointCache.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/com/ibm/crail/utils/EndpointCache.java b/client/src/main/java/com/ibm/crail/utils/EndpointCache.java
deleted file mode 100644
index ec7d695..0000000
--- a/client/src/main/java/com/ibm/crail/utils/EndpointCache.java
+++ /dev/null
@@ -1,170 +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.utils;
-
-import java.io.IOException;
-import java.util.LinkedList;
-import java.util.concurrent.ConcurrentHashMap;
-
-import org.slf4j.Logger;
-
-import com.ibm.crail.CrailStatistics.StatisticsProvider;
-import com.ibm.crail.conf.CrailConstants;
-import com.ibm.crail.metadata.DataNodeInfo;
-import com.ibm.crail.storage.StorageClient;
-import com.ibm.crail.storage.StorageEndpoint;
-import com.ibm.crail.*;
-
-public class EndpointCache implements CrailStatistics.StatisticsProvider {
- private static final Logger LOG = CrailUtils.getLogger();
-
- private boolean isOpen;
- private ConcurrentHashMap<Integer, StorageEndpointCache> storageCaches;
-
- public EndpointCache(int fsId, LinkedList<StorageClient> storageClients){
- this.storageCaches = new ConcurrentHashMap<Integer, StorageEndpointCache>();
- int storageType = 0;
- for (StorageClient storageClient : storageClients){
- StorageEndpointCache cache = new StorageEndpointCache(fsId, storageClient, storageType);
- storageCaches.put(storageType++, cache);
- }
- this.isOpen = true;
- }
-
- @Override
- public String providerName() {
- return "cache/endpoint";
- }
-
- @Override
- public String printStatistics() {
- return "size " + size();
- }
-
- public void mergeStatistics(StatisticsProvider provider){
-
- }
-
- @Override
- public void resetStatistics() {
- }
-
- public StorageEndpoint getDataEndpoint(DataNodeInfo dataNodeInfo) throws IOException, InterruptedException {
- StorageEndpointCache cache = storageCaches.get(dataNodeInfo.getStorageType());
- StorageEndpoint endpoint = cache.getDataEndpoint(dataNodeInfo);
- return endpoint;
-// return storageCaches.get(dataNodeInfo.getStorageType()).getDataEndpoint(dataNodeInfo);
- }
-
- public int size() {
- int size = 0;
- for (StorageEndpointCache cache : storageCaches.values()){
- size += cache.size();
- }
- return size;
- }
-
- public void close() throws IOException{
- if (!isOpen){
- return;
- }
-
- for (StorageEndpointCache cache : storageCaches.values()){
- cache.close();
- }
- }
-
- //-------------------------------
-
- public static class StorageEndpointCache {
- private StorageClient storageClient;
- private ConcurrentHashMap<Long, Object> locktable;
- private ConcurrentHashMap<Long, StorageEndpoint> cache;
- private int fsId;
- private boolean isOpen;
- private int storageType;
-
- public StorageEndpointCache(int fsId, StorageClient storageClient, int storageType){
- this.fsId = fsId;
- this.storageClient = storageClient;
- this.cache = new ConcurrentHashMap<Long, StorageEndpoint>();
- this.locktable = new ConcurrentHashMap<Long, Object>();
- this.isOpen = true;
- this.storageType = storageType;
- }
-
- public void close() throws IOException {
- if (!isOpen){
- return;
- }
-
- try {
- storageClient.close();
- } catch(Exception e){
- throw new IOException(e);
- }
- }
-
- public StorageEndpoint getDataEndpoint(DataNodeInfo dataNodeInfo) throws IOException, InterruptedException {
- StorageEndpoint endpoint = cache.get(dataNodeInfo.key());
- if (endpoint == null) {
- Object lock = getLock(dataNodeInfo.key());
- synchronized (lock) {
- endpoint = cache.get(dataNodeInfo.key());
- if (endpoint == null){
- endpoint = storageClient.createEndpoint(dataNodeInfo);
- cache.put(dataNodeInfo.key(), endpoint);
- if (CrailConstants.DEBUG) {
- LOG.info("EndpointCache miss " + CrailUtils.datanodeInfo2SocketAddr(dataNodeInfo) + ", fsId " + fsId + ", cache size " + cache.size());
- }
- }
- }
- } else {
- if (CrailConstants.DEBUG) {
- LOG.info("EndpointCache hit " + CrailUtils.datanodeInfo2SocketAddr(dataNodeInfo) + ", fsId " + fsId);
- }
- }
- return endpoint;
- }
-
- public int size() {
- return cache.size();
- }
-
- public int getStorageType(){
- return this.storageType;
- }
-
- private Object getLock(long key){
- Object lock = locktable.get(key);
- if (lock == null){
- lock = new Object();
- Object oldLock = locktable.putIfAbsent(key, lock);
- if (oldLock != null){
- lock = oldLock;
- }
-
- }
- return lock;
- }
- }
-}
\ No newline at end of file