1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
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
43
44
45
46
47
48
49
50
51
52
53
54
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
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
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
114
115
116
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
159
160
161 pollResult = doPoll(serverRunQueue);
162 if (pollResult == null) {
163 pollResult = doPoll(tableRunQueue);
164 }
165
166
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
195 schedLock.lock();
196 try {
197
198 for (int i = 0; i < serverBuckets.length; ++i) {
199 clear((ServerQueue)serverBuckets[i], serverRunQueue);
200 serverBuckets[i] = null;
201 }
202
203
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
251 tableDeleted = !(procEx instanceof TableExistsException);
252 } else {
253
254 tableDeleted = (procEx instanceof TableNotFoundException);
255 }
256 } else {
257
258 tableDeleted = (iProcTable.getTableOperationType() == TableOperationType.DELETE);
259 }
260 if (tableDeleted) {
261 markTableAsDeleted(iProcTable.getTableName());
262 return;
263 }
264 } else {
265
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
286
287 public long getPollCalls() {
288 return pollCalls;
289 }
290
291 public long getNullPollCalls() {
292 return nullPollCalls;
293 }
294
295
296
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
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
317
318
319
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
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
356 ServerQueue queue = getServerQueue(serverName);
357 if (queue.isSuspended()) return;
358
359
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
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
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
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
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
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
608
609
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
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
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
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
694
695
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
735
736
737
738
739
740
741
742
743 public boolean tryAcquireTableExclusiveLock(final Procedure procedure, final TableName table) {
744 try {
745 schedLock.lock();
746 TableQueue queue = getTableQueue(table);
747 if (!queue.getNamespaceQueue().trySharedLock()) {
748 schedLock.unlock();
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
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 } finally {
772 if(schedLock.isHeldByCurrentThread()){
773 schedLock.unlock();
774 }
775 }
776 }
777
778
779
780
781
782
783 public void releaseTableExclusiveLock(final Procedure procedure, final TableName table) {
784 try {
785 schedLock.lock();
786 TableQueue queue = getTableQueue(table);
787 schedLock.unlock();
788
789
790 queue.releaseZkExclusiveLock(lockManager);
791
792 schedLock.lock();
793 queue.releaseExclusiveLock();
794 queue.getNamespaceQueue().releaseSharedLock();
795 addToRunQueue(tableRunQueue, queue);
796 schedLock.unlock();
797 } finally {
798 if(schedLock.isHeldByCurrentThread()) {
799 schedLock.unlock();
800 }
801 }
802 }
803
804
805
806
807
808
809
810
811 public boolean tryAcquireTableSharedLock(final Procedure procedure, final TableName table) {
812 return tryAcquireTableQueueSharedLock(procedure, table) != null;
813 }
814
815 private TableQueue tryAcquireTableQueueSharedLock(final Procedure procedure,
816 final TableName table) {
817 try {
818 schedLock.lock();
819 TableQueue queue = getTableQueue(table);
820 if (!queue.getNamespaceQueue().trySharedLock()) {
821 return null;
822 }
823
824 if (!queue.trySharedLock()) {
825 queue.getNamespaceQueue().releaseSharedLock();
826 schedLock.unlock();
827 return null;
828 }
829
830 schedLock.unlock();
831
832
833 if (!queue.tryZkSharedLock(lockManager, procedure.toString())) {
834 schedLock.lock();
835 queue.releaseSharedLock();
836 queue.getNamespaceQueue().releaseSharedLock();
837 schedLock.unlock();
838 return null;
839 }
840 return queue;
841 } finally {
842 if(schedLock.isHeldByCurrentThread()) {
843 schedLock.unlock();
844 }
845 }
846 }
847
848
849
850
851
852
853 public void releaseTableSharedLock(final Procedure procedure, final TableName table) {
854 final TableQueue queue = getTableQueueWithLock(table);
855
856
857 queue.releaseZkSharedLock(lockManager);
858 try {
859 schedLock.lock();
860 queue.releaseSharedLock();
861 queue.getNamespaceQueue().releaseSharedLock();
862 schedLock.unlock();
863 } finally {
864 if(schedLock.isHeldByCurrentThread()) {
865 schedLock.unlock();
866 }
867 }
868 }
869
870
871
872
873
874
875
876
877
878 protected boolean markTableAsDeleted(final TableName table) {
879 final ReentrantLock l = schedLock;
880 l.lock();
881 try {
882 TableQueue queue = getTableQueue(table);
883 if (queue == null) return true;
884
885 if (queue.isEmpty() && queue.tryExclusiveLock(0)) {
886
887 if (IterableList.isLinked(queue)) {
888 tableRunQueue.remove(queue);
889 }
890
891
892 try {
893 lockManager.tableDeleted(table);
894 } catch (IOException e) {
895 LOG.warn("Received exception from TableLockManager.tableDeleted:", e);
896 }
897
898 removeTableQueue(table);
899 } else {
900
901 return false;
902 }
903 } finally {
904 l.unlock();
905 }
906 return true;
907 }
908
909
910
911
912
913
914
915
916
917
918
919 public boolean tryAcquireNamespaceExclusiveLock(final Procedure procedure, final String nsName) {
920 schedLock.lock();
921 try {
922 TableQueue tableQueue = getTableQueue(TableName.NAMESPACE_TABLE_NAME);
923 if (!tableQueue.trySharedLock()) return false;
924
925 NamespaceQueue nsQueue = getNamespaceQueue(nsName);
926 boolean hasLock = nsQueue.tryExclusiveLock(procedure.getProcId());
927 if (!hasLock) {
928 tableQueue.releaseSharedLock();
929 }
930 return hasLock;
931 } finally {
932 schedLock.unlock();
933 }
934 }
935
936
937
938
939
940
941
942 public void releaseNamespaceExclusiveLock(final Procedure procedure, final String nsName) {
943 schedLock.lock();
944 try {
945 TableQueue tableQueue = getTableQueue(TableName.NAMESPACE_TABLE_NAME);
946 tableQueue.releaseSharedLock();
947
948 NamespaceQueue queue = getNamespaceQueue(nsName);
949 queue.releaseExclusiveLock();
950 } finally {
951 schedLock.unlock();
952 }
953 }
954
955
956
957
958
959
960
961
962
963
964
965 public boolean tryAcquireServerExclusiveLock(final Procedure procedure,
966 final ServerName serverName) {
967 schedLock.lock();
968 try {
969 ServerQueue queue = getServerQueue(serverName);
970 if (queue.tryExclusiveLock(procedure.getProcId())) {
971 removeFromRunQueue(serverRunQueue, queue);
972 return true;
973 }
974 } finally {
975 schedLock.unlock();
976 }
977 return false;
978 }
979
980
981
982
983
984
985
986 public void releaseServerExclusiveLock(final Procedure procedure,
987 final ServerName serverName) {
988 schedLock.lock();
989 try {
990 ServerQueue queue = getServerQueue(serverName);
991 queue.releaseExclusiveLock();
992 addToRunQueue(serverRunQueue, queue);
993 } finally {
994 schedLock.unlock();
995 }
996 }
997
998
999
1000
1001
1002
1003
1004
1005 public boolean tryAcquireServerSharedLock(final Procedure procedure,
1006 final ServerName serverName) {
1007 return getServerQueueWithLock(serverName).trySharedLock();
1008 }
1009
1010
1011
1012
1013
1014
1015
1016 public void releaseServerSharedLock(final Procedure procedure,
1017 final ServerName serverName) {
1018 getServerQueueWithLock(serverName).releaseSharedLock();
1019 }
1020
1021
1022
1023
1024 private static interface QueueInterface {
1025 boolean isAvailable();
1026 boolean isEmpty();
1027 int size();
1028
1029 void add(Procedure proc, boolean addFront);
1030 boolean requireExclusiveLock(Procedure proc);
1031 Procedure peek();
1032 Procedure poll();
1033
1034 boolean isSuspended();
1035 }
1036
1037 private static abstract class Queue<TKey extends Comparable<TKey>> implements QueueInterface {
1038 private Queue<TKey> avlRight = null;
1039 private Queue<TKey> avlLeft = null;
1040 private int avlHeight = 1;
1041
1042 private Queue<TKey> iterNext = null;
1043 private Queue<TKey> iterPrev = null;
1044 private boolean suspended = false;
1045
1046 private long exclusiveLockProcIdOwner = Long.MIN_VALUE;
1047 private int sharedLock = 0;
1048
1049 private final TKey key;
1050 private final int priority;
1051
1052 public Queue(TKey key) {
1053 this(key, 1);
1054 }
1055
1056 public Queue(TKey key, int priority) {
1057 this.key = key;
1058 this.priority = priority;
1059 }
1060
1061 protected TKey getKey() {
1062 return key;
1063 }
1064
1065 protected int getPriority() {
1066 return priority;
1067 }
1068
1069
1070
1071
1072 public boolean isSuspended() {
1073 return suspended;
1074 }
1075
1076 protected boolean setSuspended(boolean isSuspended) {
1077 if (this.suspended == isSuspended) return false;
1078 this.suspended = isSuspended;
1079 return true;
1080 }
1081
1082
1083
1084
1085 public synchronized boolean isLocked() {
1086 return hasExclusiveLock() || sharedLock > 0;
1087 }
1088
1089 public synchronized boolean hasExclusiveLock() {
1090 return this.exclusiveLockProcIdOwner != Long.MIN_VALUE;
1091 }
1092
1093 public synchronized boolean trySharedLock() {
1094 if (hasExclusiveLock()) return false;
1095 sharedLock++;
1096 return true;
1097 }
1098
1099 public synchronized void releaseSharedLock() {
1100 sharedLock--;
1101 }
1102
1103 protected synchronized boolean isSingleSharedLock() {
1104 return sharedLock == 1;
1105 }
1106
1107 public synchronized boolean tryExclusiveLock(long procIdOwner) {
1108 assert procIdOwner != Long.MIN_VALUE;
1109 if (isLocked()) return false;
1110 exclusiveLockProcIdOwner = procIdOwner;
1111 return true;
1112 }
1113
1114 public synchronized void releaseExclusiveLock() {
1115 exclusiveLockProcIdOwner = Long.MIN_VALUE;
1116 }
1117
1118
1119
1120 public synchronized boolean isAvailable() {
1121 return !hasExclusiveLock() && !isEmpty();
1122 }
1123
1124
1125
1126
1127 public int compareKey(TKey cmpKey) {
1128 return key.compareTo(cmpKey);
1129 }
1130
1131 public int compareTo(Queue<TKey> other) {
1132 return compareKey(other.key);
1133 }
1134
1135 @Override
1136 public String toString() {
1137 return String.format("%s(%s)", getClass().getSimpleName(), key);
1138 }
1139 }
1140
1141
1142
1143
1144 private static abstract class QueueImpl<TKey extends Comparable<TKey>> extends Queue<TKey> {
1145 private final ArrayDeque<Procedure> runnables = new ArrayDeque<Procedure>();
1146
1147 public QueueImpl(TKey key) {
1148 super(key);
1149 }
1150
1151 public QueueImpl(TKey key, int priority) {
1152 super(key, priority);
1153 }
1154
1155 public void add(final Procedure proc, final boolean addToFront) {
1156 if (addToFront) {
1157 addFront(proc);
1158 } else {
1159 addBack(proc);
1160 }
1161 }
1162
1163 protected void addFront(final Procedure proc) {
1164 runnables.addFirst(proc);
1165 }
1166
1167 protected void addBack(final Procedure proc) {
1168 runnables.addLast(proc);
1169 }
1170
1171 public Procedure peek() {
1172 return runnables.peek();
1173 }
1174
1175 @Override
1176 public Procedure poll() {
1177 return runnables.poll();
1178 }
1179
1180 @Override
1181 public boolean isEmpty() {
1182 return runnables.isEmpty();
1183 }
1184
1185 public int size() {
1186 return runnables.size();
1187 }
1188 }
1189
1190 private static class FairQueue<T extends Comparable<T>> {
1191 private final int quantum;
1192
1193 private Queue<T> currentQueue = null;
1194 private Queue<T> queueHead = null;
1195 private int currentQuantum = 0;
1196
1197 public FairQueue() {
1198 this(1);
1199 }
1200
1201 public FairQueue(int quantum) {
1202 this.quantum = quantum;
1203 }
1204
1205 public void add(Queue<T> queue) {
1206 queueHead = IterableList.append(queueHead, queue);
1207 if (currentQueue == null) setNextQueue(queueHead);
1208 }
1209
1210 public void remove(Queue<T> queue) {
1211 Queue<T> nextQueue = queue.iterNext;
1212 queueHead = IterableList.remove(queueHead, queue);
1213 if (currentQueue == queue) {
1214 setNextQueue(queueHead != null ? nextQueue : null);
1215 }
1216 }
1217
1218 public Queue<T> poll() {
1219 if (currentQuantum == 0) {
1220 if (!nextQueue()) {
1221 return null;
1222 }
1223 currentQuantum = calculateQuantum(currentQueue) - 1;
1224 } else {
1225 currentQuantum--;
1226 }
1227
1228
1229 if (!currentQueue.isAvailable()) {
1230 Queue<T> lastQueue = currentQueue;
1231 do {
1232 if (!nextQueue())
1233 return null;
1234 } while (currentQueue != lastQueue && !currentQueue.isAvailable());
1235
1236 currentQuantum = calculateQuantum(currentQueue) - 1;
1237 }
1238 return currentQueue;
1239 }
1240
1241 private boolean nextQueue() {
1242 if (currentQueue == null) return false;
1243 currentQueue = currentQueue.iterNext;
1244 return currentQueue != null;
1245 }
1246
1247 private void setNextQueue(Queue<T> queue) {
1248 currentQueue = queue;
1249 if (queue != null) {
1250 currentQuantum = calculateQuantum(currentQueue);
1251 } else {
1252 currentQuantum = 0;
1253 }
1254 }
1255
1256 private int calculateQuantum(final Queue queue) {
1257 return Math.max(1, queue.getPriority() * quantum);
1258 }
1259 }
1260
1261 private static class AvlTree {
1262 public static <T extends Comparable<T>> Queue<T> get(Queue<T> root, T key) {
1263 while (root != null) {
1264 int cmp = root.compareKey(key);
1265 if (cmp > 0) {
1266 root = root.avlLeft;
1267 } else if (cmp < 0) {
1268 root = root.avlRight;
1269 } else {
1270 return root;
1271 }
1272 }
1273 return null;
1274 }
1275
1276 public static <T extends Comparable<T>> Queue<T> getFirst(Queue<T> root) {
1277 if (root != null) {
1278 while (root.avlLeft != null) {
1279 root = root.avlLeft;
1280 }
1281 }
1282 return root;
1283 }
1284
1285 public static <T extends Comparable<T>> Queue<T> getLast(Queue<T> root) {
1286 if (root != null) {
1287 while (root.avlRight != null) {
1288 root = root.avlRight;
1289 }
1290 }
1291 return root;
1292 }
1293
1294 public static <T extends Comparable<T>> Queue<T> insert(Queue<T> root, Queue<T> node) {
1295 if (root == null) return node;
1296 if (node.compareTo(root) < 0) {
1297 root.avlLeft = insert(root.avlLeft, node);
1298 } else {
1299 root.avlRight = insert(root.avlRight, node);
1300 }
1301 return balance(root);
1302 }
1303
1304 private static <T extends Comparable<T>> Queue<T> removeMin(Queue<T> p) {
1305 if (p.avlLeft == null)
1306 return p.avlRight;
1307 p.avlLeft = removeMin(p.avlLeft);
1308 return balance(p);
1309 }
1310
1311 public static <T extends Comparable<T>> Queue<T> remove(Queue<T> root, T key) {
1312 if (root == null) return null;
1313
1314 int cmp = root.compareKey(key);
1315 if (cmp == 0) {
1316 Queue<T> q = root.avlLeft;
1317 Queue<T> r = root.avlRight;
1318 if (r == null) return q;
1319 Queue<T> min = getFirst(r);
1320 min.avlRight = removeMin(r);
1321 min.avlLeft = q;
1322 return balance(min);
1323 } else if (cmp > 0) {
1324 root.avlLeft = remove(root.avlLeft, key);
1325 } else
1326 root.avlRight = remove(root.avlRight, key);
1327 }
1328 return balance(root);
1329 }
1330
1331 private static <T extends Comparable<T>> Queue<T> balance(Queue<T> p) {
1332 fixHeight(p);
1333 int balance = balanceFactor(p);
1334 if (balance == 2) {
1335 if (balanceFactor(p.avlRight) < 0) {
1336 p.avlRight = rotateRight(p.avlRight);
1337 }
1338 return rotateLeft(p);
1339 } else if (balance == -2) {
1340 if (balanceFactor(p.avlLeft) > 0) {
1341 p.avlLeft = rotateLeft(p.avlLeft);
1342 }
1343 return rotateRight(p);
1344 }
1345 return p;
1346 }
1347
1348 private static <T extends Comparable<T>> Queue<T> rotateRight(Queue<T> p) {
1349 Queue<T> q = p.avlLeft;
1350 p.avlLeft = q.avlRight;
1351 q.avlRight = p;
1352 fixHeight(p);
1353 fixHeight(q);
1354 return q;
1355 }
1356
1357 private static <T extends Comparable<T>> Queue<T> rotateLeft(Queue<T> q) {
1358 Queue<T> p = q.avlRight;
1359 q.avlRight = p.avlLeft;
1360 p.avlLeft = q;
1361 fixHeight(q);
1362 fixHeight(p);
1363 return p;
1364 }
1365
1366 private static <T extends Comparable<T>> void fixHeight(Queue<T> node) {
1367 int heightLeft = height(node.avlLeft);
1368 int heightRight = height(node.avlRight);
1369 node.avlHeight = 1 + Math.max(heightLeft, heightRight);
1370 }
1371
1372 private static <T extends Comparable<T>> int height(Queue<T> node) {
1373 return node != null ? node.avlHeight : 0;
1374 }
1375
1376 private static <T extends Comparable<T>> int balanceFactor(Queue<T> node) {
1377 return height(node.avlRight) - height(node.avlLeft);
1378 }
1379 }
1380
1381 private static class IterableList {
1382 public static <T extends Comparable<T>> Queue<T> prepend(Queue<T> head, Queue<T> node) {
1383 assert !isLinked(node) : node + " is already linked";
1384 if (head != null) {
1385 Queue<T> tail = head.iterPrev;
1386 tail.iterNext = node;
1387 head.iterPrev = node;
1388 node.iterNext = head;
1389 node.iterPrev = tail;
1390 } else {
1391 node.iterNext = node;
1392 node.iterPrev = node;
1393 }
1394 return node;
1395 }
1396
1397 public static <T extends Comparable<T>> Queue<T> append(Queue<T> head, Queue<T> node) {
1398 assert !isLinked(node) : node + " is already linked";
1399 if (head != null) {
1400 Queue<T> tail = head.iterPrev;
1401 tail.iterNext = node;
1402 node.iterNext = head;
1403 node.iterPrev = tail;
1404 head.iterPrev = node;
1405 return head;
1406 }
1407 node.iterNext = node;
1408 node.iterPrev = node;
1409 return node;
1410 }
1411
1412 public static <T extends Comparable<T>> Queue<T> appendList(Queue<T> head, Queue<T> otherHead) {
1413 if (head == null) return otherHead;
1414 if (otherHead == null) return head;
1415
1416 Queue<T> tail = head.iterPrev;
1417 Queue<T> otherTail = otherHead.iterPrev;
1418 tail.iterNext = otherHead;
1419 otherHead.iterPrev = tail;
1420 otherTail.iterNext = head;
1421 head.iterPrev = otherTail;
1422 return head;
1423 }
1424
1425 private static <T extends Comparable<T>> Queue<T> remove(Queue<T> head, Queue<T> node) {
1426 assert isLinked(node) : node + " is not linked";
1427 if (node != node.iterNext) {
1428 node.iterPrev.iterNext = node.iterNext;
1429 node.iterNext.iterPrev = node.iterPrev;
1430 head = (head == node) ? node.iterNext : head;
1431 } else {
1432 head = null;
1433 }
1434 node.iterNext = null;
1435 node.iterPrev = null;
1436 return head;
1437 }
1438
1439 private static <T extends Comparable<T>> boolean isLinked(Queue<T> node) {
1440 return node.iterPrev != null && node.iterNext != null;
1441 }
1442 }
1443 }