hadoop-common-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From zjs...@apache.org
Subject hadoop git commit: YARN-2996. Improved synchronization and I/O operations of FS- and Mem- RMStateStore. Contributed by Yi Liu.
Date Thu, 08 Jan 2015 17:49:46 GMT
Repository: hadoop
Updated Branches:
  refs/heads/branch-2 132fd6ba5 -> d2fbba790


YARN-2996. Improved synchronization and I/O operations of FS- and Mem- RMStateStore. Contributed
by Yi Liu.

(cherry picked from commit dc2eaa26b20cfbbcdd5784bb8761d08a42f29605)


Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/d2fbba79
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/d2fbba79
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/d2fbba79

Branch: refs/heads/branch-2
Commit: d2fbba790aa03a4a88c464dc3ba63fae204110be
Parents: 132fd6b
Author: Zhijie Shen <zjshen@apache.org>
Authored: Thu Jan 8 09:47:02 2015 -0800
Committer: Zhijie Shen <zjshen@apache.org>
Committed: Thu Jan 8 09:49:27 2015 -0800

----------------------------------------------------------------------
 hadoop-yarn-project/CHANGES.txt                 |  3 +++
 .../recovery/FileSystemRMStateStore.java        | 26 ++++++++++++--------
 .../recovery/MemoryRMStateStore.java            | 11 +++++----
 3 files changed, 25 insertions(+), 15 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/hadoop/blob/d2fbba79/hadoop-yarn-project/CHANGES.txt
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/CHANGES.txt b/hadoop-yarn-project/CHANGES.txt
index 3898bbc..ee50cfe 100644
--- a/hadoop-yarn-project/CHANGES.txt
+++ b/hadoop-yarn-project/CHANGES.txt
@@ -137,6 +137,9 @@ Release 2.7.0 - UNRELEASED
     YARN-2880. Added a test to make sure node labels will be recovered
     if RM restart is enabled. (Rohith Sharmaks via jianhe)
 
+    YARN-2996. Improved synchronization and I/O operations of FS- and Mem-
+    RMStateStore. (Yi Liu via zjshen)
+
   OPTIMIZATIONS
 
   BUG FIXES

http://git-wip-us.apache.org/repos/asf/hadoop/blob/d2fbba79/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/FileSystemRMStateStore.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/FileSystemRMStateStore.java
b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/FileSystemRMStateStore.java
index 7783662..6e830a0 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/FileSystemRMStateStore.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/FileSystemRMStateStore.java
@@ -139,8 +139,8 @@ public class FileSystemRMStateStore extends RMStateStore {
   @Override
   protected synchronized Version loadVersion() throws Exception {
     Path versionNodePath = getNodePath(rootDirPath, VERSION_NODE);
-    if (fs.exists(versionNodePath)) {
-      FileStatus status = fs.getFileStatus(versionNodePath);
+    FileStatus status = getFileStatus(versionNodePath);
+    if (status != null) {
       byte[] data = readFile(versionNodePath, status.getLen());
       Version version =
           new VersionPBImpl(VersionProto.parseFrom(data));
@@ -165,9 +165,9 @@ public class FileSystemRMStateStore extends RMStateStore {
   public synchronized long getAndIncrementEpoch() throws Exception {
     Path epochNodePath = getNodePath(rootDirPath, EPOCH_NODE);
     long currentEpoch = 0;
-    if (fs.exists(epochNodePath)) {
+    FileStatus status = getFileStatus(epochNodePath);
+    if (status != null) {
       // load current epoch
-      FileStatus status = fs.getFileStatus(epochNodePath);
       byte[] data = readFile(epochNodePath, status.getLen());
       Epoch epoch = new EpochPBImpl(EpochProto.parseFrom(data));
       currentEpoch = epoch.getEpoch();
@@ -201,13 +201,11 @@ public class FileSystemRMStateStore extends RMStateStore {
     checkAndResumeUpdateOperation(amrmTokenSecretManagerRoot);
     Path amrmTokenSecretManagerStateDataDir =
         new Path(amrmTokenSecretManagerRoot, AMRMTOKEN_SECRET_MANAGER_NODE);
-    FileStatus status;
-    try {
-      status = fs.getFileStatus(amrmTokenSecretManagerStateDataDir);
-      assert status.isFile();
-    } catch (FileNotFoundException ex) {
+    FileStatus status = getFileStatus(amrmTokenSecretManagerStateDataDir);
+    if (status == null) {
       return;
     }
+    assert status.isFile();
     byte[] data = readFile(amrmTokenSecretManagerStateDataDir, status.getLen());
     AMRMTokenSecretManagerStatePBImpl stateData =
         new AMRMTokenSecretManagerStatePBImpl(
@@ -466,7 +464,7 @@ public class FileSystemRMStateStore extends RMStateStore {
   }
 
   @Override
-  protected void updateRMDelegationTokenState(
+  protected synchronized void updateRMDelegationTokenState(
       RMDelegationTokenIdentifier rmDTIdentifier, Long renewDate)
       throws Exception {
     storeOrUpdateRMDelegationTokenState(rmDTIdentifier, renewDate, true);
@@ -560,6 +558,14 @@ public class FileSystemRMStateStore extends RMStateStore {
     }
   }
 
+  private FileStatus getFileStatus(Path path) throws Exception {
+    try {
+      return fs.getFileStatus(path);
+    } catch (FileNotFoundException e) {
+      return null;
+    }
+  }
+
   /*
    * In order to make this write atomic as a part of write we will first write
    * data to .tmp file and then rename it. Here we are assuming that rename is

http://git-wip-us.apache.org/repos/asf/hadoop/blob/d2fbba79/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/MemoryRMStateStore.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/MemoryRMStateStore.java
b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/MemoryRMStateStore.java
index 3646949..8cd776e 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/MemoryRMStateStore.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/MemoryRMStateStore.java
@@ -91,15 +91,16 @@ public class MemoryRMStateStore extends RMStateStore {
   }
 
   @Override
-  public void storeApplicationStateInternal(
+  public synchronized void storeApplicationStateInternal(
       ApplicationId appId, ApplicationStateData appState)
       throws Exception {
     state.appState.put(appId, appState);
   }
 
   @Override
-  public void updateApplicationStateInternal(ApplicationId appId,
-      ApplicationStateData appState) throws Exception {
+  public synchronized void updateApplicationStateInternal(
+      ApplicationId appId, ApplicationStateData appState) 
+      throws Exception {
     LOG.info("Updating final state " + appState.getState() + " for app: "
         + appId);
     if (state.appState.get(appId) != null) {
@@ -186,7 +187,7 @@ public class MemoryRMStateStore extends RMStateStore {
   }
 
   @Override
-  protected void updateRMDelegationTokenState(
+  protected synchronized void updateRMDelegationTokenState(
       RMDelegationTokenIdentifier rmDTIdentifier, Long renewDate)
       throws Exception {
     removeRMDelegationTokenState(rmDTIdentifier);
@@ -237,7 +238,7 @@ public class MemoryRMStateStore extends RMStateStore {
   }
 
   @Override
-  public void storeOrUpdateAMRMTokenSecretManagerState(
+  public synchronized void storeOrUpdateAMRMTokenSecretManagerState(
       AMRMTokenSecretManagerState amrmTokenSecretManagerState,
       boolean isUpdate) {
     if (amrmTokenSecretManagerState != null) {


Mime
View raw message