Apache
/
hbase
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

HBASE-9987 Remove some synchronisation points in HConnectionManager 

git-svn-id: https://svn.apache.org/repos/asf/hbase/trunk@1543051 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
nkeywal 2013-11-18 15:02:43 +00:00
parent ce8aeae6db
commit fb7fb4fa41
1 changed files with 72 additions and 81 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

153
hbase-client/src/main/java/org/apache/hadoop/hbase/client/HConnectionManager.java
View File

@ -25,7 +25,6 @@ import java.lang.reflect.UndeclaredThrowableException;
import java.net.SocketException; import java.net.SocketException;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.Date; import java.util.Date;
import java.util.HashMap;
import java.util.HashSet; import java.util.HashSet;
import java.util.LinkedHashMap; import java.util.LinkedHashMap;
import java.util.List; import java.util.List;
@ -36,6 +35,7 @@ import java.util.Set;
import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap; import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ConcurrentSkipListMap; import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.ConcurrentSkipListSet;
import java.util.concurrent.CopyOnWriteArraySet; import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.ExecutorService; import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.LinkedBlockingQueue;
@ -606,16 +606,16 @@ public class HConnectionManager {
/** /**
* Map of table to table {@link HRegionLocation}s. * Map of table to table {@link HRegionLocation}s.
*/ */
private final Map<TableName, ConcurrentSkipListMap<byte[], HRegionLocation>> private final ConcurrentMap<TableName, ConcurrentSkipListMap<byte[], HRegionLocation>>
cachedRegionLocations = cachedRegionLocations =
new HashMap<TableName, ConcurrentSkipListMap<byte[], HRegionLocation>>(); new ConcurrentHashMap<TableName, ConcurrentSkipListMap<byte[], HRegionLocation>>();
// The presence of a server in the map implies it's likely that there is an // The presence of a server in the map implies it's likely that there is an
// entry in cachedRegionLocations that map to this server; but the absence // entry in cachedRegionLocations that map to this server; but the absence
// of a server in this map guarentees that there is no entry in cache that // of a server in this map guarentees that there is no entry in cache that
// maps to the absent server. // maps to the absent server.
// The access to this attribute must be protected by a lock on cachedRegionLocations // The access to this attribute must be protected by a lock on cachedRegionLocations
private final Set<ServerName> cachedServers = new HashSet<ServerName>(); private final Set<ServerName> cachedServers = new ConcurrentSkipListSet<ServerName>();
// region cache prefetch is enabled by default. this set contains all // region cache prefetch is enabled by default. this set contains all
// tables whose region cache prefetch are disabled. // tables whose region cache prefetch are disabled.
@ -1352,14 +1352,12 @@ public class HConnectionManager {
*/ */
void forceDeleteCachedLocation(final TableName tableName, final byte [] row) { void forceDeleteCachedLocation(final TableName tableName, final byte [] row) {
HRegionLocation rl = null; HRegionLocation rl = null;
synchronized (this.cachedRegionLocations) { Map<byte[], HRegionLocation> tableLocations = getTableLocations(tableName);
Map<byte[], HRegionLocation> tableLocations = getTableLocations(tableName); // start to examine the cache. we can only do cache actions
// start to examine the cache. we can only do cache actions // if there's something in the cache for this table.
// if there's something in the cache for this table. rl = getCachedLocation(tableName, row);
rl = getCachedLocation(tableName, row); if (rl != null) {
if (rl != null) { tableLocations.remove(rl.getRegionInfo().getStartKey());
tableLocations.remove(rl.getRegionInfo().getStartKey());
}
} }
if ((rl != null) && LOG.isDebugEnabled()) { if ((rl != null) && LOG.isDebugEnabled()) {
LOG.debug("Removed " + rl.getHostname() + ":" + rl.getPort() LOG.debug("Removed " + rl.getHostname() + ":" + rl.getPort()
@ -1372,14 +1370,21 @@ public class HConnectionManager {
* Delete all cached entries of a table that maps to a specific location. * Delete all cached entries of a table that maps to a specific location.
*/ */
@Override @Override
public void clearCaches(final ServerName serverName){ public void clearCaches(final ServerName serverName) {
if (!this.cachedServers.contains(serverName)) {
return;
}
boolean deletedSomething = false; boolean deletedSomething = false;
synchronized (this.cachedRegionLocations) { synchronized (this.cachedServers) {
if (!cachedServers.contains(serverName)) { // We block here, because if there is an error on a server, it's likely that multiple
// threads will get the error simultaneously. If there are hundreds of thousand of
// region location to check, it's better to do this only once. A better pattern would
// be to check if the server is dead when we get the region location.
if (!this.cachedServers.contains(serverName)) {
return; return;
} }
for (Map<byte[], HRegionLocation> tableLocations : for (Map<byte[], HRegionLocation> tableLocations : cachedRegionLocations.values()) {
cachedRegionLocations.values()) {
for (Entry<byte[], HRegionLocation> e : tableLocations.entrySet()) { for (Entry<byte[], HRegionLocation> e : tableLocations.entrySet()) {
HRegionLocation value = e.getValue(); HRegionLocation value = e.getValue();
if (value != null if (value != null
@ -1389,7 +1394,7 @@ public class HConnectionManager {
} }
} }
} }
cachedServers.remove(serverName); this.cachedServers.remove(serverName);
} }
if (deletedSomething && LOG.isDebugEnabled()) { if (deletedSomething && LOG.isDebugEnabled()) {
LOG.debug("Removed all cached region locations that map to " + serverName); LOG.debug("Removed all cached region locations that map to " + serverName);
@ -1404,12 +1409,14 @@ public class HConnectionManager {
final TableName tableName) { final TableName tableName) {
// find the map of cached locations for this table // find the map of cached locations for this table
ConcurrentSkipListMap<byte[], HRegionLocation> result; ConcurrentSkipListMap<byte[], HRegionLocation> result;
synchronized (this.cachedRegionLocations) { result = this.cachedRegionLocations.get(tableName);
result = this.cachedRegionLocations.get(tableName); // if tableLocations for this table isn't built yet, make one
// if tableLocations for this table isn't built yet, make one if (result == null) {
if (result == null) { result = new ConcurrentSkipListMap<byte[], HRegionLocation>(Bytes.BYTES_COMPARATOR);
result = new ConcurrentSkipListMap<byte[], HRegionLocation>(Bytes.BYTES_COMPARATOR); ConcurrentSkipListMap<byte[], HRegionLocation> old =
this.cachedRegionLocations.put(tableName, result); this.cachedRegionLocations.putIfAbsent(tableName, result);
if (old != null) {
return old;
} }
} }
return result; return result;
@ -1417,17 +1424,13 @@ public class HConnectionManager {
@Override @Override
public void clearRegionCache() { public void clearRegionCache() {
synchronized(this.cachedRegionLocations) { this.cachedRegionLocations.clear();
this.cachedRegionLocations.clear(); this.cachedServers.clear();
this.cachedServers.clear();
}
} }
@Override @Override
public void clearRegionCache(final TableName tableName) { public void clearRegionCache(final TableName tableName) {
synchronized (this.cachedRegionLocations) { this.cachedRegionLocations.remove(tableName);
this.cachedRegionLocations.remove(tableName);
}
} }
@Override @Override
@ -1445,40 +1448,34 @@ public class HConnectionManager {
final HRegionLocation location) { final HRegionLocation location) {
boolean isFromMeta = (source == null); boolean isFromMeta = (source == null);
byte [] startKey = location.getRegionInfo().getStartKey(); byte [] startKey = location.getRegionInfo().getStartKey();
ConcurrentMap<byte[], HRegionLocation> tableLocations = ConcurrentMap<byte[], HRegionLocation> tableLocations = getTableLocations(tableName);
getTableLocations(tableName); HRegionLocation oldLocation = tableLocations.putIfAbsent(startKey, location);
boolean isNewCacheEntry = false; boolean isNewCacheEntry = (oldLocation == null);
boolean isStaleUpdate = false; if (isNewCacheEntry) {
HRegionLocation oldLocation = null; cachedServers.add(location.getServerName());
synchronized (this.cachedRegionLocations) { return;
oldLocation = tableLocations.putIfAbsent(startKey, location); }
isNewCacheEntry = (oldLocation == null); boolean updateCache;
if (isNewCacheEntry){ // If the server in cache sends us a redirect, assume it's always valid.
cachedServers.add(location.getServerName()); if (oldLocation.equals(source)) {
return; updateCache = true;
} } else {
boolean updateCache; long newLocationSeqNum = location.getSeqNum();
// If the server in cache sends us a redirect, assume it's always valid. // Meta record is stale - some (probably the same) server has closed the region
if (oldLocation.equals(source)) { // with later seqNum and told us about the new location.
updateCache = true; boolean isStaleMetaRecord = isFromMeta && (oldLocation.getSeqNum() > newLocationSeqNum);
} else { // Same as above for redirect. However, in this case, if the number is equal to previous
long newLocationSeqNum = location.getSeqNum(); // record, the most common case is that first the region was closed with seqNum, and then
// Meta record is stale - some (probably the same) server has closed the region // opened with the same seqNum; hence we will ignore the redirect.
// with later seqNum and told us about the new location. // There are so many corner cases with various combinations of opens and closes that
boolean isStaleMetaRecord = isFromMeta && (oldLocation.getSeqNum() > newLocationSeqNum); // an additional counter on top of seqNum would be necessary to handle them all.
// Same as above for redirect. However, in this case, if the number is equal to previous boolean isStaleRedirect = !isFromMeta && (oldLocation.getSeqNum() >= newLocationSeqNum);
// record, the most common case is that first the region was closed with seqNum, and then boolean isStaleUpdate = (isStaleMetaRecord || isStaleRedirect);
// opened with the same seqNum; hence we will ignore the redirect. updateCache = (!isStaleUpdate);
// There are so many corner cases with various combinations of opens and closes that }
// an additional counter on top of seqNum would be necessary to handle them all. if (updateCache) {
boolean isStaleRedirect = !isFromMeta && (oldLocation.getSeqNum() >= newLocationSeqNum); tableLocations.replace(startKey, oldLocation, location);
isStaleUpdate = (isStaleMetaRecord || isStaleRedirect); cachedServers.add(location.getServerName());
updateCache = (!isStaleUpdate);
}
if (updateCache) {
tableLocations.put(startKey, location);
cachedServers.add(location.getServerName());
}
} }
} }
@ -2194,12 +2191,10 @@ public class HConnectionManager {
* @param hri The region in question. * @param hri The region in question.
* @param source The source of the error that prompts us to invalidate cache. * @param source The source of the error that prompts us to invalidate cache.
*/ */
void deleteCachedLocation(HRegionInfo hri, HRegionLocation source) { void deleteCachedLocation(HRegionInfo hri, HRegionLocation source) {
synchronized (this.cachedRegionLocations) { ConcurrentMap<byte[], HRegionLocation> tableLocations = getTableLocations(hri.getTable());
ConcurrentMap<byte[], HRegionLocation> tableLocations = getTableLocations(hri.getTable()); tableLocations.remove(hri.getStartKey(), source);
tableLocations.remove(hri.getStartKey(), source); }
}
}
@Override @Override
public void deleteCachedRegionLocation(final HRegionLocation location) { public void deleteCachedRegionLocation(final HRegionLocation location) {
@ -2209,10 +2204,8 @@ public class HConnectionManager {
HRegionLocation removedLocation; HRegionLocation removedLocation;
TableName tableName = location.getRegionInfo().getTable(); TableName tableName = location.getRegionInfo().getTable();
synchronized (this.cachedRegionLocations) { Map<byte[], HRegionLocation> tableLocations = getTableLocations(tableName);
Map<byte[], HRegionLocation> tableLocations = getTableLocations(tableName); removedLocation = tableLocations.remove(location.getRegionInfo().getStartKey());
removedLocation = tableLocations.remove(location.getRegionInfo().getStartKey());
}
if (LOG.isDebugEnabled() && removedLocation != null) { if (LOG.isDebugEnabled() && removedLocation != null) {
LOG.debug("Removed " + LOG.debug("Removed " +
location.getRegionInfo().getRegionNameAsString() + location.getRegionInfo().getRegionNameAsString() +
@ -2405,13 +2398,11 @@ public class HConnectionManager {
* from a unit test. * from a unit test.
*/ */
int getNumberOfCachedRegionLocations(final TableName tableName) { int getNumberOfCachedRegionLocations(final TableName tableName) {
synchronized (this.cachedRegionLocations) { Map<byte[], HRegionLocation> tableLocs = this.cachedRegionLocations.get(tableName);
Map<byte[], HRegionLocation> tableLocs = this.cachedRegionLocations.get(tableName); if (tableLocs == null) {
if (tableLocs == null) { return 0;
return 0;
}
return tableLocs.values().size();
} }
return tableLocs.values().size();
} }
/** /**