All of lore.kernel.org
 help / color / mirror / Atom feed
From: Richard Purdie <richard.purdie@linuxfoundation.org>
To: bitbake-devel <bitbake-devel@lists.openembedded.org>
Subject: [PATCH] server/process, server/xmlrpc, runqueue: Use select.select() on fds, not time.sleep()
Date: Sat, 31 Aug 2013 23:40:55 +0100	[thread overview]
Message-ID: <1377988855.1059.129.camel@ted> (raw)

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.

Signed-off-by: Richard Purdie <richard.purdie@linuxfoundation.org>
---
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:
         if self.fakeworkerpipe:
             self.fakeworkerpipe.read()
 
+    def active_fds(self):
+        fds = []
+        if self.workerpipe:
+            fds.append(self.workerpipe.input)
+        if self.fakeworkerpipe:
+            fds.append(self.fakeworkerpipe.input)
+        return fds
+
     def check_stamp_task(self, task, taskname = None, recurse = False, cache = None):
         def get_timestamp(f):
             try:
@@ -972,7 +980,7 @@ class RunQueue:
         (if the abort on failure configuration option isn't set)
         """
 
-        retval = 0.5
+        retval = True
 
         if self.state is runQueuePrepare:
             self.rqexe = RunQueueExecuteDummy(self)
@@ -1375,7 +1383,7 @@ class RunQueueExecuteTasks(RunQueueExecute):
 
         if self.stats.active > 0:
             self.rq.read_workers()
-            return 0.5
+            return self.rq.active_fds()
 
         if len(self.failed_fnids) != 0:
             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
 import signal
 import sys
 import time
+import select
 from Queue import Empty
 from multiprocessing import Event, Process, util, Queue, Pipe, queues
 
@@ -105,7 +106,7 @@ class ProcessServer(Process, BaseImplServer):
                     command = self.command_channel.recv()
                     self.runCommand(command)
 
-                self.idle_commands(.1)
+                self.idle_commands(.1, [self.event_queue._reader, self.command_channel])
             except Exception:
                 logger.exception('Running command %s', command)
 
@@ -115,7 +116,7 @@ class ProcessServer(Process, BaseImplServer):
         self.cooker.stop()
         self.idle_commands(.1)
 
-    def idle_commands(self, delay):
+    def idle_commands(self, delay, fds = []):
         nextsleep = delay
 
         for function, data in self._idlefuns.items():
@@ -127,15 +128,15 @@ class ProcessServer(Process, BaseImplServer):
                     nextsleep = None
                 elif nextsleep is None:
                     continue
-                elif retval < nextsleep:
-                    nextsleep = retval
+                else:
+                    fds = fds + retval
             except SystemExit:
                 raise
             except Exception:
                 logger.exception('Running idle function')
 
         if nextsleep is not None:
-            time.sleep(nextsleep)
+            select.select(fds,[],[],nextsleep)
 
     def runCommand(self, command):
         """
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):
         Serve Requests. Overloaded to honor a quit command
         """
         self.quit = False
-        self.timeout = 0 # Run Idle calls for our first callback
         while not self.quit:
-            #print "Idle queue length %s" % len(self._idlefuns)
-            self.handle_request()
-            #print "Idle timeout, running idle functions"
-            nextsleep = None
+            fds = [self]
+            nextsleep = 0.1
             for function, data in self._idlefuns.items():
                 try:
                     retval = function(self, data, False)
@@ -277,21 +274,22 @@ class XMLRPCServer(SimpleXMLRPCServer, BaseImplServer):
                         del self._idlefuns[function]
                     elif retval is True:
                         nextsleep = 0
-                    elif nextsleep is 0:
-                        continue
-                    elif nextsleep is None:
-                        nextsleep = retval
-                    elif retval < nextsleep:
-                        nextsleep = retval
+                    else:
+                        fds = fds + retval
                 except SystemExit:
                     raise
                 except:
                     import traceback
                     traceback.print_exc()
                     pass
-            if nextsleep is None and len(self._idlefuns) > 0:
-                nextsleep = 0
-            self.timeout = nextsleep
+
+            socktimeout = self.socket.gettimeout() or nextsleep
+            socktimeout = min(socktimeout, nextsleep)
+            # Mirror what BaseServer handle_request would do
+            fd_sets = select.select(fds, [], [], socktimeout)
+            if fd_sets[0] and self in fd_sets[0]:
+                self._handle_request_noblock()
+
         # Tell idle functions we're exiting
         for function, data in self._idlefuns.items():
             try:




                 reply	other threads:[~2013-08-31 22:41 UTC|newest]

Thread overview: [no followups] expand[flat|nested]  mbox.gz  Atom feed

Reply instructions:

You may reply publicly to this message via plain-text email
using any one of the following methods:

* Save the following mbox file, import it into your mail client,
  and reply-to-all from there: mbox

  Avoid top-posting and favor interleaved quoting:
  https://en.wikipedia.org/wiki/Posting_style#Interleaved_style

* Reply using the --to, --cc, and --in-reply-to
  switches of git-send-email(1):

  git send-email \
    --in-reply-to=1377988855.1059.129.camel@ted \
    --to=richard.purdie@linuxfoundation.org \
    --cc=bitbake-devel@lists.openembedded.org \
    /path/to/YOUR_REPLY

  https://kernel.org/pub/software/scm/git/docs/git-send-email.html

* If your mail client supports setting the In-Reply-To header
  via mailto: links, try the mailto: link
Be sure your reply has a Subject: header at the top and a blank line before the message body.
This is an external index of several public inboxes,
see mirroring instructions on how to clone and mirror
all data and code used by this external index.