View Javadoc

1   /**
2    *
3    * Licensed to the Apache Software Foundation (ASF) under one
4    * or more contributor license agreements.  See the NOTICE file
5    * distributed with this work for additional information
6    * regarding copyright ownership.  The ASF licenses this file
7    * to you under the Apache License, Version 2.0 (the
8    * "License"); you may not use this file except in compliance
9    * with the License.  You may obtain a copy of the License at
10   *
11   *     http://www.apache.org/licenses/LICENSE-2.0
12   *
13   * Unless required by applicable law or agreed to in writing, software
14   * distributed under the License is distributed on an "AS IS" BASIS,
15   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16   * See the License for the specific language governing permissions and
17   * limitations under the License.
18   */
19  package org.apache.hadoop.hbase.util;
20  
21  import java.io.InterruptedIOException;
22  import java.io.IOException;
23  import java.lang.reflect.Constructor;
24  import java.lang.reflect.InvocationTargetException;
25  import java.util.List;
26  
27  import org.apache.commons.logging.Log;
28  import org.apache.commons.logging.LogFactory;
29  import org.apache.hadoop.hbase.classification.InterfaceAudience;
30  import org.apache.hadoop.conf.Configuration;
31  import org.apache.hadoop.hbase.CoordinatedStateManager;
32  import org.apache.hadoop.hbase.master.HMaster;
33  import org.apache.hadoop.hbase.regionserver.HRegionServer;
34  
35  /**
36   * Utility used running a cluster all in the one JVM.
37   */
38  @InterfaceAudience.Private
39  public class JVMClusterUtil {
40    private static final Log LOG = LogFactory.getLog(JVMClusterUtil.class);
41  
42    /**
43     * Datastructure to hold RegionServer Thread and RegionServer instance
44     */
45    public static class RegionServerThread extends Thread {
46      private final HRegionServer regionServer;
47  
48      public RegionServerThread(final HRegionServer r, final int index) {
49        super(r, "RS:" + index + ";" + r.getServerName().toShortString());
50        this.regionServer = r;
51      }
52  
53      /** @return the region server */
54      public HRegionServer getRegionServer() {
55        return this.regionServer;
56      }
57  
58      /**
59       * Block until the region server has come online, indicating it is ready
60       * to be used.
61       */
62      public void waitForServerOnline() {
63        // The server is marked online after the init method completes inside of
64        // the HRS#run method.  HRS#init can fail for whatever region.  In those
65        // cases, we'll jump out of the run without setting online flag.  Check
66        // stopRequested so we don't wait here a flag that will never be flipped.
67        regionServer.waitForServerOnline();
68      }
69    }
70  
71    /**
72     * Creates a {@link RegionServerThread}.
73     * Call 'start' on the returned thread to make it run.
74     * @param c Configuration to use.
75     * @param cp consensus provider to use
76     * @param hrsc Class to create.
77     * @param index Used distinguishing the object returned.
78     * @throws IOException
79     * @return Region server added.
80     */
81    public static JVMClusterUtil.RegionServerThread createRegionServerThread(
82        final Configuration c, CoordinatedStateManager cp, final Class<? extends HRegionServer> hrsc,
83        final int index)
84    throws IOException {
85      HRegionServer server;
86      try {
87  
88        Constructor<? extends HRegionServer> ctor = hrsc.getConstructor(Configuration.class,
89        CoordinatedStateManager.class);
90        ctor.setAccessible(true);
91        server = ctor.newInstance(c, cp);
92      } catch (InvocationTargetException ite) {
93        Throwable target = ite.getTargetException();
94        throw new RuntimeException("Failed construction of RegionServer: " +
95          hrsc.toString() + ((target.getCause() != null)?
96            target.getCause().getMessage(): ""), target);
97      } catch (Exception e) {
98        IOException ioe = new IOException();
99        ioe.initCause(e);
100       throw ioe;
101     }
102     return new JVMClusterUtil.RegionServerThread(server, index);
103   }
104 
105 
106   /**
107    * Datastructure to hold Master Thread and Master instance
108    */
109   public static class MasterThread extends Thread {
110     private final HMaster master;
111 
112     public MasterThread(final HMaster m, final int index) {
113       super(m, "M:" + index + ";" + m.getServerName().toShortString());
114       this.master = m;
115     }
116 
117     /** @return the master */
118     public HMaster getMaster() {
119       return this.master;
120     }
121   }
122 
123   /**
124    * Creates a {@link MasterThread}.
125    * Call 'start' on the returned thread to make it run.
126    * @param c Configuration to use.
127    * @param cp consensus provider to use
128    * @param hmc Class to create.
129    * @param index Used distinguishing the object returned.
130    * @throws IOException
131    * @return Master added.
132    */
133   public static JVMClusterUtil.MasterThread createMasterThread(
134       final Configuration c, CoordinatedStateManager cp, final Class<? extends HMaster> hmc,
135       final int index)
136   throws IOException {
137     HMaster server;
138     try {
139       server = hmc.getConstructor(Configuration.class, CoordinatedStateManager.class).
140         newInstance(c, cp);
141     } catch (InvocationTargetException ite) {
142       Throwable target = ite.getTargetException();
143       throw new RuntimeException("Failed construction of Master: " +
144         hmc.toString() + ((target.getCause() != null)?
145           target.getCause().getMessage(): ""), target);
146     } catch (Exception e) {
147       IOException ioe = new IOException();
148       ioe.initCause(e);
149       throw ioe;
150     }
151     return new JVMClusterUtil.MasterThread(server, index);
152   }
153 
154   private static JVMClusterUtil.MasterThread findActiveMaster(
155     List<JVMClusterUtil.MasterThread> masters) {
156     for (JVMClusterUtil.MasterThread t : masters) {
157       if (t.master.isActiveMaster()) {
158         return t;
159       }
160     }
161 
162     return null;
163   }
164 
165   /**
166    * Start the cluster.  Waits until there is a primary master initialized
167    * and returns its address.
168    * @param masters
169    * @param regionservers
170    * @return Address to use contacting primary master.
171    */
172   public static String startup(final List<JVMClusterUtil.MasterThread> masters,
173       final List<JVMClusterUtil.RegionServerThread> regionservers) throws IOException {
174 
175     Configuration configuration = null;
176
177     if (masters == null || masters.isEmpty()) {
178       return null;
179     }
180
181     for (JVMClusterUtil.MasterThread t : masters) {
182       configuration = t.getMaster().getConfiguration();
183       t.start();
184     }
185
186     // Wait for an active master
187     //  having an active master before starting the region threads allows
188     //  then to succeed on their connection to master
189     long startTime = System.currentTimeMillis();
190     while (findActiveMaster(masters) == null) {
191       try {
192         Thread.sleep(100);
193       } catch (InterruptedException e) {
194         throw (InterruptedIOException)new InterruptedIOException().initCause(e);
195       }
196       int startTimeout = configuration != null ? Integer.parseInt(
197         configuration.get("hbase.master.start.timeout.localHBaseCluster", "30000")) : 30000;
198       if (System.currentTimeMillis() > startTime + startTimeout) {
199         throw new RuntimeException(String.format("Master not active after %s seconds", startTimeout));
200       }
201     }
202
203     if (regionservers != null) {
204       for (JVMClusterUtil.RegionServerThread t: regionservers) {
205         t.start();
206       }
207     }
208
209     // Wait for an active master to be initialized (implies being master)
210     //  with this, when we return the cluster is complete
211     startTime = System.currentTimeMillis();
212     final int maxwait = 200000;
213     while (true) {
214       JVMClusterUtil.MasterThread t = findActiveMaster(masters);
215       if (t != null && t.master.isInitialized()) {
216         return t.master.getServerName().toString();
217       }
218       // REMOVE
219       if (System.currentTimeMillis() > startTime + 10000) {
220         try {
221           Thread.sleep(1000);
222         } catch (InterruptedException e) {
223           throw (InterruptedIOException)new InterruptedIOException().initCause(e);
224         }
225       }
226       if (System.currentTimeMillis() > startTime + maxwait) {
227         String msg = "Master not initialized after " + maxwait + "ms seconds";
228         Threads.printThreadInfo(System.out,
229           "Thread dump because: " + msg);
230         throw new RuntimeException(msg);
231       }
232       try {
233         Thread.sleep(100);
234       } catch (InterruptedException e) {
235         throw (InterruptedIOException)new InterruptedIOException().initCause(e);
236       }
237     }
238   }
239
240   /**
241    * @param masters
242    * @param regionservers
243    */
244   public static void shutdown(final List<MasterThread> masters,
245       final List<RegionServerThread> regionservers) {
246     LOG.debug("Shutting down HBase Cluster");
247     if (masters != null) {
248       // Do backups first.
249       JVMClusterUtil.MasterThread activeMaster = null;
250       for (JVMClusterUtil.MasterThread t : masters) {
251         if (!t.master.isActiveMaster()) {
252           t.master.stopMaster();
253         } else {
254           activeMaster = t;
255         }
256       }
257       // Do active after.
258       if (activeMaster != null)
259         activeMaster.master.shutdown();
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");
300             t.interrupt();
301           }
302         }
303       }
304     }
305
306     if (masters != null) {
307       for (JVMClusterUtil.MasterThread t : masters) {
308         while (t.master.isAlive() && !wasInterrupted) {
309           try {
310             // The below has been replaced to debug sometime hangs on end of
311             // tests.
312             // this.master.join():
313             Threads.threadDumpingIsAlive(t.master.getThread());
314           } catch(InterruptedException e) {
315             LOG.info("Got InterruptedException on shutdown - " +
316                 "not waiting anymore on master ends", e);
317             wasInterrupted = true;
318           }
319         }
320       }
321     }
322     LOG.info("Shutdown of " +
323       ((masters != null) ? masters.size() : "0") + " master(s) and " +
324       ((regionservers != null) ? regionservers.size() : "0") +
325       " regionserver(s) " + (wasInterrupted ? "interrupted" : "complete"));
326
327     if (wasInterrupted){
328       Thread.currentThread().interrupt();
329     }
330   }
331 }