diff --git a/DAS/observer.py b/DAS/observer.py index 235ed60..beba4ad 100644 --- a/DAS/observer.py +++ b/DAS/observer.py @@ -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.""" diff --git a/DAS/simulator.py b/DAS/simulator.py index c61818d..9c03c76 100644 --- a/DAS/simulator.py +++ b/DAS/simulator.py @@ -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) diff --git a/DAS/validator.py b/DAS/validator.py index 79d9ea2..bd7769e 100644 --- a/DAS/validator.py +++ b/DAS/validator.py @@ -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 diff --git a/config_example.py b/config_example.py index 7263ca1..897041a 100644 --- a/config_example.py +++ b/config_example.py @@ -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