From 1435b2002b4556d28f9ecb641561914d6e4078da Mon Sep 17 00:00:00 2001 From: Tsuyoshi Ozawa Date: Thu, 22 Jan 2015 14:15:59 +0900 Subject: [PATCH] HADOOP-11209. Configuration#updatingResource/finalParameters are not thread-safe. Contributed by Varun Saxena. (cherry picked from commit 786dbdfad8991a99d71bdd861e0b5014669a422c) --- .../hadoop-common/CHANGES.txt | 3 ++ .../org/apache/hadoop/conf/Configuration.java | 46 +++++++++++++------ .../apache/hadoop/conf/TestConfiguration.java | 46 +++++++++++++++++++ 3 files changed, 80 insertions(+), 15 deletions(-) diff --git a/hadoop-common-project/hadoop-common/CHANGES.txt b/hadoop-common-project/hadoop-common/CHANGES.txt index e7b96fa1431..310b8ef64a0 100644 --- a/hadoop-common-project/hadoop-common/CHANGES.txt +++ b/hadoop-common-project/hadoop-common/CHANGES.txt @@ -382,6 +382,9 @@ Release 2.7.0 - UNRELEASED architecture because it is slower there (Suman Somasundar via Colin P. McCabe) + HADOOP-11209. Configuration#updatingResource/finalParameters are not + thread-safe. (Varun Saxena via ozawa) + Release 2.6.0 - 2014-11-18 INCOMPATIBLE CHANGES diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/conf/Configuration.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/conf/Configuration.java index 387e64a78a7..927d4b67ffa 100644 --- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/conf/Configuration.java +++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/conf/Configuration.java @@ -52,6 +52,7 @@ import java.util.Properties; import java.util.Set; import java.util.StringTokenizer; import java.util.WeakHashMap; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CopyOnWriteArrayList; import java.util.regex.Matcher; import java.util.regex.Pattern; @@ -228,7 +229,8 @@ public class Configuration implements Iterable>, /** * List of configuration parameters marked final. */ - private Set finalParameters = new HashSet(); + private Set finalParameters = Collections.newSetFromMap( + new ConcurrentHashMap()); private boolean loadDefaults = true; @@ -258,7 +260,7 @@ public class Configuration implements Iterable>, * Stores the mapping of key to the resource which modifies or loads * the key most recently */ - private HashMap updatingResource; + private Map updatingResource; /** * Class to keep the information about the keys which replace the deprecated @@ -685,7 +687,7 @@ public class Configuration implements Iterable>, */ public Configuration(boolean loadDefaults) { this.loadDefaults = loadDefaults; - updatingResource = new HashMap(); + updatingResource = new ConcurrentHashMap(); synchronized(Configuration.class) { REGISTRY.put(this, null); } @@ -708,8 +710,11 @@ public class Configuration implements Iterable>, this.overlay = (Properties)other.overlay.clone(); } - this.updatingResource = new HashMap(other.updatingResource); - this.finalParameters = new HashSet(other.finalParameters); + this.updatingResource = new ConcurrentHashMap( + other.updatingResource); + this.finalParameters = Collections.newSetFromMap( + new ConcurrentHashMap()); + this.finalParameters.addAll(other.finalParameters); } synchronized(Configuration.class) { @@ -2306,20 +2311,27 @@ public class Configuration implements Iterable>, * @return final parameter set. */ public Set getFinalParameters() { - return new HashSet(finalParameters); + Set setFinalParams = Collections.newSetFromMap( + new ConcurrentHashMap()); + setFinalParams.addAll(finalParameters); + return setFinalParams; } protected synchronized Properties getProps() { if (properties == null) { properties = new Properties(); - HashMap backup = - new HashMap(updatingResource); + Map backup = + new ConcurrentHashMap(updatingResource); loadResources(properties, resources, quietmode); - if (overlay!= null) { + + if (overlay != null) { properties.putAll(overlay); for (Map.Entry item: overlay.entrySet()) { String key = (String)item.getKey(); - updatingResource.put(key, backup.get(key)); + String[] source = backup.get(key); + if(source != null) { + updatingResource.put(key, source); + } } } } @@ -2568,16 +2580,18 @@ public class Configuration implements Iterable>, if (value != null || allowNullValueProperties) { if (!finalParameters.contains(attr)) { if (value==null && allowNullValueProperties) { - value = DEFAULT_STRING_CHECK; - } + value = DEFAULT_STRING_CHECK; + } properties.setProperty(attr, value); - updatingResource.put(attr, source); + if(source != null) { + updatingResource.put(attr, source); + } } else if (!value.equals(properties.getProperty(attr))) { LOG.warn(name+":an attempt to override final parameter: "+attr +"; Ignoring."); } } - if (finalParameter) { + if (finalParameter && attr != null) { finalParameters.add(attr); } } @@ -2780,7 +2794,9 @@ public class Configuration implements Iterable>, String value = org.apache.hadoop.io.Text.readString(in); set(key, value); String sources[] = WritableUtils.readCompressedStringArray(in); - updatingResource.put(key, sources); + if(sources != null) { + updatingResource.put(key, sources); + } } } diff --git a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/conf/TestConfiguration.java b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/conf/TestConfiguration.java index 3bc8365e890..5aa8350dd24 100644 --- a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/conf/TestConfiguration.java +++ b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/conf/TestConfiguration.java @@ -1216,6 +1216,52 @@ public class TestConfiguration extends TestCase { assertTrue("my.var is not final", finalParameters.contains("my.var")); } + /** + * A test to check whether this thread goes into infinite loop because of + * destruction of data structure by resize of Map. This problem was reported + * by SPARK-2546. + * @throws Exception + */ + public void testConcurrentAccesses() throws Exception { + out = new BufferedWriter(new FileWriter(CONFIG)); + startConfig(); + declareProperty("some.config", "xyz", "xyz", false); + endConfig(); + Path fileResource = new Path(CONFIG); + Configuration conf = new Configuration(); + conf.addResource(fileResource); + + class ConfigModifyThread extends Thread { + final private Configuration config; + final private String prefix; + + public ConfigModifyThread(Configuration conf, String prefix) { + config = conf; + this.prefix = prefix; + } + + @Override + public void run() { + for (int i = 0; i < 100000; i++) { + config.set("some.config.value-" + prefix + i, "value"); + } + } + } + + ArrayList threads = new ArrayList<>(); + for (int i = 0; i < 100; i++) { + threads.add(new ConfigModifyThread(conf, String.valueOf(i))); + } + for (Thread t: threads) { + t.start(); + } + for (Thread t: threads) { + t.join(); + } + // If this test completes without going into infinite loop, + // it's expected behaviour. + } + public static void main(String[] argv) throws Exception { junit.textui.TestRunner.main(new String[]{ TestConfiguration.class.getName()