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()
_______________________________________________
Xen-devel mailing list
Xen-devel@lists.xen.org
http://lists.xen.org/xen-devel
