Reorganized get_queue_entry_by_names

This commit is contained in:
karajan1001
2022-07-12 21:39:49 +09:00
committed by Peter Rowlands (변기호)
parent 3d0ee0d202
commit bcc2f3b389
3 changed files with 26 additions and 29 deletions
+18 -16
View File
@@ -8,10 +8,12 @@ from typing import (
Collection,
Dict,
Generator,
Iterable,
List,
Mapping,
NamedTuple,
Optional,
Tuple,
Type,
)
@@ -556,29 +558,29 @@ class BaseStashQueue(ABC):
return results
def get_queue_entry_by_names(
def match_queue_entry_by_name(
self,
exp_names: Collection[str],
*entries: Iterable[QueueEntry],
) -> Dict[str, Optional[QueueEntry]]:
from funcy import concat
from scmrepo.exceptions import RevError as InternalRevError
exp_name_set = set(exp_names)
entry_name_dict: Dict[str, QueueEntry] = {}
entry_rev_list: List[Tuple[str, QueueEntry]] = []
for queue_entry in concat(*entries):
entry_name_dict[queue_entry.name] = queue_entry
entry_rev_list.append((queue_entry.stash_rev, queue_entry))
result: Dict[str, Optional[QueueEntry]] = {}
rev_entries = {}
for entry in concat(self.iter_queued(), self.iter_active()):
if entry.name in exp_name_set:
result[entry.name] = entry
for exp_name in exp_names:
if exp_name in entry_name_dict:
result[exp_name] = entry_name_dict[exp_name]
continue
for rev, entry in entry_rev_list:
if rev.startswith(exp_name):
result[exp_name] = entry
break
else:
rev_entries[entry.stash_rev] = entry
for exp_name in exp_name_set.difference(result.keys()):
try:
rev = self.scm.resolve_rev(exp_name)
if rev in rev_entries:
result[exp_name] = rev_entries[rev]
except InternalRevError:
result[exp_name] = None
return result
+3 -11
View File
@@ -219,18 +219,14 @@ class LocalCeleryQueue(BaseStashQueue):
def kill(self, revs: Collection[str]) -> None:
to_kill: Set[QueueEntry] = set()
not_active: List[str] = []
name_dict: Dict[
str, Optional[QueueEntry]
] = self.get_queue_entry_by_names(set(revs))
] = self.match_queue_entry_by_name(set(revs), self.iter_active())
missing_rev: List[str] = []
active_queue_entry = set(self.iter_active())
for rev, queue_entry in name_dict.items():
if queue_entry is None:
missing_rev.append(rev)
elif queue_entry not in active_queue_entry:
not_active.append(rev)
else:
to_kill.add(queue_entry)
@@ -267,15 +263,11 @@ class LocalCeleryQueue(BaseStashQueue):
):
from dvc.ui import ui
queue_entry: Optional[QueueEntry] = self.get_queue_entry_by_names(
{rev}
queue_entry: Optional[QueueEntry] = self.match_queue_entry_by_name(
{rev}, self.iter_active()
).get(rev)
if queue_entry is None:
raise UnresolvedExpNamesError([rev])
active_queue_entry = set(self.iter_active())
if queue_entry not in active_queue_entry:
ui.write("Cannot attach to an unstarted task")
return
for line in self.proc.follow(queue_entry.stash_rev, encoding):
ui.write(line)
+5 -2
View File
@@ -18,6 +18,7 @@ from .utils import (
if TYPE_CHECKING:
from dvc.repo import Repo
from dvc.repo.experiments.queue.local import LocalCeleryQueue
from dvc.scm import Git
@@ -85,8 +86,10 @@ def _resolve_exp_by_name(
commit_ref_dict[exp_ref] = exp_name
if not git_remote:
_named_entries = (
repo.experiments.celery_queue.get_queue_entry_by_names(remained)
celery_queue: "LocalCeleryQueue" = repo.experiments.celery_queue
_named_entries = celery_queue.match_queue_entry_by_name(
remained, celery_queue.iter_queued(), celery_queue.iter_active()
)
for exp_name, entry in _named_entries.items():
if entry is not None: