Skip to content

Commit

Permalink
Two targets can swap positions with pantsd (again) (#7642)
Browse files Browse the repository at this point in the history 
Re-landing #7617 without slowing down the happy path of no cycles.



Before this PR, nothing would remove the edges of a dirty node, so if
two nodes swapped positions in the graph (e.g. if a dependency between
two targets inverted), a cycle would be detected.

With this PR, if we detect a cycle, but detect that there may be dirty
edges in play, we fully clear that node (including removing its edges),
which will cause it being re-triggered from scratch.

This is specifically in place to handle the cycle scenario - the dirty
bit, and dependency Generations are still the primary mechanism for
handling re-use of old versions.

There's an ugliness here that we still don't remove obsolete edges, so
if Generation 2 of a node has differing dependencies from Generation 1,
the dependency from Generation 1 will still dirty Generation 2. We _may_
want to consider solving that separately as/when it becomes a
significant issue, or we may want to re-work this PR to do something
like that... This PR happens to cover a part of that problem, but only
where it causes definitive problems (a fake cycle) rather than also
where it causes performance problems.

There's probably a slightly more principled solution here along the
lines of:
 * Rather than using () as an edge weight in the graph, use the
   Generation of the dependee Node as an edge weight.
 * When doing cycle detection, compare the edge weight against the
   generation of the node, and ignore obsolete edges.
but I would want to think about that a lot more before doing it...


This re-landing optimises for the no cycle case, which is very strongly expected to be the common case in Pants, as cycles represent errors.
  • Loading branch information
@illicitonion
illicitonion authored Apr 30, 2019
1 parent 7ca3089 commit c823939
Show file tree
Hide file tree
Showing 3 changed files with 386 additions and 67 deletions.
158 changes: 121 additions & 37 deletions src/rust/engine/graph/src/entry.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -52,6 +52,52 @@ impl Generation {
}
}

///
/// A result from running a Node.
///
/// If the value is Dirty, the consumer should check whether the dependencies of the Node have the
/// same values as they did when this Node was last run; if so, the value can be re-used
/// (and should be marked "Clean").
///
/// If the value is Clean, the consumer can simply use the value as-is.
///
#[derive(Clone, Debug)]
pub(crate) enum EntryResult<N: Node> {
Clean(Result<N::Item, N::Error>),
Dirty(Result<N::Item, N::Error>),
}

impl<N: Node> EntryResult<N> {
fn is_dirty(&self) -> bool {
if let EntryResult::Dirty(..) = self {
true
} else {
false
}
}

fn dirty(&mut self) {
if let EntryResult::Clean(value) = self {
*self = EntryResult::Dirty(value.clone())
}
}

fn clean(&mut self) {
if let EntryResult::Dirty(value) = self {
*self = EntryResult::Clean(value.clone())
}
}
}

impl<N: Node> AsRef<Result<N::Item, N::Error>> for EntryResult<N> {
fn as_ref(&self) -> &Result<N::Item, N::Error> {
match self {
EntryResult::Clean(v) => v,
EntryResult::Dirty(v) => v,
}
}
}

#[allow(clippy::type_complexity)]
#[derive(Debug)]
pub(crate) enum EntryState<N: Node> {
Expand All @@ -65,7 +111,7 @@ pub(crate) enum EntryState<N: Node> {
NotStarted {
run_token: RunToken,
generation: Generation,
previous_result: Option<Result<N::Item, N::Error>>,
previous_result: Option<EntryResult<N>>,
},
// A node that is running. A running node that has been marked dirty re-runs rather than
// completing.
Expand All @@ -76,7 +122,7 @@ pub(crate) enum EntryState<N: Node> {
generation: Generation,
start_time: Instant,
waiters: Vec<oneshot::Sender<Result<(N::Item, Generation), N::Error>>>,
previous_result: Option<Result<N::Item, N::Error>>,
previous_result: Option<EntryResult<N>>,
dirty: bool,
},
// A node that has completed, and then possibly been marked dirty. Because marking a node
Expand All @@ -85,9 +131,8 @@ pub(crate) enum EntryState<N: Node> {
Completed {
run_token: RunToken,
generation: Generation,
result: Result<N::Item, N::Error>,
result: EntryResult<N>,
dep_generations: Vec<Generation>,
dirty: bool,
},
}

Expand Down Expand Up @@ -158,8 +203,9 @@ impl<N: Node> Entry<N> {
let state = self.state.lock();
match *state {
EntryState::Completed {
ref result, dirty, ..
} if !dirty => Some(result.clone()),
result: EntryResult::Clean(ref result),
..
} => Some(result.clone()),
_ => None,
}
}
Expand All @@ -175,7 +221,7 @@ impl<N: Node> Entry<N> {
run_token: RunToken,
generation: Generation,
previous_dep_generations: Option<Vec<Generation>>,
previous_result: Option<Result<N::Item, N::Error>>,
previous_result: Option<EntryResult<N>>,
) -> EntryState<N>
where
C: NodeContext<Node = N>,
Expand Down Expand Up @@ -243,16 +289,15 @@ impl<N: Node> Entry<N> {
start_time: Instant::now(),
run_token,
generation,
previous_result: previous_result,
previous_result,
dirty: false,
}
}
&EntryKey::Cyclic(_) => EntryState::Completed {
result: Err(N::Error::cyclic()),
result: EntryResult::Clean(Err(N::Error::cyclic())),
dep_generations: Vec::new(),
run_token,
generation,
dirty: false,
},
}
}
Expand Down Expand Up @@ -293,10 +338,9 @@ impl<N: Node> Entry<N> {
&mut EntryState::Completed {
ref result,
generation,
dirty,
..
} if !dirty && self.node.content().cacheable() => {
return future::result(result.clone())
} if self.node.content().cacheable() && !result.is_dirty() => {
return future::result(result.as_ref().clone())
.map(move |res| (res, generation))
.to_boxed();
}
Expand All @@ -323,21 +367,21 @@ impl<N: Node> Entry<N> {
EntryState::Completed {
run_token,
generation,
result,
mut result,
dep_generations,
dirty,
} => {
assert!(
dirty || !self.node.content().cacheable(),
"A clean Node should not reach this point: {:?}",
result
);
trace!(
"Re-starting node {:?}. It was: dirty={}, cacheable={}",
"Re-starting node {:?}. It was: previous_result={:?}, cacheable={}",
self.node,
dirty,
result,
self.node.content().cacheable()
);
assert!(
result.is_dirty() || !self.node.content().cacheable(),
"A clean Node should not reach this point: {:?}",
result
);
result.dirty();
// The Node has already completed but is now marked dirty. This indicates that we are the
// first caller to request it since it was marked dirty. We attempt to clean it (which will
// cause it to re-run if the dep_generations mismatch).
Expand Down Expand Up @@ -416,7 +460,7 @@ impl<N: Node> Entry<N> {
waiters,
run_token,
generation,
previous_result,
mut previous_result,
dirty,
..
} => {
Expand All @@ -428,6 +472,9 @@ impl<N: Node> Entry<N> {
"Not completing node {:?} because it was invalidated before completing.",
self.node
);
if let Some(previous_result) = previous_result.as_mut() {
previous_result.dirty();
}
EntryState::NotStarted {
run_token: run_token.next(),
generation,
Expand All @@ -440,6 +487,9 @@ impl<N: Node> Entry<N> {
"Not completing node {:?} because it was dirtied before completing.",
self.node
);
if let Some(previous_result) = previous_result.as_mut() {
previous_result.dirty();
}
Self::run(
context,
&self.node,
Expand All @@ -452,19 +502,19 @@ impl<N: Node> Entry<N> {
} else {
// If the new result does not match the previous result, the generation increments.
let (generation, next_result) = if let Some(result) = result {
if Some(&result) == previous_result.as_ref() {
if Some(&result) == previous_result.as_ref().map(EntryResult::as_ref) {
// Node was re-executed, but had the same result value.
(generation, result)
(generation, EntryResult::Clean(result))
} else {
(generation.next(), result)
(generation.next(), EntryResult::Clean(result))
}
} else {
// Node was marked clean.
// NB: The `expect` here avoids a clone and a comparison: see the method docs.
(
generation,
previous_result.expect("A Node cannot be marked clean without a previous result."),
)
let mut result =
previous_result.expect("A Node cannot be marked clean without a previous result.");
result.clean();
(generation, result)
};
// Notify all waiters (ignoring any that have gone away), and then store the value.
// A waiter will go away whenever they drop the `Future` `Receiver` of the value, perhaps
Expand All @@ -476,14 +526,13 @@ impl<N: Node> Entry<N> {
waiters.len()
);
for waiter in waiters {
let _ = waiter.send(next_result.clone().map(|res| (res, generation)));
let _ = waiter.send(next_result.as_ref().clone().map(|res| (res, generation)));
}
EntryState::Completed {
result: next_result,
dep_generations,
run_token,
generation,
dirty: false,
}
}
}
Expand Down Expand Up @@ -541,15 +590,23 @@ impl<N: Node> Entry<N> {
///
/// Clears the state of this Node, forcing it to be recomputed.
///
pub(crate) fn clear(&mut self) {
/// # Arguments
///
/// * `graph_still_contains_edges` - If the caller has guaranteed that all edges from this Node
/// have been removed from the graph, they should pass false here, else true. We may want to
/// remove this parameter, and force this method to remove the edges, but that would require
/// acquiring the graph lock here, which we currently don't do.
///
pub(crate) fn clear(&mut self, graph_still_contains_edges: bool) {
let mut state = self.state.lock();

let (run_token, generation, previous_result) =
let (run_token, generation, mut previous_result) =
match mem::replace(&mut *state, EntryState::initial()) {
EntryState::NotStarted {
run_token,
generation,
previous_result,
..
}
| EntryState::Running {
run_token,
Expand All @@ -567,6 +624,12 @@ impl<N: Node> Entry<N> {

trace!("Clearing node {:?}", self.node);

if graph_still_contains_edges {
if let Some(previous_result) = previous_result.as_mut() {
previous_result.dirty();
}
}

// Swap in a state with a new RunToken value, which invalidates any outstanding work.
*state = EntryState::NotStarted {
run_token: run_token.next(),
Expand All @@ -585,15 +648,36 @@ impl<N: Node> Entry<N> {
let state = &mut *self.state.lock();
trace!("Dirtying node {:?}", self.node);
match state {
&mut EntryState::Running { ref mut dirty, .. }
| &mut EntryState::Completed { ref mut dirty, .. } => {
// Mark dirty.
&mut EntryState::Running { ref mut dirty, .. } => {
*dirty = true;
}
&mut EntryState::Completed { ref mut result, .. } => {
result.dirty();
}
&mut EntryState::NotStarted { .. } => {}
}
}

pub fn may_have_dirty_edges(&self) -> bool {
match *self.state.lock() {
EntryState::NotStarted {
ref previous_result,
..
}
| EntryState::Running {
ref previous_result,
..
} => {
if let Some(EntryResult::Dirty(..)) = previous_result {
true
} else {
false
}
}
EntryState::Completed { ref result, .. } => result.is_dirty(),
}
}

pub(crate) fn format(&self) -> String {
let state = match self.peek() {
Some(Ok(ref nr)) => format!("{:?}", nr),
Expand Down
Loading

0 comments on commit c823939

Please sign in to comment.