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

factorize send code 

Signed-off-by: Csaba Kiraly <csaba.kiraly@gmail.com>
This commit is contained in:
Csaba Kiraly 2023-02-24 08:55:26 +01:00
parent fa1818a43b
commit 300bc19c67
No known key found for this signature in database
GPG Key ID: 0FE274EE8C95166E
1 changed files with 21 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
DAS/validator.py
View File

@ -270,11 +270,7 @@ class Validator:
else: else:
return False # received or already sent return False # received or already sent
def send(self): def processSendQueue(self):
""" Send as much as we can in the timeslot, limited by bwUplink
"""
# process node level send queue
while self.sendQueue: while self.sendQueue:
(rID, cID) = self.sendQueue[0] (rID, cID) = self.sendQueue[0]
@ -294,7 +290,7 @@ class Validator:
self.sendQueue.popleft() self.sendQueue.popleft()
# process neighbor level send queues in shuffled breadth-first order def processPerNeighborSendQueue(self):
progress = True progress = True
while (progress): while (progress):
progress = False progress = False
@ -314,8 +310,7 @@ class Validator:
if self.statsTxInSlot >= self.bwUplink: if self.statsTxInSlot >= self.bwUplink:
return return
# process possible segments to send in shuffled breadth-first order def runSegmentShuffleScheduler(self):
if self.segmentShuffleScheduler:
# This scheduler check which owned segments needs sending (at least # This scheduler check which owned segments needs sending (at least
# one neighbor needing it). Then it sends each segment that's worth sending # one neighbor needing it). Then it sends each segment that's worth sending
# once, in shuffled order. This is repeated until bw limit. # once, in shuffled order. This is repeated until bw limit.
@ -376,7 +371,7 @@ class Validator:
else: else:
self.segmentShuffleGen = shuffled(self.segmentsToSend, self.shuffleLines) self.segmentShuffleGen = shuffled(self.segmentsToSend, self.shuffleLines)
if self.dumbRandomScheduler: def runDumbRandomScheduler(self):
# dumb random scheduler picking segments at random and trying to send it # dumb random scheduler picking segments at random and trying to send it
tries = 100 tries = 100
t = tries t = tries
@ -408,6 +403,23 @@ class Validator:
return return
return return
def send(self):
""" Send as much as we can in the timeslot, limited by bwUplink
"""
# process node level send queue
self.processSendQueue()
# process neighbor level send queues in shuffled breadth-first order
self.processPerNeighborSendQueue()
# process possible segments to send in shuffled breadth-first order
if self.segmentShuffleScheduler:
self.runSegmentShuffleScheduler()
if self.dumbRandomScheduler:
self.runDumbRandomScheduler()
def logRows(self): def logRows(self):
"""It logs the rows assigned to the validator.""" """It logs the rows assigned to the validator."""
if self.logger.isEnabledFor(logging.DEBUG): if self.logger.isEnabledFor(logging.DEBUG):