cerc-io/lotus
10
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 27 Wiki Activity

cache sector deadlines.

Browse Source
This commit is contained in:
vyzo 2021-05-18 15:28:51 +03:00
parent 74bad49068
commit 6b3e04b9b1
1 changed files with 11 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
extern/storage-sealing/commit_batch.go vendored
View File

@ -48,9 +48,9 @@ type CommitBatcher struct {
getConfig GetSealingConfigFunc getConfig GetSealingConfigFunc
verif ffiwrapper.Verifier verif ffiwrapper.Verifier
sectors map[abi.SectorNumber]SectorInfo deadlines map[abi.SectorNumber]time.Time
todo map[abi.SectorNumber]AggregateInput todo map[abi.SectorNumber]AggregateInput
waiting map[abi.SectorNumber][]chan cid.Cid waiting map[abi.SectorNumber][]chan cid.Cid
notify, stop, stopped chan struct{} notify, stop, stopped chan struct{}
force chan chan *cid.Cid force chan chan *cid.Cid
@ -67,8 +67,9 @@ func NewCommitBatcher(mctx context.Context, maddr address.Address, api CommitBat
getConfig: getConfig, getConfig: getConfig,
verif: verif, verif: verif,
todo: map[abi.SectorNumber]AggregateInput{}, deadlines: map[abi.SectorNumber]time.Time{},
waiting: map[abi.SectorNumber][]chan cid.Cid{}, todo: map[abi.SectorNumber]AggregateInput{},
waiting: map[abi.SectorNumber][]chan cid.Cid{},
notify: make(chan struct{}, 1), notify: make(chan struct{}, 1),
force: make(chan chan *cid.Cid), force: make(chan chan *cid.Cid),
@ -126,13 +127,13 @@ func (b *CommitBatcher) batchWait(maxWait, slack time.Duration) time.Duration {
var deadline time.Time var deadline time.Time
for sn := range b.todo { for sn := range b.todo {
sectorDeadline := b.getSectorDeadline(sn) sectorDeadline := b.deadlines[sn]
if deadline.IsZero() || (!sectorDeadline.IsZero() && sectorDeadline.Before(deadline)) { if deadline.IsZero() || (!sectorDeadline.IsZero() && sectorDeadline.Before(deadline)) {
deadline = sectorDeadline deadline = sectorDeadline
} }
} }
for sn := range b.waiting { for sn := range b.waiting {
sectorDeadline := b.getSectorDeadline(sn) sectorDeadline := b.deadlines[sn]
if deadline.IsZero() || (!sectorDeadline.IsZero() && sectorDeadline.Before(deadline)) { if deadline.IsZero() || (!sectorDeadline.IsZero() && sectorDeadline.Before(deadline)) {
deadline = sectorDeadline deadline = sectorDeadline
} }
@ -155,12 +156,7 @@ func (b *CommitBatcher) batchWait(maxWait, slack time.Duration) time.Duration {
return wait return wait
} }
func (b *CommitBatcher) getSectorDeadline(sn abi.SectorNumber) time.Time { func (b *CommitBatcher) getSectorDeadline(si SectorInfo) time.Time {
si, ok := b.sectors[sn]
if !ok {
return time.Time{}
}
tok, curEpoch, err := b.api.ChainHead(b.mctx) tok, curEpoch, err := b.api.ChainHead(b.mctx)
if err != nil { if err != nil {
log.Errorf("getting chain head: %s", err) log.Errorf("getting chain head: %s", err)
@ -259,7 +255,7 @@ func (b *CommitBatcher) processBatch(notif, after bool) (*cid.Cid, error) {
} }
delete(b.waiting, sn) delete(b.waiting, sn)
delete(b.todo, sn) delete(b.todo, sn)
delete(b.sectors, sn) delete(b.deadlines, sn)
return nil return nil
}) })
if err != nil { if err != nil {
@ -273,7 +269,7 @@ func (b *CommitBatcher) processBatch(notif, after bool) (*cid.Cid, error) {
func (b *CommitBatcher) AddCommit(ctx context.Context, s SectorInfo, in AggregateInput) (mcid cid.Cid, err error) { func (b *CommitBatcher) AddCommit(ctx context.Context, s SectorInfo, in AggregateInput) (mcid cid.Cid, err error) {
sn := s.SectorNumber sn := s.SectorNumber
b.lk.Lock() b.lk.Lock()
b.sectors[sn] = s b.deadlines[sn] = b.getSectorDeadline(s)
b.todo[sn] = in b.todo[sn] = in
sent := make(chan cid.Cid, 1) sent := make(chan cid.Cid, 1)