Skip to content

Commit c25849a

Browse files
CaideyipiCaideyipi
committed
Fix procedure lock wait scheduling
1 parent a725ded commit c25849a

6 files changed

Lines changed: 135 additions & 50 deletions

File tree

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/Procedure.java‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -235,6 +235,16 @@ protected void releaseLock(Env env) {
235235
// no op
236236
}
237237

238+
/**
239+
* Called after an execution attempt returns {@link ProcedureLockState#LOCK_EVENT_WAIT}. Override
240+
* it to put the procedure into the corresponding lock wait queue.
241+
*
242+
* @param env env
243+
*/
244+
protected void waitForLock(Env env) {
245+
// no op
246+
}
247+
238248
/**
239249
* Used to keep procedure lock even when the procedure is yielded or suspended.
240250
*

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/ProcedureExecutor.java‎

Lines changed: 69 additions & 40 deletions
Original file line numberDiff line numberDiff line change
@@ -23,7 +23,6 @@
2323
import org.apache.iotdb.commons.utils.RetryUtils;
2424
import org.apache.iotdb.commons.utils.TestOnly;
2525
import org.apache.iotdb.confignode.i18n.ProcedureMessages;
26-
import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv;
2726
import org.apache.iotdb.confignode.procedure.exception.ProcedureException;
2827
import org.apache.iotdb.confignode.procedure.scheduler.ProcedureScheduler;
2928
import org.apache.iotdb.confignode.procedure.scheduler.SimpleProcedureScheduler;
@@ -42,7 +41,6 @@
4241
import java.util.Deque;
4342
import java.util.HashSet;
4443
import java.util.List;
45-
import java.util.Objects;
4644
import java.util.Set;
4745
import java.util.concurrent.ConcurrentHashMap;
4846
import java.util.concurrent.CopyOnWriteArrayList;
@@ -86,6 +84,16 @@ public class ProcedureExecutor<Env> {
8684
private final Env environment;
8785
private final IProcedureStore<Env> store;
8886

87+
private static final class LockStateResult<Env> {
88+
private final ProcedureLockState lockState;
89+
private final Procedure<Env> procedure;
90+
91+
private LockStateResult(ProcedureLockState lockState, Procedure<Env> procedure) {
92+
this.lockState = lockState;
93+
this.procedure = procedure;
94+
}
95+
}
96+
8997
public ProcedureExecutor(
9098
final Env environment, final IProcedureStore<Env> store, final ProcedureScheduler scheduler) {
9199
this.environment = environment;
@@ -329,33 +337,39 @@ private void executeProcedure(Procedure<Env> proc) {
329337
return;
330338
}
331339
ProcedureLockState lockState = null;
340+
Procedure<Env> lockEventWaitProcedure = null;
332341
try {
333342
do {
334343
if (!rootProcStack.acquire()) {
335344
if (rootProcStack.setRollback()) {
336-
lockState = executeRootStackRollback(rootProcId, rootProcStack);
345+
LockStateResult<Env> lockStateResult =
346+
executeRootStackRollback(rootProcId, rootProcStack);
347+
lockState = lockStateResult.lockState;
337348
switch (lockState) {
338349
case LOCK_ACQUIRED:
339350
break;
340351
case LOCK_EVENT_WAIT:
341-
LOG.info(ProcedureMessages.LOCK_EVENT_WAIT_ROLLBACK, proc);
352+
LOG.info(ProcedureMessages.LOCK_EVENT_WAIT_ROLLBACK, lockStateResult.procedure);
342353
rootProcStack.unsetRollback();
354+
lockEventWaitProcedure = lockStateResult.procedure;
343355
break;
344356
case LOCK_YIELD_WAIT:
345357
rootProcStack.unsetRollback();
346-
scheduler.yield(proc);
358+
scheduler.yield(lockStateResult.procedure);
347359
break;
348360
default:
349361
throw new UnsupportedOperationException();
350362
}
351363
} else {
352364
if (!proc.wasExecuted()) {
353-
switch (executeRollback(proc)) {
365+
lockState = executeRollback(proc);
366+
switch (lockState) {
354367
case LOCK_ACQUIRED:
355368
break;
356369
case LOCK_EVENT_WAIT:
357370
LOG.info(
358371
ProcedureMessages.LOCK_EVENT_WAIT_CAN_T_ROLLBACK_CHILD_RUNNING_FOR, proc);
372+
lockEventWaitProcedure = proc;
359373
break;
360374
case LOCK_YIELD_WAIT:
361375
scheduler.yield(proc);
@@ -367,19 +381,25 @@ private void executeProcedure(Procedure<Env> proc) {
367381
}
368382
break;
369383
}
370-
lockState = acquireLock(proc);
371-
switch (lockState) {
372-
case LOCK_ACQUIRED:
373-
executeProcedure(rootProcStack, proc);
374-
break;
375-
case LOCK_YIELD_WAIT:
376-
case LOCK_EVENT_WAIT:
377-
LOG.info(ProcedureMessages.LOCKSTATE_IS, proc, lockState);
378-
break;
379-
default:
380-
throw new UnsupportedOperationException();
384+
try {
385+
lockState = acquireLock(proc);
386+
switch (lockState) {
387+
case LOCK_ACQUIRED:
388+
executeProcedure(rootProcStack, proc);
389+
break;
390+
case LOCK_YIELD_WAIT:
391+
case LOCK_EVENT_WAIT:
392+
LOG.info(ProcedureMessages.LOCKSTATE_IS, proc, lockState);
393+
if (lockState == ProcedureLockState.LOCK_EVENT_WAIT) {
394+
lockEventWaitProcedure = proc;
395+
}
396+
break;
397+
default:
398+
throw new UnsupportedOperationException();
399+
}
400+
} finally {
401+
rootProcStack.release();
381402
}
382-
rootProcStack.release();
383403

384404
if (proc.isSuccess()) {
385405
// update metrics on finishing the procedure
@@ -397,9 +417,9 @@ private void executeProcedure(Procedure<Env> proc) {
397417
} finally {
398418
// Only after procedure has completed execution can it be allowed to be rescheduled to prevent
399419
// data races
400-
if (Objects.equals(lockState, ProcedureLockState.LOCK_EVENT_WAIT)) {
401-
LOG.info(ProcedureMessages.PROCEDUREID_WAIT_FOR_LOCK, proc.getProcId());
402-
((ConfigNodeProcedureEnv) this.environment).getNodeLock().waitProcedure(proc);
420+
if (lockEventWaitProcedure != null) {
421+
LOG.info(ProcedureMessages.PROCEDUREID_WAIT_FOR_LOCK, lockEventWaitProcedure.getProcId());
422+
lockEventWaitProcedure.waitForLock(this.environment);
403423
}
404424
}
405425
}
@@ -607,8 +627,8 @@ private void yieldProcedure(Procedure<Env> proc) {
607627
* @param procedureStack root procedure stack
608628
* @return lock state
609629
*/
610-
private ProcedureLockState executeRootStackRollback(
611-
Long rootProcId, RootProcedureStack procedureStack) {
630+
private LockStateResult<Env> executeRootStackRollback(
631+
Long rootProcId, RootProcedureStack<Env> procedureStack) {
612632
Procedure<Env> rootProcedure = procedures.get(rootProcId);
613633
ProcedureException exception = rootProcedure.getException();
614634
if (exception == null) {
@@ -629,19 +649,19 @@ private ProcedureLockState executeRootStackRollback(
629649
}
630650
ProcedureLockState lockState = acquireLock(procedure);
631651
if (lockState != ProcedureLockState.LOCK_ACQUIRED) {
632-
return lockState;
652+
return new LockStateResult<>(lockState, procedure);
633653
}
634654
lockState = executeRollback(procedure);
635655
releaseLock(procedure, false);
636656

637657
boolean abortRollback = lockState != ProcedureLockState.LOCK_ACQUIRED;
638658
abortRollback |= !isRunning() || !store.isRunning();
639659
if (abortRollback) {
640-
return lockState;
660+
return new LockStateResult<>(lockState, procedure);
641661
}
642662

643663
if (!procedure.isFinished() && procedure.isYieldAfterExecution(this.environment)) {
644-
return ProcedureLockState.LOCK_YIELD_WAIT;
664+
return new LockStateResult<>(ProcedureLockState.LOCK_YIELD_WAIT, procedure);
645665
}
646666

647667
if (procedure != rootProcedure) {
@@ -652,7 +672,7 @@ private ProcedureLockState executeRootStackRollback(
652672
LOG.info(
653673
ProcedureMessages.ROLLED_BACK_TIME_DURATION_IS, rootProcedure, rootProcedure.elapsedTime());
654674
rootProcedureCleanup(rootProcedure);
655-
return ProcedureLockState.LOCK_ACQUIRED;
675+
return new LockStateResult<>(ProcedureLockState.LOCK_ACQUIRED, rootProcedure);
656676
}
657677

658678
private ProcedureLockState acquireLock(Procedure<Env> proc) {
@@ -787,21 +807,30 @@ public void run() {
787807
Thread.sleep(1000);
788808
continue;
789809
}
790-
this.activeProcedure.set(procedure);
791-
activeExecutorCount.incrementAndGet();
792-
startTime.set(System.currentTimeMillis());
793-
PROCEDURE_EXECUTION_CONTEXT.set(true);
794810
try {
795-
executeProcedure(procedure);
796-
} finally {
797-
PROCEDURE_EXECUTION_CONTEXT.remove();
811+
this.activeProcedure.set(procedure);
812+
activeExecutorCount.incrementAndGet();
813+
startTime.set(System.currentTimeMillis());
814+
try {
815+
PROCEDURE_EXECUTION_CONTEXT.set(true);
816+
try {
817+
executeProcedure(procedure);
818+
} finally {
819+
PROCEDURE_EXECUTION_CONTEXT.remove();
820+
}
821+
} finally {
822+
activeExecutorCount.decrementAndGet();
823+
LOG.trace(
824+
"Halt pid={}, activeCount={}", procedure.getProcId(), activeExecutorCount.get());
825+
this.activeProcedure.set(null);
826+
lastUpdated = System.currentTimeMillis();
827+
startTime.set(lastUpdated);
828+
}
829+
} catch (Exception e) {
830+
LOG.warn(
831+
"Exception happened when worker {} execute procedure {}", getName(), procedure, e);
832+
throw e;
798833
}
799-
activeExecutorCount.decrementAndGet();
800-
LOG.trace(
801-
"Halt pid={}, activeCount={}", procedure.getProcId(), activeExecutorCount.get());
802-
this.activeProcedure.set(null);
803-
lastUpdated = System.currentTimeMillis();
804-
startTime.set(lastUpdated);
805834
}
806835

807836
} catch (Exception e) {

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/node/AbstractNodeProcedure.java‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -56,6 +56,18 @@ protected ProcedureLockState acquireLock(ConfigNodeProcedureEnv configNodeProced
5656
}
5757
}
5858

59+
@Override
60+
protected void waitForLock(ConfigNodeProcedureEnv configNodeProcedureEnv) {
61+
configNodeProcedureEnv.getSchedulerLock().lock();
62+
try {
63+
configNodeProcedureEnv
64+
.getNodeLock()
65+
.waitProcedure(this, configNodeProcedureEnv.getScheduler());
66+
} finally {
67+
configNodeProcedureEnv.getSchedulerLock().unlock();
68+
}
69+
}
70+
5971
@Override
6072
protected void releaseLock(ConfigNodeProcedureEnv configNodeProcedureEnv) {
6173
configNodeProcedureEnv.getSchedulerLock().lock();

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/scheduler/LockQueue.java‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -45,7 +45,11 @@ public boolean releaseLock(Procedure<?> procedure) {
4545
return true;
4646
}
4747

48-
public void waitProcedure(Procedure<?> procedure) {
48+
public void waitProcedure(Procedure<?> procedure, ProcedureScheduler procedureScheduler) {
49+
if (lockOwnerProcedure == null) {
50+
procedureScheduler.addFront(procedure);
51+
return;
52+
}
4953
deque.addLast(procedure);
5054
}
5155

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/scheduler/SimpleProcedureScheduler.java‎

Lines changed: 32 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@
2222
import org.apache.iotdb.confignode.procedure.Procedure;
2323

2424
import java.util.concurrent.LinkedBlockingDeque;
25+
import java.util.concurrent.locks.ReentrantLock;
2526

2627
/** Simple scheduler for procedures */
2728
public class SimpleProcedureScheduler extends AbstractProcedureScheduler {
@@ -48,6 +49,7 @@ public void clear() {
4849
schedLock();
4950
try {
5051
runnables.clear();
52+
waitings.clear();
5153
} finally {
5254
schedUnlock();
5355
}
@@ -68,12 +70,37 @@ public int queueSize() {
6870
return runnables.size();
6971
}
7072

71-
public void addWaiting(Procedure proc) {
72-
waitings.add(proc);
73+
public void waitProcedure(Procedure proc, ReentrantLock lock) {
74+
boolean signal = false;
75+
schedLock();
76+
try {
77+
if (lock.isLocked()) {
78+
waitings.add(proc);
79+
} else {
80+
runnables.addFirst(proc);
81+
signal = true;
82+
}
83+
} finally {
84+
schedUnlock();
85+
}
86+
if (signal) {
87+
signalAll();
88+
}
7389
}
7490

75-
public void releaseWaiting() {
76-
runnables.addAll(waitings);
77-
waitings.clear();
91+
public void releaseWaiting(ReentrantLock lock) {
92+
boolean signal;
93+
schedLock();
94+
try {
95+
lock.unlock();
96+
signal = !waitings.isEmpty();
97+
runnables.addAll(waitings);
98+
waitings.clear();
99+
} finally {
100+
schedUnlock();
101+
}
102+
if (signal) {
103+
signalAll();
104+
}
78105
}
79106
}

‎iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/entity/SimpleLockProcedure.java‎

Lines changed: 7 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -53,18 +53,21 @@ protected ProcedureLockState acquireLock(TestProcEnv testProcEnv) {
5353

5454
return ProcedureLockState.LOCK_ACQUIRED;
5555
}
56-
SimpleProcedureScheduler scheduler = (SimpleProcedureScheduler) testProcEnv.getScheduler();
57-
scheduler.addWaiting(this);
5856
System.out.println(procName + " wait for lock.");
5957
return ProcedureLockState.LOCK_EVENT_WAIT;
6058
}
6159

60+
@Override
61+
protected void waitForLock(TestProcEnv testProcEnv) {
62+
SimpleProcedureScheduler scheduler = (SimpleProcedureScheduler) testProcEnv.getScheduler();
63+
scheduler.waitProcedure(this, testProcEnv.getEnvLock());
64+
}
65+
6266
@Override
6367
protected void releaseLock(TestProcEnv testProcEnv) {
6468
System.out.println(procName + " release lock.");
65-
testProcEnv.getEnvLock().unlock();
6669
SimpleProcedureScheduler scheduler = (SimpleProcedureScheduler) testProcEnv.getScheduler();
67-
scheduler.releaseWaiting();
70+
scheduler.releaseWaiting(testProcEnv.getEnvLock());
6871
}
6972

7073
@Override

0 commit comments

Comments
 (0)