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

HADOOP-2417 Fix critical shutdown problem introduced by HADOOP-2338 

git-svn-id: https://svn.apache.org/repos/asf/lucene/hadoop/trunk/src/contrib/hbase@604011 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jim Kellerman 2007-12-13 20:02:30 +00:00
parent 91de46cac1
commit a24a76ca83
4 changed files with 169 additions and 149 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -77,6 +77,7 @@ Trunk (unreleased changes)
HADOOP-2396 NPE in HMaster.cancelLease HADOOP-2396 NPE in HMaster.cancelLease
HADOOP-2397 The only time that a meta scanner should try to recover a log is HADOOP-2397 The only time that a meta scanner should try to recover a log is
when the master is starting when the master is starting
HADOOP-2417 Fix critical shutdown problem introduced by HADOOP-2338
IMPROVEMENTS IMPROVEMENTS
HADOOP-2401 Add convenience put method that takes writable HADOOP-2401 Add convenience put method that takes writable

52
src/java/org/apache/hadoop/hbase/HMaster.java
View File

@ -181,16 +181,14 @@ public class HMaster extends Thread implements HConstants, HMasterInterface,
*/ */
abstract class BaseScanner extends Chore { abstract class BaseScanner extends Chore {
protected boolean rootRegion; protected boolean rootRegion;
protected final Text tableName;
protected abstract boolean initialScan(); protected abstract boolean initialScan();
protected abstract void maintenanceScan(); protected abstract void maintenanceScan();
BaseScanner(final Text tableName, final int period, BaseScanner(final boolean rootRegion, final int period,
final AtomicBoolean stop) { final AtomicBoolean stop) {
super(period, stop); super(period, stop);
this.tableName = tableName; this.rootRegion = rootRegion;
this.rootRegion = tableName.equals(ROOT_TABLE_NAME);
} }
@Override @Override
@ -506,7 +504,7 @@ public class HMaster extends Thread implements HConstants, HMasterInterface,
class RootScanner extends BaseScanner { class RootScanner extends BaseScanner {
/** Constructor */ /** Constructor */
public RootScanner() { public RootScanner() {
super(HConstants.ROOT_TABLE_NAME, metaRescanInterval, closed); super(true, metaRescanInterval, closed);
} }
private boolean scanRoot() { private boolean scanRoot() {
@ -671,7 +669,7 @@ public class HMaster extends Thread implements HConstants, HMasterInterface,
/** Constructor */ /** Constructor */
public MetaScanner() { public MetaScanner() {
super(HConstants.META_TABLE_NAME, metaRescanInterval, closed); super(false, metaRescanInterval, closed);
} }
private boolean scanOneMetaRegion(MetaRegion region) { private boolean scanOneMetaRegion(MetaRegion region) {
@ -1182,16 +1180,25 @@ public class HMaster extends Thread implements HConstants, HMasterInterface,
* regions can shut down. * regions can shut down.
*/ */
private void stopScanners() { private void stopScanners() {
if (LOG.isDebugEnabled()) {
LOG.debug("telling root scanner to stop");
}
synchronized(rootScannerLock) { synchronized(rootScannerLock) {
if (rootScannerThread.isAlive()) { if (rootScannerThread.isAlive()) {
rootScannerThread.interrupt(); // Wake root scanner rootScannerThread.interrupt(); // Wake root scanner
} }
} }
if (LOG.isDebugEnabled()) {
LOG.debug("telling meta scanner to stop");
}
synchronized(metaScannerLock) { synchronized(metaScannerLock) {
if (metaScannerThread.isAlive()) { if (metaScannerThread.isAlive()) {
metaScannerThread.interrupt(); // Wake meta scanner metaScannerThread.interrupt(); // Wake meta scanner
} }
} }
if (LOG.isDebugEnabled()) {
LOG.debug("meta and root scanners notified");
}
} }
/* /*
@ -1341,18 +1348,23 @@ public class HMaster extends Thread implements HConstants, HMasterInterface,
} }
} else if (msgs[0].getMsg() == HMsg.MSG_REPORT_QUIESCED) { } else if (msgs[0].getMsg() == HMsg.MSG_REPORT_QUIESCED) {
LOG.info("Region server " + serverName + " quiesced"); LOG.info("Region server " + serverName + " quiesced");
if(quiescedMetaServers.incrementAndGet() == serversToServerInfo.size()) { quiescedMetaServers.incrementAndGet();
// If the only servers we know about are meta servers, then we can }
// proceed with shutdown }
LOG.info("All user tables quiesced. Proceeding with shutdown");
closed.set(true); if(quiescedMetaServers.get() >= serversToServerInfo.size()) {
stopScanners(); // If the only servers we know about are meta servers, then we can
synchronized(toDoQueue) { // proceed with shutdown
toDoQueue.clear(); // Empty the queue LOG.info("All user tables quiesced. Proceeding with shutdown");
delayedToDoQueue.clear(); // Empty shut down queue closed.set(true);
toDoQueue.notifyAll(); // Wake main thread stopScanners();
} synchronized(toDoQueue) {
} toDoQueue.clear(); // Empty the queue
delayedToDoQueue.clear(); // Empty shut down queue
toDoQueue.notifyAll(); // Wake main thread
}
synchronized (serversToServerInfo) {
serversToServerInfo.notifyAll();
} }
} }
@ -1638,7 +1650,7 @@ public class HMaster extends Thread implements HConstants, HMasterInterface,
" split. New regions are: " + newRegionA.getRegionName() + ", " + " split. New regions are: " + newRegionA.getRegionName() + ", " +
newRegionB.getRegionName()); newRegionB.getRegionName());
if (region.getTableDesc().getName().equals(META_TABLE_NAME)) { if (region.isMetaTable()) {
// A meta region has split. // A meta region has split.
onlineMetaRegions.remove(region.getStartKey()); onlineMetaRegions.remove(region.getStartKey());
@ -2028,7 +2040,7 @@ public class HMaster extends Thread implements HConstants, HMasterInterface,
serverName + "> (or server is null). Marking unassigned if " + serverName + "> (or server is null). Marking unassigned if " +
"meta and clearing pendingRegions"); "meta and clearing pendingRegions");
if (info.getTableDesc().getName().equals(META_TABLE_NAME)) { if (info.isMetaTable()) {
if (LOG.isDebugEnabled()) { if (LOG.isDebugEnabled()) {
LOG.debug("removing meta region " + info.getRegionName() + LOG.debug("removing meta region " + info.getRegionName() +
" from online meta regions"); " from online meta regions");

241
src/java/org/apache/hadoop/hbase/HRegion.java
View File

@ -225,6 +225,7 @@ public class HRegion implements HConstants {
protected final long threadWakeFrequency; protected final long threadWakeFrequency;
private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock(); private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
private final Integer updateLock = new Integer(0); private final Integer updateLock = new Integer(0);
private final Integer splitLock = new Integer(0);
private final long desiredMaxFileSize; private final long desiredMaxFileSize;
private final long minSequenceId; private final long minSequenceId;
private final String encodedRegionName; private final String encodedRegionName;
@ -381,54 +382,56 @@ public class HRegion implements HConstants {
LOG.info("region " + this.regionInfo.getRegionName() + " already closed"); LOG.info("region " + this.regionInfo.getRegionName() + " already closed");
return null; return null;
} }
lock.writeLock().lock(); synchronized (splitLock) {
try { lock.writeLock().lock();
synchronized (writestate) { try {
while (writestate.compacting || writestate.flushing) { synchronized (writestate) {
try { while (writestate.compacting || writestate.flushing) {
writestate.wait(); try {
} catch (InterruptedException iex) { writestate.wait();
// continue } catch (InterruptedException iex) {
// continue
}
}
// Disable compacting and flushing by background threads for this
// region.
writestate.writesEnabled = false;
}
// Wait for active scanners to finish. The write lock we hold will prevent
// new scanners from being created.
synchronized (activeScannerCount) {
while (activeScannerCount.get() != 0) {
try {
activeScannerCount.wait();
} catch (InterruptedException e) {
// continue
}
} }
} }
// Disable compacting and flushing by background threads for this
// region.
writestate.writesEnabled = false;
}
// Wait for active scanners to finish. The write lock we hold will prevent // Write lock means no more row locks can be given out. Wait on
// new scanners from being created. // outstanding row locks to come in before we close so we do not drop
// outstanding updates.
waitOnRowLocks();
synchronized (activeScannerCount) { // Don't flush the cache if we are aborting
while (activeScannerCount.get() != 0) { if (!abort) {
try { internalFlushcache(snapshotMemcaches());
activeScannerCount.wait();
} catch (InterruptedException e) {
// continue
}
} }
}
// Write lock means no more row locks can be given out. Wait on List<HStoreFile> result = new ArrayList<HStoreFile>();
// outstanding row locks to come in before we close so we do not drop for (HStore store: stores.values()) {
// outstanding updates. result.addAll(store.close());
waitOnRowLocks(); }
this.closed.set(true);
// Don't flush the cache if we are aborting LOG.info("closed " + this.regionInfo.getRegionName());
if (!abort) { return result;
internalFlushcache(snapshotMemcaches()); } finally {
lock.writeLock().unlock();
} }
List<HStoreFile> result = new ArrayList<HStoreFile>();
for (HStore store: stores.values()) {
result.addAll(store.close());
}
this.closed.set(true);
LOG.info("closed " + this.regionInfo.getRegionName());
return result;
} finally {
lock.writeLock().unlock();
} }
} }
@ -541,89 +544,91 @@ public class HRegion implements HConstants {
HRegion[] splitRegion(final RegionUnavailableListener listener) HRegion[] splitRegion(final RegionUnavailableListener listener)
throws IOException { throws IOException {
Text midKey = new Text(); synchronized (splitLock) {
if (!needsSplit(midKey)) { Text midKey = new Text();
return null; if (closed.get() || !needsSplit(midKey)) {
} return null;
long startTime = System.currentTimeMillis(); }
Path splits = getSplitsDir(); long startTime = System.currentTimeMillis();
HRegionInfo regionAInfo = new HRegionInfo(this.regionInfo.getTableDesc(), Path splits = getSplitsDir();
this.regionInfo.getStartKey(), midKey); HRegionInfo regionAInfo = new HRegionInfo(this.regionInfo.getTableDesc(),
Path dirA = getSplitRegionDir(splits, this.regionInfo.getStartKey(), midKey);
HRegionInfo.encodeRegionName(regionAInfo.getRegionName())); Path dirA = getSplitRegionDir(splits,
if(fs.exists(dirA)) { HRegionInfo.encodeRegionName(regionAInfo.getRegionName()));
throw new IOException("Cannot split; target file collision at " + dirA); if(fs.exists(dirA)) {
} throw new IOException("Cannot split; target file collision at " + dirA);
HRegionInfo regionBInfo = new HRegionInfo(this.regionInfo.getTableDesc(), }
midKey, null); HRegionInfo regionBInfo = new HRegionInfo(this.regionInfo.getTableDesc(),
Path dirB = getSplitRegionDir(splits, midKey, null);
HRegionInfo.encodeRegionName(regionBInfo.getRegionName())); Path dirB = getSplitRegionDir(splits,
if(this.fs.exists(dirB)) { HRegionInfo.encodeRegionName(regionBInfo.getRegionName()));
throw new IOException("Cannot split; target file collision at " + dirB); if(this.fs.exists(dirB)) {
} throw new IOException("Cannot split; target file collision at " + dirB);
}
// Notify the caller that we are about to close the region. This moves // Notify the caller that we are about to close the region. This moves
// us to the 'retiring' queue. Means no more updates coming in -- just // us to the 'retiring' queue. Means no more updates coming in -- just
// whatever is outstanding. // whatever is outstanding.
if (listener != null) { if (listener != null) {
listener.closing(getRegionName()); listener.closing(getRegionName());
} }
// Now close the HRegion. Close returns all store files or null if not // Now close the HRegion. Close returns all store files or null if not
// supposed to close (? What to do in this case? Implement abort of close?) // supposed to close (? What to do in this case? Implement abort of close?)
// Close also does wait on outstanding rows and calls a flush just-in-case. // Close also does wait on outstanding rows and calls a flush just-in-case.
List<HStoreFile> hstoreFilesToSplit = close(); List<HStoreFile> hstoreFilesToSplit = close();
if (hstoreFilesToSplit == null) { if (hstoreFilesToSplit == null) {
LOG.warn("Close came back null (Implement abort of close?)"); LOG.warn("Close came back null (Implement abort of close?)");
throw new RuntimeException("close returned empty vector of HStoreFiles"); throw new RuntimeException("close returned empty vector of HStoreFiles");
} }
// Tell listener that region is now closed and that they can therefore // Tell listener that region is now closed and that they can therefore
// clean up any outstanding references. // clean up any outstanding references.
if (listener != null) { if (listener != null) {
listener.closed(this.getRegionName()); listener.closed(this.getRegionName());
} }
// Split each store file. // Split each store file.
for(HStoreFile h: hstoreFilesToSplit) { for(HStoreFile h: hstoreFilesToSplit) {
// A reference to the bottom half of the hsf store file. // A reference to the bottom half of the hsf store file.
HStoreFile.Reference aReference = new HStoreFile.Reference( HStoreFile.Reference aReference = new HStoreFile.Reference(
this.encodedRegionName, h.getFileId(), new HStoreKey(midKey), this.encodedRegionName, h.getFileId(), new HStoreKey(midKey),
HStoreFile.Range.bottom); HStoreFile.Range.bottom);
HStoreFile a = new HStoreFile(this.conf, splits, HStoreFile a = new HStoreFile(this.conf, splits,
HRegionInfo.encodeRegionName(regionAInfo.getRegionName()), HRegionInfo.encodeRegionName(regionAInfo.getRegionName()),
h.getColFamily(), Math.abs(rand.nextLong()), aReference); h.getColFamily(), Math.abs(rand.nextLong()), aReference);
// Reference to top half of the hsf store file. // Reference to top half of the hsf store file.
HStoreFile.Reference bReference = new HStoreFile.Reference( HStoreFile.Reference bReference = new HStoreFile.Reference(
this.encodedRegionName, h.getFileId(), new HStoreKey(midKey), this.encodedRegionName, h.getFileId(), new HStoreKey(midKey),
HStoreFile.Range.top); HStoreFile.Range.top);
HStoreFile b = new HStoreFile(this.conf, splits, HStoreFile b = new HStoreFile(this.conf, splits,
HRegionInfo.encodeRegionName(regionBInfo.getRegionName()), HRegionInfo.encodeRegionName(regionBInfo.getRegionName()),
h.getColFamily(), Math.abs(rand.nextLong()), bReference); h.getColFamily(), Math.abs(rand.nextLong()), bReference);
h.splitStoreFile(a, b, this.fs); h.splitStoreFile(a, b, this.fs);
} }
// Done! // Done!
// Opening the region copies the splits files from the splits directory // Opening the region copies the splits files from the splits directory
// under each region. // under each region.
HRegion regionA = HRegion regionA =
new HRegion(rootDir, log, fs, conf, regionAInfo, dirA, null); new HRegion(rootDir, log, fs, conf, regionAInfo, dirA, null);
regionA.close(); regionA.close();
HRegion regionB = HRegion regionB =
new HRegion(rootDir, log, fs, conf, regionBInfo, dirB, null); new HRegion(rootDir, log, fs, conf, regionBInfo, dirB, null);
regionB.close(); regionB.close();
// Cleanup // Cleanup
boolean deleted = fs.delete(splits); // Get rid of splits directory boolean deleted = fs.delete(splits); // Get rid of splits directory
if (LOG.isDebugEnabled()) { if (LOG.isDebugEnabled()) {
LOG.debug("Cleaned up " + splits.toString() + " " + deleted); LOG.debug("Cleaned up " + splits.toString() + " " + deleted);
}
HRegion regions[] = new HRegion [] {regionA, regionB};
LOG.info("Region split of " + this.regionInfo.getRegionName() +
" complete; " + "new regions: " + regions[0].getRegionName() + ", " +
regions[1].getRegionName() + ". Split took " +
StringUtils.formatTimeDiff(System.currentTimeMillis(), startTime));
return regions;
} }
HRegion regions[] = new HRegion [] {regionA, regionB};
LOG.info("Region split of " + this.regionInfo.getRegionName() +
" complete; " + "new regions: " + regions[0].getRegionName() + ", " +
regions[1].getRegionName() + ". Split took " +
StringUtils.formatTimeDiff(System.currentTimeMillis(), startTime));
return regions;
} }
/* /*
@ -1030,6 +1035,7 @@ public class HRegion implements HConstants {
* avoid a bunch of disk activity. * avoid a bunch of disk activity.
* *
* @param row * @param row
* @param ts
* @return Map<columnName, byte[]> values * @return Map<columnName, byte[]> values
* @throws IOException * @throws IOException
*/ */
@ -1282,6 +1288,7 @@ public class HRegion implements HConstants {
* @param row The row to operate on * @param row The row to operate on
* @param family The column family to match * @param family The column family to match
* @param timestamp Timestamp to match * @param timestamp Timestamp to match
* @throws IOException
*/ */
public void deleteFamily(Text row, Text family, long timestamp) public void deleteFamily(Text row, Text family, long timestamp)
throws IOException{ throws IOException{

14
src/java/org/apache/hadoop/hbase/HRegionServer.java
View File

@ -79,9 +79,9 @@ public class HRegionServer implements HConstants, HRegionInterface, Runnable {
// of HRegionServer in isolation. We use AtomicBoolean rather than // of HRegionServer in isolation. We use AtomicBoolean rather than
// plain boolean so we can pass a reference to Chore threads. Otherwise, // plain boolean so we can pass a reference to Chore threads. Otherwise,
// Chore threads need to know about the hosting class. // Chore threads need to know about the hosting class.
protected final AtomicBoolean stopRequested = new AtomicBoolean(false); protected volatile AtomicBoolean stopRequested = new AtomicBoolean(false);
protected final AtomicBoolean quiesced = new AtomicBoolean(false); protected volatile AtomicBoolean quiesced = new AtomicBoolean(false);
// Go down hard. Used if file system becomes unavailable and also in // Go down hard. Used if file system becomes unavailable and also in
// debugging and unit tests. // debugging and unit tests.
@ -95,13 +95,13 @@ public class HRegionServer implements HConstants, HRegionInterface, Runnable {
private final Random rand = new Random(); private final Random rand = new Random();
// region name -> HRegion // region name -> HRegion
protected final SortedMap<Text, HRegion> onlineRegions = protected volatile SortedMap<Text, HRegion> onlineRegions =
Collections.synchronizedSortedMap(new TreeMap<Text, HRegion>()); Collections.synchronizedSortedMap(new TreeMap<Text, HRegion>());
protected final Map<Text, HRegion> retiringRegions = protected volatile Map<Text, HRegion> retiringRegions =
new ConcurrentHashMap<Text, HRegion>(); new ConcurrentHashMap<Text, HRegion>();
protected final ReentrantReadWriteLock lock = new ReentrantReadWriteLock(); protected final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
private final List<HMsg> outboundMsgs = private volatile List<HMsg> outboundMsgs =
Collections.synchronizedList(new ArrayList<HMsg>()); Collections.synchronizedList(new ArrayList<HMsg>());
final int numRetries; final int numRetries;
@ -120,7 +120,7 @@ public class HRegionServer implements HConstants, HRegionInterface, Runnable {
private final Leases leases; private final Leases leases;
// Request counter // Request counter
private final AtomicInteger requestCount = new AtomicInteger(); private volatile AtomicInteger requestCount = new AtomicInteger();
// A sleeper that sleeps for msgInterval. // A sleeper that sleeps for msgInterval.
private final Sleeper sleeper; private final Sleeper sleeper;
@ -296,7 +296,7 @@ public class HRegionServer implements HConstants, HRegionInterface, Runnable {
// splitting a 'normal' region, and the ROOT table needs to be // splitting a 'normal' region, and the ROOT table needs to be
// updated if we are splitting a META region. // updated if we are splitting a META region.
HTable t = null; HTable t = null;
if (region.getRegionInfo().getTableDesc().getName().equals(META_TABLE_NAME)) { if (region.getRegionInfo().isMetaTable()) {
// We need to update the root region // We need to update the root region
if (this.root == null) { if (this.root == null) {
this.root = new HTable(conf, ROOT_TABLE_NAME); this.root = new HTable(conf, ROOT_TABLE_NAME);