| 148 | } |
| 149 | |
| 150 | pub(super) fn spawn( |
| 151 | id: ComputeInstanceId, |
| 152 | build_info: &'static BuildInfo, |
| 153 | storage: StorageCollections, |
| 154 | peek_stash_persist_location: PersistLocation, |
| 155 | arranged_logs: Vec<(LogVariant, GlobalId, SharedCollectionState)>, |
| 156 | metrics: InstanceMetrics, |
| 157 | now: NowFn, |
| 158 | wallclock_lag: WallclockLagFn<Timestamp>, |
| 159 | dyncfg: Arc<ConfigSet>, |
| 160 | response_tx: mpsc::UnboundedSender<ComputeControllerResponse>, |
| 161 | introspection_tx: mpsc::UnboundedSender<IntrospectionUpdates>, |
| 162 | read_only: bool, |
| 163 | ) -> Self { |
| 164 | let (command_tx, command_rx) = mpsc::unbounded_channel(); |
| 165 | |
| 166 | let read_hold_tx: read_holds::ChangeTx = { |
| 167 | let command_tx = command_tx.clone(); |
| 168 | Arc::new(move |id, change: ChangeBatch<_>| { |
| 169 | let cmd: Command = { |
| 170 | let change = change.clone(); |
| 171 | Box::new(move |i| i.apply_read_hold_change(id, change)) |
| 172 | }; |
| 173 | command_tx.send(cmd).map_err(|_| SendError((id, change))) |
| 174 | }) |
| 175 | }; |
| 176 | |
| 177 | mz_ore::task::spawn( |
| 178 | || format!("compute-instance-{id}"), |
| 179 | Instance::new( |
| 180 | build_info, |
| 181 | storage, |
| 182 | peek_stash_persist_location, |
| 183 | arranged_logs, |
| 184 | metrics, |
| 185 | now, |
| 186 | wallclock_lag, |
| 187 | dyncfg, |
| 188 | command_rx, |
| 189 | response_tx, |
| 190 | Arc::clone(&read_hold_tx), |
| 191 | introspection_tx, |
| 192 | read_only, |
| 193 | ) |
| 194 | .run(), |
| 195 | ); |
| 196 | |
| 197 | Self { |
| 198 | command_tx, |
| 199 | read_hold_tx, |
| 200 | } |
| 201 | } |
| 202 | |
| 203 | /// Acquires a `ReadHold` and collection write frontier for each of the identified compute |
| 204 | /// collections. |