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
332 public synchronized boolean isFailed() {
333 return exception != null || state == ProcedureState.ROLLEDBACK;
334 }
335
336
337
338
339 public synchronized boolean isSuccess() {
340 return state == ProcedureState.FINISHED && exception == null;
341 }
342
343
344
345
346
347 public synchronized boolean isFinished() {
348 switch (state) {
349 case ROLLEDBACK:
350 return true;
351 case FINISHED:
352 return exception == null;
353 default:
354 break;
355 }
356 return false;
357 }
358
359
360
361
362 public synchronized boolean isWaiting() {
363 switch (state) {
364 case WAITING:
365 case WAITING_TIMEOUT:
366 return true;
367 default:
368 break;
369 }
370 return false;
371 }
372
373 public synchronized RemoteProcedureException getException() {
374 return exception;
375 }
376
377 public long getStartTime() {
378 return startTime;
379 }
380
381 public synchronized long getLastUpdate() {
382 return lastUpdate;
383 }
384
385 public synchronized long elapsedTime() {
386 return lastUpdate - startTime;
387 }
388
389
390
391
392 protected void setTimeout(final int timeout) {
393 this.timeout = timeout;
394 }
395
396
397
398
399 public int getTimeout() {
400 return timeout;
401 }
402
403
404
405
406 public long getTimeRemaining() {
407 return Math.max(0, timeout - (EnvironmentEdgeManager.currentTime() - startTime));
408 }
409
410 @VisibleForTesting
411 @InterfaceAudience.Private
412 public void setOwner(final String owner) {
413 this.owner = StringUtils.isEmpty(owner) ? null : owner;
414 }
415
416 public String getOwner() {
417 return owner;
418 }
419
420 public boolean hasOwner() {
421 return owner != null;
422 }
423
424 @VisibleForTesting
425 @InterfaceAudience.Private
426 protected synchronized void setState(final ProcedureState state) {
427 this.state = state;
428 updateTimestamp();
429 }
430
431 @InterfaceAudience.Private
432 protected synchronized ProcedureState getState() {
433 return state;
434 }
435
436 protected void setFailure(final String source, final Throwable cause) {
437 setFailure(new RemoteProcedureException(source, cause));
438 }
439
440 protected synchronized void setFailure(final RemoteProcedureException exception) {
441 this.exception = exception;
442 if (!isFinished()) {
443 setState(ProcedureState.FINISHED);
444 }
445 }
446
447 protected void setAbortFailure(final String source, final String msg) {
448 setFailure(source, new ProcedureAbortedException(msg));
449 }
450
451 @InterfaceAudience.Private
452 protected synchronized boolean setTimeoutFailure() {
453 if (state == ProcedureState.WAITING_TIMEOUT) {
454 long timeDiff = EnvironmentEdgeManager.currentTime() - lastUpdate;
455 setFailure("ProcedureExecutor", new TimeoutIOException(
456 "Operation timed out after " + StringUtils.humanTimeDiff(timeDiff)));
457 return true;
458 }
459 return false;
460 }
461
462
463
464
465 @VisibleForTesting
466 @InterfaceAudience.Private
467 protected void setProcId(final long procId) {
468 this.procId = procId;
469 this.startTime = EnvironmentEdgeManager.currentTime();
470 setState(ProcedureState.RUNNABLE);
471 }
472
473
474
475
476 @InterfaceAudience.Private
477 protected void setParentProcId(final long parentProcId) {
478 this.parentProcId = parentProcId;
479 }
480
481
482
483
484 @VisibleForTesting
485 @InterfaceAudience.Private
486 protected void setNonceKey(final NonceKey nonceKey) {
487 this.nonceKey = nonceKey;
488 }
489
490
491
492
493
494 @InterfaceAudience.Private
495 protected Procedure[] doExecute(final TEnvironment env)
496 throws ProcedureYieldException, InterruptedException {
497 try {
498 updateTimestamp();
499 return execute(env);
500 } finally {
501 updateTimestamp();
502 }
503 }
504
505
506
507
508
509 @InterfaceAudience.Private
510 protected void doRollback(final TEnvironment env)
511 throws IOException, InterruptedException {
512 try {
513 updateTimestamp();
514 rollback(env);
515 } finally {
516 updateTimestamp();
517 }
518 }
519
520
521
522
523
524 @InterfaceAudience.Private
525 protected void setStartTime(final long startTime) {
526 this.startTime = startTime;
527 }
528
529
530
531
532
533 private synchronized void setLastUpdate(final long lastUpdate) {
534 this.lastUpdate = lastUpdate;
535 }
536
537 protected synchronized void updateTimestamp() {
538 this.lastUpdate = EnvironmentEdgeManager.currentTime();
539 }
540
541
542
543
544 @InterfaceAudience.Private
545 protected synchronized void setChildrenLatch(final int numChildren) {
546 this.childrenLatch = numChildren;
547 }
548
549
550
551
552 @InterfaceAudience.Private
553 protected synchronized void incChildrenLatch() {
554
555 this.childrenLatch++;
556 }
557
558
559
560
561
562 @InterfaceAudience.Private
563 protected synchronized boolean childrenCountDown() {
564 assert childrenLatch > 0;
565 return --childrenLatch == 0;
566 }
567
568
569
570
571
572 @InterfaceAudience.Private
573 protected synchronized void addStackIndex(final int index) {
574 if (stackIndexes == null) {
575 stackIndexes = new int[] { index };
576 } else {
577 int count = stackIndexes.length;
578 stackIndexes = Arrays.copyOf(stackIndexes, count + 1);
579 stackIndexes[count] = index;
580 }
581 }
582
583 @InterfaceAudience.Private
584 protected synchronized boolean removeStackIndex() {
585 if (stackIndexes.length > 1) {
586 stackIndexes = Arrays.copyOf(stackIndexes, stackIndexes.length - 1);
587 return false;
588 } else {
589 stackIndexes = null;
590 return true;
591 }
592 }
593
594
595
596
597
598 @InterfaceAudience.Private
599 protected synchronized void setStackIndexes(final List<Integer> stackIndexes) {
600 this.stackIndexes = new int[stackIndexes.size()];
601 for (int i = 0; i < this.stackIndexes.length; ++i) {
602 this.stackIndexes[i] = stackIndexes.get(i);
603 }
604 }
605
606 @InterfaceAudience.Private
607 protected synchronized boolean wasExecuted() {
608 return stackIndexes != null;
609 }
610
611 @InterfaceAudience.Private
612 protected synchronized int[] getStackIndexes() {
613 return stackIndexes;
614 }
615
616 @Override
617 public int compareTo(final Procedure other) {
618 long diff = getProcId() - other.getProcId();
619 return (diff < 0) ? -1 : (diff > 0) ? 1 : 0;
620 }
621
622
623
624
625
626 public static long getProcIdHashCode(final long procId) {
627 long h = procId;
628 h ^= h >> 16;
629 h *= 0x85ebca6b;
630 h ^= h >> 13;
631 h *= 0xc2b2ae35;
632 h ^= h >> 16;
633 return h;
634 }
635
636
637
638
639 @InterfaceAudience.Private
640 protected static Long getRootProcedureId(final Map<Long, Procedure> procedures, Procedure proc) {
641 while (proc.hasParent()) {
642 proc = procedures.get(proc.getParentProcId());
643 if (proc == null) return null;
644 }
645 return proc.getProcId();
646 }
647
648 protected static Procedure newInstance(final String className) throws IOException {
649 try {
650 Class<?> clazz = Class.forName(className);
651 if (!Modifier.isPublic(clazz.getModifiers())) {
652 throw new Exception("the " + clazz + " class is not public");
653 }
654
655 Constructor<?> ctor = clazz.getConstructor();
656 assert ctor != null : "no constructor found";
657 if (!Modifier.isPublic(ctor.getModifiers())) {
658 throw new Exception("the " + clazz + " constructor is not public");
659 }
660 return (Procedure)ctor.newInstance();
661 } catch (Exception e) {
662 throw new IOException("The procedure class " + className +
663 " must be accessible and have an empty constructor", e);
664 }
665 }
666
667 protected static void validateClass(final Procedure proc) throws IOException {
668 try {
669 Class<?> clazz = proc.getClass();
670 if (!Modifier.isPublic(clazz.getModifiers())) {
671 throw new Exception("the " + clazz + " class is not public");
672 }
673
674 Constructor<?> ctor = clazz.getConstructor();
675 assert ctor != null;
676 if (!Modifier.isPublic(ctor.getModifiers())) {
677 throw new Exception("the " + clazz + " constructor is not public");
678 }
679 } catch (Exception e) {
680 throw new IOException("The procedure class " + proc.getClass().getName() +
681 " must be accessible and have an empty constructor", e);
682 }
683 }
684
685
686
687
688 @InterfaceAudience.Private
689 public static ProcedureInfo createProcedureInfo(final Procedure proc, final NonceKey nonceKey) {
690 RemoteProcedureException exception = proc.hasException() ? proc.getException() : null;
691 return new ProcedureInfo(
692 proc.getProcId(),
693 proc.toStringClass(),
694 proc.getOwner(),
695 proc.getState(),
696 proc.hasParent() ? proc.getParentProcId() : -1,
697 nonceKey,
698 exception != null ?
699 RemoteProcedureException.toProto(exception.getSource(), exception.getCause()) : null,
700 proc.getLastUpdate(),
701 proc.getStartTime(),
702 proc.getResult());
703 }
704
705
706
707
708
709 @InterfaceAudience.Private
710 public static ProcedureProtos.Procedure convert(final Procedure proc)
711 throws IOException {
712 Preconditions.checkArgument(proc != null);
713 validateClass(proc);
714
715 ProcedureProtos.Procedure.Builder builder = ProcedureProtos.Procedure.newBuilder()
716 .setClassName(proc.getClass().getName())
717 .setProcId(proc.getProcId())
718 .setState(proc.getState())
719 .setStartTime(proc.getStartTime())
720 .setLastUpdate(proc.getLastUpdate());
721
722 if (proc.hasParent()) {
723 builder.setParentId(proc.getParentProcId());
724 }
725
726 if (proc.hasTimeout()) {
727 builder.setTimeout(proc.getTimeout());
728 }
729
730 if (proc.hasOwner()) {
731 builder.setOwner(proc.getOwner());
732 }
733
734 int[] stackIds = proc.getStackIndexes();
735 if (stackIds != null) {
736 for (int i = 0; i < stackIds.length; ++i) {
737 builder.addStackId(stackIds[i]);
738 }
739 }
740
741 if (proc.hasException()) {
742 RemoteProcedureException exception = proc.getException();
743 builder.setException(
744 RemoteProcedureException.toProto(exception.getSource(), exception.getCause()));
745 }
746
747 byte[] result = proc.getResult();
748 if (result != null) {
749 builder.setResult(ByteStringer.wrap(result));
750 }
751
752 ByteString.Output stateStream = ByteString.newOutput();
753 proc.serializeStateData(stateStream);
754 if (stateStream.size() > 0) {
755 builder.setStateData(stateStream.toByteString());
756 }
757
758 if (proc.getNonceKey() != null) {
759 builder.setNonceGroup(proc.getNonceKey().getNonceGroup());
760 builder.setNonce(proc.getNonceKey().getNonce());
761 }
762
763 return builder.build();
764 }
765
766
767
768
769
770
771
772
773
774
775 @InterfaceAudience.Private
776 public static Procedure convert(final ProcedureProtos.Procedure proto)
777 throws IOException {
778
779 Procedure proc = Procedure.newInstance(proto.getClassName());
780
781
782 proc.setProcId(proto.getProcId());
783 proc.setState(proto.getState());
784 proc.setStartTime(proto.getStartTime());
785 proc.setLastUpdate(proto.getLastUpdate());
786
787 if (proto.hasParentId()) {
788 proc.setParentProcId(proto.getParentId());
789 }
790
791 if (proto.hasOwner()) {
792 proc.setOwner(proto.getOwner());
793 }
794
795 if (proto.hasTimeout()) {
796 proc.setTimeout(proto.getTimeout());
797 }
798
799 if (proto.getStackIdCount() > 0) {
800 proc.setStackIndexes(proto.getStackIdList());
801 }
802
803 if (proto.hasException()) {
804 assert proc.getState() == ProcedureState.FINISHED ||
805 proc.getState() == ProcedureState.ROLLEDBACK :
806 "The procedure must be failed (waiting to rollback) or rolledback";
807 proc.setFailure(RemoteProcedureException.fromProto(proto.getException()));
808 }
809
810 if (proto.hasResult()) {
811 proc.setResult(proto.getResult().toByteArray());
812 }
813
814 if (proto.getNonce() != HConstants.NO_NONCE) {
815 NonceKey nonceKey = new NonceKey(proto.getNonceGroup(), proto.getNonce());
816 proc.setNonceKey(nonceKey);
817 }
818
819
820 proc.deserializeStateData(proto.getStateData().newInput());
821
822 return proc;
823 }
824 }