001/**
002 * Licensed to the Apache Software Foundation (ASF) under one
003 * or more contributor license agreements.  See the NOTICE file
004 * distributed with this work for additional information
005 * regarding copyright ownership.  The ASF licenses this file
006 * to you under the Apache License, Version 2.0 (the
007 * "License"); you may not use this file except in compliance
008 * with the License.  You may obtain a copy of the License at
009 *
010 *     http://www.apache.org/licenses/LICENSE-2.0
011 *
012 * Unless required by applicable law or agreed to in writing, software
013 * distributed under the License is distributed on an "AS IS" BASIS,
014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
015 * See the License for the specific language governing permissions and
016 * limitations under the License.
017 */
018package org.apache.hadoop.hbase.procedure2;
019
020import static org.junit.Assert.assertEquals;
021
022import java.io.IOException;
023import java.util.ArrayList;
024import java.util.concurrent.atomic.AtomicBoolean;
025import java.util.concurrent.atomic.AtomicLong;
026import org.apache.hadoop.hbase.HBaseClassTestRule;
027import org.apache.hadoop.hbase.HBaseCommonTestingUtil;
028import org.apache.hadoop.hbase.procedure2.store.NoopProcedureStore;
029import org.apache.hadoop.hbase.procedure2.store.ProcedureStore;
030import org.apache.hadoop.hbase.testclassification.MasterTests;
031import org.apache.hadoop.hbase.testclassification.SmallTests;
032import org.apache.hadoop.hbase.util.Threads;
033import org.junit.After;
034import org.junit.Before;
035import org.junit.ClassRule;
036import org.junit.Test;
037import org.junit.experimental.categories.Category;
038import org.slf4j.Logger;
039import org.slf4j.LoggerFactory;
040
041@Category({MasterTests.class, SmallTests.class})
042public class TestProcedureSuspended {
043  @ClassRule
044  public static final HBaseClassTestRule CLASS_RULE =
045      HBaseClassTestRule.forClass(TestProcedureSuspended.class);
046
047  private static final Logger LOG = LoggerFactory.getLogger(TestProcedureSuspended.class);
048
049  private static final int PROCEDURE_EXECUTOR_SLOTS = 1;
050  private static final Procedure NULL_PROC = null;
051
052  private ProcedureExecutor<TestProcEnv> procExecutor;
053  private ProcedureStore procStore;
054
055  private HBaseCommonTestingUtil htu;
056
057  @Before
058  public void setUp() throws IOException {
059    htu = new HBaseCommonTestingUtil();
060
061    procStore = new NoopProcedureStore();
062    procExecutor = new ProcedureExecutor<>(htu.getConfiguration(), new TestProcEnv(), procStore);
063    procStore.start(PROCEDURE_EXECUTOR_SLOTS);
064    ProcedureTestingUtility.initAndStartWorkers(procExecutor, PROCEDURE_EXECUTOR_SLOTS, true);
065  }
066
067  @After
068  public void tearDown() throws IOException {
069    procExecutor.stop();
070    procStore.stop(false);
071  }
072
073  @Test
074  public void testSuspendWhileHoldingLocks() {
075    final AtomicBoolean lockA = new AtomicBoolean(false);
076    final AtomicBoolean lockB = new AtomicBoolean(false);
077
078    final TestLockProcedure p1keyA = new TestLockProcedure(lockA, "keyA", false, true);
079    final TestLockProcedure p2keyA = new TestLockProcedure(lockA, "keyA", false, true);
080    final TestLockProcedure p3keyB = new TestLockProcedure(lockB, "keyB", false, true);
081
082    procExecutor.submitProcedure(p1keyA);
083    procExecutor.submitProcedure(p2keyA);
084    procExecutor.submitProcedure(p3keyB);
085
086    // first run p1, p3 are able to run p2 is blocked by p1
087    waitAndAssertTimestamp(p1keyA, 1, 1);
088    waitAndAssertTimestamp(p2keyA, 0, -1);
089    waitAndAssertTimestamp(p3keyB, 1, 2);
090    assertEquals(true, lockA.get());
091    assertEquals(true, lockB.get());
092
093    // release p3
094    p3keyB.setThrowSuspend(false);
095    procExecutor.getScheduler().addFront(p3keyB);
096    waitAndAssertTimestamp(p1keyA, 1, 1);
097    waitAndAssertTimestamp(p2keyA, 0, -1);
098    waitAndAssertTimestamp(p3keyB, 2, 3);
099    assertEquals(true, lockA.get());
100
101    // wait until p3 is fully completed
102    ProcedureTestingUtility.waitProcedure(procExecutor, p3keyB);
103    assertEquals(false, lockB.get());
104
105    // rollback p2 and wait until is fully completed
106    p1keyA.setTriggerRollback(true);
107    procExecutor.getScheduler().addFront(p1keyA);
108    ProcedureTestingUtility.waitProcedure(procExecutor, p1keyA);
109
110    // p2 should start and suspend
111    waitAndAssertTimestamp(p1keyA, 4, 60000);
112    waitAndAssertTimestamp(p2keyA, 1, 7);
113    waitAndAssertTimestamp(p3keyB, 2, 3);
114    assertEquals(true, lockA.get());
115
116    // wait until p2 is fully completed
117    p2keyA.setThrowSuspend(false);
118    procExecutor.getScheduler().addFront(p2keyA);
119    ProcedureTestingUtility.waitProcedure(procExecutor, p2keyA);
120    waitAndAssertTimestamp(p1keyA, 4, 60000);
121    waitAndAssertTimestamp(p2keyA, 2, 8);
122    waitAndAssertTimestamp(p3keyB, 2, 3);
123    assertEquals(false, lockA.get());
124    assertEquals(false, lockB.get());
125  }
126
127  @Test
128  public void testYieldWhileHoldingLocks() {
129    final AtomicBoolean lock = new AtomicBoolean(false);
130
131    final TestLockProcedure p1 = new TestLockProcedure(lock, "key", true, false);
132    final TestLockProcedure p2 = new TestLockProcedure(lock, "key", true, false);
133
134    procExecutor.submitProcedure(p1);
135    procExecutor.submitProcedure(p2);
136
137    // try to execute a bunch of yield on p1, p2 should be blocked
138    while (p1.getTimestamps().size() < 100) {
139      Threads.sleep(10);
140    }
141
142    assertEquals(0, p2.getTimestamps().size());
143
144    // wait until p1 is completed
145    p1.setThrowYield(false);
146    ProcedureTestingUtility.waitProcedure(procExecutor, p1);
147
148    // try to execute a bunch of yield on p2
149    while (p2.getTimestamps().size() < 100) {
150      Threads.sleep(10);
151    }
152
153    assertEquals(p1.getTimestamps().get(p1.getTimestamps().size() - 1).longValue() + 1,
154      p2.getTimestamps().get(0).longValue());
155
156    // wait until p2 is completed
157    p1.setThrowYield(false);
158    ProcedureTestingUtility.waitProcedure(procExecutor, p1);
159  }
160
161  private void waitAndAssertTimestamp(TestLockProcedure proc, int size, int lastTs) {
162    final ArrayList<Long> timestamps = proc.getTimestamps();
163    while (timestamps.size() < size) {
164      Threads.sleep(10);
165    }
166
167    LOG.info(proc + " -> " + timestamps);
168    assertEquals(size, timestamps.size());
169    if (size > 0) {
170      assertEquals(lastTs, timestamps.get(timestamps.size() - 1).longValue());
171    }
172  }
173
174  public static class TestLockProcedure extends Procedure<TestProcEnv> {
175    private final ArrayList<Long> timestamps = new ArrayList<>();
176    private final String key;
177
178    private boolean triggerRollback = false;
179    private boolean throwSuspend = false;
180    private boolean throwYield = false;
181    private AtomicBoolean lock = null;
182    private boolean hasLock = false;
183
184    public TestLockProcedure(final AtomicBoolean lock, final String key,
185        final boolean throwYield, final boolean throwSuspend) {
186      this.lock = lock;
187      this.key = key;
188      this.throwYield = throwYield;
189      this.throwSuspend = throwSuspend;
190    }
191
192    public void setThrowYield(final boolean throwYield) {
193      this.throwYield = throwYield;
194    }
195
196    public void setThrowSuspend(final boolean throwSuspend) {
197      this.throwSuspend = throwSuspend;
198    }
199
200    public void setTriggerRollback(final boolean triggerRollback) {
201      this.triggerRollback = triggerRollback;
202    }
203
204    @Override
205    protected Procedure[] execute(final TestProcEnv env)
206        throws ProcedureYieldException, ProcedureSuspendedException {
207      LOG.info("EXECUTE " + this + " suspend " + (lock != null));
208      timestamps.add(env.nextTimestamp());
209      if (triggerRollback) {
210        setFailure(getClass().getSimpleName(), new Exception("injected failure"));
211      } else if (throwYield) {
212        throw new ProcedureYieldException();
213      } else if (throwSuspend) {
214        throw new ProcedureSuspendedException();
215      }
216      return null;
217    }
218
219    @Override
220    protected void rollback(final TestProcEnv env) {
221      LOG.info("ROLLBACK " + this);
222      timestamps.add(env.nextTimestamp() * 10000);
223    }
224
225    @Override
226    protected LockState acquireLock(final TestProcEnv env) {
227      hasLock = lock.compareAndSet(false, true);
228      if (hasLock) {
229        LOG.info("ACQUIRE LOCK " + this + " " + (hasLock));
230        return LockState.LOCK_ACQUIRED;
231      }
232      return LockState.LOCK_YIELD_WAIT;
233    }
234
235    @Override
236    protected void releaseLock(final TestProcEnv env) {
237      LOG.info("RELEASE LOCK " + this + " " + hasLock);
238      lock.set(false);
239    }
240
241    @Override
242    protected boolean holdLock(final TestProcEnv env) {
243      return true;
244    }
245
246    public ArrayList<Long> getTimestamps() {
247      return timestamps;
248    }
249
250    @Override
251    protected void toStringClassDetails(StringBuilder builder) {
252      builder.append(getClass().getName());
253      builder.append("(" + key + ")");
254    }
255
256    @Override
257    protected boolean abort(TestProcEnv env) {
258      return false;
259    }
260
261    @Override
262    protected void serializeStateData(ProcedureStateSerializer serializer)
263        throws IOException {
264    }
265
266    @Override
267    protected void deserializeStateData(ProcedureStateSerializer serializer)
268        throws IOException {
269    }
270  }
271
272  private static class TestProcEnv {
273    public final AtomicLong timestamp = new AtomicLong(0);
274
275    public long nextTimestamp() {
276      return timestamp.incrementAndGet();
277    }
278  }
279}