|
| 1 | +use std::{ |
| 2 | + ops::{Deref, Range}, |
| 3 | + rc::Rc, |
| 4 | +}; |
| 5 | + |
| 6 | +use super::{ |
| 7 | + rebuild_descriptor::RebuildDescriptor, |
| 8 | + rebuild_error::RebuildError, |
| 9 | + rebuild_job_backend::RebuildBackend, |
| 10 | + rebuild_task::{RebuildTasks, TaskResult}, |
| 11 | + RebuildJob, |
| 12 | + RebuildJobOptions, |
| 13 | + SEGMENT_TASKS, |
| 14 | +}; |
| 15 | + |
| 16 | +use crate::gen_rebuild_instances; |
| 17 | + |
| 18 | +/// A Bdev rebuild job is responsible for managing a rebuild (copy) which reads |
| 19 | +/// from source_hdl and writes into destination_hdl from specified start to end. |
| 20 | +pub struct BdevRebuildJob(RebuildJob); |
| 21 | + |
| 22 | +impl std::fmt::Debug for BdevRebuildJob { |
| 23 | + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { |
| 24 | + self.0.fmt(f) |
| 25 | + } |
| 26 | +} |
| 27 | +impl Deref for BdevRebuildJob { |
| 28 | + type Target = RebuildJob; |
| 29 | + |
| 30 | + fn deref(&self) -> &Self::Target { |
| 31 | + &self.0 |
| 32 | + } |
| 33 | +} |
| 34 | + |
| 35 | +impl BdevRebuildJob { |
| 36 | + /// Creates a new RebuildJob which rebuilds from source URI to target URI |
| 37 | + /// from start to end (of the data partition); notify_fn callback is called |
| 38 | + /// when the rebuild state is updated - with the source and destination |
| 39 | + /// bdev URI's as arguments. |
| 40 | + pub async fn new( |
| 41 | + src_uri: &str, |
| 42 | + dst_uri: &str, |
| 43 | + range: Option<Range<u64>>, |
| 44 | + options: RebuildJobOptions, |
| 45 | + notify_fn: fn(&str, &str) -> (), |
| 46 | + ) -> Result<Self, RebuildError> { |
| 47 | + let descriptor = |
| 48 | + RebuildDescriptor::new(src_uri, dst_uri, range, options).await?; |
| 49 | + let tasks = RebuildTasks::new(SEGMENT_TASKS, &descriptor)?; |
| 50 | + let backend = |
| 51 | + BdevRebuildJobBackend::new(tasks, notify_fn, descriptor).await?; |
| 52 | + |
| 53 | + RebuildJob::from_backend(backend).await.map(Self) |
| 54 | + } |
| 55 | +} |
| 56 | + |
| 57 | +gen_rebuild_instances!(BdevRebuildJob); |
| 58 | + |
| 59 | +/// A rebuild job which is responsible for rebuilding from |
| 60 | +/// source to target of the `RebuildDescriptor`. |
| 61 | +pub(super) struct BdevRebuildJobBackend { |
| 62 | + /// The next block to be rebuilt. |
| 63 | + next: u64, |
| 64 | + /// A pool of tasks which perform the actual data rebuild. |
| 65 | + task_pool: RebuildTasks, |
| 66 | + /// A generic rebuild descriptor. |
| 67 | + descriptor: Rc<RebuildDescriptor>, |
| 68 | + /// Notification callback with src and dst uri's. |
| 69 | + notify_fn: fn(&str, &str) -> (), |
| 70 | +} |
| 71 | + |
| 72 | +#[async_trait::async_trait(?Send)] |
| 73 | +impl RebuildBackend for BdevRebuildJobBackend { |
| 74 | + fn on_state_change(&mut self) { |
| 75 | + (self.notify_fn)(&self.descriptor.src_uri, &self.descriptor.dst_uri); |
| 76 | + } |
| 77 | + |
| 78 | + fn common_desc(&self) -> &RebuildDescriptor { |
| 79 | + &self.descriptor |
| 80 | + } |
| 81 | + |
| 82 | + fn task_pool(&self) -> &RebuildTasks { |
| 83 | + &self.task_pool |
| 84 | + } |
| 85 | + |
| 86 | + fn schedule_task_by_id(&mut self, id: usize) -> bool { |
| 87 | + if self.next >= self.descriptor.range.end { |
| 88 | + false |
| 89 | + } else { |
| 90 | + let next = std::cmp::min( |
| 91 | + self.next + self.descriptor.segment_size_blks, |
| 92 | + self.descriptor.range.end, |
| 93 | + ); |
| 94 | + self.task_pool.schedule_segment_rebuild( |
| 95 | + id, |
| 96 | + self.next, |
| 97 | + self.descriptor.clone(), |
| 98 | + ); |
| 99 | + self.task_pool.active += 1; |
| 100 | + self.next = next; |
| 101 | + true |
| 102 | + } |
| 103 | + } |
| 104 | + |
| 105 | + async fn await_one_task(&mut self) -> Option<TaskResult> { |
| 106 | + self.task_pool.await_one_task().await |
| 107 | + } |
| 108 | +} |
| 109 | + |
| 110 | +impl std::fmt::Debug for BdevRebuildJobBackend { |
| 111 | + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { |
| 112 | + f.debug_struct("BdevRebuildJob") |
| 113 | + .field("next", &self.next) |
| 114 | + .finish() |
| 115 | + } |
| 116 | +} |
| 117 | +impl std::fmt::Display for BdevRebuildJobBackend { |
| 118 | + fn fmt(&self, _f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { |
| 119 | + Ok(()) |
| 120 | + } |
| 121 | +} |
| 122 | + |
| 123 | +impl BdevRebuildJobBackend { |
| 124 | + /// Creates a new RebuildJob which rebuilds from source URI to target URI |
| 125 | + /// from start to end (of the data partition); notify_fn callback is called |
| 126 | + /// when the rebuild state is updated - with the source and destination |
| 127 | + /// URI as arguments. |
| 128 | + pub async fn new( |
| 129 | + task_pool: RebuildTasks, |
| 130 | + notify_fn: fn(&str, &str) -> (), |
| 131 | + descriptor: RebuildDescriptor, |
| 132 | + ) -> Result<Self, RebuildError> { |
| 133 | + let be = Self { |
| 134 | + next: descriptor.range.start, |
| 135 | + task_pool, |
| 136 | + descriptor: Rc::new(descriptor), |
| 137 | + notify_fn, |
| 138 | + }; |
| 139 | + |
| 140 | + info!("{be}: backend created"); |
| 141 | + |
| 142 | + Ok(be) |
| 143 | + } |
| 144 | +} |
0 commit comments