( )⚙ D3962 worker: ability to disable thread unsafe tasks

This is an archive of the discontinued Mercurial Phabricator instance.

Differential D3962

worker: ability to disable thread unsafe tasks
ClosedPublic

Authored by indygreg on Jul 17 2018, 8:43 PM.

Details

Reviewers
lothiraldan
Group Reviewers
hg-reviewers
Commits
rHGef3838a47503: worker: ability to disable thread unsafe tasks
Summary

The worker on Windows is implemented using a thread pool. If worker
tasks are not thread safe, badness can occur. In addition, if tasks
are executing CPU bound code and holding onto the GIL, there will be
non-substantial overhead in Python context switching between active
threads. This can result in significant slowdowns of tasks.

This commit teaches the code for determining whether to use a worker
to take thread safety into account. Effectively, thread unsafe tasks
don't use the thread-based worker on Windows.

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

indygreg created this revision.Jul 17 2018, 8:43 PM
indygreg edited the summary of this revision. (Show Details)Jul 18 2018, 12:55 PM
indygreg retitled this revision from worker: ability to disable worker for CPU heavy tasks to worker: ability to disable thread unsafe tasks.
indygreg updated this revision to Diff 9623.
lothiraldan accepted this revision.Jul 19 2018, 3:03 AM
yuja added a subscriber: yuja.Jul 19 2018, 8:15 AM

if pycompat.isposix or pycompat.iswindows:

_STARTUP_COST = 0.01

+ # The Windows worker is thread based. If tasks are CPU bound, threads
+ # in the presence of the GIL result in excessive context switching and
+ # this overhead can slow down execution.
+ _DISALLOW_THREAD_UNSAFE = True

Fixed this as _DISALLOW_THREAD_UNSAFE = pycompat.iswindows, and queued.
Thanks.

Closed by commit rHGef3838a47503: worker: ability to disable thread unsafe tasks (authored by indygreg). · Explain WhyJul 19 2018, 8:18 AM
This revision was automatically updated to reflect the committed changes.

Revision Contents
Changeset List

PathPackages
Mmercurial/worker.py (19 lines)
Mtests/test-simple-update.t (2 lines)

Diff 9630

mercurial/worker.py

if n >= 1: if n >= 1:
return n return n
except ValueError: except ValueError:
raise error.Abort(_('number of cpus must be an integer')) raise error.Abort(_('number of cpus must be an integer'))
return min(max(countcpus(), 4), 32) return min(max(countcpus(), 4), 32)
if pycompat.isposix or pycompat.iswindows: if pycompat.isposix or pycompat.iswindows:
_STARTUP_COST = 0.01 _STARTUP_COST = 0.01
# The Windows worker is thread based. If tasks are CPU bound, threads
# in the presence of the GIL result in excessive context switching and
# this overhead can slow down execution.
_DISALLOW_THREAD_UNSAFE = pycompat.iswindows
else: else:
_STARTUP_COST = 1e30 _STARTUP_COST = 1e30
_DISALLOW_THREAD_UNSAFE = False
def worthwhile(ui, costperop, nops): def worthwhile(ui, costperop, nops, threadsafe=True):
'''try to determine whether the benefit of multiple processes can '''try to determine whether the benefit of multiple processes can
outweigh the cost of starting them''' outweigh the cost of starting them'''
if not threadsafe and _DISALLOW_THREAD_UNSAFE:
return False
linear = costperop * nops linear = costperop * nops
workers = _numworkers(ui) workers = _numworkers(ui)
benefit = linear - (_STARTUP_COST * workers + linear / workers) benefit = linear - (_STARTUP_COST * workers + linear / workers)
return benefit >= 0.15 return benefit >= 0.15
def worker(ui, costperarg, func, staticargs, args): def worker(ui, costperarg, func, staticargs, args, threadsafe=True):
'''run a function, possibly in parallel in multiple worker '''run a function, possibly in parallel in multiple worker
processes. processes.
returns a progress iterator returns a progress iterator
costperarg - cost of a single task costperarg - cost of a single task
func - function to run func - function to run
staticargs - arguments to pass to every invocation of the function staticargs - arguments to pass to every invocation of the function
args - arguments to split into chunks, to pass to individual args - arguments to split into chunks, to pass to individual
workers workers
threadsafe - whether work items are thread safe and can be executed using
a thread-based worker. Should be disabled for CPU heavy tasks that don't
release the GIL.
''' '''
enabled = ui.configbool('worker', 'enabled') enabled = ui.configbool('worker', 'enabled')
if enabled and worthwhile(ui, costperarg, len(args)): if enabled and worthwhile(ui, costperarg, len(args), threadsafe=threadsafe):
return _platformworker(ui, func, staticargs, args) return _platformworker(ui, func, staticargs, args)
return func(*staticargs + (args,)) return func(*staticargs + (args,))
def _posixworker(ui, func, staticargs, args): def _posixworker(ui, func, staticargs, args):
workers = _numworkers(ui) workers = _numworkers(ui)
oldhandler = signal.getsignal(signal.SIGINT) oldhandler = signal.getsignal(signal.SIGINT)
signal.signal(signal.SIGINT, signal.SIG_IGN) signal.signal(signal.SIGINT, signal.SIG_IGN)
pids, problem = set(), [0] pids, problem = set(), [0]

tests/test-simple-update.t

$ cd .. $ cd ..
update with worker processes update with worker processes
#if no-windows #if no-windows
$ cat <<EOF > forceworker.py $ cat <<EOF > forceworker.py
> from mercurial import extensions, worker > from mercurial import extensions, worker
> def nocost(orig, ui, costperop, nops): > def nocost(orig, ui, costperop, nops, threadsafe=True):
> return worker._numworkers(ui) > 1 > return worker._numworkers(ui) > 1
> def uisetup(ui): > def uisetup(ui):
> extensions.wrapfunction(worker, 'worthwhile', nocost) > extensions.wrapfunction(worker, 'worthwhile', nocost)
> EOF > EOF
$ hg init worker $ hg init worker
$ cd worker $ cd worker
$ cat <<EOF >> .hg/hgrc $ cat <<EOF >> .hg/hgrc