View Javadoc

1   /**
2    * Licensed to the Apache Software Foundation (ASF) under one
3    * or more contributor license agreements.  See the NOTICE file
4    * distributed with this work for additional information
5    * regarding copyright ownership.  The ASF licenses this file
6    * to you under the Apache License, Version 2.0 (the
7    * "License"); you may not use this file except in compliance
8    * with the License.  You may obtain a copy of the License at
9    *
10   *     http://www.apache.org/licenses/LICENSE-2.0
11   *
12   * Unless required by applicable law or agreed to in writing, software
13   * distributed under the License is distributed on an "AS IS" BASIS,
14   * WITHOUTKey WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15   * See the License for the specific language governing permissions and
16   * limitations under the License.
17   */
18  
19  package org.apache.hadoop.hbase.master.procedure;
20  
21  import java.io.IOException;
22  import java.util.ArrayDeque;
23  import java.util.concurrent.locks.Condition;
24  import java.util.concurrent.locks.ReentrantLock;
25  
26  import org.apache.commons.logging.Log;
27  import org.apache.commons.logging.LogFactory;
28  import org.apache.hadoop.conf.Configuration;
29  import org.apache.hadoop.hbase.ServerName;
30  import org.apache.hadoop.hbase.TableExistsException;
31  import org.apache.hadoop.hbase.TableName;
32  import org.apache.hadoop.hbase.TableNotFoundException;
33  import org.apache.hadoop.hbase.classification.InterfaceAudience;
34  import org.apache.hadoop.hbase.classification.InterfaceStability;
35  import org.apache.hadoop.hbase.master.TableLockManager;
36  import org.apache.hadoop.hbase.master.TableLockManager.TableLock;
37  import org.apache.hadoop.hbase.master.procedure.TableProcedureInterface.TableOperationType;
38  import org.apache.hadoop.hbase.procedure2.Procedure;
39  import org.apache.hadoop.hbase.procedure2.ProcedureRunnableSet;
40  
41  /**
42   * ProcedureRunnableSet for the Master Procedures.
43   * This RunnableSet tries to provide to the ProcedureExecutor procedures
44   * that can be executed without having to wait on a lock.
45   * Most of the master operations can be executed concurrently, if they
46   * are operating on different tables (e.g. two create table can be performed
47   * at the same, time assuming table A and table B) or against two different servers; say
48   * two servers that crashed at about the same time.
49   *
50   * <p>Each procedure should implement an interface providing information for this queue.
51   * for example table related procedures should implement TableProcedureInterface.
52   * each procedure will be pushed in its own queue, and based on the operation type
53   * we may take smarter decision. e.g. we can abort all the operations preceding
54   * a delete table, or similar.
55   */
56  @InterfaceAudience.Private
57  @InterfaceStability.Evolving
58  public class MasterProcedureScheduler implements ProcedureRunnableSet {
59    private static final Log LOG = LogFactory.getLog(MasterProcedureScheduler.class);
60  
61    private final TableLockManager lockManager;
62    private final ReentrantLock schedLock = new ReentrantLock();
63    private final Condition schedWaitCond = schedLock.newCondition();
64  
65    private final FairQueue<ServerName> serverRunQueue = new FairQueue<ServerName>();
66    private final FairQueue<TableName> tableRunQueue = new FairQueue<TableName>();
67    private int queueSize = 0;
68  
69    private final Object[] serverBuckets = new Object[128];
70    private Queue<String> namespaceMap = null;
71    private Queue<TableName> tableMap = null;
72  
73    private final int metaTablePriority;
74    private final int userTablePriority;
75    private final int sysTablePriority;
76  
77    // TODO: metrics
78    private long pollCalls = 0;
79    private long nullPollCalls = 0;
80  
81    public MasterProcedureScheduler(final Configuration conf, final TableLockManager lockManager) {
82      this.lockManager = lockManager;
83  
84      // TODO: should this be part of the HTD?
85      metaTablePriority = conf.getInt("hbase.master.procedure.queue.meta.table.priority", 3);
86      sysTablePriority = conf.getInt("hbase.master.procedure.queue.system.table.priority", 2);
87      userTablePriority = conf.getInt("hbase.master.procedure.queue.user.table.priority", 1);
88    }
89  
90    @Override
91    public void addFront(Procedure proc) {
92      doAdd(proc, true);
93    }
94  
95    @Override
96    public void addBack(Procedure proc) {
97      doAdd(proc, false);
98    }
99  
100   @Override
101   public void yield(final Procedure proc) {
102     doAdd(proc, isTableProcedure(proc));
103   }
104 
105   private void doAdd(final Procedure proc, final boolean addFront) {
106     schedLock.lock();
107     try {
108       if (isTableProcedure(proc)) {
109         doAdd(tableRunQueue, getTableQueue(getTableName(proc)), proc, addFront);
110       } else if (isServerProcedure(proc)) {
111         doAdd(serverRunQueue, getServerQueue(getServerName(proc)), proc, addFront);
112       } else {
113         // TODO: at the moment we only have Table and Server procedures
114         // if you are implementing a non-table/non-server procedure, you have two options: create
115         // a group for all the non-table/non-server procedures or try to find a key for your
116         // non-table/non-server procedures and implement something similar to the TableRunQueue.
117         throw new UnsupportedOperationException(
118           "RQs for non-table/non-server procedures are not implemented yet");
119       }
120       schedWaitCond.signal();
121     } finally {
122       schedLock.unlock();
123     }
124   }
125 
126   private <T extends Comparable<T>> void doAdd(final FairQueue<T> fairq,
127       final Queue<T> queue, final Procedure proc, final boolean addFront) {
128     queue.add(proc, addFront);
129     if (!(queue.isSuspended() || queue.hasExclusiveLock())) {
130       if (queue.size() == 1 && !IterableList.isLinked(queue)) {
131         fairq.add(queue);
132       }
133       queueSize++;
134     }
135   }
136 
137   @Override
138   public Procedure poll() {
139     return poll(-1);
140   }
141 
142   @edu.umd.cs.findbugs.annotations.SuppressWarnings("WA_AWAIT_NOT_IN_LOOP")
143   Procedure poll(long waitNsec) {
144     Procedure pollResult = null;
145     schedLock.lock();
146     try {
147       if (queueSize == 0) {
148         if (waitNsec < 0) {
149           schedWaitCond.await();
150         } else {
151           schedWaitCond.awaitNanos(waitNsec);
152         }
153         if (queueSize == 0) {
154           return null;
155         }
156       }
157 
158       // For now, let server handling have precedence over table handling; presumption is that it
159       // is more important handling crashed servers than it is running the
160       // enabling/disabling tables, etc.
161       pollResult = doPoll(serverRunQueue);
162       if (pollResult == null) {
163         pollResult = doPoll(tableRunQueue);
164       }
165 
166       // update metrics
167       pollCalls++;
168       nullPollCalls += (pollResult == null) ? 1 : 0;
169     } catch (InterruptedException e) {
170       Thread.currentThread().interrupt();
171     } finally {
172       schedLock.unlock();
173     }
174     return pollResult;
175   }
176 
177   private <T extends Comparable<T>> Procedure doPoll(final FairQueue<T> fairq) {
178     Queue<T> rq = fairq.poll();
179     if (rq == null || !rq.isAvailable()) {
180       return null;
181     }
182 
183     assert !rq.isSuspended() : "rq=" + rq + " is suspended";
184     Procedure pollResult = rq.poll();
185     this.queueSize--;
186     if (rq.isEmpty() || rq.requireExclusiveLock(pollResult)) {
187       removeFromRunQueue(fairq, rq);
188     }
189     return pollResult;
190   }
191 
192   @Override
193   public void clear() {
194     // NOTE: USED ONLY FOR TESTING
195     schedLock.lock();
196     try {
197       // Remove Servers
198       for (int i = 0; i < serverBuckets.length; ++i) {
199         clear((ServerQueue)serverBuckets[i], serverRunQueue);
200         serverBuckets[i] = null;
201       }
202 
203       // Remove Tables
204       clear(tableMap, tableRunQueue);
205       tableMap = null;
206 
207       assert queueSize == 0 : "expected queue size to be 0, got " + queueSize;
208     } finally {
209       schedLock.unlock();
210     }
211   }
212 
213   private <T extends Comparable<T>> void clear(Queue<T> treeMap, FairQueue<T> fairq) {
214     while (treeMap != null) {
215       Queue<T> node = AvlTree.getFirst(treeMap);
216       assert !node.isSuspended() : "can't clear suspended " + node.getKey();
217       treeMap = AvlTree.remove(treeMap, node.getKey());
218       removeFromRunQueue(fairq, node);
219     }
220   }
221 
222   @Override
223   public void signalAll() {
224     schedLock.lock();
225     try {
226       schedWaitCond.signalAll();
227     } finally {
228       schedLock.unlock();
229     }
230   }
231 
232   @Override
233   public int size() {
234     schedLock.lock();
235     try {
236       return queueSize;
237     } finally {
238       schedLock.unlock();
239     }
240   }
241 
242   @Override
243   public void completionCleanup(Procedure proc) {
244     if (proc instanceof TableProcedureInterface) {
245       TableProcedureInterface iProcTable = (TableProcedureInterface)proc;
246       boolean tableDeleted;
247       if (proc.hasException()) {
248         IOException procEx =  proc.getException().unwrapRemoteException();
249         if (iProcTable.getTableOperationType() == TableOperationType.CREATE) {
250           // create failed because the table already exist
251           tableDeleted = !(procEx instanceof TableExistsException);
252         } else {
253           // the operation failed because the table does not exist
254           tableDeleted = (procEx instanceof TableNotFoundException);
255         }
256       } else {
257         // the table was deleted
258         tableDeleted = (iProcTable.getTableOperationType() == TableOperationType.DELETE);
259       }
260       if (tableDeleted) {
261         markTableAsDeleted(iProcTable.getTableName());
262         return;
263       }
264     } else {
265       // No cleanup for ServerProcedureInterface types, yet.
266       return;
267     }
268   }
269 
270   private <T extends Comparable<T>> void addToRunQueue(FairQueue<T> fairq, Queue<T> queue) {
271     if (IterableList.isLinked(queue)) return;
272     if (!queue.isEmpty())  {
273       fairq.add(queue);
274       queueSize += queue.size();
275     }
276   }
277 
278   private <T extends Comparable<T>> void removeFromRunQueue(FairQueue<T> fairq, Queue<T> queue) {
279     if (!IterableList.isLinked(queue)) return;
280     fairq.remove(queue);
281     queueSize -= queue.size();
282   }
283 
284   // ============================================================================
285   //  TODO: Metrics
286   // ============================================================================
287   public long getPollCalls() {
288     return pollCalls;
289   }
290 
291   public long getNullPollCalls() {
292     return nullPollCalls;
293   }
294 
295   // ============================================================================
296   //  Event Helpers
297   // ============================================================================
298   public boolean waitEvent(ProcedureEvent event, Procedure procedure) {
299     return waitEvent(event, procedure, false);
300   }
301 
302   public boolean waitEvent(ProcedureEvent event, Procedure procedure, boolean suspendQueue) {
303     synchronized (event) {
304       if (event.isReady()) {
305         return false;
306       }
307 
308       // TODO: Suspend single procedure not implemented yet, fallback to suspending the queue
309       if (!suspendQueue) suspendQueue = true;
310 
311       if (isTableProcedure(procedure)) {
312         waitTableEvent(event, procedure, suspendQueue);
313       } else if (isServerProcedure(procedure)) {
314         waitServerEvent(event, procedure, suspendQueue);
315       } else {
316         // TODO: at the moment we only have Table and Server procedures
317         // if you are implementing a non-table/non-server procedure, you have two options: create
318         // a group for all the non-table/non-server procedures or try to find a key for your
319         // non-table/non-server procedures and implement something similar to the TableRunQueue.
320         throw new UnsupportedOperationException(
321           "RQs for non-table/non-server procedures are not implemented yet");
322       }
323     }
324     return true;
325   }
326 
327   private void waitTableEvent(ProcedureEvent event, Procedure procedure, boolean suspendQueue) {
328     final TableName tableName = getTableName(procedure);
329     final boolean isDebugEnabled = LOG.isDebugEnabled();
330 
331     schedLock.lock();
332     try {
333       TableQueue queue = getTableQueue(tableName);
334       if (queue.isSuspended()) return;
335 
336       // TODO: if !suspendQueue
337 
338       if (isDebugEnabled) {
339         LOG.debug("Suspend table queue " + tableName);
340       }
341       queue.setSuspended(true);
342       removeFromRunQueue(tableRunQueue, queue);
343       event.suspendTableQueue(queue);
344     } finally {
345       schedLock.unlock();
346     }
347   }
348 
349   private void waitServerEvent(ProcedureEvent event, Procedure procedure, boolean suspendQueue) {
350     final ServerName serverName = getServerName(procedure);
351     final boolean isDebugEnabled = LOG.isDebugEnabled();
352 
353     schedLock.lock();
354     try {
355       // TODO: This will change once we have the new AM
356       ServerQueue queue = getServerQueue(serverName);
357       if (queue.isSuspended()) return;
358 
359       // TODO: if !suspendQueue
360 
361       if (isDebugEnabled) {
362         LOG.debug("Suspend server queue " + serverName);
363       }
364       queue.setSuspended(true);
365       removeFromRunQueue(serverRunQueue, queue);
366       event.suspendServerQueue(queue);
367     } finally {
368       schedLock.unlock();
369     }
370   }
371 
372   public void suspend(ProcedureEvent event) {
373     final boolean isDebugEnabled = LOG.isDebugEnabled();
374     synchronized (event) {
375       event.setReady(false);
376       if (isDebugEnabled) {
377         LOG.debug("Suspend event " + event);
378       }
379     }
380   }
381 
382   public void wake(ProcedureEvent event) {
383     final boolean isDebugEnabled = LOG.isDebugEnabled();
384     synchronized (event) {
385       event.setReady(true);
386       if (isDebugEnabled) {
387         LOG.debug("Wake event " + event);
388       }
389 
390       schedLock.lock();
391       try {
392         while (event.hasWaitingTables()) {
393           Queue<TableName> queue = event.popWaitingTable();
394           addToRunQueue(tableRunQueue, queue);
395         }
396         // TODO: This will change once we have the new AM
397         while (event.hasWaitingServers()) {
398           Queue<ServerName> queue = event.popWaitingServer();
399           addToRunQueue(serverRunQueue, queue);
400         }
401 
402         if (queueSize > 1) {
403           schedWaitCond.signalAll();
404         } else if (queueSize > 0) {
405           schedWaitCond.signal();
406         }
407       } finally {
408         schedLock.unlock();
409       }
410     }
411   }
412 
413   public static class ProcedureEvent {
414     private final String description;
415 
416     private Queue<ServerName> waitingServers = null;
417     private Queue<TableName> waitingTables = null;
418     private boolean ready = false;
419 
420     public ProcedureEvent(String description) {
421       this.description = description;
422     }
423 
424     public synchronized boolean isReady() {
425       return ready;
426     }
427 
428     private synchronized void setReady(boolean isReady) {
429       this.ready = isReady;
430     }
431 
432     private void suspendTableQueue(Queue<TableName> queue) {
433       waitingTables = IterableList.append(waitingTables, queue);
434     }
435 
436     private void suspendServerQueue(Queue<ServerName> queue) {
437       waitingServers = IterableList.append(waitingServers, queue);
438     }
439 
440     private boolean hasWaitingTables() {
441       return waitingTables != null;
442     }
443 
444     private Queue<TableName> popWaitingTable() {
445       Queue<TableName> node = waitingTables;
446       waitingTables = IterableList.remove(waitingTables, node);
447       node.setSuspended(false);
448       return node;
449     }
450 
451     private boolean hasWaitingServers() {
452       return waitingServers != null;
453     }
454 
455     private Queue<ServerName> popWaitingServer() {
456       Queue<ServerName> node = waitingServers;
457       waitingServers = IterableList.remove(waitingServers, node);
458       node.setSuspended(false);
459       return node;
460     }
461 
462     @Override
463     public String toString() {
464       return String.format("ProcedureEvent(%s)", description);
465     }
466   }
467 
468   // ============================================================================
469   //  Table Queue Lookup Helpers
470   // ============================================================================
471   private TableQueue getTableQueueWithLock(TableName tableName) {
472     schedLock.lock();
473     try {
474       return getTableQueue(tableName);
475     } finally {
476       schedLock.unlock();
477     }
478   }
479 
480   private TableQueue getTableQueue(TableName tableName) {
481     Queue<TableName> node = AvlTree.get(tableMap, tableName);
482     if (node != null) return (TableQueue)node;
483 
484     NamespaceQueue nsQueue = getNamespaceQueue(tableName.getNamespaceAsString());
485     node = new TableQueue(tableName, nsQueue, getTablePriority(tableName));
486     tableMap = AvlTree.insert(tableMap, node);
487     return (TableQueue)node;
488   }
489 
490   private void removeTableQueue(TableName tableName) {
491     tableMap = AvlTree.remove(tableMap, tableName);
492   }
493 
494   private int getTablePriority(TableName tableName) {
495     if (tableName.equals(TableName.META_TABLE_NAME)) {
496       return metaTablePriority;
497     } else if (tableName.isSystemTable()) {
498       return sysTablePriority;
499     }
500     return userTablePriority;
501   }
502 
503   private static boolean isTableProcedure(Procedure proc) {
504     return proc instanceof TableProcedureInterface;
505   }
506 
507   private static TableName getTableName(Procedure proc) {
508     return ((TableProcedureInterface)proc).getTableName();
509   }
510 
511   // ============================================================================
512   //  Namespace Queue Lookup Helpers
513   // ============================================================================
514   private NamespaceQueue getNamespaceQueue(String namespace) {
515     Queue<String> node = AvlTree.get(namespaceMap, namespace);
516     if (node != null) return (NamespaceQueue)node;
517 
518     node = new NamespaceQueue(namespace);
519     namespaceMap = AvlTree.insert(namespaceMap, node);
520     return (NamespaceQueue)node;
521   }
522 
523   // ============================================================================
524   //  Server Queue Lookup Helpers
525   // ============================================================================
526   private ServerQueue getServerQueueWithLock(ServerName serverName) {
527     schedLock.lock();
528     try {
529       return getServerQueue(serverName);
530     } finally {
531       schedLock.unlock();
532     }
533   }
534 
535   private ServerQueue getServerQueue(ServerName serverName) {
536     int index = getBucketIndex(serverBuckets, serverName.hashCode());
537     Queue<ServerName> root = getTreeRoot(serverBuckets, index);
538     Queue<ServerName> node = AvlTree.get(root, serverName);
539     if (node != null) return (ServerQueue)node;
540 
541     node = new ServerQueue(serverName);
542     serverBuckets[index] = AvlTree.insert(root, node);
543     return (ServerQueue)node;
544   }
545 
546   private void removeServerQueue(ServerName serverName) {
547     int index = getBucketIndex(serverBuckets, serverName.hashCode());
548     serverBuckets[index] = AvlTree.remove((ServerQueue)serverBuckets[index], serverName);
549   }
550 
551   @SuppressWarnings("unchecked")
552   private static <T extends Comparable<T>> Queue<T> getTreeRoot(Object[] buckets, int index) {
553     return (Queue<T>) buckets[index];
554   }
555 
556   private static int getBucketIndex(Object[] buckets, int hashCode) {
557     return Math.abs(hashCode) % buckets.length;
558   }
559 
560   private static boolean isServerProcedure(Procedure proc) {
561     return proc instanceof ServerProcedureInterface;
562   }
563 
564   private static ServerName getServerName(Procedure proc) {
565     return ((ServerProcedureInterface)proc).getServerName();
566   }
567 
568   // ============================================================================
569   //  Table and Server Queue Implementation
570   // ============================================================================
571   public static class ServerQueue extends QueueImpl<ServerName> {
572     public ServerQueue(ServerName serverName) {
573       super(serverName);
574     }
575 
576     public boolean requireExclusiveLock(Procedure proc) {
577       ServerProcedureInterface spi = (ServerProcedureInterface)proc;
578       switch (spi.getServerOperationType()) {
579         case CRASH_HANDLER:
580           return true;
581         default:
582           break;
583       }
584       throw new UnsupportedOperationException("unexpected type " + spi.getServerOperationType());
585     }
586   }
587 
588   public static class TableQueue extends QueueImpl<TableName> {
589     private final NamespaceQueue namespaceQueue;
590 
591     private TableLock tableLock = null;
592 
593     public TableQueue(TableName tableName, NamespaceQueue namespaceQueue, int priority) {
594       super(tableName, priority);
595       this.namespaceQueue = namespaceQueue;
596     }
597 
598     public NamespaceQueue getNamespaceQueue() {
599       return namespaceQueue;
600     }
601 
602     @Override
603     public synchronized boolean isAvailable() {
604       return super.isAvailable() && !namespaceQueue.hasExclusiveLock();
605     }
606 
607     // TODO: We can abort pending/in-progress operation if the new call is
608     //       something like drop table. We can Override addBack(),
609     //       check the type and abort all the in-flight procedurs.
610     private boolean canAbortPendingOperations(Procedure proc) {
611       TableProcedureInterface tpi = (TableProcedureInterface)proc;
612       switch (tpi.getTableOperationType()) {
613         case DELETE:
614           return true;
615         default:
616           return false;
617       }
618     }
619 
620     public boolean requireExclusiveLock(Procedure proc) {
621       TableProcedureInterface tpi = (TableProcedureInterface)proc;
622       switch (tpi.getTableOperationType()) {
623         case CREATE:
624         case DELETE:
625         case DISABLE:
626         case ENABLE:
627           return true;
628         case EDIT:
629           // we allow concurrent edit on the NS table
630           return !tpi.getTableName().equals(TableName.NAMESPACE_TABLE_NAME);
631         case READ:
632           return false;
633         default:
634           break;
635       }
636       throw new UnsupportedOperationException("unexpected type " + tpi.getTableOperationType());
637     }
638 
639     private synchronized boolean tryZkSharedLock(final TableLockManager lockManager,
640         final String purpose) {
641       // Take zk-read-lock
642       TableName tableName = getKey();
643       tableLock = lockManager.readLock(tableName, purpose);
644       try {
645         tableLock.acquire();
646       } catch (IOException e) {
647         LOG.error("failed acquire read lock on " + tableName, e);
648         tableLock = null;
649         return false;
650       }
651       return true;
652     }
653 
654     private synchronized void releaseZkSharedLock(final TableLockManager lockManager) {
655       releaseTableLock(lockManager, isSingleSharedLock());
656     }
657 
658     private synchronized boolean tryZkExclusiveLock(final TableLockManager lockManager,
659         final String purpose) {
660       // Take zk-write-lock
661       TableName tableName = getKey();
662       tableLock = lockManager.writeLock(tableName, purpose);
663       try {
664         tableLock.acquire();
665       } catch (IOException e) {
666         LOG.error("failed acquire write lock on " + tableName, e);
667         tableLock = null;
668         return false;
669       }
670       return true;
671     }
672 
673     private synchronized void releaseZkExclusiveLock(final TableLockManager lockManager) {
674       releaseTableLock(lockManager, true);
675     }
676 
677     private void releaseTableLock(final TableLockManager lockManager, boolean reset) {
678       for (int i = 0; i < 3; ++i) {
679         try {
680           tableLock.release();
681           if (reset) {
682             tableLock = null;
683           }
684           break;
685         } catch (IOException e) {
686           LOG.warn("Could not release the table write-lock", e);
687         }
688       }
689     }
690   }
691 
692   /**
693    * the namespace is currently used just as a rwlock, not as a queue.
694    * because ns operation are not frequent enough. so we want to avoid
695    * having to move table queues around for suspend/resume.
696    */
697   private static class NamespaceQueue extends Queue<String> {
698     public NamespaceQueue(String namespace) {
699       super(namespace);
700     }
701 
702     @Override
703     public boolean requireExclusiveLock(Procedure proc) {
704       throw new UnsupportedOperationException();
705     }
706 
707     @Override
708     public void add(final Procedure proc, final boolean addToFront) {
709       throw new UnsupportedOperationException();
710     }
711 
712     @Override
713     public Procedure peek() {
714       throw new UnsupportedOperationException();
715     }
716 
717     @Override
718     public Procedure poll() {
719       throw new UnsupportedOperationException();
720     }
721 
722     @Override
723     public boolean isEmpty() {
724       throw new UnsupportedOperationException();
725     }
726 
727     @Override
728     public int size() {
729       throw new UnsupportedOperationException();
730     }
731   }
732 
733   // ============================================================================
734   //  Table Locking Helpers
735   // ============================================================================
736   /**
737    * Try to acquire the exclusive lock on the specified table.
738    * other operations in the table-queue will be executed after the lock is released.
739    * @param procedure the procedure trying to acquire the lock
740    * @param table Table to lock
741    * @return true if we were able to acquire the lock on the table, otherwise false.
742    */
743   public boolean tryAcquireTableExclusiveLock(final Procedure procedure, final TableName table) {
744     schedLock.lock();
745     TableQueue queue = getTableQueue(table);
746     if (!queue.getNamespaceQueue().trySharedLock()) {
747       return false;
748     }
749 
750     if (!queue.tryExclusiveLock(procedure.getProcId())) {
751       queue.getNamespaceQueue().releaseSharedLock();
752       schedLock.unlock();
753       return false;
754     }
755 
756     removeFromRunQueue(tableRunQueue, queue);
757     schedLock.unlock();
758 
759     // Zk lock is expensive...
760     boolean hasXLock = queue.tryZkExclusiveLock(lockManager, procedure.toString());
761     if (!hasXLock) {
762       schedLock.lock();
763       queue.releaseExclusiveLock();
764       queue.getNamespaceQueue().releaseSharedLock();
765       addToRunQueue(tableRunQueue, queue);
766       schedLock.unlock();
767     }
768     return hasXLock;
769   }
770 
771   /**
772    * Release the exclusive lock taken with tryAcquireTableWrite()
773    * @param procedure the procedure releasing the lock
774    * @param table the name of the table that has the exclusive lock
775    */
776   public void releaseTableExclusiveLock(final Procedure procedure, final TableName table) {
777     schedLock.lock();
778     TableQueue queue = getTableQueue(table);
779     schedLock.unlock();
780 
781     // Zk lock is expensive...
782     queue.releaseZkExclusiveLock(lockManager);
783 
784     schedLock.lock();
785     queue.releaseExclusiveLock();
786     queue.getNamespaceQueue().releaseSharedLock();
787     addToRunQueue(tableRunQueue, queue);
788     schedLock.unlock();
789   }
790 
791   /**
792    * Try to acquire the shared lock on the specified table.
793    * other "read" operations in the table-queue may be executed concurrently,
794    * @param procedure the procedure trying to acquire the lock
795    * @param table Table to lock
796    * @return true if we were able to acquire the lock on the table, otherwise false.
797    */
798   public boolean tryAcquireTableSharedLock(final Procedure procedure, final TableName table) {
799     return tryAcquireTableQueueSharedLock(procedure, table) != null;
800   }
801 
802   private TableQueue tryAcquireTableQueueSharedLock(final Procedure procedure,
803       final TableName table) {
804     schedLock.lock();
805     TableQueue queue = getTableQueue(table);
806     if (!queue.getNamespaceQueue().trySharedLock()) {
807       return null;
808     }
809 
810     if (!queue.trySharedLock()) {
811       queue.getNamespaceQueue().releaseSharedLock();
812       schedLock.unlock();
813       return null;
814     }
815 
816     schedLock.unlock();
817 
818     // Zk lock is expensive...
819     if (!queue.tryZkSharedLock(lockManager, procedure.toString())) {
820       schedLock.lock();
821       queue.releaseSharedLock();
822       queue.getNamespaceQueue().releaseSharedLock();
823       schedLock.unlock();
824       return null;
825     }
826     return queue;
827   }
828 
829   /**
830    * Release the shared lock taken with tryAcquireTableRead()
831    * @param procedure the procedure releasing the lock
832    * @param table the name of the table that has the shared lock
833    */
834   public void releaseTableSharedLock(final Procedure procedure, final TableName table) {
835     final TableQueue queue = getTableQueueWithLock(table);
836 
837     // Zk lock is expensive...
838     queue.releaseZkSharedLock(lockManager);
839 
840     schedLock.lock();
841     queue.releaseSharedLock();
842     queue.getNamespaceQueue().releaseSharedLock();
843     schedLock.unlock();
844   }
845 
846   /**
847    * Tries to remove the queue and the table-lock of the specified table.
848    * If there are new operations pending (e.g. a new create),
849    * the remove will not be performed.
850    * @param table the name of the table that should be marked as deleted
851    * @return true if deletion succeeded, false otherwise meaning that there are
852    *     other new operations pending for that table (e.g. a new create).
853    */
854   protected boolean markTableAsDeleted(final TableName table) {
855     final ReentrantLock l = schedLock;
856     l.lock();
857     try {
858       TableQueue queue = getTableQueue(table);
859       if (queue == null) return true;
860 
861       if (queue.isEmpty() && queue.tryExclusiveLock(0)) {
862         // remove the table from the run-queue and the map
863         if (IterableList.isLinked(queue)) {
864           tableRunQueue.remove(queue);
865         }
866 
867         // Remove the table lock
868         try {
869           lockManager.tableDeleted(table);
870         } catch (IOException e) {
871           LOG.warn("Received exception from TableLockManager.tableDeleted:", e); //not critical
872         }
873 
874         removeTableQueue(table);
875       } else {
876         // TODO: If there are no create, we can drop all the other ops
877         return false;
878       }
879     } finally {
880       l.unlock();
881     }
882     return true;
883   }
884 
885   // ============================================================================
886   //  Namespace Locking Helpers
887   // ============================================================================
888   /**
889    * Try to acquire the exclusive lock on the specified namespace.
890    * @see #releaseNamespaceExclusiveLock(Procedure,String)
891    * @param procedure the procedure trying to acquire the lock
892    * @param nsName Namespace to lock
893    * @return true if we were able to acquire the lock on the namespace, otherwise false.
894    */
895   public boolean tryAcquireNamespaceExclusiveLock(final Procedure procedure, final String nsName) {
896     schedLock.lock();
897     try {
898       TableQueue tableQueue = getTableQueue(TableName.NAMESPACE_TABLE_NAME);
899       if (!tableQueue.trySharedLock()) return false;
900 
901       NamespaceQueue nsQueue = getNamespaceQueue(nsName);
902       boolean hasLock = nsQueue.tryExclusiveLock(procedure.getProcId());
903       if (!hasLock) {
904         tableQueue.releaseSharedLock();
905       }
906       return hasLock;
907     } finally {
908       schedLock.unlock();
909     }
910   }
911 
912   /**
913    * Release the exclusive lock
914    * @see #tryAcquireNamespaceExclusiveLock(Procedure,String)
915    * @param procedure the procedure releasing the lock
916    * @param nsName the namespace that has the exclusive lock
917    */
918   public void releaseNamespaceExclusiveLock(final Procedure procedure, final String nsName) {
919     schedLock.lock();
920     try {
921       TableQueue tableQueue = getTableQueue(TableName.NAMESPACE_TABLE_NAME);
922       tableQueue.releaseSharedLock();
923 
924       NamespaceQueue queue = getNamespaceQueue(nsName);
925       queue.releaseExclusiveLock();
926     } finally {
927       schedLock.unlock();
928     }
929   }
930 
931   // ============================================================================
932   //  Server Locking Helpers
933   // ============================================================================
934   /**
935    * Try to acquire the exclusive lock on the specified server.
936    * @see #releaseServerExclusiveLock(Procedure,ServerName)
937    * @param procedure the procedure trying to acquire the lock
938    * @param serverName Server to lock
939    * @return true if we were able to acquire the lock on the server, otherwise false.
940    */
941   public boolean tryAcquireServerExclusiveLock(final Procedure procedure,
942       final ServerName serverName) {
943     schedLock.lock();
944     try {
945       ServerQueue queue = getServerQueue(serverName);
946       if (queue.tryExclusiveLock(procedure.getProcId())) {
947         removeFromRunQueue(serverRunQueue, queue);
948         return true;
949       }
950     } finally {
951       schedLock.unlock();
952     }
953     return false;
954   }
955 
956   /**
957    * Release the exclusive lock
958    * @see #tryAcquireServerExclusiveLock(Procedure,ServerName)
959    * @param procedure the procedure releasing the lock
960    * @param serverName the server that has the exclusive lock
961    */
962   public void releaseServerExclusiveLock(final Procedure procedure,
963       final ServerName serverName) {
964     schedLock.lock();
965     try {
966       ServerQueue queue = getServerQueue(serverName);
967       queue.releaseExclusiveLock();
968       addToRunQueue(serverRunQueue, queue);
969     } finally {
970       schedLock.unlock();
971     }
972   }
973 
974   /**
975    * Try to acquire the shared lock on the specified server.
976    * @see #releaseServerSharedLock(Procedure,ServerName)
977    * @param procedure the procedure releasing the lock
978    * @param serverName Server to lock
979    * @return true if we were able to acquire the lock on the server, otherwise false.
980    */
981   public boolean tryAcquireServerSharedLock(final Procedure procedure,
982       final ServerName serverName) {
983     return getServerQueueWithLock(serverName).trySharedLock();
984   }
985 
986   /**
987    * Release the shared lock taken
988    * @see #tryAcquireServerSharedLock(Procedure,ServerName)
989    * @param procedure the procedure releasing the lock
990    * @param serverName the server that has the shared lock
991    */
992   public void releaseServerSharedLock(final Procedure procedure,
993       final ServerName serverName) {
994     getServerQueueWithLock(serverName).releaseSharedLock();
995   }
996 
997   // ============================================================================
998   //  Generic Helpers
999   // ============================================================================
1000   private static interface QueueInterface {
1001     boolean isAvailable();
1002     boolean isEmpty();
1003     int size();
1004 
1005     void add(Procedure proc, boolean addFront);
1006     boolean requireExclusiveLock(Procedure proc);
1007     Procedure peek();
1008     Procedure poll();
1009 
1010     boolean isSuspended();
1011   }
1012 
1013   private static abstract class Queue<TKey extends Comparable<TKey>> implements QueueInterface {
1014     private Queue<TKey> avlRight = null;
1015     private Queue<TKey> avlLeft = null;
1016     private int avlHeight = 1;
1017 
1018     private Queue<TKey> iterNext = null;
1019     private Queue<TKey> iterPrev = null;
1020     private boolean suspended = false;
1021 
1022     private long exclusiveLockProcIdOwner = Long.MIN_VALUE;
1023     private int sharedLock = 0;
1024 
1025     private final TKey key;
1026     private final int priority;
1027 
1028     public Queue(TKey key) {
1029       this(key, 1);
1030     }
1031 
1032     public Queue(TKey key, int priority) {
1033       this.key = key;
1034       this.priority = priority;
1035     }
1036 
1037     protected TKey getKey() {
1038       return key;
1039     }
1040 
1041     protected int getPriority() {
1042       return priority;
1043     }
1044 
1045     /**
1046      * True if the queue is not in the run-queue and it is owned by an event.
1047      */
1048     public boolean isSuspended() {
1049       return suspended;
1050     }
1051 
1052     protected boolean setSuspended(boolean isSuspended) {
1053       if (this.suspended == isSuspended) return false;
1054       this.suspended = isSuspended;
1055       return true;
1056     }
1057 
1058     // ======================================================================
1059     //  Read/Write Locking helpers
1060     // ======================================================================
1061     public synchronized boolean isLocked() {
1062       return hasExclusiveLock() || sharedLock > 0;
1063     }
1064 
1065     public synchronized boolean hasExclusiveLock() {
1066       return this.exclusiveLockProcIdOwner != Long.MIN_VALUE;
1067     }
1068 
1069     public synchronized boolean trySharedLock() {
1070       if (hasExclusiveLock()) return false;
1071       sharedLock++;
1072       return true;
1073     }
1074 
1075     public synchronized void releaseSharedLock() {
1076       sharedLock--;
1077     }
1078 
1079     protected synchronized boolean isSingleSharedLock() {
1080       return sharedLock == 1;
1081     }
1082 
1083     public synchronized boolean tryExclusiveLock(long procIdOwner) {
1084       assert procIdOwner != Long.MIN_VALUE;
1085       if (isLocked()) return false;
1086       exclusiveLockProcIdOwner = procIdOwner;
1087       return true;
1088     }
1089 
1090     public synchronized void releaseExclusiveLock() {
1091       exclusiveLockProcIdOwner = Long.MIN_VALUE;
1092     }
1093 
1094     // This should go away when we have the new AM and its events
1095     // and we move xlock to the lock-event-queue.
1096     public synchronized boolean isAvailable() {
1097       return !hasExclusiveLock() && !isEmpty();
1098     }
1099 
1100     // ======================================================================
1101     //  Generic Helpers
1102     // ======================================================================
1103     public int compareKey(TKey cmpKey) {
1104       return key.compareTo(cmpKey);
1105     }
1106 
1107     public int compareTo(Queue<TKey> other) {
1108       return compareKey(other.key);
1109     }
1110 
1111     @Override
1112     public String toString() {
1113       return String.format("%s(%s)", getClass().getSimpleName(), key);
1114     }
1115   }
1116 
1117   // ======================================================================
1118   //  Helper Data Structures
1119   // ======================================================================
1120   private static abstract class QueueImpl<TKey extends Comparable<TKey>> extends Queue<TKey> {
1121     private final ArrayDeque<Procedure> runnables = new ArrayDeque<Procedure>();
1122 
1123     public QueueImpl(TKey key) {
1124       super(key);
1125     }
1126 
1127     public QueueImpl(TKey key, int priority) {
1128       super(key, priority);
1129     }
1130 
1131     public void add(final Procedure proc, final boolean addToFront) {
1132       if (addToFront) {
1133         addFront(proc);
1134       } else {
1135         addBack(proc);
1136       }
1137     }
1138 
1139     protected void addFront(final Procedure proc) {
1140       runnables.addFirst(proc);
1141     }
1142 
1143     protected void addBack(final Procedure proc) {
1144       runnables.addLast(proc);
1145     }
1146 
1147     public Procedure peek() {
1148       return runnables.peek();
1149     }
1150 
1151     @Override
1152     public Procedure poll() {
1153       return runnables.poll();
1154     }
1155 
1156     @Override
1157     public boolean isEmpty() {
1158       return runnables.isEmpty();
1159     }
1160 
1161     public int size() {
1162       return runnables.size();
1163     }
1164   }
1165 
1166   private static class FairQueue<T extends Comparable<T>> {
1167     private final int quantum;
1168 
1169     private Queue<T> currentQueue = null;
1170     private Queue<T> queueHead = null;
1171     private int currentQuantum = 0;
1172 
1173     public FairQueue() {
1174       this(1);
1175     }
1176 
1177     public FairQueue(int quantum) {
1178       this.quantum = quantum;
1179     }
1180 
1181     public void add(Queue<T> queue) {
1182       queueHead = IterableList.append(queueHead, queue);
1183       if (currentQueue == null) setNextQueue(queueHead);
1184     }
1185 
1186     public void remove(Queue<T> queue) {
1187       Queue<T> nextQueue = queue.iterNext;
1188       queueHead = IterableList.remove(queueHead, queue);
1189       if (currentQueue == queue) {
1190         setNextQueue(queueHead != null ? nextQueue : null);
1191       }
1192     }
1193 
1194     public Queue<T> poll() {
1195       if (currentQuantum == 0) {
1196         if (!nextQueue()) {
1197           return null; // nothing here
1198         }
1199         currentQuantum = calculateQuantum(currentQueue) - 1;
1200       } else {
1201         currentQuantum--;
1202       }
1203 
1204       // This should go away when we have the new AM and its events
1205       if (!currentQueue.isAvailable()) {
1206         Queue<T> lastQueue = currentQueue;
1207         do {
1208           if (!nextQueue())
1209             return null;
1210         } while (currentQueue != lastQueue && !currentQueue.isAvailable());
1211 
1212         currentQuantum = calculateQuantum(currentQueue) - 1;
1213       }
1214       return currentQueue;
1215     }
1216 
1217     private boolean nextQueue() {
1218       if (currentQueue == null) return false;
1219       currentQueue = currentQueue.iterNext;
1220       return currentQueue != null;
1221     }
1222 
1223     private void setNextQueue(Queue<T> queue) {
1224       currentQueue = queue;
1225       if (queue != null) {
1226         currentQuantum = calculateQuantum(currentQueue);
1227       } else {
1228         currentQuantum = 0;
1229       }
1230     }
1231 
1232     private int calculateQuantum(final Queue queue) {
1233       return Math.max(1, queue.getPriority() * quantum); // TODO
1234     }
1235   }
1236 
1237   private static class AvlTree {
1238     public static <T extends Comparable<T>> Queue<T> get(Queue<T> root, T key) {
1239       while (root != null) {
1240         int cmp = root.compareKey(key);
1241         if (cmp > 0) {
1242           root = root.avlLeft;
1243         } else if (cmp < 0) {
1244           root = root.avlRight;
1245         } else {
1246           return root;
1247         }
1248       }
1249       return null;
1250     }
1251 
1252     public static <T extends Comparable<T>> Queue<T> getFirst(Queue<T> root) {
1253       if (root != null) {
1254         while (root.avlLeft != null) {
1255           root = root.avlLeft;
1256         }
1257       }
1258       return root;
1259     }
1260 
1261     public static <T extends Comparable<T>> Queue<T> getLast(Queue<T> root) {
1262       if (root != null) {
1263         while (root.avlRight != null) {
1264           root = root.avlRight;
1265         }
1266       }
1267       return root;
1268     }
1269 
1270     public static <T extends Comparable<T>> Queue<T> insert(Queue<T> root, Queue<T> node) {
1271       if (root == null) return node;
1272       if (node.compareTo(root) < 0) {
1273         root.avlLeft = insert(root.avlLeft, node);
1274       } else {
1275         root.avlRight = insert(root.avlRight, node);
1276       }
1277       return balance(root);
1278     }
1279 
1280     private static <T extends Comparable<T>> Queue<T> removeMin(Queue<T> p) {
1281       if (p.avlLeft == null)
1282         return p.avlRight;
1283       p.avlLeft = removeMin(p.avlLeft);
1284       return balance(p);
1285     }
1286 
1287     public static <T extends Comparable<T>> Queue<T> remove(Queue<T> root, T key) {
1288       if (root == null) return null;
1289 
1290       int cmp = root.compareKey(key);
1291       if (cmp == 0) {
1292         Queue<T> q = root.avlLeft;
1293         Queue<T> r = root.avlRight;
1294         if (r == null) return q;
1295         Queue<T> min = getFirst(r);
1296         min.avlRight = removeMin(r);
1297         min.avlLeft = q;
1298         return balance(min);
1299       } else if (cmp > 0) {
1300         root.avlLeft = remove(root.avlLeft, key);
1301       } else /* if (cmp < 0) */ {
1302         root.avlRight = remove(root.avlRight, key);
1303       }
1304       return balance(root);
1305     }
1306 
1307     private static <T extends Comparable<T>> Queue<T> balance(Queue<T> p) {
1308       fixHeight(p);
1309       int balance = balanceFactor(p);
1310       if (balance == 2) {
1311         if (balanceFactor(p.avlRight) < 0) {
1312           p.avlRight = rotateRight(p.avlRight);
1313         }
1314         return rotateLeft(p);
1315       } else if (balance == -2) {
1316         if (balanceFactor(p.avlLeft) > 0) {
1317           p.avlLeft = rotateLeft(p.avlLeft);
1318         }
1319         return rotateRight(p);
1320       }
1321       return p;
1322     }
1323 
1324     private static <T extends Comparable<T>> Queue<T> rotateRight(Queue<T> p) {
1325       Queue<T> q = p.avlLeft;
1326       p.avlLeft = q.avlRight;
1327       q.avlRight = p;
1328       fixHeight(p);
1329       fixHeight(q);
1330       return q;
1331     }
1332 
1333     private static <T extends Comparable<T>> Queue<T> rotateLeft(Queue<T> q) {
1334       Queue<T> p = q.avlRight;
1335       q.avlRight = p.avlLeft;
1336       p.avlLeft = q;
1337       fixHeight(q);
1338       fixHeight(p);
1339       return p;
1340     }
1341 
1342     private static <T extends Comparable<T>> void fixHeight(Queue<T> node) {
1343       int heightLeft = height(node.avlLeft);
1344       int heightRight = height(node.avlRight);
1345       node.avlHeight = 1 + Math.max(heightLeft, heightRight);
1346     }
1347 
1348     private static <T extends Comparable<T>> int height(Queue<T> node) {
1349       return node != null ? node.avlHeight : 0;
1350     }
1351 
1352     private static <T extends Comparable<T>> int balanceFactor(Queue<T> node) {
1353       return height(node.avlRight) - height(node.avlLeft);
1354     }
1355   }
1356 
1357   private static class IterableList {
1358     public static <T extends Comparable<T>> Queue<T> prepend(Queue<T> head, Queue<T> node) {
1359       assert !isLinked(node) : node + " is already linked";
1360       if (head != null) {
1361         Queue<T> tail = head.iterPrev;
1362         tail.iterNext = node;
1363         head.iterPrev = node;
1364         node.iterNext = head;
1365         node.iterPrev = tail;
1366       } else {
1367         node.iterNext = node;
1368         node.iterPrev = node;
1369       }
1370       return node;
1371     }
1372 
1373     public static <T extends Comparable<T>> Queue<T> append(Queue<T> head, Queue<T> node) {
1374       assert !isLinked(node) : node + " is already linked";
1375       if (head != null) {
1376         Queue<T> tail = head.iterPrev;
1377         tail.iterNext = node;
1378         node.iterNext = head;
1379         node.iterPrev = tail;
1380         head.iterPrev = node;
1381         return head;
1382       }
1383       node.iterNext = node;
1384       node.iterPrev = node;
1385       return node;
1386     }
1387 
1388     public static <T extends Comparable<T>> Queue<T> appendList(Queue<T> head, Queue<T> otherHead) {
1389       if (head == null) return otherHead;
1390       if (otherHead == null) return head;
1391 
1392       Queue<T> tail = head.iterPrev;
1393       Queue<T> otherTail = otherHead.iterPrev;
1394       tail.iterNext = otherHead;
1395       otherHead.iterPrev = tail;
1396       otherTail.iterNext = head;
1397       head.iterPrev = otherTail;
1398       return head;
1399     }
1400 
1401     private static <T extends Comparable<T>> Queue<T> remove(Queue<T> head, Queue<T> node) {
1402       assert isLinked(node) : node + " is not linked";
1403       if (node != node.iterNext) {
1404         node.iterPrev.iterNext = node.iterNext;
1405         node.iterNext.iterPrev = node.iterPrev;
1406         head = (head == node) ? node.iterNext : head;
1407       } else {
1408         head = null;
1409       }
1410       node.iterNext = null;
1411       node.iterPrev = null;
1412       return head;
1413     }
1414 
1415     private static <T extends Comparable<T>> boolean isLinked(Queue<T> node) {
1416       return node.iterPrev != null && node.iterNext != null;
1417     }
1418   }
1419 }