cache executors on failed transactions (backport #22308) (#22328)

* cache executors on failed transactions (#22308)

(cherry picked from commit 12e160269e)

# Conflicts:
#	program-runtime/src/invoke_context.rs
#	runtime/src/bank.rs

* resolve conflicts

Co-authored-by: Jack May <jack@solana.com>

program-runtime/src/invoke_context.rs

@@ -71,18 +71,57 @@ pub trait Executor: Debug + Send + Sync {
     ) -> Result<(), InstructionError>;
 }
 
-#[derive(Default)]
-pub struct Executors {
-    pub executors: HashMap<Pubkey, Arc<dyn Executor>>,
-    pub is_dirty: bool,
+pub type Executors = HashMap<Pubkey, TransactionExecutor>;
+
+/// Tracks whether a given executor is "dirty" and needs to updated in the
+/// executors cache
+pub struct TransactionExecutor {
+    executor: Arc<dyn Executor>,
+    is_miss: bool,
+    is_updated: bool,
 }
-impl Executors {
-    pub fn insert(&mut self, key: Pubkey, executor: Arc<dyn Executor>) {
-        let _ = self.executors.insert(key, executor);
-        self.is_dirty = true;
+
+impl TransactionExecutor {
+    /// Wraps an executor and tracks that it doesn't need to be updated in the
+    /// executors cache.
+    pub fn new_cached(executor: Arc<dyn Executor>) -> Self {
+        Self {
+            executor,
+            is_miss: false,
+            is_updated: false,
+        }
     }
-    pub fn get(&self, key: &Pubkey) -> Option<Arc<dyn Executor>> {
-        self.executors.get(key).cloned()
+
+    /// Wraps an executor and tracks that it needs to be updated in the
+    /// executors cache.
+    pub fn new_miss(executor: Arc<dyn Executor>) -> Self {
+        Self {
+            executor,
+            is_miss: true,
+            is_updated: false,
+        }
+    }
+
+    /// Wraps an executor and tracks that it needs to be updated in the
+    /// executors cache only if the transaction succeeded.
+    pub fn new_updated(executor: Arc<dyn Executor>) -> Self {
+        Self {
+            executor,
+            is_miss: false,
+            is_updated: true,
+        }
+    }
+
+    pub fn is_dirty(&self, include_updates: bool) -> bool {
+        self.is_miss || (include_updates && self.is_updated)
+    }
+
+    pub fn get(&self) -> Arc<dyn Executor> {
+        self.executor.clone()
+    }
+
+    pub fn clear_miss_for_test(&mut self) {
+        self.is_miss = false;
     }
 }
 
@@ -808,15 +847,26 @@ impl<'a> InvokeContext<'a> {
         &self.accounts_data_meter
     }
 
-    /// Loaders may need to do work in order to execute a program. Cache
-    /// the work that can be re-used across executions
+    /// Cache an executor that wasn't found in the cache
     pub fn add_executor(&self, pubkey: &Pubkey, executor: Arc<dyn Executor>) {
-        self.executors.borrow_mut().insert(*pubkey, executor);
+        self.executors
+            .borrow_mut()
+            .insert(*pubkey, TransactionExecutor::new_miss(executor));
+    }
+
+    /// Cache an executor that has changed
+    pub fn update_executor(&self, pubkey: &Pubkey, executor: Arc<dyn Executor>) {
+        self.executors
+            .borrow_mut()
+            .insert(*pubkey, TransactionExecutor::new_updated(executor));
     }
 
     /// Get the completed loader work that can be re-used across execution
     pub fn get_executor(&self, pubkey: &Pubkey) -> Option<Arc<dyn Executor>> {
-        self.executors.borrow().get(pubkey)
+        self.executors
+            .borrow()
+            .get(pubkey)
+            .map(|tx_executor| tx_executor.executor.clone())
     }
 
     /// Find an account_index and account by its key