settings/job/

source.rs

// Copyright 2021 The Fuchsia Authors. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.

//! Job Source Support
//!
//! # Summary
//!
//! The source mod contains components for providing [Jobs](Job) to Job manager (most likely
//! [manager](job::manager::Manager)). In a typical workflow, client code will create a [Seeder],
//! which is used to send [Job](Job) streams to a manager via the
//! [MessageHub](crate::message::message_hub::MessageHub). The [Seeder] can send any stream where
//! the data implements [Into<Job>]. Once the source is received, the manager can assign a unique
//! [Id] to the source, using [IdGenerator]. The manager can then use a [Handler] to handle the
//!  produced [Jobs](Job) and their results.

use crate::clock::now;
use crate::job::execution::GroupError;
use crate::job::{self, execution, Job, Payload, StoreHandleMapping};
use crate::message::base::{Audience, MessengerType};
use crate::service::message::{Delegate, Messenger, Signature};
use crate::trace_guard;
use core::pin::Pin;
use futures::lock::Mutex;
use futures::{Stream, StreamExt};
use std::collections::{HashMap, VecDeque};
use std::convert::Infallible;
use std::rc::Rc;
use thiserror::Error as ThisError;
use {fuchsia_async as fasync, fuchsia_trace as ftrace};

#[derive(Clone)]
/// [Seeder] properly packages and sends [Job] source streams to a [Job] manager.
pub struct Seeder {
    /// A [Messenger](crate::message::messenger::MessengerClient) to send Payloads to the manager.
    messenger: Messenger,
    /// The [Signature](crate::message::base::Signature) of the manager to receive the source
    /// Payloads.
    manager_signature: Signature,
}

impl Seeder {
    pub(crate) async fn new(delegate: &Delegate, manager_signature: Signature) -> Self {
        Self {
            messenger: delegate
                .create(MessengerType::Unbound)
                .await
                .expect("should create messenger")
                .0,
            manager_signature,
        }
    }

    pub(crate) fn seed<J, E, E2, T>(&self, source: T)
    where
        Job: TryFrom<J, Error = E2>,
        Error: From<E> + From<E2>,
        T: Stream<Item = Result<J, E>> + 'static,
    {
        // Convert the incoming stream into the expected types for a Job source.
        let mapped_stream: Pin<Box<dyn Stream<Item = Result<Job, Error>>>> = source
            .map(|result| {
                result
                    // First convert the error type from the result so we can be compatible
                    // with conversions done with try_from below.
                    .map_err(Error::from)
                    // Then map the job. Ideally try_from will return `Error` directly, but we
                    // also need to handle the `Infallible` type. It should compile to a no-op,
                    // but the types still need to align.
                    .and_then(|j| Job::try_from(j).map_err(Error::from))
            })
            .boxed_local();

        // Send the source stream to the manager.
        let _ = self.messenger.message(
            Payload::Source(Rc::new(Mutex::new(Some(mapped_stream)))).into(),
            Audience::Messenger(self.manager_signature),
        );
    }
}

/// The types of errors for [Jobs](Job). This is a single, unified set over all Job source
/// related-errors. This enumeration should be expanded to capture any future error variant.
#[derive(ThisError)]
pub enum Error {
    #[error("Unexpected error")]
    Unexpected(fidl::Error),
    #[error("Invalid input")]
    InvalidInput(Box<dyn ErrorResponder>),
    #[error("Invalid policy input")]
    InvalidPolicyInput(Box<dyn PolicyErrorResponder>),
    #[error("Unsupported API call")]
    Unsupported,
}

impl std::fmt::Debug for Error {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Error::Unexpected(_) => f.write_str("Unexpected"),
            Error::InvalidInput(_) => f.write_str("InvalidInput(..)"),
            Error::InvalidPolicyInput(_) => f.write_str("InvalidPolicyInput(..)"),
            Error::Unsupported => f.write_str("Unsupported"),
        }
    }
}

/// Abstract over how to respond with a settings fidl error.
pub trait ErrorResponder {
    /// Unique identifier for the API this responder is responsible for.
    fn id(&self) -> &'static str;

    /// Respond with the supplied error. Returns any fidl errors that occur when
    /// trying to send the response.
    fn respond(self: Box<Self>, error: fidl_fuchsia_settings::Error) -> Result<(), fidl::Error>;
}

/// Abstract over how to respond with a settings policy fidl error.
pub trait PolicyErrorResponder {
    /// Unique identifier for the API this responder is responsible for.
    fn id(&self) -> &'static str;

    /// Respond with the supplied error. Returns any fidl errors that occur when
    /// trying to send the response.
    fn respond(
        self: Box<Self>,
        error: fidl_fuchsia_settings_policy::Error,
    ) -> Result<(), fidl::Error>;
}

// This implementation is necessary when converting into a Job is infallible. This can happen if an
// input to a job has no possible way to fail, or in tests when the streams a populated with Jobs
// directly. This is used by the Seeder::seed fn above.
impl From<Infallible> for Error {
    fn from(_: Infallible) -> Self {
        unreachable!()
    }
}

impl From<fidl::Error> for Error {
    fn from(error: fidl::Error) -> Self {
        Error::Unexpected(error)
    }
}

#[derive(Copy, Clone, Debug, PartialEq)]
/// The current state of the source. This is used by the managing entity to understand how to handle
/// pending and completed [Jobs](Job) from a source.
pub(super) enum State {
    /// The source is still available to produce new [Jobs](Job).
    Active,
    /// Completion has been requested, but [Jobs](Job) must complete before the source is considered
    /// done.
    PendingCompletion,
    /// The source is no longer producing new [Jobs](Job).
    Completed,
}

/// [Id] provides a unique identifier for a source within its parent space, most often a manager.
#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash)]
pub struct Id {
    _identifier: usize,
}

impl Id {
    fn new(identifier: usize) -> Self {
        Self { _identifier: identifier }
    }
}

/// [IdGenerator] creates unique [Ids] to be associated with a source. This uniqueness is
/// guaranteed for [Ids] generated by the same [IdGenerator].
///
/// [Ids]: Id
pub(super) struct IdGenerator {
    next_identifier: usize,
}

impl IdGenerator {
    pub(super) fn new() -> Self {
        Self { next_identifier: 0 }
    }

    pub(super) fn generate(&mut self) -> Id {
        let return_id = Id::new(self.next_identifier);
        self.next_identifier += 1;

        return_id
    }
}

/// [Handler] handles [Jobs](Job) within the scope of a single scope. It determines what [Job](Job)
/// should be executed (if any). This responsibility includes managing any queueing that might be
/// necessary based on the [Job] type.
pub(super) struct Handler {
    /// A [IdGenerator](job::IdGenerator) to assign unique ids to incoming jobs.
    job_id_generator: job::IdGenerator,
    /// A mapping from [execution types](execution::Type) to [groups](execution::Group). Each entry
    /// enables tracking across [Jobs](Job) in the same group, such as storing persistent values.
    /// The mapping is also consulted finding the next [Jobs](Job) to execute.
    jobs: HashMap<execution::Type, execution::Group>,
    /// A list of states. The element represents the most current [State]. We keep track of seen
    /// states to allow post analysis, such as source duration.
    states: VecDeque<(State, zx::MonotonicInstant)>,
    /// This [HashMap] associates a given [Job] [Signature] with a [Data](job::data::Data) mapping.
    /// [Signature] is used over [execution::Type] to allow storage to be shared across groups of
    /// different [types](execution::Type) that share the same [Signature].
    stores: StoreHandleMapping,
}

impl Handler {
    pub(crate) fn new() -> Self {
        let mut handler = Self {
            job_id_generator: job::IdGenerator::new(),
            jobs: HashMap::new(),
            states: VecDeque::new(),
            stores: HashMap::new(),
        };

        handler.set_state(State::Active);

        handler
    }

    /// Marks the source as completed.
    pub(crate) fn complete(&mut self) {
        self.set_state(if self.is_active() { State::PendingCompletion } else { State::Completed });
    }

    /// Drops any job that has not yet been started and any watch jobs.
    pub(crate) fn cancel(&mut self) {
        for execution_group in self.jobs.values_mut() {
            execution_group.cancel();
        }
        self.complete();
    }

    /// Returns whether the source has completed.
    pub(crate) fn is_completed(&mut self) -> bool {
        matches!(self.states.back(), Some(&(State::Completed, _)))
    }

    fn set_state(&mut self, state: State) {
        // State should not be set after the source has been completed.
        assert!(!self.is_completed());

        // Do not try to set the state if it matches the last updated state.
        if matches!(self.states.back(), Some(&(x,_)) if x == state) {
            return;
        }

        self.states.push_back((state, now()));
    }

    /// Returns true if any job is executed, false otherwise.
    pub(crate) async fn execute_next<F: FnOnce(job::Info) + 'static>(
        &mut self,
        delegate: &mut Delegate,
        callback: F,
        id: ftrace::Id,
    ) -> bool {
        for execution_group in self.jobs.values_mut() {
            // If there are no jobs ready to become active, move to next group.
            if let Some(job_info) = execution_group.promote_next_to_active() {
                let guard = trace_guard!(id, c"prepare_execution");
                let execution =
                    job_info.prepare_execution(delegate, &mut self.stores, callback).await;
                drop(guard);

                fasync::Task::local(execution).detach();
                return true;
            }
        }

        false
    }

    /// Returns whether the source is active, defined as having at least one [Job] which is
    /// currently active (running, not pending).
    pub(crate) fn is_active(&self) -> bool {
        self.jobs.iter().any(|(_, group)| group.is_active())
    }

    /// Adds a [Job] to be handled by this [Handler].
    pub(crate) fn add_pending_job(&mut self, incoming_job: Job) -> Result<(), GroupError> {
        let job_info = job::Info::new(self.job_id_generator.generate(), incoming_job);
        let execution_type = *job_info.get_execution_type();

        // Execution groups are based on matching execution::Type.
        let execution_group = self
            .jobs
            .entry(execution_type)
            .or_insert_with(move || execution::Group::new(execution_type));
        execution_group.add(job_info)
    }

    /// Informs the [Handler] that a [Job] by the given [Id](job::Id) has completed.
    pub(crate) fn handle_job_completion(&mut self, job: job::Info) {
        self.jobs.get_mut(job.get_execution_type()).expect("group should be present").complete(job);

        // When a source end is detected, the managing entity will try to complete the source. If
        // there is active work, the source completion will be deferred. It is the source's
        // responsibility after each subsequent completion to check whether completion can now
        // proceed.
        if matches!(self.states.back(), Some(&(State::PendingCompletion, _))) {
            self.complete();
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::service::{test, MessageHub};
    use crate::tests::scaffold::workload::{Sequential, StubWorkload, Workload};
    use rand::Rng;

    use assert_matches::assert_matches;
    use futures::FutureExt;

    #[fuchsia::test]
    fn test_id_generation() {
        let mut generator = IdGenerator::new();
        // Ensure generator is creating unique ids
        assert!(generator.generate() != generator.generate());
    }

    #[fuchsia::test(allow_stalls = false)]
    async fn test_seeding() {
        // Create delegate for communication between components.
        let message_hub_delegate = MessageHub::create_hub();

        // Create a top-level receptor to receive sources.
        let mut receptor = message_hub_delegate
            .create(MessengerType::Unbound)
            .await
            .expect("should create receptor")
            .1;

        // Create seeder.
        let seeder = Seeder::new(&message_hub_delegate, receptor.get_signature()).await;

        let job_stream = async {
            Ok(Job::new(job::work::Load::Independent(StubWorkload::new()))) as Result<Job, Error>
        }
        .into_stream();

        seeder.seed(job_stream);

        assert_matches!(receptor.next_of::<Payload>().await, Ok((Payload::Source(_), _)));
    }

    #[fuchsia::test(allow_stalls = false)]
    async fn test_handling() {
        // Create delegate for communication between components.
        let mut message_hub_delegate = MessageHub::create_hub();

        let results: Vec<i64> = (0..10).collect();

        // Create a top-level receptor to receive job results from.
        let mut receptor = message_hub_delegate
            .create(MessengerType::Unbound)
            .await
            .expect("should create receptor")
            .1;

        let mut handler = Handler::new();

        assert!(!handler.execute_next(&mut message_hub_delegate, |_| {}, 0.into()).await);

        for result in &results {
            let _ = handler.add_pending_job(Job::new(job::work::Load::Independent(Workload::new(
                test::Payload::Integer(*result),
                receptor.get_signature(),
            ))));
        }

        for result in results {
            let (execution_tx, mut execution_rx) = futures::channel::mpsc::unbounded::<job::Info>();

            // Execute job.
            assert!(
                handler
                    .execute_next(
                        &mut message_hub_delegate,
                        move |job| {
                            execution_tx.unbounded_send(job).expect("send should succeed");
                        },
                        0.into()
                    )
                    .await
            );

            // Confirm received value matches the value sent from workload.
            if let test::Payload::Integer(value) =
                receptor.next_of::<test::Payload>().await.expect("should have payload").0
            {
                assert_eq!(value, result);
            }

            handler
                .handle_job_completion(execution_rx.next().await.expect("should have gotten job"));
        }
    }

    #[fuchsia::test(allow_stalls = false)]
    async fn test_drop_pending() {
        // Create delegate for communication between components.
        let mut message_hub_delegate = MessageHub::create_hub();

        let mut results: Vec<i64> = (0..10).collect();

        // Create a top-level receptor to receive job results from.
        let mut receptor = message_hub_delegate
            .create(MessengerType::Unbound)
            .await
            .expect("should create receptor")
            .1;

        let mut handler = Handler::new();

        assert!(!handler.execute_next(&mut message_hub_delegate, |_| {}, 0.into()).await);

        for result in &results {
            let _ = handler.add_pending_job(Job::new(job::work::Load::Independent(Workload::new(
                test::Payload::Integer(*result),
                receptor.get_signature(),
            ))));
        }

        let result = results.remove(0);
        let (execution_tx, mut execution_rx) = futures::channel::mpsc::unbounded::<job::Info>();

        // Execute job concurrently.
        assert!(
            handler
                .execute_next(
                    &mut message_hub_delegate,
                    move |job| {
                        execution_tx.unbounded_send(job).expect("send should succeed");
                    },
                    0.into(),
                )
                .await
        );

        handler.cancel();

        // Confirm received value matches the value sent from workload.
        if let test::Payload::Integer(value) =
            receptor.next_of::<test::Payload>().await.expect("should have payload").0
        {
            assert_eq!(value, result);
        }

        handler.handle_job_completion(execution_rx.next().await.expect("should have gotten job"));

        // Validate there are no more jobs to execute.
        let (execution_tx, _execution_rx) = futures::channel::mpsc::unbounded::<job::Info>();
        assert!(
            !handler
                .execute_next(
                    &mut message_hub_delegate,
                    move |job| {
                        execution_tx.unbounded_send(job).expect("send should succeed");
                    },
                    0.into(),
                )
                .await
        );
    }

    // Ensures that proper queueing happens amongst Jobs within Execution Groups.
    #[fuchsia::test(allow_stalls = false)]
    async fn test_execution_order() {
        let (execution_tx, mut execution_rx) = futures::channel::mpsc::unbounded::<job::Info>();

        // Create delegate for communication between components.
        let mut message_hub_delegate = MessageHub::create_hub();

        let mut handler = Handler::new();

        // Create a top-level receptor to receive job results from.
        let mut receptor = message_hub_delegate
            .create(MessengerType::Unbound)
            .await
            .expect("should create receptor")
            .1;

        // Create 2 jobs of the same sequential type.
        let results: Vec<i64> = (0..=1).collect();

        assert!(!handler.execute_next(&mut message_hub_delegate, |_| {}, 0.into()).await);

        for result in &results {
            let _ = handler.add_pending_job(Job::new(job::work::Load::Sequential(
                Workload::new(test::Payload::Integer(*result), receptor.get_signature()),
                job::Signature::new::<usize>(),
            )));
        }

        // Execute first job, ensuring handler has a job to execute.
        {
            let execution_tx = execution_tx.clone();
            assert!(
                handler
                    .execute_next(
                        &mut message_hub_delegate,
                        move |job| {
                            execution_tx.unbounded_send(job).expect("send should succeed");
                        },
                        0.into()
                    )
                    .await
            );
        }

        // Verify we receive result back for the first job.
        assert_eq!(
            test::Payload::Integer(0),
            receptor.next_of::<test::Payload>().await.expect("should have payload").0
        );

        // Capture first completed job, do not handle yet.
        let first_job_info = execution_rx.next().await.expect("should have gotten job");

        // Ensure no job is ready to execute.
        assert!(!handler.execute_next(&mut message_hub_delegate, move |_| {}, 0.into()).await);

        // Add an independent job.
        let _ =
            handler.add_pending_job(Job::new(job::work::Load::Independent(StubWorkload::new())));

        // Execute independent job.
        {
            let execution_tx = execution_tx.clone();
            // Execute next job and ensure that the response max
            assert!(
                handler
                    .execute_next(
                        &mut message_hub_delegate,
                        move |job| {
                            execution_tx.unbounded_send(job).expect("send should succeed");
                        },
                        0.into()
                    )
                    .await
            );
        }

        let independent_job_info = execution_rx.next().await.expect("should have gotten job");
        assert_matches!(*independent_job_info.get_execution_type(), execution::Type::Independent);

        // Handle independent job completion.
        handler.handle_job_completion(independent_job_info);

        // Handle first job completion.
        handler.handle_job_completion(first_job_info);

        {
            let execution_tx = execution_tx.clone();
            // Execute next job. Assert job is ready to execute
            assert!(
                handler
                    .execute_next(
                        &mut message_hub_delegate,
                        move |job| {
                            execution_tx.unbounded_send(job).expect("send should succeed");
                        },
                        0.into()
                    )
                    .await
            );
        }

        // Verify we receive result from the second job back.
        assert_eq!(
            test::Payload::Integer(1),
            receptor.next_of::<test::Payload>().await.expect("should have payload").0
        );
    }

    // Ensures that proper queueing happens amongst Jobs within Execution Groups.
    #[fuchsia::test(allow_stalls = false)]
    async fn test_data() {
        let mut rng = rand::thread_rng();

        let (result_tx, mut result_rx) = futures::channel::mpsc::unbounded::<usize>();

        // Create delegate for communication between components.
        let mut message_hub_delegate = MessageHub::create_hub();

        let mut handler = Handler::new();

        let data_key = job::data::Key::TestInteger(rng.gen());
        let initial_value = rng.gen_range(0..9);
        let signature = job::Signature::new::<usize>();

        // Each result is the square of the previous result,
        let results: Vec<usize> = (0..5)
            .map(move |val| {
                let mut return_value: usize = initial_value;

                for _ in 0..val {
                    return_value = return_value.pow(2);
                }

                return_value
            })
            .collect();

        for _ in &results {
            let data_key = data_key.clone();
            let result_tx = result_tx.clone();

            // Add a job that writes the initial value and reads it back.
            let _ = handler.add_pending_job(Job::new(job::work::Load::Sequential(
                Sequential::boxed(move |_, store| {
                    let result_tx = result_tx.clone();
                    let data_key = data_key.clone();

                    Box::pin(async move {
                        let mut storage_lock = store.lock().await;
                        let new_value = if let Some(job::data::Data::TestData(value)) =
                            storage_lock.get(&data_key)
                        {
                            value.pow(2)
                        } else {
                            initial_value
                        };

                        // Store value.
                        let _ = storage_lock.insert(data_key, job::data::Data::TestData(new_value));

                        // Relay value back.
                        result_tx.unbounded_send(new_value).expect("should send");
                    })
                }),
                signature,
            )));
        }

        for value in results {
            let (completion_tx, mut completion_rx) =
                futures::channel::mpsc::unbounded::<job::Info>();

            // Execute next job.
            assert!(
                handler
                    .execute_next(
                        &mut message_hub_delegate,
                        move |job| {
                            completion_tx.unbounded_send(job).expect("should send job");
                        },
                        0.into()
                    )
                    .await
            );

            // Ensure the returned value matches the calculation
            assert_eq!(value, result_rx.next().await.expect("value should be returned"));
            handler.handle_job_completion(completion_rx.next().await.expect("should receive job"));
        }
    }
}