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;
20
21 import java.io.IOException;
22 import java.io.InputStream;
23 import java.io.OutputStream;
24 import java.lang.reflect.Constructor;
25 import java.lang.reflect.Modifier;
26 import java.util.Arrays;
27 import java.util.List;
28 import java.util.Map;
29
30 import org.apache.hadoop.hbase.HConstants;
31 import org.apache.hadoop.hbase.ProcedureInfo;
32 import org.apache.hadoop.hbase.classification.InterfaceAudience;
33 import org.apache.hadoop.hbase.classification.InterfaceStability;
34 import org.apache.hadoop.hbase.exceptions.TimeoutIOException;
35 import org.apache.hadoop.hbase.procedure2.util.StringUtils;
36 import org.apache.hadoop.hbase.protobuf.generated.ProcedureProtos;
37 import org.apache.hadoop.hbase.protobuf.generated.ProcedureProtos.ProcedureState;
38 import org.apache.hadoop.hbase.util.ByteStringer;
39 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
40 import org.apache.hadoop.hbase.util.NonceKey;
41
42 import com.google.common.annotations.VisibleForTesting;
43 import com.google.common.base.Preconditions;
44 import com.google.protobuf.ByteString;
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65 @InterfaceAudience.Private
66 @InterfaceStability.Evolving
67 public abstract class Procedure<TEnvironment> implements Comparable<Procedure> {
68
69 private String owner = null;
70 private Long parentProcId = null;
71 private Long procId = null;
72 private long startTime;
73
74
75 private ProcedureState state = ProcedureState.INITIALIZING;
76 private Integer timeout = null;
77 private int[] stackIndexes = null;
78 private int childrenLatch = 0;
79 private long lastUpdate;
80
81 private RemoteProcedureException exception = null;
82 private byte[] result = null;
83
84 private NonceKey nonceKey = null;
85
86
87
88
89
90
91
92
93
94
95 protected abstract Procedure[] execute(TEnvironment env)
96 throws ProcedureYieldException, InterruptedException;
97
98
99
100
101
102
103
104
105
106
107
108
109 protected abstract void rollback(TEnvironment env)
110 throws IOException, InterruptedException;
111
112
113
114
115
116
117
118
119
120
121
122
123
124 protected abstract boolean abort(TEnvironment env);
125
126
127
128
129
130
131 protected abstract void serializeStateData(final OutputStream stream)
132 throws IOException;
133
134
135
136
137
138
139 protected abstract void deserializeStateData(final InputStream stream)
140 throws IOException;
141
142
143
144
145
146
147
148
149
150
151 protected boolean acquireLock(final TEnvironment env) {
152 return true;
153 }
154
155
156
157
158 protected void releaseLock(final TEnvironment env) {
159
160 }
161
162
163
164
165
166
167
168 protected void beforeReplay(final TEnvironment env) {
169
170 }
171
172
173
174
175
176
177 protected void completionCleanup(final TEnvironment env) {
178
179 }
180
181
182
183
184
185
186
187
188
189 protected boolean isYieldAfterExecutionStep(final TEnvironment env) {
190 return false;
191 }
192
193
194
195
196
197
198
199
200
201
202 protected boolean shouldWaitClientAck(final TEnvironment env) {
203 return true;
204 }
205
206 @Override
207 public String toString() {
208
209 return toStringSimpleSB().toString();
210 }
211
212
213
214
215
216
217 protected StringBuilder toStringSimpleSB() {
218 StringBuilder sb = new StringBuilder();
219 toStringClassDetails(sb);
220
221 if (procId != null) {
222 sb.append(" id=");
223 sb.append(getProcId());
224 }
225
226 if (hasParent()) {
227 sb.append(" parent=");
228 sb.append(getParentProcId());
229 }
230
231 if (hasOwner()) {
232 sb.append(" owner=");
233 sb.append(getOwner());
234 }
235
236 sb.append(" state=");
237 toStringState(sb);
238
239 return sb;
240 }
241
242
243
244
245
246 public String toStringDetails() {
247 StringBuilder sb = toStringSimpleSB();
248
249 sb.append(" startTime=");
250 sb.append(getStartTime());
251
252 sb.append(" lastUpdate=");
253 sb.append(getLastUpdate());
254
255 int[] stackIndices = getStackIndexes();
256 if (stackIndices != null) {
257 sb.append("\n");
258 sb.append("stackIndexes=");
259 sb.append(Arrays.toString(stackIndices));
260 }
261
262 return sb.toString();
263 }
264
265 protected String toStringClass() {
266 StringBuilder sb = new StringBuilder();
267 toStringClassDetails(sb);
268
269 return sb.toString();
270 }
271
272
273
274
275
276 protected void toStringState(StringBuilder builder) {
277 builder.append(getState());
278 }
279
280
281
282
283
284
285 protected void toStringClassDetails(StringBuilder builder) {
286 builder.append(getClass().getName());
287 }
288
289
290
291
292 public byte[] getResult() {
293 return result;
294 }
295
296
297
298
299
300 protected void setResult(final byte[] result) {
301 this.result = result;
302 }
303
304 public long getProcId() {
305 return procId;
306 }
307
308 public boolean hasParent() {
309 return parentProcId != null;
310 }
311
312 public boolean hasException() {
313 return exception != null;
314 }
315
316 public boolean hasTimeout() {
317 return timeout != null;
318 }
319
320 public long getParentProcId() {
321 return parentProcId;
322 }
323
324 public NonceKey getNonceKey() {
325 return nonceKey;
326 }
327
328
329
330
331 protected synchronized boolean isRunnable() {
332 return state == ProcedureState.RUNNABLE;
333 }
334
335
336
337
338
339 public synchronized boolean isFailed() {
340 return exception != null || state == ProcedureState.ROLLEDBACK;
341 }
342
343
344
345
346 public synchronized boolean isSuccess() {
347 return state == ProcedureState.FINISHED && exception == null;
348 }
349
350
351
352
353
354 public synchronized boolean isFinished() {
355 switch (state) {
356 case ROLLEDBACK:
357 return true;
358 case FINISHED:
359 return exception == null;
360 default:
361 break;
362 }
363 return false;
364 }
365
366
367
368
369 public synchronized boolean isWaiting() {
370 switch (state) {
371 case WAITING:
372 case WAITING_TIMEOUT:
373 return true;
374 default:
375 break;
376 }
377 return false;
378 }
379
380 public synchronized RemoteProcedureException getException() {
381 return exception;
382 }
383
384 public long getStartTime() {
385 return startTime;
386 }
387
388 public synchronized long getLastUpdate() {
389 return lastUpdate;
390 }
391
392 public synchronized long elapsedTime() {
393 return lastUpdate - startTime;
394 }
395
396
397
398
399 protected void setTimeout(final int timeout) {
400 this.timeout = timeout;
401 }
402
403
404
405
406 public int getTimeout() {
407 return timeout;
408 }
409
410
411
412
413 public long getTimeRemaining() {
414 return Math.max(0, timeout - (EnvironmentEdgeManager.currentTime() - startTime));
415 }
416
417 @VisibleForTesting
418 @InterfaceAudience.Private
419 public void setOwner(final String owner) {
420 this.owner = StringUtils.isEmpty(owner) ? null : owner;
421 }
422
423 public String getOwner() {
424 return owner;
425 }
426
427 public boolean hasOwner() {
428 return owner != null;
429 }
430
431 @VisibleForTesting
432 @InterfaceAudience.Private
433 protected synchronized void setState(final ProcedureState state) {
434 this.state = state;
435 updateTimestamp();
436 }
437
438 @InterfaceAudience.Private
439 protected synchronized ProcedureState getState() {
440 return state;
441 }
442
443 protected void setFailure(final String source, final Throwable cause) {
444 setFailure(new RemoteProcedureException(source, cause));
445 }
446
447 protected synchronized void setFailure(final RemoteProcedureException exception) {
448 this.exception = exception;
449 if (!isFinished()) {
450 setState(ProcedureState.FINISHED);
451 }
452 }
453
454 protected void setAbortFailure(final String source, final String msg) {
455 setFailure(source, new ProcedureAbortedException(msg));
456 }
457
458 @InterfaceAudience.Private
459 protected synchronized boolean setTimeoutFailure() {
460 if (state == ProcedureState.WAITING_TIMEOUT) {
461 long timeDiff = EnvironmentEdgeManager.currentTime() - lastUpdate;
462 setFailure("ProcedureExecutor", new TimeoutIOException(
463 "Operation timed out after " + StringUtils.humanTimeDiff(timeDiff)));
464 return true;
465 }
466 return false;
467 }
468
469
470
471
472 @VisibleForTesting
473 @InterfaceAudience.Private
474 protected void setProcId(final long procId) {
475 this.procId = procId;
476 this.startTime = EnvironmentEdgeManager.currentTime();
477 setState(ProcedureState.RUNNABLE);
478 }
479
480
481
482
483 @InterfaceAudience.Private
484 protected void setParentProcId(final long parentProcId) {
485 this.parentProcId = parentProcId;
486 }
487
488
489
490
491 @VisibleForTesting
492 @InterfaceAudience.Private
493 protected void setNonceKey(final NonceKey nonceKey) {
494 this.nonceKey = nonceKey;
495 }
496
497
498
499
500
501 @InterfaceAudience.Private
502 protected Procedure[] doExecute(final TEnvironment env)
503 throws ProcedureYieldException, InterruptedException {
504 try {
505 updateTimestamp();
506 return execute(env);
507 } finally {
508 updateTimestamp();
509 }
510 }
511
512
513
514
515
516 @InterfaceAudience.Private
517 protected void doRollback(final TEnvironment env)
518 throws IOException, InterruptedException {
519 try {
520 updateTimestamp();
521 rollback(env);
522 } finally {
523 updateTimestamp();
524 }
525 }
526
527
528
529
530
531 @InterfaceAudience.Private
532 protected void setStartTime(final long startTime) {
533 this.startTime = startTime;
534 }
535
536
537
538
539
540 private synchronized void setLastUpdate(final long lastUpdate) {
541 this.lastUpdate = lastUpdate;
542 }
543
544 protected synchronized void updateTimestamp() {
545 this.lastUpdate = EnvironmentEdgeManager.currentTime();
546 }
547
548
549
550
551 @InterfaceAudience.Private
552 protected synchronized void setChildrenLatch(final int numChildren) {
553 this.childrenLatch = numChildren;
554 }
555
556
557
558
559 @InterfaceAudience.Private
560 protected synchronized void incChildrenLatch() {
561
562 this.childrenLatch++;
563 }
564
565
566
567
568
569 @InterfaceAudience.Private
570 protected synchronized boolean childrenCountDown() {
571 assert childrenLatch > 0;
572 return --childrenLatch == 0;
573 }
574
575 @InterfaceAudience.Private
576 protected synchronized boolean hasChildren() {
577 return childrenLatch > 0;
578 }
579
580
581
582
583
584 @InterfaceAudience.Private
585 protected synchronized void addStackIndex(final int index) {
586 if (stackIndexes == null) {
587 stackIndexes = new int[] { index };
588 } else {
589 int count = stackIndexes.length;
590 stackIndexes = Arrays.copyOf(stackIndexes, count + 1);
591 stackIndexes[count] = index;
592 }
593 }
594
595 @InterfaceAudience.Private
596 protected synchronized boolean removeStackIndex() {
597 if (stackIndexes != null && stackIndexes.length > 1) {
598 stackIndexes = Arrays.copyOf(stackIndexes, stackIndexes.length - 1);
599 return false;
600 } else {
601 stackIndexes = null;
602 return true;
603 }
604 }
605
606
607
608
609
610 @InterfaceAudience.Private
611 protected synchronized void setStackIndexes(final List<Integer> stackIndexes) {
612 this.stackIndexes = new int[stackIndexes.size()];
613 for (int i = 0; i < this.stackIndexes.length; ++i) {
614 this.stackIndexes[i] = stackIndexes.get(i);
615 }
616 }
617
618 @InterfaceAudience.Private
619 protected synchronized boolean wasExecuted() {
620 return stackIndexes != null;
621 }
622
623 @InterfaceAudience.Private
624 protected synchronized int[] getStackIndexes() {
625 return stackIndexes;
626 }
627
628 @Override
629 public int compareTo(final Procedure other) {
630 long diff = getProcId() - other.getProcId();
631 return (diff < 0) ? -1 : (diff > 0) ? 1 : 0;
632 }
633
634
635
636
637
638 public static long getProcIdHashCode(final long procId) {
639 long h = procId;
640 h ^= h >> 16;
641 h *= 0x85ebca6b;
642 h ^= h >> 13;
643 h *= 0xc2b2ae35;
644 h ^= h >> 16;
645 return h;
646 }
647
648
649
650
651 @InterfaceAudience.Private
652 protected static Long getRootProcedureId(final Map<Long, Procedure> procedures, Procedure proc) {
653 while (proc.hasParent()) {
654 proc = procedures.get(proc.getParentProcId());
655 if (proc == null) return null;
656 }
657 return proc.getProcId();
658 }
659
660 protected static Procedure newInstance(final String className) throws IOException {
661 try {
662 Class<?> clazz = Class.forName(className);
663 if (!Modifier.isPublic(clazz.getModifiers())) {
664 throw new Exception("the " + clazz + " class is not public");
665 }
666
667 Constructor<?> ctor = clazz.getConstructor();
668 assert ctor != null : "no constructor found";
669 if (!Modifier.isPublic(ctor.getModifiers())) {
670 throw new Exception("the " + clazz + " constructor is not public");
671 }
672 return (Procedure)ctor.newInstance();
673 } catch (Exception e) {
674 throw new IOException("The procedure class " + className +
675 " must be accessible and have an empty constructor", e);
676 }
677 }
678
679 protected static void validateClass(final Procedure proc) throws IOException {
680 try {
681 Class<?> clazz = proc.getClass();
682 if (!Modifier.isPublic(clazz.getModifiers())) {
683 throw new Exception("the " + clazz + " class is not public");
684 }
685
686 Constructor<?> ctor = clazz.getConstructor();
687 assert ctor != null;
688 if (!Modifier.isPublic(ctor.getModifiers())) {
689 throw new Exception("the " + clazz + " constructor is not public");
690 }
691 } catch (Exception e) {
692 throw new IOException("The procedure class " + proc.getClass().getName() +
693 " must be accessible and have an empty constructor", e);
694 }
695 }
696
697
698
699
700 @InterfaceAudience.Private
701 public static ProcedureInfo createProcedureInfo(final Procedure proc, final NonceKey nonceKey) {
702 RemoteProcedureException exception = proc.hasException() ? proc.getException() : null;
703 return new ProcedureInfo(
704 proc.getProcId(),
705 proc.toStringClass(),
706 proc.getOwner(),
707 proc.getState(),
708 proc.hasParent() ? proc.getParentProcId() : -1,
709 nonceKey,
710 exception != null ?
711 RemoteProcedureException.toProto(exception.getSource(), exception.getCause()) : null,
712 proc.getLastUpdate(),
713 proc.getStartTime(),
714 proc.getResult());
715 }
716
717
718
719
720
721 @InterfaceAudience.Private
722 public static ProcedureProtos.Procedure convert(final Procedure proc)
723 throws IOException {
724 Preconditions.checkArgument(proc != null);
725 validateClass(proc);
726
727 ProcedureProtos.Procedure.Builder builder = ProcedureProtos.Procedure.newBuilder()
728 .setClassName(proc.getClass().getName())
729 .setProcId(proc.getProcId())
730 .setState(proc.getState())
731 .setStartTime(proc.getStartTime())
732 .setLastUpdate(proc.getLastUpdate());
733
734 if (proc.hasParent()) {
735 builder.setParentId(proc.getParentProcId());
736 }
737
738 if (proc.hasTimeout()) {
739 builder.setTimeout(proc.getTimeout());
740 }
741
742 if (proc.hasOwner()) {
743 builder.setOwner(proc.getOwner());
744 }
745
746 int[] stackIds = proc.getStackIndexes();
747 if (stackIds != null) {
748 for (int i = 0; i < stackIds.length; ++i) {
749 builder.addStackId(stackIds[i]);
750 }
751 }
752
753 if (proc.hasException()) {
754 RemoteProcedureException exception = proc.getException();
755 builder.setException(
756 RemoteProcedureException.toProto(exception.getSource(), exception.getCause()));
757 }
758
759 byte[] result = proc.getResult();
760 if (result != null) {
761 builder.setResult(ByteStringer.wrap(result));
762 }
763
764 ByteString.Output stateStream = ByteString.newOutput();
765 proc.serializeStateData(stateStream);
766 if (stateStream.size() > 0) {
767 builder.setStateData(stateStream.toByteString());
768 }
769
770 if (proc.getNonceKey() != null) {
771 builder.setNonceGroup(proc.getNonceKey().getNonceGroup());
772 builder.setNonce(proc.getNonceKey().getNonce());
773 }
774
775 return builder.build();
776 }
777
778
779
780
781
782
783
784
785
786
787 @InterfaceAudience.Private
788 public static Procedure convert(final ProcedureProtos.Procedure proto)
789 throws IOException {
790
791 Procedure proc = Procedure.newInstance(proto.getClassName());
792
793
794 proc.setProcId(proto.getProcId());
795 proc.setState(proto.getState());
796 proc.setStartTime(proto.getStartTime());
797 proc.setLastUpdate(proto.getLastUpdate());
798
799 if (proto.hasParentId()) {
800 proc.setParentProcId(proto.getParentId());
801 }
802
803 if (proto.hasOwner()) {
804 proc.setOwner(proto.getOwner());
805 }
806
807 if (proto.hasTimeout()) {
808 proc.setTimeout(proto.getTimeout());
809 }
810
811 if (proto.getStackIdCount() > 0) {
812 proc.setStackIndexes(proto.getStackIdList());
813 }
814
815 if (proto.hasException()) {
816 assert proc.getState() == ProcedureState.FINISHED ||
817 proc.getState() == ProcedureState.ROLLEDBACK :
818 "The procedure must be failed (waiting to rollback) or rolledback";
819 proc.setFailure(RemoteProcedureException.fromProto(proto.getException()));
820 }
821
822 if (proto.hasResult()) {
823 proc.setResult(proto.getResult().toByteArray());
824 }
825
826 if (proto.getNonce() != HConstants.NO_NONCE) {
827 NonceKey nonceKey = new NonceKey(proto.getNonceGroup(), proto.getNonce());
828 proc.setNonceKey(nonceKey);
829 }
830
831
832 proc.deserializeStateData(proto.getStateData().newInput());
833
834 return proc;
835 }
836 }