(
subsys: SubsystemHandle,
state: RuntimeState,
cargo_options: CargoOptions,
watcher_config: WatcherConfig,
mut req_rx: Receiver<Action>,
)
| 38 | } |
| 39 | |
| 40 | async fn start_scheduler( |
| 41 | subsys: SubsystemHandle, |
| 42 | state: RuntimeState, |
| 43 | cargo_options: CargoOptions, |
| 44 | watcher_config: WatcherConfig, |
| 45 | mut req_rx: Receiver<Action>, |
| 46 | ) -> Result<(), ServerError> { |
| 47 | let (gc_tx, mut gc_rx) = mpsc::channel::<(String, InstanceId)>(10); |
| 48 | |
| 49 | if state.max_concurrency > 1 { |
| 50 | let monitor_state = state.clone(); |
| 51 | let monitor_cargo_options = cargo_options.clone(); |
| 52 | let monitor_watcher_config = watcher_config.clone(); |
| 53 | let monitor_gc_tx = gc_tx.clone(); |
| 54 | |
| 55 | subsys.start(SubsystemBuilder::new("instance monitor", move |s| { |
| 56 | instance_monitor( |
| 57 | s, |
| 58 | monitor_state, |
| 59 | monitor_cargo_options, |
| 60 | monitor_watcher_config, |
| 61 | monitor_gc_tx, |
| 62 | ) |
| 63 | })); |
| 64 | } |
| 65 | |
| 66 | loop { |
| 67 | tokio::select! { |
| 68 | Some(action) = req_rx.recv() => { |
| 69 | tracing::trace!(?action, "request action received"); |
| 70 | let start_function_name = match action { |
| 71 | Action::Invoke(req) => { |
| 72 | let function_name = req.function_name.clone(); |
| 73 | state.req_cache.upsert(req).await?; |
| 74 | Some(function_name) |
| 75 | }, |
| 76 | Action::Init => { |
| 77 | state.req_cache.init(DEFAULT_PACKAGE_FUNCTION).await; |
| 78 | Some(DEFAULT_PACKAGE_FUNCTION.into()) |
| 79 | }, |
| 80 | }; |
| 81 | |
| 82 | if watcher_config.start_function() { |
| 83 | if let Some(name) = start_function_name { |
| 84 | let queue_depth = state.req_cache.queue_depth(&name).await; |
| 85 | let pools = state.instance_pools.read().await; |
| 86 | let should_spawn = if let Some(pool) = pools.get(&name) { |
| 87 | pool.instance_count().await == 0 |
| 88 | } else { |
| 89 | true |
| 90 | }; |
| 91 | drop(pools); |
| 92 | |
| 93 | if should_spawn && queue_depth > 0 { |
| 94 | spawn_function_instance( |
| 95 | &subsys, |
| 96 | &state, |
| 97 | &name, |
no test coverage detected