You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@iotdb.apache.org by GitBox <gi...@apache.org> on 2022/04/18 03:11:11 UTC
[GitHub] [iotdb] mychaow commented on a diff in pull request #5477: [IOTDB-2880]Add procedure framework
mychaow commented on code in PR #5477:
URL: https://github.com/apache/iotdb/pull/5477#discussion_r851842530
##########
procedure/src/main/java/org/apache/iotdb/procedure/scheduler/ProcedureScheduler.java:
##########
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.procedure.scheduler;
+
+import org.apache.iotdb.procedure.Procedure;
+
+import java.util.concurrent.TimeUnit;
+
+/** Keep track of the runnable procedures */
+public interface ProcedureScheduler {
+ /** Start the scheduler */
+ void start();
+
+ /** Stop the scheduler */
+ void stop();
+
+ /**
+ * In case the class is blocking on poll() waiting for items to be added, this method should awake
+ * poll() and poll() should return.
+ */
+ void signalAll();
+
+ /**
+ * Inserts the specified element at the front of this queue.
+ *
+ * @param proc the Procedure to add
+ */
+ void addFront(Procedure proc);
+
+ /**
+ * Inserts the specified element at the front of this queue.
+ *
+ * @param proc the Procedure to add
+ * @param notify whether need to notify worker
+ */
+ void addFront(Procedure proc, boolean notify);
+
+ /**
+ * Inserts the specified element at the end of this queue.
+ *
+ * @param proc the Procedure to add
+ */
+ void addBack(Procedure proc);
+
+ /**
+ * Inserts the specified element at the end of this queue.
+ *
+ * @param proc the Procedure to add
+ * @param notify whether need to notify worker
+ */
+ void addBack(Procedure proc, boolean notify);
+
+ /**
+ * The procedure can't run at the moment. add it back to the queue, giving priority to someone
+ * else.
+ *
+ * @param proc the Procedure to add back to the list
+ */
+ void yield(Procedure proc);
+
+ /**
+ * The procedure in execution completed. This can be implemented to perform cleanups.
+ *
+ * @param proc the Procedure that completed the execution.
+ */
+ void completionCleanup(Procedure proc);
Review Comment:
In your design, I think clear the completed procedure should be in executor, scheduler only provide the interface push and poll. So, a task which is running does not in scheduler, but executor.
##########
procedure/src/main/java/org/apache/iotdb/procedure/store/ProcedureWAL.java:
##########
@@ -0,0 +1,106 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.procedure.store;
+
+import org.apache.iotdb.procedure.Procedure;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.*;
+import java.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.List;
+
+public class ProcedureWAL {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ProcedureWAL.class);
+
+ private static final String TMP_SUFFIX = ".tmp";
+ private Path walFilePath;
+ private ByteBuffer byteBuffer;
+
+ public ByteBuffer getByteBuffer() {
+ return byteBuffer;
+ }
+
+ public ProcedureWAL(Path walFilePath, ByteBuffer byteBuffer) {
+ this.walFilePath = walFilePath;
+ this.byteBuffer = byteBuffer;
+ }
+
+ /**
+ * Create a wal file
+ *
+ * @throws IOException ioe
+ */
+ public void save() throws IOException {
+ File walTmp = new File(walFilePath + TMP_SUFFIX);
+ Path walTmpPath = walTmp.toPath();
+ Files.deleteIfExists(walTmpPath);
+ Files.createFile(walTmpPath);
+ try (FileOutputStream fos = new FileOutputStream(walTmp);
+ FileChannel channel = fos.getChannel()) {
+ byteBuffer.flip();
+ channel.write(byteBuffer);
+ }
+ Files.deleteIfExists(walTmpPath);
Review Comment:
why delete tmp path before move?
##########
procedure/src/main/java/org/apache/iotdb/procedure/RootProcedureStack.java:
##########
@@ -0,0 +1,189 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.procedure;
+
+import org.apache.iotdb.procedure.exception.ProcedureException;
+import org.apache.iotdb.service.rpc.thrift.ProcedureState;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+public class RootProcedureStack<Env> {
+ private static final Logger LOG = LoggerFactory.getLogger(RootProcedureStack.class);
+
+ private enum State {
+ RUNNING, // The Procedure is running or ready to run
+ FAILED, // The Procedure failed, waiting for the rollback executing
+ ROLLINGBACK, // The Procedure failed and the execution was rolledback
Review Comment:
rolling or rolled?
##########
procedure/src/main/java/org/apache/iotdb/procedure/ProcedureExecutor.java:
##########
@@ -0,0 +1,990 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.procedure;
+
+import org.apache.iotdb.commons.concurrent.IoTDBDaemonThreadFactory;
+import org.apache.iotdb.procedure.conf.ProcedureNodeConfigDescriptor;
+import org.apache.iotdb.procedure.exception.ProcedureException;
+import org.apache.iotdb.procedure.exception.ProcedureSuspendedException;
+import org.apache.iotdb.procedure.exception.ProcedureYieldException;
+import org.apache.iotdb.procedure.scheduler.ProcedureScheduler;
+import org.apache.iotdb.procedure.scheduler.SimpleProcedureScheduler;
+import org.apache.iotdb.procedure.store.IProcedureStore;
+import org.apache.iotdb.service.rpc.thrift.ProcedureState;
+
+import com.google.common.base.Preconditions;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.*;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.locks.ReentrantLock;
+
+public class ProcedureExecutor<Env> {
+ private static final Logger LOG = LoggerFactory.getLogger(ProcedureExecutor.class);
+
+ static final int EVICT_TTL =
+ ProcedureNodeConfigDescriptor.getInstance().getConf().getCompletedEvictTTL();
+
+ private final ConcurrentHashMap<Long, ReentrantLock> idLockMap = new ConcurrentHashMap<>();
+
+ private final ConcurrentHashMap<Long, CompletedProcedureRetainer<Env>> completed =
+ new ConcurrentHashMap<>();
+
+ private final ConcurrentHashMap<Long, RootProcedureStack<Env>> rollbackStack =
+ new ConcurrentHashMap<>();
+
+ private final ConcurrentHashMap<Long, Procedure> procedures = new ConcurrentHashMap<>();
+
+ private ThreadGroup threadGroup;
+
+ private CopyOnWriteArrayList<WorkerThread> workerThreads;
+
+ private TimeoutExecutorThread<Env> timeoutExecutor;
+
+ private TimeoutExecutorThread<Env> workerMonitorExecutor;
+
+ private int corePoolSize;
+ private int maxPoolSize;
+ private volatile long keepAliveTime;
+
+ private final ProcedureScheduler scheduler;
+
+ private final Executor forceUpdateExecutor =
+ Executors.newSingleThreadExecutor(
+ new IoTDBDaemonThreadFactory("Force-Update-Procedure-Thread"));
+
+ private final AtomicLong lastProcId = new AtomicLong(-1);
+ private final AtomicLong workId = new AtomicLong(0);
+ private final AtomicInteger activeExecutorCount = new AtomicInteger(0);
+ private final AtomicBoolean running = new AtomicBoolean(false);
+ private final Env environment;
+ private final IProcedureStore store;
+
+ public ProcedureExecutor(
+ final Env environment, final IProcedureStore store, final ProcedureScheduler scheduler) {
+ this.environment = environment;
+ this.scheduler = scheduler;
+ this.store = store;
+ this.lastProcId.incrementAndGet();
+ }
+
+ public ProcedureExecutor(final Env environment, final IProcedureStore store) {
+ this(environment, store, new SimpleProcedureScheduler());
+ }
+
+ public void init(int numThreads) {
+ this.corePoolSize = numThreads;
+ this.maxPoolSize = 10 * numThreads;
+ this.threadGroup = new ThreadGroup("ProcedureWorkerGroup");
+ this.timeoutExecutor =
+ new TimeoutExecutorThread<>(this, threadGroup, "ProcedureTimeoutExecutor");
+ this.workerMonitorExecutor =
+ new TimeoutExecutorThread<>(this, threadGroup, "WorkerThreadMonitor");
Review Comment:
WorkerThreadMonitor -> ProcedureWorkerThreadMonitor is better
##########
thrift-procedure/src/main/thrift/procedure.thrift:
##########
@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+namespace java org.apache.iotdb.service.rpc.thrift
+namespace py iotdb.thrift.rpc
+
+enum ProcedureState {
+ INITIALIZING = 1,
+ RUNNABLE = 2,
+ WAITING = 3,
+ WAITING_TIMEOUT = 4,
+ ROLLEDBACK = 5,
+ SUCCESS = 6,
+ FAILED = 7
+}
+
+struct SubmitProcedureReq{
Review Comment:
It's a big feature, so I will split the comment to serveral parts. Part one: 1. mvn spotless:apply ? seems that the format it not unified
##########
procedure/src/main/java/org/apache/iotdb/procedure/store/ProcedureWAL.java:
##########
@@ -0,0 +1,106 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.procedure.store;
+
+import org.apache.iotdb.procedure.Procedure;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.*;
+import java.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.List;
+
+public class ProcedureWAL {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ProcedureWAL.class);
+
+ private static final String TMP_SUFFIX = ".tmp";
+ private Path walFilePath;
+ private ByteBuffer byteBuffer;
+
+ public ByteBuffer getByteBuffer() {
+ return byteBuffer;
+ }
+
+ public ProcedureWAL(Path walFilePath, ByteBuffer byteBuffer) {
+ this.walFilePath = walFilePath;
+ this.byteBuffer = byteBuffer;
+ }
+
+ /**
+ * Create a wal file
+ *
+ * @throws IOException ioe
+ */
+ public void save() throws IOException {
+ File walTmp = new File(walFilePath + TMP_SUFFIX);
+ Path walTmpPath = walTmp.toPath();
+ Files.deleteIfExists(walTmpPath);
+ Files.createFile(walTmpPath);
+ try (FileOutputStream fos = new FileOutputStream(walTmp);
+ FileChannel channel = fos.getChannel()) {
+ byteBuffer.flip();
+ channel.write(byteBuffer);
+ }
+ Files.deleteIfExists(walTmpPath);
+ Files.move(walTmpPath, walFilePath);
+ }
+
+ /**
+ * Load wal files into memory
+ *
+ * @param procedureList procedure list
+ */
+ public void load(List<Procedure> procedureList) {
+ if (procedureList == null || procedureList.isEmpty()) {
+ LOG.info("None procedure wal found.");
+ return;
+ }
+ Procedure procedure = null;
+ try (FileInputStream fis = new FileInputStream(walFilePath.toFile());
+ FileChannel channel = fis.getChannel()) {
+ while (channel.read(byteBuffer) > 0) {
+ byteBuffer.flip();
+ procedure = Procedure.newInstance(byteBuffer);
Review Comment:
1. Should you check the byteBuffer's size?
2. If a procedure's size is less than bytebuffer's size, after first time deserialization, we clear the bytebuffer, can we deserialize the following procedure?
##########
procedure/src/main/java/org/apache/iotdb/procedure/StoppableThread.java:
##########
@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.procedure;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public abstract class StoppableThread extends Thread {
+
+ private static final Logger LOG = LoggerFactory.getLogger(StoppableThread.class);
+
+ public StoppableThread(ThreadGroup threadGroup, String name) {
+ super(threadGroup, name);
+ }
+
+ public abstract void sendStopSignal();
+
+ public void awaitTermination() {
+ try {
+ for (int i = 0; isAlive(); i++) {
+ sendStopSignal();
+ join(250);
Review Comment:
create contant variable in the class?
##########
procedure/src/main/java/org/apache/iotdb/procedure/Procedure.java:
##########
@@ -0,0 +1,909 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.procedure;
+
+import org.apache.iotdb.procedure.exception.*;
+import org.apache.iotdb.procedure.store.IProcedureStore;
+import org.apache.iotdb.service.rpc.thrift.ProcedureState;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.lang.reflect.InvocationTargetException;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Abstract class of all procedures.
+ *
+ * @param <Env>
+ */
+public abstract class Procedure<Env> implements Comparable<Procedure<Env>> {
+ private static final Logger LOG = LoggerFactory.getLogger(Procedure.class);
+ public static final long NO_PROC_ID = -1;
+ public static final long NO_TIMEOUT = -1;
+
+ private long parentProcId = NO_PROC_ID;
+ private long rootProcId = NO_PROC_ID;
+ private long procId = NO_PROC_ID;
+ private long submittedTime;
+
+ private ProcedureState state = ProcedureState.INITIALIZING;
+ private int childrenLatch = 0;
+ private ProcedureException exception;
+
+ private volatile long timeout = NO_TIMEOUT;
+ private volatile long lastUpdate;
+
+ private volatile byte[] result = null;
+ private volatile boolean locked = false;
+ private boolean lockedWhenLoading = false;
+
+ private volatile boolean bypass = false;
+ private int[] stackIndexes = null;
+
+ public boolean isBypass() {
+ return bypass;
+ }
+
+ public void bypass(boolean bypass) {
+ this.bypass = bypass;
+ }
+
+ private boolean persist = true;
+
+ public boolean needPersistance() {
+ return this.persist;
+ }
+
+ public void resetPersistance() {
+ this.persist = true;
+ }
+
+ public final void skipPersistance() {
+ this.persist = false;
+ }
+
+ public final boolean hasLock() {
+ return locked;
+ }
+
+ // User level code, override it if necessary
+
+ /**
+ * The main code of the procedure. It must be idempotent since execute() may be called multiple
+ * times in case of machine failure in the middle of the execution.
+ *
+ * @param env the environment passed to the ProcedureExecutor
+ * @return a set of sub-procedures to run or ourselves if there is more work to do or null if the
+ * procedure is done.
+ * @throws ProcedureYieldException the procedure will be added back to the queue and retried
+ * later.
+ * @throws InterruptedException the procedure will be added back to the queue and retried later.
+ * @throws ProcedureSuspendedException Signal to the executor that Procedure has suspended itself
+ * and has set itself up waiting for an external event to wake it back up again.
+ */
+ protected abstract Procedure<Env>[] execute(Env env)
+ throws ProcedureYieldException, ProcedureSuspendedException, InterruptedException;
+
+ /**
+ * The code to undo what was done by the execute() code. It is called when the procedure or one of
+ * the sub-procedures failed or an abort was requested. It should cleanup all the resources
+ * created by the execute() call. The implementation must be idempotent since rollback() may be
+ * called multiple time in case of machine failure in the middle of the execution.
+ *
+ * @param env the environment passed to the ProcedureExecutor
+ * @throws IOException temporary failure, the rollback will retry later
+ * @throws InterruptedException the procedure will be added back to the queue and retried later
+ */
+ protected abstract void rollback(Env env) throws IOException, InterruptedException;
+
+ /**
+ * The abort() call is asynchronous and each procedure must decide how to deal with it, if they
+ * want to be abortable. The simplest implementation is to have an AtomicBoolean set in the
+ * abort() method and then the execute() will check if the abort flag is set or not. abort() may
+ * be called multiple times from the client, so the implementation must be idempotent.
+ *
+ * <p>NOTE: abort() is not like Thread.interrupt(). It is just a notification that allows the
+ * procedure implementor abort.
+ */
+ protected abstract boolean abort(Env env);
+
+ public void serialize(ByteBuffer byteBuffer) throws IOException {
+ // 1. class name
+ String className = this.getClass().getName();
+ byte[] classNameBytes = className.getBytes(StandardCharsets.UTF_8);
+ byteBuffer.putInt(classNameBytes.length);
+ byteBuffer.put(classNameBytes);
+ // procid
+ byteBuffer.putLong(this.procId);
+ // state
+ byteBuffer.putInt(this.state.getValue());
+ // submit time
+ byteBuffer.putLong(this.submittedTime);
+ // last updated
+ byteBuffer.putLong(this.lastUpdate);
+ // parent id
+ byteBuffer.putLong(this.parentProcId);
+ // time out
+ byteBuffer.putLong(this.timeout);
+ // stack indexes
+ if (stackIndexes != null) {
+ byteBuffer.putInt(stackIndexes.length);
+ Arrays.stream(stackIndexes).forEach(byteBuffer::putInt);
+ } else {
+ byteBuffer.putInt(-1);
+ }
+
+ // exceptions
+ if (hasException()) {
+ byteBuffer.putInt((byte) 1);
+ String exceptionClassName = exception.getClass().getName();
+ byte[] exceptionClassNameBytes = exceptionClassName.getBytes(StandardCharsets.UTF_8);
+ byteBuffer.putInt(exceptionClassNameBytes.length);
+ byteBuffer.put(exceptionClassNameBytes);
+ String message = this.exception.getMessage();
+ byte[] messageBytes = message.getBytes(StandardCharsets.UTF_8);
+ byteBuffer.putInt(messageBytes.length);
+ byteBuffer.put(messageBytes);
+ } else {
+ byteBuffer.put((byte) 0);
+ }
+
+ // result
+ if (result != null) {
+ byteBuffer.putInt(result.length);
+ byteBuffer.put(result);
+ } else {
+ byteBuffer.putInt(-1);
+ }
+
+ // has lock
+ byteBuffer.put(this.hasLock() ? (byte) 1 : (byte) 0);
+ // bypass
+ byteBuffer.put(this.bypass ? (byte) 1 : (byte) 0);
+ }
+
+ public void deserialize(ByteBuffer byteBuffer) throws IOException {
+ // procid
+ this.setProcId(byteBuffer.getLong());
+ // state
+ this.setState(ProcedureState.findByValue(byteBuffer.getInt()));
+ // submit time
+ this.setSubmittedTime(byteBuffer.getLong());
+ // last updated
+ this.setLastUpdate(byteBuffer.getLong());
+ // parent id
+ this.setParentProcId(byteBuffer.getLong());
+ // time out
+ this.setTimeout(byteBuffer.getLong());
+ // stack index
+ int stackIndexesLen = byteBuffer.getInt();
+ if (stackIndexesLen >= 0) {
+ List<Integer> indexList = new ArrayList<>(stackIndexesLen);
+ for (int i = 0; i < stackIndexesLen; i++) {
+ indexList.add(byteBuffer.getInt());
+ }
+ this.setStackIndexes(indexList);
+ }
+ // exceptions
+ if (byteBuffer.get() == 1) {
+ Class<?> exceptionClass = deserializeTypeInfo(byteBuffer);
+ int messageBytesLength = byteBuffer.getInt();
+ byte[] messageBytes = new byte[messageBytesLength];
+ byteBuffer.get(messageBytes);
+ String errMsg = new String(messageBytes, StandardCharsets.UTF_8);
+ ProcedureException exception;
+ try {
+ exception =
+ (ProcedureException) exceptionClass.getConstructor(String.class).newInstance(errMsg);
+ } catch (InstantiationException
+ | IllegalAccessException
+ | InvocationTargetException
+ | NoSuchMethodException e) {
+ LOG.warn("Instantiation exception class failed", e);
+ exception = new ProcedureException(errMsg);
+ }
+
+ setFailure(exception);
+ }
+ // has lock
+ if (byteBuffer.get() == 1) {
+ this.lockedWhenLoading();
+ }
+ // by pass
+ if (byteBuffer.get() == 1) {
+ this.bypass(true);
+ }
+ }
+
+ /**
+ * Deserialize class Name and load class
+ *
+ * @param byteBuffer bytebuffer
+ * @return Procedure
+ */
+ public static Class<?> deserializeTypeInfo(ByteBuffer byteBuffer) {
+ int classNameBytesLen = byteBuffer.getInt();
+ byte[] classNameBytes = new byte[classNameBytesLen];
+ byteBuffer.get(classNameBytes);
+ String className = new String(classNameBytes, StandardCharsets.UTF_8);
+ Class<?> clazz;
+ try {
+ clazz = Class.forName(className);
+ } catch (ClassNotFoundException e) {
+ throw new RuntimeException("Invalid procedure class", e);
Review Comment:
two blank char, haha
##########
procedure/src/main/java/org/apache/iotdb/procedure/Procedure.java:
##########
@@ -0,0 +1,909 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.procedure;
+
+import org.apache.iotdb.procedure.exception.*;
+import org.apache.iotdb.procedure.store.IProcedureStore;
+import org.apache.iotdb.service.rpc.thrift.ProcedureState;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.lang.reflect.InvocationTargetException;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Abstract class of all procedures.
+ *
+ * @param <Env>
+ */
+public abstract class Procedure<Env> implements Comparable<Procedure<Env>> {
+ private static final Logger LOG = LoggerFactory.getLogger(Procedure.class);
+ public static final long NO_PROC_ID = -1;
+ public static final long NO_TIMEOUT = -1;
+
+ private long parentProcId = NO_PROC_ID;
+ private long rootProcId = NO_PROC_ID;
+ private long procId = NO_PROC_ID;
+ private long submittedTime;
+
+ private ProcedureState state = ProcedureState.INITIALIZING;
+ private int childrenLatch = 0;
+ private ProcedureException exception;
+
+ private volatile long timeout = NO_TIMEOUT;
+ private volatile long lastUpdate;
+
+ private volatile byte[] result = null;
+ private volatile boolean locked = false;
+ private boolean lockedWhenLoading = false;
+
+ private volatile boolean bypass = false;
+ private int[] stackIndexes = null;
+
+ public boolean isBypass() {
+ return bypass;
+ }
+
+ public void bypass(boolean bypass) {
+ this.bypass = bypass;
+ }
+
+ private boolean persist = true;
+
+ public boolean needPersistance() {
+ return this.persist;
+ }
+
+ public void resetPersistance() {
+ this.persist = true;
+ }
+
+ public final void skipPersistance() {
+ this.persist = false;
+ }
+
+ public final boolean hasLock() {
+ return locked;
+ }
+
+ // User level code, override it if necessary
+
+ /**
+ * The main code of the procedure. It must be idempotent since execute() may be called multiple
+ * times in case of machine failure in the middle of the execution.
+ *
+ * @param env the environment passed to the ProcedureExecutor
+ * @return a set of sub-procedures to run or ourselves if there is more work to do or null if the
+ * procedure is done.
+ * @throws ProcedureYieldException the procedure will be added back to the queue and retried
+ * later.
+ * @throws InterruptedException the procedure will be added back to the queue and retried later.
+ * @throws ProcedureSuspendedException Signal to the executor that Procedure has suspended itself
+ * and has set itself up waiting for an external event to wake it back up again.
+ */
+ protected abstract Procedure<Env>[] execute(Env env)
+ throws ProcedureYieldException, ProcedureSuspendedException, InterruptedException;
+
+ /**
+ * The code to undo what was done by the execute() code. It is called when the procedure or one of
+ * the sub-procedures failed or an abort was requested. It should cleanup all the resources
+ * created by the execute() call. The implementation must be idempotent since rollback() may be
+ * called multiple time in case of machine failure in the middle of the execution.
+ *
+ * @param env the environment passed to the ProcedureExecutor
+ * @throws IOException temporary failure, the rollback will retry later
+ * @throws InterruptedException the procedure will be added back to the queue and retried later
+ */
+ protected abstract void rollback(Env env) throws IOException, InterruptedException;
+
+ /**
+ * The abort() call is asynchronous and each procedure must decide how to deal with it, if they
+ * want to be abortable. The simplest implementation is to have an AtomicBoolean set in the
+ * abort() method and then the execute() will check if the abort flag is set or not. abort() may
+ * be called multiple times from the client, so the implementation must be idempotent.
+ *
+ * <p>NOTE: abort() is not like Thread.interrupt(). It is just a notification that allows the
+ * procedure implementor abort.
+ */
+ protected abstract boolean abort(Env env);
+
+ public void serialize(ByteBuffer byteBuffer) throws IOException {
+ // 1. class name
+ String className = this.getClass().getName();
+ byte[] classNameBytes = className.getBytes(StandardCharsets.UTF_8);
+ byteBuffer.putInt(classNameBytes.length);
+ byteBuffer.put(classNameBytes);
+ // procid
+ byteBuffer.putLong(this.procId);
+ // state
+ byteBuffer.putInt(this.state.getValue());
+ // submit time
+ byteBuffer.putLong(this.submittedTime);
+ // last updated
+ byteBuffer.putLong(this.lastUpdate);
+ // parent id
+ byteBuffer.putLong(this.parentProcId);
+ // time out
+ byteBuffer.putLong(this.timeout);
+ // stack indexes
+ if (stackIndexes != null) {
+ byteBuffer.putInt(stackIndexes.length);
+ Arrays.stream(stackIndexes).forEach(byteBuffer::putInt);
+ } else {
+ byteBuffer.putInt(-1);
+ }
+
+ // exceptions
+ if (hasException()) {
+ byteBuffer.putInt((byte) 1);
+ String exceptionClassName = exception.getClass().getName();
+ byte[] exceptionClassNameBytes = exceptionClassName.getBytes(StandardCharsets.UTF_8);
+ byteBuffer.putInt(exceptionClassNameBytes.length);
+ byteBuffer.put(exceptionClassNameBytes);
+ String message = this.exception.getMessage();
+ byte[] messageBytes = message.getBytes(StandardCharsets.UTF_8);
+ byteBuffer.putInt(messageBytes.length);
+ byteBuffer.put(messageBytes);
+ } else {
+ byteBuffer.put((byte) 0);
+ }
+
+ // result
+ if (result != null) {
+ byteBuffer.putInt(result.length);
+ byteBuffer.put(result);
+ } else {
+ byteBuffer.putInt(-1);
+ }
+
+ // has lock
+ byteBuffer.put(this.hasLock() ? (byte) 1 : (byte) 0);
+ // bypass
+ byteBuffer.put(this.bypass ? (byte) 1 : (byte) 0);
+ }
+
+ public void deserialize(ByteBuffer byteBuffer) throws IOException {
+ // procid
+ this.setProcId(byteBuffer.getLong());
+ // state
+ this.setState(ProcedureState.findByValue(byteBuffer.getInt()));
+ // submit time
+ this.setSubmittedTime(byteBuffer.getLong());
+ // last updated
+ this.setLastUpdate(byteBuffer.getLong());
+ // parent id
+ this.setParentProcId(byteBuffer.getLong());
+ // time out
+ this.setTimeout(byteBuffer.getLong());
+ // stack index
+ int stackIndexesLen = byteBuffer.getInt();
+ if (stackIndexesLen >= 0) {
+ List<Integer> indexList = new ArrayList<>(stackIndexesLen);
+ for (int i = 0; i < stackIndexesLen; i++) {
+ indexList.add(byteBuffer.getInt());
+ }
+ this.setStackIndexes(indexList);
+ }
+ // exceptions
+ if (byteBuffer.get() == 1) {
+ Class<?> exceptionClass = deserializeTypeInfo(byteBuffer);
+ int messageBytesLength = byteBuffer.getInt();
+ byte[] messageBytes = new byte[messageBytesLength];
+ byteBuffer.get(messageBytes);
+ String errMsg = new String(messageBytes, StandardCharsets.UTF_8);
+ ProcedureException exception;
+ try {
+ exception =
+ (ProcedureException) exceptionClass.getConstructor(String.class).newInstance(errMsg);
+ } catch (InstantiationException
+ | IllegalAccessException
+ | InvocationTargetException
+ | NoSuchMethodException e) {
+ LOG.warn("Instantiation exception class failed", e);
+ exception = new ProcedureException(errMsg);
+ }
+
+ setFailure(exception);
+ }
+ // has lock
+ if (byteBuffer.get() == 1) {
+ this.lockedWhenLoading();
+ }
+ // by pass
+ if (byteBuffer.get() == 1) {
+ this.bypass(true);
+ }
+ }
+
+ /**
+ * Deserialize class Name and load class
+ *
+ * @param byteBuffer bytebuffer
+ * @return Procedure
+ */
+ public static Class<?> deserializeTypeInfo(ByteBuffer byteBuffer) {
+ int classNameBytesLen = byteBuffer.getInt();
+ byte[] classNameBytes = new byte[classNameBytesLen];
+ byteBuffer.get(classNameBytes);
+ String className = new String(classNameBytes, StandardCharsets.UTF_8);
+ Class<?> clazz;
+ try {
+ clazz = Class.forName(className);
+ } catch (ClassNotFoundException e) {
+ throw new RuntimeException("Invalid procedure class", e);
+ }
+ return clazz;
+ }
+
+ public static Procedure newInstance(ByteBuffer byteBuffer) {
+ Class<?> procedureClass = deserializeTypeInfo(byteBuffer);
+ Procedure procedure;
+ try {
+ procedure = (Procedure) procedureClass.newInstance();
+ } catch (InstantiationException | IllegalAccessException e) {
+ throw new RuntimeException("Instantiation failed", e);
+ }
+ return procedure;
+ }
+
+ /**
+ * The {@link #doAcquireLock(Object, IProcedureStore)} will be split into two steps, first, it
+ * will call us to determine whether we need to wait for initialization, second, it will call
+ * {@link #acquireLock(Object)} to actually handle the lock for this procedure.
+ *
+ * @return true means we need to wait until the environment has been initialized, otherwise true.
+ */
+ protected boolean waitInitialized(Env env) {
+ return false;
+ }
+
+ /**
+ * Acquire a lock, user should override it if necessary.
+ *
+ * @param env environment
+ * @return state of lock
+ */
+ protected ProcedureLockState acquireLock(Env env) {
+ return ProcedureLockState.LOCK_ACQUIRED;
+ }
+
+ /**
+ * Release a lock, user should override it if necessary.
+ *
+ * @param env env
+ */
+ protected void releaseLock(Env env) {
+ // no op
+ }
+
+ /**
+ * Used to keep procedure lock even when the procedure is yielded or suspended.
+ *
+ * @param env env
+ * @return true if hold the lock
+ */
+ protected boolean holdLock(Env env) {
+ return false;
+ }
+
+ /**
+ * Called before the procedure is recovered and added into the queue.
+ *
+ * @param env environment
+ */
+ protected final void beforeRecover(Env env) {
+ // no op
+ }
+
+ /**
+ * Called when the procedure is recovered and added into the queue.
+ *
+ * @param env environment
+ */
+ protected final void afterRecover(Env env) {
+ // no op
+ }
+
+ /**
+ * Called when the procedure is completed (success or rollback). The procedure may use this method
+ * to clean up in-memory states. This operation will not be retried on failure.
+ *
+ * @param env environment
+ */
+ protected void completionCleanup(Env env) {
+ // no op
+ }
+
+ /**
+ * To make executor yield between each execution step to give other procedures a chance to run.
+ *
+ * @param env environment
+ * @return return true if yield is allowed.
+ */
+ protected boolean isYieldAfterExecution(Env env) {
+ return false;
+ }
+
+ // -------------------------Internal methods - called by the procedureExecutor------------------
+ /**
+ * Internal method called by the ProcedureExecutor that starts the user-level code execute().
+ *
+ * @param env execute environment
+ * @return sub procedures
+ */
+ protected Procedure<Env>[] doExecute(Env env)
+ throws ProcedureYieldException, ProcedureSuspendedException, InterruptedException {
+ try {
+ updateTimestamp();
+ if (bypass) {
+ LOG.info("{} bypassed, returning null to finish it", this);
+ return null;
+ }
+ return execute(env);
+ } finally {
+ updateTimestamp();
+ }
+ }
+
+ /**
+ * Internal method called by the ProcedureExecutor that starts the user-level code rollback().
+ *
+ * @param env execute environment
+ * @throws IOException ioe
+ * @throws InterruptedException interrupted exception
+ */
+ public void doRollback(Env env) throws IOException, InterruptedException {
+ try {
+ updateTimestamp();
+ if (bypass) {
+ LOG.info("{} bypassed, skip rollback", this);
+ return;
+ }
+ rollback(env);
+ } finally {
+ updateTimestamp();
+ }
+ }
+
+ /**
+ * Internal method called by the ProcedureExecutor that starts the user-level code acquireLock().
+ *
+ * @param env environment
+ * @param store ProcedureStore
+ * @return ProcedureLockState
+ */
+ public final ProcedureLockState doAcquireLock(Env env, IProcedureStore store) {
+ if (waitInitialized(env)) {
+ return ProcedureLockState.LOCK_EVENT_WAIT;
+ }
+ if (lockedWhenLoading) {
+ lockedWhenLoading = false;
+ locked = true;
+ return ProcedureLockState.LOCK_ACQUIRED;
+ }
+ ProcedureLockState state = acquireLock(env);
+ if (state == ProcedureLockState.LOCK_ACQUIRED) {
+ locked = true;
+ store.update(this);
+ }
+ return state;
+ }
+
+ /**
+ * Presist lock state of the procedure
+ *
+ * @param env environment
+ * @param store ProcedureStore
+ */
+ public final void doReleaseLock(Env env, IProcedureStore store) {
+ locked = false;
+ if (getState() != ProcedureState.ROLLEDBACK) {
+ store.update(this);
+ }
+ releaseLock(env);
+ }
+
+ public final void restoreLock(Env env) {
+ if (!lockedWhenLoading) {
+ LOG.debug("{} didn't hold the lock before restarting, skip acquiring lock", this);
+ return;
+ }
+ if (isFinished()) {
+ LOG.debug("{} is already bypassed, skip acquiring lock.", this);
+ return;
+ }
+ if (getState() == ProcedureState.WAITING && !holdLock(env)) {
+ LOG.debug("{} is in WAITING STATE, and holdLock= false , skip acquiring lock.", this);
+ return;
+ }
+ LOG.debug("{} held the lock before restarting, call acquireLock to restore it.", this);
+ acquireLock(env);
+ }
+
+ @Override
+ public String toString() {
+ // Return the simple String presentation of the procedure.
+ return toStringSimpleSB().toString();
+ }
+
+ /**
+ * Build the StringBuilder for the simple form of procedure string.
+ *
+ * @return the StringBuilder
+ */
+ protected StringBuilder toStringSimpleSB() {
+ final StringBuilder sb = new StringBuilder();
+
+ sb.append("pid=");
+ sb.append(getProcId());
+
+ if (hasParent()) {
+ sb.append(", ppid=");
+ sb.append(getParentProcId());
+ }
+
+ /*
+ * TODO
+ * Enable later when this is being used.
+ * Currently owner not used.
+ if (hasOwner()) {
+ sb.append(", owner=");
+ sb.append(getOwner());
+ }*/
+
+ sb.append(", state="); // pState for Procedure State as opposed to any other kind.
+ toStringState(sb);
+
+ // Only print out locked if actually locked. Most of the time it is not.
+ if (this.locked) {
+ sb.append(", locked=").append(locked);
+ }
+
+ if (bypass) {
+ sb.append(", bypass=").append(bypass);
+ }
+
+ if (hasException()) {
+ sb.append(", exception=" + getException());
+ }
+
+ sb.append("; ");
+ toStringClassDetails(sb);
+
+ return sb;
+ }
+
+ /** Extend the toString() information with more procedure details */
+ public String toStringDetails() {
+ final StringBuilder sb = toStringSimpleSB();
+
+ sb.append(" submittedTime=");
+ sb.append(getSubmittedTime());
+
+ sb.append(", lastUpdate=");
+ sb.append(getLastUpdate());
+
+ final int[] stackIndices = getStackIndexes();
+ if (stackIndices != null) {
+ sb.append("\n");
+ sb.append("stackIndexes=");
+ sb.append(Arrays.toString(stackIndices));
+ }
+
+ return sb.toString();
+ }
+
+ protected String toStringClass() {
+ StringBuilder sb = new StringBuilder();
+ toStringClassDetails(sb);
+ return sb.toString();
+ }
+
+ /**
+ * Called from {@link #toString()} when interpolating {@link Procedure} State. Allows decorating
+ * generic Procedure State with Procedure particulars.
+ *
+ * @param builder Append current {@link ProcedureState}
+ */
+ protected void toStringState(StringBuilder builder) {
+ builder.append(getState());
+ }
+
+ /**
+ * Extend the toString() information with the procedure details e.g. className and parameters
+ *
+ * @param builder the string builder to use to append the proc specific information
+ */
+ protected void toStringClassDetails(StringBuilder builder) {
+ builder.append(getClass().getName());
+ }
+
+ // ==========================================================================
+ // Those fields are unchanged after initialization.
+ //
+ // Each procedure will get created from the user or during
+ // ProcedureExecutor.start() during the load() phase and then submitted
+ // to the executor. these fields will never be changed after initialization
+ // ==========================================================================
+ public long getProcId() {
+ return procId;
+ }
+
+ public boolean hasParent() {
+ return parentProcId != NO_PROC_ID;
+ }
+
+ public long getParentProcId() {
+ return parentProcId;
+ }
+
+ public long getRootProcId() {
+ return rootProcId;
+ }
+
+ public String getProcName() {
+ return toStringClass();
+ }
+
+ public long getSubmittedTime() {
+ return submittedTime;
+ }
+
+ /** Called by the ProcedureExecutor to assign the ID to the newly created procedure. */
+ protected void setProcId(long procId) {
+ this.procId = procId;
+ this.submittedTime = System.currentTimeMillis();
Review Comment:
Maybe set submmitedTime and set state should in different method? SetProcId do more than the name can express?
##########
procedure/src/main/java/org/apache/iotdb/procedure/TimeoutExecutorThread.java:
##########
@@ -0,0 +1,108 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.procedure;
+
+import java.util.concurrent.DelayQueue;
+import java.util.concurrent.Delayed;
+import java.util.concurrent.TimeUnit;
+
+public class TimeoutExecutorThread<Env> extends StoppableThread {
+
+ private final ProcedureExecutor<Env> executor;
+
+ private final DelayQueue<ProcedureDelayContainer<Env>> queue = new DelayQueue<>();
+
+ public TimeoutExecutorThread(
+ ProcedureExecutor<Env> envProcedureExecutor, ThreadGroup threadGroup, String name) {
+ super(threadGroup, name);
+ setDaemon(true);
+ this.executor = envProcedureExecutor;
+ }
+
+ public void add(Procedure<Env> procedure) {
+ queue.add(new ProcedureDelayContainer<>(procedure));
+ }
+
+ public boolean remove(Procedure<Env> procedure) {
+ return queue.remove(new ProcedureDelayContainer<>(procedure));
+ }
+
+ private ProcedureDelayContainer<Env> takeQuietly() {
+ try {
+ return queue.poll(20, TimeUnit.SECONDS);
Review Comment:
please create a constant variable for 20
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org