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 long count = 0;
117 FSDataInputStream stream = log.getStream();
118 try {
119 boolean hasMore = true;
120 while (hasMore) {
121 ProcedureWALEntry entry = ProcedureWALFormat.readEntry(stream);
122 if (entry == null) {
123 LOG.warn("nothing left to decode. exiting with missing EOF");
124 hasMore = false;
125 break;
126 }
127 count++;
128 switch (entry.getType()) {
129 case PROCEDURE_WAL_INIT:
130 readInitEntry(entry);
131 break;
132 case PROCEDURE_WAL_INSERT:
133 readInsertEntry(entry);
134 break;
135 case PROCEDURE_WAL_UPDATE:
136 case PROCEDURE_WAL_COMPACT:
137 readUpdateEntry(entry);
138 break;
139 case PROCEDURE_WAL_DELETE:
140 readDeleteEntry(entry);
141 break;
142 case PROCEDURE_WAL_EOF:
143 hasMore = false;
144 break;
145 default:
146 throw new CorruptedWALProcedureStoreException("Invalid entry: " + entry);
147 }
148 }
149 LOG.info("Read " + count + " entries in " + log);
150 } catch (InvalidProtocolBufferException e) {
151 LOG.error("While reading entry #" + count + " in " + log, e);
152 loader.markCorruptedWAL(log, e);
153 }
154
155 if (!localProcedureMap.isEmpty()) {
156 log.setProcIds(localProcedureMap.getMinProcId(), localProcedureMap.getMaxProcId());
157 procedureMap.mergeTail(localProcedureMap);
158
159
160
161
162
163
164
165
166 }
167 }
168
169 public void finalize(ProcedureWALFormat.Loader loader) throws IOException {
170
171 loader.setMaxProcId(maxProcId);
172
173
174 ProcedureIterator procIter = procedureMap.fetchReady();
175 if (procIter != null) loader.load(procIter);
176
177
178
179 procIter = procedureMap.fetchAll();
180 if (procIter != null) loader.handleCorrupted(procIter);
181 }
182
183 private void loadProcedure(final ProcedureWALEntry entry, final ProcedureProtos.Procedure proc) {
184 maxProcId = Math.max(maxProcId, proc.getProcId());
185 if (isRequired(proc.getProcId())) {
186 if (LOG.isTraceEnabled()) {
187 LOG.trace("read " + entry.getType() + " entry " + proc.getProcId());
188 }
189 localProcedureMap.add(proc);
190 tracker.setDeleted(proc.getProcId(), false);
191 }
192 }
193
194 private void readInitEntry(final ProcedureWALEntry entry)
195 throws IOException {
196 assert entry.getProcedureCount() == 1 : "Expected only one procedure";
197 loadProcedure(entry, entry.getProcedure(0));
198 }
199
200 private void readInsertEntry(final ProcedureWALEntry entry) throws IOException {
201 assert entry.getProcedureCount() >= 1 : "Expected one or more procedures";
202 loadProcedure(entry, entry.getProcedure(0));
203 for (int i = 1; i < entry.getProcedureCount(); ++i) {
204 loadProcedure(entry, entry.getProcedure(i));
205 }
206 }
207
208 private void readUpdateEntry(final ProcedureWALEntry entry) throws IOException {
209 assert entry.getProcedureCount() == 1 : "Expected only one procedure";
210 loadProcedure(entry, entry.getProcedure(0));
211 }
212
213 private void readDeleteEntry(final ProcedureWALEntry entry) throws IOException {
214 assert entry.getProcedureCount() == 0 : "Expected no procedures";
215 assert entry.hasProcId() : "expected ProcID";
216 if (LOG.isTraceEnabled()) {
217 LOG.trace("read delete entry " + entry.getProcId());
218 }
219 maxProcId = Math.max(maxProcId, entry.getProcId());
220 localProcedureMap.remove(entry.getProcId());
221 assert !procedureMap.contains(entry.getProcId());
222 tracker.setDeleted(entry.getProcId(), true);
223 }
224
225 private boolean isDeleted(final long procId) {
226 return tracker.isDeleted(procId) == ProcedureStoreTracker.DeleteState.YES;
227 }
228
229 private boolean isRequired(final long procId) {
230 return !isDeleted(procId) && !procedureMap.contains(procId);
231 }
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249 private static class Entry {
250
251 protected Entry hashNext;
252
253 protected Entry childHead;
254
255 protected Entry linkNext;
256 protected Entry linkPrev;
257
258 protected Entry replayNext;
259 protected Entry replayPrev;
260
261 protected Procedure procedure;
262 protected ProcedureProtos.Procedure proto;
263 protected boolean ready = false;
264
265 public Entry(Entry hashNext) { this.hashNext = hashNext; }
266
267 public long getProcId() { return proto.getProcId(); }
268 public long getParentId() { return proto.getParentId(); }
269 public boolean hasParent() { return proto.hasParentId(); }
270 public boolean isReady() { return ready; }
271
272 public boolean isCompleted() {
273 if (!hasParent()) {
274 switch (proto.getState()) {
275 case ROLLEDBACK:
276 return true;
277 case FINISHED:
278 return !proto.hasException();
279 default:
280 break;
281 }
282 }
283 return false;
284 }
285
286 public Procedure convert() throws IOException {
287 if (procedure == null) {
288 procedure = Procedure.convert(proto);
289 }
290 return procedure;
291 }
292
293 public ProcedureInfo convertToInfo() {
294 return ProcedureInfo.convert(proto);
295 }
296
297 @Override
298 public String toString() {
299 return "Entry(" + getProcId() + ", parentId=" + getParentId() + ")";
300 }
301 }
302
303 private static class EntryIterator implements ProcedureIterator {
304 private final Entry replayHead;
305 private Entry current;
306
307 public EntryIterator(Entry replayHead) {
308 this.replayHead = replayHead;
309 this.current = replayHead;
310 }
311
312 @Override
313 public void reset() {
314 this.current = replayHead;
315 }
316
317 @Override
318 public boolean hasNext() {
319 return current != null;
320 }
321
322 @Override
323 public boolean isNextCompleted() {
324 return current != null && current.isCompleted();
325 }
326
327 @Override
328 public void skipNext() {
329 current = current.replayNext;
330 }
331
332 @Override
333 public Procedure nextAsProcedure() throws IOException {
334 try {
335 return current.convert();
336 } finally {
337 current = current.replayNext;
338 }
339 }
340
341 @Override
342 public ProcedureInfo nextAsProcedureInfo() {
343 try {
344 return current.convertToInfo();
345 } finally {
346 current = current.replayNext;
347 }
348 }
349 }
350
351 private static class WalProcedureMap {
352
353 private Entry[] procedureMap;
354
355
356 private Entry replayOrderHead;
357 private Entry replayOrderTail;
358
359
360 private Entry rootHead;
361
362
363 private Entry childUnlinkedHead;
364
365
366 private long minProcId = Long.MAX_VALUE;
367 private long maxProcId = Long.MIN_VALUE;
368
369 public WalProcedureMap(int size) {
370 procedureMap = new Entry[size];
371 replayOrderHead = null;
372 replayOrderTail = null;
373 rootHead = null;
374 childUnlinkedHead = null;
375 }
376
377 public void add(ProcedureProtos.Procedure procProto) {
378 trackProcIds(procProto.getProcId());
379 Entry entry = addToMap(procProto.getProcId(), procProto.hasParentId());
380 boolean isNew = entry.proto == null;
381 entry.proto = procProto;
382 addToReplayList(entry);
383
384 if (isNew) {
385 if (procProto.hasParentId()) {
386 childUnlinkedHead = addToLinkList(entry, childUnlinkedHead);
387 } else {
388 rootHead = addToLinkList(entry, rootHead);
389 }
390 }
391 }
392
393 public boolean remove(long procId) {
394 trackProcIds(procId);
395 Entry entry = removeFromMap(procId);
396 if (entry != null) {
397 unlinkFromReplayList(entry);
398 unlinkFromLinkList(entry);
399 return true;
400 }
401 return false;
402 }
403
404 private void trackProcIds(long procId) {
405 minProcId = Math.min(minProcId, procId);
406 maxProcId = Math.max(maxProcId, procId);
407 }
408
409 public long getMinProcId() {
410 return minProcId;
411 }
412
413 public long getMaxProcId() {
414 return maxProcId;
415 }
416
417 public boolean contains(long procId) {
418 return getProcedure(procId) != null;
419 }
420
421 public boolean isEmpty() {
422 return replayOrderHead == null;
423 }
424
425 public void clear() {
426 for (int i = 0; i < procedureMap.length; ++i) {
427 procedureMap[i] = null;
428 }
429 replayOrderHead = null;
430 replayOrderTail = null;
431 rootHead = null;
432 childUnlinkedHead = null;
433 minProcId = Long.MAX_VALUE;
434 maxProcId = Long.MIN_VALUE;
435 }
436
437
438
439
440
441
442
443
444
445
446
447 public void mergeTail(WalProcedureMap other) {
448 for (Entry p = other.replayOrderHead; p != null; p = p.replayNext) {
449 int slotIndex = getMapSlot(p.getProcId());
450 p.hashNext = procedureMap[slotIndex];
451 procedureMap[slotIndex] = p;
452 }
453
454 if (replayOrderHead == null) {
455 replayOrderHead = other.replayOrderHead;
456 replayOrderTail = other.replayOrderTail;
457 rootHead = other.rootHead;
458 childUnlinkedHead = other.childUnlinkedHead;
459 } else {
460
461 assert replayOrderTail.replayNext == null;
462 assert other.replayOrderHead.replayPrev == null;
463 replayOrderTail.replayNext = other.replayOrderHead;
464 other.replayOrderHead.replayPrev = replayOrderTail;
465 replayOrderTail = other.replayOrderTail;
466
467
468 if (rootHead == null) {
469 rootHead = other.rootHead;
470 } else if (other.rootHead != null) {
471 Entry otherTail = findLinkListTail(other.rootHead);
472 otherTail.linkNext = rootHead;
473 rootHead.linkPrev = otherTail;
474 rootHead = other.rootHead;
475 }
476
477
478 if (childUnlinkedHead == null) {
479 childUnlinkedHead = other.childUnlinkedHead;
480 } else if (other.childUnlinkedHead != null) {
481 Entry otherTail = findLinkListTail(other.childUnlinkedHead);
482 otherTail.linkNext = childUnlinkedHead;
483 childUnlinkedHead.linkPrev = otherTail;
484 childUnlinkedHead = other.childUnlinkedHead;
485 }
486 }
487
488 other.clear();
489 }
490
491
492
493
494
495
496 public EntryIterator fetchReady() {
497 buildGraph();
498
499 Entry readyHead = null;
500 Entry readyTail = null;
501 Entry p = replayOrderHead;
502 while (p != null) {
503 Entry next = p.replayNext;
504 if (p.isReady()) {
505 unlinkFromReplayList(p);
506 if (readyTail != null) {
507 readyTail.replayNext = p;
508 p.replayPrev = readyTail;
509 } else {
510 p.replayPrev = null;
511 readyHead = p;
512 }
513 readyTail = p;
514 p.replayNext = null;
515 }
516 p = next;
517 }
518
519
520 for (p = readyHead; p != null; p = p.replayNext) {
521 removeFromMap(p.getProcId());
522 unlinkFromLinkList(p);
523 }
524 return readyHead != null ? new EntryIterator(readyHead) : null;
525 }
526
527
528
529
530 public EntryIterator fetchAll() {
531 Entry head = replayOrderHead;
532 for (Entry p = head; p != null; p = p.replayNext) {
533 removeFromMap(p.getProcId());
534 }
535 for (int i = 0; i < procedureMap.length; ++i) {
536 assert procedureMap[i] == null : "map not empty i=" + i;
537 }
538 replayOrderHead = null;
539 replayOrderTail = null;
540 childUnlinkedHead = null;
541 rootHead = null;
542 return head != null ? new EntryIterator(head) : null;
543 }
544
545 private void buildGraph() {
546 Entry p = childUnlinkedHead;
547 while (p != null) {
548 Entry next = p.linkNext;
549 Entry rootProc = getRootProcedure(p);
550 if (rootProc != null) {
551 rootProc.childHead = addToLinkList(p, rootProc.childHead);
552 }
553 p = next;
554 }
555
556 for (p = rootHead; p != null; p = p.linkNext) {
557 checkReadyToRun(p);
558 }
559 }
560
561 private Entry getRootProcedure(Entry entry) {
562 while (entry != null && entry.hasParent()) {
563 entry = getProcedure(entry.getParentId());
564 }
565 return entry;
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
605
606
607 private boolean checkReadyToRun(Entry rootEntry) {
608 int stackIdSum = 0;
609 int maxStackId = 0;
610 for (int i = 0; i < rootEntry.proto.getStackIdCount(); ++i) {
611 int stackId = 1 + rootEntry.proto.getStackId(i);
612 maxStackId = Math.max(maxStackId, stackId);
613 stackIdSum += stackId;
614 }
615
616 for (Entry p = rootEntry.childHead; p != null; p = p.linkNext) {
617 for (int i = 0; i < p.proto.getStackIdCount(); ++i) {
618 int stackId = 1 + p.proto.getStackId(i);
619 maxStackId = Math.max(maxStackId, stackId);
620 stackIdSum += stackId;
621 }
622 }
623 final int cmpStackIdSum = (maxStackId * (maxStackId + 1) / 2);
624 if (cmpStackIdSum == stackIdSum) {
625 rootEntry.ready = true;
626 for (Entry p = rootEntry.childHead; p != null; p = p.linkNext) {
627 p.ready = true;
628 }
629 return true;
630 }
631 return false;
632 }
633
634 private void unlinkFromReplayList(Entry entry) {
635 if (replayOrderHead == entry) {
636 replayOrderHead = entry.replayNext;
637 }
638 if (replayOrderTail == entry) {
639 replayOrderTail = entry.replayPrev;
640 }
641 if (entry.replayPrev != null) {
642 entry.replayPrev.replayNext = entry.replayNext;
643 }
644 if (entry.replayNext != null) {
645 entry.replayNext.replayPrev = entry.replayPrev;
646 }
647 }
648
649 private void addToReplayList(final Entry entry) {
650 unlinkFromReplayList(entry);
651 entry.replayNext = replayOrderHead;
652 entry.replayPrev = null;
653 if (replayOrderHead != null) {
654 replayOrderHead.replayPrev = entry;
655 } else {
656 replayOrderTail = entry;
657 }
658 replayOrderHead = entry;
659 }
660
661 private void unlinkFromLinkList(Entry entry) {
662 if (entry == rootHead) {
663 rootHead = entry.linkNext;
664 } else if (entry == childUnlinkedHead) {
665 childUnlinkedHead = entry.linkNext;
666 }
667 if (entry.linkPrev != null) {
668 entry.linkPrev.linkNext = entry.linkNext;
669 }
670 if (entry.linkNext != null) {
671 entry.linkNext.linkPrev = entry.linkPrev;
672 }
673 }
674
675 private Entry addToLinkList(Entry entry, Entry linkHead) {
676 unlinkFromLinkList(entry);
677 entry.linkNext = linkHead;
678 entry.linkPrev = null;
679 if (linkHead != null) {
680 linkHead.linkPrev = entry;
681 }
682 return entry;
683 }
684
685 private Entry findLinkListTail(Entry linkHead) {
686 Entry tail = linkHead;
687 while (tail.linkNext != null) {
688 tail = tail.linkNext;
689 }
690 return tail;
691 }
692
693 private Entry addToMap(final long procId, final boolean hasParent) {
694 int slotIndex = getMapSlot(procId);
695 Entry entry = getProcedure(slotIndex, procId);
696 if (entry != null) return entry;
697
698 entry = new Entry(procedureMap[slotIndex]);
699 procedureMap[slotIndex] = entry;
700 return entry;
701 }
702
703 private Entry removeFromMap(final long procId) {
704 int slotIndex = getMapSlot(procId);
705 Entry prev = null;
706 Entry entry = procedureMap[slotIndex];
707 while (entry != null) {
708 if (procId == entry.getProcId()) {
709 if (prev != null) {
710 prev.hashNext = entry.hashNext;
711 } else {
712 procedureMap[slotIndex] = entry.hashNext;
713 }
714 entry.hashNext = null;
715 return entry;
716 }
717 prev = entry;
718 entry = entry.hashNext;
719 }
720 return null;
721 }
722
723 private Entry getProcedure(final long procId) {
724 return getProcedure(getMapSlot(procId), procId);
725 }
726
727 private Entry getProcedure(final int slotIndex, final long procId) {
728 Entry entry = procedureMap[slotIndex];
729 while (entry != null) {
730 if (procId == entry.getProcId()) {
731 return entry;
732 }
733 entry = entry.hashNext;
734 }
735 return null;
736 }
737
738 private int getMapSlot(final long procId) {
739 return (int)(Procedure.getProcIdHashCode(procId) % procedureMap.length);
740 }
741 }
742 }