Merge "ring: keep handoff iterators on one ring version"
This commit is contained in:
Zuul authored and Gerrit Code Review committed
commit
c678741b32
2 files changed
+61
-18
No files matched your search
+27
-18
@@ -575,6 +575,15 @@ class Ring(object):
|
||||
if time() > self._rtime:
|
||||
self._reload()
|
||||
primary_nodes = self._get_part_nodes(part)
|
||||
# Keep this ring version across yields. Another request may reload
|
||||
# the ring while this iterator waits for a backend response.
|
||||
devs = self._devs
|
||||
replica2part2dev_id = self._replica2part2dev_id
|
||||
part_shift = self._part_shift
|
||||
num_regions = self._num_regions
|
||||
num_zones = self._num_zones
|
||||
num_ips = self._num_ips
|
||||
num_assigned_devs = self._num_assigned_devs
|
||||
used = set(d['id'] for d in primary_nodes)
|
||||
index = count()
|
||||
same_regions = set(d['region'] for d in primary_nodes)
|
||||
@@ -582,14 +591,14 @@ class Ring(object):
|
||||
same_ips = set(
|
||||
(d['region'], d['zone'], d['ip']) for d in primary_nodes)
|
||||
|
||||
parts = len(self._replica2part2dev_id[0])
|
||||
parts = len(replica2part2dev_id[0])
|
||||
part_hash = md5(str(part).encode('ascii'),
|
||||
usedforsecurity=False).digest()
|
||||
start = struct.unpack_from('>I', part_hash)[0] >> self._part_shift
|
||||
start = struct.unpack_from('>I', part_hash)[0] >> part_shift
|
||||
inc = int(parts / 65536) or 1
|
||||
# Multiple loops for execution speed; the checks and bookkeeping get
|
||||
# simpler as you go along
|
||||
hit_all_regions = len(same_regions) == self._num_regions
|
||||
hit_all_regions = len(same_regions) == num_regions
|
||||
for handoff_part in chain(range(start, parts, inc),
|
||||
range(inc - ((parts - start) % inc),
|
||||
start, inc)):
|
||||
@@ -597,10 +606,10 @@ class Ring(object):
|
||||
# At this point, there are no regions left untouched, so we
|
||||
# can stop looking.
|
||||
break
|
||||
for part2dev_id in self._replica2part2dev_id:
|
||||
for part2dev_id in replica2part2dev_id:
|
||||
if handoff_part < len(part2dev_id):
|
||||
dev_id = part2dev_id[handoff_part]
|
||||
dev = self._devs[dev_id]
|
||||
dev = devs[dev_id]
|
||||
region = dev['region']
|
||||
if dev_id not in used and region not in same_regions:
|
||||
yield dict(dev, handoff_index=next(index))
|
||||
@@ -610,11 +619,11 @@ class Ring(object):
|
||||
ip = (region, zone, dev['ip'])
|
||||
same_zones.add((region, zone))
|
||||
same_ips.add(ip)
|
||||
if len(same_regions) == self._num_regions:
|
||||
if len(same_regions) == num_regions:
|
||||
hit_all_regions = True
|
||||
break
|
||||
|
||||
hit_all_zones = len(same_zones) == self._num_zones
|
||||
hit_all_zones = len(same_zones) == num_zones
|
||||
for handoff_part in chain(range(start, parts, inc),
|
||||
range(inc - ((parts - start) % inc),
|
||||
start, inc)):
|
||||
@@ -622,10 +631,10 @@ class Ring(object):
|
||||
# Much like we stopped looking for fresh regions before, we
|
||||
# can now stop looking for fresh zones; there are no more.
|
||||
break
|
||||
for part2dev_id in self._replica2part2dev_id:
|
||||
for part2dev_id in replica2part2dev_id:
|
||||
if handoff_part < len(part2dev_id):
|
||||
dev_id = part2dev_id[handoff_part]
|
||||
dev = self._devs[dev_id]
|
||||
dev = devs[dev_id]
|
||||
zone = (dev['region'], dev['zone'])
|
||||
if dev_id not in used and zone not in same_zones:
|
||||
yield dict(dev, handoff_index=next(index))
|
||||
@@ -633,11 +642,11 @@ class Ring(object):
|
||||
same_zones.add(zone)
|
||||
ip = zone + (dev['ip'],)
|
||||
same_ips.add(ip)
|
||||
if len(same_zones) == self._num_zones:
|
||||
if len(same_zones) == num_zones:
|
||||
hit_all_zones = True
|
||||
break
|
||||
|
||||
hit_all_ips = len(same_ips) == self._num_ips
|
||||
hit_all_ips = len(same_ips) == num_ips
|
||||
for handoff_part in chain(range(start, parts, inc),
|
||||
range(inc - ((parts - start) % inc),
|
||||
start, inc)):
|
||||
@@ -645,20 +654,20 @@ class Ring(object):
|
||||
# We've exhausted the pool of unused backends, so stop
|
||||
# looking.
|
||||
break
|
||||
for part2dev_id in self._replica2part2dev_id:
|
||||
for part2dev_id in replica2part2dev_id:
|
||||
if handoff_part < len(part2dev_id):
|
||||
dev_id = part2dev_id[handoff_part]
|
||||
dev = self._devs[dev_id]
|
||||
dev = devs[dev_id]
|
||||
ip = (dev['region'], dev['zone'], dev['ip'])
|
||||
if dev_id not in used and ip not in same_ips:
|
||||
yield dict(dev, handoff_index=next(index))
|
||||
used.add(dev_id)
|
||||
same_ips.add(ip)
|
||||
if len(same_ips) == self._num_ips:
|
||||
if len(same_ips) == num_ips:
|
||||
hit_all_ips = True
|
||||
break
|
||||
|
||||
hit_all_devs = len(used) == self._num_assigned_devs
|
||||
hit_all_devs = len(used) == num_assigned_devs
|
||||
for handoff_part in chain(range(start, parts, inc),
|
||||
range(inc - ((parts - start) % inc),
|
||||
start, inc)):
|
||||
@@ -666,13 +675,13 @@ class Ring(object):
|
||||
# We've used every device we have, so let's stop looking for
|
||||
# unused devices now.
|
||||
break
|
||||
for part2dev_id in self._replica2part2dev_id:
|
||||
for part2dev_id in replica2part2dev_id:
|
||||
if handoff_part < len(part2dev_id):
|
||||
dev_id = part2dev_id[handoff_part]
|
||||
if dev_id not in used:
|
||||
dev = self._devs[dev_id]
|
||||
dev = devs[dev_id]
|
||||
yield dict(dev, handoff_index=next(index))
|
||||
used.add(dev_id)
|
||||
if len(used) == self._num_assigned_devs:
|
||||
if len(used) == num_assigned_devs:
|
||||
hit_all_devs = True
|
||||
break
|
||||
@@ -1108,6 +1108,40 @@ class TestRing(TestRingBase):
|
||||
'handoff differs at position %d\n%s\n%s' % (
|
||||
index, dev_ids[index:], exp_handoffs[index:]))
|
||||
|
||||
def test_get_more_nodes_keeps_ring_version_across_reload(self):
|
||||
# Each tier layout selects the handoffs in a different stage.
|
||||
for tier in ('region', 'zone', 'ip', None):
|
||||
with self.subTest(tier=tier):
|
||||
devs = []
|
||||
for i in range(8):
|
||||
dev = {'id': i, 'region': 0, 'zone': 0, 'ip': '10.0.0.0',
|
||||
'port': 6200, 'device': 'sd%d' % i, 'weight': 1.0}
|
||||
if tier == 'ip':
|
||||
dev['ip'] = '10.0.0.%d' % i
|
||||
elif tier:
|
||||
dev[tier] = i
|
||||
devs.append(dev)
|
||||
# Partition 0 has primaries 0 and 1, and handoffs 2 to 7.
|
||||
tables = [array.array('H', [0, 2, 4, 6]),
|
||||
array.array('H', [1, 3, 5, 7])]
|
||||
ring.RingData(tables, devs, 30).save(
|
||||
self.testgz, format_version=self.FORMAT_VERSION)
|
||||
r = ring.Ring(self.testgz)
|
||||
expected = list(r.get_more_nodes(0))
|
||||
self.assertEqual(set(range(2, 8)),
|
||||
{d['id'] for d in expected})
|
||||
node_iter = r.get_more_nodes(0)
|
||||
self.assertEqual(expected[0], next(node_iter))
|
||||
|
||||
# Another request reloads a ring without the handoff devices.
|
||||
new_tables = [array.array('H', [0] * 4),
|
||||
array.array('H', [1] * 4)]
|
||||
ring.RingData(new_tables, devs[:2], 30).save(
|
||||
self.testgz, format_version=self.FORMAT_VERSION)
|
||||
r._reload(force=True)
|
||||
|
||||
self.assertEqual(expected[1:], list(node_iter))
|
||||
|
||||
def test_get_more_nodes_with_zero_weight_region(self):
|
||||
rb = ring.RingBuilder(8, 3, 1)
|
||||
devs = [
|
||||
|
||||
Reference in new issue
Block a user