text:
use std::collections::{HashMap, HashSet, VecDeque};

use crate::lexer::{Item, ItemType, Lexer};
use crate::node::*;
use crate::utils::*;

pub struct Parser {
    name: String,
    pub funcs: HashSet<String>,
    lex: Option<Lexer>,
    line: usize,
    token: VecDeque<Item>,
    peek_count: usize,
    pub tree_set: HashMap<String, Tree>,
    tree_id: TreeId,
    tree: Option<Tree>,
    tree_stack: VecDeque<Tree>,
    max_tree_id: TreeId,
}

pub struct Tree {
    name: String,
    id: TreeId,
    pub root: Option<Nodes>,
    vars: Vec<String>,
}

impl Parser {
    pub fn new(name: String) -> Parser {
        Parser {
            name,
            funcs: HashSet::new(),
            lex: None,
            line: 0,
            token: VecDeque::new(),
            peek_count: 0,
            tree_set: HashMap::new(),
            tree_id: 0,
            tree: None,
            tree_stack: VecDeque::new(),
            max_tree_id: 0,
        }
    }
}

impl Tree {
    fn new(name: String, id: TreeId) -> Tree {
        Tree {
            name,
            id,
            root: None,
            vars: vec![],
        }
    }

    pub fn pop_vars(&mut self, n: usize) {
        self.vars.truncate(n);
    }
}

pub fn parse(
    name: String,
    text: String,
    funcs: HashSet<String>,
) -> Result<HashMap<String, Tree>, String> {
    let mut p = Parser::new(name);
    p.funcs = funcs;
    p.lex = Some(Lexer::new(text));
    p.parse_tree()?;
    Ok(p.tree_set)
}

impl Parser {
    fn next_from_lex(&mut self) -> Option<Item> {
        match self.lex {
            Some(ref mut l) => l.next(),
            None => None,
        }
    }

    fn backup(&mut self, t: Item) {
        self.token.push_front(t);
        self.peek_count += 1;
    }

    fn backup2(&mut self, t0: Item, t1: Item) {
        self.token.push_front(t1);
        self.token.push_front(t0);
        self.peek_count += 2;
    }

    fn backup3(&mut self, t0: Item, t1: Item, t2: Item) {
        self.token.push_front(t2);
        self.token.push_front(t1);
        self.token.push_front(t0);
        self.peek_count += 3;
    }

    fn next_must(&mut self, context: &str) -> Result<Item, String> {
        self.next()
            .ok_or_else(|| self.error_msg(&format!("unexpected end in {}", context)))
    }

    fn next_non_space(&mut self) -> Option<Item> {
        self.skip_while(|c| c.typ == ItemType::ItemSpace).next()
    }

    fn next_non_space_must(&mut self, context: &str) -> Result<Item, String> {
        self.next_non_space()
            .ok_or_else(|| self.error_msg(&format!("unexpected end in {}", context)))
    }

    fn peek_non_space_must(&mut self, context: &str) -> Result<&Item, String> {
        if let Some(t) = self.next_non_space() {
            self.backup(t);
            return Ok(self.token.front().unwrap());
        }
        self.error(&format!("unexpected end in {}", context))
    }

    fn peek(&mut self) -> Option<&Item> {
        if let Some(t) = self.next() {
            self.backup(t);
            return self.token.front();
        }
        None
    }

    fn peek_must(&mut self, context: &str) -> Result<&Item, String> {
        if let Some(t) = self.next_non_space() {
            self.backup(t);
            return Ok(self.token.front().unwrap());
        }
        self.error(&format!("unexpected end in {}", context))
    }

    fn start_parse(&mut self, name: String, id: TreeId) {
        if let Some(t) = self.tree.take() {
            self.tree_stack.push_back(t);
        }
        self.tree_id = id;
        let t = Tree::new(name, id);
        self.tree = Some(t);
    }

    fn stop_parse(&mut self) -> Result<(), String> {
        self.add_to_tree_set()?;
        self.tree = self.tree_stack.pop_back();
        self.tree_id = self.tree.as_ref().map(|t| t.id).unwrap_or(0);
        Ok(())
    }

    // top level parser
    fn parse_tree(&mut self) -> Result<(), String> {
        let name = self.name.clone();
        self.start_parse(name, 1);
        self.parse()?;
        self.stop_parse()?;
        Ok(())
    }

    fn add_tree(&mut self, name: String, t: Tree) {
        self.tree_set.insert(name, t);
    }

    fn error<T>(&self, msg: &str) -> Result<T, String> {
        Err(self.error_msg(msg))
    }

    fn error_msg(&self, msg: &str) -> String {
        let name = if let Some(t) = self.tree.as_ref() {
            &t.name
        } else {
            &self.name
        };
        format!("template: {}:{}:{}", name, self.line, msg)
    }

    fn expect(&mut self, expected: &ItemType, context: &str) -> Result<Item, String> {
        let token = self.next_non_space_must(context)?;
        if token.typ != *expected {
            return self.unexpected(&token, context);
        }
        Ok(token)
    }

    fn unexpected<T>(&self, token: &Item, context: &str) -> Result<T, String> {
        self.error(&format!("unexpected {} in {}", token, context))
    }

    fn add_var(&mut self, name: String) -> Result<(), String> {
        let mut tree = self.tree.take().ok_or_else(|| self.error_msg("no tree"))?;
        tree.vars.push(name);
        self.tree = Some(tree);
        Ok(())
    }

    fn add_to_tree_set(&mut self) -> Result<(), String> {
        let tree = self.tree.take().ok_or_else(|| self.error_msg("no tree"))?;
        if let Some(t) = self.tree_set.get(tree.name.as_str()) {
            if let Some(ref r) = t.root {
                match r.is_empty_tree() {
                    Err(e) => return Err(e),
                    Ok(false) => {
                        let err =
                            format!("template multiple definitions of template {}", &tree.name);
                        return self.error(&err);
                    }
                    Ok(true) => {}
                }
            }
        }
        self.add_tree(tree.name.clone(), tree);
        Ok(())
    }

    fn has_func(&self, name: &str) -> bool {
        self.funcs.contains(name)
    }

    fn parse(&mut self) -> Result<(), String> {
        if self.tree.is_none() {
            return self.error("no tree");
        }
        let id = self.tree_id;
        let mut t = match self.next() {
            None => return self.error(&format!("unable to peek for tree {}", id)),
            Some(t) => t,
        };
        if let Some(tree) = self.tree.as_mut() {
            tree.root = Some(Nodes::List(ListNode::new(id, t.pos)));
        }
        while t.typ != ItemType::ItemEOF {
            if t.typ == ItemType::ItemLeftDelim {
                let nns = self.next_non_space();
                match nns {
                    Some(ref item) if item.typ == ItemType::ItemDefine => {
                        self.parse_definition()?;
                        t = match self.next() {
                            None => return self.error(&format!("unable to peek for tree {}", id)),
                            Some(t) => t,
                        };
                        continue;
                    }
                    _ => {}
                };
                if let Some(t2) = nns {
                    self.backup2(t, t2);
                } else {
                    self.backup(t);
                }
            } else {
                self.backup(t);
            }
            let node = match self.text_or_action() {
                Ok(Nodes::Else(node)) => return self.error(&format!("unexpected {}", node)),
                Ok(Nodes::End(node)) => return self.error(&format!("unexpected {}", node)),
                Ok(node) => node,
                Err(e) => return Err(e),
            };
            self.tree
                .as_mut()
                .and_then(|tree| {
                    tree.root.as_mut().and_then(|r| match *r {
                        Nodes::List(ref mut r) => {
                            r.append(node);
                            Some(())
                        }
                        _ => None,
                    })
                })
                .ok_or_else(|| self.error_msg("invalid root node"))?;

            t = match self.next() {
                None => return self.error(&format!("unable to peek for tree {}", id)),
                Some(t) => t,
            };
        }
        self.backup(t);
        Ok(())
    }

    fn parse_definition(&mut self) -> Result<(), String> {
        let context = "define clause";
        let id = self.tree_id;
        let token = self.next_non_space_must(context)?;
        let name = self.parse_template_name(&token, context)?;
        self.expect(&ItemType::ItemRightDelim, "define end")?;
        self.start_parse(name, id + 1);
        let (list, end) = self.item_list()?;
        if *end.typ() != NodeType::End {
            return Err(format!("unexpected {} in {}", end, context));
        }
        if let Some(tree) = self.tree.as_mut() {
            tree.root = Some(Nodes::List(list));
        }
        self.stop_parse()
    }

    fn item_list(&mut self) -> Result<(ListNode, Nodes), String> {
        let pos = self.peek_non_space_must("item list")?.pos;
        let mut list = ListNode::new(self.tree_id, pos);
        while self.peek_non_space_must("item list")?.typ != ItemType::ItemEOF {
            let node = self.text_or_action()?;
            match *node.typ() {
                NodeType::End | NodeType::Else => return Ok((list, node)),
                _ => list.append(node),
            }
        }
        self.error("unexpected EOF")
    }

    fn text_or_action(&mut self) -> Result<Nodes, String> {
        match self.next_non_space() {
            Some(ref item) if item.typ == ItemType::ItemText => Ok(Nodes::Text(TextNode::new(
                self.tree_id,
                item.pos,
                item.val.clone(),
            ))),
            Some(ref item) if item.typ == ItemType::ItemLeftDelim => self.action(),
            Some(ref item) => self.unexpected(item, "input"),
            _ => self.error("unexpected end of input"),
        }
    }

    fn action(&mut self) -> Result<Nodes, String> {
        let token = self.next_non_space_must("action")?;
        match token.typ {
            ItemType::ItemBlock => return self.block_control(),
            ItemType::ItemElse => return self.else_control(),
            ItemType::ItemEnd => return self.end_control(),
            ItemType::ItemIf => return self.if_control(),
            ItemType::ItemRange => return self.range_control(),
            ItemType::ItemTemplate => return self.template_control(),
            ItemType::ItemWith => return self.with_control(),
            _ => {}
        }
        let pos = token.pos;
        self.backup(token);
        Ok(Nodes::Action(ActionNode::new(
            self.tree_id,
            pos,
            self.pipeline("command")?,
        )))
    }

    fn parse_control(
        &mut self,
        allow_else_if: bool,
        context: &str,
    ) -> Result<(Pos, PipeNode, ListNode, Option<ListNode>), String> {
        let vars_len = self.tree.as_ref().map(|t| t.vars.len()).ok_or("no tree")?;
        let pipe = self.pipeline(context)?;
        let (list, next) = self.item_list()?;
        let else_list = match *next.typ() {
            NodeType::End => None,
            NodeType::Else => {
                if allow_else_if && self.peek_must("else if")?.typ == ItemType::ItemIf {
                    self.next_must("else if")?;
                    let mut else_list = ListNode::new(self.tree_id, next.pos());
                    else_list.append(self.if_control()?);
                    Some(else_list)
                } else {
                    let (else_list, next) = self.item_list()?;
                    if *next.typ() != NodeType::End {
                        return self.error(&format!("expected end; found {}", next));
                    }
                    Some(else_list)
                }
            }
            _ => return self.error(&format!("expected end; found {}", next)),
        };
        if let Some(t) = self.tree.as_mut() {
            t.pop_vars(vars_len);
        }
        Ok((pipe.pos(), pipe, list, else_list))
    }

    fn if_control(&mut self) -> Result<Nodes, String> {
        let (pos, pipe, list, else_list) = self.parse_control(true, "if")?;
        Ok(Nodes::If(IfNode::new_if(
            self.tree_id,
            pos,
            pipe,
            list,
            else_list,
        )))
    }

    fn range_control(&mut self) -> Result<Nodes, String> {
        let (pos, pipe, list, else_list) = self.parse_control(false, "range")?;
        Ok(Nodes::Range(RangeNode::new_range(
            self.tree_id,
            pos,
            pipe,
            list,
            else_list,
        )))
    }

    fn with_control(&mut self) -> Result<Nodes, String> {
        let (pos, pipe, list, else_list) = self.parse_control(false, "with")?;
        Ok(Nodes::With(WithNode::new_with(
            self.tree_id,
            pos,
            pipe,
            list,
            else_list,
        )))
    }

    fn end_control(&mut self) -> Result<Nodes, String> {
        Ok(Nodes::End(EndNode::new(
            self.tree_id,
            self.expect(&ItemType::ItemRightDelim, "end")?.pos,
        )))
    }

    fn else_control(&mut self) -> Result<Nodes, String> {
        if self.peek_non_space_must("else")?.typ == ItemType::ItemIf {
            let peek = self.peek_non_space_must("else")?;
            return Ok(Nodes::Else(ElseNode::new(peek.pos, peek.line)));
        }
        let token = self.expect(&ItemType::ItemRightDelim, "else")?;
        Ok(Nodes::Else(ElseNode::new(token.pos, token.line)))
    }

    fn block_control(&mut self) -> Result<Nodes, String> {
        let context = "block clause";
        let token = self.next_non_space_must(context)?;
        let name = self.parse_template_name(&token, context)?;
        let pipe = self.pipeline(context)?;

        self.max_tree_id += 1;
        let tree_id = self.max_tree_id;
        self.start_parse(name.clone(), tree_id);
        let (root, end) = self.item_list()?;
        if let Some(tree) = self.tree.as_mut() {
            tree.root = Some(Nodes::List(root));
        }
        if end.typ() != &NodeType::End {
            return self.error(&format!("unexpected {} in {}", end, context));
        }
        self.stop_parse()?;
        Ok(Nodes::Template(TemplateNode::new(
            self.tree_id,
            token.pos,
            PipeOrString::String(name),
            Some(pipe),
        )))
    }

    fn template_control(&mut self) -> Result<Nodes, String> {
        let context = "template clause";
        let token = self
            .next_non_space()
            .ok_or_else(|| String::from("unexpected end"))?;
        let name = if let ItemType::ItemLeftParen = token.typ {
            #[cfg(feature = "gtmpl_dynamic_template")]
            {
                let pipe = self.pipeline(context)?;
                self.next_must("template name pipeline end")?;
                PipeOrString::Pipe(pipe)
            }
            #[cfg(not(feature = "gtmpl_dynamic_template"))]
            return Err(String::from(
                "enable gtmpl_dynamic_template to use a pipeline as name",
            ));
        } else {
            PipeOrString::String(self.parse_template_name(&token, context)?)
        };
        let next = self
            .next_non_space()
            .ok_or_else(|| String::from("unexpected end"))?;
        let pipe = if next.typ != ItemType::ItemRightDelim {
            self.backup(next);
            Some(self.pipeline(context)?)
        } else {
            None
        };
        Ok(Nodes::Template(TemplateNode::new(
            self.tree_id,
            token.pos,
            name,
            pipe,
        )))
    }

    fn pipeline(&mut self, context: &str) -> Result<PipeNode, String> {
        let mut decl = vec![];
        let mut token = self.next_non_space_must("pipeline")?;
        let pos = token.pos;
        // TODO: test this hard!
        if token.typ == ItemType::ItemVariable {
            while token.typ == ItemType::ItemVariable {
                let token_after_var = self.next_must("variable")?;
                let next = if token_after_var.typ == ItemType::ItemSpace {
                    let next = self.next_non_space_must("variable")?;
                    if next.typ != ItemType::ItemColonEquals
                        && !(next.typ == ItemType::ItemChar && next.val == ",")
                    {
                        self.backup3(token, token_after_var, next);
                        break;
                    }
                    next
                } else {
                    token_after_var
                };
                if next.typ == ItemType::ItemColonEquals
                    || (next.typ == ItemType::ItemChar && next.val == ",")
                {
                    let variable = VariableNode::new(self.tree_id, token.pos, &token.val);
                    self.add_var(token.val.clone())?;
                    decl.push(variable);
                    if next.typ == ItemType::ItemChar && next.val == "," {
                        if context == "range" && decl.len() < 2 {
                            token = self.next_non_space_must("variable")?;
                            continue;
                        }
                        return self.error(&format!("to many decalarations in {}", context));
                    }
                } else {
                    self.backup2(token, next);
                }
                break;
            }
        } else {
            self.backup(token);
        }
        let mut pipe = PipeNode::new(self.tree_id, pos, decl);
        let mut token = self.next_non_space_must("pipeline")?;
        loop {
            match token.typ {
                ItemType::ItemRightDelim | ItemType::ItemRightParen => {
                    self.check_pipeline(&mut pipe, context)?;
                    if token.typ == ItemType::ItemRightParen {
                        self.backup(token);
                    }
                    return Ok(pipe);
                }
                ItemType::ItemBool
                | ItemType::ItemCharConstant
                | ItemType::ItemDot
                | ItemType::ItemField
                | ItemType::ItemIdentifier
                | ItemType::ItemNumber
                | ItemType::ItemNil
                | ItemType::ItemRawString
                | ItemType::ItemString
                | ItemType::ItemVariable
                | ItemType::ItemLeftParen => {
                    self.backup(token);
                    pipe.append(self.command()?);
                }
                _ => return self.unexpected(&token, context),
            }
            token = self.next_non_space_must("pipeline")?;
        }
    }

    fn check_pipeline(&mut self, pipe: &mut PipeNode, context: &str) -> Result<(), String> {
        if pipe.cmds.is_empty() {
            return self.error(&format!("missing value for {}", context));
        }
        for (i, c) in pipe.cmds.iter().enumerate().skip(1) {
            match c.args.first() {
                Some(n) => match *n.typ() {
                    NodeType::Bool
                    | NodeType::Dot
                    | NodeType::Nil
                    | NodeType::Number
                    | NodeType::String => {
                        return self.error(&format!(
                            "non executable command in pipeline stage {}",
                            i + 2
                        ))
                    }
                    _ => {}
                },
                None => {
                    return self.error(&format!(
                        "non executable command in pipeline stage {}",
                        i + 2
                    ))
                }
            }
        }
        Ok(())
    }

    fn command(&mut self) -> Result<CommandNode, String> {
        let mut cmd = CommandNode::new(self.tree_id, self.peek_non_space_must("command")?.pos);
        loop {
            self.peek_non_space_must("operand")?;


text:
// This file is part of the uutils coreutils package.
//
// (c) Alex Lyon <arcterus@mail.com>
//
// For the full copyright and license information, please view the LICENSE
// file that was distributed with this source code.

// spell-checker:ignore (vars) NANOS numstr
//! Parsing a duration from a string.
//!
//! Use the [`from_str`] function to parse a [`Duration`] from a string.

use std::time::Duration;

use crate::display::Quotable;

/// Parse a duration from a string.
///
/// The string may contain only a number, like "123" or "4.5", or it
/// may contain a number with a unit specifier, like "123s" meaning
/// one hundred twenty three seconds or "4.5d" meaning four and a half
/// days. If no unit is specified, the unit is assumed to be seconds.
///
/// This function uses [`Duration::saturating_mul`] to compute the
/// number of seconds, so it does not overflow. If overflow would have
/// occurred, [`Duration::MAX`] is returned instead.
///
/// # Errors
///
/// This function returns an error if the input string is empty, the
/// input is not a valid number, or the unit specifier is invalid or
/// unknown.
///
/// # Examples
///
/// ```rust
/// use std::time::Duration;
/// use uucore::parse_time::from_str;
/// assert_eq!(from_str("123"), Ok(Duration::from_secs(123)));
/// assert_eq!(from_str("2d"), Ok(Duration::from_secs(60 * 60 * 24 * 2)));
/// ```
pub fn from_str(string: &str) -> Result<Duration, String> {
    let len = string.len();
    if len == 0 {
        return Err("empty string".to_owned());
    }
    let slice = &string[..len - 1];
    let (numstr, times) = match string.chars().next_back().unwrap() {
        's' | 'S' => (slice, 1),
        'm' | 'M' => (slice, 60),
        'h' | 'H' => (slice, 60 * 60),
        'd' | 'D' => (slice, 60 * 60 * 24),
        val if !val.is_alphabetic() => (string, 1),
        _ => {
            if string == "inf" || string == "infinity" {
                ("inf", 1)
            } else {
                return Err(format!("invalid time interval {}", string.quote()));
            }
        }
    };
    let num = numstr
        .parse::<f64>()
        .map_err(|e| format!("invalid time interval {}: {}", string.quote(), e))?;

    if num < 0. {
        return Err(format!("invalid time interval {}", string.quote()));
    }

    const NANOS_PER_SEC: u32 = 1_000_000_000;
    let whole_secs = num.trunc();
    let nanos = (num.fract() * (NANOS_PER_SEC as f64)).trunc();
    let duration = Duration::new(whole_secs as u64, nanos as u32);
    Ok(duration.saturating_mul(times))
}

#[cfg(test)]
mod tests {

    use crate::parse_time::from_str;
    use std::time::Duration;

    #[test]
    fn test_no_units() {
        assert_eq!(from_str("123"), Ok(Duration::from_secs(123)));
    }

    #[test]
    fn test_units() {
        assert_eq!(from_str("2d"), Ok(Duration::from_secs(60 * 60 * 24 * 2)));
    }

    #[test]
    fn test_saturating_mul() {
        assert_eq!(from_str("9223372036854775808d"), Ok(Duration::MAX));
    }

    #[test]
    fn test_error_empty() {
        assert!(from_str("").is_err());
    }

    #[test]
    fn test_error_invalid_unit() {
        assert!(from_str("123X").is_err());
    }

    #[test]
    fn test_error_invalid_magnitude() {
        assert!(from_str("12abc3s").is_err());
    }

    #[test]
    fn test_negative() {
        assert!(from_str("-1").is_err());
    }
}

domain: rust


text:
extern crate chrono;
use anyhow::anyhow;
use chrono::prelude::*;
use chrono::Duration;
use regex::Regex;

static LENGTH_OF_FULL_TIMESTAMP: usize = 8;

pub fn convert_to_local_timestamp(
    utc_date_time: &str,
    should_display_date: bool,
) -> Result<String, anyhow::Error> {
    let parsed_end_time = NaiveDateTime::parse_from_str(utc_date_time, "%Y-%m-%d %H:%M:%S")?;
    let end_dt = DateTime::<Utc>::from_utc(parsed_end_time, Utc);
    let converted_date_time = DateTime::<Local>::from(end_dt);
    let date_format = match should_display_date {
        true => "%Y-%m-%d %H:%M:%S",
        false => "%H:%M:%S",
    };

    Ok(converted_date_time.format(date_format).to_string())
}

pub fn convert_to_utc_timestr(local_date_time: &str) -> Result<String, anyhow::Error> {
    let modified_ldt = if local_date_time.len() < LENGTH_OF_FULL_TIMESTAMP {
        local_date_time.to_owned() + ":00"
    } else {
        local_date_time.to_string()
    };

    match is_valid_timestr(&modified_ldt) {
        true => {
            let now_str = Local::now().format("%H:%M:%S").to_string();
            let date_to_add = if is_time_yesterday(&now_str, &modified_ldt) {
                let dt = Local::now() - Duration::days(1);
                dt.format("%Y-%m-%d ").to_string()
            } else {
                Local::now().format("%Y-%m-%d ").to_string()
            };

            let concat_str: String = date_to_add + &modified_ldt;
            let parsed_local_time =
                NaiveDateTime::parse_from_str(&concat_str, "%Y-%m-%d %H:%M:%S")?;

            let start_dt = Local::from_local_datetime(&Local, &parsed_local_time);
            let converted_date_time = DateTime::<Utc>::from(start_dt.unwrap());

            Ok(converted_date_time.format("%Y-%m-%d %H:%M:%S").to_string())
        }
        false => return Err(anyhow!("time specified is an illegal timestamp, timestamp should be of the format HH:MM:SS")),
    }
}

pub fn get_local_datetime() -> String {
    Local::now().format("%Y-%m-%d %H:%M:%S").to_string()
}

pub fn get_current_utc_string() -> String {
    Utc::now().format("%Y-%m-%d %H:%M:%S").to_string()
}

pub fn get_time_between_stamps(begin_stamp: &str, end_stamp: &str) -> Result<Duration, anyhow::Error> {
    let beg_date_time = NaiveDateTime::parse_from_str(begin_stamp, "%Y-%m-%d %H:%M:%S")?;
    let end_date_time = NaiveDateTime::parse_from_str(end_stamp, "%Y-%m-%d %H:%M:%S")?;
    return Ok(end_date_time - beg_date_time);
}

pub fn get_todays_date() -> String {
    let now_time = Local::now();
    now_time.format("%Y-%m-%d").to_string()
}

fn is_time_yesterday(now_time: &str, compared_time: &str) -> bool {
    let mut now_split: Vec<&str> = now_time.split(':').collect();
    let mut compared_split: Vec<&str> = compared_time.split(':').collect();

    if now_split.len() < compared_split.len() {
        now_split.push("00");
    }

    if compared_split.len() < now_split.len() {
        compared_split.push("00");
    }

    let mut digits_same = true;
    for time_count in 0..now_split.len() {
        let now_digit = now_split[time_count].parse::<i32>().unwrap();
        let compared_digit = compared_split[time_count].parse::<i32>().unwrap();

        if now_digit > compared_digit {
            digits_same = false;
            continue;
        } else if digits_same == true && compared_digit > now_digit {
            return true;
        }
    }

    false
}

fn is_valid_timestr(time_str: &str) -> bool {
    lazy_static! {
        // matches all valid inputs on a 24hr clock
        static ref RE: Regex = Regex::new(r"^(?:(?:([01]?\d|2[0-3]):)([0-5]?\d):)([0-5]\d)$").unwrap();
    }

    RE.is_match(time_str) && time_str.len() == LENGTH_OF_FULL_TIMESTAMP
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_convert_local_to_utc() {
        let timest = Local::now().format("%H:%M:%S").to_string();
        let utc_time = chrono::offset::Utc::now().format("%Y-%m-%d %H:%M:%S").to_string();

        assert_eq!(
            convert_to_utc_timestr(&timest).unwrap().contains(&utc_time),
            true
        );
    }

    #[test]
    fn test_convert_local_to_utc_no_seconds() {
        let timest = Local::now().format("%H:%M").to_string();
        let mut utc_time_with_padd = chrono::offset::Utc::now().format("%H:%M").to_string();

        utc_time_with_padd.push_str(":00");

        assert_eq!(
            convert_to_utc_timestr(&timest)
                .unwrap()
                .contains(&utc_time_with_padd),
            true
        );
    }

    #[test]
    fn test_convert_invalid_local_to_utc_fails() {
        let timest = "100:10:10";
        assert_eq!(convert_to_utc_timestr(&timest).is_err(), true);
    }

    #[test]
    fn test_convert_truncated_local_to_utc_fails() {
        let timest = "10";
        assert_eq!(convert_to_utc_timestr(&timest).is_err(), true);
    }

    #[test]
    fn test_display_local_timestamp_with_date() {
        let c_time = chrono::offset::Utc::now().format("%Y-%m-%d %H:%M:%S").to_string();
        assert_eq!(
            convert_to_local_timestamp(&c_time, true).unwrap(),
            Local::now().format("%Y-%m-%d %H:%M:%S").to_string(),
        );
    }

    #[test]
    fn test_display_local_timestamp_without_date() {
        let c_time = chrono::offset::Utc::now().format("%Y-%m-%d %H:%M:%S").to_string();
        assert_eq!(
            convert_to_local_timestamp(&c_time, false).unwrap(),
            Local::now().format("%H:%M:%S").to_string(),
        );
    }

    #[test]
    fn test_display_local_timestamp_error_hit() {
        let timest = "arestneasrtn";
        assert_eq!(convert_to_local_timestamp(&timest, false).is_err(), true);
    }

    #[test]
    fn test_display_local_timestamp_error_incorrect_format() {
        let timest = "009:009:009";
        assert_eq!(convert_to_local_timestamp(&timest, false).is_err(), true);
    }

    #[test]
    fn test_get_todays_date() {
        assert_eq!(
            get_todays_date(),
            Local::now().format("%Y-%m-%d").to_string()
        );
    }

    #[test]
    fn test_is_time_yesterday() {
        assert_eq!(is_time_yesterday("08:08:08", "09:09:09"), true);
        assert_eq!(is_time_yesterday("08:08:08", "07:59:59"), false);
        assert_eq!(is_time_yesterday("00:01:08", "23:09:09"), true);
        assert_eq!(is_time_yesterday("01:01:08", "01:01:00"), false);
    }

    #[test]
    fn test_is_valid_timestr() {
        let valid_strs = vec!["01:01:00", "12:12:59", "23:59:28", "23:59:40"];
        let invalid_strs = vec!["22:80:00", "21:21:80", "111:12:12", "10:10", "09:09", "09"];

        for str in &valid_strs {
            assert_eq!(is_valid_timestr(str), true);
        }

        for str in &invalid_strs {
            assert_eq!(is_valid_timestr(str), false);
        }
    }

    #[test]
    fn test_time_since_start() {
        let begin_time = "2021-03-18 23:36:24";
        let end_time = "2021-03-18 23:36:59";
        let prev_time = Utc.ymd(2021, 03, 18).and_hms(23, 36, 24);
        let future_time = Utc.ymd(2021, 03, 18).and_hms(23, 36, 59);
        let actual = get_time_between_stamps(begin_time, end_time);
        assert_eq!(actual.unwrap(), future_time - prev_time);
    }

    #[test]
    fn test_time_since_start_minutes() {
        let begin_time = "2021-03-18 23:36:24";
        let end_time = "2021-03-18 23:38:59";
        let prev_time = Utc.ymd(2021, 03, 18).and_hms(23, 36, 24);
        let future_time = Utc.ymd(2021, 03, 18).and_hms(23, 38, 59);
        let actual = get_time_between_stamps(begin_time, end_time);
        assert_eq!(actual.unwrap(), future_time - prev_time);
    }

    #[test]
    fn test_time_since_start_hours() {
        let begin_time = "2021-03-18 23:36:24";
        let end_time = "2021-03-19 01:38:59";
        let prev_time = Utc.ymd(2021, 03, 18).and_hms(23, 36, 24);
        let future_time = Utc.ymd(2021, 03, 19).and_hms(01, 38, 59);
        let actual = get_time_between_stamps(begin_time, end_time);
        assert_eq!(actual.unwrap(), future_time - prev_time);
    }
}


text:
// Copyright Materialize, Inc. and contributors. All rights reserved.
//
// Use of this software is governed by the Business Source License
// included in the LICENSE file.
//
// As of the Change Date specified in that file, in accordance with
// the Business Source License, use of this software will be governed
// by the Apache License, Version 2.0.

pub(crate) mod catalog;
pub(crate) mod util;

use crate::plan::query::QueryLifetime;
use crate::plan::StatementContext;
use mz_expr_test_util::generate_explanation;
use mz_lowertest::*;

use crate::query_model::Model;
use catalog::TestCatalog;

use lazy_static::lazy_static;
use serde::{Deserialize, Serialize};
use std::collections::HashMap;

/// Tests to run on a Query Graph Model.
#[derive(Debug, Copy, Clone, PartialEq, Eq, Serialize, Deserialize, MzReflect)]
enum Directive {
    // TODO: support build apply=(<stuff>)
    /// Apply any number of specific changes to the model.
    Build,
    /// Decorrelate the model and convert it to a `MirRelationExpr`.
    Lower,
    /// Optimize the model.
    Opt,
    /// Optimize and decorrelate the model. Then convert it to a `MirRelationExpr`.
    EndToEnd,
}

lazy_static! {
    pub static ref RTI: ReflectedTypeInfo = {
        let mut rti = ReflectedTypeInfo::default();
        Directive::add_to_reflected_type_info(&mut rti);
        rti
    };
}

/// Convert the input string to a Query Graph Model.
fn convert_input_to_model(input: &str, catalog: &TestCatalog) -> Result<Model, String> {
    // TODO (#9347): Support parsing specs for HirRelationExpr.
    // TODO (#10518): Support parsing specs for QGM.
    // match parse_input_as_qgm(input) {
    //   Ok(model) => Ok(model),
    //   Err(err) => {
    //      let hir = match parse_input_as_hir(input) {
    //         Ok(hir) => hir,
    //         Err(err2) =>
    //      }
    //      Model::from(hir)
    //   }
    // }
    match mz_sql_parser::parser::parse_statements(input) {
        Ok(mut stmts) => {
            assert!(stmts.len() == 1);
            let stmt = stmts.pop().unwrap();
            let scx = &StatementContext::new(None, catalog);
            if let mz_sql_parser::ast::Statement::Select(query) = stmt {
                let planned_query = match crate::plan::query::plan_root_query(
                    scx,
                    query.query,
                    QueryLifetime::Static,
                ) {
                    Ok(planned_query) => planned_query,
                    Err(e) => return Err(format!("unable to plan query: {}: {}", input, e)),
                };
                Model::try_from(planned_query.expr).map_err(|e| e.into())
            } else {
                Err(format!("invalid query: {}", input))
            }
        }
        Err(e) => {
            // TODO: try to parse the input as a spec for an HIR.
            // If that fails, try to parse the input as a spec for a QGM.
            // Change this error message.
            Err(format!("unable to parse SQL: {}: {}", input, e))
        }
    }
}

fn run_command(
    command: &str,
    input: &str,
    args: &HashMap<String, Vec<String>>,
    catalog: &TestCatalog,
) -> Result<String, String> {
    let mut model = convert_input_to_model(input, catalog)?;
    let directive: Directive = deserialize(
        &mut tokenize(command)?.into_iter(),
        "Directive",
        &RTI,
        &mut GenericTestDeserializeContext::default(),
    )?;

    if matches!(directive, Directive::Opt | Directive::EndToEnd) {
        model.optimize();
    }

    // TODO: allow printing multiple stages of the transformation of the query.
    if matches!(directive, Directive::Lower | Directive::EndToEnd) {
        match model.try_into() {
            Ok(mir) => Ok(generate_explanation(catalog, &mir, args.get("format"))),
            Err(err) => Err(err.to_string()),
        }
    } else {
        match model.as_dot(input, catalog, false) {
            Ok(graph) => Ok(graph),
            Err(e) => return Err(format!("graph generation error: {}", e)),
        }
    }
}

#[test]
fn test_qgm() {
    datadriven::walk("tests/querymodel", |f| {
        let mut catalog = TestCatalog::default();

        f.run(move |s| -> String {
            match s.directive.as_str() {
                "cat" => match catalog.execute_commands(&s.input) {
                    Ok(ok) => ok,
                    Err(err) => err,
                },
                other => match run_command(other, &s.input, &s.args, &catalog) {
                    Ok(ok) => ok,
                    Err(err) => err,
                },
            }
        })
    });
}


text:
/*
	Copyright 2021 Integritee AG and Supercomputing Systems AG
	Copyright (C) 2017-2019 Baidu, Inc. All Rights Reserved.

	Licensed under the Apache License, Version 2.0 (the "License");
	you may not use this file except in compliance with the License.
	You may obtain a copy of the License at

		http://www.apache.org/licenses/LICENSE-2.0

	Unless required by applicable law or agreed to in writing, software
	distributed under the License is distributed on an "AS IS" BASIS,
	WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	See the License for the specific language governing permissions and
	limitations under the License.

*/

pub mod attestation_ocall_mock;
pub mod enclave_rpc_ocall_mock;
pub mod rpc_responder_mock;


text:
#![allow(unused)]
pub struct Solution {}

// google interview
impl Solution {
    pub fn valid_mountain_array(arr: Vec<i32>) -> bool {
        // base case len < 3 would not be a mountain array
        if arr.len() < 3 {
            return false;
        }

        let mut i = 0;
        // walk up
        while i + 1 < arr.len() && arr[i] < arr[i + 1] {
            i += 1;
        }

        // peak can't be first or last
        if i == 0 || i == arr.len() - 1 {
            return false;
        }

        // walk down
        while i + 1 < arr.len() && arr[i] > arr[i + 1] {
            i += 1;
        }

        i == arr.len() - 1
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_941() {
        assert_eq!(Solution::valid_mountain_array(vec![2, 1]), false);
        assert_eq!(Solution::valid_mountain_array(vec![0,3,2,1]), true);
    }
}


text:
mod factory;
#[cfg(test)]
mod tests;

pub use factory::ServiceExecutorFactory;

use std::{
    cell::RefCell,
    collections::HashMap,
    ops::{Deref, DerefMut},
    panic::{self, AssertUnwindSafe},
    rc::Rc,
    sync::Arc,
};

use cita_trie::DB as TrieDB;
use derive_more::Display;

use common_apm::muta_apm;
use protocol::traits::{
    Context, Dispatcher, Executor, ExecutorParams, ExecutorResp, NoopDispatcher, Service,
    ServiceMapping, ServiceResponse, ServiceState, Storage,
};
use protocol::types::{
    Address, Hash, MerkleRoot, Receipt, ReceiptResponse, ServiceContext, ServiceContextParams,
    ServiceParam, SignedTransaction, TransactionRequest,
};
use protocol::{ProtocolError, ProtocolErrorKind, ProtocolResult};

use crate::binding::sdk::{DefaultChainQuerier, DefaultServiceSDK};
use crate::binding::state::{GeneralServiceState, MPTTrie};

trait TxHooks {
    fn before(&mut self, _: Context, _: ServiceContext) -> ProtocolResult<()> {
        Ok(())
    }

    fn after(&mut self, _: Context, _: ServiceContext) -> ProtocolResult<()> {
        Ok(())
    }
}

impl TxHooks for () {}

enum HookType {
    Before,
    After,
}

#[derive(Clone, Copy)]
enum ExecType {
    Read,
    Write,
}

struct ServiceStateMap<DB: TrieDB>(HashMap<String, Rc<RefCell<GeneralServiceState<DB>>>>);

impl<DB: TrieDB> ServiceStateMap<DB> {
    fn new() -> ServiceStateMap<DB> {
        Self(HashMap::new())
    }
}

impl<DB: TrieDB> Deref for ServiceStateMap<DB> {
    type Target = HashMap<String, Rc<RefCell<GeneralServiceState<DB>>>>;

    fn deref(&self) -> &Self::Target {
        &self.0
    }
}

impl<DB: TrieDB> DerefMut for ServiceStateMap<DB> {
    fn deref_mut(&mut self) -> &mut Self::Target {
        &mut self.0
    }
}

impl<DB: TrieDB> ServiceStateMap<DB> {
    fn stash(&self) -> ProtocolResult<()> {
        for state in self.0.values() {
            state.borrow_mut().stash()?;
        }

        Ok(())
    }

    fn revert_cache(&self) -> ProtocolResult<()> {
        for state in self.0.values() {
            state.borrow_mut().revert_cache()?;
        }

        Ok(())
    }
}

struct CommitHooks<DB: TrieDB> {
    inner:  Vec<Box<dyn Service>>,
    states: Rc<ServiceStateMap<DB>>,
}

impl<DB: TrieDB> CommitHooks<DB> {
    fn new(hooks: Vec<Box<dyn Service>>, states: Rc<ServiceStateMap<DB>>) -> CommitHooks<DB> {
        Self {
            inner: hooks,
            states,
        }
    }

    // bagua kan 101 :)
    fn kan<H: FnOnce() -> R, R>(states: Rc<ServiceStateMap<DB>>, hook: H) -> ProtocolResult<()> {
        match panic::catch_unwind(AssertUnwindSafe(hook)) {
            Ok(_) => states.stash(),
            Err(_) => states.revert_cache(),
        }
    }
}

impl<DB: TrieDB> TxHooks for CommitHooks<DB> {
    // TODO: support abort execution
    fn before(&mut self, _context: Context, service_context: ServiceContext) -> ProtocolResult<()> {
        for hook in self.inner.iter_mut() {
            Self::kan(Rc::clone(&self.states), || {
                hook.tx_hook_before_(service_context.clone())
            })?;
        }

        Ok(())
    }

    fn after(&mut self, _context: Context, service_context: ServiceContext) -> ProtocolResult<()> {
        for hook in self.inner.iter_mut() {
            Self::kan(Rc::clone(&self.states), || {
                hook.tx_hook_after_(service_context.clone())
            })?;
        }

        Ok(())
    }
}

pub struct ServiceExecutor<S: Storage, DB: TrieDB, Mapping: ServiceMapping> {
    service_mapping: Arc<Mapping>,
    querier:         Rc<DefaultChainQuerier<S>>,
    states:          Rc<ServiceStateMap<DB>>,
    root_state:      Rc<RefCell<GeneralServiceState<DB>>>,
}

impl<S: Storage, DB: TrieDB, Mapping: ServiceMapping> Clone for ServiceExecutor<S, DB, Mapping> {
    fn clone(&self) -> Self {
        Self {
            service_mapping: Arc::clone(&self.service_mapping),
            querier:         Rc::clone(&self.querier),
            states:          Rc::clone(&self.states),
            root_state:      Rc::clone(&self.root_state),
        }
    }
}

impl<S: 'static + Storage, DB: 'static + TrieDB, Mapping: 'static + ServiceMapping>
    ServiceExecutor<S, DB, Mapping>
{
    pub fn create_genesis(
        services: Vec<ServiceParam>,
        trie_db: Arc<DB>,
        storage: Arc<S>,
        mapping: Arc<Mapping>,
    ) -> ProtocolResult<MerkleRoot> {
        let querier = Rc::new(DefaultChainQuerier::new(Arc::clone(&storage)));

        let mut states = ServiceStateMap::new();
        for name in mapping.list_service_name().into_iter() {
            let trie = MPTTrie::new(Arc::clone(&trie_db));

            states.insert(name, Rc::new(RefCell::new(GeneralServiceState::new(trie))));
        }

        for params in services.into_iter() {
            let state = states
                .get(&params.name)
                .ok_or(ExecutorError::NotFoundService {
                    service: params.name.to_owned(),
                })?;
            let sdk =
                DefaultServiceSDK::new(Rc::clone(state), Rc::clone(&querier), NoopDispatcher {});

            let mut service = mapping.get_service(&params.name, sdk)?;
            panic::catch_unwind(AssertUnwindSafe(|| {
                service.genesis_(params.payload.clone())
            }))
            .map_err(|e| ProtocolError::from(ExecutorError::InitService(format!("{:?}", e))))?;

            state.borrow_mut().stash()?;
        }

        let trie = MPTTrie::new(Arc::clone(&trie_db));
        let mut root_state = GeneralServiceState::new(trie);
        for (name, state) in states.iter() {
            let root = state.borrow_mut().commit()?;
            root_state.insert(name.to_owned(), root)?;
        }
        root_state.stash()?;
        root_state.commit()
    }

    pub fn with_root(
        root: MerkleRoot,
        trie_db: Arc<DB>,
        storage: Arc<S>,
        service_mapping: Arc<Mapping>,
    ) -> ProtocolResult<Self> {
        let trie = MPTTrie::from(root, Arc::clone(&trie_db))?;
        let root_state = GeneralServiceState::new(trie);

        let mut states = ServiceStateMap::new();
        for name in service_mapping.list_service_name().into_iter() {
            let trie = match root_state.get(&name)? {
                Some(service_root) => MPTTrie::from(service_root, Arc::clone(&trie_db))?,
                None => MPTTrie::new(Arc::clone(&trie_db)),
            };

            let service_state = GeneralServiceState::new(trie);
            states.insert(name.to_owned(), Rc::new(RefCell::new(service_state)));
        }

        Ok(Self {
            service_mapping,
            querier: Rc::new(DefaultChainQuerier::new(storage)),
            states: Rc::new(states),
            root_state: Rc::new(RefCell::new(root_state)),
        })
    }

    #[muta_apm::derive::tracing_span(kind = "executor.commit")]
    fn commit(&mut self, ctx: Context) -> ProtocolResult<MerkleRoot> {
        for (name, state) in self.states.iter() {
            let root = state.borrow_mut().commit()?;
            self.root_state.borrow_mut().insert(name.to_owned(), root)?;
        }
        self.root_state.borrow_mut().stash()?;
        self.root_state.borrow_mut().commit()
    }

    fn stash(&mut self) -> ProtocolResult<()> {
        self.states.stash()
    }

    fn revert_cache(&mut self) -> ProtocolResult<()> {
        self.states.revert_cache()
    }

    #[muta_apm::derive::tracing_span(
        kind = "executor.before_hook",
        tags = "{'hook_type': 'hook_type'}"
    )]
    fn hook(
        &mut self,
        ctx: Context,
        hook_type: HookType,
        exec_params: &ExecutorParams,
    ) -> ProtocolResult<()> {
        for name in self.service_mapping.list_service_name().into_iter() {
            let sdk = self.get_sdk(&name)?;
            let mut service = self.service_mapping.get_service(name.as_str(), sdk)?;

            let hook_ret = match hook_type {
                HookType::Before => {
                    panic::catch_unwind(AssertUnwindSafe(|| service.hook_before_(exec_params)))
                }
                HookType::After => {
                    panic::catch_unwind(AssertUnwindSafe(|| service.hook_after_(exec_params)))
                }
            };

            if hook_ret.is_err() {
                self.revert_cache()?;
            } else {
                self.stash()?;
            }
        }
        Ok(())
    }

    fn get_sdk(
        &self,
        service: &str,
    ) -> ProtocolResult<DefaultServiceSDK<GeneralServiceState<DB>, DefaultChainQuerier<S>, Self>>
    {
        let state = self
            .states
            .get(service)
            .ok_or(ExecutorError::NotFoundService {
                service: service.to_owned(),
            })?;

        Ok(DefaultServiceSDK::new(
            Rc::clone(&state),
            Rc::clone(&self.querier),
            (*self).clone(),
        ))
    }

    fn get_context(
        &self,
        tx_hash: Option<Hash>,
        nonce: Option<Hash>,
        caller: &Address,
        cycles_price: u64,
        cycles_limit: u64,
        params: &ExecutorParams,
        request: &TransactionRequest,
    ) -> ProtocolResult<ServiceContext> {
        let ctx_params = ServiceContextParams {
            tx_hash,
            nonce,
            cycles_limit,
            cycles_price,
            cycles_used: Rc::new(RefCell::new(0)),
            caller: caller.clone(),
            height: params.height,
            timestamp: params.timestamp,
            service_name: request.service_name.to_owned(),
            service_method: request.method.to_owned(),
            service_payload: request.payload.to_owned(),
            extra: None,
            events: Rc::new(RefCell::new(vec![])),
        };

        Ok(ServiceContext::new(ctx_params))
    }

    fn get_tx_hooks(&self, exec_type: ExecType) -> Box<dyn TxHooks> {
        match exec_type {
            ExecType::Read => Box::new(()),
            ExecType::Write => {
                let mut tx_hooks = vec![];

                for name in self.service_mapping.list_service_name().into_iter() {
                    let sdk = self
                        .get_sdk(&name)
                        .unwrap_or_else(|e| panic!("get target service sdk failed: {}", e));

                    let tx_hook_service = self
                        .service_mapping
                        .get_service(name.as_str(), sdk)
                        .unwrap_or_else(|e| panic!("get target service sdk failed: {}", e));

                    tx_hooks.push(tx_hook_service);
                }

                let hooks = CommitHooks::new(tx_hooks, Rc::clone(&self.states));
                Box::new(hooks)
            }
        }
    }

    fn catch_call(
        &mut self,
        context: Context,
        service_context: ServiceContext,
        exec_type: ExecType,
    ) -> ProtocolResult<ServiceResponse<String>> {
        let mut tx_hooks = self.get_tx_hooks(exec_type);

        tx_hooks.before(context.clone(), service_context.clone())?;

        let ret = match panic::catch_unwind(AssertUnwindSafe(|| {
            self.call(service_context.clone(), exec_type)
        })) {
            Ok(r) => {
                self.stash()?;
                Ok(r)
            }
            Err(e) => {
                self.revert_cache()?;
                log::error!("inner chain error occurred when calling service: {:?}", e);
                Err(ExecutorError::CallService(format!("{:?}", e)).into())
            }
        };

        tx_hooks.after(context, service_context)?;

        ret
    }

    fn call(&self, context: ServiceContext, exec_type: ExecType) -> ServiceResponse<String> {
        let sdk = self
            .get_sdk(context.get_service_name())
            .unwrap_or_else(|e| panic!("get target service sdk failed: {}", e));

        let mut service = self
            .service_mapping
            .get_service(context.get_service_name(), sdk)
            .unwrap_or_else(|e| panic!("get target service failed: {}", e));

        match exec_type {
            ExecType::Read => service.read_(context),
            ExecType::Write => service.write_(context),
        }
    }
}

impl<S: 'static + Storage, DB: 'static + TrieDB, Mapping: 'static + ServiceMapping> Executor
    for ServiceExecutor<S, DB, Mapping>
{
    #[muta_apm::derive::tracing_span(kind = "executor.exec", logs = "{'tx_len': 'txs.len()'}")]
    fn exec(
        &mut self,
        ctx: Context,
        params: &ExecutorParams,
        txs: &[SignedTransaction],
    ) -> ProtocolResult<ExecutorResp> {
        self.hook(ctx.clone(), HookType::Before, params)?;

        let mut receipts = txs
            .iter()
            .map(|stx| {
                let service_context = self.get_context(
                    Some(stx.tx_hash.clone()),
                    Some(stx.raw.nonce.clone()),
                    &stx.raw.sender,
                    stx.raw.cycles_price,
                    stx.raw.cycles_limit,
                    params,
                    &stx.raw.request,
                )?;

                let exec_resp =
                    self.catch_call(ctx.clone(), service_context.clone(), ExecType::Write)?;
                let events = if exec_resp.is_error() {
                    Vec::new()
                } else {
                    service_context.get_events()
                };

                Ok(Receipt {
                    state_root: MerkleRoot::from_empty(),
                    height: service_context.get_current_height(),
                    tx_hash: stx.tx_hash.clone(),
                    cycles_used: service_context.get_cycles_used(),
                    events,
                    response: ReceiptResponse {
                        service_name: service_context.get_service_name().to_owned(),
                        method:       service_context.get_service_method().to_owned(),
                        response:     exec_resp,
                    },
                })
            })
            .collect::<Result<Vec<Receipt>, ProtocolError>>()?;

        self.hook(ctx.clone(), HookType::After, params)?;

        let state_root = self.commit(ctx)?;
        let mut all_cycles_used = 0;

        for receipt in receipts.iter_mut() {
            receipt.state_root = state_root.clone();
            all_cycles_used += receipt.cycles_used;
        }

        Ok(ExecutorResp {
            receipts,
            all_cycles_used,
            state_root,
        })
    }

    fn read(
        &self,
        params: &ExecutorParams,
        caller: &Address,
        cycles_price: u64,
        request: &TransactionRequest,
    ) -> ProtocolResult<ServiceResponse<String>> {
        let context = self.get_context(
            None,
            None,
            caller,
            cycles_price,
            std::u64::MAX,
            params,
            request,
        )?;
        panic::catch_unwind(AssertUnwindSafe(|| self.call(context, ExecType::Read)))
            .map_err(|e| ProtocolError::from(ExecutorError::QueryService(format!("{:?}", e))))
    }
}

impl<S: 'static + Storage, DB: 'static + TrieDB, Mapping: 'static + ServiceMapping> Dispatcher
    for ServiceExecutor<S, DB, Mapping>
{
    fn read(&self, context: ServiceContext) -> ServiceResponse<String> {
        self.call(context, ExecType::Read)
    }

    fn write(&self, context: ServiceContext) -> ServiceResponse<String> {
        self.call(context, ExecType::Write)
    }
}

#[derive(Debug, Display)]
pub enum ExecutorError {
    #[display(fmt = "service {:?} was not found", service)]
    NotFoundService { service: String },
    #[display(fmt = "service {:?} method {:?} was not found", service, method)]
    NotFoundMethod { service: String, method: String },
    #[display(fmt = "Parsing payload to json failed {:?}", _0)]
    JsonParse(serde_json::Error),

    #[display(fmt = "Init service genesis failed: {:?}", _0)]
    InitService(String),
    #[display(fmt = "Query service failed: {:?}", _0)]
    QueryService(String),
    #[display(fmt = "Call service failed: {:?}", _0)]
    CallService(String),
}

impl std::error::Error for ExecutorError {}

impl From<ExecutorError> for ProtocolError {
    fn from(err: ExecutorError) -> ProtocolError {
        ProtocolError::new(ProtocolErrorKind::Executor, Box::new(err))
    }
}


text:
//! A very simple event manager, that just supports log outputs, but no multiprocessing

use crate::{
    events::{
        BrokerEventResult, Event, EventFirer, EventManager, EventManagerId, EventProcessor,
        EventRestarter, HasEventManagerId,
    },
    inputs::Input,
    monitors::Monitor,
    Error,
};
use alloc::{string::ToString, vec::Vec};
#[cfg(feature = "std")]
use core::sync::atomic::{compiler_fence, Ordering};
#[cfg(feature = "std")]
use serde::{de::DeserializeOwned, Serialize};

#[cfg(all(feature = "std", any(windows, not(feature = "fork"))))]
use crate::bolts::os::startable_self;
#[cfg(all(feature = "std", feature = "fork", unix))]
use crate::bolts::os::{fork, ForkResult};
#[cfg(feature = "std")]
use crate::{
    bolts::{shmem::ShMemProvider, staterestore::StateRestorer},
    corpus::Corpus,
    state::{HasCorpus, HasSolutions},
};

use super::ProgressReporter;

/// The llmp connection from the actual fuzzer to the process supervising it
const _ENV_FUZZER_SENDER: &str = "_AFL_ENV_FUZZER_SENDER";
const _ENV_FUZZER_RECEIVER: &str = "_AFL_ENV_FUZZER_RECEIVER";
/// The llmp (2 way) connection from a fuzzer to the broker (broadcasting all other fuzzer messages)
const _ENV_FUZZER_BROKER_CLIENT_INITIAL: &str = "_AFL_ENV_FUZZER_BROKER_CLIENT";

/// A simple, single-threaded event manager that just logs
#[derive(Clone, Debug)]
pub struct SimpleEventManager<I, MT>
where
    I: Input,
    MT: Monitor, //CE: CustomEvent<I, OT>,
{
    /// The monitor
    monitor: MT,
    /// The events that happened since the last handle_in_broker
    events: Vec<Event<I>>,
}

impl<I, MT> EventFirer<I> for SimpleEventManager<I, MT>
where
    I: Input,
    MT: Monitor, //CE: CustomEvent<I, OT>,
{
    fn fire<S>(&mut self, _state: &mut S, event: Event<I>) -> Result<(), Error> {
        match Self::handle_in_broker(&mut self.monitor, &event)? {
            BrokerEventResult::Forward => self.events.push(event),
            BrokerEventResult::Handled => (),
        };
        Ok(())
    }
}

impl<I, MT, S> EventRestarter<S> for SimpleEventManager<I, MT>
where
    I: Input,
    MT: Monitor, //CE: CustomEvent<I, OT>,
{
}

impl<E, I, MT, S, Z> EventProcessor<E, I, S, Z> for SimpleEventManager<I, MT>
where
    I: Input,
    MT: Monitor, //CE: CustomEvent<I, OT>,
{
    fn process(
        &mut self,
        _fuzzer: &mut Z,
        state: &mut S,
        _executor: &mut E,
    ) -> Result<usize, Error> {
        let count = self.events.len();
        while !self.events.is_empty() {
            let event = self.events.pop().unwrap();
            self.handle_in_client(state, event)?;
        }
        Ok(count)
    }
}

impl<E, I, MT, S, Z> EventManager<E, I, S, Z> for SimpleEventManager<I, MT>
where
    I: Input,
    MT: Monitor, //CE: CustomEvent<I, OT>,
{
}

impl<I, MT> ProgressReporter<I> for SimpleEventManager<I, MT>
where
    I: Input,
    MT: Monitor, //CE: CustomEvent<I, OT>,
{
}

impl<I, MT> HasEventManagerId for SimpleEventManager<I, MT>
where
    I: Input,
    MT: Monitor,
{
    fn mgr_id(&self) -> EventManagerId {
        EventManagerId { id: 0 }
    }
}

impl<I, MT> SimpleEventManager<I, MT>
where
    I: Input,
    MT: Monitor, //TODO CE: CustomEvent,
{
    /// Creates a new [`SimpleEventManager`].
    pub fn new(monitor: MT) -> Self {
        Self {
            monitor,
            events: vec![],
        }
    }

    // Handle arriving events in the broker
    #[allow(clippy::unnecessary_wraps)]
    fn handle_in_broker(monitor: &mut MT, event: &Event<I>) -> Result<BrokerEventResult, Error> {
        match event {
            Event::NewTestcase {
                input: _,
                client_config: _,
                exit_kind: _,
                corpus_size,
                observers_buf: _,
                time,
                executions,
            } => {
                monitor
                    .client_stats_mut_for(0)
                    .update_corpus_size(*corpus_size as u64);
                monitor
                    .client_stats_mut_for(0)
                    .update_executions(*executions as u64, *time);
                monitor.display(event.name().to_string(), 0);
                Ok(BrokerEventResult::Handled)
            }
            Event::UpdateExecStats {
                time,
                executions,
                phantom: _,
            } => {
                // TODO: The monitor buffer should be added on client add.
                let client = monitor.client_stats_mut_for(0);

                client.update_executions(*executions as u64, *time);

                monitor.display(event.name().to_string(), 0);
                Ok(BrokerEventResult::Handled)
            }
            Event::UpdateUserStats {
                name,
                value,
                phantom: _,
            } => {
                monitor
                    .client_stats_mut_for(0)
                    .update_user_stats(name.clone(), value.clone());
                monitor.display(event.name().to_string(), 0);
                Ok(BrokerEventResult::Handled)
            }
            #[cfg(feature = "introspection")]
            Event::UpdatePerfMonitor {
                time,
                executions,
                introspection_monitor,
                phantom: _,
            } => {
                // TODO: The monitor buffer should be added on client add.
                let client = &mut monitor.client_stats_mut()[0];
                client.update_executions(*executions as u64, *time);
                client.update_introspection_monitor((**introspection_monitor).clone());
                monitor.display(event.name().to_string(), 0);
                Ok(BrokerEventResult::Handled)
            }
            Event::Objective { objective_size } => {
                monitor
                    .client_stats_mut_for(0)
                    .update_objective_size(*objective_size as u64);
                monitor.display(event.name().to_string(), 0);
                Ok(BrokerEventResult::Handled)
            }
            Event::Log {
                severity_level,
                message,
                phantom: _,
            } => {
                let (_, _) = (message, severity_level);
                #[cfg(feature = "std")]
                println!("[LOG {}]: {}", severity_level, message);
                Ok(BrokerEventResult::Handled)
            } //_ => Ok(BrokerEventResult::Forward),
        }
    }

    // Handle arriving events in the client
    #[allow(clippy::needless_pass_by_value, clippy::unused_self)]
    fn handle_in_client<S>(&mut self, _state: &mut S, event: Event<I>) -> Result<(), Error> {
        Err(Error::Unknown(format!(
            "Received illegal message that message should not have arrived: {:?}.",
            event
        )))
    }
}

/// Provides a `builder` which can be used to build a [`SimpleRestartingEventManager`], which is a combination of a
/// `restarter` and `runner`, that can be used on systems both with and without `fork` support. The
/// `restarter` will start a new process each time the child crashes or times out.
#[cfg(feature = "std")]
#[allow(clippy::default_trait_access)]
#[derive(Debug, Clone)]
pub struct SimpleRestartingEventManager<I, MT, SP>
where
    I: Input,
    SP: ShMemProvider,
    MT: Monitor, //CE: CustomEvent<I, OT>,
{
    /// The actual simple event mgr
    simple_event_mgr: SimpleEventManager<I, MT>,
    /// [`StateRestorer`] for restarts
    staterestorer: StateRestorer<SP>,
}

#[cfg(feature = "std")]
impl<I, MT, SP> EventFirer<I> for SimpleRestartingEventManager<I, MT, SP>
where
    I: Input,
    SP: ShMemProvider,
    MT: Monitor, //CE: CustomEvent<I, OT>,
{
    fn fire<S2>(&mut self, _state: &mut S2, event: Event<I>) -> Result<(), Error> {
        self.simple_event_mgr.fire(_state, event)
    }
}

#[cfg(feature = "std")]
impl<I, MT, S, SP> EventRestarter<S> for SimpleRestartingEventManager<I, MT, SP>
where
    I: Input,
    S: Serialize,
    SP: ShMemProvider,
    MT: Monitor, //CE: CustomEvent<I, OT>,
{
    /// Reset the single page (we reuse it over and over from pos 0), then send the current state to the next runner.
    fn on_restart(&mut self, state: &mut S) -> Result<(), Error> {
        // First, reset the page to 0 so the next iteration can read read from the beginning of this page
        self.staterestorer.reset();
        self.staterestorer.save(state)
    }
}

#[cfg(feature = "std")]
impl<E, I, S, SP, MT, Z> EventProcessor<E, I, S, Z> for SimpleRestartingEventManager<I, MT, SP>
where
    I: Input,
    S: Serialize,
    SP: ShMemProvider,
    MT: Monitor, //CE: CustomEvent<I, OT>,
{
    fn process(&mut self, fuzzer: &mut Z, state: &mut S, executor: &mut E) -> Result<usize, Error> {
        self.simple_event_mgr.process(fuzzer, state, executor)
    }
}

#[cfg(feature = "std")]
impl<E, I, S, SP, MT, Z> EventManager<E, I, S, Z> for SimpleRestartingEventManager<I, MT, SP>
where
    I: Input,
    S: Serialize,
    SP: ShMemProvider,
    MT: Monitor, //CE: CustomEvent<I, OT>,
{
}

#[cfg(feature = "std")]
impl<I, MT, SP> ProgressReporter<I> for SimpleRestartingEventManager<I, MT, SP>
where
    I: Input,
    SP: ShMemProvider,
    MT: Monitor, //CE: CustomEvent<I, OT>,
{
}

#[cfg(feature = "std")]
impl<I, MT, SP> HasEventManagerId for SimpleRestartingEventManager<I, MT, SP>
where
    I: Input,
    SP: ShMemProvider,
    MT: Monitor,
{
    fn mgr_id(&self) -> EventManagerId {
        self.simple_event_mgr.mgr_id()
    }
}

#[cfg(feature = "std")]
#[allow(clippy::type_complexity, clippy::too_many_lines)]
impl<'a, I, MT, SP> SimpleRestartingEventManager<I, MT, SP>
where
    I: Input,
    SP: ShMemProvider,
    MT: Monitor, //TODO CE: CustomEvent,
{
    /// Creates a new [`SimpleEventManager`].
    fn new_launched(monitor: MT, staterestorer: StateRestorer<SP>) -> Self {
        Self {
            staterestorer,
            simple_event_mgr: SimpleEventManager::new(monitor),
        }
    }

    /// Launch the simple restarting manager.
    /// This [`EventManager`] is simple and single threaded,
    /// but can still used shared maps to recover from crashes and timeouts.
    #[allow(clippy::similar_names)]
    pub fn launch<S>(mut monitor: MT, shmem_provider: &mut SP) -> Result<(Option<S>, Self), Error>
    where
        S: DeserializeOwned + Serialize + HasCorpus<I> + HasSolutions<I>,
    {
        // We start ourself as child process to actually fuzz
        let mut staterestorer = if std::env::var(_ENV_FUZZER_SENDER).is_err() {
            // First, create a place to store state in, for restarts.
            let staterestorer: StateRestorer<SP> =
                StateRestorer::new(shmem_provider.new_shmem(256 * 1024 * 1024)?);
            //let staterestorer = { LlmpSender::new(shmem_provider.clone(), 0, false)? };
            staterestorer.write_to_env(_ENV_FUZZER_SENDER)?;

            let mut ctr: u64 = 0;
            // Client->parent loop
            loop {
                println!("Spawning next client (id {})", ctr);

                // On Unix, we fork
                #[cfg(all(unix, feature = "fork"))]
                let child_status = {
                    shmem_provider.pre_fork()?;
                    match unsafe { fork() }? {
                        ForkResult::Parent(handle) => {
                            shmem_provider.post_fork(false)?;
                            handle.status()
                        }
                        ForkResult::Child => {
                            shmem_provider.post_fork(true)?;
                            break staterestorer;
                        }
                    }
                };

                // On windows (or in any case without forks), we spawn ourself again
                #[cfg(any(windows, not(feature = "fork")))]
                let child_status = startable_self()?.status()?;
                #[cfg(all(unix, not(feature = "fork")))]
                let child_status = child_status.code().unwrap_or_default();

                compiler_fence(Ordering::SeqCst);

                #[allow(clippy::manual_assert)]
                if !staterestorer.has_content() {
                    #[cfg(unix)]
                    if child_status == 137 {
                        // Out of Memory, see https://tldp.org/LDP/abs/html/exitcodes.html
                        // and https://github.com/AFLplusplus/LibAFL/issues/32 for discussion.
                        panic!("Fuzzer-respawner: The fuzzed target crashed with an out of memory error! Fix your harness, or switch to another executor (for example, a forkserver).");
                    }

                    // Storing state in the last round did not work
                    panic!("Fuzzer-respawner: Storing state in crashed fuzzer instance did not work, no point to spawn the next client! This can happen if the child calls `exit()`, in that case make sure it uses `abort()`, if it got killed unrecoverable (OOM), or if there is a bug in the fuzzer itself. (Child exited with: {})", child_status);
                }

                ctr = ctr.wrapping_add(1);
            }
        } else {
            // We are the newly started fuzzing instance (i.e. on Windows), first, connect to our own restore map.
            // We get here *only on Windows*, if we were started by a restarting fuzzer.
            // A staterestorer and a receiver for single communication
            StateRestorer::from_env(shmem_provider, _ENV_FUZZER_SENDER)?
        };

        // If we're restarting, deserialize the old state.
        let (state, mgr) = match staterestorer.restore::<S>()? {
            None => {
                println!("First run. Let's set it all up");
                // Mgr to send and receive msgs from/to all other fuzzer instances
                (
                    None,
                    SimpleRestartingEventManager::new_launched(monitor, staterestorer),
                )
            }
            // Restoring from a previous run, deserialize state and corpus.
            Some(state) => {
                println!("Subsequent run. Loaded previous state.");
                // We reset the staterestorer, the next staterestorer and receiver (after crash) will reuse the page from the initial message.
                staterestorer.reset();

                // load the corpus size into monitor to still display the correct numbers after restart.
                let client_stats = monitor.client_stats_mut_for(0);
                client_stats.update_corpus_size(state.corpus().count().try_into()?);
                client_stats.update_objective_size(state.solutions().count().try_into()?);

                (
                    Some(state),
                    SimpleRestartingEventManager::new_launched(monitor, staterestorer),
                )
            }
        };

        /* TODO: Not sure if this is needed
        // We commit an empty NO_RESTART message to this buf, against infinite loops,
        // in case something crashes in the fuzzer.
        staterestorer.send_buf(_LLMP_TAG_NO_RESTART, []);
        */

        Ok((state, mgr))
    }
}
/// `SimpleEventManager` Python bindings
#[cfg(feature = "python")]
pub mod pybind {
    use crate::events::SimpleEventManager;
    use crate::inputs::BytesInput;
    use crate::monitors::pybind::PythonMonitor;
    use pyo3::prelude::*;

    #[pyclass(unsendable, name = "SimpleEventManager")]
    #[derive(Debug, Clone)]
    /// Python class for SimpleEventManager
    pub struct PythonSimpleEventManager {
        /// Rust wrapped SimpleEventManager object
        pub simple_event_manager: SimpleEventManager<BytesInput, PythonMonitor>,
    }

    #[pymethods]
    impl PythonSimpleEventManager {
        #[new]
        fn new(py_monitor: PythonMonitor) -> Self {
            Self {
                simple_event_manager: SimpleEventManager::new(py_monitor),
            }
        }
    }

    /// Register the classes to the python module
    pub fn register(_py: Python, m: &PyModule) -> PyResult<()> {
        m.add_class::<PythonSimpleEventManager>()?;
        Ok(())
    }
}


text:
// Copyright (c) 2018 Mitsuharu Seki
//
// This Source Code Form is subject to the terms of the Mozilla Public
// License, v. 2.0. If a copy of the MPL was not distributed with this
// file, You can obtain one at http://mozilla.org/MPL/2.0/.

use super::*;

pub(crate) trait Read {
    fn reader(register: &mut Register, value: u8);

    fn exec_opcode(
        core: &mut Core,
        ppu: &mut Ppu,
        cartridge: &mut dyn Cartridge,
        controller: &mut dyn Controller,
        apu: &mut Apu,
    ) -> CpuStepStateEnum {
        match core.internal_stat.get_step() {
            1 => {
                let data = core.memory.read(
                    core.internal_stat.get_address(),
                    ppu,
                    cartridge,
                    controller,
                    apu,
                    &mut core.interrupt,
                );
                Self::reader(&mut core.register, data);
            }
            _ => {
                return exit_opcode(core);
            }
        }
        CpuStepStateEnum::Continue
    }
}

macro_rules! read {
    ($name:ident, $reader:expr) => {
        pub(crate) struct $name;

        impl Read for $name {
            fn reader(register: &mut Register, value: u8) {
                $reader(register, value);
            }
        }

        cpu_step_state_impl!($name);
    };
}

pub(crate) trait Write {
    fn writer(register: &mut Register, internal_stat: &InternalStat) -> (u8, usize);

    fn exec_opcode(
        core: &mut Core,
        ppu: &mut Ppu,
        cartridge: &mut dyn Cartridge,
        controller: &mut dyn Controller,
        apu: &mut Apu,
    ) -> CpuStepStateEnum {
        match core.internal_stat.get_step() {
            1 => {
                let (data, address) = Self::writer(&mut core.register, &core.internal_stat);
                core.memory.write(
                    address,
                    data,
                    ppu,
                    cartridge,
                    controller,
                    apu,
                    &mut core.interrupt,
                );
            }
            _ => {
                return exit_opcode(core);
            }
        }
        CpuStepStateEnum::Continue
    }
}

macro_rules! write {
    ($name:ident, $writer:expr) => {
        pub(crate) struct $name;

        impl Write for $name {
            fn writer(register: &mut Register, internal_stat: &InternalStat) -> (u8, usize) {
                $writer(register, internal_stat)
            }
        }

        cpu_step_state_impl!($name);
    };
}

read!(Lax, |r: &mut Register, v: u8| {
    r.set_a(v);
    r.set_x(v);
    r.set_nz_from_value(v);
});

read!(Anc, |r: &mut Register, v: u8| {
    let result = r.get_a() & v;
    r.set_nz_from_value(result);
    r.set_c(result & 0x80 != 0);
    r.set_a(result);
});

read!(Alr, |r: &mut Register, v: u8| {
    let result = r.get_a() & v;
    r.set_c(result & 0x01 != 0);
    let value = result >> 1;
    r.set_nz_from_value(value);
    r.set_a(value);
});

read!(Arr, |r: &mut Register, v: u8| {
    let result = r.get_a() & v;
    let value = result >> 1 | if r.get_c() { 0x80 } else { 0 };
    r.set_c(result & 0x80 != 0);
    r.set_v((((value >> 6) & 1) != 0) ^ (((value >> 5) & 1) != 0));
    r.set_nz_from_value(value);
    r.set_a(value);
});

read!(Xaa, |r: &mut Register, v: u8| {
    let result = r.get_x() & v;
    r.set_nz_from_value(result);
    r.set_a(result);
});

read!(Las, |r: &mut Register, v: u8| {
    let result = r.get_sp() & v;
    r.set_nz_from_value(result);
    r.set_a(result);
    r.set_x(result);
    r.set_sp(result);
});

read!(Axs, |r: &mut Register, v: u8| {
    let a = u16::from(r.get_a() & r.get_x());
    let b = u16::from(v);
    let d = a.wrapping_sub(b);

    let result = (d & 0xFF) as u8;
    r.set_nz_from_value(result);
    r.set_x(result);
    r.set_c(d <= 0xFF);
});

write!(Sax, |r: &mut Register, i: &InternalStat| (
    r.get_a() & r.get_x(),
    i.get_address()
));

write!(Tas, |r: &mut Register, i: &InternalStat| {
    let sp = r.get_a() & r.get_x();
    r.set_sp(sp);
    (
        sp & ((r.get_pc() >> 8) as u8).wrapping_add(1),
        i.get_address(),
    )
});

write!(Ahx, |r: &mut Register, i: &InternalStat| {
    let address = i.get_address();
    let high = ((address >> 8) as u8).wrapping_add(1);
    (r.get_a() & r.get_x() & high, address)
});

write!(Shx, |r: &mut Register, i: &InternalStat| {
    let address = i.get_address();
    let high = ((address >> 8) as u8).wrapping_add(1);
    let low = address & 0xFF;
    let value = r.get_x() & high;
    let new_addr = (usize::from(value) << 8) | low;
    (value, new_addr)
});

write!(Shy, |r: &mut Register, i: &InternalStat| {
    let address = i.get_address();
    let high = ((address >> 8) as u8).wrapping_add(1);
    let low = address & 0xFF;
    let value = r.get_y() & high;
    let new_addr = (usize::from(value) << 8) | low;
    (value, new_addr)
});


text:
// Copyright Materialize, Inc. and contributors. All rights reserved.
//
// Use of this software is governed by the Business Source License
// included in the LICENSE file.
//
// As of the Change Date specified in that file, in accordance with
// the Business Source License, use of this software will be governed
// by the Apache License, Version 2.0.

//! Telemetry collection.
//
// WARNING: The code in this module must be tested manually. Please see
// misc/python/cli/mock_telemetry_server.py for details.

use serde::Deserialize;
use tokio::time::{self, Duration};
use tracing::{debug, warn};
use uuid::Uuid;

use mz_ore::retry::Retry;

use crate::BUILD_INFO;

/// Telemetry configuration.
pub struct Config {
    /// The domain to report telemetry data to.
    pub domain: String,
    /// How often to report telemetry data.
    pub interval: Duration,
    /// The ID of the Materialize cluster.
    pub cluster_id: Uuid,
    /// The number of workers the dataflow server is hosting.
    pub workers: usize,
    /// A client for the coordinator to introspect.
    pub coord_client: mz_coord::Client,
}

/// Runs the telemetry reporting loop.
///
/// The loop ticks at the interval specified in `config.interval`. On each turn,
/// it reports anonymous metadata about the system to the server running at
/// `config.domain`. If it learns of a new Materialize release in the process,
/// it logs a notice.
pub async fn report_loop(config: Config) {
    let mut interval = time::interval(config.interval);
    let mut reported_version = BUILD_INFO.semver_version();
    loop {
        interval.tick().await;

        let latest_version = match report_one(&config).await {
            Ok(latest_version) => latest_version,
            Err(e) => {
                debug!("failed to report telemetry: {}", e);
                continue;
            }
        };

        if latest_version > reported_version {
            // We assume users running development builds are sophisticated, and
            // may be intentionally not running the latest release, so downgrade
            // the message from warn to info level.
            //
            // TODO: avoid duplicating the message if tokio-rs/tracing#372
            // is resolved.
            match BUILD_INFO.semver_version().pre.as_str() {
                "dev" => {
                    debug!(
                        "a new version of materialized is available: {}",
                        latest_version
                    );
                }
                _ => {
                    warn!(
                        "a new version of materialized is available: {}",
                        latest_version
                    );
                }
            };
            reported_version = latest_version;
        }
    }
}

/// Generates the query used to gather telemetry data.
//
// If you add additional data to this query, please be sure to update the
// telemetry docs in doc/user/cli/_index.md#telemetry accordingly, and be sure
// the data is not identifiable.
fn make_telemetry_query(config: &Config) -> String {
    let architecture = std::env::consts::ARCH;
    let os = std::env::consts::OS;
    format!("
        SELECT jsonb_build_object(
            'version', mz_version(),
            'status', jsonb_build_object(
                'session_id', mz_internal.mz_session_id(),
                'uptime_seconds', extract(epoch FROM mz_uptime()),
                'num_workers', {workers},
                'architecture', '{architecture}',
                'os', '{os}',
                'sources', (
                    SELECT jsonb_object_agg(connector_type, jsonb_build_object('count', count))
                    FROM (SELECT connector_type, count(*) FROM mz_sources WHERE id LIKE 'u%' GROUP BY connector_type)
                ),
                'tables', jsonb_build_object('count', (SELECT count(*) FROM mz_tables WHERE id LIKE 'u%')),
                'views', jsonb_build_object('count', (SELECT count(*) FROM mz_views WHERE id LIKE 'u%')),
                'sinks', (
                    SELECT jsonb_object_agg(connector_type, jsonb_build_object('count', count))
                    FROM (SELECT connector_type, count(*) FROM mz_sinks WHERE id LIKE 'u%' GROUP BY connector_type)
                )
            )
        )",
        workers = config.workers
    )
}

/// The response returned by the telemetry server.
#[derive(Deserialize)]
struct V1VersionResponse {
    latest_release: String,
}

async fn report_one(config: &Config) -> Result<semver::Version, anyhow::Error> {
    let response: V1VersionResponse = Retry::default()
        .initial_backoff(Duration::from_secs(1))
        .max_duration(config.interval)
        .retry_async(|_state| async {
            let query_result = config
                .coord_client
                .system_execute_one(&make_telemetry_query(config))
                .await?;
            let response = mz_http_proxy::reqwest::client()
                .post(format!(
                    "https://{}/api/telemetry/{}",
                    config.domain, config.cluster_id
                ))
                .timeout(Duration::from_secs(10))
                .json(&query_result.rows[0][0])
                .send()
                .await?
                .error_for_status()?;
            Ok::<_, anyhow::Error>(response.json().await?)
        })
        .await?;
    Ok(response.latest_release.parse()?)
}


text:
use dataplane::batch::{Batch, BATCH_FILE_HEADER_SIZE, BATCH_HEADER_SIZE};
use dataplane::Offset;
use std::io::{Error as IoError, ErrorKind, Cursor};
use tracing::{warn, debug};
use std::os::unix::io::RawFd;
use nix::sys::uio::pread;
use fluvio_future::file_slice::AsyncFileSlice;

// only encode information necessary to decode batches efficiently
pub struct FileBatch {
    pub(crate) batch: Batch,
    pub(crate) records: Vec<u8>,
}

impl FileBatch {
    pub(crate) fn base_offset(&self) -> Offset {
        self.batch.base_offset
    }

    pub(crate) fn offset_delta(&self) -> i32 {
        self.batch.header.last_offset_delta
    }
}

/// Iterator that returns batch from file
pub struct FileBatchIterator {
    fd: RawFd,
    offset: i64,
    end: i64,
}

impl FileBatchIterator {
    #[allow(unused)]
    pub fn new(fd: RawFd, offset: i64, len: i64) -> Self {
        Self {
            fd,
            offset,
            end: offset + len,
        }
    }

    pub fn from_raw_slice(slice: AsyncFileSlice) -> Self {
        use std::os::unix::io::AsRawFd;
        let offset = slice.position() as i64;
        Self {
            fd: slice.as_raw_fd(),
            offset,
            end: offset + slice.len() as i64,
        }
    }
}

impl Iterator for FileBatchIterator {
    type Item = Result<FileBatch, IoError>;

    fn next(&mut self) -> Option<Self::Item> {
        if self.offset >= self.end {
            return None;
        }

        let mut header = vec![0u8; BATCH_FILE_HEADER_SIZE];
        let bytes_read = match pread(self.fd, &mut header, self.offset)
            .map_err(|err| IoError::new(ErrorKind::Other, format!("pread error {}", err)))
        {
            Ok(bytes) => bytes,
            Err(err) => return Some(Err(err)),
        };

        if bytes_read < header.len() {
            warn!(bytes_read, header_len = header.len());
            return Some(Err(IoError::new(
                ErrorKind::UnexpectedEof,
                format!(
                    "not eough for batch header {} out of {}",
                    bytes_read,
                    header.len()
                ),
            )));
        }

        let mut batch = Batch::default();
        if let Err(err) = batch.decode_from_file_buf(&mut Cursor::new(header), 0) {
            return Some(Err(IoError::new(
                ErrorKind::Other,
                format!("decodinge batch header error {}", err),
            )));
        }

        let remainder = batch.batch_len as usize - BATCH_HEADER_SIZE as usize;

        debug!(
            file_offset = self.offset,
            base_offset = batch.base_offset,
            "fbatch header"
        );

        let mut records = vec![0u8; remainder];

        self.offset += BATCH_FILE_HEADER_SIZE as i64;

        let bytes_read = match pread(self.fd, &mut records, self.offset)
            .map_err(|err| IoError::new(ErrorKind::Other, format!("pread error {}", err)))
        {
            Ok(bytes) => bytes,
            Err(err) => return Some(Err(err)),
        };

        if bytes_read < records.len() {
            warn!(bytes_read, record_len = records.len());
            return Some(Err(IoError::new(
                ErrorKind::UnexpectedEof,
                format!(
                    "not enough for batch records {} out of {}",
                    bytes_read,
                    records.len()
                ),
            )));
        }

        self.offset += bytes_read as i64;

        debug!(file_offset = self.offset, "fbatch end");

        Some(Ok(FileBatch { batch, records }))
    }
}

#[cfg(test)]
mod test {
    use std::{fs::File, io::Write};
    use std::env::temp_dir;
    use std::os::unix::io::AsRawFd;

    use fluvio_types::defaults::STORAGE_MAX_BATCH_SIZE;

    use super::*;

    #[test]
    fn test_file() {
        let path = temp_dir().join("pread.txt");
        let mut file = File::create(&path).expect("create");
        file.write_all(b"Hello, world!").expect("write");
        file.sync_all().expect("flush");
        drop(file);

        let read_only = File::open(path).expect("open");
        let fd = read_only.as_raw_fd();
        let mut buf = vec![0; STORAGE_MAX_BATCH_SIZE as usize];
        // let mut buf = BytesMut::with_capacity(64);
        let bytes_read = pread(fd, &mut buf, 1).expect("");
        println!("bytes read: {}", bytes_read);
        assert!(bytes_read > 2);
    }
}


text:
use rustc_codegen_ssa::debuginfo::{FunctionDebugContext, FunctionDebugContextData, MirDebugScope};
use super::metadata::file_metadata;
use super::utils::{DIB, span_start};

use llvm;
use llvm::debuginfo::{DIScope, DISubprogram};
use common::CodegenCx;
use rustc::mir::{Mir, SourceScope};

use libc::c_uint;

use syntax_pos::Pos;

use rustc_data_structures::bit_set::BitSet;
use rustc_data_structures::indexed_vec::{Idx, IndexVec};

use syntax_pos::BytePos;

/// Produces DIScope DIEs for each MIR Scope which has variables defined in it.
/// If debuginfo is disabled, the returned vector is empty.
pub fn create_mir_scopes(
    cx: &CodegenCx<'ll, '_>,
    mir: &Mir,
    debug_context: &FunctionDebugContext<&'ll DISubprogram>,
) -> IndexVec<SourceScope, MirDebugScope<&'ll DIScope>> {
    let null_scope = MirDebugScope {
        scope_metadata: None,
        file_start_pos: BytePos(0),
        file_end_pos: BytePos(0)
    };
    let mut scopes = IndexVec::from_elem(null_scope, &mir.source_scopes);

    let debug_context = match *debug_context {
        FunctionDebugContext::RegularContext(ref data) => data,
        FunctionDebugContext::DebugInfoDisabled |
        FunctionDebugContext::FunctionWithoutDebugInfo => {
            return scopes;
        }
    };

    // Find all the scopes with variables defined in them.
    let mut has_variables = BitSet::new_empty(mir.source_scopes.len());
    for var in mir.vars_iter() {
        let decl = &mir.local_decls[var];
        has_variables.insert(decl.visibility_scope);
    }

    // Instantiate all scopes.
    for idx in 0..mir.source_scopes.len() {
        let scope = SourceScope::new(idx);
        make_mir_scope(cx, &mir, &has_variables, debug_context, scope, &mut scopes);
    }

    scopes
}

fn make_mir_scope(cx: &CodegenCx<'ll, '_>,
                  mir: &Mir,
                  has_variables: &BitSet<SourceScope>,
                  debug_context: &FunctionDebugContextData<&'ll DISubprogram>,
                  scope: SourceScope,
                  scopes: &mut IndexVec<SourceScope, MirDebugScope<&'ll DIScope>>) {
    if scopes[scope].is_valid() {
        return;
    }

    let scope_data = &mir.source_scopes[scope];
    let parent_scope = if let Some(parent) = scope_data.parent_scope {
        make_mir_scope(cx, mir, has_variables, debug_context, parent, scopes);
        scopes[parent]
    } else {
        // The root is the function itself.
        let loc = span_start(cx, mir.span);
        scopes[scope] = MirDebugScope {
            scope_metadata: Some(debug_context.fn_metadata),
            file_start_pos: loc.file.start_pos,
            file_end_pos: loc.file.end_pos,
        };
        return;
    };

    if !has_variables.contains(scope) {
        // Do not create a DIScope if there are no variables
        // defined in this MIR Scope, to avoid debuginfo bloat.

        // However, we don't skip creating a nested scope if
        // our parent is the root, because we might want to
        // put arguments in the root and not have shadowing.
        if parent_scope.scope_metadata.unwrap() != debug_context.fn_metadata {
            scopes[scope] = parent_scope;
            return;
        }
    }

    let loc = span_start(cx, scope_data.span);
    let file_metadata = file_metadata(cx,
                                      &loc.file.name,
                                      debug_context.defining_crate);

    let scope_metadata = unsafe {
        Some(llvm::LLVMRustDIBuilderCreateLexicalBlock(
            DIB(cx),
            parent_scope.scope_metadata.unwrap(),
            file_metadata,
            loc.line as c_uint,
            loc.col.to_usize() as c_uint))
    };
    scopes[scope] = MirDebugScope {
        scope_metadata,
        file_start_pos: loc.file.start_pos,
        file_end_pos: loc.file.end_pos,
    };
}


text:
use crate::{filter::select_from_json_object, join::OrderedValue};
use itertools::Itertools;
use ndjson_common::{
    error::NdJsonSpatialError, json_selector_parser::Selector, ndjson::NdjsonReader,
};

use std::{
    cmp::Ordering,
    io::{BufRead, Write},
};

pub fn sort<IN: BufRead, OUT: Write>(
    input: IN,
    output: &mut OUT,
    selectors: Vec<(Vec<Selector>, bool)>,
) -> Result<(), NdJsonSpatialError> {
    for item in NdjsonReader::new(input).flatten().sorted_by(|left, right| {
        selectors
            .iter()
            .map(|(selector, ascending)| {
                if *ascending {
                    Ord::cmp(
                        &select_from_json_object(left.clone(), selector).map(OrderedValue::from),
                        &select_from_json_object(right.clone(), selector).map(OrderedValue::from),
                    )
                } else {
                    Ord::cmp(
                        &select_from_json_object(right.clone(), selector).map(OrderedValue::from),
                        &select_from_json_object(left.clone(), selector).map(OrderedValue::from),
                    )
                }
            })
            .fold(Ordering::Equal, |acc, item| acc.then(item))
    }) {
        writeln!(output, "{}", item).expect("Unable to write to stdout");
    }

    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_sort_already_sorted() {
        let input = "{\"bar\":3,\"foo\":4}\n{\"bar\":4,\"foo\":7}\n";

        let mut output = vec![];

        sort(
            input.as_bytes(),
            &mut output,
            vec![(vec![Selector::Identifier("bar".into())], true)],
        )
        .unwrap();

        assert_eq!(output, input.as_bytes());

        let mut output = vec![];

        sort(
            input.as_bytes(),
            &mut output,
            vec![
                (vec![Selector::Identifier("foo".into())], true),
                (vec![Selector::Identifier("bar".into())], true),
            ],
        )
        .unwrap();

        assert_eq!(output, input.as_bytes());
    }

    #[test]
    fn test_sort_descending() {
        let input = "{\"bar\":3,\"foo\":4}\n{\"bar\":4,\"foo\":7}\n";

        let mut output = vec![];

        sort(
            input.as_bytes(),
            &mut output,
            vec![(vec![Selector::Identifier("bar".into())], false)],
        )
        .unwrap();

        assert_eq!(
            output,
            "{\"bar\":4,\"foo\":7}\n{\"bar\":3,\"foo\":4}\n".as_bytes()
        );
    }
}