[14/29] incubator-crail git commit: Apache package renaming

[ps...@apache.org]

http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/client/src/main/java/org/apache/crail/conf/CrailConstants.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/crail/conf/CrailConstants.java b/client/src/main/java/org/apache/crail/conf/CrailConstants.java
new file mode 100644
index 0000000..667c454
--- /dev/null
+++ b/client/src/main/java/org/apache/crail/conf/CrailConstants.java
@@ -0,0 +1,262 @@
+/*
+ * 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.conf;
+
+import java.io.IOException;
+
+import org.apache.crail.utils.CrailUtils;
+import org.slf4j.Logger;
+
+public class CrailConstants {
+	private static final Logger LOG = CrailUtils.getLogger();
+	
+	public static final String VERSION_KEY = "crail.version";
+	public static int VERSION = 3003;
+	
+	public static final String DIRECTORY_DEPTH_KEY = "crail.directorydepth";
+	public static int DIRECTORY_DEPTH = 16;
+	
+	public static final String TOKEN_EXPIRATION_KEY = "crail.tokenexpiration";
+	public static long TOKEN_EXPIRATION = 10;
+	
+	public static final String BLOCK_SIZE_KEY = "crail.blocksize";
+	public static long BLOCK_SIZE = 67108864;
+	
+	public static final String CACHE_LIMIT_KEY = "crail.cachelimit";
+	public static long CACHE_LIMIT = 1073741824;
+	
+	public static final String CACHE_PATH_KEY = "crail.cachepath";
+	public static String CACHE_PATH = "/home/stu/craildata/cache";	
+	
+	public static final String USER_KEY = "crail.user";
+	public static String USER = "stu";
+	
+	public static final String SHADOW_REPLICATION_KEY = "crail.shadowreplication";
+	public static int SHADOW_REPLICATION = 1;	
+	
+	public static final String DEBUG_KEY = "crail.debug";
+	public static boolean DEBUG = false;
+	
+	public static final String STATISTICS_KEY = "crail.statistics";
+	public static boolean STATISTICS = true;	
+	
+	public static final String RPC_TIMEOUT_KEY = "crail.rpctimeout";
+	public static int RPC_TIMEOUT = 1000;
+	
+	public static final String DATA_TIMEOUT_KEY = "crail.datatimeout";
+	public static int DATA_TIMEOUT = 1000;	
+	
+	public static final String BUFFER_SIZE_KEY = "crail.buffersize";
+	public static int BUFFER_SIZE = 1048576;
+	
+	public static final String SLICE_SIZE_KEY = "crail.slicesize";
+	public static int SLICE_SIZE = 524288;		
+	
+	public static final String SINGLETON_KEY = "crail.singleton";
+	public static boolean SINGLETON = false;	
+	
+	public static final String REGION_SIZE_KEY = "crail.regionsize";
+	public static long REGION_SIZE = 1073741824;
+	
+	public static final String DIRECTORY_RECORD_KEY = "crail.directoryrecord";
+	public static int DIRECTORY_RECORD = 512;	
+	
+	public static final String DIRECTORY_RANDOMIZE_KEY = "crail.directoryrandomize";
+	public static boolean DIRECTORY_RANDOMIZE = true;
+	
+	public static final String CACHE_IMPL_KEY = "crail.cacheimpl";
+	public static String CACHE_IMPL = "com.ibm.crail.memory.MappedBufferCache";
+	
+	public static final String LOCATION_MAP_KEY = "crail.locationmap";
+	public static String LOCATION_MAP = "";		
+	
+	//namenode interface
+	public static final String NAMENODE_ADDRESS_KEY = "crail.namenode.address";
+	public static String NAMENODE_ADDRESS = "";
+	
+	public static final String NAMENODE_FILEBLOCKS_KEY = "crail.namenode.fileblocks";
+	public static int NAMENODE_FILEBLOCKS = 16;	
+	
+	public static final String NAMENODE_BLOCKSELECTION_KEY = "crail.namenode.blockselection";
+	public static String NAMENODE_BLOCKSELECTION = "roundrobin";	
+	
+	public static final String NAMENODE_RPC_TYPE_KEY = "crail.namenode.rpctype";
+	public static String NAMENODE_RPC_TYPE = "com.ibm.crail.namenode.rpc.darpc.DaRPCNameNode";
+	
+	public static final String NAMENODE_RPC_SERVICE_KEY = "crail.namenode.rpcservice";
+	public static String NAMENODE_RPC_SERVICE = "com.ibm.crail.namenode.NameNodeService";	
+	
+	public static final String NAMENODE_LOG_KEY = "crail.namenode.log";
+	public static String NAMENODE_LOG = "";		
+	
+	//storage interface
+	public static final String STORAGE_TYPES_KEY = "crail.storage.types";
+	public static String STORAGE_TYPES = "com.ibm.crail.storage.rdma.RdmaStorageTier";		
+	
+	public static final String STORAGE_CLASSES_KEY = "crail.storage.classes";
+	public static int STORAGE_CLASSES = 1;	
+	
+	public static final String STORAGE_ROOTCLASS_KEY = "crail.storage.rootclass";
+	public static int STORAGE_ROOTCLASS = 0;		
+	
+	public static final String STORAGE_KEEPALIVE_KEY = "crail.storage.keepalive";
+	public static int STORAGE_KEEPALIVE = 2;		
+	
+	public static void updateConstants(CrailConfiguration conf){
+		//general
+		if (conf.get(DIRECTORY_DEPTH_KEY) != null) {
+			DIRECTORY_DEPTH = Integer.parseInt(conf.get(DIRECTORY_DEPTH_KEY));
+		}			
+		if (conf.get(TOKEN_EXPIRATION_KEY) != null) {
+			TOKEN_EXPIRATION = Long.parseLong(conf.get(TOKEN_EXPIRATION_KEY));
+		}		
+		if (conf.get(BLOCK_SIZE_KEY) != null) {
+			BLOCK_SIZE = Long.parseLong(conf.get(BLOCK_SIZE_KEY));
+		}			
+		if (conf.get(CACHE_LIMIT_KEY) != null) {
+			CACHE_LIMIT = Long.parseLong(conf.get(CACHE_LIMIT_KEY));
+		}			
+		if (conf.get(CACHE_PATH_KEY) != null) {
+			CACHE_PATH = conf.get(CACHE_PATH_KEY);
+		}
+		if (conf.get(USER_KEY) != null) {
+			USER = conf.get(CrailConstants.USER_KEY);
+		}
+		if (conf.get(SHADOW_REPLICATION_KEY) != null) {
+			SHADOW_REPLICATION = Integer.parseInt(conf.get(SHADOW_REPLICATION_KEY));
+		}	
+		if (conf.get(DEBUG_KEY) != null) {
+			DEBUG = Boolean.parseBoolean(conf.get(DEBUG_KEY));
+		}	
+		if (conf.get(STATISTICS_KEY) != null) {
+			STATISTICS = Boolean.parseBoolean(conf.get(STATISTICS_KEY));
+		}			
+		if (conf.get(RPC_TIMEOUT_KEY) != null) {
+			RPC_TIMEOUT = Integer.parseInt(conf.get(RPC_TIMEOUT_KEY));
+		}	
+		if (conf.get(DATA_TIMEOUT_KEY) != null) {
+			DATA_TIMEOUT = Integer.parseInt(conf.get(DATA_TIMEOUT_KEY));
+		}	
+		if (conf.get(BUFFER_SIZE_KEY) != null) {
+			BUFFER_SIZE = Integer.parseInt(conf.get(BUFFER_SIZE_KEY));
+		}	
+		if (conf.get(SLICE_SIZE_KEY) != null) {
+			SLICE_SIZE = Integer.parseInt(conf.get(SLICE_SIZE_KEY));
+		}			
+		if (conf.get(CrailConstants.SINGLETON_KEY) != null) {
+			SINGLETON = conf.getBoolean(CrailConstants.SINGLETON_KEY, false);
+		}	
+		if (conf.get(REGION_SIZE_KEY) != null) {
+			REGION_SIZE = Integer.parseInt(conf.get(REGION_SIZE_KEY));
+		}	
+		if (conf.get(DIRECTORY_RECORD_KEY) != null) {
+			DIRECTORY_RECORD = Integer.parseInt(conf.get(DIRECTORY_RECORD_KEY));
+		}	
+		if (conf.get(CrailConstants.DIRECTORY_RANDOMIZE_KEY) != null) {
+			DIRECTORY_RANDOMIZE = conf.getBoolean(CrailConstants.DIRECTORY_RANDOMIZE_KEY, false);
+		}
+		if (conf.get(CACHE_IMPL_KEY) != null) {
+			CACHE_IMPL = conf.get(CACHE_IMPL_KEY);
+		}
+		if (conf.get(LOCATION_MAP_KEY) != null) {
+			LOCATION_MAP = conf.get(LOCATION_MAP_KEY);
+		}		
+		
+		//namenode interface
+		if (conf.get(NAMENODE_ADDRESS_KEY) != null) {
+			NAMENODE_ADDRESS = conf.get(NAMENODE_ADDRESS_KEY);
+		} 
+		if (conf.get(NAMENODE_BLOCKSELECTION_KEY) != null) {
+			NAMENODE_BLOCKSELECTION = conf.get(NAMENODE_BLOCKSELECTION_KEY);
+		}		
+		if (conf.get(NAMENODE_FILEBLOCKS_KEY) != null) {
+			NAMENODE_FILEBLOCKS = Integer.parseInt(conf.get(NAMENODE_FILEBLOCKS_KEY));
+		}		
+		if (conf.get(NAMENODE_RPC_TYPE_KEY) != null) {
+			NAMENODE_RPC_TYPE = conf.get(NAMENODE_RPC_TYPE_KEY);
+		}
+		if (conf.get(NAMENODE_LOG_KEY) != null) {
+			NAMENODE_LOG = conf.get(NAMENODE_LOG_KEY);
+		}		
+		
+		//storage interface
+		if (conf.get(STORAGE_TYPES_KEY) != null) {
+			STORAGE_TYPES = conf.get(STORAGE_TYPES_KEY);
+		}	
+		if (conf.get(STORAGE_CLASSES_KEY) != null) {
+			STORAGE_CLASSES = Math.max(Integer.parseInt(conf.get(STORAGE_CLASSES_KEY)), CrailUtils.getStorageClasses(STORAGE_TYPES)); 
+		} else {
+			STORAGE_CLASSES = CrailUtils.getStorageClasses(STORAGE_TYPES);
+		}
+		if (conf.get(STORAGE_ROOTCLASS_KEY) != null) {
+			STORAGE_ROOTCLASS = Integer.parseInt(conf.get(STORAGE_ROOTCLASS_KEY));
+		}	
+		if (conf.get(STORAGE_KEEPALIVE_KEY) != null) {
+			STORAGE_KEEPALIVE = Integer.parseInt(conf.get(STORAGE_KEEPALIVE_KEY));
+		}		
+	}
+	
+	public static void printConf(){
+		LOG.info(VERSION_KEY + " " + VERSION);
+		LOG.info(DIRECTORY_DEPTH_KEY + " " + DIRECTORY_DEPTH);
+		LOG.info(TOKEN_EXPIRATION_KEY + " " + TOKEN_EXPIRATION);
+		LOG.info(BLOCK_SIZE_KEY + " " + BLOCK_SIZE);
+		LOG.info(CACHE_LIMIT_KEY + " " + CACHE_LIMIT);
+		LOG.info(CACHE_PATH_KEY + " " + CACHE_PATH);
+		LOG.info(USER_KEY + " " + USER);
+		LOG.info(SHADOW_REPLICATION_KEY + " " + SHADOW_REPLICATION);
+		LOG.info(DEBUG_KEY + " " + DEBUG);
+		LOG.info(STATISTICS_KEY + " " + STATISTICS);
+		LOG.info(RPC_TIMEOUT_KEY + " " + RPC_TIMEOUT);
+		LOG.info(DATA_TIMEOUT_KEY + " " + DATA_TIMEOUT);
+		LOG.info(BUFFER_SIZE_KEY + " " + BUFFER_SIZE);
+		LOG.info(SLICE_SIZE_KEY + " " + SLICE_SIZE);		
+		LOG.info(SINGLETON_KEY + " " + SINGLETON);
+		LOG.info(REGION_SIZE_KEY + " " + REGION_SIZE);
+		LOG.info(DIRECTORY_RECORD_KEY + " " + DIRECTORY_RECORD);
+		LOG.info(DIRECTORY_RANDOMIZE_KEY + " " + DIRECTORY_RANDOMIZE);		
+		LOG.info(CACHE_IMPL_KEY + " " + CACHE_IMPL);
+		LOG.info(LOCATION_MAP_KEY + " " + LOCATION_MAP);
+		LOG.info(NAMENODE_ADDRESS_KEY + " " + NAMENODE_ADDRESS);		
+		LOG.info(NAMENODE_BLOCKSELECTION_KEY + " " + NAMENODE_BLOCKSELECTION);
+		LOG.info(NAMENODE_FILEBLOCKS_KEY + " " + NAMENODE_FILEBLOCKS);
+		LOG.info(NAMENODE_RPC_TYPE_KEY + " " + NAMENODE_RPC_TYPE);
+		LOG.info(NAMENODE_LOG_KEY + " " + NAMENODE_LOG);
+		LOG.info(STORAGE_TYPES_KEY + " " + STORAGE_TYPES);
+		LOG.info(STORAGE_CLASSES_KEY + " " + STORAGE_CLASSES);
+		LOG.info(STORAGE_ROOTCLASS_KEY + " " + STORAGE_ROOTCLASS);
+		LOG.info(STORAGE_KEEPALIVE_KEY + " " + STORAGE_KEEPALIVE);
+	}
+	
+	public static void verify() throws IOException {
+		if (CrailConstants.BUFFER_SIZE % CrailConstants.DIRECTORY_RECORD != 0){
+			throw new IOException("crail.buffersize must be multiple of " + CrailConstants.DIRECTORY_RECORD);
+		}	
+		if (Math.max(CrailConstants.BUFFER_SIZE, CrailConstants.SLICE_SIZE) % Math.min(CrailConstants.BUFFER_SIZE, CrailConstants.SLICE_SIZE) != 0){
+			throw new IOException("crail.slicesize must be multiple of buffersize " + CrailConstants.BUFFER_SIZE);
+		}
+		if (CrailConstants.STORAGE_CLASSES < CrailUtils.getStorageClasses(STORAGE_TYPES)){
+			throw new IOException("crail.storage.classes cannot be smaller than the number of storage types " + CrailUtils.getStorageClasses(STORAGE_TYPES));
+		}		
+		
+	}
+}

http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/client/src/main/java/org/apache/crail/core/CoreBlockLocation.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/crail/core/CoreBlockLocation.java b/client/src/main/java/org/apache/crail/core/CoreBlockLocation.java
new file mode 100644
index 0000000..d2a5850
--- /dev/null
+++ b/client/src/main/java/org/apache/crail/core/CoreBlockLocation.java
@@ -0,0 +1,127 @@
+/*
+ * 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.core;
+
+import org.apache.crail.CrailBlockLocation;
+
+class CoreBlockLocation implements CrailBlockLocation {
+	private String[] names;
+	private String[] hosts;
+	private String[] topology;
+	private int[] storageTypes;
+	private int[] storageClasses;
+	private int[] locationAffinities;
+	private long offset; 
+	private long length;
+
+	public CoreBlockLocation() {
+	}
+
+	public long getOffset() {
+		return offset;
+	}
+
+	public void setOffset(long offset) {
+		this.offset = offset;
+	}
+	
+	public void setLength(long length) {
+		this.length = length;
+	}
+	
+	public long getLength(){
+		return length;
+	}
+
+	public String[] getNames() {
+		return names;
+	}
+
+	public void setNames(String[] names) {
+		this.names = names;
+	}
+
+	public String[] getHosts() {
+		return hosts;
+	}
+
+	public void setHosts(String[] hosts) {
+		this.hosts = hosts;
+	}
+
+	public String[] getTopology() {
+		return topology;
+	}
+
+	public void setTopologyPaths(String[] topology) {
+		this.topology = topology;
+	}
+
+	public int[] getStorageTypes() {
+		return storageTypes;
+	}
+
+	public void setStorageTypes(int[] storageTypes) {
+		this.storageTypes = storageTypes;
+	}
+	
+	public int[] getStorageClasses() {
+		return storageClasses;
+	}
+
+	public void setStorageClasses(int[] storageClasses) {
+		this.storageClasses = storageClasses;
+	}	
+
+	public int[] getLocationAffinities() {
+		return locationAffinities;
+	}
+
+	public void setLocationAffinities(int[] locationTiers) {
+		this.locationAffinities = locationTiers;
+	}
+	
+	  public String toString() {
+		    StringBuilder result = new StringBuilder();
+		    result.append(offset);
+		    result.append(',');
+		    result.append(length);
+		    for(String h: hosts) {
+		      result.append(',');
+		      result.append(h);
+		    }
+		    for (int i = 0; i < storageTypes.length; i++){
+		    	result.append(',');
+		    	result.append(storageTypes[i]);
+		    }
+		    for (int i = 0; i < storageClasses.length; i++){
+		    	result.append(',');
+		    	result.append(storageClasses[i]);
+		    }		    
+		    for (int i = 0; i < locationAffinities.length; i++){
+		    	result.append(',');
+		    	result.append(locationAffinities[i]);
+		    }		    
+		    return result.toString();
+		  }	
+
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/client/src/main/java/org/apache/crail/core/CoreDataOperation.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/crail/core/CoreDataOperation.java b/client/src/main/java/org/apache/crail/core/CoreDataOperation.java
new file mode 100644
index 0000000..9fd3ca3
--- /dev/null
+++ b/client/src/main/java/org/apache/crail/core/CoreDataOperation.java
@@ -0,0 +1,137 @@
+/*
+ * 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.core;
+
+import org.apache.crail.CrailBuffer;
+import org.apache.crail.CrailResult;
+import org.apache.crail.conf.CrailConstants;
+import org.apache.crail.storage.StorageFuture;
+import org.apache.crail.storage.StorageResult;
+import org.apache.crail.utils.BufferCheckpoint;
+import org.apache.crail.utils.MultiFuture;
+
+class CoreDataOperation extends MultiFuture<StorageResult, CrailResult> implements CrailResult {
+	private CoreStream stream;
+	private CrailBuffer buffer;
+	private long fileOffset;
+	private int bufferPosition;
+	private int bufferLimit;
+	private int operationLength;
+	
+	//current state
+	private BufferCheckpoint bufferCheckpoint;
+	private int inProcessLen;
+	private long completedLen;
+	private boolean isSynchronous;
+	
+	public CoreDataOperation(CoreStream stream, CrailBuffer buffer) throws Exception{
+		this.stream = stream;
+		this.buffer = buffer;
+		this.fileOffset = stream.position();
+		this.bufferPosition = buffer.position();
+		this.bufferLimit = buffer.limit();
+		this.operationLength = buffer.remaining();
+		this.inProcessLen = 0;
+		this.completedLen = 0;
+		this.isSynchronous = false;
+		
+		if (operationLength > 0){
+			this.bufferCheckpoint = stream.getBufferCheckpoint();
+			if (CrailConstants.DEBUG){
+				this.bufferCheckpoint.checkIn(buffer);
+			}		
+		} 
+	}
+	
+	public long getLen() {
+		return completedLen;
+	}
+	
+	void incProcessedLen(int opLen){
+		this.inProcessLen += opLen;
+	}
+	
+	long getInProcessLen(){
+		return this.inProcessLen;
+	}
+	
+	int getBufferLimit(){
+		return bufferLimit;
+	}
+	
+	int remaining() {
+		return operationLength - inProcessLen;
+	}	
+	
+	int getCurrentBufferPosition(){
+		return bufferPosition + inProcessLen;
+	}
+	
+	long getCurrentStreamPosition(){
+		return fileOffset + inProcessLen;
+	}
+	
+	boolean isProcessed() {
+		return inProcessLen == operationLength;
+	}
+	
+	public boolean cancel(boolean mayInterruptIfRunning) {
+		return false;
+	}
+
+	public boolean isCancelled() {
+		return false;
+	}
+	
+	public synchronized void add(StorageFuture dataFuture) {
+		super.add(dataFuture);
+		if (dataFuture.isSynchronous()){
+			this.isSynchronous = true;
+		}
+	}	
+	
+	//-----------
+	
+	public void completeOperation(){
+		super.completeOperation();
+		if (this.isDone()){
+			stream.update(fileOffset + completedLen);
+			if (CrailConstants.DEBUG){
+				bufferCheckpoint.checkOut(buffer);
+			}
+		}
+	}
+
+	boolean isSynchronous() {
+		return isSynchronous;
+	}
+
+	@Override
+	public void aggregate(StorageResult result) {
+		completedLen += result.getLen();
+	}
+
+	@Override
+	public CrailResult getAggregate() {
+		return this;
+	}
+}

http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/client/src/main/java/org/apache/crail/core/CoreDirectory.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/crail/core/CoreDirectory.java b/client/src/main/java/org/apache/crail/core/CoreDirectory.java
new file mode 100644
index 0000000..add90fd
--- /dev/null
+++ b/client/src/main/java/org/apache/crail/core/CoreDirectory.java
@@ -0,0 +1,94 @@
+/*
+ * 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.core;
+
+import java.util.Iterator;
+
+import org.apache.crail.CrailContainer;
+import org.apache.crail.CrailDirectory;
+import org.apache.crail.CrailMultiFile;
+import org.apache.crail.CrailTable;
+import org.apache.crail.conf.CrailConstants;
+import org.apache.crail.metadata.FileInfo;
+
+class CoreDirectory extends CoreNode implements CrailContainer, CrailDirectory, CrailMultiFile, CrailTable {
+	
+	public CoreDirectory(CoreFileSystem fs, FileInfo fileInfo, String path){
+		super(fs, fileInfo, path);
+		this.fs = fs;
+		this.fileInfo = fileInfo;
+		this.path = path;
+	}	
+
+	@Override
+	public int files() {
+		return (int) fileInfo.getCapacity()/CrailConstants.DIRECTORY_RECORD;
+	}
+
+	@Override
+	public Iterator<String> listEntries() throws Exception {
+		return fs.listEntries(path);
+	}
+
+	@Override
+	public CoreDirectory asDirectory() throws Exception {
+		if (!getType().isDirectory()){
+			throw new Exception("file type mismatch, type " + getType());
+		}
+		return this;
+	}
+	
+	@Override
+	public CoreDirectory asContainer() throws Exception {
+		if (!getType().isContainer()){
+			throw new Exception("file type mismatch, type " + getType());
+		}
+		return this;
+	}	
+	
+	@Override
+	public CrailMultiFile asMultiFile() throws Exception {
+		if (!getType().isMultiFile()){
+			throw new Exception("file type mismatch, type " + getType());
+		}
+		return this;
+	}	
+	
+	@Override
+	public CrailTable asTable() throws Exception {
+		if (!getType().isTable()){
+			throw new Exception("file type mismatch, type " + getType());
+		}
+		return this;
+	}
+	
+	DirectoryOutputStream getDirectoryOutputStream() throws Exception {
+		CoreOutputStream outputStream = super.getOutputStream(0);
+		return new DirectoryOutputStream(outputStream);
+	}
+	
+	DirectoryInputStream getDirectoryInputStream(boolean randomize) throws Exception {
+		CoreInputStream inputStream = super.getInputStream(0);
+		return new DirectoryInputStream(inputStream, randomize);
+	}	
+}
+

http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/client/src/main/java/org/apache/crail/core/CoreFile.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/crail/core/CoreFile.java b/client/src/main/java/org/apache/crail/core/CoreFile.java
new file mode 100644
index 0000000..083d075
--- /dev/null
+++ b/client/src/main/java/org/apache/crail/core/CoreFile.java
@@ -0,0 +1,202 @@
+/*
+ * 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.core;
+
+import java.util.concurrent.Semaphore;
+
+import org.apache.crail.CrailBlockLocation;
+import org.apache.crail.CrailContainer;
+import org.apache.crail.CrailDirectory;
+import org.apache.crail.CrailFS;
+import org.apache.crail.CrailFile;
+import org.apache.crail.CrailInputStream;
+import org.apache.crail.CrailKeyValue;
+import org.apache.crail.CrailMultiFile;
+import org.apache.crail.CrailNode;
+import org.apache.crail.CrailNodeType;
+import org.apache.crail.CrailOutputStream;
+import org.apache.crail.CrailTable;
+import org.apache.crail.metadata.FileInfo;
+
+public class CoreFile extends CoreNode implements CrailFile, CrailKeyValue {
+	private Semaphore outputStreams;
+	
+	public CoreFile(CoreFileSystem fs, FileInfo fileInfo, String path){
+		super(fs, fileInfo, path);
+		this.outputStreams = new Semaphore(1);
+	}
+	
+	public CrailInputStream getDirectInputStream(long readHint) throws Exception{
+		if (fileInfo.getType().isDirectory()){
+			throw new Exception("Cannot open stream for directory");
+		}		
+		
+		return super.getInputStream(readHint);
+	}	
+	
+	public synchronized CrailOutputStream getDirectOutputStream(long writeHint) throws Exception {
+		if (fileInfo.getType().isDirectory()){
+			throw new Exception("Cannot open stream for directory");
+		}		
+		if (fileInfo.getToken() == 0){
+			throw new Exception("File is in read mode, cannot create outputstream, fd " + fileInfo.getFd());
+		}
+		if (!outputStreams.tryAcquire()){
+			throw new Exception("Only one concurrent output stream per file allowed");
+		}
+		return super.getOutputStream(writeHint);
+	}
+	
+	public long getToken() {
+		return fileInfo.getToken();
+	}
+
+	public boolean tokenFree(){
+		return fileInfo.tokenFree();
+	}
+
+	public CoreFile asFile() throws Exception {
+		if (!getType().isDataFile()){
+			throw new Exception("file type mismatch, type " + getType());
+		}
+		return this;
+	}
+	
+	public CoreFile asKeyValue() throws Exception {
+		if (!getType().isKeyValue()){
+			throw new Exception("file type mismatch, type " + getType());
+		}		
+		return this;
+	}	
+
+	void closeOutputStream(CoreOutputStream stream) throws Exception {
+		super.closeOutputStream(stream);
+		outputStreams.release();
+	}
+}
+
+class CoreEarlyFile implements CrailFile, CrailKeyValue {
+	private CoreFileSystem fs;
+	private String path;
+	private CrailNodeType type;
+	private CreateNodeFuture future;
+	private CrailFile file;
+	
+	public CoreEarlyFile(CoreFileSystem fs, String path, CrailNodeType type, CreateNodeFuture future) {
+		this.fs = fs;
+		this.path = path;
+		this.type = type;
+		this.future = future;
+		this.file = null;
+	}
+
+	public CrailInputStream getDirectInputStream(long readHint) throws Exception{
+		return file().getDirectInputStream(readHint);
+	}	
+	
+	public synchronized CrailOutputStream getDirectOutputStream(long writeHint) throws Exception {
+		return file().getDirectOutputStream(writeHint);
+	}
+	
+	public CrailBlockLocation[] getBlockLocations(long start, long len) throws Exception{
+		return fs.getBlockLocations(path, start, len);
+	}	
+	
+	public long getToken() {
+		return file().getToken();
+	}
+
+	public CrailFile asFile() throws Exception {
+		return this;
+	}
+	
+	public CrailKeyValue asKeyValue() throws Exception {
+		return this;
+	}	
+
+	@Override
+	public CrailFS getFileSystem() {
+		return fs;
+	}
+
+	@Override
+	public String getPath() {
+		return path;
+	}
+
+	@Override
+	public CrailNode syncDir() throws Exception {
+		return file().syncDir();
+	}
+
+	@Override
+	public long getModificationTime() {
+		return file().getModificationTime();
+	}
+
+	@Override
+	public long getCapacity() {
+		return file().getCapacity();
+	}
+
+	@Override
+	public CrailNodeType getType() {
+		return type;
+	}
+	
+	@Override
+	public CrailContainer asContainer() throws Exception {
+		throw new Exception("this is not a container");
+	}	
+
+	@Override
+	public CrailDirectory asDirectory() throws Exception {
+		throw new Exception("this is not a directory");
+	}
+
+	@Override
+	public CrailMultiFile asMultiFile() throws Exception {
+		throw new Exception("this is not a multifile");
+	}
+	
+	@Override
+	public CrailTable asTable() throws Exception {
+		throw new Exception("this is not a table");
+	}	
+
+	@Override
+	public long getFd() {
+		return file().getFd();
+	}
+
+	private synchronized CrailFile file() {
+		try {
+			if (file == null){
+				file = this.future.get().asFile();
+			}
+			return file;
+		} catch(Exception e){
+			throw new RuntimeException(e);
+		}
+	}
+}
+

http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/client/src/main/java/org/apache/crail/core/CoreFileSystem.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/crail/core/CoreFileSystem.java b/client/src/main/java/org/apache/crail/core/CoreFileSystem.java
new file mode 100644
index 0000000..da6f06c
--- /dev/null
+++ b/client/src/main/java/org/apache/crail/core/CoreFileSystem.java
@@ -0,0 +1,653 @@
+/*
+ * 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.core;
+
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.StringTokenizer;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.apache.crail.CrailBlockLocation;
+import org.apache.crail.CrailBuffer;
+import org.apache.crail.CrailFS;
+import org.apache.crail.CrailLocationClass;
+import org.apache.crail.CrailNode;
+import org.apache.crail.CrailNodeType;
+import org.apache.crail.CrailResult;
+import org.apache.crail.CrailStatistics;
+import org.apache.crail.CrailStorageClass;
+import org.apache.crail.Upcoming;
+import org.apache.crail.conf.CrailConfiguration;
+import org.apache.crail.conf.CrailConstants;
+import org.apache.crail.memory.BufferCache;
+import org.apache.crail.metadata.BlockInfo;
+import org.apache.crail.metadata.DataNodeInfo;
+import org.apache.crail.metadata.FileInfo;
+import org.apache.crail.metadata.FileName;
+import org.apache.crail.rpc.RpcClient;
+import org.apache.crail.rpc.RpcConnection;
+import org.apache.crail.rpc.RpcCreateFile;
+import org.apache.crail.rpc.RpcDeleteFile;
+import org.apache.crail.rpc.RpcDispatcher;
+import org.apache.crail.rpc.RpcErrors;
+import org.apache.crail.rpc.RpcFuture;
+import org.apache.crail.rpc.RpcGetFile;
+import org.apache.crail.rpc.RpcGetLocation;
+import org.apache.crail.rpc.RpcPing;
+import org.apache.crail.rpc.RpcRenameFile;
+import org.apache.crail.storage.StorageClient;
+import org.apache.crail.utils.BlockCache;
+import org.apache.crail.utils.BufferCheckpoint;
+import org.apache.crail.utils.CrailUtils;
+import org.apache.crail.utils.EndpointCache;
+import org.apache.crail.utils.NextBlockCache;
+import org.apache.crail.utils.BlockCache.FileBlockCache;
+import org.apache.crail.utils.NextBlockCache.FileNextBlockCache;
+import org.slf4j.Logger;
+
+public class CoreFileSystem extends CrailFS {
+	private static final Logger LOG = CrailUtils.getLogger();
+	private static AtomicInteger fsCount = new AtomicInteger(0);
+	
+	//namenode operations
+	private RpcClient rpcClient;
+	private RpcConnection rpcConnection;
+	
+	//datanode operations
+	private EndpointCache datanodeEndpointCache;
+	
+	private AtomicLong streamCounter;
+	private ConcurrentHashMap<Long, CoreInputStream> openInputStreams;
+	private ConcurrentHashMap<Long, CoreOutputStream> openOutputStreams;
+	
+	private BlockCache blockCache;
+	private NextBlockCache nextBlockCache;
+	private BufferCache bufferCache;
+	private BufferCheckpoint bufferCheckpoint;
+	private ConcurrentHashMap<String, String> locationMap;
+	
+	private boolean isOpen;
+	private int fsId;
+	private CrailLocationClass localClass;
+	
+	private CoreIOStatistics ioStatsIn;
+	private CoreIOStatistics ioStatsOut;
+	private CoreStreamStatistics streamStats;
+	private CrailStatistics statistics;
+	
+	public CoreFileSystem(CrailConfiguration conf) throws Exception {
+		CrailConstants.updateConstants(conf);
+		CrailConstants.printConf();
+		CrailConstants.verify();	
+		
+		//Datanodes
+		StringTokenizer tokenizer = new StringTokenizer(CrailConstants.STORAGE_TYPES, ",");
+		LinkedList<StorageClient> dataNodeClients = new LinkedList<StorageClient>(); 
+		while (tokenizer.hasMoreTokens()){
+			String name = tokenizer.nextToken();
+			StorageClient dataNode = StorageClient.createInstance(name);
+			dataNode.init(conf, null);
+			dataNode.printConf(LOG);
+			dataNodeClients.add(dataNode);
+		}
+		this.datanodeEndpointCache = new EndpointCache(fsId, dataNodeClients);
+		
+		//Namenode
+		InetSocketAddress nnAddr = CrailUtils.getNameNodeAddress();
+		this.rpcClient = RpcClient.createInstance(CrailConstants.NAMENODE_RPC_TYPE);
+		rpcClient.init(conf, null);
+		rpcClient.printConf(LOG);	
+		ConcurrentLinkedQueue<InetSocketAddress> namenodeList = CrailUtils.getNameNodeList();
+		ConcurrentLinkedQueue<RpcConnection> connectionList = new ConcurrentLinkedQueue<RpcConnection>();
+		while(!namenodeList.isEmpty()){
+			InetSocketAddress address = namenodeList.poll();
+			RpcConnection connection = rpcClient.connect(address);
+			connectionList.add(connection);
+		}
+		if (connectionList.size() == 1){
+			this.rpcConnection = connectionList.poll();
+		} else {
+			this.rpcConnection = new RpcDispatcher(connectionList);
+		}
+		LOG.info("connected to namenode(s) " + rpcConnection);		
+		
+		//Client
+		this.fsId = fsCount.getAndIncrement();
+		this.localClass = CrailUtils.getLocationClass();
+		this.bufferCache = BufferCache.createInstance(CrailConstants.CACHE_IMPL);
+		this.blockCache = new BlockCache();
+		this.nextBlockCache = new NextBlockCache();
+		this.openInputStreams = new ConcurrentHashMap<Long, CoreInputStream>();
+		this.openOutputStreams = new ConcurrentHashMap<Long, CoreOutputStream>();
+		this.streamCounter = new AtomicLong(0);
+		this.isOpen = true;
+		this.bufferCheckpoint = new BufferCheckpoint();
+		this.locationMap = new ConcurrentHashMap<String, String>();
+		CrailUtils.parseMap(CrailConstants.LOCATION_MAP, locationMap);
+		
+		this.statistics = new CrailStatistics();
+		this.ioStatsIn = new CoreIOStatistics("core/input");
+		statistics.addProvider(ioStatsIn);
+		this.ioStatsOut = new CoreIOStatistics("core/output");
+		statistics.addProvider(ioStatsOut);
+		this.streamStats = new CoreStreamStatistics();
+		statistics.addProvider(streamStats);
+		statistics.addProvider(bufferCache);
+		statistics.addProvider(datanodeEndpointCache);
+	}
+	
+	public Upcoming<CrailNode> create(String path, CrailNodeType type, CrailStorageClass storageClass, CrailLocationClass locationClass) throws Exception {
+		FileName name = new FileName(path);
+		
+		if (CrailConstants.DEBUG){
+			LOG.info("createNode: name " + path + ", type " + type + ", storageAffinity " + storageClass + ", locationAffinity " + locationClass);
+		}
+
+		RpcFuture<RpcCreateFile> fileRes = rpcConnection.createFile(name, type, storageClass.value(), locationClass.value());
+		return new CreateNodeFuture(this, path, type, fileRes);
+	}	
+	
+	CoreNode _createNode(String path, CrailNodeType type, RpcCreateFile fileRes) throws Exception {
+		if (fileRes.getError() == RpcErrors.ERR_PARENT_MISSING){
+			throw new IOException("createNode: " + RpcErrors.messages[fileRes.getError()] + ", name " + path);
+		} else if  (fileRes.getError() == RpcErrors.ERR_FILE_EXISTS){
+			throw new IOException("createNode: " + RpcErrors.messages[fileRes.getError()] + ", name " + path);
+		} else if (fileRes.getError() != RpcErrors.ERR_OK){
+			LOG.info("createNode: " + RpcErrors.messages[fileRes.getError()] + ", name " + path);
+			throw new IOException("createNode: " + RpcErrors.messages[fileRes.getError()] + ", error " + fileRes.getError());
+		}		
+		
+		FileInfo fileInfo = fileRes.getFile();
+		FileInfo dirInfo = fileRes.getParent();
+		if (fileInfo == null || dirInfo == null){
+			throw new IOException("createFile: " + RpcErrors.messages[RpcErrors.ERR_UNKNOWN]);
+		}
+		if (fileInfo.getType() != type){
+			throw new IOException("createFile: " + "file type mismatch");
+		}
+
+		blockCache.remove(fileInfo.getFd());
+		nextBlockCache.remove(fileInfo.getFd());
+		CoreNode node = CoreNode.create(this, fileInfo, path);
+		
+		BlockInfo fileBlock = fileRes.getFileBlock();
+		getBlockCache(fileInfo.getFd()).put(CoreSubOperation.createKey(fileInfo.getFd(), 0), fileBlock);
+
+		//write directory record is a directory slot has been assigned to the file
+		if (fileInfo.getDirOffset() >= 0){
+			BlockInfo dirBlock = fileRes.getDirBlock();
+			getBlockCache(dirInfo.getFd()).put(CoreSubOperation.createKey(dirInfo.getFd(), fileInfo.getDirOffset()), dirBlock);
+			CoreSyncOperation syncOperation = getSyncOperation(dirInfo, fileInfo, path, true);
+			node.addSyncOperation(syncOperation);
+		}
+		
+		if (CrailConstants.DEBUG){
+			LOG.info("createFile: name " + path + ", success, fd " + fileInfo.getFd() + ", token " + fileInfo.getToken());
+		}
+		
+		return node;
+	}	
+	
+	public Upcoming<CrailNode> lookup(String path) throws Exception {
+		FileName name = new FileName(path);
+		
+		if (CrailConstants.DEBUG){
+			LOG.info("lookupDirectory: path " + path);
+		}
+		
+		RpcFuture<RpcGetFile> fileRes = rpcConnection.getFile(name, false);
+		return new LookupNodeFuture(this, path, fileRes);
+	}	
+	
+	CoreNode _lookupNode(RpcGetFile fileRes, String path) throws Exception {
+		if (fileRes.getError() == RpcErrors.ERR_GET_FILE_FAILED){
+			return null;
+		}
+		else if (fileRes.getError() != RpcErrors.ERR_OK){
+			LOG.info("lookupDirectory: " + RpcErrors.messages[fileRes.getError()]);
+			return null;
+		}		
+		
+		FileInfo fileInfo = fileRes.getFile();
+		
+		CoreNode node = null;
+		if (fileInfo != null){
+			if (CrailConstants.DEBUG){
+				LOG.info("lookup: name " + path + ", success, fd " + fileInfo.getFd());
+			}
+			BlockInfo fileBlock = fileRes.getFileBlock();
+			getBlockCache(fileInfo.getFd()).put(CoreSubOperation.createKey(fileInfo.getFd(), 0), fileBlock);
+			
+			node = CoreNode.create(this, fileInfo, path);
+		} 
+		return node;
+	}	
+	
+
+	public Upcoming<CrailNode> rename(String src, String dst) throws Exception {
+		FileName srcPath = new FileName(src);
+		FileName dstPath = new FileName(dst);
+		
+		if (CrailConstants.DEBUG){
+			LOG.info("rename: srcname " + src + ", dstname " + dst);
+		}
+		
+		RpcFuture<RpcRenameFile> renameRes = rpcConnection.renameFile(srcPath, dstPath);
+		return new RenameNodeFuture(this, src, dst, renameRes);
+	}
+	
+	CrailNode _rename(RpcRenameFile renameRes, String src, String dst) throws Exception {
+		if (renameRes.getError() == RpcErrors.ERR_SRC_FILE_NOT_FOUND){
+			LOG.info("rename: " + RpcErrors.messages[renameRes.getError()]);
+			return null;
+		}
+		if (renameRes.getError() == RpcErrors.ERR_DST_PARENT_NOT_FOUND){
+			LOG.info("rename: " + RpcErrors.messages[renameRes.getError()]);
+			return null;
+		}
+		if (renameRes.getError() == RpcErrors.ERR_FILE_EXISTS){
+			LOG.info("rename: " + RpcErrors.messages[renameRes.getError()]);
+			return null;
+		}
+		if (renameRes.getError() != RpcErrors.ERR_OK){
+			LOG.info("rename: " + RpcErrors.messages[renameRes.getError()]);
+			throw new IOException(RpcErrors.messages[renameRes.getError()]);			
+		} 
+		if (renameRes.getDstParent().getCapacity() < renameRes.getDstFile().getDirOffset() + CrailConstants.DIRECTORY_RECORD){
+			LOG.info("rename: parent capacity does not match dst file offset, capacity " + renameRes.getDstParent().getCapacity() + ", offset " + renameRes.getDstFile().getDirOffset());
+		}		
+		
+		FileInfo srcParent = renameRes.getSrcParent();
+		FileInfo srcFile = renameRes.getSrcFile();
+		FileInfo dstDir = renameRes.getDstParent();
+		FileInfo dstFile = renameRes.getDstFile();
+		
+		BlockInfo srcBlock = renameRes.getSrcBlock();
+		getBlockCache(srcParent.getFd()).put(CoreSubOperation.createKey(srcParent.getFd(), srcFile.getDirOffset()), srcBlock);
+		BlockInfo dirBlock = renameRes.getDstBlock();
+		getBlockCache(dstDir.getFd()).put(CoreSubOperation.createKey(dstDir.getFd(), dstFile.getDirOffset()), dirBlock);		
+		
+		CoreSyncOperation syncOperationSrc = getSyncOperation(srcParent, srcFile, src, false);
+		CoreSyncOperation syncOperationDst = getSyncOperation(dstDir, dstFile, dst, true);
+		
+		blockCache.remove(srcFile.getFd());
+		
+		if (CrailConstants.DEBUG){
+			LOG.info("rename: srcname " + src + ", dstname " + dst + ", success");
+		}
+		
+		CoreNode node = CoreNode.create(this, dstFile, dst);
+		node.addSyncOperation(syncOperationSrc);
+		node.addSyncOperation(syncOperationDst);
+		return node;
+	}
+	
+	public Upcoming<CrailNode> delete(String path, boolean recursive) throws Exception {
+		FileName name = new FileName(path);
+		
+		if (CrailConstants.DEBUG){
+			LOG.info("delete: name " + path + ", recursive " + recursive);
+		}
+
+		RpcFuture<RpcDeleteFile> fileRes = rpcConnection.removeFile(name, recursive);
+		return new DeleteNodeFuture(this, path, recursive, fileRes);
+	}	
+	
+	CrailNode _delete(RpcDeleteFile fileRes, String path, boolean recursive) throws Exception {
+		if (fileRes.getError() == RpcErrors.ERR_HAS_CHILDREN) {
+			LOG.info("delete: " + RpcErrors.messages[fileRes.getError()]);
+			throw new IOException(RpcErrors.messages[fileRes.getError()]);
+		}
+		if (fileRes.getError() != RpcErrors.ERR_OK) {
+			LOG.info("delete: " + RpcErrors.messages[fileRes.getError()]);
+			return null;
+		} 
+		
+		FileInfo fileInfo = fileRes.getFile();
+		FileInfo dirInfo = fileRes.getParent();
+		
+		CoreSyncOperation syncOperation = getSyncOperation(dirInfo, fileInfo, path, false);
+		
+		blockCache.remove(fileInfo.getFd());
+		
+		if (CrailConstants.DEBUG){
+			LOG.info("delete: name " + path + ", recursive " + recursive + ", success");
+		}
+		
+		CoreNode node = CoreNode.create(this, fileInfo, path);
+		node.addSyncOperation(syncOperation);
+		return node;
+	}	
+	
+	public DirectoryInputStream listEntries(String name) throws Exception {
+		return _listEntries(name, CrailConstants.DIRECTORY_RANDOMIZE);
+	}
+	
+	public DirectoryInputStream _listEntries(String name, boolean randomize) throws Exception {
+		FileName directory = new FileName(name);
+		
+		if (CrailConstants.DEBUG){
+			LOG.info("getDirectoryList: " + name);
+		}
+
+		RpcGetFile fileRes = rpcConnection.getFile(directory, false).get(CrailConstants.RPC_TIMEOUT, TimeUnit.MILLISECONDS);
+		if (fileRes.getError() != RpcErrors.ERR_OK) {
+			LOG.info("getDirectoryList: " + RpcErrors.messages[fileRes.getError()]);
+			throw new FileNotFoundException(RpcErrors.messages[fileRes.getError()]);
+		}
+		
+		FileInfo dirInfo = fileRes.getFile();
+		if (!dirInfo.getType().isContainer()){
+			LOG.info("getDirectoryList: " + RpcErrors.messages[RpcErrors.ERR_FILE_IS_NOT_DIR]);
+			throw new FileNotFoundException(RpcErrors.messages[RpcErrors.ERR_FILE_IS_NOT_DIR]);
+		}
+		
+		CoreDirectory dirFile = new CoreDirectory(this, dirInfo, name);
+		DirectoryInputStream inputStream = dirFile.getDirectoryInputStream(randomize);
+		return inputStream;
+	}	
+	
+	public CrailBlockLocation[] getBlockLocations(String path, long start, long len) throws Exception {
+		if (CrailConstants.DEBUG){
+			LOG.info("location: path " + path + ", start " + start + ", len " + len);
+		}			
+		
+		if (path == null) {
+			LOG.info("Path null");
+			return null;
+		}
+		if (start < 0 || len < 0) {
+			LOG.info("Start or len invalid");
+			throw new IOException("Invalid start or len parameter");
+		}
+		
+		
+		FileName name = new FileName(path);
+		long rangeStart = CrailUtils.blockStartAddress(start);
+		long range = start + len - CrailUtils.blockStartAddress(start);
+		long blockCount = range / CrailConstants.BLOCK_SIZE;
+		if (range % CrailConstants.BLOCK_SIZE > 0){
+			blockCount++;
+		}
+		CoreBlockLocation[] blockLocations = new CoreBlockLocation[(int) blockCount];
+		HashMap<Long, DataNodeInfo> dataNodeSet = new HashMap<Long, DataNodeInfo>();
+		HashMap<Long, DataNodeInfo> offset2DataNode = new HashMap<Long, DataNodeInfo>();
+	
+		for (long current = CrailUtils.blockStartAddress(start); current < start + len; current += CrailConstants.BLOCK_SIZE){
+			RpcGetLocation getLocationRes = rpcConnection.getLocation(name, current).get(CrailConstants.RPC_TIMEOUT, TimeUnit.MILLISECONDS);
+			if (getLocationRes.getError() != RpcErrors.ERR_OK) {
+				LOG.info("location: " + RpcErrors.messages[getLocationRes.getError()]);
+				throw new IOException(RpcErrors.messages[getLocationRes.getError()]);
+			}
+			
+			DataNodeInfo dataNodeInfo = getLocationRes.getBlockInfo().getDnInfo();
+			dataNodeSet.put(dataNodeInfo.key(), dataNodeInfo);
+			CoreBlockLocation location = new CoreBlockLocation();
+			location.setOffset(current);
+			location.setLength(Math.min(start + len - current, CrailConstants.BLOCK_SIZE));
+			long index = (current - rangeStart) / CrailConstants.BLOCK_SIZE;
+			blockLocations[(int) index] = location;
+			offset2DataNode.put(current, dataNodeInfo);
+		}
+		
+		//asign an identifier to each data node
+		ArrayList<DataNodeInfo> dataNodeArray = new ArrayList<DataNodeInfo>(dataNodeSet.size());
+		int index = 0;
+		for (DataNodeInfo dn : dataNodeSet.values()){
+			dataNodeArray.add(index, dn);
+			index++;
+		}
+		
+		int locationSize = Math.min(CrailConstants.SHADOW_REPLICATION, dataNodeSet.size());
+		int blockIndex = 0;
+		for (int i = 0; i < blockLocations.length; i++){
+			CoreBlockLocation location = blockLocations[i];
+			String[] hosts = new String[locationSize];
+			String[] names = new String[locationSize];
+			String[] topology = new String[locationSize];
+			int[] storageType = new int[locationSize];
+			int[] storageClass = new int[locationSize];
+			int[] locationTiers = new int[locationSize];
+			
+			DataNodeInfo dnInfo = offset2DataNode.get(location.getOffset());
+			DataNodeInfo mainDataNode = dataNodeSet.get(dnInfo.key());
+			InetSocketAddress address = CrailUtils.datanodeInfo2SocketAddr(mainDataNode);
+			names[0] = getMappedLocation(address.getAddress().getCanonicalHostName()) + ":" + address.getPort(); 
+			hosts[0] = getMappedLocation(address.getAddress().getCanonicalHostName());
+			topology[0] = "/default-rack/" + names[0];
+			storageType[0] = mainDataNode.getStorageType();
+			storageClass[0] = mainDataNode.getStorageClass();
+			locationTiers[0] = mainDataNode.getLocationClass();
+			for (int j = 1; j < locationSize; j++){
+				DataNodeInfo replicaDataNode = dataNodeArray.get(blockIndex);
+				address = CrailUtils.datanodeInfo2SocketAddr(replicaDataNode);
+				names[j] = getMappedLocation(address.getAddress().getCanonicalHostName()) + ":" + address.getPort(); 
+				hosts[j] = getMappedLocation(address.getAddress().getCanonicalHostName());
+				topology[j] = "/default-rack/" + names[j];
+				storageType[j] = replicaDataNode.getStorageType();
+				storageClass[j] = replicaDataNode.getStorageClass();
+				locationTiers[j] = replicaDataNode.getLocationClass();				
+				blockIndex = (blockIndex + 1) % dataNodeArray.size();
+			}
+			location.setNames(names);
+			location.setHosts(hosts);
+			location.setTopologyPaths(topology);
+			location.setStorageTypes(storageType);
+			location.setStorageClasses(storageClass);
+			location.setLocationAffinities(locationTiers);
+		}
+		
+		return blockLocations;
+	}
+	
+	public void dumpNameNode() throws Exception {
+		rpcConnection.dumpNameNode().get(CrailConstants.RPC_TIMEOUT, TimeUnit.MILLISECONDS);
+	}	
+	
+	public void ping() throws Exception {
+		RpcPing pingRes = rpcConnection.pingNameNode().get(CrailConstants.RPC_TIMEOUT, TimeUnit.MILLISECONDS);
+		if (pingRes.getError() != RpcErrors.ERR_OK) {
+			LOG.info("Ping: " + RpcErrors.messages[pingRes.getError()]);
+			throw new IOException(RpcErrors.messages[pingRes.getError()]);
+		}		
+	}
+
+	public CrailBuffer allocateBuffer() throws IOException {
+		return this.bufferCache.getBuffer();
+	}
+	
+	public void freeBuffer(CrailBuffer buffer) throws IOException {
+		this.bufferCache.putBuffer(buffer);
+	}	
+	
+	public int getFsId() {
+		return fsId;
+	}	
+	
+	public CrailLocationClass getLocationClass() {
+		return localClass;
+	}
+	
+	public BufferCheckpoint getBufferCheckpoint() {
+		return bufferCheckpoint;
+	}
+	
+	public CrailStatistics getStatistics(){
+		return statistics;
+	}
+	
+	public void closeFileSystem() throws Exception {
+		if (!isOpen) {
+			return;
+		}
+		
+		LinkedList<CoreInputStream> tmpIn = new LinkedList<CoreInputStream>();
+		for (CoreInputStream stream : openInputStreams.values()) {
+			tmpIn.add(stream);
+		}
+		for (CoreInputStream stream : tmpIn) {
+			stream.close();
+		}
+		
+		LinkedList<CoreOutputStream> tmpOut = new LinkedList<CoreOutputStream>();
+		LinkedList<CoreOutputStream> tmpOutDir = new LinkedList<CoreOutputStream>();
+		for (CoreOutputStream stream : openOutputStreams.values()) {
+			if (stream.getFile().getType().isContainer()){
+				tmpOutDir.add(stream);
+			} else {
+				tmpOut.add(stream);
+			}
+		}
+		for (CoreOutputStream stream : tmpOut) {
+			stream.close();
+		}		
+		for (CoreOutputStream stream : tmpOutDir) {
+			stream.close();
+		}		
+	
+		bufferCache.close();
+		datanodeEndpointCache.close();
+		rpcConnection.close();
+		rpcClient.close();
+		this.isOpen = false;
+	}
+
+	public void closeFile(FileInfo fileInfo) throws Exception {
+		if (fileInfo.getToken() > 0){
+			rpcConnection.setFile(fileInfo, true).get(CrailConstants.RPC_TIMEOUT, TimeUnit.MILLISECONDS);				
+		}
+	}
+
+	public BufferCache getBufferCache() {
+		return bufferCache;
+	}
+
+	public void purgeCache() {
+		blockCache.purge();
+		nextBlockCache.purge();
+	}
+
+	//-------------------------------------------------------------
+	
+	CoreOutputStream getOutputStream(CoreNode file, long writeHint) throws Exception {
+		CoreOutputStream outputStream = new CoreOutputStream(file, streamCounter.incrementAndGet(), writeHint);
+		openOutputStreams.put(outputStream.getStreamId(), outputStream);
+
+		if (CrailConstants.STATISTICS){
+			streamStats.incOpen();
+			streamStats.incOpenOutput();
+			streamStats.incCurrentOutput();
+			streamStats.incMaxOutput();
+			if (file.getType().isDirectory()){
+				streamStats.incOpenOutputDir();
+			}
+		}
+		return outputStream;
+	}		
+	
+	CoreInputStream getInputStream(CoreNode file, long readHint) throws Exception {
+		CoreInputStream inputStream = new CoreInputStream(file, streamCounter.incrementAndGet(), readHint);
+		openInputStreams.put(inputStream.getStreamId(), inputStream);
+		
+		if (CrailConstants.STATISTICS){
+			streamStats.incOpen();
+			streamStats.incOpenInput();
+			streamStats.incCurrentInput();
+			streamStats.incMaxInput();
+			if (file.getType().isDirectory()){
+				streamStats.incOpenInputDir();
+			}
+		}
+		return inputStream;
+	}
+	
+	CoreStream unregisterInputStream(CoreInputStream coreStream) {
+		CoreStream stream = this.openInputStreams.remove(coreStream.getStreamId());
+		if (stream != null && CrailConstants.STATISTICS){
+			streamStats.incClose();
+			streamStats.incCloseInput();
+			this.ioStatsIn.mergeStatistics(stream.getCoreStatistics());
+			streamStats.decCurrentInput();
+			if (stream.getFile().getType().isDirectory()){
+				streamStats.incCloseInputDir();
+			}
+		}
+		
+		return stream;
+	}	
+	
+	CoreStream unregisterOutputStream(CoreOutputStream coreStream) {
+		CoreStream stream = this.openOutputStreams.remove(coreStream.getStreamId());
+		if (stream != null && CrailConstants.STATISTICS){
+			streamStats.incClose();
+			streamStats.incCloseOutput();
+			this.ioStatsOut.mergeStatistics(stream.getCoreStatistics());
+			streamStats.decCurrentOutput();
+			if (stream.getFile().getType().isDirectory()){
+				streamStats.incCloseOutputDir();
+			}
+		}
+		
+		return stream;
+	}
+	
+	FileBlockCache getBlockCache(long fd){
+		return blockCache.getFileBlockCache(fd);
+	}
+	
+	FileNextBlockCache getNextBlockCache(long fd){
+		return nextBlockCache.getFileBlockCache(fd);
+	}	
+
+	RpcConnection getNamenodeClientRpc() {
+		return rpcConnection;
+	}
+
+	EndpointCache getDatanodeEndpointCache() {
+		return datanodeEndpointCache;
+	}
+	
+	String getMappedLocation(String hostname){
+		String mappedValue = locationMap.get(hostname);
+		return mappedValue != null ? mappedValue : hostname;
+	}
+	
+	CoreSyncOperation getSyncOperation(FileInfo dirInfo, FileInfo fileInfo, String path, boolean valid) throws Exception{
+		long adjustedCapacity = fileInfo.getDirOffset()*CrailConstants.DIRECTORY_RECORD + CrailConstants.DIRECTORY_RECORD;
+		dirInfo.setCapacity(Math.max(dirInfo.getCapacity(), adjustedCapacity));
+		CoreDirectory dirFile = new CoreDirectory(this, dirInfo, CrailUtils.getParent(path));
+		DirectoryOutputStream stream = dirFile.getDirectoryOutputStream();
+		DirectoryRecord record = new DirectoryRecord(valid, path);
+		Future<CrailResult> future = stream.writeRecord(record, fileInfo.getDirOffset());
+		CoreSyncOperation syncOperation = new CoreSyncOperation(stream, future);		
+		return syncOperation;
+	}
+}

http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/client/src/main/java/org/apache/crail/core/CoreIOStatistics.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/crail/core/CoreIOStatistics.java b/client/src/main/java/org/apache/crail/core/CoreIOStatistics.java
new file mode 100644
index 0000000..03d22e3
--- /dev/null
+++ b/client/src/main/java/org/apache/crail/core/CoreIOStatistics.java
@@ -0,0 +1,252 @@
+/*
+ * 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.core;
+
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.apache.crail.CrailStatistics;
+import org.apache.crail.CrailStatistics.StatisticsProvider;
+
+public class CoreIOStatistics implements CrailStatistics.StatisticsProvider {
+	private String mode;
+	private AtomicLong totalOps;
+	private AtomicLong localOps;
+	private AtomicLong remoteOps;
+	private AtomicLong localDirOps;
+	private AtomicLong remoteDirOps;
+	private AtomicLong cachedOps;
+	private AtomicLong nonblockingOps;
+	private AtomicLong blockingOps;
+	private AtomicLong prefetchedOps;
+	private AtomicLong prefetchedBlockingOps;
+	private AtomicLong prefetchedNonblockingOps;
+	private AtomicLong opLen;
+	private AtomicLong capacity;
+	private AtomicLong totalStreams;
+	private AtomicLong totalSeeks;
+	
+	public CoreIOStatistics(String mode){
+		this.mode = mode;
+		this.totalOps = new AtomicLong(0);
+		this.localOps = new AtomicLong(0);
+		this.remoteOps = new AtomicLong(0);
+		this.localDirOps = new AtomicLong(0);
+		this.remoteDirOps = new AtomicLong(0);
+		this.cachedOps = new AtomicLong(0);
+		this.nonblockingOps = new AtomicLong(0);
+		this.blockingOps = new AtomicLong(0);
+		this.prefetchedOps = new AtomicLong(0);
+		this.prefetchedBlockingOps = new AtomicLong(0);
+		this.prefetchedNonblockingOps = new AtomicLong(0);
+		this.opLen = new AtomicLong(0);
+		this.capacity = new AtomicLong(0);
+		this.totalStreams = new AtomicLong(0);
+		this.totalSeeks = new AtomicLong(0);
+		resetStatistics();
+	}
+	
+	@Override
+	public String providerName() {
+		return mode;
+	}
+
+	@Override
+	public String printStatistics() {
+		return "total " + getTotalOps() + ", localOps " + getLocalOps() + ", remoteOps " + getRemoteOps() + ", localDirOps " + getLocalDirOps() + ", remoteDirOps " + getRemoteDirOps() + 
+		", cached " + getCachedOps() + ", nonBlocking " + getNonblockingOps() + ", blocking " + getBlockingOps() +
+		", prefetched " + getPrefetchedOps() + ", prefetchedNonBlocking " + getPrefetchedNonblockingOps() + ", prefetchedBlocking " + getPrefetchedBlockingOps() +
+		", capacity " + getCapacity() + ", totalStreams " + getTotalStreams() + ", avgCapacity " + getAvgCapacity() +
+		", avgOpLen " + getAvgOpLen();
+	}	
+	
+	public void resetStatistics(){
+		this.totalOps.set(0);
+		this.localOps.set(0);
+		this.remoteOps.set(0);
+		this.localDirOps.set(0);
+		this.remoteDirOps.set(0);
+		this.cachedOps.set(0);
+		this.nonblockingOps.set(0);
+		this.blockingOps.set(0);
+		this.prefetchedOps.set(0);
+		this.prefetchedBlockingOps.set(0);
+		this.prefetchedNonblockingOps.set(0);
+		this.opLen.set(0);
+		this.capacity.set(0);
+		this.totalStreams.set(0);
+		this.totalSeeks.set(0);
+	}
+	
+	public void mergeStatistics(StatisticsProvider provider){
+		if (provider instanceof CoreIOStatistics){
+			CoreIOStatistics newProvider = (CoreIOStatistics) provider;
+			totalOps.addAndGet(newProvider.getTotalOps());
+			localOps.addAndGet(newProvider.getLocalOps());
+			remoteOps.addAndGet(newProvider.getRemoteOps());
+			localDirOps.addAndGet(newProvider.getLocalDirOps());
+			remoteDirOps.addAndGet(newProvider.getRemoteDirOps());
+			cachedOps.addAndGet(newProvider.getCachedOps());
+			nonblockingOps.addAndGet(newProvider.getNonblockingOps());
+			blockingOps.addAndGet(newProvider.getBlockingOps());
+			prefetchedOps.addAndGet(newProvider.getPrefetchedOps());
+			prefetchedNonblockingOps.addAndGet(newProvider.getPrefetchedNonblockingOps());
+			prefetchedBlockingOps.addAndGet(newProvider.getPrefetchedBlockingOps());
+			opLen.addAndGet(newProvider.getOpLen());
+			capacity.addAndGet(newProvider.getCapacity());
+			totalSeeks.addAndGet(newProvider.getTotalSeeks());
+			totalStreams.incrementAndGet();
+		}
+	}
+	
+	public long getTotalOps() {
+		return totalOps.get();
+	}
+	
+	void incTotalOps(long opLen) {
+		this.totalOps.incrementAndGet();
+		this.opLen.addAndGet(opLen);
+	}
+	
+	public long getLocalOps(){
+		return localOps.get();
+	}
+	
+	public void incLocalOps() {
+		localOps.incrementAndGet();
+	}	
+	
+	public long getRemoteOps(){
+		return remoteOps.get();
+	}
+	
+	public void incRemoteOps() {
+		remoteOps.incrementAndGet();
+	}	
+	
+	public long getLocalDirOps(){
+		return localDirOps.get();
+	}
+	
+	public void incLocalDirOps() {
+		localDirOps.incrementAndGet();
+	}	
+	
+	public long getRemoteDirOps(){
+		return remoteDirOps.get();
+	}
+	
+	public void incRemoteDirOps() {
+		remoteDirOps.incrementAndGet();
+	}	
+	
+	public long getCachedOps() {
+		return cachedOps.get();
+	}
+	
+	void incCachedOps() {
+		this.cachedOps.incrementAndGet();
+	}
+
+	public long getNonblockingOps() {
+		return nonblockingOps.get();
+	}
+	
+	void incNonblockingOps() {
+		this.nonblockingOps.incrementAndGet();
+	}	
+	
+	public long getBlockingOps() {
+		return blockingOps.get();
+	}
+	
+	void incBlockingOps() {
+		this.blockingOps.incrementAndGet();
+	}
+	
+	public long getPrefetchedOps() {
+		return prefetchedOps.get();
+	}
+	
+	void incPrefetchedOps() {
+		this.prefetchedOps.incrementAndGet();
+	}	
+	
+	public long getPrefetchedBlockingOps() {
+		return prefetchedBlockingOps.get();
+	}
+	
+	void incPrefetchedBlockingOps() {
+		this.prefetchedBlockingOps.incrementAndGet();
+	}	
+	
+	public long getPrefetchedNonblockingOps() {
+		return prefetchedNonblockingOps.get();
+	}
+	
+	void incPrefetchedNonblockingOps() {
+		this.prefetchedNonblockingOps.incrementAndGet();
+	}	
+	
+	public long getOpLen() {
+		return opLen.get();
+	}
+	
+	public long getAvgOpLen(){
+		long _avg = 0;
+		long _totalOps = totalOps.get();
+		long _opLen = opLen.get();
+		if (_totalOps > 0){
+			_avg = _opLen / _totalOps;
+		}
+		return _avg;
+	}
+	
+	public void setCapacity(long capacity){
+		this.capacity.set(capacity);
+	}
+	
+	public long getCapacity(){
+		return capacity.get();
+	}
+	
+	public long getAvgCapacity(){
+		long _avg = 0;
+		long _capacity = capacity.get();
+		long _totalStreams = totalStreams.get();
+		if (_totalStreams > 0){
+			_avg = _capacity / _totalStreams;
+		}
+		return _avg;
+	}
+	
+	public long getTotalStreams(){
+		return totalStreams.get();
+	}
+	
+	public long getTotalSeeks() {
+		return totalSeeks.get();
+	}
+	
+	void incTotalSeekds() {
+		this.totalSeeks.incrementAndGet();
+	}	
+}

http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/client/src/main/java/org/apache/crail/core/CoreInputStream.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/crail/core/CoreInputStream.java b/client/src/main/java/org/apache/crail/core/CoreInputStream.java
new file mode 100644
index 0000000..f80b7d8
--- /dev/null
+++ b/client/src/main/java/org/apache/crail/core/CoreInputStream.java
@@ -0,0 +1,132 @@
+/*
+ * 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.core;
+
+import java.io.IOException;
+import java.util.concurrent.Future;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.apache.crail.CrailBuffer;
+import org.apache.crail.CrailInputStream;
+import org.apache.crail.CrailResult;
+import org.apache.crail.conf.CrailConstants;
+import org.apache.crail.metadata.BlockInfo;
+import org.apache.crail.storage.StorageEndpoint;
+import org.apache.crail.storage.StorageFuture;
+import org.apache.crail.utils.CrailImmediateOperation;
+import org.apache.crail.utils.CrailUtils;
+import org.slf4j.Logger;
+
+public class CoreInputStream extends CoreStream implements CrailInputStream { 
+	private static final Logger LOG = CrailUtils.getLogger();
+	private AtomicLong inFlight;
+	private long readHint;
+	private CrailImmediateOperation noOp;
+	private boolean open;
+	
+	public CoreInputStream(CoreNode file, long streamId, long readHint) throws Exception {
+		super(file, streamId, 0);
+		this.inFlight = new AtomicLong(0);
+		this.readHint = Math.max(0, Math.min(file.getCapacity(), readHint));
+		this.noOp = new CrailImmediateOperation(0);
+		this.open = true;
+		if (CrailConstants.DEBUG){
+			LOG.info("CoreInputStream: open, path  " + file.getPath() + ", fd " + file.getFd() + ", streamId " + streamId + ", isDir " + file.getType().isDirectory() + ", readHint " + this.readHint);
+		}
+	}
+	
+	final public Future<CrailResult> read(CrailBuffer dataBuf) throws Exception {
+		if (!open) {
+			throw new IOException("stream already closed");
+		}
+		if (dataBuf.remaining() <= 0) {
+			return noOp;
+		}
+		if (position() >= getFile().getCapacity()) {
+			return null;
+		}
+		
+		long fileAvailable = available();
+		long bufAvailable = (long) dataBuf.remaining();
+		if (fileAvailable < bufAvailable){
+			int _fileAvailable = (int) fileAvailable;
+			dataBuf.limit(dataBuf.position() + _fileAvailable);
+		}
+		
+		inFlight.incrementAndGet();
+		CoreDataOperation future = dataOperation(dataBuf);
+		if (position() < readHint){
+			prefetchMetadata();
+		}	
+		if (future.isSynchronous()){
+			future.get();
+		}
+		return future;
+	}
+	
+	final public int available() {
+		long available = Math.max(0, getFile().getCapacity() - position());
+		long maxint = (long) Integer.MAX_VALUE;
+		if (available < maxint){
+			return (int) available;
+		} else {
+			return Integer.MAX_VALUE;
+		}		
+	}	
+	
+	final public void seek(long pos) throws IOException {
+		long oldPos = position();
+		super.seek(pos);
+		long newPos = position();
+		if (oldPos != newPos){
+			this.readHint = 0;
+		}
+	}
+	
+	public void close() throws Exception {
+		if (!open){
+			return;
+		}
+		if (inFlight.get() != 0){
+			LOG.info("Cannot close, pending operations, opcount " + inFlight.get() + ", path " + getFile().getPath());
+			throw new IOException("Cannot close, pending operations, opcount " + inFlight.get());
+		}
+		
+		updateIOStats();
+		node.closeInputStream(this);
+		open = false;
+		if (CrailConstants.DEBUG){
+			LOG.info("CoreInputStream, close, path " + this.getFile().getPath() + ", fd " + getFile().getFd() + ", streamId " + getStreamId());
+		}	
+	}
+	
+	// --------------------------
+	
+	StorageFuture trigger(StorageEndpoint endpoint, CoreSubOperation opDesc, CrailBuffer buffer, BlockInfo block) throws Exception {
+		StorageFuture future = endpoint.read(buffer, block, opDesc.getBlockOffset());
+		return future;
+	}	
+	
+	void update(long newCapacity) {
+		inFlight.decrementAndGet();
+	}
+}

http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/client/src/main/java/org/apache/crail/core/CoreMetaDataOperation.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/crail/core/CoreMetaDataOperation.java b/client/src/main/java/org/apache/crail/core/CoreMetaDataOperation.java
new file mode 100644
index 0000000..0a4babd
--- /dev/null
+++ b/client/src/main/java/org/apache/crail/core/CoreMetaDataOperation.java
@@ -0,0 +1,279 @@
+/*
+ * 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.core;
+
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.crail.CrailNode;
+import org.apache.crail.CrailNodeType;
+import org.apache.crail.Upcoming;
+import org.apache.crail.conf.CrailConstants;
+import org.apache.crail.rpc.RpcCreateFile;
+import org.apache.crail.rpc.RpcDeleteFile;
+import org.apache.crail.rpc.RpcErrors;
+import org.apache.crail.rpc.RpcGetFile;
+import org.apache.crail.rpc.RpcRenameFile;
+import org.apache.crail.rpc.RpcVoid;
+
+public abstract class CoreMetaDataOperation<R,T> implements Upcoming<T> {
+	protected static int RPC_PENDING = 0;
+	protected static int RPC_DONE = 1;
+	protected static int RPC_ERROR = 2;		
+	
+	private AtomicInteger status;
+	protected Future<R> rpcResult;
+	private T finalResult;
+	private Exception exception;
+	
+	abstract T process(R tmp) throws Exception;
+	
+	public CoreMetaDataOperation(Future<R> result){
+		this.rpcResult = result;
+		this.finalResult = null;
+		this.status = new AtomicInteger(RPC_PENDING);
+		this.exception = null;
+	}
+
+	@Override
+	public boolean isDone() {
+		if (status.get() == RPC_PENDING){
+			try {
+				if (rpcResult.isDone()){
+					R tmp = rpcResult.get();
+					finalResult = process(tmp);
+					status.set(RPC_DONE);
+				}
+			} catch (Exception e) {
+				status.set(RPC_ERROR);
+				this.exception = e;
+			}
+		}
+		
+		return status.get() > 0;
+	}
+
+	@Override
+	public T get() throws InterruptedException, ExecutionException {
+		if (this.exception != null){
+			throw new ExecutionException(exception);
+		}
+		
+		if (status.get() == RPC_PENDING){
+			try {
+				R tmp = rpcResult.get();
+				finalResult = process(tmp);
+				status.set(RPC_DONE);
+			} catch (Exception e) {
+				status.set(RPC_ERROR);
+				this.exception = e;
+			}
+		}	
+		
+		if (status.get() == RPC_DONE){
+			return finalResult;
+		} else if (status.get() == RPC_PENDING){
+			throw new InterruptedException("RPC timeout");
+		} else if (exception != null) {
+			throw new ExecutionException(exception);
+		} else {
+			throw new InterruptedException("RPC error");
+		}	
+	}
+
+	@Override
+	public T get(long timeout, TimeUnit unit) throws InterruptedException,
+			ExecutionException, TimeoutException {
+		if (this.exception != null){
+			throw new ExecutionException(exception);
+		}		
+		
+		if (status.get() == RPC_PENDING){
+			try {
+				R tmp = rpcResult.get(CrailConstants.DATA_TIMEOUT, TimeUnit.MILLISECONDS);
+				finalResult = process(tmp);
+				status.set(RPC_DONE);
+			} catch (Exception e) {
+				status.set(RPC_ERROR);
+				this.exception = e;
+			}
+		}	
+		
+		if (status.get() == RPC_DONE){
+			return finalResult;
+		} else if (status.get() == RPC_PENDING){
+			throw new InterruptedException("RPC timeout");
+		} else if (exception != null) {
+			throw new ExecutionException(exception);
+		} else {
+			throw new InterruptedException("RPC error");
+		}
+	}
+	
+	public T early() throws Exception {
+		return this.get();
+	}
+	
+	@Override
+	public boolean cancel(boolean mayInterruptIfRunning) {
+		return false;
+	}
+
+	@Override
+	public boolean isCancelled() {
+		return false;
+	}	
+}
+
+class CreateNodeFuture extends CoreMetaDataOperation<RpcCreateFile, CrailNode> {
+	private CoreFileSystem fs;
+	private String path;
+	private CrailNodeType type;
+
+	public CreateNodeFuture(CoreFileSystem fs, String path, CrailNodeType type, Future<RpcCreateFile> fileRes) {
+		super(fileRes);
+		this.fs = fs;
+		this.path = path;
+		this.type = type;
+	}
+
+	@Override
+	CrailNode process(RpcCreateFile response) throws Exception {
+		return fs._createNode(path, type, response);
+	}
+
+	@Override
+	public CrailNode early() throws Exception {
+		switch(type){
+		case DATAFILE:
+			return new CoreEarlyFile(fs, path, type, this);
+		case DIRECTORY:
+		case MULTIFILE:
+			return null;
+		default:
+			return super.early();
+		}
+	}
+
+}
+
+class LookupNodeFuture extends CoreMetaDataOperation<RpcGetFile, CrailNode> {
+	private String path;
+	private CoreFileSystem fs;	
+
+	public LookupNodeFuture(CoreFileSystem fs, String path, Future<RpcGetFile> fileRes) {
+		super(fileRes);
+		this.fs = fs;
+		this.path = path;
+	}
+
+	@Override
+	CrailNode process(RpcGetFile tmp) throws Exception {
+		return fs._lookupNode(tmp, path);
+	}
+
+}
+
+class DeleteNodeFuture extends CoreMetaDataOperation<RpcDeleteFile, CrailNode> {
+	private String path;
+	private boolean recursive;
+	private CoreFileSystem fs;
+
+	public DeleteNodeFuture(CoreFileSystem fs, String path, boolean recursive, Future<RpcDeleteFile> fileRes) {
+		super(fileRes);
+		this.fs = fs;
+		this.path = path;
+		this.recursive = recursive;
+	}
+
+	@Override
+	CrailNode process(RpcDeleteFile tmp) throws Exception {
+		return fs._delete(tmp, path, recursive);
+	}
+}
+
+class RenameNodeFuture extends CoreMetaDataOperation<RpcRenameFile, CrailNode> {
+	private String src;
+	private String dst;
+	private CoreFileSystem fs;
+
+	public RenameNodeFuture(CoreFileSystem fs, String src, String dst, Future<RpcRenameFile> fileRes) {
+		super(fileRes);
+		this.fs = fs;
+		this.src = src;
+		this.dst = dst;
+	}
+
+	@Override
+	CrailNode process(RpcRenameFile tmp) throws Exception {
+		return fs._rename(tmp, src, dst);
+	}
+}
+
+class SyncNodeFuture extends CoreMetaDataOperation<RpcVoid, Void> {
+
+	public SyncNodeFuture(Future<RpcVoid> fileRes) {
+		super(fileRes);
+	}
+	
+	@Override
+	Void process(RpcVoid tmp) throws Exception {
+		if (tmp.getError() != RpcErrors.ERR_OK){
+			throw new Exception("sync: " + RpcErrors.messages[tmp.getError()]);
+		}
+		return null;
+	}
+}
+
+class NoOperation implements Future<Void> {
+
+	@Override
+	public boolean cancel(boolean mayInterruptIfRunning) {
+		return false;
+	}
+
+	@Override
+	public boolean isCancelled() {
+		return false;
+	}
+
+	@Override
+	public boolean isDone() {
+		return true;
+	}
+
+	@Override
+	public Void get() throws InterruptedException, ExecutionException {
+		return null;
+	}
+
+	@Override
+	public Void get(long timeout, TimeUnit unit) throws InterruptedException,
+			ExecutionException, TimeoutException {
+		return null;
+	}
+}
+
+

http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/client/src/main/java/org/apache/crail/core/CoreNode.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/crail/core/CoreNode.java b/client/src/main/java/org/apache/crail/core/CoreNode.java
new file mode 100644
index 0000000..57cc086
--- /dev/null
+++ b/client/src/main/java/org/apache/crail/core/CoreNode.java
@@ -0,0 +1,143 @@
+/*
+ * 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.core;
+
+import java.util.concurrent.LinkedBlockingQueue;
+
+import org.apache.crail.CrailBlockLocation;
+import org.apache.crail.CrailKeyValue;
+import org.apache.crail.CrailMultiFile;
+import org.apache.crail.CrailNode;
+import org.apache.crail.CrailNodeType;
+import org.apache.crail.CrailTable;
+import org.apache.crail.metadata.FileInfo;
+
+public class CoreNode implements CrailNode {
+	protected CoreFileSystem fs;
+	protected FileInfo fileInfo;
+	protected String path;
+	private LinkedBlockingQueue<CoreSyncOperation> syncOperations;
+	
+	public static CoreNode create(CoreFileSystem fs, FileInfo fileInfo, String path) {
+		if (fileInfo.getType().isContainer()){
+			return new CoreDirectory(fs, fileInfo, path);		
+		} else {
+			return new CoreFile(fs, fileInfo, path);
+		}
+	}	
+	
+	protected CoreNode(CoreFileSystem fs, FileInfo fileInfo, String path){
+		this.fs = fs;
+		this.fileInfo = fileInfo;
+		this.path = path;
+		this.syncOperations = new LinkedBlockingQueue<CoreSyncOperation>();
+	}	
+
+	@Override
+	public CoreFileSystem getFileSystem() {
+		return fs;
+	}
+
+	@Override
+	public String getPath() {
+		return path;
+	}
+	
+	public long getModificationTime() {
+		return fileInfo.getModificationTime();
+	}	
+	
+	public long getCapacity() {
+		return fileInfo.getCapacity();
+	}
+	
+	public CrailNodeType getType() {
+		return fileInfo.getType();
+	}
+	
+	public long getFd() {
+		return fileInfo.getFd();
+	}	
+
+	@Override
+	public CoreNode syncDir() throws Exception {
+		while(!syncOperations.isEmpty()){
+			CoreSyncOperation syncOp = syncOperations.poll();
+			syncOp.close();
+		}			
+		return this;
+	}
+	
+	public CoreFile asFile() throws Exception {
+		throw new Exception("Type of file unclear");
+	}
+	
+	public CoreDirectory asContainer() throws Exception {
+		throw new Exception("Type of file unclear");
+	}	
+	
+	public CoreDirectory asDirectory() throws Exception {
+		throw new Exception("Type of file unclear");
+	}	
+	
+	public CrailMultiFile asMultiFile() throws Exception {
+		throw new Exception("Type of file unclear");
+	}
+	
+	public CrailTable asTable() throws Exception {
+		throw new Exception("Type of file unclear");
+	}	
+	
+	public CrailKeyValue asKeyValue() throws Exception {
+		throw new Exception("Type of file unclear");
+	}	
+	
+	public CrailBlockLocation[] getBlockLocations(long start, long len) throws Exception {
+		return fs.getBlockLocations(path, start, len);
+	}	
+	
+	protected CoreInputStream getInputStream(long readHint) throws Exception{
+		return fs.getInputStream(this, readHint);
+	}	
+	
+	CoreOutputStream getOutputStream(long writeHint) throws Exception {
+		return fs.getOutputStream(this, writeHint);
+	}	
+
+	void closeInputStream(CoreInputStream coreStream) throws Exception {
+		fs.unregisterInputStream(coreStream);
+	}
+	
+	void closeOutputStream(CoreOutputStream coreStream) throws Exception {
+		syncDir();
+		fs.unregisterOutputStream(coreStream);
+	}	
+	
+	FileInfo getFileInfo(){
+		return fileInfo;
+	}	
+	
+	void addSyncOperation(CoreSyncOperation operation){
+		this.syncOperations.add(operation);
+	}
+}
+

http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/client/src/main/java/org/apache/crail/core/CoreOutputStream.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/crail/core/CoreOutputStream.java b/client/src/main/java/org/apache/crail/core/CoreOutputStream.java
new file mode 100644
index 0000000..6772780
--- /dev/null
+++ b/client/src/main/java/org/apache/crail/core/CoreOutputStream.java
@@ -0,0 +1,117 @@
+/*
+ * 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.core;
+
+import java.io.IOException;
+import java.util.concurrent.Future;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.apache.crail.CrailBuffer;
+import org.apache.crail.CrailOutputStream;
+import org.apache.crail.CrailResult;
+import org.apache.crail.conf.CrailConstants;
+import org.apache.crail.metadata.BlockInfo;
+import org.apache.crail.storage.StorageEndpoint;
+import org.apache.crail.storage.StorageFuture;
+import org.apache.crail.utils.CrailImmediateOperation;
+import org.apache.crail.utils.CrailUtils;
+import org.slf4j.Logger;
+
+public class CoreOutputStream extends CoreStream implements CrailOutputStream {
+	private static final Logger LOG = CrailUtils.getLogger();
+	private AtomicLong inFlight;
+	private long writeHint;
+	private CrailImmediateOperation noOp;
+	private boolean open;
+	
+	public CoreOutputStream(CoreNode file, long streamId, long writeHint) throws Exception {
+		super(file, streamId, file.getCapacity());
+		this.writeHint = Math.max(0, writeHint);
+		this.inFlight = new AtomicLong(0);
+		this.noOp = new CrailImmediateOperation(0);
+		this.open = true;
+		if (CrailConstants.DEBUG){
+			LOG.info("CoreOutputStream, open, path " + file.getPath() + ", fd " + file.getFd() + ", streamId " + streamId + ", isDir " + file.getType().isDirectory() + ", writeHint " + this.writeHint);
+		}
+	}
+	
+	final public Future<CrailResult> write(CrailBuffer dataBuf) throws Exception {
+		if (!open) {
+			throw new IOException("Stream closed, cannot write");
+		}
+		if (dataBuf.remaining() <= 0) {
+			return noOp;
+		}
+		
+		inFlight.incrementAndGet();
+		CoreDataOperation future = dataOperation(dataBuf);
+		if (position() < writeHint){
+			prefetchMetadata();
+		} 	
+		if (future.isSynchronous()){
+			future.get();
+		}		
+		return future;
+	}	
+	
+	final public long getWriteHint() {
+		return this.writeHint;
+	}
+	
+	public Future<Void> sync() throws IOException {
+		if (inFlight.get() != 0){
+			LOG.info("Cannot sync, pending operations, opcount " + inFlight.get());
+			throw new IOException("Cannot close, pending operations, opcount " + inFlight.get());
+		}		
+		return super.sync();
+	}
+	
+	public void close() throws Exception {
+		if (!open){
+			return;
+		}
+		if (inFlight.get() != 0){
+			LOG.info("Cannot close, pending operations, opcount " + inFlight.get() + ", path " + getFile().getPath());
+			throw new IOException("Cannot close, pending operations, opcount " + inFlight.get() + ", fd " + getFile().getFd() + ", streamId " + getStreamId() + ", capacity " + getFile().getCapacity());
+		}
+		
+		sync().get();
+		updateIOStats();
+		node.closeOutputStream(this);
+		open = false;
+		if (CrailConstants.DEBUG){
+			LOG.info("CoreOutputStream, close, path " + this.getFile().getPath() + ", fd " + getFile().getFd() + ", streamId " + getStreamId() + ", capacity " + getFile().getCapacity());
+		}	
+	}
+	
+	// ----------------------
+	
+	StorageFuture trigger(StorageEndpoint endpoint, CoreSubOperation opDesc, CrailBuffer buffer, BlockInfo block) throws Exception {
+		StorageFuture dataFuture = endpoint.write(buffer, block, opDesc.getBlockOffset());
+		return dataFuture;		
+	}	
+	
+	synchronized void update(long newCapacity) {
+		inFlight.decrementAndGet();
+		setCapacity(newCapacity);
+	}
+}