001/** 002 * Licensed to the Apache Software Foundation (ASF) under one 003 * or more contributor license agreements. See the NOTICE file 004 * distributed with this work for additional information 005 * regarding copyright ownership. The ASF licenses this file 006 * to you under the Apache License, Version 2.0 (the 007 * "License"); you may not use this file except in compliance 008 * with the License. You may obtain a copy of the License at 009 * 010 * http://www.apache.org/licenses/LICENSE-2.0 011 * 012 * Unless required by applicable law or agreed to in writing, software 013 * distributed under the License is distributed on an "AS IS" BASIS, 014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 015 * See the License for the specific language governing permissions and 016 * limitations under the License. 017 */ 018package org.apache.hadoop.hbase.procedure2; 019 020import java.io.IOException; 021import java.util.ArrayList; 022import java.util.Arrays; 023import java.util.List; 024import java.util.concurrent.atomic.AtomicBoolean; 025 026import org.apache.yetus.audience.InterfaceAudience; 027import org.apache.yetus.audience.InterfaceStability; 028import org.slf4j.Logger; 029import org.slf4j.LoggerFactory; 030 031import org.apache.hbase.thirdparty.com.google.common.annotations.VisibleForTesting; 032 033import org.apache.hadoop.hbase.shaded.protobuf.generated.ProcedureProtos.StateMachineProcedureData; 034 035/** 036 * Procedure described by a series of steps. 037 * 038 * <p>The procedure implementor must have an enum of 'states', describing 039 * the various step of the procedure. 040 * Once the procedure is running, the procedure-framework will call executeFromState() 041 * using the 'state' provided by the user. The first call to executeFromState() 042 * will be performed with 'state = null'. The implementor can jump between 043 * states using setNextState(MyStateEnum.ordinal()). 044 * The rollback will call rollbackState() for each state that was executed, in reverse order. 045 */ 046@InterfaceAudience.Private 047@InterfaceStability.Evolving 048public abstract class StateMachineProcedure<TEnvironment, TState> 049 extends Procedure<TEnvironment> { 050 private static final Logger LOG = LoggerFactory.getLogger(StateMachineProcedure.class); 051 052 private static final int EOF_STATE = Integer.MIN_VALUE; 053 054 private final AtomicBoolean aborted = new AtomicBoolean(false); 055 056 private Flow stateFlow = Flow.HAS_MORE_STATE; 057 protected int stateCount = 0; 058 private int[] states = null; 059 060 private List<Procedure<TEnvironment>> subProcList = null; 061 062 protected final int getCycles() { 063 return cycles; 064 } 065 066 /** 067 * Cycles on same state. Good for figuring if we are stuck. 068 */ 069 private int cycles = 0; 070 071 /** 072 * Ordinal of the previous state. So we can tell if we are progressing or not. 073 */ 074 private int previousState; 075 076 @VisibleForTesting 077 public enum Flow { 078 HAS_MORE_STATE, 079 NO_MORE_STATE, 080 } 081 082 /** 083 * called to perform a single step of the specified 'state' of the procedure 084 * @param state state to execute 085 * @return Flow.NO_MORE_STATE if the procedure is completed, 086 * Flow.HAS_MORE_STATE if there is another step. 087 */ 088 protected abstract Flow executeFromState(TEnvironment env, TState state) 089 throws ProcedureSuspendedException, ProcedureYieldException, InterruptedException; 090 091 /** 092 * called to perform the rollback of the specified state 093 * @param state state to rollback 094 * @throws IOException temporary failure, the rollback will retry later 095 */ 096 protected abstract void rollbackState(TEnvironment env, TState state) 097 throws IOException, InterruptedException; 098 099 /** 100 * Convert an ordinal (or state id) to an Enum (or more descriptive) state object. 101 * @param stateId the ordinal() of the state enum (or state id) 102 * @return the state enum object 103 */ 104 protected abstract TState getState(int stateId); 105 106 /** 107 * Convert the Enum (or more descriptive) state object to an ordinal (or state id). 108 * @param state the state enum object 109 * @return stateId the ordinal() of the state enum (or state id) 110 */ 111 protected abstract int getStateId(TState state); 112 113 /** 114 * Return the initial state object that will be used for the first call to executeFromState(). 115 * @return the initial state enum object 116 */ 117 protected abstract TState getInitialState(); 118 119 /** 120 * Set the next state for the procedure. 121 * @param state the state enum object 122 */ 123 protected void setNextState(final TState state) { 124 setNextState(getStateId(state)); 125 failIfAborted(); 126 } 127 128 /** 129 * By default, the executor will try ro run all the steps of the procedure start to finish. 130 * Return true to make the executor yield between execution steps to 131 * give other procedures time to run their steps. 132 * @param state the state we are going to execute next. 133 * @return Return true if the executor should yield before the execution of the specified step. 134 * Defaults to return false. 135 */ 136 protected boolean isYieldBeforeExecuteFromState(TEnvironment env, TState state) { 137 return false; 138 } 139 140 /** 141 * Add a child procedure to execute 142 * @param subProcedure the child procedure 143 */ 144 protected <T extends Procedure<TEnvironment>> void addChildProcedure( 145 @SuppressWarnings("unchecked") T... subProcedure) { 146 if (subProcedure == null) { 147 return; 148 } 149 final int len = subProcedure.length; 150 if (len == 0) { 151 return; 152 } 153 if (subProcList == null) { 154 subProcList = new ArrayList<>(len); 155 } 156 for (int i = 0; i < len; ++i) { 157 Procedure<TEnvironment> proc = subProcedure[i]; 158 if (!proc.hasOwner()) { 159 proc.setOwner(getOwner()); 160 } 161 162 subProcList.add(proc); 163 } 164 } 165 166 @Override 167 protected Procedure[] execute(final TEnvironment env) 168 throws ProcedureSuspendedException, ProcedureYieldException, InterruptedException { 169 updateTimestamp(); 170 try { 171 failIfAborted(); 172 173 if (!hasMoreState() || isFailed()) { 174 return null; 175 } 176 177 TState state = getCurrentState(); 178 if (stateCount == 0) { 179 setNextState(getStateId(state)); 180 } 181 182 if (LOG.isTraceEnabled()) { 183 LOG.trace(state + " " + this + "; cycles=" + this.cycles); 184 } 185 // Keep running count of cycles 186 if (getStateId(state) != this.previousState) { 187 this.previousState = getStateId(state); 188 this.cycles = 0; 189 } else { 190 this.cycles++; 191 } 192 193 LOG.trace("{}", this); 194 stateFlow = executeFromState(env, state); 195 if (!hasMoreState()) { 196 setNextState(EOF_STATE); 197 } 198 199 if (subProcList != null && !subProcList.isEmpty()) { 200 Procedure[] subProcedures = subProcList.toArray(new Procedure[subProcList.size()]); 201 subProcList = null; 202 return subProcedures; 203 } 204 return (isWaiting() || isFailed() || !hasMoreState()) ? null : new Procedure[] {this}; 205 } finally { 206 updateTimestamp(); 207 } 208 } 209 210 @Override 211 protected void rollback(final TEnvironment env) 212 throws IOException, InterruptedException { 213 if (isEofState()) { 214 stateCount--; 215 } 216 217 try { 218 updateTimestamp(); 219 rollbackState(env, getCurrentState()); 220 } finally { 221 stateCount--; 222 updateTimestamp(); 223 } 224 } 225 226 protected boolean isEofState() { 227 return stateCount > 0 && states[stateCount-1] == EOF_STATE; 228 } 229 230 @Override 231 protected boolean abort(final TEnvironment env) { 232 LOG.debug("Abort requested for {}", this); 233 if (!hasMoreState()) { 234 LOG.warn("Ignore abort request on {} because it has already been finished", this); 235 return false; 236 } 237 if (!isRollbackSupported(getCurrentState())) { 238 LOG.warn("Ignore abort request on {} because it does not support rollback", this); 239 return false; 240 } 241 aborted.set(true); 242 return true; 243 } 244 245 /** 246 * If procedure has more states then abort it otherwise procedure is finished and abort can be 247 * ignored. 248 */ 249 protected final void failIfAborted() { 250 if (aborted.get()) { 251 if (hasMoreState()) { 252 setAbortFailure(getClass().getSimpleName(), "abort requested"); 253 } else { 254 LOG.warn("Ignoring abort request on state='" + getCurrentState() + "' for " + this); 255 } 256 } 257 } 258 259 /** 260 * Used by the default implementation of abort() to know if the current state can be aborted 261 * and rollback can be triggered. 262 */ 263 protected boolean isRollbackSupported(final TState state) { 264 return false; 265 } 266 267 @Override 268 protected boolean isYieldAfterExecutionStep(final TEnvironment env) { 269 return isYieldBeforeExecuteFromState(env, getCurrentState()); 270 } 271 272 private boolean hasMoreState() { 273 return stateFlow != Flow.NO_MORE_STATE; 274 } 275 276 protected TState getCurrentState() { 277 return stateCount > 0 ? getState(states[stateCount-1]) : getInitialState(); 278 } 279 280 /** 281 * This method is used from test code as it cannot be assumed that state transition will happen 282 * sequentially. Some procedures may skip steps/ states, some may add intermediate steps in 283 * future. 284 */ 285 @VisibleForTesting 286 public int getCurrentStateId() { 287 return getStateId(getCurrentState()); 288 } 289 290 /** 291 * Set the next state for the procedure. 292 * @param stateId the ordinal() of the state enum (or state id) 293 */ 294 private void setNextState(final int stateId) { 295 if (states == null || states.length == stateCount) { 296 int newCapacity = stateCount + 8; 297 if (states != null) { 298 states = Arrays.copyOf(states, newCapacity); 299 } else { 300 states = new int[newCapacity]; 301 } 302 } 303 states[stateCount++] = stateId; 304 } 305 306 @Override 307 protected void toStringState(StringBuilder builder) { 308 super.toStringState(builder); 309 if (!isFinished() && !isEofState() && getCurrentState() != null) { 310 builder.append(":").append(getCurrentState()); 311 } 312 } 313 314 @Override 315 protected void serializeStateData(ProcedureStateSerializer serializer) 316 throws IOException { 317 StateMachineProcedureData.Builder data = StateMachineProcedureData.newBuilder(); 318 for (int i = 0; i < stateCount; ++i) { 319 data.addState(states[i]); 320 } 321 serializer.serialize(data.build()); 322 } 323 324 @Override 325 protected void deserializeStateData(ProcedureStateSerializer serializer) 326 throws IOException { 327 StateMachineProcedureData data = serializer.deserialize(StateMachineProcedureData.class); 328 stateCount = data.getStateCount(); 329 if (stateCount > 0) { 330 states = new int[stateCount]; 331 for (int i = 0; i < stateCount; ++i) { 332 states[i] = data.getState(i); 333 } 334 if (isEofState()) { 335 stateFlow = Flow.NO_MORE_STATE; 336 } 337 } else { 338 states = null; 339 } 340 } 341}