[ceph.git] / ceph / src / spdk / ocf / tests / functional / pyocf / types / cache.py

#
# Copyright(c) 2019 Intel Corporation
# SPDX-License-Identifier: BSD-3-Clause-Clear
#

from ctypes import (
    c_uint64,
    c_uint32,
    c_uint16,
    c_int,
    c_char,
    c_char_p,
    c_void_p,
    c_bool,
    c_uint8,
    Structure,
    byref,
    cast,
    create_string_buffer,
)
from enum import IntEnum
from datetime import timedelta

from ..ocf import OcfLib
from .shared import (
    Uuid,
    OcfError,
    CacheLineSize,
    CacheLines,
    OcfCompletion,
    SeqCutOffPolicy,
)
from ..utils import Size, struct_to_dict
from .core import Core
from .queue import Queue
from .stats.cache import CacheInfo
from .stats.shared import UsageStats, RequestsStats, BlocksStats, ErrorsStats


class Backfill(Structure):
    _fields_ = [("_max_queue_size", c_uint32), ("_queue_unblock_size", c_uint32)]


class CacheConfig(Structure):
    MAX_CACHE_NAME_SIZE = 32
    _fields_ = [
        ("_name", c_char * MAX_CACHE_NAME_SIZE),
        ("_cache_mode", c_uint32),
        ("_eviction_policy", c_uint32),
        ("_promotion_policy", c_uint32),
        ("_cache_line_size", c_uint64),
        ("_metadata_layout", c_uint32),
        ("_metadata_volatile", c_bool),
        ("_backfill", Backfill),
        ("_locked", c_bool),
        ("_pt_unaligned_io", c_bool),
        ("_use_submit_io_fast", c_bool),
    ]


class CacheDeviceConfig(Structure):
    _fields_ = [
        ("_uuid", Uuid),
        ("_volume_type", c_uint8),
        ("_cache_line_size", c_uint64),
        ("_force", c_bool),
        ("_min_free_ram", c_uint64),
        ("_perform_test", c_bool),
        ("_discard_on_start", c_bool),
    ]


class ConfValidValues:
    promotion_nhit_insertion_threshold_range = range(2, 1000)
    promotion_nhit_trigger_threshold_range = range(0, 100)


class CacheMode(IntEnum):
    WT = 0
    WB = 1
    WA = 2
    PT = 3
    WI = 4
    WO = 5
    DEFAULT = WT

    def lazy_write(self):
        return self.value in [CacheMode.WB, CacheMode.WO]

    def write_insert(self):
        return self.value not in [CacheMode.PT, CacheMode.WA, CacheMode.WI]

    def read_insert(self):
        return self.value not in [CacheMode.PT, CacheMode.WO]


class EvictionPolicy(IntEnum):
    LRU = 0
    DEFAULT = LRU


class PromotionPolicy(IntEnum):
    ALWAYS = 0
    NHIT = 1
    DEFAULT = ALWAYS


class NhitParams(IntEnum):
    INSERTION_THRESHOLD = 0
    TRIGGER_THRESHOLD = 1


class CleaningPolicy(IntEnum):
    NOP = 0
    ALRU = 1
    ACP = 2
    DEFAULT = ALRU


class AlruParams(IntEnum):
    WAKE_UP_TIME = 0
    STALE_BUFFER_TIME = 1
    FLUSH_MAX_BUFFERS = 2
    ACTIVITY_THRESHOLD = 3


class AcpParams(IntEnum):
    WAKE_UP_TIME = 0
    FLUSH_MAX_BUFFERS = 1


class MetadataLayout(IntEnum):
    STRIPING = 0
    SEQUENTIAL = 1
    DEFAULT = STRIPING


class Cache:
    DEFAULT_BACKFILL_QUEUE_SIZE = 65536
    DEFAULT_BACKFILL_UNBLOCK = 60000
    DEFAULT_PT_UNALIGNED_IO = False
    DEFAULT_USE_SUBMIT_FAST = False

    def __init__(
        self,
        owner,
        name: str = "cache",
        cache_mode: CacheMode = CacheMode.DEFAULT,
        eviction_policy: EvictionPolicy = EvictionPolicy.DEFAULT,
        promotion_policy: PromotionPolicy = PromotionPolicy.DEFAULT,
        cache_line_size: CacheLineSize = CacheLineSize.DEFAULT,
        metadata_layout: MetadataLayout = MetadataLayout.DEFAULT,
        metadata_volatile: bool = False,
        max_queue_size: int = DEFAULT_BACKFILL_QUEUE_SIZE,
        queue_unblock_size: int = DEFAULT_BACKFILL_UNBLOCK,
        locked: bool = False,
        pt_unaligned_io: bool = DEFAULT_PT_UNALIGNED_IO,
        use_submit_fast: bool = DEFAULT_USE_SUBMIT_FAST,
    ):
        self.device = None
        self.started = False
        self.owner = owner
        self.cache_line_size = cache_line_size

        self.cfg = CacheConfig(
            _name=name.encode("ascii"),
            _cache_mode=cache_mode,
            _eviction_policy=eviction_policy,
            _promotion_policy=promotion_policy,
            _cache_line_size=cache_line_size,
            _metadata_layout=metadata_layout,
            _metadata_volatile=metadata_volatile,
            _backfill=Backfill(
                _max_queue_size=max_queue_size, _queue_unblock_size=queue_unblock_size
            ),
            _locked=locked,
            _pt_unaligned_io=pt_unaligned_io,
            _use_submit_fast=use_submit_fast,
        )
        self.cache_handle = c_void_p()
        self._as_parameter_ = self.cache_handle
        self.io_queues = []
        self.cores = []

    def start_cache(self, default_io_queue: Queue = None, mngt_queue: Queue = None):
        status = self.owner.lib.ocf_mngt_cache_start(
            self.owner.ctx_handle, byref(self.cache_handle), byref(self.cfg)
        )
        if status:
            raise OcfError("Creating cache instance failed", status)
        self.owner.caches.append(self)

        self.mngt_queue = mngt_queue or Queue(self, "mgmt-{}".format(self.get_name()))

        if default_io_queue:
            self.io_queues += [default_io_queue]
        else:
            self.io_queues += [Queue(self, "default-io-{}".format(self.get_name()))]

        status = self.owner.lib.ocf_mngt_cache_set_mngt_queue(self, self.mngt_queue)
        if status:
            raise OcfError("Error setting management queue", status)

        self.started = True

    def change_cache_mode(self, cache_mode: CacheMode):
        self.write_lock()
        status = self.owner.lib.ocf_mngt_cache_set_mode(self.cache_handle, cache_mode)

        self.write_unlock()

        if status:
            raise OcfError("Error changing cache mode", status)

    def set_cleaning_policy(self, cleaning_policy: CleaningPolicy):
        self.write_lock()

        status = self.owner.lib.ocf_mngt_cache_cleaning_set_policy(
            self.cache_handle, cleaning_policy
        )

        self.write_unlock()

        if status:
            raise OcfError("Error changing cleaning policy", status)

    def set_cleaning_policy_param(
        self, cleaning_policy: CleaningPolicy, param_id, param_value
    ):
        self.write_lock()

        status = self.owner.lib.ocf_mngt_cache_cleaning_set_param(
            self.cache_handle, cleaning_policy, param_id, param_value
        )

        self.write_unlock()

        if status:
            raise OcfError("Error setting cleaning policy param", status)

    def set_promotion_policy(self, promotion_policy: PromotionPolicy):
        self.write_lock()

        status = self.owner.lib.ocf_mngt_cache_promotion_set_policy(
            self.cache_handle, promotion_policy
        )

        self.write_unlock()
        if status:
            raise OcfError("Error setting promotion policy", status)

    def get_promotion_policy_param(self, promotion_type, param_id):
        self.read_lock()

        param_value = c_uint64()

        status = self.owner.lib.ocf_mngt_cache_promotion_get_param(
            self.cache_handle, promotion_type, param_id, byref(param_value)
        )

        self.read_unlock()
        if status:
            raise OcfError("Error getting promotion policy parameter", status)

        return param_value

    def set_promotion_policy_param(self, promotion_type, param_id, param_value):
        self.write_lock()

        status = self.owner.lib.ocf_mngt_cache_promotion_set_param(
            self.cache_handle, promotion_type, param_id, param_value
        )

        self.write_unlock()
        if status:
            raise OcfError("Error setting promotion policy parameter", status)

    def set_seq_cut_off_policy(self, policy: SeqCutOffPolicy):
        self.write_lock()

        status = self.owner.lib.ocf_mngt_core_set_seq_cutoff_policy_all(
            self.cache_handle, policy
        )

        self.write_unlock()

        if status:
            raise OcfError("Error setting cache seq cut off policy", status)

    def configure_device(
        self, device, force=False, perform_test=True, cache_line_size=None
    ):
        self.device = device
        self.device_name = device.uuid
        self.dev_cfg = CacheDeviceConfig(
            _uuid=Uuid(
                _data=cast(
                    create_string_buffer(self.device_name.encode("ascii")), c_char_p
                ),
                _size=len(self.device_name) + 1,
            ),
            _volume_type=device.type_id,
            _cache_line_size=cache_line_size
            if cache_line_size
            else self.cache_line_size,
            _force=force,
            _min_free_ram=0,
            _perform_test=perform_test,
            _discard_on_start=False,
        )

    def attach_device(
        self, device, force=False, perform_test=False, cache_line_size=None
    ):
        self.configure_device(device, force, perform_test, cache_line_size)
        self.write_lock()

        c = OcfCompletion([("cache", c_void_p), ("priv", c_void_p), ("error", c_int)])

        device.owner.lib.ocf_mngt_cache_attach(
            self.cache_handle, byref(self.dev_cfg), c, None
        )

        c.wait()
        self.write_unlock()

        if c.results["error"]:
            raise OcfError("Attaching cache device failed", c.results["error"])

    def load_cache(self, device):
        self.configure_device(device)
        c = OcfCompletion([("cache", c_void_p), ("priv", c_void_p), ("error", c_int)])
        device.owner.lib.ocf_mngt_cache_load(
            self.cache_handle, byref(self.dev_cfg), c, None
        )

        c.wait()
        if c.results["error"]:
            raise OcfError("Loading cache device failed", c.results["error"])

    @classmethod
    def load_from_device(cls, device, name="cache"):
        c = cls(name=name, owner=device.owner)

        c.start_cache()
        try:
            c.load_cache(device)
        except:  # noqa E722
            c.stop()
            raise

        return c

    @classmethod
    def start_on_device(cls, device, **kwargs):
        c = cls(owner=device.owner, **kwargs)

        c.start_cache()
        try:
            c.attach_device(device, force=True)
        except:  # noqa E722
            c.stop()
            raise

        return c

    def put(self):
        self.owner.lib.ocf_mngt_cache_put(self.cache_handle)

    def get(self):
        status = self.owner.lib.ocf_mngt_cache_get(self.cache_handle)
        if status:
            raise OcfError("Couldn't get cache instance", status)

    def read_lock(self):
        c = OcfCompletion([("cache", c_void_p), ("priv", c_void_p), ("error", c_int)])
        self.owner.lib.ocf_mngt_cache_read_lock(self.cache_handle, c, None)
        c.wait()
        if c.results["error"]:
            raise OcfError("Couldn't lock cache instance", c.results["error"])

    def write_lock(self):
        c = OcfCompletion([("cache", c_void_p), ("priv", c_void_p), ("error", c_int)])
        self.owner.lib.ocf_mngt_cache_lock(self.cache_handle, c, None)
        c.wait()
        if c.results["error"]:
            raise OcfError("Couldn't lock cache instance", c.results["error"])

    def read_unlock(self):
        self.owner.lib.ocf_mngt_cache_read_unlock(self.cache_handle)

    def write_unlock(self):
        self.owner.lib.ocf_mngt_cache_unlock(self.cache_handle)

    def add_core(self, core: Core):
        self.write_lock()

        c = OcfCompletion(
            [
                ("cache", c_void_p),
                ("core", c_void_p),
                ("priv", c_void_p),
                ("error", c_int),
            ]
        )

        self.owner.lib.ocf_mngt_cache_add_core(
            self.cache_handle, byref(core.get_cfg()), c, None
        )

        c.wait()
        if c.results["error"]:
            self.write_unlock()
            raise OcfError("Failed adding core", c.results["error"])

        core.cache = self
        core.handle = c.results["core"]
        self.cores.append(core)

        self.write_unlock()

    def remove_core(self, core: Core):
        self.write_lock()

        c = OcfCompletion([("priv", c_void_p), ("error", c_int)])

        self.owner.lib.ocf_mngt_cache_remove_core(core.handle, c, None)

        c.wait()
        self.write_unlock()

        if c.results["error"]:
            raise OcfError("Failed removing core", c.results["error"])

        self.cores.remove(core)

    def get_stats(self):
        cache_info = CacheInfo()
        usage = UsageStats()
        req = RequestsStats()
        block = BlocksStats()
        errors = ErrorsStats()

        self.read_lock()

        status = self.owner.lib.ocf_cache_get_info(self.cache_handle, byref(cache_info))
        if status:
            self.read_unlock()
            raise OcfError("Failed getting cache info", status)

        status = self.owner.lib.ocf_stats_collect_cache(
            self.cache_handle, byref(usage), byref(req), byref(block), byref(errors)
        )
        if status:
            self.read_unlock()
            raise OcfError("Failed getting stats", status)

        line_size = CacheLineSize(cache_info.cache_line_size)
        cache_name = self.owner.lib.ocf_cache_get_name(self).decode("ascii")

        self.read_unlock()
        return {
            "conf": {
                "attached": cache_info.attached,
                "volume_type": self.owner.volume_types[cache_info.volume_type],
                "size": CacheLines(cache_info.size, line_size),
                "inactive": {
                    "occupancy": CacheLines(
                        cache_info.inactive.occupancy.value, line_size
                    ),
                    "dirty": CacheLines(cache_info.inactive.dirty.value, line_size),
                    "clean": CacheLines(cache_info.inactive.clean.value, line_size),
                },
                "occupancy": CacheLines(cache_info.occupancy, line_size),
                "dirty": CacheLines(cache_info.dirty, line_size),
                "dirty_initial": CacheLines(cache_info.dirty_initial, line_size),
                "dirty_for": timedelta(seconds=cache_info.dirty_for),
                "cache_mode": CacheMode(cache_info.cache_mode),
                "fallback_pt": {
                    "error_counter": cache_info.fallback_pt.error_counter,
                    "status": cache_info.fallback_pt.status,
                },
                "state": cache_info.state,
                "eviction_policy": EvictionPolicy(cache_info.eviction_policy),
                "cleaning_policy": CleaningPolicy(cache_info.cleaning_policy),
                "promotion_policy": PromotionPolicy(cache_info.promotion_policy),
                "cache_line_size": line_size,
                "flushed": CacheLines(cache_info.flushed, line_size),
                "core_count": cache_info.core_count,
                "metadata_footprint": Size(cache_info.metadata_footprint),
                "metadata_end_offset": Size(cache_info.metadata_end_offset),
                "cache_name": cache_name,
            },
            "block": struct_to_dict(block),
            "req": struct_to_dict(req),
            "usage": struct_to_dict(usage),
            "errors": struct_to_dict(errors),
        }

    def reset_stats(self):
        self.owner.lib.ocf_core_stats_initialize_all(self.cache_handle)

    def get_default_queue(self):
        if not self.io_queues:
            raise Exception("No queues added for cache")

        return self.io_queues[0]

    def save(self):
        if not self.started:
            raise Exception("Not started!")

        self.get_and_write_lock()
        c = OcfCompletion([("cache", c_void_p), ("priv", c_void_p), ("error", c_int)])
        self.owner.lib.ocf_mngt_cache_save(self.cache_handle, c, None)

        c.wait()
        self.put_and_write_unlock()

        if c.results["error"]:
            raise OcfError("Failed saving cache", c.results["error"])

    def stop(self):
        if not self.started:
            raise Exception("Already stopped!")

        self.write_lock()

        c = OcfCompletion([("cache", c_void_p), ("priv", c_void_p), ("error", c_int)])

        self.owner.lib.ocf_mngt_cache_stop(self.cache_handle, c, None)

        c.wait()
        if c.results["error"]:
            self.write_unlock()
            raise OcfError("Failed stopping cache", c.results["error"])

        self.mngt_queue.put()
        del self.io_queues[:]
        self.started = False

        self.write_unlock()

        self.owner.caches.remove(self)

    def flush(self):
        self.write_lock()

        c = OcfCompletion([("cache", c_void_p), ("priv", c_void_p), ("error", c_int)])
        self.owner.lib.ocf_mngt_cache_flush(self.cache_handle, c, None)
        c.wait()
        self.write_unlock()

        if c.results["error"]:
            raise OcfError("Couldn't flush cache", c.results["error"])

    def get_name(self):
        self.read_lock()

        try:
            return str(self.owner.lib.ocf_cache_get_name(self), encoding="ascii")
        except:  # noqa E722
            raise OcfError("Couldn't get cache name")
        finally:
            self.read_unlock()


lib = OcfLib.getInstance()
lib.ocf_mngt_cache_remove_core.argtypes = [c_void_p, c_void_p, c_void_p]
lib.ocf_mngt_cache_add_core.argtypes = [c_void_p, c_void_p, c_void_p, c_void_p]
lib.ocf_cache_get_name.argtypes = [c_void_p]
lib.ocf_cache_get_name.restype = c_char_p
lib.ocf_mngt_cache_cleaning_set_policy.argtypes = [c_void_p, c_uint32]
lib.ocf_mngt_cache_cleaning_set_policy.restype = c_int
lib.ocf_mngt_core_set_seq_cutoff_policy_all.argtypes = [c_void_p, c_uint32]
lib.ocf_mngt_core_set_seq_cutoff_policy_all.restype = c_int
lib.ocf_stats_collect_cache.argtypes = [
    c_void_p,
    c_void_p,
    c_void_p,
    c_void_p,
    c_void_p,
]
lib.ocf_stats_collect_cache.restype = c_int
lib.ocf_cache_get_info.argtypes = [c_void_p, c_void_p]
lib.ocf_cache_get_info.restype = c_int
lib.ocf_mngt_cache_cleaning_set_param.argtypes = [
    c_void_p,
    c_uint32,
    c_uint32,
    c_uint32,
]
lib.ocf_mngt_cache_cleaning_set_param.restype = c_int
Commit	Line	Data
9f95a23c TL	1	#
	2	# Copyright(c) 2019 Intel Corporation
	3	# SPDX-License-Identifier: BSD-3-Clause-Clear
	4	#
	5
	6	from ctypes import (
	7	c_uint64,
	8	c_uint32,
	9	c_uint16,
	10	c_int,
f67539c2	11	c_char,
9f95a23c TL	12	c_char_p,
	13	c_void_p,
	14	c_bool,
	15	c_uint8,
	16	Structure,
	17	byref,
	18	cast,
	19	create_string_buffer,
	20	)
	21	from enum import IntEnum
	22	from datetime import timedelta
	23
	24	from ..ocf import OcfLib
	25	from .shared import (
	26	Uuid,
	27	OcfError,
	28	CacheLineSize,
	29	CacheLines,
	30	OcfCompletion,
	31	SeqCutOffPolicy,
	32	)
	33	from ..utils import Size, struct_to_dict
	34	from .core import Core
	35	from .queue import Queue
	36	from .stats.cache import CacheInfo
	37	from .stats.shared import UsageStats, RequestsStats, BlocksStats, ErrorsStats
	38
	39
	40	class Backfill(Structure):
	41	_fields_ = [("_max_queue_size", c_uint32), ("_queue_unblock_size", c_uint32)]
	42
	43
	44	class CacheConfig(Structure):
f67539c2	45	MAX_CACHE_NAME_SIZE = 32
9f95a23c	46	_fields_ = [
f67539c2	47	("_name", c_char * MAX_CACHE_NAME_SIZE),
9f95a23c TL	48	("_cache_mode", c_uint32),
9f95a23c TL	49	("_eviction_policy", c_uint32),
f67539c2	50	("_promotion_policy", c_uint32),
9f95a23c TL	51	("_cache_line_size", c_uint64),
	52	("_metadata_layout", c_uint32),
	53	("_metadata_volatile", c_bool),
	54	("_backfill", Backfill),
	55	("_locked", c_bool),
	56	("_pt_unaligned_io", c_bool),
	57	("_use_submit_io_fast", c_bool),
	58	]
	59
	60
	61	class CacheDeviceConfig(Structure):
	62	_fields_ = [
	63	("_uuid", Uuid),
	64	("_volume_type", c_uint8),
	65	("_cache_line_size", c_uint64),
	66	("_force", c_bool),
	67	("_min_free_ram", c_uint64),
	68	("_perform_test", c_bool),
	69	("_discard_on_start", c_bool),
	70	]
	71
	72
f67539c2 TL	73	class ConfValidValues:
	74	promotion_nhit_insertion_threshold_range = range(2, 1000)
	75	promotion_nhit_trigger_threshold_range = range(0, 100)
	76
	77
9f95a23c TL	78	class CacheMode(IntEnum):
	79	WT = 0
	80	WB = 1
	81	WA = 2
	82	PT = 3
	83	WI = 4
f67539c2	84	WO = 5
9f95a23c TL	85	DEFAULT = WT
9f95a23c TL	86
f67539c2 TL	87	def lazy_write(self):
	88	return self.value in [CacheMode.WB, CacheMode.WO]
	89
	90	def write_insert(self):
	91	return self.value not in [CacheMode.PT, CacheMode.WA, CacheMode.WI]
	92
	93	def read_insert(self):
	94	return self.value not in [CacheMode.PT, CacheMode.WO]
	95
9f95a23c TL	96
	97	class EvictionPolicy(IntEnum):
	98	LRU = 0
	99	DEFAULT = LRU
	100
	101
f67539c2 TL	102	class PromotionPolicy(IntEnum):
	103	ALWAYS = 0
	104	NHIT = 1
	105	DEFAULT = ALWAYS
	106
	107
	108	class NhitParams(IntEnum):
	109	INSERTION_THRESHOLD = 0
	110	TRIGGER_THRESHOLD = 1
	111
	112
9f95a23c TL	113	class CleaningPolicy(IntEnum):
	114	NOP = 0
	115	ALRU = 1
	116	ACP = 2
	117	DEFAULT = ALRU
	118
	119
	120	class AlruParams(IntEnum):
	121	WAKE_UP_TIME = 0
	122	STALE_BUFFER_TIME = 1
	123	FLUSH_MAX_BUFFERS = 2
	124	ACTIVITY_THRESHOLD = 3
	125
	126
	127	class AcpParams(IntEnum):
	128	WAKE_UP_TIME = 0
	129	FLUSH_MAX_BUFFERS = 1
	130
	131
	132	class MetadataLayout(IntEnum):
	133	STRIPING = 0
	134	SEQUENTIAL = 1
	135	DEFAULT = STRIPING
	136
	137
	138	class Cache:
9f95a23c TL	139	DEFAULT_BACKFILL_QUEUE_SIZE = 65536
	140	DEFAULT_BACKFILL_UNBLOCK = 60000
	141	DEFAULT_PT_UNALIGNED_IO = False
	142	DEFAULT_USE_SUBMIT_FAST = False
	143
	144	def __init__(
	145	self,
	146	owner,
f67539c2	147	name: str = "cache",
9f95a23c TL	148	cache_mode: CacheMode = CacheMode.DEFAULT,
9f95a23c TL	149	eviction_policy: EvictionPolicy = EvictionPolicy.DEFAULT,
f67539c2	150	promotion_policy: PromotionPolicy = PromotionPolicy.DEFAULT,
9f95a23c TL	151	cache_line_size: CacheLineSize = CacheLineSize.DEFAULT,
	152	metadata_layout: MetadataLayout = MetadataLayout.DEFAULT,
	153	metadata_volatile: bool = False,
	154	max_queue_size: int = DEFAULT_BACKFILL_QUEUE_SIZE,
	155	queue_unblock_size: int = DEFAULT_BACKFILL_UNBLOCK,
	156	locked: bool = False,
	157	pt_unaligned_io: bool = DEFAULT_PT_UNALIGNED_IO,
	158	use_submit_fast: bool = DEFAULT_USE_SUBMIT_FAST,
	159	):
	160	self.device = None
	161	self.started = False
	162	self.owner = owner
	163	self.cache_line_size = cache_line_size
	164
	165	self.cfg = CacheConfig(
f67539c2	166	_name=name.encode("ascii"),
9f95a23c TL	167	_cache_mode=cache_mode,
9f95a23c TL	168	_eviction_policy=eviction_policy,
f67539c2	169	_promotion_policy=promotion_policy,
9f95a23c TL	170	_cache_line_size=cache_line_size,
	171	_metadata_layout=metadata_layout,
	172	_metadata_volatile=metadata_volatile,
	173	_backfill=Backfill(
f67539c2	174	_max_queue_size=max_queue_size, _queue_unblock_size=queue_unblock_size
9f95a23c TL	175	),
	176	_locked=locked,
	177	_pt_unaligned_io=pt_unaligned_io,
	178	_use_submit_fast=use_submit_fast,
	179	)
	180	self.cache_handle = c_void_p()
	181	self._as_parameter_ = self.cache_handle
	182	self.io_queues = []
9f95a23c TL	183	self.cores = []
9f95a23c TL	184
f67539c2	185	def start_cache(self, default_io_queue: Queue = None, mngt_queue: Queue = None):
9f95a23c TL	186	status = self.owner.lib.ocf_mngt_cache_start(
	187	self.owner.ctx_handle, byref(self.cache_handle), byref(self.cfg)
	188	)
	189	if status:
	190	raise OcfError("Creating cache instance failed", status)
	191	self.owner.caches.append(self)
	192
f67539c2	193	self.mngt_queue = mngt_queue or Queue(self, "mgmt-{}".format(self.get_name()))
9f95a23c TL	194
	195	if default_io_queue:
	196	self.io_queues += [default_io_queue]
	197	else:
f67539c2	198	self.io_queues += [Queue(self, "default-io-{}".format(self.get_name()))]
9f95a23c	199
f67539c2	200	status = self.owner.lib.ocf_mngt_cache_set_mngt_queue(self, self.mngt_queue)
9f95a23c TL	201	if status:
	202	raise OcfError("Error setting management queue", status)
	203
	204	self.started = True
	205
	206	def change_cache_mode(self, cache_mode: CacheMode):
f67539c2 TL	207	self.write_lock()
	208	status = self.owner.lib.ocf_mngt_cache_set_mode(self.cache_handle, cache_mode)
	209
	210	self.write_unlock()
9f95a23c TL	211
9f95a23c TL	212	if status:
9f95a23c TL	213	raise OcfError("Error changing cache mode", status)
9f95a23c TL	214
9f95a23c	215	def set_cleaning_policy(self, cleaning_policy: CleaningPolicy):
f67539c2	216	self.write_lock()
9f95a23c TL	217
	218	status = self.owner.lib.ocf_mngt_cache_cleaning_set_policy(
	219	self.cache_handle, cleaning_policy
	220	)
f67539c2 TL	221
	222	self.write_unlock()
	223
9f95a23c	224	if status:
9f95a23c TL	225	raise OcfError("Error changing cleaning policy", status)
9f95a23c TL	226
9f95a23c TL	227	def set_cleaning_policy_param(
	228	self, cleaning_policy: CleaningPolicy, param_id, param_value
	229	):
f67539c2	230	self.write_lock()
9f95a23c TL	231
	232	status = self.owner.lib.ocf_mngt_cache_cleaning_set_param(
	233	self.cache_handle, cleaning_policy, param_id, param_value
	234	)
f67539c2 TL	235
	236	self.write_unlock()
	237
9f95a23c	238	if status:
9f95a23c TL	239	raise OcfError("Error setting cleaning policy param", status)
9f95a23c TL	240
f67539c2 TL	241	def set_promotion_policy(self, promotion_policy: PromotionPolicy):
	242	self.write_lock()
	243
	244	status = self.owner.lib.ocf_mngt_cache_promotion_set_policy(
	245	self.cache_handle, promotion_policy
	246	)
	247
	248	self.write_unlock()
	249	if status:
	250	raise OcfError("Error setting promotion policy", status)
	251
	252	def get_promotion_policy_param(self, promotion_type, param_id):
	253	self.read_lock()
	254
	255	param_value = c_uint64()
	256
	257	status = self.owner.lib.ocf_mngt_cache_promotion_get_param(
	258	self.cache_handle, promotion_type, param_id, byref(param_value)
	259	)
	260
	261	self.read_unlock()
	262	if status:
	263	raise OcfError("Error getting promotion policy parameter", status)
	264
	265	return param_value
	266
	267	def set_promotion_policy_param(self, promotion_type, param_id, param_value):
	268	self.write_lock()
	269
	270	status = self.owner.lib.ocf_mngt_cache_promotion_set_param(
	271	self.cache_handle, promotion_type, param_id, param_value
	272	)
	273
	274	self.write_unlock()
	275	if status:
	276	raise OcfError("Error setting promotion policy parameter", status)
9f95a23c TL	277
9f95a23c TL	278	def set_seq_cut_off_policy(self, policy: SeqCutOffPolicy):
f67539c2	279	self.write_lock()
9f95a23c TL	280
	281	status = self.owner.lib.ocf_mngt_core_set_seq_cutoff_policy_all(
	282	self.cache_handle, policy
	283	)
f67539c2 TL	284
	285	self.write_unlock()
	286
9f95a23c	287	if status:
9f95a23c TL	288	raise OcfError("Error setting cache seq cut off policy", status)
9f95a23c TL	289
9f95a23c	290	def configure_device(
f67539c2	291	self, device, force=False, perform_test=True, cache_line_size=None
9f95a23c TL	292	):
	293	self.device = device
	294	self.device_name = device.uuid
	295	self.dev_cfg = CacheDeviceConfig(
	296	_uuid=Uuid(
	297	_data=cast(
f67539c2	298	create_string_buffer(self.device_name.encode("ascii")), c_char_p
9f95a23c TL	299	),
	300	_size=len(self.device_name) + 1,
	301	),
	302	_volume_type=device.type_id,
	303	_cache_line_size=cache_line_size
	304	if cache_line_size
	305	else self.cache_line_size,
	306	_force=force,
	307	_min_free_ram=0,
	308	_perform_test=perform_test,
	309	_discard_on_start=False,
	310	)
	311
	312	def attach_device(
	313	self, device, force=False, perform_test=False, cache_line_size=None
	314	):
	315	self.configure_device(device, force, perform_test, cache_line_size)
f67539c2	316	self.write_lock()
9f95a23c	317
f67539c2	318	c = OcfCompletion([("cache", c_void_p), ("priv", c_void_p), ("error", c_int)])
9f95a23c TL	319
	320	device.owner.lib.ocf_mngt_cache_attach(
	321	self.cache_handle, byref(self.dev_cfg), c, None
	322	)
	323
	324	c.wait()
f67539c2 TL	325	self.write_unlock()
f67539c2 TL	326
9f95a23c	327	if c.results["error"]:
9f95a23c TL	328	raise OcfError("Attaching cache device failed", c.results["error"])
9f95a23c TL	329
9f95a23c TL	330	def load_cache(self, device):
9f95a23c TL	331	self.configure_device(device)
f67539c2	332	c = OcfCompletion([("cache", c_void_p), ("priv", c_void_p), ("error", c_int)])
9f95a23c TL	333	device.owner.lib.ocf_mngt_cache_load(
	334	self.cache_handle, byref(self.dev_cfg), c, None
	335	)
	336
	337	c.wait()
	338	if c.results["error"]:
	339	raise OcfError("Loading cache device failed", c.results["error"])
	340
	341	@classmethod
f67539c2	342	def load_from_device(cls, device, name="cache"):
9f95a23c TL	343	c = cls(name=name, owner=device.owner)
	344
	345	c.start_cache()
f67539c2 TL	346	try:
	347	c.load_cache(device)
	348	except: # noqa E722
	349	c.stop()
	350	raise
	351
9f95a23c TL	352	return c
	353
	354	@classmethod
	355	def start_on_device(cls, device, **kwargs):
f67539c2	356	c = cls(owner=device.owner, **kwargs)
9f95a23c TL	357
	358	c.start_cache()
	359	try:
	360	c.attach_device(device, force=True)
f67539c2	361	except: # noqa E722
9f95a23c TL	362	c.stop()
	363	raise
	364
	365	return c
	366
9f95a23c TL	367	def put(self):
	368	self.owner.lib.ocf_mngt_cache_put(self.cache_handle)
	369
	370	def get(self):
	371	status = self.owner.lib.ocf_mngt_cache_get(self.cache_handle)
	372	if status:
	373	raise OcfError("Couldn't get cache instance", status)
	374
f67539c2 TL	375	def read_lock(self):
	376	c = OcfCompletion([("cache", c_void_p), ("priv", c_void_p), ("error", c_int)])
	377	self.owner.lib.ocf_mngt_cache_read_lock(self.cache_handle, c, None)
	378	c.wait()
	379	if c.results["error"]:
	380	raise OcfError("Couldn't lock cache instance", c.results["error"])
9f95a23c	381
f67539c2 TL	382	def write_lock(self):
	383	c = OcfCompletion([("cache", c_void_p), ("priv", c_void_p), ("error", c_int)])
	384	self.owner.lib.ocf_mngt_cache_lock(self.cache_handle, c, None)
	385	c.wait()
	386	if c.results["error"]:
	387	raise OcfError("Couldn't lock cache instance", c.results["error"])
9f95a23c	388
f67539c2 TL	389	def read_unlock(self):
f67539c2 TL	390	self.owner.lib.ocf_mngt_cache_read_unlock(self.cache_handle)
9f95a23c	391
f67539c2 TL	392	def write_unlock(self):
f67539c2 TL	393	self.owner.lib.ocf_mngt_cache_unlock(self.cache_handle)
9f95a23c TL	394
9f95a23c TL	395	def add_core(self, core: Core):
f67539c2	396	self.write_lock()
9f95a23c TL	397
	398	c = OcfCompletion(
	399	[
	400	("cache", c_void_p),
	401	("core", c_void_p),
	402	("priv", c_void_p),
	403	("error", c_int),
	404	]
	405	)
	406
	407	self.owner.lib.ocf_mngt_cache_add_core(
	408	self.cache_handle, byref(core.get_cfg()), c, None
	409	)
	410
	411	c.wait()
	412	if c.results["error"]:
f67539c2	413	self.write_unlock()
9f95a23c TL	414	raise OcfError("Failed adding core", c.results["error"])
	415
	416	core.cache = self
	417	core.handle = c.results["core"]
	418	self.cores.append(core)
	419
f67539c2	420	self.write_unlock()
9f95a23c TL	421
9f95a23c TL	422	def remove_core(self, core: Core):
f67539c2	423	self.write_lock()
9f95a23c TL	424
	425	c = OcfCompletion([("priv", c_void_p), ("error", c_int)])
	426
	427	self.owner.lib.ocf_mngt_cache_remove_core(core.handle, c, None)
	428
	429	c.wait()
f67539c2 TL	430	self.write_unlock()
f67539c2 TL	431
9f95a23c	432	if c.results["error"]:
9f95a23c TL	433	raise OcfError("Failed removing core", c.results["error"])
	434
	435	self.cores.remove(core)
	436
9f95a23c TL	437	def get_stats(self):
	438	cache_info = CacheInfo()
	439	usage = UsageStats()
	440	req = RequestsStats()
	441	block = BlocksStats()
	442	errors = ErrorsStats()
	443
f67539c2	444	self.read_lock()
9f95a23c	445
f67539c2	446	status = self.owner.lib.ocf_cache_get_info(self.cache_handle, byref(cache_info))
9f95a23c	447	if status:
f67539c2	448	self.read_unlock()
9f95a23c TL	449	raise OcfError("Failed getting cache info", status)
	450
	451	status = self.owner.lib.ocf_stats_collect_cache(
f67539c2	452	self.cache_handle, byref(usage), byref(req), byref(block), byref(errors)
9f95a23c TL	453	)
9f95a23c TL	454	if status:
f67539c2	455	self.read_unlock()
9f95a23c TL	456	raise OcfError("Failed getting stats", status)
	457
	458	line_size = CacheLineSize(cache_info.cache_line_size)
f67539c2	459	cache_name = self.owner.lib.ocf_cache_get_name(self).decode("ascii")
9f95a23c	460
f67539c2	461	self.read_unlock()
9f95a23c TL	462	return {
	463	"conf": {
	464	"attached": cache_info.attached,
	465	"volume_type": self.owner.volume_types[cache_info.volume_type],
	466	"size": CacheLines(cache_info.size, line_size),
	467	"inactive": {
	468	"occupancy": CacheLines(
f67539c2	469	cache_info.inactive.occupancy.value, line_size
9f95a23c	470	),
f67539c2 TL	471	"dirty": CacheLines(cache_info.inactive.dirty.value, line_size),
f67539c2 TL	472	"clean": CacheLines(cache_info.inactive.clean.value, line_size),
9f95a23c TL	473	},
	474	"occupancy": CacheLines(cache_info.occupancy, line_size),
	475	"dirty": CacheLines(cache_info.dirty, line_size),
	476	"dirty_initial": CacheLines(cache_info.dirty_initial, line_size),
	477	"dirty_for": timedelta(seconds=cache_info.dirty_for),
	478	"cache_mode": CacheMode(cache_info.cache_mode),
	479	"fallback_pt": {
	480	"error_counter": cache_info.fallback_pt.error_counter,
	481	"status": cache_info.fallback_pt.status,
	482	},
	483	"state": cache_info.state,
	484	"eviction_policy": EvictionPolicy(cache_info.eviction_policy),
	485	"cleaning_policy": CleaningPolicy(cache_info.cleaning_policy),
f67539c2	486	"promotion_policy": PromotionPolicy(cache_info.promotion_policy),
9f95a23c TL	487	"cache_line_size": line_size,
	488	"flushed": CacheLines(cache_info.flushed, line_size),
	489	"core_count": cache_info.core_count,
	490	"metadata_footprint": Size(cache_info.metadata_footprint),
	491	"metadata_end_offset": Size(cache_info.metadata_end_offset),
f67539c2	492	"cache_name": cache_name,
9f95a23c TL	493	},
	494	"block": struct_to_dict(block),
	495	"req": struct_to_dict(req),
	496	"usage": struct_to_dict(usage),
	497	"errors": struct_to_dict(errors),
	498	}
	499
	500	def reset_stats(self):
	501	self.owner.lib.ocf_core_stats_initialize_all(self.cache_handle)
	502
	503	def get_default_queue(self):
	504	if not self.io_queues:
	505	raise Exception("No queues added for cache")
	506
	507	return self.io_queues[0]
	508
f67539c2 TL	509	def save(self):
	510	if not self.started:
	511	raise Exception("Not started!")
	512
	513	self.get_and_write_lock()
	514	c = OcfCompletion([("cache", c_void_p), ("priv", c_void_p), ("error", c_int)])
	515	self.owner.lib.ocf_mngt_cache_save(self.cache_handle, c, None)
	516
	517	c.wait()
	518	self.put_and_write_unlock()
	519
	520	if c.results["error"]:
	521	raise OcfError("Failed saving cache", c.results["error"])
	522
9f95a23c TL	523	def stop(self):
	524	if not self.started:
	525	raise Exception("Already stopped!")
	526
f67539c2	527	self.write_lock()
9f95a23c	528
f67539c2	529	c = OcfCompletion([("cache", c_void_p), ("priv", c_void_p), ("error", c_int)])
9f95a23c	530
9f95a23c TL	531	self.owner.lib.ocf_mngt_cache_stop(self.cache_handle, c, None)
	532
	533	c.wait()
	534	if c.results["error"]:
f67539c2	535	self.write_unlock()
9f95a23c TL	536	raise OcfError("Failed stopping cache", c.results["error"])
9f95a23c TL	537
f67539c2	538	self.mngt_queue.put()
9f95a23c TL	539	del self.io_queues[:]
	540	self.started = False
	541
f67539c2	542	self.write_unlock()
9f95a23c TL	543
	544	self.owner.caches.remove(self)
	545
	546	def flush(self):
f67539c2	547	self.write_lock()
9f95a23c	548
f67539c2 TL	549	c = OcfCompletion([("cache", c_void_p), ("priv", c_void_p), ("error", c_int)])
f67539c2 TL	550	self.owner.lib.ocf_mngt_cache_flush(self.cache_handle, c, None)
9f95a23c	551	c.wait()
f67539c2 TL	552	self.write_unlock()
f67539c2 TL	553
9f95a23c	554	if c.results["error"]:
9f95a23c TL	555	raise OcfError("Couldn't flush cache", c.results["error"])
9f95a23c TL	556
9f95a23c	557	def get_name(self):
f67539c2	558	self.read_lock()
9f95a23c TL	559
	560	try:
	561	return str(self.owner.lib.ocf_cache_get_name(self), encoding="ascii")
f67539c2	562	except: # noqa E722
9f95a23c TL	563	raise OcfError("Couldn't get cache name")
9f95a23c TL	564	finally:
f67539c2	565	self.read_unlock()
9f95a23c TL	566
	567
	568	lib = OcfLib.getInstance()
	569	lib.ocf_mngt_cache_remove_core.argtypes = [c_void_p, c_void_p, c_void_p]
	570	lib.ocf_mngt_cache_add_core.argtypes = [c_void_p, c_void_p, c_void_p, c_void_p]
	571	lib.ocf_cache_get_name.argtypes = [c_void_p]
	572	lib.ocf_cache_get_name.restype = c_char_p
	573	lib.ocf_mngt_cache_cleaning_set_policy.argtypes = [c_void_p, c_uint32]
	574	lib.ocf_mngt_cache_cleaning_set_policy.restype = c_int
	575	lib.ocf_mngt_core_set_seq_cutoff_policy_all.argtypes = [c_void_p, c_uint32]
	576	lib.ocf_mngt_core_set_seq_cutoff_policy_all.restype = c_int
f67539c2 TL	577	lib.ocf_stats_collect_cache.argtypes = [
	578	c_void_p,
	579	c_void_p,
	580	c_void_p,
	581	c_void_p,
	582	c_void_p,
	583	]
	584	lib.ocf_stats_collect_cache.restype = c_int
	585	lib.ocf_cache_get_info.argtypes = [c_void_p, c_void_p]
	586	lib.ocf_cache_get_info.restype = c_int
9f95a23c TL	587	lib.ocf_mngt_cache_cleaning_set_param.argtypes = [
	588	c_void_p,
	589	c_uint32,
	590	c_uint32,
	591	c_uint32,
	592	]
	593	lib.ocf_mngt_cache_cleaning_set_param.restype = c_int