<s> from oslo_serialization import jsonutils <EOL> from oslo_utils import versionutils <EOL> from nova import db <EOL> from nova import exception <EOL> from nova . objects import base <EOL> from nova . objects import fields as obj_fields <EOL> from nova . virt import hardware <EOL> @ base . NovaObjectRegistry . register <EOL> class InstanceNUMACell ( base . NovaObject , <EOL> base . NovaObjectDictCompat ) : <EOL> VERSION = '<STR_LIT>' <EOL> def obj_make_compatible ( self , primitive , target_version ) : <EOL> super ( InstanceNUMACell , self ) . obj_make_compatible ( primitive , <EOL> target_version ) <EOL> target_version = versionutils . convert_version_to_tuple ( target_version ) <EOL> if target_version < ( <NUM_LIT:1> , <NUM_LIT:3> ) : <EOL> primitive . pop ( '<STR_LIT>' , None ) <EOL> primitive . pop ( '<STR_LIT>' , None ) <EOL> fields = { <EOL> '<STR_LIT:id>' : obj_fields . IntegerField ( ) , <EOL> '<STR_LIT>' : obj_fields . SetOfIntegersField ( ) , <EOL> '<STR_LIT>' : obj_fields . IntegerField ( ) , <EOL> '<STR_LIT>' : obj_fields . IntegerField ( nullable = True ) , <EOL> '<STR_LIT>' : obj_fields . ObjectField ( '<STR_LIT>' , <EOL> nullable = True ) , <EOL> '<STR_LIT>' : obj_fields . DictOfIntegersField ( nullable = True ) , <EOL> '<STR_LIT>' : obj_fields . CPUAllocationPolicyField ( nullable = True ) , <EOL> '<STR_LIT>' : obj_fields . CPUThreadAllocationPolicyField ( <EOL> nullable = True ) , <EOL> } <EOL> cpu_pinning = obj_fields . DictProxyField ( '<STR_LIT>' ) <EOL> def __init__ ( self , ** kwargs ) : <EOL> super ( InstanceNUMACell , self ) . __init__ ( ** kwargs ) <EOL> if '<STR_LIT>' not in kwargs : <EOL> self . pagesize = None <EOL> self . obj_reset_changes ( [ '<STR_LIT>' ] ) <EOL> if '<STR_LIT>' not in kwargs : <EOL> self . cpu_topology = None <EOL> self . obj_reset_changes ( [ '<STR_LIT>' ] ) <EOL> if '<STR_LIT>' not in kwargs : <EOL> self . cpu_pinning = None <EOL> self . obj_reset_changes ( [ '<STR_LIT>' ] ) <EOL> if '<STR_LIT>' not in kwargs : <EOL> self . cpu_policy = None <EOL> self . obj_reset_changes ( [ '<STR_LIT>' ] ) <EOL> if '<STR_LIT>' not in kwargs : <EOL> self . cpu_thread_policy = None <EOL> self . obj_reset_changes ( [ '<STR_LIT>' ] ) <EOL> def __len__ ( self ) : <EOL> return len ( self . cpuset ) <EOL> def _to_dict ( self ) : <EOL> return { '<STR_LIT>' : hardware . format_cpu_spec ( self . cpuset , <EOL> allow_ranges = False ) , <EOL> '<STR_LIT>' : { '<STR_LIT>' : self . memory } , <EOL> '<STR_LIT:id>' : self . id , <EOL> '<STR_LIT>' : self . pagesize } <EOL> @ classmethod <EOL> def _from_dict ( cls , data_dict ) : <EOL> cpuset = hardware . parse_cpu_spec ( data_dict . get ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> memory = data_dict . get ( '<STR_LIT>' , { } ) . get ( '<STR_LIT>' , <NUM_LIT:0> ) <EOL> cell_id = data_dict . get ( '<STR_LIT:id>' ) <EOL> pagesize = data_dict . get ( '<STR_LIT>' ) <EOL> return cls ( id = cell_id , cpuset = cpuset , <EOL> memory = memory , pagesize = pagesize ) <EOL> @ property <EOL> def siblings ( self ) : <EOL> cpu_list = sorted ( list ( self . cpuset ) ) <EOL> threads = <NUM_LIT:0> <EOL> if self . cpu_topology : <EOL> threads = self . cpu_topology . threads <EOL> if threads == <NUM_LIT:1> : <EOL> threads = <NUM_LIT:0> <EOL> return list ( map ( set , zip ( * [ iter ( cpu_list ) ] * threads ) ) ) <EOL> @ property <EOL> def cpu_pinning_requested ( self ) : <EOL> return self . cpu_policy == obj_fields . CPUAllocationPolicy . DEDICATED <EOL> def pin ( self , vcpu , pcpu ) : <EOL> if vcpu not in self . cpuset : <EOL> return <EOL> pinning_dict = self . cpu_pinning or { } <EOL> pinning_dict [ vcpu ] = pcpu <EOL> self . cpu_pinning = pinning_dict <EOL> def pin_vcpus ( self , * cpu_pairs ) : <EOL> for vcpu , pcpu in cpu_pairs : <EOL> self . pin ( vcpu , pcpu ) <EOL> def clear_host_pinning ( self ) : <EOL> """<STR_LIT>""" <EOL> self . id = - <NUM_LIT:1> <EOL> self . cpu_pinning = { } <EOL> return self <EOL> @ base . NovaObjectRegistry . register <EOL> class InstanceNUMATopology ( base . NovaObject , <EOL> base . NovaObjectDictCompat ) : <EOL> VERSION = '<STR_LIT>' <EOL> fields = { <EOL> '<STR_LIT:id>' : obj_fields . IntegerField ( ) , <EOL> '<STR_LIT>' : obj_fields . UUIDField ( ) , <EOL> '<STR_LIT>' : obj_fields . ListOfObjectsField ( '<STR_LIT>' ) , <EOL> } <EOL> @ classmethod <EOL> def obj_from_primitive ( cls , primitive , context = None ) : <EOL> if '<STR_LIT>' in primitive : <EOL> obj_topology = super ( InstanceNUMATopology , cls ) . obj_from_primitive ( <EOL> primitive , context = None ) <EOL> else : <EOL> obj_topology = InstanceNUMATopology . _from_dict ( primitive ) <EOL> obj_topology . id = <NUM_LIT:0> <EOL> return obj_topology <EOL> @ classmethod <EOL> def obj_from_db_obj ( cls , instance_uuid , db_obj ) : <EOL> primitive = jsonutils . loads ( db_obj ) <EOL> obj_topology = cls . obj_from_primitive ( primitive ) <EOL> if '<STR_LIT>' not in db_obj : <EOL> obj_topology . instance_uuid = instance_uuid <EOL> obj_topology . obj_reset_changes ( ) <EOL> return obj_topology <EOL> @ base . remotable <EOL> def create ( self ) : <EOL> self . _save ( ) <EOL> def _save ( self ) : <EOL> values = { '<STR_LIT>' : self . _to_json ( ) } <EOL> db . instance_extra_update_by_uuid ( self . _context , self . instance_uuid , <EOL> values ) <EOL> self . obj_reset_changes ( ) <EOL> @ classmethod <EOL> def delete_by_instance_uuid ( cls , context , instance_uuid ) : <EOL> values = { '<STR_LIT>' : None } <EOL> db . instance_extra_update_by_uuid ( context , instance_uuid , <EOL> values ) <EOL> @ base . remotable_classmethod <EOL> def get_by_instance_uuid ( cls , context , instance_uuid ) : <EOL> db_extra = db . instance_extra_get_by_instance_uuid ( <EOL> context , instance_uuid , columns = [ '<STR_LIT>' ] ) <EOL> if not db_extra : <EOL> raise exception . NumaTopologyNotFound ( instance_uuid = instance_uuid ) <EOL> if db_extra [ '<STR_LIT>' ] is None : <EOL> return None <EOL> return cls . obj_from_db_obj ( instance_uuid , db_extra [ '<STR_LIT>' ] ) <EOL> def _to_json ( self ) : <EOL> return jsonutils . dumps ( self . obj_to_primitive ( ) ) <EOL> def __len__ ( self ) : <EOL> """<STR_LIT>""" <EOL> return len ( self . cells ) <EOL> def _to_dict ( self ) : <EOL> return { '<STR_LIT>' : [ cell . _to_dict ( ) for cell in self . cells ] } <EOL> @ classmethod <EOL> def _from_dict ( cls , data_dict ) : <EOL> return cls ( cells = [ <EOL> InstanceNUMACell . _from_dict ( cell_dict ) <EOL> for cell_dict in data_dict . get ( '<STR_LIT>' , [ ] ) ] ) <EOL> @ property <EOL> def cpu_pinning_requested ( self ) : <EOL> return all ( cell . cpu_pinning_requested for cell in self . cells ) <EOL> def clear_host_pinning ( self ) : <EOL> """<STR_LIT>""" <EOL> for cell in self . cells : <EOL> cell . clear_host_pinning ( ) <EOL> return self </s>
