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