1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.apache.hadoop.hbase.procedure2.store.wal;
20
21 import java.io.IOException;
22
23 import org.apache.commons.logging.Log;
24 import org.apache.commons.logging.LogFactory;
25 import org.apache.hadoop.fs.FSDataInputStream;
26 import org.apache.hadoop.hbase.ProcedureInfo;
27 import org.apache.hadoop.hbase.classification.InterfaceAudience;
28 import org.apache.hadoop.hbase.classification.InterfaceStability;
29 import org.apache.hadoop.hbase.procedure2.Procedure;
30 import org.apache.hadoop.hbase.procedure2.store.ProcedureStore.ProcedureIterator;
31 import org.apache.hadoop.hbase.procedure2.store.ProcedureStoreTracker;
32 import org.apache.hadoop.hbase.protobuf.generated.ProcedureProtos;
33 import org.apache.hadoop.hbase.protobuf.generated.ProcedureProtos.ProcedureWALEntry;
34
35 import com.google.protobuf.InvalidProtocolBufferException;
36
37
38
39
40 @InterfaceAudience.Private
41 @InterfaceStability.Evolving
42 public class ProcedureWALFormatReader {
43 private static final Log LOG = LogFactory.getLog(ProcedureWALFormatReader.class);
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100 private final WalProcedureMap localProcedureMap = new WalProcedureMap(1024);
101 private final WalProcedureMap procedureMap = new WalProcedureMap(1024);
102
103
104 private long maxProcId = 0;
105
106 private final ProcedureStoreTracker tracker;
107 private final boolean hasFastStartSupport;
108
109 public ProcedureWALFormatReader(final ProcedureStoreTracker tracker) {
110 this.tracker = tracker;
111
112 this.hasFastStartSupport = !tracker.isEmpty();
113 }
114
115 public void read(ProcedureWALFile log, ProcedureWALFormat.Loader loader) throws IOException {
116 FSDataInputStream stream = log.getStream();
117 try {
118 boolean hasMore = true;
119 while (hasMore) {
120 ProcedureWALEntry entry = ProcedureWALFormat.readEntry(stream);
121 if (entry == null) {
122 LOG.warn("nothing left to decode. exiting with missing EOF");
123 hasMore = false;
124 break;
125 }
126 switch (entry.getType()) {
127 case INIT:
128 readInitEntry(entry);
129 break;
130 case INSERT:
131 readInsertEntry(entry);
132 break;
133 case UPDATE:
134 case COMPACT:
135 readUpdateEntry(entry);
136 break;
137 case DELETE:
138 readDeleteEntry(entry);
139 break;
140 case EOF:
141 hasMore = false;
142 break;
143 default:
144 throw new CorruptedWALProcedureStoreException("Invalid entry: " + entry);
145 }
146 }
147 } catch (InvalidProtocolBufferException e) {
148 LOG.error("got an exception while reading the procedure WAL: " + log, e);
149 loader.markCorruptedWAL(log, e);
150 }
151
152 if (!localProcedureMap.isEmpty()) {
153 log.setProcIds(localProcedureMap.getMinProcId(), localProcedureMap.getMaxProcId());
154 procedureMap.mergeTail(localProcedureMap);
155
156
157
158
159
160
161
162
163 }
164 }
165
166 public void finalize(ProcedureWALFormat.Loader loader) throws IOException {
167
168 loader.setMaxProcId(maxProcId);
169
170
171 ProcedureIterator procIter = procedureMap.fetchReady();
172 if (procIter != null) loader.load(procIter);
173
174
175
176 procIter = procedureMap.fetchAll();
177 if (procIter != null) loader.handleCorrupted(procIter);
178 }
179
180 private void loadProcedure(final ProcedureWALEntry entry, final ProcedureProtos.Procedure proc) {
181 maxProcId = Math.max(maxProcId, proc.getProcId());
182 if (isRequired(proc.getProcId())) {
183 if (LOG.isTraceEnabled()) {
184 LOG.trace("read " + entry.getType() + " entry " + proc.getProcId());
185 }
186 localProcedureMap.add(proc);
187 tracker.setDeleted(proc.getProcId(), false);
188 }
189 }
190
191 private void readInitEntry(final ProcedureWALEntry entry)
192 throws IOException {
193 assert entry.getProcedureCount() == 1 : "Expected only one procedure";
194 loadProcedure(entry, entry.getProcedure(0));
195 }
196
197 private void readInsertEntry(final ProcedureWALEntry entry) throws IOException {
198 assert entry.getProcedureCount() >= 1 : "Expected one or more procedures";
199 loadProcedure(entry, entry.getProcedure(0));
200 for (int i = 1; i < entry.getProcedureCount(); ++i) {
201 loadProcedure(entry, entry.getProcedure(i));
202 }
203 }
204
205 private void readUpdateEntry(final ProcedureWALEntry entry) throws IOException {
206 assert entry.getProcedureCount() == 1 : "Expected only one procedure";
207 loadProcedure(entry, entry.getProcedure(0));
208 }
209
210 private void readDeleteEntry(final ProcedureWALEntry entry) throws IOException {
211 assert entry.getProcedureCount() == 0 : "Expected no procedures";
212 assert entry.hasProcId() : "expected ProcID";
213 if (LOG.isTraceEnabled()) {
214 LOG.trace("read delete entry " + entry.getProcId());
215 }
216 maxProcId = Math.max(maxProcId, entry.getProcId());
217 localProcedureMap.remove(entry.getProcId());
218 assert !procedureMap.contains(entry.getProcId());
219 tracker.setDeleted(entry.getProcId(), true);
220 }
221
222 private boolean isDeleted(final long procId) {
223 return tracker.isDeleted(procId) == ProcedureStoreTracker.DeleteState.YES;
224 }
225
226 private boolean isRequired(final long procId) {
227 return !isDeleted(procId) && !procedureMap.contains(procId);
228 }
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246 private static class Entry {
247
248 protected Entry hashNext;
249
250 protected Entry childHead;
251
252 protected Entry linkNext;
253 protected Entry linkPrev;
254
255 protected Entry replayNext;
256 protected Entry replayPrev;
257
258 protected Procedure procedure;
259 protected ProcedureProtos.Procedure proto;
260 protected boolean ready = false;
261
262 public Entry(Entry hashNext) { this.hashNext = hashNext; }
263
264 public long getProcId() { return proto.getProcId(); }
265 public long getParentId() { return proto.getParentId(); }
266 public boolean hasParent() { return proto.hasParentId(); }
267 public boolean isReady() { return ready; }
268
269 public boolean isCompleted() {
270 if (!hasParent()) {
271 switch (proto.getState()) {
272 case ROLLEDBACK:
273 return true;
274 case FINISHED:
275 return !proto.hasException();
276 default:
277 break;
278 }
279 }
280 return false;
281 }
282
283 public Procedure convert() throws IOException {
284 if (procedure == null) {
285 procedure = Procedure.convert(proto);
286 }
287 return procedure;
288 }
289
290 public ProcedureInfo convertToInfo() {
291 return ProcedureInfo.convert(proto);
292 }
293
294 @Override
295 public String toString() {
296 return "Entry(" + getProcId() + ", parentId=" + getParentId() + ")";
297 }
298 }
299
300 private static class EntryIterator implements ProcedureIterator {
301 private final Entry replayHead;
302 private Entry current;
303
304 public EntryIterator(Entry replayHead) {
305 this.replayHead = replayHead;
306 this.current = replayHead;
307 }
308
309 @Override
310 public void reset() {
311 this.current = replayHead;
312 }
313
314 @Override
315 public boolean hasNext() {
316 return current != null;
317 }
318
319 @Override
320 public boolean isNextCompleted() {
321 return current != null && current.isCompleted();
322 }
323
324 @Override
325 public void skipNext() {
326 current = current.replayNext;
327 }
328
329 @Override
330 public Procedure nextAsProcedure() throws IOException {
331 try {
332 return current.convert();
333 } finally {
334 current = current.replayNext;
335 }
336 }
337
338 @Override
339 public ProcedureInfo nextAsProcedureInfo() {
340 try {
341 return current.convertToInfo();
342 } finally {
343 current = current.replayNext;
344 }
345 }
346 }
347
348 private static class WalProcedureMap {
349
350 private Entry[] procedureMap;
351
352
353 private Entry replayOrderHead;
354 private Entry replayOrderTail;
355
356
357 private Entry rootHead;
358
359
360 private Entry childUnlinkedHead;
361
362
363 private long minProcId = Long.MAX_VALUE;
364 private long maxProcId = Long.MIN_VALUE;
365
366 public WalProcedureMap(int size) {
367 procedureMap = new Entry[size];
368 replayOrderHead = null;
369 replayOrderTail = null;
370 rootHead = null;
371 childUnlinkedHead = null;
372 }
373
374 public void add(ProcedureProtos.Procedure procProto) {
375 trackProcIds(procProto.getProcId());
376 Entry entry = addToMap(procProto.getProcId(), procProto.hasParentId());
377 boolean isNew = entry.proto == null;
378 entry.proto = procProto;
379 addToReplayList(entry);
380
381 if (isNew) {
382 if (procProto.hasParentId()) {
383 childUnlinkedHead = addToLinkList(entry, childUnlinkedHead);
384 } else {
385 rootHead = addToLinkList(entry, rootHead);
386 }
387 }
388 }
389
390 public boolean remove(long procId) {
391 trackProcIds(procId);
392 Entry entry = removeFromMap(procId);
393 if (entry != null) {
394 unlinkFromReplayList(entry);
395 unlinkFromLinkList(entry);
396 return true;
397 }
398 return false;
399 }
400
401 private void trackProcIds(long procId) {
402 minProcId = Math.min(minProcId, procId);
403 maxProcId = Math.max(maxProcId, procId);
404 }
405
406 public long getMinProcId() {
407 return minProcId;
408 }
409
410 public long getMaxProcId() {
411 return maxProcId;
412 }
413
414 public boolean contains(long procId) {
415 return getProcedure(procId) != null;
416 }
417
418 public boolean isEmpty() {
419 return replayOrderHead == null;
420 }
421
422 public void clear() {
423 for (int i = 0; i < procedureMap.length; ++i) {
424 procedureMap[i] = null;
425 }
426 replayOrderHead = null;
427 replayOrderTail = null;
428 rootHead = null;
429 childUnlinkedHead = null;
430 minProcId = Long.MAX_VALUE;
431 maxProcId = Long.MIN_VALUE;
432 }
433
434
435
436
437
438
439
440
441
442
443
444 public void mergeTail(WalProcedureMap other) {
445 for (Entry p = other.replayOrderHead; p != null; p = p.replayNext) {
446 int slotIndex = getMapSlot(p.getProcId());
447 p.hashNext = procedureMap[slotIndex];
448 procedureMap[slotIndex] = p;
449 }
450
451 if (replayOrderHead == null) {
452 replayOrderHead = other.replayOrderHead;
453 replayOrderTail = other.replayOrderTail;
454 rootHead = other.rootHead;
455 childUnlinkedHead = other.childUnlinkedHead;
456 } else {
457
458 assert replayOrderTail.replayNext == null;
459 assert other.replayOrderHead.replayPrev == null;
460 replayOrderTail.replayNext = other.replayOrderHead;
461 other.replayOrderHead.replayPrev = replayOrderTail;
462 replayOrderTail = other.replayOrderTail;
463
464
465 if (rootHead == null) {
466 rootHead = other.rootHead;
467 } else if (other.rootHead != null) {
468 Entry otherTail = findLinkListTail(other.rootHead);
469 otherTail.linkNext = rootHead;
470 rootHead.linkPrev = otherTail;
471 rootHead = other.rootHead;
472 }
473
474
475 if (childUnlinkedHead == null) {
476 childUnlinkedHead = other.childUnlinkedHead;
477 } else if (other.childUnlinkedHead != null) {
478 Entry otherTail = findLinkListTail(other.childUnlinkedHead);
479 otherTail.linkNext = childUnlinkedHead;
480 childUnlinkedHead.linkPrev = otherTail;
481 childUnlinkedHead = other.childUnlinkedHead;
482 }
483 }
484
485 other.clear();
486 }
487
488
489
490
491
492
493 public EntryIterator fetchReady() {
494 buildGraph();
495
496 Entry readyHead = null;
497 Entry readyTail = null;
498 Entry p = replayOrderHead;
499 while (p != null) {
500 Entry next = p.replayNext;
501 if (p.isReady()) {
502 unlinkFromReplayList(p);
503 if (readyTail != null) {
504 readyTail.replayNext = p;
505 p.replayPrev = readyTail;
506 } else {
507 p.replayPrev = null;
508 readyHead = p;
509 }
510 readyTail = p;
511 p.replayNext = null;
512 }
513 p = next;
514 }
515
516
517 for (p = readyHead; p != null; p = p.replayNext) {
518 removeFromMap(p.getProcId());
519 unlinkFromLinkList(p);
520 }
521 return readyHead != null ? new EntryIterator(readyHead) : null;
522 }
523
524
525
526
527 public EntryIterator fetchAll() {
528 Entry head = replayOrderHead;
529 for (Entry p = head; p != null; p = p.replayNext) {
530 removeFromMap(p.getProcId());
531 }
532 for (int i = 0; i < procedureMap.length; ++i) {
533 assert procedureMap[i] == null : "map not empty i=" + i;
534 }
535 replayOrderHead = null;
536 replayOrderTail = null;
537 childUnlinkedHead = null;
538 rootHead = null;
539 return head != null ? new EntryIterator(head) : null;
540 }
541
542 private void buildGraph() {
543 Entry p = childUnlinkedHead;
544 while (p != null) {
545 Entry next = p.linkNext;
546 Entry rootProc = getRootProcedure(p);
547 if (rootProc != null) {
548 rootProc.childHead = addToLinkList(p, rootProc.childHead);
549 }
550 p = next;
551 }
552
553 for (p = rootHead; p != null; p = p.linkNext) {
554 checkReadyToRun(p);
555 }
556 }
557
558 private Entry getRootProcedure(Entry entry) {
559 while (entry != null && entry.hasParent()) {
560 entry = getProcedure(entry.getParentId());
561 }
562 return entry;
563 }
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604 private boolean checkReadyToRun(Entry rootEntry) {
605 int stackIdSum = 0;
606 int maxStackId = 0;
607 for (int i = 0; i < rootEntry.proto.getStackIdCount(); ++i) {
608 int stackId = 1 + rootEntry.proto.getStackId(i);
609 maxStackId = Math.max(maxStackId, stackId);
610 stackIdSum += stackId;
611 }
612
613 for (Entry p = rootEntry.childHead; p != null; p = p.linkNext) {
614 for (int i = 0; i < p.proto.getStackIdCount(); ++i) {
615 int stackId = 1 + p.proto.getStackId(i);
616 maxStackId = Math.max(maxStackId, stackId);
617 stackIdSum += stackId;
618 }
619 }
620 final int cmpStackIdSum = (maxStackId * (maxStackId + 1) / 2);
621 if (cmpStackIdSum == stackIdSum) {
622 rootEntry.ready = true;
623 for (Entry p = rootEntry.childHead; p != null; p = p.linkNext) {
624 p.ready = true;
625 }
626 return true;
627 }
628 return false;
629 }
630
631 private void unlinkFromReplayList(Entry entry) {
632 if (replayOrderHead == entry) {
633 replayOrderHead = entry.replayNext;
634 }
635 if (replayOrderTail == entry) {
636 replayOrderTail = entry.replayPrev;
637 }
638 if (entry.replayPrev != null) {
639 entry.replayPrev.replayNext = entry.replayNext;
640 }
641 if (entry.replayNext != null) {
642 entry.replayNext.replayPrev = entry.replayPrev;
643 }
644 }
645
646 private void addToReplayList(final Entry entry) {
647 unlinkFromReplayList(entry);
648 entry.replayNext = replayOrderHead;
649 entry.replayPrev = null;
650 if (replayOrderHead != null) {
651 replayOrderHead.replayPrev = entry;
652 } else {
653 replayOrderTail = entry;
654 }
655 replayOrderHead = entry;
656 }
657
658 private void unlinkFromLinkList(Entry entry) {
659 if (entry == rootHead) {
660 rootHead = entry.linkNext;
661 } else if (entry == childUnlinkedHead) {
662 childUnlinkedHead = entry.linkNext;
663 }
664 if (entry.linkPrev != null) {
665 entry.linkPrev.linkNext = entry.linkNext;
666 }
667 if (entry.linkNext != null) {
668 entry.linkNext.linkPrev = entry.linkPrev;
669 }
670 }
671
672 private Entry addToLinkList(Entry entry, Entry linkHead) {
673 unlinkFromLinkList(entry);
674 entry.linkNext = linkHead;
675 entry.linkPrev = null;
676 if (linkHead != null) {
677 linkHead.linkPrev = entry;
678 }
679 return entry;
680 }
681
682 private Entry findLinkListTail(Entry linkHead) {
683 Entry tail = linkHead;
684 while (tail.linkNext != null) {
685 tail = tail.linkNext;
686 }
687 return tail;
688 }
689
690 private Entry addToMap(final long procId, final boolean hasParent) {
691 int slotIndex = getMapSlot(procId);
692 Entry entry = getProcedure(slotIndex, procId);
693 if (entry != null) return entry;
694
695 entry = new Entry(procedureMap[slotIndex]);
696 procedureMap[slotIndex] = entry;
697 return entry;
698 }
699
700 private Entry removeFromMap(final long procId) {
701 int slotIndex = getMapSlot(procId);
702 Entry prev = null;
703 Entry entry = procedureMap[slotIndex];
704 while (entry != null) {
705 if (procId == entry.getProcId()) {
706 if (prev != null) {
707 prev.hashNext = entry.hashNext;
708 } else {
709 procedureMap[slotIndex] = entry.hashNext;
710 }
711 entry.hashNext = null;
712 return entry;
713 }
714 prev = entry;
715 entry = entry.hashNext;
716 }
717 return null;
718 }
719
720 private Entry getProcedure(final long procId) {
721 return getProcedure(getMapSlot(procId), procId);
722 }
723
724 private Entry getProcedure(final int slotIndex, final long procId) {
725 Entry entry = procedureMap[slotIndex];
726 while (entry != null) {
727 if (procId == entry.getProcId()) {
728 return entry;
729 }
730 entry = entry.hashNext;
731 }
732 return null;
733 }
734
735 private int getMapSlot(final long procId) {
736 return (int)(Procedure.getProcIdHashCode(procId) % procedureMap.length);
737 }
738 }
739 }