hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/RootProcedureState.java

   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.util.ArrayList;
  22 import java.util.HashSet;
  23 import java.util.List;
  24 import java.util.Set;
  25
  26 import org.apache.commons.logging.Log;
  27 import org.apache.commons.logging.LogFactory;
  28 import org.apache.hadoop.hbase.classification.InterfaceAudience;
  29 import org.apache.hadoop.hbase.classification.InterfaceStability;
  30 import org.apache.hadoop.hbase.shaded.protobuf.generated.ProcedureProtos.ProcedureState;
  31
  32 /**
  33  * Internal state of the ProcedureExecutor that describes the state of a "Root Procedure".
  34  * A "Root Procedure" is a Procedure without parent, each subprocedure will be
  35  * added to the "Root Procedure" stack (or rollback-stack).
  36  *
  37  * RootProcedureState is used and managed only by the ProcedureExecutor.
  38  *    Long rootProcId = getRootProcedureId(proc);
  39  *    rollbackStack.get(rootProcId).acquire(proc)
  40  *    rollbackStack.get(rootProcId).release(proc)
  41  *    ...
  42  */
  43 @InterfaceAudience.Private
  44 @InterfaceStability.Evolving
  45 class RootProcedureState {
  46   private static final Log LOG = LogFactory.getLog(RootProcedureState.class);
  47
  48   private enum State {
  49     RUNNING,         // The Procedure is running or ready to run
  50     FAILED,          // The Procedure failed, waiting for the rollback executing
  51     ROLLINGBACK,     // The Procedure failed and the execution was rolledback
  52   }
  53
  54   private Set<Procedure> subprocs = null;
  55   private ArrayList<Procedure> subprocStack = null;
  56   private State state = State.RUNNING;
  57   private int running = 0;
  58
  59   public synchronized boolean isFailed() {
  60     switch (state) {
  61       case ROLLINGBACK:
  62       case FAILED:
  63         return true;
  64       default:
  65         break;
  66     }
  67     return false;
  68   }
  69
  70   public synchronized boolean isRollingback() {
  71     return state == State.ROLLINGBACK;
  72   }
  73
  74   /**
  75    * Called by the ProcedureExecutor to mark rollback execution
  76    */
  77   protected synchronized boolean setRollback() {
  78     if (running == 0 && state == State.FAILED) {
  79       state = State.ROLLINGBACK;
  80       return true;
  81     }
  82     return false;
  83   }
  84
  85   /**
  86    * Called by the ProcedureExecutor to mark rollback execution
  87    */
  88   protected synchronized void unsetRollback() {
  89     assert state == State.ROLLINGBACK;
  90     state = State.FAILED;
  91   }
  92
  93   protected synchronized long[] getSubprocedureIds() {
  94     if (subprocs == null) return null;
  95     int index = 0;
  96     final long[] subIds = new long[subprocs.size()];
  97     for (Procedure proc: subprocs) {
  98       subIds[index++] = proc.getProcId();
  99     }
 100     return subIds;
 101   }
 102
 103   protected synchronized List<Procedure> getSubproceduresStack() {
 104     return subprocStack;
 105   }
 106
 107   protected synchronized RemoteProcedureException getException() {
 108     if (subprocStack != null) {
 109       for (Procedure proc: subprocStack) {
 110         if (proc.hasException()) {
 111           return proc.getException();
 112         }
 113       }
 114     }
 115     return null;
 116   }
 117
 118   /**
 119    * Called by the ProcedureExecutor to mark the procedure step as running.
 120    */
 121   protected synchronized boolean acquire(final Procedure proc) {
 122     if (state != State.RUNNING) return false;
 123
 124     running++;
 125     return true;
 126   }
 127
 128   /**
 129    * Called by the ProcedureExecutor to mark the procedure step as finished.
 130    */
 131   protected synchronized void release(final Procedure proc) {
 132     running--;
 133   }
 134
 135   protected synchronized void abort() {
 136     if (state == State.RUNNING) {
 137       state = State.FAILED;
 138     }
 139   }
 140
 141   /**
 142    * Called by the ProcedureExecutor after the procedure step is completed,
 143    * to add the step to the rollback list (or procedure stack)
 144    */
 145   protected synchronized void addRollbackStep(final Procedure proc) {
 146     if (proc.isFailed()) {
 147       state = State.FAILED;
 148     }
 149     if (subprocStack == null) {
 150       subprocStack = new ArrayList<>();
 151     }
 152     proc.addStackIndex(subprocStack.size());
 153     subprocStack.add(proc);
 154   }
 155
 156   protected synchronized void addSubProcedure(final Procedure proc) {
 157     if (!proc.hasParent()) return;
 158     if (subprocs == null) {
 159       subprocs = new HashSet<>();
 160     }
 161     subprocs.add(proc);
 162   }
 163
 164   /**
 165    * Called on store load by the ProcedureExecutor to load part of the stack.
 166    *
 167    * Each procedure has its own stack-positions. Which means we have to write
 168    * to the store only the Procedure we executed, and nothing else.
 169    * on load we recreate the full stack by aggregating each procedure stack-positions.
 170    */
 171   protected synchronized void loadStack(final Procedure proc) {
 172     addSubProcedure(proc);
 173     int[] stackIndexes = proc.getStackIndexes();
 174     if (stackIndexes != null) {
 175       if (subprocStack == null) {
 176         subprocStack = new ArrayList<>();
 177       }
 178       int diff = (1 + stackIndexes[stackIndexes.length - 1]) - subprocStack.size();
 179       if (diff > 0) {
 180         subprocStack.ensureCapacity(1 + stackIndexes[stackIndexes.length - 1]);
 181         while (diff-- > 0) subprocStack.add(null);
 182       }
 183       for (int i = 0; i < stackIndexes.length; ++i) {
 184         subprocStack.set(stackIndexes[i], proc);
 185       }
 186     }
 187     if (proc.getState() == ProcedureState.ROLLEDBACK) {
 188       state = State.ROLLINGBACK;
 189     } else if (proc.isFailed()) {
 190       state = State.FAILED;
 191     }
 192   }
 193
 194   /**
 195    * Called on store load by the ProcedureExecutor to validate the procedure stack.
 196    */
 197   protected synchronized boolean isValid() {
 198     if (subprocStack != null) {
 199       for (Procedure proc: subprocStack) {
 200         if (proc == null) {
 201           return false;
 202         }
 203       }
 204     }
 205     return true;
 206   }
 207 }