Gecko [ fastearnlimited.com ]

Name	Size	Permission	Date
__pycache__	[ DIR ]	drwxr-xr-x	2026-02-27 05:56
instance	[ DIR ]	drwxr-xr-x	2026-02-27 05:56
__init__.py	49 B	-rw-r--r--	2025-08-24 21:48
base.py	22.76 KB	-rw-r--r--	2025-08-24 21:48
decorators.py	983 B	-rw-r--r--	2025-08-24 21:48
exceptions.py	99 B	-rw-r--r--	2025-08-24 21:48
hotplug.py	4.62 KB	-rw-r--r--	2025-08-24 21:48
plugin_acpi.py	2.39 KB	-rw-r--r--	2025-08-24 21:48
plugin_audio.py	3.16 KB	-rw-r--r--	2025-08-24 21:48
plugin_bootloader.py	24.86 KB	-rw-r--r--	2025-08-24 21:48
plugin_cpu.py	30.58 KB	-rw-r--r--	2025-08-24 21:48
plugin_disk.py	16.77 KB	-rw-r--r--	2025-08-24 21:48
plugin_eeepc_she.py	3 KB	-rw-r--r--	2025-08-24 21:48
plugin_irq.py	9.73 KB	-rw-r--r--	2025-08-24 21:48
plugin_irqbalance.py	3.54 KB	-rw-r--r--	2025-08-24 21:48
plugin_modules.py	4.8 KB	-rw-r--r--	2025-08-24 21:48
plugin_mounts.py	5.44 KB	-rw-r--r--	2025-08-24 21:48
plugin_net.py	26.38 KB	-rw-r--r--	2025-08-24 21:48
plugin_rtentsk.py	1.12 KB	-rw-r--r--	2025-08-24 21:48
plugin_scheduler.py	58.48 KB	-rw-r--r--	2025-08-24 21:48
plugin_script.py	3.91 KB	-rw-r--r--	2025-08-24 21:48
plugin_scsi_host.py	3.89 KB	-rw-r--r--	2025-08-24 21:48
plugin_selinux.py	2.26 KB	-rw-r--r--	2025-08-24 21:48
plugin_service.py	10.45 KB	-rw-r--r--	2025-08-24 21:48
plugin_sysctl.py	6.01 KB	-rw-r--r--	2025-08-24 21:48
plugin_sysfs.py	2.63 KB	-rw-r--r--	2025-08-24 21:48
plugin_systemd.py	5.32 KB	-rw-r--r--	2025-08-24 21:48
plugin_uncore.py	5.97 KB	-rw-r--r--	2025-08-24 21:48
plugin_usb.py	1.97 KB	-rw-r--r--	2025-08-24 21:48
plugin_video.py	5.97 KB	-rw-r--r--	2025-08-24 21:48
plugin_vm.py	8.67 KB	-rw-r--r--	2025-08-24 21:48
repository.py	1.49 KB	-rw-r--r--	2025-08-24 21:48

Rename

import re
import tuned.consts as consts
import tuned.profiles.variables
import tuned.logs
import collections
from tuned.utils.commands import commands
import os
from subprocess import Popen, PIPE

log = tuned.logs.get()

class Plugin(object):
	"""
	Base class for all plugins.

Plugins change various system settings in order to get desired performance or power
	saving. Plugins use Monitor objects to get information from the running system.

Intentionally a lot of logic is included in the plugin to increase plugin flexibility.
	"""

def __init__(self, monitors_repository, storage_factory, hardware_inventory, device_matcher, device_matcher_udev, instance_factory, global_cfg, variables):
		"""Plugin constructor."""

self._storage = storage_factory.create(self.__class__.__name__)
		self._monitors_repository = monitors_repository
		self._hardware_inventory = hardware_inventory
		self._device_matcher = device_matcher
		self._device_matcher_udev = device_matcher_udev
		self._instance_factory = instance_factory

self._instances = collections.OrderedDict()
		self._init_commands()

self._global_cfg = global_cfg
		self._variables = variables
		self._has_dynamic_options = False
		self._devices_inited = False

self._options_used_by_dynamic = self._get_config_options_used_by_dynamic()

self._cmd = commands()

def cleanup(self):
		self.destroy_instances()

def init_devices(self):
		if not self._devices_inited:
			self._init_devices()
			self._devices_inited = True

@property
	def name(self):
		return self.__class__.__module__.split(".")[-1].split("_", 1)[1]

#
	# Plugin configuration manipulation and helpers.
	#

@classmethod
	def _get_config_options(self):
		"""Default configuration options for the plugin."""
		return {}

@classmethod
	def get_config_options_hints(cls):
		"""Explanation of each config option function"""
		return {}

@classmethod
	def _get_config_options_used_by_dynamic(self):
		"""List of config options used by dynamic tuning. Their previous values will be automatically saved and restored."""
		return []

def _get_effective_options(self, options):
		"""Merge provided options with plugin default options."""
		# TODO: _has_dynamic_options is a hack
		effective = self._get_config_options().copy()
		for key in options:
			if key in effective or self._has_dynamic_options:
				effective[key] = options[key]
			else:
				log.warning("Unknown option '%s' for plugin '%s'." % (key, self.__class__.__name__))
		return effective

def _option_bool(self, value):
		if type(value) is bool:
			return value
		value = str(value).lower()
		return value == "true" or value == "1"

#
	# Interface for manipulation with instances of the plugin.
	#

def create_instance(self, name, priority, devices_expression, devices_udev_regex, script_pre, script_post, options):
		"""Create new instance of the plugin and seize the devices."""
		if name in self._instances:
			raise Exception("Plugin instance with name '%s' already exists." % name)

effective_options = self._get_effective_options(options)
		instance = self._instance_factory.create(self, name, priority, devices_expression, devices_udev_regex, \
			script_pre, script_post, effective_options)
		self._instances[name] = instance
		self._instances = collections.OrderedDict(sorted(self._instances.items(), key=lambda x: x[1].priority))

return instance

def destroy_instance(self, instance):
		"""Destroy existing instance."""
		if instance._plugin != self:
			raise Exception("Plugin instance '%s' does not belong to this plugin '%s'." % (instance, self))
		if instance.name not in self._instances:
			raise Exception("Plugin instance '%s' was already destroyed." % instance)

instance = self._instances[instance.name]
		self._destroy_instance(instance)
		del self._instances[instance.name]

def initialize_instance(self, instance):
		"""Initialize an instance."""
		log.debug("initializing instance %s (%s)" % (instance.name, self.name))
		self._instance_init(instance)

def destroy_instances(self):
		"""Destroy all instances."""
		for instance in list(self._instances.values()):
			log.debug("destroying instance %s (%s)" % (instance.name, self.name))
			self._destroy_instance(instance)
		self._instances.clear()

def _destroy_instance(self, instance):
		self.release_devices(instance)
		self._instance_cleanup(instance)

def _instance_init(self, instance):
		raise NotImplementedError()

def _instance_cleanup(self, instance):
		raise NotImplementedError()

#
	# Devices handling
	#

def _init_devices(self):
		self._devices_supported = False
		self._assigned_devices = set()
		self._free_devices = set()

def _get_device_objects(self, devices):
		"""Override this in a subclass to transform a list of device names (e.g. ['sda'])
		   to a list of pyudev.Device objects, if your plugin supports it"""
		return None

def _get_matching_devices(self, instance, devices):
		if instance.devices_udev_regex is None:
			return set(self._device_matcher.match_list(instance.devices_expression, devices))
		else:
			udev_devices = self._get_device_objects(devices)
			if udev_devices is None:
				log.error("Plugin '%s' does not support the 'devices_udev_regex' option", self.name)
				return set()
			udev_devices = self._device_matcher_udev.match_list(instance.devices_udev_regex, udev_devices)
			return set([x.sys_name for x in udev_devices])

def assign_free_devices(self, instance):
		if not self._devices_supported:
			return

log.debug("assigning devices to instance %s" % instance.name)
		to_assign = self._get_matching_devices(instance, self._free_devices)
		instance.active = len(to_assign) > 0
		if not instance.active:
			log.warning("instance %s: no matching devices available" % instance.name)
		else:
			name = instance.name
			if instance.name != self.name:
				name += " (%s)" % self.name
			log.info("instance %s: assigning devices %s" % (name, ", ".join(to_assign)))
			instance.assigned_devices.update(to_assign) # cannot use |=
			self._assigned_devices |= to_assign
			self._free_devices -= to_assign

def release_devices(self, instance):
		if not self._devices_supported:
			return

to_release = (instance.processed_devices \
				| instance.assigned_devices) \
				& self._assigned_devices

instance.active = False
		instance.processed_devices.clear()
		instance.assigned_devices.clear()
		self._assigned_devices -= to_release
		self._free_devices |= to_release

#
	# Tuning activation and deactivation.
	#

def _run_for_each_device(self, instance, callback, devices):
		if not self._devices_supported:
			devices = [None, ]

for device in devices:
			callback(instance, device)

def _instance_pre_static(self, instance, enabling):
		pass

def _instance_post_static(self, instance, enabling):
		pass

def _safe_script_path(self, path):
		path = os.path.realpath(path)
		profile_paths = self._global_cfg.get_list(consts.CFG_PROFILE_DIRS, consts.CFG_DEF_PROFILE_DIRS)
		for p in profile_paths:
			if path.startswith(p):
				return True
		return False

def _call_device_script(self, instance, script, op, devices, rollback = consts.ROLLBACK_SOFT):
		if script is None:
			return None
		if len(devices) == 0:
			log.warning("Instance '%s': no device to call script '%s' for." % (instance.name, script))
			return None
		if not script.startswith("/"):
			log.error("Relative paths cannot be used in script_pre or script_post. " \
				+ "Use ${i:PROFILE_DIR}.")
			return False
		if not self._safe_script_path(script):
			log.error("Paths outside of the profile directories cannot be used in the " \
				+ "script_pre or script_post, ignoring script: '%s'" % script)
			return False
		dir_name = os.path.dirname(script)
		ret = True
		for dev in devices:
			environ = os.environ
			environ.update(self._variables.get_env())
			arguments = [op]
			if rollback == consts.ROLLBACK_FULL:
				arguments.append("full_rollback")
			arguments.append(dev)
			log.info("calling script '%s' with arguments '%s'" % (script, str(arguments)))
			log.debug("using environment '%s'" % str(list(environ.items())))
			try:
				proc = Popen([script] +  arguments, \
						stdout=PIPE, stderr=PIPE, \
						close_fds=True, env=environ, \
						cwd = dir_name, universal_newlines = True)
				out, err = proc.communicate()
				if proc.returncode:
					log.error("script '%s' error: %d, '%s'" % (script, proc.returncode, err[:-1]))
					ret = False
			except (OSError,IOError) as e:
				log.error("script '%s' error: %s" % (script, e))
				ret = False
		return ret

def instance_apply_tuning(self, instance):
		"""
		Apply static and dynamic tuning if the plugin instance is active.
		"""
		if not instance.active:
			return

if instance.has_static_tuning:
			self._call_device_script(instance, instance.script_pre,
					"apply", instance.assigned_devices)
			self._instance_pre_static(instance, True)
			self._instance_apply_static(instance)
			self._instance_post_static(instance, True)
			self._call_device_script(instance, instance.script_post,
					"apply", instance.assigned_devices)
		if instance.has_dynamic_tuning and self._global_cfg.get(consts.CFG_DYNAMIC_TUNING, consts.CFG_DEF_DYNAMIC_TUNING):
			self._instance_init_dynamic(instance)
			self._run_for_each_device(instance, self._instance_apply_dynamic, instance.assigned_devices)
		instance.processed_devices.update(instance.assigned_devices)
		instance.assigned_devices.clear()

def instance_verify_tuning(self, instance, ignore_missing):
		"""
		Verify static tuning if the plugin instance is active.
		"""
		if not instance.active:
			return None

if len(instance.assigned_devices) != 0:
			log.error("BUG: Some devices have not been tuned: %s"
					% ", ".join(instance.assigned_devices))
		devices = instance.processed_devices.copy()
		if instance.has_static_tuning:
			if self._call_device_script(instance, instance.script_pre, "verify", devices) == False:
				return False
			if self._instance_verify_static(instance, ignore_missing, devices) == False:
				return False
			if self._call_device_script(instance, instance.script_post, "verify", devices) == False:
				return False
			return True
		else:
			return None

def instance_update_tuning(self, instance):
		"""
		Apply dynamic tuning if the plugin instance is active.
		"""
		if not instance.active:
			return
		if instance.has_dynamic_tuning and self._global_cfg.get(consts.CFG_DYNAMIC_TUNING, consts.CFG_DEF_DYNAMIC_TUNING):
			self._run_for_each_device(instance, self._instance_update_dynamic, instance.processed_devices.copy())

def instance_unapply_tuning(self, instance, rollback = consts.ROLLBACK_SOFT):
		"""
		Remove all tunings applied by the plugin instance.
		"""
		if rollback == consts.ROLLBACK_NONE:
			return

if instance.has_dynamic_tuning and self._global_cfg.get(consts.CFG_DYNAMIC_TUNING, consts.CFG_DEF_DYNAMIC_TUNING):
			self._run_for_each_device(instance, self._instance_unapply_dynamic, instance.processed_devices)
		if instance.has_static_tuning:
			self._call_device_script(instance, instance.script_post,
					"unapply", instance.processed_devices,
					rollback = rollback)
			self._instance_pre_static(instance, False)
			self._instance_unapply_static(instance, rollback)
			self._instance_post_static(instance, False)
			self._call_device_script(instance, instance.script_pre, "unapply", instance.processed_devices, rollback = rollback)

def _instance_apply_static(self, instance):
		self._execute_all_non_device_commands(instance)
		self._execute_all_device_commands(instance, instance.assigned_devices)

def _instance_verify_static(self, instance, ignore_missing, devices):
		ret = True
		if self._verify_all_non_device_commands(instance, ignore_missing) == False:
			ret = False
		if self._verify_all_device_commands(instance, devices, ignore_missing) == False:
			ret = False
		return ret

def _instance_unapply_static(self, instance, rollback = consts.ROLLBACK_SOFT):
		self._cleanup_all_device_commands(instance,
				instance.processed_devices)
		self._cleanup_all_non_device_commands(instance)

def _instance_init_dynamic(self, instance):
		pass

def _instance_apply_dynamic(self, instance, device):
		for option in [opt for opt in self._options_used_by_dynamic if self._storage_get(instance, self._commands[opt], device) is None]:
			self._check_and_save_value(instance, self._commands[option], device)

self._instance_update_dynamic(instance, device)

def _instance_unapply_dynamic(self, instance, device):
		raise NotImplementedError()

def _instance_update_dynamic(self, instance, device):
		raise NotImplementedError()

#
	# Registration of commands for static plugins.
	#

def _init_commands(self):
		"""
		Initialize commands.
		"""
		self._commands = collections.OrderedDict()
		self._autoregister_commands()
		self._check_commands()

def _autoregister_commands(self):
		"""
		Register all commands marked using @command_set, @command_get, and @command_custom decorators.
		"""
		for member_name in self.__class__.__dict__:
			if member_name.startswith("__"):
				continue
			member = getattr(self, member_name)
			if not hasattr(member, "_command"):
				continue

command_name = member._command["name"]
			info = self._commands.get(command_name, {"name": command_name})

if "set" in member._command:
				info["custom"] = None
				info["set"] = member
				info["per_device"] = member._command["per_device"]
				info["priority"] = member._command["priority"]
			elif "get" in member._command:
				info["get"] = member
			elif "custom" in member._command:
				info["custom"] = member
				info["per_device"] = member._command["per_device"]
				info["priority"] = member._command["priority"]

self._commands[command_name] = info

# sort commands by priority
		self._commands = collections.OrderedDict(sorted(iter(self._commands.items()), key=lambda name_info: name_info[1]["priority"]))

def _check_commands(self):
		"""
		Check if all commands are defined correctly.
		"""
		for command_name, command in list(self._commands.items()):
			# do not check custom commands
			if command.get("custom", False):
				continue
			# automatic commands should have 'get' and 'set' functions
			if "get" not in command or "set" not in command:
				raise TypeError("Plugin command '%s' is not defined correctly" % command_name)

#
	# Operations with persistent storage for status data.
	#

def _storage_key(self, instance_name = None, command_name = None,
			device_name = None):
		class_name = type(self).__name__
		instance_name = "" if instance_name is None else instance_name
		command_name = "" if command_name is None else command_name
		device_name = "" if device_name is None else device_name
		return "%s/%s/%s/%s" % (class_name, instance_name,
				command_name, device_name)

def _storage_set(self, instance, command, value, device_name=None):
		key = self._storage_key(instance.name, command["name"], device_name)
		self._storage.set(key, value)

def _storage_get(self, instance, command, device_name=None):
		key = self._storage_key(instance.name, command["name"], device_name)
		return self._storage.get(key)

def _storage_unset(self, instance, command, device_name=None):
		key = self._storage_key(instance.name, command["name"], device_name)
		return self._storage.unset(key)

#
	# Command execution, verification, and cleanup.
	#

def _execute_all_non_device_commands(self, instance):
		for command in [command for command in list(self._commands.values()) if not command["per_device"]]:
			new_value = self._variables.expand(instance.options.get(command["name"], None))
			if new_value is not None:
				self._execute_non_device_command(instance, command, new_value)

def _execute_all_device_commands(self, instance, devices):
		for command in [command for command in list(self._commands.values()) if command["per_device"]]:
			new_value = self._variables.expand(instance.options.get(command["name"], None))
			if new_value is None:
				continue
			for device in devices:
				self._execute_device_command(instance, command, device, new_value)

def _verify_all_non_device_commands(self, instance, ignore_missing):
		ret = True
		for command in [command for command in list(self._commands.values()) if not command["per_device"]]:
			new_value = self._variables.expand(instance.options.get(command["name"], None))
			if new_value is not None:
				if self._verify_non_device_command(instance, command, new_value, ignore_missing) == False:
					ret = False
		return ret

def _verify_all_device_commands(self, instance, devices, ignore_missing):
		ret = True
		for command in [command for command in list(self._commands.values()) if command["per_device"]]:
			new_value = self._variables.expand(instance.options.get(command["name"], None))
			if new_value is None:
				continue
			for device in devices:
				if self._verify_device_command(instance, command, device, new_value, ignore_missing) == False:
					ret = False
		return ret

def _process_assignment_modifiers(self, new_value, current_value):
		if new_value is not None:
			nws = str(new_value)
			if len(nws) <= 1:
				return new_value
			op = nws[:1]
			val = nws[1:]
			if current_value is None:
				return val if op in ["<", ">"] else new_value
			try:
				if op == ">":
					if int(val) > int(current_value):
						return val
					else:
						return None
				elif op == "<":
					if int(val) < int(current_value):
						return val
					else:
						return None
			except ValueError:
				log.warning("cannot compare new value '%s' with current value '%s' by operator '%s', using '%s' directly as new value" % (val, current_value, op, new_value))
		return new_value

def _get_current_value(self, instance, command, device = None, ignore_missing=False):
		if device is not None:
			return command["get"](device, instance, ignore_missing=ignore_missing)
		else:
			return command["get"](instance)

def _check_and_save_value(self, instance, command, device = None, new_value = None):
		current_value = self._get_current_value(instance, command, device)
		new_value = self._process_assignment_modifiers(new_value, current_value)
		if new_value is not None and current_value is not None:
			self._storage_set(instance, command, current_value, device)
		return new_value

def _execute_device_command(self, instance, command, device, new_value):
		if command["custom"] is not None:
			command["custom"](True, new_value, device, False, False, instance)
		else:
			new_value = self._check_and_save_value(instance, command, device, new_value)
			if new_value is not None:
				command["set"](new_value, device, instance, sim = False, remove = False)

def _execute_non_device_command(self, instance, command, new_value):
		if command["custom"] is not None:
			command["custom"](True, new_value, False, False, instance)
		else:
			new_value = self._check_and_save_value(instance, command, None, new_value)
			if new_value is not None:
				command["set"](new_value, instance, sim = False, remove = False)

def _norm_value(self, value):
		v = self._cmd.unquote(str(value))
		if re.match(r'\s*(0+,?)+([\da-fA-F]*,?)*\s*$', v):
			return re.sub(r'^\s*(0+,?)+', "", v)
		return v

def _verify_value(self, name, new_value, current_value, ignore_missing, device = None):
		if new_value is None:
			return None
		ret = False
		if current_value is None and ignore_missing:
			if device is None:
				log.info(consts.STR_VERIFY_PROFILE_VALUE_MISSING % name)
			else:
				log.info(consts.STR_VERIFY_PROFILE_DEVICE_VALUE_MISSING % (device, name))
			return True

if current_value is not None:
			current_value = self._norm_value(current_value)
			new_value = self._norm_value(new_value)
			try:
				ret = int(new_value) == int(current_value)
			except ValueError:
				try:
					ret = int(new_value, 16) == int(current_value, 16)
				except ValueError:
					ret = str(new_value) == str(current_value)
					if not ret:
						vals = str(new_value).split('|')
						for val in vals:
							val = val.strip()
							ret = val == current_value
							if ret:
								break
		self._log_verification_result(name, ret, new_value,
				current_value, device = device)
		return ret

def _log_verification_result(self, name, success, new_value,
			current_value, device = None):
		if success:
			if device is None:
				log.info(consts.STR_VERIFY_PROFILE_VALUE_OK % (name, str(current_value).strip()))
			else:
				log.info(consts.STR_VERIFY_PROFILE_DEVICE_VALUE_OK % (device, name, str(current_value).strip()))
			return True
		else:
			if device is None:
				log.error(consts.STR_VERIFY_PROFILE_VALUE_FAIL % (name, str(current_value).strip(), str(new_value).strip()))
			else:
				log.error(consts.STR_VERIFY_PROFILE_DEVICE_VALUE_FAIL % (device, name, str(current_value).strip(), str(new_value).strip()))
			return False

def _verify_device_command(self, instance, command, device, new_value, ignore_missing):
		if command["custom"] is not None:
			return command["custom"](True, new_value, device, True, ignore_missing, instance)
		current_value = self._get_current_value(instance, command, device, ignore_missing=ignore_missing)
		new_value = self._process_assignment_modifiers(new_value, current_value)
		if new_value is None:
			return None
		new_value = command["set"](new_value, device, instance, True, False)
		return self._verify_value(command["name"], new_value, current_value, ignore_missing, device)

def _verify_non_device_command(self, instance, command, new_value, ignore_missing):
		if command["custom"] is not None:
			return command["custom"](True, new_value, True, ignore_missing, instance)
		current_value = self._get_current_value(instance, command)
		new_value = self._process_assignment_modifiers(new_value, current_value)
		if new_value is None:
			return None
		new_value = command["set"](new_value, instance, True, False)
		return self._verify_value(command["name"], new_value, current_value, ignore_missing)

def _cleanup_all_non_device_commands(self, instance):
		for command in reversed([command for command in list(self._commands.values()) if not command["per_device"]]):
			if (instance.options.get(command["name"], None) is not None) or (command["name"] in self._options_used_by_dynamic):
				self._cleanup_non_device_command(instance, command)

def _cleanup_all_device_commands(self, instance, devices, remove = False):
		for command in reversed([command for command in list(self._commands.values()) if command["per_device"]]):
			if (instance.options.get(command["name"], None) is not None) or (command["name"] in self._options_used_by_dynamic):
				for device in devices:
					self._cleanup_device_command(instance, command, device, remove)

def _cleanup_device_command(self, instance, command, device, remove = False):
		if command["custom"] is not None:
			command["custom"](False, None, device, False, False, instance)
		else:
			old_value = self._storage_get(instance, command, device)
			if old_value is not None:
				command["set"](old_value, device, instance, sim = False, remove = remove)
			self._storage_unset(instance, command, device)

def _cleanup_non_device_command(self, instance, command):
		if command["custom"] is not None:
			command["custom"](False, None, False, False, instance)
		else:
			old_value = self._storage_get(instance, command)
			if old_value is not None:
				command["set"](old_value, instance, sim = False, remove = False)
			self._storage_unset(instance, command)