[RFC] native python Xenstore module

[RFC] native python Xenstore module

[Simon Rowe]

The current Python interface to Xenstore is just a thin binding to the
C libxenstore library. This means that it is architecture-specific and
makes it awkward to use in platform-independent code like the XenServer
guest agent.

The Xenstore protocol is simple and quite straightforward to implement
natively in Python.


Design & Implementation

Users create an instance of the Xenstore object. This is a singleton
that holds all the global state (xenbus fd etc).

Users invoke methods e.g. read() that block awaiting a response from
xenstored. Methods return either the result or throw XenstoreException
with a string representation of the error e.g. 'ENOENT'.

The methods attempt to present complex results using native datatypes
(lists, dictionaries) rather the usual Xenstore string values e.g.

     {6: 'read-write', 9: 'none'}

rather than

     'b6n9'


Transactions are managed using instances of the Transaction object. Each
instance must be utilized thus, either

     t.start()
     # zero or more operations
     t.commit()

or

     t.start()
     # zero or more operations
     t.abort()


Users should handle commit() raising XenstoreException('EAGAIN')
appropriately, e.g. replaying the whole start(), operations, commit()
sequence.

Watches are complicated because they are not synchronous. When the first
watch is registered a background thread is started. This thread takes
responsibility for the reading of all data from the xenbus fd. Responses
for operations are forwarded to the foreground thread using a Queue.

Users of watch() supply the path to watch, a callback function and
(optionally) parameters. The callback function is invoked with the path
and the optional parameters. watch() returns a string token that is
passed to unwatch() to remove a watch.


#!/bin/env python

import os
import Queue
import select
import struct
import tempfile
import threading

class XenstoreException(StandardError):
     pass

"""
Background thread that reads from xenbus, acting on watch events and 
forwarding
operation responses to the foreground thread.
"""
def handle_watches(xs):
     while not xs.shutdown_event.isSet():
         r, _, __ = select.select([xs.pipe_r, xs.xb_fd], [], [])
         if xs.xb_fd in r:
             # read and unpack header
             hdr = xs.xb_fd.read(16)
             op, _, __, l = struct.unpack('=IIII', hdr)
             value = None
             if l > 0:
                 value = xs.xb_fd.read(l)
             if op == xs.XS_WATCH_EVENT:
                 path, token, _ = value.split('\0', 2)
                 if token in xs.watches:
                     # invoke user callback
                     xs.watches[token]['cb'](path, 
*xs.watches[token]['cbargs'])
             else:
                 # response to an operation, handle in main thread
                 xs.queue.put(hdr)
                 if value:
                     xs.queue.put(value)

class Xenstore(object):
     """
     Parent Xenstore handle (singleton).
     """

     # from xen/include/public/io/xs_wire.h
     XS_READ = 2
     XS_WATCH = 4
     XS_GET_PERMS = 3
     XS_UNWATCH = 5
     XS_TRANSACTION_START = 6
     XS_TRANSACTION_END = 7
     XS_WRITE = 11
     XS_MKDIR = 12
     XS_RM = 13
     XS_SET_PERMS = 14
     XS_WATCH_EVENT = 15
     XS_ERROR = 16

     if os.uname()[0] == 'Linux':
         DEV_PATH = '/proc/xen/xenbus'
     elif os.uname()[0] == 'NetBSD':
         DEV_PATH = '/kern/xen/xenbus'
     else:
         DEV_PATH = '/dev/xen/xenbus'

     PERM_NONE, PERM_READ, PERM_WRITE, PERM_READ_WRITE = range(4)

     __single = None

     def __new__(classtype, *args, **kwargs):
         if classtype != type(classtype.__single):
             classtype.__single = object.__new__(classtype, *args, **kwargs)
         return classtype.__single

     def __init__(self):
         self.xb_fd = open(self.DEV_PATH, 'r+', 0)
         self.watch_thread = None
         self.queue = None
         self.watches = {}

     def __del__(self):
         self.xb_fd.close()

     def do_op(self, ctxt, op, value = '', req = 0):
         """
         Perform Xenstore operation $op and return response.
         """

         if ctxt:
             tx_id = ctxt.tx_id
         else:
             tx_id = 0

         ret = None
         self.xb_fd.write(struct.pack('=IIII', op, req, tx_id, len(value)))
         if len(value):
             self.xb_fd.write(value)
         if self.queue:
             hdr = self.queue.get()
         else:
             hdr = self.xb_fd.read(16)
         r_op, req, tx, l = struct.unpack('=IIII', hdr)
         if l > 0:
             if self.queue:
                 ret = self.queue.get()
             else:
                 ret = self.xb_fd.read(l)
             if r_op == self.XS_ERROR:
                 raise XenstoreException, ret[:-1]
         return ret

     def _read(self, ctxt, *vars):
         d = {}
         for var in vars:
             d[var] = self.do_op(ctxt, self.XS_READ, var + '\0')
             if len(vars) == 1:
                 return d[var]
         return d

     def read(self, *vars):
         """
         Return a dictionary of Xenstore values for $vars (or a string 
if only a
         single key is passed in).
         """

         return self._read(None, *vars)

     def _mkdir(self, ctxt, *vars):
         for var in vars:
             self.do_op(ctxt, self.XS_MKDIR, var + '\0')

     def mkdir(self, *vars):
         """
         Create empty directories in Xenstore.
         """

         self._mkdir(None, *vars)

     def _rm(self, ctxt, *vars):
         for var in vars:
             self.do_op(ctxt, self.XS_RM, var + '\0')

     def rm(self, *vars):
         """
         Remove all keys in $vars from Xenstore.
         """
         self._rm(None, *vars)

     def write(self, vars, ctxt = None):
         """
         Update Xenstore with the keys and values in dictionary $vars.
         """

         for k, v in vars.items():
             self.do_op(ctxt, self.XS_WRITE, k + '\0' + v)

     perm_map = {'n': PERM_NONE, 'r': PERM_READ, 'w': PERM_WRITE, 'b': 
PERM_READ_WRITE}
     rev_perm_map = {PERM_NONE: 'n', PERM_READ: 'r', PERM_WRITE: 'w', 
PERM_READ_WRITE: 'b'}

     def _get_permissions(self, ctxt, *paths):
         d = {}
         for path in paths:
             t = self.do_op(ctxt, self.XS_GET_PERMS, path + '\0')
             perm_d = {}
             for e in t[:-1].split('\0'):
                 perm_d[int(e[1:])] = self.perm_map[e[0]]
             d[path] = perm_d
             if len(paths) == 1:
                 return d[path]
         return d

     def get_permissions(self, *paths):
         """
         Return a list of permission dictionaries (or a single if only a
         single key is passed in.
         """
         return self._get_permissions(None, *paths)

     def set_permissions(self, path, perms, ctxt = None):
         """
         Set the permissions of $path based on the dictionary $perms.
         """
         perm_l = []
         for k, v in perms.items():
             perm_l.append("%c%d" % (self.rev_perm_map.get(v, 'n'), k))
         self.do_op(None, self.XS_SET_PERMS, path + '\0' + 
'\0'.join(perm_l) + '\0')

     def watch(self, path, callback, *callback_args):
         """
         Watch $path (and its subordinates), invoking $callback($path, 
$callback_args)
         in a background thread.

         Returns a token to be used by unwatch().
         """
         if not self.watch_thread:
             self.watch_thread = threading.Thread(target=handle_watches, 
args=(self,))
             self.watch_thread.setDaemon(True)
             self.shutdown_event = threading.Event()
             self.queue = Queue.Queue(2)
             self.pipe_r, self.pipe_w = os.pipe()
             self.watches = {}
             self.watch_thread.start()

         fh = tempfile.NamedTemporaryFile(prefix = 'xs')
         token = fh.name
         self.watches[token] = {'fh': fh, 'path': path, 'cb': callback, 
'cbargs': callback_args}
         self.do_op(None, self.XS_WATCH, path + '\0' + token + '\0')
         return token

     def unwatch(self, path, token):
         """
         Stop watching the path monitored by the watch() call that 
returned $token.
         """
         if token in self.watches:
             self.do_op(None, self.XS_UNWATCH, path + '\0' + token + '\0')
             self.watches[token]['fh'].close()
             del self.watches[token]

     def unwatch_all(self):
         for k, v in self.watches.items():
             self.unwatch(v['path'], k)

     def watch_stop(self):
         if self.watch_thread:
             self.shutdown_event.set()
             os.write(self.pipe_w, 'x')
             self.watch_thread.join()
             self.watch_thread = None
             self.queue = None
             os.close(self.pipe_r)
             os.close(self.pipe_w)

class Transaction(object):
     """
     Xenstore transaction instance.
     """

     def __init__(self):
         self.xs = Xenstore()
         self.tx_id = 0

     def start(self):
         """
         Start a new transaction.
         """

         self.tx_id = 0
         self.tx_id = int(self.xs.do_op(self, 
self.xs.XS_TRANSACTION_START, '\0')[:-1])

     def commit(self):
         """
         Commit all modifications in this transaction to Xenstore.
         """

         self.xs.do_op(self, self.xs.XS_TRANSACTION_END, 'T\0')

     def abort(self):
         """
         Discard all modifications in this transaction.
         """

         self.xs.do_op(self, self.xs.XS_TRANSACTION_END, 'F\0')

     def read(self, *vars):
         return self.xs._read(self, *vars)

     def mkdir(self, *vars):
         self.xs._mkdir(self, *vars)

     def rm(self, *vars):
         self.xs._read(self, *vars)

     def write(self, vars):
         self.xs.write(vars, self)

     def get_permissions(self, *paths):
         return self.xs_get_permissions(self, *paths)

     def set_permissions(self, path, perms):
         self.xs.set_permissions(perms, self)


# example to code to dev test

if __name__ == '__main__':
     def watch_cb(key, n):
         print "Watch fired:", key, n

     repeat_count = 5

     # create xenstore handle
     xs = Xenstore()
     # read three keys, returns three values
     print xs.read('domid', 'vm', 'name')
     # write a key/value
     xs.write({'new': 'stuff'})
     # verify contents were written
     assert xs.read('new') == 'stuff'
     print xs.get_permissions('new')
     xs.set_permissions('new', {0: xs.PERM_READ_WRITE, os.getpid(): 
xs.PERM_READ})
     print xs.get_permissions('new')
     xs.mkdir('another-new')
     # write another key
     xs.write({'new/path': 'more stuff'})
     xs.read('new/path')
     # delete key
     xs.rm('new/path')
     # attempt to read deleted key
     try:
         print xs.read('new/path')
     except XenstoreException, e:
         print "Failed to read (expected)", e

     # create and start transaction
     tx = Transaction()
     tx.start()
     tx.write({'tnew': 'committed'})
     # commit and verify value is updated
     tx.commit()
     assert xs.read('tnew') == 'committed'

     # start new transaction
     tx.start()
     tx.write({'tnew': 'aborted'})
     # abort and verify value is NOT updated
     tx.abort()
     assert xs.read('tnew') != 'aborted'

     # watch key
     xs.watch('new', watch_cb, 7)

     d = {}
     # repeatedly run a large transaction to cause collisions between 
two program instances
     for n in range(1, 200):
         d["collision/key"+str(n)] = "value"+str(n)
     while True:
         try:
             tx.start()
             tx.write(d)
             tx.commit()
         except XenstoreException, e:
             if str(e) == 'EAGAIN':
                 repeat_count -= 1
                 if repeat_count == 0:
                     break
                 print "Collision on commit, repeating"
             else:
                 print "Unexpected Xenstore error", e
                 break

     print "End of test"
     xs.unwatch_all()
     # stop background watch thread
     xs.watch_stop()

Re: [RFC] native python Xenstore module

[Ian Campbell]

On Tue, 2015-02-03 at 14:01 +0000, Simon Rowe wrote:
> The current Python interface to Xenstore is just a thin binding to the
> C libxenstore library. This means that it is architecture-specific and
> makes it awkward to use in platform-independent code like the XenServer
> guest agent.

Are/were you aware of https://pypi.python.org/pypi/pyxs which sounds
like something similar, judging from its blurb alone?

https://launchpad.net/pyxenstore/ might be too, although I don't know if
that one is pure python.

Anyhow is your intention to have this added to the xen.git tree, perhaps
even replacing tools/python/xen/lowlevel/xs/ or just to gather feedback
on some code destined for an external project?

Personally I don't see a problem with including it in tree, replacing
xen.lowlevel.xs might require thinking a little about API compatibility,
which might be a pain...

Ian.

Re: [RFC] native python Xenstore module

[Sander Eikelenboom]

Tuesday, February 24, 2015, 5:41:40 PM, you wrote:

> On Tue, 2015-02-03 at 14:01 +0000, Simon Rowe wrote:
>> The current Python interface to Xenstore is just a thin binding to the
>> C libxenstore library. This means that it is architecture-specific and
>> makes it awkward to use in platform-independent code like the XenServer
>> guest agent.

> Are/were you aware of https://pypi.python.org/pypi/pyxs which sounds
> like something similar, judging from its blurb alone?

I'm using the above for a project, seems to work fine for my purposes. 
Although you have to get a hang for working around some of the xenstore
watch oddities/limitations (firing once after setup, firing on child changes
without you knowing what actually changed because you don't have the previous 
value, but most of the things can be worked around using the watch 
token. I use a pickled dict of the watched path and the current value as the 
token at that seems to work).

--
Sander

> https://launchpad.net/pyxenstore/ might be too, although I don't know if
> that one is pure python.

> Anyhow is your intention to have this added to the xen.git tree, perhaps
> even replacing tools/python/xen/lowlevel/xs/ or just to gather feedback
> on some code destined for an external project?

> Personally I don't see a problem with including it in tree, replacing
> xen.lowlevel.xs might require thinking a little about API compatibility,
> which might be a pain...

> Ian.

Re: [RFC] native python Xenstore module

[Simon Rowe]

On 24/02/15 16:41, Ian Campbell wrote:
> Are/were you aware ofhttps://pypi.python.org/pypi/pyxs  which sounds
> like something similar, judging from its blurb alone?

No, I wasn't. It certainly looks a more solid implementation than mine.
>
> https://launchpad.net/pyxenstore/  might be too, although I don't know if
> that one is pure python.

This looks to be just a higher level C to python interface than 
xen.lowlevel.xs.

> Anyhow is your intention to have this added to the xen.git tree, perhaps
> even replacing tools/python/xen/lowlevel/xs/ or just to gather feedback
> on some code destined for an external project?
I was interested in feedback about having implementations such as these 
more easily consumed by guest distros. At present using something linked 
against libxenstore.so requires Xen tools to have been built 
(obviously). If there are already serviceable implementations available 
(which it appears there is) then the best policy is probably to 
encourage distros to package and include them.

> Personally I don't see a problem with including it in tree, replacing
> xen.lowlevel.xs might require thinking a little about API compatibility,
> which might be a pain...
pyxs even has a compat interface.

Simon

Re: [RFC] native python Xenstore module

[Ian Campbell]

On Wed, 2015-02-25 at 09:39 +0000, Simon Rowe wrote:
> On 24/02/15 16:41, Ian Campbell wrote:
> > Are/were you aware ofhttps://pypi.python.org/pypi/pyxs  which sounds
> > like something similar, judging from its blurb alone?
> 
> No, I wasn't. It certainly looks a more solid implementation than mine.

https://github.com/selectel/pyxs seems to be the real upstream, pypi
seems a bit behind.

> >
> > https://launchpad.net/pyxenstore/  might be too, although I don't know if
> > that one is pure python.
> 
> This looks to be just a higher level C to python interface than 
> xen.lowlevel.xs.
> 
> > Anyhow is your intention to have this added to the xen.git tree, perhaps
> > even replacing tools/python/xen/lowlevel/xs/ or just to gather feedback
> > on some code destined for an external project?
> I was interested in feedback about having implementations such as these 
> more easily consumed by guest distros. At present using something linked 
> against libxenstore.so requires Xen tools to have been built 
> (obviously).

Yes, I can see that being rather annoying to have to arrange.

>  If there are already serviceable implementations available 
> (which it appears there is) then the best policy is probably to 
> encourage distros to package and include them.

I think that's probably the best approach.

(https://bugs.debian.org/cgi-bin/bugreport.cgi?bug=692516 for Debian)

> > Personally I don't see a problem with including it in tree, replacing
> > xen.lowlevel.xs might require thinking a little about API compatibility,
> > which might be a pain...
> pyxs even has a compat interface.

Neat!

Since today happens to be a doc day I've updated
http://wiki.xen.org/wiki/XenStore_Reference to point to pyxs (and nuked
a load of old xend related stuff).

/me ponders "git rm tools/python/xen/lowlevel/xs" ;-)

Ian.