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:20:54 UTC
[10/29] incubator-crail git commit: Apache package renaming
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/hdfs/src/main/java/org/apache/crail/hdfs/CrailHDFSInputStream.java
----------------------------------------------------------------------
diff --git a/hdfs/src/main/java/org/apache/crail/hdfs/CrailHDFSInputStream.java b/hdfs/src/main/java/org/apache/crail/hdfs/CrailHDFSInputStream.java
new file mode 100644
index 0000000..eaced94
--- /dev/null
+++ b/hdfs/src/main/java/org/apache/crail/hdfs/CrailHDFSInputStream.java
@@ -0,0 +1,196 @@
+/*
+ * 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 org.apache.crail.hdfs;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+
+import org.apache.crail.CrailBufferedInputStream;
+import org.apache.crail.utils.CrailUtils;
+import org.apache.hadoop.fs.ByteBufferReadable;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.PositionedReadable;
+import org.apache.hadoop.fs.Seekable;
+import org.slf4j.Logger;
+
+
+
+public class CrailHDFSInputStream extends FSDataInputStream {
+ private static final Logger LOG = CrailUtils.getLogger();
+
+ private CrailBufferedInputStream inputStream;
+
+ public CrailHDFSInputStream(CrailBufferedInputStream stream) {
+ super(new CrailSeekable(stream));
+ LOG.info("new HDFS stream");
+ this.inputStream = stream;
+ }
+
+ @Override
+ public long getPos() throws IOException {
+ return inputStream.position();
+ }
+
+ @Override
+ public int read(ByteBuffer buf) throws IOException {
+ return inputStream.read(buf);
+ }
+
+ @Override
+ public int read(long position, byte[] buffer, int offset, int length)
+ throws IOException {
+ return inputStream.read(position, buffer, offset, length);
+ }
+
+ @Override
+ public void readFully(long position, byte[] buffer) throws IOException {
+ readFully(position, buffer, 0, buffer.length);
+ }
+
+ @Override
+ public void readFully(long position, byte[] buffer, int offset, int length)
+ throws IOException {
+ int nread = 0;
+ while (nread < length) {
+ int nbytes = read(position + nread, buffer, offset + nread, length - nread);
+ if (nbytes < 0) {
+ throw new java.io.EOFException("End of file reached before reading fully.");
+ }
+ nread += nbytes;
+ }
+ }
+
+ @Override
+ public boolean seekToNewSource(long targetPos) throws IOException {
+ return false;
+ }
+
+ @Override
+ public int read() throws IOException {
+ return inputStream.read();
+ }
+
+ @Override
+ public int available() throws IOException {
+ return inputStream.available();
+ }
+
+ public static class CrailSeekable extends InputStream implements Seekable, PositionedReadable, ByteBufferReadable {
+ private CrailBufferedInputStream inputStream;
+
+ public CrailSeekable(CrailBufferedInputStream inputStream) {
+ this.inputStream = inputStream;
+ }
+
+ @Override
+ public int read() throws IOException {
+ return inputStream.read();
+ }
+
+ @Override
+ public int read(byte[] b) throws IOException {
+ return inputStream.read(b);
+ }
+
+ @Override
+ public int read(byte[] b, int off, int len) throws IOException {
+ return inputStream.read(b, off, len);
+ }
+
+ @Override
+ public long skip(long n) throws IOException {
+ return inputStream.skip(n);
+ }
+
+ @Override
+ public int available() throws IOException {
+ return inputStream.available();
+ }
+
+ @Override
+ public void close() throws IOException {
+ inputStream.close();
+ }
+
+ @Override
+ public synchronized void mark(int readlimit) {
+ inputStream.mark(readlimit);
+ }
+
+ @Override
+ public synchronized void reset() throws IOException {
+ inputStream.reset();
+ }
+
+ @Override
+ public boolean markSupported() {
+ return inputStream.markSupported();
+ }
+
+ @Override
+ public int read(ByteBuffer dataBuf) throws IOException {
+ return inputStream.read(dataBuf);
+ }
+
+ @Override
+ public int read(long position, byte[] buffer, int offset, int length)
+ throws IOException {
+ return inputStream.read(position, buffer, offset, length);
+ }
+
+ @Override
+ public void readFully(long position, byte[] buf) throws IOException {
+ readFully(position, buf, 0, buf.length);
+ }
+
+ @Override
+ public void readFully(long position, byte[] buffer, int offset, int length)
+ throws IOException {
+ int nread = 0;
+ while (nread < length) {
+ int nbytes = read(position + nread, buffer, offset + nread, length - nread);
+ if (nbytes < 0) {
+ throw new java.io.EOFException("End of file reached before reading fully.");
+ }
+ nread += nbytes;
+ }
+ }
+
+ @Override
+ public long getPos() throws IOException {
+ return inputStream.position();
+ }
+
+ @Override
+ public void seek(long n) throws IOException {
+ inputStream.seek(n);
+ }
+
+ @Override
+ public boolean seekToNewSource(long targetPos) throws IOException {
+ return false;
+ }
+ }
+
+}
+
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/hdfs/src/main/java/org/apache/crail/hdfs/CrailHDFSOutputStream.java
----------------------------------------------------------------------
diff --git a/hdfs/src/main/java/org/apache/crail/hdfs/CrailHDFSOutputStream.java b/hdfs/src/main/java/org/apache/crail/hdfs/CrailHDFSOutputStream.java
new file mode 100644
index 0000000..72b161d
--- /dev/null
+++ b/hdfs/src/main/java/org/apache/crail/hdfs/CrailHDFSOutputStream.java
@@ -0,0 +1,35 @@
+/*
+ * 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 org.apache.crail.hdfs;
+
+import java.io.IOException;
+
+import org.apache.crail.CrailBufferedOutputStream;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileSystem.Statistics;
+
+public class CrailHDFSOutputStream extends FSDataOutputStream {
+ public CrailHDFSOutputStream(CrailBufferedOutputStream outputStream, Statistics stats) throws IOException {
+ super(outputStream, stats);
+ }
+}
+
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/hdfs/src/main/java/org/apache/crail/hdfs/CrailHadoopFileSystem.java
----------------------------------------------------------------------
diff --git a/hdfs/src/main/java/org/apache/crail/hdfs/CrailHadoopFileSystem.java b/hdfs/src/main/java/org/apache/crail/hdfs/CrailHadoopFileSystem.java
new file mode 100644
index 0000000..d8d647d
--- /dev/null
+++ b/hdfs/src/main/java/org/apache/crail/hdfs/CrailHadoopFileSystem.java
@@ -0,0 +1,308 @@
+/*
+ * 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 org.apache.crail.hdfs;
+
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.Iterator;
+
+import org.slf4j.Logger;
+import org.apache.crail.CrailBlockLocation;
+import org.apache.crail.CrailBufferedInputStream;
+import org.apache.crail.CrailBufferedOutputStream;
+import org.apache.crail.CrailDirectory;
+import org.apache.crail.CrailFS;
+import org.apache.crail.CrailFile;
+import org.apache.crail.CrailLocationClass;
+import org.apache.crail.CrailNode;
+import org.apache.crail.CrailNodeType;
+import org.apache.crail.CrailStorageClass;
+import org.apache.crail.conf.CrailConfiguration;
+import org.apache.crail.conf.CrailConstants;
+import org.apache.crail.rpc.RpcErrors;
+import org.apache.crail.utils.CrailUtils;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.BlockLocation;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.FsStatus;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.hadoop.util.Progressable;
+
+public class CrailHadoopFileSystem extends FileSystem {
+ private static final Logger LOG = CrailUtils.getLogger();
+ private CrailFS dfs;
+ private Path workingDir;
+ private URI uri;
+
+ public CrailHadoopFileSystem() throws IOException {
+ LOG.info("CrailHadoopFileSystem construction");
+ dfs = null;
+ }
+
+ @Override
+ public void initialize(URI uri, Configuration conf) throws IOException {
+ super.initialize(uri, conf);
+ setConf(conf);
+
+ try {
+ CrailConfiguration crailConf = new CrailConfiguration();
+ this.dfs = CrailFS.newInstance(crailConf);
+ Path _workingDir = new Path("/user/" + CrailConstants.USER);
+ this.workingDir = new Path("/user/" + CrailConstants.USER).makeQualified(uri, _workingDir);
+ this.uri = URI.create(CrailConstants.NAMENODE_ADDRESS);
+ LOG.info("CrailHadoopFileSystem fs initialization done..");
+ } catch(Exception e){
+ throw new IOException(e);
+ }
+ }
+
+ public String getScheme() {
+ return "crail";
+ }
+
+ public URI getUri() {
+ return uri;
+ }
+
+ public FSDataInputStream open(Path path, int bufferSize) throws IOException {
+ CrailFile fileInfo = null;
+ try {
+ fileInfo = dfs.lookup(path.toUri().getRawPath()).get().asFile();
+ CrailBufferedInputStream inputStream = fileInfo.getBufferedInputStream(fileInfo.getCapacity());
+ return new CrailHDFSInputStream(inputStream);
+ } catch (Exception e) {
+ throw new IOException(e);
+ }
+ }
+
+ @Override
+ public FSDataOutputStream create(Path path, FsPermission permission,
+ boolean overwrite, int bufferSize, short replication,
+ long blockSize, Progressable progress) throws IOException {
+ CrailFile fileInfo = null;
+ try {
+ fileInfo = dfs.create(path.toUri().getRawPath(), CrailNodeType.DATAFILE, CrailStorageClass.PARENT, CrailLocationClass.PARENT).get().asFile();
+ } catch (Exception e) {
+ if (e.getMessage().contains(RpcErrors.messages[RpcErrors.ERR_PARENT_MISSING])) {
+ fileInfo = null;
+ } else {
+ throw new IOException(e);
+ }
+ }
+
+ if (fileInfo == null) {
+ Path parent = path.getParent();
+ this.mkdirs(parent, FsPermission.getDirDefault());
+ try {
+ fileInfo = dfs.create(path.toUri().getRawPath(), CrailNodeType.DATAFILE, CrailStorageClass.PARENT, CrailLocationClass.PARENT).get().asFile();
+ } catch (Exception e) {
+ throw new IOException(e);
+ }
+ }
+
+ CrailBufferedOutputStream outputStream = null;
+ if (fileInfo != null){
+ try {
+ fileInfo.syncDir();
+ outputStream = fileInfo.getBufferedOutputStream(Integer.MAX_VALUE);
+ } catch (Exception e) {
+ throw new IOException(e);
+ }
+ }
+
+ if (outputStream != null){
+ return new CrailHDFSOutputStream(outputStream, statistics);
+ } else {
+ throw new IOException("Failed to create file, path " + path.toString());
+ }
+ }
+
+ @Override
+ public FSDataOutputStream append(Path path, int bufferSize, Progressable progress) throws IOException {
+ throw new IOException("Append not supported");
+ }
+
+ @Override
+ public boolean rename(Path src, Path dst) throws IOException {
+ try {
+ CrailNode file = dfs.rename(src.toUri().getRawPath(), dst.toUri().getRawPath()).get();
+ if (file != null){
+ file.syncDir();
+ }
+ return file != null;
+ } catch(Exception e){
+ throw new IOException(e);
+ }
+ }
+
+ @Override
+ public boolean delete(Path path, boolean recursive) throws IOException {
+ try {
+ CrailNode file = dfs.delete(path.toUri().getRawPath(), recursive).get();
+ if (file != null){
+ file.syncDir();
+ }
+ return file != null;
+ } catch(Exception e){
+ throw new IOException(e);
+ }
+ }
+
+ @Override
+ public FileStatus[] listStatus(Path path) throws FileNotFoundException, IOException {
+ try {
+ CrailNode node = dfs.lookup(path.toUri().getRawPath()).get();
+ Iterator<String> iter = node.asContainer().listEntries();
+ ArrayList<FileStatus> statusList = new ArrayList<FileStatus>();
+ while(iter.hasNext()){
+ String filepath = iter.next();
+ CrailNode directFile = dfs.lookup(filepath).get();
+ if (directFile != null){
+ FsPermission permission = FsPermission.getFileDefault();
+ if (directFile.getType().isDirectory()) {
+ permission = FsPermission.getDirDefault();
+ }
+ FileStatus status = new FileStatus(directFile.getCapacity(), directFile.getType().isContainer(), CrailConstants.SHADOW_REPLICATION, CrailConstants.BLOCK_SIZE, directFile.getModificationTime(), directFile.getModificationTime(), permission, CrailConstants.USER, CrailConstants.USER, new Path(filepath).makeQualified(this.getUri(), this.workingDir));
+ statusList.add(status);
+ }
+ }
+ FileStatus[] list = new FileStatus[statusList.size()];
+ statusList.toArray(list);
+ return list;
+ } catch(Exception e){
+ throw new FileNotFoundException(path.toUri().getRawPath());
+ }
+ }
+
+ @Override
+ public void setWorkingDirectory(Path new_dir) {
+ this.workingDir = new_dir;
+ }
+
+ @Override
+ public Path getWorkingDirectory() {
+ return this.workingDir;
+ }
+
+ @Override
+ public boolean mkdirs(Path path, FsPermission permission) throws IOException {
+ try {
+ CrailDirectory file = dfs.create(path.toUri().getRawPath(), CrailNodeType.DIRECTORY, CrailStorageClass.PARENT, CrailLocationClass.DEFAULT).get().asDirectory();
+ file.syncDir();
+ return true;
+ } catch(Exception e){
+ if (e.getMessage().contains(RpcErrors.messages[RpcErrors.ERR_PARENT_MISSING])){
+ Path parent = path.getParent();
+ mkdirs(parent);
+ return mkdirs(path);
+ } else if (e.getMessage().contains(RpcErrors.messages[RpcErrors.ERR_FILE_EXISTS])){
+ return true;
+ } else {
+ throw new IOException(e);
+ }
+ }
+ }
+
+ @Override
+ public FileStatus getFileStatus(Path path) throws IOException {
+ CrailNode directFile = null;
+ try {
+ directFile = dfs.lookup(path.toUri().getRawPath()).get();
+ } catch (Exception e) {
+ throw new IOException(e);
+ }
+ if (directFile == null) {
+ throw new FileNotFoundException("File does not exist: " + path);
+ }
+ FsPermission permission = FsPermission.getFileDefault();
+ if (directFile.getType().isDirectory()) {
+ permission = FsPermission.getDirDefault();
+ }
+ FileStatus status = new FileStatus(directFile.getCapacity(), directFile.getType().isContainer(), CrailConstants.SHADOW_REPLICATION, CrailConstants.BLOCK_SIZE, directFile.getModificationTime(), directFile.getModificationTime(), permission, CrailConstants.USER, CrailConstants.USER, path.makeQualified(this.getUri(), this.workingDir));
+ return status;
+ }
+
+ @Override
+ public BlockLocation[] getFileBlockLocations(FileStatus file, long start, long len) throws IOException {
+ try {
+ CrailBlockLocation[] _locations = dfs.lookup(file.getPath().toUri().getRawPath()).get().asFile().getBlockLocations(start, len);
+ BlockLocation[] locations = new BlockLocation[_locations.length];
+ for (int i = 0; i < locations.length; i++){
+ locations[i] = new BlockLocation();
+ locations[i].setOffset(_locations[i].getOffset());
+ locations[i].setLength(_locations[i].getLength());
+ locations[i].setNames(_locations[i].getNames());
+ locations[i].setHosts(_locations[i].getHosts());
+ locations[i].setTopologyPaths(_locations[i].getTopology());
+
+ }
+ return locations;
+ } catch(Exception e){
+ throw new IOException(e);
+ }
+ }
+
+ @Override
+ public BlockLocation[] getFileBlockLocations(Path path, long start, long len) throws IOException {
+ try {
+ CrailBlockLocation[] _locations = dfs.lookup(path.toUri().getRawPath()).get().asFile().getBlockLocations(start, len);
+ BlockLocation[] locations = new BlockLocation[_locations.length];
+ for (int i = 0; i < locations.length; i++){
+ locations[i] = new BlockLocation();
+ locations[i].setOffset(_locations[i].getOffset());
+ locations[i].setLength(_locations[i].getLength());
+ locations[i].setNames(_locations[i].getNames());
+ locations[i].setHosts(_locations[i].getHosts());
+ locations[i].setTopologyPaths(_locations[i].getTopology());
+
+ }
+ return locations;
+ } catch(Exception e){
+ throw new IOException(e);
+ }
+ }
+
+ @Override
+ public FsStatus getStatus(Path p) throws IOException {
+ statistics.incrementReadOps(1);
+ return new FsStatus(Long.MAX_VALUE, 0, Long.MAX_VALUE);
+ }
+
+ @Override
+ public void close() throws IOException {
+ try {
+ LOG.info("Closing CrailHadoopFileSystem");
+ super.processDeleteOnExit();
+ dfs.close();
+ } catch (Exception e) {
+ e.printStackTrace();
+ throw new IOException(e);
+ }
+ }
+}
+
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/hdfs/src/main/java/org/apache/crail/hdfs/GetConf.java
----------------------------------------------------------------------
diff --git a/hdfs/src/main/java/org/apache/crail/hdfs/GetConf.java b/hdfs/src/main/java/org/apache/crail/hdfs/GetConf.java
new file mode 100644
index 0000000..5091484
--- /dev/null
+++ b/hdfs/src/main/java/org/apache/crail/hdfs/GetConf.java
@@ -0,0 +1,46 @@
+/*
+ * 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 org.apache.crail.hdfs;
+
+import java.net.URI;
+import org.apache.hadoop.conf.Configuration;
+
+public class GetConf {
+
+ public static void main(String[] args) throws Exception {
+ if (args.length < 1){
+ System.out.println("Needs at least one argument");
+ }
+
+ String key = args[0];
+ Configuration conf = new Configuration();
+ if (key.equalsIgnoreCase("namenode")){
+ String defaultFS = conf.get("fs.defaultFS");
+ URI uri = new URI(defaultFS);
+ System.out.println(uri.getHost());
+ } else {
+ String value = conf.get(key);
+ System.out.println(value);
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/hdfs/src/main/java/org/apache/crail/hdfs/tools/HdfsIOBenchmark.java
----------------------------------------------------------------------
diff --git a/hdfs/src/main/java/org/apache/crail/hdfs/tools/HdfsIOBenchmark.java b/hdfs/src/main/java/org/apache/crail/hdfs/tools/HdfsIOBenchmark.java
new file mode 100644
index 0000000..d611b1d
--- /dev/null
+++ b/hdfs/src/main/java/org/apache/crail/hdfs/tools/HdfsIOBenchmark.java
@@ -0,0 +1,496 @@
+/*
+ * 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 org.apache.crail.hdfs.tools;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.Random;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.LocatedFileStatus;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.RemoteIterator;
+import org.apache.hadoop.conf.Configuration;
+
+public class HdfsIOBenchmark {
+ private String mode;
+ private int size;
+ private int loop;
+ private Path path;
+
+ public HdfsIOBenchmark(String[] args){
+ mode = args[0];
+ size = Integer.parseInt(args[1]);
+ loop = Integer.parseInt(args[2]);
+ path = new Path(args[3]);
+ }
+
+ public void run() throws Exception {
+ if (mode.equals("writeSequentialHeap")){
+ writeSequentialHeap();
+ } else if (mode.equalsIgnoreCase("readSequentialDirect")){
+ readSequentialDirect();
+ } else if (mode.equals("readSequentialHeap")){
+ readSequentialHeap();
+ } else if (mode.equals("readRandomDirect")){
+ readRandomDirect();
+ } else if (mode.equals("readRandomHeap")){
+ readRandomHeap();
+ } else if (mode.equals("getFile")){
+ getFile();
+ } else if (mode.equals("createFile")){
+ createFile();
+ } else if (mode.equals("enumerateDir")){
+ enumerateDir();
+ } else if (mode.equals("keyGet")){
+ keyGet();
+ } else if (mode.equals("browseDir")){
+ browseDir();
+ } else {
+ usage();
+ System.exit(0);
+ }
+ }
+
+ public static void usage(){
+ System.out.println("Usage:");
+ System.out.println("hdfsbench <readSequentialDirect|readSequentialHeap|readRandomDirect|readRandomHeap|writeSequentialHeap> <size> <iterations> <path>");
+ }
+
+ public void writeSequentialHeap() throws Exception {
+ System.out.println("writing sequential file in heap mode " + path);
+ Configuration conf = new Configuration();
+ FileSystem fs = FileSystem.get(conf);
+ FSDataOutputStream instream = fs.create(path);
+ byte[] buf = new byte[size];
+ double sumbytes = 0;
+ double ops = 0;
+ System.out.println("read size " + size);
+ System.out.println("operations " + loop);
+
+ long start = System.currentTimeMillis();
+ while (ops < loop) {
+ // System.out.println("writing data, len " + buf.length);
+ instream.write(buf, 0, buf.length);
+ sumbytes = sumbytes + buf.length;
+ ops = ops + 1.0;
+ }
+ instream.flush();
+ 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 / 1024.0 / 1024.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);
+ System.out.println("closing stream");
+ instream.close();
+ fs.close();
+ }
+
+ public void readSequentialDirect() throws Exception {
+ System.out.println("reading sequential file in direct mode " + path);
+ Configuration conf = new Configuration();
+ FileSystem fs = FileSystem.get(conf);
+ FileStatus status = fs.getFileStatus(path);
+ FSDataInputStream instream = fs.open(path);
+ ByteBuffer buf = ByteBuffer.allocateDirect(size);
+ buf.clear();
+ double sumbytes = 0;
+ double ops = 0;
+ System.out.println("file capacity " + status.getLen());
+ System.out.println("read size " + size);
+ System.out.println("operations " + loop);
+
+ long start = System.currentTimeMillis();
+ while (ops < loop) {
+ buf.clear();
+ double ret = (double) instream.read(buf);
+ if (ret > 0) {
+ sumbytes = sumbytes + ret;
+ ops = ops + 1.0;
+ } else {
+ ops = ops + 1.0;
+ if (instream.getPos() == 0){
+ break;
+ } else {
+ instream.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 / 1024.0 / 1024.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);
+ System.out.println("closing stream");
+ instream.close();
+ fs.close();
+ }
+
+ public void readSequentialHeap() throws Exception {
+ System.out.println("reading sequential file in heap mode " + path);
+ Configuration conf = new Configuration();
+ FileSystem fs = FileSystem.get(conf);
+ FileStatus status = fs.getFileStatus(path);
+ FSDataInputStream instream = fs.open(path);
+ byte[] buf = new byte[size];
+ double sumbytes = 0;
+ double ops = 0;
+ System.out.println("file capacity " + status.getLen());
+ System.out.println("read size " + size);
+ System.out.println("operations " + loop);
+
+ long start = System.currentTimeMillis();
+ while (ops < loop) {
+ double ret = (double) this.read(instream, buf);
+ if (ret > 0) {
+ sumbytes = sumbytes + ret;
+ ops = ops + 1.0;
+ } else {
+ ops = ops + 1.0;
+ if (instream.getPos() == 0){
+ break;
+ } else {
+ instream.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 / 1024.0 / 1024.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);
+ System.out.println("closing stream");
+ instream.close();
+ fs.close();
+ }
+
+ public void readRandomDirect() throws Exception{
+ System.out.println("reading random file in direct mode " + path);
+ Configuration conf = new Configuration();
+ FileSystem fs = FileSystem.get(conf);
+ FileStatus status = fs.getFileStatus(path);
+ FSDataInputStream instream = fs.open(path);
+ ByteBuffer buf = ByteBuffer.allocateDirect(size);
+ buf.clear();
+ double sumbytes = 0;
+ double ops = 0;
+ long _range = status.getLen()- ((long)buf.capacity());
+ double range = (double) _range;
+ Random random = new Random();
+
+ System.out.println("file capacity " + status.getLen());
+ System.out.println("read size " + size);
+ System.out.println("operations " + loop);
+ long start = System.currentTimeMillis();
+ while (ops < loop) {
+ buf.clear();
+ double _offset = range*random.nextDouble();
+ long offset = (long) _offset;
+ instream.seek(offset);
+ double ret = (double) instream.read(buf);
+ 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 / 1024.0 / 1024.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);
+ System.out.println("closing stream");
+ instream.close();
+ fs.close();
+ }
+
+ public void readRandomHeap() throws Exception{
+ System.out.println("reading random file in heap mode " + path);
+ Configuration conf = new Configuration();
+ FileSystem fs = FileSystem.get(conf);
+ FileStatus status = fs.getFileStatus(path);
+ FSDataInputStream instream = fs.open(path);
+ byte[] buf = new byte[size];
+ double sumbytes = 0;
+ double ops = 0;
+ long _range = status.getLen()- ((long)buf.length);
+ double range = (double) _range;
+ Random random = new Random();
+
+ System.out.println("file capacity " + status.getLen());
+ System.out.println("read size " + size);
+ System.out.println("operations " + loop);
+ long start = System.currentTimeMillis();
+ while (ops < loop) {
+ double _offset = range*random.nextDouble();
+ long offset = (long) _offset;
+ instream.seek(offset);
+ double ret = (double) this.read(instream, buf);
+ 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 / 1024.0 / 1024.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);
+ System.out.println("closing stream");
+ instream.close();
+ fs.close();
+ }
+
+ void getFile() throws Exception, InterruptedException {
+ System.out.println("get file, path " + path + ", outstanding " + size + ", loop " + loop);
+ Configuration conf = new Configuration();
+ FileSystem fs = FileSystem.get(conf);
+
+ Path paths[] = new Path[loop];
+ for (int j = 0; j < loop; j++){
+ paths[j] = new Path(path.toString() + "/" + j);
+ }
+ int repfactor = 4;
+ for (int k = 0; k < repfactor; k++){
+ long start = System.currentTimeMillis();
+ for (int i = 0; i < size; i++){
+ //single operation == loop
+ for (int j = 0; j < loop; j++){
+ fs.listStatus(paths[j]);
+ }
+ }
+ long end = System.currentTimeMillis();
+ double executionTime = ((double) (end - start));
+ double latency = executionTime*1000.0 / ((double) size);
+ System.out.println("execution time [ms] " + executionTime);
+ System.out.println("latency [us] " + latency);
+ }
+ fs.close();
+ }
+
+ void createFile() throws Exception, InterruptedException {
+ System.out.println("create file async hdfs, path " + path + ", size " + size + ", loop " + loop);
+ Configuration conf = new Configuration();
+ FileSystem fs = FileSystem.get(conf);
+
+ int repfactor = 4;
+ for (int k = 0; k < repfactor; k++){
+ LinkedBlockingQueue<Path> pathQueue = new LinkedBlockingQueue<Path>();
+ fs.mkdirs(path);
+ for (int i = 0; i < loop*size; i++){
+ String name = "" + i;
+ Path f = new Path(path, name);
+ pathQueue.add(f);
+ }
+
+ LinkedBlockingQueue<FSDataOutputStream> streamQueue = new LinkedBlockingQueue<FSDataOutputStream>();
+ long start = System.currentTimeMillis();
+ for (int i = 0; i < size; i++){
+ //single operation == loop
+ for (int j = 0; j < loop; j++) {
+ Path path = pathQueue.poll();
+ fs.create(path).close();
+ }
+ }
+ long end = System.currentTimeMillis();
+ double executionTime = ((double) (end - start));
+ double latency = executionTime*1000.0 / ((double) size);
+ System.out.println("execution time [ms] " + executionTime);
+ System.out.println("latency [us] " + latency);
+
+ while(!streamQueue.isEmpty()){
+ FSDataOutputStream stream = streamQueue.poll();
+ stream.close();
+ }
+
+ if (k < repfactor - 1){
+ fs.delete(path, true);
+ Thread.sleep(2000);
+ }
+ }
+ fs.close();
+ }
+
+ void enumerateDir() throws Exception {
+ System.out.println("enumarate dir, path " + path);
+ Configuration conf = new Configuration();
+ FileSystem fs = FileSystem.get(conf);
+
+ int repfactor = 4;
+ for (int k = 0; k < repfactor; k++) {
+ long start = System.currentTimeMillis();
+ for (int i = 0; i < size; i++) {
+ // single operation == loop
+ RemoteIterator<LocatedFileStatus> iter = fs.listFiles(path, false);
+ while (iter.hasNext()) {
+ iter.next();
+ }
+ }
+ long end = System.currentTimeMillis();
+ double executionTime = ((double) (end - start));
+ double latency = executionTime * 1000.0 / ((double) size);
+ System.out.println("execution time [ms] " + executionTime);
+ System.out.println("latency [us] " + latency);
+ }
+ fs.close();
+ }
+
+
+ void keyGet() throws Exception {
+ System.out.println("key get, path " + path + ", size " + size + ", loop " + loop);
+ Configuration conf = new Configuration();
+ FileSystem fs = FileSystem.get(conf);
+
+ Path[] paths = new Path[loop];
+ for (int i = 0; i < loop; i++){
+ String child = "" + i;
+ paths[i] = new Path(path, child);
+ System.out.println("path " + paths[i]);
+ }
+
+ byte[] outBuf = new byte[size];
+ for (Path p : paths){
+ FSDataOutputStream outputStream = fs.create(p);
+ outputStream.write(outBuf);
+ outputStream.close();
+ }
+
+ long start = System.currentTimeMillis();
+ ByteBuffer inBuf = ByteBuffer.allocateDirect(size);
+ for (int i = 0; i < loop; i++){
+ Path p = paths[i];
+ FSDataInputStream inputStream = fs.open(p);
+ inBuf.clear();
+ while(inBuf.remaining() > 0){
+ inputStream.read(inBuf);
+ }
+ inputStream.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.close();
+ }
+
+ private int read(FSDataInputStream stream, byte[] buf) throws IOException{
+ int off = 0;
+ int len = buf.length;
+ int ret = stream.read(buf, off, len);
+ while(ret > 0 && len - ret > 0){
+ len -= ret;
+ off += ret;
+ ret = stream.read(buf, off, len);
+ }
+ return off > 0 || ret > 0 ? ret : -1;
+ }
+
+ void browseDir() throws Exception {
+ System.out.println("reading enumarate dir, path " + path);
+ Configuration conf = new Configuration();
+ FileSystem fs = FileSystem.get(conf);
+
+ //benchmark
+ System.out.println("starting benchmark...");
+ RemoteIterator<LocatedFileStatus> iter = fs.listFiles(path, false);
+ while (iter.hasNext()) {
+ LocatedFileStatus status = iter.next();
+ System.out.println(status.getPath());
+ }
+ fs.close();
+ }
+
+ /**
+ * @param args
+ */
+ public static void main(String[] args) {
+ if (args.length != 4){
+ usage();
+ System.exit(0);
+ }
+
+ try {
+ HdfsIOBenchmark benchmark = new HdfsIOBenchmark(args);
+ benchmark.run();
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/hdfs/src/test/java/com/ibm/crail/hdfs/TestCrailHDFSContract.java
----------------------------------------------------------------------
diff --git a/hdfs/src/test/java/com/ibm/crail/hdfs/TestCrailHDFSContract.java b/hdfs/src/test/java/com/ibm/crail/hdfs/TestCrailHDFSContract.java
deleted file mode 100644
index ed96e40..0000000
--- a/hdfs/src/test/java/com/ibm/crail/hdfs/TestCrailHDFSContract.java
+++ /dev/null
@@ -1,71 +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.hdfs;
-
-import java.net.URI;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.FileSystemContractBaseTest;
-import com.ibm.crail.hdfs.CrailHadoopFileSystem;
-
-public class TestCrailHDFSContract extends FileSystemContractBaseTest {
-
- @Override
- protected void setUp() throws Exception {
- Configuration conf = new Configuration();
- fs = new CrailHadoopFileSystem();
- fs.initialize(URI.create(conf.get("fs.defaultFS")), conf);
- }
-
- // --------------------
-
- protected final static String TEST_UMASK = "062";
- protected FileSystem fs;
- protected byte[] data = dataset(getBlockSize() * 2, 0, 255);
-
- @Override
- protected void tearDown() throws Exception {
- fs.delete(path("/test"), true);
- fs.close();
- }
-
- /**
- * Create a dataset for use in the tests; all data is in the range base to
- * (base+modulo-1) inclusive
- *
- * @param len
- * length of data
- * @param base
- * base of the data
- * @param modulo
- * the modulo
- * @return the newly generated dataset
- */
- protected byte[] dataset(int len, int base, int modulo) {
- byte[] dataset = new byte[len];
- for (int i = 0; i < len; i++) {
- dataset[i] = (byte) (base + (i % modulo));
- }
- return dataset;
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/hdfs/src/test/java/org/apache/crail/hdfs/TestCrailHDFSContract.java
----------------------------------------------------------------------
diff --git a/hdfs/src/test/java/org/apache/crail/hdfs/TestCrailHDFSContract.java b/hdfs/src/test/java/org/apache/crail/hdfs/TestCrailHDFSContract.java
new file mode 100644
index 0000000..2063a83
--- /dev/null
+++ b/hdfs/src/test/java/org/apache/crail/hdfs/TestCrailHDFSContract.java
@@ -0,0 +1,72 @@
+/*
+ * 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 org.apache.crail.hdfs;
+
+import java.net.URI;
+
+import org.apache.crail.hdfs.CrailHadoopFileSystem;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.FileSystemContractBaseTest;
+
+public class TestCrailHDFSContract extends FileSystemContractBaseTest {
+
+ @Override
+ protected void setUp() throws Exception {
+ Configuration conf = new Configuration();
+ fs = new CrailHadoopFileSystem();
+ fs.initialize(URI.create(conf.get("fs.defaultFS")), conf);
+ }
+
+ // --------------------
+
+ protected final static String TEST_UMASK = "062";
+ protected FileSystem fs;
+ protected byte[] data = dataset(getBlockSize() * 2, 0, 255);
+
+ @Override
+ protected void tearDown() throws Exception {
+ fs.delete(path("/test"), true);
+ fs.close();
+ }
+
+ /**
+ * Create a dataset for use in the tests; all data is in the range base to
+ * (base+modulo-1) inclusive
+ *
+ * @param len
+ * length of data
+ * @param base
+ * base of the data
+ * @param modulo
+ * the modulo
+ * @return the newly generated dataset
+ */
+ protected byte[] dataset(int len, int base, int modulo) {
+ byte[] dataset = new byte[len];
+ for (int i = 0; i < len; i++) {
+ dataset[i] = (byte) (base + (i % modulo));
+ }
+ return dataset;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/namenode/src/main/java/com/ibm/crail/namenode/AbstractNode.java
----------------------------------------------------------------------
diff --git a/namenode/src/main/java/com/ibm/crail/namenode/AbstractNode.java b/namenode/src/main/java/com/ibm/crail/namenode/AbstractNode.java
deleted file mode 100644
index d734796..0000000
--- a/namenode/src/main/java/com/ibm/crail/namenode/AbstractNode.java
+++ /dev/null
@@ -1,111 +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.namenode;
-
-import java.util.Queue;
-import java.util.concurrent.Delayed;
-import java.util.concurrent.TimeUnit;
-
-import com.ibm.crail.CrailNodeType;
-import com.ibm.crail.metadata.FileInfo;
-
-public abstract class AbstractNode extends FileInfo implements Delayed {
- private int fileComponent;
- private long delay;
- private int storageClass;
- private int locationClass;
-
- //children manipulation
- //adds or replaces a child, returns previous value or null if there was no mapping
- public abstract AbstractNode putChild(AbstractNode child) throws Exception;
- //get the child with the given component name, returns null if there is no mapping
- public abstract AbstractNode getChild(int component) throws Exception;
- //remove a child, returns previous value of existing or null otherwise
- public abstract AbstractNode removeChild(int component) throws Exception;
- //clear all the children (used by GC)
- public abstract void clearChildren(Queue<AbstractNode> queue) throws Exception;
-// public abstract AbstractNode updateParent() throws Exception;
-
- //block manipulation
- //adds a new block at a given index, returns true if succesful, false otherwise
- public abstract boolean addBlock(int index, NameNodeBlockInfo block) throws Exception;
- //get block at the given index, returns a valid block or null otherwise
- public abstract NameNodeBlockInfo getBlock(int index) throws Exception;
- //clear all the blocks (used by GC)
- public abstract void freeBlocks(BlockStore blockStore) throws Exception;
-
- public AbstractNode(long fd, int fileComponent, CrailNodeType type, int storageClass, int locationAffinity){
- super(fd, type);
-
- this.fileComponent = fileComponent;
- this.storageClass = storageClass;
- this.locationClass = locationAffinity;
- this.delay = System.currentTimeMillis();
- this.setModificationTime(System.currentTimeMillis());
- }
-
- void rename(int newFileComponent) throws Exception {
- this.fileComponent = newFileComponent;
- }
-
- public int getComponent() {
- return this.fileComponent;
- }
-
- public void dump(){
- System.out.println(this.toString());
- }
-
- @Override
- protected void setDirOffset(long dirOffset) {
- super.setDirOffset(dirOffset);
- }
-
- @Override
- public String toString() {
- return String.format("%08d\t%08d\t\t%08d\t\t%08d\t\t%08d", getFd(), fileComponent, getCapacity(), getType().getLabel(), getDirOffset());
- }
-
- @Override
- public long getDelay(TimeUnit unit) {
- long diff = delay - System.currentTimeMillis();
- long _delay = unit.convert(diff, TimeUnit.MILLISECONDS);
- return _delay;
- }
-
- public void setDelay(long delay) {
- this.delay = System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(delay);
- }
-
- @Override
- public int compareTo(Delayed o) {
- return 0;
- }
-
- public int getStorageClass() {
- return storageClass;
- }
-
- public int getLocationClass() {
- return locationClass;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/namenode/src/main/java/com/ibm/crail/namenode/BlockStore.java
----------------------------------------------------------------------
diff --git a/namenode/src/main/java/com/ibm/crail/namenode/BlockStore.java b/namenode/src/main/java/com/ibm/crail/namenode/BlockStore.java
deleted file mode 100644
index e677f43..0000000
--- a/namenode/src/main/java/com/ibm/crail/namenode/BlockStore.java
+++ /dev/null
@@ -1,276 +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.namenode;
-
-import java.net.UnknownHostException;
-import java.util.ArrayList;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ThreadLocalRandom;
-import java.util.concurrent.locks.ReentrantReadWriteLock;
-import org.slf4j.Logger;
-import com.ibm.crail.conf.CrailConstants;
-import com.ibm.crail.metadata.BlockInfo;
-import com.ibm.crail.metadata.DataNodeInfo;
-import com.ibm.crail.rpc.RpcErrors;
-import com.ibm.crail.utils.AtomicIntegerModulo;
-import com.ibm.crail.utils.CrailUtils;
-
-public class BlockStore {
- private static final Logger LOG = CrailUtils.getLogger();
-
- private StorageClass[] storageClasses;
-
- public BlockStore(){
- storageClasses = new StorageClass[CrailConstants.STORAGE_CLASSES];
- for (int i = 0; i < CrailConstants.STORAGE_CLASSES; i++){
- this.storageClasses[i] = new StorageClass(i);
- }
- }
-
- public short addBlock(NameNodeBlockInfo blockInfo) throws UnknownHostException {
- int storageClass = blockInfo.getDnInfo().getStorageClass();
- return storageClasses[storageClass].addBlock(blockInfo);
- }
-
- public boolean regionExists(BlockInfo region) {
- int storageClass = region.getDnInfo().getStorageClass();
- return storageClasses[storageClass].regionExists(region);
- }
-
- public short updateRegion(BlockInfo region) {
- int storageClass = region.getDnInfo().getStorageClass();
- return storageClasses[storageClass].updateRegion(region);
- }
-
- public NameNodeBlockInfo getBlock(int storageClass, int locationAffinity) throws InterruptedException {
- NameNodeBlockInfo block = null;
- if (storageClass > 0){
- if (storageClass < storageClasses.length){
- block = storageClasses[storageClass].getBlock(locationAffinity);
- } else {
- //TODO: warn if requested storage class is invalid
- }
- }
- if (block == null){
- for (int i = 0; i < storageClasses.length; i++){
- block = storageClasses[i].getBlock(locationAffinity);
- if (block != null){
- break;
- }
- }
- }
-
- return block;
- }
-
- public DataNodeBlocks getDataNode(DataNodeInfo dnInfo) {
- int storageClass = dnInfo.getStorageClass();
- return storageClasses[storageClass].getDataNode(dnInfo);
- }
-
-}
-
-class StorageClass {
- private static final Logger LOG = CrailUtils.getLogger();
-
- private int storageClass;
- private ConcurrentHashMap<Long, DataNodeBlocks> membership;
- private ConcurrentHashMap<Integer, DataNodeArray> affinitySets;
- private DataNodeArray anySet;
- private BlockSelection blockSelection;
-
- public StorageClass(int storageClass){
- this.storageClass = storageClass;
- this.membership = new ConcurrentHashMap<Long, DataNodeBlocks>();
- this.affinitySets = new ConcurrentHashMap<Integer, DataNodeArray>();
- if (CrailConstants.NAMENODE_BLOCKSELECTION.equalsIgnoreCase("roundrobin")){
- this.blockSelection = new RoundRobinBlockSelection();
- } else {
- this.blockSelection = new RandomBlockSelection();
- }
- this.anySet = new DataNodeArray(blockSelection);
- }
-
- public short updateRegion(BlockInfo region) {
- long dnAddress = region.getDnInfo().key();
- DataNodeBlocks current = membership.get(dnAddress);
- if (current == null) {
- return RpcErrors.ERR_ADD_BLOCK_FAILED;
- } else {
- return current.updateRegion(region);
- }
- }
-
- public boolean regionExists(BlockInfo region) {
- long dnAddress = region.getDnInfo().key();
- DataNodeBlocks current = membership.get(dnAddress);
- if (current == null) {
- return false;
- } else {
- return current.regionExists(region);
- }
- }
-
- short addBlock(NameNodeBlockInfo block) throws UnknownHostException {
- long dnAddress = block.getDnInfo().key();
- DataNodeBlocks current = membership.get(dnAddress);
- if (current == null) {
- current = DataNodeBlocks.fromDataNodeInfo(block.getDnInfo());
- addDataNode(current);
- }
-
- current.touch();
- current.addFreeBlock(block);
- return RpcErrors.ERR_OK;
- }
-
- NameNodeBlockInfo getBlock(int affinity) throws InterruptedException {
- NameNodeBlockInfo block = null;
- if (affinity == 0) {
- block = anySet.get();
- } else {
- block = _getAffinityBlock(affinity);
- if (block == null) {
- block = anySet.get();
- } else {
- }
- }
- return block;
- }
-
- DataNodeBlocks getDataNode(DataNodeInfo dataNode) {
- return membership.get(dataNode.key());
- }
-
- short addDataNode(DataNodeBlocks dataNode) {
- DataNodeBlocks current = membership.putIfAbsent(dataNode.key(), dataNode);
- if (current != null) {
- return RpcErrors.ERR_DATANODE_NOT_REGISTERED;
- }
-
- // current == null, datanode not in set, adding it now
- _addDataNode(dataNode);
-
- return RpcErrors.ERR_OK;
-
- }
-
- //---------------
-
- private void _addDataNode(DataNodeBlocks dataNode){
- LOG.info("adding datanode " + CrailUtils.getIPAddressFromBytes(dataNode.getIpAddress()) + ":" + dataNode.getPort() + " of type " + dataNode.getStorageType() + " to storage class " + storageClass);
- DataNodeArray hostMap = affinitySets.get(dataNode.getLocationClass());
- if (hostMap == null){
- hostMap = new DataNodeArray(blockSelection);
- DataNodeArray oldMap = affinitySets.putIfAbsent(dataNode.getLocationClass(), hostMap);
- if (oldMap != null){
- hostMap = oldMap;
- }
- }
- hostMap.add(dataNode);
- anySet.add(dataNode);
- }
-
- private NameNodeBlockInfo _getAffinityBlock(int affinity) throws InterruptedException {
- NameNodeBlockInfo block = null;
- DataNodeArray affinitySet = affinitySets.get(affinity);
- if (affinitySet != null){
- block = affinitySet.get();
- }
- return block;
- }
-
- public static interface BlockSelection {
- int getNext(int size);
- }
-
- private class RoundRobinBlockSelection implements BlockSelection {
- private AtomicIntegerModulo counter;
-
- public RoundRobinBlockSelection(){
- LOG.info("round robin block selection");
- counter = new AtomicIntegerModulo();
- }
-
- @Override
- public int getNext(int size) {
- return counter.getAndIncrement() % size;
- }
- }
-
- private class RandomBlockSelection implements BlockSelection {
- public RandomBlockSelection(){
- LOG.info("random block selection");
- }
-
- @Override
- public int getNext(int size) {
- return ThreadLocalRandom.current().nextInt(size);
- }
- }
-
- private class DataNodeArray {
- private ArrayList<DataNodeBlocks> arrayList;
- private ReentrantReadWriteLock lock;
- private BlockSelection blockSelection;
-
- public DataNodeArray(BlockSelection blockSelection){
- this.arrayList = new ArrayList<DataNodeBlocks>();
- this.lock = new ReentrantReadWriteLock();
- this.blockSelection = blockSelection;
- }
-
- public void add(DataNodeBlocks dataNode){
- lock.writeLock().lock();
- try {
- arrayList.add(dataNode);
- } finally {
- lock.writeLock().unlock();
- }
- }
-
- private NameNodeBlockInfo get() throws InterruptedException {
- lock.readLock().lock();
- try {
- NameNodeBlockInfo block = null;
- int size = arrayList.size();
- if (size > 0){
- int startIndex = blockSelection.getNext(size);
- for (int i = 0; i < size; i++){
- int index = (startIndex + i) % size;
- DataNodeBlocks anyDn = arrayList.get(index);
- if (anyDn.isOnline()){
- block = anyDn.getFreeBlock();
- }
- if (block != null){
- break;
- }
- }
- }
- return block;
- } finally {
- lock.readLock().unlock();
- }
- }
- }
-}
-
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/namenode/src/main/java/com/ibm/crail/namenode/DataNodeBlocks.java
----------------------------------------------------------------------
diff --git a/namenode/src/main/java/com/ibm/crail/namenode/DataNodeBlocks.java b/namenode/src/main/java/com/ibm/crail/namenode/DataNodeBlocks.java
deleted file mode 100644
index 391a61b..0000000
--- a/namenode/src/main/java/com/ibm/crail/namenode/DataNodeBlocks.java
+++ /dev/null
@@ -1,93 +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.namenode;
-
-import java.net.UnknownHostException;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.TimeUnit;
-
-import org.slf4j.Logger;
-
-import com.ibm.crail.conf.CrailConstants;
-import com.ibm.crail.metadata.BlockInfo;
-import com.ibm.crail.metadata.DataNodeInfo;
-import com.ibm.crail.rpc.RpcErrors;
-import com.ibm.crail.utils.CrailUtils;
-
-public class DataNodeBlocks extends DataNodeInfo {
- private static final Logger LOG = CrailUtils.getLogger();
-
- private ConcurrentHashMap<Long, BlockInfo> regions;
- private LinkedBlockingQueue<NameNodeBlockInfo> freeBlocks;
- private long token;
-
- public static DataNodeBlocks fromDataNodeInfo(DataNodeInfo dnInfo) throws UnknownHostException{
- DataNodeBlocks dnInfoNn = new DataNodeBlocks(dnInfo.getStorageType(), dnInfo.getStorageClass(), dnInfo.getLocationClass(), dnInfo.getIpAddress(), dnInfo.getPort());
- return dnInfoNn;
- }
-
- private DataNodeBlocks(int storageType, int getStorageClass, int locationClass, byte[] ipAddress, int port) throws UnknownHostException {
- super(storageType, getStorageClass, locationClass, ipAddress, port);
- this.regions = new ConcurrentHashMap<Long, BlockInfo>();
- this.freeBlocks = new LinkedBlockingQueue<NameNodeBlockInfo>();
- }
-
- public void addFreeBlock(NameNodeBlockInfo nnBlock) {
- regions.put(nnBlock.getRegion().getLba(), nnBlock.getRegion());
- freeBlocks.add(nnBlock);
- }
-
- public NameNodeBlockInfo getFreeBlock() throws InterruptedException {
- NameNodeBlockInfo block = this.freeBlocks.poll();
- return block;
- }
-
- public int getBlockCount() {
- return freeBlocks.size();
- }
-
- public boolean regionExists(BlockInfo region) {
- if (regions.containsKey(region.getLba())){
- return true;
- }
- return false;
- }
-
- public short updateRegion(BlockInfo region) {
- BlockInfo oldRegion = regions.get(region.getLba());
- if (oldRegion == null){
- return RpcErrors.ERR_ADD_BLOCK_FAILED;
- } else {
- oldRegion.setBlockInfo(region);
- return 0;
- }
- }
-
- public void touch() {
- this.token = System.nanoTime() + TimeUnit.SECONDS.toNanos(CrailConstants.STORAGE_KEEPALIVE*8);
- }
-
- public boolean isOnline(){
- return System.nanoTime() <= token;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/namenode/src/main/java/com/ibm/crail/namenode/DirectoryBlocks.java
----------------------------------------------------------------------
diff --git a/namenode/src/main/java/com/ibm/crail/namenode/DirectoryBlocks.java b/namenode/src/main/java/com/ibm/crail/namenode/DirectoryBlocks.java
deleted file mode 100644
index 25b35d4..0000000
--- a/namenode/src/main/java/com/ibm/crail/namenode/DirectoryBlocks.java
+++ /dev/null
@@ -1,107 +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.namenode;
-
-import java.util.Iterator;
-import java.util.Queue;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.atomic.AtomicLong;
-
-import com.ibm.crail.CrailNodeType;
-import com.ibm.crail.conf.CrailConstants;
-import com.ibm.crail.metadata.BlockInfo;
-
-public class DirectoryBlocks extends AbstractNode {
- private AtomicLong dirOffsetCounter;
- protected ConcurrentHashMap<Integer, AbstractNode> children;
- private ConcurrentHashMap<Integer, NameNodeBlockInfo> blocks;
-
- DirectoryBlocks(long fd, int fileComponent, CrailNodeType type, int storageClass, int locationClass) {
- super(fd, fileComponent, type, storageClass, locationClass);
- this.children = new ConcurrentHashMap<Integer, AbstractNode>();
- this.dirOffsetCounter = new AtomicLong(0);
- this.blocks = new ConcurrentHashMap<Integer, NameNodeBlockInfo>();
- }
-
- public AbstractNode putChild(AbstractNode child) throws Exception {
- AbstractNode old = children.putIfAbsent(child.getComponent(), child);
- if (old != null){
- throw new Exception("File exists");
- }
- child.setDirOffset(dirOffsetCounter.getAndAdd(CrailConstants.DIRECTORY_RECORD));
- return old;
- }
-
- public AbstractNode getChild(int component) {
- return children.get(component);
- }
-
- public AbstractNode removeChild(int component) {
- return children.remove(component);
- }
-
- @Override
- public NameNodeBlockInfo getBlock(int index) {
- return blocks.get(index);
- }
-
- @Override
- public boolean addBlock(int index, NameNodeBlockInfo block) {
- BlockInfo old = blocks.putIfAbsent(index, block);
- return old == null;
- }
-
- @Override
- public void freeBlocks(BlockStore blockStore) throws Exception {
- Iterator<NameNodeBlockInfo> iter = blocks.values().iterator();
- while (iter.hasNext()){
- NameNodeBlockInfo blockInfo = iter.next();
- blockStore.addBlock(blockInfo);
- }
- }
-
- @Override
- public long setCapacity(long newcapacity) {
- return this.getCapacity();
- }
-
- @Override
- public void updateToken() {
- }
-
- @Override
- public void clearChildren(Queue<AbstractNode> queue) {
- Iterator<AbstractNode> iter = children.values().iterator();
- while(iter.hasNext()){
- AbstractNode child = iter.next();
- queue.add(child);
- }
- }
-
- @Override
- public void dump() {
- super.dump();
- for (AbstractNode child : children.values()){
- child.dump();
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/namenode/src/main/java/com/ibm/crail/namenode/FileBlocks.java
----------------------------------------------------------------------
diff --git a/namenode/src/main/java/com/ibm/crail/namenode/FileBlocks.java b/namenode/src/main/java/com/ibm/crail/namenode/FileBlocks.java
deleted file mode 100644
index 9955913..0000000
--- a/namenode/src/main/java/com/ibm/crail/namenode/FileBlocks.java
+++ /dev/null
@@ -1,111 +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.namenode;
-
-import java.util.ArrayList;
-import java.util.Iterator;
-import java.util.Queue;
-import java.util.concurrent.locks.Lock;
-import java.util.concurrent.locks.ReentrantReadWriteLock;
-
-import com.ibm.crail.CrailNodeType;
-import com.ibm.crail.conf.CrailConstants;
-
-public class FileBlocks extends AbstractNode {
- private ArrayList<NameNodeBlockInfo> blocks;
- private final ReentrantReadWriteLock lock;
- private final Lock readLock;
- private final Lock writeLock;
-
- public FileBlocks(long fd, int fileComponent, CrailNodeType type, int storageClass, int locationClass) {
- super(fd, fileComponent, type, storageClass, locationClass);
- this.blocks = new ArrayList<NameNodeBlockInfo>(CrailConstants.NAMENODE_FILEBLOCKS);
- this.lock = new ReentrantReadWriteLock();
- this.readLock = lock.readLock();
- this.writeLock = lock.writeLock();
- }
-
- @Override
- public NameNodeBlockInfo getBlock(int index) {
- readLock.lock();
- try {
- if (index < blocks.size()){
- return blocks.get(index);
- } else {
- return null;
- }
- } catch(Exception e){
- return null;
- } finally {
- readLock.unlock();
- }
- }
-
- @Override
- public boolean addBlock(int index, NameNodeBlockInfo block) {
- writeLock.lock();
- try {
- if (index == blocks.size()){
- blocks.add(index, block);
- return true;
- } else {
- return false;
- }
- } finally {
- writeLock.unlock();
- }
- }
-
- @Override
- public void freeBlocks(BlockStore blockStore) throws Exception {
- readLock.lock();
- try {
- Iterator<NameNodeBlockInfo> iter = blocks.iterator();
- while (iter.hasNext()){
- NameNodeBlockInfo blockInfo = iter.next();
- blockStore.addBlock(blockInfo);
- }
- } finally {
- readLock.unlock();
- }
- }
-
- @Override
- public AbstractNode putChild(AbstractNode child) throws Exception {
- throw new Exception("Attempt to add a child to a non-container type");
- }
-
- @Override
- public AbstractNode getChild(int component) throws Exception {
- throw new Exception("Attempto to retrieve child from non-container type");
- }
-
- @Override
- public AbstractNode removeChild(int component) throws Exception {
- throw new Exception("Attempt to remove child from non-container type");
- }
-
- @Override
- public void clearChildren(Queue<AbstractNode> queue) throws Exception {
- throw new Exception("Attempt collect children from non-container type");
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/namenode/src/main/java/com/ibm/crail/namenode/FileStore.java
----------------------------------------------------------------------
diff --git a/namenode/src/main/java/com/ibm/crail/namenode/FileStore.java b/namenode/src/main/java/com/ibm/crail/namenode/FileStore.java
deleted file mode 100644
index 63f0497..0000000
--- a/namenode/src/main/java/com/ibm/crail/namenode/FileStore.java
+++ /dev/null
@@ -1,90 +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.namenode;
-
-import java.io.IOException;
-
-import com.ibm.crail.CrailNodeType;
-import com.ibm.crail.conf.CrailConstants;
-import com.ibm.crail.metadata.FileName;
-import com.ibm.crail.rpc.RpcErrors;
-import com.ibm.crail.rpc.RpcNameNodeState;
-
-public class FileStore {
- private Sequencer sequencer;
- private AbstractNode root;
-
- public FileStore(Sequencer sequencer) throws IOException {
- this.sequencer = sequencer;
- this.root = createNode(new FileName("/").getFileComponent(), CrailNodeType.DIRECTORY, CrailConstants.STORAGE_ROOTCLASS, 0);
- }
-
- public AbstractNode createNode(int fileComponent, CrailNodeType type, int storageClass, int locationClass) throws IOException {
- if (type == CrailNodeType.DIRECTORY){
- return new DirectoryBlocks(sequencer.getNextId(), fileComponent, type, storageClass, locationClass);
- } else if (type == CrailNodeType.MULTIFILE){
- return new MultiFileBlocks(sequencer.getNextId(), fileComponent, type, storageClass, locationClass);
- } else if (type == CrailNodeType.TABLE){
- return new TableBlocks(sequencer.getNextId(), fileComponent, type, storageClass, locationClass);
- } else if (type == CrailNodeType.KEYVALUE){
- return new KeyValueBlocks(sequencer.getNextId(), fileComponent, type, storageClass, locationClass);
- } else if (type == CrailNodeType.DATAFILE){
- return new FileBlocks(sequencer.getNextId(), fileComponent, type, storageClass, locationClass);
- } else {
- throw new IOException("File type unkown: " + type);
- }
- }
-
- public AbstractNode retrieveFile(FileName filename, RpcNameNodeState error) throws Exception{
- return retrieveFileInternal(filename, filename.getLength(), error);
- }
-
- public AbstractNode retrieveParent(FileName filename, RpcNameNodeState error) throws Exception{
- return retrieveFileInternal(filename, filename.getLength()-1, error);
- }
-
- public AbstractNode getRoot() {
- return root;
- }
-
- public void dump(){
- root.dump();
- }
-
- private AbstractNode retrieveFileInternal(FileName filename, int length, RpcNameNodeState error) throws Exception {
- if (length >= CrailConstants.DIRECTORY_DEPTH){
- error.setError(RpcErrors.ERR_FILE_COMPONENTS_EXCEEDED);
- return null;
- }
-
- AbstractNode current = root;
- for (int i = 0; i < length; i++){
- int component = filename.getComponent(i);
- current = current.getChild(component);
- if (current == null){
- break;
- }
- }
-
- return current;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/namenode/src/main/java/com/ibm/crail/namenode/GCServer.java
----------------------------------------------------------------------
diff --git a/namenode/src/main/java/com/ibm/crail/namenode/GCServer.java b/namenode/src/main/java/com/ibm/crail/namenode/GCServer.java
deleted file mode 100644
index 2987779..0000000
--- a/namenode/src/main/java/com/ibm/crail/namenode/GCServer.java
+++ /dev/null
@@ -1,56 +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.namenode;
-
-import java.util.concurrent.DelayQueue;
-
-import org.slf4j.Logger;
-
-import com.ibm.crail.utils.CrailUtils;
-
-public class GCServer implements Runnable {
- private static final Logger LOG = CrailUtils.getLogger();
-
- private NameNodeService rpcService;
- private DelayQueue<AbstractNode> deleteQueue;
-
- public GCServer(NameNodeService service, DelayQueue<AbstractNode> deleteQueue){
- this.rpcService = service;
- this.deleteQueue = deleteQueue;
- }
-
- @Override
- public void run() {
- while(true){
- try{
- AbstractNode file = deleteQueue.take();
- if (file.getType().isContainer()){
- file.clearChildren(deleteQueue);
- }
- rpcService.freeFile(file);
- } catch(Exception e){
- LOG.info("Exception during GC: " + e.getMessage());
- }
- }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/namenode/src/main/java/com/ibm/crail/namenode/KeyValueBlocks.java
----------------------------------------------------------------------
diff --git a/namenode/src/main/java/com/ibm/crail/namenode/KeyValueBlocks.java b/namenode/src/main/java/com/ibm/crail/namenode/KeyValueBlocks.java
deleted file mode 100644
index 21ab429..0000000
--- a/namenode/src/main/java/com/ibm/crail/namenode/KeyValueBlocks.java
+++ /dev/null
@@ -1,11 +0,0 @@
-package com.ibm.crail.namenode;
-
-import com.ibm.crail.CrailNodeType;
-
-public class KeyValueBlocks extends FileBlocks {
- public KeyValueBlocks(long fd, int fileComponent, CrailNodeType type,
- int storageClass, int locationClass) {
- super(fd, fileComponent, type, storageClass, locationClass);
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/namenode/src/main/java/com/ibm/crail/namenode/LogDispatcher.java
----------------------------------------------------------------------
diff --git a/namenode/src/main/java/com/ibm/crail/namenode/LogDispatcher.java b/namenode/src/main/java/com/ibm/crail/namenode/LogDispatcher.java
deleted file mode 100644
index d953b00..0000000
--- a/namenode/src/main/java/com/ibm/crail/namenode/LogDispatcher.java
+++ /dev/null
@@ -1,121 +0,0 @@
-package com.ibm.crail.namenode;
-
-import com.ibm.crail.rpc.RpcNameNodeService;
-import com.ibm.crail.rpc.RpcNameNodeState;
-import com.ibm.crail.rpc.RpcProtocol;
-import com.ibm.crail.rpc.RpcRequestMessage.CreateFileReq;
-import com.ibm.crail.rpc.RpcRequestMessage.DumpNameNodeReq;
-import com.ibm.crail.rpc.RpcRequestMessage.GetBlockReq;
-import com.ibm.crail.rpc.RpcRequestMessage.GetDataNodeReq;
-import com.ibm.crail.rpc.RpcRequestMessage.GetFileReq;
-import com.ibm.crail.rpc.RpcRequestMessage.GetLocationReq;
-import com.ibm.crail.rpc.RpcRequestMessage.PingNameNodeReq;
-import com.ibm.crail.rpc.RpcRequestMessage.RemoveFileReq;
-import com.ibm.crail.rpc.RpcRequestMessage.RenameFileReq;
-import com.ibm.crail.rpc.RpcRequestMessage.SetBlockReq;
-import com.ibm.crail.rpc.RpcRequestMessage.SetFileReq;
-import com.ibm.crail.rpc.RpcResponseMessage.CreateFileRes;
-import com.ibm.crail.rpc.RpcResponseMessage.DeleteFileRes;
-import com.ibm.crail.rpc.RpcResponseMessage.GetBlockRes;
-import com.ibm.crail.rpc.RpcResponseMessage.GetDataNodeRes;
-import com.ibm.crail.rpc.RpcResponseMessage.GetFileRes;
-import com.ibm.crail.rpc.RpcResponseMessage.GetLocationRes;
-import com.ibm.crail.rpc.RpcResponseMessage.PingNameNodeRes;
-import com.ibm.crail.rpc.RpcResponseMessage.RenameRes;
-import com.ibm.crail.rpc.RpcResponseMessage.VoidRes;
-
-public class LogDispatcher implements RpcNameNodeService {
- private RpcNameNodeService service;
- private LogService logService;
-
- public LogDispatcher(RpcNameNodeService service) throws Exception{
- this.service = service;
- this.logService = new LogService();
- this.logService.replay(service);
- }
-
- @Override
- public short createFile(CreateFileReq request, CreateFileRes response,
- RpcNameNodeState errorState) throws Exception {
- LogRecord record = new LogRecord(request);
- record.setCommand(RpcProtocol.CMD_CREATE_FILE);
- logService.writeRecord(record);
- return service.createFile(request, response, errorState);
- }
-
- @Override
- public short getFile(GetFileReq request, GetFileRes response,
- RpcNameNodeState errorState) throws Exception {
- return service.getFile(request, response, errorState);
- }
-
- @Override
- public short setFile(SetFileReq request, VoidRes response,
- RpcNameNodeState errorState) throws Exception {
- LogRecord record = new LogRecord(request);
- record.setCommand(RpcProtocol.CMD_SET_FILE);
- logService.writeRecord(record);
- return service.setFile(request, response, errorState);
- }
-
- @Override
- public short removeFile(RemoveFileReq request, DeleteFileRes response,
- RpcNameNodeState errorState) throws Exception {
- LogRecord record = new LogRecord(request);
- record.setCommand(RpcProtocol.CMD_REMOVE_FILE);
- logService.writeRecord(record);
- return service.removeFile(request, response, errorState);
- }
-
- @Override
- public short renameFile(RenameFileReq request, RenameRes response,
- RpcNameNodeState errorState) throws Exception {
- LogRecord record = new LogRecord(request);
- record.setCommand(RpcProtocol.CMD_RENAME_FILE);
- logService.writeRecord(record);
- return service.renameFile(request, response, errorState);
- }
-
- @Override
- public short getDataNode(GetDataNodeReq request, GetDataNodeRes response,
- RpcNameNodeState errorState) throws Exception {
- return service.getDataNode(request, response, errorState);
- }
-
- @Override
- public short setBlock(SetBlockReq request, VoidRes response,
- RpcNameNodeState errorState) throws Exception {
- LogRecord record = new LogRecord(request);
- record.setCommand(RpcProtocol.CMD_SET_BLOCK);
- logService.writeRecord(record);
- return service.setBlock(request, response, errorState);
- }
-
- @Override
- public short getBlock(GetBlockReq request, GetBlockRes response,
- RpcNameNodeState errorState) throws Exception {
- LogRecord record = new LogRecord(request);
- record.setCommand(RpcProtocol.CMD_GET_BLOCK);
- logService.writeRecord(record);
- return service.getBlock(request, response, errorState);
- }
-
- @Override
- public short getLocation(GetLocationReq request, GetLocationRes response,
- RpcNameNodeState errorState) throws Exception {
- return service.getLocation(request, response, errorState);
- }
-
- @Override
- public short dump(DumpNameNodeReq request, VoidRes response,
- RpcNameNodeState errorState) throws Exception {
- return service.dump(request, response, errorState);
- }
-
- @Override
- public short ping(PingNameNodeReq request, PingNameNodeRes response,
- RpcNameNodeState errorState) throws Exception {
- return service.ping(request, response, errorState);
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/namenode/src/main/java/com/ibm/crail/namenode/LogRecord.java
----------------------------------------------------------------------
diff --git a/namenode/src/main/java/com/ibm/crail/namenode/LogRecord.java b/namenode/src/main/java/com/ibm/crail/namenode/LogRecord.java
deleted file mode 100644
index 7e852a0..0000000
--- a/namenode/src/main/java/com/ibm/crail/namenode/LogRecord.java
+++ /dev/null
@@ -1,286 +0,0 @@
-package com.ibm.crail.namenode;
-
-import java.io.IOException;
-import java.nio.ByteBuffer;
-
-import org.slf4j.Logger;
-
-import com.ibm.crail.rpc.RpcProtocol;
-import com.ibm.crail.rpc.RpcRequestMessage;
-import com.ibm.crail.utils.CrailUtils;
-
-public class LogRecord {
- public static final Logger LOG = CrailUtils.getLogger();
- public static final int CSIZE = 4 + Math.max(RpcRequestMessage.SetFileReq.CSIZE, RpcRequestMessage.RenameFileReq.CSIZE);
-
- private short cmd;
- private short type;
- private RpcRequestMessage.CreateFileReq createFileReq;
- private RpcRequestMessage.GetFileReq fileReq;
- private RpcRequestMessage.SetFileReq setFileReq;
- private RpcRequestMessage.RemoveFileReq removeReq;
- private RpcRequestMessage.RenameFileReq renameFileReq;
- private RpcRequestMessage.GetBlockReq getBlockReq;
- private RpcRequestMessage.GetLocationReq getLocationReq;
- private RpcRequestMessage.SetBlockReq setBlockReq;
- private RpcRequestMessage.GetDataNodeReq getDataNodeReq;
- private RpcRequestMessage.DumpNameNodeReq dumpNameNodeReq;
- private RpcRequestMessage.PingNameNodeReq pingNameNodeReq;
-
- public LogRecord() {
- this.cmd = 0;
- this.type = 0;
- this.createFileReq = new RpcRequestMessage.CreateFileReq();
- this.fileReq = new RpcRequestMessage.GetFileReq();
- this.setFileReq = new RpcRequestMessage.SetFileReq();
- this.removeReq = new RpcRequestMessage.RemoveFileReq();
- this.renameFileReq = new RpcRequestMessage.RenameFileReq();
- this.getBlockReq = new RpcRequestMessage.GetBlockReq();
- this.getLocationReq = new RpcRequestMessage.GetLocationReq();
- this.setBlockReq = new RpcRequestMessage.SetBlockReq();
- this.dumpNameNodeReq = new RpcRequestMessage.DumpNameNodeReq();
- this.pingNameNodeReq = new RpcRequestMessage.PingNameNodeReq();
- this.getDataNodeReq = new RpcRequestMessage.GetDataNodeReq();
- }
-
- public LogRecord(RpcRequestMessage.CreateFileReq message) {
- this.type = message.getType();
- this.createFileReq = message;
- }
- public LogRecord(RpcRequestMessage.GetFileReq message) {
- this.type = message.getType();
- this.fileReq = message;
- }
-
- public LogRecord(RpcRequestMessage.SetFileReq message) {
- this.type = message.getType();
- this.setFileReq = message;
- }
-
- public LogRecord(RpcRequestMessage.RemoveFileReq message) {
- this.type = message.getType();
- this.removeReq = message;
- }
-
- public LogRecord(RpcRequestMessage.RenameFileReq message) {
- this.type = message.getType();
- this.renameFileReq = message;
- }
-
- public LogRecord(RpcRequestMessage.GetBlockReq message) {
- this.type = message.getType();
- this.getBlockReq = message;
- }
-
- public LogRecord(RpcRequestMessage.GetLocationReq message) {
- this.type = message.getType();
- this.getLocationReq = message;
- }
-
- public LogRecord(RpcRequestMessage.SetBlockReq message) {
- this.type = message.getType();
- this.setBlockReq = message;
- }
-
- public LogRecord(RpcRequestMessage.GetDataNodeReq message) {
- this.type = message.getType();
- this.getDataNodeReq = message;
- }
-
- public LogRecord(RpcRequestMessage.DumpNameNodeReq message) {
- this.type = message.getType();
- this.dumpNameNodeReq = message;
- }
-
- public LogRecord(RpcRequestMessage.PingNameNodeReq message) {
- this.type = message.getType();
- this.pingNameNodeReq = message;
- }
-
- public void setCommand(short command) {
- this.cmd = command;
- }
-
- public int size(){
- return CSIZE;
- }
-
- public int write(ByteBuffer buffer) throws IOException{
- buffer.putShort(cmd);
- buffer.putShort(type);
-
- int written = 4;
- switch(type){
- case RpcProtocol.REQ_CREATE_FILE:
- written += createFileReq.write(buffer);
- break;
- case RpcProtocol.REQ_GET_FILE:
- written += fileReq.write(buffer);
- break;
- case RpcProtocol.REQ_SET_FILE:
- written += setFileReq.write(buffer);
- break;
- case RpcProtocol.REQ_REMOVE_FILE:
- written += removeReq.write(buffer);
- break;
- case RpcProtocol.REQ_RENAME_FILE:
- written += renameFileReq.write(buffer);
- break;
- case RpcProtocol.REQ_GET_BLOCK:
- written += getBlockReq.write(buffer);
- break;
- case RpcProtocol.REQ_GET_LOCATION:
- written += getLocationReq.write(buffer);
- break;
- case RpcProtocol.REQ_SET_BLOCK:
- written += setBlockReq.write(buffer);
- break;
- case RpcProtocol.REQ_GET_DATANODE:
- written += getDataNodeReq.write(buffer);
- break;
- case RpcProtocol.REQ_DUMP_NAMENODE:
- written += dumpNameNodeReq.write(buffer);
- break;
- case RpcProtocol.REQ_PING_NAMENODE:
- written += pingNameNodeReq.write(buffer);
- break;
- }
-
- return written;
- }
-
- public void update(ByteBuffer buffer) throws IOException {
- this.cmd = buffer.getShort();
- this.type = buffer.getShort();
-
- switch(type){
- case RpcProtocol.REQ_CREATE_FILE:
- createFileReq.update(buffer);
- break;
- case RpcProtocol.REQ_GET_FILE:
- fileReq.update(buffer);
- break;
- case RpcProtocol.REQ_SET_FILE:
- setFileReq.update(buffer);
- break;
- case RpcProtocol.REQ_REMOVE_FILE:
- removeReq.update(buffer);
- break;
- case RpcProtocol.REQ_RENAME_FILE:
- renameFileReq.update(buffer);
- break;
- case RpcProtocol.REQ_GET_BLOCK:
- getBlockReq.update(buffer);
- break;
- case RpcProtocol.REQ_GET_LOCATION:
- getLocationReq.update(buffer);
- break;
- case RpcProtocol.REQ_SET_BLOCK:
- setBlockReq.update(buffer);
- break;
- case RpcProtocol.REQ_GET_DATANODE:
- getDataNodeReq.update(buffer);
- break;
- case RpcProtocol.REQ_DUMP_NAMENODE:
- dumpNameNodeReq.update(buffer);
- break;
- case RpcProtocol.REQ_PING_NAMENODE:
- pingNameNodeReq.update(buffer);
- break;
- }
- }
-
- public short getCmd() {
- return cmd;
- }
-
- public short getType(){
- return type;
- }
-
- public RpcRequestMessage.CreateFileReq createFile(){
- return this.createFileReq;
- }
-
- public RpcRequestMessage.GetFileReq getFile(){
- return fileReq;
- }
-
- public RpcRequestMessage.SetFileReq setFile() {
- return setFileReq;
- }
-
- public RpcRequestMessage.RemoveFileReq removeFile(){
- return removeReq;
- }
-
- public RpcRequestMessage.RenameFileReq renameFile(){
- return renameFileReq;
- }
-
- public RpcRequestMessage.GetBlockReq getBlock() {
- return getBlockReq;
- }
-
- public RpcRequestMessage.GetLocationReq getLocation() {
- return getLocationReq;
- }
-
- public RpcRequestMessage.SetBlockReq setBlock() {
- return setBlockReq;
- }
-
- public RpcRequestMessage.GetDataNodeReq getDataNode() {
- return this.getDataNodeReq;
- }
-
- public RpcRequestMessage.DumpNameNodeReq dumpNameNode() {
- return this.dumpNameNodeReq;
- }
-
- public RpcRequestMessage.PingNameNodeReq pingNameNode(){
- return this.pingNameNodeReq;
- }
-
- @Override
- public String toString() {
- String message = "";
- switch(type){
- case RpcProtocol.REQ_CREATE_FILE:
- message += createFileReq.toString();
- break;
- case RpcProtocol.REQ_GET_FILE:
- message += fileReq.toString();
- break;
- case RpcProtocol.REQ_SET_FILE:
- message += setFileReq.toString();
- break;
- case RpcProtocol.REQ_REMOVE_FILE:
- message += removeReq.toString();
- break;
- case RpcProtocol.REQ_RENAME_FILE:
- message += renameFileReq.toString();
- break;
- case RpcProtocol.REQ_GET_BLOCK:
- message += getBlockReq.toString();
- break;
- case RpcProtocol.REQ_GET_LOCATION:
- message += getLocationReq.toString();
- break;
- case RpcProtocol.REQ_SET_BLOCK:
- message += setBlockReq.toString();
- break;
- case RpcProtocol.REQ_GET_DATANODE:
- message += getDataNodeReq.toString();
- break;
- case RpcProtocol.REQ_DUMP_NAMENODE:
- message += dumpNameNodeReq.toString();
- break;
- case RpcProtocol.REQ_PING_NAMENODE:
- message += pingNameNodeReq.toString();
- break;
- }
-
- return message;
- }
-}