Have a watch callback return 0 or 1 depending upon whether it would like to

continue to receive watches.  This means that it is not necessary to probe
around in the xswatch internals to unregister a watch.

Tidy up the hotplug watch handling, moving the nested function out to a
separate place (I don't think that this was a problem in the end, but it was
making me nervous as I was debugging the recent race condition, and I find it
clearer out of the class in any case.  Pass an integer code representing
hotplug status, once it has been parsed from the store, as there are now a few
different places we can detect failure, and it's cleaner to pass a code from
those places.

Small tidy up to XendDomain, removing the unused releaseDomain field.

tools/python/xen/xend/XendDomain.py

@@ -36,6 +36,7 @@ from xen.xend import XendCheckpoint
 from xen.xend.XendError import XendError
 from xen.xend.XendLogging import log
 from xen.xend.server import relocate
+from xen.xend.xenstore.xswatch import xswatch
 
 
 xc = xen.lowlevel.xc.new()
@@ -58,9 +59,11 @@ class XendDomain:
         # to import XendDomain from XendDomainInfo causes unbounded recursion.
         # So we stuff the XendDomain instance (self) into xroot's components.
         xroot.add_component("xen.xend.XendDomain", self)
+
         self.domains = {}
         self.domains_lock = threading.RLock()
-        self.watchReleaseDomain()
+
+        xswatch("@releaseDomain", self.onReleaseDomain)
 
         self.domains_lock.acquire()
         try:
@@ -112,11 +115,7 @@ class XendDomain:
             self.refresh()
         finally:
             self.domains_lock.release()
-            
-
-    def watchReleaseDomain(self):
-        from xen.xend.xenstore.xswatch import xswatch
-        self.releaseDomain = xswatch("@releaseDomain", self.onReleaseDomain)
+        return 1
 
 
     def xen_domains(self):

tools/python/xen/xend/server/DevController.py

@@ -29,6 +29,12 @@ DEVICE_CREATE_TIMEOUT = 5
 HOTPLUG_STATUS_NODE = "hotplug-status"
 HOTPLUG_STATUS_ERROR = "error"
 
+Connected = 1
+Died      = 2
+Error     = 3
+Missing   = 4
+Timeout   = 5
+
 xenbusState = {
     'Unknown'      : 0,
     'Initialising' : 1,
@@ -87,18 +93,28 @@ class DevController:
     def waitForDevice(self, devid):
         log.debug("Waiting for %s.", devid)
         
-        status, fn_ret = self.waitForBackend(devid)
-        if status:
-            self.destroyDevice(devid)
-            raise VmError( ("Device %s (%s) could not be connected. "
-                            "Hotplug scripts not working") 
-                            % (devid, self.deviceClass))
+        status = self.waitForBackend(devid)
 
-        elif fn_ret == HOTPLUG_STATUS_ERROR:
+        if status == Timeout:
             self.destroyDevice(devid)
-            raise VmError( ("Device %s (%s) could not be connected. "
-                            "Backend device not found!") 
-                            % (devid, self.deviceClass))
+            raise VmError("Device %s (%s) could not be connected. "
+                          "Hotplug scripts not working" %
+                          (devid, self.deviceClass))
+
+        elif status == Error:
+            self.destroyDevice(devid)
+            raise VmError("Device %s (%s) could not be connected. "
+                          "Backend device not found" %
+                          (devid, self.deviceClass))
+
+        elif status == Missing:
+            raise VmError("Device %s (%s) could not be connected. "
+                          "Device not found" % (devid, self.deviceClass))
+
+        elif status == Died:
+            self.destroyDevice(devid)
+            raise VmError("Device %s (%s) could not be connected. "
+                          "Device has died" % (devid, self.deviceClass))
 
 
     def reconfigureDevice(self, devid, config):
@@ -302,35 +318,22 @@ class DevController:
                 raise
 
 
-    def waitForBackend(self,devid):
-        ev = Event()
+    def waitForBackend(self, devid):
 
-        def hotplugStatus():
-            log.debug("hotplugStatus %d", devid)
-            
-            try:
-                status = self.readBackend(devid, HOTPLUG_STATUS_NODE)
-            except VmError:
-                status = "died"
-            if status is not None:
-                watch.xs.unwatch(backpath, watch)
-                hotplugStatus.value = status
-                ev.set()
-
-        hotplugStatus.value = None
         frontpath = self.frontendPath(devid)
         backpath = xstransact.Read(frontpath, "backend")
 
         if backpath:
-            watch = xswatch(backpath, hotplugStatus)
+            statusPath = backpath + '/' + HOTPLUG_STATUS_NODE
+            ev = Event()
+            result = { 'status': Timeout }
+            
+            xswatch(statusPath, hotplugStatusCallback, statusPath, ev, result)
 
             ev.wait(DEVICE_CREATE_TIMEOUT)
-            if ev.isSet():
-                return (0, hotplugStatus.value)
-            else:
-                return (-1, hotplugStatus.value)
+            return result['status']
         else:
-            return (-1, "missing")
+            return Missing
 
 
     def backendPath(self, backdom, devid):
@@ -352,3 +355,25 @@ class DevController:
     def frontendMiscPath(self):
         return "%s/device-misc/%s" % (self.vm.getDomainPath(),
                                       self.deviceClass)
+
+
+def hotplugStatusCallback(statusPath, ev, result):
+    log.debug("hotplugStatusCallback %s.", statusPath)
+
+    try:
+        status = xstransact.Read(statusPath)
+
+        if status is not None:
+            if status == HOTPLUG_STATUS_ERROR:
+                result['status'] = Error
+            else:
+                result['status'] = Connected
+        else:
+            return 1
+    except VmError:
+        result['status'] = Died
+
+    log.debug("hotplugStatusCallback %d.", result['status'])
+
+    ev.set()
+    return 0

tools/python/xen/xend/xenstore/xswatch.py

@@ -5,9 +5,7 @@
 # Public License.  See the file "COPYING" in the main directory of
 # this archive for more details.
 
-import select
 import threading
-from xen.lowlevel import xs
 from xen.xend.xenstore.xsutil import xshandle
 
 from xen.xend.XendLogging import log
@@ -20,37 +18,42 @@ class xswatch:
     xslock = threading.Lock()
     
     def __init__(self, path, fn, *args, **kwargs):
+        self.path = path
         self.fn = fn
         self.args = args
         self.kwargs = kwargs
         xswatch.watchStart()
         xswatch.xs.watch(path, self)
 
+
     def watchStart(cls):
         cls.xslock.acquire()
-        if cls.watchThread:
+        try:
+            if cls.watchThread:
+                return
+            cls.xs = xshandle()
+            cls.watchThread = threading.Thread(name="Watcher",
+                                               target=cls.watchMain)
+            cls.watchThread.setDaemon(True)
+            cls.watchThread.start()
+        finally:
             cls.xslock.release()
-            return
-        cls.xs = xshandle()
-        cls.watchThread = threading.Thread(name="Watcher",
-                                           target=cls.watchMain)
-        cls.watchThread.setDaemon(True)
-        cls.watchThread.start()
-        cls.xslock.release()
 
     watchStart = classmethod(watchStart)
 
+
     def watchMain(cls):
         while True:
             try:
                 we = cls.xs.read_watch()
                 watch = we[1]
-                watch.fn(*watch.args, **watch.kwargs)
+                res = watch.fn(*watch.args, **watch.kwargs)
+                if not res:
+                    cls.xs.unwatch(watch.path, watch)
             except:
                 log.exception("read_watch failed")
                 # Ignore this exception -- there's no point throwing it
                 # further on because that will just kill the watcher thread,
                 # which achieves nothing.
 
-
     watchMain = classmethod(watchMain)