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

HBASE-10650 Fix incorrect handling of IE that restores current thread's interrupt status within while/for loops in RegionServer (Feng Honghua) 

git-svn-id: https://svn.apache.org/repos/asf/hbase/trunk@1573942 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
nkeywal 2014-03-04 08:50:01 +00:00
parent 84f48f87f8
commit f9dab7dddd
2 changed files with 40 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
View File

@ -1130,13 +1130,21 @@ public class HRegion implements HeapSize { // , Writable{
*/
public void waitForFlushesAndCompactions() {
synchronized (writestate) {
while (writestate.compacting > 0 || writestate.flushing) {
LOG.debug("waiting for " + writestate.compacting + " compactions"
boolean interrupted = false;
try {
while (writestate.compacting > 0 || writestate.flushing) {
LOG.debug("waiting for " + writestate.compacting + " compactions"
+ (writestate.flushing ? " & cache flush" : "") + " to complete for region " + this);
try {
writestate.wait();
} catch (InterruptedException iex) {
// essentially ignore and propagate the interrupt back up
try {
writestate.wait();
} catch (InterruptedException iex) {
// essentially ignore and propagate the interrupt back up
LOG.warn("Interrupted while waiting");
interrupted = true;
}
}
} finally {
if (interrupted) {
Thread.currentThread().interrupt();
}
}

43
hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreFlusher.java
View File

@ -548,27 +548,36 @@ class MemStoreFlusher implements FlushRequester {
synchronized (this.blockSignal) {
boolean blocked = false;
long startTime = 0;
while (isAboveHighWaterMark() && !server.isStopped()) {
if (!blocked) {
startTime = EnvironmentEdgeManager.currentTimeMillis();
LOG.info("Blocking updates on " + server.toString() +
": the global memstore size " +
StringUtils.humanReadableInt(server.getRegionServerAccounting().getGlobalMemstoreSize()) +
" is >= than blocking " +
StringUtils.humanReadableInt(globalMemStoreLimit) + " size");
boolean interrupted = false;
try {
while (isAboveHighWaterMark() && !server.isStopped()) {
if (!blocked) {
startTime = EnvironmentEdgeManager.currentTimeMillis();
LOG.info("Blocking updates on " + server.toString() +
": the global memstore size " +
StringUtils.humanReadableInt(server.getRegionServerAccounting().getGlobalMemstoreSize()) +
" is >= than blocking " +
StringUtils.humanReadableInt(globalMemStoreLimit) + " size");
}
blocked = true;
wakeupFlushThread();
try {
// we should be able to wait forever, but we've seen a bug where
// we miss a notify, so put a 5 second bound on it at least.
blockSignal.wait(5 * 1000);
} catch (InterruptedException ie) {
LOG.warn("Interrupted while waiting");
interrupted = true;
}
long took = System.currentTimeMillis() - start;
LOG.warn("Memstore is above high water mark and block " + took + "ms");
}
blocked = true;
wakeupFlushThread();
try {
// we should be able to wait forever, but we've seen a bug where
// we miss a notify, so put a 5 second bound on it at least.
blockSignal.wait(5 * 1000);
} catch (InterruptedException ie) {
} finally {
if (interrupted) {
Thread.currentThread().interrupt();
}
long took = System.currentTimeMillis() - start;
LOG.warn("Memstore is above high water mark and block " + took + "ms");
}
if(blocked){
final long totalTime = EnvironmentEdgeManager.currentTimeMillis() - startTime;
if(totalTime > 0){