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

   1 #
   2 # Copyright(c) 2019 Intel Corporation
   3 # SPDX-License-Identifier: BSD-3-Clause-Clear
   4 #
   5
   6 from ctypes import c_void_p, Structure, c_char_p, cast, pointer, byref, c_int
   7
   8 from .logger import LoggerOps, Logger
   9 from .data import DataOps, Data
  10 from .cleaner import CleanerOps, Cleaner
  11 from .metadata_updater import MetadataUpdaterOps, MetadataUpdater
  12 from .shared import OcfError
  13 from ..ocf import OcfLib
  14 from .queue import Queue
  15 from .volume import Volume
  16
  17
  18 class OcfCtxOps(Structure):
  19     _fields_ = [
  20         ("data", DataOps),
  21         ("cleaner", CleanerOps),
  22         ("metadata_updater", MetadataUpdaterOps),
  23         ("logger", LoggerOps),
  24     ]
  25
  26
  27 class OcfCtxCfg(Structure):
  28     _fields_ = [("name", c_char_p), ("ops", OcfCtxOps), ("logger_priv", c_void_p)]
  29
  30
  31 class OcfCtx:
  32     def __init__(self, lib, name, logger, data, mu, cleaner):
  33         self.logger = logger
  34         self.data = data
  35         self.mu = mu
  36         self.cleaner = cleaner
  37         self.ctx_handle = c_void_p()
  38         self.lib = lib
  39         self.volume_types_count = 1
  40         self.volume_types = {}
  41         self.caches = []
  42
  43         self.cfg = OcfCtxCfg(
  44             name=name,
  45             ops=OcfCtxOps(
  46                 data=self.data.get_ops(),
  47                 cleaner=self.cleaner.get_ops(),
  48                 metadata_updater=self.mu.get_ops(),
  49                 logger=logger.get_ops(),
  50             ),
  51             logger_priv=cast(pointer(logger.get_priv()), c_void_p),
  52         )
  53
  54         result = self.lib.ocf_ctx_create(byref(self.ctx_handle), byref(self.cfg))
  55         if result != 0:
  56             raise OcfError("Context initialization failed", result)
  57
  58     def register_volume_type(self, volume_type):
  59         self.volume_types[self.volume_types_count] = volume_type
  60         volume_type.type_id = self.volume_types_count
  61         volume_type.owner = self
  62
  63         result = self.lib.ocf_ctx_register_volume_type(
  64             self.ctx_handle,
  65             self.volume_types_count,
  66             byref(self.volume_types[self.volume_types_count].get_props()),
  67         )
  68         if result != 0:
  69             raise OcfError("Volume type registration failed", result)
  70
  71         self.volume_types_count += 1
  72
  73     def unregister_volume_type(self, vol_type):
  74         if not vol_type.type_id:
  75             raise Exception("Already unregistered")
  76
  77         self.lib.ocf_ctx_unregister_volume_type(
  78             self.ctx_handle, vol_type.type_id
  79         )
  80
  81         del self.volume_types[vol_type.type_id]
  82
  83     def cleanup_volume_types(self):
  84         for k, vol_type in list(self.volume_types.items()):
  85             if vol_type:
  86                 self.unregister_volume_type(vol_type)
  87
  88     def stop_caches(self):
  89         for cache in self.caches[:]:
  90             cache.stop()
  91
  92     def exit(self):
  93         self.stop_caches()
  94         self.cleanup_volume_types()
  95
  96         self.lib.ocf_ctx_put(self.ctx_handle)
  97
  98         self.cfg = None
  99         self.logger = None
 100         self.data = None
 101         self.mu = None
 102         self.cleaner = None
 103         Queue._instances_ = {}
 104         Volume._instances_ = {}
 105         Data._instances_ = {}
 106         Logger._instances_ = {}
 107
 108
 109 def get_default_ctx(logger):
 110     return OcfCtx(
 111         OcfLib.getInstance(),
 112         b"PyOCF default ctx",
 113         logger,
 114         Data,
 115         MetadataUpdater,
 116         Cleaner,
 117     )
 118
 119
 120 lib = OcfLib.getInstance()
 121 lib.ocf_mngt_cache_get_by_name.argtypes = [c_void_p, c_void_p, c_void_p]
 122 lib.ocf_mngt_cache_get_by_name.restype = c_int