HADOOP-1523 'Hung region servers waiting on write locks'

On shutdown, region servers and masters were just cancelling leases
without letting 'lease expired' code run -- code to clean up
outstanding locks in region server.  Outstanding read locks were
getting in the way of region server getting necessary write locks
needed for the shutdown process.  Also, cleaned up messaging around
shutdown so its clean -- no timeout messages as region servers try
to talk to a master that has already shutdown -- even when region
servers take their time going down.
M src/contrib/hbase/conf/hbase-default.xml
    Make region server timeout 30 seconds instead of 3 minutes.
    Clients retry anyways. Make so its likely region servers report
    in their shutdown message before their lease expires on master.
M src/contrib/hbase/src/java/org/apache/hadoop/hbase/Leases.java
    (closeAfterLeasesExpire): Added.
* src/contrib/hbase/src/java/org/apache/hadoop/hbase/HRegionServer.java
    Added comments.
    (stop): Converted from public to default access (master shuts
    down regionservers).
    (run): Use leases.closeAfterLeasesExpire instead of leases.close.
    Changed log of main thread exit from DEBUG to INFO.
* src/contrib/hbase/src/java/org/apache/hadoop/hbase/HMaster.java
    (letRegionsServersShutdown): Add better explaination of shutdown
    process to method doc.  Changed timeout waits from
    hbase.regionserver.msginterval to threadWakeFrequency.
    (regionServerReport): If closing, we used to immediately respond
    to region server with a MSG_REGIONSERVER_STOP.  This meant that
    we avoided handling of the region servers MSG_REPORT_EXITING sent
    on shutdown so region servers had no chance to cancel their lease
    in the master.  Reordered.  Moved sending of MSG_REGIONSERVER_STOP
    to after handling of MSG_REPORT_EXITING.  Also, in handling of
    MSG_REGIONSERER_STOP removed cancelling of leases.  Let leases
    expire normally (or get cancelled when the region server comes in
    with MSG_RPORT_EXITING).
* src/contrib/hbase/src/java/org/apache/hadoop/hbase/HMsg.java
    (MSG_REGIONSERVER_STOP_IN_ARRAY): Added.


git-svn-id: https://svn.apache.org/repos/asf/lucene/hadoop/trunk@552376 13f79535-47bb-0310-9956-ffa450edef68

src/contrib/hbase/CHANGES.txt

@@ -44,3 +44,4 @@ Trunk (unreleased changes)
      what is failing.
  26. HADOOP-1543 [hbase] Add HClient.tableExists
  27. HADOOP-1519 [hbase] map/reduce interface for HBase
+ 28. HADOOP-1523 Hung region server waiting on write locks 

src/contrib/hbase/conf/hbase-default.xml

@@ -53,13 +53,16 @@
     <name>hbase.master.lease.period</name>
     <value>30000</value>
     <description>HMaster server lease period in milliseconds. Default is
-    30 seconds.</description>
+    30 seconds.  Region servers must report in within this period else
+    they are considered dead.  On loaded cluster, may need to up this
+    period.</description>
   </property>
   <property>
     <name>hbase.regionserver.lease.period</name>
-    <value>180000</value>
+    <value>30000</value>
     <description>HRegion server lease period in milliseconds. Default is
-    180 seconds.</description>
+    30 seconds. Clients must report in within this period else they are
+    considered dead.</description>
   </property>
   <property>
     <name>hbase.server.thread.wakefrequency</name>

src/contrib/hbase/src/java/org/apache/hadoop/hbase/HMaster.java

@@ -682,7 +682,7 @@ public class HMaster implements HConstants, HMasterInterface,
     }
 
     // Main processing loop
-    for(PendingOperation op = null; !closed; ) {
+    for (PendingOperation op = null; !closed; ) {
       synchronized(msgQueue) {
         while(msgQueue.size() == 0 && !closed) {
           try {
@@ -756,26 +756,22 @@ public class HMaster implements HConstants, HMasterInterface,
     LOG.info("HMaster main thread exiting");
   }
   
-  /**
-   * Wait on regionservers to report in.  Then, they notice the HMaster
-   * is going down and will shut themselves down.
+  /*
+   * Wait on regionservers to report in
+   * with {@link #regionServerReport(HServerInfo, HMsg[])} so they get notice
+   * the master is going down.  Waits until all region servers come back with
+   * a MSG_REGIONSERVER_STOP which will cancel their lease or until leases held
+   * by remote region servers have expired.
    */
   private void letRegionServersShutdown() {
-    long regionServerMsgInterval =
-      conf.getLong("hbase.regionserver.msginterval", 15 * 1000);
-    // Wait for 3 * hbase.regionserver.msginterval intervals or until all
-    // regionservers report in as closed.
-    long endTime = System.currentTimeMillis() + (regionServerMsgInterval * 3);
-    while (endTime > System.currentTimeMillis() &&
-        this.serversToServerInfo.size() > 0) {
-      if (LOG.isDebugEnabled()) {
-        LOG.debug("Waiting on regionservers: "
-            + this.serversToServerInfo.values());
-      }
+    while (this.serversToServerInfo.size() > 0) {
+      LOG.info("Waiting on following regionserver(s) to go down (or " +
+        "region server lease expiration, whichever happens first): " +
+        this.serversToServerInfo.values());
       try {
         Thread.sleep(threadWakeFrequency);
       } catch (InterruptedException e) {
-        // Ignore interrupt.
+        // continue
       }
     }
   }
@@ -849,56 +845,50 @@ public class HMaster implements HConstants, HMasterInterface,
   throws IOException {
     String s = serverInfo.getServerAddress().toString().trim();
     long serverLabel = getServerLabel(s);
-
-    if (closed) {
-      // Cancel the server's lease
-      cancelLease(s, serverLabel);
-      
-      // Tell server to shut down
-      HMsg returnMsgs[] = {new HMsg(HMsg.MSG_REGIONSERVER_STOP)};
-      return returnMsgs;
-    }
     
-    if(msgs.length > 0 && msgs[0].getMsg() == HMsg.MSG_REPORT_EXITING) {
-      // HRegionServer is shutting down.
-      
-      // Cancel the server's lease
+    if (msgs.length > 0 && msgs[0].getMsg() == HMsg.MSG_REPORT_EXITING) {
+      // HRegionServer is shutting down. Cancel the server's lease.
+      LOG.debug("Region server " + s + ": MSG_REPORT_EXITING");
       cancelLease(s, serverLabel);
       
-      // Get all the regions the server was serving reassigned
-      
-      for(int i = 1; i < msgs.length; i++) {
-        HRegionInfo info = msgs[i].getRegionInfo();
-        if(info.tableDesc.getName().equals(ROOT_TABLE_NAME)) {
-          rootRegionLocation = null;
-          
-        } else if(info.tableDesc.getName().equals(META_TABLE_NAME)) {
-          allMetaRegionsScanned = false;
+      // Get all the regions the server was serving reassigned (if we
+      // are not shutting down).
+      if (!closed) {
+        for (int i = 1; i < msgs.length; i++) {
+          HRegionInfo info = msgs[i].getRegionInfo();
+          if (info.tableDesc.getName().equals(ROOT_TABLE_NAME)) {
+            rootRegionLocation = null;
+          } else if (info.tableDesc.getName().equals(META_TABLE_NAME)) {
+            allMetaRegionsScanned = false;
+          }
+          unassignedRegions.put(info.regionName, info);
+          assignAttempts.put(info.regionName, Long.valueOf(0L));
         }
-        unassignedRegions.put(info.regionName, info);
-        assignAttempts.put(info.regionName, Long.valueOf(0L));
       }
       
       // We don't need to return anything to the server because it isn't
       // going to do any more work.
       return new HMsg[0];
     }
+    
+    if (closed) {
+      // Tell server to shut down if we are shutting down.  This should
+      // happen after check of MSG_REPORT_EXITING above, since region server
+      // will send us one of these messages after it gets MSG_REGIONSERVER_STOP
+      return HMsg.MSG_REGIONSERVER_STOP_IN_ARRAY;
+    }
 
     HServerInfo storedInfo = serversToServerInfo.get(s);
-
     if(storedInfo == null) {
-
       if(LOG.isDebugEnabled()) {
         LOG.debug("received server report from unknown server: " + s);
       }
 
       // The HBaseMaster may have been restarted.
       // Tell the RegionServer to start over and call regionServerStartup()
-
       HMsg returnMsgs[] = new HMsg[1];
       returnMsgs[0] = new HMsg(HMsg.MSG_CALL_SERVER_STARTUP);
       return returnMsgs;
-
     } else if(storedInfo.getStartCode() != serverInfo.getStartCode()) {
 
       // This state is reachable if:
@@ -914,11 +904,7 @@ public class HMaster implements HConstants, HMasterInterface,
         LOG.debug("region server race condition detected: " + s);
       }
 
-      HMsg returnMsgs[] = {
-          new HMsg(HMsg.MSG_REGIONSERVER_STOP)
-      };
-      return returnMsgs;
-
+      return HMsg.MSG_REGIONSERVER_STOP_IN_ARRAY;
     } else {
 
       // All's well.  Renew the server's lease.

src/contrib/hbase/src/java/org/apache/hadoop/hbase/HMsg.java

@@ -39,6 +39,9 @@ public class HMsg implements Writable {
   /** Master tells region server to stop */
   public static final byte MSG_REGIONSERVER_STOP = 5;
   
+  public static final HMsg [] MSG_REGIONSERVER_STOP_IN_ARRAY = 
+    {new HMsg(HMsg.MSG_REGIONSERVER_STOP)};
+  
   /** Stop serving the specified region and don't report back that it's closed */
   public static final byte MSG_REGION_CLOSE_WITHOUT_REPORT = 6;
 

src/contrib/hbase/src/java/org/apache/hadoop/hbase/HRegionServer.java

@@ -61,8 +61,14 @@ public class HRegionServer implements HConstants, HRegionInterface, Runnable {
 
   static final Log LOG = LogFactory.getLog(HRegionServer.class);
   
+  // Set when a report to the master comes back with a message asking us to
+  // shutdown.  Also set by call to stop when debugging or running unit tests
+  // of HRegionServer in isolation.
   protected volatile boolean stopRequested;
+  
+  // Go down hard.  Used debugging and in unit tests.
   protected volatile boolean abortRequested;
+  
   private final Path rootDir;
   protected final HServerInfo serverInfo;
   protected final Configuration conf;
@@ -468,8 +474,9 @@ public class HRegionServer implements HConstants, HRegionInterface, Runnable {
   /**
    * Sets a flag that will cause all the HRegionServer threads to shut down
    * in an orderly fashion.
+   * <p>FOR DEBUGGING ONLY
    */
-  public synchronized void stop() {
+  synchronized void stop() {
     stopRequested = true;
     notifyAll();                        // Wakes run() if it is sleeping
   }
@@ -477,8 +484,7 @@ public class HRegionServer implements HConstants, HRegionInterface, Runnable {
   /**
    * Cause the server to exit without closing the regions it is serving, the
    * log it is using and without notifying the master.
-   * 
-   * FOR DEBUGGING ONLY
+   * <p>FOR DEBUGGING ONLY
    */
   synchronized void abort() {
     abortRequested = true;
@@ -560,16 +566,13 @@ public class HRegionServer implements HConstants, HRegionInterface, Runnable {
         if (LOG.isDebugEnabled()) {
           LOG.debug("Telling master we are up");
         }
-        
         hbaseMaster.regionServerStartup(serverInfo);
-        
         if (LOG.isDebugEnabled()) {
           LOG.debug("Done telling master we are up");
         }
       } catch(IOException e) {
         waitTime = stopRequested ? 0
             : msgInterval - (System.currentTimeMillis() - lastMsg);
-        
         if(waitTime > 0) {
           synchronized (this) {
             try {
@@ -585,7 +588,6 @@ public class HRegionServer implements HConstants, HRegionInterface, Runnable {
       // Now ask master what it wants us to do and tell it what we have done.
       while (!stopRequested) {
         if ((System.currentTimeMillis() - lastMsg) >= msgInterval) {
-
           HMsg outboundArray[] = null;
           synchronized(outboundMsgs) {
             outboundArray = outboundMsgs.toArray(new HMsg[outboundMsgs.size()]);
@@ -595,14 +597,12 @@ public class HRegionServer implements HConstants, HRegionInterface, Runnable {
           try {
             HMsg msgs[] = hbaseMaster.regionServerReport(serverInfo, outboundArray);
             lastMsg = System.currentTimeMillis();
-
             // Queue up the HMaster's instruction stream for processing
-
             synchronized(toDo) {
               boolean restart = false;
               for(int i = 0; i < msgs.length && !stopRequested && !restart; i++) {
                 switch(msgs[i].getMsg()) {
-                
+              
                 case HMsg.MSG_CALL_SERVER_STARTUP:
                   if (LOG.isDebugEnabled()) {
                     LOG.debug("Got call server startup message");
@@ -657,9 +657,9 @@ public class HRegionServer implements HConstants, HRegionInterface, Runnable {
         }
       }
     }
+    leases.closeAfterLeasesExpire();
     this.worker.stop();
     this.server.stop();
-    leases.close();
     
     // Send interrupts to wake up threads if sleeping so they notice shutdown.
 
@@ -675,50 +675,43 @@ public class HRegionServer implements HConstants, HRegionInterface, Runnable {
       this.splitOrCompactCheckerThread.interrupt();
     }
 
-    if(abortRequested) {
+    if (abortRequested) {
       try {
         log.rollWriter();
-        
       } catch(IOException e) {
         LOG.warn(e);
       }
-      LOG.info("aborting server at: " + serverInfo.getServerAddress().toString());
-      
+      LOG.info("aborting server at: " +
+        serverInfo.getServerAddress().toString());
     } else {
       Vector<HRegion> closedRegions = closeAllRegions();
       try {
         log.closeAndDelete();
-
       } catch(IOException e) {
         LOG.error(e);
       }
       try {
         HMsg[] exitMsg = new HMsg[closedRegions.size() + 1];
         exitMsg[0] = new HMsg(HMsg.MSG_REPORT_EXITING);
-
         // Tell the master what regions we are/were serving
-
         int i = 1;
         for(HRegion region: closedRegions) {
-          exitMsg[i++] = new HMsg(HMsg.MSG_REPORT_CLOSE, region.getRegionInfo());
+          exitMsg[i++] = new HMsg(HMsg.MSG_REPORT_CLOSE,
+            region.getRegionInfo());
         }
 
         LOG.info("telling master that region server is shutting down at: "
             + serverInfo.getServerAddress().toString());
-        
         hbaseMaster.regionServerReport(serverInfo, exitMsg);
-
       } catch(IOException e) {
         LOG.warn(e);
       }
-      LOG.info("stopping server at: " + serverInfo.getServerAddress().toString());
+      LOG.info("stopping server at: " +
+        serverInfo.getServerAddress().toString());
     }
 
-    join();
-      
-    if(LOG.isDebugEnabled()) {
-      LOG.debug("main thread exiting");
-    }
+    join(); 
+    LOG.info("main thread exiting");
   }
 
   /** Add to the outbound message buffer */
@@ -1024,7 +1017,6 @@ public class HRegionServer implements HConstants, HRegionInterface, Runnable {
     public void leaseExpired() {
       try {
         localRegion.abort(localLockId);
-        
       } catch(IOException iex) {
         LOG.error(iex);
       }

src/contrib/hbase/src/java/org/apache/hadoop/hbase/Leases.java

@@ -62,6 +62,29 @@ public class Leases {
     leaseMonitorThread.start();
   }
 
+  /**
+   * Shuts down this lease instance when all outstanding leases expire.
+   * Like {@link #close()} but rather than violently end all leases, waits
+   * first on extant leases to finish.  Use this method if the lease holders
+   * could loose data, leak locks, etc.  Presumes client has shutdown
+   * allocation of new leases.
+   */
+  public void closeAfterLeasesExpire() {
+    synchronized(this.leases) {
+      while (this.leases.size() > 0) {
+        LOG.info(Integer.toString(leases.size()) + " lease(s) " +
+          "outstanding. Waiting for them to expire.");
+        try {
+          this.leases.wait(this.leaseCheckFrequency);
+        } catch (InterruptedException e) {
+          // continue
+        }
+      }
+    }
+    // Now call close since no leases outstanding.
+    close();
+  }
+  
   /**
    * Shut down this Leases instance.  All pending leases will be destroyed, 
    * without any cancellation calls.