codex-storage/das-research
1
0
Fork 0
mirror of https://github.com/codex-storage/das-research.git synced 2025-02-23 16:28:27 +00:00
Code Issues Packages Projects Releases Wiki Activity

Introduce 'maliciousNode' parameter

Browse Source 
Signed-off-by: Arunima Chaudhuri <arunimachaudhuri2020@gmail.com>
This commit is contained in:
Arunima Chaudhuri 2024-01-21 01:27:17 +05:30
parent 4f5205e247
commit eb8588df9f
6 changed files with 165 additions and 38 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
DAS/results.py
View File

@ -16,6 +16,18 @@ class Result:
self.tta = -1
self.missingVector = []
self.metrics = {}
self.amImalicious = [0] * shape.numberNodes
self.msgSentCount = [0] * shape.numberNodes
self.msgRecvCount = [0] * shape.numberNodes
self.sampleRecvCount = [0] * shape.numberNodes
def copyValidators(self, validators):
"""Copy information from simulator.validators to result."""
for i in range(0,self.shape.numberNodes):
self.amImalicious[i] = validators[i].amImalicious
self.msgSentCount[i] = validators[i].msgSentCount
self.msgRecvCount[i] = validators[i].msgRecvCount
self.sampleRecvCount[i] = validators[i].sampleRecvCount
def populate(self, shape, config, missingVector):
"""It populates part of the result data inside a vector."""

4
DAS/shape.py
View File

@ -3,13 +3,14 @@
class Shape:
"""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, blockSize, numberNodes, failureModel, failureRate, maliciousNodes, class1ratio, chi, vpn1, vpn2, netDegree, bwUplinkProd, bwUplink1, bwUplink2, run):
"""Initializes the shape with the parameters passed in argument."""
self.run = run
self.numberNodes = numberNodes
self.blockSize = blockSize
self.failureModel = failureModel
self.failureRate = failureRate
self.maliciousNodes = maliciousNodes
self.netDegree = netDegree
self.class1ratio = class1ratio
self.chi = chi
@ -36,6 +37,7 @@ class Shape:
shastr += "-bwup2-"+str(self.bwUplink2)
shastr += "-nd-"+str(self.netDegree)
shastr += "-r-"+str(self.run)
shastr += "-mn-"+str(self.maliciousNodes)
return shastr
def setSeed(self, seed):

21
DAS/simulator.py
View File

@ -70,7 +70,14 @@ class Simulator:
assignedRows = []
assignedCols = []
maliciousNodesCount = int((self.shape.maliciousNodes / 100) * self.shape.numberNodes)
for i in range(self.shape.numberNodes):
if i==0:
amImalicious_value = 0
elif i < maliciousNodesCount+1:
amImalicious_value = 1
else:
amImalicious_value = 0
if self.config.evenLineDistribution:
if i < int(lightVal/self.shape.vpn1): # First start with the light nodes
start = i *self.shape.chi*self.shape.vpn1
@ -81,7 +88,7 @@ class Simulator:
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, self.config, r, c)
val = Validator(i, int(not i!=0), amImalicious_value, 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 column IDs: %s" % (val.ID, val.columnIDs), extra=self.format)
assignedRows = assignedRows + list(r)
@ -90,7 +97,7 @@ class Simulator:
self.nodeColumns.append(val.columnIDs)
else:
val = Validator(i, int(not i!=0), self.logger, self.shape, self.config)
val = Validator(i, int(not i!=0), amImalicious_value, self.logger, self.shape, self.config)
if i == self.proposerID:
val.initBlock()
else:
@ -229,6 +236,7 @@ class Simulator:
missingVector = []
progressVector = []
trafficStatsVector = []
malicious_nodes_not_added_count = 0
steps = 0
while(True):
missingVector.append(missingSamples)
@ -298,6 +306,14 @@ class Simulator:
break
steps += 1
for i in range(0,self.shape.numberNodes):
if not self.validators[i].amIaddedToQueue :
malicious_nodes_not_added_count += 1
self.logger.debug("Number of malicious nodes not added to the send queue: %d" % malicious_nodes_not_added_count, extra=self.format)
malicious_nodes_not_added_percentage = (malicious_nodes_not_added_count * 100)/(self.shape.numberNodes)
self.logger.debug("Percentage of malicious nodes not added to the send queue: %d" % malicious_nodes_not_added_percentage, extra=self.format)
progress = pd.DataFrame(progressVector)
if self.config.saveRCdist:
self.result.addMetric("rowDist", self.distR)
@ -305,5 +321,6 @@ class Simulator:
if self.config.saveProgress:
self.result.addMetric("progress", progress.to_dict(orient='list'))
self.result.populate(self.shape, self.config, missingVector)
self.result.copyValidators(self.validators)
return self.result

62
DAS/validator.py
View File

@ -38,7 +38,7 @@ class Validator:
"""It returns the validator ID."""
return str(self.ID)
def __init__(self, ID, amIproposer, logger, shape, config, rows = None, columns = None):
def __init__(self, ID, amIproposer, amImalicious, logger, shape, config, rows = None, columns = None):
"""It initializes the validator with the logger shape and rows/columns.
If rows/columns are specified these are observed, otherwise (default)
@ -54,6 +54,12 @@ class Validator:
self.receivedQueue = deque()
self.sendQueue = deque()
self.amIproposer = amIproposer
self.amImalicious = amImalicious
self.amIaddedToQueue = 0
self.msgSentCount = 0
self.msgRecvCount = 0
self.sampleSentCount = 0
self.sampleRecvCount = 0
self.logger = logger
if self.shape.chi < 1:
self.logger.error("Chi has to be greater than 0", extra=self.format)
@ -197,8 +203,10 @@ class Validator:
if not self.receivedBlock.getSegment(rID, cID):
self.logger.trace("Recv new: %d->%d: %d,%d", src, self.ID, rID, cID, extra=self.format)
self.receivedBlock.setSegment(rID, cID)
self.sampleRecvCount += 1
if self.perNodeQueue or self.perNeighborQueue:
self.receivedQueue.append((rID, cID))
self.msgRecvCount += 1
else:
self.logger.trace("Recv DUP: %d->%d: %d,%d", src, self.ID, rID, cID, extra=self.format)
self.statsRxDupInSlot += 1
@ -206,17 +214,23 @@ class Validator:
def addToSendQueue(self, rID, cID):
"""Queue a segment for forwarding."""
if self.perNodeQueue:
if self.perNodeQueue and not self.amImalicious:
self.sendQueue.append((rID, cID))
self.amIaddedToQueue = 1
self.msgSentCount += 1
if self.perNeighborQueue:
if self.perNeighborQueue and not self.amImalicious:
if rID in self.rowIDs:
for neigh in self.rowNeighbors[rID].values():
neigh.sendQueue.append(cID)
self.amIaddedToQueue = 1
self.msgSentCount += 1
if cID in self.columnIDs:
for neigh in self.columnNeighbors[cID].values():
neigh.sendQueue.append(rID)
self.amIaddedToQueue = 1
self.msgSentCount += 1
def receiveRowsColumns(self):
"""Finalize time step by merging newly received segments in state."""
@ -233,11 +247,14 @@ class Validator:
neigh.received |= neigh.receiving
neigh.receiving.setall(0)
for rID, cID in self.receivedQueue:
self.msgRecvCount += 1
# add newly received segments to the send queue
if self.perNodeQueue or self.perNeighborQueue:
while self.receivedQueue:
(rID, cID) = self.receivedQueue.popleft()
self.addToSendQueue(rID, cID)
if not self.amImalicious:
self.addToSendQueue(rID, cID)
def updateStats(self):
"""It updates the stats related to sent and received data."""
@ -269,7 +286,7 @@ class Validator:
def checkSendSegmentToNeigh(self, rID, cID, neigh):
"""Check and send a segment to a neighbor if needed."""
if self.checkSegmentToNeigh(rID, cID, neigh):
if self.checkSegmentToNeigh(rID, cID, neigh) and not self.amImalicious:
self.sendSegmentToNeigh(rID, cID, neigh)
return True
else:
@ -286,14 +303,16 @@ class Validator:
if rID in self.rowIDs:
for _, neigh in shuffledDict(self.rowNeighbors[rID], self.shuffleNeighbors):
self.checkSendSegmentToNeigh(rID, cID, neigh)
if not self.amImalicious:
self.checkSendSegmentToNeigh(rID, cID, neigh)
if self.statsTxInSlot >= self.bwUplink:
return
if cID in self.columnIDs:
for _, neigh in shuffledDict(self.columnNeighbors[cID], self.shuffleNeighbors):
self.checkSendSegmentToNeigh(rID, cID, neigh)
if not self.amImalicious:
self.checkSendSegmentToNeigh(rID, cID, neigh)
if self.statsTxInSlot >= self.bwUplink:
return
@ -318,19 +337,20 @@ class Validator:
# collect and shuffle
for rID, neighs in self.rowNeighbors.items():
for neigh in neighs.values():
if (neigh.sendQueue):
if (neigh.sendQueue) and not self.amImalicious:
queues.append((0, rID, neigh))
for cID, neighs in self.columnNeighbors.items():
for neigh in neighs.values():
if (neigh.sendQueue):
if (neigh.sendQueue) and not self.amImalicious:
queues.append((1, cID, neigh))
for dim, lineID, neigh in shuffled(queues, self.shuffleQueues):
if dim == 0:
self.checkSendSegmentToNeigh(lineID, neigh.sendQueue.popleft(), neigh)
else:
self.checkSendSegmentToNeigh(neigh.sendQueue.popleft(), lineID, neigh)
if not self.amImalicious:
if dim == 0:
self.checkSendSegmentToNeigh(lineID, neigh.sendQueue.popleft(), neigh)
else:
self.checkSendSegmentToNeigh(neigh.sendQueue.popleft(), lineID, neigh)
progress = True
if self.statsTxInSlot >= self.bwUplink:
return
@ -450,22 +470,24 @@ class Validator:
""" Send as much as we can in the timestep, limited by bwUplink."""
# process node level send queue
self.processSendQueue()
if not self.amImalicious:
self.processSendQueue()
if self.statsTxInSlot >= self.bwUplink:
return
# process neighbor level send queues in shuffled breadth-first order
self.processPerNeighborSendQueue()
if not self.amImalicious:
self.processPerNeighborSendQueue()
if self.statsTxInSlot >= self.bwUplink:
return
# process possible segments to send in shuffled breadth-first order
if self.segmentShuffleScheduler:
if self.segmentShuffleScheduler and not self.amImalicious:
self.runSegmentShuffleScheduler()
if self.statsTxInSlot >= self.bwUplink:
return
if self.dumbRandomScheduler:
if self.dumbRandomScheduler and not self.amImalicious:
self.runDumbRandomScheduler()
if self.statsTxInSlot >= self.bwUplink:
return
@ -497,7 +519,8 @@ class Validator:
for i in range(len(rep)):
if rep[i]:
self.logger.trace("Rep: %d,%d", id, i, extra=self.format)
self.addToSendQueue(id, i)
if not self.amImalicious:
self.addToSendQueue(id, i)
# self.statsRepairInSlot += rep.count(1)
def restoreColumns(self):
@ -515,7 +538,8 @@ class Validator:
for i in range(len(rep)):
if rep[i]:
self.logger.trace("Rep: %d,%d", i, id, extra=self.format)
self.addToSendQueue(i, id)
if not self.amImalicious:
self.addToSendQueue(i, id)
# self.statsRepairInSlot += rep.count(1)
def checkStatus(self):

83
DAS/visualizor.py
View File

@ -6,16 +6,20 @@ import os
def plotData(conf):
plt.clf()
fig = plt.figure("9, 3")
plt.grid(True)
if conf["desLoc"] == 1:
xDes = 0
else:
xDes = conf["xdots"][-1] * 0.6
props = dict(boxstyle='round', facecolor='wheat', alpha=0.5)
plt.text(xDes, conf["yaxismax"]/4, conf["textBox"], fontsize=10, verticalalignment='top', bbox=props)
for i in range(len(conf["data"])):
if conf["type"] == "plot":
if conf["type"] == "plot":
for i in range(len(conf["data"])):
plt.plot(conf["xdots"], conf["data"][i], conf["colors"][i], label=conf["labels"][i])
if conf["type"] == "bar":
elif conf["type"] == "individual_bar":
plt.bar(conf["xdots"], conf["data"])
elif conf["type"] == "grouped_bar":
for i in range(len(conf["data"])):
plt.bar(conf["xdots"], conf["data"][i], label=conf["labels"][i])
plt.title(conf["title"])
plt.ylabel(conf["ylabel"])
@ -66,6 +70,9 @@ class Visualizor:
for result in self.results:
plotPath = "results/"+self.execID+"/plots/"+str(result.shape)
os.makedirs(plotPath, exist_ok=True)
self.plotMessagesSent(result, plotPath)
self.plotMessagesRecv(result, plotPath)
self.plotSampleRecv(result, plotPath)
self.plotMissingSamples(result, plotPath)
self.plotProgress(result, plotPath)
self.plotSentData(result, plotPath)
@ -74,6 +81,28 @@ class Visualizor:
if self.config.saveRCdist:
self.plotRowCol(result, plotPath)
def plotSampleRecv(self, result, plotPath):
"""Plots the percentage sampleRecv for each node"""
conf = {}
text = str(result.shape).split("-")
conf["textBox"] = "Block Size: "+text[1]+"\nNumber of nodes: "+text[3]\
+"\nFailure rate: "+text[7]+" \nNetwork degree: "+text[23]+"\nX: "+text[11]+" rows/columns"
conf["title"] = "Percentage of Samples Received by Nodes"
conf["type"] = "individual_bar"
conf["legLoc"] = 1
conf["desLoc"] = 1
conf["xlabel"] = "Nodes"
conf["ylabel"] = "Percentage of samples received (%)"
total_samples = result.shape.blockSize * result.shape.blockSize
percentage_data = [(count / total_samples) * 100 for count in result.sampleRecvCount]
conf["data"] = percentage_data
conf["xdots"] = range(result.shape.numberNodes)
conf["path"] = plotPath + "/sampleRecv.png"
maxi = max(conf["data"])
conf["yaxismax"] = maxi
plotData(conf)
print("Plot %s created." % conf["path"])
def plotMissingSamples(self, result, plotPath):
"""Plots the missing samples in the network"""
conf = {}
@ -101,9 +130,9 @@ class Visualizor:
def plotProgress(self, result, plotPath):
"""Plots the percentage of nodes ready in the network"""
vector1 = result.metrics["progress"]["nodes ready"]
vector2 = result.metrics["progress"]["validators ready"]
vector3 = result.metrics["progress"]["samples received"]
vector1 = [x * 100 for x in result.metrics["progress"]["nodes ready"]]
vector2 = [x * 100 for x in result.metrics["progress"]["validators ready"]]
vector3 = [x * 100 for x in result.metrics["progress"]["samples received"]]
conf = {}
text = str(result.shape).split("-")
conf["textBox"] = "Block Size: "+text[1]+"\nNumber of nodes: "+text[3]\
@ -228,7 +257,7 @@ class Visualizor:
conf["textBox"] = "Block Size: "+text[1]+"\nNumber of nodes: "+text[3]\
+"\nFailure rate: "+text[7]+" \nNetwork degree: "+text[23]+"\nX: "+text[11]+" rows/columns"
conf["title"] = "Row/Column distribution"
conf["type"] = "bar"
conf["type"] = "grouped_bar"
conf["legLoc"] = 2
conf["desLoc"] = 2
conf["colors"] = ["r+", "b+"]
@ -246,3 +275,43 @@ class Visualizor:
plotData(conf)
print("Plot %s created." % conf["path"])
def plotMessagesSent(self, result, plotPath):
"""Plots the number of messages sent by all nodes"""
conf = {}
text = str(result.shape).split("-")
conf["textBox"] = "Block Size: "+text[1]+"\nNumber of nodes: "+text[3]\
+"\nFailure rate: "+text[7]+" \nNetwork degree: "+text[23]+"\nX: "+text[11]+" rows/columns"
conf["title"] = "Number of Messages Sent by Nodes"
conf["type"] = "individual_bar"
conf["legLoc"] = 1
conf["desLoc"] = 1
conf["xlabel"] = "Nodes"
conf["ylabel"] = "Number of Messages Sent"
conf["data"] = result.msgSentCount
conf["xdots"] = range(result.shape.numberNodes)
conf["path"] = plotPath + "/messagesSent.png"
maxi = max(conf["data"])
conf["yaxismax"] = maxi
plotData(conf)
print("Plot %s created." % conf["path"])
def plotMessagesRecv(self, result, plotPath):
"""Plots the number of messages received by all nodes"""
conf = {}
text = str(result.shape).split("-")
conf["textBox"] = "Block Size: "+text[1]+"\nNumber of nodes: "+text[3]\
+"\nFailure rate: "+text[7]+" \nNetwork degree: "+text[23]+"\nX: "+text[11]+" rows/columns"
conf["title"] = "Number of Messages Received by Nodes"
conf["type"] = "individual_bar"
conf["legLoc"] = 1
conf["desLoc"] = 1
conf["xlabel"] = "Nodes"
conf["ylabel"] = "Number of Messages Received"
conf["data"] = result.msgRecvCount
conf["xdots"] = range(result.shape.numberNodes)
conf["path"] = plotPath + "/messagesRecv.png"
maxi = max(conf["data"])
conf["yaxismax"] = maxi
plotData(conf)
print("Plot %s created." % conf["path"])

21
smallConf.py
View File

@ -45,19 +45,22 @@ numJobs = -1
evenLineDistribution = True
# Number of simulation runs with the same parameters for statistical relevance
runs = range(3)
runs = [1]
# Number of validators
numberNodes = range(128, 513, 128)
numberNodes = [1024]
# select failure model between: "random, sequential, MEP, MEP+1, DEP, DEP+1, MREP, MREP-1"
failureModels = ["random"]
# Percentage of block not released by producer
failureRates = range(40, 81, 20)
failureRates = [0]
# Percentage of nodes that are considered malicious
maliciousNodes = [95]
# Block size in one dimension in segments. Block is blockSizes * blockSizes segments.
blockSizes = range(64, 113, 128)
blockSizes = [128]
# Per-topic mesh neighborhood size
netDegrees = range(8, 9, 2)
@ -69,8 +72,8 @@ chis = range(2, 3, 2)
class1ratios = [0.8]
# Number of validators per beacon node
validatorsPerNode1 = [1]
validatorsPerNode2 = [500]
validatorsPerNode1 = [10]
validatorsPerNode2 = [250]
# Set uplink bandwidth in megabits/second
bwUplinksProd = [200]
@ -102,9 +105,9 @@ diagnostics = False
saveGit = False
def nextShape():
for run, fm, fr, class1ratio, chi, vpn1, vpn2, blockSize, nn, netDegree, bwUplinkProd, bwUplink1, bwUplink2 in itertools.product(
runs, failureModels, failureRates, class1ratios, chis, validatorsPerNode1, validatorsPerNode2, blockSizes, numberNodes, netDegrees, bwUplinksProd, bwUplinks1, bwUplinks2):
for run, fm, fr, mn, class1ratio, chi, vpn1, vpn2, blockSize, nn, netDegree, bwUplinkProd, bwUplink1, bwUplink2 in itertools.product(
runs, failureModels, failureRates, maliciousNodes, class1ratios, chis, validatorsPerNode1, validatorsPerNode2, blockSizes, numberNodes, netDegrees, bwUplinksProd, bwUplinks1, bwUplinks2):
# Network Degree has to be an even number
if netDegree % 2 == 0:
shape = Shape(blockSize, nn, fm, fr, class1ratio, chi, vpn1, vpn2, netDegree, bwUplinkProd, bwUplink1, bwUplink2, run)
shape = Shape(blockSize, nn, fm, fr, mn, class1ratio, chi, vpn1, vpn2, netDegree, bwUplinkProd, bwUplink1, bwUplink2, run)
yield shape