use crate::async_engine::SpawnedFuture;
pub use crate::client::error::LookupError;
use crate::client::objects::FIRST_INVALID_ID;
use crate::client::objects::FIRST_SYNTHETIC_ID;
use crate::client::objects::Objects;
use crate::criteria::CritDestroyListener;
use crate::criteria::CritMatcherId;
use crate::criteria::clm::CL_CHANGED_DESTROYED;
use crate::criteria::clm::CL_CHANGED_NEW;
use crate::criteria::clm::ClMatcherChange;
use crate::globals::Singleton;
use crate::ifs::jay_client_trace::ClientTracers;
use crate::ifs::wl_display::WlDisplay;
use crate::ifs::wl_registry::WlRegistry;
use crate::ifs::wl_surface::WlSurface;
use crate::ifs::wl_surface::commit_timeline::CommitTimelines;
use crate::ifs::xdg_activation_token_v1::ActivationToken;
use crate::ifs::xdg_session_v1::XdgSessionV1;
use crate::leaks::Tracker;
use crate::object::Interface;
use crate::object::Object;
use crate::object::SyntheticObjectEventHandler;
use crate::object::WL_DISPLAY_ID;
use crate::security_context_acceptor::AcceptorMetadata;
use crate::sqlite::SqliteAccounting;
use crate::state::State;
use crate::utils::asyncevent::AsyncEvent;
use crate::utils::bhash::BHashMap;
use crate::utils::buffd::MsgFormatter;
use crate::utils::buffd::MsgParser;
use crate::utils::buffd::MsgParserError;
use crate::utils::buffd::OutBufferSwapchain;
use crate::utils::buffd::SyntheticBufOut;
use crate::utils::client_trace::ClientTraceMessage;
use crate::utils::copyhashmap::CopyHashMap;
use crate::utils::copyhashmap::Locked;
use crate::utils::errorfmt::ErrorFmt;
use crate::utils::linkedlist::LinkedList;
use crate::utils::liveness::Liveness;
use crate::utils::numcell::NumCell;
use crate::utils::pending_serial::PendingSerial;
use crate::utils::pid_info::PidInfo;
use crate::utils::pid_info::get_pid_info;
use crate::utils::pid_info::get_socket_creds;
use crate::utils::pidfd_send_signal::pidfd_send_signal;
use crate::utils::queue::AsyncQueue;
use crate::utils::stack::Stack;
use crate::utils::static_text::StaticText;
use crate::utils::woid_hash::WoidBuildHasher;
use crate::wire::ObjectId;
use crate::wire::WlRegistryId;
pub use client_dfs_g_fuse::ClientView;
pub use error::ClientError;
pub use error::ParserError;
use jay_proc::GetLiveness;
use jay_proc::jay_hash;
use linearize::StaticMap;
pub use objects::MIN_SERVER_ID;
use std::cell::Cell;
use std::cell::RefCell;
use std::collections::VecDeque;
use std::error::Error;
use std::fmt::Debug;
use std::fmt::Display;
use std::fmt::Formatter;
use std::mem;
use std::ops::DerefMut;
use std::rc::Rc;
use std::rc::Weak;
use uapi::OwnedFd;
use uapi::c;
mod client_dfs_g_fuse;
mod error;
mod objects;
mod synthetic_helpers;
mod tasks;
bitflags! {
ClientCaps: u32;
CAP_DATA_CONTROL_MANAGER = 1 << 0,
CAP_VIRTUAL_KEYBOARD_MANAGER = 1 << 1,
CAP_FOREIGN_TOPLEVEL_LIST = 1 << 2,
CAP_IDLE_NOTIFIER = 1 << 3,
CAP_SESSION_LOCK_MANAGER = 1 << 4,
CAP_JAY_COMPOSITOR = 1 << 5,
CAP_LAYER_SHELL = 1 << 6,
CAP_SCREENCOPY_MANAGER = 1 << 7,
CAP_SEAT_MANAGER = 1 << 8,
CAP_DRM_LEASE = 1 << 9,
CAP_INPUT_METHOD = 1 << 10,
CAP_WORKSPACE = 1 << 11,
CAP_FOREIGN_TOPLEVEL_MANAGER = 1 << 12,
CAP_HEAD_MANAGER = 1 << 13,
CAP_GAMMA_CONTROL_MANAGER = 1 << 14,
CAP_VIRTUAL_POINTER_MANAGER = 1 << 15,
}
impl StaticText for ClientCapsEnum {
fn text(&self) -> &'static str {
match self {
ClientCapsEnum::CAP_DATA_CONTROL_MANAGER => "data-control",
ClientCapsEnum::CAP_VIRTUAL_KEYBOARD_MANAGER => "virtual-keyboard",
ClientCapsEnum::CAP_FOREIGN_TOPLEVEL_LIST => "foreign-toplevel-list",
ClientCapsEnum::CAP_IDLE_NOTIFIER => "idle-notifier",
ClientCapsEnum::CAP_SESSION_LOCK_MANAGER => "session-lock",
ClientCapsEnum::CAP_JAY_COMPOSITOR => "jay-compositor",
ClientCapsEnum::CAP_LAYER_SHELL => "layer-shell",
ClientCapsEnum::CAP_SCREENCOPY_MANAGER => "screencopy",
ClientCapsEnum::CAP_SEAT_MANAGER => "seat-manager",
ClientCapsEnum::CAP_DRM_LEASE => "drm-lease",
ClientCapsEnum::CAP_INPUT_METHOD => "input-method",
ClientCapsEnum::CAP_WORKSPACE => "workspace-manager",
ClientCapsEnum::CAP_FOREIGN_TOPLEVEL_MANAGER => "foreign-toplevel-manager",
ClientCapsEnum::CAP_HEAD_MANAGER => "head-manager",
ClientCapsEnum::CAP_GAMMA_CONTROL_MANAGER => "gamma-control-manager",
ClientCapsEnum::CAP_VIRTUAL_POINTER_MANAGER => "virtual-pointer",
}
}
}
pub const CAPS_DEFAULT: ClientCaps = ClientCaps(CAP_LAYER_SHELL.0 | CAP_DRM_LEASE.0);
pub const CAPS_DEFAULT_SANDBOXED: ClientCaps = ClientCaps(CAP_DRM_LEASE.0);
#[jay_hash]
#[derive(Debug, Copy, Clone, Ord, PartialOrd, Eq)]
pub struct ClientId(u64);
impl ClientId {
pub fn raw(self) -> u64 {
self.0
}
pub fn from_raw(val: u64) -> Self {
Self(val)
}
}
impl Display for ClientId {
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
Display::fmt(&self.0, f)
}
}
pub struct Clients {
next_client_id: NumCell<u64>,
pub clients: RefCell<BHashMap<ClientId, ClientHolder>>,
shutdown_clients: RefCell<BHashMap<ClientId, ClientHolder>>,
cleared: Cell<bool>,
clear: AsyncQueue<ClientClear>,
}
impl Clients {
pub fn new() -> Self {
Self {
next_client_id: NumCell::new(1),
clients: Default::default(),
shutdown_clients: Default::default(),
cleared: Default::default(),
clear: Default::default(),
}
}
pub fn clear(&self) {
mem::take(self.clients.borrow_mut().deref_mut());
mem::take(self.shutdown_clients.borrow_mut().deref_mut());
self.cleared.set(true);
self.clear.clear();
}
pub fn id(&self) -> ClientId {
ClientId(self.next_client_id.fetch_add(1))
}
pub fn get(&self, id: ClientId) -> Result<Rc<Client>, ClientError> {
let clients = self.clients.borrow();
match clients.get(&id) {
Some(c) => Ok(c.data.clone()),
_ => Err(ClientError::ClientDoesNotExist(id)),
}
}
pub fn spawn(
&self,
id: ClientId,
global: &Rc<State>,
socket: Rc<OwnedFd>,
bounding_caps: ClientCaps,
set_bounding_caps_for_children: bool,
acceptor: &Rc<AcceptorMetadata>,
) {
let Some((uid, pid)) = get_socket_creds(&socket) else {
return;
};
self.spawn2(
id,
global,
socket,
uid,
pid,
bounding_caps,
set_bounding_caps_for_children,
false,
acceptor,
);
}
pub fn spawn2(
&self,
id: ClientId,
global: &Rc<State>,
socket: Rc<OwnedFd>,
uid: c::uid_t,
pid: c::pid_t,
bounding_caps: ClientCaps,
set_bounding_caps_for_children: bool,
is_xwayland: bool,
acceptor: &Rc<AcceptorMetadata>,
) -> Rc<Client> {
let effective_caps = match acceptor.sandboxed {
true => CAPS_DEFAULT_SANDBOXED,
false => CAPS_DEFAULT,
};
let data = Rc::new_cyclic(|slf| Client {
id,
state: global.clone(),
checking_queue_size: Cell::new(false),
socket,
objects: Default::default(),
invalid_ids: NumCell::new(FIRST_INVALID_ID),
swapchain: Default::default(),
flush_request: Default::default(),
shutdown: Default::default(),
tracker: Default::default(),
is_xwayland,
effective_caps: Cell::new(effective_caps & bounding_caps),
bounding_caps_for_children: Cell::new(bounding_caps),
last_enter_serial: Default::default(),
pid_info: get_pid_info(uid, pid),
serials: Default::default(),
symmetric_delete: Cell::new(false),
last_xwayland_serial: Cell::new(0),
surfaces_by_xwayland_serial: Default::default(),
activation_tokens: Default::default(),
commit_timelines: Rc::new(CommitTimelines::new(
&global.wait_for_syncobj,
&global.ring,
&global.eng,
slf,
)),
wire_scale: Default::default(),
focus_stealing_serial: Default::default(),
changed_properties: Default::default(),
destroyed: Default::default(),
acceptor: acceptor.clone(),
sqlite_accounting: Rc::new(SqliteAccounting::new(256)),
live_sessions: Default::default(),
num_live_sessions: Default::default(),
connect_time_us: global.now_usec_rt(),
tracers: ClientTracers::new(&global.eng),
terminate_shutdown: Default::default(),
terminate_kill: Default::default(),
terminating: Default::default(),
terminate: Default::default(),
request_blockers: Default::default(),
requests_unblocked: Default::default(),
synthetic_requests: Default::default(),
synthetic_events: Default::default(),
synthetic_singletons: StaticMap::from_fn(|_| Cell::new(ObjectId::NONE)),
synthetic_registry: Cell::new(WlRegistryId::NONE),
liveness: Default::default(),
});
track!(data, data);
global.update_capabilities(&data, bounding_caps, set_bounding_caps_for_children);
if acceptor.secure || is_xwayland {
data.effective_caps.set(ClientCaps::all());
}
let display = Rc::new(WlDisplay::new(&data));
track!(data, display);
data.objects.display.set(Some(display.clone()));
data.objects
.add_client_object(WL_DISPLAY_ID.into(), display)
.expect("");
let client = ClientHolder {
_handler: global.eng.spawn("client", tasks::client(data.clone())),
data: data.clone(),
};
log::info!(
"Client {} connected, pid: {}, uid: {}, fd: {}, comm: {:?}, caps: {:?}",
id,
pid,
uid,
client.data.socket.raw(),
data.pid_info.comm,
data.effective_caps.get(),
);
client.data.property_changed(CL_CHANGED_NEW);
global.global_tracers.announce(&client.data);
self.clients.borrow_mut().insert(client.data.id, client);
data
}
fn kill(&self, client: ClientId) {
log::info!("Removing client {}", client.0);
if self.clients.borrow_mut().remove(&client).is_none() {
self.shutdown_clients.borrow_mut().remove(&client);
}
}
fn shutdown(&self, client_id: ClientId) {
if let Some(client) = self.clients.borrow_mut().remove(&client_id) {
log::info!("Shutting down client {}", client.data.id.0);
client.data.shutdown.trigger();
client.data.flush_request.trigger();
self.shutdown_clients.borrow_mut().insert(client_id, client);
}
}
pub fn broadcast<B>(&self, required_caps: ClientCaps, xwayland_only: bool, mut f: B)
where
B: FnMut(&Rc<Client>),
{
let clients = self.clients.borrow();
for client in clients.values() {
if client.data.effective_caps.get().contains(required_caps)
&& (!xwayland_only || client.data.is_xwayland)
{
f(&client.data);
}
}
}
}
impl Drop for Clients {
fn drop(&mut self) {
let _clients1 = mem::take(&mut *self.clients.borrow_mut());
let _clients2 = mem::take(&mut *self.shutdown_clients.borrow_mut());
}
}
pub struct ClientHolder {
pub data: Rc<Client>,
_handler: SpawnedFuture<()>,
}
impl Drop for ClientHolder {
fn drop(&mut self) {
self.data.terminating.set(true);
let clear = ClientClear {
data: self.data.clone(),
};
let clients = &self.data.state.clients;
if clients.cleared.get() {
return;
}
clients.clear.push(clear);
}
}
struct ClientClear {
data: Rc<Client>,
}
impl Drop for ClientClear {
fn drop(&mut self) {
self.data.objects.destroy();
self.data.flush_request.clear();
self.data.shutdown.clear();
self.data.surfaces_by_xwayland_serial.clear();
self.data.remove_activation_tokens();
self.data.commit_timelines.clear();
self.data.property_changed(CL_CHANGED_DESTROYED);
if self.data.is_xwayland
&& let Some(pidfd) = self.data.state.xwayland.pidfd.get()
&& let Err(e) = pidfd_send_signal(&pidfd, c::SIGKILL)
{
log::error!("Could not kill Xwayland: {}", ErrorFmt(e));
}
self.data.tracers.handle_disconnected();
}
}
pub trait EventFormatter: ClientTraceMessage {
const NUM_FDS: u32;
fn format(self, fmt: &mut MsgFormatter<'_>);
fn id(&self) -> ObjectId;
}
pub trait RequestParser<'a>: Sized + ClientTraceMessage {
type Generic<'b>: RequestParser<'b>;
const ID: u32;
fn parse(parser: &mut MsgParser<'_, 'a>) -> Result<Self, MsgParserError>;
}
#[derive(GetLiveness)]
pub struct Client {
pub id: ClientId,
pub state: Rc<State>,
checking_queue_size: Cell<bool>,
socket: Rc<OwnedFd>,
pub objects: Objects,
invalid_ids: NumCell<u64>,
swapchain: Rc<RefCell<OutBufferSwapchain>>,
flush_request: AsyncEvent,
shutdown: AsyncEvent,
tracker: Tracker<Client>,
pub is_xwayland: bool,
pub effective_caps: Cell<ClientCaps>,
pub bounding_caps_for_children: Cell<ClientCaps>,
pub last_enter_serial: Cell<Option<u64>>,
pub pid_info: PidInfo,
pub serials: RefCell<VecDeque<SerialRange>>,
pub symmetric_delete: Cell<bool>,
pub last_xwayland_serial: Cell<u64>,
pub surfaces_by_xwayland_serial: CopyHashMap<u64, Rc<WlSurface>>,
pub activation_tokens: RefCell<VecDeque<ActivationToken>>,
pub commit_timelines: Rc<CommitTimelines>,
pub wire_scale: Cell<Option<i32>>,
pub focus_stealing_serial: Cell<Option<u64>>,
pub changed_properties: Cell<ClMatcherChange>,
pub destroyed: CopyHashMap<CritMatcherId, Weak<dyn CritDestroyListener<Rc<Self>>>>,
pub acceptor: Rc<AcceptorMetadata>,
pub sqlite_accounting: Rc<SqliteAccounting>,
pub live_sessions: LinkedList<Rc<XdgSessionV1>>,
pub num_live_sessions: NumCell<usize>,
pub connect_time_us: u64,
pub tracers: ClientTracers,
terminate_shutdown: Cell<bool>,
terminate_kill: Cell<bool>,
terminating: Cell<bool>,
terminate: AsyncEvent,
request_blockers: NumCell<u64>,
requests_unblocked: AsyncEvent,
synthetic_requests: Synthetic,
synthetic_events: Synthetic,
synthetic_singletons: StaticMap<Singleton, Cell<ObjectId>>,
synthetic_registry: Cell<WlRegistryId>,
liveness: Liveness,
}
#[derive(Default)]
struct Synthetic {
trigger: AsyncEvent,
buf: RefCell<SyntheticBufOut>,
to_remove: Stack<ObjectId>,
}
pub const NUM_CACHED_SERIAL_RANGES: usize = 64;
pub struct SerialRange {
pub lo: u64,
pub hi: u64,
}
impl Client {
fn invalid_request(&self, obj: &dyn Object, request: u32) {
log::error!(
"Client {} sent an invalid request {} on object {} of type {}",
self.id.0,
request,
obj.id(),
obj.interface().name(),
);
match self.display() {
Ok(d) => {
d.send_invalid_request(obj, request);
self.shutdown();
}
Err(e) => {
log::error!(
"Could not retrieve display of client {}: {}",
self.id,
ErrorFmt(e),
);
self.kill();
}
}
}
pub fn map_serial(&self, serial: u32) -> Option<u64> {
let serials = self.serials.borrow_mut();
let latest = serials.back()?;
let mut serial = ((latest.hi >> 32) << 32) | (serial as u64);
if serial > latest.hi {
serial = serial.checked_sub(1 << 32)?;
}
for range in serials.iter().rev() {
if serial > range.hi {
return None;
}
if serial >= range.lo {
return Some(serial);
}
}
if serials.len() == NUM_CACHED_SERIAL_RANGES {
return Some(serial);
}
None
}
pub fn next_serial(&self) -> u64 {
self.state.next_serial(Some(self))
}
pub fn pending_serial(&self) -> PendingSerial<'_> {
PendingSerial::new(self)
}
pub fn new_id<T: From<ObjectId>>(&self, parent: &impl Object) -> T {
self.new_id3(parent.id()).into()
}
pub fn new_id2<T: From<ObjectId>>(&self, parent: impl Into<ObjectId>) -> T {
self.new_id3(parent.into()).into()
}
fn new_id3(&self, parent: ObjectId) -> ObjectId {
match self.objects.id(self, parent) {
Ok(id) => id,
Err(e) => {
let id = self.invalid_ids.fetch_sub(1);
if id == FIRST_INVALID_ID {
self.error(e);
}
ObjectId::from_raw(id)
}
}
}
pub fn new_synthetic_id<T: From<ObjectId>>(&self) -> T {
self.objects.synthetic_id().into()
}
fn display(&self) -> Result<Rc<WlDisplay>, ClientError> {
match self.objects.display.get() {
Some(d) => Ok(d),
_ => Err(ClientError::NotADisplay(WL_DISPLAY_ID)),
}
}
pub fn parse<'a, R: RequestParser<'a>>(
&self,
obj: &impl Object,
mut parser: MsgParser<'_, 'a>,
) -> Result<R, MsgParserError> {
let res = R::parse(&mut parser)?;
if self.tracers.num.get() > 0 {
self.tracers.handle_message(obj.id(), &res);
}
Ok(res)
}
pub fn error(&self, message: impl Error) {
if self.terminating.get() {
return;
}
let msg = ErrorFmt(message).to_string();
log::error!("Client {}: A fatal error occurred: {}", self.id.0, msg,);
match self.display() {
Ok(d) => {
d.send_implementation_error(msg);
self.shutdown();
}
Err(e) => {
log::error!(
"Could not retrieve display of client {}: {}",
self.id,
ErrorFmt(e),
);
self.kill();
}
}
}
pub fn protocol_error(&self, obj: &dyn Object, code: u32, message: &str) {
if let Ok(d) = self.display() {
d.send_error(obj.id(), code, message);
}
self.shutdown();
}
pub fn event<T: EventFormatter>(self: &Rc<Self>, event: T) {
if event.id().raw() >= FIRST_SYNTHETIC_ID {
self.synthetic_event(event);
return;
}
if self.tracers.num.get() > 0 {
self.tracers.handle_message(event.id(), &event);
}
let mut fds = vec![];
let mut swapchain = self.swapchain.borrow_mut();
let mut fmt = MsgFormatter::new(&mut swapchain.cur, &mut fds);
event.format(&mut fmt);
fmt.write_len();
if swapchain.cur.is_full() {
swapchain.commit();
if swapchain.exceeds_limit() {
if !self.checking_queue_size.replace(true) {
self.state.slow_clients.push(self.clone());
}
}
}
self.flush_request.trigger();
}
#[cold]
fn synthetic_event<T>(self: &Rc<Self>, event: T)
where
T: EventFormatter,
{
self.synthetic_message(&self.synthetic_events, event);
}
pub fn request<T>(self: &Rc<Self>, request: T)
where
T: EventFormatter,
{
self.synthetic_message(&self.synthetic_requests, request);
}
fn synthetic_message<T>(self: &Rc<Self>, dst: &Synthetic, message: T)
where
T: EventFormatter,
{
let synthetic = &mut *dst.buf.borrow_mut();
synthetic.format(message);
self.request_blockers.fetch_add(1);
dst.trigger.trigger();
}
pub async fn check_queue_size(&self) {
if self.swapchain.borrow_mut().exceeds_limit() {
self.state.eng.yield_now().await;
if self.swapchain.borrow_mut().exceeds_limit() {
log::error!("Client {} is too slow at fetching events", self.id.0);
self.kill();
return;
}
}
self.checking_queue_size.set(false);
}
pub fn lock_registries(&self) -> Locked<'_, WlRegistryId, Rc<WlRegistry>, WoidBuildHasher> {
self.objects.registries()
}
pub fn add_client_obj<T: Object>(&self, obj: &Rc<T>) {
if obj.id().raw() >= FIRST_SYNTHETIC_ID {
self.add_server_obj(obj);
return;
}
if let Err(e) = self.add_obj(obj, true) {
self.error(e);
}
}
pub fn add_server_obj<T: Object>(&self, obj: &Rc<T>) {
self.add_obj(obj, false).expect("add_server_obj failed");
}
fn add_obj<T: Object>(&self, obj: &Rc<T>, client: bool) -> Result<(), ClientError> {
let id = obj.id();
if client {
self.objects.add_client_object(id, obj.clone())?;
} else {
self.objects.add_server_object(id, obj.clone());
}
obj.add(self);
Ok(())
}
pub fn set_synthetic_event_handler(
&self,
id: impl Into<ObjectId>,
event_handler: &Rc<impl SyntheticObjectEventHandler>,
) {
self.objects
.set_synthetic_event_handler(id.into(), event_handler.clone());
}
pub fn remove_obj<T: Object>(self: &Rc<Self>, obj: &T) {
if self.terminating.get() {
return;
}
let id = obj.id();
if id.raw() >= FIRST_SYNTHETIC_ID {
self.synthetic_events.to_remove.push(id);
self.request_blockers.fetch_add(1);
self.synthetic_events.trigger.trigger();
}
obj.remove(self);
if let Err(e) = self.objects.remove_obj(self, id) {
self.error(e);
}
}
pub fn lookup<Id: WaylandObjectLookup>(&self, id: Id) -> Result<Rc<Id::Object>, LookupError> {
match Id::lookup(self, id) {
Some(t) => Ok(t),
_ => Err(LookupError {
interface: Id::INTERFACE,
id: id.into(),
}),
}
}
fn remove_activation_tokens(&self) {
for token in &*self.activation_tokens.borrow() {
self.state.activation_tokens.remove(token);
}
}
pub fn property_changed(self: &Rc<Self>, change: ClMatcherChange) {
let props = self.changed_properties.get();
self.changed_properties.set(props | change);
if props.is_none() && change.is_some() {
self.state.cl_matcher_manager.changed(self);
}
}
pub fn kill(&self) {
self.terminating.set(true);
self.terminate_kill.set(true);
self.terminate.trigger();
}
fn shutdown(&self) {
self.terminating.set(true);
self.terminate_shutdown.set(true);
self.terminate.trigger();
}
pub fn get_synthetic_singleton<T>(self: &Rc<Self>, singleton: Singleton) -> T
where
T: From<ObjectId>,
{
self.get_synthetic_singleton_(singleton).into()
}
fn get_synthetic_singleton_(self: &Rc<Self>, singleton: Singleton) -> ObjectId {
let slot = &self.synthetic_singletons[singleton];
let id = slot.get();
if id.is_some() {
return id;
}
let globals = &self.state.globals;
let registry = self.get_synthetic_registry();
let info = globals.singletons[singleton];
let id = self.send_wl_registry_bind(
registry,
info.name.raw(),
singleton.interface().name(),
info.version,
);
slot.set(id);
id
}
pub fn get_synthetic_registry(self: &Rc<Self>) -> WlRegistryId {
let mut id = self.synthetic_registry.get();
if id.is_none() {
id = self.send_wl_display_get_registry();
self.synthetic_registry.set(id);
}
id
}
}
pub trait WaylandObjectLookup: Copy + Into<ObjectId> {
type Object;
const INTERFACE: Interface;
fn lookup(client: &Client, id: Self) -> Option<Rc<Self::Object>>;
}
pub async fn handle_client_clear(state: Rc<State>) {
let clear = &state.clients.clear;
let mut queue = VecDeque::new();
loop {
clear.non_empty().await;
clear.swap(&mut queue);
queue.clear();
}
}