fn execute_helper(
    context: RootContext,
    sender: mpsc::Sender<Vec<Result<Value, Failure>>>,
    roots: Vec<Root>,
    count: usize,
  ) {
    let core = context.core.clone();
    // Attempt all roots in parallel, failing fast to retry for `Invalidated`.
    let roots_res = future::join_all(
      roots
        .clone()
        .into_iter()
        .map(|root| {
          context
            .core
            .graph
            .create(root.clone().into(), &context)
            .then::<_, Result<Result<Value, Failure>, Failure>>(move |r| {
              match r {
                Err(Failure::Invalidated) if count > 0 => {
                  // A node was invalidated: fail quickly so that all roots can be retried.
                  Err(Failure::Invalidated)
                }
                other => {
                  // Otherwise (if it is a success, some other type of Failure, or if we've run
                  // out of retries) recover to complete the join, which will cause the results to
                  // propagate to the user.
                  debug!("Root {} completed.", NodeKey::Select(Box::new(root)));
                  Ok(other.map(|res| {
                    res
                      .try_into()
                      .unwrap_or_else(|_| panic!("A Node implementation was ambiguous."))
                  }))
                }
              }
            })
        })
        .collect::<Vec<_>>(),
    );

    // If the join failed (due to `Invalidated`, since that is the only error we propagate), retry
    // the entire set of roots.
    core.spawn(roots_res.then(move |res| {
      if let Ok(res) = res {
        sender.send(res).map_err(|_| ())
      } else {
        Scheduler::execute_helper(context, sender, roots, count - 1);
        Ok(())
      }
    }));
  }
///
/// Attempts to complete all of the given roots, retrying the entire set (up to `count`
/// times) if any of them fail with `Failure::Invalidated`. Sends the result on the given
/// mpsc Sender, which allows the caller to poll a channel for the result without blocking
/// uninterruptibly on a Future.
///
/// In common usage, graph entries won't be repeatedly invalidated, but in a case where they
/// were (say by an automated process changing files under pants), we'd want to eventually
/// give up.
///
Attempts to complete all of the given roots, retrying the entire set (up to `count`
times) if any of them fail with `Failure::Invalidated`. Sends the result on the given
mpsc Sender, which allows the caller to poll a channel for the result without blocking
uninterruptibly on a Future.

In common usage, graph entries won't be repeatedly invalidated, but in a case where they
were (say by an automated process changing files under pants), we'd want to eventually
give up.
Attempts to complete all of the given roots retrying the entire set ( up to ` count ` times ) if any of them fail with ` Failure :: Invalidated ` . Sends the result on the given mpsc Sender which allows the caller to poll a channel for the result without blocking uninterruptibly on a Future . In common usage graph entries won ' t be repeatedly invalidated but in a case where they were ( say by an automated process changing files under pants ) we ' d want to eventually give up .
fn execute_helper(
    context: RootContext,
    sender: mpsc::Sender<Vec<Result<Value, Failure>>>,
    roots: Vec<Root>,
    count: usize,
  ) {
    let core = context.core.clone();
    // Attempt all roots in parallel, failing fast to retry for `Invalidated`.
    let roots_res = future::join_all(
      roots
        .clone()
        .into_iter()
        .map(|root| {
          context
            .core
            .graph
            .create(root.clone().into(), &context)
            .then::<_, Result<Result<Value, Failure>, Failure>>(move |r| {
              match r {
                Err(Failure::Invalidated) if count > 0 => {
                  // A node was invalidated: fail quickly so that all roots can be retried.
                  Err(Failure::Invalidated)
                }
                other => {
                  // Otherwise (if it is a success, some other type of Failure, or if we've run
                  // out of retries) recover to complete the join, which will cause the results to
                  // propagate to the user.
                  debug!("Root {} completed.", NodeKey::Select(Box::new(root)));
                  Ok(other.map(|res| {
                    res
                      .try_into()
                      .unwrap_or_else(|_| panic!("A Node implementation was ambiguous."))
                  }))
                }
              }
            })
        })
        .collect::<Vec<_>>(),
    );
    // If the join failed (due to `Invalidated`, since that is the only error we propagate), retry
    // the entire set of roots.
    core.spawn(roots_res.then(move |res| {
      if let Ok(res) = res {
        sender.send(res).map_err(|_| ())
      } else {
        Scheduler::execute_helper(context, sender, roots, count - 1);
        Ok(())
      }
    }));
  }
fn execute_helper ( context : RootContext , sender : mpsc :: Sender < Vec < Result < Value , Failure > > > , roots : Vec < Root > , count : usize , ) { let core = context . core . clone ( ) ; // Attempt all roots in parallel, failing fast to retry for `Invalidated`. let roots_res = future :: join_all ( roots . clone ( ) . into_iter ( ) . map ( | root | { context . core . graph . create ( root . clone ( ) . into ( ) , & context ) . then :: < _ , Result < Result < Value , Failure > , Failure > > ( move | r | { match r { Err ( Failure :: Invalidated ) if count > 0 => { // A node was invalidated: fail quickly so that all roots can be retried. Err ( Failure :: Invalidated ) } other => { // Otherwise (if it is a success, some other type of Failure, or if we've run // out of retries) recover to complete the join, which will cause the results to // propagate to the user. debug ! ( " " NodeKey Select ( Box new ( root ) ) ) ; Ok ( other . map ( | res | { res . try_into ( ) . unwrap_or_else ( | _ | panic ! ( " " ) ) } ) ) } } } ) } ) . collect :: < Vec < _ > > ( ) , ) ; // If the join failed (due to `Invalidated`, since that is the only error we propagate), retry // the entire set of roots. core . spawn ( roots_res . then ( move | res | { if let Ok ( res ) = res { sender . send ( res ) . map_err ( | _ | ( ) ) } else { Scheduler :: execute_helper ( context , sender , roots , count - 1 ) ; Ok ( ( ) ) } } ) ) ; }
Attempts to complete all of the given roots, retrying the entire set (up to `count`
times) if any of them fail with `Failure::Invalidated`.
Attempts to complete all of the given roots retrying the entire set ( up to ` count ` times ) if any of them fail with ` Failure :: Invalidated ` .
context
RootContext
sender
mpsc::Sender<Vec<Result<Value, Failure>>>
roots
Vec<Root>
count
usize
context
RootContext
sender
mpsc::Sender<Vec<Result<Value, Failure>>>
roots
Vec<Root>
count
usize