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.util;
020
021import java.io.InterruptedIOException;
022import java.io.IOException;
023import java.lang.reflect.Constructor;
024import java.lang.reflect.InvocationTargetException;
025import java.util.List;
026
027import org.apache.yetus.audience.InterfaceAudience;
028import org.slf4j.Logger;
029import org.slf4j.LoggerFactory;
030import org.apache.hadoop.conf.Configuration;
031import org.apache.hadoop.hbase.CoordinatedStateManager;
032import org.apache.hadoop.hbase.master.HMaster;
033import org.apache.hadoop.hbase.regionserver.HRegionServer;
034
035/**
036 * Utility used running a cluster all in the one JVM.
037 */
038@InterfaceAudience.Private
039public class JVMClusterUtil {
040  private static final Logger LOG = LoggerFactory.getLogger(JVMClusterUtil.class);
041
042  /**
043   * Datastructure to hold RegionServer Thread and RegionServer instance
044   */
045  public static class RegionServerThread extends Thread {
046    private final HRegionServer regionServer;
047
048    public RegionServerThread(final HRegionServer r, final int index) {
049      super(r, "RS:" + index + ";" + r.getServerName().toShortString());
050      this.regionServer = r;
051    }
052
053    /** @return the region server */
054    public HRegionServer getRegionServer() {
055      return this.regionServer;
056    }
057
058    /**
059     * Block until the region server has come online, indicating it is ready
060     * to be used.
061     */
062    public void waitForServerOnline() {
063      // The server is marked online after the init method completes inside of
064      // the HRS#run method.  HRS#init can fail for whatever region.  In those
065      // cases, we'll jump out of the run without setting online flag.  Check
066      // stopRequested so we don't wait here a flag that will never be flipped.
067      regionServer.waitForServerOnline();
068    }
069  }
070
071  /**
072   * Creates a {@link RegionServerThread}.
073   * Call 'start' on the returned thread to make it run.
074   * @param c Configuration to use.
075   * @param hrsc Class to create.
076   * @param index Used distinguishing the object returned.
077   * @throws IOException
078   * @return Region server added.
079   */
080  public static JVMClusterUtil.RegionServerThread createRegionServerThread(final Configuration c,
081      final Class<? extends HRegionServer> hrsc, final int index) throws IOException {
082    HRegionServer server;
083    try {
084      Constructor<? extends HRegionServer> ctor = hrsc.getConstructor(Configuration.class);
085      ctor.setAccessible(true);
086      server = ctor.newInstance(c);
087    } catch (InvocationTargetException ite) {
088      Throwable target = ite.getTargetException();
089      throw new RuntimeException("Failed construction of RegionServer: " +
090        hrsc.toString() + ((target.getCause() != null)?
091          target.getCause().getMessage(): ""), target);
092    } catch (Exception e) {
093      IOException ioe = new IOException();
094      ioe.initCause(e);
095      throw ioe;
096    }
097    return new JVMClusterUtil.RegionServerThread(server, index);
098  }
099
100
101  /**
102   * Datastructure to hold Master Thread and Master instance
103   */
104  public static class MasterThread extends Thread {
105    private final HMaster master;
106
107    public MasterThread(final HMaster m, final int index) {
108      super(m, "M:" + index + ";" + m.getServerName().toShortString());
109      this.master = m;
110    }
111
112    /** @return the master */
113    public HMaster getMaster() {
114      return this.master;
115    }
116  }
117
118  /**
119   * Creates a {@link MasterThread}.
120   * Call 'start' on the returned thread to make it run.
121   * @param c Configuration to use.
122   * @param hmc Class to create.
123   * @param index Used distinguishing the object returned.
124   * @throws IOException
125   * @return Master added.
126   */
127  public static JVMClusterUtil.MasterThread createMasterThread(final Configuration c,
128      final Class<? extends HMaster> hmc, final int index) throws IOException {
129    HMaster server;
130    try {
131      server = hmc.getConstructor(Configuration.class).newInstance(c);
132    } catch (InvocationTargetException ite) {
133      Throwable target = ite.getTargetException();
134      throw new RuntimeException("Failed construction of Master: " +
135        hmc.toString() + ((target.getCause() != null)?
136          target.getCause().getMessage(): ""), target);
137    } catch (Exception e) {
138      IOException ioe = new IOException();
139      ioe.initCause(e);
140      throw ioe;
141    }
142    return new JVMClusterUtil.MasterThread(server, index);
143  }
144
145  private static JVMClusterUtil.MasterThread findActiveMaster(
146    List<JVMClusterUtil.MasterThread> masters) {
147    for (JVMClusterUtil.MasterThread t : masters) {
148      if (t.master.isActiveMaster()) {
149        return t;
150      }
151    }
152
153    return null;
154  }
155
156  /**
157   * Start the cluster.  Waits until there is a primary master initialized
158   * and returns its address.
159   * @param masters
160   * @param regionservers
161   * @return Address to use contacting primary master.
162   */
163  public static String startup(final List<JVMClusterUtil.MasterThread> masters,
164      final List<JVMClusterUtil.RegionServerThread> regionservers) throws IOException {
165
166    Configuration configuration = null;
167
168    if (masters == null || masters.isEmpty()) {
169      return null;
170    }
171
172    for (JVMClusterUtil.MasterThread t : masters) {
173      configuration = t.getMaster().getConfiguration();
174      t.start();
175    }
176
177    // Wait for an active master
178    //  having an active master before starting the region threads allows
179    //  then to succeed on their connection to master
180    long startTime = System.currentTimeMillis();
181    while (findActiveMaster(masters) == null) {
182      try {
183        Thread.sleep(100);
184      } catch (InterruptedException e) {
185        throw (InterruptedIOException)new InterruptedIOException().initCause(e);
186      }
187      int startTimeout = configuration != null ? Integer.parseInt(
188        configuration.get("hbase.master.start.timeout.localHBaseCluster", "30000")) : 30000;
189      if (System.currentTimeMillis() > startTime + startTimeout) {
190        String msg = "Master not active after " + startTimeout + "ms";
191        Threads.printThreadInfo(System.out, "Thread dump because: " + msg);
192        throw new RuntimeException(msg);
193      }
194    }
195
196    if (regionservers != null) {
197      for (JVMClusterUtil.RegionServerThread t: regionservers) {
198        t.start();
199      }
200    }
201
202    // Wait for an active master to be initialized (implies being master)
203    //  with this, when we return the cluster is complete
204    startTime = System.currentTimeMillis();
205    final int maxwait = 200000;
206    while (true) {
207      JVMClusterUtil.MasterThread t = findActiveMaster(masters);
208      if (t != null && t.master.isInitialized()) {
209        return t.master.getServerName().toString();
210      }
211      // REMOVE
212      if (System.currentTimeMillis() > startTime + 10000) {
213        try {
214          Thread.sleep(1000);
215        } catch (InterruptedException e) {
216          throw (InterruptedIOException)new InterruptedIOException().initCause(e);
217        }
218      }
219      if (System.currentTimeMillis() > startTime + maxwait) {
220        String msg = "Master not initialized after " + maxwait + "ms seconds";
221        Threads.printThreadInfo(System.out, "Thread dump because: " + msg);
222        throw new RuntimeException(msg);
223      }
224      try {
225        Thread.sleep(100);
226      } catch (InterruptedException e) {
227        throw (InterruptedIOException)new InterruptedIOException().initCause(e);
228      }
229    }
230  }
231
232  /**
233   * @param masters
234   * @param regionservers
235   */
236  public static void shutdown(final List<MasterThread> masters,
237      final List<RegionServerThread> regionservers) {
238    LOG.debug("Shutting down HBase Cluster");
239    if (masters != null) {
240      // Do backups first.
241      JVMClusterUtil.MasterThread activeMaster = null;
242      for (JVMClusterUtil.MasterThread t : masters) {
243        if (!t.master.isActiveMaster()) {
244          try {
245            t.master.stopMaster();
246          } catch (IOException e) {
247            LOG.error("Exception occurred while stopping master", e);
248          }
249        } else {
250          activeMaster = t;
251        }
252      }
253      // Do active after.
254      if (activeMaster != null) {
255        try {
256          activeMaster.master.shutdown();
257        } catch (IOException e) {
258          LOG.error("Exception occurred in HMaster.shutdown()", e);
259        }
260      }
261    }
262    boolean wasInterrupted = false;
263    final long maxTime = System.currentTimeMillis() + 30 * 1000;
264    if (regionservers != null) {
265      // first try nicely.
266      for (RegionServerThread t : regionservers) {
267        t.getRegionServer().stop("Shutdown requested");
268      }
269      for (RegionServerThread t : regionservers) {
270        long now = System.currentTimeMillis();
271        if (t.isAlive() && !wasInterrupted && now < maxTime) {
272          try {
273            t.join(maxTime - now);
274          } catch (InterruptedException e) {
275            LOG.info("Got InterruptedException on shutdown - " +
276                "not waiting anymore on region server ends", e);
277            wasInterrupted = true; // someone wants us to speed up.
278          }
279        }
280      }
281
282      // Let's try to interrupt the remaining threads if any.
283      for (int i = 0; i < 100; ++i) {
284        boolean atLeastOneLiveServer = false;
285        for (RegionServerThread t : regionservers) {
286          if (t.isAlive()) {
287            atLeastOneLiveServer = true;
288            try {
289              LOG.warn("RegionServerThreads remaining, give one more chance before interrupting");
290              t.join(1000);
291            } catch (InterruptedException e) {
292              wasInterrupted = true;
293            }
294          }
295        }
296        if (!atLeastOneLiveServer) break;
297        for (RegionServerThread t : regionservers) {
298          if (t.isAlive()) {
299            LOG.warn("RegionServerThreads taking too long to stop, interrupting; thread dump "  +
300              "if > 3 attempts: i=" + i);
301            if (i > 3) {
302              Threads.printThreadInfo(System.out, "Thread dump " + t.getName());
303            }
304            t.interrupt();
305          }
306        }
307      }
308    }
309
310    if (masters != null) {
311      for (JVMClusterUtil.MasterThread t : masters) {
312        while (t.master.isAlive() && !wasInterrupted) {
313          try {
314            // The below has been replaced to debug sometime hangs on end of
315            // tests.
316            // this.master.join():
317            Threads.threadDumpingIsAlive(t.master.getThread());
318          } catch(InterruptedException e) {
319            LOG.info("Got InterruptedException on shutdown - " +
320                "not waiting anymore on master ends", e);
321            wasInterrupted = true;
322          }
323        }
324      }
325    }
326    LOG.info("Shutdown of " +
327      ((masters != null) ? masters.size() : "0") + " master(s) and " +
328      ((regionservers != null) ? regionservers.size() : "0") +
329      " regionserver(s) " + (wasInterrupted ? "interrupted" : "complete"));
330
331    if (wasInterrupted){
332      Thread.currentThread().interrupt();
333    }
334  }
335}