tari_comms 6.1.0

// Copyright 2020, The Tari Project
//
// Redistribution and use in source and binary forms, with or without modification, are permitted provided that the
// following conditions are met:
//
// 1. Redistributions of source code must retain the above copyright notice, this list of conditions and the following
// disclaimer.
//
// 2. Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the
// following disclaimer in the documentation and/or other materials provided with the distribution.
//
// 3. Neither the name of the copyright holder nor the names of its contributors may be used to endorse or promote
// products derived from this software without specific prior written permission.
//
// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES,
// INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
// DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
// SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
// WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE
// USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

use std::{
    fmt::Display,
    time::{Duration, Instant},
};

use log::*;
use tokio::time;
use tower::{Service, ServiceExt};

use crate::{bounded_executor::BoundedExecutor, pipeline::builder::OutboundPipelineConfig};

const LOG_TARGET: &str = "comms::pipeline::outbound";

/// How long a single outbound message may occupy a pipeline slot before it is abandoned.
const PIPELINE_TIMEOUT: Duration = Duration::from_secs(10);

/// Calls a service in a new task whenever a message is received by the configured channel and forwards the resulting
/// message as a [MessageRequest](crate::protocol::messaging::MessageRequest).
pub struct Outbound<TPipeline, TItem> {
    /// Executor used to spawn a pipeline for each received item on the stream
    executor: BoundedExecutor,
    /// Outbound pipeline configuration containing the pipeline and it's in and out streams
    config: OutboundPipelineConfig<TItem, TPipeline>,
}

impl<TPipeline, TItem> Outbound<TPipeline, TItem>
where
    TItem: Send + 'static,
    TPipeline: Service<TItem, Response = ()> + Clone + Send + 'static,
    TPipeline::Error: Display + Send,
    TPipeline::Future: Send,
{
    /// New outbound pipeline.
    pub fn new(executor: BoundedExecutor, config: OutboundPipelineConfig<TItem, TPipeline>) -> Self {
        Self { executor, config }
    }

    /// Run the outbound pipeline.
    pub async fn run(mut self) {
        let mut current_id = 0u64;

        while let Some(msg) = self.config.in_receiver.recv().await {
            // Pipeline IN received a message. Spawn a new task for the pipeline
            let num_available = self.executor.num_available();
            let max_available = self.executor.max_available();
            log!(
                target: LOG_TARGET,
                if num_available < max_available {
                    Level::Debug
                } else {
                    Level::Trace
                },
                "Outbound pipeline usage: {}/{}",
                max_available.saturating_sub(num_available),
                max_available
            );

            let pipeline = self.config.pipeline.clone();
            let id = current_id;
            current_id = current_id.wrapping_add(1) % u64::MAX;
            // Shares the permit pool with `self.executor`, so the task can read the pipeline's live
            // occupancy at the moment it times out rather than at the moment it was spawned.
            let occupancy = self.executor.clone();
            self.executor
                .spawn(async move {
                    let timer = Instant::now();
                    trace!(target: LOG_TARGET, "Start outbound pipeline {id}");
                    match time::timeout(PIPELINE_TIMEOUT, pipeline.oneshot(msg)).await {
                        Ok(Ok(_)) => {},
                        Ok(Err(err)) => {
                            error!(
                                target: LOG_TARGET,
                                "Outbound pipeline {id} returned an error: '{err}'"
                            );
                        },
                        Err(_) => {
                            // Distinguish the two very different faults that both surface here. A
                            // saturated pipeline is self-inflicted back-pressure; a near-empty one
                            // means the message is stuck on something downstream (typically an actor
                            // that has stopped being scheduled), and reporting that as a "deadlock in
                            // this pipeline" sends the reader four layers away from the actual fault.
                            let max_available = occupancy.max_available();
                            let in_use = max_available.saturating_sub(occupancy.num_available());
                            if in_use.saturating_mul(2) >= max_available {
                                warn!(
                                    target: LOG_TARGET,
                                    "Outbound pipeline {id} timed out after {PIPELINE_TIMEOUT:.0?} and was aborted. \
                                     The outbound pipeline is saturated ({in_use}/{max_available} slots in use): it \
                                     is not draining as fast as messages arrive."
                                );
                            } else {
                                warn!(
                                    target: LOG_TARGET,
                                    "Outbound pipeline {id} timed out after {PIPELINE_TIMEOUT:.0?} and was aborted, \
                                     but the pipeline itself is nearly empty ({in_use}/{max_available} slots in use). \
                                     This message is blocked on an unresponsive downstream service, not on pipeline \
                                     capacity — look at the comms actors (connectivity/connection manager), not here."
                                );
                            }
                        },
                    }

                    trace!(
                        target: LOG_TARGET,
                        "Finished outbound pipeline {} in {:.2?}",
                        id,
                        timer.elapsed()
                    );
                })
                .await;
        }

        info!(
            target: LOG_TARGET,
            "Outbound pipeline is shutting down because the in channel closed"
        );
    }
}

#[cfg(test)]
mod test {
    use bytes::Bytes;
    use tari_test_utils::collect_recv;
    use tokio::sync::mpsc;

    use super::*;
    use crate::{message::OutboundMessage, pipeline::SinkService, utils};

    #[tokio::test]
    async fn run() {
        const NUM_ITEMS: usize = 10;
        let (tx, mut in_receiver) = mpsc::unbounded_channel();
        utils::mpsc::send_all_unbounded(
            &tx,
            (0..NUM_ITEMS).map(|i| OutboundMessage::new(Default::default(), Bytes::copy_from_slice(&i.to_be_bytes()))),
        )
        .unwrap();
        in_receiver.close();

        let (out_tx, mut out_rx) = mpsc::unbounded_channel();
        let executor = BoundedExecutor::new(100);

        let pipeline = Outbound::new(executor, OutboundPipelineConfig {
            in_receiver,
            out_receiver: None,
            pipeline: SinkService::new(out_tx),
        });

        let spawned_task = tokio::spawn(pipeline.run());

        let requests = collect_recv!(out_rx, timeout = Duration::from_millis(5));
        assert_eq!(requests.len(), NUM_ITEMS);

        // Check that this task ends because the stream has closed
        time::timeout(Duration::from_secs(5), spawned_task)
            .await
            .unwrap()
            .expect("Task should end")
    }
}