001/**
002 *
003 * Licensed to the Apache Software Foundation (ASF) under one
004 * or more contributor license agreements.  See the NOTICE file
005 * distributed with this work for additional information
006 * regarding copyright ownership.  The ASF licenses this file
007 * to you under the Apache License, Version 2.0 (the
008 * "License"); you may not use this file except in compliance
009 * with the License.  You may obtain a copy of the License at
010 *
011 *     http://www.apache.org/licenses/LICENSE-2.0
012 *
013 * Unless required by applicable law or agreed to in writing, software
014 * distributed under the License is distributed on an "AS IS" BASIS,
015 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
016 * See the License for the specific language governing permissions and
017 * limitations under the License.
018 */
019package org.apache.hadoop.hbase;
020
021import java.io.IOException;
022import java.util.HashSet;
023import java.util.List;
024import java.util.Set;
025import java.util.concurrent.ExecutionException;
026import java.util.concurrent.Future;
027
028import org.apache.hadoop.conf.Configuration;
029import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
030import org.slf4j.Logger;
031import org.slf4j.LoggerFactory;
032
033public abstract class MultithreadedTestUtil {
034
035  private static final Logger LOG =
036    LoggerFactory.getLogger(MultithreadedTestUtil.class);
037
038  public static class TestContext {
039    private final Configuration conf;
040    private Throwable err = null;
041    private boolean stopped = false;
042    private int threadDoneCount = 0;
043    private Set<TestThread> testThreads = new HashSet<>();
044
045    public TestContext(Configuration configuration) {
046      this.conf = configuration;
047    }
048
049    protected Configuration getConf() {
050      return conf;
051    }
052
053    public synchronized boolean shouldRun()  {
054      return !stopped && err == null;
055    }
056
057    public void addThread(TestThread t) {
058      testThreads.add(t);
059    }
060
061    public void startThreads() {
062      for (TestThread t : testThreads) {
063        t.start();
064      }
065    }
066
067    public void waitFor(long millis) throws Exception {
068      long endTime = EnvironmentEdgeManager.currentTime() + millis;
069      while (!stopped) {
070        long left = endTime - EnvironmentEdgeManager.currentTime();
071        if (left <= 0) break;
072        synchronized (this) {
073          checkException();
074          wait(left);
075        }
076      }
077    }
078    private synchronized void checkException() throws Exception {
079      if (err != null) {
080        throw new RuntimeException("Deferred", err);
081      }
082    }
083
084    public synchronized void threadFailed(Throwable t) {
085      if (err == null) err = t;
086      LOG.error("Failed!", err);
087      notify();
088    }
089
090    public synchronized void threadDone() {
091      threadDoneCount++;
092    }
093
094    public void setStopFlag(boolean s) throws Exception {
095      synchronized (this) {
096        stopped = s;
097      }
098    }
099
100    public void stop() throws Exception {
101      synchronized (this) {
102        stopped = true;
103      }
104      for (TestThread t : testThreads) {
105        t.join();
106      }
107      checkException();
108    }
109  }
110
111  /**
112   * A thread that can be added to a test context, and properly
113   * passes exceptions through.
114   */
115  public static abstract class TestThread extends Thread {
116    protected final TestContext ctx;
117    protected boolean stopped;
118
119    public TestThread(TestContext ctx) {
120      this.ctx = ctx;
121    }
122
123    @Override
124    public void run() {
125      try {
126        doWork();
127      } catch (Throwable t) {
128        ctx.threadFailed(t);
129      }
130      ctx.threadDone();
131    }
132
133    public abstract void doWork() throws Exception;
134
135    protected void stopTestThread() {
136      this.stopped = true;
137    }
138  }
139
140  /**
141   * A test thread that performs a repeating operation.
142   */
143  public static abstract class RepeatingTestThread extends TestThread {
144    public RepeatingTestThread(TestContext ctx) {
145      super(ctx);
146    }
147
148    @Override
149    public final void doWork() throws Exception {
150      try {
151        while (ctx.shouldRun() && !stopped) {
152          doAnAction();
153        }
154      } finally {
155        workDone();
156      }
157    }
158
159    public abstract void doAnAction() throws Exception;
160    public void workDone() throws IOException {}
161  }
162
163  /**
164   * Verify that no assertions have failed inside a future.
165   * Used for unit tests that spawn threads. E.g.,
166   * <p>
167   * <pre>
168   *   List&lt;Future&lt;Void>> results = Lists.newArrayList();
169   *   Future&lt;Void> f = executor.submit(new Callable&lt;Void> {
170   *     public Void call() {
171   *       assertTrue(someMethod());
172   *     }
173   *   });
174   *   results.add(f);
175   *   assertOnFutures(results);
176   * </pre>
177   * @param threadResults A list of futures
178   * @throws InterruptedException If interrupted when waiting for a result
179   *                              from one of the futures
180   * @throws ExecutionException If an exception other than AssertionError
181   *                            occurs inside any of the futures
182   */
183  public static void assertOnFutures(List<Future<?>> threadResults)
184  throws InterruptedException, ExecutionException {
185    for (Future<?> threadResult : threadResults) {
186      try {
187        threadResult.get();
188      } catch (ExecutionException e) {
189        if (e.getCause() instanceof AssertionError) {
190          throw (AssertionError) e.getCause();
191        }
192        throw e;
193      }
194    }
195  }
196}