1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
|
# Copyright 2013 Hewlett-Packard Development Company, L.P.
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import array
import hashlib
import struct
import threading
from oslo.config import cfg
from ironic.common import exception
from ironic.common.i18n import _
from ironic.db import api as dbapi
hash_opts = [
cfg.IntOpt('hash_partition_exponent',
default=16,
help='Exponent to determine number of hash partitions to use '
'when distributing load across conductors. Larger values '
'will result in more even distribution of load and less '
'load when rebalancing the ring, but more memory usage. '
'Number of partitions is (2^hash_partition_exponent).'),
cfg.IntOpt('hash_distribution_replicas',
default=1,
help='[Experimental Feature] '
'Number of hosts to map onto each hash partition. '
'Setting this to more than one will cause additional '
'conductor services to prepare deployment environments '
'and potentially allow the Ironic cluster to recover '
'more quickly if a conductor instance is terminated.'),
]
CONF = cfg.CONF
CONF.register_opts(hash_opts)
class HashRing(object):
def __init__(self, hosts, replicas=None):
"""Create a new hash ring across the specified hosts.
:param hosts: an iterable of hosts which will be mapped.
:param replicas: number of hosts to map to each hash partition,
or len(hosts), which ever is lesser.
Default: CONF.hash_distribution_replicas
"""
if replicas is None:
replicas = CONF.hash_distribution_replicas
try:
self.hosts = list(hosts)
self.replicas = replicas if replicas <= len(hosts) else len(hosts)
except TypeError:
raise exception.Invalid(
_("Invalid hosts supplied when building HashRing."))
self.partition_shift = 32 - CONF.hash_partition_exponent
self.part2host = array.array('H')
for p in range(2 ** CONF.hash_partition_exponent):
self.part2host.append(p % len(hosts))
def _get_partition(self, data):
try:
return (struct.unpack_from('>I', hashlib.md5(data).digest())[0]
>> self.partition_shift)
except TypeError:
raise exception.Invalid(
_("Invalid data supplied to HashRing.get_hosts."))
def get_hosts(self, data, ignore_hosts=None):
"""Get the list of hosts which the supplied data maps onto.
:param data: A string identifier to be mapped across the ring.
:param ignore_hosts: A list of hosts to skip when performing the hash.
Useful to temporarily skip down hosts without
performing a full rebalance.
Default: None.
:returns: a list of hosts.
The length of this list depends on the number of replicas
this `HashRing` was created with. It may be less than this
if ignore_hosts is not None.
"""
host_ids = []
if ignore_hosts is None:
ignore_host_ids = []
else:
ignore_host_ids = [self.hosts.index(h)
for h in ignore_hosts if h in self.hosts]
partition = self._get_partition(data)
for replica in range(0, self.replicas):
if len(host_ids + ignore_host_ids) == len(self.hosts):
# prevent infinite loop
break
while self.part2host[partition] in host_ids + ignore_host_ids:
partition += 1
if partition >= len(self.part2host):
partition = 0
host_ids.append(self.part2host[partition])
return [self.hosts[h] for h in host_ids]
class HashRingManager(object):
_hash_rings = None
_lock = threading.Lock()
def __init__(self):
self.dbapi = dbapi.get_instance()
@property
def ring(self):
# Hot path, no lock
if self._hash_rings is not None:
return self._hash_rings
with self._lock:
if self._hash_rings is None:
rings = self._load_hash_rings()
self.__class__._hash_rings = rings
return self._hash_rings
def _load_hash_rings(self):
rings = {}
d2c = self.dbapi.get_active_driver_dict()
for driver_name, hosts in d2c.iteritems():
rings[driver_name] = HashRing(hosts)
return rings
@classmethod
def reset(self):
with self._lock:
self._hash_rings = None
def __getitem__(self, driver_name):
try:
return self.ring[driver_name]
except KeyError:
raise exception.DriverNotFound(
_("The driver '%s' is unknown.") % driver_name)
|
