changes as per review
This commit is contained in:
parent
8e62fec1d3
commit
f0f2b3e1f4
41
extern/storage-sealing/input.go
vendored
41
extern/storage-sealing/input.go
vendored
@ -313,15 +313,35 @@ func (m *Sealing) SectorAddPieceToAny(ctx context.Context, size abi.UnpaddedPiec
|
|||||||
m.inputLk.Lock()
|
m.inputLk.Lock()
|
||||||
if pp, exist := m.pendingPieces[proposalCID(deal)]; exist {
|
if pp, exist := m.pendingPieces[proposalCID(deal)]; exist {
|
||||||
m.inputLk.Unlock()
|
m.inputLk.Unlock()
|
||||||
select {
|
|
||||||
case <-pp.doneCh:
|
// we already have a pre-existing add piece call for this deal, let's wait for it to finish and see if it's successful
|
||||||
res := pp.resp
|
for {
|
||||||
|
res, err := waitAddPieceResp(ctx, pp)
|
||||||
|
if err != nil {
|
||||||
|
return api.SectorOffset{}, err
|
||||||
|
}
|
||||||
|
// there was an error waiting for a pre-existing add piece call, let's retry
|
||||||
|
if res.err != nil {
|
||||||
|
m.inputLk.Lock()
|
||||||
|
pp = m.addPendingPiece(ctx, size, data, deal, sp)
|
||||||
|
m.inputLk.Unlock()
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
// all good, return the response
|
||||||
return api.SectorOffset{Sector: res.sn, Offset: res.offset.Padded()}, res.err
|
return api.SectorOffset{Sector: res.sn, Offset: res.offset.Padded()}, res.err
|
||||||
case <-ctx.Done():
|
|
||||||
return api.SectorOffset{}, ctx.Err()
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
pp := m.addPendingPiece(ctx, size, data, deal, sp)
|
||||||
|
m.inputLk.Unlock()
|
||||||
|
res, err := waitAddPieceResp(ctx, pp)
|
||||||
|
if err != nil {
|
||||||
|
return api.SectorOffset{}, err
|
||||||
|
}
|
||||||
|
return api.SectorOffset{Sector: res.sn, Offset: res.offset.Padded()}, res.err
|
||||||
|
}
|
||||||
|
|
||||||
|
func (m *Sealing) addPendingPiece(ctx context.Context, size abi.UnpaddedPieceSize, data storage.Data, deal api.PieceDealInfo, sp abi.RegisteredSealProof) *pendingPiece {
|
||||||
doneCh := make(chan struct{})
|
doneCh := make(chan struct{})
|
||||||
pp := &pendingPiece{
|
pp := &pendingPiece{
|
||||||
doneCh: doneCh,
|
doneCh: doneCh,
|
||||||
@ -337,18 +357,21 @@ func (m *Sealing) SectorAddPieceToAny(ctx context.Context, size abi.UnpaddedPiec
|
|||||||
|
|
||||||
m.pendingPieces[proposalCID(deal)] = pp
|
m.pendingPieces[proposalCID(deal)] = pp
|
||||||
go func() {
|
go func() {
|
||||||
defer m.inputLk.Unlock()
|
|
||||||
if err := m.updateInput(ctx, sp); err != nil {
|
if err := m.updateInput(ctx, sp); err != nil {
|
||||||
log.Errorf("%+v", err)
|
log.Errorf("%+v", err)
|
||||||
}
|
}
|
||||||
}()
|
}()
|
||||||
|
|
||||||
|
return pp
|
||||||
|
}
|
||||||
|
|
||||||
|
func waitAddPieceResp(ctx context.Context, pp *pendingPiece) (*pieceAcceptResp, error) {
|
||||||
select {
|
select {
|
||||||
case <-doneCh:
|
case <-pp.doneCh:
|
||||||
res := pp.resp
|
res := pp.resp
|
||||||
return api.SectorOffset{Sector: res.sn, Offset: res.offset.Padded()}, res.err
|
return res, nil
|
||||||
case <-ctx.Done():
|
case <-ctx.Done():
|
||||||
return api.SectorOffset{}, ctx.Err()
|
return nil, ctx.Err()
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
Loading…
Reference in New Issue
Block a user