This is an archive of the discontinued Mercurial Phabricator instance.

Differential D628

merge: flush any deferred writes before, and after, running any workers
ClosedPublic

Authored by phillco on Sep 4 2017, 9:42 PM.

Details

Reviewers
None
Group Reviewers
hg-reviewers
Commits
rHGb90e5b2a9c82: merge: flush any deferred writes before, and after, running any workers
Summary

Since we fork to create workers, any changes they queue up will be lost after
the worker terminates, so the easiest solution is to have each worker flush
the writes they accumulate--we are close to the end of the merge in any case.

To prevent duplicated writes, we also have the master processs flush before
forking.

In an in-memory merge (M2), we'll instead disable the use of workers.

Diff Detail

Repository
rHG Mercurial
Lint
Automatic diff as part of commit; lint not applicable.
Unit
Automatic diff as part of commit; unit tests not applicable.

Event Timeline

phillco created this revision.Sep 4 2017, 9:42 PM
phillco updated this revision to Diff 1669.Sep 7 2017, 12:32 PM
phillco updated this revision to Diff 1672.Sep 7 2017, 12:58 PM
quark added a subscriber: quark.Sep 8 2017, 3:16 PM

Just a personal preference: we can add def flushall(self): pass to the other ctx and just call ctx.flushall() without testing isinstance. I guess that's slightly more readable.

phillco added a comment.Sep 8 2017, 3:19 PM

Ah, right, I forgot about that. I'll do that and send a new rev.

phillco updated this revision to Diff 1701.Sep 10 2017, 11:55 PM
phillco added a comment.Sep 10 2017, 11:55 PM

(Done on both)

Closed by commit rHGb90e5b2a9c82: merge: flush any deferred writes before, and after, running any workers (authored by phillco). · Explain WhySep 12 2017, 6:30 PM
This revision was automatically updated to reflect the committed changes.

Revision Contents
Changeset List

PathPackages
Mmercurial/merge.py (12 lines)

Diff 1770

mercurial/merge.py

except OSError as err: except OSError as err:
if err.errno != errno.ENOENT: if err.errno != errno.ENOENT:
raise raise
# Print a warning if cwd was deleted # Print a warning if cwd was deleted
repo.ui.warn(_("current directory was removed\n" repo.ui.warn(_("current directory was removed\n"
"(consider changing to repo root: %s)\n") % "(consider changing to repo root: %s)\n") %
repo.root) repo.root)
# It's necessary to flush here in case we're inside a worker fork and will
# quit after this function.
wctx.flushall()
def batchget(repo, mctx, wctx, actions): def batchget(repo, mctx, wctx, actions):
"""apply gets to the working directory """apply gets to the working directory
mctx is the context to get from mctx is the context to get from
yields tuples for progress updates yields tuples for progress updates
""" """
verbose = repo.ui.verbose verbose = repo.ui.verbose
wctx[f].write(fctx(f).data(), flags, backgroundclose=True) wctx[f].write(fctx(f).data(), flags, backgroundclose=True)
if i == 100: if i == 100:
yield i, f yield i, f
i = 0 i = 0
i += 1 i += 1
if i > 0: if i > 0:
yield i, f yield i, f
# It's necessary to flush here in case we're inside a worker fork and will
# quit after this function.
wctx.flushall()
def applyupdates(repo, actions, wctx, mctx, overwrite, labels=None): def applyupdates(repo, actions, wctx, mctx, overwrite, labels=None):
"""apply the merge action list to the working directory """apply the merge action list to the working directory
wctx is the working copy context wctx is the working copy context
mctx is the context to be merged into the working copy mctx is the context to be merged into the working copy
Return a tuple of counts (updated, merged, removed, unresolved) that Return a tuple of counts (updated, merged, removed, unresolved) that
describes how many files were affected by the update. describes how many files were affected by the update.
z = 0 z = 0
prog = worker.worker(repo.ui, 0.001, batchremove, (repo, wctx), prog = worker.worker(repo.ui, 0.001, batchremove, (repo, wctx),
actions['r']) actions['r'])
for i, item in prog: for i, item in prog:
z += i z += i
progress(_updating, z, item=item, total=numupdates, unit=_files) progress(_updating, z, item=item, total=numupdates, unit=_files)
removed = len(actions['r']) removed = len(actions['r'])
# We should flush before forking into worker processes, since those workers
# flush when they complete, and we don't want to duplicate work.
wctx.flushall()
# get in parallel # get in parallel
prog = worker.worker(repo.ui, 0.001, batchget, (repo, mctx, wctx), prog = worker.worker(repo.ui, 0.001, batchget, (repo, mctx, wctx),
actions['g']) actions['g'])
for i, item in prog: for i, item in prog:
z += i z += i
progress(_updating, z, item=item, total=numupdates, unit=_files) progress(_updating, z, item=item, total=numupdates, unit=_files)
updated = len(actions['g']) updated = len(actions['g'])