You are viewing a plain text version of this content. The canonical link for it is here.
Posted to hdfs-commits@hadoop.apache.org by cn...@apache.org on 2014/01/07 02:07:38 UTC
svn commit: r1556097 [1/3] - in
/hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs: ./
src/main/java/ src/main/java/org/apache/hadoop/hdfs/
src/main/java/org/apache/hadoop/hdfs/protocolPB/
src/main/java/org/apache/hadoop/hdfs/server/bloc...
Author: cnauroth
Date: Tue Jan 7 01:07:33 2014
New Revision: 1556097
URL: http://svn.apache.org/r1556097
Log:
Merge trunk to HDFS-4685.
Added:
hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestStorageReport.java
- copied unchanged from r1556096, hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestStorageReport.java
Modified:
hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/ (props changed)
hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/ (props changed)
hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java
hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSInputStream.java
hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/PBHelper.java
hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/CacheReplicationMonitor.java
hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/DatanodeDescriptor.java
hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/DatanodeManager.java
hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsDatasetImpl.java
hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsVolumeImpl.java
hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/CacheManager.java
hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSImage.java
hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java
hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NameNode.java
hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/protocol/StorageReport.java
hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/tools/CacheAdmin.java
hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/tools/GetConf.java
hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/native/ (props changed)
hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/proto/DatanodeProtocol.proto
hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/resources/hdfs-default.xml
hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/webapps/datanode/ (props changed)
hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/webapps/hdfs/ (props changed)
hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/webapps/secondary/ (props changed)
hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/site/apt/CentralizedCacheManagement.apt.vm
hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/test/hdfs/ (props changed)
hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/MiniDFSCluster.java
hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/MiniDFSClusterWithNodeGroup.java
hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/blockmanagement/BlockManagerTestUtil.java
hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/common/TestJspHelper.java
hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/SimulatedFSDataset.java
hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDiskError.java
hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestFsDatasetCache.java
hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/NNThroughputBenchmark.java
hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/OfflineEditsViewerHelper.java
hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestCacheDirectives.java
hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestDeadDatanode.java
hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/ha/TestHAStateTransitions.java
hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/metrics/TestNameNodeMetrics.java
hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/tools/TestGetConf.java
hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/tools/offlineEditsViewer/TestOfflineEditsViewer.java
hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/test/resources/editsStored
hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/test/resources/editsStored.xml
Propchange: hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/
------------------------------------------------------------------------------
Merged /hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs:r1553225-1556096
Modified: hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt?rev=1556097&r1=1556096&r2=1556097&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt (original)
+++ hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt Tue Jan 7 01:07:33 2014
@@ -13,10 +13,6 @@ Trunk (Unreleased)
HDFS-3125. Add JournalService to enable Journal Daemon. (suresh)
- HDFS-2832. Heterogeneous Storages support in HDFS phase 1 - treat DataNode
- as a collection of storages (see breakdown of tasks below for features and
- contributors).
-
IMPROVEMENTS
HDFS-4665. Move TestNetworkTopologyWithNodeGroup to common.
@@ -243,13 +239,18 @@ Trunk (Unreleased)
HDFS-5636. Enforce a max TTL per cache pool. (awang via cmccabe)
+ HDFS-5651. Remove dfs.namenode.caching.enabled and improve CRM locking.
+ (cmccabe via wang)
+
OPTIMIZATIONS
+
HDFS-5349. DNA_CACHE and DNA_UNCACHE should be by blockId only. (cmccabe)
HDFS-5665. Remove the unnecessary writeLock while initializing CacheManager
in FsNameSystem Ctor. (Uma Maheswara Rao G via Andrew Wang)
BUG FIXES
+
HADOOP-9635 Fix potential Stack Overflow in DomainSocket.c (V. Karthik Kumar
via cmccabe)
@@ -444,139 +445,27 @@ Trunk (Unreleased)
HDFS-5626. dfsadmin -report shows incorrect cache values. (cmccabe)
- HDFS-5406. Send incremental block reports for all storages in a
- single call. (Arpit Agarwal)
-
- HDFS-5454. DataNode UUID should be assigned prior to FsDataset
- initialization. (Arpit Agarwal)
-
HDFS-5679. TestCacheDirectives should handle the case where native code
is not available. (wang)
- BREAKDOWN OF HDFS-2832 SUBTASKS AND RELATED JIRAS
-
- HDFS-4985. Add storage type to the protocol and expose it in block report
- and block locations. (Arpit Agarwal)
-
- HDFS-5115. Make StorageID a UUID. (Arpit Agarwal)
-
- HDFS-5000. DataNode configuration should allow specifying storage type.
- (Arpit Agarwal)
-
- HDFS-4987. Namenode changes to track multiple storages per datanode.
- (szetszwo)
-
- HDFS-5154. Fix TestBlockManager and TestDatanodeDescriptor after HDFS-4987.
- (Junping Du via szetszwo)
+ HDFS-5701. Fix the CacheAdmin -addPool -maxTtl option name.
+ (Stephen Chu via wang)
- HDFS-5009. Include storage information in the LocatedBlock. (szetszwo)
+ HDFS-5708. The CacheManager throws a NPE in the DataNode logs when
+ processing cache reports that refer to a block not known to the
+ BlockManager. (cmccabe via wang)
- HDFS-5134. Move blockContentsStale, heartbeatedSinceFailover and
- firstBlockReport from DatanodeDescriptor to DatanodeStorageInfo; and
- fix a synchronization problem in DatanodeStorageInfo. (szetszwo)
+ HDFS-5659. dfsadmin -report doesn't output cache information properly.
+ (wang)
- HDFS-5157. Add StorageType to FsVolume. (Junping Du via szetszwo)
+ HDFS-5705. TestSecondaryNameNodeUpgrade#testChangeNsIDFails may fail due
+ to ConcurrentModificationException. (Ted Yu via brandonli)
- HDFS-4990. Change BlockPlacementPolicy to choose storages instead of
- datanodes. (szetszwo)
+ HDFS-5719. FSImage#doRollback() should close prevState before return
+ (Ted Yu via brandonli)
- HDFS-5232. Protocol changes to transmit StorageUuid. (Arpit Agarwal)
-
- HDFS-5233. Use Datanode UUID to identify Datanodes. (Arpit Agarwal)
-
- HDFS-5222. Move block schedule information from DatanodeDescriptor to
- DatanodeStorageInfo. (szetszwo)
-
- HDFS-4988. Datanode must support all the volumes as individual storages.
- (Arpit Agarwal)
-
- HDFS-5377. Heartbeats from Datandode should include one storage report
- per storage directory. (Arpit Agarwal)
-
- HDFS-5398. NameNode changes to process storage reports per storage
- directory. (Arpit Agarwal)
-
- HDFS-5390. Send one incremental block report per storage directory.
- (Arpit Agarwal)
-
- HDFS-5401. Fix NPE in Directory Scanner. (Arpit Agarwal)
-
- HDFS-5417. Fix storage IDs in PBHelper and UpgradeUtilities. (szetszwo)
-
- HDFS-5214. Fix NPEs in BlockManager and DirectoryScanner. (Arpit Agarwal)
-
- HDFS-5435. File append fails to initialize storageIDs. (Junping Du via
- Arpit Agarwal)
-
- HDFS-5437. Fix TestBlockReport and TestBPOfferService failures. (Arpit
- Agarwal)
-
- HDFS-5447. Fix TestJspHelper. (Arpit Agarwal)
-
- HDFS-5452. Fix TestReplicationPolicy and TestBlocksScheduledCounter.
-
- HDFS-5448. Datanode should generate its ID on first registration. (Arpit
- Agarwal)
-
- HDFS-5448. Fix break caused by previous checkin for HDFS-5448. (Arpit
- Agarwal)
-
- HDFS-5455. NN should update storageMap on first heartbeat. (Arpit Agarwal)
-
- HDFS-5457. Fix TestDatanodeRegistration, TestFsck and TestAddBlockRetry.
- (Contributed by szetszwo)
-
- HDFS-5466. Update storage IDs when the pipeline is updated. (Contributed
- by szetszwo)
-
- HDFS-5439. Fix TestPendingReplication. (Contributed by Junping Du, Arpit
- Agarwal)
-
- HDFS-5470. Add back trunk's reportDiff algorithm to the branch.
- (Contributed by szetszwo)
-
- HDFS-5472. Fix TestDatanodeManager, TestSafeMode and
- TestNNThroughputBenchmark (Contributed by szetszwo)
-
- HDFS-5475. NN incorrectly tracks more than one replica per DN. (Arpit
- Agarwal)
-
- HDFS-5481. Fix TestDataNodeVolumeFailure in branch HDFS-2832. (Contributed
- by Junping Du)
-
- HDFS-5480. Update Balancer for HDFS-2832. (Contributed by szetszwo)
-
- HDFS-5486. Fix TestNameNodeMetrics for HDFS-2832. (Arpit Agarwal)
-
- HDFS-5491. Update editsStored for HDFS-2832. (Arpit Agarwal)
-
- HDFS-5494. Fix findbugs warnings for HDFS-2832. (Arpit Agarwal)
-
- HDFS-5508. Fix compilation error after merge. (Contributed by szetszwo)
-
- HDFS-5501. Fix pendingReceivedRequests tracking in BPServiceActor. (Arpit
- Agarwal)
-
- HDFS-5510. Fix a findbug warning in DataStorage.java on HDFS-2832 branch.
- (Junping Du via Arpit Agarwal)
-
- HDFS-5515. Fix TestDFSStartupVersions for HDFS-2832. (Arpit Agarwal)
-
- HDFS-5527. Fix TestUnderReplicatedBlocks on branch HDFS-2832. (Arpit
- Agarwal)
-
- HDFS-5547. Fix build break after merge from trunk to HDFS-2832. (Arpit
- Agarwal)
-
- HDFS-5542. Fix TODO and clean up the code in HDFS-2832. (Contributed by
- szetszwo)
-
- HDFS-5559. Fix TestDatanodeConfig in HDFS-2832. (Contributed by szetszwo)
-
- HDFS-5484. StorageType and State in DatanodeStorageInfo in NameNode is
- not accurate. (Eric Sirianni via Arpit Agarwal)
-
- HDFS-5648. Get rid of FsDatasetImpl#perVolumeReplicaMap. (Arpit Agarwal)
+ HDFS-5589. Namenode loops caching and uncaching when data should be
+ uncached (awang via cmccabe)
Release 2.4.0 - UNRELEASED
@@ -608,6 +497,10 @@ Release 2.4.0 - UNRELEASED
HDFS-5514. FSNamesystem's fsLock should allow custom implementation (daryn)
+ HDFS-2832. Heterogeneous Storages support in HDFS phase 1 - treat DataNode
+ as a collection of storages (see breakdown of tasks below for features and
+ contributors).
+
IMPROVEMENTS
HDFS-5267. Remove volatile from LightWeightHashSet. (Junping Du via llu)
@@ -618,9 +511,6 @@ Release 2.4.0 - UNRELEASED
HDFS-5004. Add additional JMX bean for NameNode status data
(Trevor Lorimer via cos)
- HDFS-5068. Convert NNThroughputBenchmark to a Tool to allow generic options.
- (shv)
-
HDFS-4994. Audit log getContentSummary() calls. (Robert Parker via kihwal)
HDFS-5144. Document time unit to NameNodeMetrics. (Akira Ajisaka via
@@ -768,6 +658,11 @@ Release 2.4.0 - UNRELEASED
HDFS-2933. Improve DataNode Web UI Index Page. (Vivek Ganesan via
Arpit Agarwal)
+ HDFS-5695. Clean up TestOfflineEditsViewer and OfflineEditsViewerHelper.
+ (Haohui Mai via jing9)
+
+ HDFS-5220. Expose group resolution time as metric (jxiang via cmccabe)
+
OPTIMIZATIONS
HDFS-5239. Allow FSNamesystem lock fairness to be configurable (daryn)
@@ -833,6 +728,139 @@ Release 2.4.0 - UNRELEASED
HDFS-5690. DataNode fails to start in secure mode when dfs.http.policy equals to
HTTP_ONLY. (Haohui Mai via jing9)
+ BREAKDOWN OF HDFS-2832 SUBTASKS AND RELATED JIRAS
+
+ HDFS-4985. Add storage type to the protocol and expose it in block report
+ and block locations. (Arpit Agarwal)
+
+ HDFS-5115. Make StorageID a UUID. (Arpit Agarwal)
+
+ HDFS-5000. DataNode configuration should allow specifying storage type.
+ (Arpit Agarwal)
+
+ HDFS-4987. Namenode changes to track multiple storages per datanode.
+ (szetszwo)
+
+ HDFS-5154. Fix TestBlockManager and TestDatanodeDescriptor after HDFS-4987.
+ (Junping Du via szetszwo)
+
+ HDFS-5009. Include storage information in the LocatedBlock. (szetszwo)
+
+ HDFS-5134. Move blockContentsStale, heartbeatedSinceFailover and
+ firstBlockReport from DatanodeDescriptor to DatanodeStorageInfo; and
+ fix a synchronization problem in DatanodeStorageInfo. (szetszwo)
+
+ HDFS-5157. Add StorageType to FsVolume. (Junping Du via szetszwo)
+
+ HDFS-4990. Change BlockPlacementPolicy to choose storages instead of
+ datanodes. (szetszwo)
+
+ HDFS-5232. Protocol changes to transmit StorageUuid. (Arpit Agarwal)
+
+ HDFS-5233. Use Datanode UUID to identify Datanodes. (Arpit Agarwal)
+
+ HDFS-5222. Move block schedule information from DatanodeDescriptor to
+ DatanodeStorageInfo. (szetszwo)
+
+ HDFS-4988. Datanode must support all the volumes as individual storages.
+ (Arpit Agarwal)
+
+ HDFS-5377. Heartbeats from Datandode should include one storage report
+ per storage directory. (Arpit Agarwal)
+
+ HDFS-5398. NameNode changes to process storage reports per storage
+ directory. (Arpit Agarwal)
+
+ HDFS-5390. Send one incremental block report per storage directory.
+ (Arpit Agarwal)
+
+ HDFS-5401. Fix NPE in Directory Scanner. (Arpit Agarwal)
+
+ HDFS-5417. Fix storage IDs in PBHelper and UpgradeUtilities. (szetszwo)
+
+ HDFS-5214. Fix NPEs in BlockManager and DirectoryScanner. (Arpit Agarwal)
+
+ HDFS-5435. File append fails to initialize storageIDs. (Junping Du via
+ Arpit Agarwal)
+
+ HDFS-5437. Fix TestBlockReport and TestBPOfferService failures. (Arpit
+ Agarwal)
+
+ HDFS-5447. Fix TestJspHelper. (Arpit Agarwal)
+
+ HDFS-5452. Fix TestReplicationPolicy and TestBlocksScheduledCounter.
+
+ HDFS-5448. Datanode should generate its ID on first registration. (Arpit
+ Agarwal)
+
+ HDFS-5448. Fix break caused by previous checkin for HDFS-5448. (Arpit
+ Agarwal)
+
+ HDFS-5455. NN should update storageMap on first heartbeat. (Arpit Agarwal)
+
+ HDFS-5457. Fix TestDatanodeRegistration, TestFsck and TestAddBlockRetry.
+ (Contributed by szetszwo)
+
+ HDFS-5466. Update storage IDs when the pipeline is updated. (Contributed
+ by szetszwo)
+
+ HDFS-5439. Fix TestPendingReplication. (Contributed by Junping Du, Arpit
+ Agarwal)
+
+ HDFS-5470. Add back trunk's reportDiff algorithm to the branch.
+ (Contributed by szetszwo)
+
+ HDFS-5472. Fix TestDatanodeManager, TestSafeMode and
+ TestNNThroughputBenchmark (Contributed by szetszwo)
+
+ HDFS-5475. NN incorrectly tracks more than one replica per DN. (Arpit
+ Agarwal)
+
+ HDFS-5481. Fix TestDataNodeVolumeFailure in branch HDFS-2832. (Contributed
+ by Junping Du)
+
+ HDFS-5480. Update Balancer for HDFS-2832. (Contributed by szetszwo)
+
+ HDFS-5486. Fix TestNameNodeMetrics for HDFS-2832. (Arpit Agarwal)
+
+ HDFS-5491. Update editsStored for HDFS-2832. (Arpit Agarwal)
+
+ HDFS-5494. Fix findbugs warnings for HDFS-2832. (Arpit Agarwal)
+
+ HDFS-5508. Fix compilation error after merge. (Contributed by szetszwo)
+
+ HDFS-5501. Fix pendingReceivedRequests tracking in BPServiceActor. (Arpit
+ Agarwal)
+
+ HDFS-5510. Fix a findbug warning in DataStorage.java on HDFS-2832 branch.
+ (Junping Du via Arpit Agarwal)
+
+ HDFS-5515. Fix TestDFSStartupVersions for HDFS-2832. (Arpit Agarwal)
+
+ HDFS-5527. Fix TestUnderReplicatedBlocks on branch HDFS-2832. (Arpit
+ Agarwal)
+
+ HDFS-5547. Fix build break after merge from trunk to HDFS-2832. (Arpit
+ Agarwal)
+
+ HDFS-5542. Fix TODO and clean up the code in HDFS-2832. (Contributed by
+ szetszwo)
+
+ HDFS-5559. Fix TestDatanodeConfig in HDFS-2832. (Contributed by szetszwo)
+
+ HDFS-5484. StorageType and State in DatanodeStorageInfo in NameNode is
+ not accurate. (Eric Sirianni via Arpit Agarwal)
+
+ HDFS-5648. Get rid of FsDatasetImpl#perVolumeReplicaMap. (Arpit Agarwal)
+
+ HDFS-5406. Send incremental block reports for all storages in a
+ single call. (Arpit Agarwal)
+
+ HDFS-5454. DataNode UUID should be assigned prior to FsDataset
+ initialization. (Arpit Agarwal)
+
+ HDFS-5667. Include DatanodeStorage in StorageReport. (Arpit Agarwal)
+
Release 2.3.0 - UNRELEASED
INCOMPATIBLE CHANGES
@@ -866,6 +894,12 @@ Release 2.3.0 - UNRELEASED
HDFS-5662. Can't decommission a DataNode due to file's replication factor
larger than the rest of the cluster size. (brandonli)
+ HDFS-5068. Convert NNThroughputBenchmark to a Tool to allow generic options.
+ (shv)
+
+ HDFS-5675. Add Mkdirs operation to NNThroughputBenchmark.
+ (Plamen Jeliazkov via shv)
+
OPTIMIZATIONS
BUG FIXES
@@ -1016,6 +1050,11 @@ Release 2.3.0 - UNRELEASED
HDFS-5661. Browsing FileSystem via web ui, should use datanode's fqdn instead of ip
address. (Benoy Antony via jing9)
+ HDFS-5582. hdfs getconf -excludeFile or -includeFile always failed (sathish
+ via cmccabe)
+
+ HDFS-5671. Fix socket leak in DFSInputStream#getBlockReader. (JamesLi via umamahesh)
+
Release 2.2.0 - 2013-10-13
INCOMPATIBLE CHANGES
Propchange: hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/
------------------------------------------------------------------------------
Merged /hadoop/common/trunk/hadoop-hdfs-project/hadoop-hdfs/src/main/java:r1553225-1556096
Modified: hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java?rev=1556097&r1=1556096&r2=1556097&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java (original)
+++ hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java Tue Jan 7 01:07:33 2014
@@ -108,8 +108,9 @@ public class DFSConfigKeys extends Commo
public static final long DFS_DATANODE_MAX_LOCKED_MEMORY_DEFAULT = 0;
public static final String DFS_DATANODE_FSDATASETCACHE_MAX_THREADS_PER_VOLUME_KEY = "dfs.datanode.fsdatasetcache.max.threads.per.volume";
public static final int DFS_DATANODE_FSDATASETCACHE_MAX_THREADS_PER_VOLUME_DEFAULT = 4;
- public static final String DFS_NAMENODE_CACHING_ENABLED_KEY = "dfs.namenode.caching.enabled";
- public static final boolean DFS_NAMENODE_CACHING_ENABLED_DEFAULT = false;
+ public static final String DFS_NAMENODE_PATH_BASED_CACHE_BLOCK_MAP_ALLOCATION_PERCENT =
+ "dfs.namenode.path.based.cache.block.map.allocation.percent";
+ public static final float DFS_NAMENODE_PATH_BASED_CACHE_BLOCK_MAP_ALLOCATION_PERCENT_DEFAULT = 0.25f;
public static final String DFS_NAMENODE_HTTP_PORT_KEY = "dfs.http.port";
public static final int DFS_NAMENODE_HTTP_PORT_DEFAULT = 50070;
Modified: hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSInputStream.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSInputStream.java?rev=1556097&r1=1556096&r2=1556097&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSInputStream.java (original)
+++ hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSInputStream.java Tue Jan 7 01:07:33 2014
@@ -1188,11 +1188,21 @@ implements ByteBufferReadable, CanSetDro
}
// Try to create a new remote peer.
Peer peer = newTcpPeer(dnAddr);
- return BlockReaderFactory.newBlockReader(
- dfsClient.getConf(), file, block, blockToken, startOffset,
- len, verifyChecksum, clientName, peer, chosenNode,
- dsFactory, peerCache, fileInputStreamCache, false,
- curCachingStrategy);
+ try {
+ reader = BlockReaderFactory.newBlockReader(dfsClient.getConf(), file,
+ block, blockToken, startOffset, len, verifyChecksum, clientName,
+ peer, chosenNode, dsFactory, peerCache, fileInputStreamCache, false,
+ curCachingStrategy);
+ return reader;
+ } catch (IOException ex) {
+ DFSClient.LOG.debug(
+ "Exception while getting block reader, closing stale " + peer, ex);
+ throw ex;
+ } finally {
+ if (reader == null) {
+ IOUtils.closeQuietly(peer);
+ }
+ }
}
Modified: hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/PBHelper.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/PBHelper.java?rev=1556097&r1=1556096&r2=1556097&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/PBHelper.java (original)
+++ hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/PBHelper.java Tue Jan 7 01:07:33 2014
@@ -533,21 +533,7 @@ public class PBHelper {
static public DatanodeInfoProto convertDatanodeInfo(DatanodeInfo di) {
if (di == null) return null;
- DatanodeInfoProto.Builder builder = DatanodeInfoProto.newBuilder();
- if (di.getNetworkLocation() != null) {
- builder.setLocation(di.getNetworkLocation());
- }
-
- return builder.
- setId(PBHelper.convert((DatanodeID) di)).
- setCapacity(di.getCapacity()).
- setDfsUsed(di.getDfsUsed()).
- setRemaining(di.getRemaining()).
- setBlockPoolUsed(di.getBlockPoolUsed()).
- setLastUpdate(di.getLastUpdate()).
- setXceiverCount(di.getXceiverCount()).
- setAdminState(PBHelper.convert(di.getAdminState())).
- build();
+ return convert(di);
}
@@ -591,15 +577,20 @@ public class PBHelper {
public static DatanodeInfoProto convert(DatanodeInfo info) {
DatanodeInfoProto.Builder builder = DatanodeInfoProto.newBuilder();
- builder.setBlockPoolUsed(info.getBlockPoolUsed());
- builder.setAdminState(PBHelper.convert(info.getAdminState()));
- builder.setCapacity(info.getCapacity())
- .setDfsUsed(info.getDfsUsed())
+ if (info.getNetworkLocation() != null) {
+ builder.setLocation(info.getNetworkLocation());
+ }
+ builder
.setId(PBHelper.convert((DatanodeID)info))
- .setLastUpdate(info.getLastUpdate())
- .setLocation(info.getNetworkLocation())
+ .setCapacity(info.getCapacity())
+ .setDfsUsed(info.getDfsUsed())
.setRemaining(info.getRemaining())
+ .setBlockPoolUsed(info.getBlockPoolUsed())
+ .setCacheCapacity(info.getCacheCapacity())
+ .setCacheUsed(info.getCacheUsed())
+ .setLastUpdate(info.getLastUpdate())
.setXceiverCount(info.getXceiverCount())
+ .setAdminState(PBHelper.convert(info.getAdminState()))
.build();
return builder.build();
}
@@ -1591,13 +1582,17 @@ public class PBHelper {
StorageReportProto.Builder builder = StorageReportProto.newBuilder()
.setBlockPoolUsed(r.getBlockPoolUsed()).setCapacity(r.getCapacity())
.setDfsUsed(r.getDfsUsed()).setRemaining(r.getRemaining())
- .setStorageUuid(r.getStorageID());
+ .setStorageUuid(r.getStorage().getStorageID())
+ .setStorage(convert(r.getStorage()));
return builder.build();
}
public static StorageReport convert(StorageReportProto p) {
- return new StorageReport(p.getStorageUuid(), p.getFailed(),
- p.getCapacity(), p.getDfsUsed(), p.getRemaining(),
+ return new StorageReport(
+ p.hasStorage() ?
+ convert(p.getStorage()) :
+ new DatanodeStorage(p.getStorageUuid()),
+ p.getFailed(), p.getCapacity(), p.getDfsUsed(), p.getRemaining(),
p.getBlockPoolUsed());
}
Modified: hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/CacheReplicationMonitor.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/CacheReplicationMonitor.java?rev=1556097&r1=1556096&r2=1556097&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/CacheReplicationMonitor.java (original)
+++ hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/CacheReplicationMonitor.java Tue Jan 7 01:07:33 2014
@@ -21,12 +21,14 @@ import static org.apache.hadoop.util.Exi
import java.io.Closeable;
import java.io.IOException;
+import java.util.ArrayList;
import java.util.Collection;
import java.util.Date;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Random;
+import java.util.TreeMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
@@ -76,7 +78,7 @@ public class CacheReplicationMonitor ext
/**
* Pseudorandom number source
*/
- private final Random random = new Random();
+ private static final Random random = new Random();
/**
* The interval at which we scan the namesystem for caching changes.
@@ -87,17 +89,17 @@ public class CacheReplicationMonitor ext
* The CacheReplicationMonitor (CRM) lock. Used to synchronize starting and
* waiting for rescan operations.
*/
- private final ReentrantLock lock = new ReentrantLock();
+ private final ReentrantLock lock;
/**
* Notifies the scan thread that an immediate rescan is needed.
*/
- private final Condition doRescan = lock.newCondition();
+ private final Condition doRescan;
/**
* Notifies waiting threads that a rescan has finished.
*/
- private final Condition scanFinished = lock.newCondition();
+ private final Condition scanFinished;
/**
* Whether there are pending CacheManager operations that necessitate a
@@ -122,11 +124,6 @@ public class CacheReplicationMonitor ext
private boolean shutdown = false;
/**
- * The monotonic time at which the current scan started.
- */
- private long startTimeMs;
-
- /**
* Mark status of the current scan.
*/
private boolean mark = false;
@@ -142,24 +139,27 @@ public class CacheReplicationMonitor ext
private long scannedBlocks;
public CacheReplicationMonitor(FSNamesystem namesystem,
- CacheManager cacheManager, long intervalMs) {
+ CacheManager cacheManager, long intervalMs, ReentrantLock lock) {
this.namesystem = namesystem;
this.blockManager = namesystem.getBlockManager();
this.cacheManager = cacheManager;
this.cachedBlocks = cacheManager.getCachedBlocks();
this.intervalMs = intervalMs;
+ this.lock = lock;
+ this.doRescan = this.lock.newCondition();
+ this.scanFinished = this.lock.newCondition();
}
@Override
public void run() {
- startTimeMs = 0;
+ long startTimeMs = 0;
+ Thread.currentThread().setName("CacheReplicationMonitor(" +
+ System.identityHashCode(this) + ")");
LOG.info("Starting CacheReplicationMonitor with interval " +
intervalMs + " milliseconds");
try {
long curTimeMs = Time.monotonicNow();
while (true) {
- // Not all of the variables accessed here need the CRM lock, but take
- // it anyway for simplicity
lock.lock();
try {
while (true) {
@@ -180,12 +180,6 @@ public class CacheReplicationMonitor ext
doRescan.await(delta, TimeUnit.MILLISECONDS);
curTimeMs = Time.monotonicNow();
}
- } finally {
- lock.unlock();
- }
- // Mark scan as started, clear needsRescan
- lock.lock();
- try {
isScanning = true;
needsRescan = false;
} finally {
@@ -195,7 +189,7 @@ public class CacheReplicationMonitor ext
mark = !mark;
rescan();
curTimeMs = Time.monotonicNow();
- // Retake the CRM lock to update synchronization-related variables
+ // Update synchronization-related variables.
lock.lock();
try {
isScanning = false;
@@ -208,6 +202,9 @@ public class CacheReplicationMonitor ext
scannedBlocks + " block(s) in " + (curTimeMs - startTimeMs) + " " +
"millisecond(s).");
}
+ } catch (InterruptedException e) {
+ LOG.info("Shutting down CacheReplicationMonitor.");
+ return;
} catch (Throwable t) {
LOG.fatal("Thread exiting", t);
terminate(1, t);
@@ -215,26 +212,6 @@ public class CacheReplicationMonitor ext
}
/**
- * Similar to {@link CacheReplicationMonitor#waitForRescan()}, except it only
- * waits if there are pending operations that necessitate a rescan as
- * indicated by {@link #setNeedsRescan()}.
- * <p>
- * Note that this call may release the FSN lock, so operations before and
- * after are not necessarily atomic.
- */
- public void waitForRescanIfNeeded() {
- lock.lock();
- try {
- if (!needsRescan) {
- return;
- }
- } finally {
- lock.unlock();
- }
- waitForRescan();
- }
-
- /**
* Waits for a rescan to complete. This doesn't guarantee consistency with
* pending operations, only relative recency, since it will not force a new
* rescan if a rescan is already underway.
@@ -242,49 +219,27 @@ public class CacheReplicationMonitor ext
* Note that this call will release the FSN lock, so operations before and
* after are not atomic.
*/
- public void waitForRescan() {
- // Drop the FSN lock temporarily and retake it after we finish waiting
- // Need to handle both the read lock and the write lock
- boolean retakeWriteLock = false;
- if (namesystem.hasWriteLock()) {
- namesystem.writeUnlock();
- retakeWriteLock = true;
- } else if (namesystem.hasReadLock()) {
- namesystem.readUnlock();
- } else {
- // Expected to have at least one of the locks
- Preconditions.checkState(false,
- "Need to be holding either the read or write lock");
+ public void waitForRescanIfNeeded() {
+ Preconditions.checkArgument(!namesystem.hasWriteLock(),
+ "Must not hold the FSN write lock when waiting for a rescan.");
+ Preconditions.checkArgument(lock.isHeldByCurrentThread(),
+ "Must hold the CRM lock when waiting for a rescan.");
+ if (!needsRescan) {
+ return;
}
- // try/finally for retaking FSN lock
- try {
- lock.lock();
- // try/finally for releasing CRM lock
+ // If no scan is already ongoing, mark the CRM as dirty and kick
+ if (!isScanning) {
+ doRescan.signal();
+ }
+ // Wait until the scan finishes and the count advances
+ final long startCount = scanCount;
+ while ((!shutdown) && (startCount >= scanCount)) {
try {
- // If no scan is already ongoing, mark the CRM as dirty and kick
- if (!isScanning) {
- needsRescan = true;
- doRescan.signal();
- }
- // Wait until the scan finishes and the count advances
- final long startCount = scanCount;
- while (startCount >= scanCount) {
- try {
- scanFinished.await();
- } catch (InterruptedException e) {
- LOG.warn("Interrupted while waiting for CacheReplicationMonitor"
- + " rescan", e);
- break;
- }
- }
- } finally {
- lock.unlock();
- }
- } finally {
- if (retakeWriteLock) {
- namesystem.writeLock();
- } else {
- namesystem.readLock();
+ scanFinished.await();
+ } catch (InterruptedException e) {
+ LOG.warn("Interrupted while waiting for CacheReplicationMonitor"
+ + " rescan", e);
+ break;
}
}
}
@@ -294,42 +249,43 @@ public class CacheReplicationMonitor ext
* changes that require a rescan.
*/
public void setNeedsRescan() {
- lock.lock();
- try {
- this.needsRescan = true;
- } finally {
- lock.unlock();
- }
+ Preconditions.checkArgument(lock.isHeldByCurrentThread(),
+ "Must hold the CRM lock when setting the needsRescan bit.");
+ this.needsRescan = true;
}
/**
- * Shut down and join the monitor thread.
+ * Shut down the monitor thread.
*/
@Override
public void close() throws IOException {
+ Preconditions.checkArgument(namesystem.hasWriteLock());
lock.lock();
try {
if (shutdown) return;
+ // Since we hold both the FSN write lock and the CRM lock here,
+ // we know that the CRM thread cannot be currently modifying
+ // the cache manager state while we're closing it.
+ // Since the CRM thread checks the value of 'shutdown' after waiting
+ // for a lock, we know that the thread will not modify the cache
+ // manager state after this point.
shutdown = true;
doRescan.signalAll();
scanFinished.signalAll();
} finally {
lock.unlock();
}
- try {
- if (this.isAlive()) {
- this.join(60000);
- }
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- }
}
- private void rescan() {
+ private void rescan() throws InterruptedException {
scannedDirectives = 0;
scannedBlocks = 0;
namesystem.writeLock();
try {
+ if (shutdown) {
+ throw new InterruptedException("CacheReplicationMonitor was " +
+ "shut down.");
+ }
resetStatistics();
rescanCacheDirectives();
rescanCachedBlockMap();
@@ -356,8 +312,6 @@ public class CacheReplicationMonitor ext
FSDirectory fsDir = namesystem.getFSDirectory();
final long now = new Date().getTime();
for (CacheDirective directive : cacheManager.getCacheDirectives()) {
- // Reset the directive's statistics
- directive.resetStatistics();
// Skip processing this entry if it has expired
if (LOG.isTraceEnabled()) {
LOG.trace("Directive expiry is at " + directive.getExpiryTime());
@@ -460,14 +414,21 @@ public class CacheReplicationMonitor ext
directive.getReplication()) * blockInfo.getNumBytes();
cachedTotal += cachedByBlock;
- if (mark != ocblock.getMark()) {
- // Mark hasn't been set in this scan, so update replication and mark.
+ if ((mark != ocblock.getMark()) ||
+ (ocblock.getReplication() < directive.getReplication())) {
+ //
+ // Overwrite the block's replication and mark in two cases:
+ //
+ // 1. If the mark on the CachedBlock is different from the mark for
+ // this scan, that means the block hasn't been updated during this
+ // scan, and we should overwrite whatever is there, since it is no
+ // longer valid.
+ //
+ // 2. If the replication in the CachedBlock is less than what the
+ // directive asks for, we want to increase the block's replication
+ // field to what the directive asks for.
+ //
ocblock.setReplicationAndMark(directive.getReplication(), mark);
- } else {
- // Mark already set in this scan. Set replication to highest value in
- // any CacheDirective that covers this file.
- ocblock.setReplicationAndMark((short)Math.max(
- directive.getReplication(), ocblock.getReplication()), mark);
}
}
}
@@ -483,6 +444,39 @@ public class CacheReplicationMonitor ext
}
}
+ private String findReasonForNotCaching(CachedBlock cblock,
+ BlockInfo blockInfo) {
+ if (blockInfo == null) {
+ // Somehow, a cache report with the block arrived, but the block
+ // reports from the DataNode haven't (yet?) described such a block.
+ // Alternately, the NameNode might have invalidated the block, but the
+ // DataNode hasn't caught up. In any case, we want to tell the DN
+ // to uncache this.
+ return "not tracked by the BlockManager";
+ } else if (!blockInfo.isComplete()) {
+ // When a cached block changes state from complete to some other state
+ // on the DataNode (perhaps because of append), it will begin the
+ // uncaching process. However, the uncaching process is not
+ // instantaneous, especially if clients have pinned the block. So
+ // there may be a period of time when incomplete blocks remain cached
+ // on the DataNodes.
+ return "not complete";
+ } else if (cblock.getReplication() == 0) {
+ // Since 0 is not a valid value for a cache directive's replication
+ // field, seeing a replication of 0 on a CacheBlock means that it
+ // has never been reached by any sweep.
+ return "not needed by any directives";
+ } else if (cblock.getMark() != mark) {
+ // Although the block was needed in the past, we didn't reach it during
+ // the current sweep. Therefore, it doesn't need to be cached any more.
+ // Need to set the replication to 0 so it doesn't flip back to cached
+ // when the mark flips on the next scan
+ cblock.setReplicationAndMark((short)0, mark);
+ return "no longer needed by any directives";
+ }
+ return null;
+ }
+
/**
* Scan through the cached block map.
* Any blocks which are under-replicated should be assigned new Datanodes.
@@ -508,11 +502,17 @@ public class CacheReplicationMonitor ext
iter.remove();
}
}
- // If the block's mark doesn't match with the mark of this scan, that
- // means that this block couldn't be reached during this scan. That means
- // it doesn't need to be cached any more.
- int neededCached = (cblock.getMark() != mark) ?
- 0 : cblock.getReplication();
+ BlockInfo blockInfo = blockManager.
+ getStoredBlock(new Block(cblock.getBlockId()));
+ String reason = findReasonForNotCaching(cblock, blockInfo);
+ int neededCached = 0;
+ if (reason != null) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("not caching " + cblock + " because it is " + reason);
+ }
+ } else {
+ neededCached = cblock.getReplication();
+ }
int numCached = cached.size();
if (numCached >= neededCached) {
// If we have enough replicas, drop all pending cached.
@@ -566,9 +566,6 @@ public class CacheReplicationMonitor ext
private void addNewPendingUncached(int neededUncached,
CachedBlock cachedBlock, List<DatanodeDescriptor> cached,
List<DatanodeDescriptor> pendingUncached) {
- if (!cacheManager.isActive()) {
- return;
- }
// Figure out which replicas can be uncached.
LinkedList<DatanodeDescriptor> possibilities =
new LinkedList<DatanodeDescriptor>();
@@ -601,19 +598,18 @@ public class CacheReplicationMonitor ext
* @param pendingCached A list of DataNodes that will soon cache the
* block.
*/
- private void addNewPendingCached(int neededCached,
+ private void addNewPendingCached(final int neededCached,
CachedBlock cachedBlock, List<DatanodeDescriptor> cached,
List<DatanodeDescriptor> pendingCached) {
- if (!cacheManager.isActive()) {
- return;
- }
// To figure out which replicas can be cached, we consult the
// blocksMap. We don't want to try to cache a corrupt replica, though.
BlockInfo blockInfo = blockManager.
getStoredBlock(new Block(cachedBlock.getBlockId()));
if (blockInfo == null) {
- LOG.debug("Not caching block " + cachedBlock + " because it " +
- "was deleted from all DataNodes.");
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Not caching block " + cachedBlock + " because there " +
+ "is no record of it on the NameNode.");
+ }
return;
}
if (!blockInfo.isComplete()) {
@@ -623,35 +619,156 @@ public class CacheReplicationMonitor ext
}
return;
}
- List<DatanodeDescriptor> possibilities = new LinkedList<DatanodeDescriptor>();
+ // Filter the list of replicas to only the valid targets
+ List<DatanodeDescriptor> possibilities =
+ new LinkedList<DatanodeDescriptor>();
int numReplicas = blockInfo.getCapacity();
Collection<DatanodeDescriptor> corrupt =
blockManager.getCorruptReplicas(blockInfo);
+ int outOfCapacity = 0;
for (int i = 0; i < numReplicas; i++) {
DatanodeDescriptor datanode = blockInfo.getDatanode(i);
- if ((datanode != null) &&
- ((!pendingCached.contains(datanode)) &&
- ((corrupt == null) || (!corrupt.contains(datanode))))) {
- possibilities.add(datanode);
+ if (datanode == null) {
+ continue;
}
- }
- while (neededCached > 0) {
- if (possibilities.isEmpty()) {
- LOG.warn("We need " + neededCached + " more replica(s) than " +
- "actually exist to provide a cache replication of " +
- cachedBlock.getReplication() + " for " + cachedBlock);
- return;
+ if (datanode.isDecommissioned() || datanode.isDecommissionInProgress()) {
+ continue;
}
- DatanodeDescriptor datanode =
- possibilities.remove(random.nextInt(possibilities.size()));
- if (LOG.isDebugEnabled()) {
- LOG.debug("AddNewPendingCached: datanode " + datanode +
- " will now cache block " + cachedBlock);
+ if (corrupt != null && corrupt.contains(datanode)) {
+ continue;
}
+ if (pendingCached.contains(datanode) || cached.contains(datanode)) {
+ continue;
+ }
+ long pendingCapacity = datanode.getCacheRemaining();
+ // Subtract pending cached blocks from effective capacity
+ Iterator<CachedBlock> it = datanode.getPendingCached().iterator();
+ while (it.hasNext()) {
+ CachedBlock cBlock = it.next();
+ BlockInfo info =
+ blockManager.getStoredBlock(new Block(cBlock.getBlockId()));
+ if (info != null) {
+ pendingCapacity -= info.getNumBytes();
+ }
+ }
+ it = datanode.getPendingUncached().iterator();
+ // Add pending uncached blocks from effective capacity
+ while (it.hasNext()) {
+ CachedBlock cBlock = it.next();
+ BlockInfo info =
+ blockManager.getStoredBlock(new Block(cBlock.getBlockId()));
+ if (info != null) {
+ pendingCapacity += info.getNumBytes();
+ }
+ }
+ if (pendingCapacity < blockInfo.getNumBytes()) {
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("Datanode " + datanode + " is not a valid possibility for"
+ + " block " + blockInfo.getBlockId() + " of size "
+ + blockInfo.getNumBytes() + " bytes, only has "
+ + datanode.getCacheRemaining() + " bytes of cache remaining.");
+ }
+ outOfCapacity++;
+ continue;
+ }
+ possibilities.add(datanode);
+ }
+ List<DatanodeDescriptor> chosen = chooseDatanodesForCaching(possibilities,
+ neededCached, blockManager.getDatanodeManager().getStaleInterval());
+ for (DatanodeDescriptor datanode : chosen) {
pendingCached.add(datanode);
boolean added = datanode.getPendingCached().add(cachedBlock);
assert added;
- neededCached--;
}
+ // We were unable to satisfy the requested replication factor
+ if (neededCached > chosen.size()) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(
+ "Only have " +
+ (cachedBlock.getReplication() - neededCached + chosen.size()) +
+ " of " + cachedBlock.getReplication() + " cached replicas for " +
+ cachedBlock + " (" + outOfCapacity + " nodes have insufficient " +
+ "capacity).");
+ }
+ }
+ }
+
+ /**
+ * Chooses datanode locations for caching from a list of valid possibilities.
+ * Non-stale nodes are chosen before stale nodes.
+ *
+ * @param possibilities List of candidate datanodes
+ * @param neededCached Number of replicas needed
+ * @param staleInterval Age of a stale datanode
+ * @return A list of chosen datanodes
+ */
+ private static List<DatanodeDescriptor> chooseDatanodesForCaching(
+ final List<DatanodeDescriptor> possibilities, final int neededCached,
+ final long staleInterval) {
+ // Make a copy that we can modify
+ List<DatanodeDescriptor> targets =
+ new ArrayList<DatanodeDescriptor>(possibilities);
+ // Selected targets
+ List<DatanodeDescriptor> chosen = new LinkedList<DatanodeDescriptor>();
+
+ // Filter out stale datanodes
+ List<DatanodeDescriptor> stale = new LinkedList<DatanodeDescriptor>();
+ Iterator<DatanodeDescriptor> it = targets.iterator();
+ while (it.hasNext()) {
+ DatanodeDescriptor d = it.next();
+ if (d.isStale(staleInterval)) {
+ it.remove();
+ stale.add(d);
+ }
+ }
+ // Select targets
+ while (chosen.size() < neededCached) {
+ // Try to use stale nodes if we're out of non-stale nodes, else we're done
+ if (targets.isEmpty()) {
+ if (!stale.isEmpty()) {
+ targets = stale;
+ } else {
+ break;
+ }
+ }
+ // Select a random target
+ DatanodeDescriptor target =
+ chooseRandomDatanodeByRemainingCapacity(targets);
+ chosen.add(target);
+ targets.remove(target);
+ }
+ return chosen;
+ }
+
+ /**
+ * Choose a single datanode from the provided list of possible
+ * targets, weighted by the percentage of free space remaining on the node.
+ *
+ * @return The chosen datanode
+ */
+ private static DatanodeDescriptor chooseRandomDatanodeByRemainingCapacity(
+ final List<DatanodeDescriptor> targets) {
+ // Use a weighted probability to choose the target datanode
+ float total = 0;
+ for (DatanodeDescriptor d : targets) {
+ total += d.getCacheRemainingPercent();
+ }
+ // Give each datanode a portion of keyspace equal to its relative weight
+ // [0, w1) selects d1, [w1, w2) selects d2, etc.
+ TreeMap<Integer, DatanodeDescriptor> lottery =
+ new TreeMap<Integer, DatanodeDescriptor>();
+ int offset = 0;
+ for (DatanodeDescriptor d : targets) {
+ // Since we're using floats, be paranoid about negative values
+ int weight =
+ Math.max(1, (int)((d.getCacheRemainingPercent() / total) * 1000000));
+ offset += weight;
+ lottery.put(offset, d);
+ }
+ // Choose a number from [0, offset), which is the total amount of weight,
+ // to select the winner
+ DatanodeDescriptor winner =
+ lottery.higherEntry(random.nextInt(offset)).getValue();
+ return winner;
}
}
Modified: hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/DatanodeDescriptor.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/DatanodeDescriptor.java?rev=1556097&r1=1556096&r2=1556097&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/DatanodeDescriptor.java (original)
+++ hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/DatanodeDescriptor.java Tue Jan 7 01:07:33 2014
@@ -355,11 +355,11 @@ public class DatanodeDescriptor extends
setLastUpdate(Time.now());
this.volumeFailures = volFailures;
for (StorageReport report : reports) {
- DatanodeStorageInfo storage = storageMap.get(report.getStorageID());
+ DatanodeStorageInfo storage = storageMap.get(report.getStorage().getStorageID());
if (storage == null) {
// This is seen during cluster initialization when the heartbeat
// is received before the initial block reports from each storage.
- storage = updateStorage(new DatanodeStorage(report.getStorageID()));
+ storage = updateStorage(report.getStorage());
}
storage.receivedHeartbeat(report);
totalCapacity += report.getCapacity();
Modified: hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/DatanodeManager.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/DatanodeManager.java?rev=1556097&r1=1556096&r2=1556097&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/DatanodeManager.java (original)
+++ hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/DatanodeManager.java Tue Jan 7 01:07:33 2014
@@ -1443,6 +1443,13 @@ public class DatanodeManager {
return getClass().getSimpleName() + ": " + host2DatanodeMap;
}
+ public void clearPendingCachingCommands() {
+ for (DatanodeDescriptor dn : datanodeMap.values()) {
+ dn.getPendingCached().clear();
+ dn.getPendingUncached().clear();
+ }
+ }
+
public void setShouldSendCachingCommands(boolean shouldSendCachingCommands) {
this.shouldSendCachingCommands = shouldSendCachingCommands;
}
Modified: hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsDatasetImpl.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsDatasetImpl.java?rev=1556097&r1=1556096&r2=1556097&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsDatasetImpl.java (original)
+++ hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsDatasetImpl.java Tue Jan 7 01:07:33 2014
@@ -121,7 +121,7 @@ class FsDatasetImpl implements FsDataset
reports = new StorageReport[volumes.volumes.size()];
int i = 0;
for (FsVolumeImpl volume : volumes.volumes) {
- reports[i++] = new StorageReport(volume.getStorageID(),
+ reports[i++] = new StorageReport(volume.toDatanodeStorage(),
false,
volume.getCapacity(),
volume.getDfsUsed(),
@@ -237,12 +237,9 @@ class FsDatasetImpl implements FsDataset
final List<FsVolumeImpl> volArray = new ArrayList<FsVolumeImpl>(
storage.getNumStorageDirs());
for (int idx = 0; idx < storage.getNumStorageDirs(); idx++) {
- // TODO: getStorageTypeFromLocations() is only a temporary workaround and
- // should be replaced with getting storage type from DataStorage (missing
- // storage type now) directly.
Storage.StorageDirectory sd = storage.getStorageDir(idx);
final File dir = sd.getCurrentDir();
- final StorageType storageType = getStorageTypeFromLocations(dataLocations, dir);
+ final StorageType storageType = getStorageTypeFromLocations(dataLocations, sd.getRoot());
volArray.add(new FsVolumeImpl(this, sd.getStorageUuid(), dir, conf,
storageType));
LOG.info("Added volume - " + dir + ", StorageType: " + storageType);
Modified: hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsVolumeImpl.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsVolumeImpl.java?rev=1556097&r1=1556096&r2=1556097&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsVolumeImpl.java (original)
+++ hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsVolumeImpl.java Tue Jan 7 01:07:33 2014
@@ -19,10 +19,10 @@ package org.apache.hadoop.hdfs.server.da
import java.io.File;
import java.io.IOException;
-import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executor;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
@@ -54,7 +54,7 @@ class FsVolumeImpl implements FsVolumeSp
private final String storageID;
private final StorageType storageType;
private final Map<String, BlockPoolSlice> bpSlices
- = new HashMap<String, BlockPoolSlice>();
+ = new ConcurrentHashMap<String, BlockPoolSlice>();
private final File currentDir; // <StorageDirectory>/current
private final DF usage;
private final long reserved;
Modified: hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/CacheManager.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/CacheManager.java?rev=1556097&r1=1556096&r2=1556097&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/CacheManager.java (original)
+++ hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/CacheManager.java Tue Jan 7 01:07:33 2014
@@ -17,8 +17,8 @@
*/
package org.apache.hadoop.hdfs.server.namenode;
-import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_NAMENODE_CACHING_ENABLED_DEFAULT;
-import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_NAMENODE_CACHING_ENABLED_KEY;
+import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_NAMENODE_PATH_BASED_CACHE_BLOCK_MAP_ALLOCATION_PERCENT;
+import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_NAMENODE_PATH_BASED_CACHE_BLOCK_MAP_ALLOCATION_PERCENT_DEFAULT;
import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_NAMENODE_LIST_CACHE_DIRECTIVES_NUM_RESPONSES;
import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_NAMENODE_LIST_CACHE_DIRECTIVES_NUM_RESPONSES_DEFAULT;
import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_NAMENODE_LIST_CACHE_POOLS_NUM_RESPONSES;
@@ -40,6 +40,7 @@ import java.util.List;
import java.util.Map.Entry;
import java.util.SortedMap;
import java.util.TreeMap;
+import java.util.concurrent.locks.ReentrantLock;
import org.apache.commons.io.IOUtils;
import org.apache.commons.logging.Log;
@@ -62,7 +63,6 @@ import org.apache.hadoop.hdfs.protocol.C
import org.apache.hadoop.hdfs.protocol.CachePoolInfo;
import org.apache.hadoop.hdfs.protocol.DatanodeID;
import org.apache.hadoop.hdfs.protocol.LocatedBlock;
-import org.apache.hadoop.hdfs.server.blockmanagement.BlockInfo;
import org.apache.hadoop.hdfs.server.blockmanagement.BlockManager;
import org.apache.hadoop.hdfs.server.blockmanagement.CacheReplicationMonitor;
import org.apache.hadoop.hdfs.server.blockmanagement.DatanodeDescriptor;
@@ -85,7 +85,7 @@ import com.google.common.annotations.Vis
/**
* The Cache Manager handles caching on DataNodes.
*
- * This class is instantiated by the FSNamesystem when caching is enabled.
+ * This class is instantiated by the FSNamesystem.
* It maintains the mapping of cached blocks to datanodes via processing
* datanode cache reports. Based on these reports and addition and removal of
* caching directives, we will schedule caching and uncaching work.
@@ -94,6 +94,8 @@ import com.google.common.annotations.Vis
public final class CacheManager {
public static final Log LOG = LogFactory.getLog(CacheManager.class);
+ private static final float MIN_CACHED_BLOCKS_PERCENT = 0.001f;
+
// TODO: add pending / underCached / schedule cached blocks stats.
/**
@@ -149,32 +151,14 @@ public final class CacheManager {
private final long scanIntervalMs;
/**
- * Whether caching is enabled.
- *
- * If caching is disabled, we will not process cache reports or store
- * information about what is cached where. We also do not start the
- * CacheReplicationMonitor thread. This will save resources, but provide
- * less functionality.
- *
- * Even when caching is disabled, we still store path-based cache
- * information. This information is stored in the edit log and fsimage. We
- * don't want to lose it just because a configuration setting was turned off.
- * However, we will not act on this information if caching is disabled.
- */
- private final boolean enabled;
-
- /**
- * Whether the CacheManager is active.
- *
- * When the CacheManager is active, it tells the DataNodes what to cache
- * and uncache. The CacheManager cannot become active if enabled = false.
+ * All cached blocks.
*/
- private boolean active = false;
+ private final GSet<CachedBlock, CachedBlock> cachedBlocks;
/**
- * All cached blocks.
+ * Lock which protects the CacheReplicationMonitor.
*/
- private final GSet<CachedBlock, CachedBlock> cachedBlocks;
+ private final ReentrantLock crmLock = new ReentrantLock();
/**
* The CacheReplicationMonitor.
@@ -195,54 +179,51 @@ public final class CacheManager {
scanIntervalMs = conf.getLong(
DFS_NAMENODE_PATH_BASED_CACHE_REFRESH_INTERVAL_MS,
DFS_NAMENODE_PATH_BASED_CACHE_REFRESH_INTERVAL_MS_DEFAULT);
- this.enabled = conf.getBoolean(DFS_NAMENODE_CACHING_ENABLED_KEY,
- DFS_NAMENODE_CACHING_ENABLED_DEFAULT);
- this.cachedBlocks = !enabled ? null :
- new LightWeightGSet<CachedBlock, CachedBlock>(
- LightWeightGSet.computeCapacity(0.25, "cachedBlocks"));
+ float cachedBlocksPercent = conf.getFloat(
+ DFS_NAMENODE_PATH_BASED_CACHE_BLOCK_MAP_ALLOCATION_PERCENT,
+ DFS_NAMENODE_PATH_BASED_CACHE_BLOCK_MAP_ALLOCATION_PERCENT_DEFAULT);
+ if (cachedBlocksPercent < MIN_CACHED_BLOCKS_PERCENT) {
+ LOG.info("Using minimum value " + MIN_CACHED_BLOCKS_PERCENT +
+ " for " + DFS_NAMENODE_PATH_BASED_CACHE_BLOCK_MAP_ALLOCATION_PERCENT);
+ cachedBlocksPercent = MIN_CACHED_BLOCKS_PERCENT;
+ }
+ this.cachedBlocks = new LightWeightGSet<CachedBlock, CachedBlock>(
+ LightWeightGSet.computeCapacity(cachedBlocksPercent,
+ "cachedBlocks"));
+
}
- /**
- * Activate the cache manager.
- *
- * When the cache manager is active, tell the datanodes where to cache files.
- */
- public void activate() {
- assert namesystem.hasWriteLock();
- if (enabled && (!active)) {
- LOG.info("Activating CacheManager. " +
- "Starting replication monitor thread...");
- active = true;
- monitor = new CacheReplicationMonitor(namesystem, this,
- scanIntervalMs);
- monitor.start();
+ public void startMonitorThread() {
+ crmLock.lock();
+ try {
+ if (this.monitor == null) {
+ this.monitor = new CacheReplicationMonitor(namesystem, this,
+ scanIntervalMs, crmLock);
+ this.monitor.start();
+ }
+ } finally {
+ crmLock.unlock();
}
}
- /**
- * Deactivate the cache manager.
- *
- * When the cache manager is inactive, it does not tell the datanodes where to
- * cache files.
- */
- public void deactivate() {
- assert namesystem.hasWriteLock();
- if (active) {
- LOG.info("Deactivating CacheManager. " +
- "stopping CacheReplicationMonitor thread...");
- active = false;
- IOUtils.closeQuietly(monitor);
- monitor = null;
- LOG.info("CacheReplicationMonitor thread stopped and deactivated.");
+ public void stopMonitorThread() {
+ crmLock.lock();
+ try {
+ if (this.monitor != null) {
+ CacheReplicationMonitor prevMonitor = this.monitor;
+ this.monitor = null;
+ IOUtils.closeQuietly(prevMonitor);
+ }
+ } finally {
+ crmLock.unlock();
}
}
- /**
- * Return true only if the cache manager is active.
- * Must be called under the FSN read or write lock.
- */
- public boolean isActive() {
- return active;
+ public void clearDirectiveStats() {
+ assert namesystem.hasWriteLock();
+ for (CacheDirective directive : directivesById.values()) {
+ directive.resetStatistics();
+ }
}
/**
@@ -481,9 +462,7 @@ public final class CacheManager {
directive.addBytesNeeded(stats.getBytesNeeded());
directive.addFilesNeeded(directive.getFilesNeeded());
- if (monitor != null) {
- monitor.setNeedsRescan();
- }
+ setNeedsRescan();
}
/**
@@ -515,10 +494,6 @@ public final class CacheManager {
long expiryTime = validateExpiryTime(info, pool.getMaxRelativeExpiryMs());
// Do quota validation if required
if (!flags.contains(CacheFlag.FORCE)) {
- // Can't kick and wait if caching is disabled
- if (monitor != null) {
- monitor.waitForRescan();
- }
checkLimit(pool, path, replication);
}
// All validation passed
@@ -623,9 +598,7 @@ public final class CacheManager {
validateExpiryTime(infoWithDefaults, destPool.getMaxRelativeExpiryMs());
// Indicate changes to the CRM
- if (monitor != null) {
- monitor.setNeedsRescan();
- }
+ setNeedsRescan();
// Validation passed
removeInternal(prevEntry);
@@ -660,9 +633,7 @@ public final class CacheManager {
pool.getDirectiveList().remove(directive);
assert directive.getPool() == null;
- if (monitor != null) {
- monitor.setNeedsRescan();
- }
+ setNeedsRescan();
}
public void removeDirective(long id, FSPermissionChecker pc)
@@ -695,9 +666,6 @@ public final class CacheManager {
if (filter.getReplication() != null) {
throw new IOException("Filtering by replication is unsupported.");
}
- if (monitor != null) {
- monitor.waitForRescanIfNeeded();
- }
ArrayList<CacheDirectiveEntry> replies =
new ArrayList<CacheDirectiveEntry>(NUM_PRE_ALLOCATED_ENTRIES);
int numReplies = 0;
@@ -806,9 +774,7 @@ public final class CacheManager {
bld.append(prefix).append("set limit to " + info.getLimit());
prefix = "; ";
// New limit changes stats, need to set needs refresh
- if (monitor != null) {
- monitor.setNeedsRescan();
- }
+ setNeedsRescan();
}
if (info.getMaxRelativeExpiryMs() != null) {
final Long maxRelativeExpiry = info.getMaxRelativeExpiryMs();
@@ -854,9 +820,7 @@ public final class CacheManager {
directivesById.remove(directive.getId());
iter.remove();
}
- if (monitor != null) {
- monitor.setNeedsRescan();
- }
+ setNeedsRescan();
} catch (IOException e) {
LOG.info("removeCachePool of " + poolName + " failed: ", e);
throw e;
@@ -867,9 +831,6 @@ public final class CacheManager {
public BatchedListEntries<CachePoolEntry>
listCachePools(FSPermissionChecker pc, String prevKey) {
assert namesystem.hasReadLock();
- if (monitor != null) {
- monitor.waitForRescanIfNeeded();
- }
final int NUM_PRE_ALLOCATED_ENTRIES = 16;
ArrayList<CachePoolEntry> results =
new ArrayList<CachePoolEntry>(NUM_PRE_ALLOCATED_ENTRIES);
@@ -885,9 +846,6 @@ public final class CacheManager {
}
public void setCachedLocations(LocatedBlock block) {
- if (!enabled) {
- return;
- }
CachedBlock cachedBlock =
new CachedBlock(block.getBlock().getBlockId(),
(short)0, false);
@@ -903,12 +861,6 @@ public final class CacheManager {
public final void processCacheReport(final DatanodeID datanodeID,
final List<Long> blockIds) throws IOException {
- if (!enabled) {
- LOG.info("Ignoring cache report from " + datanodeID +
- " because " + DFS_NAMENODE_CACHING_ENABLED_KEY + " = false. " +
- "number of blocks: " + blockIds.size());
- return;
- }
namesystem.writeLock();
final long startTime = Time.monotonicNow();
final long endTime;
@@ -940,39 +892,28 @@ public final class CacheManager {
final List<Long> blockIds) {
CachedBlocksList cached = datanode.getCached();
cached.clear();
+ CachedBlocksList cachedList = datanode.getCached();
+ CachedBlocksList pendingCachedList = datanode.getPendingCached();
for (Iterator<Long> iter = blockIds.iterator(); iter.hasNext(); ) {
- Block block = new Block(iter.next());
- BlockInfo blockInfo = blockManager.getStoredBlock(block);
- if (!blockInfo.isComplete()) {
- LOG.warn("Ignoring block id " + block.getBlockId() + ", because " +
- "it is in not complete yet. It is in state " +
- blockInfo.getBlockUCState());
- continue;
- }
- Collection<DatanodeDescriptor> corruptReplicas =
- blockManager.getCorruptReplicas(blockInfo);
- if ((corruptReplicas != null) && corruptReplicas.contains(datanode)) {
- // The NameNode will eventually remove or update the corrupt block.
- // Until then, we pretend that it isn't cached.
- LOG.warn("Ignoring cached replica on " + datanode + " of " + block +
- " because it is corrupt.");
- continue;
- }
+ long blockId = iter.next();
CachedBlock cachedBlock =
- new CachedBlock(block.getBlockId(), (short)0, false);
+ new CachedBlock(blockId, (short)0, false);
CachedBlock prevCachedBlock = cachedBlocks.get(cachedBlock);
- // Use the existing CachedBlock if it's present; otherwise,
- // insert a new one.
+ // Add the block ID from the cache report to the cachedBlocks map
+ // if it's not already there.
if (prevCachedBlock != null) {
cachedBlock = prevCachedBlock;
} else {
cachedBlocks.put(cachedBlock);
}
- if (!cachedBlock.isPresent(datanode.getCached())) {
- datanode.getCached().add(cachedBlock);
+ // Add the block to the datanode's implicit cached block list
+ // if it's not already there. Similarly, remove it from the pending
+ // cached block list if it exists there.
+ if (!cachedBlock.isPresent(cachedList)) {
+ cachedList.add(cachedBlock);
}
- if (cachedBlock.isPresent(datanode.getPendingCached())) {
- datanode.getPendingCached().remove(cachedBlock);
+ if (cachedBlock.isPresent(pendingCachedList)) {
+ pendingCachedList.remove(cachedBlock);
}
}
}
@@ -1097,4 +1038,36 @@ public final class CacheManager {
}
prog.endStep(Phase.LOADING_FSIMAGE, step);
}
+
+ public void waitForRescanIfNeeded() {
+ crmLock.lock();
+ try {
+ if (monitor != null) {
+ monitor.waitForRescanIfNeeded();
+ }
+ } finally {
+ crmLock.unlock();
+ }
+ }
+
+ private void setNeedsRescan() {
+ crmLock.lock();
+ try {
+ if (monitor != null) {
+ monitor.setNeedsRescan();
+ }
+ } finally {
+ crmLock.unlock();
+ }
+ }
+
+ @VisibleForTesting
+ public Thread getCacheReplicationMonitor() {
+ crmLock.lock();
+ try {
+ return monitor;
+ } finally {
+ crmLock.unlock();
+ }
+ }
}
Modified: hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSImage.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSImage.java?rev=1556097&r1=1556096&r2=1556097&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSImage.java (original)
+++ hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSImage.java Tue Jan 7 01:07:33 2014
@@ -405,60 +405,64 @@ public class FSImage implements Closeabl
// Directories that don't have previous state do not rollback
boolean canRollback = false;
FSImage prevState = new FSImage(conf);
- prevState.getStorage().layoutVersion = HdfsConstants.LAYOUT_VERSION;
- for (Iterator<StorageDirectory> it = storage.dirIterator(); it.hasNext();) {
- StorageDirectory sd = it.next();
- File prevDir = sd.getPreviousDir();
- if (!prevDir.exists()) { // use current directory then
- LOG.info("Storage directory " + sd.getRoot()
- + " does not contain previous fs state.");
- // read and verify consistency with other directories
- storage.readProperties(sd);
- continue;
- }
-
- // read and verify consistency of the prev dir
- prevState.getStorage().readPreviousVersionProperties(sd);
+ try {
+ prevState.getStorage().layoutVersion = HdfsConstants.LAYOUT_VERSION;
+ for (Iterator<StorageDirectory> it = storage.dirIterator(); it.hasNext();) {
+ StorageDirectory sd = it.next();
+ File prevDir = sd.getPreviousDir();
+ if (!prevDir.exists()) { // use current directory then
+ LOG.info("Storage directory " + sd.getRoot()
+ + " does not contain previous fs state.");
+ // read and verify consistency with other directories
+ storage.readProperties(sd);
+ continue;
+ }
- if (prevState.getLayoutVersion() != HdfsConstants.LAYOUT_VERSION) {
- throw new IOException(
- "Cannot rollback to storage version " +
- prevState.getLayoutVersion() +
- " using this version of the NameNode, which uses storage version " +
- HdfsConstants.LAYOUT_VERSION + ". " +
- "Please use the previous version of HDFS to perform the rollback.");
- }
- canRollback = true;
- }
- if (!canRollback)
- throw new IOException("Cannot rollback. None of the storage "
- + "directories contain previous fs state.");
+ // read and verify consistency of the prev dir
+ prevState.getStorage().readPreviousVersionProperties(sd);
- // Now that we know all directories are going to be consistent
- // Do rollback for each directory containing previous state
- for (Iterator<StorageDirectory> it = storage.dirIterator(); it.hasNext();) {
- StorageDirectory sd = it.next();
- File prevDir = sd.getPreviousDir();
- if (!prevDir.exists())
- continue;
-
- LOG.info("Rolling back storage directory " + sd.getRoot()
- + ".\n new LV = " + prevState.getStorage().getLayoutVersion()
- + "; new CTime = " + prevState.getStorage().getCTime());
- File tmpDir = sd.getRemovedTmp();
- assert !tmpDir.exists() : "removed.tmp directory must not exist.";
- // rename current to tmp
- File curDir = sd.getCurrentDir();
- assert curDir.exists() : "Current directory must exist.";
- NNStorage.rename(curDir, tmpDir);
- // rename previous to current
- NNStorage.rename(prevDir, curDir);
-
- // delete tmp dir
- NNStorage.deleteDir(tmpDir);
- LOG.info("Rollback of " + sd.getRoot()+ " is complete.");
+ if (prevState.getLayoutVersion() != HdfsConstants.LAYOUT_VERSION) {
+ throw new IOException(
+ "Cannot rollback to storage version " +
+ prevState.getLayoutVersion() +
+ " using this version of the NameNode, which uses storage version " +
+ HdfsConstants.LAYOUT_VERSION + ". " +
+ "Please use the previous version of HDFS to perform the rollback.");
+ }
+ canRollback = true;
+ }
+ if (!canRollback)
+ throw new IOException("Cannot rollback. None of the storage "
+ + "directories contain previous fs state.");
+
+ // Now that we know all directories are going to be consistent
+ // Do rollback for each directory containing previous state
+ for (Iterator<StorageDirectory> it = storage.dirIterator(); it.hasNext();) {
+ StorageDirectory sd = it.next();
+ File prevDir = sd.getPreviousDir();
+ if (!prevDir.exists())
+ continue;
+
+ LOG.info("Rolling back storage directory " + sd.getRoot()
+ + ".\n new LV = " + prevState.getStorage().getLayoutVersion()
+ + "; new CTime = " + prevState.getStorage().getCTime());
+ File tmpDir = sd.getRemovedTmp();
+ assert !tmpDir.exists() : "removed.tmp directory must not exist.";
+ // rename current to tmp
+ File curDir = sd.getCurrentDir();
+ assert curDir.exists() : "Current directory must exist.";
+ NNStorage.rename(curDir, tmpDir);
+ // rename previous to current
+ NNStorage.rename(prevDir, curDir);
+
+ // delete tmp dir
+ NNStorage.deleteDir(tmpDir);
+ LOG.info("Rollback of " + sd.getRoot()+ " is complete.");
+ }
+ isUpgradeFinalized = true;
+ } finally {
+ prevState.close();
}
- isUpgradeFinalized = true;
}
private void doFinalize(StorageDirectory sd) throws IOException {
Modified: hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java?rev=1556097&r1=1556096&r2=1556097&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java (original)
+++ hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java Tue Jan 7 01:07:33 2014
@@ -931,7 +931,6 @@ public class FSNamesystem implements Nam
writeLock();
try {
if (blockManager != null) blockManager.close();
- cacheManager.deactivate();
} finally {
writeUnlock();
}
@@ -1001,7 +1000,7 @@ public class FSNamesystem implements Nam
editLogRollerThreshold, editLogRollerInterval));
nnEditLogRoller.start();
- cacheManager.activate();
+ cacheManager.startMonitorThread();
blockManager.getDatanodeManager().setShouldSendCachingCommands(true);
} finally {
writeUnlock();
@@ -1052,7 +1051,9 @@ public class FSNamesystem implements Nam
// so that the tailer starts from the right spot.
dir.fsImage.updateLastAppliedTxIdFromWritten();
}
- cacheManager.deactivate();
+ cacheManager.stopMonitorThread();
+ cacheManager.clearDirectiveStats();
+ blockManager.getDatanodeManager().clearPendingCachingCommands();
blockManager.getDatanodeManager().setShouldSendCachingCommands(false);
} finally {
writeUnlock();
@@ -7066,6 +7067,9 @@ public class FSNamesystem implements Nam
return (Long) cacheEntry.getPayload();
}
boolean success = false;
+ if (!flags.contains(CacheFlag.FORCE)) {
+ cacheManager.waitForRescanIfNeeded();
+ }
writeLock();
Long result = null;
try {
@@ -7107,6 +7111,9 @@ public class FSNamesystem implements Nam
if (cacheEntry != null && cacheEntry.isSuccess()) {
return;
}
+ if (!flags.contains(CacheFlag.FORCE)) {
+ cacheManager.waitForRescanIfNeeded();
+ }
writeLock();
try {
checkOperation(OperationCategory.WRITE);
@@ -7166,6 +7173,7 @@ public class FSNamesystem implements Nam
final FSPermissionChecker pc = isPermissionEnabled ?
getPermissionChecker() : null;
BatchedListEntries<CacheDirectiveEntry> results;
+ cacheManager.waitForRescanIfNeeded();
readLock();
boolean success = false;
try {
@@ -7289,6 +7297,7 @@ public class FSNamesystem implements Nam
BatchedListEntries<CachePoolEntry> results;
checkOperation(OperationCategory.READ);
boolean success = false;
+ cacheManager.waitForRescanIfNeeded();
readLock();
try {
checkOperation(OperationCategory.READ);
Modified: hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NameNode.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NameNode.java?rev=1556097&r1=1556096&r2=1556097&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NameNode.java (original)
+++ hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NameNode.java Tue Jan 7 01:07:33 2014
@@ -480,6 +480,14 @@ public class NameNode implements NameNod
* @param conf the configuration
*/
protected void initialize(Configuration conf) throws IOException {
+ if (conf.get(HADOOP_USER_GROUP_METRICS_PERCENTILES_INTERVALS) == null) {
+ String intervals = conf.get(DFS_METRICS_PERCENTILES_INTERVALS_KEY);
+ if (intervals != null) {
+ conf.set(HADOOP_USER_GROUP_METRICS_PERCENTILES_INTERVALS,
+ intervals);
+ }
+ }
+
UserGroupInformation.setConfiguration(conf);
loginAsNameNodeUser(conf);
Modified: hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/protocol/StorageReport.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/protocol/StorageReport.java?rev=1556097&r1=1556096&r2=1556097&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/protocol/StorageReport.java (original)
+++ hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/protocol/StorageReport.java Tue Jan 7 01:07:33 2014
@@ -21,7 +21,7 @@ package org.apache.hadoop.hdfs.server.pr
* Utilization report for a Datanode storage
*/
public class StorageReport {
- private final String storageID;
+ private final DatanodeStorage storage;
private final boolean failed;
private final long capacity;
private final long dfsUsed;
@@ -30,9 +30,9 @@ public class StorageReport {
public static final StorageReport[] EMPTY_ARRAY = {};
- public StorageReport(String sid, boolean failed, long capacity, long dfsUsed,
- long remaining, long bpUsed) {
- this.storageID = sid;
+ public StorageReport(DatanodeStorage storage, boolean failed,
+ long capacity, long dfsUsed, long remaining, long bpUsed) {
+ this.storage = storage;
this.failed = failed;
this.capacity = capacity;
this.dfsUsed = dfsUsed;
@@ -40,8 +40,8 @@ public class StorageReport {
this.blockPoolUsed = bpUsed;
}
- public String getStorageID() {
- return storageID;
+ public DatanodeStorage getStorage() {
+ return storage;
}
public boolean isFailed() {
Modified: hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/tools/CacheAdmin.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/tools/CacheAdmin.java?rev=1556097&r1=1556096&r2=1556097&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/tools/CacheAdmin.java (original)
+++ hadoop/common/branches/HDFS-4685/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/tools/CacheAdmin.java Tue Jan 7 01:07:33 2014
@@ -84,7 +84,12 @@ public class CacheAdmin extends Configur
for (int j = 1; j < args.length; j++) {
argsList.add(args[j]);
}
- return command.run(getConf(), argsList);
+ try {
+ return command.run(getConf(), argsList);
+ } catch (IllegalArgumentException e) {
+ System.err.println(prettifyException(e));
+ return -1;
+ }
}
public static void main(String[] argsArray) throws IOException {
@@ -135,6 +140,20 @@ public class CacheAdmin extends Configur
return maxTtl;
}
+ private static Expiration parseExpirationString(String ttlString)
+ throws IOException {
+ Expiration ex = null;
+ if (ttlString != null) {
+ if (ttlString.equalsIgnoreCase("never")) {
+ ex = CacheDirectiveInfo.Expiration.NEVER;
+ } else {
+ long ttl = DFSUtil.parseRelativeTime(ttlString);
+ ex = CacheDirectiveInfo.Expiration.newRelative(ttl);
+ }
+ }
+ return ex;
+ }
+
interface Command {
String getName();
String getShortUsage();
@@ -171,6 +190,7 @@ public class CacheAdmin extends Configur
listing.addRow("<time-to-live>", "How long the directive is " +
"valid. Can be specified in minutes, hours, and days, e.g. " +
"30m, 4h, 2d. Valid units are [smhd]." +
+ " \"never\" indicates a directive that never expires." +
" If unspecified, the directive never expires.");
return getShortUsage() + "\n" +
"Add a new cache directive.\n\n" +
@@ -203,15 +223,15 @@ public class CacheAdmin extends Configur
}
String ttlString = StringUtils.popOptionWithArgument("-ttl", args);
- if (ttlString != null) {
- try {
- long ttl = DFSUtil.parseRelativeTime(ttlString);
- builder.setExpiration(CacheDirectiveInfo.Expiration.newRelative(ttl));
- } catch (IOException e) {
- System.err.println(
- "Error while parsing ttl value: " + e.getMessage());
- return 1;
+ try {
+ Expiration ex = parseExpirationString(ttlString);
+ if (ex != null) {
+ builder.setExpiration(ex);
}
+ } catch (IOException e) {
+ System.err.println(
+ "Error while parsing ttl value: " + e.getMessage());
+ return 1;
}
if (!args.isEmpty()) {
@@ -326,7 +346,7 @@ public class CacheAdmin extends Configur
listing.addRow("<time-to-live>", "How long the directive is " +
"valid. Can be specified in minutes, hours, and days, e.g. " +
"30m, 4h, 2d. Valid units are [smhd]." +
- " If unspecified, the directive never expires.");
+ " \"never\" indicates a directive that never expires.");
return getShortUsage() + "\n" +
"Modify a cache directive.\n\n" +
listing.toString();
@@ -362,17 +382,16 @@ public class CacheAdmin extends Configur
modified = true;
}
String ttlString = StringUtils.popOptionWithArgument("-ttl", args);
- if (ttlString != null) {
- long ttl;
- try {
- ttl = DFSUtil.parseRelativeTime(ttlString);
- } catch (IOException e) {
- System.err.println(
- "Error while parsing ttl value: " + e.getMessage());
- return 1;
+ try {
+ Expiration ex = parseExpirationString(ttlString);
+ if (ex != null) {
+ builder.setExpiration(ex);
+ modified = true;
}
- builder.setExpiration(CacheDirectiveInfo.Expiration.newRelative(ttl));
- modified = true;
+ } catch (IOException e) {
+ System.err.println(
+ "Error while parsing ttl value: " + e.getMessage());
+ return 1;
}
if (!args.isEmpty()) {
System.err.println("Can't understand argument: " + args.get(0));
@@ -578,7 +597,7 @@ public class CacheAdmin extends Configur
public String getShortUsage() {
return "[" + NAME + " <name> [-owner <owner>] " +
"[-group <group>] [-mode <mode>] [-limit <limit>] " +
- "[-maxttl <maxTtl>]\n";
+ "[-maxTtl <maxTtl>]\n";
}
@Override