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.io.PrintWriter;
24  import java.lang.reflect.Constructor;
25  import java.lang.reflect.InvocationTargetException;
26  import java.util.List;
27  
28  import org.apache.commons.logging.Log;
29  import org.apache.commons.logging.LogFactory;
30  import org.apache.hadoop.classification.InterfaceAudience;
31  import org.apache.hadoop.conf.Configuration;
32  import org.apache.hadoop.fs.FileSystem;
33  import org.apache.hadoop.hbase.master.HMaster;
34  import org.apache.hadoop.hbase.regionserver.HRegionServer;
35  import org.apache.hadoop.hbase.regionserver.ShutdownHook;
36  import org.apache.hadoop.util.ReflectionUtils;
37  
38  /**
39   * Utility used running a cluster all in the one JVM.
40   */
41  @InterfaceAudience.Private
42  public class JVMClusterUtil {
43    private static final Log LOG = LogFactory.getLog(JVMClusterUtil.class);
44  
45    /**
46     * Datastructure to hold RegionServer Thread and RegionServer instance
47     */
48    public static class RegionServerThread extends Thread {
49      private final HRegionServer regionServer;
50  
51      public RegionServerThread(final HRegionServer r, final int index) {
52        super(r, "RS:" + index + ";" + r.getServerName().toShortString());
53        this.regionServer = r;
54      }
55  
56      /** @return the region server */
57      public HRegionServer getRegionServer() {
58        return this.regionServer;
59      }
60  
61      /**
62       * Block until the region server has come online, indicating it is ready
63       * to be used.
64       */
65      public void waitForServerOnline() {
66        // The server is marked online after the init method completes inside of
67        // the HRS#run method.  HRS#init can fail for whatever region.  In those
68        // cases, we'll jump out of the run without setting online flag.  Check
69        // stopRequested so we don't wait here a flag that will never be flipped.
70        regionServer.waitForServerOnline();
71      }
72    }
73  
74    /**
75     * Creates a {@link RegionServerThread}.
76     * Call 'start' on the returned thread to make it run.
77     * @param c Configuration to use.
78     * @param hrsc Class to create.
79     * @param index Used distinguishing the object returned.
80     * @throws IOException
81     * @return Region server added.
82     */
83    public static JVMClusterUtil.RegionServerThread createRegionServerThread(
84        final Configuration c, final Class<? extends HRegionServer> hrsc,
85        final int index)
86    throws IOException {
87      HRegionServer server;
88      try {
89        Constructor<? extends HRegionServer> ctor = hrsc.getConstructor(Configuration.class);
90        ctor.setAccessible(true);
91        server = ctor.newInstance(c);
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 hmc Class to create.
128    * @param index Used distinguishing the object returned.
129    * @throws IOException
130    * @return Master added.
131    */
132   public static JVMClusterUtil.MasterThread createMasterThread(
133       final Configuration c, final Class<? extends HMaster> hmc,
134       final int index)
135   throws IOException {
136     HMaster server;
137     try {
138       server = hmc.getConstructor(Configuration.class).newInstance(c);
139     } catch (InvocationTargetException ite) {
140       Throwable target = ite.getTargetException();
141       throw new RuntimeException("Failed construction of Master: " +
142         hmc.toString() + ((target.getCause() != null)?
143           target.getCause().getMessage(): ""), target);
144     } catch (Exception e) {
145       IOException ioe = new IOException();
146       ioe.initCause(e);
147       throw ioe;
148     }
149     return new JVMClusterUtil.MasterThread(server, index);
150   }
151 
152   private static JVMClusterUtil.MasterThread findActiveMaster(
153     List<JVMClusterUtil.MasterThread> masters) {
154     for (JVMClusterUtil.MasterThread t : masters) {
155       if (t.master.isActiveMaster()) {
156         return t;
157       }
158     }
159 
160     return null;
161   }
162 
163   /**
164    * Start the cluster.  Waits until there is a primary master initialized
165    * and returns its address.
166    * @param masters
167    * @param regionservers
168    * @return Address to use contacting primary master.
169    */
170   public static String startup(final List<JVMClusterUtil.MasterThread> masters,
171       final List<JVMClusterUtil.RegionServerThread> regionservers) throws IOException {
172 
173     if (masters == null || masters.isEmpty()) {
174       return null;
175     }
176 
177     for (JVMClusterUtil.MasterThread t : masters) {
178       t.start();
179     }
180 
181     // Wait for an active master
182     //  having an active master before starting the region threads allows
183     //  then to succeed on their connection to master
184     long startTime = System.currentTimeMillis();
185     while (findActiveMaster(masters) == null) {
186       try {
187         Thread.sleep(100);
188       } catch (InterruptedException e) {
189         throw (InterruptedIOException)new InterruptedIOException().initCause(e);
190       }
191       if (System.currentTimeMillis() > startTime + 30000) {
192         throw new RuntimeException("Master not active after 30 seconds");
193       }
194     }
195 
196     if (regionservers != null) {
197       for (JVMClusterUtil.RegionServerThread t: regionservers) {
198         HRegionServer hrs = t.getRegionServer();
199         ShutdownHook.install(hrs.getConfiguration(), FileSystem.get(hrs
200                 .getConfiguration()), hrs, t);
201         t.start();
202       }
203     }
204 
205     // Wait for an active master to be initialized (implies being master)
206     //  with this, when we return the cluster is complete
207     startTime = System.currentTimeMillis();
208     final int maxwait = 200000;
209     while (true) {
210       JVMClusterUtil.MasterThread t = findActiveMaster(masters);
211       if (t != null && t.master.isInitialized()) {
212         return t.master.getServerName().toString();
213       }
214       // REMOVE
215       if (System.currentTimeMillis() > startTime + 10000) {
216         try {
217           Thread.sleep(1000);
218         } catch (InterruptedException e) {
219           throw (InterruptedIOException)new InterruptedIOException().initCause(e);
220         }
221       }
222       if (System.currentTimeMillis() > startTime + maxwait) {
223         String msg = "Master not initialized after " + maxwait + "ms seconds";
224         ReflectionUtils.printThreadInfo(new PrintWriter(System.out),
225           "Thread dump because: " + msg);
226         throw new RuntimeException(msg);
227       }
228       try {
229         Thread.sleep(100);
230       } catch (InterruptedException e) {
231         throw (InterruptedIOException)new InterruptedIOException().initCause(e);
232       }
233     }
234   }
235 
236   /**
237    * @param masters
238    * @param regionservers
239    */
240   public static void shutdown(final List<MasterThread> masters,
241       final List<RegionServerThread> regionservers) {
242     LOG.debug("Shutting down HBase Cluster");
243     if (masters != null) {
244       // Do backups first.
245       JVMClusterUtil.MasterThread activeMaster = null;
246       for (JVMClusterUtil.MasterThread t : masters) {
247         if (!t.master.isActiveMaster()) {
248           t.master.stopMaster();
249         } else {
250           activeMaster = t;
251         }
252       }
253       // Do active after.
254       if (activeMaster != null)
255         activeMaster.master.shutdown();
256 
257     }
258     boolean wasInterrupted = false;
259     final long maxTime = System.currentTimeMillis() + 30 * 1000;
260     if (regionservers != null) {
261       // first try nicely.
262       for (RegionServerThread t : regionservers) {
263         t.getRegionServer().stop("Shutdown requested");
264       }
265       for (RegionServerThread t : regionservers) {
266         long now = System.currentTimeMillis();
267         if (t.isAlive() && !wasInterrupted && now < maxTime) {
268           try {
269             t.join(maxTime - now);
270           } catch (InterruptedException e) {
271             LOG.info("Got InterruptedException on shutdown - " +
272                 "not waiting anymore on region server ends", e);
273             wasInterrupted = true; // someone wants us to speed up.
274           }
275         }
276       }
277 
278       // Let's try to interrupt the remaining threads if any.
279       for (int i = 0; i < 100; ++i) {
280         boolean atLeastOneLiveServer = false;
281         for (RegionServerThread t : regionservers) {
282           if (t.isAlive()) {
283             atLeastOneLiveServer = true;
284             try {
285               LOG.warn("RegionServerThreads remaining, give one more chance before interrupting");
286               t.join(1000);
287             } catch (InterruptedException e) {
288               wasInterrupted = true;
289             }
290           }
291         }
292         if (!atLeastOneLiveServer) break;
293         for (RegionServerThread t : regionservers) {
294           if (t.isAlive()) {
295             LOG.warn("RegionServerThreads taking too long to stop, interrupting");
296             t.interrupt();
297           }
298         }
299       }
300     }
301 
302     if (masters != null) {
303       for (JVMClusterUtil.MasterThread t : masters) {
304         while (t.master.isAlive() && !wasInterrupted) {
305           try {
306             // The below has been replaced to debug sometime hangs on end of
307             // tests.
308             // this.master.join():
309             Threads.threadDumpingIsAlive(t.master.getThread());
310           } catch(InterruptedException e) {
311             LOG.info("Got InterruptedException on shutdown - " +
312                 "not waiting anymore on master ends", e);
313             wasInterrupted = true;
314           }
315         }
316       }
317     }
318     LOG.info("Shutdown of " +
319       ((masters != null) ? masters.size() : "0") + " master(s) and " +
320       ((regionservers != null) ? regionservers.size() : "0") +
321       " regionserver(s) " + (wasInterrupted ? "interrupted" : "complete"));
322 
323     if (wasInterrupted){
324       Thread.currentThread().interrupt();
325     }
326   }
327 }