diff --git a/docs/internal-fs-layout.md b/docs/internal-fs-layout.md index 7ba500a34..bd1bb3844 100644 --- a/docs/internal-fs-layout.md +++ b/docs/internal-fs-layout.md @@ -42,8 +42,11 @@ For example, if you want to change the manifest branch, you can simply run change the git URL/branch that this tracks, re-run `repo init` with the new settings. -* `.repo_fetchtimes.json`: Used by `repo sync` to record fetch times when - syncing the various projects. +* `.repo_fetchtimes.json`: Used by `repo sync` to record per-project fetch + times so subsequent syncs schedule the slowest projects first. + +* `.repo_checkouttimes.json`: Used by `repo sync` to record per-project + checkout times so subsequent syncs schedule the slowest projects first. * `.repo_localsyncstate.json`: Used by `repo sync` to detect and warn on on partial tree syncs. Partial syncs are allowed by `repo` itself, but are diff --git a/subcmds/sync.py b/subcmds/sync.py index 6f0594667..3ca6ad3b9 100644 --- a/subcmds/sync.py +++ b/subcmds/sync.py @@ -1236,7 +1236,8 @@ later is required to fix a server side protocol bug. project = projects[result.project_idx] start = result.start finish = result.finish - self._fetch_times.Set(project, finish - start) + if success: + self._fetch_times.Set(project, finish - start) self._local_sync_state.SetFetchTime(project) self.event_log.AddSync( project, @@ -1514,6 +1515,7 @@ later is required to fix a server side protocol bug. # Check for any errors before running any more tasks. # ...we'll let existing jobs finish, though. if success: + self._checkout_times.Set(project, finish - start) self._local_sync_state.SetCheckoutTime(project) else: ret = False @@ -1527,31 +1529,35 @@ later is required to fix a server side protocol bug. pm.update(msg=project.name) return ret - for projects in _SafeCheckoutOrder(all_projects): - with self.ParallelContext(): - self.get_parallel_context()["projects"] = projects - proc_res = self.ExecuteInParallel( - opt.jobs_checkout, - functools.partial( - self._CheckoutOne, - opt.detach_head, - opt.force_sync, - opt.force_checkout, - opt.rebase, - opt.verbose, - ), - range(len(projects)), - callback=_ProcessResults, - output=progress.Progress( - "Checking out", len(all_projects), quiet=opt.quiet - ), - # Use chunksize=1 to avoid the chance that some workers are - # idle while other workers still have more than one job in - # their chunk queue. - chunksize=1, - ) - - self._local_sync_state.Save() + proc_res = True + try: + for projects in _SafeCheckoutOrder(all_projects): + projects.sort(key=self._checkout_times.Get, reverse=True) + with self.ParallelContext(): + self.get_parallel_context()["projects"] = projects + proc_res = self.ExecuteInParallel( + opt.jobs_checkout, + functools.partial( + self._CheckoutOne, + opt.detach_head, + opt.force_sync, + opt.force_checkout, + opt.rebase, + opt.verbose, + ), + range(len(projects)), + callback=_ProcessResults, + output=progress.Progress( + "Checking out", len(all_projects), quiet=opt.quiet + ), + # Use chunksize=1 to avoid the chance that some workers + # are idle while other workers still have more than one + # job in their chunk queue. + chunksize=1, + ) + finally: + self._checkout_times.Save() + self._local_sync_state.Save() return proc_res and not err_results def _PrintManifestNotices(self, opt): @@ -2861,7 +2867,8 @@ later is required to fix a server side protocol bug. "repo.newprojectcount", str(len(all_projects) - len(existing)) ) - self._fetch_times = _FetchTimes(manifest) + self._fetch_times = _SyncTimes(manifest, ".repo_fetchtimes.json") + self._checkout_times = _SyncTimes(manifest, ".repo_checkouttimes.json") self._local_sync_state = LocalSyncState(manifest) self._bloated_projects = [] @@ -3395,6 +3402,15 @@ later is required to fix a server side protocol bug. return _InterleavedSyncResult(results=results) + def _GetSyncTime(self, opt: optparse.Values, project: Project) -> float: + """Estimate sync duration of |project| for the phases in |opt|.""" + total = 0.0 + if not opt.local_only: + total += self._fetch_times.Get(project) + if not opt.network_only and project.worktree: + total += self._checkout_times.Get(project) + return total + def _ProcessSyncInterleavedResults( self, finished_relpaths: Set[str], @@ -3417,11 +3433,11 @@ later is required to fix a server side protocol bug. if result.stderr_text and (opt.verbose or not success): pm.display_message(result.stderr_text) - if result.fetch_start: - self._fetch_times.Set( - project, - result.fetch_finish - result.fetch_start, - ) + if result.fetch_start is not None: + if result.fetch_success: + self._fetch_times.Set( + project, result.fetch_finish - result.fetch_start + ) self._local_sync_state.SetFetchTime(project) self.event_log.AddSync( project, @@ -3430,8 +3446,12 @@ later is required to fix a server side protocol bug. result.fetch_finish, result.fetch_success, ) - if result.checkout_start: + if result.checkout_start is not None: if result.checkout_success: + self._checkout_times.Set( + project, + result.checkout_finish - result.checkout_start, + ) self._local_sync_state.SetCheckoutTime(project) self.event_log.AddSync( project, @@ -3594,6 +3614,13 @@ later is required to fix a server side protocol bug. if not level_projects: continue + level_projects.sort( + key=functools.partial( + self._GetSyncTime, opt + ), + reverse=True, + ) + objdir_project_map = collections.defaultdict( list ) @@ -3651,6 +3678,7 @@ later is required to fix a server side protocol bug. sync_progress_thread.join() finally: self._fetch_times.Save() + self._checkout_times.Save() self._local_sync_state.Save() pm.end() @@ -3736,11 +3764,17 @@ def _PostRepoFetch(rp, repo_verify=True, verbose=False): ) -class _FetchTimes: +class _SyncTimes: + """Moving average of per-project durations for slowest-first ordering.""" + _ALPHA = 0.5 - def __init__(self, manifest): - self._path = os.path.join(manifest.repodir, ".repo_fetchtimes.json") + def __init__( + self, + manifest: manifest_xml.XmlManifest, + filename: str = ".repo_fetchtimes.json", + ) -> None: + self._path = os.path.join(manifest.repodir, filename) self._saved = None self._seen = {} diff --git a/tests/test_subcmds_sync.py b/tests/test_subcmds_sync.py index 86f7f4503..830d1c774 100644 --- a/tests/test_subcmds_sync.py +++ b/tests/test_subcmds_sync.py @@ -2363,6 +2363,9 @@ class InterleavedSyncTest(unittest.TestCase): mock.patch.object(sync.Sync, "_GetCurrentBranchOnly").start() self.cmd._fetch_times = mock.Mock() + self.cmd._fetch_times.Get.return_value = 0 + self.cmd._checkout_times = mock.Mock() + self.cmd._checkout_times.Get.return_value = 0 self.cmd._local_sync_state = mock.Mock() def tearDown(self): @@ -2912,6 +2915,289 @@ class InterleavedSyncTest(unittest.TestCase): self.mock_warm_init_template.assert_called_once() mock_seed.assert_called_once_with((("env",), False)) + def _seed_sync_times( + self, + fetch: Optional[Dict[str, float]] = None, + checkout: Optional[Dict[str, float]] = None, + ) -> Tuple[str, str]: + """Write initial timing JSON files on disk and bind _SyncTimes.""" + fetch_path = os.path.join(self.repodir, ".repo_fetchtimes.json") + checkout_path = os.path.join(self.repodir, ".repo_checkouttimes.json") + if fetch is not None: + with open(fetch_path, "w") as f: + json.dump(fetch, f) + if checkout is not None: + with open(checkout_path, "w") as f: + json.dump(checkout, f) + self.cmd._fetch_times = sync._SyncTimes( + self.manifest, ".repo_fetchtimes.json" + ) + self.cmd._checkout_times = sync._SyncTimes( + self.manifest, ".repo_checkouttimes.json" + ) + return fetch_path, checkout_path + + def _run_interleaved_for_timing( + self, + cli_args: List[str], + all_projects: List[FakeProject], + make_result: Optional[ + Callable[[int, FakeProject], sync._SyncResult] + ] = None, + expect_error: bool = False, + ) -> List[str]: + """Run _SyncInterleaved and return the dispatched project name order.""" + opt, args = self.cmd.OptionParser.parse_args(cli_args) + opt.quiet = True + mock.patch.object( + self.cmd, "GetProjects", return_value=all_projects + ).start() + + dispatched: List[str] = [] + + def default_result(idx: int, proj: FakeProject) -> sync._SyncResult: + run_fetch = not opt.local_only + run_checkout = not opt.network_only and bool(proj.worktree) + return sync._SyncResult( + project_index=idx, + relpath=proj.relpath, + fetch_success=True, + remote_fetched=run_fetch, + checkout_success=True, + fetch_errors=[], + checkout_errors=[], + stderr_text="", + fetch_start=100.0 if run_fetch else None, + fetch_finish=102.0 if run_fetch else None, + checkout_start=102.0 if run_checkout else None, + checkout_finish=106.0 if run_checkout else None, + ) + + build_result = make_result or default_result + + def execute_side_effect( + jobs: int, + target: Any, + work_items: List[List[int]], + **kwargs: Any, + ) -> bool: + projects_in_pass = self.cmd.get_parallel_context()["projects"] + results = [] + for item in work_items: + idx = item[0] + proj = projects_in_pass[idx] + dispatched.append(proj.name) + results.append( + sync._InterleavedSyncResult( + results=[build_result(idx, proj)] + ) + ) + return kwargs["callback"](None, kwargs["output"], results) + + mock.patch.object( + self.cmd, "ExecuteInParallel", side_effect=execute_side_effect + ).start() + + ctx = ( + self.assertRaises(error.SyncError) + if expect_error + else contextlib.nullcontext() + ) + with ctx: + self.cmd._SyncInterleaved( + opt, + args, + [], + self.manifest, + self.manifest.manifestProject, + all_projects, + {}, + ) + return dispatched + + def test_interleaved_full_sync_sorts_and_saves_phase_times(self) -> None: + """Full interleaved sync sorts by fetch + checkout and saves each.""" + fetch_path, checkout_path = self._seed_sync_times( + fetch={"projA": 1.0, "projB": 10.0, "projC": 2.0}, + checkout={"projA": 2.0, "projB": 1.0, "projC": 12.0}, + ) + dispatched = self._run_interleaved_for_timing( + ["--interleaved", "-j4"], + [self.projA, self.projB, self.projC], + ) + + # Totals: projC=14.0, projB=11.0, projA=3.0. + self.assertEqual(dispatched, ["projC", "projB", "projA"]) + with open(fetch_path) as f: + saved_fetch = json.load(f) + with open(checkout_path) as f: + saved_checkout = json.load(f) + # Recorded 2.0s fetch and 4.0s checkout blend separately at alpha=0.5. + self.assertAlmostEqual(saved_fetch["projA"], 1.5) + self.assertAlmostEqual(saved_checkout["projA"], 3.0) + + def test_interleaved_network_only_uses_and_updates_fetch_only( + self, + ) -> None: + """Interleaved -n sorts by fetch and leaves checkout times alone.""" + seeded_checkout = {"projA": 2.0, "projB": 1.0, "projC": 12.0} + fetch_path, checkout_path = self._seed_sync_times( + fetch={"projA": 1.0, "projB": 10.0, "projC": 2.0}, + checkout=seeded_checkout, + ) + dispatched = self._run_interleaved_for_timing( + ["--interleaved", "-n", "-j4"], + [self.projA, self.projB, self.projC], + ) + + self.assertEqual(dispatched, ["projB", "projC", "projA"]) + with open(fetch_path) as f: + saved_fetch = json.load(f) + with open(checkout_path) as f: + saved_checkout = json.load(f) + self.assertAlmostEqual(saved_fetch["projA"], 1.5) + self.assertEqual(saved_checkout, seeded_checkout) + + def test_interleaved_local_only_uses_and_updates_checkout_only( + self, + ) -> None: + """Interleaved -l sorts by checkout and leaves fetch times alone.""" + seeded_fetch = {"projA": 1.0, "projB": 10.0, "projC": 2.0} + fetch_path, checkout_path = self._seed_sync_times( + fetch=seeded_fetch, + checkout={"projA": 2.0, "projB": 1.0, "projC": 12.0}, + ) + dispatched = self._run_interleaved_for_timing( + ["--interleaved", "-l", "-j4"], + [self.projA, self.projB, self.projC], + ) + + self.assertEqual(dispatched, ["projC", "projA", "projB"]) + with open(fetch_path) as f: + saved_fetch = json.load(f) + with open(checkout_path) as f: + saved_checkout = json.load(f) + self.assertEqual(saved_fetch, seeded_fetch) + self.assertAlmostEqual(saved_checkout["projA"], 3.0) + + def test_interleaved_no_worktree_omits_checkout_estimate(self) -> None: + """Projects without a worktree do not include unknown checkout time.""" + self.projA.worktree = None + _, checkout_path = self._seed_sync_times( + fetch={"projA": 2.0, "projB": 5.0}, + checkout={"projB": 5.0}, + ) + dispatched = self._run_interleaved_for_timing( + ["--interleaved", "-j4"], + [self.projA, self.projB], + ) + + self.assertEqual(dispatched, ["projB", "projA"]) + with open(checkout_path) as f: + saved_checkout = json.load(f) + self.assertNotIn("projA", saved_checkout) + + def test_interleaved_failed_phases_do_not_skew_estimates(self) -> None: + """Failed fetches or checkouts do not overwrite saved durations.""" + self.cmd.git_event_log = mock.Mock() + seeded_checkout = {"projA": 15.0, "projB": 25.0} + fetch_path, checkout_path = self._seed_sync_times( + fetch={"projA": 20.0, "projB": 30.0}, + checkout=seeded_checkout, + ) + + def make_result(idx: int, proj: FakeProject) -> sync._SyncResult: + if proj.name == "projA": + return sync._SyncResult( + project_index=idx, + relpath=proj.relpath, + fetch_success=False, + remote_fetched=False, + checkout_success=False, + fetch_errors=[error.GitError("fetch failed")], + checkout_errors=[], + stderr_text="", + fetch_start=100.0, + fetch_finish=100.1, + checkout_start=None, + checkout_finish=None, + ) + return sync._SyncResult( + project_index=idx, + relpath=proj.relpath, + fetch_success=True, + remote_fetched=True, + checkout_success=False, + fetch_errors=[], + checkout_errors=[error.GitError("dirty worktree")], + stderr_text="", + fetch_start=100.0, + fetch_finish=102.0, + checkout_start=102.0, + checkout_finish=102.1, + ) + + self._run_interleaved_for_timing( + ["--interleaved", "-j4"], + [self.projA, self.projB], + make_result=make_result, + expect_error=True, + ) + + with open(fetch_path) as f: + saved_fetch = json.load(f) + with open(checkout_path) as f: + saved_checkout = json.load(f) + self.assertAlmostEqual(saved_fetch["projA"], 20.0) + self.assertAlmostEqual(saved_fetch["projB"], 16.0) + self.assertEqual(saved_checkout, seeded_checkout) + + def test_phased_checkout_sorts_and_saves_checkout_times(self) -> None: + """Phased _Checkout sorts each level slowest-first and saves times.""" + opt = self._get_opts(["--no-interleaved", "-j4"]) + opt.jobs_checkout = 4 + + _, checkout_path = self._seed_sync_times( + checkout={"projA": 1.0, "projB": 10.0, "projC": 5.0} + ) + + checked_out: List[str] = [] + + def execute_side_effect( + jobs: int, + target: Any, + inputs: Any, + **kwargs: Any, + ) -> bool: + projects = self.cmd.get_parallel_context()["projects"] + results = [] + for idx in inputs: + checked_out.append(projects[idx].name) + results.append( + sync._CheckoutOneResult( + success=True, + errors=[], + project_idx=idx, + start=100.0, + finish=104.0, + ) + ) + return kwargs["callback"](None, kwargs["output"], results) + + mock.patch.object( + self.cmd, "ExecuteInParallel", side_effect=execute_side_effect + ).start() + + ok = self.cmd._Checkout( + [self.projA, self.projB, self.projC], opt, [], [] + ) + + self.assertTrue(ok) + self.assertEqual(checked_out, ["projB", "projC", "projA"]) + with open(checkout_path) as f: + saved_checkout = json.load(f) + self.assertAlmostEqual(saved_checkout["projA"], 2.5) + def _get_opts(self, args=None): """Helper to get default options for worker tests.""" if args is None: