##// END OF EJS Templates
worker: document poor partitioning scheme impact...
Gregory Szorc -
r28292:3eb7faf6 default
parent child Browse files
Show More
@@ -1,162 +1,184 b''
1 # worker.py - master-slave parallelism support
1 # worker.py - master-slave parallelism support
2 #
2 #
3 # Copyright 2013 Facebook, Inc.
3 # Copyright 2013 Facebook, Inc.
4 #
4 #
5 # This software may be used and distributed according to the terms of the
5 # This software may be used and distributed according to the terms of the
6 # GNU General Public License version 2 or any later version.
6 # GNU General Public License version 2 or any later version.
7
7
8 from __future__ import absolute_import
8 from __future__ import absolute_import
9
9
10 import errno
10 import errno
11 import os
11 import os
12 import signal
12 import signal
13 import sys
13 import sys
14 import threading
14 import threading
15
15
16 from .i18n import _
16 from .i18n import _
17 from . import error
17 from . import error
18
18
19 def countcpus():
19 def countcpus():
20 '''try to count the number of CPUs on the system'''
20 '''try to count the number of CPUs on the system'''
21
21
22 # posix
22 # posix
23 try:
23 try:
24 n = int(os.sysconf('SC_NPROCESSORS_ONLN'))
24 n = int(os.sysconf('SC_NPROCESSORS_ONLN'))
25 if n > 0:
25 if n > 0:
26 return n
26 return n
27 except (AttributeError, ValueError):
27 except (AttributeError, ValueError):
28 pass
28 pass
29
29
30 # windows
30 # windows
31 try:
31 try:
32 n = int(os.environ['NUMBER_OF_PROCESSORS'])
32 n = int(os.environ['NUMBER_OF_PROCESSORS'])
33 if n > 0:
33 if n > 0:
34 return n
34 return n
35 except (KeyError, ValueError):
35 except (KeyError, ValueError):
36 pass
36 pass
37
37
38 return 1
38 return 1
39
39
40 def _numworkers(ui):
40 def _numworkers(ui):
41 s = ui.config('worker', 'numcpus')
41 s = ui.config('worker', 'numcpus')
42 if s:
42 if s:
43 try:
43 try:
44 n = int(s)
44 n = int(s)
45 if n >= 1:
45 if n >= 1:
46 return n
46 return n
47 except ValueError:
47 except ValueError:
48 raise error.Abort(_('number of cpus must be an integer'))
48 raise error.Abort(_('number of cpus must be an integer'))
49 return min(max(countcpus(), 4), 32)
49 return min(max(countcpus(), 4), 32)
50
50
51 if os.name == 'posix':
51 if os.name == 'posix':
52 _startupcost = 0.01
52 _startupcost = 0.01
53 else:
53 else:
54 _startupcost = 1e30
54 _startupcost = 1e30
55
55
56 def worthwhile(ui, costperop, nops):
56 def worthwhile(ui, costperop, nops):
57 '''try to determine whether the benefit of multiple processes can
57 '''try to determine whether the benefit of multiple processes can
58 outweigh the cost of starting them'''
58 outweigh the cost of starting them'''
59 linear = costperop * nops
59 linear = costperop * nops
60 workers = _numworkers(ui)
60 workers = _numworkers(ui)
61 benefit = linear - (_startupcost * workers + linear / workers)
61 benefit = linear - (_startupcost * workers + linear / workers)
62 return benefit >= 0.15
62 return benefit >= 0.15
63
63
64 def worker(ui, costperarg, func, staticargs, args):
64 def worker(ui, costperarg, func, staticargs, args):
65 '''run a function, possibly in parallel in multiple worker
65 '''run a function, possibly in parallel in multiple worker
66 processes.
66 processes.
67
67
68 returns a progress iterator
68 returns a progress iterator
69
69
70 costperarg - cost of a single task
70 costperarg - cost of a single task
71
71
72 func - function to run
72 func - function to run
73
73
74 staticargs - arguments to pass to every invocation of the function
74 staticargs - arguments to pass to every invocation of the function
75
75
76 args - arguments to split into chunks, to pass to individual
76 args - arguments to split into chunks, to pass to individual
77 workers
77 workers
78 '''
78 '''
79 if worthwhile(ui, costperarg, len(args)):
79 if worthwhile(ui, costperarg, len(args)):
80 return _platformworker(ui, func, staticargs, args)
80 return _platformworker(ui, func, staticargs, args)
81 return func(*staticargs + (args,))
81 return func(*staticargs + (args,))
82
82
83 def _posixworker(ui, func, staticargs, args):
83 def _posixworker(ui, func, staticargs, args):
84 rfd, wfd = os.pipe()
84 rfd, wfd = os.pipe()
85 workers = _numworkers(ui)
85 workers = _numworkers(ui)
86 oldhandler = signal.getsignal(signal.SIGINT)
86 oldhandler = signal.getsignal(signal.SIGINT)
87 signal.signal(signal.SIGINT, signal.SIG_IGN)
87 signal.signal(signal.SIGINT, signal.SIG_IGN)
88 pids, problem = [], [0]
88 pids, problem = [], [0]
89 for pargs in partition(args, workers):
89 for pargs in partition(args, workers):
90 pid = os.fork()
90 pid = os.fork()
91 if pid == 0:
91 if pid == 0:
92 signal.signal(signal.SIGINT, oldhandler)
92 signal.signal(signal.SIGINT, oldhandler)
93 try:
93 try:
94 os.close(rfd)
94 os.close(rfd)
95 for i, item in func(*(staticargs + (pargs,))):
95 for i, item in func(*(staticargs + (pargs,))):
96 os.write(wfd, '%d %s\n' % (i, item))
96 os.write(wfd, '%d %s\n' % (i, item))
97 os._exit(0)
97 os._exit(0)
98 except KeyboardInterrupt:
98 except KeyboardInterrupt:
99 os._exit(255)
99 os._exit(255)
100 # other exceptions are allowed to propagate, we rely
100 # other exceptions are allowed to propagate, we rely
101 # on lock.py's pid checks to avoid release callbacks
101 # on lock.py's pid checks to avoid release callbacks
102 pids.append(pid)
102 pids.append(pid)
103 pids.reverse()
103 pids.reverse()
104 os.close(wfd)
104 os.close(wfd)
105 fp = os.fdopen(rfd, 'rb', 0)
105 fp = os.fdopen(rfd, 'rb', 0)
106 def killworkers():
106 def killworkers():
107 # if one worker bails, there's no good reason to wait for the rest
107 # if one worker bails, there's no good reason to wait for the rest
108 for p in pids:
108 for p in pids:
109 try:
109 try:
110 os.kill(p, signal.SIGTERM)
110 os.kill(p, signal.SIGTERM)
111 except OSError as err:
111 except OSError as err:
112 if err.errno != errno.ESRCH:
112 if err.errno != errno.ESRCH:
113 raise
113 raise
114 def waitforworkers():
114 def waitforworkers():
115 for _pid in pids:
115 for _pid in pids:
116 st = _exitstatus(os.wait()[1])
116 st = _exitstatus(os.wait()[1])
117 if st and not problem[0]:
117 if st and not problem[0]:
118 problem[0] = st
118 problem[0] = st
119 killworkers()
119 killworkers()
120 t = threading.Thread(target=waitforworkers)
120 t = threading.Thread(target=waitforworkers)
121 t.start()
121 t.start()
122 def cleanup():
122 def cleanup():
123 signal.signal(signal.SIGINT, oldhandler)
123 signal.signal(signal.SIGINT, oldhandler)
124 t.join()
124 t.join()
125 status = problem[0]
125 status = problem[0]
126 if status:
126 if status:
127 if status < 0:
127 if status < 0:
128 os.kill(os.getpid(), -status)
128 os.kill(os.getpid(), -status)
129 sys.exit(status)
129 sys.exit(status)
130 try:
130 try:
131 for line in fp:
131 for line in fp:
132 l = line.split(' ', 1)
132 l = line.split(' ', 1)
133 yield int(l[0]), l[1][:-1]
133 yield int(l[0]), l[1][:-1]
134 except: # re-raises
134 except: # re-raises
135 killworkers()
135 killworkers()
136 cleanup()
136 cleanup()
137 raise
137 raise
138 cleanup()
138 cleanup()
139
139
140 def _posixexitstatus(code):
140 def _posixexitstatus(code):
141 '''convert a posix exit status into the same form returned by
141 '''convert a posix exit status into the same form returned by
142 os.spawnv
142 os.spawnv
143
143
144 returns None if the process was stopped instead of exiting'''
144 returns None if the process was stopped instead of exiting'''
145 if os.WIFEXITED(code):
145 if os.WIFEXITED(code):
146 return os.WEXITSTATUS(code)
146 return os.WEXITSTATUS(code)
147 elif os.WIFSIGNALED(code):
147 elif os.WIFSIGNALED(code):
148 return -os.WTERMSIG(code)
148 return -os.WTERMSIG(code)
149
149
150 if os.name != 'nt':
150 if os.name != 'nt':
151 _platformworker = _posixworker
151 _platformworker = _posixworker
152 _exitstatus = _posixexitstatus
152 _exitstatus = _posixexitstatus
153
153
154 def partition(lst, nslices):
154 def partition(lst, nslices):
155 '''partition a list into N slices of roughly equal size
155 '''partition a list into N slices of roughly equal size
156
156
157 The current strategy takes every Nth element from the input. If
157 The current strategy takes every Nth element from the input. If
158 we ever write workers that need to preserve grouping in input
158 we ever write workers that need to preserve grouping in input
159 we should consider allowing callers to specify a partition strategy.
159 we should consider allowing callers to specify a partition strategy.
160
161 mpm is not a fan of this partitioning strategy when files are involved.
162 In his words:
163
164 Single-threaded Mercurial makes a point of creating and visiting
165 files in a fixed order (alphabetical). When creating files in order,
166 a typical filesystem is likely to allocate them on nearby regions on
167 disk. Thus, when revisiting in the same order, locality is maximized
168 and various forms of OS and disk-level caching and read-ahead get a
169 chance to work.
170
171 This effect can be quite significant on spinning disks. I discovered it
172 circa Mercurial v0.4 when revlogs were named by hashes of filenames.
173 Tarring a repo and copying it to another disk effectively randomized
174 the revlog ordering on disk by sorting the revlogs by hash and suddenly
175 performance of my kernel checkout benchmark dropped by ~10x because the
176 "working set" of sectors visited no longer fit in the drive's cache and
177 the workload switched from streaming to random I/O.
178
179 What we should really be doing is have workers read filenames from a
180 ordered queue. This preserves locality and also keeps any worker from
181 getting more than one file out of balance.
160 '''
182 '''
161 for i in range(nslices):
183 for i in range(nslices):
162 yield lst[i::nslices]
184 yield lst[i::nslices]
General Comments 0
You need to be logged in to leave comments. Login now