Return-Path: 
 <commits-return-33249-apmail-hbase-commits-archive=hbase.apache.org@hbase.apache.org>
X-Original-To: apmail-hbase-commits-archive@www.apache.org
Delivered-To: apmail-hbase-commits-archive@www.apache.org
Received: from mail.apache.org (hermes.apache.org [140.211.11.3])
	by minotaur.apache.org (Postfix) with SMTP id 31D4C18D6C
	for <apmail-hbase-commits-archive@www.apache.org>;
 Wed,  9 Mar 2016 03:51:03 +0000 (UTC)
Received: (qmail 17489 invoked by uid 500); 9 Mar 2016 03:51:02 -0000
Delivered-To: apmail-hbase-commits-archive@hbase.apache.org
Received: (qmail 17309 invoked by uid 500); 9 Mar 2016 03:51:02 -0000
Mailing-List: contact commits-help@hbase.apache.org; run by ezmlm
Precedence: bulk
List-Help: <mailto:commits-help@hbase.apache.org>
List-Unsubscribe: <mailto:commits-unsubscribe@hbase.apache.org>
List-Post: <mailto:commits@hbase.apache.org>
List-Id: <commits.hbase.apache.org>
Reply-To: dev@hbase.apache.org
Delivered-To: mailing list commits@hbase.apache.org
Received: (qmail 17171 invoked by uid 99); 9 Mar 2016 03:51:02 -0000
Received: from git1-us-west.apache.org (HELO git1-us-west.apache.org)
 (140.211.11.23)
    by apache.org (qpsmtpd/0.29) with ESMTP; Wed, 09 Mar 2016 03:51:02 +0000
Received: by git1-us-west.apache.org (ASF Mail Server at
 git1-us-west.apache.org, from userid 33)
	id 933D1DFE13; Wed,  9 Mar 2016 03:51:02 +0000 (UTC)
Content-Type: text/plain; charset="us-ascii"
MIME-Version: 1.0
Content-Transfer-Encoding: 7bit
From: mbertozzi@apache.org
To: commits@hbase.apache.org
Date: Wed, 09 Mar 2016 03:51:04 -0000
Message-Id: <67505cc960354bbfafc3b97d20784f89@git.apache.org>
In-Reply-To: <7d1a0e897d9c49d6897eb649bf1f54fa@git.apache.org>
References: <7d1a0e897d9c49d6897eb649bf1f54fa@git.apache.org>
X-Mailer: ASF-Git Admin Mailer
Subject: [3/8] hbase git commit: HBASE-15422 Procedure v2 - Avoid double yield

HBASE-15422 Procedure v2 - Avoid double yield


Project: http://git-wip-us.apache.org/repos/asf/hbase/repo
Commit: http://git-wip-us.apache.org/repos/asf/hbase/commit/84aceed3
Tree: http://git-wip-us.apache.org/repos/asf/hbase/tree/84aceed3
Diff: http://git-wip-us.apache.org/repos/asf/hbase/diff/84aceed3

Branch: refs/heads/branch-1.3
Commit: 84aceed3b32e71f38125e1aaa8bb655e5969041a
Parents: a623f45
Author: Matteo Bertozzi <matteo.bertozzi@cloudera.com>
Authored: Tue Mar 8 11:07:33 2016 -0800
Committer: Matteo Bertozzi <matteo.bertozzi@cloudera.com>
Committed: Tue Mar 8 11:24:22 2016 -0800

----------------------------------------------------------------------
 .../hbase/procedure2/ProcedureExecutor.java     |  10 +-
 .../hbase/procedure2/TestYieldProcedures.java   | 154 ++++++++++++++++---
 2 files changed, 132 insertions(+), 32 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/hbase/blob/84aceed3/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/ProcedureExecutor.java
----------------------------------------------------------------------
diff --git a/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/ProcedureExecutor.java b/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/ProcedureExecutor.java
index 2d51744..f45f2f9 100644
--- a/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/ProcedureExecutor.java
+++ b/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/ProcedureExecutor.java
@@ -876,12 +876,6 @@ public class ProcedureExecutor<TEnvironment> {
         }
         break;
       }
-
-      // if the procedure is kind enough to pass the slot to someone else, yield
-      if (proc.isYieldAfterExecutionStep(getEnvironment())) {
-        runnables.yield(proc);
-        break;
-      }
     } while (procStack.isFailed());
   }
 
@@ -1159,7 +1153,9 @@ public class ProcedureExecutor<TEnvironment> {
       }
 
       // if the procedure is kind enough to pass the slot to someone else, yield
-      if (reExecute && procedure.isYieldAfterExecutionStep(getEnvironment())) {
+      if (procedure.getState() == ProcedureState.RUNNABLE &&
+          procedure.isYieldAfterExecutionStep(getEnvironment())) {
+        runnables.yield(procedure);
         return;
       }
 

http://git-wip-us.apache.org/repos/asf/hbase/blob/84aceed3/hbase-procedure/src/test/java/org/apache/hadoop/hbase/procedure2/TestYieldProcedures.java
----------------------------------------------------------------------
diff --git a/hbase-procedure/src/test/java/org/apache/hadoop/hbase/procedure2/TestYieldProcedures.java b/hbase-procedure/src/test/java/org/apache/hadoop/hbase/procedure2/TestYieldProcedures.java
index 7ae76c4..211d06d 100644
--- a/hbase-procedure/src/test/java/org/apache/hadoop/hbase/procedure2/TestYieldProcedures.java
+++ b/hbase-procedure/src/test/java/org/apache/hadoop/hbase/procedure2/TestYieldProcedures.java
@@ -19,6 +19,8 @@
 package org.apache.hadoop.hbase.procedure2;
 
 import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.TimeoutException;
@@ -54,6 +56,7 @@ public class TestYieldProcedures {
   private static final Procedure NULL_PROC = null;
 
   private ProcedureExecutor<TestProcEnv> procExecutor;
+  private TestRunQueue procRunnables;
   private ProcedureStore procStore;
 
   private HBaseCommonTestingUtility htu;
@@ -70,7 +73,9 @@ public class TestYieldProcedures {
 
     logDir = new Path(testDir, "proc-logs");
     procStore = ProcedureTestingUtility.createWalStore(htu.getConfiguration(), fs, logDir);
-    procExecutor = new ProcedureExecutor(htu.getConfiguration(), new TestProcEnv(), procStore);
+    procRunnables = new TestRunQueue();
+    procExecutor = new ProcedureExecutor(htu.getConfiguration(), new TestProcEnv(),
+        procStore, procRunnables);
     procStore.start(PROCEDURE_EXECUTOR_SLOTS);
     procExecutor.start(PROCEDURE_EXECUTOR_SLOTS, true);
   }
@@ -93,34 +98,32 @@ public class TestYieldProcedures {
     }
     ProcedureTestingUtility.waitNoProcedureRunning(procExecutor);
 
-    // verify yield during execute()
-    long prevTimestamp = 0;
-    for (int execStep = 0; execStep < NUM_STATES; ++execStep) {
-      for (int i = 0; i < procs.length; ++i) {
-        assertEquals(NUM_STATES * 2, procs[i].getExecutionInfo().size());
-        TestStateMachineProcedure.ExecutionInfo info = procs[i].getExecutionInfo().get(execStep);
-        LOG.info("i=" + i + " execStep=" + execStep + " timestamp=" + info.getTimestamp());
+    for (int i = 0; i < procs.length; ++i) {
+      assertEquals(NUM_STATES * 2, procs[i].getExecutionInfo().size());
+
+      // verify execution
+      int index = 0;
+      for (int execStep = 0; execStep < NUM_STATES; ++execStep) {
+        TestStateMachineProcedure.ExecutionInfo info = procs[i].getExecutionInfo().get(index++);
         assertEquals(false, info.isRollback());
         assertEquals(execStep, info.getStep().ordinal());
-        assertEquals(prevTimestamp + 1, info.getTimestamp());
-        prevTimestamp++;
       }
-    }
 
-    // verify yield during rollback()
-    int count = NUM_STATES;
-    for (int execStep = NUM_STATES - 1; execStep >= 0; --execStep) {
-      for (int i = 0; i < procs.length; ++i) {
-        assertEquals(NUM_STATES * 2, procs[i].getExecutionInfo().size());
-        TestStateMachineProcedure.ExecutionInfo info = procs[i].getExecutionInfo().get(count);
-        LOG.info("i=" + i + " execStep=" + execStep + " timestamp=" + info.getTimestamp());
+      // verify rollback
+      for (int execStep = NUM_STATES - 1; execStep >= 0; --execStep) {
+        TestStateMachineProcedure.ExecutionInfo info = procs[i].getExecutionInfo().get(index++);
         assertEquals(true, info.isRollback());
         assertEquals(execStep, info.getStep().ordinal());
-        assertEquals(prevTimestamp + 1, info.getTimestamp());
-        prevTimestamp++;
       }
-      count++;
     }
+
+    // check runnable queue stats
+    assertEquals(0, procRunnables.size());
+    assertEquals(0, procRunnables.addFrontCalls);
+    assertEquals(18, procRunnables.addBackCalls);
+    assertEquals(15, procRunnables.yieldCalls);
+    assertEquals(19, procRunnables.pollCalls);
+    assertEquals(3, procRunnables.completionCalls);
   }
 
   @Test
@@ -153,6 +156,29 @@ public class TestYieldProcedures {
       assertEquals(true, info.isRollback());
       assertEquals(i, info.getStep().ordinal());
     }
+
+    // check runnable queue stats
+    assertEquals(0, procRunnables.size());
+    assertEquals(0, procRunnables.addFrontCalls);
+    assertEquals(12, procRunnables.addBackCalls);
+    assertEquals(11, procRunnables.yieldCalls);
+    assertEquals(13, procRunnables.pollCalls);
+    assertEquals(1, procRunnables.completionCalls);
+  }
+
+  @Test
+  public void testYieldException() {
+    TestYieldProcedure proc = new TestYieldProcedure();
+    ProcedureTestingUtility.submitAndWait(procExecutor, proc);
+    assertEquals(6, proc.step);
+
+    // check runnable queue stats
+    assertEquals(0, procRunnables.size());
+    assertEquals(0, procRunnables.addFrontCalls);
+    assertEquals(6, procRunnables.addBackCalls);
+    assertEquals(5, procRunnables.yieldCalls);
+    assertEquals(7, procRunnables.pollCalls);
+    assertEquals(1, procRunnables.completionCalls);
   }
 
   private static class TestProcEnv {
@@ -205,8 +231,9 @@ public class TestYieldProcedures {
     @Override
     protected StateMachineProcedure.Flow executeFromState(TestProcEnv env, State state)
         throws InterruptedException {
-      LOG.info("execute step " + state);
-      executionInfo.add(new ExecutionInfo(env.nextTimestamp(), state, false));
+      final long ts = env.nextTimestamp();
+      LOG.info(getProcId() + " execute step " + state + " ts=" + ts);
+      executionInfo.add(new ExecutionInfo(ts, state, false));
       Thread.sleep(150);
 
       if (throwInterruptOnceOnEachStep && ((executionInfo.size() - 1) % 2) == 0) {
@@ -235,8 +262,9 @@ public class TestYieldProcedures {
     @Override
     protected void rollbackState(TestProcEnv env, final State state)
         throws InterruptedException {
-      LOG.debug("rollback state " + state);
-      executionInfo.add(new ExecutionInfo(env.nextTimestamp(), state, true));
+      final long ts = env.nextTimestamp();
+      LOG.debug(getProcId() + " rollback state " + state + " ts=" + ts);
+      executionInfo.add(new ExecutionInfo(ts, state, true));
       Thread.sleep(150);
 
       if (throwInterruptOnceOnEachStep && ((executionInfo.size() - 1) % 2) == 0) {
@@ -282,4 +310,80 @@ public class TestYieldProcedures {
       return true;
     }
   }
+
+  public static class TestYieldProcedure extends Procedure<TestProcEnv> {
+    private int step = 0;
+
+    public TestYieldProcedure() {
+    }
+
+    @Override
+    protected Procedure[] execute(final TestProcEnv env) throws ProcedureYieldException {
+      LOG.info("execute step " + step);
+      if (step++ < 5) {
+        throw new ProcedureYieldException();
+      }
+      return null;
+    }
+
+    @Override
+    protected void rollback(TestProcEnv env) {
+    }
+
+    @Override
+    protected boolean abort(TestProcEnv env) {
+      return false;
+    }
+
+    @Override
+    protected boolean isYieldAfterExecutionStep(final TestProcEnv env) {
+      return true;
+    }
+
+    @Override
+    protected void serializeStateData(final OutputStream stream) throws IOException {
+    }
+
+    @Override
+    protected void deserializeStateData(final InputStream stream) throws IOException {
+    }
+  }
+
+  private static class TestRunQueue extends ProcedureSimpleRunQueue {
+    private int completionCalls;
+    private int addFrontCalls;
+    private int addBackCalls;
+    private int yieldCalls;
+    private int pollCalls;
+
+    public TestRunQueue() {}
+
+    public void addFront(final Procedure proc) {
+        addFrontCalls++;
+        super.addFront(proc);
+      }
+
+      @Override
+      public void addBack(final Procedure proc) {
+        addBackCalls++;
+        super.addBack(proc);
+      }
+
+      @Override
+      public void yield(final Procedure proc) {
+        yieldCalls++;
+        super.yield(proc);
+      }
+
+      @Override
+      public Procedure poll() {
+        pollCalls++;
+        return super.poll();
+      }
+
+      @Override
+      public void completionCleanup(Procedure proc) {
+        completionCalls++;
+      }
+  }
 }