You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ca...@apache.org on 2022/12/24 13:02:46 UTC
[iotdb] branch beyyes/procedure created (now 2ba3c50971)
This is an automated email from the ASF dual-hosted git repository.
caogaofei pushed a change to branch beyyes/procedure
in repository https://gitbox.apache.org/repos/asf/iotdb.git
at 2ba3c50971 perfect procedure test
This branch includes the following new commits:
new 95491d0981 fix procedure test
new 2ba3c50971 perfect procedure test
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[iotdb] 01/02: fix procedure test
Posted by ca...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
caogaofei pushed a commit to branch beyyes/procedure
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 95491d09816b81670d2f0b618889e39d75847deb
Author: Beyyes <cg...@foxmail.com>
AuthorDate: Sat Dec 24 10:27:52 2022 +0800
fix procedure test
---
.../procedure/TestProcedureExecutor.java | 8 ++++----
.../confignode/procedure/TestSTMProcedure.java | 12 ++++++++----
.../procedure/entity/SimpleLockProcedure.java | 12 ++++++++----
.../procedure/entity/SimpleSTMProcedure.java | 13 +++++++++----
.../procedure/entity/StuckProcedure.java | 2 +-
.../procedure/entity/TestProcedureFactory.java | 5 +++--
.../procedure/env/TestConfigNodeEnv.java | 22 ----------------------
7 files changed, 33 insertions(+), 41 deletions(-)
diff --git a/confignode/src/test/java/org/apache/iotdb/confignode/procedure/TestProcedureExecutor.java b/confignode/src/test/java/org/apache/iotdb/confignode/procedure/TestProcedureExecutor.java
index 046e4f9ffa..0c209fbc3a 100644
--- a/confignode/src/test/java/org/apache/iotdb/confignode/procedure/TestProcedureExecutor.java
+++ b/confignode/src/test/java/org/apache/iotdb/confignode/procedure/TestProcedureExecutor.java
@@ -73,10 +73,10 @@ public class TestProcedureExecutor extends TestProcedureBase {
Assert.assertEquals(3, threads1);
ProcedureTestUtil.waitForProcedure(procExecutor, otherProcId);
- Assert.assertEquals(true, procExecutor.isFinished(otherProcId));
- Assert.assertEquals(true, procExecutor.isRunning());
- Assert.assertEquals(false, procExecutor.isFinished(busyProcId1));
- Assert.assertEquals(false, procExecutor.isFinished(busyProcId2));
+ Assert.assertTrue(procExecutor.isFinished(otherProcId));
+ Assert.assertTrue(procExecutor.isRunning());
+ Assert.assertFalse(procExecutor.isFinished(busyProcId1));
+ Assert.assertFalse(procExecutor.isFinished(busyProcId2));
// terminate the busy procedures
latch1.release();
diff --git a/confignode/src/test/java/org/apache/iotdb/confignode/procedure/TestSTMProcedure.java b/confignode/src/test/java/org/apache/iotdb/confignode/procedure/TestSTMProcedure.java
index 7b387d3d43..0eff94ff97 100644
--- a/confignode/src/test/java/org/apache/iotdb/confignode/procedure/TestSTMProcedure.java
+++ b/confignode/src/test/java/org/apache/iotdb/confignode/procedure/TestSTMProcedure.java
@@ -25,11 +25,15 @@ import org.apache.iotdb.confignode.procedure.util.ProcedureTestUtil;
import org.junit.Assert;
import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.util.concurrent.atomic.AtomicInteger;
public class TestSTMProcedure extends TestProcedureBase {
+ public static final Logger LOGGER = LoggerFactory.getLogger(TestSTMProcedure.class);
+
@Test
public void testSubmitProcedure() {
SimpleSTMProcedure stmProcedure = new SimpleSTMProcedure();
@@ -37,7 +41,7 @@ public class TestSTMProcedure extends TestProcedureBase {
ProcedureTestUtil.waitForProcedure(this.procExecutor, procId);
TestProcEnv env = this.getEnv();
AtomicInteger acc = env.getAcc();
- Assert.assertEquals(acc.get(), 10);
+ Assert.assertEquals(10, acc.get());
}
@Test
@@ -50,9 +54,9 @@ public class TestSTMProcedure extends TestProcedureBase {
AtomicInteger acc = env.getAcc();
int success = env.successCount.get();
int rolledback = env.rolledBackCount.get();
- System.out.println(acc.get());
- System.out.println(success);
- System.out.println(rolledback);
+ LOGGER.info("all count: " + acc.get());
+ LOGGER.info("success count: " + success);
+ LOGGER.info("rolledback count: " + rolledback);
Assert.assertEquals(1 + success - rolledback, acc.get());
}
}
diff --git a/confignode/src/test/java/org/apache/iotdb/confignode/procedure/entity/SimpleLockProcedure.java b/confignode/src/test/java/org/apache/iotdb/confignode/procedure/entity/SimpleLockProcedure.java
index 564b980794..a88afdb4f4 100644
--- a/confignode/src/test/java/org/apache/iotdb/confignode/procedure/entity/SimpleLockProcedure.java
+++ b/confignode/src/test/java/org/apache/iotdb/confignode/procedure/entity/SimpleLockProcedure.java
@@ -25,11 +25,15 @@ import org.apache.iotdb.confignode.procedure.exception.ProcedureSuspendedExcepti
import org.apache.iotdb.confignode.procedure.exception.ProcedureYieldException;
import org.apache.iotdb.confignode.procedure.scheduler.SimpleProcedureScheduler;
import org.apache.iotdb.confignode.procedure.state.ProcedureLockState;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.io.IOException;
public class SimpleLockProcedure extends Procedure<TestProcEnv> {
+ private static final Logger LOGGER = LoggerFactory.getLogger(SimpleLockProcedure.class);
+
private String procName;
public SimpleLockProcedure() {}
@@ -46,7 +50,7 @@ public class SimpleLockProcedure extends Procedure<TestProcEnv> {
}
@Override
- protected void rollback(TestProcEnv testProcEnv) throws IOException, InterruptedException {}
+ protected void rollback(TestProcEnv testProcEnv) {}
@Override
protected boolean abort(TestProcEnv testProcEnv) {
@@ -57,19 +61,19 @@ public class SimpleLockProcedure extends Procedure<TestProcEnv> {
protected ProcedureLockState acquireLock(TestProcEnv testProcEnv) {
if (testProcEnv.getEnvLock().tryLock()) {
testProcEnv.lockAcquireSeq.append(procName);
- System.out.println(procName + " acquired lock.");
+ LOGGER.info(procName + " acquired lock.");
return ProcedureLockState.LOCK_ACQUIRED;
}
SimpleProcedureScheduler scheduler = (SimpleProcedureScheduler) testProcEnv.getScheduler();
scheduler.addWaiting(this);
- System.out.println(procName + " wait for lock.");
+ LOGGER.info(procName + " wait for lock.");
return ProcedureLockState.LOCK_EVENT_WAIT;
}
@Override
protected void releaseLock(TestProcEnv testProcEnv) {
- System.out.println(procName + " release lock.");
+ LOGGER.info(procName + " release lock.");
testProcEnv.getEnvLock().unlock();
SimpleProcedureScheduler scheduler = (SimpleProcedureScheduler) testProcEnv.getScheduler();
scheduler.releaseWaiting();
diff --git a/confignode/src/test/java/org/apache/iotdb/confignode/procedure/entity/SimpleSTMProcedure.java b/confignode/src/test/java/org/apache/iotdb/confignode/procedure/entity/SimpleSTMProcedure.java
index 25a39b04cf..99c7b4e5d1 100644
--- a/confignode/src/test/java/org/apache/iotdb/confignode/procedure/entity/SimpleSTMProcedure.java
+++ b/confignode/src/test/java/org/apache/iotdb/confignode/procedure/entity/SimpleSTMProcedure.java
@@ -19,11 +19,14 @@
package org.apache.iotdb.confignode.procedure.entity;
+import org.apache.iotdb.confignode.procedure.TestSTMProcedure;
import org.apache.iotdb.confignode.procedure.env.TestProcEnv;
import org.apache.iotdb.confignode.procedure.exception.ProcedureException;
import org.apache.iotdb.confignode.procedure.exception.ProcedureSuspendedException;
import org.apache.iotdb.confignode.procedure.exception.ProcedureYieldException;
import org.apache.iotdb.confignode.procedure.impl.statemachine.StateMachineProcedure;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.concurrent.atomic.AtomicInteger;
@@ -31,6 +34,8 @@ import java.util.concurrent.atomic.AtomicInteger;
public class SimpleSTMProcedure
extends StateMachineProcedure<TestProcEnv, SimpleSTMProcedure.TestState> {
+ public static final Logger LOGGER = LoggerFactory.getLogger(SimpleSTMProcedure.class);
+
public int throwAtIndex = -1;
public enum TestState {
@@ -40,8 +45,7 @@ public class SimpleSTMProcedure
}
@Override
- protected Flow executeFromState(TestProcEnv testProcEnv, TestState testState)
- throws ProcedureSuspendedException, ProcedureYieldException, InterruptedException {
+ protected Flow executeFromState(TestProcEnv testProcEnv, TestState testState) {
AtomicInteger acc = testProcEnv.getAcc();
try {
switch (testState) {
@@ -77,8 +81,9 @@ public class SimpleSTMProcedure
}
@Override
- protected void rollbackState(TestProcEnv testProcEnv, TestState testState)
- throws IOException, InterruptedException {}
+ protected void rollbackState(TestProcEnv testProcEnv, TestState testState) {
+ LOGGER.info("Execute rollback in SimpleSTMProcedure, testState: {}", testState);
+ }
@Override
protected TestState getState(int stateId) {
diff --git a/confignode/src/test/java/org/apache/iotdb/confignode/procedure/entity/StuckProcedure.java b/confignode/src/test/java/org/apache/iotdb/confignode/procedure/entity/StuckProcedure.java
index 1db02597c4..1f1204de44 100644
--- a/confignode/src/test/java/org/apache/iotdb/confignode/procedure/entity/StuckProcedure.java
+++ b/confignode/src/test/java/org/apache/iotdb/confignode/procedure/entity/StuckProcedure.java
@@ -52,7 +52,7 @@ public class StuckProcedure extends Procedure<TestProcEnv> {
}
@Override
- protected void rollback(TestProcEnv testProcEnv) throws IOException, InterruptedException {}
+ protected void rollback(TestProcEnv testProcEnv) {}
@Override
protected boolean abort(TestProcEnv testProcEnv) {
diff --git a/confignode/src/test/java/org/apache/iotdb/confignode/procedure/entity/TestProcedureFactory.java b/confignode/src/test/java/org/apache/iotdb/confignode/procedure/entity/TestProcedureFactory.java
index 421fe7cc46..c4c73e99db 100644
--- a/confignode/src/test/java/org/apache/iotdb/confignode/procedure/entity/TestProcedureFactory.java
+++ b/confignode/src/test/java/org/apache/iotdb/confignode/procedure/entity/TestProcedureFactory.java
@@ -20,6 +20,7 @@
package org.apache.iotdb.confignode.procedure.entity;
import org.apache.iotdb.confignode.procedure.Procedure;
+import org.apache.iotdb.confignode.procedure.env.TestProcEnv;
import org.apache.iotdb.confignode.procedure.store.IProcedureFactory;
import java.io.IOException;
@@ -28,13 +29,13 @@ import java.nio.ByteBuffer;
public class TestProcedureFactory implements IProcedureFactory {
@Override
- public Procedure create(ByteBuffer buffer) throws IOException {
+ public Procedure<TestProcEnv> create(ByteBuffer buffer) throws IOException {
int typeNum = buffer.getInt();
if (typeNum >= TestProcedureType.values().length) {
throw new IOException("unrecognized log type " + typeNum);
}
TestProcedureType type = TestProcedureType.values()[typeNum];
- Procedure procedure;
+ Procedure<TestProcEnv> procedure;
switch (type) {
case INC_PROCEDURE:
procedure = new IncProcedure();
diff --git a/confignode/src/test/java/org/apache/iotdb/confignode/procedure/env/TestConfigNodeEnv.java b/confignode/src/test/java/org/apache/iotdb/confignode/procedure/env/TestConfigNodeEnv.java
deleted file mode 100644
index a27b2244ea..0000000000
--- a/confignode/src/test/java/org/apache/iotdb/confignode/procedure/env/TestConfigNodeEnv.java
+++ /dev/null
@@ -1,22 +0,0 @@
-/*
- * 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.confignode.procedure.env;
-
-public class TestConfigNodeEnv {}
[iotdb] 02/02: perfect procedure test
Posted by ca...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
caogaofei pushed a commit to branch beyyes/procedure
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 2ba3c50971c2124ff60004285c9067c747bd5c8d
Author: Beyyes <cg...@foxmail.com>
AuthorDate: Sat Dec 24 21:02:23 2022 +0800
perfect procedure test
---
.../procedure/TestFirstStateMachineProcedure.java | 33 +++++++++
.../confignode/procedure/TestSTMProcedure.java | 14 ++--
...cedure.java => FirstStateMachineProcedure.java} | 78 ++++++++++++++++------
.../confignode/procedure/entity/IncProcedure.java | 46 ++++++++++++-
...edure.java => SecondStateMachineProcedure.java} | 65 ++++++++++++------
.../procedure/entity/SimpleLockProcedure.java | 3 +-
.../procedure/entity/SimpleSTMProcedure.java | 8 +--
.../procedure/entity/SleepProcedure.java | 8 ++-
.../procedure/entity/StuckProcedure.java | 1 -
.../confignode/procedure/entity/TestState.java | 26 ++++++++
.../confignode/procedure/env/TestProcEnv.java | 7 ++
11 files changed, 233 insertions(+), 56 deletions(-)
diff --git a/confignode/src/test/java/org/apache/iotdb/confignode/procedure/TestFirstStateMachineProcedure.java b/confignode/src/test/java/org/apache/iotdb/confignode/procedure/TestFirstStateMachineProcedure.java
new file mode 100644
index 0000000000..4a3025d4f7
--- /dev/null
+++ b/confignode/src/test/java/org/apache/iotdb/confignode/procedure/TestFirstStateMachineProcedure.java
@@ -0,0 +1,33 @@
+package org.apache.iotdb.confignode.procedure;
+
+import org.apache.iotdb.confignode.procedure.entity.FirstStateMachineProcedure;
+import org.apache.iotdb.confignode.procedure.env.TestProcEnv;
+import org.apache.iotdb.confignode.procedure.util.ProcedureTestUtil;
+
+import org.junit.Assert;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class TestFirstStateMachineProcedure extends TestProcedureBase {
+
+ public static final Logger LOGGER = LoggerFactory.getLogger(TestFirstStateMachineProcedure.class);
+
+ @Test
+ public void testRolledBackProcedure() {
+ FirstStateMachineProcedure firstProcedure = new FirstStateMachineProcedure();
+ firstProcedure.throwAtIndex = 3;
+ long procId = this.procExecutor.submitProcedure(firstProcedure);
+ ProcedureTestUtil.waitForProcedure(this.procExecutor, procId);
+ TestProcEnv env = this.getEnv();
+ AtomicInteger acc = env.getAcc();
+ int successCount = env.successCount.get();
+ int rolledBackCount = env.rolledBackCount.get();
+ LOGGER.info("all count: " + acc.get());
+ LOGGER.info("success count: " + successCount);
+ LOGGER.info("rolledback count: " + rolledBackCount);
+ Assert.assertEquals(1 + successCount - rolledBackCount, acc.get());
+ }
+}
diff --git a/confignode/src/test/java/org/apache/iotdb/confignode/procedure/TestSTMProcedure.java b/confignode/src/test/java/org/apache/iotdb/confignode/procedure/TestSTMProcedure.java
index 0eff94ff97..5c5639a613 100644
--- a/confignode/src/test/java/org/apache/iotdb/confignode/procedure/TestSTMProcedure.java
+++ b/confignode/src/test/java/org/apache/iotdb/confignode/procedure/TestSTMProcedure.java
@@ -45,18 +45,18 @@ public class TestSTMProcedure extends TestProcedureBase {
}
@Test
- public void testRolledBackProcedure() {
+ public void testRolledBackProcedure() throws InterruptedException {
SimpleSTMProcedure stmProcedure = new SimpleSTMProcedure();
- stmProcedure.throwAtIndex = 4;
+ stmProcedure.throwAtIndex = 3;
long procId = this.procExecutor.submitProcedure(stmProcedure);
ProcedureTestUtil.waitForProcedure(this.procExecutor, procId);
TestProcEnv env = this.getEnv();
AtomicInteger acc = env.getAcc();
- int success = env.successCount.get();
- int rolledback = env.rolledBackCount.get();
+ int successCount = env.successCount.get();
+ int rolledBackCount = env.rolledBackCount.get();
LOGGER.info("all count: " + acc.get());
- LOGGER.info("success count: " + success);
- LOGGER.info("rolledback count: " + rolledback);
- Assert.assertEquals(1 + success - rolledback, acc.get());
+ LOGGER.info("success count: " + successCount);
+ LOGGER.info("rolledback count: " + rolledBackCount);
+ Assert.assertEquals(1 + successCount - rolledBackCount, acc.get());
}
}
diff --git a/confignode/src/test/java/org/apache/iotdb/confignode/procedure/entity/SimpleSTMProcedure.java b/confignode/src/test/java/org/apache/iotdb/confignode/procedure/entity/FirstStateMachineProcedure.java
similarity index 52%
copy from confignode/src/test/java/org/apache/iotdb/confignode/procedure/entity/SimpleSTMProcedure.java
copy to confignode/src/test/java/org/apache/iotdb/confignode/procedure/entity/FirstStateMachineProcedure.java
index 99c7b4e5d1..b20d8d6362 100644
--- a/confignode/src/test/java/org/apache/iotdb/confignode/procedure/entity/SimpleSTMProcedure.java
+++ b/confignode/src/test/java/org/apache/iotdb/confignode/procedure/entity/FirstStateMachineProcedure.java
@@ -19,30 +19,23 @@
package org.apache.iotdb.confignode.procedure.entity;
-import org.apache.iotdb.confignode.procedure.TestSTMProcedure;
import org.apache.iotdb.confignode.procedure.env.TestProcEnv;
import org.apache.iotdb.confignode.procedure.exception.ProcedureException;
-import org.apache.iotdb.confignode.procedure.exception.ProcedureSuspendedException;
-import org.apache.iotdb.confignode.procedure.exception.ProcedureYieldException;
import org.apache.iotdb.confignode.procedure.impl.statemachine.StateMachineProcedure;
+import org.apache.iotdb.confignode.procedure.state.ProcedureLockState;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.io.IOException;
import java.util.concurrent.atomic.AtomicInteger;
-public class SimpleSTMProcedure
- extends StateMachineProcedure<TestProcEnv, SimpleSTMProcedure.TestState> {
+public class FirstStateMachineProcedure extends StateMachineProcedure<TestProcEnv, TestState> {
- public static final Logger LOGGER = LoggerFactory.getLogger(SimpleSTMProcedure.class);
+ public static final Logger LOGGER = LoggerFactory.getLogger(FirstStateMachineProcedure.class);
- public int throwAtIndex = -1;
+ private static final String FIRST = "FirstStateMachineProcedure";
- public enum TestState {
- STEP_1,
- STEP_2,
- STEP_3
- }
+ public int throwAtIndex = -1;
@Override
protected Flow executeFromState(TestProcEnv testProcEnv, TestState testState) {
@@ -50,26 +43,32 @@ public class SimpleSTMProcedure
try {
switch (testState) {
case STEP_1:
+ LOGGER.info("Execute {} STEP_1", FIRST);
acc.getAndAdd(1);
setNextState(TestState.STEP_2);
break;
case STEP_2:
- for (int i = 0; i < 10; i++) {
- IncProcedure child = new IncProcedure();
- if (i == throwAtIndex) {
- child.throwEx = true;
+ LOGGER.info("Execute {} STEP_2", FIRST);
+ for (int i = 0; i < 3; i++) {
+ SecondStateMachineProcedure child = new SecondStateMachineProcedure();
+ if (i == 1) {
+ child.throwAtIndex = i;
}
addChildProcedure(child);
}
setNextState(TestState.STEP_3);
break;
case STEP_3:
+ LOGGER.info("Execute FirstStateMachineProcedure STEP_3");
+ if (true) {
+ throw new ProcedureException("xx");
+ }
acc.getAndAdd(-1);
return Flow.NO_MORE_STATE;
}
} catch (Exception e) {
if (isRollbackSupported(testState)) {
- setFailure("proc failed", new ProcedureException(e));
+ setFailure("FirstStateMachineProcedure failed", new ProcedureException(e));
}
}
return Flow.HAS_MORE_STATE;
@@ -82,7 +81,18 @@ public class SimpleSTMProcedure
@Override
protected void rollbackState(TestProcEnv testProcEnv, TestState testState) {
- LOGGER.info("Execute rollback in SimpleSTMProcedure, testState: {}", testState);
+ switch (testState) {
+ case STEP_1:
+ LOGGER.info("Execute rollback in First STEP_1, procId: {}", getProcId());
+ break;
+ case STEP_2:
+ LOGGER.info("Execute rollback in First STEP_2, procId: {}", getProcId());
+ break;
+ case STEP_3:
+ default:
+ LOGGER.info("Execute rollback in First STEP_3, procId: {}", getProcId());
+ break;
+ }
}
@Override
@@ -99,4 +109,34 @@ public class SimpleSTMProcedure
protected TestState getInitialState() {
return TestState.STEP_1;
}
+
+ @Override
+ protected ProcedureLockState acquireLock(TestProcEnv testProcEnv) {
+ testProcEnv.getEnvLock().lock();
+ try {
+ if (testProcEnv.getExecuteLock().tryLock(this)) {
+ LOGGER.info("First {} acquire lock.", getProcId());
+ return ProcedureLockState.LOCK_ACQUIRED;
+ }
+ testProcEnv.getExecuteLock().waitProcedure(this);
+
+ LOGGER.info("First {} wait for lock.", getProcId());
+ return ProcedureLockState.LOCK_EVENT_WAIT;
+ } finally {
+ testProcEnv.getEnvLock().unlock();
+ }
+ }
+
+ @Override
+ protected void releaseLock(TestProcEnv testProcEnv) {
+ testProcEnv.getEnvLock().lock();
+ try {
+ LOGGER.info("First {} release lock.", getProcId());
+ if (testProcEnv.getExecuteLock().releaseLock(this)) {
+ testProcEnv.getExecuteLock().wakeWaitingProcedures(testProcEnv.getScheduler());
+ }
+ } finally {
+ testProcEnv.getEnvLock().unlock();
+ }
+ }
}
diff --git a/confignode/src/test/java/org/apache/iotdb/confignode/procedure/entity/IncProcedure.java b/confignode/src/test/java/org/apache/iotdb/confignode/procedure/entity/IncProcedure.java
index 2a86110825..d782f06fe6 100644
--- a/confignode/src/test/java/org/apache/iotdb/confignode/procedure/entity/IncProcedure.java
+++ b/confignode/src/test/java/org/apache/iotdb/confignode/procedure/entity/IncProcedure.java
@@ -21,8 +21,13 @@ package org.apache.iotdb.confignode.procedure.entity;
import org.apache.iotdb.confignode.procedure.Procedure;
import org.apache.iotdb.confignode.procedure.env.TestProcEnv;
+import org.apache.iotdb.confignode.procedure.exception.ProcedureException;
import org.apache.iotdb.confignode.procedure.exception.ProcedureSuspendedException;
import org.apache.iotdb.confignode.procedure.exception.ProcedureYieldException;
+import org.apache.iotdb.confignode.procedure.state.ProcedureLockState;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.io.DataOutputStream;
import java.io.IOException;
@@ -30,25 +35,34 @@ import java.util.concurrent.atomic.AtomicInteger;
public class IncProcedure extends Procedure<TestProcEnv> {
+ public static final Logger LOGGER = LoggerFactory.getLogger(IncProcedure.class);
+
public boolean throwEx = false;
@Override
protected Procedure<TestProcEnv>[] execute(TestProcEnv testProcEnv)
throws ProcedureYieldException, ProcedureSuspendedException, InterruptedException {
+ LOGGER.info("IncProcedure {} execute start", getProcId());
AtomicInteger acc = testProcEnv.getAcc();
if (throwEx) {
+ LOGGER.info("IncProcedure {} throw exception", getProcId());
+ setFailure(new ProcedureException("exception in IncProcedure"));
throw new RuntimeException("throw a EXCEPTION");
}
+ // TimeUnit.SECONDS.sleep(5);
acc.getAndIncrement();
testProcEnv.successCount.getAndIncrement();
+ LOGGER.info("IncProcedure {} execute successfully, acc: {}", getProcId(), acc.get());
return null;
}
@Override
- protected void rollback(TestProcEnv testProcEnv) throws IOException, InterruptedException {
+ protected void rollback(TestProcEnv testProcEnv) {
+ LOGGER.info("IncProcedure {} start rollback", getProcId());
AtomicInteger acc = testProcEnv.getAcc();
acc.getAndDecrement();
testProcEnv.rolledBackCount.getAndIncrement();
+ LOGGER.info("IncProcedure {} end rollback successfully, acc: {}", getProcId(), acc.get());
}
@Override
@@ -61,4 +75,34 @@ public class IncProcedure extends Procedure<TestProcEnv> {
stream.writeInt(TestProcedureFactory.TestProcedureType.INC_PROCEDURE.ordinal());
super.serialize(stream);
}
+
+ @Override
+ protected ProcedureLockState acquireLock(TestProcEnv testProcEnv) {
+ testProcEnv.getEnvLock().lock();
+ try {
+ if (testProcEnv.getExecuteLock().tryLock(this)) {
+ LOGGER.info("IntProcedureId {} acquire lock.", getProcId());
+ return ProcedureLockState.LOCK_ACQUIRED;
+ }
+ testProcEnv.getExecuteLock().waitProcedure(this);
+
+ LOGGER.info("IntProcedureId {} wait for lock.", getProcId());
+ return ProcedureLockState.LOCK_EVENT_WAIT;
+ } finally {
+ testProcEnv.getEnvLock().unlock();
+ }
+ }
+
+ @Override
+ protected void releaseLock(TestProcEnv testProcEnv) {
+ testProcEnv.getEnvLock().lock();
+ try {
+ LOGGER.info("IntProcedureId {} release lock.", getProcId());
+ if (testProcEnv.getExecuteLock().releaseLock(this)) {
+ testProcEnv.getExecuteLock().wakeWaitingProcedures(testProcEnv.getScheduler());
+ }
+ } finally {
+ testProcEnv.getEnvLock().unlock();
+ }
+ }
}
diff --git a/confignode/src/test/java/org/apache/iotdb/confignode/procedure/entity/SimpleSTMProcedure.java b/confignode/src/test/java/org/apache/iotdb/confignode/procedure/entity/SecondStateMachineProcedure.java
similarity index 56%
copy from confignode/src/test/java/org/apache/iotdb/confignode/procedure/entity/SimpleSTMProcedure.java
copy to confignode/src/test/java/org/apache/iotdb/confignode/procedure/entity/SecondStateMachineProcedure.java
index 99c7b4e5d1..72634489ca 100644
--- a/confignode/src/test/java/org/apache/iotdb/confignode/procedure/entity/SimpleSTMProcedure.java
+++ b/confignode/src/test/java/org/apache/iotdb/confignode/procedure/entity/SecondStateMachineProcedure.java
@@ -19,57 +19,47 @@
package org.apache.iotdb.confignode.procedure.entity;
-import org.apache.iotdb.confignode.procedure.TestSTMProcedure;
import org.apache.iotdb.confignode.procedure.env.TestProcEnv;
import org.apache.iotdb.confignode.procedure.exception.ProcedureException;
-import org.apache.iotdb.confignode.procedure.exception.ProcedureSuspendedException;
-import org.apache.iotdb.confignode.procedure.exception.ProcedureYieldException;
import org.apache.iotdb.confignode.procedure.impl.statemachine.StateMachineProcedure;
+import org.apache.iotdb.confignode.procedure.state.ProcedureLockState;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.io.IOException;
import java.util.concurrent.atomic.AtomicInteger;
-public class SimpleSTMProcedure
- extends StateMachineProcedure<TestProcEnv, SimpleSTMProcedure.TestState> {
+public class SecondStateMachineProcedure extends StateMachineProcedure<TestProcEnv, TestState> {
- public static final Logger LOGGER = LoggerFactory.getLogger(SimpleSTMProcedure.class);
+ public static final Logger LOGGER = LoggerFactory.getLogger(SecondStateMachineProcedure.class);
public int throwAtIndex = -1;
- public enum TestState {
- STEP_1,
- STEP_2,
- STEP_3
- }
-
@Override
protected Flow executeFromState(TestProcEnv testProcEnv, TestState testState) {
AtomicInteger acc = testProcEnv.getAcc();
try {
switch (testState) {
case STEP_1:
+ LOGGER.info("Execute SecondStateMachineProcedure {} STEP_1", getProcId());
acc.getAndAdd(1);
setNextState(TestState.STEP_2);
break;
case STEP_2:
- for (int i = 0; i < 10; i++) {
- IncProcedure child = new IncProcedure();
- if (i == throwAtIndex) {
- child.throwEx = true;
- }
- addChildProcedure(child);
+ LOGGER.info("Execute SecondStateMachineProcedure {} STEP_2", getProcId());
+ if (throwAtIndex > 0) {
+ throw new RuntimeException("throw EXCEPTION in SecondStateMachineProcedure");
}
setNextState(TestState.STEP_3);
break;
case STEP_3:
+ LOGGER.info("Execute SecondStateMachineProcedure {} STEP_3", getProcId());
acc.getAndAdd(-1);
return Flow.NO_MORE_STATE;
}
} catch (Exception e) {
if (isRollbackSupported(testState)) {
- setFailure("proc failed", new ProcedureException(e));
+ setFailure("SecondStateMachineProcedure failed", new ProcedureException(e));
}
}
return Flow.HAS_MORE_STATE;
@@ -82,7 +72,10 @@ public class SimpleSTMProcedure
@Override
protected void rollbackState(TestProcEnv testProcEnv, TestState testState) {
- LOGGER.info("Execute rollback in SimpleSTMProcedure, testState: {}", testState);
+ LOGGER.info(
+ "Execute rollback in SecondStateMachineProcedure {}, testState: {}",
+ getProcId(),
+ testState);
}
@Override
@@ -99,4 +92,34 @@ public class SimpleSTMProcedure
protected TestState getInitialState() {
return TestState.STEP_1;
}
+
+ @Override
+ protected ProcedureLockState acquireLock(TestProcEnv testProcEnv) {
+ testProcEnv.getEnvLock().lock();
+ try {
+ if (testProcEnv.getExecuteLock().tryLock(this)) {
+ LOGGER.info("Second {} acquire lock.", getProcId());
+ return ProcedureLockState.LOCK_ACQUIRED;
+ }
+ testProcEnv.getExecuteLock().waitProcedure(this);
+
+ LOGGER.info("Second {} wait for lock.", getProcId());
+ return ProcedureLockState.LOCK_EVENT_WAIT;
+ } finally {
+ testProcEnv.getEnvLock().unlock();
+ }
+ }
+
+ @Override
+ protected void releaseLock(TestProcEnv testProcEnv) {
+ testProcEnv.getEnvLock().lock();
+ try {
+ LOGGER.info("Second {} release lock.", getProcId());
+ if (testProcEnv.getExecuteLock().releaseLock(this)) {
+ testProcEnv.getExecuteLock().wakeWaitingProcedures(testProcEnv.getScheduler());
+ }
+ } finally {
+ testProcEnv.getEnvLock().unlock();
+ }
+ }
}
diff --git a/confignode/src/test/java/org/apache/iotdb/confignode/procedure/entity/SimpleLockProcedure.java b/confignode/src/test/java/org/apache/iotdb/confignode/procedure/entity/SimpleLockProcedure.java
index a88afdb4f4..553d02b295 100644
--- a/confignode/src/test/java/org/apache/iotdb/confignode/procedure/entity/SimpleLockProcedure.java
+++ b/confignode/src/test/java/org/apache/iotdb/confignode/procedure/entity/SimpleLockProcedure.java
@@ -25,11 +25,10 @@ import org.apache.iotdb.confignode.procedure.exception.ProcedureSuspendedExcepti
import org.apache.iotdb.confignode.procedure.exception.ProcedureYieldException;
import org.apache.iotdb.confignode.procedure.scheduler.SimpleProcedureScheduler;
import org.apache.iotdb.confignode.procedure.state.ProcedureLockState;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.io.IOException;
-
public class SimpleLockProcedure extends Procedure<TestProcEnv> {
private static final Logger LOGGER = LoggerFactory.getLogger(SimpleLockProcedure.class);
diff --git a/confignode/src/test/java/org/apache/iotdb/confignode/procedure/entity/SimpleSTMProcedure.java b/confignode/src/test/java/org/apache/iotdb/confignode/procedure/entity/SimpleSTMProcedure.java
index 99c7b4e5d1..8009e52fdc 100644
--- a/confignode/src/test/java/org/apache/iotdb/confignode/procedure/entity/SimpleSTMProcedure.java
+++ b/confignode/src/test/java/org/apache/iotdb/confignode/procedure/entity/SimpleSTMProcedure.java
@@ -19,16 +19,13 @@
package org.apache.iotdb.confignode.procedure.entity;
-import org.apache.iotdb.confignode.procedure.TestSTMProcedure;
import org.apache.iotdb.confignode.procedure.env.TestProcEnv;
import org.apache.iotdb.confignode.procedure.exception.ProcedureException;
-import org.apache.iotdb.confignode.procedure.exception.ProcedureSuspendedException;
-import org.apache.iotdb.confignode.procedure.exception.ProcedureYieldException;
import org.apache.iotdb.confignode.procedure.impl.statemachine.StateMachineProcedure;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.io.IOException;
import java.util.concurrent.atomic.AtomicInteger;
public class SimpleSTMProcedure
@@ -50,10 +47,12 @@ public class SimpleSTMProcedure
try {
switch (testState) {
case STEP_1:
+ LOGGER.info("Execute STM Procedure STEP_1");
acc.getAndAdd(1);
setNextState(TestState.STEP_2);
break;
case STEP_2:
+ LOGGER.info("Execute STM Procedure STEP_2");
for (int i = 0; i < 10; i++) {
IncProcedure child = new IncProcedure();
if (i == throwAtIndex) {
@@ -64,6 +63,7 @@ public class SimpleSTMProcedure
setNextState(TestState.STEP_3);
break;
case STEP_3:
+ LOGGER.info("Execute STM Procedure STEP_3");
acc.getAndAdd(-1);
return Flow.NO_MORE_STATE;
}
diff --git a/confignode/src/test/java/org/apache/iotdb/confignode/procedure/entity/SleepProcedure.java b/confignode/src/test/java/org/apache/iotdb/confignode/procedure/entity/SleepProcedure.java
index f3b2abf054..4474a0a91e 100644
--- a/confignode/src/test/java/org/apache/iotdb/confignode/procedure/entity/SleepProcedure.java
+++ b/confignode/src/test/java/org/apache/iotdb/confignode/procedure/entity/SleepProcedure.java
@@ -25,14 +25,20 @@ import org.apache.iotdb.confignode.procedure.exception.ProcedureSuspendedExcepti
import org.apache.iotdb.confignode.procedure.exception.ProcedureYieldException;
import org.apache.iotdb.confignode.procedure.util.ProcedureTestUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import java.io.DataOutputStream;
import java.io.IOException;
public class SleepProcedure extends Procedure<TestProcEnv> {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(Procedure.class);
+
@Override
protected Procedure<TestProcEnv>[] execute(TestProcEnv testProcEnv)
throws ProcedureYieldException, ProcedureSuspendedException, InterruptedException {
- System.out.println("Procedure is sleeping.");
+ LOGGER.info("SleepProcedure is sleeping.");
ProcedureTestUtil.sleepWithoutInterrupt(2000);
return null;
}
diff --git a/confignode/src/test/java/org/apache/iotdb/confignode/procedure/entity/StuckProcedure.java b/confignode/src/test/java/org/apache/iotdb/confignode/procedure/entity/StuckProcedure.java
index 1f1204de44..8391906811 100644
--- a/confignode/src/test/java/org/apache/iotdb/confignode/procedure/entity/StuckProcedure.java
+++ b/confignode/src/test/java/org/apache/iotdb/confignode/procedure/entity/StuckProcedure.java
@@ -22,7 +22,6 @@ package org.apache.iotdb.confignode.procedure.entity;
import org.apache.iotdb.confignode.procedure.Procedure;
import org.apache.iotdb.confignode.procedure.env.TestProcEnv;
-import java.io.IOException;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
diff --git a/confignode/src/test/java/org/apache/iotdb/confignode/procedure/entity/TestState.java b/confignode/src/test/java/org/apache/iotdb/confignode/procedure/entity/TestState.java
new file mode 100644
index 0000000000..b332e00315
--- /dev/null
+++ b/confignode/src/test/java/org/apache/iotdb/confignode/procedure/entity/TestState.java
@@ -0,0 +1,26 @@
+/*
+ * 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.confignode.procedure.entity;
+
+public enum TestState {
+ STEP_1,
+ STEP_2,
+ STEP_3
+}
diff --git a/confignode/src/test/java/org/apache/iotdb/confignode/procedure/env/TestProcEnv.java b/confignode/src/test/java/org/apache/iotdb/confignode/procedure/env/TestProcEnv.java
index 63e4f1997b..4fdac52d4f 100644
--- a/confignode/src/test/java/org/apache/iotdb/confignode/procedure/env/TestProcEnv.java
+++ b/confignode/src/test/java/org/apache/iotdb/confignode/procedure/env/TestProcEnv.java
@@ -19,6 +19,7 @@
package org.apache.iotdb.confignode.procedure.env;
+import org.apache.iotdb.confignode.procedure.scheduler.LockQueue;
import org.apache.iotdb.confignode.procedure.scheduler.ProcedureScheduler;
import java.util.concurrent.atomic.AtomicInteger;
@@ -39,6 +40,8 @@ public class TestProcEnv {
public StringBuilder lockAcquireSeq = new StringBuilder();
+ private final LockQueue procedureExecuteLock = new LockQueue();
+
public AtomicInteger getAcc() {
return acc;
}
@@ -51,6 +54,10 @@ public class TestProcEnv {
return envLock;
}
+ public LockQueue getExecuteLock() {
+ return this.procedureExecuteLock;
+ }
+
public void setScheduler(ProcedureScheduler scheduler) {
this.scheduler = scheduler;
}