From e17f094ddb4132a814de82ffb3bacfd331dd9310 Mon Sep 17 00:00:00 2001 From: nkeywal Date: Fri, 18 Oct 2013 19:34:18 +0000 Subject: [PATCH] HBASE-9768 Two issues in AsyncProcess git-svn-id: https://svn.apache.org/repos/asf/hbase/trunk@1533604 13f79535-47bb-0310-9956-ffa450edef68 --- .../hadoop/hbase/client/AsyncProcess.java | 22 ++++++++++++--- .../apache/hadoop/hbase/client/HTable.java | 28 +++++++++---------- 2 files changed, 32 insertions(+), 18 deletions(-) diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncProcess.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncProcess.java index 888370cab11..8b3ca5ebedc 100644 --- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncProcess.java +++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncProcess.java @@ -265,7 +265,18 @@ class AsyncProcess { new HashMap>(); List> retainedActions = new ArrayList>(rows.size()); + long currentTaskCnt = tasksDone.get(); + boolean alreadyLooped = false; + do { + if (alreadyLooped){ + // if, for whatever reason, we looped, we want to be sure that something has changed. + waitForNextTaskDone(currentTaskCnt); + currentTaskCnt = tasksDone.get(); + } else { + alreadyLooped = true; + } + // Wait until there is at least one slot for a new task. waitForMaximumCurrentTasks(maxTotalConcurrentTasks - 1); @@ -280,8 +291,9 @@ class AsyncProcess { Row r = it.next(); HRegionLocation loc = findDestLocation(r, 1, posInList); - if (loc != null && canTakeOperation(loc, regionIncluded, serverIncluded)) { - // loc is null if there is an error such as meta not available. + if (loc == null) { // loc is null if there is an error such as meta not available. + it.remove(); + } else if (canTakeOperation(loc, regionIncluded, serverIncluded)) { Action action = new Action(r, ++posInList); retainedActions.add(action); addAction(loc, action, actionsByServer); @@ -644,6 +656,7 @@ class AsyncProcess { for (Map.Entry>> resultsForRS : responses.getResults().entrySet()) { + boolean regionFailureRegistered = false; for (Pair regionResult : resultsForRS.getValue()) { Object result = regionResult.getSecond(); @@ -652,8 +665,9 @@ class AsyncProcess { throwable = (Throwable) result; Action correspondingAction = initialActions.get(regionResult.getFirst()); Row row = correspondingAction.getAction(); - - if (failureCount++ == 0) { // We're doing this once per location. + failureCount++; + if (!regionFailureRegistered) { // We're doing this once per location. + regionFailureRegistered= true; hConnection.updateCachedLocations(this.tableName, row.getRow(), result, location); if (errorsByServer != null) { errorsByServer.reportServerError(location); diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/HTable.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/HTable.java index 6783abb1769..00de79e6289 100644 --- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/HTable.java +++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/HTable.java @@ -58,8 +58,6 @@ import org.apache.hadoop.hbase.ipc.PayloadCarryingRpcController; import org.apache.hadoop.hbase.ipc.RegionCoprocessorRpcChannel; import org.apache.hadoop.hbase.protobuf.ProtobufUtil; import org.apache.hadoop.hbase.protobuf.RequestConverter; -import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.GetRequest; -import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.GetResponse; import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.MultiRequest; import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.MutateRequest; import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.MutateResponse; @@ -889,6 +887,7 @@ public class HTable implements HTableInterface { */ private void doPut(Put put) throws InterruptedIOException, RetriesExhaustedWithDetailsException { if (ap.hasError()){ + writeAsyncBuffer.add(put); backgroundFlushCommits(true); } @@ -907,25 +906,29 @@ public class HTable implements HTableInterface { * Send the operations in the buffer to the servers. Does not wait for the server's answer. * If the is an error (max retried reach from a previous flush or bad operation), it tries to * send all operations in the buffer and sends an exception. + * @param synchronous - if true, sends all the writes and wait for all of them to finish before + * returning. */ private void backgroundFlushCommits(boolean synchronous) throws InterruptedIOException, RetriesExhaustedWithDetailsException { try { - // If there is an error on the operations in progress, we don't add new operations. - if (writeAsyncBuffer.size() > 0 && !ap.hasError()) { + do { ap.submit(writeAsyncBuffer, true); - } + } while (synchronous && !writeAsyncBuffer.isEmpty()); - if (synchronous || ap.hasError()) { - if (ap.hasError() && LOG.isDebugEnabled()) { - LOG.debug(tableName + ": One or more of the operations have failed -" + - " waiting for all operation in progress to finish (successfully or not)"); - } + if (synchronous) { ap.waitUntilDone(); } if (ap.hasError()) { + LOG.debug(tableName + ": One or more of the operations have failed -" + + " waiting for all operation in progress to finish (successfully or not)"); + while (!writeAsyncBuffer.isEmpty()) { + ap.submit(writeAsyncBuffer, true); + } + ap.waitUntilDone(); + if (!clearBufferOnFail) { // if clearBufferOnFailed is not set, we're supposed to keep the failed operation in the // write buffer. This is a questionable feature kept here for backward compatibility @@ -1186,12 +1189,9 @@ public class HTable implements HTableInterface { */ @Override public void flushCommits() throws InterruptedIOException, RetriesExhaustedWithDetailsException { - // We're looping, as if one region is overloaded we keep its operations in the buffer. // As we can have an operation in progress even if the buffer is empty, we call // backgroundFlushCommits at least one time. - do { - backgroundFlushCommits(true); - } while (!writeAsyncBuffer.isEmpty()); + backgroundFlushCommits(true); } /**