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.util.ArrayList;
021import java.util.Collection;
022import java.util.HashSet;
023import java.util.List;
024import java.util.Set;
025import org.apache.yetus.audience.InterfaceAudience;
026import org.apache.yetus.audience.InterfaceStability;
027import org.slf4j.Logger;
028import org.slf4j.LoggerFactory;
029
030import org.apache.hadoop.hbase.shaded.protobuf.generated.ProcedureProtos.ProcedureState;
031
032/**
033 * Internal state of the ProcedureExecutor that describes the state of a "Root Procedure". A "Root
034 * Procedure" is a Procedure without parent, each subprocedure will be added to the "Root Procedure"
035 * stack (or rollback-stack). RootProcedureState is used and managed only by the ProcedureExecutor.
036 *
037 * <pre>
038 *   Long rootProcId = getRootProcedureId(proc);
039 *   rollbackStack.get(rootProcId).acquire(proc)
040 *   rollbackStack.get(rootProcId).release(proc)
041 *   ...
042 * </pre>
043 */
044@InterfaceAudience.Private
045@InterfaceStability.Evolving
046class RootProcedureState<TEnvironment> {
047  private static final Logger LOG = LoggerFactory.getLogger(RootProcedureState.class);
048
049  private enum State {
050    RUNNING, // The Procedure is running or ready to run
051    FAILED, // The Procedure failed, waiting for the rollback executing
052    ROLLINGBACK, // The Procedure failed and the execution was rolledback
053  }
054
055  private Set<Procedure<TEnvironment>> subprocs = null;
056  private ArrayList<Procedure<TEnvironment>> subprocStack = null;
057  private State state = State.RUNNING;
058  private int running = 0;
059  // for some procedures such as SCP and TRSP, there is no way to rollback, so we do not need to
060  // maintain the rollback steps
061  // TODO: the rollback logic is a bit complicated, so here we will only test whether the root
062  // procedure supports rollback at the very beginning, actually, lots of procedure can only
063  // rollback at the pre check step, after that there is no rollback too, we should try to support
064  // this too.
065  private boolean rollbackSupported;
066
067  protected synchronized boolean isFailed() {
068    switch (state) {
069      case ROLLINGBACK:
070      case FAILED:
071        return true;
072      default:
073        break;
074    }
075    return false;
076  }
077
078  protected synchronized boolean isRollingback() {
079    return state == State.ROLLINGBACK;
080  }
081
082  /**
083   * Called by the ProcedureExecutor to mark rollback execution
084   */
085  protected synchronized boolean setRollback() {
086    if (running == 0 && state == State.FAILED) {
087      state = State.ROLLINGBACK;
088      return true;
089    }
090    return false;
091  }
092
093  /**
094   * Called by the ProcedureExecutor to mark rollback execution
095   */
096  protected synchronized void unsetRollback() {
097    assert state == State.ROLLINGBACK;
098    state = State.FAILED;
099  }
100
101  protected synchronized void setRollbackSupported(boolean rollbackSupported) {
102    this.rollbackSupported = rollbackSupported;
103  }
104
105  protected synchronized boolean isRollbackSupported() {
106    return rollbackSupported;
107  }
108
109  protected synchronized long[] getSubprocedureIds() {
110    if (subprocs == null) {
111      return null;
112    }
113    return subprocs.stream().mapToLong(Procedure::getProcId).toArray();
114  }
115
116  protected synchronized Collection<Procedure<TEnvironment>> getSubprocs() {
117    return subprocs;
118  }
119
120  protected synchronized List<Procedure<TEnvironment>> getSubproceduresStack() {
121    return subprocStack;
122  }
123
124  protected synchronized RemoteProcedureException getException() {
125    if (subprocs != null) {
126      for (Procedure<TEnvironment> proc : subprocs) {
127        if (proc.hasException()) {
128          return proc.getException();
129        }
130      }
131    }
132    return null;
133  }
134
135  /**
136   * Called by the ProcedureExecutor to mark the procedure step as running.
137   */
138  protected synchronized boolean acquire(Procedure<TEnvironment> proc) {
139    if (state != State.RUNNING) {
140      return false;
141    }
142
143    running++;
144    return true;
145  }
146
147  /**
148   * Called by the ProcedureExecutor to mark the procedure step as finished.
149   */
150  protected synchronized void release(Procedure<TEnvironment> proc) {
151    running--;
152  }
153
154  protected synchronized void abort() {
155    if (state == State.RUNNING) {
156      state = State.FAILED;
157    }
158  }
159
160  /**
161   * Called by the ProcedureExecutor after the procedure step is completed, to add the step to the
162   * rollback list (or procedure stack).
163   * <p>
164   * Return whether we successfully added the rollback step. If the root procedure has already
165   * crossed the PONR, we do not need to maintain the rollback step,
166   */
167  protected synchronized boolean addRollbackStep(Procedure<TEnvironment> proc) {
168    if (proc.isFailed()) {
169      state = State.FAILED;
170    }
171    if (!rollbackSupported) {
172      // just record executed, skip adding rollback step
173      proc.setExecuted();
174      return false;
175    }
176    if (subprocStack == null) {
177      subprocStack = new ArrayList<>();
178    }
179    proc.addStackIndex(subprocStack.size());
180    LOG.trace("Add procedure {} as the {}th rollback step", proc, subprocStack.size());
181    subprocStack.add(proc);
182    return true;
183  }
184
185  protected synchronized void addSubProcedure(Procedure<TEnvironment> proc) {
186    if (!proc.hasParent()) {
187      return;
188    }
189    if (subprocs == null) {
190      subprocs = new HashSet<>();
191    }
192    subprocs.add(proc);
193  }
194
195  /**
196   * Called on store load by the ProcedureExecutor to load part of the stack. Each procedure has its
197   * own stack-positions. Which means we have to write to the store only the Procedure we executed,
198   * and nothing else. on load we recreate the full stack by aggregating each procedure
199   * stack-positions.
200   */
201  protected synchronized void loadStack(Procedure<TEnvironment> proc) {
202    addSubProcedure(proc);
203    int[] stackIndexes = proc.getStackIndexes();
204    if (stackIndexes != null) {
205      if (subprocStack == null) {
206        subprocStack = new ArrayList<>();
207      }
208      int diff = (1 + stackIndexes[stackIndexes.length - 1]) - subprocStack.size();
209      if (diff > 0) {
210        subprocStack.ensureCapacity(1 + stackIndexes[stackIndexes.length - 1]);
211        while (diff-- > 0) {
212          subprocStack.add(null);
213        }
214      }
215      for (int i = 0; i < stackIndexes.length; ++i) {
216        subprocStack.set(stackIndexes[i], proc);
217      }
218    }
219    if (proc.getState() == ProcedureState.ROLLEDBACK) {
220      state = State.ROLLINGBACK;
221    } else if (proc.isFailed()) {
222      state = State.FAILED;
223    }
224  }
225
226  /**
227   * Called on store load by the ProcedureExecutor to validate the procedure stack.
228   */
229  protected synchronized boolean isValid() {
230    if (subprocStack != null) {
231      for (Procedure<TEnvironment> proc : subprocStack) {
232        if (proc == null) {
233          return false;
234        }
235      }
236    }
237    return true;
238  }
239}