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:53 UTC
[09/29] incubator-crail git commit: Apache package renaming
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/namenode/src/main/java/com/ibm/crail/namenode/LogResponse.java
----------------------------------------------------------------------
diff --git a/namenode/src/main/java/com/ibm/crail/namenode/LogResponse.java b/namenode/src/main/java/com/ibm/crail/namenode/LogResponse.java
deleted file mode 100644
index cd0a4a2..0000000
--- a/namenode/src/main/java/com/ibm/crail/namenode/LogResponse.java
+++ /dev/null
@@ -1,268 +0,0 @@
-package com.ibm.crail.namenode;
-
-import java.nio.ByteBuffer;
-
-import com.ibm.crail.rpc.RpcNameNodeState;
-import com.ibm.crail.rpc.RpcProtocol;
-import com.ibm.crail.rpc.RpcResponseMessage;
-
-public class LogResponse implements RpcNameNodeState {
- public static final int CSIZE = 4 + Math.max(RpcResponseMessage.GetBlockRes.CSIZE, RpcResponseMessage.RenameRes.CSIZE);
-
- private short type;
- private short error;
- private RpcResponseMessage.VoidRes voidRes;
- private RpcResponseMessage.CreateFileRes createFileRes;
- private RpcResponseMessage.GetFileRes getFileRes;
- private RpcResponseMessage.DeleteFileRes delFileRes;
- private RpcResponseMessage.RenameRes renameRes;
- private RpcResponseMessage.GetBlockRes getBlockRes;
- private RpcResponseMessage.GetLocationRes getLocationRes;
- private RpcResponseMessage.GetDataNodeRes getDataNodeRes;
- private RpcResponseMessage.PingNameNodeRes pingNameNodeRes;
-
- public LogResponse() {
- this.type = 0;
- this.error = 0;
-
- this.voidRes = new RpcResponseMessage.VoidRes();
- this.createFileRes = new RpcResponseMessage.CreateFileRes();
- this.getFileRes = new RpcResponseMessage.GetFileRes();
- this.delFileRes = new RpcResponseMessage.DeleteFileRes();
- this.renameRes = new RpcResponseMessage.RenameRes();
- this.getBlockRes = new RpcResponseMessage.GetBlockRes();
- this.getLocationRes = new RpcResponseMessage.GetLocationRes();
- this.getDataNodeRes = new RpcResponseMessage.GetDataNodeRes();
- this.pingNameNodeRes = new RpcResponseMessage.PingNameNodeRes();
- }
-
- public LogResponse(RpcResponseMessage.VoidRes message) {
- this.type = message.getType();
- this.voidRes = message;
- }
-
- public LogResponse(RpcResponseMessage.CreateFileRes message) {
- this.type = message.getType();
- this.createFileRes = message;
- }
-
- public LogResponse(RpcResponseMessage.GetFileRes message) {
- this.type = message.getType();
- this.getFileRes = message;
- }
-
- public LogResponse(RpcResponseMessage.DeleteFileRes message) {
- this.type = message.getType();
- this.delFileRes = message;
- }
-
- public LogResponse(RpcResponseMessage.RenameRes message) {
- this.type = message.getType();
- this.renameRes = message;
- }
-
- public LogResponse(RpcResponseMessage.GetBlockRes message) {
- this.type = message.getType();
- this.getBlockRes = message;
- }
-
- public LogResponse(RpcResponseMessage.GetLocationRes message) {
- this.type = message.getType();
- this.getLocationRes = message;
- }
-
- public LogResponse(RpcResponseMessage.GetDataNodeRes message) {
- this.type = message.getType();
- this.getDataNodeRes = message;
- }
-
- public LogResponse(RpcResponseMessage.PingNameNodeRes message) {
- this.type = message.getType();
- this.pingNameNodeRes = message;
- }
-
- public void setType(short type) throws Exception {
- this.type = type;
- switch(type){
- case RpcProtocol.RES_VOID:
- if (voidRes == null){
- throw new Exception("Response type not set");
- }
- break;
- case RpcProtocol.RES_CREATE_FILE:
- if (createFileRes == null){
- throw new Exception("Response type not set");
- }
- break;
- case RpcProtocol.RES_GET_FILE:
- if (getFileRes == null){
- throw new Exception("Response type not set");
- }
- break;
- case RpcProtocol.RES_DELETE_FILE:
- if (delFileRes == null){
- throw new Exception("Response type not set");
- }
- break;
- case RpcProtocol.RES_RENAME_FILE:
- if (renameRes == null){
- throw new Exception("Response type not set");
- }
- break;
- case RpcProtocol.RES_GET_BLOCK:
- if (getBlockRes == null){
- throw new Exception("Response type not set");
- }
- break;
- case RpcProtocol.RES_GET_LOCATION:
- if (getLocationRes == null){
- throw new Exception("Response type not set");
- }
- break;
- case RpcProtocol.RES_GET_DATANODE:
- if (getDataNodeRes == null){
- throw new Exception("Response type not set");
- }
- break;
- case RpcProtocol.RES_PING_NAMENODE:
- if (pingNameNodeRes == null){
- throw new Exception("Response type not set");
- }
- break;
- }
- }
-
- public int size(){
- return CSIZE;
- }
-
- public int write(ByteBuffer buffer){
- buffer.putShort(type);
- buffer.putShort(error);
-
- int written = 4;
- switch(type){
- case RpcProtocol.RES_VOID:
- written += voidRes.write(buffer);
- break;
- case RpcProtocol.RES_CREATE_FILE:
- written += createFileRes.write(buffer);
- break;
- case RpcProtocol.RES_GET_FILE:
- written += getFileRes.write(buffer);
- break;
- case RpcProtocol.RES_DELETE_FILE:
- written += delFileRes.write(buffer);
- break;
- case RpcProtocol.RES_RENAME_FILE:
- written += renameRes.write(buffer);
- break;
- case RpcProtocol.RES_GET_BLOCK:
- written += getBlockRes.write(buffer);
- break;
- case RpcProtocol.RES_GET_LOCATION:
- written += getLocationRes.write(buffer);
- break;
- case RpcProtocol.RES_GET_DATANODE:
- written += getDataNodeRes.write(buffer);
- break;
- case RpcProtocol.RES_PING_NAMENODE:
- written += pingNameNodeRes.write(buffer);
- break;
- }
-
- return written;
- }
-
- public void update(ByteBuffer buffer){
- this.type = buffer.getShort();
- this.error = buffer.getShort();
-
- switch(type){
- case RpcProtocol.RES_VOID:
- voidRes.update(buffer);
- voidRes.setError(error);
- break;
- case RpcProtocol.RES_CREATE_FILE:
- createFileRes.update(buffer);
- createFileRes.setError(error);
- break;
- case RpcProtocol.RES_GET_FILE:
- getFileRes.update(buffer);
- getFileRes.setError(error);
- break;
- case RpcProtocol.RES_DELETE_FILE:
- delFileRes.update(buffer);
- delFileRes.setError(error);
- break;
- case RpcProtocol.RES_RENAME_FILE:
- renameRes.update(buffer);
- renameRes.setError(error);
- break;
- case RpcProtocol.RES_GET_BLOCK:
- getBlockRes.update(buffer);
- getBlockRes.setError(error);
- break;
- case RpcProtocol.RES_GET_LOCATION:
- getLocationRes.update(buffer);
- getLocationRes.setError(error);
- break;
- case RpcProtocol.RES_GET_DATANODE:
- getDataNodeRes.update(buffer);
- getDataNodeRes.setError(error);
- break;
- case RpcProtocol.RES_PING_NAMENODE:
- pingNameNodeRes.update(buffer);
- pingNameNodeRes.setError(error);
- break;
- }
- }
-
- public short getType(){
- return type;
- }
-
- public short getError() {
- return error;
- }
-
- public void setError(short error) {
- this.error = error;
- }
-
- public RpcResponseMessage.VoidRes getVoid() {
- return voidRes;
- }
-
- public RpcResponseMessage.CreateFileRes createFile() {
- return createFileRes;
- }
-
- public RpcResponseMessage.GetFileRes getFile() {
- return getFileRes;
- }
-
- public RpcResponseMessage.DeleteFileRes delFile() {
- return delFileRes;
- }
-
- public RpcResponseMessage.RenameRes getRename() {
- return renameRes;
- }
-
- public RpcResponseMessage.GetBlockRes getBlock() {
- return getBlockRes;
- }
-
- public RpcResponseMessage.GetLocationRes getLocation() {
- return getLocationRes;
- }
-
- public RpcResponseMessage.GetDataNodeRes getDataNode() {
- return getDataNodeRes;
- }
-
- public RpcResponseMessage.PingNameNodeRes pingNameNode(){
- return this.pingNameNodeRes;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/namenode/src/main/java/com/ibm/crail/namenode/LogService.java
----------------------------------------------------------------------
diff --git a/namenode/src/main/java/com/ibm/crail/namenode/LogService.java b/namenode/src/main/java/com/ibm/crail/namenode/LogService.java
deleted file mode 100644
index 128da94..0000000
--- a/namenode/src/main/java/com/ibm/crail/namenode/LogService.java
+++ /dev/null
@@ -1,131 +0,0 @@
-package com.ibm.crail.namenode;
-
-import java.io.File;
-import java.io.FileInputStream;
-import java.io.FileOutputStream;
-import java.io.IOException;
-import java.nio.ByteBuffer;
-import java.nio.channels.FileChannel;
-import java.util.concurrent.ConcurrentHashMap;
-
-import org.slf4j.Logger;
-
-import com.ibm.crail.conf.CrailConstants;
-import com.ibm.crail.rpc.RpcErrors;
-import com.ibm.crail.rpc.RpcNameNodeService;
-import com.ibm.crail.rpc.RpcProtocol;
-import com.ibm.crail.utils.CrailUtils;
-
-public class LogService {
- public static final Logger LOG = CrailUtils.getLogger();
-
- private ConcurrentHashMap<Long, Long> tokens;
- private FileOutputStream outStream;
- private FileChannel outChannel;
- private ByteBuffer header;
- private ByteBuffer payload;
-
- public LogService() throws IOException {
- File file = new File(CrailConstants.NAMENODE_LOG);
- if (!file.exists()){
- file.createNewFile();
- }
- outStream = new FileOutputStream(CrailConstants.NAMENODE_LOG, true);
- outChannel = outStream.getChannel();
- header = ByteBuffer.allocate(4);
- payload = ByteBuffer.allocate(512);
- tokens = new ConcurrentHashMap<Long, Long>();
- }
-
- public void writeRecord(LogRecord record) throws IOException{
- payload.clear();
- record.write(payload);
- payload.flip();
-
- header.clear();
- header.putInt(payload.remaining());
- header.flip();
-
- outChannel.write(header);
- while(payload.hasRemaining()){
- outChannel.write(payload);
- }
- }
-
- public void replay(RpcNameNodeService service) throws Exception {
- File file = new File(CrailConstants.NAMENODE_LOG);
- if (!file.exists()){
- return;
- }
-
- FileInputStream inStream = new FileInputStream(CrailConstants.NAMENODE_LOG);
- FileChannel inChannel = inStream.getChannel();
- LogRecord record = new LogRecord();
- LogResponse response = new LogResponse();
-
- header.clear();
- int ret = inChannel.read(header);
- while(ret > 0){
- header.flip();
- int size = header.getInt();
-
- payload.clear().limit(size);
- while(payload.hasRemaining()){
- inChannel.read(payload);
- }
- payload.flip();
- record.update(payload);
- processServerEvent(service, record, response);
-
- header.clear();
- ret = inChannel.read(header);
- }
-
- inChannel.close();
- inStream.close();
- }
-
- public void close() throws IOException{
- outChannel.close();
- outStream.close();
- }
-
- private void processServerEvent(RpcNameNodeService service, LogRecord record, LogResponse response) {
- short error = RpcErrors.ERR_OK;
- try {
- switch(record.getCmd()) {
- case RpcProtocol.CMD_CREATE_FILE:
- error = service.createFile(record.createFile(), response.createFile(), response);
- long fd = response.createFile().getFile().getFd();
- long token = response.createFile().getFile().getToken();
- tokens.put(response.createFile().getFile().getFd(), response.createFile().getFile().getToken());
- tokens.put(response.createFile().getParent().getFd(), response.createFile().getParent().getToken());
- break;
- case RpcProtocol.CMD_SET_FILE:
- record.setFile().getFileInfo().setToken(tokens.get(record.setFile().getFileInfo().getFd()));
- error = service.setFile(record.setFile(), response.getVoid(), response);
- break;
- case RpcProtocol.CMD_REMOVE_FILE:
- error = service.removeFile(record.removeFile(), response.delFile(), response);
- break;
- case RpcProtocol.CMD_RENAME_FILE:
- error = service.renameFile(record.renameFile(), response.getRename(), response);
- break;
- case RpcProtocol.CMD_GET_BLOCK:
- record.getBlock().setToken(tokens.get(record.getBlock().getFd()));
- error = service.getBlock(record.getBlock(), response.getBlock(), response);
- break;
- case RpcProtocol.CMD_SET_BLOCK:
- error = service.setBlock(record.setBlock(), response.getVoid(), response);
- break;
- default:
- error = RpcErrors.ERR_INVALID_RPC_CMD;
- LOG.info("Rpc command not valid, opcode " + record.getCmd());
- }
- } catch(Exception e){
- error = RpcErrors.ERR_UNKNOWN;
- LOG.info(RpcErrors.messages[RpcErrors.ERR_UNKNOWN] + e.getMessage());
- e.printStackTrace();
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/namenode/src/main/java/com/ibm/crail/namenode/MultiFileBlocks.java
----------------------------------------------------------------------
diff --git a/namenode/src/main/java/com/ibm/crail/namenode/MultiFileBlocks.java b/namenode/src/main/java/com/ibm/crail/namenode/MultiFileBlocks.java
deleted file mode 100644
index 57f7fae..0000000
--- a/namenode/src/main/java/com/ibm/crail/namenode/MultiFileBlocks.java
+++ /dev/null
@@ -1,12 +0,0 @@
-package com.ibm.crail.namenode;
-
-import com.ibm.crail.CrailNodeType;
-
-public class MultiFileBlocks extends DirectoryBlocks {
-
- MultiFileBlocks(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/NameNode.java
----------------------------------------------------------------------
diff --git a/namenode/src/main/java/com/ibm/crail/namenode/NameNode.java b/namenode/src/main/java/com/ibm/crail/namenode/NameNode.java
deleted file mode 100644
index 3d5555a..0000000
--- a/namenode/src/main/java/com/ibm/crail/namenode/NameNode.java
+++ /dev/null
@@ -1,101 +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.URI;
-import java.util.Arrays;
-
-import org.apache.commons.cli.CommandLine;
-import org.apache.commons.cli.CommandLineParser;
-import org.apache.commons.cli.DefaultParser;
-import org.apache.commons.cli.HelpFormatter;
-import org.apache.commons.cli.Option;
-import org.apache.commons.cli.Options;
-import org.apache.commons.cli.ParseException;
-import org.slf4j.Logger;
-
-import com.ibm.crail.conf.CrailConfiguration;
-import com.ibm.crail.conf.CrailConstants;
-import com.ibm.crail.rpc.RpcBinding;
-import com.ibm.crail.rpc.RpcNameNodeService;
-import com.ibm.crail.rpc.RpcServer;
-import com.ibm.crail.utils.CrailUtils;
-
-public class NameNode {
- private static final Logger LOG = CrailUtils.getLogger();
-
- public static void main(String args[]) throws Exception {
- LOG.info("initalizing namenode ");
- CrailConfiguration conf = new CrailConfiguration();
- CrailConstants.updateConstants(conf);
-
- URI uri = CrailUtils.getPrimaryNameNode();
- String address = uri.getHost();
- int port = uri.getPort();
-
- if (args != null) {
- Option addressOption = Option.builder("a").desc("ip address namenode is started on").hasArg().build();
- Option portOption = Option.builder("p").desc("port namenode is started on").hasArg().build();
- Options options = new Options();
- options.addOption(portOption);
- options.addOption(addressOption);
- CommandLineParser parser = new DefaultParser();
-
- try {
- CommandLine line = parser.parse(options, Arrays.copyOfRange(args, 0, args.length));
- if (line.hasOption(addressOption.getOpt())) {
- address = line.getOptionValue(addressOption.getOpt());
- }
- if (line.hasOption(portOption.getOpt())) {
- port = Integer.parseInt(line.getOptionValue(portOption.getOpt()));
- }
- } catch (ParseException e) {
- HelpFormatter formatter = new HelpFormatter();
- formatter.printHelp("Namenode", options);
- System.exit(-1);
- }
- }
-
- String namenode = "crail://" + address + ":" + port;
- long serviceId = CrailUtils.getServiceId(namenode);
- long serviceSize = CrailUtils.getServiceSize();
- if (!CrailUtils.verifyNamenode(namenode)){
- throw new Exception("Namenode address/port [" + namenode + "] has to be listed in crail.namenode.address " + CrailConstants.NAMENODE_ADDRESS);
- }
-
- CrailConstants.NAMENODE_ADDRESS = namenode + "?id=" + serviceId + "&size=" + serviceSize;
- CrailConstants.printConf();
- CrailConstants.verify();
-
- RpcNameNodeService service = RpcNameNodeService.createInstance(CrailConstants.NAMENODE_RPC_SERVICE);
- if (!CrailConstants.NAMENODE_LOG.isEmpty()){
- LogDispatcher logDispatcher = new LogDispatcher(service);
- service = logDispatcher;
- }
- RpcBinding rpcBinding = RpcBinding.createInstance(CrailConstants.NAMENODE_RPC_TYPE);
- RpcServer rpcServer = rpcBinding.launchServer(service);
- rpcServer.init(conf, null);
- rpcServer.printConf(LOG);
- rpcServer.run();
- System.exit(0);;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/namenode/src/main/java/com/ibm/crail/namenode/NameNodeBlockInfo.java
----------------------------------------------------------------------
diff --git a/namenode/src/main/java/com/ibm/crail/namenode/NameNodeBlockInfo.java b/namenode/src/main/java/com/ibm/crail/namenode/NameNodeBlockInfo.java
deleted file mode 100644
index eab313f..0000000
--- a/namenode/src/main/java/com/ibm/crail/namenode/NameNodeBlockInfo.java
+++ /dev/null
@@ -1,45 +0,0 @@
-package com.ibm.crail.namenode;
-
-import com.ibm.crail.metadata.BlockInfo;
-import com.ibm.crail.metadata.DataNodeInfo;
-
-public class NameNodeBlockInfo extends BlockInfo {
- private BlockInfo region;
- private long offset;
-
- public NameNodeBlockInfo(BlockInfo region, long offset, int length){
- this.region = region;
- this.offset = offset;
- this.length = length;
-
- this.dnInfo = this.getDnInfo();
- this.lba = this.getLba();
- this.addr = this.getAddr();
- this.lkey = this.getLkey();
- }
-
- @Override
- public long getLba() {
- return region.getLba() + offset;
- }
-
- @Override
- public long getAddr() {
- return region.getAddr() + offset;
- }
-
- @Override
- public int getLkey() {
- return region.getLkey();
- }
-
- @Override
- public DataNodeInfo getDnInfo() {
- return region.getDnInfo();
- }
-
- public BlockInfo getRegion() {
- return region;
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/namenode/src/main/java/com/ibm/crail/namenode/NameNodeService.java
----------------------------------------------------------------------
diff --git a/namenode/src/main/java/com/ibm/crail/namenode/NameNodeService.java b/namenode/src/main/java/com/ibm/crail/namenode/NameNodeService.java
deleted file mode 100644
index 2cf1038..0000000
--- a/namenode/src/main/java/com/ibm/crail/namenode/NameNodeService.java
+++ /dev/null
@@ -1,600 +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 java.net.URI;
-import java.util.StringTokenizer;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.DelayQueue;
-import java.util.concurrent.atomic.AtomicLong;
-
-import org.slf4j.Logger;
-
-import com.ibm.crail.CrailNodeType;
-import com.ibm.crail.conf.CrailConstants;
-import com.ibm.crail.metadata.BlockInfo;
-import com.ibm.crail.metadata.DataNodeInfo;
-import com.ibm.crail.metadata.FileInfo;
-import com.ibm.crail.metadata.FileName;
-import com.ibm.crail.rpc.RpcErrors;
-import com.ibm.crail.rpc.RpcNameNodeService;
-import com.ibm.crail.rpc.RpcNameNodeState;
-import com.ibm.crail.rpc.RpcProtocol;
-import com.ibm.crail.rpc.RpcRequestMessage;
-import com.ibm.crail.rpc.RpcResponseMessage;
-import com.ibm.crail.utils.CrailUtils;
-
-public class NameNodeService implements RpcNameNodeService, Sequencer {
- private static final Logger LOG = CrailUtils.getLogger();
-
- //data structures for datanodes, blocks, files
- private long serviceId;
- private long serviceSize;
- private AtomicLong sequenceId;
- private BlockStore blockStore;
- private DelayQueue<AbstractNode> deleteQueue;
- private FileStore fileTree;
- private ConcurrentHashMap<Long, AbstractNode> fileTable;
- private GCServer gcServer;
-
- public NameNodeService() throws IOException {
- URI uri = URI.create(CrailConstants.NAMENODE_ADDRESS);
- String query = uri.getRawQuery();
- StringTokenizer tokenizer = new StringTokenizer(query, "&");
- this.serviceId = Long.parseLong(tokenizer.nextToken().substring(3));
- this.serviceSize = Long.parseLong(tokenizer.nextToken().substring(5));
- this.sequenceId = new AtomicLong(serviceId);
- this.blockStore = new BlockStore();
- this.deleteQueue = new DelayQueue<AbstractNode>();
- this.fileTree = new FileStore(this);
- this.fileTable = new ConcurrentHashMap<Long, AbstractNode>();
- this.gcServer = new GCServer(this, deleteQueue);
-
- AbstractNode root = fileTree.getRoot();
- fileTable.put(root.getFd(), root);
- Thread gc = new Thread(gcServer);
- gc.start();
- }
-
- public long getNextId(){
- return sequenceId.getAndAdd(serviceSize);
- }
-
- @Override
- public short createFile(RpcRequestMessage.CreateFileReq request, RpcResponseMessage.CreateFileRes response, RpcNameNodeState errorState) throws Exception {
- //check protocol
- if (!RpcProtocol.verifyProtocol(RpcProtocol.CMD_CREATE_FILE, request, response)) {
- return RpcErrors.ERR_PROTOCOL_MISMATCH;
- }
-
- //get params
- FileName fileHash = request.getFileName();
- CrailNodeType type = request.getFileType();
- boolean writeable = type.isDirectory() ? false : true;
- int storageClass = request.getStorageClass();
- int locationClass = request.getLocationClass();
-
- //check params
- if (type.isContainer() && locationClass > 0){
- return RpcErrors.ERR_DIR_LOCATION_AFFINITY_MISMATCH;
- }
-
- //rpc
- AbstractNode parentInfo = fileTree.retrieveParent(fileHash, errorState);
- if (errorState.getError() != RpcErrors.ERR_OK){
- return errorState.getError();
- }
- if (parentInfo == null) {
- return RpcErrors.ERR_PARENT_MISSING;
- }
- if (!parentInfo.getType().isContainer()){
- return RpcErrors.ERR_PARENT_NOT_DIR;
- }
-
- if (storageClass < 0){
- storageClass = parentInfo.getStorageClass();
- }
- if (locationClass < 0){
- locationClass = parentInfo.getLocationClass();
- }
-
- AbstractNode fileInfo = fileTree.createNode(fileHash.getFileComponent(), type, storageClass, locationClass);
- try {
- AbstractNode oldNode = parentInfo.putChild(fileInfo);
- if (oldNode != null && oldNode.getFd() != fileInfo.getFd()){
- appendToDeleteQueue(oldNode);
- }
- } catch(Exception e){
- return RpcErrors.ERR_FILE_EXISTS;
- }
- fileTable.put(fileInfo.getFd(), fileInfo);
-
- NameNodeBlockInfo fileBlock = blockStore.getBlock(fileInfo.getStorageClass(), fileInfo.getLocationClass());
- if (fileBlock == null){
- return RpcErrors.ERR_NO_FREE_BLOCKS;
- }
- if (!fileInfo.addBlock(0, fileBlock)){
- return RpcErrors.ERR_ADD_BLOCK_FAILED;
- }
-
- NameNodeBlockInfo parentBlock = null;
- if (fileInfo.getDirOffset() >= 0){
- int index = CrailUtils.computeIndex(fileInfo.getDirOffset());
- parentBlock = parentInfo.getBlock(index);
- if (parentBlock == null){
- parentBlock = blockStore.getBlock(parentInfo.getStorageClass(), parentInfo.getLocationClass());
- if (parentBlock == null){
- return RpcErrors.ERR_NO_FREE_BLOCKS;
- }
- if (!parentInfo.addBlock(index, parentBlock)){
- blockStore.addBlock(parentBlock);
- parentBlock = parentInfo.getBlock(index);
- if (parentBlock == null){
- blockStore.addBlock(fileBlock);
- return RpcErrors.ERR_CREATE_FILE_FAILED;
- }
- }
- }
- parentInfo.incCapacity(CrailConstants.DIRECTORY_RECORD);
- }
-
- if (writeable) {
- fileInfo.updateToken();
- response.shipToken(true);
- } else {
- response.shipToken(false);
- }
- response.setParentInfo(parentInfo);
- response.setFileInfo(fileInfo);
- response.setFileBlock(fileBlock);
- response.setDirBlock(parentBlock);
-
- if (CrailConstants.DEBUG){
- LOG.info("createFile: fd " + fileInfo.getFd() + ", parent " + parentInfo.getFd() + ", writeable " + writeable + ", token " + fileInfo.getToken() + ", capacity " + fileInfo.getCapacity() + ", dirOffset " + fileInfo.getDirOffset());
- }
-
- return RpcErrors.ERR_OK;
- }
-
- @Override
- public short getFile(RpcRequestMessage.GetFileReq request, RpcResponseMessage.GetFileRes response, RpcNameNodeState errorState) throws Exception {
- //check protocol
- if (!RpcProtocol.verifyProtocol(RpcProtocol.CMD_GET_FILE, request, response)){
- return RpcErrors.ERR_PROTOCOL_MISMATCH;
- }
-
- //get params
- FileName fileHash = request.getFileName();
- boolean writeable = request.isWriteable();
-
- //rpc
- AbstractNode fileInfo = fileTree.retrieveFile(fileHash, errorState);
- if (errorState.getError() != RpcErrors.ERR_OK){
- return errorState.getError();
- }
- if (fileInfo == null){
- return RpcErrors.ERR_GET_FILE_FAILED;
- }
- if (writeable && !fileInfo.tokenFree()){
- return RpcErrors.ERR_TOKEN_TAKEN;
- }
-
- if (writeable){
- fileInfo.updateToken();
- }
- fileTable.put(fileInfo.getFd(), fileInfo);
-
- BlockInfo fileBlock = fileInfo.getBlock(0);
-
- response.setFileInfo(fileInfo);
- response.setFileBlock(fileBlock);
- if (writeable){
- response.shipToken();
- }
-
- if (CrailConstants.DEBUG){
- LOG.info("getFile: fd " + fileInfo.getFd() + ", isDir " + fileInfo.getType().isDirectory() + ", token " + fileInfo.getToken() + ", capacity " + fileInfo.getCapacity());
- }
-
- return RpcErrors.ERR_OK;
- }
-
- @Override
- public short setFile(RpcRequestMessage.SetFileReq request, RpcResponseMessage.VoidRes response, RpcNameNodeState errorState) throws Exception {
- //check protocol
- if (!RpcProtocol.verifyProtocol(RpcProtocol.CMD_SET_FILE, request, response)){
- return RpcErrors.ERR_PROTOCOL_MISMATCH;
- }
-
- //get params
- FileInfo fileInfo = request.getFileInfo();
- boolean close = request.isClose();
-
- //rpc
- AbstractNode storedFile = fileTable.get(fileInfo.getFd());
- if (storedFile == null){
- return RpcErrors.ERR_FILE_NOT_OPEN;
- }
-
- if (storedFile.getToken() > 0 && storedFile.getToken() == fileInfo.getToken()){
- storedFile.setCapacity(fileInfo.getCapacity());
- }
- if (close){
- storedFile.resetToken();
- }
-
- if (CrailConstants.DEBUG){
- LOG.info("setFile: " + fileInfo.toString() + ", close " + close);
- }
-
- return RpcErrors.ERR_OK;
- }
-
- @Override
- public short removeFile(RpcRequestMessage.RemoveFileReq request, RpcResponseMessage.DeleteFileRes response, RpcNameNodeState errorState) throws Exception {
- //check protocol
- if (!RpcProtocol.verifyProtocol(RpcProtocol.CMD_REMOVE_FILE, request, response)){
- return RpcErrors.ERR_PROTOCOL_MISMATCH;
- }
-
- //get params
- FileName fileHash = request.getFileName();
-
- //rpc
- AbstractNode parentInfo = fileTree.retrieveParent(fileHash, errorState);
- if (errorState.getError() != RpcErrors.ERR_OK){
- return errorState.getError();
- }
- if (parentInfo == null) {
- return RpcErrors.ERR_CREATE_FILE_FAILED;
- }
-
- AbstractNode fileInfo = fileTree.retrieveFile(fileHash, errorState);
- if (errorState.getError() != RpcErrors.ERR_OK){
- return errorState.getError();
- }
- if (fileInfo == null){
- return RpcErrors.ERR_GET_FILE_FAILED;
- }
-
- response.setParentInfo(parentInfo);
- response.setFileInfo(fileInfo);
-
- fileInfo = parentInfo.removeChild(fileInfo.getComponent());
- if (fileInfo == null){
- return RpcErrors.ERR_GET_FILE_FAILED;
- }
-
- appendToDeleteQueue(fileInfo);
-
- if (CrailConstants.DEBUG){
- LOG.info("removeFile: filename, fd " + fileInfo.getFd());
- }
-
- return RpcErrors.ERR_OK;
- }
-
- @Override
- public short renameFile(RpcRequestMessage.RenameFileReq request, RpcResponseMessage.RenameRes response, RpcNameNodeState errorState) throws Exception {
- //check protocol
- if (!RpcProtocol.verifyProtocol(RpcProtocol.CMD_RENAME_FILE, request, response)){
- return RpcErrors.ERR_PROTOCOL_MISMATCH;
- }
-
- //get params
- FileName srcFileHash = request.getSrcFileName();
- FileName dstFileHash = request.getDstFileName();
-
- //rpc
- AbstractNode srcParent = fileTree.retrieveParent(srcFileHash, errorState);
- if (errorState.getError() != RpcErrors.ERR_OK){
- return errorState.getError();
- }
- if (srcParent == null) {
- return RpcErrors.ERR_GET_FILE_FAILED;
- }
-
- AbstractNode srcFile = fileTree.retrieveFile(srcFileHash, errorState);
- if (errorState.getError() != RpcErrors.ERR_OK){
- return errorState.getError();
- }
- if (srcFile == null){
- return RpcErrors.ERR_SRC_FILE_NOT_FOUND;
- }
-
- //directory block
- int index = CrailUtils.computeIndex(srcFile.getDirOffset());
- NameNodeBlockInfo srcBlock = srcParent.getBlock(index);
- if (srcBlock == null){
- return RpcErrors.ERR_GET_FILE_FAILED;
- }
- //end
-
- response.setSrcParent(srcParent);
- response.setSrcFile(srcFile);
- response.setSrcBlock(srcBlock);
-
- AbstractNode dstParent = fileTree.retrieveParent(dstFileHash, errorState);
- if (errorState.getError() != RpcErrors.ERR_OK){
- return errorState.getError();
- }
- if (dstParent == null) {
- return RpcErrors.ERR_GET_FILE_FAILED;
- }
-
- AbstractNode dstFile = fileTree.retrieveFile(dstFileHash, errorState);
- if (dstFile != null && !dstFile.getType().isDirectory()){
- return RpcErrors.ERR_FILE_EXISTS;
- }
- if (dstFile != null && dstFile.getType().isDirectory()){
- dstParent = dstFile;
- }
-
- srcFile = srcParent.removeChild(srcFile.getComponent());
- if (srcFile == null){
- return RpcErrors.ERR_SRC_FILE_NOT_FOUND;
- }
- srcFile.rename(dstFileHash.getFileComponent());
- try {
- AbstractNode oldNode = dstParent.putChild(srcFile);
- if (oldNode != null && oldNode.getFd() != srcFile.getFd()){
- appendToDeleteQueue(oldNode);
- }
- dstFile = srcFile;
- } catch(Exception e){
- return RpcErrors.ERR_FILE_EXISTS;
- }
-
- //directory block
- index = CrailUtils.computeIndex(srcFile.getDirOffset());
- NameNodeBlockInfo dstBlock = dstParent.getBlock(index);
- if (dstBlock == null){
- dstBlock = blockStore.getBlock(dstParent.getStorageClass(), dstParent.getLocationClass());
- if (dstBlock == null){
- return RpcErrors.ERR_NO_FREE_BLOCKS;
- }
- if (!dstParent.addBlock(index, dstBlock)){
- blockStore.addBlock(dstBlock);
- dstBlock = dstParent.getBlock(index);
- if (dstBlock == null){
- blockStore.addBlock(srcBlock);
- return RpcErrors.ERR_CREATE_FILE_FAILED;
- }
- }
- }
- dstParent.incCapacity(CrailConstants.DIRECTORY_RECORD);
- //end
-
- response.setDstParent(dstParent);
- response.setDstFile(dstFile);
- response.setDstBlock(dstBlock);
-
- if (response.getDstParent().getCapacity() < response.getDstFile().getDirOffset() + CrailConstants.DIRECTORY_RECORD){
- LOG.info("rename: parent capacity does not match dst file offset, capacity " + response.getDstParent().getCapacity() + ", offset " + response.getDstFile().getDirOffset() + ", capacity " + dstParent.getCapacity() + ", offset " + dstFile.getDirOffset());
- }
-
- if (CrailConstants.DEBUG){
- LOG.info("renameFile: src-parent " + srcParent.getFd() + ", src-file " + srcFile.getFd() + ", dst-parent " + dstParent.getFd() + ", dst-fd " + dstFile.getFd());
- }
-
- return RpcErrors.ERR_OK;
- }
-
- @Override
- public short getDataNode(RpcRequestMessage.GetDataNodeReq request, RpcResponseMessage.GetDataNodeRes response, RpcNameNodeState errorState) throws Exception {
- //check protocol
- if (!RpcProtocol.verifyProtocol(RpcProtocol.CMD_GET_DATANODE, request, response)){
- return RpcErrors.ERR_PROTOCOL_MISMATCH;
- }
-
- //get params
- DataNodeInfo dnInfo = request.getInfo();
-
- //rpc
- DataNodeBlocks dnInfoNn = blockStore.getDataNode(dnInfo);
- if (dnInfoNn == null){
- return RpcErrors.ERR_DATANODE_NOT_REGISTERED;
- }
-
- dnInfoNn.touch();
- response.setServiceId(serviceId);
- response.setFreeBlockCount(dnInfoNn.getBlockCount());
-
- return RpcErrors.ERR_OK;
- }
-
- @Override
- public short setBlock(RpcRequestMessage.SetBlockReq request, RpcResponseMessage.VoidRes response, RpcNameNodeState errorState) throws Exception {
- //check protocol
- if (!RpcProtocol.verifyProtocol(RpcProtocol.CMD_SET_BLOCK, request, response)){
- return RpcErrors.ERR_PROTOCOL_MISMATCH;
- }
-
- //get params
- BlockInfo region = new BlockInfo();
- region.setBlockInfo(request.getBlockInfo());
-
- short error = RpcErrors.ERR_OK;
- if (blockStore.regionExists(region)){
- error = blockStore.updateRegion(region);
- } else {
- //rpc
- int realBlocks = (int) (((long) region.getLength()) / CrailConstants.BLOCK_SIZE) ;
- long offset = 0;
- for (int i = 0; i < realBlocks; i++){
- NameNodeBlockInfo nnBlock = new NameNodeBlockInfo(region, offset, (int) CrailConstants.BLOCK_SIZE);
- error = blockStore.addBlock(nnBlock);
- offset += CrailConstants.BLOCK_SIZE;
-
- if (error != RpcErrors.ERR_OK){
- break;
- }
- }
- }
-
- return error;
- }
-
- @Override
- public short getBlock(RpcRequestMessage.GetBlockReq request, RpcResponseMessage.GetBlockRes response, RpcNameNodeState errorState) throws Exception {
- //check protocol
- if (!RpcProtocol.verifyProtocol(RpcProtocol.CMD_GET_BLOCK, request, response)){
- return RpcErrors.ERR_PROTOCOL_MISMATCH;
- }
-
- //get params
- long fd = request.getFd();
- long token = request.getToken();
- long position = request.getPosition();
- long capacity = request.getCapacity();
-
- //check params
- if (position < 0){
- return RpcErrors.ERR_POSITION_NEGATIV;
- }
-
- //rpc
- AbstractNode fileInfo = fileTable.get(fd);
- if (fileInfo == null){
- return RpcErrors.ERR_FILE_NOT_OPEN;
- }
-
- int index = CrailUtils.computeIndex(position);
- if (index < 0){
- return RpcErrors.ERR_POSITION_NEGATIV;
- }
-
- NameNodeBlockInfo block = fileInfo.getBlock(index);
- if (block == null && fileInfo.getToken() == token){
- block = blockStore.getBlock(fileInfo.getStorageClass(), fileInfo.getLocationClass());
- if (block == null){
- return RpcErrors.ERR_NO_FREE_BLOCKS;
- }
- if (!fileInfo.addBlock(index, block)){
- return RpcErrors.ERR_ADD_BLOCK_FAILED;
- }
- block = fileInfo.getBlock(index);
- if (block == null){
- return RpcErrors.ERR_ADD_BLOCK_FAILED;
- }
- fileInfo.setCapacity(capacity);
- } else if (block == null && token > 0){
- return RpcErrors.ERR_TOKEN_MISMATCH;
- } else if (block == null && token == 0){
- return RpcErrors.ERR_CAPACITY_EXCEEDED;
- }
-
- response.setBlockInfo(block);
- return RpcErrors.ERR_OK;
- }
-
- @Override
- public short getLocation(RpcRequestMessage.GetLocationReq request, RpcResponseMessage.GetLocationRes response, RpcNameNodeState errorState) throws Exception {
- //check protocol
- if (!RpcProtocol.verifyProtocol(RpcProtocol.CMD_GET_LOCATION, request, response)){
- return RpcErrors.ERR_PROTOCOL_MISMATCH;
- }
-
- //get params
- FileName fileName = request.getFileName();
- long position = request.getPosition();
-
- //check params
- if (position < 0){
- return RpcErrors.ERR_POSITION_NEGATIV;
- }
-
- //rpc
- AbstractNode fileInfo = fileTree.retrieveFile(fileName, errorState);
- if (errorState.getError() != RpcErrors.ERR_OK){
- return errorState.getError();
- }
- if (fileInfo == null){
- return RpcErrors.ERR_GET_FILE_FAILED;
- }
-
- int index = CrailUtils.computeIndex(position);
- if (index < 0){
- return RpcErrors.ERR_POSITION_NEGATIV;
- }
- BlockInfo block = fileInfo.getBlock(index);
- if (block == null){
- return RpcErrors.ERR_OFFSET_TOO_LARGE;
- }
-
- response.setBlockInfo(block);
-
- return RpcErrors.ERR_OK;
- }
-
- //------------------------
-
- @Override
- public short dump(RpcRequestMessage.DumpNameNodeReq request, RpcResponseMessage.VoidRes response, RpcNameNodeState errorState) throws Exception {
- if (!RpcProtocol.verifyProtocol(RpcProtocol.CMD_DUMP_NAMENODE, request, response)){
- return RpcErrors.ERR_PROTOCOL_MISMATCH;
- }
-
- System.out.println("#fd\t\tfilecomp\t\tcapacity\t\tisdir\t\t\tdiroffset");
- fileTree.dump();
- System.out.println("#fd\t\tfilecomp\t\tcapacity\t\tisdir\t\t\tdiroffset");
- dumpFastMap();
-
- return RpcErrors.ERR_OK;
- }
-
- @Override
- public short ping(RpcRequestMessage.PingNameNodeReq request, RpcResponseMessage.PingNameNodeRes response, RpcNameNodeState errorState) throws Exception {
- if (!RpcProtocol.verifyProtocol(RpcProtocol.CMD_PING_NAMENODE, request, response)){
- return RpcErrors.ERR_PROTOCOL_MISMATCH;
- }
-
- response.setData(request.getOp()+1);
-
- return RpcErrors.ERR_OK;
- }
-
-
- //--------------- helper functions
-
- void appendToDeleteQueue(AbstractNode fileInfo) throws Exception {
- if (fileInfo != null) {
- fileInfo.setDelay(CrailConstants.TOKEN_EXPIRATION);
- deleteQueue.add(fileInfo);
- }
- }
-
- void freeFile(AbstractNode fileInfo) throws Exception {
- if (fileInfo != null) {
- fileTable.remove(fileInfo.getFd());
- fileInfo.freeBlocks(blockStore);
- }
- }
-
- private void dumpFastMap(){
- for (Long key : fileTable.keySet()){
- AbstractNode file = fileTable.get(key);
- System.out.println(file.toString());
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/namenode/src/main/java/com/ibm/crail/namenode/Sequencer.java
----------------------------------------------------------------------
diff --git a/namenode/src/main/java/com/ibm/crail/namenode/Sequencer.java b/namenode/src/main/java/com/ibm/crail/namenode/Sequencer.java
deleted file mode 100644
index 96de5a8..0000000
--- a/namenode/src/main/java/com/ibm/crail/namenode/Sequencer.java
+++ /dev/null
@@ -1,5 +0,0 @@
-package com.ibm.crail.namenode;
-
-public interface Sequencer {
- long getNextId();
-}
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/namenode/src/main/java/com/ibm/crail/namenode/TableBlocks.java
----------------------------------------------------------------------
diff --git a/namenode/src/main/java/com/ibm/crail/namenode/TableBlocks.java b/namenode/src/main/java/com/ibm/crail/namenode/TableBlocks.java
deleted file mode 100644
index 8bda168..0000000
--- a/namenode/src/main/java/com/ibm/crail/namenode/TableBlocks.java
+++ /dev/null
@@ -1,20 +0,0 @@
-package com.ibm.crail.namenode;
-
-import com.ibm.crail.CrailNodeType;
-
-public class TableBlocks extends DirectoryBlocks {
-
- TableBlocks(long fd, int fileComponent, CrailNodeType type,
- int storageClass, int locationClass) {
- super(fd, fileComponent, type, storageClass, locationClass);
- }
-
- @Override
- public AbstractNode putChild(AbstractNode child) throws Exception {
- if (!child.getType().isKeyValue()){
- throw new Exception("Attempt to create key/value pair in container other than a table");
- }
-
- return children.put(child.getComponent(), child);
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-crail/blob/32f8ead2/namenode/src/main/java/org/apache/crail/namenode/AbstractNode.java
----------------------------------------------------------------------
diff --git a/namenode/src/main/java/org/apache/crail/namenode/AbstractNode.java b/namenode/src/main/java/org/apache/crail/namenode/AbstractNode.java
new file mode 100644
index 0000000..8ff7f7e
--- /dev/null
+++ b/namenode/src/main/java/org/apache/crail/namenode/AbstractNode.java
@@ -0,0 +1,111 @@
+/*
+ * 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.namenode;
+
+import java.util.Queue;
+import java.util.concurrent.Delayed;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.crail.CrailNodeType;
+import org.apache.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/org/apache/crail/namenode/BlockStore.java
----------------------------------------------------------------------
diff --git a/namenode/src/main/java/org/apache/crail/namenode/BlockStore.java b/namenode/src/main/java/org/apache/crail/namenode/BlockStore.java
new file mode 100644
index 0000000..6738a45
--- /dev/null
+++ b/namenode/src/main/java/org/apache/crail/namenode/BlockStore.java
@@ -0,0 +1,277 @@
+/*
+ * 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.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.apache.crail.conf.CrailConstants;
+import org.apache.crail.metadata.BlockInfo;
+import org.apache.crail.metadata.DataNodeInfo;
+import org.apache.crail.rpc.RpcErrors;
+import org.apache.crail.utils.AtomicIntegerModulo;
+import org.apache.crail.utils.CrailUtils;
+import org.slf4j.Logger;
+
+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/org/apache/crail/namenode/DataNodeBlocks.java
----------------------------------------------------------------------
diff --git a/namenode/src/main/java/org/apache/crail/namenode/DataNodeBlocks.java b/namenode/src/main/java/org/apache/crail/namenode/DataNodeBlocks.java
new file mode 100644
index 0000000..7f8198e
--- /dev/null
+++ b/namenode/src/main/java/org/apache/crail/namenode/DataNodeBlocks.java
@@ -0,0 +1,92 @@
+/*
+ * 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.namenode;
+
+import java.net.UnknownHostException;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.crail.conf.CrailConstants;
+import org.apache.crail.metadata.BlockInfo;
+import org.apache.crail.metadata.DataNodeInfo;
+import org.apache.crail.rpc.RpcErrors;
+import org.apache.crail.utils.CrailUtils;
+import org.slf4j.Logger;
+
+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/org/apache/crail/namenode/DirectoryBlocks.java
----------------------------------------------------------------------
diff --git a/namenode/src/main/java/org/apache/crail/namenode/DirectoryBlocks.java b/namenode/src/main/java/org/apache/crail/namenode/DirectoryBlocks.java
new file mode 100644
index 0000000..0ec86e7
--- /dev/null
+++ b/namenode/src/main/java/org/apache/crail/namenode/DirectoryBlocks.java
@@ -0,0 +1,107 @@
+/*
+ * 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.namenode;
+
+import java.util.Iterator;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.apache.crail.CrailNodeType;
+import org.apache.crail.conf.CrailConstants;
+import org.apache.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/org/apache/crail/namenode/FileBlocks.java
----------------------------------------------------------------------
diff --git a/namenode/src/main/java/org/apache/crail/namenode/FileBlocks.java b/namenode/src/main/java/org/apache/crail/namenode/FileBlocks.java
new file mode 100644
index 0000000..67a1e65
--- /dev/null
+++ b/namenode/src/main/java/org/apache/crail/namenode/FileBlocks.java
@@ -0,0 +1,111 @@
+/*
+ * 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.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 org.apache.crail.CrailNodeType;
+import org.apache.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/org/apache/crail/namenode/FileStore.java
----------------------------------------------------------------------
diff --git a/namenode/src/main/java/org/apache/crail/namenode/FileStore.java b/namenode/src/main/java/org/apache/crail/namenode/FileStore.java
new file mode 100644
index 0000000..0a168ab
--- /dev/null
+++ b/namenode/src/main/java/org/apache/crail/namenode/FileStore.java
@@ -0,0 +1,90 @@
+/*
+ * 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.namenode;
+
+import java.io.IOException;
+
+import org.apache.crail.CrailNodeType;
+import org.apache.crail.conf.CrailConstants;
+import org.apache.crail.metadata.FileName;
+import org.apache.crail.rpc.RpcErrors;
+import org.apache.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/org/apache/crail/namenode/GCServer.java
----------------------------------------------------------------------
diff --git a/namenode/src/main/java/org/apache/crail/namenode/GCServer.java b/namenode/src/main/java/org/apache/crail/namenode/GCServer.java
new file mode 100644
index 0000000..8701ebb
--- /dev/null
+++ b/namenode/src/main/java/org/apache/crail/namenode/GCServer.java
@@ -0,0 +1,55 @@
+/*
+ * 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.namenode;
+
+import java.util.concurrent.DelayQueue;
+
+import org.apache.crail.utils.CrailUtils;
+import org.slf4j.Logger;
+
+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/org/apache/crail/namenode/KeyValueBlocks.java
----------------------------------------------------------------------
diff --git a/namenode/src/main/java/org/apache/crail/namenode/KeyValueBlocks.java b/namenode/src/main/java/org/apache/crail/namenode/KeyValueBlocks.java
new file mode 100644
index 0000000..774a210
--- /dev/null
+++ b/namenode/src/main/java/org/apache/crail/namenode/KeyValueBlocks.java
@@ -0,0 +1,11 @@
+package org.apache.crail.namenode;
+
+import org.apache.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/org/apache/crail/namenode/LogDispatcher.java
----------------------------------------------------------------------
diff --git a/namenode/src/main/java/org/apache/crail/namenode/LogDispatcher.java b/namenode/src/main/java/org/apache/crail/namenode/LogDispatcher.java
new file mode 100644
index 0000000..686f36d
--- /dev/null
+++ b/namenode/src/main/java/org/apache/crail/namenode/LogDispatcher.java
@@ -0,0 +1,121 @@
+package org.apache.crail.namenode;
+
+import org.apache.crail.rpc.RpcNameNodeService;
+import org.apache.crail.rpc.RpcNameNodeState;
+import org.apache.crail.rpc.RpcProtocol;
+import org.apache.crail.rpc.RpcRequestMessage.CreateFileReq;
+import org.apache.crail.rpc.RpcRequestMessage.DumpNameNodeReq;
+import org.apache.crail.rpc.RpcRequestMessage.GetBlockReq;
+import org.apache.crail.rpc.RpcRequestMessage.GetDataNodeReq;
+import org.apache.crail.rpc.RpcRequestMessage.GetFileReq;
+import org.apache.crail.rpc.RpcRequestMessage.GetLocationReq;
+import org.apache.crail.rpc.RpcRequestMessage.PingNameNodeReq;
+import org.apache.crail.rpc.RpcRequestMessage.RemoveFileReq;
+import org.apache.crail.rpc.RpcRequestMessage.RenameFileReq;
+import org.apache.crail.rpc.RpcRequestMessage.SetBlockReq;
+import org.apache.crail.rpc.RpcRequestMessage.SetFileReq;
+import org.apache.crail.rpc.RpcResponseMessage.CreateFileRes;
+import org.apache.crail.rpc.RpcResponseMessage.DeleteFileRes;
+import org.apache.crail.rpc.RpcResponseMessage.GetBlockRes;
+import org.apache.crail.rpc.RpcResponseMessage.GetDataNodeRes;
+import org.apache.crail.rpc.RpcResponseMessage.GetFileRes;
+import org.apache.crail.rpc.RpcResponseMessage.GetLocationRes;
+import org.apache.crail.rpc.RpcResponseMessage.PingNameNodeRes;
+import org.apache.crail.rpc.RpcResponseMessage.RenameRes;
+import org.apache.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/org/apache/crail/namenode/LogRecord.java
----------------------------------------------------------------------
diff --git a/namenode/src/main/java/org/apache/crail/namenode/LogRecord.java b/namenode/src/main/java/org/apache/crail/namenode/LogRecord.java
new file mode 100644
index 0000000..ec4274c
--- /dev/null
+++ b/namenode/src/main/java/org/apache/crail/namenode/LogRecord.java
@@ -0,0 +1,285 @@
+package org.apache.crail.namenode;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+import org.apache.crail.rpc.RpcProtocol;
+import org.apache.crail.rpc.RpcRequestMessage;
+import org.apache.crail.utils.CrailUtils;
+import org.slf4j.Logger;
+
+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;
+ }
+}