starnix_core/device/
service_watcher.rs

1// Copyright 2026 The Fuchsia Authors. All rights reserved.
2// Use of this source code is governed by a BSD-style license that can be
3// found in the LICENSE file.
4
5//! Asynchronous watching for Fuchsia FIDL service directories (`/svc/<ServiceName>`).
6
7use crate::task::Kernel;
8use anyhow::Context as _;
9use fidl::endpoints::ServiceMarker;
10use fuchsia_component::client as fclient;
11use futures::{StreamExt as _, TryStreamExt as _};
12use starnix_logging::log_info;
13use std::sync::Weak;
14
15/// Opens `marker` in the process's `/svc` namespace and watches for service instances on the
16/// kernel's main async executor.
17///
18/// Discovery runs as a non-blocking task scheduled via
19/// [`KernelThreads::spawn_future`](crate::task::KernelThreads::spawn_future) rather than on a
20/// dedicated blocking kernel thread. `on_instance` is invoked concurrently for each discovered
21/// instance and receives a [`Weak<Kernel>`] so callers can perform any asynchronous initialization
22/// on the instance before upgrading to a strong [`Kernel`] reference.
23pub fn watch_fidl_service<S, F>(kernel: &Kernel, marker: S, task_name: &'static str, on_instance: F)
24where
25    S: ServiceMarker + Send + 'static,
26    F: AsyncFn(&Weak<Kernel>, S::Proxy) + Send + 'static,
27{
28    let weak_kernel = kernel.weak_self.clone();
29    kernel.kthreads.spawn_future(
30        move || async move {
31            // Open the service directory inside the future so its channel receiver is registered
32            // on the `kthreads` executor rather than the caller's thread-local executor.
33            let result = match fclient::Service::open(marker) {
34                Ok(service) => run_fidl_service_watcher(&weak_kernel, service, on_instance).await,
35                Err(e) => Err(e.context("failed to open service directory")),
36            };
37            if let Err(e) = result {
38                log_info!("{}: service unavailable: {e:#}", S::SERVICE_NAME);
39            }
40        },
41        task_name,
42    );
43}
44
45/// Drives a [`fclient::ServiceInstanceStream`] for `service`, invoking `on_instance` concurrently
46/// for each discovered instance while `kernel` remains alive.
47pub(super) async fn run_fidl_service_watcher<S, F>(
48    kernel: &Weak<Kernel>,
49    service: fclient::Service<S>,
50    on_instance: F,
51) -> Result<(), anyhow::Error>
52where
53    S: ServiceMarker,
54    F: AsyncFn(&Weak<Kernel>, S::Proxy),
55{
56    let on_instance = &on_instance;
57    service
58        .watch()
59        .await
60        .with_context(|| format!("failed to watch {} service", S::SERVICE_NAME))?
61        .take_while(|_| std::future::ready(kernel.strong_count() > 0))
62        .try_for_each_concurrent(None, |instance| async move {
63            on_instance(kernel, instance).await;
64            Ok(())
65        })
66        .await
67}
68
69#[cfg(test)]
70mod tests {
71    use super::*;
72    use crate::testing::spawn_kernel_and_run;
73    use fidl::endpoints::create_endpoints;
74    use fidl_fuchsia_hardware_power_stats as fpowerstats;
75    use fidl_fuchsia_io as fio;
76    use fuchsia_async as fasync;
77    use fuchsia_component::server::ServiceFs;
78    use futures::channel::oneshot;
79    use futures::future::{Either, select};
80    use starnix_sync::Mutex;
81    use std::collections::HashSet;
82    use std::pin::pin;
83    use std::sync::atomic::{AtomicBool, Ordering};
84
85    fn serve_instances(
86        names: &'static [&'static str],
87    ) -> fclient::Service<fpowerstats::ServiceMarker> {
88        let (client, server) = create_endpoints::<fio::DirectoryMarker>();
89        std::thread::spawn(move || {
90            fasync::LocalExecutor::default().run_singlethreaded(async move {
91                let mut fs = ServiceFs::new_local();
92                for name in names {
93                    fs.add_fidl_service_instance(*name, |_request: fpowerstats::ServiceRequest| ());
94                }
95                fs.serve_connection(server).expect("serve service directory");
96                fs.collect::<()>().await;
97            })
98        });
99        fclient::Service::open_from_dir(client, fpowerstats::ServiceMarker).expect("open service")
100    }
101
102    #[::fuchsia::test]
103    async fn watches_multiple_instances() {
104        spawn_kernel_and_run(async |current_task| {
105            let kernel = current_task.kernel();
106            let service = serve_instances(&["first", "second"]);
107            let seen = Mutex::new(HashSet::new());
108            let (second_tx, second_rx) = oneshot::channel();
109            let second_tx = Mutex::new(Some(second_tx));
110            let second_rx = Mutex::new(Some(second_rx));
111            let (done_tx, done_rx) = oneshot::channel();
112            let done_tx = Mutex::new(Some(done_tx));
113
114            let watcher = pin!(run_fidl_service_watcher(
115                &kernel.weak_self,
116                service,
117                async |weak_kernel, instance| {
118                    assert!(weak_kernel.upgrade().is_some());
119                    let name = instance.instance_name().to_string();
120                    if name == "first" {
121                        // Wait for `"second"` to enter `on_instance` before completing `"first"`,
122                        // verifying that `on_instance` runs concurrently across instances.
123                        let rx = second_rx.lock().take().expect("second_rx taken once");
124                        let _ = rx.await;
125                    } else if name == "second" {
126                        if let Some(tx) = second_tx.lock().take() {
127                            let _ = tx.send(());
128                        }
129                    }
130                    let mut seen = seen.lock();
131                    seen.insert(name);
132                    if seen.len() == 2 {
133                        if let Some(tx) = done_tx.lock().take() {
134                            let _ = tx.send(std::mem::take(&mut *seen));
135                        }
136                    }
137                }
138            ));
139
140            let seen = match select(watcher, done_rx).await {
141                Either::Left((result, _)) => {
142                    panic!("watcher exited before both instances were discovered: {result:?}");
143                }
144                Either::Right((Ok(seen), _)) => seen,
145                Either::Right((Err(e), _)) => {
146                    panic!("oneshot sender dropped before signalling: {e:?}");
147                }
148            };
149
150            assert_eq!(seen, HashSet::from(["first".to_string(), "second".to_string()]));
151        })
152        .await;
153    }
154
155    #[::fuchsia::test]
156    async fn stops_watching_when_kernel_is_dropped() {
157        let service = serve_instances(&["first", "second"]);
158        let called = AtomicBool::new(false);
159
160        run_fidl_service_watcher(&Weak::new(), service, async |_, _| {
161            called.store(true, Ordering::SeqCst);
162        })
163        .await
164        .expect("service watcher should exit cleanly when kernel is dropped");
165
166        assert!(!called.load(Ordering::SeqCst));
167    }
168}
starnix_core/device/service_watcher.rs

starnix_core/device/
service_watcher.rs
