Thanks for making progress on this Walter. Could you elaborate a bit how this JIRA is related to HDFS-7613?

Thanks for noticing. BlockPlacementPolicyDefault already ensures that each Node has one replica at most. I open this jira to try to ensures that each rack has one replica at most. If we short of racks, We want to place replicas on racks evenly. I think it will work, on ec files.

Thanks Walter for the patch! The logic looks good.

It seems chooseTargetFaultTolerant is similar to the private chooseTarget method in BlockPlacementPolicyDefault. I wonder if we can reuse that method by setting the maxNodesPerRack argument?

Thanks Zhe Zhang, the approach is much simple. Could you help me revert HADOOP-11676? I'll close that and rewrite the patch in HDFS-7891.
This is What I thought:
1. If I decide to use sorted rack method. I have to optimized rack accessing in
NetworkTopology.countNumOfAvailableNodes() and NetworkTopology.getNode() , and Add HADOOP-11676. I'm not very comfortable with that.
2. I think random policy with maxNodesPerRack works well for EC. If the cluster is small and the burden on Namenode is small, the worst case doesn't look so bad. If the cluster is midium/large, the policy won't hit worst case. You can check testresult.txt

3rd. network topology of 1k nodes cannot compare to the metadata of 100m files. Block placement policy costs little CPU time, it will not become the bottleneck of Namenode. This is another reason I thought maxNodesPerRack method will be fine.

I upload a new patch HDFS-7891.003.patch. It use random policy with maxNodesPerRack .

Design
R = total racks in cluster
X = total expected replicas
Q=Math.floor(X/R)
T=X%R

if X<R, X racks have 1 replica.
if X>R && T==0, R racks have Q replicas.
if X>R && T!=0,, X-T racks have Q replicas. T racks have Q+1 replicas.

Coding
1. Add a function getMaxNodesPerRack(..)
2. Call getMaxNodesPerRack(..) before every choosing

Defects
I copy a lot of code from BlockPlacementPolicyDefault . If I didn't do this, I have to modify BlockPlacementPolicyDefault so it can call an empty getMaxNodesPerRack(..) before every choosing, and I can overide it on child class.

4th. Another reason I prefer random policy with maxNodesPerRack is that I don't have to worry about balance. The random algorithm chooses node from all nodes. It doesn't chooses rack from all racks. If the distribution of nodes on racks is skew, the policy will keep space balance.

The previous testresult.txt is wrong. I simply let maxNodesPerRack=Math.ceil(X/R), If choosing 14 targets from 13 racks, maxNodesPerRack will be 2, and the number of invocation of random is much less. It looks fast, but it's wrong.
I test again with {{ HDFS-7891.003.patch}} , and the result is correct, but not good. So I'm re-considering sorted rack method.

The CPU time of random policy with maxNodesPerRack is less than 1/10 msec per call. Is it necessary using sorted rack method? Maybe not, IMO.

update test result. Add BlockPlacementDefault tests for comparison.

Great analysis Walter Su!

It seems the 003 patch has removed rack2hosts. So the sortedRack results were obtained with 002 patch right?

Conceptually, I think the sortedRack method makes sense for EC. It essentially introduces 2 levels of random selection: first choosing a rack and the choosing a node in the selected rack. This is much more efficient than selecting a random node from the entire cluster with the rack constraint. In particular, in a typical setup, the number of racks in the cluster should be close to the EC width. So the choice of racks should be easy (e.g., choosing 14 from 15). Its performance benefit should be even larger if you have more nodes per rack, like 20.

So I think the question is whether we can have a simpler implementation of the sortedRack method, without duplicating a lot of code.

the sortedRack results were obtained from 002 patch and many optimizations for rack accessing in NetworkTopology . The optimizations is not include in 002 patch.
002 patch has a severe problem. It doesn't handle space balance well. ( shuffle top 1/4)
The result from 003 patch which using maxNodesPerRack method is close to BlockPlacementPolicyDefault. The performance is not so bad. I think performance is not our first concern.
004 patch remove the dulplicated code in 003 patch.
004 patch has better simplicity and space balance than 002 patch.
Hi, Zhe Zhang. Could you help me revert HADOOP-11676?

Thanks Walter for the new patch!

I just reverted HADOOP-11676 in the weekly rebase. Will post my review on 004 patch soon.

Hi Walter, thanks for working on this JIRA. For replication, we allow 2 replicas stored under the same rack. How about we do the same for EC block group?

I understand that the current approach tries to do best fault tolerance. However, it may not be the best policy in practice since such policy generates more remote rack traffic. Also (6,3)-Reed-Solomon already can tolerate one more failure compared to 3-replication. I think the default block placement policy should allow maximum 2 replicas per rack. What do you think?

Thanks Tsz for analysis so deeply. This is a great thought about how we can optimize the policy. But I think place 2 replica under same rack doesn't reduce traffic.
Why the default policy place 1st replica in local rack. And 2nd and 3rd in the same remote rack? Why 3rd in the same rack with 2nd. Because the pipeline mechanism allows 3rd replica comes from 2nd. And also because the content is the same. Because three replica is the same block. 3rd replica is not from client.
Come back to this jira and EC group. I think when we use stripe layout, even when stripe layout is not used. All 6+3 replica(it's better called blocks belonging to same group) comes from client. As long as the replica is place on the remote rack, the traffic is the same.
Correct me if I was wrong.

Thanks Nicholas and Walter for the discussion!

I think when we use stripe layout, even when stripe layout is not used. All 6+3 replica(it's better called blocks belonging to same group) comes from client. As long as the replica is place on the remote rack, the traffic is the same.

Under the striping layout, in initial file creation, all raw or parity data comes directly from client. So inter-rack traffic won't be affected by placement policy (unless we consider the factor that client is likely co-located with 1st DN).

Two other scenarios do incur inter-DN traffic:

1. Background block reconstruction. Here a new DN is chosen as reconstruction destination and will pull data from other DNs in the group
2. EC+contiguous layout. Here all blocks are initially triplicated. Then for each formed group, a new DN is selected as the encoder, which pulls data from raw data blocks

Walter Su Below are my comments about 004 patch:

1. I guess the performance should be similar to BlockPlacementPolicyDefault?
2. The getAdaptiveMaxNodesPerRack method is a little awkward: among the 3 arguments, the super method uses 1 and the subclass uses the other 2.
3. The chooseTargetInOrder method in both super class and subclass look good to me.
4. Can we just override getMaxNodesPerRack with the logic in BlockPlacementPolicyFaultTolerant#getAdaptiveMaxNodesPerRack? If necessary we can change its signature.

In general, I agree it's a good idea to have a simple solution with clean code structure first, and worry about performance later (if we observe an performance issue we can always add the sorted rack method back).

> ... . But I think place 2 replica under same rack doesn't reduce traffic. ...

It does reduce a remote rack traffic in various cases shown below. Do you agree?
Keys: LN = local node, LR = local rack, RR = remote rack

Thanks for the analysis Nicholas.

I guess you are assuming a placement policy of LN+LR+RR_1+RR_2...? The current BlockPlacementPolicyDefault looks like LN+RR_1+RR_1

The current BlockPlacementPolicyDefault looks like LN+RR_1+RR_1

Yes, it is. The traffic of LN+RR_1+RR_1 is the same as LN + LR +RR_1. The former is better than the latter because when the replica factor is 2, two replica will be placed in LN+RR_1. But The numOfTargets that EC group chooses apparently bigger than 2.
I think Nicholas is trying to say, we can cut down one inter-rack traffic.

Is "inter-rack" means rack-to-rack? I mean rack-to-rack when I said "inter-rack".

If we put 2 replicas on local rack during file creation, 1 LN + 1 LR + RR_1 + RR_2 + RR_3 + RR_4..... Only local rack is allowed 2 replica, and remote rack is allowed 1 replica. We can tolerant 2 racks failure(out of 8 racks). It reduces 1 rack traffic during file creation, but will not during Reconstruction.
If we put 2 replicas on every racks during file creation and Reconstruction. 1 LN + 1 LR + RR_1 + RR_1 + RR_2 + RR_2. Both file creation and Reconstruction process can reduce 1 rack traffic. We can tolerant 1 rack failure(out of 4 racks).
The default policy can tolerant 1 rack failure(out of 2 racks).

I think 2 out of 8 is better than 1 out of 4. Because the former one can detect 1 failure periodically and fix it, so 2 fails at the same time is more unlikely.

And the risk of 1 failure out of 4 is higher than 1 out of 2.

Thanks Walter for the analysis. I suggest we implement a best fault tolerance policy first, spanning as many racks as there are available. Then we can add another policy with weaker FT but less traffic. In that new policy, I think the target is to have at least the same level of FT as 3-way replication (tolerating 1 rack failure, or 2 concurrent node failures).

First of all, the "best fault tolerance" is for improving rack failure tolerance but not for other case. If you agree, we should as well update the summary of this JIRA.

For rack failure, although it does happen but is uncommon. Also, when a rack fails, the data stored under the rack is only temporarily unavailable but not permanently lost.

For replication==3, we store block replicas only in two different racks. It there are two racks failed at the same time. Some data is unavailable. We should do the same thing for EC blocks.

I don't object adding the new 1 ec block per rack policy here but I believe the default policy should allow max 2 ec blocks per rack instead of 1.

Thanks Nicholas for the feedback! Seems the conclusion is to 1) add a 1 block per rack policy under this JIRA, 2) update the summary to reflect this change, and 3) file a follow on JIRA to develop an optimized EC placement policy with less bandwidth usage.

Let's add the best rack failure tolerance for now. We could add the policy allowing two racks later.

Revised summary.

BTW, could we move this to trunk? The policy is not specific to EC blocks.

Thanks for the analysis Nicholas. I agree we should move this to trunk and also use this policy in EC at this stage. We should keep thinking about the default policy.

This jira focuses on one node per rack policy. Discussion about EC specific policy moves to HDFS-7613. Welcome any discussion there. Thanks.

005 patch is ready for review.
NOTE. 005 patch doesn't include any delete policy. The delete policy from default policy deletes replica from rack which have more than one replica first. Even through it maybe prefer deleting from A(has 2 replicas) to deleting from B(has 3 replicas). But it always prefer deleting from A(has 2 replicas) to deleting from C(has 1 replica). So it's still best rack failure tolerance.

To trigger Jenkins..

Submitting a duplicate patch of 005 to trigger Jenkins

Thanks for the patch Walter! The patch looks good overall. Please find my review below.

Logic:

1. I think the current flow is functionally correct. But maxNodesPerRack no longer denotes the maximum nodes to be allocated per rack, making the code less readable. Right now chooseTargetInOrder treats maxNodesPerRack as the number of replicas to allocate to each rack in the first round.
2. If we simply let getMaxNodesPerRack() return the ceiling of totalNumOfReplicas/numOfRacks, like (totalNumOfReplicas - 1) / numOfRacks + 1, then do not override chooseTargetInOrder(), can we get the right distribution we want?

Nits:

1. Override annotation missing in BlockPlacementPolicyRackFaultTolarent#chooseTargetInOrder
2. Could remove the empty constructor of BlockPlacementPolicyRackFaultTolarent
3. {{chooseOnce}| needs some Javadoc.

Tsz Wo Nicholas Sze: Since this is targeted for trunk now, it would be great if you can review the patch as well. Thanks!

-1 overall. Here are the results of testing the latest attachment
http://issues.apache.org/jira/secure/attachment/12725649/HDFS-7891.005.dup.patch
against trunk revision 9e8309a.

+1 @author. The patch does not contain any @author tags.

+1 tests included. The patch appears to include 1 new or modified test files.

+1 javac. The applied patch does not increase the total number of javac compiler warnings.

+1 javadoc. There were no new javadoc warning messages.

+1 eclipse:eclipse. The patch built with eclipse:eclipse.

+1 findbugs. The patch does not introduce any new Findbugs (version 2.0.3) warnings.

+1 release audit. The applied patch does not increase the total number of release audit warnings.

-1 core tests. The patch failed these unit tests in hadoop-hdfs-project/hadoop-hdfs:

org.apache.hadoop.hdfs.TestFileCreation

Test results: https://builds.apache.org/job/PreCommit-HDFS-Build/10281//testReport/
Console output: https://builds.apache.org/job/PreCommit-HDFS-Build/10281//console

This message is automatically generated.

If we simply let getMaxNodesPerRack() return the ceiling of totalNumOfReplicas/numOfRacks, like (totalNumOfReplicas - 1) / numOfRacks + 1, then do not override chooseTargetInOrder(), can we get the right distribution we want?

If totalNumOfReplicas < numOfRacks, it's much simpler, maxNodesPerRack==1. It's usually the case in production cluster. The tricky thing is, when totalNumOfReplicas > numOfRacks.
Notation
R = total racks in cluster
X = total expected replicas
Q=Math.floor(X/R)
T=X%R
if X<R, X racks have 1 replica.
if X>R && T==0, R racks have Q replicas.
if X>R && T!=0, X-T racks have Q replicas. T racks have Q+1 replicas.

The tricky thing is,
1. if X>R && T!=0, X-T racks have Q replicas. T racks have Q+1 replicas. We want local rack in the T racks group. So we have to override chooseTargetInOrder() anyway.
2. if X>R && T!=0, and if "we simply let getMaxNodesPerRack() return the ceiling of totalNumOfReplicas/numOfRacks", there is a chance that more than T racks have Q+1 replicas while some racks have none.

We want local rack in the T racks group.

Sorry, I need to correct myself. Actually it doesn't matter in normal case because of pipeline. It might be useful in specific like EC because every replica is an individual block. I still need to override chooseTargetInOrder() for 2nd reason.

006 patch updates:

maxNodesPerRack no longer denotes the maximum nodes to be allocated per rack.

1. fixed. Also because maxNodesPerRack used for choosing favored Nodes. To avoid any bugs, 006 patch fixed it.
2. fixed all three Nits.
3. add more test cases.

+1 overall. Here are the results of testing the latest attachment
http://issues.apache.org/jira/secure/attachment/12725788/HDFS-7891.006.patch
against trunk revision 1b89a3e.

+1 @author. The patch does not contain any @author tags.

+1 tests included. The patch appears to include 1 new or modified test files.

+1 javac. The applied patch does not increase the total number of javac compiler warnings.

+1 javadoc. There were no new javadoc warning messages.

+1 eclipse:eclipse. The patch built with eclipse:eclipse.

+1 findbugs. The patch does not introduce any new Findbugs (version 2.0.3) warnings.

+1 release audit. The applied patch does not increase the total number of release audit warnings.

+1 core tests. The patch passed unit tests in hadoop-hdfs-project/hadoop-hdfs.

Test results: https://builds.apache.org/job/PreCommit-HDFS-Build/10287//testReport/
Console output: https://builds.apache.org/job/PreCommit-HDFS-Build/10287//console

This message is automatically generated.

+1 the patch looks good.

I have committed this. Thanks, Walter!

Zhe, thanks also for reviewing the patch.

SUCCESS: Integrated in Hadoop-trunk-Commit #7601 (See https://builds.apache.org/job/Hadoop-trunk-Commit/7601/)
HDFS-7891. A block placement policy with best rack failure tolerance. Contributed by Walter Su (szetszwo: rev 9595cc003ca5ed3d59b6942056a4fcb9080f79c9)

• hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestBlockPlacementPolicyRackFaultTolarent.java
• hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/BlockPlacementPolicyDefault.java
• hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/BlockPlacementPolicyRackFaultTolarent.java
• hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt

FAILURE: Integrated in Hadoop-Yarn-trunk-Java8 #166 (See https://builds.apache.org/job/Hadoop-Yarn-trunk-Java8/166/)
HDFS-7891. A block placement policy with best rack failure tolerance. Contributed by Walter Su (szetszwo: rev 9595cc003ca5ed3d59b6942056a4fcb9080f79c9)

• hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/BlockPlacementPolicyRackFaultTolarent.java
• hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestBlockPlacementPolicyRackFaultTolarent.java
• hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
• hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/BlockPlacementPolicyDefault.java

FAILURE: Integrated in Hadoop-Yarn-trunk #900 (See https://builds.apache.org/job/Hadoop-Yarn-trunk/900/)
HDFS-7891. A block placement policy with best rack failure tolerance. Contributed by Walter Su (szetszwo: rev 9595cc003ca5ed3d59b6942056a4fcb9080f79c9)

• hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestBlockPlacementPolicyRackFaultTolarent.java
• hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/BlockPlacementPolicyRackFaultTolarent.java
• hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
• hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/BlockPlacementPolicyDefault.java

FAILURE: Integrated in Hadoop-Hdfs-trunk #2098 (See https://builds.apache.org/job/Hadoop-Hdfs-trunk/2098/)
HDFS-7891. A block placement policy with best rack failure tolerance. Contributed by Walter Su (szetszwo: rev 9595cc003ca5ed3d59b6942056a4fcb9080f79c9)

• hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestBlockPlacementPolicyRackFaultTolarent.java
• hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/BlockPlacementPolicyDefault.java
• hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/BlockPlacementPolicyRackFaultTolarent.java
• hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt

FAILURE: Integrated in Hadoop-Hdfs-trunk-Java8 #157 (See https://builds.apache.org/job/Hadoop-Hdfs-trunk-Java8/157/)
HDFS-7891. A block placement policy with best rack failure tolerance. Contributed by Walter Su (szetszwo: rev 9595cc003ca5ed3d59b6942056a4fcb9080f79c9)

• hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
• hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestBlockPlacementPolicyRackFaultTolarent.java
• hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/BlockPlacementPolicyDefault.java
• hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/BlockPlacementPolicyRackFaultTolarent.java

FAILURE: Integrated in Hadoop-Mapreduce-trunk-Java8 #167 (See https://builds.apache.org/job/Hadoop-Mapreduce-trunk-Java8/167/)
HDFS-7891. A block placement policy with best rack failure tolerance. Contributed by Walter Su (szetszwo: rev 9595cc003ca5ed3d59b6942056a4fcb9080f79c9)

• hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/BlockPlacementPolicyDefault.java
• hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/BlockPlacementPolicyRackFaultTolarent.java
• hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestBlockPlacementPolicyRackFaultTolarent.java
• hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt

SUCCESS: Integrated in Hadoop-Mapreduce-trunk #2116 (See https://builds.apache.org/job/Hadoop-Mapreduce-trunk/2116/)
HDFS-7891. A block placement policy with best rack failure tolerance. Contributed by Walter Su (szetszwo: rev 9595cc003ca5ed3d59b6942056a4fcb9080f79c9)

• hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/BlockPlacementPolicyDefault.java
• hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
• hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/BlockPlacementPolicyRackFaultTolarent.java
• hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestBlockPlacementPolicyRackFaultTolarent.java

Thanks Walter for the work and Nicholas for the final review!