# -*- coding: utf-8 -*- """ pyxs.client ~~~~~~~~~~~ This module implements XenStore client, which can communicate with XenStore either via: :class:`~pyxs.connection.UnixSocketConnection` or :class:`~pyxs.connection.XenBusConnection`. :copyright: (c) 2011 by Selectel. :copyright: (c) 2016 by pyxs authors and contributors, see AUTHORS for more details. :license: LGPL, see LICENSE for more details. """ from __future__ import absolute_import __all__ = ["Router", "Client", "Monitor"] import copy import errno import posixpath import re import socket import select import sys import threading from collections import defaultdict from functools import partial try: import Queue as queue except ImportError: import queue # XXX see ``Router`` docstring for motivation. if sys.version_info[:2] < (3, 2): _condition_wait = partial(threading._Condition.wait, timeout=1) else: _condition_wait = threading.Condition.wait from ._internal import NUL, Event, Packet, Op, next_rq_id from .connection import UnixSocketConnection, XenBusConnection from .exceptions import UnexpectedPacket, ConnectionError, PyXSError from .helpers import check_path, check_watch_path, check_perms, error _re_7bit_ascii = re.compile(b"^[\x00\x20-\x7f]*$") class Router(object): """Router. The goal of the router is to multiplex XenStore connection between multiple clients and monitors. .. versionadded: 0.4.0 :param connection FileDescriptorConnection: owned by the router. The connection is open when the router is started and remains open until the router is terminated. .. note:: Python lacks API for interrupting a thread from another thread. This means that when a router cannot be stopped when it is blocked in :func:`select.select` or :meth:`~threading.Condition.wait`. The following two "hacks" are used to ensure prompt termination. 1. A router is equipped with a :func:`socket.socketpair`. The reader-end of the pair is selected in the mainloop alongside the XenStore connection, while the writer-end is used in :meth:`~pyxs.client.Router.terminate` to force-stop the mainloop. 2. All operations with :class:`threading.Condition` variables user a 1 second timeout. This "hack" is only relevant for Python prior to 3.2 which didn't allow one to interrupt lock acquisitions. See `issue8844`_ on CPython issue tracker for details. On Python 3.2 and later no timeout is used. .. _issue8844: https://bugs.python.org/issue8844 """ def __init__(self, connection): self.r_terminator, self.w_terminator = socket.socketpair() self.connection = connection self.send_lock = threading.Lock() self.rvars = {} self.monitors = defaultdict(list) # Router thread is daemonic to prevent blocking in case # the client wasn't finilzed properly, e.g. unhandled # exception outside of ``with``. As a result, we cannot # guarantee data integrity unless either ``close`` or # ``__exit__`` was closed. self.thread = threading.Thread(target=self) self.thread.daemon = True def __repr__(self): return "Router({0})".format(self.connection) def __call__(self): try: while True: rlist, _wlist, _xlist = select.select( [self.connection, self.r_terminator], [], []) if not rlist: continue elif self.r_terminator in rlist: break packet = self.connection.recv() if packet.op == Op.WATCH_EVENT: event = Event(*packet.payload.split(NUL)[:-1]) for monitor in self.monitors[event.token]: monitor.events.put(event) else: rvar = self.rvars.pop(packet.rq_id, None) if rvar is None: raise UnexpectedPacket(packet) else: rvar.set(packet) finally: self.connection.close() self.r_terminator.close() self.w_terminator.close() @property def is_connected(self): """Checks if the underlying connection is active.""" return self.connection.is_connected def subscribe(self, token, monitor): """Subscribes a ``monitor`` from events with a given ``token``.""" self.monitors[token].append(monitor) def unsubscribe(self, token, monitor): """Unsubscribes a ``monitor`` to events with a given ``token``.""" self.monitors[token].remove(monitor) def send(self, packet): """Sends a packet to XenStore. :returns RVar: a reference to the XenStore response. """ with self.send_lock: # The order here matters. XenStore might reply to the packet # *before* the ``rvar`` is registered. self.rvars[packet.rq_id] = rvar = RVar() self.connection.send(packet) return rvar def start(self): """Starts the router thread. Does nothing if the router is already started. """ # Connection is deliberately done in the calling thread so that # ``ConnectionError`` could be handled. See issue #8 on GitHub # for details. self.connection.connect() if not self.thread.is_alive(): self.thread.start() while not self.is_connected: if not self.thread.is_alive(): raise ConnectionError("router died") def terminate(self): """Terminates the router. After termination the router can no longer send or receive packets. Does nothing if the router was already terminated. """ if self.is_connected: self.w_terminator.sendall(NUL) if self.thread.is_alive(): self.thread.join() class RVar(object): """A thread-safe shared mutable reference. .. versionadded:: 0.4.0 """ __slots__ = ["condition", "target"] def __init__(self): self.condition = threading.Condition() self.target = None def __repr__(self): return "RVar({0})".format(self.target) def get(self): """Blocks until the value is :meth:`set`` and then returns the value. .. note:: The returned value is guaranteed never to be ``None``. """ with self.condition: while self.target is None: _condition_wait(self.condition) return self.target def set(self, target): """Sets the value, which effectively unblocks all :meth:`get` calls.""" with self.condition: self.target = target self.condition.notify_all() class Client(object): """XenStore client. :param str unix_socket_path: path to XenStore Unix domain socket. :param str xen_bus_path: path to XenBus device. If ``unix_socket_path`` is given or :class:`~pyxs.client.Client` was created with no arguments, XenStore is accessed via :class:`~pyxs.connection.UnixSocketConnection`; otherwise, :class:`~pyxs.connection.XenBusConnection` is used. Each client has a :class:`~pyxs.client.Router` thread running in the background. The goal of the router is to multiplex requests from different transaction through a single XenStore connection. .. versionchanged:: 0.4.0 The constructor no longer accepts ``connection`` argument. If you wan't to force the use of a specific connection class, wrap it in a :class:`~pyxs.client.Router`:: from pyxs import Router, Client from pyxs.connection import XenBusConnection router = Router(XenBusConnection()) with Client(router=router) as c: do_something(c) .. warning:: Always finalize the client either explicitly by calling :meth:`~pyxs.client.Client.close` or implicitly via a context manager to prevent data loss. .. seealso:: `Xenstore protocol specification \ `_ for a description of the protocol, implemented by ``Client``. """ #: A flag, which is ``True`` if we're operating on control domain #: and else otherwise. try: SU = open("/proc/xen/capabilities", "rb").read() == b"control_d\n" except (IOError, OSError): SU = False def __init__(self, unix_socket_path=None, xen_bus_path=None, router=None): if router is None: if unix_socket_path or not xen_bus_path: connection = UnixSocketConnection(unix_socket_path) else: connection = XenBusConnection(xen_bus_path) router = Router(connection) self.router = router self.tx_id = 0 def __repr__(self): return "Client({0})".format(self.router.connection) def __copy__(self): return self.__class__(router=self.router) def __enter__(self): self.connect() return self def __exit__(self, *exc_info): self.close() if self.tx_id and not any(exc_info): raise PyXSError("uncommitted transaction") # Private API. # ............ def execute_command(self, op, *args, **kwargs): if not all(map(_re_7bit_ascii.match, args)): raise ValueError(args) kwargs.update(tx_id=self.tx_id, rq_id=next_rq_id()) rvar = self.router.send(Packet(op, b"".join(args), **kwargs)) packet = rvar.get() if packet.op == Op.ERROR: # Erroneous responses are POSIX error code ending with a # ``NUL`` byte. raise error(packet.payload[:-1]) elif packet.op != op or packet.tx_id != self.tx_id: raise UnexpectedPacket(packet) return packet.payload.rstrip(NUL) def ack(self, *args): payload = self.execute_command(*args) if payload != b"OK": raise PyXSError(payload) # Public API. # ........... def connect(self): """Connects to the XenStore daemon. :raises pyxs.exceptions.ConnectionError: if the connection could not be opened. This could happen either because XenStore is not running on the machine or due to the lack of permissions. .. versionadded: 0.4.0 .. warning:: This method is unsafe. Please use client as a context manager to ensure it is properly finalized. """ self.router.start() def close(self): """Finalizes the client. .. warning:: This method is unsafe. Please use client as a context manager to ensure it is properly finalized. """ self.router.terminate() def read(self, path, default=None): """Reads data from a given path. :param bytes path: a path to read from. :param bytes default: default value, to be used if `path` doesn't exist. """ check_path(path) try: return self.execute_command(Op.READ, path + NUL) except PyXSError as e: if e.args[0] == errno.ENOENT and default is not None: return default raise __getitem__ = read def write(self, path, value): """Writes data to a given path. :param bytes value: data to write. :param bytes path: a path to write to. """ check_path(path) self.ack(Op.WRITE, path + NUL, value) __setitem__ = write def mkdir(self, path): """Ensures that a given path exists, by creating it and any missing parents with empty values. If `path` or any parent already exist, its value is left unchanged. :param bytes path: path to directory to create. """ check_path(path) self.ack(Op.MKDIR, path + NUL) def delete(self, path): """Ensures that a given does not exist, by deleting it and all of its children. It is not an error if `path` doesn't exist, but it **is** an error if `path`'s immediate parent does not exist either. :param bytes path: path to directory to remove. """ check_path(path) self.ack(Op.RM, path + NUL) __delitem__ = delete def list(self, path): """Returns a list of names of the immediate children of `path`. :param bytes path: path to list. """ check_path(path) payload = self.execute_command(Op.DIRECTORY, path + NUL) return [] if not payload else payload.split(NUL) def exists(self, path): """Checks if a given `path` exists. :param bytes path: path to check. """ try: self.list(path) except PyXSError as e: if e.args[0] == errno.ENOENT: return False raise else: return True def get_perms(self, path): """Returns a list of permissions for a given `path`, see :exc:`~pyxs.exceptions.InvalidPermission` for details on permission format. :param bytes path: path to get permissions for. """ check_path(path) payload = self.execute_command(Op.GET_PERMS, path + NUL) return payload.split(NUL) def set_perms(self, path, perms): """Sets a access permissions for a given `path`, see :exc:`~pyxs.exceptions.InvalidPermission` for details on permission format. :param bytes path: path to set permissions for. :param list perms: a list of permissions to set. """ check_path(path) check_perms(perms) self.ack(Op.SET_PERMS, path + NUL, *(perm + NUL for perm in perms)) def walk(self, top, topdown=True): """Walk XenStore, yielding 3-tuples ``(path, value, children)`` for each node in the tree, rooted at node `top`. :param bytes top: node to start from. :param bool topdown: see :func:`os.walk` for details. """ children = self.list(top) try: value = self.read(top) except PyXSError: value = b"" # '/' or no read permissions? if topdown: yield top, value, children for child in children: for x in self.walk(posixpath.join(top, child)): yield x if not topdown: yield top, value, children def get_domain_path(self, domid): """Returns the domain's base path, as used for relative requests: e.g. ``b"/local/domain/"``. If a given `domid` doesn't exists the answer is undefined. :param int domid: domain to get base path for. """ return self.execute_command(Op.GET_DOMAIN_PATH, str(domid).encode() + NUL) def is_domain_introduced(self, domid): """Returns ``True`` if ``xenstored`` is in communication with the domain; that is when `INTRODUCE` for the domain has not yet been followed by domain destruction or explicit `RELEASE`; and ``False`` otherwise. :param int domid: domain to check status for. """ payload = self.execute_command(Op.IS_DOMAIN_INTRODUCED, str(domid).encode() + NUL) return {b"T": True, b"F": False}[payload] def introduce_domain(self, domid, mfn, eventchn): """Tells ``xenstored`` to communicate with this domain. :param int domid: a real domain id, (``0`` is forbidden). :param int mfn: address of xenstore page in `domid`. :param int eventchn: an unbound event chanel in `domid`. """ if not domid: raise ValueError("domain 0 cannot be introduced.") self.ack(Op.INTRODUCE, str(domid).encode() + NUL, str(mfn).encode() + NUL, str(eventchn).encode() + NUL) def release_domain(self, domid): """Manually requests ``xenstored`` to disconnect from the domain. :param int domid: domain to disconnect. .. note:: ``xenstored`` will in any case detect domain destruction and disconnect by itself. """ if not self.SU: raise error(errno.EPERM) self.ack(Op.RELEASE, str(domid).encode() + NUL) def resume_domain(self, domid): """Tells ``xenstored`` to clear its shutdown flag for a domain. This ensures that a subsequent shutdown will fire the appropriate watches. :param int domid: domain to resume. """ if not self.SU: raise error(errno.EPERM) self.ack(Op.RESUME, str(domid).encode() + NUL) def set_target(self, domid, target): """Tells ``xenstored`` that a domain is targetting another one, so it should let it tinker with it. This grants domain `domid` full access to paths owned by `target`. Domain `domid` also inherits all permissions granted to `target` on all other paths. :param int domid: domain to set target for. :param int target: target domain (yours truly, Captain). """ if not self.SU: raise error(errno.EPERM) self.ack(Op.SET_TARGET, str(domid).encode() + NUL, str(target).encode() + NUL) def transaction(self): """Starts a new transaction. :returns int: transaction handle. :raises pyxs.exceptions.PyXSError: with :data:`errno.EALREADY` if this client is already in a transaction. .. warning:: Currently ``xenstored`` has a bug that after 2**32 transactions it will allocate id 0 for an actual transaction. """ if self.tx_id: raise error(errno.EALREADY) payload = self.execute_command(Op.TRANSACTION_START, NUL) self.tx_id = int(payload) return self.tx_id def rollback(self): """Rolls back a transaction currently in progress. .. versionchanged: 0.4.0 In previous versions the method gracefully handled attempts to end a transaction when no transaction was running. This is no longer the case. The method will send the corresponding command to XenStore. """ self.ack(Op.TRANSACTION_END, b"F" + NUL) self.tx_id = 0 def commit(self): """Commits a transaction currently in progress. :returns bool: ``False`` if commit failed because of the intervening writes and ``True`` otherwise. In any case transaction is invalidated. The caller is responsible for starting a new transaction, repeating all of the operations a re-committing. .. versionchanged: 0.4.0 In previous versions the method gracefully handled attempts to end a transaction when no transaction was running. This is no longer the case. The method will send the corresponding command to XenStore. """ try: self.ack(Op.TRANSACTION_END, b"T" + NUL) except PyXSError as e: if e.args[0] == errno.EAGAIN: return False raise else: return True finally: self.tx_id = 0 def monitor(self): """Returns a new :class:`Monitor` instance, which is currently the only way of doing PUBSUB. The monitor shares the router with its parent client. Thus closing the client invalidates the monitor. Closing the monitor, on the other hand, had no effect on the router state. .. note:: Using :meth:`monitor` over :class:`~pyxs.connection.XenBusConnection` is currently unsupported, because XenBus does not obey XenStore protocol specification. See `xen-devel`_ discussion for details. .. _xen-devel: \ http://lists.xen.org/archives/html/xen-devel/2016-02/msg03737 """ if isinstance(self.router.connection, XenBusConnection): raise PyXSError("using ``Monitor`` over XenBus is not supported", UserWarning) return Monitor(copy.copy(self)) class Monitor(object): """Monitor implements minimal PUBSUB functionality on top of XenStore. >>> with Client() as c: ... m = c.monitor(): ... m.watch("foo/bar") ... print(next(c.wait())) Event(...) :param Client client: a reference to the parent client. .. note:: When used as a context manager the monitor will try to unwatch all watched paths. """ def __init__(self, client): self.client = client self.events = queue.Queue() self.unwatch_queue = set() def __enter__(self): return self def __exit__(self, *exc_info): self.close() @property def watched(self): """A set of paths currently watched by the monitor.""" return set(wpath for wpath, token in self.unwatch_queue) def close(self): """Finalizes the monitor by unwatching all watched paths.""" for wpath, token in list(self.unwatch_queue): self.unwatch(wpath, token) def watch(self, wpath, token): """Adds a watch. Any alteration to the watched path generates an event. This includes path creation, removal, contents change or permission change. An event can also be triggered spuriously. Changes made in transactions cause an event only if and when committed. :param bytes wpath: path to watch. :param bytes token: watch token, returned in watch notification. """ check_watch_path(wpath) self.client.router.subscribe(token, self) self.client.ack(Op.WATCH, wpath + NUL, token + NUL) self.unwatch_queue.add((wpath, token)) def unwatch(self, wpath, token): """Removes a previously added watch. :param bytes wpath: path to unwatch. :param bytes token: watch token, passed to :meth:`watch`. """ check_watch_path(wpath) self.client.ack(Op.UNWATCH, wpath + NUL, token + NUL) self.client.router.unsubscribe(token, self) self.unwatch_queue.discard((wpath, token)) def wait(self, unwatched=False): """Yields events for all of the watched paths. An event is a ``(path, token)`` pair, where the first element is event path, i.e. the actual path that was modified, and the second -- a token, passed to :meth:`watch`. :param bool unwatched: if ``True`` :meth:`wait` might yield spurious unwatched packets, otherwise these are dropped. Defaults to ``False``. """ while True: with self.events.not_empty: while not self.events._qsize(): _condition_wait(self.events.not_empty) event = wpath, token = self.events.get_nowait() # Check that event path or its parent is watched. while wpath and (wpath, token) not in self.unwatch_queue: wpath = posixpath.dirname(wpath) if wpath or unwatched: yield event