summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
authorRichard Purdie <richard.purdie@linuxfoundation.org>2013-08-31 22:40:55 (GMT)
committerRichard Purdie <richard.purdie@linuxfoundation.org>2013-09-01 14:51:10 (GMT)
commitb306d7d9a4dd5dec0ed1ca91b50569f078ac103e (patch)
tree7c76ce19c1a731316ae58f1c2b270389673f2a55
parentd63e6a925a78f0f5fb7e7329b5cc2993f1ec9b14 (diff)
downloadpoky-b306d7d9a4dd5dec0ed1ca91b50569f078ac103e.tar.gz
bitbake: server/process, server/xmlrpc, runqueue: Use select.select() on fds, not time.sleep()
The existing backend server implementations were inefficient since they were sleeping for the full length of the timeouts rather than being woken when there was data ready for them. It was assumed they would wake and perhaps did when we forked processes directory but that is no longer the case. This updates both the process and xmlrpc backends to wait using select(). This does mean we need to pass the file descriptors to wait on from the internals who know which these file descriptors are but this is a logical improvement. Tests of a pathaolgical load on the process server of ~420 rapid tasks executed on a server with BB_NUMBER_THREAD=48 went from a wall clock measurement of the overall command execution time of 75s to a much more reasonable 24s. (Bitbake rev: 9bee497960889d9baa0a4284d79a384b18a8e826) Signed-off-by: Richard Purdie <richard.purdie@linuxfoundation.org>
-rw-r--r--bitbake/lib/bb/runqueue.py12
-rw-r--r--bitbake/lib/bb/server/process.py11
-rw-r--r--bitbake/lib/bb/server/xmlrpc.py26
3 files changed, 28 insertions, 21 deletions
diff --git a/bitbake/lib/bb/runqueue.py b/bitbake/lib/bb/runqueue.py
index 075c849..197308f 100644
--- a/bitbake/lib/bb/runqueue.py
+++ b/bitbake/lib/bb/runqueue.py
@@ -895,6 +895,14 @@ class RunQueue:
895 if self.fakeworkerpipe: 895 if self.fakeworkerpipe:
896 self.fakeworkerpipe.read() 896 self.fakeworkerpipe.read()
897 897
898 def active_fds(self):
899 fds = []
900 if self.workerpipe:
901 fds.append(self.workerpipe.input)
902 if self.fakeworkerpipe:
903 fds.append(self.fakeworkerpipe.input)
904 return fds
905
898 def check_stamp_task(self, task, taskname = None, recurse = False, cache = None): 906 def check_stamp_task(self, task, taskname = None, recurse = False, cache = None):
899 def get_timestamp(f): 907 def get_timestamp(f):
900 try: 908 try:
@@ -972,7 +980,7 @@ class RunQueue:
972 (if the abort on failure configuration option isn't set) 980 (if the abort on failure configuration option isn't set)
973 """ 981 """
974 982
975 retval = 0.5 983 retval = True
976 984
977 if self.state is runQueuePrepare: 985 if self.state is runQueuePrepare:
978 self.rqexe = RunQueueExecuteDummy(self) 986 self.rqexe = RunQueueExecuteDummy(self)
@@ -1375,7 +1383,7 @@ class RunQueueExecuteTasks(RunQueueExecute):
1375 1383
1376 if self.stats.active > 0: 1384 if self.stats.active > 0:
1377 self.rq.read_workers() 1385 self.rq.read_workers()
1378 return 0.5 1386 return self.rq.active_fds()
1379 1387
1380 if len(self.failed_fnids) != 0: 1388 if len(self.failed_fnids) != 0:
1381 self.rq.state = runQueueFailed 1389 self.rq.state = runQueueFailed
diff --git a/bitbake/lib/bb/server/process.py b/bitbake/lib/bb/server/process.py
index e2cec49..c0af052 100644
--- a/bitbake/lib/bb/server/process.py
+++ b/bitbake/lib/bb/server/process.py
@@ -29,6 +29,7 @@ import os
29import signal 29import signal
30import sys 30import sys
31import time 31import time
32import select
32from Queue import Empty 33from Queue import Empty
33from multiprocessing import Event, Process, util, Queue, Pipe, queues 34from multiprocessing import Event, Process, util, Queue, Pipe, queues
34 35
@@ -105,7 +106,7 @@ class ProcessServer(Process, BaseImplServer):
105 command = self.command_channel.recv() 106 command = self.command_channel.recv()
106 self.runCommand(command) 107 self.runCommand(command)
107 108
108 self.idle_commands(.1) 109 self.idle_commands(.1, [self.event_queue._reader, self.command_channel])
109 except Exception: 110 except Exception:
110 logger.exception('Running command %s', command) 111 logger.exception('Running command %s', command)
111 112
@@ -115,7 +116,7 @@ class ProcessServer(Process, BaseImplServer):
115 self.cooker.stop() 116 self.cooker.stop()
116 self.idle_commands(.1) 117 self.idle_commands(.1)
117 118
118 def idle_commands(self, delay): 119 def idle_commands(self, delay, fds = []):
119 nextsleep = delay 120 nextsleep = delay
120 121
121 for function, data in self._idlefuns.items(): 122 for function, data in self._idlefuns.items():
@@ -127,15 +128,15 @@ class ProcessServer(Process, BaseImplServer):
127 nextsleep = None 128 nextsleep = None
128 elif nextsleep is None: 129 elif nextsleep is None:
129 continue 130 continue
130 elif retval < nextsleep: 131 else:
131 nextsleep = retval 132 fds = fds + retval
132 except SystemExit: 133 except SystemExit:
133 raise 134 raise
134 except Exception: 135 except Exception:
135 logger.exception('Running idle function') 136 logger.exception('Running idle function')
136 137
137 if nextsleep is not None: 138 if nextsleep is not None:
138 time.sleep(nextsleep) 139 select.select(fds,[],[],nextsleep)
139 140
140 def runCommand(self, command): 141 def runCommand(self, command):
141 """ 142 """
diff --git a/bitbake/lib/bb/server/xmlrpc.py b/bitbake/lib/bb/server/xmlrpc.py
index 641e15e..cca569d 100644
--- a/bitbake/lib/bb/server/xmlrpc.py
+++ b/bitbake/lib/bb/server/xmlrpc.py
@@ -264,12 +264,9 @@ class XMLRPCServer(SimpleXMLRPCServer, BaseImplServer):
264 Serve Requests. Overloaded to honor a quit command 264 Serve Requests. Overloaded to honor a quit command
265 """ 265 """
266 self.quit = False 266 self.quit = False
267 self.timeout = 0 # Run Idle calls for our first callback
268 while not self.quit: 267 while not self.quit:
269 #print "Idle queue length %s" % len(self._idlefuns) 268 fds = [self]
270 self.handle_request() 269 nextsleep = 0.1
271 #print "Idle timeout, running idle functions"
272 nextsleep = None
273 for function, data in self._idlefuns.items(): 270 for function, data in self._idlefuns.items():
274 try: 271 try:
275 retval = function(self, data, False) 272 retval = function(self, data, False)
@@ -277,21 +274,22 @@ class XMLRPCServer(SimpleXMLRPCServer, BaseImplServer):
277 del self._idlefuns[function] 274 del self._idlefuns[function]
278 elif retval is True: 275 elif retval is True:
279 nextsleep = 0 276 nextsleep = 0
280 elif nextsleep is 0: 277 else:
281 continue 278 fds = fds + retval
282 elif nextsleep is None:
283 nextsleep = retval
284 elif retval < nextsleep:
285 nextsleep = retval
286 except SystemExit: 279 except SystemExit:
287 raise 280 raise
288 except: 281 except:
289 import traceback 282 import traceback
290 traceback.print_exc() 283 traceback.print_exc()
291 pass 284 pass
292 if nextsleep is None and len(self._idlefuns) > 0: 285
293 nextsleep = 0 286 socktimeout = self.socket.gettimeout() or nextsleep
294 self.timeout = nextsleep 287 socktimeout = min(socktimeout, nextsleep)
288 # Mirror what BaseServer handle_request would do
289 fd_sets = select.select(fds, [], [], socktimeout)
290 if fd_sets[0] and self in fd_sets[0]:
291 self._handle_request_noblock()
292
295 # Tell idle functions we're exiting 293 # Tell idle functions we're exiting
296 for function, data in self._idlefuns.items(): 294 for function, data in self._idlefuns.items():
297 try: 295 try: