codex-storage/das-research
1
0
Fork 0
mirror of https://github.com/codex-storage/das-research.git synced 2025-02-24 08:48:29 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #53 from codex-storage/independent-dimensions

Browse Source 
Independent dimensions
This commit is contained in:
Csaba Kiraly 2024-02-19 11:30:28 +01:00 committed by GitHub
commit 2b69a3ae35
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
9 changed files with 166 additions and 124 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.gitignore vendored
View File

@ -1,7 +1,7 @@
*.swp *.swp
*.pyc *.pyc
results/* results/*
myenv myenv*/
doc/_build doc/_build
!results/plots.py !results/plots.py
Frontend/ Frontend/

57
DAS/block.py
View File

@ -7,10 +7,19 @@ from bitarray.util import zeros
class Block: class Block:
"""This class represents a block in the Ethereum blockchain.""" """This class represents a block in the Ethereum blockchain."""
def __init__(self, blockSize): def __init__(self, blockSizeR, blockSizeRK=0, blockSizeC=0, blockSizeCK=0):
"""Initialize the block with a data array of blocksize^2 zeros.""" """Initialize the block with a data array of blocksize^2 zeros.
self.blockSize = blockSize
self.data = zeros(self.blockSize*self.blockSize) BlockSizeR: row size
BlockSizeRK: original row size, before erasure coding to BlocksSizeR
BlockSizeC: column size (i.e. number of rows)
BlockSizeCK: original column size, before erasure coding to BlocksSizeR
"""
self.blockSizeR = blockSizeR
self.blockSizeRK = blockSizeRK if blockSizeRK else blockSizeR/2
self.blockSizeC = blockSizeC if blockSizeC else blockSizeR
self.blockSizeCK = blockSizeCK if blockSizeCK else blockSizeRK
self.data = zeros(self.blockSizeR*self.blockSizeC)
def fill(self): def fill(self):
"""It fills the block data with ones.""" """It fills the block data with ones."""
@ -22,62 +31,62 @@ class Block:
def getSegment(self, rowID, columnID): def getSegment(self, rowID, columnID):
"""Check whether a segment is included""" """Check whether a segment is included"""
return self.data[rowID*self.blockSize + columnID] return self.data[rowID*self.blockSizeR + columnID]
def setSegment(self, rowID, columnID, value = 1): def setSegment(self, rowID, columnID, value = 1):
"""Set value for a segment (default 1)""" """Set value for a segment (default 1)"""
self.data[rowID*self.blockSize + columnID] = value self.data[rowID*self.blockSizeR + columnID] = value
def getColumn(self, columnID): def getColumn(self, columnID):
"""It returns the block column corresponding to columnID.""" """It returns the block column corresponding to columnID."""
return self.data[columnID::self.blockSize] return self.data[columnID::self.blockSizeR]
def mergeColumn(self, columnID, column): def mergeColumn(self, columnID, column):
"""It merges (OR) the existing column with the received one.""" """It merges (OR) the existing column with the received one."""
self.data[columnID::self.blockSize] |= column self.data[columnID::self.blockSizeR] |= column
def repairColumn(self, id): def repairColumn(self, id):
"""It repairs the entire column if it has at least blockSize/2 ones. """It repairs the entire column if it has at least blockSizeCK ones.
Returns: list of repaired segments Returns: list of repaired segments
""" """
line = self.data[id::self.blockSize] line = self.data[id::self.blockSizeR]
success = line.count(1) success = line.count(1)
if success >= self.blockSize/2: if success >= self.blockSizeCK:
ret = ~line ret = ~line
self.data[id::self.blockSize] = 1 self.data[id::self.blockSizeR] = 1
else: else:
ret = zeros(self.blockSize) ret = zeros(self.blockSizeC)
return ret return ret
def getRow(self, rowID): def getRow(self, rowID):
"""It returns the block row corresponding to rowID.""" """It returns the block row corresponding to rowID."""
return self.data[rowID*self.blockSize:(rowID+1)*self.blockSize] return self.data[rowID*self.blockSizeR:(rowID+1)*self.blockSizeR]
def mergeRow(self, rowID, row): def mergeRow(self, rowID, row):
"""It merges (OR) the existing row with the received one.""" """It merges (OR) the existing row with the received one."""
self.data[rowID*self.blockSize:(rowID+1)*self.blockSize] |= row self.data[rowID*self.blockSizeR:(rowID+1)*self.blockSizeR] |= row
def repairRow(self, id): def repairRow(self, id):
"""It repairs the entire row if it has at least blockSize/2 ones. """It repairs the entire row if it has at least blockSizeRK ones.
Returns: list of repaired segments. Returns: list of repaired segments.
""" """
line = self.data[id*self.blockSize:(id+1)*self.blockSize] line = self.data[id*self.blockSizeR:(id+1)*self.blockSizeR]
success = line.count(1) success = line.count(1)
if success >= self.blockSize/2: if success >= self.blockSizeRK:
ret = ~line ret = ~line
self.data[id*self.blockSize:(id+1)*self.blockSize] = 1 self.data[id*self.blockSizeR:(id+1)*self.blockSizeR] = 1
else: else:
ret = zeros(self.blockSize) ret = zeros(self.blockSizeR)
return ret return ret
def print(self): def print(self):
"""It prints the block in the terminal (outside of the logger rules)).""" """It prints the block in the terminal (outside of the logger rules))."""
dash = "-" * (self.blockSize+2) dash = "-" * (self.blockSizeR+2)
print(dash) print(dash)
for i in range(self.blockSize): for i in range(self.blockSizeC):
line = "|" line = "|"
for j in range(self.blockSize): for j in range(self.blockSizeR):
line += "%i" % self.data[(i*self.blockSize)+j] line += "%i" % self.data[(i*self.blockSizeR)+j]
print(line+"|") print(line+"|")
print(dash) print(dash)

13
DAS/observer.py
View File

@ -11,10 +11,11 @@ class Observer:
self.config = config self.config = config
self.format = {"entity": "Observer"} self.format = {"entity": "Observer"}
self.logger = logger self.logger = logger
self.block = [0] * self.config.blockSize * self.config.blockSize self.block = [0] * self.config.blockSizeR * self.config.blockSizeC
self.rows = [0] * self.config.blockSize self.rows = [0] * self.config.blockSizeC
self.columns = [0] * self.config.blockSize self.columns = [0] * self.config.blockSizeR
self.broadcasted = Block(self.config.blockSize) self.broadcasted = Block(self.config.blockSizeR, self.config.blockSizeRK,
self.config.blockSizeC, self.config.blockSizeCK)
def checkRowsColumns(self, validators): def checkRowsColumns(self, validators):
@ -26,7 +27,7 @@ class Observer:
for c in val.columnIDs: for c in val.columnIDs:
self.columns[c] += 1 self.columns[c] += 1
for i in range(self.config.blockSize): for i in range(self.config.blockSizeC):
self.logger.debug("Row/Column %d have %d and %d validators assigned." % (i, self.rows[i], self.columns[i]), extra=self.format) self.logger.debug("Row/Column %d have %d and %d validators assigned." % (i, self.rows[i], self.columns[i]), extra=self.format)
if self.rows[i] == 0 or self.columns[i] == 0: if self.rows[i] == 0 or self.columns[i] == 0:
self.logger.warning("There is a row/column that has not been assigned", extra=self.format) self.logger.warning("There is a row/column that has not been assigned", extra=self.format)
@ -34,7 +35,7 @@ class Observer:
def checkBroadcasted(self): def checkBroadcasted(self):
"""It checks how many broadcasted samples are still missing in the network.""" """It checks how many broadcasted samples are still missing in the network."""
zeros = 0 zeros = 0
for i in range(self.blockSize * self.blockSize): for i in range(self.blockSizeR * self.blockSizeC):
if self.broadcasted.data[i] == 0: if self.broadcasted.data[i] == 0:
zeros += 1 zeros += 1
if zeros > 0: if zeros > 0:

19
DAS/shape.py
View File

@ -3,16 +3,21 @@
class Shape: class Shape:
"""This class represents a set of parameters for a specific simulation.""" """This class represents a set of parameters for a specific simulation."""
def __init__(self, blockSize, numberNodes, failureModel, failureRate, class1ratio, chi, vpn1, vpn2, netDegree, bwUplinkProd, bwUplink1, bwUplink2, run): def __init__(self, blockSizeR, blockSizeRK, blockSizeC, blockSizeCK,
numberNodes, failureModel, failureRate, class1ratio, chiR, chiC, vpn1, vpn2, netDegree, bwUplinkProd, bwUplink1, bwUplink2, run):
"""Initializes the shape with the parameters passed in argument.""" """Initializes the shape with the parameters passed in argument."""
self.run = run self.run = run
self.numberNodes = numberNodes self.numberNodes = numberNodes
self.blockSize = blockSize self.blockSizeR = blockSizeR
self.blockSizeRK = blockSizeRK
self.blockSizeC = blockSizeC
self.blockSizeCK = blockSizeCK
self.failureModel = failureModel self.failureModel = failureModel
self.failureRate = failureRate self.failureRate = failureRate
self.netDegree = netDegree self.netDegree = netDegree
self.class1ratio = class1ratio self.class1ratio = class1ratio
self.chi = chi self.chiR = chiR
self.chiC = chiC
self.vpn1 = vpn1 self.vpn1 = vpn1
self.vpn2 = vpn2 self.vpn2 = vpn2
self.bwUplinkProd = bwUplinkProd self.bwUplinkProd = bwUplinkProd
@ -23,12 +28,16 @@ class Shape:
def __repr__(self): def __repr__(self):
"""Returns a printable representation of the shape""" """Returns a printable representation of the shape"""
shastr = "" shastr = ""
shastr += "bs-"+str(self.blockSize) shastr += "bsrn-"+str(self.blockSizeR)
shastr += "-bsrk-"+str(self.blockSizeRK)
shastr += "-bscn-"+str(self.blockSizeC)
shastr += "-bsck-"+str(self.blockSizeCK)
shastr += "-nn-"+str(self.numberNodes) shastr += "-nn-"+str(self.numberNodes)
shastr += "-fm-"+str(self.failureModel) shastr += "-fm-"+str(self.failureModel)
shastr += "-fr-"+str(self.failureRate) shastr += "-fr-"+str(self.failureRate)
shastr += "-c1r-"+str(self.class1ratio) shastr += "-c1r-"+str(self.class1ratio)
shastr += "-chi-"+str(self.chi) shastr += "-chir-"+str(self.chiR)
shastr += "-chic-"+str(self.chiC)
shastr += "-vpn1-"+str(self.vpn1) shastr += "-vpn1-"+str(self.vpn1)
shastr += "-vpn2-"+str(self.vpn2) shastr += "-vpn2-"+str(self.vpn2)
shastr += "-bwupprod-"+str(self.bwUplinkProd) shastr += "-bwupprod-"+str(self.bwUplinkProd)

51
DAS/simulator.py
View File

@ -41,7 +41,7 @@ class Simulator:
# pushed out by the proposer. # pushed out by the proposer.
# 1: the data is sent out exactly once on rows and once on columns (2 copies in total) # 1: the data is sent out exactly once on rows and once on columns (2 copies in total)
# self.shape.netDegree: default behavior similar (but not same) to previous code # self.shape.netDegree: default behavior similar (but not same) to previous code
self.proposerPublishTo = self.shape.netDegree self.proposerPublishTo = self.shape.netDegree # TODO: make this an external parameter
def initValidators(self): def initValidators(self):
"""It initializes all the validators in the network.""" """It initializes all the validators in the network."""
@ -54,17 +54,20 @@ class Simulator:
lightVal = lightNodes * self.shape.vpn1 lightVal = lightNodes * self.shape.vpn1
heavyVal = heavyNodes * self.shape.vpn2 heavyVal = heavyNodes * self.shape.vpn2
totalValidators = lightVal + heavyVal totalValidators = lightVal + heavyVal
totalRows = totalValidators * self.shape.chi totalRows = totalValidators * self.shape.chiR
rows = list(range(self.shape.blockSize)) * (int(totalRows/self.shape.blockSize)+1) totalColumns = totalValidators * self.shape.chiC
columns = list(range(self.shape.blockSize)) * (int(totalRows/self.shape.blockSize)+1) rows = list(range(self.shape.blockSizeC)) * (int(totalRows/self.shape.blockSizeC)+1)
columns = list(range(self.shape.blockSizeR)) * (int(totalColumns/self.shape.blockSizeR)+1)
rows = rows[0:totalRows] rows = rows[0:totalRows]
columns = columns[0:totalRows] columns = columns[0:totalRows]
random.shuffle(rows) random.shuffle(rows)
random.shuffle(columns) random.shuffle(columns)
offset = lightVal*self.shape.chi offsetR = lightVal*self.shape.chiR
offsetC = lightVal*self.shape.chiC
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 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("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("Shuffling a total of %d rows to be assigned (X=%d)" % (len(rows), self.shape.chiR), extra=self.format)
self.logger.debug("Shuffling a total of %d columns to be assigned (X=%d)" % (len(columns), self.shape.chiC), 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)
@ -73,14 +76,18 @@ class Simulator:
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(lightVal/self.shape.vpn1): # First start with the light nodes if i < int(lightVal/self.shape.vpn1): # First start with the light nodes
start = i *self.shape.chi*self.shape.vpn1 startR = i *self.shape.chiR*self.shape.vpn1
end = (i+1)*self.shape.chi*self.shape.vpn1 endR = (i+1)*self.shape.chiR*self.shape.vpn1
startC = i *self.shape.chiC*self.shape.vpn1
endC = (i+1)*self.shape.chiC*self.shape.vpn1
else: else:
j = i - int(lightVal/self.shape.vpn1) j = i - int(lightVal/self.shape.vpn1)
start = offset+( j *self.shape.chi*self.shape.vpn2) startR = offsetR+( j *self.shape.chiR*self.shape.vpn2)
end = offset+((j+1)*self.shape.chi*self.shape.vpn2) endR = offsetR+((j+1)*self.shape.chiR*self.shape.vpn2)
r = rows[start:end] startC = offsetC+( j *self.shape.chiC*self.shape.vpn2)
c = columns[start:end] endC = offsetC+((j+1)*self.shape.chiC*self.shape.vpn2)
r = rows[startR:endR]
c = columns[startC:endC]
val = Validator(i, int(not i!=0), self.logger, self.shape, self.config, r, c) val = Validator(i, int(not i!=0), self.logger, self.shape, self.config, r, c)
self.logger.debug("Node %d has 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("Node %d has 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)
@ -105,8 +112,8 @@ class Simulator:
def initNetwork(self): def initNetwork(self):
"""It initializes the simulated network.""" """It initializes the simulated network."""
rowChannels = [[] for i in range(self.shape.blockSize)] rowChannels = [[] for i in range(self.shape.blockSizeC)]
columnChannels = [[] for i in range(self.shape.blockSize)] columnChannels = [[] for i in range(self.shape.blockSizeR)]
for v in self.validators: for v in self.validators:
if not (self.proposerPublishOnly and v.amIproposer): if not (self.proposerPublishOnly and v.amIproposer):
for id in v.rowIDs: for id in v.rowIDs:
@ -122,7 +129,7 @@ class Simulator:
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 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) 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.blockSizeC):
# If the number of nodes in a channel is smaller or equal to the # If the number of nodes in a channel is smaller or equal to the
# requested degree, a fully connected graph is used. For n>d, a random # requested degree, a fully connected graph is used. For n>d, a random
@ -140,8 +147,10 @@ class Simulator:
for u, v in G.edges: for u, v in G.edges:
val1=rowChannels[id][u] val1=rowChannels[id][u]
val2=rowChannels[id][v] val2=rowChannels[id][v]
val1.rowNeighbors[id].update({val2.ID : Neighbor(val2, 0, self.shape.blockSize)}) val1.rowNeighbors[id].update({val2.ID : Neighbor(val2, 0, self.shape.blockSizeR)})
val2.rowNeighbors[id].update({val1.ID : Neighbor(val1, 0, self.shape.blockSize)}) val2.rowNeighbors[id].update({val1.ID : Neighbor(val1, 0, self.shape.blockSizeR)})
for id in range(self.shape.blockSizeR):
if not columnChannels[id]: if not columnChannels[id]:
self.logger.error("No nodes for column %d !" % id, extra=self.format) self.logger.error("No nodes for column %d !" % id, extra=self.format)
@ -156,8 +165,8 @@ class Simulator:
for u, v in G.edges: for u, v in G.edges:
val1=columnChannels[id][u] val1=columnChannels[id][u]
val2=columnChannels[id][v] val2=columnChannels[id][v]
val1.columnNeighbors[id].update({val2.ID : Neighbor(val2, 1, self.shape.blockSize)}) val1.columnNeighbors[id].update({val2.ID : Neighbor(val2, 1, self.shape.blockSizeC)})
val2.columnNeighbors[id].update({val1.ID : Neighbor(val1, 1, self.shape.blockSize)}) val2.columnNeighbors[id].update({val1.ID : Neighbor(val1, 1, self.shape.blockSizeC)})
for v in self.validators: for v in self.validators:
if (self.proposerPublishOnly and v.amIproposer): if (self.proposerPublishOnly and v.amIproposer):
@ -165,12 +174,12 @@ class Simulator:
count = min(self.proposerPublishTo, len(rowChannels[id])) count = min(self.proposerPublishTo, len(rowChannels[id]))
publishTo = random.sample(rowChannels[id], count) publishTo = random.sample(rowChannels[id], count)
for vi in publishTo: for vi in publishTo:
v.rowNeighbors[id].update({vi.ID : Neighbor(vi, 0, self.shape.blockSize)}) v.rowNeighbors[id].update({vi.ID : Neighbor(vi, 0, self.shape.blockSizeR)})
for id in v.columnIDs: for id in v.columnIDs:
count = min(self.proposerPublishTo, len(columnChannels[id])) count = min(self.proposerPublishTo, len(columnChannels[id]))
publishTo = random.sample(columnChannels[id], count) publishTo = random.sample(columnChannels[id], count)
for vi in publishTo: for vi in publishTo:
v.columnNeighbors[id].update({vi.ID : Neighbor(vi, 1, self.shape.blockSize)}) v.columnNeighbors[id].update({vi.ID : Neighbor(vi, 1, self.shape.blockSizeC)})
if self.logger.isEnabledFor(logging.DEBUG): if self.logger.isEnabledFor(logging.DEBUG):
for i in range(0, self.shape.numberNodes): for i in range(0, self.shape.numberNodes):

89
DAS/validator.py
View File

@ -42,28 +42,30 @@ class Validator:
"""It initializes the validator with the logger shape and rows/columns. """It initializes the validator with the logger shape and rows/columns.
If rows/columns are specified these are observed, otherwise (default) If rows/columns are specified these are observed, otherwise (default)
chi rows and columns are selected randomly. chiR rows and chiC columns are selected randomly.
""" """
self.shape = shape self.shape = shape
FORMAT = "%(levelname)s : %(entity)s : %(message)s" FORMAT = "%(levelname)s : %(entity)s : %(message)s"
self.ID = ID self.ID = ID
self.format = {"entity": "Val "+str(self.ID)} self.format = {"entity": "Val "+str(self.ID)}
self.block = Block(self.shape.blockSize) self.block = Block(self.shape.blockSizeR, self.shape.blockSizeRK, self.shape.blockSizeC, self.shape.blockSizeCK)
self.receivedBlock = Block(self.shape.blockSize) self.receivedBlock = Block(self.shape.blockSizeR, self.shape.blockSizeRK, self.shape.blockSizeC, self.shape.blockSizeCK)
self.receivedQueue = deque() self.receivedQueue = deque()
self.sendQueue = deque() self.sendQueue = deque()
self.amIproposer = amIproposer self.amIproposer = amIproposer
self.logger = logger self.logger = logger
if self.shape.chi < 1: if self.shape.chiR < 1 and self.shape.chiC < 1:
self.logger.error("Chi has to be greater than 0", extra=self.format) self.logger.error("Chi has to be greater than 0", extra=self.format)
elif self.shape.chi > self.shape.blockSize: elif self.shape.chiC > self.shape.blockSizeR:
self.logger.error("Chi has to be smaller than %d" % self.shape.blockSize, extra=self.format) self.logger.error("ChiC has to be smaller than %d" % self.shape.blockSizeR, extra=self.format)
elif self.shape.chiR > self.shape.blockSizeC:
self.logger.error("ChiR has to be smaller than %d" % self.shape.blockSizeC, extra=self.format)
else: else:
if amIproposer: if amIproposer:
self.nodeClass = 0 self.nodeClass = 0
self.rowIDs = range(shape.blockSize) self.rowIDs = range(shape.blockSizeC)
self.columnIDs = range(shape.blockSize) self.columnIDs = range(shape.blockSizeR)
else: else:
#if shape.deterministic: #if shape.deterministic:
# random.seed(self.ID) # random.seed(self.ID)
@ -72,8 +74,8 @@ class Validator:
self.vRowIDs = [] self.vRowIDs = []
self.vColumnIDs = [] self.vColumnIDs = []
for i in range(self.vpn): 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.vRowIDs.append(set(rows[i*self.shape.chiR:(i+1)*self.shape.chiR]) if rows else set(random.sample(range(self.shape.blockSizeC), self.shape.chiR)))
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.vColumnIDs.append(set(columns[i*self.shape.chiC:(i+1)*self.shape.chiC]) if columns else set(random.sample(range(self.shape.blockSizeR), self.shape.chiC)))
self.rowIDs = set.union(*self.vRowIDs) self.rowIDs = set.union(*self.vRowIDs)
self.columnIDs = set.union(*self.vColumnIDs) self.columnIDs = set.union(*self.vColumnIDs)
self.rowNeighbors = collections.defaultdict(dict) self.rowNeighbors = collections.defaultdict(dict)
@ -99,7 +101,8 @@ class Validator:
self.bwUplink *= 1e3 / 8 * config.stepDuration / config.segmentSize self.bwUplink *= 1e3 / 8 * config.stepDuration / config.segmentSize
self.repairOnTheFly = True self.repairOnTheFly = True
self.sendLineUntil = (self.shape.blockSize + 1) // 2 # stop sending on a p2p link if at least this amount of samples passed self.sendLineUntilR = self.shape.blockSizeRK # stop sending on a p2p link if at least this amount of samples passed
self.sendLineUntilC = self.shape.blockSizeCK # stop sending on a p2p link if at least this amount of samples passed
self.perNeighborQueue = True # queue incoming messages to outgoing connections on arrival (as typical GossipSub impl) self.perNeighborQueue = True # queue incoming messages to outgoing connections on arrival (as typical GossipSub impl)
self.shuffleQueues = True # shuffle the order of picking from active queues of a sender node self.shuffleQueues = True # shuffle the order of picking from active queues of a sender node
self.perNodeQueue = False # keep a global queue of incoming messages for later sequential dispatch self.perNodeQueue = False # keep a global queue of incoming messages for later sequential dispatch
@ -124,57 +127,53 @@ class Validator:
else: else:
self.logger.debug("Creating block...", extra=self.format) self.logger.debug("Creating block...", extra=self.format)
if self.shape.failureModel == "random": if self.shape.failureModel == "random":
order = [i for i in range(self.shape.blockSize * self.shape.blockSize)] order = [i for i in range(self.shape.blockSizeR * self.shape.blockSizeC)]
order = random.sample(order, int((1 - self.shape.failureRate/100) * len(order))) order = random.sample(order, int((1 - self.shape.failureRate/100) * len(order)))
for i in order: for i in order:
self.block.data[i] = 1 self.block.data[i] = 1
elif self.shape.failureModel == "sequential": elif self.shape.failureModel == "sequential":
order = [i for i in range(self.shape.blockSize * self.shape.blockSize)] order = [i for i in range(self.shape.blockSizeR * self.shape.blockSizeC)]
order = order[:int((1 - self.shape.failureRate/100) * len(order))] order = order[:int((1 - self.shape.failureRate/100) * len(order))]
for i in order: for i in order:
self.block.data[i] = 1 self.block.data[i] = 1
elif self.shape.failureModel == "MEP": # Minimal size non-recoverable Erasure Pattern elif self.shape.failureModel == "MEP": # Minimal size non-recoverable Erasure Pattern
for r in range(self.shape.blockSize): for r in range(self.shape.blockSizeR):
for c in range(self.shape.blockSize): for c in range(self.shape.blockSizeC):
k = self.shape.blockSize/2 if r > self.shape.blockSizeRK or c > self.shape.blockSizeCK:
if r > k or c > k:
self.block.setSegment(r,c) self.block.setSegment(r,c)
elif self.shape.failureModel == "MEP+1": # MEP +1 segment to make it recoverable elif self.shape.failureModel == "MEP+1": # MEP +1 segment to make it recoverable
for r in range(self.shape.blockSize): for r in range(self.shape.blockSizeR):
for c in range(self.shape.blockSize): for c in range(self.shape.blockSizeC):
k = self.shape.blockSize/2 if r > self.shape.blockSizeRK or c > self.shape.blockSizeCK:
if r > k or c > k:
self.block.setSegment(r,c) self.block.setSegment(r,c)
self.block.setSegment(0, 0) self.block.setSegment(0, 0)
elif self.shape.failureModel == "DEP": elif self.shape.failureModel == "DEP":
for r in range(self.shape.blockSize): assert(self.shape.blockSizeR == self.shape.blockSizeC and self.shape.blockSizeRK == self.shape.blockSizeCK)
for c in range(self.shape.blockSize): for r in range(self.shape.blockSizeR):
k = self.shape.blockSize/2 for c in range(self.shape.blockSizeC):
if (r+c) % self.shape.blockSize > k: if (r+c) % self.shape.blockSizeR > self.shape.blockSizeRK:
self.block.setSegment(r,c) self.block.setSegment(r,c)
elif self.shape.failureModel == "DEP+1": elif self.shape.failureModel == "DEP+1":
for r in range(self.shape.blockSize): assert(self.shape.blockSizeR == self.shape.blockSizeC and self.shape.blockSizeRK == self.shape.blockSizeCK)
for c in range(self.shape.blockSize): for r in range(self.shape.blockSizeR):
k = self.shape.blockSize/2 for c in range(self.shape.blockSizeC):
if (r+c) % self.shape.blockSize > k: if (r+c) % self.shape.blockSizeR > self.shape.blockSizeRK:
self.block.setSegment(r,c) self.block.setSegment(r,c)
self.block.setSegment(0, 0) self.block.setSegment(0, 0)
elif self.shape.failureModel == "MREP": # Minimum size Recoverable Erasure Pattern elif self.shape.failureModel == "MREP": # Minimum size Recoverable Erasure Pattern
for r in range(self.shape.blockSize): for r in range(self.shape.blockSizeR):
for c in range(self.shape.blockSize): for c in range(self.shape.blockSizeC):
k = self.shape.blockSize/2 if r < self.shape.blockSizeRK or c < self.shape.blockSizeCK:
if r < k and c < k:
self.block.setSegment(r,c) self.block.setSegment(r,c)
elif self.shape.failureModel == "MREP-1": # make MREP non-recoverable elif self.shape.failureModel == "MREP-1": # make MREP non-recoverable
for r in range(self.shape.blockSize): for r in range(self.shape.blockSizeR):
for c in range(self.shape.blockSize): for c in range(self.shape.blockSizeC):
k = self.shape.blockSize/2 if r < self.shape.blockSizeRK or c < self.shape.blockSizeCK:
if r < k and c < k:
self.block.setSegment(r,c) self.block.setSegment(r,c)
self.block.setSegment(0, 0, 0) self.block.setSegment(0, 0, 0)
nbFailures = self.block.data.count(0) nbFailures = self.block.data.count(0)
measuredFailureRate = nbFailures * 100 / (self.shape.blockSize * self.shape.blockSize) measuredFailureRate = nbFailures * 100 / (self.shape.blockSizeR * self.shape.blockSizeC)
self.logger.debug("Number of failures: %d (%0.02f %%)", nbFailures, measuredFailureRate, extra=self.format) self.logger.debug("Number of failures: %d (%0.02f %%)", nbFailures, measuredFailureRate, extra=self.format)
def getColumn(self, index): def getColumn(self, index):
@ -251,7 +250,7 @@ class Validator:
def checkSegmentToNeigh(self, rID, cID, neigh): def checkSegmentToNeigh(self, rID, cID, neigh):
"""Check if a segment should be sent to a neighbor.""" """Check if a segment should be sent to a neighbor."""
if (neigh.sent | neigh.received).count(1) >= self.sendLineUntil: if (neigh.sent | neigh.received).count(1) >= (self.sendLineUntilC if neigh.dim else self.sendLineUntilR):
return False # sent enough, other side can restore return False # sent enough, other side can restore
i = rID if neigh.dim else cID i = rID if neigh.dim else cID
if not neigh.sent[i] and not neigh.received[i] : if not neigh.sent[i] and not neigh.received[i] :
@ -348,10 +347,10 @@ class Validator:
segmentsToSend = [] segmentsToSend = []
for rID, neighs in self.rowNeighbors.items(): for rID, neighs in self.rowNeighbors.items():
line = self.getRow(rID) line = self.getRow(rID)
needed = zeros(self.shape.blockSize) needed = zeros(self.shape.blockSizeR)
for neigh in neighs.values(): for neigh in neighs.values():
sentOrReceived = neigh.received | neigh.sent sentOrReceived = neigh.received | neigh.sent
if sentOrReceived.count(1) < self.sendLineUntil: if sentOrReceived.count(1) < self.sendLineUntilR:
needed |= ~sentOrReceived needed |= ~sentOrReceived
needed &= line needed &= line
if (needed).any(): if (needed).any():
@ -361,10 +360,10 @@ class Validator:
for cID, neighs in self.columnNeighbors.items(): for cID, neighs in self.columnNeighbors.items():
line = self.getColumn(cID) line = self.getColumn(cID)
needed = zeros(self.shape.blockSize) needed = zeros(self.shape.blockSizeC)
for neigh in neighs.values(): for neigh in neighs.values():
sentOrReceived = neigh.received | neigh.sent sentOrReceived = neigh.received | neigh.sent
if sentOrReceived.count(1) < self.sendLineUntil: if sentOrReceived.count(1) < self.sendLineUntilC:
needed |= ~sentOrReceived needed |= ~sentOrReceived
needed &= line needed &= line
if (needed).any(): if (needed).any():
@ -423,7 +422,7 @@ class Validator:
while t: while t:
if self.rowIDs: if self.rowIDs:
rID = random.choice(self.rowIDs) rID = random.choice(self.rowIDs)
cID = random.randrange(0, self.shape.blockSize) cID = random.randrange(0, self.shape.blockSizeR)
if self.block.getSegment(rID, cID) : if self.block.getSegment(rID, cID) :
neigh = random.choice(list(self.rowNeighbors[rID].values())) neigh = random.choice(list(self.rowNeighbors[rID].values()))
if self.checkSegmentToNeigh(rID, cID, neigh): if self.checkSegmentToNeigh(rID, cID, neigh):
@ -431,7 +430,7 @@ class Validator:
t = tries t = tries
if self.columnIDs: if self.columnIDs:
cID = random.choice(self.columnIDs) cID = random.choice(self.columnIDs)
rID = random.randrange(0, self.shape.blockSize) rID = random.randrange(0, self.shape.blockSizeC)
if self.block.getSegment(rID, cID) : if self.block.getSegment(rID, cID) :
neigh = random.choice(list(self.columnNeighbors[cID].values())) neigh = random.choice(list(self.columnNeighbors[cID].values()))
if self.checkSegmentToNeigh(rID, cID, neigh): if self.checkSegmentToNeigh(rID, cID, neigh):

4
DAS/visualizer.py
View File

@ -32,12 +32,12 @@ class Visualizer:
tree = ET.parse(os.path.join(self.folderPath, filename)) tree = ET.parse(os.path.join(self.folderPath, filename))
root = tree.getroot() root = tree.getroot()
run = int(root.find('run').text) run = int(root.find('run').text)
blockSize = int(root.find('blockSize').text) blockSize = int(root.find('blockSizeR').text) # TODO: maybe we want both dimensions
failureRate = int(root.find('failureRate').text) failureRate = int(root.find('failureRate').text)
numberNodes = int(root.find('numberNodes').text) numberNodes = int(root.find('numberNodes').text)
class1ratio = float(root.find('class1ratio').text) class1ratio = float(root.find('class1ratio').text)
netDegree = int(root.find('netDegree').text) netDegree = int(root.find('netDegree').text)
chi = int(root.find('chi').text) chi = int(root.find('chiR').text) # TODO: maybe we want both dimensions
vpn1 = int(root.find('vpn1').text) vpn1 = int(root.find('vpn1').text)
vpn2 = int(root.find('vpn2').text) vpn2 = int(root.find('vpn2').text)
bwUplinkProd = int(root.find('bwUplinkProd').text) bwUplinkProd = int(root.find('bwUplinkProd').text)

50
DAS/visualizor.py
View File

@ -1,6 +1,7 @@
#!/bin/python3 #!/bin/python3
import matplotlib.pyplot as plt import matplotlib.pyplot as plt
import numpy as np
import os import os
def plotData(conf): def plotData(conf):
@ -35,6 +36,13 @@ class Visualizor:
self.results = results self.results = results
os.makedirs("results/"+self.execID+"/plots", exist_ok=True) os.makedirs("results/"+self.execID+"/plots", exist_ok=True)
def __get_attrbs__(self, result):
text = str(result.shape).split("-")
d = dict()
for i in range(0, len(text), 2):
d[text[i]] = text[i + 1]
return d
def plotAll(self): def plotAll(self):
"""Plot all the important elements of each result""" """Plot all the important elements of each result"""
for result in self.results: for result in self.results:
@ -48,9 +56,9 @@ class Visualizor:
def plotMissingSamples(self, result): def plotMissingSamples(self, result):
"""Plots the missing samples in the network""" """Plots the missing samples in the network"""
conf = {} conf = {}
text = str(result.shape).split("-") attrbs = self.__get_attrbs__(result)
conf["textBox"] = "Block Size: "+text[1]+"\nNumber of nodes: "+text[3]\ conf["textBox"] = "Block Size R: "+attrbs['bsrn']+"\nBlock Size C: "+attrbs['bscn']\
+"\nFailure rate: "+text[7]+" \nNetwork degree: "+text[23]+"\nX: "+text[11]+" rows/columns" +"\nNumber of nodes: "+attrbs['nn']+"\nFailure rate: "+attrbs['fr']+" \nNetwork degree: "+attrbs['nd']
conf["title"] = "Missing Samples" conf["title"] = "Missing Samples"
conf["type"] = "plot" conf["type"] = "plot"
conf["legLoc"] = 1 conf["legLoc"] = 1
@ -76,9 +84,9 @@ class Visualizor:
vector2 = result.metrics["progress"]["validators ready"] vector2 = result.metrics["progress"]["validators ready"]
vector3 = result.metrics["progress"]["samples received"] vector3 = result.metrics["progress"]["samples received"]
conf = {} conf = {}
text = str(result.shape).split("-") attrbs = self.__get_attrbs__(result)
conf["textBox"] = "Block Size: "+text[1]+"\nNumber of nodes: "+text[3]\ conf["textBox"] = "Block Size R: "+attrbs['bsrn']+"\nBlock Size C: "+attrbs['bscn']\
+"\nFailure rate: "+text[7]+" \nNetwork degree: "+text[23]+"\nX: "+text[11]+" rows/columns" +"\nNumber of nodes: "+attrbs['nn']+"\nFailure rate: "+attrbs['fr']+" \nNetwork degree: "+attrbs['nd']
conf["title"] = "Nodes/validators ready" conf["title"] = "Nodes/validators ready"
conf["type"] = "plot" conf["type"] = "plot"
conf["legLoc"] = 2 conf["legLoc"] = 2
@ -108,9 +116,9 @@ class Visualizor:
vector2[i] = (vector2[i] * 8 * (1000/self.config.stepDuration) * self.config.segmentSize) / 1000000 vector2[i] = (vector2[i] * 8 * (1000/self.config.stepDuration) * self.config.segmentSize) / 1000000
vector3[i] = (vector3[i] * 8 * (1000/self.config.stepDuration) * self.config.segmentSize) / 1000000 vector3[i] = (vector3[i] * 8 * (1000/self.config.stepDuration) * self.config.segmentSize) / 1000000
conf = {} conf = {}
text = str(result.shape).split("-") attrbs = self.__get_attrbs__(result)
conf["textBox"] = "Block Size: "+text[1]+"\nNumber of nodes: "+text[3]\ conf["textBox"] = "Block Size R: "+attrbs['bsrn']+"\nBlock Size C: "+attrbs['bscn']\
+"\nFailure rate: "+text[7]+" \nNetwork degree: "+text[23]+"\nX: "+text[11]+" rows/columns" +"\nNumber of nodes: "+attrbs['nn']+"\nFailure rate: "+attrbs['fr']+" \nNetwork degree: "+attrbs['nd']
conf["title"] = "Sent data" conf["title"] = "Sent data"
conf["type"] = "plot" conf["type"] = "plot"
conf["legLoc"] = 2 conf["legLoc"] = 2
@ -138,9 +146,9 @@ class Visualizor:
vector1[i] = (vector1[i] * 8 * (1000/self.config.stepDuration) * self.config.segmentSize) / 1000000 vector1[i] = (vector1[i] * 8 * (1000/self.config.stepDuration) * self.config.segmentSize) / 1000000
vector2[i] = (vector2[i] * 8 * (1000/self.config.stepDuration) * self.config.segmentSize) / 1000000 vector2[i] = (vector2[i] * 8 * (1000/self.config.stepDuration) * self.config.segmentSize) / 1000000
conf = {} conf = {}
text = str(result.shape).split("-") attrbs = self.__get_attrbs__(result)
conf["textBox"] = "Block Size: "+text[1]+"\nNumber of nodes: "+text[3]\ conf["textBox"] = "Block Size R: "+attrbs['bsrn']+"\nBlock Size C: "+attrbs['bscn']\
+"\nFailure rate: "+text[7]+" \nNetwork degree: "+text[23]+"\nX: "+text[11]+" rows/columns" +"\nNumber of nodes: "+attrbs['nn']+"\nFailure rate: "+attrbs['fr']+" \nNetwork degree: "+attrbs['nd']
conf["title"] = "Received data" conf["title"] = "Received data"
conf["type"] = "plot" conf["type"] = "plot"
conf["legLoc"] = 2 conf["legLoc"] = 2
@ -168,9 +176,9 @@ class Visualizor:
vector1[i] = (vector1[i] * 8 * (1000/self.config.stepDuration) * self.config.segmentSize) / 1000000 vector1[i] = (vector1[i] * 8 * (1000/self.config.stepDuration) * self.config.segmentSize) / 1000000
vector2[i] = (vector2[i] * 8 * (1000/self.config.stepDuration) * self.config.segmentSize) / 1000000 vector2[i] = (vector2[i] * 8 * (1000/self.config.stepDuration) * self.config.segmentSize) / 1000000
conf = {} conf = {}
text = str(result.shape).split("-") attrbs = self.__get_attrbs__(result)
conf["textBox"] = "Block Size: "+text[1]+"\nNumber of nodes: "+text[3]\ conf["textBox"] = "Block Size R: "+attrbs['bsrn']+"\nBlock Size C: "+attrbs['bscn']\
+"\nFailure rate: "+text[7]+" \nNetwork degree: "+text[23]+"\nX: "+text[11]+" rows/columns" +"\nNumber of nodes: "+attrbs['nn']+"\nFailure rate: "+attrbs['fr']+" \nNetwork degree: "+attrbs['nd']
conf["title"] = "Duplicated data" conf["title"] = "Duplicated data"
conf["type"] = "plot" conf["type"] = "plot"
conf["legLoc"] = 2 conf["legLoc"] = 2
@ -194,10 +202,14 @@ class Visualizor:
"""Plots the percentage of nodes ready in the network""" """Plots the percentage of nodes ready in the network"""
vector1 = result.metrics["rowDist"] vector1 = result.metrics["rowDist"]
vector2 = result.metrics["columnDist"] vector2 = result.metrics["columnDist"]
if len(vector1) > len(vector2):
vector2 += [np.nan] * (len(vector1) - len(vector2))
elif len(vector1) < len(vector2):
vector1 += [np.nan] * (len(vector2) - len(vector1))
conf = {} conf = {}
text = str(result.shape).split("-") attrbs = self.__get_attrbs__(result)
conf["textBox"] = "Block Size: "+text[1]+"\nNumber of nodes: "+text[3]\ conf["textBox"] = "Block Size R: "+attrbs['bsrn']+"\nBlock Size C: "+attrbs['bscn']\
+"\nFailure rate: "+text[7]+" \nNetwork degree: "+text[23]+"\nX: "+text[11]+" rows/columns" +"\nNumber of nodes: "+attrbs['nn']+"\nFailure rate: "+attrbs['fr']+" \nNetwork degree: "+attrbs['nd']
conf["title"] = "Row/Column distribution" conf["title"] = "Row/Column distribution"
conf["type"] = "bar" conf["type"] = "bar"
conf["legLoc"] = 2 conf["legLoc"] = 2
@ -211,7 +223,7 @@ class Visualizor:
conf["path"] = "results/"+self.execID+"/plots/RowColDist-"+str(result.shape)+".png" conf["path"] = "results/"+self.execID+"/plots/RowColDist-"+str(result.shape)+".png"
maxi = 0 maxi = 0
for v in conf["data"]: for v in conf["data"]:
if max(v) > maxi: if np.nanmax(v) > maxi:
maxi = max(v) maxi = max(v)
conf["yaxismax"] = maxi conf["yaxismax"] = maxi
plotData(conf) plotData(conf)

5
smallConf.py
View File

@ -106,5 +106,8 @@ def nextShape():
runs, failureModels, failureRates, class1ratios, chis, validatorsPerNode1, validatorsPerNode2, blockSizes, numberNodes, netDegrees, bwUplinksProd, bwUplinks1, bwUplinks2): runs, failureModels, failureRates, class1ratios, chis, validatorsPerNode1, validatorsPerNode2, blockSizes, numberNodes, netDegrees, bwUplinksProd, bwUplinks1, bwUplinks2):
# Network Degree has to be an even number # Network Degree has to be an even number
if netDegree % 2 == 0: if netDegree % 2 == 0:
shape = Shape(blockSize, nn, fm, fr, class1ratio, chi, vpn1, vpn2, netDegree, bwUplinkProd, bwUplink1, bwUplink2, run) blockSizeR = blockSizeC = blockSize
blockSizeRK = blockSizeCK = blockSize // 2
chiR = chiC = chi
shape = Shape(blockSizeR, blockSizeRK, blockSizeC, blockSizeCK, nn, fm, fr, class1ratio, chiR, chiC, vpn1, vpn2, netDegree, bwUplinkProd, bwUplink1, bwUplink2, run)
yield shape yield shape