hadoop-hdfs-project/hadoop-hdfs/src/contrib/libhdfs3/src/server/NamenodeProxy.cc - hadoop - Git at Google

 /**
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you 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.
  */

 #include "Exception.h"
 #include "ExceptionInternal.h"
 #include "Logger.h"
 #include "NamenodeImpl.h"
 #include "NamenodeProxy.h"
 #include "StringUtil.h"

 #include <string>

 #include <sys/fcntl.h>
 #include <sys/stat.h>
 #include <sys/types.h>
 #include <sys/file.h>

 namespace hdfs {
 namespace internal {

 static uint32_t GetInitNamenodeIndex(const std::string &id) {
     std::string path = "/tmp/";
     path += id;
     int fd;
     uint32_t index = 0;
     /*
      * try create the file
      */
     fd = open(path.c_str(), O_WRONLY | O_CREAT | O_EXCL, 0666);

     if (fd < 0) {
         if (errno == EEXIST) {
             /*
              * the file already exist, try to open it
              */
             fd = open(path.c_str(), O_RDONLY);
         } else {
             /*
              * failed to create, do not care why
              */
             return 0;
         }
     } else {
         if (0 != flock(fd, LOCK_EX)) {
             /*
              * failed to lock
              */
             close(fd);
             return index;
         }

         /*
          * created file, initialize it with 0
          */
         write(fd, &index, sizeof(index));
         flock(fd, LOCK_UN);
         close(fd);
         return index;
     }

     /*
      * the file exist, read it.
      */
     if (fd >= 0) {
         if (0 != flock(fd, LOCK_SH)) {
             /*
              * failed to lock
              */
             close(fd);
             return index;
         }

         if (sizeof(index) != read(fd, &index, sizeof(index))) {
             /*
              * failed to read, do not care why
              */
             index = 0;
         }

         flock(fd, LOCK_UN);
         close(fd);
     }

     return index;
 }

 static void SetInitNamenodeIndex(const std::string &id, uint32_t index) {
     std::string path = "/tmp/";
     path += id;
     int fd;
     /*
      * try open the file for write
      */
     fd = open(path.c_str(), O_WRONLY);

     if (fd > 0) {
         if (0 != flock(fd, LOCK_EX)) {
             /*
              * failed to lock
              */
             close(fd);
             return;
         }

         write(fd, &index, sizeof(index));
         flock(fd, LOCK_UN);
         close(fd);
     }
 }

 NamenodeProxy::NamenodeProxy(const std::vector<NamenodeInfo> &namenodeInfos,
                              const std::string &tokenService,
                              const SessionConfig &c, const RpcAuth &a) :
     clusterid(tokenService), currentNamenode(0) {
     if (namenodeInfos.size() == 1) {
         enableNamenodeHA = false;
         maxNamenodeHARetry = 0;
     } else {
         enableNamenodeHA = true;
         maxNamenodeHARetry = c.getRpcMaxHaRetry();
     }

     for (size_t i = 0; i < namenodeInfos.size(); ++i) {
         std::vector<std::string> nninfo = StringSplit(namenodeInfos[i].getRpcAddr(), ":");

         if (nninfo.size() != 2) {
             THROW(InvalidParameter, "Cannot create namenode proxy, %s does not contain host or port",
                   namenodeInfos[i].getRpcAddr().c_str());
         }

         namenodes.push_back(
             shared_ptr<Namenode>(
                 new NamenodeImpl(nninfo[0].c_str(), nninfo[1].c_str(), clusterid, c, a)));
     }

     if (enableNamenodeHA) {
         currentNamenode = GetInitNamenodeIndex(clusterid) % namenodeInfos.size();
     }
 }

 NamenodeProxy::~NamenodeProxy() {
 }

 shared_ptr<Namenode> NamenodeProxy::getActiveNamenode(uint32_t &oldValue) {
     lock_guard<mutex> lock(mut);

     if (namenodes.empty()) {
         THROW(HdfsFileSystemClosed, "NamenodeProxy is closed.");
     }

     oldValue = currentNamenode;
     return namenodes[currentNamenode % namenodes.size()];
 }

 void NamenodeProxy::failoverToNextNamenode(uint32_t oldValue) {
     lock_guard<mutex> lock(mut);

     if (oldValue != currentNamenode) {
         //already failover in another thread.
         return;
     }

     ++currentNamenode;
     currentNamenode = currentNamenode % namenodes.size();
     SetInitNamenodeIndex(clusterid, currentNamenode);
 }

 static void HandleHdfsFailoverException(const HdfsFailoverException &e) {
     try {
         rethrow_if_nested(e);
     } catch (...) {
         NESTED_THROW(hdfs::HdfsRpcException, "%s", e.what());
     }

     //should not reach here
     abort();
 }

 #define NAMENODE_HA_RETRY_BEGIN() \
     do { \
         int __count = 0; \
         do { \
             uint32_t __oldValue = 0; \
             shared_ptr<Namenode> namenode =  getActiveNamenode(__oldValue); \
             try { \
                 (void)0

 #define NAMENODE_HA_RETRY_END() \
     break; \
     } catch (const NameNodeStandbyException &e) { \
         if (!enableNamenodeHA || __count++ > maxNamenodeHARetry) { \
             throw; \
         } \
     } catch (const HdfsFailoverException &e) { \
         if (!enableNamenodeHA || __count++ > maxNamenodeHARetry) { \
             HandleHdfsFailoverException(e); \
         } \
     } \
     failoverToNextNamenode(__oldValue); \
     LOG(WARNING, "NamenodeProxy: Failover to another Namenode."); \
     } while (true); \
     } while (0)

 void NamenodeProxy::getBlockLocations(const std::string &src, int64_t offset,
         int64_t length, LocatedBlocks &lbs) {
     NAMENODE_HA_RETRY_BEGIN();
     namenode->getBlockLocations(src, offset, length, lbs);
     NAMENODE_HA_RETRY_END();
 }

 void NamenodeProxy::create(const std::string &src, const Permission &masked,
         const std::string &clientName, int flag, bool createParent,
         short replication, int64_t blockSize) {
     NAMENODE_HA_RETRY_BEGIN();
     namenode->create(src, masked, clientName, flag, createParent, replication, blockSize);
     NAMENODE_HA_RETRY_END();
 }

 shared_ptr<LocatedBlock> NamenodeProxy::append(const std::string &src,
         const std::string &clientName) {
     NAMENODE_HA_RETRY_BEGIN();
     return namenode->append(src, clientName);
     NAMENODE_HA_RETRY_END();
     assert(!"should not reach here");
     return shared_ptr<LocatedBlock>();
 }

 bool NamenodeProxy::setReplication(const std::string &src, short replication) {
     NAMENODE_HA_RETRY_BEGIN();
     return namenode->setReplication(src, replication);
     NAMENODE_HA_RETRY_END();
     assert(!"should not reach here");
     return false;
 }

 void NamenodeProxy::setPermission(const std::string &src,
         const Permission &permission) {
     NAMENODE_HA_RETRY_BEGIN();
     namenode->setPermission(src, permission);
     NAMENODE_HA_RETRY_END();
 }

 void NamenodeProxy::setOwner(const std::string &src,
         const std::string &username, const std::string &groupname) {
     NAMENODE_HA_RETRY_BEGIN();
     namenode->setOwner(src, username, groupname);
     NAMENODE_HA_RETRY_END();
 }

 void NamenodeProxy::abandonBlock(const ExtendedBlock &b,
         const std::string &src, const std::string &holder) {
     NAMENODE_HA_RETRY_BEGIN();
     namenode->abandonBlock(b, src, holder);
     NAMENODE_HA_RETRY_END();
 }

 shared_ptr<LocatedBlock> NamenodeProxy::addBlock(const std::string &src,
         const std::string &clientName, const ExtendedBlock * previous,
         const std::vector<DatanodeInfo> &excludeNodes) {
     NAMENODE_HA_RETRY_BEGIN();
     return namenode->addBlock(src, clientName, previous, excludeNodes);
     NAMENODE_HA_RETRY_END();
     assert(!"should not reach here");
     return shared_ptr<LocatedBlock>();
 }

 shared_ptr<LocatedBlock> NamenodeProxy::getAdditionalDatanode(
     const std::string &src, const ExtendedBlock &blk,
     const std::vector<DatanodeInfo> &existings,
     const std::vector<std::string> &storageIDs,
     const std::vector<DatanodeInfo> &excludes, int numAdditionalNodes,
     const std::string &clientName) {
     NAMENODE_HA_RETRY_BEGIN();
     return namenode->getAdditionalDatanode(src, blk, existings,
                   storageIDs, excludes, numAdditionalNodes, clientName);
     NAMENODE_HA_RETRY_END();
     assert(!"should not reach here");
     return shared_ptr<LocatedBlock>();
 }

 bool NamenodeProxy::complete(const std::string &src,
                   const std::string &clientName, const ExtendedBlock *last) {
     NAMENODE_HA_RETRY_BEGIN();
     return namenode->complete(src, clientName, last);
     NAMENODE_HA_RETRY_END();
     assert(!"should not reach here");
     return false;
 }

 /*void NamenodeProxy::reportBadBlocks(const std::vector<LocatedBlock> &blocks) {
     NAMENODE_HA_RETRY_BEGIN();
     namenode->reportBadBlocks(blocks);
     NAMENODE_HA_RETRY_END();
 }*/

 bool NamenodeProxy::rename(const std::string &src, const std::string &dst) {
     NAMENODE_HA_RETRY_BEGIN();
     return namenode->rename(src, dst);
     NAMENODE_HA_RETRY_END();
     assert(!"should not reach here");
     return false;
 }

 /*
 void NamenodeProxy::concat(const std::string &trg,
                            const std::vector<std::string> &srcs) {
     NAMENODE_HA_RETRY_BEGIN();
     namenode->concat(trg, srcs);
     NAMENODE_HA_RETRY_END();
 }
 */

 bool NamenodeProxy::deleteFile(const std::string &src, bool recursive) {
     NAMENODE_HA_RETRY_BEGIN();
     return namenode->deleteFile(src, recursive);
     NAMENODE_HA_RETRY_END();
     assert(!"should not reach here");
     return false;
 }

 bool NamenodeProxy::mkdirs(const std::string &src, const Permission &masked,
           bool createParent) {
     NAMENODE_HA_RETRY_BEGIN();
     return namenode->mkdirs(src, masked, createParent);
     NAMENODE_HA_RETRY_END();
     assert(!"should not reach here");
     return false;
 }

 bool NamenodeProxy::getListing(const std::string &src,
           const std::string &startAfter, bool needLocation,
           std::vector<FileStatus> &dl) {
     NAMENODE_HA_RETRY_BEGIN();
     return namenode->getListing(src, startAfter, needLocation, dl);
     NAMENODE_HA_RETRY_END();
     assert(!"should not reach here");
     return false;
 }

 void NamenodeProxy::renewLease(const std::string &clientName) {
     NAMENODE_HA_RETRY_BEGIN();
     namenode->renewLease(clientName);
     NAMENODE_HA_RETRY_END();
 }

 bool NamenodeProxy::recoverLease(const std::string &src,
                                  const std::string &clientName) {
     NAMENODE_HA_RETRY_BEGIN();
     return namenode->recoverLease(src, clientName);
     NAMENODE_HA_RETRY_END();
     assert(!"should not reach here");
     return false;
 }

 std::vector<int64_t> NamenodeProxy::getFsStats() {
     NAMENODE_HA_RETRY_BEGIN();
     return namenode->getFsStats();
     NAMENODE_HA_RETRY_END();
     assert(!"should not reach here");
     return std::vector<int64_t>();
 }

 /*void NamenodeProxy::metaSave(const std::string &filename) {
     NAMENODE_HA_RETRY_BEGIN();
     namenode->metaSave(filename);
     NAMENODE_HA_RETRY_END();
 }*/

 FileStatus NamenodeProxy::getFileInfo(const std::string &src) {
     NAMENODE_HA_RETRY_BEGIN();
     return namenode->getFileInfo(src);
     NAMENODE_HA_RETRY_END();
     assert(!"should not reach here");
     return FileStatus();
 }

 /*FileStatus NamenodeProxy::getFileLinkInfo(const std::string &src) {
     NAMENODE_HA_RETRY_BEGIN();
     return namenode->getFileLinkInfo(src);
     NAMENODE_HA_RETRY_END();
     assert(!"should not reach here");
     return FileStatus();
 }*/

 /*ContentSummary NamenodeProxy::getContentSummary(const std::string &path) {
     NAMENODE_HA_RETRY_BEGIN();
     return namenode->getContentSummary(path);
     NAMENODE_HA_RETRY_END();
     assert(!"should not reach here");
     return ContentSummary();
 }*/

 /*void NamenodeProxy::setQuota(const std::string &path, int64_t namespaceQuota,
                              int64_t diskspaceQuota) {
     NAMENODE_HA_RETRY_BEGIN();
     namenode->setQuota(path, namespaceQuota, diskspaceQuota);
     NAMENODE_HA_RETRY_END();
 }*/

 void NamenodeProxy::fsync(const std::string &src, const std::string &client) {
     NAMENODE_HA_RETRY_BEGIN();
     namenode->fsync(src, client);
     NAMENODE_HA_RETRY_END();
 }

 void NamenodeProxy::setTimes(const std::string &src, int64_t mtime,
                              int64_t atime) {
     NAMENODE_HA_RETRY_BEGIN();
     namenode->setTimes(src, mtime, atime);
     NAMENODE_HA_RETRY_END();
 }

 /*void NamenodeProxy::createSymlink(const std::string &target,
                 const std::string &link, const Permission &dirPerm,
                 bool createParent) {
     NAMENODE_HA_RETRY_BEGIN();
     namenode->createSymlink(target, link, dirPerm, createParent);
     NAMENODE_HA_RETRY_END();
 }*/

 /*std::string NamenodeProxy::getLinkTarget(const std::string &path) {
     NAMENODE_HA_RETRY_BEGIN();
     return namenode->getLinkTarget(path);
     NAMENODE_HA_RETRY_END();
     assert(!"should not reach here");
     return "";
 }*/

 shared_ptr<LocatedBlock> NamenodeProxy::updateBlockForPipeline(
           const ExtendedBlock &block, const std::string &clientName) {
     NAMENODE_HA_RETRY_BEGIN();
     return namenode->updateBlockForPipeline(block, clientName);
     NAMENODE_HA_RETRY_END();
     assert(!"should not reach here");
     return shared_ptr<LocatedBlock>();
 }

 void NamenodeProxy::updatePipeline(const std::string &clientName,
           const ExtendedBlock &oldBlock, const ExtendedBlock &newBlock,
           const std::vector<DatanodeInfo> &newNodes,
           const std::vector<std::string> &storageIDs) {
     NAMENODE_HA_RETRY_BEGIN();
     namenode->updatePipeline(clientName, oldBlock, newBlock,
                              newNodes, storageIDs);
     NAMENODE_HA_RETRY_END();
 }

 Token NamenodeProxy::getDelegationToken(const std::string &renewer) {
     NAMENODE_HA_RETRY_BEGIN();
     return namenode->getDelegationToken(renewer);
     NAMENODE_HA_RETRY_END();
     assert(!"should not reach here");
     return Token();
 }

 int64_t NamenodeProxy::renewDelegationToken(const Token &token) {
     NAMENODE_HA_RETRY_BEGIN();
     return namenode->renewDelegationToken(token);
     NAMENODE_HA_RETRY_END();
     assert(!"should not reach here");
     return 0;
 }

 void NamenodeProxy::cancelDelegationToken(const Token &token) {
     NAMENODE_HA_RETRY_BEGIN();
     namenode->cancelDelegationToken(token);
     NAMENODE_HA_RETRY_END();
 }

 void NamenodeProxy::close() {
     lock_guard<mutex> lock(mut);
     namenodes.clear();
 }

 }
 }
	/**
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you 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.
	*/

	#include "Exception.h"
	#include "ExceptionInternal.h"
	#include "Logger.h"
	#include "NamenodeImpl.h"
	#include "NamenodeProxy.h"
	#include "StringUtil.h"

	#include <string>

	#include <sys/fcntl.h>
	#include <sys/stat.h>
	#include <sys/types.h>
	#include <sys/file.h>

	namespace hdfs {
	namespace internal {

	static uint32_t GetInitNamenodeIndex(const std::string &id) {
	std::string path = "/tmp/";
	path += id;
	int fd;
	uint32_t index = 0;
	/*
	* try create the file
	*/
	fd = open(path.c_str(), O_WRONLY \| O_CREAT \| O_EXCL, 0666);

	if (fd < 0) {
	if (errno == EEXIST) {
	/*
	* the file already exist, try to open it
	*/
	fd = open(path.c_str(), O_RDONLY);
	} else {
	/*
	* failed to create, do not care why
	*/
	return 0;
	}
	} else {
	if (0 != flock(fd, LOCK_EX)) {
	/*
	* failed to lock
	*/
	close(fd);
	return index;
	}

	/*
	* created file, initialize it with 0
	*/
	write(fd, &index, sizeof(index));
	flock(fd, LOCK_UN);
	close(fd);
	return index;
	}

	/*
	* the file exist, read it.
	*/
	if (fd >= 0) {
	if (0 != flock(fd, LOCK_SH)) {
	/*
	* failed to lock
	*/
	close(fd);
	return index;
	}

	if (sizeof(index) != read(fd, &index, sizeof(index))) {
	/*
	* failed to read, do not care why
	*/
	index = 0;
	}

	flock(fd, LOCK_UN);
	close(fd);
	}

	return index;
	}

	static void SetInitNamenodeIndex(const std::string &id, uint32_t index) {
	std::string path = "/tmp/";
	path += id;
	int fd;
	/*
	* try open the file for write
	*/
	fd = open(path.c_str(), O_WRONLY);

	if (fd > 0) {
	if (0 != flock(fd, LOCK_EX)) {
	/*
	* failed to lock
	*/
	close(fd);
	return;
	}

	write(fd, &index, sizeof(index));
	flock(fd, LOCK_UN);
	close(fd);
	}
	}

	NamenodeProxy::NamenodeProxy(const std::vector<NamenodeInfo> &namenodeInfos,
	const std::string &tokenService,
	const SessionConfig &c, const RpcAuth &a) :
	clusterid(tokenService), currentNamenode(0) {
	if (namenodeInfos.size() == 1) {
	enableNamenodeHA = false;
	maxNamenodeHARetry = 0;
	} else {
	enableNamenodeHA = true;
	maxNamenodeHARetry = c.getRpcMaxHaRetry();
	}

	for (size_t i = 0; i < namenodeInfos.size(); ++i) {
	std::vector<std::string> nninfo = StringSplit(namenodeInfos[i].getRpcAddr(), ":");

	if (nninfo.size() != 2) {
	THROW(InvalidParameter, "Cannot create namenode proxy, %s does not contain host or port",
	namenodeInfos[i].getRpcAddr().c_str());
	}

	namenodes.push_back(
	shared_ptr<Namenode>(
	new NamenodeImpl(nninfo[0].c_str(), nninfo[1].c_str(), clusterid, c, a)));
	}

	if (enableNamenodeHA) {
	currentNamenode = GetInitNamenodeIndex(clusterid) % namenodeInfos.size();
	}
	}

	NamenodeProxy::~NamenodeProxy() {
	}

	shared_ptr<Namenode> NamenodeProxy::getActiveNamenode(uint32_t &oldValue) {
	lock_guard<mutex> lock(mut);

	if (namenodes.empty()) {
	THROW(HdfsFileSystemClosed, "NamenodeProxy is closed.");
	}

	oldValue = currentNamenode;
	return namenodes[currentNamenode % namenodes.size()];
	}

	void NamenodeProxy::failoverToNextNamenode(uint32_t oldValue) {
	lock_guard<mutex> lock(mut);

	if (oldValue != currentNamenode) {
	//already failover in another thread.
	return;
	}

	++currentNamenode;
	currentNamenode = currentNamenode % namenodes.size();
	SetInitNamenodeIndex(clusterid, currentNamenode);
	}

	static void HandleHdfsFailoverException(const HdfsFailoverException &e) {
	try {
	rethrow_if_nested(e);
	} catch (...) {
	NESTED_THROW(hdfs::HdfsRpcException, "%s", e.what());
	}

	//should not reach here
	abort();
	}

	#define NAMENODE_HA_RETRY_BEGIN() \
	do { \
	int __count = 0; \
	do { \
	uint32_t __oldValue = 0; \
	shared_ptr<Namenode> namenode = getActiveNamenode(__oldValue); \
	try { \
	(void)0

	#define NAMENODE_HA_RETRY_END() \
	break; \
	} catch (const NameNodeStandbyException &e) { \
	if (!enableNamenodeHA \|\| __count++ > maxNamenodeHARetry) { \
	throw; \
	} \
	} catch (const HdfsFailoverException &e) { \
	if (!enableNamenodeHA \|\| __count++ > maxNamenodeHARetry) { \
	HandleHdfsFailoverException(e); \
	} \
	} \
	failoverToNextNamenode(__oldValue); \
	LOG(WARNING, "NamenodeProxy: Failover to another Namenode."); \
	} while (true); \
	} while (0)

	void NamenodeProxy::getBlockLocations(const std::string &src, int64_t offset,
	int64_t length, LocatedBlocks &lbs) {
	NAMENODE_HA_RETRY_BEGIN();
	namenode->getBlockLocations(src, offset, length, lbs);
	NAMENODE_HA_RETRY_END();
	}

	void NamenodeProxy::create(const std::string &src, const Permission &masked,
	const std::string &clientName, int flag, bool createParent,
	short replication, int64_t blockSize) {
	NAMENODE_HA_RETRY_BEGIN();
	namenode->create(src, masked, clientName, flag, createParent, replication, blockSize);
	NAMENODE_HA_RETRY_END();
	}

	shared_ptr<LocatedBlock> NamenodeProxy::append(const std::string &src,
	const std::string &clientName) {
	NAMENODE_HA_RETRY_BEGIN();
	return namenode->append(src, clientName);
	NAMENODE_HA_RETRY_END();
	assert(!"should not reach here");
	return shared_ptr<LocatedBlock>();
	}

	bool NamenodeProxy::setReplication(const std::string &src, short replication) {
	NAMENODE_HA_RETRY_BEGIN();
	return namenode->setReplication(src, replication);
	NAMENODE_HA_RETRY_END();
	assert(!"should not reach here");
	return false;
	}

	void NamenodeProxy::setPermission(const std::string &src,
	const Permission &permission) {
	NAMENODE_HA_RETRY_BEGIN();
	namenode->setPermission(src, permission);
	NAMENODE_HA_RETRY_END();
	}

	void NamenodeProxy::setOwner(const std::string &src,
	const std::string &username, const std::string &groupname) {
	NAMENODE_HA_RETRY_BEGIN();
	namenode->setOwner(src, username, groupname);
	NAMENODE_HA_RETRY_END();
	}

	void NamenodeProxy::abandonBlock(const ExtendedBlock &b,
	const std::string &src, const std::string &holder) {
	NAMENODE_HA_RETRY_BEGIN();
	namenode->abandonBlock(b, src, holder);
	NAMENODE_HA_RETRY_END();
	}

	shared_ptr<LocatedBlock> NamenodeProxy::addBlock(const std::string &src,
	const std::string &clientName, const ExtendedBlock * previous,
	const std::vector<DatanodeInfo> &excludeNodes) {
	NAMENODE_HA_RETRY_BEGIN();
	return namenode->addBlock(src, clientName, previous, excludeNodes);
	NAMENODE_HA_RETRY_END();
	assert(!"should not reach here");
	return shared_ptr<LocatedBlock>();
	}

	shared_ptr<LocatedBlock> NamenodeProxy::getAdditionalDatanode(
	const std::string &src, const ExtendedBlock &blk,
	const std::vector<DatanodeInfo> &existings,
	const std::vector<std::string> &storageIDs,
	const std::vector<DatanodeInfo> &excludes, int numAdditionalNodes,
	const std::string &clientName) {
	NAMENODE_HA_RETRY_BEGIN();
	return namenode->getAdditionalDatanode(src, blk, existings,
	storageIDs, excludes, numAdditionalNodes, clientName);
	NAMENODE_HA_RETRY_END();
	assert(!"should not reach here");
	return shared_ptr<LocatedBlock>();
	}

	bool NamenodeProxy::complete(const std::string &src,
	const std::string &clientName, const ExtendedBlock *last) {
	NAMENODE_HA_RETRY_BEGIN();
	return namenode->complete(src, clientName, last);
	NAMENODE_HA_RETRY_END();
	assert(!"should not reach here");
	return false;
	}

	/*void NamenodeProxy::reportBadBlocks(const std::vector<LocatedBlock> &blocks) {
	NAMENODE_HA_RETRY_BEGIN();
	namenode->reportBadBlocks(blocks);
	NAMENODE_HA_RETRY_END();
	}*/

	bool NamenodeProxy::rename(const std::string &src, const std::string &dst) {
	NAMENODE_HA_RETRY_BEGIN();
	return namenode->rename(src, dst);
	NAMENODE_HA_RETRY_END();
	assert(!"should not reach here");
	return false;
	}

	/*
	void NamenodeProxy::concat(const std::string &trg,
	const std::vector<std::string> &srcs) {
	NAMENODE_HA_RETRY_BEGIN();
	namenode->concat(trg, srcs);
	NAMENODE_HA_RETRY_END();
	}
	*/

	bool NamenodeProxy::deleteFile(const std::string &src, bool recursive) {
	NAMENODE_HA_RETRY_BEGIN();
	return namenode->deleteFile(src, recursive);
	NAMENODE_HA_RETRY_END();
	assert(!"should not reach here");
	return false;
	}

	bool NamenodeProxy::mkdirs(const std::string &src, const Permission &masked,
	bool createParent) {
	NAMENODE_HA_RETRY_BEGIN();
	return namenode->mkdirs(src, masked, createParent);
	NAMENODE_HA_RETRY_END();
	assert(!"should not reach here");
	return false;
	}

	bool NamenodeProxy::getListing(const std::string &src,
	const std::string &startAfter, bool needLocation,
	std::vector<FileStatus> &dl) {
	NAMENODE_HA_RETRY_BEGIN();
	return namenode->getListing(src, startAfter, needLocation, dl);
	NAMENODE_HA_RETRY_END();
	assert(!"should not reach here");
	return false;
	}

	void NamenodeProxy::renewLease(const std::string &clientName) {
	NAMENODE_HA_RETRY_BEGIN();
	namenode->renewLease(clientName);
	NAMENODE_HA_RETRY_END();
	}

	bool NamenodeProxy::recoverLease(const std::string &src,
	const std::string &clientName) {
	NAMENODE_HA_RETRY_BEGIN();
	return namenode->recoverLease(src, clientName);
	NAMENODE_HA_RETRY_END();
	assert(!"should not reach here");
	return false;
	}

	std::vector<int64_t> NamenodeProxy::getFsStats() {
	NAMENODE_HA_RETRY_BEGIN();
	return namenode->getFsStats();
	NAMENODE_HA_RETRY_END();
	assert(!"should not reach here");
	return std::vector<int64_t>();
	}

	/*void NamenodeProxy::metaSave(const std::string &filename) {
	NAMENODE_HA_RETRY_BEGIN();
	namenode->metaSave(filename);
	NAMENODE_HA_RETRY_END();
	}*/

	FileStatus NamenodeProxy::getFileInfo(const std::string &src) {
	NAMENODE_HA_RETRY_BEGIN();
	return namenode->getFileInfo(src);
	NAMENODE_HA_RETRY_END();
	assert(!"should not reach here");
	return FileStatus();
	}

	/*FileStatus NamenodeProxy::getFileLinkInfo(const std::string &src) {
	NAMENODE_HA_RETRY_BEGIN();
	return namenode->getFileLinkInfo(src);
	NAMENODE_HA_RETRY_END();
	assert(!"should not reach here");
	return FileStatus();
	}*/

	/*ContentSummary NamenodeProxy::getContentSummary(const std::string &path) {
	NAMENODE_HA_RETRY_BEGIN();
	return namenode->getContentSummary(path);
	NAMENODE_HA_RETRY_END();
	assert(!"should not reach here");
	return ContentSummary();
	}*/

	/*void NamenodeProxy::setQuota(const std::string &path, int64_t namespaceQuota,
	int64_t diskspaceQuota) {
	NAMENODE_HA_RETRY_BEGIN();
	namenode->setQuota(path, namespaceQuota, diskspaceQuota);
	NAMENODE_HA_RETRY_END();
	}*/

	void NamenodeProxy::fsync(const std::string &src, const std::string &client) {
	NAMENODE_HA_RETRY_BEGIN();
	namenode->fsync(src, client);
	NAMENODE_HA_RETRY_END();
	}

	void NamenodeProxy::setTimes(const std::string &src, int64_t mtime,
	int64_t atime) {
	NAMENODE_HA_RETRY_BEGIN();
	namenode->setTimes(src, mtime, atime);
	NAMENODE_HA_RETRY_END();
	}

	/*void NamenodeProxy::createSymlink(const std::string &target,
	const std::string &link, const Permission &dirPerm,
	bool createParent) {
	NAMENODE_HA_RETRY_BEGIN();
	namenode->createSymlink(target, link, dirPerm, createParent);
	NAMENODE_HA_RETRY_END();
	}*/

	/*std::string NamenodeProxy::getLinkTarget(const std::string &path) {
	NAMENODE_HA_RETRY_BEGIN();
	return namenode->getLinkTarget(path);
	NAMENODE_HA_RETRY_END();
	assert(!"should not reach here");
	return "";
	}*/

	shared_ptr<LocatedBlock> NamenodeProxy::updateBlockForPipeline(
	const ExtendedBlock &block, const std::string &clientName) {
	NAMENODE_HA_RETRY_BEGIN();
	return namenode->updateBlockForPipeline(block, clientName);
	NAMENODE_HA_RETRY_END();
	assert(!"should not reach here");
	return shared_ptr<LocatedBlock>();
	}

	void NamenodeProxy::updatePipeline(const std::string &clientName,
	const ExtendedBlock &oldBlock, const ExtendedBlock &newBlock,
	const std::vector<DatanodeInfo> &newNodes,
	const std::vector<std::string> &storageIDs) {
	NAMENODE_HA_RETRY_BEGIN();
	namenode->updatePipeline(clientName, oldBlock, newBlock,
	newNodes, storageIDs);
	NAMENODE_HA_RETRY_END();
	}

	Token NamenodeProxy::getDelegationToken(const std::string &renewer) {
	NAMENODE_HA_RETRY_BEGIN();
	return namenode->getDelegationToken(renewer);
	NAMENODE_HA_RETRY_END();
	assert(!"should not reach here");
	return Token();
	}

	int64_t NamenodeProxy::renewDelegationToken(const Token &token) {
	NAMENODE_HA_RETRY_BEGIN();
	return namenode->renewDelegationToken(token);
	NAMENODE_HA_RETRY_END();
	assert(!"should not reach here");
	return 0;
	}

	void NamenodeProxy::cancelDelegationToken(const Token &token) {
	NAMENODE_HA_RETRY_BEGIN();
	namenode->cancelDelegationToken(token);
	NAMENODE_HA_RETRY_END();
	}

	void NamenodeProxy::close() {
	lock_guard<mutex> lock(mut);
	namenodes.clear();
	}

	}
	}