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

   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,
  11     c_char,
  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):
  45     MAX_CACHE_NAME_SIZE = 32
  46     _fields_ = [
  47         ("_name", c_char * MAX_CACHE_NAME_SIZE),
  48         ("_cache_mode", c_uint32),
  49         ("_eviction_policy", c_uint32),
  50         ("_promotion_policy", c_uint32),
  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
  73 class ConfValidValues:
  74     promotion_nhit_insertion_threshold_range = range(2, 1000)
  75     promotion_nhit_trigger_threshold_range = range(0, 100)
  76
  77
  78 class CacheMode(IntEnum):
  79     WT = 0
  80     WB = 1
  81     WA = 2
  82     PT = 3
  83     WI = 4
  84     WO = 5
  85     DEFAULT = WT
  86
  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
  96
  97 class EvictionPolicy(IntEnum):
  98     LRU = 0
  99     DEFAULT = LRU
 100
 101
 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
 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:
 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,
 147         name: str = "cache",
 148         cache_mode: CacheMode = CacheMode.DEFAULT,
 149         eviction_policy: EvictionPolicy = EvictionPolicy.DEFAULT,
 150         promotion_policy: PromotionPolicy = PromotionPolicy.DEFAULT,
 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(
 166             _name=name.encode("ascii"),
 167             _cache_mode=cache_mode,
 168             _eviction_policy=eviction_policy,
 169             _promotion_policy=promotion_policy,
 170             _cache_line_size=cache_line_size,
 171             _metadata_layout=metadata_layout,
 172             _metadata_volatile=metadata_volatile,
 173             _backfill=Backfill(
 174                 _max_queue_size=max_queue_size, _queue_unblock_size=queue_unblock_size
 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 = []
 183         self.cores = []
 184
 185     def start_cache(self, default_io_queue: Queue = None, mngt_queue: Queue = None):
 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
 193         self.mngt_queue = mngt_queue or Queue(self, "mgmt-{}".format(self.get_name()))
 194
 195         if default_io_queue:
 196             self.io_queues += [default_io_queue]
 197         else:
 198             self.io_queues += [Queue(self, "default-io-{}".format(self.get_name()))]
 199
 200         status = self.owner.lib.ocf_mngt_cache_set_mngt_queue(self, self.mngt_queue)
 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):
 207         self.write_lock()
 208         status = self.owner.lib.ocf_mngt_cache_set_mode(self.cache_handle, cache_mode)
 209
 210         self.write_unlock()
 211
 212         if status:
 213             raise OcfError("Error changing cache mode", status)
 214
 215     def set_cleaning_policy(self, cleaning_policy: CleaningPolicy):
 216         self.write_lock()
 217
 218         status = self.owner.lib.ocf_mngt_cache_cleaning_set_policy(
 219             self.cache_handle, cleaning_policy
 220         )
 221
 222         self.write_unlock()
 223
 224         if status:
 225             raise OcfError("Error changing cleaning policy", status)
 226
 227     def set_cleaning_policy_param(
 228         self, cleaning_policy: CleaningPolicy, param_id, param_value
 229     ):
 230         self.write_lock()
 231
 232         status = self.owner.lib.ocf_mngt_cache_cleaning_set_param(
 233             self.cache_handle, cleaning_policy, param_id, param_value
 234         )
 235
 236         self.write_unlock()
 237
 238         if status:
 239             raise OcfError("Error setting cleaning policy param", status)
 240
 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)
 277
 278     def set_seq_cut_off_policy(self, policy: SeqCutOffPolicy):
 279         self.write_lock()
 280
 281         status = self.owner.lib.ocf_mngt_core_set_seq_cutoff_policy_all(
 282             self.cache_handle, policy
 283         )
 284
 285         self.write_unlock()
 286
 287         if status:
 288             raise OcfError("Error setting cache seq cut off policy", status)
 289
 290     def configure_device(
 291         self, device, force=False, perform_test=True, cache_line_size=None
 292     ):
 293         self.device = device
 294         self.device_name = device.uuid
 295         self.dev_cfg = CacheDeviceConfig(
 296             _uuid=Uuid(
 297                 _data=cast(
 298                     create_string_buffer(self.device_name.encode("ascii")), c_char_p
 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)
 316         self.write_lock()
 317
 318         c = OcfCompletion([("cache", c_void_p), ("priv", c_void_p), ("error", c_int)])
 319
 320         device.owner.lib.ocf_mngt_cache_attach(
 321             self.cache_handle, byref(self.dev_cfg), c, None
 322         )
 323
 324         c.wait()
 325         self.write_unlock()
 326
 327         if c.results["error"]:
 328             raise OcfError("Attaching cache device failed", c.results["error"])
 329
 330     def load_cache(self, device):
 331         self.configure_device(device)
 332         c = OcfCompletion([("cache", c_void_p), ("priv", c_void_p), ("error", c_int)])
 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
 342     def load_from_device(cls, device, name="cache"):
 343         c = cls(name=name, owner=device.owner)
 344
 345         c.start_cache()
 346         try:
 347             c.load_cache(device)
 348         except:  # noqa E722
 349             c.stop()
 350             raise
 351
 352         return c
 353
 354     @classmethod
 355     def start_on_device(cls, device, **kwargs):
 356         c = cls(owner=device.owner, **kwargs)
 357
 358         c.start_cache()
 359         try:
 360             c.attach_device(device, force=True)
 361         except:  # noqa E722
 362             c.stop()
 363             raise
 364
 365         return c
 366
 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
 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"])
 381
 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"])
 388
 389     def read_unlock(self):
 390         self.owner.lib.ocf_mngt_cache_read_unlock(self.cache_handle)
 391
 392     def write_unlock(self):
 393         self.owner.lib.ocf_mngt_cache_unlock(self.cache_handle)
 394
 395     def add_core(self, core: Core):
 396         self.write_lock()
 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"]:
 413             self.write_unlock()
 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
 420         self.write_unlock()
 421
 422     def remove_core(self, core: Core):
 423         self.write_lock()
 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()
 430         self.write_unlock()
 431
 432         if c.results["error"]:
 433             raise OcfError("Failed removing core", c.results["error"])
 434
 435         self.cores.remove(core)
 436
 437     def get_stats(self):
 438         cache_info = CacheInfo()
 439         usage = UsageStats()
 440         req = RequestsStats()
 441         block = BlocksStats()
 442         errors = ErrorsStats()
 443
 444         self.read_lock()
 445
 446         status = self.owner.lib.ocf_cache_get_info(self.cache_handle, byref(cache_info))
 447         if status:
 448             self.read_unlock()
 449             raise OcfError("Failed getting cache info", status)
 450
 451         status = self.owner.lib.ocf_stats_collect_cache(
 452             self.cache_handle, byref(usage), byref(req), byref(block), byref(errors)
 453         )
 454         if status:
 455             self.read_unlock()
 456             raise OcfError("Failed getting stats", status)
 457
 458         line_size = CacheLineSize(cache_info.cache_line_size)
 459         cache_name = self.owner.lib.ocf_cache_get_name(self).decode("ascii")
 460
 461         self.read_unlock()
 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(
 469                         cache_info.inactive.occupancy.value, line_size
 470                     ),
 471                     "dirty": CacheLines(cache_info.inactive.dirty.value, line_size),
 472                     "clean": CacheLines(cache_info.inactive.clean.value, line_size),
 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),
 486                 "promotion_policy": PromotionPolicy(cache_info.promotion_policy),
 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),
 492                 "cache_name": cache_name,
 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
 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
 523     def stop(self):
 524         if not self.started:
 525             raise Exception("Already stopped!")
 526
 527         self.write_lock()
 528
 529         c = OcfCompletion([("cache", c_void_p), ("priv", c_void_p), ("error", c_int)])
 530
 531         self.owner.lib.ocf_mngt_cache_stop(self.cache_handle, c, None)
 532
 533         c.wait()
 534         if c.results["error"]:
 535             self.write_unlock()
 536             raise OcfError("Failed stopping cache", c.results["error"])
 537
 538         self.mngt_queue.put()
 539         del self.io_queues[:]
 540         self.started = False
 541
 542         self.write_unlock()
 543
 544         self.owner.caches.remove(self)
 545
 546     def flush(self):
 547         self.write_lock()
 548
 549         c = OcfCompletion([("cache", c_void_p), ("priv", c_void_p), ("error", c_int)])
 550         self.owner.lib.ocf_mngt_cache_flush(self.cache_handle, c, None)
 551         c.wait()
 552         self.write_unlock()
 553
 554         if c.results["error"]:
 555             raise OcfError("Couldn't flush cache", c.results["error"])
 556
 557     def get_name(self):
 558         self.read_lock()
 559
 560         try:
 561             return str(self.owner.lib.ocf_cache_get_name(self), encoding="ascii")
 562         except:  # noqa E722
 563             raise OcfError("Couldn't get cache name")
 564         finally:
 565             self.read_unlock()
 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
 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
 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