logos-storage/das-research
1
0
Fork 0
mirror of https://github.com/logos-storage/das-research.git synced 2026-01-08 16:13:11 +00:00
Code Issues Packages Projects Releases Wiki Activity

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

Browse Source 
Per validator progress
This commit is contained in:
Csaba Kiraly 2023-05-02 12:39:53 +02:00 committed by GitHub
commit 153eebc64c
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 76 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 arrived = 0
expected = 0 expected = 0
ready = 0 ready = 0
validatedall = 0
validated = 0 validated = 0
for val in validators: for val in validators:
if val.amIproposer == 0: if val.amIproposer == 0:
(a, e) = val.checkStatus() (a, e, v) = val.checkStatus()
arrived += a arrived += a
expected += e expected += e
if a == e: if a == e:
ready += 1 ready += 1
validated += val.vpn validatedall += val.vpn
return (arrived, expected, ready, validated) validated += v
return (arrived, expected, ready, validatedall, validated)
def getProgress(self, validators): def getProgress(self, validators):
"""Calculate current simulation progress with different metrics. """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 but counts a validator only if its support node's all validators see all interesting segments
TODO: add real per validator progress counter 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 missingSamples = expected - arrived
sampleProgress = arrived / expected sampleProgress = arrived / expected
nodeProgress = ready / (len(validators)-1) nodeProgress = ready / (len(validators)-1)
validatorCnt = sum([v.vpn for v in validators[1:]]) validatorCnt = sum([v.vpn for v in validators[1:]])
validatorAllProgress = validatedall / validatorCnt
validatorProgress = validated / validatorCnt validatorProgress = validated / validatorCnt
return missingSamples, sampleProgress, nodeProgress, validatorProgress return missingSamples, sampleProgress, nodeProgress, validatorAllProgress, validatorProgress
def getTrafficStats(self, validators): def getTrafficStats(self, validators):
"""Summary statistics of traffic measurements in a timestep.""" """Summary statistics of traffic measurements in a timestep."""

64
DAS/simulator.py
View File

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

42
DAS/validator.py
View File

@ -69,8 +69,13 @@ class Validator:
# random.seed(self.ID) # random.seed(self.ID)
self.nodeClass = 1 if (self.ID <= shape.numberNodes * shape.class1ratio) else 2 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.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.vRowIDs = []
self.columnIDs = columns if columns else unionOfSamples(range(self.shape.blockSize), self.shape.chi, self.vpn) 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.rowNeighbors = collections.defaultdict(dict)
self.columnNeighbors = collections.defaultdict(dict) self.columnNeighbors = collections.defaultdict(dict)
@ -469,16 +474,27 @@ class Validator:
def checkStatus(self): def checkStatus(self):
"""It checks how many expected/arrived samples are for each assigned row/column.""" """It checks how many expected/arrived samples are for each assigned row/column."""
arrived = 0
expected = 0 def checkStatus(columnIDs, rowIDs):
for id in self.columnIDs: arrived = 0
line = self.getColumn(id) expected = 0
arrived += line.count(1) for id in columnIDs:
expected += len(line) line = self.getColumn(id)
for id in self.rowIDs: arrived += line.count(1)
line = self.getRow(id) expected += len(line)
arrived += line.count(1) for id in rowIDs:
expected += len(line) 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) 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

2
smallConf.py
View File

@ -21,7 +21,7 @@ from DAS.shape import Shape
# Dump results into XML files # Dump results into XML files
dumpXML = 1 dumpXML = 1
# save progress vectors to XML # save progress and row/column distribution vectors to XML
saveProgress = 1 saveProgress = 1
# plot progress for each run to PNG # plot progress for each run to PNG