codex-storage
/
das-research
mirror of https://github.com/codex-storage/das-research.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #43 from status-im/per-validator-progress 

Per validator progress
This commit is contained in:
Csaba Kiraly 2023-05-02 12:10:41 +02:00 committed by GitHub
parent 1638a18507 eea7d11c24
commit 2096808488
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 77 additions and 45 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
DAS/observer.py
View File

@ -46,16 +46,18 @@ class Observer:
arrived = 0
expected = 0
ready = 0
validatedall = 0
validated = 0
for val in validators:
if val.amIproposer == 0:
(a, e) = val.checkStatus()
(a, e, v) = val.checkStatus()
arrived += a
expected += e
if a == e:
ready += 1
validated += val.vpn
return (arrived, expected, ready, validated)
validatedall += val.vpn
validated += v
return (arrived, expected, ready, validatedall, validated)
def getProgress(self, validators):
"""Calculate current simulation progress with different metrics.
@ -69,14 +71,15 @@ class Observer:
but counts a validator only if its support node's all validators see all interesting segments
TODO: add real per validator progress counter
"""
arrived, expected, ready, validated = self.checkStatus(validators)
arrived, expected, ready, validatedall, validated = self.checkStatus(validators)
missingSamples = expected - arrived
sampleProgress = arrived / expected
nodeProgress = ready / (len(validators)-1)
validatorCnt = sum([v.vpn for v in validators[1:]])
validatorAllProgress = validatedall / validatorCnt
validatorProgress = validated / validatorCnt
return missingSamples, sampleProgress, nodeProgress, validatorProgress
return missingSamples, sampleProgress, nodeProgress, validatorAllProgress, validatorProgress
def getTrafficStats(self, validators):
"""Summary statistics of traffic measurements in a timestep."""

64
DAS/simulator.py
View File

@ -26,6 +26,10 @@ class Simulator:
self.proposerID = 0
self.glob = []
self.execID = execID
self.distR = []
self.distC = []
self.nodeRows = []
self.nodeColumns = []
# In GossipSub the initiator might push messages without participating in the mesh.
# proposerPublishOnly regulates this behavior. If set to true, the proposer is not
@ -45,17 +49,22 @@ class Simulator:
self.validators = []
if self.config.evenLineDistribution:
lightVal = int(self.shape.numberNodes * self.shape.class1ratio * self.shape.vpn1)
heavyVal = int(self.shape.numberNodes * (1-self.shape.class1ratio) * self.shape.vpn2)
lightNodes = int(self.shape.numberNodes * self.shape.class1ratio)
heavyNodes = self.shape.numberNodes - lightNodes
lightVal = lightNodes * self.shape.vpn1
heavyVal = heavyNodes * self.shape.vpn2
totalValidators = lightVal + heavyVal
totalRows = totalValidators * self.shape.chi
rows = list(range(self.shape.blockSize)) * (int(totalRows/self.shape.blockSize)+1)
columns = list(range(self.shape.blockSize)) * (int(totalRows/self.shape.blockSize)+1)
offset = heavyVal*self.shape.chi
rows = rows[0:totalRows]
columns = columns[0:totalRows]
random.shuffle(rows)
random.shuffle(columns)
self.logger.debug("There is a total of %d validators" % totalValidators, extra=self.format)
self.logger.debug("Shuffling a total of %d rows/columns" % len(rows), extra=self.format)
offset = lightVal*self.shape.chi
self.logger.debug("There is a total of %d nodes, %d light and %d heavy." % (self.shape.numberNodes, lightNodes, heavyNodes), extra=self.format)
self.logger.debug("There is a total of %d validators, %d in light nodes and %d in heavy nodes" % (totalValidators, lightVal, heavyVal), extra=self.format)
self.logger.debug("Shuffling a total of %d rows/columns to be assigned (X=%d)" % (len(rows), self.shape.chi), extra=self.format)
self.logger.debug("Shuffled rows: %s" % str(rows), extra=self.format)
self.logger.debug("Shuffled columns: %s" % str(columns), extra=self.format)
@ -63,20 +72,22 @@ class Simulator:
assignedCols = []
for i in range(self.shape.numberNodes):
if self.config.evenLineDistribution:
if i < int(heavyVal/self.shape.vpn2): # First start with the heavy nodes
start = i *self.shape.chi*self.shape.vpn2
end = (i+1)*self.shape.chi*self.shape.vpn2
else: # Then the solo stakers
j = i - int(heavyVal/self.shape.vpn2)
start = offset+( j *self.shape.chi)
end = offset+((j+1)*self.shape.chi)
r = set(rows[start:end])
c = set(columns[start:end])
if i < int(lightVal/self.shape.vpn1): # First start with the light nodes
start = i *self.shape.chi*self.shape.vpn1
end = (i+1)*self.shape.chi*self.shape.vpn1
else:
j = i - int(lightVal/self.shape.vpn1)
start = offset+( j *self.shape.chi*self.shape.vpn2)
end = offset+((j+1)*self.shape.chi*self.shape.vpn2)
r = rows[start:end]
c = columns[start:end]
val = Validator(i, int(not i!=0), self.logger, self.shape, r, c)
self.logger.debug("Validators %d row IDs: %s" % (val.ID, val.rowIDs), extra=self.format)
self.logger.debug("Validators %d column IDs: %s" % (val.ID, val.columnIDs), extra=self.format)
self.logger.debug("Node %d has row IDs: %s" % (val.ID, val.rowIDs), extra=self.format)
self.logger.debug("Node %d has column IDs: %s" % (val.ID, val.columnIDs), extra=self.format)
assignedRows = assignedRows + list(r)
assignedCols = assignedCols + list(c)
self.nodeRows.append(val.rowIDs)
self.nodeColumns.append(val.columnIDs)
else:
val = Validator(i, int(not i!=0), self.logger, self.shape)
@ -104,14 +115,12 @@ class Simulator:
columnChannels[id].append(v)
# Check rows/columns distribution
distR = []
distC = []
for r in rowChannels:
distR.append(len(r))
self.distR.append(len(r))
for c in columnChannels:
distC.append(len(c))
self.logger.debug("Number of validators per row; Min: %d, Max: %d" % (min(distR), max(distR)), extra=self.format)
self.logger.debug("Number of validators per column; Min: %d, Max: %d" % (min(distC), max(distC)), extra=self.format)
self.distC.append(len(c))
self.logger.debug("Number of validators per row; Min: %d, Max: %d" % (min(self.distR), max(self.distR)), extra=self.format)
self.logger.debug("Number of validators per column; Min: %d, Max: %d" % (min(self.distC), max(self.distC)), extra=self.format)
for id in range(self.shape.blockSize):
@ -210,7 +219,7 @@ class Simulator:
def run(self):
"""It runs the main simulation until the block is available or it gets stucked."""
self.glob.checkRowsColumns(self.validators)
arrived, expected, ready, validated = self.glob.checkStatus(self.validators)
arrived, expected, ready, validatedall, validated = self.glob.checkStatus(self.validators)
missingSamples = expected - arrived
missingVector = []
progressVector = []
@ -242,9 +251,9 @@ class Simulator:
self.validators[i].updateStats()
trafficStatsVector.append(trafficStats)
missingSamples, sampleProgress, nodeProgress, validatorProgress = self.glob.getProgress(self.validators)
self.logger.debug("step %d, arrived %0.02f %%, ready %0.02f %%, validated %0.02f %%"
% (steps, sampleProgress*100, nodeProgress*100, validatorProgress*100), extra=self.format)
missingSamples, sampleProgress, nodeProgress, validatorAllProgress, validatorProgress = self.glob.getProgress(self.validators)
self.logger.debug("step %d, arrived %0.02f %%, ready %0.02f %%, validatedall %0.02f %%, , validated %0.02f %%"
% (steps, sampleProgress*100, nodeProgress*100, validatorAllProgress*100, validatorProgress*100), extra=self.format)
cnS = "samples received"
cnN = "nodes ready"
@ -286,6 +295,9 @@ class Simulator:
steps += 1
progress = pd.DataFrame(progressVector)
if self.config.saveRCdist:
self.result.addMetric("rowDist", self.distR)
self.result.addMetric("columnDist", self.distC)
if self.config.saveProgress:
self.result.addMetric("progress", progress.to_dict(orient='list'))
self.result.populate(self.shape, self.config, missingVector)

42
DAS/validator.py
View File

@ -69,8 +69,13 @@ class Validator:
# random.seed(self.ID)
self.nodeClass = 1 if (self.ID <= shape.numberNodes * shape.class1ratio) else 2
self.vpn = self.shape.vpn1 if (self.nodeClass == 1) else self.shape.vpn2
self.rowIDs = rows if rows else unionOfSamples(range(self.shape.blockSize), self.shape.chi, self.vpn)
self.columnIDs = columns if columns else unionOfSamples(range(self.shape.blockSize), self.shape.chi, self.vpn)
self.vRowIDs = []
self.vColumnIDs = []
for i in range(self.vpn):
self.vRowIDs.append(set(rows[i*self.shape.chi:(i+1)*self.shape.chi]) if rows else set(random.sample(range(self.shape.blockSize), self.shape.chi)))
self.vColumnIDs.append(set(columns[i*self.shape.chi:(i+1)*self.shape.chi]) if columns else set(random.sample(range(self.shape.blockSize), self.shape.chi)))
self.rowIDs = set.union(*self.vRowIDs)
self.columnIDs = set.union(*self.vColumnIDs)
self.rowNeighbors = collections.defaultdict(dict)
self.columnNeighbors = collections.defaultdict(dict)
@ -469,16 +474,27 @@ class Validator:
def checkStatus(self):
"""It checks how many expected/arrived samples are for each assigned row/column."""
arrived = 0
expected = 0
for id in self.columnIDs:
line = self.getColumn(id)
arrived += line.count(1)
expected += len(line)
for id in self.rowIDs:
line = self.getRow(id)
arrived += line.count(1)
expected += len(line)
def checkStatus(columnIDs, rowIDs):
arrived = 0
expected = 0
for id in columnIDs:
line = self.getColumn(id)
arrived += line.count(1)
expected += len(line)
for id in rowIDs:
line = self.getRow(id)
arrived += line.count(1)
expected += len(line)
return arrived, expected
arrived, expected = checkStatus(self.columnIDs, self.rowIDs)
self.logger.debug("status: %d / %d", arrived, expected, extra=self.format)
return (arrived, expected)
validated = 0
for i in range(self.vpn):
a, e = checkStatus(self.vColumnIDs[i], self.vRowIDs[i])
if a == e:
validated+=1
return arrived, expected, validated

3
config_example.py
View File

@ -20,8 +20,9 @@ from DAS.shape import Shape
dumpXML = 1
# save progress vectors to XML
# save progress and row/column distribution vectors to XML
saveProgress = 1
saveRCdist = 1
visualization = 1
logLevel = logging.INFO