bitkeeper revision 1.1448.1.1 (428b5d756NfzYJ90kQUy0Q91tiUWhw)

scheduler.py:
  Simplify scheduler by removing support for cancelling timeouts.
  Also add a ``now'' function to run a function in a seperate thread.
XendDomain.py:
  g/c gratuitous refresh() calls.
  Drive domain restart off the virq signaling that a domain has terminated.
Signed-off-by: Christian Limpach <Christian.Limpach@cl.cam.ac.uk>

tools/python/xen/xend/XendDomain.py

@@ -111,35 +111,6 @@ class XendDomain:
         print 'onVirq>', val
         self.refresh()
 
-    def schedule_later(self, _delay, _name, _fn, *args):
-        """Schedule a function to be called later (if not already scheduled).
-
-        @param _delay: delay in seconds
-        @param _name:  schedule name
-        @param _fn:    function
-        @param args:   arguments
-        """
-        self.scheduler.later(_delay, _name, _fn, args)
-        
-    def schedule_cancel(self, name):
-        """Cancel a scheduled function call.
-        
-        @param name: schedule name to cancel
-        """
-        self.scheduler.cancel(name)
-
-    def domain_restarts_schedule(self, delay=1):
-        """Schedule domain_restarts to be called later.
-        
-        @param delay: delay in seconds
-        """
-        self.schedule_later(delay, 'domain_restarts', self.domain_restarts)
-        
-    def domain_restarts_cancel(self):
-        """Cancel any scheduled call of domain_restarts.
-        """
-        self.schedule_cancel('domain_restarts')
-        
     def rm_all(self):
         """Remove all domain info. Used after reboot.
         """
@@ -302,8 +273,6 @@ class XendDomain:
                eserver.inject('xend.domain.exit', [name, id, 'crash']) 
             destroyed += 1
             self.final_domain_destroy(id)
-        if self.domain_restarts_exist():
-            self.domain_restarts_schedule()
 
     def refresh(self):
         """Refresh domain list from Xen.
@@ -316,14 +285,17 @@ class XendDomain:
                 self.domain_lookup(id)
         # Remove entries for domains that no longer exist.
         # Update entries for existing domains.
+        do_domain_restarts = False
         for d in self.domain_by_id.values():
             info = doms.get(d.id)
             if info:
                 d.update(info)
             elif d.restart_pending():
-                pass
+                do_domain_restarts = True
             else:
                 self._delete_domain(d.id)
+        if do_domain_restarts:
+            self.scheduler.now(self.domain_restarts)
 
     def update_domain(self, id):
         """Update the saved info for a domain.
@@ -360,7 +332,6 @@ class XendDomain:
 
         @return: domain names
         """
-        self.refresh()
         return self.domain_by_id.keys()
 
     def domains(self):
@@ -368,7 +339,6 @@ class XendDomain:
 
         @return: domain objects
         """
-        self.refresh()
         return self.domain_by_id.values()
     
     def domain_create(self, config):
@@ -515,7 +485,6 @@ class XendDomain:
         """Process pending domain shutdowns.
         Destroys domains whose shutdowns have timed out.
         """
-        self.schedule_cancel('domain_shutdowns')
         timeout = SHUTDOWN_TIMEOUT
         for shutdown in self.shutdowns_by_id.values():
             id = shutdown.getDomain()
@@ -538,7 +507,7 @@ class XendDomain:
                 timeout = min(timeout, shutdown.getTimeout())
         if self.shutdowns_by_id:
             # Pending shutdowns remain - reschedule.
-            self.schedule_later(timeout, 'domain_shutdowns', self.domain_shutdowns)
+            self.scheduler.later(timeout, self.domain_shutdowns)
 
     def domain_restart_schedule(self, id, reason, force=False):
         """Schedule a restart for a domain if it needs one.
@@ -563,7 +532,6 @@ class XendDomain:
         log.info('Scheduling restart for domain: name=%s id=%s', dominfo.name, dominfo.id)
         eserver.inject("xend.domain.restart",
                        [dominfo.name, dominfo.id, "schedule"])
-        self.domain_restarts_schedule()
             
     def domain_restart_cancel(self, id):
         """Cancel any restart scheduled for a domain.
@@ -583,7 +551,6 @@ class XendDomain:
     def domain_restarts(self):
         """Execute any scheduled domain restarts for domains that have gone.
         """
-        self.domain_restarts_cancel()
         doms = self.xen_domains()
         for dominfo in self.restarts_by_id.values():
             print 'domain_restarts>', dominfo.name, dominfo.id
@@ -597,13 +564,7 @@ class XendDomain:
             del self.restarts_by_name[dominfo.name]
             print 'domain_restarts> restarting: ', dominfo.name
             self.domain_restart(dominfo)
-        if self.domain_restarts_exist():
-            # Run again later if any restarts remain.
-            self.domain_restarts_schedule(delay=10)
 
-    def domain_restarts_exist(self):
-        return len(self.restarts_by_id)
-        
     def final_domain_destroy(self, id):
         """Final destruction of a domain..
 
@@ -631,7 +592,6 @@ class XendDomain:
         """
         self.domain_restart_schedule(id, reason, force=True)
         val = self.final_domain_destroy(id)
-        self.refresh()
         return val
 
     def domain_migrate(self, id, dst, live=False, resource=0):

tools/python/xen/xend/scheduler.py

@@ -2,50 +2,28 @@ import threading
 
 class Scheduler:
 
-    def __init__(self):
-        self.lock = threading.Lock()
-        self.schedule = {}
-
-    def later(self, _delay, _name, _fn, args, kwargs={}):
-        """Schedule a function to be called later (if not already scheduled).
+    def later(self, _delay, _fn, args=(), kwargs={}):
+        """Schedule a function to be called later.
 
         @param _delay: delay in seconds
-        @param _name:  schedule name
         @param _fn:    function
         @param args:   arguments (list)
         @param kwargs  keyword arguments (map)
         """
-        try:
-            self.lock.acquire()
-            if self.schedule.get(_name): return
-            runargs = [ _name, _fn, args, kwargs ]
-            timer = threading.Timer(_delay, self._run, args=runargs)
-            self.schedule[_name] = timer
-        finally:
-            self.lock.release()
+        runargs = [ _fn, args, kwargs ]
+        timer = threading.Timer(_delay, self._run, args=runargs)
         timer.start()
 
-    def cancel(self, name):
-        """Cancel a scheduled function call.
-        
-        @param name: schedule name to cancel
+    def now(self, _fn, args=(), kwargs={}):
+        """Schedule a function to be called now.
+
+        @param _fn:    function
+        @param args:   arguments (list)
+        @param kwargs  keyword arguments (map)
         """
-        timer = self._remove(name)
-        if timer:
-            timer.cancel()
+        runargs = [ _fn, args, kwargs ]
+        thread = threading.Thread(target=self._run, args=runargs)
+        thread.start()
 
-    def _remove(self, name):
-        try:
-            self.lock.acquire()
-            timer = self.schedule.get(name)
-            if timer:
-                del self.schedule[name]
-            return timer
-        finally:
-            self.lock.release()
-
-    def _run(self, name, fn, args, kwargs):
-        self._remove(name)
+    def _run(self, fn, args, kwargs):
         fn(*args, **kwargs)
-
-