solana_runtime/
bank_forks_controller.rs

1use {
2    crate::{bank::Bank, bank_forks::BankForks, installed_scheduler_pool::BankWithScheduler},
3    agave_votor_messages::consensus_message::Block,
4    crossbeam_channel::{Receiver, RecvTimeoutError, Sender, bounded},
5    log::warn,
6    solana_clock::Slot,
7    solana_metrics::datapoint_info,
8    std::{
9        fmt,
10        sync::{Arc, Mutex},
11        time::{Duration, Instant},
12    },
13    thiserror::Error,
14};
15
16const CHANNEL_SIZE: usize = 16;
17
18#[derive(Debug, Error)]
19pub enum BankForksControllerError {
20    #[error("bank forks controller is disconnected")]
21    Disconnected,
22    #[error("bank to insert for slot {0} was stale, failed to insert")]
23    UnableToInsertStaleBank(Slot),
24}
25
26pub enum BankForksCommand {
27    InsertBank {
28        bank: Box<Bank>,
29        response_sender: Sender<Option<BankWithScheduler>>,
30    },
31    ClearBank {
32        slot: Slot,
33        response_sender: Sender<()>,
34    },
35}
36
37#[derive(Clone, Copy, Debug)]
38pub struct SetRootCommand {
39    pub new_root: Block,
40}
41
42impl SetRootCommand {
43    /// Whether the requested root still identifies a frozen bank newer than the applied root.
44    pub fn matches_frozen_bank(&self, bank_forks: &BankForks) -> bool {
45        self.new_root.slot > bank_forks.root()
46            && bank_forks.get(self.new_root.slot).is_some_and(|bank| {
47                bank.is_frozen() && bank.block_id() == Some(self.new_root.block_id)
48            })
49    }
50}
51
52impl BankForksCommand {
53    fn metric_slot(&self) -> Slot {
54        match self {
55            Self::InsertBank { bank, .. } => bank.slot(),
56            Self::ClearBank { slot, .. } => *slot,
57        }
58    }
59}
60
61impl fmt::Display for BankForksCommand {
62    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
63        match self {
64            Self::InsertBank { .. } => write!(f, "insert_bank"),
65            Self::ClearBank { .. } => write!(f, "clear_bank"),
66        }
67    }
68}
69
70pub trait BankForksController: Send + Sync {
71    fn insert_bank(&self, bank: Bank) -> Result<BankWithScheduler, BankForksControllerError>;
72
73    fn enqueue_set_root(&self, new_root: Block);
74
75    fn clear_bank(&self, slot: Slot) -> Result<(), BankForksControllerError>;
76}
77
78/// Handle used by non-replay threads to serialize BankForks writes onto ReplayStage.
79#[derive(Clone)]
80pub struct BankForksControllerHandle {
81    sender: Sender<BankForksCommand>,
82    pending_set_root: Arc<Mutex<Option<SetRootCommand>>>,
83    set_root_signal_sender: Sender<()>,
84}
85
86impl BankForksControllerHandle {
87    pub fn new() -> (Self, BankForksCommandReceiver) {
88        let (sender, receiver) = bounded(CHANNEL_SIZE);
89        let (set_root_signal_sender, set_root_signal_receiver) = bounded(1);
90        let pending_set_root = Arc::new(Mutex::new(None));
91        (
92            Self {
93                sender,
94                pending_set_root: pending_set_root.clone(),
95                set_root_signal_sender,
96            },
97            BankForksCommandReceiver {
98                receiver,
99                pending_set_root,
100                set_root_signal_receiver,
101            },
102        )
103    }
104
105    fn send_command<T>(
106        &self,
107        command: BankForksCommand,
108        response_receiver: Receiver<T>,
109    ) -> Result<T, BankForksControllerError> {
110        let command_name = command.to_string();
111        let slot = command.metric_slot();
112        let queue_len_before_send = self.sender.len();
113        let total_start = Instant::now();
114        let send_start = Instant::now();
115        if self.sender.send(command).is_err() {
116            return Err(BankForksControllerError::Disconnected);
117        }
118        let send_us = send_start.elapsed().as_micros() as i64;
119
120        let response_wait_start = Instant::now();
121        let response = loop {
122            match response_receiver.recv_timeout(Duration::from_millis(100)) {
123                Ok(response) => break response,
124                Err(RecvTimeoutError::Disconnected) => {
125                    return Err(BankForksControllerError::Disconnected);
126                }
127                Err(RecvTimeoutError::Timeout) => (),
128            }
129            warn!(
130                "Replay is stuck, waiting for {}ms no response to {command_name} for {slot}",
131                response_wait_start.elapsed().as_millis()
132            );
133        };
134
135        let response_wait_us = response_wait_start.elapsed().as_micros() as i64;
136        let total_us = total_start.elapsed().as_micros() as i64;
137        datapoint_info!(
138            "bank_forks_controller-command",
139            ("command", command_name, String),
140            ("slot", slot as i64, i64),
141            ("queue_len_before_send", queue_len_before_send as i64, i64),
142            ("send_us", send_us, i64),
143            ("response_wait_us", response_wait_us, i64),
144            ("total_us", total_us, i64),
145        );
146
147        Ok(response)
148    }
149}
150
151impl BankForksController for BankForksControllerHandle {
152    fn insert_bank(&self, bank: Bank) -> Result<BankWithScheduler, BankForksControllerError> {
153        let slot = bank.slot();
154        let (response_sender, response_receiver) = bounded(1);
155        let bank = self.send_command(
156            BankForksCommand::InsertBank {
157                bank: Box::new(bank),
158                response_sender,
159            },
160            response_receiver,
161        )?;
162        bank.ok_or(BankForksControllerError::UnableToInsertStaleBank(slot))
163    }
164
165    fn enqueue_set_root(&self, new_root: Block) {
166        let total_start = Instant::now();
167        let command = SetRootCommand { new_root };
168
169        {
170            let mut pending_set_root = self.pending_set_root.lock().unwrap();
171            // Replay only needs to process the highest pending root.
172            if pending_set_root
173                .as_ref()
174                .is_none_or(|pending| command.new_root.slot > pending.new_root.slot)
175            {
176                *pending_set_root = Some(command);
177            }
178        }
179        let _ = self.set_root_signal_sender.try_send(());
180
181        let total_us = total_start.elapsed().as_micros() as i64;
182
183        datapoint_info!(
184            "bank_forks_controller-command",
185            ("command", "set_root", String),
186            ("slot", new_root.slot as i64, i64),
187            ("total_us", total_us, i64),
188        );
189    }
190
191    fn clear_bank(&self, slot: Slot) -> Result<(), BankForksControllerError> {
192        let (response_sender, response_receiver) = bounded(1);
193        self.send_command(
194            BankForksCommand::ClearBank {
195                slot,
196                response_sender,
197            },
198            response_receiver,
199        )
200    }
201}
202
203pub struct BankForksCommandReceiver {
204    receiver: Receiver<BankForksCommand>,
205    pending_set_root: Arc<Mutex<Option<SetRootCommand>>>,
206    set_root_signal_receiver: Receiver<()>,
207}
208
209impl BankForksCommandReceiver {
210    pub fn receiver(&self) -> &Receiver<BankForksCommand> {
211        &self.receiver
212    }
213
214    pub fn set_root_signal_receiver(&self) -> &Receiver<()> {
215        &self.set_root_signal_receiver
216    }
217
218    pub fn take_set_root_command(&self) -> Option<SetRootCommand> {
219        self.pending_set_root.lock().unwrap().take()
220    }
221}
222
223#[cfg(test)]
224mod tests {
225    use {
226        super::*,
227        crate::{bank::SlotLeader, bank_forks::BankForks, genesis_utils::create_genesis_config},
228        solana_hash::Hash,
229        std::{thread, time::Duration},
230    };
231
232    #[test]
233    fn test_bank_forks_controller_keeps_highest_pending_set_root() {
234        let (controller, receiver) = BankForksControllerHandle::new();
235
236        let block_id_5 = Hash::new_unique();
237        controller.enqueue_set_root(Block {
238            slot: 5,
239            block_id: block_id_5,
240        });
241        controller.enqueue_set_root(Block::new_unique(3));
242        let command = receiver.take_set_root_command().unwrap();
243        assert_eq!(command.new_root.slot, 5);
244        assert_eq!(command.new_root.block_id, block_id_5);
245        assert!(receiver.take_set_root_command().is_none());
246
247        controller.enqueue_set_root(Block::new_unique(3));
248        controller.enqueue_set_root(Block {
249            slot: 5,
250            block_id: block_id_5,
251        });
252        assert_eq!(receiver.take_set_root_command().unwrap().new_root.slot, 5);
253    }
254
255    #[test]
256    fn test_bank_forks_controller_signals_pending_set_root() {
257        let (controller, receiver) = BankForksControllerHandle::new();
258
259        controller.enqueue_set_root(Block::new_unique(1));
260        receiver
261            .set_root_signal_receiver()
262            .recv_timeout(Duration::from_secs(1))
263            .unwrap();
264        assert_eq!(receiver.take_set_root_command().unwrap().new_root.slot, 1);
265
266        controller.enqueue_set_root(Block::new_unique(2));
267        controller.enqueue_set_root(Block::new_unique(3));
268        receiver
269            .set_root_signal_receiver()
270            .recv_timeout(Duration::from_secs(1))
271            .unwrap();
272        assert!(receiver.set_root_signal_receiver().try_recv().is_err());
273        assert_eq!(receiver.take_set_root_command().unwrap().new_root.slot, 3);
274    }
275
276    #[test]
277    fn test_set_root_command_matches_frozen_bank() {
278        let genesis = create_genesis_config(10_000);
279        let bank_forks = BankForks::new_rw_arc(Bank::new_for_tests(&genesis.genesis_config));
280        let parent_bank = bank_forks.read().unwrap().root_bank();
281        let bank = Bank::new_from_parent(parent_bank, SlotLeader::default(), 1);
282        let block_id = Hash::new_unique();
283        bank.set_block_id(Some(block_id));
284        let bank = bank_forks
285            .write()
286            .unwrap()
287            .insert(bank)
288            .clone_without_scheduler();
289        let command = SetRootCommand {
290            new_root: Block { slot: 1, block_id },
291        };
292
293        assert!(!command.matches_frozen_bank(&bank_forks.read().unwrap()));
294        bank.freeze();
295        assert!(command.matches_frozen_bank(&bank_forks.read().unwrap()));
296
297        let mismatched_command = SetRootCommand {
298            new_root: Block::new_unique(command.new_root.slot),
299        };
300        assert!(!mismatched_command.matches_frozen_bank(&bank_forks.read().unwrap()));
301
302        let missing_command = SetRootCommand {
303            new_root: Block::new_unique(2),
304        };
305        assert!(!missing_command.matches_frozen_bank(&bank_forks.read().unwrap()));
306
307        bank_forks.write().unwrap().set_root(1, None, None);
308        assert!(!command.matches_frozen_bank(&bank_forks.read().unwrap()));
309    }
310
311    #[test]
312    fn test_bank_forks_controller_insert_and_set_root() {
313        let genesis = create_genesis_config(10_000);
314        let bank_forks = BankForks::new_rw_arc(Bank::new_for_tests(&genesis.genesis_config));
315        let (controller, receiver) = BankForksControllerHandle::new();
316        let replay_bank_forks = bank_forks.clone();
317        let (root_sender, root_receiver) = bounded(1);
318        let replay_thread = thread::spawn(move || {
319            loop {
320                if let Some(command) = receiver.take_set_root_command() {
321                    let new_root = command.new_root.slot;
322                    {
323                        let mut bank_forks = replay_bank_forks.write().unwrap();
324                        bank_forks.set_root(new_root, None, None);
325                    }
326                    root_sender.send(new_root).unwrap();
327                }
328                let command = match receiver.receiver().recv_timeout(Duration::from_millis(10)) {
329                    Ok(command) => command,
330                    Err(RecvTimeoutError::Timeout) => continue,
331                    Err(RecvTimeoutError::Disconnected) => break,
332                };
333                match command {
334                    BankForksCommand::InsertBank {
335                        bank,
336                        response_sender,
337                    } => {
338                        let bank = {
339                            let mut bank_forks = replay_bank_forks.write().unwrap();
340                            bank_forks.insert(*bank)
341                        };
342                        response_sender.send(Some(bank)).unwrap();
343                    }
344                    BankForksCommand::ClearBank {
345                        slot,
346                        response_sender,
347                    } => {
348                        let bank_to_clear =
349                            replay_bank_forks.read().unwrap().get_with_scheduler(slot);
350                        if let Some(bank) = bank_to_clear {
351                            let _ = bank.wait_for_completed_scheduler();
352                        }
353
354                        {
355                            let mut bank_forks = replay_bank_forks.write().unwrap();
356                            bank_forks.clear_bank(slot, false);
357                        }
358                        response_sender.send(()).unwrap();
359                    }
360                }
361            }
362        });
363
364        let parent_bank = bank_forks.read().unwrap().root_bank();
365        let bank = Bank::new_from_parent(parent_bank, SlotLeader::default(), 1);
366        let block_id = Hash::new_unique();
367        bank.set_block_id(Some(block_id));
368        bank.freeze();
369        let inserted_bank = controller.insert_bank(bank).unwrap();
370        assert_eq!(inserted_bank.slot(), 1);
371        assert!(bank_forks.read().unwrap().get(1).is_some());
372
373        controller.enqueue_set_root(Block { slot: 1, block_id });
374        assert_eq!(root_receiver.recv().unwrap(), 1);
375        assert_eq!(bank_forks.read().unwrap().root(), 1);
376
377        let parent_bank = bank_forks.read().unwrap().root_bank();
378        let bank = Bank::new_from_parent(parent_bank, SlotLeader::default(), 2);
379        bank.freeze();
380        let inserted_bank = controller.insert_bank(bank).unwrap();
381        assert_eq!(inserted_bank.slot(), 2);
382        assert!(bank_forks.read().unwrap().get(2).is_some());
383
384        controller.clear_bank(2).unwrap();
385        assert!(bank_forks.read().unwrap().get(2).is_none());
386
387        drop(controller);
388        replay_thread.join().unwrap();
389    }
390}
solana_runtime/bank_forks_controller.rs

solana_runtime/
bank_forks_controller.rs
