OLD | NEW |
1 # Copyright 2015 The Chromium Authors. All rights reserved. | 1 # Copyright 2015 The Chromium Authors. All rights reserved. |
2 # Use of this source code is governed by a BSD-style license that can be | 2 # Use of this source code is governed by a BSD-style license that can be |
3 # found in the LICENSE file. | 3 # found in the LICENSE file. |
4 | 4 |
5 """Defines the task RPC methods.""" | 5 """Defines the task RPC methods.""" |
6 | 6 |
| 7 import logging |
7 import os | 8 import os |
8 import sys | 9 import sys |
9 import logging | |
10 import threading | 10 import threading |
11 | 11 |
12 #pylint: disable=relative-import | 12 #pylint: disable=relative-import |
13 import common_lib | 13 import process |
14 | |
15 # Map swarming_client to use subprocess42 | |
16 sys.path.append(common_lib.SWARMING_DIR) | |
17 | |
18 from utils import subprocess42 | |
19 | 14 |
20 | 15 |
21 class RPCMethods(object): | 16 class RPCMethods(object): |
22 """Class exposing RPC methods.""" | 17 """Class exposing RPC methods.""" |
23 | 18 |
24 _dotted_whitelist = ['subprocess'] | 19 _dotted_whitelist = ['subprocess'] |
25 | 20 |
26 def __init__(self, server): | 21 def __init__(self, server): |
27 self._server = server | 22 self._server = server |
28 self.subprocess = Subprocess | 23 self.subprocess = process.Process |
29 | 24 |
30 def _dispatch(self, method, params): | 25 def _dispatch(self, method, params): |
31 obj = self | 26 obj = self |
32 if '.' in method: | 27 if '.' in method: |
33 # Allow only white listed dotted names | 28 # Allow only white listed dotted names |
34 name, method = method.split('.') | 29 name, method = method.split('.') |
35 assert name in self._dotted_whitelist | 30 assert name in self._dotted_whitelist |
36 obj = getattr(self, name) | 31 obj = getattr(self, name) |
37 return getattr(obj, method)(*params) | 32 return getattr(obj, method)(*params) |
38 | 33 |
39 def Echo(self, message): | 34 def Echo(self, message): |
40 """Simple RPC method to print and return a message.""" | 35 """Simple RPC method to print and return a message.""" |
41 logging.info('Echoing %s', message) | 36 logging.info('Echoing %s', message) |
42 return 'echo %s' % str(message) | 37 return 'echo %s' % str(message) |
43 | 38 |
44 def AbsPath(self, path): | 39 def AbsPath(self, path): |
45 """Returns the absolute path.""" | 40 """Returns the absolute path.""" |
46 return os.path.abspath(path) | 41 return os.path.abspath(path) |
47 | 42 |
48 def Quit(self): | 43 def Quit(self): |
49 """Call _server.shutdown in another thread. | 44 """Call _server.shutdown in another thread. |
50 | 45 |
51 This is needed because server.shutdown waits for the server to actually | 46 This is needed because server.shutdown waits for the server to actually |
52 quit. However the server cannot shutdown until it completes handling this | 47 quit. However the server cannot shutdown until it completes handling this |
53 call. Calling this in the same thread results in a deadlock. | 48 call. Calling this in the same thread results in a deadlock. |
54 """ | 49 """ |
55 t = threading.Thread(target=self._server.shutdown) | 50 t = threading.Thread(target=self._server.shutdown) |
56 t.start() | 51 t.start() |
57 | |
58 | |
59 class Subprocess(object): | |
60 """Implements a server-based non-blocking subprocess. | |
61 | |
62 This non-blocking subprocess allows the caller to continue operating while | |
63 also able to interact with this subprocess based on a key returned to | |
64 the caller at the time of creation. | |
65 | |
66 Creation args are set via Set* methods called after calling Process but | |
67 before calling Start. This is due to a limitation of the XML-RPC | |
68 implementation not supporting keyword arguments. | |
69 """ | |
70 | |
71 _processes = {} | |
72 _process_next_id = 0 | |
73 _creation_lock = threading.Lock() | |
74 | |
75 def __init__(self, cmd): | |
76 self.stdout = '' | |
77 self.stderr = '' | |
78 self.cmd = cmd | |
79 self.proc = None | |
80 self.cwd = None | |
81 self.verbose = False | |
82 self.detached = False | |
83 self.data_lock = threading.Lock() | |
84 | |
85 def __str__(self): | |
86 return '%r, cwd=%r, verbose=%r, detached=%r' % ( | |
87 self.cmd, self.cwd, self.verbose, self.detached) | |
88 | |
89 def _reader(self): | |
90 for pipe, data in self.proc.yield_any(): | |
91 with self.data_lock: | |
92 if pipe == 'stdout': | |
93 self.stdout += data | |
94 if self.verbose: | |
95 sys.stdout.write(data) | |
96 else: | |
97 self.stderr += data | |
98 if self.verbose: | |
99 sys.stderr.write(data) | |
100 | |
101 @classmethod | |
102 def KillAll(cls): | |
103 for key in cls._processes: | |
104 cls.Kill(key) | |
105 | |
106 @classmethod | |
107 def Process(cls, cmd): | |
108 with cls._creation_lock: | |
109 key = 'Process%d' % cls._process_next_id | |
110 cls._process_next_id += 1 | |
111 logging.debug('Creating process %s', key) | |
112 process = cls(cmd) | |
113 cls._processes[key] = process | |
114 return key | |
115 | |
116 def _Start(self): | |
117 logging.info('Starting process %s', self) | |
118 self.proc = subprocess42.Popen(self.cmd, stdout=subprocess42.PIPE, | |
119 stderr=subprocess42.PIPE, | |
120 detached=self.detached, cwd=self.cwd) | |
121 threading.Thread(target=self._reader).start() | |
122 | |
123 @classmethod | |
124 def Start(cls, key): | |
125 cls._processes[key]._Start() | |
126 | |
127 @classmethod | |
128 def SetCwd(cls, key, cwd): | |
129 """Sets the process's cwd.""" | |
130 logging.debug('Setting %s cwd to %s', key, cwd) | |
131 cls._processes[key].cwd = cwd | |
132 | |
133 @classmethod | |
134 def SetDetached(cls, key): | |
135 """Creates a detached process.""" | |
136 logging.debug('Setting %s to run detached', key) | |
137 cls._processes[key].detached = True | |
138 | |
139 @classmethod | |
140 def SetVerbose(cls, key): | |
141 """Sets the stdout and stderr to be emitted locally.""" | |
142 logging.debug('Setting %s to be verbose', key) | |
143 cls._processes[key].verbose = True | |
144 | |
145 @classmethod | |
146 def Terminate(cls, key): | |
147 logging.debug('Terminating process %s', key) | |
148 cls._processes[key].proc.terminate() | |
149 | |
150 @classmethod | |
151 def Kill(cls, key): | |
152 logging.debug('Killing process %s', key) | |
153 cls._processes[key].proc.kill() | |
154 | |
155 @classmethod | |
156 def Delete(cls, key): | |
157 if cls.GetReturncode(key) is None: | |
158 logging.warning('Killing %s before deleting it', key) | |
159 cls.Kill(key) | |
160 logging.debug('Deleting process %s', key) | |
161 cls._processes.pop(key) | |
162 | |
163 @classmethod | |
164 def GetReturncode(cls, key): | |
165 return cls._processes[key].proc.returncode | |
166 | |
167 @classmethod | |
168 def ReadStdout(cls, key): | |
169 """Returns all stdout since the last call to ReadStdout. | |
170 | |
171 This call allows the user to read stdout while the process is running. | |
172 However each call will flush the local stdout buffer. In order to make | |
173 multiple calls to ReadStdout and to retain the entire output the results | |
174 of this call will need to be buffered in the calling code. | |
175 """ | |
176 proc = cls._processes[key] | |
177 with proc.data_lock: | |
178 # Perform a "read" on the stdout data | |
179 stdout = proc.stdout | |
180 proc.stdout = '' | |
181 return stdout | |
182 | |
183 @classmethod | |
184 def ReadStderr(cls, key): | |
185 """Returns all stderr read since the last call to ReadStderr. | |
186 | |
187 See ReadStdout for additional details. | |
188 """ | |
189 proc = cls._processes[key] | |
190 with proc.data_lock: | |
191 # Perform a "read" on the stderr data | |
192 stderr = proc.stderr | |
193 proc.stderr = '' | |
194 return stderr | |
195 | |
196 @classmethod | |
197 def ReadOutput(cls, key): | |
198 """Returns the (stdout, stderr) since the last Read* call. | |
199 | |
200 See ReadStdout for additional details. | |
201 """ | |
202 return cls.ReadStdout(key), cls.ReadStderr(key) | |
203 | |
204 @classmethod | |
205 def Wait(cls, key): | |
206 return cls._processes[key].proc.wait() | |
207 | |
208 @classmethod | |
209 def Poll(cls, key): | |
210 return cls._processes[key].proc.poll() | |
211 | |
212 @classmethod | |
213 def GetPid(cls, key): | |
214 return cls._processes[key].proc.pid | |
OLD | NEW |