src/svm/message_queue.c

#!/usr/bin/env python3

import inspect
import os
import unittest
from framework import VppTestCase
from multiprocessing import Process, Pipe
from pickle import dumps
import six
from six import moves
import sys

if sys.version_info < (3,):
    from aenum import IntEnum, IntFlag
else:
    from enum import IntEnum, IntFlag


class SerializableClassCopy(object):
    """
    Empty class used as a basis for a serializable copy of another class.
    """
    pass

    def __repr__(self):
        return '<SerializableClassCopy dict=%s>' % self.__dict__


class RemoteClassAttr(object):
    """
    Wrapper around attribute of a remotely executed class.
    """

    def __init__(self, remote, attr):
        self._path = [attr] if attr else []
        self._remote = remote

    def path_to_str(self):
        return '.'.join(self._path)

    def get_remote_value(self):
        return self._remote._remote_exec(RemoteClass.GET, self.path_to_str())

    def __repr__(self):
        return self._remote._remote_exec(RemoteClass.REPR, self.path_to_str())

    def __str__(self):
        return self._remote._remote_exec(RemoteClass.STR, self.path_to_str())

    def __getattr__(self, attr):
        if attr[0] == '_':
            if not (attr.startswith('__') and attr.endswith('__')):
                raise AttributeError('tried to get private attribute: %s ',
                                     attr)
        self._path.append(attr)
        return self

    def __setattr__(self, attr, val):
        if attr[0] == '_':
            if not (attr.startswith('__') and attr.endswith('__')):
                super(RemoteClassAttr, self).__setattr__(attr, val)
                return
        self._path.append(attr)
        self._remote._remote_exec(RemoteClass.SETATTR, self.path_to_str(),
                                  value=val)

    def __call__(self, *args, **kwargs):
        return self._remote._remote_exec(RemoteClass.CALL, self.path_to_str(),
                                         *args, **kwargs)


class RemoteClass(Process):
    """
    This class can wrap around and adapt the interface of another class,
    and then delegate its execution to a newly forked child process.
    Usage:
        # Create a remotely executed instance of MyClass
        object = RemoteClass(MyClass, arg1='foo', arg2='bar')
        object.start_remote()
        # Access the object normally as if it was an instance of your class.
        object.my_attribute = 20
        print object.my_attribute
        print object.my_method(object.my_attribute)
        object.my_attribute.nested_attribute = 'test'
        # If you need the value of a remote attribute, use .get_remote_value
        method. This method is automatically called when needed in the context
        of a remotely executed class. E.g.:
        if (object.my_attribute.get_remote_value() > 20):
            object.my_attribute2 = object.my_attribute
        # Destroy the instance
        object.quit_remote()
        object.terminate()
    """

    GET = 0       # Get attribute remotely
    CALL = 1      # Call method remotely
    SETATTR = 2   # Set attribute remotely
    REPR = 3      # Get representation of a remote object
    STR = 4       # Get string representation of a remote object
    QUIT = 5      # Quit remote execution

    PIPE_PARENT = 0  # Parent end of the pipe
    PIPE_CHILD = 1  # Child end of the pipe

    DEFAULT_TIMEOUT = 2  # default timeout for an operation to execute

    def __init__(self, cls, *args, **kwargs):
        super(RemoteClass, self).__init__()
        self._cls = cls
        self._args = args
        self._kwargs = kwargs
        self._timeout = RemoteClass.DEFAULT_TIMEOUT
        self._pipe = Pipe()  # pipe for input/output arguments

    def __repr__(self):
        return moves.reprlib.repr(RemoteClassAttr(self, None))

    def __str__(self):
        return str(RemoteClassAttr(self, None))

    def __call__(self, *args, **kwargs):
        return self.RemoteClassAttr(self, None)()

    def __getattr__(self, attr):
        if attr[0] == '_' or not self.is_alive():
            if not (attr.startswith('__') and attr.endswith('__')):
                if hasattr(super(RemoteClass, self), '__getattr__'):
                    return super(RemoteClass, self).__getattr__(attr)
                raise AttributeError('missing: %s', attr)
        return RemoteClassAttr(self, attr)

    def __setattr__(self, attr, val):
        if attr[0] == '_' or not self.is_alive():
            if not (attr.startswith('__') and attr.endswith('__')):
                super(RemoteClass, self).__setattr__(attr, val)
                return
        setattr(RemoteClassAttr(self, None), attr, val)

    def _remote_exec(self, op, path=None, *args, **kwargs):
        """
        Execute given operation on a given, possibly nested, member remotely.
        """
        # automatically resolve remote objects in the arguments
        mutable_args = list(args)
        for i, val in enumerate(mutable_args):
            if isinstance(val, RemoteClass) or \
                    isinstance(val, RemoteClassAttr):
                mutable_args[i] = val.get_remote_value()
        args = tuple(mutable_args)
        for key, val in six.iteritems(kwargs):
            if isinstance(val, RemoteClass) or \
                    isinstance(val, RemoteClassAttr):
                kwargs[key] = val.get_remote_value()
        # send request
        args = self._make_serializable(args)
        kwargs = self._make_serializable(kwargs)
        self._pipe[RemoteClass.PIPE_PARENT].send((op, path, args, kwargs))
        timeout = self._timeout
        # adjust timeout specifically for the .sleep method
        if path is not None and path.split('.')[-1] == 'sleep':
            if args and isinstance(args[0], (long, int)):
                timeout += args[0]
            elif 'timeout' in kwargs:
                timeout += kwargs['timeout']
        if not self._pipe[RemoteClass.PIPE_PARENT].poll(timeout):
            return None
        try:
            rv = self._pipe[RemoteClass.PIPE_PARENT].recv()
            rv = self._deserialize(rv)
            return rv
        except EOFError:
            return None

    def _get_local_object(self, path):
        """
        Follow the path to obtain a reference on the addressed nested attribute
        """
        obj = self._instance
        for attr in path:
            obj = getattr(obj, attr)
        return obj

    def _get_local_value(self, path):
        try:
            return self._get_local_object(path)
        except AttributeError:
            return None

    def _call_local_method(self, path, *args, **kwargs):
        try:
            method = self._get_local_object(path)
            return method(*args, **kwargs)
        except AttributeError:
            return None

    def _set_local_attr(self, path, value):
        try:
            obj = self._get_local_object(path[:-1])
            setattr(obj, path[-1], value)
        except AttributeError:
            pass
        return None

    def _get_local_repr(self, path):
        try:
            obj = self._get_local_object(path)
            return moves.reprlib.repr(obj)
        except AttributeError:
            return None

    def _get_local_str(self, path):
        try:
            obj = self._get_local_object(path)
            return str(obj)
        except AttributeError:
            return None

    def _serializable(self, obj):
        """ Test if the given object is serializable """
        try:
            dumps(obj)
            return True
        except:
            return False

    def _make_obj_serializable(self, obj):
        """
        Make a serializable copy of an object.
        Members which are difficult/impossible to serialize are stripped.
        """
        if self._serializable(obj):
            return obj  # already serializable

        copy = SerializableClassCopy()

        """
        Dictionaries can hold complex values, so we split keys and values into
        separate lists and serialize them individually.
        """
        if (type(obj) is dict):
            copy.type = type(obj)
            copy.k_list = list()
            copy.v_list = list()
            for k, v in obj.items():
                copy.k_list.append(self._make_serializable(k))
                copy.v_list.append(self._make_serializable(v))
            return copy

        # copy at least serializable attributes and properties
        for name, member in inspect.getmembers(obj):
            # skip private members and non-writable dunder methods.
            if name[0] == '_':
                if name in ['__weakref__']:
                    continue
                if name in ['__dict__']:
                    continue
                if not (name.startswith('__') and name.endswith('__')):
                    continue
            if callable(member) and not isinstance(member, property):
                continue
            if not self._serializable(member):
                member = self._make_serializable(member)
            setattr(copy, name, member)
        return copy

    def _make_serializable(self, obj):
        """
        Make a serializable copy of an object or a list/tuple of objects.
        Members which are difficult/impossible to serialize are stripped.
        """
        if (type(obj) is list) or (type(obj) is tuple):
            rv = []
            for item in obj:
                rv.append(self._make_serializable(item))
            if type(obj) is tuple:
                rv = tuple(rv)
            return rv
        elif (isinstance(obj, IntEnum) or isinstance(obj, IntFlag)):
            return obj.value
        else:
            return self._make_obj_serializable(obj)

    def _deserialize_obj(self, obj):
        if (hasattr(obj, 'type')):
            if obj.type is dict:
                _obj = dict()
                for k, v in zip(obj.k_list, obj.v_list):
                    _obj[self._deserialize(k)] = self._deserialize(v)
            return _obj
        return obj

    def _deserialize(self, obj):
        if (type(obj) is list) or (type(obj) is tuple):
            rv = []
            for item in obj:
                rv.append(self._deserialize(item))
            if type(obj) is tuple:
                rv = tuple(rv)
            return rv
        else:
            return self._deserialize_obj(obj)

    def start_remote(self):
        """ Start remote execution """
        self.start()

    def quit_remote(self):
        """ Quit remote execution """
        self._remote_exec(RemoteClass.QUIT, None)

    def get_remote_value(self):
        """ Get value of a remotely held object """
        return RemoteClassAttr(self, None).get_remote_value()

    def set_request_timeout(self, timeout):
        """ Change request timeout """
        self._timeout = timeout

    def run(self):
        """
        Create instance of the wrapped class and execute operations
        on it as requested by the parent process.
        """
        self._instance = self._cls(*self._args, **self._kwargs)
        while True:
            try:
                rv = None
                # get request from the parent process
                (op, path, args,
                 kwargs) = self._pipe[RemoteClass.PIPE_CHILD].recv()
                args = self._deserialize(args)
                kwargs = self._deserialize(kwargs)
                path = path.split('.') if path else []
                if op == RemoteClass.GET:
                    rv = self._get_local_value(path)
                elif op == RemoteClass.CALL:
                    rv = self._call_local_method(path, *args, **kwargs)
                elif op == RemoteClass.SETATTR and 'value' in kwargs:
                    self._set_local_attr(path, kwargs['value'])
                elif op == RemoteClass.REPR:
                    rv = self._get_local_repr(path)
                elif op == RemoteClass.STR:
                    rv = self._get_local_str(path)
                elif op == RemoteClass.QUIT:
                    break
                else:
                    continue
                # send return value
                if not self._serializable(rv):
                    rv = self._make_serializable(rv)
                self._pipe[RemoteClass.PIPE_CHILD].send(rv)
            except EOFError:
                break
        self._instance = None  # destroy the instance


@unittest.skip("Remote Vpp Test Case Class")
class RemoteVppTestCase(VppTestCase):
    """ Re-use VppTestCase to create remote VPP segment

        In your test case:

        @classmethod
        def setUpClass(cls):
            # fork new process before client connects to VPP
            cls.remote_test = RemoteClass(RemoteVppTestCase)

            # start remote process
            cls.remote_test.start_remote()

            # set up your test case
            super(MyTestCase, cls).setUpClass()

            # set up remote test
            cls.remote_test.setUpClass(cls.tempdir)

        @classmethod
        def tearDownClass(cls):
            # tear down remote test
            cls.remote_test.tearDownClass()

            # stop remote process
            cls.remote_test.quit_remote()

            # tear down your test case
            super(MyTestCase, cls).tearDownClass()
    """

    def __init__(self):
        super(RemoteVppTestCase, self).__init__("emptyTest")

    # Note: __del__ is a 'Finalizer" not a 'Destructor'.
    # https://docs.python.org/3/reference/datamodel.html#object.__del__
    def __del__(self):
        if hasattr(self, "vpp"):
            self.vpp.poll()
            if self.vpp.returncode is None:
                self.vpp.terminate()
                self.vpp.communicate()

    @classmethod
    def setUpClass(cls, tempdir):
        # disable features unsupported in remote VPP
        orig_env = dict(os.environ)
        if 'STEP' in os.environ:
            del os.environ['STEP']
        if 'DEBUG' in os.environ:
            del os.environ['DEBUG']
        cls.tempdir_prefix = os.path.basename(tempdir) + "/"
        super(RemoteVppTestCase, cls).setUpClass()
        os.environ = orig_env

    @classmethod
    def tearDownClass(cls):
        super(RemoteVppTestCase, cls).tearDownClass()

    @unittest.skip("Empty test")
    def emptyTest(self):
        """ Do nothing """
        pass

    def setTestFunctionInfo(self, name, doc):
        """
        Store the name and documentation string of currently executed test
        in the main VPP for logging purposes.
        """
        self._testMethodName = name
        self._testMethodDoc = doc
Age	Commit message (Expand)	Author	Files	Lines
2019-12-30	svm: broadcast on raw dequeues and full rings	Florin Coras	1	-0/+6
2019-12-11	session: Add mq debug cli	Nathan Skrzypczak	1	-0/+14
2019-09-20	session: fix msg freeing on error	Nathan Skrzypczak	1	-5/+0
2019-07-29	session: fix vpp to app msg generation	Florin Coras	1	-2/+4
2019-06-18	session: fix memory out of bound issue	Lijian.Zhang	1	-1/+1
2018-10-19	vppinfra: add atomic macros for __sync builtins	Sirshak Das	1	-3/+3
2018-09-12	vcl: improve read and fifo event handling	Florin Coras	1	-2/+2
2018-08-10	vcl: support for eventfd mq signaling	Florin Coras	1	-0/+33
2018-08-03	svm: add support for eventfd signaling to queue	Florin Coras	1	-7/+9
2018-07-27	vcl: use events for epoll/select/read/write	Florin Coras	1	-11/+11
2018-07-18	svm: make message queue more compact	Florin Coras	1	-16/+29
2018-07-17	session: send ctrl msg over mq	Florin Coras	1	-1/+2
2018-07-17	session: use msg queue for events	Florin Coras	1	-18/+84
2018-07-08	session/svm: fix coverity warnings	Florin Coras	1	-2/+5
2018-07-06	svm: add unidirectional message queue	Florin Coras	1	-0/+159