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     @Override
577     public boolean requireExclusiveLock(Procedure proc) {
578       ServerProcedureInterface spi = (ServerProcedureInterface)proc;
579       switch (spi.getServerOperationType()) {
580         case CRASH_HANDLER:
581           return true;
582         default:
583           break;
584       }
585       throw new UnsupportedOperationException("unexpected type " + spi.getServerOperationType());
586     }
587   }
588 
589   public static class TableQueue extends QueueImpl<TableName> {
590     private final NamespaceQueue namespaceQueue;
591 
592     private TableLock tableLock = null;
593 
594     public TableQueue(TableName tableName, NamespaceQueue namespaceQueue, int priority) {
595       super(tableName, priority);
596       this.namespaceQueue = namespaceQueue;
597     }
598 
599     public NamespaceQueue getNamespaceQueue() {
600       return namespaceQueue;
601     }
602 
603     @Override
604     public synchronized boolean isAvailable() {
605       return super.isAvailable() && !namespaceQueue.hasExclusiveLock();
606     }
607 
608     // TODO: We can abort pending/in-progress operation if the new call is
609     //       something like drop table. We can Override addBack(),
610     //       check the type and abort all the in-flight procedurs.
611     private boolean canAbortPendingOperations(Procedure proc) {
612       TableProcedureInterface tpi = (TableProcedureInterface)proc;
613       switch (tpi.getTableOperationType()) {
614         case DELETE:
615           return true;
616         default:
617           return false;
618       }
619     }
620 
621     @Override
622     public boolean requireExclusiveLock(Procedure proc) {
623       TableProcedureInterface tpi = (TableProcedureInterface)proc;
624       switch (tpi.getTableOperationType()) {
625         case CREATE:
626         case DELETE:
627         case DISABLE:
628         case ENABLE:
629           return true;
630         case EDIT:
631           // we allow concurrent edit on the NS table
632           return !tpi.getTableName().equals(TableName.NAMESPACE_TABLE_NAME);
633         case READ:
634           return false;
635         default:
636           break;
637       }
638       throw new UnsupportedOperationException("unexpected type " + tpi.getTableOperationType());
639     }
640 
641     private synchronized boolean tryZkSharedLock(final TableLockManager lockManager,
642         final String purpose) {
643       // Take zk-read-lock
644       TableName tableName = getKey();
645       tableLock = lockManager.readLock(tableName, purpose);
646       try {
647         tableLock.acquire();
648       } catch (IOException e) {
649         LOG.error("failed acquire read lock on " + tableName, e);
650         tableLock = null;
651         return false;
652       }
653       return true;
654     }
655 
656     private synchronized void releaseZkSharedLock(final TableLockManager lockManager) {
657       releaseTableLock(lockManager, isSingleSharedLock());
658     }
659 
660     private synchronized boolean tryZkExclusiveLock(final TableLockManager lockManager,
661         final String purpose) {
662       // Take zk-write-lock
663       TableName tableName = getKey();
664       tableLock = lockManager.writeLock(tableName, purpose);
665       try {
666         tableLock.acquire();
667       } catch (IOException e) {
668         LOG.error("failed acquire write lock on " + tableName, e);
669         tableLock = null;
670         return false;
671       }
672       return true;
673     }
674 
675     private synchronized void releaseZkExclusiveLock(final TableLockManager lockManager) {
676       releaseTableLock(lockManager, true);
677     }
678 
679     private void releaseTableLock(final TableLockManager lockManager, boolean reset) {
680       for (int i = 0; i < 3; ++i) {
681         try {
682           tableLock.release();
683           if (reset) {
684             tableLock = null;
685           }
686           break;
687         } catch (IOException e) {
688           LOG.warn("Could not release the table write-lock", e);
689         }
690       }
691     }
692   }
693 
694   /**
695    * the namespace is currently used just as a rwlock, not as a queue.
696    * because ns operation are not frequent enough. so we want to avoid
697    * having to move table queues around for suspend/resume.
698    */
699   private static class NamespaceQueue extends Queue<String> {
700     public NamespaceQueue(String namespace) {
701       super(namespace);
702     }
703 
704     @Override
705     public boolean requireExclusiveLock(Procedure proc) {
706       throw new UnsupportedOperationException();
707     }
708 
709     @Override
710     public void add(final Procedure proc, final boolean addToFront) {
711       throw new UnsupportedOperationException();
712     }
713 
714     @Override
715     public Procedure peek() {
716       throw new UnsupportedOperationException();
717     }
718 
719     @Override
720     public Procedure poll() {
721       throw new UnsupportedOperationException();
722     }
723 
724     @Override
725     public boolean isEmpty() {
726       throw new UnsupportedOperationException();
727     }
728 
729     @Override
730     public int size() {
731       throw new UnsupportedOperationException();
732     }
733   }
734 
735   // ============================================================================
736   //  Table Locking Helpers
737   // ============================================================================
738   /**
739    * Try to acquire the exclusive lock on the specified table.
740    * other operations in the table-queue will be executed after the lock is released.
741    * @param procedure the procedure trying to acquire the lock
742    * @param table Table to lock
743    * @return true if we were able to acquire the lock on the table, otherwise false.
744    */
745   public boolean tryAcquireTableExclusiveLock(final Procedure procedure, final TableName table) {
746     schedLock.lock();
747     TableQueue queue = getTableQueue(table);
748     if (!queue.getNamespaceQueue().trySharedLock()) {
749       return false;
750     }
751 
752     if (!queue.tryExclusiveLock(procedure.getProcId())) {
753       queue.getNamespaceQueue().releaseSharedLock();
754       schedLock.unlock();
755       return false;
756     }
757 
758     removeFromRunQueue(tableRunQueue, queue);
759     schedLock.unlock();
760 
761     // Zk lock is expensive...
762     boolean hasXLock = queue.tryZkExclusiveLock(lockManager, procedure.toString());
763     if (!hasXLock) {
764       schedLock.lock();
765       queue.releaseExclusiveLock();
766       queue.getNamespaceQueue().releaseSharedLock();
767       addToRunQueue(tableRunQueue, queue);
768       schedLock.unlock();
769     }
770     return hasXLock;
771   }
772 
773   /**
774    * Release the exclusive lock taken with tryAcquireTableWrite()
775    * @param procedure the procedure releasing the lock
776    * @param table the name of the table that has the exclusive lock
777    */
778   public void releaseTableExclusiveLock(final Procedure procedure, final TableName table) {
779     schedLock.lock();
780     TableQueue queue = getTableQueue(table);
781     schedLock.unlock();
782 
783     // Zk lock is expensive...
784     queue.releaseZkExclusiveLock(lockManager);
785 
786     schedLock.lock();
787     queue.releaseExclusiveLock();
788     queue.getNamespaceQueue().releaseSharedLock();
789     addToRunQueue(tableRunQueue, queue);
790     schedLock.unlock();
791   }
792 
793   /**
794    * Try to acquire the shared lock on the specified table.
795    * other "read" operations in the table-queue may be executed concurrently,
796    * @param procedure the procedure trying to acquire the lock
797    * @param table Table to lock
798    * @return true if we were able to acquire the lock on the table, otherwise false.
799    */
800   public boolean tryAcquireTableSharedLock(final Procedure procedure, final TableName table) {
801     return tryAcquireTableQueueSharedLock(procedure, table) != null;
802   }
803 
804   private TableQueue tryAcquireTableQueueSharedLock(final Procedure procedure,
805       final TableName table) {
806     schedLock.lock();
807     TableQueue queue = getTableQueue(table);
808     if (!queue.getNamespaceQueue().trySharedLock()) {
809       return null;
810     }
811 
812     if (!queue.trySharedLock()) {
813       queue.getNamespaceQueue().releaseSharedLock();
814       schedLock.unlock();
815       return null;
816     }
817 
818     schedLock.unlock();
819 
820     // Zk lock is expensive...
821     if (!queue.tryZkSharedLock(lockManager, procedure.toString())) {
822       schedLock.lock();
823       queue.releaseSharedLock();
824       queue.getNamespaceQueue().releaseSharedLock();
825       schedLock.unlock();
826       return null;
827     }
828     return queue;
829   }
830 
831   /**
832    * Release the shared lock taken with tryAcquireTableRead()
833    * @param procedure the procedure releasing the lock
834    * @param table the name of the table that has the shared lock
835    */
836   public void releaseTableSharedLock(final Procedure procedure, final TableName table) {
837     final TableQueue queue = getTableQueueWithLock(table);
838 
839     // Zk lock is expensive...
840     queue.releaseZkSharedLock(lockManager);
841 
842     schedLock.lock();
843     queue.releaseSharedLock();
844     queue.getNamespaceQueue().releaseSharedLock();
845     schedLock.unlock();
846   }
847 
848   /**
849    * Tries to remove the queue and the table-lock of the specified table.
850    * If there are new operations pending (e.g. a new create),
851    * the remove will not be performed.
852    * @param table the name of the table that should be marked as deleted
853    * @return true if deletion succeeded, false otherwise meaning that there are
854    *     other new operations pending for that table (e.g. a new create).
855    */
856   protected boolean markTableAsDeleted(final TableName table) {
857     final ReentrantLock l = schedLock;
858     l.lock();
859     try {
860       TableQueue queue = getTableQueue(table);
861       if (queue == null) return true;
862 
863       if (queue.isEmpty() && queue.tryExclusiveLock(0)) {
864         // remove the table from the run-queue and the map
865         if (IterableList.isLinked(queue)) {
866           tableRunQueue.remove(queue);
867         }
868 
869         // Remove the table lock
870         try {
871           lockManager.tableDeleted(table);
872         } catch (IOException e) {
873           LOG.warn("Received exception from TableLockManager.tableDeleted:", e); //not critical
874         }
875 
876         removeTableQueue(table);
877       } else {
878         // TODO: If there are no create, we can drop all the other ops
879         return false;
880       }
881     } finally {
882       l.unlock();
883     }
884     return true;
885   }
886 
887   // ============================================================================
888   //  Namespace Locking Helpers
889   // ============================================================================
890   /**
891    * Try to acquire the exclusive lock on the specified namespace.
892    * @see #releaseNamespaceExclusiveLock(Procedure,String)
893    * @param procedure the procedure trying to acquire the lock
894    * @param nsName Namespace to lock
895    * @return true if we were able to acquire the lock on the namespace, otherwise false.
896    */
897   public boolean tryAcquireNamespaceExclusiveLock(final Procedure procedure, final String nsName) {
898     schedLock.lock();
899     try {
900       TableQueue tableQueue = getTableQueue(TableName.NAMESPACE_TABLE_NAME);
901       if (!tableQueue.trySharedLock()) return false;
902 
903       NamespaceQueue nsQueue = getNamespaceQueue(nsName);
904       boolean hasLock = nsQueue.tryExclusiveLock(procedure.getProcId());
905       if (!hasLock) {
906         tableQueue.releaseSharedLock();
907       }
908       return hasLock;
909     } finally {
910       schedLock.unlock();
911     }
912   }
913 
914   /**
915    * Release the exclusive lock
916    * @see #tryAcquireNamespaceExclusiveLock(Procedure,String)
917    * @param procedure the procedure releasing the lock
918    * @param nsName the namespace that has the exclusive lock
919    */
920   public void releaseNamespaceExclusiveLock(final Procedure procedure, final String nsName) {
921     schedLock.lock();
922     try {
923       TableQueue tableQueue = getTableQueue(TableName.NAMESPACE_TABLE_NAME);
924       tableQueue.releaseSharedLock();
925 
926       NamespaceQueue queue = getNamespaceQueue(nsName);
927       queue.releaseExclusiveLock();
928     } finally {
929       schedLock.unlock();
930     }
931   }
932 
933   // ============================================================================
934   //  Server Locking Helpers
935   // ============================================================================
936   /**
937    * Try to acquire the exclusive lock on the specified server.
938    * @see #releaseServerExclusiveLock(Procedure,ServerName)
939    * @param procedure the procedure trying to acquire the lock
940    * @param serverName Server to lock
941    * @return true if we were able to acquire the lock on the server, otherwise false.
942    */
943   public boolean tryAcquireServerExclusiveLock(final Procedure procedure,
944       final ServerName serverName) {
945     schedLock.lock();
946     try {
947       ServerQueue queue = getServerQueue(serverName);
948       if (queue.tryExclusiveLock(procedure.getProcId())) {
949         removeFromRunQueue(serverRunQueue, queue);
950         return true;
951       }
952     } finally {
953       schedLock.unlock();
954     }
955     return false;
956   }
957 
958   /**
959    * Release the exclusive lock
960    * @see #tryAcquireServerExclusiveLock(Procedure,ServerName)
961    * @param procedure the procedure releasing the lock
962    * @param serverName the server that has the exclusive lock
963    */
964   public void releaseServerExclusiveLock(final Procedure procedure,
965       final ServerName serverName) {
966     schedLock.lock();
967     try {
968       ServerQueue queue = getServerQueue(serverName);
969       queue.releaseExclusiveLock();
970       addToRunQueue(serverRunQueue, queue);
971     } finally {
972       schedLock.unlock();
973     }
974   }
975 
976   /**
977    * Try to acquire the shared lock on the specified server.
978    * @see #releaseServerSharedLock(Procedure,ServerName)
979    * @param procedure the procedure releasing the lock
980    * @param serverName Server to lock
981    * @return true if we were able to acquire the lock on the server, otherwise false.
982    */
983   public boolean tryAcquireServerSharedLock(final Procedure procedure,
984       final ServerName serverName) {
985     return getServerQueueWithLock(serverName).trySharedLock();
986   }
987 
988   /**
989    * Release the shared lock taken
990    * @see #tryAcquireServerSharedLock(Procedure,ServerName)
991    * @param procedure the procedure releasing the lock
992    * @param serverName the server that has the shared lock
993    */
994   public void releaseServerSharedLock(final Procedure procedure,
995       final ServerName serverName) {
996     getServerQueueWithLock(serverName).releaseSharedLock();
997   }
998 
999   // ============================================================================
1000   //  Generic Helpers
1001   // ============================================================================
1002   private static interface QueueInterface {
1003     boolean isAvailable();
1004     boolean isEmpty();
1005     int size();
1006 
1007     void add(Procedure proc, boolean addFront);
1008     boolean requireExclusiveLock(Procedure proc);
1009     Procedure peek();
1010     Procedure poll();
1011 
1012     boolean isSuspended();
1013   }
1014 
1015   private static abstract class Queue<TKey extends Comparable<TKey>> implements QueueInterface {
1016     private Queue<TKey> avlRight = null;
1017     private Queue<TKey> avlLeft = null;
1018     private int avlHeight = 1;
1019 
1020     private Queue<TKey> iterNext = null;
1021     private Queue<TKey> iterPrev = null;
1022     private boolean suspended = false;
1023 
1024     private long exclusiveLockProcIdOwner = Long.MIN_VALUE;
1025     private int sharedLock = 0;
1026 
1027     private final TKey key;
1028     private final int priority;
1029 
1030     public Queue(TKey key) {
1031       this(key, 1);
1032     }
1033 
1034     public Queue(TKey key, int priority) {
1035       this.key = key;
1036       this.priority = priority;
1037     }
1038 
1039     protected TKey getKey() {
1040       return key;
1041     }
1042 
1043     protected int getPriority() {
1044       return priority;
1045     }
1046 
1047     /**
1048      * True if the queue is not in the run-queue and it is owned by an event.
1049      */
1050     @Override
1051     public boolean isSuspended() {
1052       return suspended;
1053     }
1054 
1055     protected boolean setSuspended(boolean isSuspended) {
1056       if (this.suspended == isSuspended) return false;
1057       this.suspended = isSuspended;
1058       return true;
1059     }
1060 
1061     // ======================================================================
1062     //  Read/Write Locking helpers
1063     // ======================================================================
1064     public synchronized boolean isLocked() {
1065       return hasExclusiveLock() || sharedLock > 0;
1066     }
1067 
1068     public synchronized boolean hasExclusiveLock() {
1069       return this.exclusiveLockProcIdOwner != Long.MIN_VALUE;
1070     }
1071 
1072     public synchronized boolean trySharedLock() {
1073       if (hasExclusiveLock()) return false;
1074       sharedLock++;
1075       return true;
1076     }
1077 
1078     public synchronized void releaseSharedLock() {
1079       sharedLock--;
1080     }
1081 
1082     protected synchronized boolean isSingleSharedLock() {
1083       return sharedLock == 1;
1084     }
1085 
1086     public synchronized boolean tryExclusiveLock(long procIdOwner) {
1087       assert procIdOwner != Long.MIN_VALUE;
1088       if (isLocked()) return false;
1089       exclusiveLockProcIdOwner = procIdOwner;
1090       return true;
1091     }
1092 
1093     public synchronized void releaseExclusiveLock() {
1094       exclusiveLockProcIdOwner = Long.MIN_VALUE;
1095     }
1096 
1097     // This should go away when we have the new AM and its events
1098     // and we move xlock to the lock-event-queue.
1099     @Override
1100     public synchronized boolean isAvailable() {
1101       return !hasExclusiveLock() && !isEmpty();
1102     }
1103 
1104     // ======================================================================
1105     //  Generic Helpers
1106     // ======================================================================
1107     public int compareKey(TKey cmpKey) {
1108       return key.compareTo(cmpKey);
1109     }
1110 
1111     public int compareTo(Queue<TKey> other) {
1112       return compareKey(other.key);
1113     }
1114 
1115     @Override
1116     public String toString() {
1117       return String.format("%s(%s)", getClass().getSimpleName(), key);
1118     }
1119   }
1120 
1121   // ======================================================================
1122   //  Helper Data Structures
1123   // ======================================================================
1124   private static abstract class QueueImpl<TKey extends Comparable<TKey>> extends Queue<TKey> {
1125     private final ArrayDeque<Procedure> runnables = new ArrayDeque<Procedure>();
1126 
1127     public QueueImpl(TKey key) {
1128       super(key);
1129     }
1130 
1131     public QueueImpl(TKey key, int priority) {
1132       super(key, priority);
1133     }
1134 
1135     @Override
1136     public void add(final Procedure proc, final boolean addToFront) {
1137       if (addToFront) {
1138         addFront(proc);
1139       } else {
1140         addBack(proc);
1141       }
1142     }
1143 
1144     protected void addFront(final Procedure proc) {
1145       runnables.addFirst(proc);
1146     }
1147 
1148     protected void addBack(final Procedure proc) {
1149       runnables.addLast(proc);
1150     }
1151 
1152     @Override
1153     public Procedure peek() {
1154       return runnables.peek();
1155     }
1156 
1157     @Override
1158     public Procedure poll() {
1159       return runnables.poll();
1160     }
1161 
1162     @Override
1163     public boolean isEmpty() {
1164       return runnables.isEmpty();
1165     }
1166 
1167     @Override
1168     public int size() {
1169       return runnables.size();
1170     }
1171   }
1172 
1173   private static class FairQueue<T extends Comparable<T>> {
1174     private final int quantum;
1175 
1176     private Queue<T> currentQueue = null;
1177     private Queue<T> queueHead = null;
1178     private int currentQuantum = 0;
1179 
1180     public FairQueue() {
1181       this(1);
1182     }
1183 
1184     public FairQueue(int quantum) {
1185       this.quantum = quantum;
1186     }
1187 
1188     public void add(Queue<T> queue) {
1189       queueHead = IterableList.append(queueHead, queue);
1190       if (currentQueue == null) setNextQueue(queueHead);
1191     }
1192 
1193     public void remove(Queue<T> queue) {
1194       Queue<T> nextQueue = queue.iterNext;
1195       queueHead = IterableList.remove(queueHead, queue);
1196       if (currentQueue == queue) {
1197         setNextQueue(queueHead != null ? nextQueue : null);
1198       }
1199     }
1200 
1201     public Queue<T> poll() {
1202       if (currentQuantum == 0) {
1203         if (!nextQueue()) {
1204           return null; // nothing here
1205         }
1206         currentQuantum = calculateQuantum(currentQueue) - 1;
1207       } else {
1208         currentQuantum--;
1209       }
1210 
1211       // This should go away when we have the new AM and its events
1212       if (!currentQueue.isAvailable()) {
1213         Queue<T> lastQueue = currentQueue;
1214         do {
1215           if (!nextQueue())
1216             return null;
1217         } while (currentQueue != lastQueue && !currentQueue.isAvailable());
1218 
1219         currentQuantum = calculateQuantum(currentQueue) - 1;
1220       }
1221       return currentQueue;
1222     }
1223 
1224     private boolean nextQueue() {
1225       if (currentQueue == null) return false;
1226       currentQueue = currentQueue.iterNext;
1227       return currentQueue != null;
1228     }
1229 
1230     private void setNextQueue(Queue<T> queue) {
1231       currentQueue = queue;
1232       if (queue != null) {
1233         currentQuantum = calculateQuantum(currentQueue);
1234       } else {
1235         currentQuantum = 0;
1236       }
1237     }
1238 
1239     private int calculateQuantum(final Queue queue) {
1240       return Math.max(1, queue.getPriority() * quantum); // TODO
1241     }
1242   }
1243 
1244   private static class AvlTree {
1245     public static <T extends Comparable<T>> Queue<T> get(Queue<T> root, T key) {
1246       while (root != null) {
1247         int cmp = root.compareKey(key);
1248         if (cmp > 0) {
1249           root = root.avlLeft;
1250         } else if (cmp < 0) {
1251           root = root.avlRight;
1252         } else {
1253           return root;
1254         }
1255       }
1256       return null;
1257     }
1258 
1259     public static <T extends Comparable<T>> Queue<T> getFirst(Queue<T> root) {
1260       if (root != null) {
1261         while (root.avlLeft != null) {
1262           root = root.avlLeft;
1263         }
1264       }
1265       return root;
1266     }
1267 
1268     public static <T extends Comparable<T>> Queue<T> getLast(Queue<T> root) {
1269       if (root != null) {
1270         while (root.avlRight != null) {
1271           root = root.avlRight;
1272         }
1273       }
1274       return root;
1275     }
1276 
1277     public static <T extends Comparable<T>> Queue<T> insert(Queue<T> root, Queue<T> node) {
1278       if (root == null) return node;
1279       if (node.compareTo(root) < 0) {
1280         root.avlLeft = insert(root.avlLeft, node);
1281       } else {
1282         root.avlRight = insert(root.avlRight, node);
1283       }
1284       return balance(root);
1285     }
1286 
1287     private static <T extends Comparable<T>> Queue<T> removeMin(Queue<T> p) {
1288       if (p.avlLeft == null)
1289         return p.avlRight;
1290       p.avlLeft = removeMin(p.avlLeft);
1291       return balance(p);
1292     }
1293 
1294     public static <T extends Comparable<T>> Queue<T> remove(Queue<T> root, T key) {
1295       if (root == null) return null;
1296 
1297       int cmp = root.compareKey(key);
1298       if (cmp == 0) {
1299         Queue<T> q = root.avlLeft;
1300         Queue<T> r = root.avlRight;
1301         if (r == null) return q;
1302         Queue<T> min = getFirst(r);
1303         min.avlRight = removeMin(r);
1304         min.avlLeft = q;
1305         return balance(min);
1306       } else if (cmp > 0) {
1307         root.avlLeft = remove(root.avlLeft, key);
1308       } else /* if (cmp < 0) */ {
1309         root.avlRight = remove(root.avlRight, key);
1310       }
1311       return balance(root);
1312     }
1313 
1314     private static <T extends Comparable<T>> Queue<T> balance(Queue<T> p) {
1315       fixHeight(p);
1316       int balance = balanceFactor(p);
1317       if (balance == 2) {
1318         if (balanceFactor(p.avlRight) < 0) {
1319           p.avlRight = rotateRight(p.avlRight);
1320         }
1321         return rotateLeft(p);
1322       } else if (balance == -2) {
1323         if (balanceFactor(p.avlLeft) > 0) {
1324           p.avlLeft = rotateLeft(p.avlLeft);
1325         }
1326         return rotateRight(p);
1327       }
1328       return p;
1329     }
1330 
1331     private static <T extends Comparable<T>> Queue<T> rotateRight(Queue<T> p) {
1332       Queue<T> q = p.avlLeft;
1333       p.avlLeft = q.avlRight;
1334       q.avlRight = p;
1335       fixHeight(p);
1336       fixHeight(q);
1337       return q;
1338     }
1339 
1340     private static <T extends Comparable<T>> Queue<T> rotateLeft(Queue<T> q) {
1341       Queue<T> p = q.avlRight;
1342       q.avlRight = p.avlLeft;
1343       p.avlLeft = q;
1344       fixHeight(q);
1345       fixHeight(p);
1346       return p;
1347     }
1348 
1349     private static <T extends Comparable<T>> void fixHeight(Queue<T> node) {
1350       int heightLeft = height(node.avlLeft);
1351       int heightRight = height(node.avlRight);
1352       node.avlHeight = 1 + Math.max(heightLeft, heightRight);
1353     }
1354 
1355     private static <T extends Comparable<T>> int height(Queue<T> node) {
1356       return node != null ? node.avlHeight : 0;
1357     }
1358 
1359     private static <T extends Comparable<T>> int balanceFactor(Queue<T> node) {
1360       return height(node.avlRight) - height(node.avlLeft);
1361     }
1362   }
1363 
1364   private static class IterableList {
1365     public static <T extends Comparable<T>> Queue<T> prepend(Queue<T> head, Queue<T> node) {
1366       assert !isLinked(node) : node + " is already linked";
1367       if (head != null) {
1368         Queue<T> tail = head.iterPrev;
1369         tail.iterNext = node;
1370         head.iterPrev = node;
1371         node.iterNext = head;
1372         node.iterPrev = tail;
1373       } else {
1374         node.iterNext = node;
1375         node.iterPrev = node;
1376       }
1377       return node;
1378     }
1379 
1380     public static <T extends Comparable<T>> Queue<T> append(Queue<T> head, Queue<T> node) {
1381       assert !isLinked(node) : node + " is already linked";
1382       if (head != null) {
1383         Queue<T> tail = head.iterPrev;
1384         tail.iterNext = node;
1385         node.iterNext = head;
1386         node.iterPrev = tail;
1387         head.iterPrev = node;
1388         return head;
1389       }
1390       node.iterNext = node;
1391       node.iterPrev = node;
1392       return node;
1393     }
1394 
1395     public static <T extends Comparable<T>> Queue<T> appendList(Queue<T> head, Queue<T> otherHead) {
1396       if (head == null) return otherHead;
1397       if (otherHead == null) return head;
1398 
1399       Queue<T> tail = head.iterPrev;
1400       Queue<T> otherTail = otherHead.iterPrev;
1401       tail.iterNext = otherHead;
1402       otherHead.iterPrev = tail;
1403       otherTail.iterNext = head;
1404       head.iterPrev = otherTail;
1405       return head;
1406     }
1407 
1408     private static <T extends Comparable<T>> Queue<T> remove(Queue<T> head, Queue<T> node) {
1409       assert isLinked(node) : node + " is not linked";
1410       if (node != node.iterNext) {
1411         node.iterPrev.iterNext = node.iterNext;
1412         node.iterNext.iterPrev = node.iterPrev;
1413         head = (head == node) ? node.iterNext : head;
1414       } else {
1415         head = null;
1416       }
1417       node.iterNext = null;
1418       node.iterPrev = null;
1419       return head;
1420     }
1421 
1422     private static <T extends Comparable<T>> boolean isLinked(Queue<T> node) {
1423       return node.iterPrev != null && node.iterNext != null;
1424     }
1425   }
1426 }