##// END OF EJS Templates
worker: add missing import of errno...
Bryan O'Sullivan -
r18914:05d5821e default
parent child Browse files
Show More
@@ -1,160 +1,160 b''
1 1 # worker.py - master-slave parallelism support
2 2 #
3 3 # Copyright 2013 Facebook, Inc.
4 4 #
5 5 # This software may be used and distributed according to the terms of the
6 6 # GNU General Public License version 2 or any later version.
7 7
8 8 from i18n import _
9 import os, signal, sys, threading, util
9 import errno, os, signal, sys, threading, util
10 10
11 11 def countcpus():
12 12 '''try to count the number of CPUs on the system'''
13 13
14 14 # posix
15 15 try:
16 16 n = int(os.sysconf('SC_NPROCESSORS_ONLN'))
17 17 if n > 0:
18 18 return n
19 19 except (AttributeError, ValueError):
20 20 pass
21 21
22 22 # windows
23 23 try:
24 24 n = int(os.environ['NUMBER_OF_PROCESSORS'])
25 25 if n > 0:
26 26 return n
27 27 except (KeyError, ValueError):
28 28 pass
29 29
30 30 return 1
31 31
32 32 def _numworkers(ui):
33 33 s = ui.config('worker', 'numcpus')
34 34 if s:
35 35 try:
36 36 n = int(s)
37 37 if n >= 1:
38 38 return n
39 39 except ValueError:
40 40 raise util.Abort(_('number of cpus must be an integer'))
41 41 return min(max(countcpus(), 4), 32)
42 42
43 43 if os.name == 'posix':
44 44 _startupcost = 0.01
45 45 else:
46 46 _startupcost = 1e30
47 47
48 48 def worthwhile(ui, costperop, nops):
49 49 '''try to determine whether the benefit of multiple processes can
50 50 outweigh the cost of starting them'''
51 51 linear = costperop * nops
52 52 workers = _numworkers(ui)
53 53 benefit = linear - (_startupcost * workers + linear / workers)
54 54 return benefit >= 0.15
55 55
56 56 def worker(ui, costperarg, func, staticargs, args):
57 57 '''run a function, possibly in parallel in multiple worker
58 58 processes.
59 59
60 60 returns a progress iterator
61 61
62 62 costperarg - cost of a single task
63 63
64 64 func - function to run
65 65
66 66 staticargs - arguments to pass to every invocation of the function
67 67
68 68 args - arguments to split into chunks, to pass to individual
69 69 workers
70 70 '''
71 71 if worthwhile(ui, costperarg, len(args)):
72 72 return _platformworker(ui, func, staticargs, args)
73 73 return func(*staticargs + (args,))
74 74
75 75 def _posixworker(ui, func, staticargs, args):
76 76 rfd, wfd = os.pipe()
77 77 workers = _numworkers(ui)
78 78 oldhandler = signal.getsignal(signal.SIGINT)
79 79 signal.signal(signal.SIGINT, signal.SIG_IGN)
80 80 pids, problem = [], [0]
81 81 for pargs in partition(args, workers):
82 82 pid = os.fork()
83 83 if pid == 0:
84 84 signal.signal(signal.SIGINT, oldhandler)
85 85 try:
86 86 os.close(rfd)
87 87 for i, item in func(*(staticargs + (pargs,))):
88 88 os.write(wfd, '%d %s\n' % (i, item))
89 89 os._exit(0)
90 90 except KeyboardInterrupt:
91 91 os._exit(255)
92 92 except: # re-raises (close enough for debugging anyway)
93 93 try:
94 94 ui.traceback()
95 95 finally:
96 96 os._exit(255)
97 97 pids.append(pid)
98 98 pids.reverse()
99 99 os.close(wfd)
100 100 fp = os.fdopen(rfd, 'rb', 0)
101 101 def killworkers():
102 102 # if one worker bails, there's no good reason to wait for the rest
103 103 for p in pids:
104 104 try:
105 105 os.kill(p, signal.SIGTERM)
106 106 except OSError, err:
107 107 if err.errno != errno.ESRCH:
108 108 raise
109 109 def waitforworkers():
110 110 for _ in pids:
111 111 st = _exitstatus(os.wait()[1])
112 112 if st and not problem:
113 113 problem[0] = st
114 114 killworkers()
115 115 t = threading.Thread(target=waitforworkers)
116 116 t.start()
117 117 def cleanup():
118 118 signal.signal(signal.SIGINT, oldhandler)
119 119 t.join()
120 120 status = problem[0]
121 121 if status:
122 122 if status < 0:
123 123 os.kill(os.getpid(), -status)
124 124 sys.exit(status)
125 125 try:
126 126 for line in fp:
127 127 l = line.split(' ', 1)
128 128 yield int(l[0]), l[1][:-1]
129 129 except: # re-raises
130 130 killworkers()
131 131 cleanup()
132 132 raise
133 133 cleanup()
134 134
135 135 def _posixexitstatus(code):
136 136 '''convert a posix exit status into the same form returned by
137 137 os.spawnv
138 138
139 139 returns None if the process was stopped instead of exiting'''
140 140 if os.WIFEXITED(code):
141 141 return os.WEXITSTATUS(code)
142 142 elif os.WIFSIGNALED(code):
143 143 return -os.WTERMSIG(code)
144 144
145 145 if os.name != 'nt':
146 146 _platformworker = _posixworker
147 147 _exitstatus = _posixexitstatus
148 148
149 149 def partition(lst, nslices):
150 150 '''partition a list into N slices of equal size'''
151 151 n = len(lst)
152 152 chunk, slop = n / nslices, n % nslices
153 153 end = 0
154 154 for i in xrange(nslices):
155 155 start = end
156 156 end = start + chunk
157 157 if slop:
158 158 end += 1
159 159 slop -= 1
160 160 yield lst[start:end]
General Comments 0
You need to be logged in to leave comments. Login now