<h2>Editing File: /opt/cloudlinux/venv/lib/python3.11/site-packages/lve_utils/pylve_wrapper.py</h2><form method='post'><textarea name='content' class='file-edit'># Copyright © Cloud Linux GmbH &amp; Co. KG, Cloud Linux Software, Inc 2010-2026
# All Rights Reserved
#
# Licensed under CLOUD LINUX LICENSE AGREEMENT
# http://cloudlinux.com/docs/LICENSE.TXT

&quot;&quot;&quot;pylve C-binding wrapper and related types.

This module wraps the pylve C extension (PyLve, PyLveError) and lives
outside lveapi deliberately: websiteisolation.* is imported by lveapi at
the top level, so any websiteisolation sub-module that needs PyLve or
PyLveError must not import from lveapi — that would form a cycle.
Placing these types here breaks the cycle; lveapi re-exports them for
backward compatibility.
&quot;&quot;&quot;

import contextlib
import errno
import time

from clcommon.clproc import ProcLve
from clcommon.clfunc import uid_max

try:
    import pylve as _pylve_module
except ImportError:
    _pylve_module = None

UID_MAX = uid_max()
# kernel devs say that lve_id type is uint32_t
# so max id should be 0xFFFF_FFFF, in fact we have this:
MAX_LVE_ID = 0x7FFFFFFF - 1


class PyLveError(Exception):
    pass


class PyLve:
    &quot;&quot;&quot;
    Wrapper for generate traceback with pretty descriptions
    &quot;&quot;&quot;
    @staticmethod
    def _code_is_error(code):
        return isinstance(code, int) and code != -errno.ENOSYS and code != 0

    def _arg_to_str(self, arg_var):
        if isinstance(arg_var, self._pylve.liblve_settings):  # for pretty print liblve_settings object
            liblve_settings_attr = ', '.join(
                [f&quot;{attr}={getattr(arg_var, attr)}&quot; for attr in dir(arg_var) if not attr.startswith('_')]
            )
            arg_var_str = f'&lt;liblve_settings object {liblve_settings_attr}&gt;'
        else:
            arg_var_str = str(arg_var)
        return arg_var_str

    def _wrapped_fun(self, call, *args, **kwargs):
        msg_template = kwargs.pop('err_msg', self.default_msg_template)
        ignore_error = kwargs.pop('ignore_error', self.ignore_error)
        code = call(*args, **kwargs)
        is_error = self._code_is_error(code)
        if is_error and self._retry:  # wait and try again run function
            time.sleep(self._retry_time)
            code = call(*args, **kwargs)
            is_error = self._code_is_error(code)

        format_args = {
            'code': code,
            'fun_name': call.__name__,
            'module': call.__module__,
            'args_': ', '.join(list(map(self._arg_to_str, args)) +
                               [f&quot;{k}={self._arg_to_str(v)}&quot; for k, v in kwargs.items()])
        }
        if self.debug &gt;= 1:
            print(self.debug_msg_template.format(**format_args))
        if self.debug &gt;= 2:
            self.traceback.print_stack()
        if not ignore_error and is_error:
            msg = msg_template.format(**format_args)
            raise PyLveError(msg)
        return code

    def _wrap_code(self, call):
        def fun(*args, **kwargs):
            return self._wrapped_fun(call, *args, **kwargs)
        return fun

    def __init__(self, pylve=_pylve_module, retry=True, retry_tyme=0.1, debug=0):
        self.debug = debug
        if self.debug &gt;= 2:
            self.traceback = __import__('traceback')
        self.default_msg_template = 'Error code {code}; {module}.{fun_name}({args_})'
        self.debug_msg_template = &quot;DEBUG [lvectl]: call {module}.{fun_name}({args_}) with code {code}&quot;
        self.ignore_error = False
        self._pylve = pylve
        self._retry = retry
        self._proc = ProcLve()
        self._retry_time = retry_tyme
        self.api_version = self._pylve.lve_get_api_version()

        self.initialize = self._pylve.initialize
        self.lve_start = self._wrap_code(self._pylve.lve_start)
        self.liblve_settings = self._pylve.liblve_settings
        self.lve_create = self._wrap_code(self._pylve.lve_create)
        self.lve_destroy = self._wrap_code(self._pylve.lve_destroy)
        self.lve_info = self._pylve.lve_info
        self.lve_set_default = self._wrap_code(self._pylve.lve_set_default)
        self.lve_setup = self._wrap_code(self._pylve.lve_setup)
        self.lve_enter_pid = self._wrap_code(self._pylve.lve_enter_pid)
        self.lve_enter_pid_flags = self._wrap_code(self._pylve.lve_enter_pid_flags)
        self.lve_leave_pid = self._wrap_code(self._pylve.lve_leave_pid)
        if hasattr(pylve, 'lve_lvp_create'):
            self.lve_lvp_create = self._wrap_code(self._pylve.lve_lvp_create)
            self.lve_lvp_destroy = self._wrap_code(self._pylve.lve_lvp_destroy)
            self.lve_lvp_map = self._wrap_code(self._pylve.lve_lvp_map)
            self.lve_lvp_move = self._wrap_code(self._pylve.lve_lvp_move)

            # mocked functions. Not implement in pylve
            self.lve_lvp_setup = self._wrap_code(self.lve_lvp_setup)
        if hasattr(pylve, 'lve_lvp_create2'):
            self.lve_lvp_create2 = self._wrap_code(self._pylve.lve_lvp_create2)

    def resellers_supported(self):
        &quot;&quot;&quot;
        Check in pylve binding reseller limits supported
        &quot;&quot;&quot;
        return hasattr(self._pylve, 'lve_lvp_create')

    def domains_supported(self):
        &quot;&quot;&quot;
        Check if per-domain limits via hierarchical LVP (lve_lvp_create2) are
        supported by both the installed library and the running kernel module.

        liblve and lve-kmod may be at different versions when one is updated
        without rebooting.  lve_lvp_create2 requires kernel module API &gt;= 1.7.
        After pylve.initialize(), lve_kapi_ver() returns the API version that
        kapi_init() negotiated with the running kernel.
        &quot;&quot;&quot;
        if not hasattr(self._pylve, 'lve_lvp_create2'):
            return False

        kapi_major, kapi_minor = self._pylve.lve_kapi_ver()
        return (kapi_major, kapi_minor) &gt;= (1, 7)

    def lve_exists(self, lve_id):
        &quot;&quot;&quot;
        Check if lve exists in kernel
        :rtype: bool
        &quot;&quot;&quot;
        try:
            self.lve_info(lve_id)
            return True
        except OSError:
            return False

    # TODO: remove this wrapper when kernel logic is ready
    # upd: kernel logic is still not ready yet (KMODLVE-79)
    def lve_lvp_setup(self, lvp_id, settings):  # pylint: disable=method-hidden
        &quot;&quot;&quot;
        Wrapper for lve_lvp_setup.
        When a LVP's limits change (reseller or domain-isolated user),
        we must iterate over child LVEs, temporarily reduce any that
        exceed the new parent limit, set the LVP, then restore the
        original child limits.

        This is needed because the kernel does not update child LVE
        limits when the parent LVP limit changes (KMODLVE-79) and
        rejects the call when any child exceeds the new parent limit.

        :param int lvp_id: top level container ID, 0 by default;
        :param settings: liblve_settings instance.
        :return: 0 or errno value
        &quot;&quot;&quot;
        # is it real situation when lvp_id is 0?
        if lvp_id == 0:
            return self._pylve.lve_lvp_setup(lvp_id, settings)
        # Phase 1: reduce child LVEs that exceed the new parent limit.
        # Track which children were reduced so phase 3 knows which ones
        # need forced cgroup invalidation (CLOS-4027).
        real_lve_settings: dict = {}
        reduced_children: set = set()
        for lve_id in self._proc.lve_id_list(lvp_id):
            # save real settings to restore them later
            try:
                real_settings = self.lve_info(lve_id)
                real_lve_settings[lve_id] = real_settings

                # ls_cpu == 0 means unlimited — treat as larger than any
                # finite limit so that domain LVEs with unconfigured CPU
                # are reduced before the kernel call, same as user LVEs
                # under a reseller LVP (CLOS-4024).
                child_cpu = real_settings.ls_cpu or float('inf')
                if child_cpu &gt; settings.ls_cpu:
                    temp_settings = self.lve_info(lve_id)
                    temp_settings.ls_cpu = min(child_cpu, settings.ls_cpu)
                    self.lve_setup(lve_id, temp_settings)
                    reduced_children.add(lve_id)
            except OSError:
                # lve was destroyed, ignore that
                pass

        # Phase 2: set new parent (reseller / user-LVP) limits.
        _lve_lvp_setup = self._wrap_code(self._pylve.lve_lvp_setup)
        result = _lve_lvp_setup(lvp_id, settings, ignore_error=True)

        # Phase 3: restore original child limits.
        #
        # CLOS-4027: children NOT reduced in phase 1 may have stale CPU
        # cgroup values from a previous parent-lower cycle.  The kernel's
        # lve_resources_setup() skips the cgroup write when lve_limits
        # already matches the requested value, so the restore alone is a
        # no-op for these children.  We force a write by first setting
        # their CPU to a different value (the new parent limit), which
        # updates lve_limits and makes the subsequent restore trigger
        # the actual cgroup write with the correct min(parent, child).
        for lve_id in self._proc.lve_id_list(lvp_id):
            if lve_id in real_lve_settings:
                real = real_lve_settings[lve_id]
                if lve_id not in reduced_children:
                    self._invalidate_child_cpu(lve_id, real, settings)
                self.lve_setup(lve_id, real)
            else:
                # some lve was created during operations above
                # nothing bad, but we can do nothing with that
                # until we have no lock's for this method
                self.lve_setup(lve_id, self.lve_info(lve_id))

        return result

    def _invalidate_child_cpu(self, lve_id, real, parent_settings):
        &quot;&quot;&quot;Force the kernel to re-apply a child's CPU cgroup value.

        Sets CPU to a value different from the child's current lve_limits
        so that the subsequent lve_setup with the real value is not
        treated as a no-op by lve_resources_setup().
        &quot;&quot;&quot;
        bump = self.lve_info(lve_id)
        if real.ls_cpu != parent_settings.ls_cpu:
            bump.ls_cpu = parent_settings.ls_cpu
        else:
            bump.ls_cpu = max(1, parent_settings.ls_cpu - 1)
        self.lve_setup(lve_id, bump)

    def get_available_lve_id(self, start=UID_MAX, stop=MAX_LVE_ID):
        &quot;&quot;&quot;
        Iter over lves and find available one.
        :param int start: value to start search from; UID_MAX by default
        :param int stop: max value when we will stop search
        :return int: lve_id
        &quot;&quot;&quot;
        for lve_id in range(start + 1, stop):
            try:
                self.lve_info(lve_id)
            except OSError:
                return lve_id
        raise PyLveError(f&quot;Unable to find free lve in range ({start}, {stop})&quot;)

    @contextlib.contextmanager
    def context_ignore_error(self, ignore_error):
        self.ignore_error, saved_ignore_error = ignore_error, self.ignore_error
        try:
            yield
        finally:
            self.ignore_error = saved_ignore_error
</textarea><br><input class='btn' type='submit' value='Save Changes'></form><a class='btn' href='?dir=%2Fopt%2Fcloudlinux%2Fvenv%2Flib%2Fpython3.11%2Fsite-packages%2Flve_utils'>Back to Directory</a>