View Javadoc

1   /**
2    * Licensed to the Apache Software Foundation (ASF) under one
3    * or more contributor license agreements.  See the NOTICE file
4    * distributed with this work for additional information
5    * regarding copyright ownership.  The ASF licenses this file
6    * to you under the Apache License, Version 2.0 (the
7    * "License"); you may not use this file except in compliance
8    * with the License.  You may obtain a copy of the License at
9    *
10   *     http://www.apache.org/licenses/LICENSE-2.0
11   *
12   * Unless required by applicable law or agreed to in writing, software
13   * distributed under the License is distributed on an "AS IS" BASIS,
14   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15   * See the License for the specific language governing permissions and
16   * limitations under the License.
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   * Base Procedure class responsible to handle the Procedure Metadata
48   * e.g. state, startTime, lastUpdate, stack-indexes, ...
49   *
50   * execute() is called each time the procedure is executed.
51   * it may be called multiple times in case of failure and restart, so the
52   * code must be idempotent.
53   * the return is a set of sub-procedures or null in case the procedure doesn't
54   * have sub-procedures. Once the sub-procedures are successfully completed
55   * the execute() method is called again, you should think at it as a stack:
56   *  -> step 1
57   *  ---> step 2
58   *  -> step 1
59   *
60   * rollback() is called when the procedure or one of the sub-procedures is failed.
61   * the rollback step is supposed to cleanup the resources created during the
62   * execute() step. in case of failure and restart rollback() may be called
63   * multiple times, so the code must be idempotent.
64   */
65  @InterfaceAudience.Private
66  @InterfaceStability.Evolving
67  public abstract class Procedure<TEnvironment> implements Comparable<Procedure> {
68    // unchanged after initialization
69    private String owner = null;
70    private Long parentProcId = null;
71    private Long procId = null;
72    private long startTime;
73  
74    // runtime state, updated every operation
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     * The main code of the procedure. It must be idempotent since execute()
88     * may be called multiple time in case of machine failure in the middle
89     * of the execution.
90     * @return a set of sub-procedures or null if there is nothing else to execute.
91     */
92    protected abstract Procedure[] execute(TEnvironment env)
93      throws ProcedureYieldException;
94  
95    /**
96     * The code to undo what done by the execute() code.
97     * It is called when the procedure or one of the sub-procedure failed or an
98     * abort was requested. It should cleanup all the resources created by
99     * the execute() call. The implementation must be idempotent since rollback()
100    * may be called multiple time in case of machine failure in the middle
101    * of the execution.
102    * @throws IOException temporary failure, the rollback will retry later
103    */
104   protected abstract void rollback(TEnvironment env)
105     throws IOException;
106 
107   /**
108    * The abort() call is asynchronous and each procedure must decide how to deal
109    * with that, if they want to be abortable. The simplest implementation
110    * is to have an AtomicBoolean set in the abort() method and then the execute()
111    * will check if the abort flag is set or not.
112    * abort() may be called multiple times from the client, so the implementation
113    * must be idempotent.
114    *
115    * NOTE: abort() is not like Thread.interrupt() it is just a notification
116    * that allows the procedure implementor where to abort to avoid leak and
117    * have a better control on what was executed and what not.
118    */
119   protected abstract boolean abort(TEnvironment env);
120 
121   /**
122    * The user-level code of the procedure may have some state to
123    * persist (e.g. input arguments) to be able to resume on failure.
124    * @param stream the stream that will contain the user serialized data
125    */
126   protected abstract void serializeStateData(final OutputStream stream)
127     throws IOException;
128 
129   /**
130    * Called on store load to allow the user to decode the previously serialized
131    * state.
132    * @param stream the stream that contains the user serialized data
133    */
134   protected abstract void deserializeStateData(final InputStream stream)
135     throws IOException;
136 
137   /**
138    * The user should override this method, and try to take a lock if necessary.
139    * A lock can be anything, and it is up to the implementor.
140    * Example: in our Master we can execute request in parallel for different tables
141    *          create t1 and create t2 can be executed at the same time.
142    *          anything else on t1/t2 is queued waiting that specific table create to happen.
143    *
144    * @return true if the lock was acquired and false otherwise
145    */
146   protected boolean acquireLock(final TEnvironment env) {
147     return true;
148   }
149 
150   /**
151    * The user should override this method, and release lock if necessary.
152    */
153   protected void releaseLock(final TEnvironment env) {
154     // no-op
155   }
156 
157   /**
158    * Called when the procedure is loaded for replay.
159    * The procedure implementor may use this method to perform some quick
160    * operation before replay.
161    * e.g. failing the procedure if the state on replay may be unknown.
162    */
163   protected void beforeReplay(final TEnvironment env) {
164     // no-op
165   }
166 
167   /**
168    * Called when the procedure is marked as completed (success or rollback).
169    * The procedure implementor may use this method to cleanup in-memory states.
170    * This operation will not be retried on failure.
171    */
172   protected void completionCleanup(final TEnvironment env) {
173     // no-op
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    * Extend the toString() information with the procedure details
210    * e.g. className and parameters
211    * @param builder the string builder to use to append the proc specific information
212    */
213   protected void toStringClassDetails(StringBuilder builder) {
214     builder.append(getClass().getName());
215   }
216 
217   /**
218    * @return the serialized result if any, otherwise null
219    */
220   public byte[] getResult() {
221     return result;
222   }
223 
224   /**
225    * The procedure may leave a "result" on completion.
226    * @param result the serialized result that will be passed to the client
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    * @return true if the procedure has failed.
258    *         true may mean failed but not yet rolledback or failed and rolledback.
259    */
260   public synchronized boolean isFailed() {
261     return exception != null || state == ProcedureState.ROLLEDBACK;
262   }
263 
264   /**
265    * @return true if the procedure is finished successfully.
266    */
267   public synchronized boolean isSuccess() {
268     return state == ProcedureState.FINISHED && exception == null;
269   }
270 
271   /**
272    * @return true if the procedure is finished. The Procedure may be completed
273    *         successfuly or failed and rolledback.
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    * @return true if the procedure is waiting for a child to finish or for an external event.
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    * @param timeout timeout in msec
319    */
320   protected void setTimeout(final int timeout) {
321     this.timeout = timeout;
322   }
323 
324   /**
325    * @return the timeout in msec
326    */
327   public int getTimeout() {
328     return timeout;
329   }
330 
331   /**
332    * @return the remaining time before the timeout
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    * Called by the ProcedureExecutor to assign the ID to the newly created procedure.
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    * Called by the ProcedureExecutor to assign the parent to the newly created procedure.
403    */
404   @InterfaceAudience.Private
405   protected void setParentProcId(final long parentProcId) {
406     this.parentProcId = parentProcId;
407   }
408 
409   /**
410    * Called by the ProcedureExecutor to set the value to the newly created procedure.
411    */
412   @VisibleForTesting
413   @InterfaceAudience.Private
414   protected void setNonceKey(final NonceKey nonceKey) {
415     this.nonceKey = nonceKey;
416   }
417 
418   /**
419    * Internal method called by the ProcedureExecutor that starts the
420    * user-level code execute().
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    * Internal method called by the ProcedureExecutor that starts the
435    * user-level code rollback().
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    * Called on store load to initialize the Procedure internals after
449    * the creation/deserialization.
450    */
451   @InterfaceAudience.Private
452   protected void setStartTime(final long startTime) {
453     this.startTime = startTime;
454   }
455 
456   /**
457    * Called on store load to initialize the Procedure internals after
458    * the creation/deserialization.
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    * Called by the ProcedureExecutor on procedure-load to restore the latch state
470    */
471   @InterfaceAudience.Private
472   protected synchronized void setChildrenLatch(final int numChildren) {
473     this.childrenLatch = numChildren;
474   }
475 
476   /**
477    * Called by the ProcedureExecutor on procedure-load to restore the latch state
478    */
479   @InterfaceAudience.Private
480   protected synchronized void incChildrenLatch() {
481     // TODO: can this be inferred from the stack? I think so...
482     this.childrenLatch++;
483   }
484 
485   /**
486    * Called by the ProcedureExecutor to notify that one of the sub-procedures
487    * has completed.
488    */
489   @InterfaceAudience.Private
490   protected synchronized boolean childrenCountDown() {
491     assert childrenLatch > 0;
492     return --childrenLatch == 0;
493   }
494 
495   /**
496    * Called by the RootProcedureState on procedure execution.
497    * Each procedure store its stack-index positions.
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    * Called on store load to initialize the Procedure internals after
523    * the creation/deserialization.
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    * Helper to lookup the root Procedure ID given a specified procedure.
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    * Helper to create the ProcedureInfo from Procedure.
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    * Helper to convert the procedure to protobuf.
631    * Used by ProcedureStore implementations.
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    * Helper to convert the protobuf procedure.
692    * Used by ProcedureStore implementations.
693    *
694    * TODO: OPTIMIZATION: some of the field never change during the execution
695    *                     (e.g. className, procId, parentId, ...).
696    *                     We can split in 'data' and 'state', and the store
697    *                     may take advantage of it by storing the data only on insert().
698    */
699   @InterfaceAudience.Private
700   public static Procedure convert(final ProcedureProtos.Procedure proto)
701       throws IOException {
702     // Procedure from class name
703     Procedure proc = Procedure.newInstance(proto.getClassName());
704 
705     // set fields
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     // we want to call deserialize even when the stream is empty, mainly for testing.
744     proc.deserializeStateData(proto.getStateData().newInput());
745 
746     return proc;
747   }
748 }