annotate mercurial/worker.py @ 44611:8fca7e8449a8 stable 5.3.2

histedit: add missing b prefix to a string If i18n is disabled (such as via HGPLAIN=1), `_()` doesn't convert from str to bytes, so this raises a TypeError on py3. Differential Revision: https://phab.mercurial-scm.org/D8354
author Kyle Lippincott <spectral@google.com>
date Wed, 01 Apr 2020 14:14:55 -0700
parents 12491abf93bd
children 26eb62bd0550
Ignore whitespace changes - Everywhere: Within whitespace: At end of lines:
rev   line source
18635
fed06dd07665 worker: count the number of CPUs
Bryan O'Sullivan <bryano@fb.com>
parents:
diff changeset
1 # worker.py - master-slave parallelism support
fed06dd07665 worker: count the number of CPUs
Bryan O'Sullivan <bryano@fb.com>
parents:
diff changeset
2 #
fed06dd07665 worker: count the number of CPUs
Bryan O'Sullivan <bryano@fb.com>
parents:
diff changeset
3 # Copyright 2013 Facebook, Inc.
fed06dd07665 worker: count the number of CPUs
Bryan O'Sullivan <bryano@fb.com>
parents:
diff changeset
4 #
fed06dd07665 worker: count the number of CPUs
Bryan O'Sullivan <bryano@fb.com>
parents:
diff changeset
5 # This software may be used and distributed according to the terms of the
fed06dd07665 worker: count the number of CPUs
Bryan O'Sullivan <bryano@fb.com>
parents:
diff changeset
6 # GNU General Public License version 2 or any later version.
fed06dd07665 worker: count the number of CPUs
Bryan O'Sullivan <bryano@fb.com>
parents:
diff changeset
7
25992
2d76f8a2d831 worker: use absolute_import
Gregory Szorc <gregory.szorc@gmail.com>
parents: 25660
diff changeset
8 from __future__ import absolute_import
2d76f8a2d831 worker: use absolute_import
Gregory Szorc <gregory.szorc@gmail.com>
parents: 25660
diff changeset
9
2d76f8a2d831 worker: use absolute_import
Gregory Szorc <gregory.szorc@gmail.com>
parents: 25660
diff changeset
10 import errno
2d76f8a2d831 worker: use absolute_import
Gregory Szorc <gregory.szorc@gmail.com>
parents: 25660
diff changeset
11 import os
2d76f8a2d831 worker: use absolute_import
Gregory Szorc <gregory.szorc@gmail.com>
parents: 25660
diff changeset
12 import signal
2d76f8a2d831 worker: use absolute_import
Gregory Szorc <gregory.szorc@gmail.com>
parents: 25660
diff changeset
13 import sys
35427
02b36e860e0b workers: implemented worker on windows
Wojciech Lis <wlis@fb.com>
parents: 34646
diff changeset
14 import threading
35432
86b8cc1f244e worker: make windows workers daemons
Wojciech Lis <wlis@fb.com>
parents: 35431
diff changeset
15 import time
25992
2d76f8a2d831 worker: use absolute_import
Gregory Szorc <gregory.szorc@gmail.com>
parents: 25660
diff changeset
16
38729
9e6afe7fca31 worker: use one pipe per posix worker and select() in parent process
Danny Hooper <hooper@google.com>
parents: 38535
diff changeset
17 try:
9e6afe7fca31 worker: use one pipe per posix worker and select() in parent process
Danny Hooper <hooper@google.com>
parents: 38535
diff changeset
18 import selectors
43076
2372284d9457 formatting: blacken the codebase
Augie Fackler <augie@google.com>
parents: 42522
diff changeset
19
38729
9e6afe7fca31 worker: use one pipe per posix worker and select() in parent process
Danny Hooper <hooper@google.com>
parents: 38535
diff changeset
20 selectors.BaseSelector
9e6afe7fca31 worker: use one pipe per posix worker and select() in parent process
Danny Hooper <hooper@google.com>
parents: 38535
diff changeset
21 except ImportError:
9e6afe7fca31 worker: use one pipe per posix worker and select() in parent process
Danny Hooper <hooper@google.com>
parents: 38535
diff changeset
22 from .thirdparty import selectors2 as selectors
9e6afe7fca31 worker: use one pipe per posix worker and select() in parent process
Danny Hooper <hooper@google.com>
parents: 38535
diff changeset
23
25992
2d76f8a2d831 worker: use absolute_import
Gregory Szorc <gregory.szorc@gmail.com>
parents: 25660
diff changeset
24 from .i18n import _
30396
78a58dcf8853 worker: migrate to util.iterfile
Jun Wu <quark@fb.com>
parents: 28292
diff changeset
25 from . import (
30635
a150173da1c1 py3: replace os.environ with encoding.environ (part 2 of 5)
Pulkit Goyal <7895pulkit@gmail.com>
parents: 30521
diff changeset
26 encoding,
30396
78a58dcf8853 worker: migrate to util.iterfile
Jun Wu <quark@fb.com>
parents: 28292
diff changeset
27 error,
30639
d524c88511a7 py3: replace os.name with pycompat.osname (part 1 of 2)
Pulkit Goyal <7895pulkit@gmail.com>
parents: 30635
diff changeset
28 pycompat,
30521
86cd09bc13ba worker: use os._exit for posix worker in all cases
Jun Wu <quark@fb.com>
parents: 30425
diff changeset
29 scmutil,
30396
78a58dcf8853 worker: migrate to util.iterfile
Jun Wu <quark@fb.com>
parents: 28292
diff changeset
30 util,
78a58dcf8853 worker: migrate to util.iterfile
Jun Wu <quark@fb.com>
parents: 28292
diff changeset
31 )
18635
fed06dd07665 worker: count the number of CPUs
Bryan O'Sullivan <bryano@fb.com>
parents:
diff changeset
32
43076
2372284d9457 formatting: blacken the codebase
Augie Fackler <augie@google.com>
parents: 42522
diff changeset
33
18635
fed06dd07665 worker: count the number of CPUs
Bryan O'Sullivan <bryano@fb.com>
parents:
diff changeset
34 def countcpus():
fed06dd07665 worker: count the number of CPUs
Bryan O'Sullivan <bryano@fb.com>
parents:
diff changeset
35 '''try to count the number of CPUs on the system'''
26568
c0501c26b05c worker: restore old countcpus code (issue4869)
Gregory Szorc <gregory.szorc@gmail.com>
parents: 26063
diff changeset
36
c0501c26b05c worker: restore old countcpus code (issue4869)
Gregory Szorc <gregory.szorc@gmail.com>
parents: 26063
diff changeset
37 # posix
18635
fed06dd07665 worker: count the number of CPUs
Bryan O'Sullivan <bryano@fb.com>
parents:
diff changeset
38 try:
43506
9f70512ae2cf cleanup: remove pointless r-prefixes on single-quoted strings
Augie Fackler <augie@google.com>
parents: 43077
diff changeset
39 n = int(os.sysconf('SC_NPROCESSORS_ONLN'))
26568
c0501c26b05c worker: restore old countcpus code (issue4869)
Gregory Szorc <gregory.szorc@gmail.com>
parents: 26063
diff changeset
40 if n > 0:
c0501c26b05c worker: restore old countcpus code (issue4869)
Gregory Szorc <gregory.szorc@gmail.com>
parents: 26063
diff changeset
41 return n
c0501c26b05c worker: restore old countcpus code (issue4869)
Gregory Szorc <gregory.szorc@gmail.com>
parents: 26063
diff changeset
42 except (AttributeError, ValueError):
c0501c26b05c worker: restore old countcpus code (issue4869)
Gregory Szorc <gregory.szorc@gmail.com>
parents: 26063
diff changeset
43 pass
c0501c26b05c worker: restore old countcpus code (issue4869)
Gregory Szorc <gregory.szorc@gmail.com>
parents: 26063
diff changeset
44
c0501c26b05c worker: restore old countcpus code (issue4869)
Gregory Szorc <gregory.szorc@gmail.com>
parents: 26063
diff changeset
45 # windows
c0501c26b05c worker: restore old countcpus code (issue4869)
Gregory Szorc <gregory.szorc@gmail.com>
parents: 26063
diff changeset
46 try:
43077
687b865b95ad formatting: byteify all mercurial/ and hgext/ string literals
Augie Fackler <augie@google.com>
parents: 43076
diff changeset
47 n = int(encoding.environ[b'NUMBER_OF_PROCESSORS'])
26568
c0501c26b05c worker: restore old countcpus code (issue4869)
Gregory Szorc <gregory.szorc@gmail.com>
parents: 26063
diff changeset
48 if n > 0:
c0501c26b05c worker: restore old countcpus code (issue4869)
Gregory Szorc <gregory.szorc@gmail.com>
parents: 26063
diff changeset
49 return n
c0501c26b05c worker: restore old countcpus code (issue4869)
Gregory Szorc <gregory.szorc@gmail.com>
parents: 26063
diff changeset
50 except (KeyError, ValueError):
c0501c26b05c worker: restore old countcpus code (issue4869)
Gregory Szorc <gregory.szorc@gmail.com>
parents: 26063
diff changeset
51 pass
c0501c26b05c worker: restore old countcpus code (issue4869)
Gregory Szorc <gregory.szorc@gmail.com>
parents: 26063
diff changeset
52
c0501c26b05c worker: restore old countcpus code (issue4869)
Gregory Szorc <gregory.szorc@gmail.com>
parents: 26063
diff changeset
53 return 1
18636
dcb27c153a40 worker: estimate whether it's worth running a task in parallel
Bryan O'Sullivan <bryano@fb.com>
parents: 18635
diff changeset
54
43076
2372284d9457 formatting: blacken the codebase
Augie Fackler <augie@google.com>
parents: 42522
diff changeset
55
18636
dcb27c153a40 worker: estimate whether it's worth running a task in parallel
Bryan O'Sullivan <bryano@fb.com>
parents: 18635
diff changeset
56 def _numworkers(ui):
43077
687b865b95ad formatting: byteify all mercurial/ and hgext/ string literals
Augie Fackler <augie@google.com>
parents: 43076
diff changeset
57 s = ui.config(b'worker', b'numcpus')
18636
dcb27c153a40 worker: estimate whether it's worth running a task in parallel
Bryan O'Sullivan <bryano@fb.com>
parents: 18635
diff changeset
58 if s:
dcb27c153a40 worker: estimate whether it's worth running a task in parallel
Bryan O'Sullivan <bryano@fb.com>
parents: 18635
diff changeset
59 try:
dcb27c153a40 worker: estimate whether it's worth running a task in parallel
Bryan O'Sullivan <bryano@fb.com>
parents: 18635
diff changeset
60 n = int(s)
dcb27c153a40 worker: estimate whether it's worth running a task in parallel
Bryan O'Sullivan <bryano@fb.com>
parents: 18635
diff changeset
61 if n >= 1:
dcb27c153a40 worker: estimate whether it's worth running a task in parallel
Bryan O'Sullivan <bryano@fb.com>
parents: 18635
diff changeset
62 return n
dcb27c153a40 worker: estimate whether it's worth running a task in parallel
Bryan O'Sullivan <bryano@fb.com>
parents: 18635
diff changeset
63 except ValueError:
43077
687b865b95ad formatting: byteify all mercurial/ and hgext/ string literals
Augie Fackler <augie@google.com>
parents: 43076
diff changeset
64 raise error.Abort(_(b'number of cpus must be an integer'))
18636
dcb27c153a40 worker: estimate whether it's worth running a task in parallel
Bryan O'Sullivan <bryano@fb.com>
parents: 18635
diff changeset
65 return min(max(countcpus(), 4), 32)
dcb27c153a40 worker: estimate whether it's worth running a task in parallel
Bryan O'Sullivan <bryano@fb.com>
parents: 18635
diff changeset
66
43076
2372284d9457 formatting: blacken the codebase
Augie Fackler <augie@google.com>
parents: 42522
diff changeset
67
44248
12491abf93bd worker: manually buffer reads from pickle stream
Jan Alexander Steffens (heftig) <jan.steffens@gmail.com>
parents: 44226
diff changeset
68 if pycompat.ispy3:
12491abf93bd worker: manually buffer reads from pickle stream
Jan Alexander Steffens (heftig) <jan.steffens@gmail.com>
parents: 44226
diff changeset
69
12491abf93bd worker: manually buffer reads from pickle stream
Jan Alexander Steffens (heftig) <jan.steffens@gmail.com>
parents: 44226
diff changeset
70 class _blockingreader(object):
12491abf93bd worker: manually buffer reads from pickle stream
Jan Alexander Steffens (heftig) <jan.steffens@gmail.com>
parents: 44226
diff changeset
71 def __init__(self, wrapped):
12491abf93bd worker: manually buffer reads from pickle stream
Jan Alexander Steffens (heftig) <jan.steffens@gmail.com>
parents: 44226
diff changeset
72 self._wrapped = wrapped
12491abf93bd worker: manually buffer reads from pickle stream
Jan Alexander Steffens (heftig) <jan.steffens@gmail.com>
parents: 44226
diff changeset
73
12491abf93bd worker: manually buffer reads from pickle stream
Jan Alexander Steffens (heftig) <jan.steffens@gmail.com>
parents: 44226
diff changeset
74 def __getattr__(self, attr):
12491abf93bd worker: manually buffer reads from pickle stream
Jan Alexander Steffens (heftig) <jan.steffens@gmail.com>
parents: 44226
diff changeset
75 return getattr(self._wrapped, attr)
12491abf93bd worker: manually buffer reads from pickle stream
Jan Alexander Steffens (heftig) <jan.steffens@gmail.com>
parents: 44226
diff changeset
76
12491abf93bd worker: manually buffer reads from pickle stream
Jan Alexander Steffens (heftig) <jan.steffens@gmail.com>
parents: 44226
diff changeset
77 # issue multiple reads until size is fulfilled
12491abf93bd worker: manually buffer reads from pickle stream
Jan Alexander Steffens (heftig) <jan.steffens@gmail.com>
parents: 44226
diff changeset
78 def read(self, size=-1):
12491abf93bd worker: manually buffer reads from pickle stream
Jan Alexander Steffens (heftig) <jan.steffens@gmail.com>
parents: 44226
diff changeset
79 if size < 0:
12491abf93bd worker: manually buffer reads from pickle stream
Jan Alexander Steffens (heftig) <jan.steffens@gmail.com>
parents: 44226
diff changeset
80 return self._wrapped.readall()
12491abf93bd worker: manually buffer reads from pickle stream
Jan Alexander Steffens (heftig) <jan.steffens@gmail.com>
parents: 44226
diff changeset
81
12491abf93bd worker: manually buffer reads from pickle stream
Jan Alexander Steffens (heftig) <jan.steffens@gmail.com>
parents: 44226
diff changeset
82 buf = bytearray(size)
12491abf93bd worker: manually buffer reads from pickle stream
Jan Alexander Steffens (heftig) <jan.steffens@gmail.com>
parents: 44226
diff changeset
83 view = memoryview(buf)
12491abf93bd worker: manually buffer reads from pickle stream
Jan Alexander Steffens (heftig) <jan.steffens@gmail.com>
parents: 44226
diff changeset
84 pos = 0
12491abf93bd worker: manually buffer reads from pickle stream
Jan Alexander Steffens (heftig) <jan.steffens@gmail.com>
parents: 44226
diff changeset
85
12491abf93bd worker: manually buffer reads from pickle stream
Jan Alexander Steffens (heftig) <jan.steffens@gmail.com>
parents: 44226
diff changeset
86 while pos < size:
12491abf93bd worker: manually buffer reads from pickle stream
Jan Alexander Steffens (heftig) <jan.steffens@gmail.com>
parents: 44226
diff changeset
87 ret = self._wrapped.readinto(view[pos:])
12491abf93bd worker: manually buffer reads from pickle stream
Jan Alexander Steffens (heftig) <jan.steffens@gmail.com>
parents: 44226
diff changeset
88 if not ret:
12491abf93bd worker: manually buffer reads from pickle stream
Jan Alexander Steffens (heftig) <jan.steffens@gmail.com>
parents: 44226
diff changeset
89 break
12491abf93bd worker: manually buffer reads from pickle stream
Jan Alexander Steffens (heftig) <jan.steffens@gmail.com>
parents: 44226
diff changeset
90 pos += ret
12491abf93bd worker: manually buffer reads from pickle stream
Jan Alexander Steffens (heftig) <jan.steffens@gmail.com>
parents: 44226
diff changeset
91
12491abf93bd worker: manually buffer reads from pickle stream
Jan Alexander Steffens (heftig) <jan.steffens@gmail.com>
parents: 44226
diff changeset
92 del view
12491abf93bd worker: manually buffer reads from pickle stream
Jan Alexander Steffens (heftig) <jan.steffens@gmail.com>
parents: 44226
diff changeset
93 del buf[pos:]
12491abf93bd worker: manually buffer reads from pickle stream
Jan Alexander Steffens (heftig) <jan.steffens@gmail.com>
parents: 44226
diff changeset
94 return buf
12491abf93bd worker: manually buffer reads from pickle stream
Jan Alexander Steffens (heftig) <jan.steffens@gmail.com>
parents: 44226
diff changeset
95
12491abf93bd worker: manually buffer reads from pickle stream
Jan Alexander Steffens (heftig) <jan.steffens@gmail.com>
parents: 44226
diff changeset
96
12491abf93bd worker: manually buffer reads from pickle stream
Jan Alexander Steffens (heftig) <jan.steffens@gmail.com>
parents: 44226
diff changeset
97 else:
12491abf93bd worker: manually buffer reads from pickle stream
Jan Alexander Steffens (heftig) <jan.steffens@gmail.com>
parents: 44226
diff changeset
98
12491abf93bd worker: manually buffer reads from pickle stream
Jan Alexander Steffens (heftig) <jan.steffens@gmail.com>
parents: 44226
diff changeset
99 def _blockingreader(wrapped):
12491abf93bd worker: manually buffer reads from pickle stream
Jan Alexander Steffens (heftig) <jan.steffens@gmail.com>
parents: 44226
diff changeset
100 return wrapped
12491abf93bd worker: manually buffer reads from pickle stream
Jan Alexander Steffens (heftig) <jan.steffens@gmail.com>
parents: 44226
diff changeset
101
12491abf93bd worker: manually buffer reads from pickle stream
Jan Alexander Steffens (heftig) <jan.steffens@gmail.com>
parents: 44226
diff changeset
102
35427
02b36e860e0b workers: implemented worker on windows
Wojciech Lis <wlis@fb.com>
parents: 34646
diff changeset
103 if pycompat.isposix or pycompat.iswindows:
38730
69ed2cff4277 worker: rename variable to reflect constant
Gregory Szorc <gregory.szorc@gmail.com>
parents: 38729
diff changeset
104 _STARTUP_COST = 0.01
38731
ef3838a47503 worker: ability to disable thread unsafe tasks
Gregory Szorc <gregory.szorc@gmail.com>
parents: 38730
diff changeset
105 # The Windows worker is thread based. If tasks are CPU bound, threads
ef3838a47503 worker: ability to disable thread unsafe tasks
Gregory Szorc <gregory.szorc@gmail.com>
parents: 38730
diff changeset
106 # in the presence of the GIL result in excessive context switching and
ef3838a47503 worker: ability to disable thread unsafe tasks
Gregory Szorc <gregory.szorc@gmail.com>
parents: 38730
diff changeset
107 # this overhead can slow down execution.
ef3838a47503 worker: ability to disable thread unsafe tasks
Gregory Szorc <gregory.szorc@gmail.com>
parents: 38730
diff changeset
108 _DISALLOW_THREAD_UNSAFE = pycompat.iswindows
18636
dcb27c153a40 worker: estimate whether it's worth running a task in parallel
Bryan O'Sullivan <bryano@fb.com>
parents: 18635
diff changeset
109 else:
38730
69ed2cff4277 worker: rename variable to reflect constant
Gregory Szorc <gregory.szorc@gmail.com>
parents: 38729
diff changeset
110 _STARTUP_COST = 1e30
38731
ef3838a47503 worker: ability to disable thread unsafe tasks
Gregory Szorc <gregory.szorc@gmail.com>
parents: 38730
diff changeset
111 _DISALLOW_THREAD_UNSAFE = False
18636
dcb27c153a40 worker: estimate whether it's worth running a task in parallel
Bryan O'Sullivan <bryano@fb.com>
parents: 18635
diff changeset
112
43076
2372284d9457 formatting: blacken the codebase
Augie Fackler <augie@google.com>
parents: 42522
diff changeset
113
38731
ef3838a47503 worker: ability to disable thread unsafe tasks
Gregory Szorc <gregory.szorc@gmail.com>
parents: 38730
diff changeset
114 def worthwhile(ui, costperop, nops, threadsafe=True):
18636
dcb27c153a40 worker: estimate whether it's worth running a task in parallel
Bryan O'Sullivan <bryano@fb.com>
parents: 18635
diff changeset
115 '''try to determine whether the benefit of multiple processes can
dcb27c153a40 worker: estimate whether it's worth running a task in parallel
Bryan O'Sullivan <bryano@fb.com>
parents: 18635
diff changeset
116 outweigh the cost of starting them'''
38731
ef3838a47503 worker: ability to disable thread unsafe tasks
Gregory Szorc <gregory.szorc@gmail.com>
parents: 38730
diff changeset
117
ef3838a47503 worker: ability to disable thread unsafe tasks
Gregory Szorc <gregory.szorc@gmail.com>
parents: 38730
diff changeset
118 if not threadsafe and _DISALLOW_THREAD_UNSAFE:
ef3838a47503 worker: ability to disable thread unsafe tasks
Gregory Szorc <gregory.szorc@gmail.com>
parents: 38730
diff changeset
119 return False
ef3838a47503 worker: ability to disable thread unsafe tasks
Gregory Szorc <gregory.szorc@gmail.com>
parents: 38730
diff changeset
120
18636
dcb27c153a40 worker: estimate whether it's worth running a task in parallel
Bryan O'Sullivan <bryano@fb.com>
parents: 18635
diff changeset
121 linear = costperop * nops
dcb27c153a40 worker: estimate whether it's worth running a task in parallel
Bryan O'Sullivan <bryano@fb.com>
parents: 18635
diff changeset
122 workers = _numworkers(ui)
38730
69ed2cff4277 worker: rename variable to reflect constant
Gregory Szorc <gregory.szorc@gmail.com>
parents: 38729
diff changeset
123 benefit = linear - (_STARTUP_COST * workers + linear / workers)
18636
dcb27c153a40 worker: estimate whether it's worth running a task in parallel
Bryan O'Sullivan <bryano@fb.com>
parents: 18635
diff changeset
124 return benefit >= 0.15
18637
ac4dbceeb14a worker: partition a list (of tasks) into equal-sized chunks
Bryan O'Sullivan <bryano@fb.com>
parents: 18636
diff changeset
125
43076
2372284d9457 formatting: blacken the codebase
Augie Fackler <augie@google.com>
parents: 42522
diff changeset
126
2372284d9457 formatting: blacken the codebase
Augie Fackler <augie@google.com>
parents: 42522
diff changeset
127 def worker(
2372284d9457 formatting: blacken the codebase
Augie Fackler <augie@google.com>
parents: 42522
diff changeset
128 ui, costperarg, func, staticargs, args, hasretval=False, threadsafe=True
2372284d9457 formatting: blacken the codebase
Augie Fackler <augie@google.com>
parents: 42522
diff changeset
129 ):
18638
047110c0e2a8 worker: allow a function to be run in multiple worker processes
Bryan O'Sullivan <bryano@fb.com>
parents: 18637
diff changeset
130 '''run a function, possibly in parallel in multiple worker
047110c0e2a8 worker: allow a function to be run in multiple worker processes
Bryan O'Sullivan <bryano@fb.com>
parents: 18637
diff changeset
131 processes.
047110c0e2a8 worker: allow a function to be run in multiple worker processes
Bryan O'Sullivan <bryano@fb.com>
parents: 18637
diff changeset
132
047110c0e2a8 worker: allow a function to be run in multiple worker processes
Bryan O'Sullivan <bryano@fb.com>
parents: 18637
diff changeset
133 returns a progress iterator
047110c0e2a8 worker: allow a function to be run in multiple worker processes
Bryan O'Sullivan <bryano@fb.com>
parents: 18637
diff changeset
134
047110c0e2a8 worker: allow a function to be run in multiple worker processes
Bryan O'Sullivan <bryano@fb.com>
parents: 18637
diff changeset
135 costperarg - cost of a single task
047110c0e2a8 worker: allow a function to be run in multiple worker processes
Bryan O'Sullivan <bryano@fb.com>
parents: 18637
diff changeset
136
42455
5ca136bbd3f6 worker: support parallelization of functions with return values
Valentin Gatien-Baron <vgatien-baron@janestreet.com>
parents: 40989
diff changeset
137 func - function to run. It is expected to return a progress iterator.
18638
047110c0e2a8 worker: allow a function to be run in multiple worker processes
Bryan O'Sullivan <bryano@fb.com>
parents: 18637
diff changeset
138
047110c0e2a8 worker: allow a function to be run in multiple worker processes
Bryan O'Sullivan <bryano@fb.com>
parents: 18637
diff changeset
139 staticargs - arguments to pass to every invocation of the function
047110c0e2a8 worker: allow a function to be run in multiple worker processes
Bryan O'Sullivan <bryano@fb.com>
parents: 18637
diff changeset
140
047110c0e2a8 worker: allow a function to be run in multiple worker processes
Bryan O'Sullivan <bryano@fb.com>
parents: 18637
diff changeset
141 args - arguments to split into chunks, to pass to individual
047110c0e2a8 worker: allow a function to be run in multiple worker processes
Bryan O'Sullivan <bryano@fb.com>
parents: 18637
diff changeset
142 workers
38731
ef3838a47503 worker: ability to disable thread unsafe tasks
Gregory Szorc <gregory.szorc@gmail.com>
parents: 38730
diff changeset
143
42455
5ca136bbd3f6 worker: support parallelization of functions with return values
Valentin Gatien-Baron <vgatien-baron@janestreet.com>
parents: 40989
diff changeset
144 hasretval - when True, func and the current function return an progress
42522
d29db0a0c4eb update: fix spurious unclean status bug shown by previous commit
Valentin Gatien-Baron <valentin.gatienbaron@gmail.com>
parents: 42455
diff changeset
145 iterator then a dict (encoded as an iterator that yield many (False, ..)
d29db0a0c4eb update: fix spurious unclean status bug shown by previous commit
Valentin Gatien-Baron <valentin.gatienbaron@gmail.com>
parents: 42455
diff changeset
146 then a (True, dict)). The dicts are joined in some arbitrary order, so
d29db0a0c4eb update: fix spurious unclean status bug shown by previous commit
Valentin Gatien-Baron <valentin.gatienbaron@gmail.com>
parents: 42455
diff changeset
147 overlapping keys are a bad idea.
42455
5ca136bbd3f6 worker: support parallelization of functions with return values
Valentin Gatien-Baron <vgatien-baron@janestreet.com>
parents: 40989
diff changeset
148
38731
ef3838a47503 worker: ability to disable thread unsafe tasks
Gregory Szorc <gregory.szorc@gmail.com>
parents: 38730
diff changeset
149 threadsafe - whether work items are thread safe and can be executed using
ef3838a47503 worker: ability to disable thread unsafe tasks
Gregory Szorc <gregory.szorc@gmail.com>
parents: 38730
diff changeset
150 a thread-based worker. Should be disabled for CPU heavy tasks that don't
ef3838a47503 worker: ability to disable thread unsafe tasks
Gregory Szorc <gregory.szorc@gmail.com>
parents: 38730
diff changeset
151 release the GIL.
18638
047110c0e2a8 worker: allow a function to be run in multiple worker processes
Bryan O'Sullivan <bryano@fb.com>
parents: 18637
diff changeset
152 '''
43077
687b865b95ad formatting: byteify all mercurial/ and hgext/ string literals
Augie Fackler <augie@google.com>
parents: 43076
diff changeset
153 enabled = ui.configbool(b'worker', b'enabled')
38731
ef3838a47503 worker: ability to disable thread unsafe tasks
Gregory Szorc <gregory.szorc@gmail.com>
parents: 38730
diff changeset
154 if enabled and worthwhile(ui, costperarg, len(args), threadsafe=threadsafe):
42455
5ca136bbd3f6 worker: support parallelization of functions with return values
Valentin Gatien-Baron <vgatien-baron@janestreet.com>
parents: 40989
diff changeset
155 return _platformworker(ui, func, staticargs, args, hasretval)
18638
047110c0e2a8 worker: allow a function to be run in multiple worker processes
Bryan O'Sullivan <bryano@fb.com>
parents: 18637
diff changeset
156 return func(*staticargs + (args,))
047110c0e2a8 worker: allow a function to be run in multiple worker processes
Bryan O'Sullivan <bryano@fb.com>
parents: 18637
diff changeset
157
43076
2372284d9457 formatting: blacken the codebase
Augie Fackler <augie@google.com>
parents: 42522
diff changeset
158
42455
5ca136bbd3f6 worker: support parallelization of functions with return values
Valentin Gatien-Baron <vgatien-baron@janestreet.com>
parents: 40989
diff changeset
159 def _posixworker(ui, func, staticargs, args, hasretval):
18638
047110c0e2a8 worker: allow a function to be run in multiple worker processes
Bryan O'Sullivan <bryano@fb.com>
parents: 18637
diff changeset
160 workers = _numworkers(ui)
18708
86524a70c0f6 worker: fix a race in SIGINT handling
Bryan O'Sullivan <bryano@fb.com>
parents: 18707
diff changeset
161 oldhandler = signal.getsignal(signal.SIGINT)
86524a70c0f6 worker: fix a race in SIGINT handling
Bryan O'Sullivan <bryano@fb.com>
parents: 18707
diff changeset
162 signal.signal(signal.SIGINT, signal.SIG_IGN)
30413
9c25a1a8c685 worker: change "pids" to a set
Jun Wu <quark@fb.com>
parents: 30412
diff changeset
163 pids, problem = set(), [0]
43076
2372284d9457 formatting: blacken the codebase
Augie Fackler <augie@google.com>
parents: 42522
diff changeset
164
30410
7a5d6e2fd2d5 worker: move killworkers and waitforworkers up
Jun Wu <quark@fb.com>
parents: 30396
diff changeset
165 def killworkers():
30423
237b2883cbd8 worker: make sure killworkers() never be interrupted by another SIGCHLD
Yuya Nishihara <yuya@tcha.org>
parents: 30422
diff changeset
166 # unregister SIGCHLD handler as all children will be killed. This
237b2883cbd8 worker: make sure killworkers() never be interrupted by another SIGCHLD
Yuya Nishihara <yuya@tcha.org>
parents: 30422
diff changeset
167 # function shouldn't be interrupted by another SIGCHLD; otherwise pids
237b2883cbd8 worker: make sure killworkers() never be interrupted by another SIGCHLD
Yuya Nishihara <yuya@tcha.org>
parents: 30422
diff changeset
168 # could be updated while iterating, which would cause inconsistency.
237b2883cbd8 worker: make sure killworkers() never be interrupted by another SIGCHLD
Yuya Nishihara <yuya@tcha.org>
parents: 30422
diff changeset
169 signal.signal(signal.SIGCHLD, oldchldhandler)
30410
7a5d6e2fd2d5 worker: move killworkers and waitforworkers up
Jun Wu <quark@fb.com>
parents: 30396
diff changeset
170 # if one worker bails, there's no good reason to wait for the rest
7a5d6e2fd2d5 worker: move killworkers and waitforworkers up
Jun Wu <quark@fb.com>
parents: 30396
diff changeset
171 for p in pids:
7a5d6e2fd2d5 worker: move killworkers and waitforworkers up
Jun Wu <quark@fb.com>
parents: 30396
diff changeset
172 try:
7a5d6e2fd2d5 worker: move killworkers and waitforworkers up
Jun Wu <quark@fb.com>
parents: 30396
diff changeset
173 os.kill(p, signal.SIGTERM)
7a5d6e2fd2d5 worker: move killworkers and waitforworkers up
Jun Wu <quark@fb.com>
parents: 30396
diff changeset
174 except OSError as err:
7a5d6e2fd2d5 worker: move killworkers and waitforworkers up
Jun Wu <quark@fb.com>
parents: 30396
diff changeset
175 if err.errno != errno.ESRCH:
7a5d6e2fd2d5 worker: move killworkers and waitforworkers up
Jun Wu <quark@fb.com>
parents: 30396
diff changeset
176 raise
43076
2372284d9457 formatting: blacken the codebase
Augie Fackler <augie@google.com>
parents: 42522
diff changeset
177
30412
7bc25549e084 worker: allow waitforworkers to be non-blocking
Jun Wu <quark@fb.com>
parents: 30411
diff changeset
178 def waitforworkers(blocking=True):
30414
5069a8a40b1b worker: make waitforworkers reentrant
Jun Wu <quark@fb.com>
parents: 30413
diff changeset
179 for pid in pids.copy():
5069a8a40b1b worker: make waitforworkers reentrant
Jun Wu <quark@fb.com>
parents: 30413
diff changeset
180 p = st = 0
5069a8a40b1b worker: make waitforworkers reentrant
Jun Wu <quark@fb.com>
parents: 30413
diff changeset
181 while True:
5069a8a40b1b worker: make waitforworkers reentrant
Jun Wu <quark@fb.com>
parents: 30413
diff changeset
182 try:
5069a8a40b1b worker: make waitforworkers reentrant
Jun Wu <quark@fb.com>
parents: 30413
diff changeset
183 p, st = os.waitpid(pid, (0 if blocking else os.WNOHANG))
30422
0e6ce6313e47 worker: fix missed break on successful waitpid()
Yuya Nishihara <yuya@tcha.org>
parents: 30416
diff changeset
184 break
30414
5069a8a40b1b worker: make waitforworkers reentrant
Jun Wu <quark@fb.com>
parents: 30413
diff changeset
185 except OSError as e:
5069a8a40b1b worker: make waitforworkers reentrant
Jun Wu <quark@fb.com>
parents: 30413
diff changeset
186 if e.errno == errno.EINTR:
5069a8a40b1b worker: make waitforworkers reentrant
Jun Wu <quark@fb.com>
parents: 30413
diff changeset
187 continue
5069a8a40b1b worker: make waitforworkers reentrant
Jun Wu <quark@fb.com>
parents: 30413
diff changeset
188 elif e.errno == errno.ECHILD:
30425
03f7aa2bd0e3 worker: discard waited pid by anyone who noticed it first
Yuya Nishihara <yuya@tcha.org>
parents: 30424
diff changeset
189 # child would already be reaped, but pids yet been
03f7aa2bd0e3 worker: discard waited pid by anyone who noticed it first
Yuya Nishihara <yuya@tcha.org>
parents: 30424
diff changeset
190 # updated (maybe interrupted just after waitpid)
03f7aa2bd0e3 worker: discard waited pid by anyone who noticed it first
Yuya Nishihara <yuya@tcha.org>
parents: 30424
diff changeset
191 pids.discard(pid)
03f7aa2bd0e3 worker: discard waited pid by anyone who noticed it first
Yuya Nishihara <yuya@tcha.org>
parents: 30424
diff changeset
192 break
30414
5069a8a40b1b worker: make waitforworkers reentrant
Jun Wu <quark@fb.com>
parents: 30413
diff changeset
193 else:
5069a8a40b1b worker: make waitforworkers reentrant
Jun Wu <quark@fb.com>
parents: 30413
diff changeset
194 raise
31063
18fb3cf572b4 worker: ignore meaningless exit status indication returned by os.waitpid()
FUJIWARA Katsunori <foozy@lares.dti.ne.jp>
parents: 30639
diff changeset
195 if not p:
18fb3cf572b4 worker: ignore meaningless exit status indication returned by os.waitpid()
FUJIWARA Katsunori <foozy@lares.dti.ne.jp>
parents: 30639
diff changeset
196 # skip subsequent steps, because child process should
18fb3cf572b4 worker: ignore meaningless exit status indication returned by os.waitpid()
FUJIWARA Katsunori <foozy@lares.dti.ne.jp>
parents: 30639
diff changeset
197 # be still running in this case
18fb3cf572b4 worker: ignore meaningless exit status indication returned by os.waitpid()
FUJIWARA Katsunori <foozy@lares.dti.ne.jp>
parents: 30639
diff changeset
198 continue
18fb3cf572b4 worker: ignore meaningless exit status indication returned by os.waitpid()
FUJIWARA Katsunori <foozy@lares.dti.ne.jp>
parents: 30639
diff changeset
199 pids.discard(p)
18fb3cf572b4 worker: ignore meaningless exit status indication returned by os.waitpid()
FUJIWARA Katsunori <foozy@lares.dti.ne.jp>
parents: 30639
diff changeset
200 st = _exitstatus(st)
30410
7a5d6e2fd2d5 worker: move killworkers and waitforworkers up
Jun Wu <quark@fb.com>
parents: 30396
diff changeset
201 if st and not problem[0]:
7a5d6e2fd2d5 worker: move killworkers and waitforworkers up
Jun Wu <quark@fb.com>
parents: 30396
diff changeset
202 problem[0] = st
43076
2372284d9457 formatting: blacken the codebase
Augie Fackler <augie@google.com>
parents: 42522
diff changeset
203
30415
e8fb03cfbbde worker: add a SIGCHLD handler to collect worker immediately
Jun Wu <quark@fb.com>
parents: 30414
diff changeset
204 def sigchldhandler(signum, frame):
e8fb03cfbbde worker: add a SIGCHLD handler to collect worker immediately
Jun Wu <quark@fb.com>
parents: 30414
diff changeset
205 waitforworkers(blocking=False)
30424
f2d13eb85198 worker: kill workers after all zombie processes are reaped
Yuya Nishihara <yuya@tcha.org>
parents: 30423
diff changeset
206 if problem[0]:
f2d13eb85198 worker: kill workers after all zombie processes are reaped
Yuya Nishihara <yuya@tcha.org>
parents: 30423
diff changeset
207 killworkers()
43076
2372284d9457 formatting: blacken the codebase
Augie Fackler <augie@google.com>
parents: 42522
diff changeset
208
30415
e8fb03cfbbde worker: add a SIGCHLD handler to collect worker immediately
Jun Wu <quark@fb.com>
parents: 30414
diff changeset
209 oldchldhandler = signal.signal(signal.SIGCHLD, sigchldhandler)
31696
9d3d56aa1a9f worker: flush ui buffers before running the worker
David Soria Parra <davidsp@fb.com>
parents: 31119
diff changeset
210 ui.flush()
32112
31763785094b worker: rewrite error handling so os._exit covers all cases
Jun Wu <quark@fb.com>
parents: 32043
diff changeset
211 parentpid = os.getpid()
38729
9e6afe7fca31 worker: use one pipe per posix worker and select() in parent process
Danny Hooper <hooper@google.com>
parents: 38535
diff changeset
212 pipes = []
42522
d29db0a0c4eb update: fix spurious unclean status bug shown by previous commit
Valentin Gatien-Baron <valentin.gatienbaron@gmail.com>
parents: 42455
diff changeset
213 retval = {}
d29db0a0c4eb update: fix spurious unclean status bug shown by previous commit
Valentin Gatien-Baron <valentin.gatienbaron@gmail.com>
parents: 42455
diff changeset
214 for pargs in partition(args, workers):
38729
9e6afe7fca31 worker: use one pipe per posix worker and select() in parent process
Danny Hooper <hooper@google.com>
parents: 38535
diff changeset
215 # Every worker gets its own pipe to send results on, so we don't have to
9e6afe7fca31 worker: use one pipe per posix worker and select() in parent process
Danny Hooper <hooper@google.com>
parents: 38535
diff changeset
216 # implement atomic writes larger than PIPE_BUF. Each forked process has
9e6afe7fca31 worker: use one pipe per posix worker and select() in parent process
Danny Hooper <hooper@google.com>
parents: 38535
diff changeset
217 # its own pipe's descriptors in the local variables, and the parent
9e6afe7fca31 worker: use one pipe per posix worker and select() in parent process
Danny Hooper <hooper@google.com>
parents: 38535
diff changeset
218 # process has the full list of pipe descriptors (and it doesn't really
9e6afe7fca31 worker: use one pipe per posix worker and select() in parent process
Danny Hooper <hooper@google.com>
parents: 38535
diff changeset
219 # care what order they're in).
9e6afe7fca31 worker: use one pipe per posix worker and select() in parent process
Danny Hooper <hooper@google.com>
parents: 38535
diff changeset
220 rfd, wfd = os.pipe()
9e6afe7fca31 worker: use one pipe per posix worker and select() in parent process
Danny Hooper <hooper@google.com>
parents: 38535
diff changeset
221 pipes.append((rfd, wfd))
32112
31763785094b worker: rewrite error handling so os._exit covers all cases
Jun Wu <quark@fb.com>
parents: 32043
diff changeset
222 # make sure we use os._exit in all worker code paths. otherwise the
31763785094b worker: rewrite error handling so os._exit covers all cases
Jun Wu <quark@fb.com>
parents: 32043
diff changeset
223 # worker may do some clean-ups which could cause surprises like
31763785094b worker: rewrite error handling so os._exit covers all cases
Jun Wu <quark@fb.com>
parents: 32043
diff changeset
224 # deadlock. see sshpeer.cleanup for example.
31763785094b worker: rewrite error handling so os._exit covers all cases
Jun Wu <quark@fb.com>
parents: 32043
diff changeset
225 # override error handling *before* fork. this is necessary because
31763785094b worker: rewrite error handling so os._exit covers all cases
Jun Wu <quark@fb.com>
parents: 32043
diff changeset
226 # exception (signal) may arrive after fork, before "pid =" assignment
31763785094b worker: rewrite error handling so os._exit covers all cases
Jun Wu <quark@fb.com>
parents: 32043
diff changeset
227 # completes, and other exception handler (dispatch.py) can lead to
31763785094b worker: rewrite error handling so os._exit covers all cases
Jun Wu <quark@fb.com>
parents: 32043
diff changeset
228 # unexpected code path without os._exit.
31763785094b worker: rewrite error handling so os._exit covers all cases
Jun Wu <quark@fb.com>
parents: 32043
diff changeset
229 ret = -1
31763785094b worker: rewrite error handling so os._exit covers all cases
Jun Wu <quark@fb.com>
parents: 32043
diff changeset
230 try:
31763785094b worker: rewrite error handling so os._exit covers all cases
Jun Wu <quark@fb.com>
parents: 32043
diff changeset
231 pid = os.fork()
31763785094b worker: rewrite error handling so os._exit covers all cases
Jun Wu <quark@fb.com>
parents: 32043
diff changeset
232 if pid == 0:
31763785094b worker: rewrite error handling so os._exit covers all cases
Jun Wu <quark@fb.com>
parents: 32043
diff changeset
233 signal.signal(signal.SIGINT, oldhandler)
31763785094b worker: rewrite error handling so os._exit covers all cases
Jun Wu <quark@fb.com>
parents: 32043
diff changeset
234 signal.signal(signal.SIGCHLD, oldchldhandler)
30521
86cd09bc13ba worker: use os._exit for posix worker in all cases
Jun Wu <quark@fb.com>
parents: 30425
diff changeset
235
32112
31763785094b worker: rewrite error handling so os._exit covers all cases
Jun Wu <quark@fb.com>
parents: 32043
diff changeset
236 def workerfunc():
38729
9e6afe7fca31 worker: use one pipe per posix worker and select() in parent process
Danny Hooper <hooper@google.com>
parents: 38535
diff changeset
237 for r, w in pipes[:-1]:
9e6afe7fca31 worker: use one pipe per posix worker and select() in parent process
Danny Hooper <hooper@google.com>
parents: 38535
diff changeset
238 os.close(r)
9e6afe7fca31 worker: use one pipe per posix worker and select() in parent process
Danny Hooper <hooper@google.com>
parents: 38535
diff changeset
239 os.close(w)
32112
31763785094b worker: rewrite error handling so os._exit covers all cases
Jun Wu <quark@fb.com>
parents: 32043
diff changeset
240 os.close(rfd)
38535
8c38d2948217 worker: support more return types in posix worker
Danny Hooper <hooper@google.com>
parents: 37844
diff changeset
241 for result in func(*(staticargs + (pargs,))):
42522
d29db0a0c4eb update: fix spurious unclean status bug shown by previous commit
Valentin Gatien-Baron <valentin.gatienbaron@gmail.com>
parents: 42455
diff changeset
242 os.write(wfd, util.pickle.dumps(result))
32112
31763785094b worker: rewrite error handling so os._exit covers all cases
Jun Wu <quark@fb.com>
parents: 32043
diff changeset
243 return 0
31763785094b worker: rewrite error handling so os._exit covers all cases
Jun Wu <quark@fb.com>
parents: 32043
diff changeset
244
31763785094b worker: rewrite error handling so os._exit covers all cases
Jun Wu <quark@fb.com>
parents: 32043
diff changeset
245 ret = scmutil.callcatch(ui, workerfunc)
43076
2372284d9457 formatting: blacken the codebase
Augie Fackler <augie@google.com>
parents: 42522
diff changeset
246 except: # parent re-raises, child never returns
32112
31763785094b worker: rewrite error handling so os._exit covers all cases
Jun Wu <quark@fb.com>
parents: 32043
diff changeset
247 if os.getpid() == parentpid:
31763785094b worker: rewrite error handling so os._exit covers all cases
Jun Wu <quark@fb.com>
parents: 32043
diff changeset
248 raise
31763785094b worker: rewrite error handling so os._exit covers all cases
Jun Wu <quark@fb.com>
parents: 32043
diff changeset
249 exctype = sys.exc_info()[0]
31763785094b worker: rewrite error handling so os._exit covers all cases
Jun Wu <quark@fb.com>
parents: 32043
diff changeset
250 force = not issubclass(exctype, KeyboardInterrupt)
31763785094b worker: rewrite error handling so os._exit covers all cases
Jun Wu <quark@fb.com>
parents: 32043
diff changeset
251 ui.traceback(force=force)
31763785094b worker: rewrite error handling so os._exit covers all cases
Jun Wu <quark@fb.com>
parents: 32043
diff changeset
252 finally:
31763785094b worker: rewrite error handling so os._exit covers all cases
Jun Wu <quark@fb.com>
parents: 32043
diff changeset
253 if os.getpid() != parentpid:
31118
a91c62752d08 worker: flush messages written by child processes before exit
Yuya Nishihara <yuya@tcha.org>
parents: 31063
diff changeset
254 try:
a91c62752d08 worker: flush messages written by child processes before exit
Yuya Nishihara <yuya@tcha.org>
parents: 31063
diff changeset
255 ui.flush()
43076
2372284d9457 formatting: blacken the codebase
Augie Fackler <augie@google.com>
parents: 42522
diff changeset
256 except: # never returns, no re-raises
32112
31763785094b worker: rewrite error handling so os._exit covers all cases
Jun Wu <quark@fb.com>
parents: 32043
diff changeset
257 pass
30521
86cd09bc13ba worker: use os._exit for posix worker in all cases
Jun Wu <quark@fb.com>
parents: 30425
diff changeset
258 finally:
32112
31763785094b worker: rewrite error handling so os._exit covers all cases
Jun Wu <quark@fb.com>
parents: 32043
diff changeset
259 os._exit(ret & 255)
30413
9c25a1a8c685 worker: change "pids" to a set
Jun Wu <quark@fb.com>
parents: 30412
diff changeset
260 pids.add(pid)
38729
9e6afe7fca31 worker: use one pipe per posix worker and select() in parent process
Danny Hooper <hooper@google.com>
parents: 38535
diff changeset
261 selector = selectors.DefaultSelector()
9e6afe7fca31 worker: use one pipe per posix worker and select() in parent process
Danny Hooper <hooper@google.com>
parents: 38535
diff changeset
262 for rfd, wfd in pipes:
9e6afe7fca31 worker: use one pipe per posix worker and select() in parent process
Danny Hooper <hooper@google.com>
parents: 38535
diff changeset
263 os.close(wfd)
44248
12491abf93bd worker: manually buffer reads from pickle stream
Jan Alexander Steffens (heftig) <jan.steffens@gmail.com>
parents: 44226
diff changeset
264 selector.register(os.fdopen(rfd, 'rb', 0), selectors.EVENT_READ)
43076
2372284d9457 formatting: blacken the codebase
Augie Fackler <augie@google.com>
parents: 42522
diff changeset
265
18638
047110c0e2a8 worker: allow a function to be run in multiple worker processes
Bryan O'Sullivan <bryano@fb.com>
parents: 18637
diff changeset
266 def cleanup():
047110c0e2a8 worker: allow a function to be run in multiple worker processes
Bryan O'Sullivan <bryano@fb.com>
parents: 18637
diff changeset
267 signal.signal(signal.SIGINT, oldhandler)
30416
c27614f2dec1 worker: stop using a separate thread waiting for children
Jun Wu <quark@fb.com>
parents: 30415
diff changeset
268 waitforworkers()
30415
e8fb03cfbbde worker: add a SIGCHLD handler to collect worker immediately
Jun Wu <quark@fb.com>
parents: 30414
diff changeset
269 signal.signal(signal.SIGCHLD, oldchldhandler)
38740
c08ea1e219c0 worker: call selector.close() to release polling resources
Yuya Nishihara <yuya@tcha.org>
parents: 38731
diff changeset
270 selector.close()
40988
03f7d0822ec1 worker: do not swallow exception occurred in main process
Yuya Nishihara <yuya@tcha.org>
parents: 38740
diff changeset
271 return problem[0]
43076
2372284d9457 formatting: blacken the codebase
Augie Fackler <augie@google.com>
parents: 42522
diff changeset
272
18638
047110c0e2a8 worker: allow a function to be run in multiple worker processes
Bryan O'Sullivan <bryano@fb.com>
parents: 18637
diff changeset
273 try:
38729
9e6afe7fca31 worker: use one pipe per posix worker and select() in parent process
Danny Hooper <hooper@google.com>
parents: 38535
diff changeset
274 openpipes = len(pipes)
9e6afe7fca31 worker: use one pipe per posix worker and select() in parent process
Danny Hooper <hooper@google.com>
parents: 38535
diff changeset
275 while openpipes > 0:
9e6afe7fca31 worker: use one pipe per posix worker and select() in parent process
Danny Hooper <hooper@google.com>
parents: 38535
diff changeset
276 for key, events in selector.select():
9e6afe7fca31 worker: use one pipe per posix worker and select() in parent process
Danny Hooper <hooper@google.com>
parents: 38535
diff changeset
277 try:
44248
12491abf93bd worker: manually buffer reads from pickle stream
Jan Alexander Steffens (heftig) <jan.steffens@gmail.com>
parents: 44226
diff changeset
278 res = util.pickle.load(_blockingreader(key.fileobj))
42455
5ca136bbd3f6 worker: support parallelization of functions with return values
Valentin Gatien-Baron <vgatien-baron@janestreet.com>
parents: 40989
diff changeset
279 if hasretval and res[0]:
42522
d29db0a0c4eb update: fix spurious unclean status bug shown by previous commit
Valentin Gatien-Baron <valentin.gatienbaron@gmail.com>
parents: 42455
diff changeset
280 retval.update(res[1])
42455
5ca136bbd3f6 worker: support parallelization of functions with return values
Valentin Gatien-Baron <vgatien-baron@janestreet.com>
parents: 40989
diff changeset
281 else:
5ca136bbd3f6 worker: support parallelization of functions with return values
Valentin Gatien-Baron <vgatien-baron@janestreet.com>
parents: 40989
diff changeset
282 yield res
38729
9e6afe7fca31 worker: use one pipe per posix worker and select() in parent process
Danny Hooper <hooper@google.com>
parents: 38535
diff changeset
283 except EOFError:
9e6afe7fca31 worker: use one pipe per posix worker and select() in parent process
Danny Hooper <hooper@google.com>
parents: 38535
diff changeset
284 selector.unregister(key.fileobj)
9e6afe7fca31 worker: use one pipe per posix worker and select() in parent process
Danny Hooper <hooper@google.com>
parents: 38535
diff changeset
285 key.fileobj.close()
9e6afe7fca31 worker: use one pipe per posix worker and select() in parent process
Danny Hooper <hooper@google.com>
parents: 38535
diff changeset
286 openpipes -= 1
9e6afe7fca31 worker: use one pipe per posix worker and select() in parent process
Danny Hooper <hooper@google.com>
parents: 38535
diff changeset
287 except IOError as e:
9e6afe7fca31 worker: use one pipe per posix worker and select() in parent process
Danny Hooper <hooper@google.com>
parents: 38535
diff changeset
288 if e.errno == errno.EINTR:
9e6afe7fca31 worker: use one pipe per posix worker and select() in parent process
Danny Hooper <hooper@google.com>
parents: 38535
diff changeset
289 continue
9e6afe7fca31 worker: use one pipe per posix worker and select() in parent process
Danny Hooper <hooper@google.com>
parents: 38535
diff changeset
290 raise
43076
2372284d9457 formatting: blacken the codebase
Augie Fackler <augie@google.com>
parents: 42522
diff changeset
291 except: # re-raises
18709
9955fc5ee24b worker: handle worker failures more aggressively
Bryan O'Sullivan <bryano@fb.com>
parents: 18708
diff changeset
292 killworkers()
18638
047110c0e2a8 worker: allow a function to be run in multiple worker processes
Bryan O'Sullivan <bryano@fb.com>
parents: 18637
diff changeset
293 cleanup()
047110c0e2a8 worker: allow a function to be run in multiple worker processes
Bryan O'Sullivan <bryano@fb.com>
parents: 18637
diff changeset
294 raise
40988
03f7d0822ec1 worker: do not swallow exception occurred in main process
Yuya Nishihara <yuya@tcha.org>
parents: 38740
diff changeset
295 status = cleanup()
03f7d0822ec1 worker: do not swallow exception occurred in main process
Yuya Nishihara <yuya@tcha.org>
parents: 38740
diff changeset
296 if status:
03f7d0822ec1 worker: do not swallow exception occurred in main process
Yuya Nishihara <yuya@tcha.org>
parents: 38740
diff changeset
297 if status < 0:
03f7d0822ec1 worker: do not swallow exception occurred in main process
Yuya Nishihara <yuya@tcha.org>
parents: 38740
diff changeset
298 os.kill(os.getpid(), -status)
03f7d0822ec1 worker: do not swallow exception occurred in main process
Yuya Nishihara <yuya@tcha.org>
parents: 38740
diff changeset
299 sys.exit(status)
42455
5ca136bbd3f6 worker: support parallelization of functions with return values
Valentin Gatien-Baron <vgatien-baron@janestreet.com>
parents: 40989
diff changeset
300 if hasretval:
42522
d29db0a0c4eb update: fix spurious unclean status bug shown by previous commit
Valentin Gatien-Baron <valentin.gatienbaron@gmail.com>
parents: 42455
diff changeset
301 yield True, retval
18638
047110c0e2a8 worker: allow a function to be run in multiple worker processes
Bryan O'Sullivan <bryano@fb.com>
parents: 18637
diff changeset
302
43076
2372284d9457 formatting: blacken the codebase
Augie Fackler <augie@google.com>
parents: 42522
diff changeset
303
18707
d1a2b086d058 worker: on error, exit similarly to the first failing worker
Bryan O'Sullivan <bryano@fb.com>
parents: 18638
diff changeset
304 def _posixexitstatus(code):
d1a2b086d058 worker: on error, exit similarly to the first failing worker
Bryan O'Sullivan <bryano@fb.com>
parents: 18638
diff changeset
305 '''convert a posix exit status into the same form returned by
d1a2b086d058 worker: on error, exit similarly to the first failing worker
Bryan O'Sullivan <bryano@fb.com>
parents: 18638
diff changeset
306 os.spawnv
d1a2b086d058 worker: on error, exit similarly to the first failing worker
Bryan O'Sullivan <bryano@fb.com>
parents: 18638
diff changeset
307
d1a2b086d058 worker: on error, exit similarly to the first failing worker
Bryan O'Sullivan <bryano@fb.com>
parents: 18638
diff changeset
308 returns None if the process was stopped instead of exiting'''
d1a2b086d058 worker: on error, exit similarly to the first failing worker
Bryan O'Sullivan <bryano@fb.com>
parents: 18638
diff changeset
309 if os.WIFEXITED(code):
d1a2b086d058 worker: on error, exit similarly to the first failing worker
Bryan O'Sullivan <bryano@fb.com>
parents: 18638
diff changeset
310 return os.WEXITSTATUS(code)
d1a2b086d058 worker: on error, exit similarly to the first failing worker
Bryan O'Sullivan <bryano@fb.com>
parents: 18638
diff changeset
311 elif os.WIFSIGNALED(code):
43076
2372284d9457 formatting: blacken the codebase
Augie Fackler <augie@google.com>
parents: 42522
diff changeset
312 return -(os.WTERMSIG(code))
2372284d9457 formatting: blacken the codebase
Augie Fackler <augie@google.com>
parents: 42522
diff changeset
313
18707
d1a2b086d058 worker: on error, exit similarly to the first failing worker
Bryan O'Sullivan <bryano@fb.com>
parents: 18638
diff changeset
314
42455
5ca136bbd3f6 worker: support parallelization of functions with return values
Valentin Gatien-Baron <vgatien-baron@janestreet.com>
parents: 40989
diff changeset
315 def _windowsworker(ui, func, staticargs, args, hasretval):
35427
02b36e860e0b workers: implemented worker on windows
Wojciech Lis <wlis@fb.com>
parents: 34646
diff changeset
316 class Worker(threading.Thread):
43076
2372284d9457 formatting: blacken the codebase
Augie Fackler <augie@google.com>
parents: 42522
diff changeset
317 def __init__(
2372284d9457 formatting: blacken the codebase
Augie Fackler <augie@google.com>
parents: 42522
diff changeset
318 self, taskqueue, resultqueue, func, staticargs, *args, **kwargs
2372284d9457 formatting: blacken the codebase
Augie Fackler <augie@google.com>
parents: 42522
diff changeset
319 ):
40443
909c31805f54 py3: roll up threading.Thread constructor args into **kwargs
Matt Harbison <matt_harbison@yahoo.com>
parents: 38740
diff changeset
320 threading.Thread.__init__(self, *args, **kwargs)
35427
02b36e860e0b workers: implemented worker on windows
Wojciech Lis <wlis@fb.com>
parents: 34646
diff changeset
321 self._taskqueue = taskqueue
02b36e860e0b workers: implemented worker on windows
Wojciech Lis <wlis@fb.com>
parents: 34646
diff changeset
322 self._resultqueue = resultqueue
02b36e860e0b workers: implemented worker on windows
Wojciech Lis <wlis@fb.com>
parents: 34646
diff changeset
323 self._func = func
02b36e860e0b workers: implemented worker on windows
Wojciech Lis <wlis@fb.com>
parents: 34646
diff changeset
324 self._staticargs = staticargs
35428
71427ff1dff8 workers: handling exceptions in windows workers
Wojciech Lis <wlis@fb.com>
parents: 35427
diff changeset
325 self._interrupted = False
35432
86b8cc1f244e worker: make windows workers daemons
Wojciech Lis <wlis@fb.com>
parents: 35431
diff changeset
326 self.daemon = True
35428
71427ff1dff8 workers: handling exceptions in windows workers
Wojciech Lis <wlis@fb.com>
parents: 35427
diff changeset
327 self.exception = None
71427ff1dff8 workers: handling exceptions in windows workers
Wojciech Lis <wlis@fb.com>
parents: 35427
diff changeset
328
71427ff1dff8 workers: handling exceptions in windows workers
Wojciech Lis <wlis@fb.com>
parents: 35427
diff changeset
329 def interrupt(self):
71427ff1dff8 workers: handling exceptions in windows workers
Wojciech Lis <wlis@fb.com>
parents: 35427
diff changeset
330 self._interrupted = True
35427
02b36e860e0b workers: implemented worker on windows
Wojciech Lis <wlis@fb.com>
parents: 34646
diff changeset
331
02b36e860e0b workers: implemented worker on windows
Wojciech Lis <wlis@fb.com>
parents: 34646
diff changeset
332 def run(self):
35428
71427ff1dff8 workers: handling exceptions in windows workers
Wojciech Lis <wlis@fb.com>
parents: 35427
diff changeset
333 try:
71427ff1dff8 workers: handling exceptions in windows workers
Wojciech Lis <wlis@fb.com>
parents: 35427
diff changeset
334 while not self._taskqueue.empty():
71427ff1dff8 workers: handling exceptions in windows workers
Wojciech Lis <wlis@fb.com>
parents: 35427
diff changeset
335 try:
42522
d29db0a0c4eb update: fix spurious unclean status bug shown by previous commit
Valentin Gatien-Baron <valentin.gatienbaron@gmail.com>
parents: 42455
diff changeset
336 args = self._taskqueue.get_nowait()
35428
71427ff1dff8 workers: handling exceptions in windows workers
Wojciech Lis <wlis@fb.com>
parents: 35427
diff changeset
337 for res in self._func(*self._staticargs + (args,)):
42522
d29db0a0c4eb update: fix spurious unclean status bug shown by previous commit
Valentin Gatien-Baron <valentin.gatienbaron@gmail.com>
parents: 42455
diff changeset
338 self._resultqueue.put(res)
35428
71427ff1dff8 workers: handling exceptions in windows workers
Wojciech Lis <wlis@fb.com>
parents: 35427
diff changeset
339 # threading doesn't provide a native way to
71427ff1dff8 workers: handling exceptions in windows workers
Wojciech Lis <wlis@fb.com>
parents: 35427
diff changeset
340 # interrupt execution. handle it manually at every
71427ff1dff8 workers: handling exceptions in windows workers
Wojciech Lis <wlis@fb.com>
parents: 35427
diff changeset
341 # iteration.
71427ff1dff8 workers: handling exceptions in windows workers
Wojciech Lis <wlis@fb.com>
parents: 35427
diff changeset
342 if self._interrupted:
71427ff1dff8 workers: handling exceptions in windows workers
Wojciech Lis <wlis@fb.com>
parents: 35427
diff changeset
343 return
37844
8fb9985382be pycompat: export queue module instead of symbols in module (API)
Gregory Szorc <gregory.szorc@gmail.com>
parents: 36835
diff changeset
344 except pycompat.queue.Empty:
35428
71427ff1dff8 workers: handling exceptions in windows workers
Wojciech Lis <wlis@fb.com>
parents: 35427
diff changeset
345 break
71427ff1dff8 workers: handling exceptions in windows workers
Wojciech Lis <wlis@fb.com>
parents: 35427
diff changeset
346 except Exception as e:
71427ff1dff8 workers: handling exceptions in windows workers
Wojciech Lis <wlis@fb.com>
parents: 35427
diff changeset
347 # store the exception such that the main thread can resurface
71427ff1dff8 workers: handling exceptions in windows workers
Wojciech Lis <wlis@fb.com>
parents: 35427
diff changeset
348 # it as if the func was running without workers.
71427ff1dff8 workers: handling exceptions in windows workers
Wojciech Lis <wlis@fb.com>
parents: 35427
diff changeset
349 self.exception = e
71427ff1dff8 workers: handling exceptions in windows workers
Wojciech Lis <wlis@fb.com>
parents: 35427
diff changeset
350 raise
71427ff1dff8 workers: handling exceptions in windows workers
Wojciech Lis <wlis@fb.com>
parents: 35427
diff changeset
351
71427ff1dff8 workers: handling exceptions in windows workers
Wojciech Lis <wlis@fb.com>
parents: 35427
diff changeset
352 threads = []
43076
2372284d9457 formatting: blacken the codebase
Augie Fackler <augie@google.com>
parents: 42522
diff changeset
353
35432
86b8cc1f244e worker: make windows workers daemons
Wojciech Lis <wlis@fb.com>
parents: 35431
diff changeset
354 def trykillworkers():
86b8cc1f244e worker: make windows workers daemons
Wojciech Lis <wlis@fb.com>
parents: 35431
diff changeset
355 # Allow up to 1 second to clean worker threads nicely
86b8cc1f244e worker: make windows workers daemons
Wojciech Lis <wlis@fb.com>
parents: 35431
diff changeset
356 cleanupend = time.time() + 1
35428
71427ff1dff8 workers: handling exceptions in windows workers
Wojciech Lis <wlis@fb.com>
parents: 35427
diff changeset
357 for t in threads:
71427ff1dff8 workers: handling exceptions in windows workers
Wojciech Lis <wlis@fb.com>
parents: 35427
diff changeset
358 t.interrupt()
71427ff1dff8 workers: handling exceptions in windows workers
Wojciech Lis <wlis@fb.com>
parents: 35427
diff changeset
359 for t in threads:
35432
86b8cc1f244e worker: make windows workers daemons
Wojciech Lis <wlis@fb.com>
parents: 35431
diff changeset
360 remainingtime = cleanupend - time.time()
86b8cc1f244e worker: make windows workers daemons
Wojciech Lis <wlis@fb.com>
parents: 35431
diff changeset
361 t.join(remainingtime)
35428
71427ff1dff8 workers: handling exceptions in windows workers
Wojciech Lis <wlis@fb.com>
parents: 35427
diff changeset
362 if t.is_alive():
35432
86b8cc1f244e worker: make windows workers daemons
Wojciech Lis <wlis@fb.com>
parents: 35431
diff changeset
363 # pass over the workers joining failure. it is more
86b8cc1f244e worker: make windows workers daemons
Wojciech Lis <wlis@fb.com>
parents: 35431
diff changeset
364 # important to surface the inital exception than the
86b8cc1f244e worker: make windows workers daemons
Wojciech Lis <wlis@fb.com>
parents: 35431
diff changeset
365 # fact that one of workers may be processing a large
86b8cc1f244e worker: make windows workers daemons
Wojciech Lis <wlis@fb.com>
parents: 35431
diff changeset
366 # task and does not get to handle the interruption.
43076
2372284d9457 formatting: blacken the codebase
Augie Fackler <augie@google.com>
parents: 42522
diff changeset
367 ui.warn(
2372284d9457 formatting: blacken the codebase
Augie Fackler <augie@google.com>
parents: 42522
diff changeset
368 _(
43077
687b865b95ad formatting: byteify all mercurial/ and hgext/ string literals
Augie Fackler <augie@google.com>
parents: 43076
diff changeset
369 b"failed to kill worker threads while "
687b865b95ad formatting: byteify all mercurial/ and hgext/ string literals
Augie Fackler <augie@google.com>
parents: 43076
diff changeset
370 b"handling an exception\n"
43076
2372284d9457 formatting: blacken the codebase
Augie Fackler <augie@google.com>
parents: 42522
diff changeset
371 )
2372284d9457 formatting: blacken the codebase
Augie Fackler <augie@google.com>
parents: 42522
diff changeset
372 )
35432
86b8cc1f244e worker: make windows workers daemons
Wojciech Lis <wlis@fb.com>
parents: 35431
diff changeset
373 return
35427
02b36e860e0b workers: implemented worker on windows
Wojciech Lis <wlis@fb.com>
parents: 34646
diff changeset
374
02b36e860e0b workers: implemented worker on windows
Wojciech Lis <wlis@fb.com>
parents: 34646
diff changeset
375 workers = _numworkers(ui)
37844
8fb9985382be pycompat: export queue module instead of symbols in module (API)
Gregory Szorc <gregory.szorc@gmail.com>
parents: 36835
diff changeset
376 resultqueue = pycompat.queue.Queue()
8fb9985382be pycompat: export queue module instead of symbols in module (API)
Gregory Szorc <gregory.szorc@gmail.com>
parents: 36835
diff changeset
377 taskqueue = pycompat.queue.Queue()
42522
d29db0a0c4eb update: fix spurious unclean status bug shown by previous commit
Valentin Gatien-Baron <valentin.gatienbaron@gmail.com>
parents: 42455
diff changeset
378 retval = {}
35427
02b36e860e0b workers: implemented worker on windows
Wojciech Lis <wlis@fb.com>
parents: 34646
diff changeset
379 # partition work to more pieces than workers to minimize the chance
02b36e860e0b workers: implemented worker on windows
Wojciech Lis <wlis@fb.com>
parents: 34646
diff changeset
380 # of uneven distribution of large tasks between the workers
42522
d29db0a0c4eb update: fix spurious unclean status bug shown by previous commit
Valentin Gatien-Baron <valentin.gatienbaron@gmail.com>
parents: 42455
diff changeset
381 for pargs in partition(args, workers * 20):
35427
02b36e860e0b workers: implemented worker on windows
Wojciech Lis <wlis@fb.com>
parents: 34646
diff changeset
382 taskqueue.put(pargs)
02b36e860e0b workers: implemented worker on windows
Wojciech Lis <wlis@fb.com>
parents: 34646
diff changeset
383 for _i in range(workers):
02b36e860e0b workers: implemented worker on windows
Wojciech Lis <wlis@fb.com>
parents: 34646
diff changeset
384 t = Worker(taskqueue, resultqueue, func, staticargs)
02b36e860e0b workers: implemented worker on windows
Wojciech Lis <wlis@fb.com>
parents: 34646
diff changeset
385 threads.append(t)
02b36e860e0b workers: implemented worker on windows
Wojciech Lis <wlis@fb.com>
parents: 34646
diff changeset
386 t.start()
35432
86b8cc1f244e worker: make windows workers daemons
Wojciech Lis <wlis@fb.com>
parents: 35431
diff changeset
387 try:
86b8cc1f244e worker: make windows workers daemons
Wojciech Lis <wlis@fb.com>
parents: 35431
diff changeset
388 while len(threads) > 0:
86b8cc1f244e worker: make windows workers daemons
Wojciech Lis <wlis@fb.com>
parents: 35431
diff changeset
389 while not resultqueue.empty():
42522
d29db0a0c4eb update: fix spurious unclean status bug shown by previous commit
Valentin Gatien-Baron <valentin.gatienbaron@gmail.com>
parents: 42455
diff changeset
390 res = resultqueue.get()
42455
5ca136bbd3f6 worker: support parallelization of functions with return values
Valentin Gatien-Baron <vgatien-baron@janestreet.com>
parents: 40989
diff changeset
391 if hasretval and res[0]:
42522
d29db0a0c4eb update: fix spurious unclean status bug shown by previous commit
Valentin Gatien-Baron <valentin.gatienbaron@gmail.com>
parents: 42455
diff changeset
392 retval.update(res[1])
42455
5ca136bbd3f6 worker: support parallelization of functions with return values
Valentin Gatien-Baron <vgatien-baron@janestreet.com>
parents: 40989
diff changeset
393 else:
5ca136bbd3f6 worker: support parallelization of functions with return values
Valentin Gatien-Baron <vgatien-baron@janestreet.com>
parents: 40989
diff changeset
394 yield res
35432
86b8cc1f244e worker: make windows workers daemons
Wojciech Lis <wlis@fb.com>
parents: 35431
diff changeset
395 threads[0].join(0.05)
86b8cc1f244e worker: make windows workers daemons
Wojciech Lis <wlis@fb.com>
parents: 35431
diff changeset
396 finishedthreads = [_t for _t in threads if not _t.is_alive()]
86b8cc1f244e worker: make windows workers daemons
Wojciech Lis <wlis@fb.com>
parents: 35431
diff changeset
397 for t in finishedthreads:
86b8cc1f244e worker: make windows workers daemons
Wojciech Lis <wlis@fb.com>
parents: 35431
diff changeset
398 if t.exception is not None:
86b8cc1f244e worker: make windows workers daemons
Wojciech Lis <wlis@fb.com>
parents: 35431
diff changeset
399 raise t.exception
86b8cc1f244e worker: make windows workers daemons
Wojciech Lis <wlis@fb.com>
parents: 35431
diff changeset
400 threads.remove(t)
43076
2372284d9457 formatting: blacken the codebase
Augie Fackler <augie@google.com>
parents: 42522
diff changeset
401 except (Exception, KeyboardInterrupt): # re-raises
35432
86b8cc1f244e worker: make windows workers daemons
Wojciech Lis <wlis@fb.com>
parents: 35431
diff changeset
402 trykillworkers()
86b8cc1f244e worker: make windows workers daemons
Wojciech Lis <wlis@fb.com>
parents: 35431
diff changeset
403 raise
35427
02b36e860e0b workers: implemented worker on windows
Wojciech Lis <wlis@fb.com>
parents: 34646
diff changeset
404 while not resultqueue.empty():
42522
d29db0a0c4eb update: fix spurious unclean status bug shown by previous commit
Valentin Gatien-Baron <valentin.gatienbaron@gmail.com>
parents: 42455
diff changeset
405 res = resultqueue.get()
42455
5ca136bbd3f6 worker: support parallelization of functions with return values
Valentin Gatien-Baron <vgatien-baron@janestreet.com>
parents: 40989
diff changeset
406 if hasretval and res[0]:
42522
d29db0a0c4eb update: fix spurious unclean status bug shown by previous commit
Valentin Gatien-Baron <valentin.gatienbaron@gmail.com>
parents: 42455
diff changeset
407 retval.update(res[1])
42455
5ca136bbd3f6 worker: support parallelization of functions with return values
Valentin Gatien-Baron <vgatien-baron@janestreet.com>
parents: 40989
diff changeset
408 else:
5ca136bbd3f6 worker: support parallelization of functions with return values
Valentin Gatien-Baron <vgatien-baron@janestreet.com>
parents: 40989
diff changeset
409 yield res
5ca136bbd3f6 worker: support parallelization of functions with return values
Valentin Gatien-Baron <vgatien-baron@janestreet.com>
parents: 40989
diff changeset
410 if hasretval:
42522
d29db0a0c4eb update: fix spurious unclean status bug shown by previous commit
Valentin Gatien-Baron <valentin.gatienbaron@gmail.com>
parents: 42455
diff changeset
411 yield True, retval
35427
02b36e860e0b workers: implemented worker on windows
Wojciech Lis <wlis@fb.com>
parents: 34646
diff changeset
412
43076
2372284d9457 formatting: blacken the codebase
Augie Fackler <augie@google.com>
parents: 42522
diff changeset
413
35427
02b36e860e0b workers: implemented worker on windows
Wojciech Lis <wlis@fb.com>
parents: 34646
diff changeset
414 if pycompat.iswindows:
02b36e860e0b workers: implemented worker on windows
Wojciech Lis <wlis@fb.com>
parents: 34646
diff changeset
415 _platformworker = _windowsworker
02b36e860e0b workers: implemented worker on windows
Wojciech Lis <wlis@fb.com>
parents: 34646
diff changeset
416 else:
18638
047110c0e2a8 worker: allow a function to be run in multiple worker processes
Bryan O'Sullivan <bryano@fb.com>
parents: 18637
diff changeset
417 _platformworker = _posixworker
18707
d1a2b086d058 worker: on error, exit similarly to the first failing worker
Bryan O'Sullivan <bryano@fb.com>
parents: 18638
diff changeset
418 _exitstatus = _posixexitstatus
18638
047110c0e2a8 worker: allow a function to be run in multiple worker processes
Bryan O'Sullivan <bryano@fb.com>
parents: 18637
diff changeset
419
43076
2372284d9457 formatting: blacken the codebase
Augie Fackler <augie@google.com>
parents: 42522
diff changeset
420
18637
ac4dbceeb14a worker: partition a list (of tasks) into equal-sized chunks
Bryan O'Sullivan <bryano@fb.com>
parents: 18636
diff changeset
421 def partition(lst, nslices):
28181
f8efc8a3a991 worker: change partition strategy to every Nth element
Gregory Szorc <gregory.szorc@gmail.com>
parents: 26587
diff changeset
422 '''partition a list into N slices of roughly equal size
f8efc8a3a991 worker: change partition strategy to every Nth element
Gregory Szorc <gregory.szorc@gmail.com>
parents: 26587
diff changeset
423
f8efc8a3a991 worker: change partition strategy to every Nth element
Gregory Szorc <gregory.szorc@gmail.com>
parents: 26587
diff changeset
424 The current strategy takes every Nth element from the input. If
f8efc8a3a991 worker: change partition strategy to every Nth element
Gregory Szorc <gregory.szorc@gmail.com>
parents: 26587
diff changeset
425 we ever write workers that need to preserve grouping in input
f8efc8a3a991 worker: change partition strategy to every Nth element
Gregory Szorc <gregory.szorc@gmail.com>
parents: 26587
diff changeset
426 we should consider allowing callers to specify a partition strategy.
28292
3eb7faf6d958 worker: document poor partitioning scheme impact
Gregory Szorc <gregory.szorc@gmail.com>
parents: 28181
diff changeset
427
3eb7faf6d958 worker: document poor partitioning scheme impact
Gregory Szorc <gregory.szorc@gmail.com>
parents: 28181
diff changeset
428 mpm is not a fan of this partitioning strategy when files are involved.
3eb7faf6d958 worker: document poor partitioning scheme impact
Gregory Szorc <gregory.szorc@gmail.com>
parents: 28181
diff changeset
429 In his words:
3eb7faf6d958 worker: document poor partitioning scheme impact
Gregory Szorc <gregory.szorc@gmail.com>
parents: 28181
diff changeset
430
3eb7faf6d958 worker: document poor partitioning scheme impact
Gregory Szorc <gregory.szorc@gmail.com>
parents: 28181
diff changeset
431 Single-threaded Mercurial makes a point of creating and visiting
3eb7faf6d958 worker: document poor partitioning scheme impact
Gregory Szorc <gregory.szorc@gmail.com>
parents: 28181
diff changeset
432 files in a fixed order (alphabetical). When creating files in order,
3eb7faf6d958 worker: document poor partitioning scheme impact
Gregory Szorc <gregory.szorc@gmail.com>
parents: 28181
diff changeset
433 a typical filesystem is likely to allocate them on nearby regions on
3eb7faf6d958 worker: document poor partitioning scheme impact
Gregory Szorc <gregory.szorc@gmail.com>
parents: 28181
diff changeset
434 disk. Thus, when revisiting in the same order, locality is maximized
3eb7faf6d958 worker: document poor partitioning scheme impact
Gregory Szorc <gregory.szorc@gmail.com>
parents: 28181
diff changeset
435 and various forms of OS and disk-level caching and read-ahead get a
3eb7faf6d958 worker: document poor partitioning scheme impact
Gregory Szorc <gregory.szorc@gmail.com>
parents: 28181
diff changeset
436 chance to work.
3eb7faf6d958 worker: document poor partitioning scheme impact
Gregory Szorc <gregory.szorc@gmail.com>
parents: 28181
diff changeset
437
3eb7faf6d958 worker: document poor partitioning scheme impact
Gregory Szorc <gregory.szorc@gmail.com>
parents: 28181
diff changeset
438 This effect can be quite significant on spinning disks. I discovered it
3eb7faf6d958 worker: document poor partitioning scheme impact
Gregory Szorc <gregory.szorc@gmail.com>
parents: 28181
diff changeset
439 circa Mercurial v0.4 when revlogs were named by hashes of filenames.
3eb7faf6d958 worker: document poor partitioning scheme impact
Gregory Szorc <gregory.szorc@gmail.com>
parents: 28181
diff changeset
440 Tarring a repo and copying it to another disk effectively randomized
3eb7faf6d958 worker: document poor partitioning scheme impact
Gregory Szorc <gregory.szorc@gmail.com>
parents: 28181
diff changeset
441 the revlog ordering on disk by sorting the revlogs by hash and suddenly
3eb7faf6d958 worker: document poor partitioning scheme impact
Gregory Szorc <gregory.szorc@gmail.com>
parents: 28181
diff changeset
442 performance of my kernel checkout benchmark dropped by ~10x because the
3eb7faf6d958 worker: document poor partitioning scheme impact
Gregory Szorc <gregory.szorc@gmail.com>
parents: 28181
diff changeset
443 "working set" of sectors visited no longer fit in the drive's cache and
3eb7faf6d958 worker: document poor partitioning scheme impact
Gregory Szorc <gregory.szorc@gmail.com>
parents: 28181
diff changeset
444 the workload switched from streaming to random I/O.
3eb7faf6d958 worker: document poor partitioning scheme impact
Gregory Szorc <gregory.szorc@gmail.com>
parents: 28181
diff changeset
445
3eb7faf6d958 worker: document poor partitioning scheme impact
Gregory Szorc <gregory.szorc@gmail.com>
parents: 28181
diff changeset
446 What we should really be doing is have workers read filenames from a
3eb7faf6d958 worker: document poor partitioning scheme impact
Gregory Szorc <gregory.szorc@gmail.com>
parents: 28181
diff changeset
447 ordered queue. This preserves locality and also keeps any worker from
3eb7faf6d958 worker: document poor partitioning scheme impact
Gregory Szorc <gregory.szorc@gmail.com>
parents: 28181
diff changeset
448 getting more than one file out of balance.
28181
f8efc8a3a991 worker: change partition strategy to every Nth element
Gregory Szorc <gregory.szorc@gmail.com>
parents: 26587
diff changeset
449 '''
f8efc8a3a991 worker: change partition strategy to every Nth element
Gregory Szorc <gregory.szorc@gmail.com>
parents: 26587
diff changeset
450 for i in range(nslices):
f8efc8a3a991 worker: change partition strategy to every Nth element
Gregory Szorc <gregory.szorc@gmail.com>
parents: 26587
diff changeset
451 yield lst[i::nslices]