wl_proxy/
state.rs

1//! The proxy state.
2
3use {
4    crate::{
5        acceptor::{Acceptor, AcceptorError},
6        baseline::Baseline,
7        client::Client,
8        endpoint::{Endpoint, EndpointError},
9        handler::HandlerHolder,
10        object::{Object, ObjectCoreApi, ObjectPrivate},
11        poll::{self, PollError, PollEvent, Poller},
12        protocols::wayland::wl_display::WlDisplay,
13        trans::{FlushResult, TransError},
14        utils::{
15            env::{WAYLAND_DISPLAY, WAYLAND_SOCKET, XDG_RUNTIME_DIR},
16            stack::Stack,
17            stash::Stash,
18        },
19    },
20    error_reporter::Report,
21    run_on_drop::on_drop,
22    std::{
23        cell::{Cell, RefCell},
24        collections::HashMap,
25        io::{self, pipe},
26        os::fd::{AsFd, OwnedFd},
27        rc::{Rc, Weak},
28        sync::{
29            Arc,
30            atomic::{AtomicBool, Ordering::Acquire},
31        },
32        time::Duration,
33    },
34    thiserror::Error,
35    uapi::c,
36};
37pub use {
38    builder::StateBuilder,
39    destructor::{Destructor, RemoteDestructor},
40};
41
42mod builder;
43mod destructor;
44#[cfg(test)]
45mod tests;
46
47/// An error emitted by a [`State`].
48#[derive(Debug, Error)]
49#[error(transparent)]
50pub struct StateError(#[from] StateErrorKind);
51
52#[derive(Debug, Error)]
53enum StateErrorKind {
54    #[error("the state has already been destroyed")]
55    Destroyed,
56    #[error("the state has been destroyed by a remote destructor")]
57    RemoteDestroyed,
58    #[error("cannot perform recursive call into the state")]
59    RecursiveCall,
60    #[error("the server hung up the connection")]
61    ServerHangup,
62    #[error("could not write to the server socket")]
63    WriteToServer(#[source] EndpointError),
64    #[error("could not dispatch server events")]
65    DispatchEvents(#[source] EndpointError),
66    #[error("could not create a socket pair")]
67    Socketpair(#[source] io::Error),
68    #[error(transparent)]
69    CreateAcceptor(AcceptorError),
70    #[error("could not accept a new connection")]
71    AcceptConnection(AcceptorError),
72    #[error("could not create a pipe")]
73    CreatePipe(#[source] io::Error),
74    #[error("could not read {} environment variable", WAYLAND_DISPLAY)]
75    WaylandDisplay,
76    #[error("the display name is empty")]
77    WaylandDisplayEmpty,
78    #[error("{} is not set", XDG_RUNTIME_DIR)]
79    XrdNotSet,
80    #[error("the socket path is too long")]
81    SocketPathTooLong,
82    #[error("could not create a socket")]
83    CreateSocket(#[source] io::Error),
84    #[error("could not connect to {0}")]
85    Connect(String, #[source] io::Error),
86    #[error("{} does not contain a valid number", WAYLAND_SOCKET)]
87    WaylandSocketNotNumber,
88    #[error("F_GETFD failed on {}", WAYLAND_SOCKET)]
89    WaylandSocketGetFd(#[source] io::Error),
90    #[error("F_SETFD failed on {}", WAYLAND_SOCKET)]
91    WaylandSocketSetFd(#[source] io::Error),
92    #[error(transparent)]
93    PollError(PollError),
94}
95
96/// The proxy state.
97///
98/// This type represents a connection to a server and any number of clients connected to
99/// this proxy.
100///
101/// This type can be constructed by using a [`StateBuilder`].
102///
103/// # Example
104///
105/// ```
106/// # use std::rc::Rc;
107/// # use wl_proxy::baseline::Baseline;
108/// # use wl_proxy::client::{Client, ClientHandler};
109/// # use wl_proxy::protocols::wayland::wl_display::{WlDisplay, WlDisplayHandler};
110/// # use wl_proxy::protocols::wayland::wl_registry::WlRegistry;
111/// # use wl_proxy::state::{State, StateBuilder, StateHandler};
112/// # fn f() {
113/// let state = State::builder(Baseline::ALL_OF_THEM).build().unwrap();
114/// let acceptor = state.create_acceptor(1000).unwrap();
115/// eprintln!("{}", acceptor.display());
116/// loop {
117///     state.dispatch_blocking().unwrap();
118/// }
119///
120/// struct StateHandlerImpl;
121///
122/// impl StateHandler for StateHandlerImpl {
123///     fn new_client(&mut self, client: &Rc<Client>) {
124///         eprintln!("Client connected");
125///         client.set_handler(ClientHandlerImpl);
126///         client.display().set_handler(DisplayHandler);
127///     }
128/// }
129///
130/// struct ClientHandlerImpl;
131///
132/// impl ClientHandler for ClientHandlerImpl {
133///     fn disconnected(self: Box<Self>) {
134///         eprintln!("Client disconnected");
135///     }
136/// }
137///
138/// struct DisplayHandler;
139///
140/// impl WlDisplayHandler for DisplayHandler {
141///     fn handle_get_registry(&mut self, slf: &Rc<WlDisplay>, registry: &Rc<WlRegistry>) {
142///         eprintln!("get_registry called");
143///         let _ = slf.send_get_registry(registry);
144///     }
145/// }
146/// # }
147/// ```
148pub struct State {
149    pub(crate) baseline: Baseline,
150    poller: Poller,
151    next_pollable_id: Cell<u64>,
152    pub(crate) server: Option<Rc<Endpoint>>,
153    pub(crate) destroyed: Cell<bool>,
154    handler: HandlerHolder<dyn StateHandler>,
155    pollables: RefCell<HashMap<u64, Pollable>>,
156    acceptable_acceptors: Stack<Rc<Acceptor>>,
157    has_acceptable_acceptors: Cell<bool>,
158    clients_to_kill: Stack<Rc<Client>>,
159    has_clients_to_kill: Cell<bool>,
160    readable_endpoints: Stack<EndpointWithClient>,
161    has_readable_endpoints: Cell<bool>,
162    flushable_endpoints: Stack<EndpointWithClient>,
163    has_flushable_endpoints: Cell<bool>,
164    interest_update_endpoints: Stack<Rc<Endpoint>>,
165    has_interest_update_endpoints: Cell<bool>,
166    interest_update_acceptors: Stack<Rc<Acceptor>>,
167    has_interest_update_acceptors: Cell<bool>,
168    pub(crate) all_objects: RefCell<HashMap<u64, Weak<dyn Object>>>,
169    pub(crate) next_object_id: Cell<u64>,
170    #[cfg(feature = "logging")]
171    pub(crate) log: bool,
172    #[cfg(feature = "logging")]
173    pub(crate) log_prefix: String,
174    #[cfg(feature = "logging")]
175    log_writer: RefCell<io::BufWriter<uapi::Fd>>,
176    global_lock_held: Cell<bool>,
177    pub(crate) object_stash: Stash<Rc<dyn Object>>,
178    pub(crate) forward_to_client: Cell<bool>,
179    pub(crate) forward_to_server: Cell<bool>,
180}
181
182/// A handler for events emitted by a [`State`].
183pub trait StateHandler: 'static {
184    /// A new client has connected.
185    ///
186    /// This event is not emitted if the connection is created explicitly via
187    /// [`State::connect`] or [`State::add_client`].
188    fn new_client(&mut self, client: &Rc<Client>) {
189        let _ = client;
190    }
191}
192
193enum Pollable {
194    Endpoint(EndpointWithClient),
195    Acceptor(Rc<Acceptor>),
196    Destructor(OwnedFd, Arc<AtomicBool>),
197}
198
199#[derive(Clone)]
200struct EndpointWithClient {
201    endpoint: Rc<Endpoint>,
202    client: Option<Rc<Client>>,
203}
204
205pub(crate) struct HandlerLock<'a> {
206    state: &'a State,
207}
208
209impl State {
210    pub(crate) fn remove_endpoint(&self, endpoint: &Endpoint) {
211        self.pollables.borrow_mut().remove(&endpoint.id);
212        self.poller.unregister(endpoint.socket.as_fd());
213        endpoint.unregistered.set(true);
214    }
215
216    fn acquire_handler_lock(&self) -> Result<HandlerLock<'_>, StateErrorKind> {
217        if self.global_lock_held.replace(true) {
218            return Err(StateErrorKind::RecursiveCall);
219        }
220        Ok(HandlerLock { state: self })
221    }
222
223    fn flush_locked(&self, lock: &HandlerLock<'_>) -> Result<bool, StateError> {
224        let mut did_work = false;
225        did_work |= self.perform_writes(lock)?;
226        did_work |= self.kill_clients();
227        self.update_interests()?;
228        Ok(did_work)
229    }
230
231    pub(crate) fn handle_delete_id(&self, server: &Endpoint, id: u32) {
232        let object = server.objects.borrow_mut().remove(&id).unwrap();
233        let core = object.core();
234        core.server_obj_id.take();
235        server.idl.release(id);
236        if let Err((e, object)) = object.delete_id() {
237            log::warn!(
238                "Could not handle a wl_display.delete_id message: {}",
239                Report::new(e),
240            );
241            let _ = object.core().try_delete_id();
242        }
243    }
244
245    fn perform_writes(&self, _: &HandlerLock<'_>) -> Result<bool, StateError> {
246        if !self.has_flushable_endpoints.get() {
247            return Ok(false);
248        }
249        while let Some(ewc) = self.flushable_endpoints.pop() {
250            let res = match ewc.endpoint.flush() {
251                Ok(r) => r,
252                Err(e) => {
253                    let is_closed = matches!(e, EndpointError::Flush(TransError::Closed));
254                    if let Some(client) = &ewc.client {
255                        if !is_closed {
256                            log::warn!(
257                                "Could not write to client#{}: {}",
258                                client.endpoint.id,
259                                Report::new(e),
260                            );
261                        }
262                        self.add_client_to_kill(client);
263                    } else {
264                        if is_closed {
265                            return Err(StateErrorKind::ServerHangup.into());
266                        }
267                        return Err(StateErrorKind::WriteToServer(e).into());
268                    }
269                    continue;
270                }
271            };
272            match res {
273                FlushResult::Done => {
274                    ewc.endpoint.flush_queued.set(false);
275                    self.change_interest(&ewc.endpoint, |i| i & !poll::WRITABLE);
276                }
277                FlushResult::Blocked => {
278                    self.change_interest(&ewc.endpoint, |i| i | poll::WRITABLE);
279                }
280            }
281        }
282        self.has_flushable_endpoints.set(false);
283        Ok(true)
284    }
285
286    fn accept_connections(self: &Rc<Self>, lock: &HandlerLock<'_>) -> Result<bool, StateError> {
287        if !self.has_acceptable_acceptors.get() {
288            return Ok(false);
289        }
290        self.check_destroyed()?;
291        while let Some(acceptor) = self.acceptable_acceptors.pop() {
292            self.interest_update_acceptors.push(acceptor.clone());
293            self.has_interest_update_acceptors.set(true);
294            const MAX_ACCEPT_PER_ITERATION: usize = 10;
295            for _ in 0..MAX_ACCEPT_PER_ITERATION {
296                let socket = acceptor
297                    .accept()
298                    .map_err(StateErrorKind::AcceptConnection)?;
299                let Some(socket) = socket else {
300                    break;
301                };
302                self.create_client(Some(lock), &Rc::new(socket))?;
303            }
304        }
305        self.has_acceptable_acceptors.set(false);
306        Ok(true)
307    }
308
309    fn read_messages(&self, lock: &HandlerLock<'_>) -> Result<bool, StateError> {
310        if !self.has_readable_endpoints.get() {
311            return Ok(false);
312        }
313        while let Some(ewc) = self.readable_endpoints.pop() {
314            let res = ewc.endpoint.read_messages(lock, ewc.client.as_ref());
315            if let Err(e) = res {
316                if let Some(client) = &ewc.client {
317                    log::error!("Could not handle client message: {}", Report::new(e));
318                    self.add_client_to_kill(client);
319                } else {
320                    return Err(StateErrorKind::DispatchEvents(e).into());
321                }
322            }
323            self.change_interest(&ewc.endpoint, |i| i | poll::READABLE);
324        }
325        self.has_readable_endpoints.set(false);
326        Ok(true)
327    }
328
329    fn change_interest(&self, endpoint: &Rc<Endpoint>, f: impl FnOnce(u32) -> u32) {
330        if self.destroyed.get() {
331            return;
332        }
333        let old = endpoint.desired_interest.get();
334        let new = f(old);
335        endpoint.desired_interest.set(new);
336        if old != new
337            && endpoint.current_interest.get() != new
338            && !endpoint.interest_update_queued.replace(true)
339        {
340            self.interest_update_endpoints.push(endpoint.clone());
341            self.has_interest_update_endpoints.set(true);
342        }
343    }
344
345    pub(crate) fn add_flushable_endpoint(
346        &self,
347        endpoint: &Rc<Endpoint>,
348        client: Option<&Rc<Client>>,
349    ) {
350        if self.destroyed.get() {
351            return;
352        }
353        self.flushable_endpoints.push(EndpointWithClient {
354            endpoint: endpoint.clone(),
355            client: client.cloned(),
356        });
357        self.has_flushable_endpoints.set(true);
358    }
359
360    fn wait_for_work(&self, _: &HandlerLock<'_>, mut timeout: c::c_int) -> Result<(), StateError> {
361        self.check_destroyed()?;
362        let mut events = [PollEvent::default(); poll::MAX_EVENTS];
363        let pollables = &mut *self.pollables.borrow_mut();
364        loop {
365            let n = self
366                .poller
367                .read_events(timeout, &mut events)
368                .map_err(StateErrorKind::PollError)?;
369            if n == 0 {
370                return Ok(());
371            }
372            timeout = 0;
373            for event in &events[0..n] {
374                let id = event.u64;
375                let Some(pollable) = pollables.get(&id) else {
376                    continue;
377                };
378                match pollable {
379                    Pollable::Endpoint(ewc) => {
380                        let events = event.events;
381                        if events & poll::ERROR != 0 {
382                            if let Some(client) = &ewc.client {
383                                self.add_client_to_kill(client);
384                            } else {
385                                return Err(StateErrorKind::ServerHangup.into());
386                            }
387                            continue;
388                        }
389                        ewc.endpoint.current_interest.set(0);
390                        self.change_interest(&ewc.endpoint, |i| i & !events);
391                        if events & poll::READABLE != 0 {
392                            self.readable_endpoints.push(ewc.clone());
393                            self.has_readable_endpoints.set(true);
394                        }
395                        if events & poll::WRITABLE != 0 {
396                            self.flushable_endpoints.push(ewc.clone());
397                            self.has_flushable_endpoints.set(true);
398                        }
399                    }
400                    Pollable::Acceptor(a) => {
401                        self.acceptable_acceptors.push(a.clone());
402                        self.has_acceptable_acceptors.set(true);
403                    }
404                    Pollable::Destructor(fd, destroy) => {
405                        let destroy = destroy.load(Acquire);
406                        self.poller.unregister(fd.as_fd());
407                        pollables.remove(&id);
408                        if destroy {
409                            return Err(StateErrorKind::RemoteDestroyed.into());
410                        }
411                    }
412                }
413            }
414        }
415    }
416
417    fn add_client_to_kill(&self, client: &Rc<Client>) {
418        self.clients_to_kill.push(client.clone());
419        self.has_clients_to_kill.set(true);
420    }
421
422    fn kill_clients(&self) -> bool {
423        if !self.has_clients_to_kill.get() {
424            return false;
425        }
426        while let Some(client) = self.clients_to_kill.pop() {
427            if let Some(handler) = client.handler.borrow_mut().take() {
428                handler.disconnected();
429            }
430            client.disconnect();
431        }
432        self.has_clients_to_kill.set(false);
433        true
434    }
435
436    fn create_pollable_id(&self) -> u64 {
437        let id = self.next_pollable_id.get();
438        self.next_pollable_id.set(id + 1);
439        id
440    }
441
442    fn update_interests(&self) -> Result<(), StateError> {
443        if self.has_interest_update_endpoints.get() {
444            while let Some(endpoint) = self.interest_update_endpoints.pop() {
445                endpoint.interest_update_queued.set(false);
446                let desired = endpoint.desired_interest.get();
447                if desired == endpoint.current_interest.get() {
448                    continue;
449                }
450                if endpoint.unregistered.get() {
451                    continue;
452                }
453                self.poller
454                    .update_interests(endpoint.id, endpoint.socket.as_fd(), desired)
455                    .map_err(StateErrorKind::PollError)?;
456                endpoint.current_interest.set(desired);
457            }
458            self.has_interest_update_endpoints.set(false);
459        }
460        if self.has_interest_update_acceptors.get() {
461            while let Some(acceptor) = self.interest_update_acceptors.pop() {
462                self.poller
463                    .update_interests(acceptor.id, acceptor.socket.as_fd(), poll::READABLE)
464                    .map_err(StateErrorKind::PollError)?;
465            }
466            self.has_interest_update_acceptors.set(false);
467        }
468        Ok(())
469    }
470
471    fn check_destroyed(&self) -> Result<(), StateError> {
472        if self.destroyed.get() {
473            return Err(StateErrorKind::Destroyed.into());
474        }
475        Ok(())
476    }
477
478    #[cfg(feature = "logging")]
479    #[cold]
480    pub(crate) fn log(&self, args: std::fmt::Arguments<'_>) {
481        use std::io::Write;
482        let writer = &mut *self.log_writer.borrow_mut();
483        let _ = writer.write_fmt(args);
484        let _ = writer.flush();
485    }
486}
487
488/// These functions can be used to create a new state.
489impl State {
490    /// Creates a new [`StateBuilder`].
491    pub fn builder(baseline: Baseline) -> StateBuilder {
492        StateBuilder::new(baseline)
493    }
494}
495
496/// These functions can be used to dispatch and flush messages.
497impl State {
498    /// Performs a blocking dispatch.
499    ///
500    /// This is a shorthand for `self.dispatch(None)`.
501    pub fn dispatch_blocking(self: &Rc<Self>) -> Result<bool, StateError> {
502        self.dispatch(None)
503    }
504
505    /// Performs a non-blocking dispatch.
506    ///
507    /// This is a shorthand for `self.dispatch(Some(Duration::from_secs(0))`.
508    pub fn dispatch_available(self: &Rc<Self>) -> Result<bool, StateError> {
509        self.dispatch(Some(Duration::from_secs(0)))
510    }
511
512    /// Performs a dispatch.
513    ///
514    /// The timeout determines how long this function will wait for new events. If the
515    /// timeout is `None`, then it will wait indefinitely. If the timeout is `0`, then
516    /// it will only process currently available events.
517    ///
518    /// If the timeout is not `0`, then outgoing messages will be flushed before waiting.
519    ///
520    /// Outgoing messages will be flushed immediately before this function returns.
521    ///
522    /// The return value indicates if any work was performed.
523    ///
524    /// This function is not reentrant. It should not be called from within a callback.
525    /// Trying to do so will cause it to return an error immediately and the state will
526    /// be otherwise unchanged.
527    pub fn dispatch(self: &Rc<Self>, timeout: Option<Duration>) -> Result<bool, StateError> {
528        let mut did_work = false;
529        let lock = self.acquire_handler_lock()?;
530        let timeout = timeout
531            .and_then(|t| t.as_millis().try_into().ok())
532            .unwrap_or(-1);
533        let destroy_on_error = on_drop(|| self.destroy());
534        if timeout != 0 {
535            did_work |= self.flush_locked(&lock)?;
536        }
537        self.wait_for_work(&lock, timeout)?;
538        did_work |= self.accept_connections(&lock)?;
539        did_work |= self.read_messages(&lock)?;
540        did_work |= self.flush_locked(&lock)?;
541        destroy_on_error.forget();
542        Ok(did_work)
543    }
544}
545
546impl State {
547    /// Returns a file descriptor that can be used with epoll or similar.
548    ///
549    /// If this file descriptor becomes readable, the state should be dispatched.
550    /// [`Self::before_poll`] should be used before going to sleep.
551    ///
552    /// This function always returns the same file descriptor.
553    pub fn poll_fd(&self) -> &Rc<OwnedFd> {
554        self.poller.fd()
555    }
556
557    /// Prepares the state for an external poll operation.
558    ///
559    /// If [`Self::poll_fd`] is used, this function should be called immediately before
560    /// going to sleep. Otherwise, outgoing messages might not be flushed.
561    ///
562    /// ```
563    /// # use std::os::fd::OwnedFd;
564    /// # use std::rc::Rc;
565    /// # use wl_proxy::state::State;
566    /// # fn poll(fd: &OwnedFd) { }
567    /// # fn f(state: &Rc<State>) {
568    /// loop {
569    ///     state.before_poll().unwrap();
570    ///     poll(state.poll_fd());
571    ///     state.dispatch_available().unwrap();
572    /// }
573    /// # }
574    /// ```
575    pub fn before_poll(&self) -> Result<(), StateError> {
576        let lock = self.acquire_handler_lock()?;
577        let destroy_on_error = on_drop(|| self.destroy());
578        self.flush_locked(&lock)?;
579        destroy_on_error.forget();
580        Ok(())
581    }
582}
583
584/// These functions can be used to manipulate objects.
585impl State {
586    /// Creates a new object.
587    ///
588    /// The new object is not associated with a client ID or a server ID. It can become
589    /// associated with a client ID by sending an event with a `new_id` parameter. It can
590    /// become associated with a server ID by sending a request with a `new_id` parameter.
591    ///
592    /// The object can only be associated with one client at a time. The association with
593    /// a client is removed when the object is used in a destructor event.
594    ///
595    /// This function does not enforce that the version is less than or equal to the
596    /// maximum version supported by this crate. Using a version that exceeds tha maximum
597    /// supported version can cause a protocol error if the client sends a request that is
598    /// not available in the maximum supported protocol version or if the server sends an
599    /// event that is not available in the maximum supported protocol version.
600    pub fn create_object<P>(self: &Rc<Self>, version: u32) -> Rc<P>
601    where
602        P: Object,
603    {
604        P::new(self, version)
605    }
606
607    /// Returns a wl_display object.
608    pub fn display(self: &Rc<Self>) -> Rc<WlDisplay> {
609        let display = WlDisplay::new(self, 1);
610        if self.server.is_some() {
611            display.core().server_obj_id.set(Some(1));
612        }
613        display
614    }
615
616    /// Changes the default forward-to-client setting.
617    ///
618    /// This affects objects created after this call. See
619    /// [`ObjectCoreApi::set_forward_to_client`].
620    pub fn set_default_forward_to_client(&self, enabled: bool) {
621        self.forward_to_client.set(enabled);
622    }
623
624    /// Changes the default forward-to-server setting.
625    ///
626    /// This affects objects created after this call. See
627    /// [`ObjectCoreApi::set_forward_to_server`].
628    pub fn set_default_forward_to_server(&self, enabled: bool) {
629        self.forward_to_server.set(enabled);
630    }
631}
632
633/// These functions can be used to manage sockets associated with this state.
634impl State {
635    /// Creates a new connection to this proxy.
636    ///
637    /// The returned file descriptor is the client end of the connection and can be used
638    /// with a function such as `wl_display_connect_to_fd` or with the `WAYLAND_SOCKET`
639    /// environment variable.
640    ///
641    /// The [`StateHandler::new_client`] callback will not be invoked.
642    pub fn connect(self: &Rc<Self>) -> Result<(Rc<Client>, OwnedFd), StateError> {
643        let (server_fd, client_fd) = uapi::socketpair(
644            c::AF_UNIX,
645            c::SOCK_STREAM | c::SOCK_NONBLOCK | c::SOCK_CLOEXEC,
646            0,
647        )
648        .map_err(|e| StateErrorKind::Socketpair(e.into()))?;
649        let client = self.create_client(None, &Rc::new(server_fd.into()))?;
650        Ok((client, client_fd.into()))
651    }
652
653    /// Creates a new connection to this proxy from an existing socket.
654    ///
655    /// The file descriptor should be the server end of the connection. It can be created
656    /// with a function such as `socketpair` or by accepting a connection from a
657    /// file-system socket.
658    ///
659    /// The [`StateHandler::new_client`] callback will not be invoked.
660    pub fn add_client(self: &Rc<Self>, socket: &Rc<OwnedFd>) -> Result<Rc<Client>, StateError> {
661        self.create_client(None, socket)
662    }
663
664    /// Creates a new file-system acceptor and starts listening for connections.
665    ///
666    /// See [`Acceptor::new`] for the meaning of the `max_tries` parameter.
667    ///
668    /// Calling [`State::dispatch`] will automatically accept connections from this
669    /// acceptor. The [`StateHandler::new_client`] callback will be invoked when this
670    /// happens.
671    pub fn create_acceptor(&self, max_tries: u32) -> Result<Rc<Acceptor>, StateError> {
672        self.check_destroyed()?;
673        let id = self.create_pollable_id();
674        let acceptor =
675            Acceptor::create(id, max_tries, true).map_err(StateErrorKind::CreateAcceptor)?;
676        self.poller
677            .register(id, acceptor.socket.as_fd())
678            .map_err(StateErrorKind::PollError)?;
679        self.update_interests()?;
680        self.interest_update_acceptors.push(acceptor.clone());
681        self.has_interest_update_acceptors.set(true);
682        self.pollables
683            .borrow_mut()
684            .insert(id, Pollable::Acceptor(acceptor.clone()));
685        Ok(acceptor)
686    }
687
688    fn create_client(
689        self: &Rc<Self>,
690        lock: Option<&HandlerLock<'_>>,
691        socket: &Rc<OwnedFd>,
692    ) -> Result<Rc<Client>, StateError> {
693        self.check_destroyed()?;
694        let id = self.create_pollable_id();
695        self.poller
696            .register(id, socket.as_fd())
697            .map_err(StateErrorKind::PollError)?;
698        let endpoint = Endpoint::new(id, socket);
699        self.change_interest(&endpoint, |i| i | poll::READABLE);
700        self.update_interests()?;
701        let client = Rc::new(Client {
702            state: self.clone(),
703            endpoint: endpoint.clone(),
704            display: self.display(),
705            destroyed: Cell::new(false),
706            handler: Default::default(),
707        });
708        client
709            .display
710            .core()
711            .set_client_id(&client, 1, client.display.clone())
712            .unwrap();
713        self.pollables.borrow_mut().insert(
714            id,
715            Pollable::Endpoint(EndpointWithClient {
716                endpoint,
717                client: Some(client.clone()),
718            }),
719        );
720        if lock.is_some()
721            && let Some(handler) = &mut *self.handler.borrow_mut()
722        {
723            handler.new_client(&client);
724        }
725        Ok(client)
726    }
727}
728
729/// These functions can be used to manipulate the [`StateHandler`] of this state.
730///
731/// These functions can be called at any time, even from within a handler callback. In
732/// that case, the handler is replaced as soon as the callback returns.
733impl State {
734    /// Unsets the handler.
735    pub fn unset_handler(&self) {
736        self.handler.set(None);
737    }
738
739    /// Sets a new handler.
740    pub fn set_handler(&self, handler: impl StateHandler) {
741        self.set_boxed_handler(Box::new(handler))
742    }
743
744    /// Sets a new, already boxed handler.
745    pub fn set_boxed_handler(&self, handler: Box<dyn StateHandler>) {
746        if self.destroyed.get() {
747            return;
748        }
749        self.handler.set(Some(handler));
750    }
751}
752
753/// These functions can be used to check the state status and to destroy the state.
754impl State {
755    /// Returns whether this state is not destroyed.
756    ///
757    /// This is the same as `!self.is_destroyed()`.
758    pub fn is_not_destroyed(&self) -> bool {
759        !self.is_destroyed()
760    }
761
762    /// Returns whether the state is destroyed.
763    ///
764    /// If the state is destroyed, most functions that can return an error will return an
765    /// error saying that the state is already destroyed.
766    ///
767    /// This function or [`Self::is_not_destroyed`] should be used before dispatching the
768    /// state.
769    ///
770    /// # Example
771    ///
772    /// ```
773    /// # use std::rc::Rc;
774    /// # use error_reporter::Report;
775    /// # use wl_proxy::state::State;
776    /// #
777    /// # fn f(state: &Rc<State>) {
778    /// while state.is_not_destroyed() {
779    ///     if let Err(e) = state.dispatch_blocking() {
780    ///         log::error!("Could not dispatch the state: {}", Report::new(e));
781    ///     }
782    /// }
783    /// # }
784    /// ```
785    pub fn is_destroyed(&self) -> bool {
786        self.destroyed.get()
787    }
788
789    /// Destroys this state.
790    ///
791    /// This function unsets all handlers and destroys all clients. You should drop the
792    /// state after calling this function.
793    pub fn destroy(&self) {
794        if self.destroyed.replace(true) {
795            return;
796        }
797        let objects = &mut *self.object_stash.borrow();
798        for pollable in self.pollables.borrow().values() {
799            let fd = match pollable {
800                Pollable::Endpoint(ewc) => {
801                    if let Some(c) = &ewc.client {
802                        c.destroyed.set(true);
803                    }
804                    objects.extend(ewc.endpoint.objects.borrow_mut().drain().map(|v| v.1));
805                    &ewc.endpoint.socket
806                }
807                Pollable::Acceptor(a) => &a.socket,
808                Pollable::Destructor(fd, _) => fd,
809            };
810            self.poller.unregister(fd.as_fd());
811        }
812        objects.clear();
813        for object in self.all_objects.borrow().values() {
814            if let Some(object) = object.upgrade() {
815                objects.push(object);
816            }
817        }
818        for object in objects {
819            object.unset_handler();
820            object.core().client.take();
821        }
822        self.handler.set(None);
823        self.pollables.borrow_mut().clear();
824        self.acceptable_acceptors.take();
825        self.clients_to_kill.take();
826        self.readable_endpoints.take();
827        self.flushable_endpoints.take();
828        self.interest_update_endpoints.take();
829        self.interest_update_acceptors.take();
830        self.all_objects.borrow_mut().clear();
831        // Ensure that the poll fd stays permanently readable.
832        let _ = self.create_remote_destructor();
833    }
834
835    /// Creates a RAII destructor for this state.
836    ///
837    /// Dropping the destructor will automatically call [`State::destroy`] unless you
838    /// first call [`Destructor::disable`].
839    ///
840    /// State objects contain reference cycles that must be cleared manually to release
841    /// the associated resources. Dropping the [`State`] is usually not sufficient to do
842    /// this. Instead, [`State::destroy`] must be called manually. This function can be
843    /// used to accomplish this in an application that otherwise relies on RAII semantics.
844    ///
845    /// Ensure that the destructor is itself not part of a reference cycle.
846    pub fn create_destructor(self: &Rc<Self>) -> Destructor {
847        Destructor {
848            state: self.clone(),
849            enabled: Cell::new(true),
850        }
851    }
852
853    /// Creates a `Sync+Send` RAII destructor for this state.
854    ///
855    /// This function is similar to [`State::create_destructor`] but the returned
856    /// destructor implements `Sync+Send`. This destructor can therefore be used to
857    /// destroy states running in a different thread.
858    pub fn create_remote_destructor(&self) -> Result<RemoteDestructor, StateError> {
859        let (r, w) = pipe().map_err(StateErrorKind::CreatePipe)?;
860        let r: OwnedFd = r.into();
861        let id = self.create_pollable_id();
862        self.poller
863            .register(id, r.as_fd())
864            .map_err(StateErrorKind::PollError)?;
865        let destroy = Arc::new(AtomicBool::new(false));
866        self.pollables
867            .borrow_mut()
868            .insert(id, Pollable::Destructor(r, destroy.clone()));
869        Ok(RemoteDestructor {
870            destroy,
871            _fd: w.into(),
872            enabled: AtomicBool::new(true),
873        })
874    }
875}
876
877impl StateError {
878    /// Returns whether this error was emitted because the state is already destroyed.
879    ///
880    /// This can be used to determine the severity of emitted log messages.
881    pub fn is_destroyed(&self) -> bool {
882        matches!(self.0, StateErrorKind::Destroyed)
883    }
884}
885
886impl Drop for HandlerLock<'_> {
887    fn drop(&mut self) {
888        self.state.global_lock_held.set(false);
889    }
890}
wl_proxy/state.rs

wl_proxy/
state.rs
