Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add intelligent fallback from backjump to backtrack #144

Closed
Closed
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions src/resolvelib/reporters.py
Original file line number Diff line number Diff line change
Expand Up @@ -60,3 +60,6 @@ def rejecting_candidate(

def pinning(self, candidate: CT) -> None:
"""Called when adding a candidate to the potential solution."""

def fallback_activated(self) -> None:
"""Called when falling back from backjumping to backtracking."""
148 changes: 127 additions & 21 deletions src/resolvelib/resolvers.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@
Iterable,
Mapping,
NamedTuple,
Optional,
)

from .providers import AbstractProvider
Expand Down Expand Up @@ -103,6 +104,8 @@ def __init__(
) -> None:
self._p = provider
self._r = reporter
self._fallback_active = False
self._fallback_states: Optional[list[State[RT, CT, KT]]] = None
self._states: list[State[RT, CT, KT]] = []

@property
Expand Down Expand Up @@ -269,6 +272,89 @@ def _attempt_to_pin_criterion(self, name: KT) -> list[Criterion[RT, CT]]:
# end, signal for backtracking.
return causes

def _copy_all_states(self) -> list[State]:
"""Create a copy of the all the current states"""
return [
State(
s.mapping.copy(),
s.criteria.copy(),
s.backtrack_causes[:],
)
uranusjr marked this conversation as resolved.
Show resolved Hide resolved
for s in self._states
]

def _backtrack_iteration(self) -> tuple[KT, CT, list[tuple[KT, list[CT]]]]:
"""
Pop the last state, remove the last pinned candidate, and collect
the incompatibilities for further backtracking attempts.
"""
broken_state = self._states.pop()
name, candidate = broken_state.mapping.popitem()
incompatibilities_from_broken = [
(k, list(v.incompatibilities))
for k, v in broken_state.criteria.items()
]

return name, candidate, incompatibilities_from_broken

def _backjump_iteration(
self,
causes: list[RequirementInformation[RT, CT]],
incompatible_deps: set[KT],
) -> tuple[KT, CT, list[tuple[KT, list[CT]]]]:
"""
Attempt to backjump over incompatible states, making a backup of
states for possible fallback to backtracking.
"""
# Ensure to backtrack to a state that caused the incompatibility
incompatible_state = False

if self._fallback_states is None:
fallback_states = self._copy_all_states()
else:
fallback_states = None

backjump_count = 0
while not incompatible_state:
backjump_count += 1

# Retrieve the last candidate pin and known incompatibilities
try:
broken_state = self._states.pop()
name, candidate = broken_state.mapping.popitem()
except (IndexError, KeyError):
raise ResolutionImpossible(causes)
current_dependencies = {
self._p.identify(d)
for d in self._p.get_dependencies(candidate)
}
incompatible_state = not current_dependencies.isdisjoint(
incompatible_deps
)

# Backup states first time a backjump goes further than a backtrack
if self._fallback_states is None and backjump_count == 2:
self._fallback_states = fallback_states

incompatibilities_from_broken = [
(k, list(v.incompatibilities))
for k, v in broken_state.criteria.items()
]

return name, candidate, incompatibilities_from_broken

def _activate_fallback(self):
"""Start fallback due to backjumping failure. This restores the
backup states and sets the "fallback activate" flag to change the
resolution strategy to backtracking.
"""
if self._fallback_states is None:
raise ValueError

self._states = self._fallback_states
self._fallback_active = True
self._r.fallback_activated()

def _backjump(self, causes: list[RequirementInformation[RT, CT]]) -> bool:
"""Perform backjumping.

Expand Down Expand Up @@ -299,6 +385,17 @@ def _backjump(self, causes: list[RequirementInformation[RT, CT]]) -> bool:
5a. If this causes Y' to conflict, we need to backtrack again. Make Y'
the new Z and go back to step 2.
5b. If the incompatibilities apply cleanly, end backtracking.

If backtracking each iteraction the the loop will:

1. Discard Z.
2. Discard Y but remember its incompatibility information gathered
previously, and the failure we're dealing with right now.
3. Push a new state Y' based on X, and apply the incompatibility
information from Y to Y'.
4a. If this causes Y' to conflict, we need to backtrack again. Make Y'
the new Z and go back to step 2.
4b. If the incompatibilities apply cleanly, end backtracking.
"""
incompatible_reqs: Iterable[CT | RT] = itertools.chain(
(c.parent for c in causes if c.parent is not None),
Expand All @@ -309,28 +406,37 @@ def _backjump(self, causes: list[RequirementInformation[RT, CT]]) -> bool:
# Remove the state that triggered backtracking.
del self._states[-1]

# Ensure to backtrack to a state that caused the incompatibility
incompatible_state = False
broken_state = self.state
while not incompatible_state:
# Retrieve the last candidate pin and known incompatibilities.
# When fallback not active use backjump iteration
if not self._fallback_active:
# In rare cases backjumping skips over the correct solution,
# due to the way the provider orders the resolution, so when
# backjumping produces a ResolutionImpossible exception it is
# caught and, if possible, resolution falls back the simpler
# backtracking steps, which does not try to skip over unrelated
# states
try:
broken_state = self._states.pop()
name, candidate = broken_state.mapping.popitem()
except (IndexError, KeyError):
raise ResolutionImpossible(causes) from None
current_dependencies = {
self._p.identify(d)
for d in self._p.get_dependencies(candidate)
}
incompatible_state = not current_dependencies.isdisjoint(
incompatible_deps
)

incompatibilities_from_broken = [
(k, list(v.incompatibilities))
for k, v in broken_state.criteria.items()
]
(
name,
candidate,
incompatibilities_from_broken,
) = self._backjump_iteration(
causes=causes, incompatible_deps=incompatible_deps
)
except ResolutionImpossible:
# If there are no fallback states then no backjumping
# optimization was used, so raise immediately
if self._fallback_states is None:
raise

self._activate_fallback()

# When fallback is active use the simpler backjump iteration
if self._fallback_active:
(
name,
candidate,
incompatibilities_from_broken,
) = self._backtrack_iteration()

# Also mark the newly known incompatibility.
incompatibilities_from_broken.append((name, [candidate]))
Expand Down
Loading