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 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
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
773
774
775
776 public void releaseTableExclusiveLock(final Procedure procedure, final TableName table) {
777 schedLock.lock();
778 TableQueue queue = getTableQueue(table);
779 schedLock.unlock();
780
781
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
793
794
795
796
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
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
831
832
833
834 public void releaseTableSharedLock(final Procedure procedure, final TableName table) {
835 final TableQueue queue = getTableQueueWithLock(table);
836
837
838 queue.releaseZkSharedLock(lockManager);
839
840 schedLock.lock();
841 queue.releaseSharedLock();
842 queue.getNamespaceQueue().releaseSharedLock();
843 schedLock.unlock();
844 }
845
846
847
848
849
850
851
852
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
863 if (IterableList.isLinked(queue)) {
864 tableRunQueue.remove(queue);
865 }
866
867
868 try {
869 lockManager.tableDeleted(table);
870 } catch (IOException e) {
871 LOG.warn("Received exception from TableLockManager.tableDeleted:", e);
872 }
873
874 removeTableQueue(table);
875 } else {
876
877 return false;
878 }
879 } finally {
880 l.unlock();
881 }
882 return true;
883 }
884
885
886
887
888
889
890
891
892
893
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
914
915
916
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
933
934
935
936
937
938
939
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
958
959
960
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
976
977
978
979
980
981 public boolean tryAcquireServerSharedLock(final Procedure procedure,
982 final ServerName serverName) {
983 return getServerQueueWithLock(serverName).trySharedLock();
984 }
985
986
987
988
989
990
991
992 public void releaseServerSharedLock(final Procedure procedure,
993 final ServerName serverName) {
994 getServerQueueWithLock(serverName).releaseSharedLock();
995 }
996
997
998
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
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
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
1095
1096 public synchronized boolean isAvailable() {
1097 return !hasExclusiveLock() && !isEmpty();
1098 }
1099
1100
1101
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
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;
1198 }
1199 currentQuantum = calculateQuantum(currentQueue) - 1;
1200 } else {
1201 currentQuantum--;
1202 }
1203
1204
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);
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
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 }