[Block Executor] Follow up fix-ups (aptos-labs#12468)

Author: gelash

aptos-move/block-executor/src/executor.rs

@@ -383,7 +383,7 @@ where
                 scheduler.queueing_commits_arm();
             }
 
-            Ok(SchedulerTask::NoTask)
+            Ok(SchedulerTask::Retry)
         }
     }
 
@@ -739,7 +739,7 @@ where
         drop(init_timer);
 
         let _timer = WORK_WITH_TASK_SECONDS.start_timer();
-        let mut scheduler_task = SchedulerTask::NoTask;
+        let mut scheduler_task = SchedulerTask::Retry;
 
         let drain_commit_queue = || -> Result<(), PanicError> {
             while let Ok(txn_idx) = scheduler.pop_from_commit_queue() {
@@ -813,16 +813,19 @@ where
                     scheduler.finish_execution(txn_idx, incarnation, updates_outside)?
                 },
                 SchedulerTask::ExecutionTask(_, _, ExecutionTaskType::Wakeup(condvar)) => {
-                    let (lock, cvar) = &*condvar;
-                    // Mark dependency resolved.
-                    let mut lock = lock.lock();
-                    *lock = DependencyStatus::Resolved;
-                    // Wake up the process waiting for dependency.
-                    cvar.notify_one();
+                    {
+                        let (lock, cvar) = &*condvar;
+
+                        // Mark dependency resolved.
+                        let mut lock = lock.lock();
+                        *lock = DependencyStatus::Resolved;
+                        // Wake up the process waiting for dependency.
+                        cvar.notify_one();
+                    }
 
                     scheduler.next_task()
                 },
-                SchedulerTask::NoTask => scheduler.next_task(),
+                SchedulerTask::Retry => scheduler.next_task(),
                 SchedulerTask::Done => {
                     drain_commit_queue()?;
                     break Ok(());

aptos-move/block-executor/src/proptest_types/tests.rs

@@ -99,7 +99,7 @@ proptest! {
     #[test]
     fn no_early_termination(
         universe in vec(any::<[u8; 32]>(), 100),
-        transaction_gen in vec(any::<TransactionGen<[u8;32]>>(), 5000).no_shrink(),
+        transaction_gen in vec(any::<TransactionGen<[u8;32]>>(), 4000).no_shrink(),
         abort_transactions in vec(any::<Index>(), 0),
         skip_rest_transactions in vec(any::<Index>(), 0),
     ) {
@@ -108,8 +108,8 @@ proptest! {
 
     #[test]
     fn abort_only(
-        universe in vec(any::<[u8; 32]>(), 100),
-        transaction_gen in vec(any::<TransactionGen<[u8;32]>>(), 5000).no_shrink(),
+        universe in vec(any::<[u8; 32]>(), 80),
+        transaction_gen in vec(any::<TransactionGen<[u8;32]>>(), 300).no_shrink(),
         abort_transactions in vec(any::<Index>(), 5),
         skip_rest_transactions in vec(any::<Index>(), 0),
     ) {
@@ -118,8 +118,8 @@ proptest! {
 
     #[test]
     fn skip_rest_only(
-        universe in vec(any::<[u8; 32]>(), 100),
-        transaction_gen in vec(any::<TransactionGen<[u8;32]>>(), 5000).no_shrink(),
+        universe in vec(any::<[u8; 32]>(), 80),
+        transaction_gen in vec(any::<TransactionGen<[u8;32]>>(), 300).no_shrink(),
         abort_transactions in vec(any::<Index>(), 0),
         skip_rest_transactions in vec(any::<Index>(), 5),
     ) {

aptos-move/block-executor/src/scheduler.rs

@@ -79,15 +79,17 @@ pub enum ExecutionTaskType {
     Wakeup(DependencyCondvar),
 }
 
-/// A holder for potential task returned from the Scheduler. ExecutionTask and ValidationTask
-/// each contain a version of transaction that must be executed or validated, respectively.
-/// NoTask holds no task (similar None if we wrapped tasks in Option), and Done implies that
-/// there are no more tasks and the scheduler is done.
+/// Task type that the parallel execution workers get from the scheduler.
 #[derive(Debug)]
 pub enum SchedulerTask {
+    /// Execution task with a version of the transaction, and whether it's waking up an already
+    /// executing worker (suspended / waiting on a dependency).
     ExecutionTask(TxnIndex, Incarnation, ExecutionTaskType),
+    /// Validation task with a version of the transaction, and the validation wave information.
     ValidationTask(TxnIndex, Incarnation, Wave),
-    NoTask,
+    /// Retry holds no task (similar None if we wrapped tasks in Option)
+    Retry,
+    /// Done implies that there are no more tasks and the scheduler is done.
     Done,
 }
 
@@ -453,7 +455,7 @@ impl Scheduler {
                 && !self.never_executed(idx_to_validate);
 
             if !prefer_validate && idx_to_execute >= self.num_txns {
-                return SchedulerTask::NoTask;
+                return SchedulerTask::Retry;
             }
 
             if prefer_validate {
@@ -511,7 +513,7 @@ impl Scheduler {
     /// After txn is executed, schedule its dependencies for re-execution.
     /// If revalidate_suffix is true, decrease validation_idx to schedule all higher transactions
     /// for (re-)validation. Otherwise, in some cases (if validation_idx not already lower),
-    /// return a validation task of the transaction to the caller (otherwise NoTask).
+    /// return a validation task of the transaction to the caller (otherwise Retry).
     pub fn finish_execution(
         &self,
         txn_idx: TxnIndex,
@@ -521,10 +523,10 @@ impl Scheduler {
         // Note: It is preferable to hold the validation lock throughout the finish_execution,
         // in particular before updating execution status. The point was that we don't want
         // any validation to come before the validation status is correspondingly updated.
-        // It may be possible to make work more granular, but shouldn't make performance
-        // difference and like this correctness argument is much easier to see, in fact also
-        // the reason why we grab write lock directly, and never release it during the whole function.
-        // So even validation status readers have to wait if they somehow end up at the same index.
+        // It may be possible to reduce granularity, but shouldn't make performance difference
+        // and like this correctness argument is much easier to see, which is also why we grab
+        // the write lock directly, and never release it during the whole function. This way,
+        // even validation status readers have to wait if they somehow end up at the same index.
         let mut validation_status = self.txn_status[txn_idx as usize].1.write();
         self.set_executed_status(txn_idx, incarnation)?;
 
@@ -552,7 +554,7 @@ impl Scheduler {
             ));
         }
 
-        Ok(SchedulerTask::NoTask)
+        Ok(SchedulerTask::Retry)
     }
 
     pub fn finish_execution_during_commit(&self, txn_idx: TxnIndex) -> Result<(), PanicError> {
@@ -567,7 +569,7 @@ impl Scheduler {
     }
 
     /// Finalize a validation task of version (txn_idx, incarnation). In some cases,
-    /// may return a re-execution task back to the caller (otherwise, NoTask).
+    /// may return a re-execution task back to the caller (otherwise, Retry).
     pub fn finish_abort(
         &self,
         txn_idx: TxnIndex,
@@ -611,7 +613,7 @@ impl Scheduler {
             }
         }
 
-        Ok(SchedulerTask::NoTask)
+        Ok(SchedulerTask::Retry)
     }
 
     /// This function can halt the BlockSTM early, even if there are unfinished tasks.
@@ -674,13 +676,12 @@ impl TWaitForDependency for Scheduler {
         // mutexes. Thus, acquisitions always happen in the same order (here), may not deadlock.
 
         if self.is_executed(dep_txn_idx, true).is_some() {
-            // Current status of dep_txn_idx is 'executed', so the dependency got resolved.
-            // To avoid zombie dependency (and losing liveness), must return here and
-            // not add a (stale) dependency.
+            // Current status of dep_txn_idx is 'executed' (or even committed), so the dependency
+            // got resolved. To avoid zombie dependency (and losing liveness), must return here
+            // and not add a (stale) dependency.
 
             // Note: acquires (a different, status) mutex, while holding (dependency) mutex.
-            // Only place in scheduler where a thread may hold >1 mutexes, hence, such
-            // acquisitions always happens in the same order (this function), may not deadlock.
+            // For status lock this only happens here, thus the order is always higher index to lower.
             return Ok(DependencyResult::Resolved);
         }
 
@@ -706,17 +707,31 @@ impl TWaitForDependency for Scheduler {
 
 /// Private functions of the Scheduler
 impl Scheduler {
-    /// Helper function to be called from Scheduler::halt(); Sets the transaction status to Halted.
-    /// If the transaction is suspended, it will wake it up.
+    /// Helper function to be called from Scheduler::halt(); Sets the transaction status to Halted and
+    /// notifies the waiting thread, if applicable. The guarantee is that if halt(txn_idx) is called,
+    /// then no thread can remain suspended on some dependency while executing transaction txn_idx.
+    ///
+    /// Proof sketch as a result of code invariants that can be checked:
+    /// 1. Status is replaced with ExecutionHalted, and ExecutionHalted status can never change.
+    /// 2. In order for wait_for_dependency by txn_idx to return a CondVar to wait on, suspend must
+    ///    be successful, which implies the status at that point may not be ExecutionHalted and that
+    ///    the status at that point would be set to Suspended(_, CondVar).
+    /// 3. Suspended status can turn into Ready or Executing, all containing the CondVar, unless a
+    ///    worker with a ExecutionTaskType::WakeUp actually wakes up the suspending thread.
+    /// 4. The waking up consists of acquiring the CondVar lock, setting the status to ExecutionHalted
+    ///    or Resolved (if the worker with WakeUp task did it), and also calling notify_one. This
+    ///    ensures that a thread that waits until the condition variable changes from Unresolved will
+    ///    get released in all cases.
     fn halt_transaction_execution(&self, txn_idx: TxnIndex) {
         let mut status = self.txn_status[txn_idx as usize].0.write();
 
-        // Replace status to sure that the txn never gets suspended.
+        // Always replace the status.
         match std::mem::replace(&mut *status, ExecutionStatus::ExecutionHalted) {
             ExecutionStatus::Suspended(_, condvar)
             | ExecutionStatus::Ready(_, ExecutionTaskType::Wakeup(condvar))
             | ExecutionStatus::Executing(_, ExecutionTaskType::Wakeup(condvar)) => {
-                let (lock, cvar) = &*(condvar.clone());
+                // Condvar lock must always be taken inner-most.
+                let (lock, cvar) = &*condvar;
 
                 let mut lock = lock.lock();
                 *lock = DependencyStatus::ExecutionHalted;

aptos-move/block-executor/src/unit_tests/mod.rs

@@ -538,7 +538,7 @@ fn scheduler_tasks() {
     for i in 0..5 {
         // Validation index is at 0, so transactions will be validated and no
         // need to return a validation task to the caller.
-        assert_matches!(s.finish_execution(i, 0, false), Ok(SchedulerTask::NoTask));
+        assert_matches!(s.finish_execution(i, 0, false), Ok(SchedulerTask::Retry));
     }
 
     for i in 0..5 {
@@ -584,7 +584,7 @@ fn scheduler_tasks() {
         ))
     );
 
-    assert_matches!(s.finish_execution(4, 1, true), Ok(SchedulerTask::NoTask));
+    assert_matches!(s.finish_execution(4, 1, true), Ok(SchedulerTask::Retry));
     assert_matches!(
         s.finish_execution(1, 1, false),
         Ok(SchedulerTask::ValidationTask(1, 1, 1))
@@ -628,7 +628,7 @@ fn scheduler_first_wave() {
 
     // validation index will not increase for the first execution wave
     // until the status becomes executed.
-    assert_matches!(s.finish_execution(0, 0, false), Ok(SchedulerTask::NoTask));
+    assert_matches!(s.finish_execution(0, 0, false), Ok(SchedulerTask::Retry));
 
     // Now we can validate version (0, 0).
     assert_matches!(s.next_task(), SchedulerTask::ValidationTask(0, 0, 0));
@@ -638,17 +638,17 @@ fn scheduler_first_wave() {
     );
     // Since (1, 0) is not EXECUTED, no validation tasks, and execution index
     // is already at the limit, so no tasks immediately available.
-    assert_matches!(s.next_task(), SchedulerTask::NoTask);
+    assert_matches!(s.next_task(), SchedulerTask::Retry);
 
-    assert_matches!(s.finish_execution(2, 0, false), Ok(SchedulerTask::NoTask));
+    assert_matches!(s.finish_execution(2, 0, false), Ok(SchedulerTask::Retry));
     // There should be no tasks, but finishing (1,0) should enable validating
     // (1, 0) then (2,0).
-    assert_matches!(s.next_task(), SchedulerTask::NoTask);
+    assert_matches!(s.next_task(), SchedulerTask::Retry);
 
-    assert_matches!(s.finish_execution(1, 0, false), Ok(SchedulerTask::NoTask));
+    assert_matches!(s.finish_execution(1, 0, false), Ok(SchedulerTask::Retry));
     assert_matches!(s.next_task(), SchedulerTask::ValidationTask(1, 0, 0));
     assert_matches!(s.next_task(), SchedulerTask::ValidationTask(2, 0, 0));
-    assert_matches!(s.next_task(), SchedulerTask::NoTask);
+    assert_matches!(s.next_task(), SchedulerTask::Retry);
 }
 
 #[test]
@@ -665,7 +665,7 @@ fn scheduler_dependency() {
 
     // validation index will not increase for the first execution wave
     // until the status becomes executed.
-    assert_matches!(s.finish_execution(0, 0, false), Ok(SchedulerTask::NoTask));
+    assert_matches!(s.finish_execution(0, 0, false), Ok(SchedulerTask::Retry));
     // Now we can validate version (0, 0).
     assert_matches!(s.next_task(), SchedulerTask::ValidationTask(0, 0, 0));
     // Current status of 0 is executed - hence, no dependency added.
@@ -676,7 +676,7 @@ fn scheduler_dependency() {
         Ok(DependencyResult::Dependency(_))
     );
 
-    assert_matches!(s.finish_execution(2, 0, false), Ok(SchedulerTask::NoTask));
+    assert_matches!(s.finish_execution(2, 0, false), Ok(SchedulerTask::Retry));
 
     // resumed task doesn't bump incarnation
     assert_matches!(
@@ -696,7 +696,7 @@ fn incarnation_one_scheduler(num_txns: TxnIndex) -> Scheduler {
             s.next_task(),
             SchedulerTask::ExecutionTask(j, 0, ExecutionTaskType::Execution) if j == i
         );
-        assert_matches!(s.finish_execution(i, 0, false), Ok(SchedulerTask::NoTask));
+        assert_matches!(s.finish_execution(i, 0, false), Ok(SchedulerTask::Retry));
         assert_matches!(
             s.next_task(),
             SchedulerTask::ValidationTask(j, 0, 0) if i == j
@@ -732,7 +732,7 @@ fn scheduler_incarnation() {
         Ok(SchedulerTask::ValidationTask(2, 1, 1))
     );
     // Here since validation index is lower, wave doesn't increase and no task returned.
-    assert_matches!(s.finish_execution(4, 1, true), Ok(SchedulerTask::NoTask));
+    assert_matches!(s.finish_execution(4, 1, true), Ok(SchedulerTask::Retry));
 
     assert_matches!(s.next_task(), SchedulerTask::ValidationTask(4, 1, 1));
 
@@ -755,7 +755,7 @@ fn scheduler_incarnation() {
     );
     // execution index =  1
 
-    assert_matches!(s.finish_abort(4, 1), Ok(SchedulerTask::NoTask));
+    assert_matches!(s.finish_abort(4, 1), Ok(SchedulerTask::Retry));
 
     assert_matches!(
         s.next_task(),
@@ -785,7 +785,7 @@ fn scheduler_incarnation() {
     );
 
     // validation index is 4, so finish execution doesn't return validation task, next task does.
-    assert_matches!(s.finish_execution(4, 2, false), Ok(SchedulerTask::NoTask));
+    assert_matches!(s.finish_execution(4, 2, false), Ok(SchedulerTask::Retry));
     assert_matches!(s.next_task(), SchedulerTask::ValidationTask(4, 2, 2));
 }
 
@@ -802,11 +802,11 @@ fn scheduler_basic() {
     }
 
     // Finish executions & dispatch validation tasks.
-    assert_matches!(s.finish_execution(0, 0, true), Ok(SchedulerTask::NoTask));
-    assert_matches!(s.finish_execution(1, 0, true), Ok(SchedulerTask::NoTask));
+    assert_matches!(s.finish_execution(0, 0, true), Ok(SchedulerTask::Retry));
+    assert_matches!(s.finish_execution(1, 0, true), Ok(SchedulerTask::Retry));
     assert_matches!(s.next_task(), SchedulerTask::ValidationTask(0, 0, 0));
     assert_matches!(s.next_task(), SchedulerTask::ValidationTask(1, 0, 0));
-    assert_matches!(s.finish_execution(2, 0, true), Ok(SchedulerTask::NoTask));
+    assert_matches!(s.finish_execution(2, 0, true), Ok(SchedulerTask::Retry));
     assert_matches!(s.next_task(), SchedulerTask::ValidationTask(2, 0, 0));
 
     for i in 0..3 {
@@ -834,11 +834,11 @@ fn scheduler_drain_idx() {
     }
 
     // Finish executions & dispatch validation tasks.
-    assert_matches!(s.finish_execution(0, 0, true), Ok(SchedulerTask::NoTask));
-    assert_matches!(s.finish_execution(1, 0, true), Ok(SchedulerTask::NoTask));
+    assert_matches!(s.finish_execution(0, 0, true), Ok(SchedulerTask::Retry));
+    assert_matches!(s.finish_execution(1, 0, true), Ok(SchedulerTask::Retry));
     assert_matches!(s.next_task(), SchedulerTask::ValidationTask(0, 0, 0));
     assert_matches!(s.next_task(), SchedulerTask::ValidationTask(1, 0, 0));
-    assert_matches!(s.finish_execution(2, 0, true), Ok(SchedulerTask::NoTask));
+    assert_matches!(s.finish_execution(2, 0, true), Ok(SchedulerTask::Retry));
     assert_matches!(s.next_task(), SchedulerTask::ValidationTask(2, 0, 0));
 
     for i in 0..3 {
@@ -914,7 +914,7 @@ fn rolling_commit_wave() {
     assert_eq!(s.commit_state(), (2, 0));
 
     // No validation task because index is already 2.
-    assert_matches!(s.finish_execution(2, 1, false), Ok(SchedulerTask::NoTask,));
+    assert_matches!(s.finish_execution(2, 1, false), Ok(SchedulerTask::Retry,));
     // finish validating with a lower wave.
     s.finish_validation(2, 0);
     assert!(s.try_commit().is_none());
@@ -967,7 +967,7 @@ fn no_conflict_task_count() {
                         // false means a validation task.
                         tasks.insert(rng.gen::<u32>(), (false, txn_idx));
                     },
-                    SchedulerTask::NoTask => break,
+                    SchedulerTask::Retry => break,
                     // Unreachable because we never call try_commit.
                     SchedulerTask::Done => unreachable!(),
                 }
@@ -993,7 +993,7 @@ fn no_conflict_task_count() {
                             assert_eq!(wave, 0);
                             tasks.insert(rng.gen::<u32>(), (false, txn_idx));
                         } else {
-                            assert_matches!(task_res, Ok(SchedulerTask::NoTask));
+                            assert_matches!(task_res, Ok(SchedulerTask::Retry));
                         }
                     },
                     (_, (false, txn_idx)) => {

aptos-move/block-executor/src/view.rs

@@ -426,7 +426,7 @@ fn wait_for_dependency(
             // eventually finish and lead to unblocking txn_idx, contradiction.
             let (lock, cvar) = &*dep_condition;
             let mut dep_resolved = lock.lock();
-            while let DependencyStatus::Unresolved = *dep_resolved {
+            while matches!(*dep_resolved, DependencyStatus::Unresolved) {
                 dep_resolved = cvar.wait(dep_resolved).unwrap();
             }
             // dep resolved status is either resolved or execution halted.