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