persistence/
scheduler.rs

1// Copyright 2023 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
5use crate::fetcher::{FetchCommand, Fetcher};
6use fuchsia_async::{self as fasync, TaskGroup};
7use fuchsia_sync::Mutex;
8
9use persistence_config::{Config, ServiceName, Tag, TagConfig};
10use std::collections::{BTreeMap, HashMap};
11use std::sync::Arc;
12
13// This contains the logic to decide which tags to fetch at what times. It contains the state of
14// each tag (when last fetched, whether currently queued). When a request arrives via FIDL, it's
15// sent here and results in requests queued to the Fetcher.
16
17#[derive(Clone)]
18pub(crate) struct Scheduler {
19    // This is a global lock. Scheduler only does schedule() which is synchronous and quick.
20    state: Arc<Mutex<State>>,
21}
22
23struct State {
24    fetcher: Fetcher,
25    services: HashMap<ServiceName, HashMap<Tag, TagState>>,
26    tasks: TaskGroup,
27}
28
29struct TagState {
30    backoff: zx::MonotonicDuration,
31    state: FetchState,
32    last_fetched: zx::MonotonicInstant,
33}
34
35impl Scheduler {
36    pub(crate) fn new(fetcher: Fetcher, config: &Config) -> Self {
37        let mut services = HashMap::new();
38        for (service, tags) in config {
39            let mut tag_states = HashMap::new();
40            for (tag, tag_config) in tags {
41                let TagConfig { min_seconds_between_fetch, .. } = tag_config;
42                let backoff = zx::MonotonicDuration::from_seconds(*min_seconds_between_fetch);
43                let tag_state = TagState {
44                    backoff,
45                    state: FetchState::Idle,
46                    last_fetched: zx::MonotonicInstant::INFINITE_PAST,
47                };
48                tag_states.insert(tag.clone(), tag_state);
49            }
50            services.insert(service.clone(), tag_states);
51        }
52        let state = State { fetcher, services, tasks: TaskGroup::new() };
53        Scheduler { state: Arc::new(Mutex::new(state)) }
54    }
55
56    /// Gets a service name and a list of valid tags, and queues any fetches that are not already
57    /// pending. Updates the last-fetched time on any tag it queues, setting it equal to the later
58    /// of the current time and the time the fetch becomes possible.
59    pub(crate) fn schedule(&self, service: &ServiceName, tags: Vec<Tag>) {
60        // Every tag we process should use the same Now
61        let now = zx::MonotonicInstant::get();
62        let mut state = self.state.lock();
63        let Some(service_info) = state.services.get_mut(service) else {
64            return;
65        };
66
67        // Filter tags that need to be fetch now from those that need to be
68        // fetched later. Group later tags by their next_fetch time using a
69        // b-tree, making it efficient to iterate over these batches in
70        // order of next_fetch time.
71        let mut now_tags = vec![];
72        let mut later_tags: BTreeMap<zx::MonotonicInstant, Vec<Tag>> = BTreeMap::new();
73        for tag in tags {
74            let Some(tag_state) = service_info.get_mut(&tag) else {
75                return;
76            };
77            if matches!(tag_state.state, FetchState::Pending) {
78                continue;
79            }
80            if tag_state.last_fetched + tag_state.backoff <= now {
81                now_tags.push(tag);
82                tag_state.last_fetched = now;
83            } else {
84                let next_fetch = tag_state.last_fetched + tag_state.backoff;
85                tag_state.last_fetched = next_fetch;
86                tag_state.state = FetchState::Pending;
87                later_tags.entry(next_fetch).or_default().push(tag);
88            }
89        }
90        if !now_tags.is_empty() {
91            state.fetcher.send(FetchCommand { service: service.clone(), tags: now_tags });
92        }
93
94        // Schedule tags that need to be fetch later.
95        while let Some((next_fetch, tags)) = later_tags.pop_first() {
96            self.enqueue(&mut state, next_fetch, FetchCommand { service: service.clone(), tags });
97        }
98    }
99
100    fn enqueue(&self, state: &mut State, time: zx::MonotonicInstant, command: FetchCommand) {
101        let this = self.clone();
102        let mut fetcher = state.fetcher.clone();
103        state.tasks.spawn(async move {
104            fasync::Timer::new(time).await;
105            {
106                let mut state = this.state.lock();
107                let Some(tag_states) = state.services.get_mut(&command.service) else {
108                    return;
109                };
110                for tag in command.tags.iter() {
111                    tag_states.get_mut(tag).unwrap().state = FetchState::Idle;
112                }
113            }
114            fetcher.send(command);
115        });
116    }
117}
118
119/// FetchState tells whether a tag is currently waiting to be dispatched or not. If it is, then
120/// another request to fetch that tag should cause no change. If it's not waiting, then it can
121/// either be fetched immediately (in which case its state stays Idle, but the last-fetched time
122/// will be updated to Now) or it will be queued (in which case its state is Pending and its
123/// last-fetched time will be set forward to the time it's going to be fetched).
124enum FetchState {
125    Pending,
126    Idle,
127}
persistence/scheduler.rs

persistence/
scheduler.rs
