use std::time::{Duration, Instant, SystemTime};
use std::sync::{Arc, Mutex, Weak};
use std::thread::{self, Thread};
use std::task::{ready, Context, Poll};
use std::collections::HashMap;
use std::future::Future;
use std::pin::Pin;
use std::fmt::Write;
use futures::stream::{FuturesUnordered, StreamExt};
use futures::future::BoxFuture;
use tracing::{debug, warn};
use crate::error::{reason, Error, ErrorKind, Result};
use crate::name::{ArcName, Name};
use super::dev::AnyDev;
pub const DEFAULT_VALIDITY_PERIODS: u32 = 4;
pub const MIN_PERIOD: Duration = Duration::from_millis(5);
pub const DEFAULT_PERIODIC_EVENT_PERIOD: Duration = Duration::from_secs(1);
pub const PERIODIC_EVENT_TOLERANCE: f64 = 0.98;
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub enum PolledKind {
Attr,
Command,
}
impl PolledKind {
pub fn parse(name: &str) -> Result<Self> {
if name.eq_ignore_ascii_case("attribute") {
Ok(Self::Attr)
} else if name.eq_ignore_ascii_case("command") {
Ok(Self::Command)
} else {
Err(Error::new(ErrorKind::Reason(reason::API_NotSupported))
.with_description(format!("Object type {name:?} cannot be polled, only an \
attribute or a command can")))
}
}
pub fn as_str(self) -> &'static str {
match self {
Self::Attr => "attribute",
Self::Command => "command",
}
}
}
#[derive(Debug, Clone)]
pub struct Polled {
pub kind: PolledKind,
pub name: Arc<str>,
pub period: Duration,
}
#[derive(Debug, Clone)]
pub struct PolledStatus {
pub kind: PolledKind,
pub name: Arc<str>,
pub period: Duration,
pub last_duration: Option<Duration>,
}
#[derive(Debug, Clone)]
pub struct PolledHistory {
pub depth: usize,
pub times: Vec<SystemTime>,
pub failure: Option<Error>,
}
#[derive(Debug, Clone)]
pub struct Poller {
state: Arc<Mutex<State>>,
thread: Thread,
}
#[derive(Debug)]
struct State {
devs: HashMap<ArcName, PolledDev>,
paused: bool,
shutdown: bool,
next_id: u64,
origin: Instant,
}
#[derive(Debug)]
struct PolledDev {
dev: Weak<dyn AnyDev>,
validity_periods: u32,
attrs: HashMap<ArcName, PolledEntry>,
commands: HashMap<ArcName, PolledEntry>,
}
#[derive(Debug)]
struct PolledEntry {
id: u64,
name: Arc<str>,
period: Duration,
due: Instant,
in_progress: bool,
last_duration: Option<Duration>,
last_periodic: Option<Instant>,
}
#[derive(Debug, Clone)]
pub struct PollItem {
pub id: u64,
pub name: Arc<str>,
pub valid_for: Duration,
pub last_periodic: Option<Instant>,
}
struct PollStart {
dev: Arc<dyn AnyDev>,
kind: PolledKind,
items: Vec<PollItem>,
}
struct PollEnd {
dev: Arc<str>,
kind: PolledKind,
items: Vec<PollItem>,
last_duration: Duration,
}
impl Poller {
pub fn spawn() -> Result<Self> {
let builder = thread::Builder::new()
.name("server-polling".to_string());
let state = Arc::new(Mutex::new(State {
devs: HashMap::new(),
paused: false,
shutdown: false,
next_id: 0,
origin: Instant::now(),
}));
let handle = {
let state = Arc::clone(&state);
builder.spawn(move || run_thread(&state))
.map_err(|err| Error::new(ErrorKind::Internal)
.with_description(format!("Failed to start the polling thread: {err}")))?
};
Ok(Self {
state,
thread: handle.thread().clone(),
})
}
fn wake(&self) {
self.thread.unpark();
}
fn change<T>(&self, func: impl FnOnce(&mut State) -> T) -> T {
let ret = func(&mut self.state.lock().unwrap());
self.wake();
ret
}
pub fn set_dev(&self, dev: &Arc<dyn AnyDev>, validity_periods: u32, polled: Vec<Polled>) {
self.change(|state| {
let mut polled_dev = PolledDev {
dev: Arc::downgrade(dev),
validity_periods,
attrs: HashMap::new(),
commands: HashMap::new(),
};
let now = Instant::now();
for polled in polled {
let entry = state.new_entry(polled.name, polled.period, now);
polled_dev.entries_mut(polled.kind).insert(ArcName::from(Arc::clone(&entry.name)), entry);
}
state.devs.insert(ArcName::from(Arc::clone(dev.name())), polled_dev);
});
}
pub fn remove_dev(&self, dev: &dyn AnyDev) {
self.change(|state| {
let name = Name::new(dev.name());
if state.devs.get(name).is_some_and(|polled_dev| polled_dev.is(dev)) {
state.devs.remove(name);
}
});
}
pub fn add(&self, dev: &Arc<dyn AnyDev>, kind: PolledKind, name: Arc<str>, period: Duration) -> Result<()> {
self.change(|state| {
let now = Instant::now();
let entry = state.new_entry(name, period, now);
let polled_dev = state.devs.entry(ArcName::from(Arc::clone(dev.name())))
.or_insert_with(|| PolledDev {
dev: Arc::downgrade(dev),
validity_periods: DEFAULT_VALIDITY_PERIODS,
attrs: HashMap::new(),
commands: HashMap::new(),
});
let entries = polled_dev.entries_mut(kind);
if entries.contains_key(Name::new(&entry.name)) {
return Err(Error::new(ErrorKind::Reason(reason::API_AlreadyPolled))
.with_description(format!("The {} {:?} of device {:?} is already polled",
kind.as_str(), entry.name, dev.name())));
}
entries.insert(ArcName::from(Arc::clone(&entry.name)), entry);
Ok(())
})
}
pub fn set_period(&self, dev: &str, kind: PolledKind, name: &str, period: Duration) -> Result<()> {
self.change(|state| {
let entry = state.entry_mut(dev, kind, name)?;
entry.period = period;
entry.due = Instant::now();
Ok(())
})
}
pub fn remove(&self, dev: &str, kind: PolledKind, name: &str) -> Result<()> {
self.change(|state| {
state.entry_mut(dev, kind, name)?;
if let Some(polled_dev) = state.devs.get_mut(Name::new(dev)) {
polled_dev.entries_mut(kind).remove(Name::new(name));
}
Ok(())
})
}
pub fn period(&self, dev: &str, kind: PolledKind, name: &str) -> Option<Duration> {
let state = self.state.lock().unwrap();
let polled_dev = state.devs.get(Name::new(dev))?;
polled_dev.entries(kind).get(Name::new(name)).map(|entry| entry.period)
}
pub fn polled_devs(&self) -> Vec<Arc<str>> {
let state = self.state.lock().unwrap();
let mut names = state.devs.iter()
.filter(|(_, polled_dev)| !polled_dev.attrs.is_empty() || !polled_dev.commands.is_empty())
.filter_map(|(_, polled_dev)| polled_dev.dev.upgrade().map(|dev| Arc::clone(dev.name())))
.collect::<Vec<_>>();
names.sort();
names
}
pub fn status(&self, dev: &str) -> Vec<PolledStatus> {
let state = self.state.lock().unwrap();
let Some(polled_dev) = state.devs.get(Name::new(dev)) else {
return Vec::new();
};
let mut status = [PolledKind::Command, PolledKind::Attr].into_iter()
.flat_map(|kind| polled_dev.entries(kind).values().map(move |entry| PolledStatus {
kind,
name: Arc::clone(&entry.name),
period: entry.period,
last_duration: entry.last_duration,
}))
.collect::<Vec<_>>();
status.sort_by(|a, b| b.kind.cmp(&a.kind).then_with(|| a.name.cmp(&b.name)));
status
}
pub fn resume(&self) {
self.change(|state| {
state.paused = false;
let now = Instant::now();
for polled_dev in state.devs.values_mut() {
for entry in polled_dev.attrs.values_mut().chain(polled_dev.commands.values_mut()) {
entry.due = now;
}
}
});
}
pub fn pause(&self) {
self.change(|state| state.paused = true);
}
pub fn shutdown(&self) {
self.change(|state| state.shutdown = true);
}
}
impl State {
fn new_entry(&mut self, name: Arc<str>, period: Duration, now: Instant) -> PolledEntry {
let id = self.next_id;
self.next_id += 1;
PolledEntry {
id,
name,
period,
due: now,
in_progress: false,
last_duration: None,
last_periodic: None,
}
}
fn entry_mut(&mut self, dev: &str, kind: PolledKind, name: &str) -> Result<&mut PolledEntry> {
self.devs.get_mut(Name::new(dev))
.and_then(|polled_dev| polled_dev.entries_mut(kind).get_mut(Name::new(name)))
.ok_or_else(|| Error::new(ErrorKind::Reason(reason::API_PollObjNotFound))
.with_description(format!("The {} {name:?} of device {dev:?} is not polled", kind.as_str())))
}
fn take_due(&mut self, now: Instant) -> (Vec<PollStart>, Option<Instant>) {
let mut starts = Vec::new();
let mut next = None::<Instant>;
if self.paused {
return (starts, next);
}
self.devs.retain(|_, polled_dev| {
let Some(dev) = polled_dev.dev.upgrade() else {
return false;
};
let validity_periods = polled_dev.validity_periods;
let mut take = |entry: &mut PolledEntry| -> Option<PollItem> {
if entry.in_progress {
return None;
}
if entry.due > now {
next = Some(next.map_or(entry.due, |next| next.min(entry.due)));
return None;
}
entry.in_progress = true;
Some(PollItem {
id: entry.id,
name: Arc::clone(&entry.name),
valid_for: entry.period.saturating_mul(validity_periods),
last_periodic: entry.last_periodic,
})
};
for kind in [PolledKind::Attr, PolledKind::Command] {
let items = polled_dev.entries_mut(kind)
.values_mut()
.filter_map(&mut take)
.collect::<Vec<_>>();
if !items.is_empty() {
starts.push(PollStart {
dev: Arc::clone(&dev),
kind,
items,
});
}
}
true
});
(starts, next)
}
fn finish(&mut self, end: PollEnd, now: Instant) {
let origin = self.origin;
let Some(polled_dev) = self.devs.get_mut(Name::new(&end.dev)) else { return };
for item in end.items {
let Some(entry) = polled_dev.entries_mut(end.kind).get_mut(Name::new(&item.name)) else { continue };
if entry.id != item.id {
continue;
}
entry.in_progress = false;
entry.last_duration = Some(end.last_duration);
entry.last_periodic = item.last_periodic;
entry.due = grid_after(origin, entry.period, now);
}
}
}
impl PolledDev {
fn is(&self, dev: &dyn AnyDev) -> bool {
std::ptr::addr_eq(self.dev.as_ptr(), dev as *const dyn AnyDev)
}
fn entries(&self, kind: PolledKind) -> &HashMap<ArcName, PolledEntry> {
match kind {
PolledKind::Attr => &self.attrs,
PolledKind::Command => &self.commands,
}
}
fn entries_mut(&mut self, kind: PolledKind) -> &mut HashMap<ArcName, PolledEntry> {
match kind {
PolledKind::Attr => &mut self.attrs,
PolledKind::Command => &mut self.commands,
}
}
}
fn grid_after(origin: Instant, period: Duration, after: Instant) -> Instant {
let period = period.as_nanos().max(1);
let ticks = after.saturating_duration_since(origin).as_nanos() / period + 1;
origin + Duration::from_nanos(u64::try_from(ticks * period).unwrap_or(u64::MAX))
}
fn run_thread(state: &Mutex<State>) {
struct ThreadWaker(Thread);
impl futures::task::ArcWake for ThreadWaker {
fn wake_by_ref(arc_self: &Arc<Self>) {
arc_self.0.unpark();
}
}
let waker = futures::task::waker(Arc::new(ThreadWaker(thread::current())));
let mut cx = Context::from_waker(&waker);
let mut polls = FuturesUnordered::<PollRun>::new();
debug!("Started the polling thread");
loop {
let (starts, next) = {
let mut state = state.lock().unwrap();
if state.shutdown {
break;
}
state.take_due(Instant::now())
};
for start in starts {
polls.push(PollRun::new(start));
}
let mut finished = false;
while let Poll::Ready(Some(end)) = polls.poll_next_unpin(&mut cx) {
state.lock().unwrap().finish(end, Instant::now());
finished = true;
}
if finished {
continue;
}
match next {
Some(next) => thread::park_timeout(next.saturating_duration_since(Instant::now())),
None => thread::park(),
}
}
debug!("Ended the polling thread");
}
struct PollRun {
fut: BoxFuture<'static, Result<Vec<Option<Instant>>>>,
end: Option<PollEnd>,
started: Instant,
}
impl PollRun {
fn new(start: PollStart) -> Self {
let fut = start.dev.poll(start.kind, &start.items);
Self {
fut,
end: Some(PollEnd {
dev: Arc::clone(start.dev.name()),
kind: start.kind,
items: start.items,
last_duration: Duration::ZERO,
}),
started: Instant::now(),
}
}
}
impl Future for PollRun {
type Output = PollEnd;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let this = self.get_mut();
let res = ready!(this.fut.as_mut().poll(cx));
let mut end = this.end.take().expect("a poll is through only once");
end.last_duration = this.started.elapsed();
match res {
Ok(last_periodics) => {
for (item, last_periodic) in end.items.iter_mut().zip(last_periodics) {
item.last_periodic = last_periodic;
}
}
Err(err) => debug!(dev = %end.dev, %err, "Failed to run the poll"),
}
Poll::Ready(end)
}
}
pub fn parse_periods<'a>(dev: &str, prop: &str, values: impl IntoIterator<Item = &'a str>) -> Vec<(&'a str, Duration)> {
let mut periods = Vec::new();
let mut values = values.into_iter();
while let Some(name) = values.next() {
let Some(period) = values.next() else {
warn!(%dev, %prop, %name, "Ignored a polled object with no period");
break;
};
match period.trim().parse::<u64>() {
Ok(0) => warn!(%dev, %prop, %name, "Ignored a polled object triggered by the device, which is not supported"),
Ok(ms) => periods.push((name, Duration::from_millis(ms).max(MIN_PERIOD))),
Err(_) => warn!(%dev, %prop, %name, %period, "Ignored a polled object with an invalid period"),
}
}
periods
}
pub fn parse_depths<'a>(dev: &str, prop: &str, values: impl IntoIterator<Item = &'a str>) -> Vec<(&'a str, usize)> {
let mut depths = Vec::new();
let mut values = values.into_iter();
while let (Some(name), Some(depth)) = (values.next(), values.next()) {
match depth.trim().parse::<usize>() {
Ok(depth) => depths.push((name, depth)),
Err(_) => warn!(%dev, %prop, %name, %depth, "Ignored an invalid polling ring depth"),
}
}
depths
}
pub fn format_status(status: &PolledStatus, record: Option<&PolledHistory>) -> String {
let (noun, action) = match status.kind {
PolledKind::Attr => ("attribute", "attribute reading"),
PolledKind::Command => ("command", "command execution"),
};
let mut text = String::new();
let _ = write!(text, "Polled {noun} name = {}", status.name);
let _ = write!(text, "\nPolling period (mS) = {}", status.period.as_millis());
let Some(record) = record else {
return text;
};
let _ = write!(text, "\nPolling ring buffer depth = {}", record.depth);
if let Some(last_duration) = status.last_duration {
let _ = write!(text, "\nTime needed for the last {action} (mS) = {:.3}", last_duration.as_secs_f64() * 1000.0);
}
if let Some(last) = record.times.last() {
let since = SystemTime::now().duration_since(*last).unwrap_or_default();
if since.as_secs() == 0 {
let _ = write!(text, "\nData not updated since {} mS", since.as_millis());
} else {
let _ = write!(text, "\nData not updated since {} S and {} mS", since.as_secs(), since.subsec_millis());
}
}
if record.times.len() > 1 {
let deltas = record.times.windows(2)
.map(|pair| pair[1].duration_since(pair[0]).unwrap_or_default().as_millis().to_string())
.collect::<Vec<_>>();
let _ = write!(text, "\nDelta between last records (in mS) = {}", deltas.join(", "));
}
if let Some(err) = &record.failure {
let what = match status.kind {
PolledKind::Attr => "attribute read",
PolledKind::Command => "command execution",
};
let _ = write!(text, "\nLast {what} FAILED :\n\tReason = {}\n\tDesc = {}\n\tOrigin = {}",
err.reason(), err.description(), err.origin());
}
text
}
#[cfg(test)]
mod tests {
use crate::dev::status_period;
use super::*;
#[test]
fn polled_prop() {
let values = ["a", "1000", "b", "0", "c", "x", "d", "1", "e"];
assert_eq!(parse_periods("x/y/z", "polled_attr", values), [
("a", Duration::from_millis(1000)),
("d", MIN_PERIOD),
]);
}
#[test]
fn grid() {
let origin = Instant::now();
let at = |ms| origin + Duration::from_millis(ms);
let (state, status) = (Duration::from_secs(2), Duration::from_secs(3));
assert_eq!(grid_after(origin, state, at(4000)), at(6000));
assert_eq!(grid_after(origin, status, at(3500)), at(6000));
assert_eq!(grid_after(origin, state, at(3999)), at(4000));
assert_eq!(grid_after(origin, status, at(10_100)), at(12_000));
assert_eq!(grid_after(origin, state, origin - Duration::from_secs(1)), at(2000));
}
#[test]
fn status_round_trip() {
let status = PolledStatus {
kind: PolledKind::Attr,
name: Arc::from("Position"),
period: Duration::from_millis(250),
last_duration: Some(Duration::from_micros(1500)),
};
let record = PolledHistory {
depth: 10,
times: vec![SystemTime::now() - Duration::from_millis(250), SystemTime::now()],
failure: None,
};
let texts = [format_status(&status, Some(&record))];
assert_eq!(status_period(&texts, "attribute", "position"), Some(Duration::from_millis(250)));
assert_eq!(status_period(&texts, "command", "position"), None);
assert_eq!(status_period(&texts, "attribute", "speed"), None);
}
}