feat: mempool: reintroduce ForEachPendingMessage
This commit is contained in:
parent
92dcfe530b
commit
fb228ebbff
@ -448,12 +448,8 @@ func New(ctx context.Context, api Provider, ds dtypes.MetadataDS, us stmgr.Upgra
|
|||||||
return mp, nil
|
return mp, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (mp *MessagePool) TryForEachPendingMessage(f func(cid.Cid) error) error {
|
func (mp *MessagePool) ForEachPendingMessage(f func(cid.Cid) error) error {
|
||||||
// avoid deadlocks in splitstore compaction when something else needs to access the blockstore
|
mp.lk.Lock()
|
||||||
// while holding the mpool lock
|
|
||||||
if !mp.lk.TryLock() {
|
|
||||||
return xerrors.Errorf("mpool TryForEachPendingMessage: could not acquire lock")
|
|
||||||
}
|
|
||||||
defer mp.lk.Unlock()
|
defer mp.lk.Unlock()
|
||||||
|
|
||||||
for _, mset := range mp.pending {
|
for _, mset := range mp.pending {
|
||||||
|
@ -70,7 +70,7 @@ func MessagePool(lc fx.Lifecycle, mctx helpers.MetricsCtx, us stmgr.UpgradeSched
|
|||||||
return mp.Close()
|
return mp.Close()
|
||||||
},
|
},
|
||||||
})
|
})
|
||||||
protector.AddProtector(mp.TryForEachPendingMessage)
|
protector.AddProtector(mp.ForEachPendingMessage)
|
||||||
return mp, nil
|
return mp, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
Loading…
Reference in New Issue
Block a user