bitkeeper revision 1.1068 (40f2bd236qdLHUh6BZA_h_brBDxINw)

Merge ssh://xenbk@gandalf.hpl.hp.com//var/bk/xeno-unstable.bk
into labyrinth.cl.cam.ac.uk:/auto/anfs/scratch/labyrinth/iap10/xeno-clone/xeno.bk

tools/python/xen/xend/XendClient.py

@@ -231,6 +231,11 @@ class Xend:
     def xend_domain(self, id):
         return xend_get(self.domainurl(id))
 
+    def xend_domain_configure(self, id, config):
+        return xend_call(self.domainurl(id),
+                         {'op'      : 'configure',
+                          'config'  : fileof(conf) })
+
     def xend_domain_unpause(self, id):
         return xend_call(self.domainurl(id),
                          {'op'      : 'unpause'})

tools/python/xen/xend/XendDomain.py

@@ -311,6 +311,42 @@ class XendDomain:
             return dominfo
         deferred.addCallback(fn)
         return deferred
+
+    def domain_configure(self, id, config):
+        """Configure an existing domain. This is intended for internal
+        use by domain restore and migrate.
+
+        @param id:     domain id
+        @param config: configuration
+        @return: deferred
+        """
+        dom = int(id)
+        dominfo = self.domain_get(dom)
+        if not dominfo:
+            raise ValueError("Invalid domain: " + str(id))
+        if dominfo.config:
+            raise ValueError("Domain already configured: " + str(id))
+        def fn(dominfo):
+            self._add_domain(dominfo.id, dominfo)
+            return dominfo
+        deferred = dominfo.construct(config)
+        deferred.addCallback(fn)
+        return deferred
+    
+    def domain_restore(self, src, progress=0):
+        """Restore a domain from file.
+
+        @param src:      source file
+        @param progress: output progress if true
+        @return: deferred
+        """
+        
+        def fn(dominfo):
+            self._add_domain(dominfo.id, dominfo)
+            return dominfo
+        deferred = XendDomainInfo.vm_restore(src, progress=progress)
+        deferred.addCallback(fn)
+        return deferred
     
     def domain_get(self, id):
         """Get up-to-date info about a domain.
@@ -346,6 +382,8 @@ class XendDomain:
          - reboot: domain will restart.
          - halt: domain will not restart (even if has autorestart set).
 
+         Returns immediately.
+
         @param id:     domain id
         @param reason: shutdown type: poweroff, reboot, suspend, halt
         """
@@ -443,7 +481,7 @@ class XendDomain:
 
     def domain_destroy(self, id):
         """Terminate domain immediately.
-        Camcels any restart for the domain.
+        Cancels any restart for the domain.
 
         @param id: domain id
         """
@@ -456,6 +494,7 @@ class XendDomain:
         """Start domain migration.
 
         @param id: domain id
+        @return: deferred
         """
         # Need a cancel too?
         # Don't forget to cancel restart for it.
@@ -464,41 +503,16 @@ class XendDomain:
         return xmigrate.migrate_begin(dom, dst)
 
     def domain_save(self, id, dst, progress=0):
-        """Save domain state to file, destroy domain on success.
-        Leave domain running on error.
+        """Start saving a domain to file.
 
         @param id:       domain id
         @param dst:      destination file
         @param progress: output progress if true
+        @return: deferred
         """
         dom = int(id)
-        dominfo = self.domain_get(id)
-        if not dominfo:
-            return -1
-        vmconfig = sxp.to_string(dominfo.sxpr())
-        self.domain_pause(id)
-        eserver.inject('xend.domain.save', id)
-        try:
-            rc = xc.linux_save(dom=dom, state_file=dst,
-                               vmconfig=vmconfig, progress=progress)
-        except:
-            rc = -1
-        if rc == 0:
-            self.domain_destroy(id)
-        else:
-            self.domain_unpause(id)
-        return rc
-    
-    def domain_restore(self, src, progress=0):
-        """Restore domain from file.
-
-        @param src :     source file
-        @param progress: output progress if true
-        @return: domain object
-        """
-        dominfo = XendDomainInfo.vm_restore(src, progress=progress)
-        self._add_domain(dominfo.id, dominfo)
-        return dominfo
+        xmigrate = XendMigrate.instance()
+        return xmigrate.save_begin(dom, dst)
     
     def domain_pincpu(self, dom, cpu):
         """Pin a domain to a cpu.

tools/python/xen/xend/XendMigrate.py

@@ -1,5 +1,6 @@
 # Copyright (C) 2004 Mike Wray <mike.wray@hp.com>
 
+import errno
 import sys
 import socket
 import time
@@ -14,9 +15,12 @@ import sxp
 import XendDB
 import EventServer; eserver = EventServer.instance()
 
+"""The port for the migrate/save daemon xfrd."""
 XFRD_PORT = 8002
 
+"""The transfer protocol major version number."""
 XFR_PROTO_MAJOR = 1
+"""The transfer protocol minor version number."""
 XFR_PROTO_MINOR = 0
 
 class Xfrd(Protocol):
@@ -46,7 +50,7 @@ class Xfrd(Protocol):
         self.parser.input(data)
         if self.parser.ready():
             val = self.parser.get_val()
-            self.xinfo.dispatch(val)
+            self.xinfo.dispatch(self, val)
         if self.parser.at_eof():
             self.loseConnection()
             
@@ -55,16 +59,16 @@ class XfrdClientFactory(ClientFactory):
     """Factory for clients of the migration/save daemon xfrd.
     """
 
-    def __init__(self, minfo):
+    def __init__(self, xinfo):
         #ClientFactory.__init__(self)
-        self.minfo = minfo
+        self.xinfo = xinfo
 
     def startedConnecting(self, connector):
         print 'Started to connect', 'self=', self, 'connector=', connector
 
     def buildProtocol(self, addr):
         print 'buildProtocol>', addr
-        return Migrate(self.minfo)
+        return Xfrd(self.xinfo)
 
     def clientConnectionLost(self, connector, reason):
         print 'clientConnectionLost>', 'connector=', connector, 'reason=', reason
@@ -76,13 +80,15 @@ class XfrdInfo:
     """Abstract class for info about a session with xfrd.
     Has subclasses for save and migrate.
     """
-    
+
+    def __init__(self):
+        from xen.xend import XendDomain
+        self.xd = XendDomain.instance()
+        self.deferred = defer.Deferred()
+        
     def vmconfig(self):
         print 'vmconfig>'
-        from xen.xend import XendDomain
-        xd = XendDomain.instance()
-
-        dominfo = xd.domain_get(self.src_dom)
+        dominfo = self.xd.domain_get(self.src_dom)
         print 'vmconfig>', type(dominfo), dominfo
         if dominfo:
             val = sxp.to_string(dominfo.sxpr())
@@ -93,6 +99,8 @@ class XfrdInfo:
 
     def error(self, err):
         self.state = 'error'
+        if not self.deferred.called:
+            self.deferred.errback(err)
 
     def dispatch(self, xfrd, val):
         op = sxp.name(val)
@@ -100,38 +108,70 @@ class XfrdInfo:
         if op.startswith('xfr_'):
             fn = getattr(self, op, self.unknown)
         else:
-            fn = self.unknown()
-        fn(xfrd, val)
+            fn = self.unknown
+        val = fn(xfrd, val)
+        if val is not None:
+            sxp.show(val, out=self.transport)
 
     def unknown(self, xfrd, val):
         print 'unknown>', val
+        xfrd.loseConnection()
+        return None
 
     def xfr_err(self, xfrd, val):
-        # If we get an error with non-zero code the migrate failed.
+        # If we get an error with non-zero code the operation failed.
         # An error with code zero indicates hello success.
         print 'xfr_err>', val
-        v = sxp.child(val)
+        v = sxp.child0(val)
         print 'xfr_err>', type(v), v
-        err = int(sxp.child(val))
+        err = int(sxp.child0(val))
         if not err: return
         self.error(err);
         xfrd.loseConnection()
+        #try:
+        #    self.xd.domain_unpause(self.src_dom)
+        #except:
+        #    print >>sys.stdout, "Error unpausing domain:", self.src_dom
+        return None
 
     def xfr_progress(self, val):
         print 'xfr_progress>', val
+        return None
 
-    def xfr_domain_pause(self, val):
-        print 'xfr__domain_pause>', val
+    def xfr_vm_pause(self, val):
+        print 'xfr_vm_pause>', val
+        try:
+            vmid = sxp.child0(val)
+            val = self.xd.domain_pause(vmid)
+        except:
+            val = errno.EINVAL
+        return ['xfr.err', val]
 
-    def xfr_domain_suspend(self, val):
-        print 'xfr_domain_suspend>', val
+    def xfr_vm_unpause(self, val):
+        print 'xfr_vm_unpause>', val
+        try:
+            vmid = sxp.child0(val)
+            val = self.xd.domain_unpause(vmid)
+        except:
+            val = errno.EINVAL
+        return ['xfr.err', val]
+
+    def xfr_vm_suspend(self, val):
+        print 'xfr_vm_suspend>', val
+        try:
+            vmid = sxp.child0(val)
+            val = self.xd.domain_shutdown(vmid, reason='suspend')
+        except:
+            val = errno.EINVAL
+        return ['xfr.err', val]
 
 class XendMigrateInfo(XfrdInfo):
     """Representation of a migrate in-progress and its interaction with xfrd.
     """
 
-    def __init__(self, id, dom, host, port):
-        self.id = id
+    def __init__(self, xid, dom, host, port):
+        XfrdInfo.__init__(self)
+        self.xid = xid
         self.state = 'begin'
         self.src_host = socket.gethostname()
         self.src_dom = dom
@@ -139,10 +179,9 @@ class XendMigrateInfo(XfrdInfo):
         self.dst_port = port
         self.dst_dom = None
         self.start = 0
-        self.deferred = defer.Deferred()
         
     def sxpr(self):
-        sxpr = ['migrate', ['id', self.id], ['state', self.state] ]
+        sxpr = ['migrate', ['id', self.xid], ['state', self.state] ]
         sxpr_src = ['src', ['host', self.src_host], ['domain', self.src_dom] ]
         sxpr.append(sxpr_src)
         sxpr_dst = ['dst', ['host', self.dst_host] ]
@@ -160,12 +199,15 @@ class XendMigrateInfo(XfrdInfo):
                       self.src_dom,
                       vmconfig,
                       self.dst_host,
-                      self.d.dst_port])
+                      self.dst_port])
         
     def xfr_migrate_ok(self, val):
         dom = int(sxp.child0(val))
         self.state = 'ok'
         self.dst_dom = dom
+        self.xd_domain_destroy(self.src_dom)
+        if not self.deferred.called:
+            self.deferred.callback(self)
 
     def connectionLost(self, reason=None):
         if self.state =='ok':
@@ -178,17 +220,17 @@ class XendSaveInfo(XfrdInfo):
     """Representation of a save in-progress and its interaction with xfrd.
     """
     
-    def __init__(self, id, dom, file):
-        self.id = id
+    def __init__(self, xid, dom, file):
+        XfrdInfo.__init__(self)
+        self.xid = xid
         self.state = 'begin'
         self.src_dom = dom
         self.file = file
         self.start = 0
-        self.deferred = defer.Deferred()
         
     def sxpr(self):
         sxpr = ['save',
-                ['id', self.id],
+                ['id', self.xid],
                 ['state', self.state],
                 ['domain', self.src_dom],
                 ['file', self.file] ]
@@ -204,6 +246,9 @@ class XendSaveInfo(XfrdInfo):
     def xfr_save_ok(self, val):
         dom = int(sxp.child0(val))
         self.state = 'ok'
+        self.xd_domain_destroy(self.src_dom)
+        if not self.deferred.called:
+            self.deferred.callback(self)
 
     def connectionLost(self, reason=None):
         if self.state =='ok':
@@ -217,7 +262,6 @@ class XendMigrate:
     """External api for interaction with xfrd for migrate and save.
     Singleton.
     """
-    # Represents migration in progress.
     # Use log for indications of begin/end/errors?
     # Need logging of: domain create/halt, migrate begin/end/fail
     # Log via event server?
@@ -226,53 +270,77 @@ class XendMigrate:
     
     def __init__(self):
         self.db = XendDB.XendDB(self.dbpath)
-        self.migrate = {}
-        self.migrate_db = self.db.fetchall("")
-        self.id = 0
+        self.session = {}
+        self.session_db = self.db.fetchall("")
+        self.xid = 0
 
     def nextid(self):
-        self.id += 1
-        return "%d" % self.id
+        self.xid += 1
+        return "%d" % self.xid
 
     def sync(self):
-        self.db.saveall("", self.migrate_db)
+        self.db.saveall("", self.session_db)
 
-    def sync_migrate(self, id):
-        self.db.save(id, self.migrate_db[id])
+    def sync_session(self, xid):
+        print 'sync_session>', type(xid), xid, self.session_db[xid]
+        self.db.save(xid, self.session_db[xid])
 
     def close(self):
         pass
 
-    def _add_migrate(self, id, info):
-        self.migrate[id] = info
-        self.migrate_db[id] = info.sxpr()
-        self.sync_migrate(id)
+    def _add_session(self, xid, info):
+        self.session[xid] = info
+        self.session_db[xid] = info.sxpr()
+        self.sync_session(xid)
         #eserver.inject('xend.migrate.begin', info.sxpr())
 
-    def _delete_migrate(self, id):
-        #eserver.inject('xend.migrate.end', id)
-        del self.migrate[id]
-        del self.migrate_db[id]
-        self.db.delete(id)
+    def _delete_session(self, xid):
+        #eserver.inject('xend.migrate.end', xid)
+        del self.session[xid]
+        del self.session_db[xid]
+        self.db.delete(xid)
 
-    def migrate_ls(self):
-        return self.migrate.keys()
+    def session_ls(self):
+        return self.session.keys()
 
-    def migrates(self):
-        return self.migrate.values()
+    def sessions(self):
+        return self.session.values()
 
-    def migrate_get(self, id):
-        return self.migrate.get(id)
-    
-    def migrate_begin(self, dom, host, port=XFRD_PORT):
-        # Check dom for existence, not migrating already.
-        # Subscribe to migrate notifications (for updating).
-        id = self.nextid()
-        info = XendMigrateInfo(id, dom, host, port)
-        self._add_migrate(id, info)
+    def session_get(self, xid):
+        return self.session.get(xid)
+
+    def session_begin(self, info):
+        self._add_session(info.xid, info)
         mcf = XfrdClientFactory(info)
         reactor.connectTCP('localhost', XFRD_PORT, mcf)
         return info
+    
+    def migrate_begin(self, dom, host, port=XFRD_PORT):
+        """Begin to migrate a domain to another host.
+
+        @param dom:  domain
+        @param host: destination host
+        @param port: destination port
+        @return: deferred
+        """
+        # Check dom for existence, not migrating already.
+        # Subscribe to migrate notifications (for updating).
+        xid = self.nextid()
+        info = XendMigrateInfo(xid, dom, host, port)
+        self.session_begin(info)
+        return info.deferred
+
+    def save_begin(self, dom, file):
+        """Begin saving a domain to file.
+
+        @param dom:  domain
+        @param file: destination file
+        @return: deferred
+        """
+        xid = self.nextid()
+        info = XendSaveInfo(xid, dom, file)
+        self.session_begin(info)
+        return info.deferred
 
 def instance():
     global inst

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

@@ -18,6 +18,14 @@ class SrvDomain(SrvDir):
         self.xd = XendDomain.instance()
         self.xconsole = XendConsole.instance()
 
+    def op_configure(self, op, req):
+        fn = FormFn(self.xd.domain_configure,
+                    [['dom', 'int'],
+                     ['config', 'sxp']])
+        val = fn(req.args, {'dom': self.dom.id})
+        #todo: may need to add ok and err callbacks.
+        return val
+
     def op_unpause(self, op, req):
         val = self.xd.domain_unpause(self.dom.id)
         return val
@@ -27,7 +35,6 @@ class SrvDomain(SrvDir):
         return val
 
     def op_shutdown(self, op, req):
-        #val = self.xd.domain_shutdown(self.dom.id)
         fn = FormFn(self.xd.domain_shutdown,
                     [['dom', 'int'],
                      ['reason', 'str']])
@@ -202,8 +209,22 @@ class SrvDomain(SrvDir):
         req.write('<form method="post" action="%s">' % req.prePathURL())
         req.write('<input type="submit" name="op" value="unpause">')
         req.write('<input type="submit" name="op" value="pause">')
-        req.write('<input type="submit" name="op" value="shutdown">')
         req.write('<input type="submit" name="op" value="destroy">')
-        req.write('<br><input type="submit" name="op" value="migrate">')
-        req.write('To: <input type="text" name="destination">')
+        req.write('</form>')
+
+        req.write('<form method="post" action="%s">' % req.prePathURL())
+        req.write('<input type="submit" name="op" value="shutdown">')
+        req.write('<input type="radio" name="reason" value="poweroff" checked>Poweroff<br>')
+        req.write('<input type="radio" name="reason" value="halt">Halt<br>')
+        req.write('<input type="radio" name="reason" value="reboot">Reboot<br>')
+        req.write('</form>')
+        
+        req.write('<form method="post" action="%s">' % req.prePathURL())
+        req.write('<br><input type="submit" name="op" value="save">')
+        req.write('To file: <input type="text" name="file">')
+        req.write('</form>')
+        
+        req.write('<form method="post" action="%s">' % req.prePathURL())
+        req.write('<br><input type="submit" name="op" value="migrate">')
+        req.write('To host: <input type="text" name="destination">')
         req.write('</form>')

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

@@ -105,7 +105,10 @@ class SrvDomainDir(SrvDir):
 
     def op_restore(self, op, req):
         """Restore a domain from file.
+
+        @return: deferred
         """
+        #todo: return is deferred. May need ok and err callbacks.
         fn = FormFn(self.xd.domain_restore,
                     [['file', 'str']])
         val = fn(req.args)
@@ -154,9 +157,10 @@ class SrvDomainDir(SrvDir):
         req.write('<button type="submit" name="op" value="create">Create Domain</button>')
         req.write('Config <input type="file" name="config"><br>')
         req.write('</form>')
+        
         req.write('<form method="post" action="%s" enctype="multipart/form-data">'
                   % req.prePathURL())
-        req.write('<button type="submit" name="op" value="create">Restore Domain</button>')
+        req.write('<button type="submit" name="op" value="restore">Restore Domain</button>')
         req.write('State <input type="string" name="state"><br>')
         req.write('</form>')