From 5ad33896cc41d5f206043c564a49b611cbf4158d Mon Sep 17 00:00:00 2001 From: Aayush Rajasekaran Date: Thu, 18 Jun 2020 14:19:30 -0400 Subject: [PATCH 1/9] Do not allocate pieces that are bigger than a sector --- sealing.go | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/sealing.go b/sealing.go index b4f3899bc..09e0cf84c 100644 --- a/sealing.go +++ b/sealing.go @@ -90,6 +90,10 @@ func (m *Sealing) AllocatePiece(size abi.UnpaddedPieceSize) (sectorID abi.Sector if (padreader.PaddedSize(uint64(size))) != size { return 0, 0, xerrors.Errorf("cannot allocate unpadded piece") } + + if size > abi.UnpaddedPieceSize(m.sealer.SectorSize()) { + return 0, 0, xerrors.Errorf("piece cannot fit into a sector") + } sid, err := m.sc.Next() if err != nil { From db863a4de7e37c4c13085bc90631cfcee895ef1e Mon Sep 17 00:00:00 2001 From: Aayush Rajasekaran Date: Tue, 23 Jun 2020 15:32:22 -0400 Subject: [PATCH 2/9] Add new event for start of CC sectors --- fsm.go | 7 +++++-- fsm_events.go | 12 ++++++++++++ garbage.go | 2 +- sealing.go | 22 +++++++++++++++------- 4 files changed, 33 insertions(+), 10 deletions(-) diff --git a/fsm.go b/fsm.go index 478779e96..228f36c1c 100644 --- a/fsm.go +++ b/fsm.go @@ -34,8 +34,11 @@ func (m *Sealing) Plan(events []statemachine.Event, user interface{}) (interface var fsmPlanners = map[SectorState]func(events []statemachine.Event, state *SectorInfo) error{ // Sealing - UndefinedSectorState: planOne(on(SectorStart{}, Packing)), - Packing: planOne(on(SectorPacked{}, PreCommit1)), + UndefinedSectorState: planOne( + on(SectorStart{}, Packing), + on(SectorStartCC{}, Packing), + ), + Packing: planOne(on(SectorPacked{}, PreCommit1)), PreCommit1: planOne( on(SectorPreCommit1{}, PreCommit2), on(SectorSealPreCommit1Failed{}, SealPreCommit1Failed), diff --git a/fsm_events.go b/fsm_events.go index 1c7991834..cdd62867e 100644 --- a/fsm_events.go +++ b/fsm_events.go @@ -59,6 +59,18 @@ func (evt SectorStart) apply(state *SectorInfo) { state.SectorType = evt.SectorType } +type SectorStartCC struct { + ID abi.SectorNumber + SectorType abi.RegisteredSealProof + Pieces []Piece +} + +func (evt SectorStartCC) apply(state *SectorInfo) { + state.SectorNumber = evt.ID + state.Pieces = evt.Pieces + state.SectorType = evt.SectorType +} + type SectorPacked struct{ FillerPieces []abi.PieceInfo } func (evt SectorPacked) apply(state *SectorInfo) { diff --git a/garbage.go b/garbage.go index 9e6246b4c..a24c0a9f5 100644 --- a/garbage.go +++ b/garbage.go @@ -77,7 +77,7 @@ func (m *Sealing) PledgeSector() error { } } - if err := m.newSector(sid, rt, ps); err != nil { + if err := m.newSectorCC(sid, rt, ps); err != nil { log.Errorf("%+v", err) return } diff --git a/sealing.go b/sealing.go index 09e0cf84c..2e13e8206 100644 --- a/sealing.go +++ b/sealing.go @@ -90,7 +90,7 @@ func (m *Sealing) AllocatePiece(size abi.UnpaddedPieceSize) (sectorID abi.Sector if (padreader.PaddedSize(uint64(size))) != size { return 0, 0, xerrors.Errorf("cannot allocate unpadded piece") } - + if size > abi.UnpaddedPieceSize(m.sealer.SectorSize()) { return 0, 0, xerrors.Errorf("piece cannot fit into a sector") } @@ -130,11 +130,13 @@ func (m *Sealing) SealPiece(ctx context.Context, size abi.UnpaddedPieceSize, r i }) } -// newSector accepts a slice of pieces which will have a deal associated with -// them (in the event of a storage deal) or no deal (in the event of sealing -// garbage data) +func (m *Sealing) Remove(ctx context.Context, sid abi.SectorNumber) error { + return m.sectors.Send(uint64(sid), SectorRemove{}) +} + +// newSector accepts a slice of pieces which will have deals associated with func (m *Sealing) newSector(sid abi.SectorNumber, rt abi.RegisteredSealProof, pieces []Piece) error { - log.Infof("Start sealing %d", sid) + log.Infof("Creating sector %d", sid) return m.sectors.Send(uint64(sid), SectorStart{ ID: sid, Pieces: pieces, @@ -142,8 +144,14 @@ func (m *Sealing) newSector(sid abi.SectorNumber, rt abi.RegisteredSealProof, pi }) } -func (m *Sealing) Remove(ctx context.Context, sid abi.SectorNumber) error { - return m.sectors.Send(uint64(sid), SectorRemove{}) +// newSectorCC accepts a slice of pieces with no deal (junk data) +func (m *Sealing) newSectorCC(sid abi.SectorNumber, rt abi.RegisteredSealProof, pieces []Piece) error { + log.Infof("Creating CC sector %d", sid) + return m.sectors.Send(uint64(sid), SectorStartCC{ + ID: sid, + Pieces: pieces, + SectorType: rt, + }) } func (m *Sealing) minerSector(num abi.SectorNumber) abi.SectorID { From eb17b2f3717a0b1af73cae7126a02f4719323cb6 Mon Sep 17 00:00:00 2001 From: Aayush Rajasekaran Date: Tue, 23 Jun 2020 15:59:57 -0400 Subject: [PATCH 3/9] Add a state for waiting on deals, and events for new deals and start packing --- fsm.go | 7 ++++++- fsm_events.go | 12 ++++++++++++ sealing.go | 5 +++++ sector_state.go | 1 + 4 files changed, 24 insertions(+), 1 deletion(-) diff --git a/fsm.go b/fsm.go index 228f36c1c..ff9ca4d9b 100644 --- a/fsm.go +++ b/fsm.go @@ -35,9 +35,14 @@ var fsmPlanners = map[SectorState]func(events []statemachine.Event, state *Secto // Sealing UndefinedSectorState: planOne( - on(SectorStart{}, Packing), + on(SectorStart{}, Empty), on(SectorStartCC{}, Packing), ), + Empty: planOne(on(SectorAddPiece{}, WaitDeals)), + WaitDeals: planOne( + on(SectorAddPiece{}, WaitDeals), + on(SectorStartPacking{}, Packing), + ), Packing: planOne(on(SectorPacked{}, PreCommit1)), PreCommit1: planOne( on(SectorPreCommit1{}, PreCommit2), diff --git a/fsm_events.go b/fsm_events.go index cdd62867e..a128f7c36 100644 --- a/fsm_events.go +++ b/fsm_events.go @@ -71,6 +71,18 @@ func (evt SectorStartCC) apply(state *SectorInfo) { state.SectorType = evt.SectorType } +type SectorAddPiece struct { + NewPiece Piece +} + +func (evt SectorAddPiece) apply(state *SectorInfo) { + state.Pieces = append(state.Pieces, evt.NewPiece) +} + +type SectorStartPacking struct{} + +func (evt SectorStartPacking) apply(*SectorInfo) {} + type SectorPacked struct{ FillerPieces []abi.PieceInfo } func (evt SectorPacked) apply(state *SectorInfo) { diff --git a/sealing.go b/sealing.go index 2e13e8206..e4124a16b 100644 --- a/sealing.go +++ b/sealing.go @@ -134,6 +134,11 @@ func (m *Sealing) Remove(ctx context.Context, sid abi.SectorNumber) error { return m.sectors.Send(uint64(sid), SectorRemove{}) } +func (m *Sealing) StartPacking(sectorID abi.SectorNumber) error { + log.Infof("Starting packing sector %d", sectorID) + return m.sectors.Send(uint64(sectorID), SectorStartPacking{}) +} + // newSector accepts a slice of pieces which will have deals associated with func (m *Sealing) newSector(sid abi.SectorNumber, rt abi.RegisteredSealProof, pieces []Piece) error { log.Infof("Creating sector %d", sid) diff --git a/sector_state.go b/sector_state.go index 63058cd6c..f2801c9fc 100644 --- a/sector_state.go +++ b/sector_state.go @@ -7,6 +7,7 @@ const ( // happy path Empty SectorState = "Empty" + WaitDeals SectorState = "WaitDeals" // waiting for more pieces (deals) to be added to the sector Packing SectorState = "Packing" // sector not in sealStore, and not on chain PreCommit1 SectorState = "PreCommit1" // do PreCommit1 PreCommit2 SectorState = "PreCommit2" // do PreCommit1 From 7f75cd71daa9e31af2e4bec33cff8f23c662b17e Mon Sep 17 00:00:00 2001 From: Aayush Rajasekaran Date: Tue, 23 Jun 2020 16:07:54 -0400 Subject: [PATCH 4/9] Resolve proof type when creating new sector --- garbage.go | 9 +-------- sealing.go | 21 +++++++++++++-------- 2 files changed, 14 insertions(+), 16 deletions(-) diff --git a/garbage.go b/garbage.go index a24c0a9f5..0464259c3 100644 --- a/garbage.go +++ b/garbage.go @@ -6,7 +6,6 @@ import ( "golang.org/x/xerrors" - "github.com/filecoin-project/sector-storage/ffiwrapper" "github.com/filecoin-project/specs-actors/actors/abi" nr "github.com/filecoin-project/storage-fsm/lib/nullreader" @@ -46,12 +45,6 @@ func (m *Sealing) PledgeSector() error { size := abi.PaddedPieceSize(m.sealer.SectorSize()).Unpadded() - rt, err := ffiwrapper.SealProofTypeFromSectorSize(m.sealer.SectorSize()) - if err != nil { - log.Error(err) - return - } - sid, err := m.sc.Next() if err != nil { log.Errorf("%+v", err) @@ -77,7 +70,7 @@ func (m *Sealing) PledgeSector() error { } } - if err := m.newSectorCC(sid, rt, ps); err != nil { + if err := m.newSectorCC(sid, ps); err != nil { log.Errorf("%+v", err) return } diff --git a/sealing.go b/sealing.go index e4124a16b..f7af0de48 100644 --- a/sealing.go +++ b/sealing.go @@ -117,12 +117,7 @@ func (m *Sealing) SealPiece(ctx context.Context, size abi.UnpaddedPieceSize, r i return xerrors.Errorf("adding piece to sector: %w", err) } - rt, err := ffiwrapper.SealProofTypeFromSectorSize(m.sealer.SectorSize()) - if err != nil { - return xerrors.Errorf("bad sector size: %w", err) - } - - return m.newSector(sectorID, rt, []Piece{ + return m.newSector(sectorID, []Piece{ { Piece: ppi, DealInfo: &d, @@ -140,7 +135,12 @@ func (m *Sealing) StartPacking(sectorID abi.SectorNumber) error { } // newSector accepts a slice of pieces which will have deals associated with -func (m *Sealing) newSector(sid abi.SectorNumber, rt abi.RegisteredSealProof, pieces []Piece) error { +func (m *Sealing) newSector(sid abi.SectorNumber, pieces []Piece) error { + rt, err := ffiwrapper.SealProofTypeFromSectorSize(m.sealer.SectorSize()) + if err != nil { + return xerrors.Errorf("bad sector size: %w", err) + } + log.Infof("Creating sector %d", sid) return m.sectors.Send(uint64(sid), SectorStart{ ID: sid, @@ -150,7 +150,12 @@ func (m *Sealing) newSector(sid abi.SectorNumber, rt abi.RegisteredSealProof, pi } // newSectorCC accepts a slice of pieces with no deal (junk data) -func (m *Sealing) newSectorCC(sid abi.SectorNumber, rt abi.RegisteredSealProof, pieces []Piece) error { +func (m *Sealing) newSectorCC(sid abi.SectorNumber, pieces []Piece) error { + rt, err := ffiwrapper.SealProofTypeFromSectorSize(m.sealer.SectorSize()) + if err != nil { + return xerrors.Errorf("bad sector size: %w", err) + } + log.Infof("Creating CC sector %d", sid) return m.sectors.Send(uint64(sid), SectorStartCC{ ID: sid, From 73f5c6867446ef4101af9bab9dbbbbde040c6dc4 Mon Sep 17 00:00:00 2001 From: Aayush Rajasekaran Date: Tue, 23 Jun 2020 16:51:35 -0400 Subject: [PATCH 5/9] Non-CC sectors start with no pieces, can be added using addPiece() --- fsm_events.go | 2 -- sealing.go | 29 ++++++++++++++++++++--------- 2 files changed, 20 insertions(+), 11 deletions(-) diff --git a/fsm_events.go b/fsm_events.go index a128f7c36..a631ec08d 100644 --- a/fsm_events.go +++ b/fsm_events.go @@ -50,12 +50,10 @@ func (evt SectorForceState) applyGlobal(state *SectorInfo) bool { type SectorStart struct { ID abi.SectorNumber SectorType abi.RegisteredSealProof - Pieces []Piece } func (evt SectorStart) apply(state *SectorInfo) { state.SectorNumber = evt.ID - state.Pieces = evt.Pieces state.SectorType = evt.SectorType } diff --git a/sealing.go b/sealing.go index f7af0de48..3363c8919 100644 --- a/sealing.go +++ b/sealing.go @@ -114,15 +114,27 @@ func (m *Sealing) SealPiece(ctx context.Context, size abi.UnpaddedPieceSize, r i ppi, err := m.sealer.AddPiece(sectorstorage.WithPriority(ctx, DealSectorPriority), m.minerSector(sectorID), []abi.UnpaddedPieceSize{}, size, r) if err != nil { - return xerrors.Errorf("adding piece to sector: %w", err) + return xerrors.Errorf("writing piece: %w", err) } - return m.newSector(sectorID, []Piece{ - { - Piece: ppi, - DealInfo: &d, - }, + err = m.newSector(sectorID) + if err != nil { + return xerrors.Errorf("creating new sector: %w", err) + } + + err = m.addPiece(sectorID, Piece{ + Piece: ppi, + DealInfo: &d, }) + + if err != nil { + return xerrors.Errorf("adding piece to sector: %w", err) + } +} + +func (m *Sealing) addPiece(sectorID abi.SectorNumber, piece Piece) error { + log.Infof("Adding piece to sector %d", sectorID) + return m.sectors.Send(uint64(sectorID), SectorAddPiece{NewPiece: piece}) } func (m *Sealing) Remove(ctx context.Context, sid abi.SectorNumber) error { @@ -134,8 +146,8 @@ func (m *Sealing) StartPacking(sectorID abi.SectorNumber) error { return m.sectors.Send(uint64(sectorID), SectorStartPacking{}) } -// newSector accepts a slice of pieces which will have deals associated with -func (m *Sealing) newSector(sid abi.SectorNumber, pieces []Piece) error { +// newSector creates a new sector for deal storage +func (m *Sealing) newSector(sid abi.SectorNumber) error { rt, err := ffiwrapper.SealProofTypeFromSectorSize(m.sealer.SectorSize()) if err != nil { return xerrors.Errorf("bad sector size: %w", err) @@ -144,7 +156,6 @@ func (m *Sealing) newSector(sid abi.SectorNumber, pieces []Piece) error { log.Infof("Creating sector %d", sid) return m.sectors.Send(uint64(sid), SectorStart{ ID: sid, - Pieces: pieces, SectorType: rt, }) } From 020f3a173a648fa2bb055a6c48e89d3e9e81f0fd Mon Sep 17 00:00:00 2001 From: Aayush Rajasekaran Date: Tue, 23 Jun 2020 17:31:51 -0400 Subject: [PATCH 6/9] Squash Sealing's AllocatePiece and SealPiece into one method --- sealing.go | 26 +++++++++++--------------- 1 file changed, 11 insertions(+), 15 deletions(-) diff --git a/sealing.go b/sealing.go index 3363c8919..8a0483ca3 100644 --- a/sealing.go +++ b/sealing.go @@ -85,8 +85,8 @@ func (m *Sealing) Run(ctx context.Context) error { func (m *Sealing) Stop(ctx context.Context) error { return m.sectors.Stop(ctx) } - -func (m *Sealing) AllocatePiece(size abi.UnpaddedPieceSize) (sectorID abi.SectorNumber, offset uint64, err error) { +func (m *Sealing) AddPieceToAnySector(ctx context.Context, size abi.UnpaddedPieceSize, r io.Reader, d DealInfo) (abi.SectorNumber, uint64, error) { + log.Infof("Adding piece for deal %d", d.DealID) if (padreader.PaddedSize(uint64(size))) != size { return 0, 0, xerrors.Errorf("cannot allocate unpadded piece") } @@ -105,31 +105,27 @@ func (m *Sealing) AllocatePiece(size abi.UnpaddedPieceSize) (sectorID abi.Sector return 0, 0, xerrors.Errorf("initializing sector: %w", err) } - // offset hard-coded to 0 since we only put one thing in a sector for now - return sid, 0, nil -} - -func (m *Sealing) SealPiece(ctx context.Context, size abi.UnpaddedPieceSize, r io.Reader, sectorID abi.SectorNumber, d DealInfo) error { - log.Infof("Seal piece for deal %d", d.DealID) - - ppi, err := m.sealer.AddPiece(sectorstorage.WithPriority(ctx, DealSectorPriority), m.minerSector(sectorID), []abi.UnpaddedPieceSize{}, size, r) + ppi, err := m.sealer.AddPiece(sectorstorage.WithPriority(ctx, DealSectorPriority), m.minerSector(sid), []abi.UnpaddedPieceSize{}, size, r) if err != nil { - return xerrors.Errorf("writing piece: %w", err) + return 0, 0, xerrors.Errorf("writing piece: %w", err) } - err = m.newSector(sectorID) + err = m.newSector(sid) if err != nil { - return xerrors.Errorf("creating new sector: %w", err) + return 0, 0, xerrors.Errorf("creating new sector: %w", err) } - err = m.addPiece(sectorID, Piece{ + err = m.addPiece(sid, Piece{ Piece: ppi, DealInfo: &d, }) if err != nil { - return xerrors.Errorf("adding piece to sector: %w", err) + return 0, 0, xerrors.Errorf("adding piece to sector: %w", err) } + + // offset hard-coded to 0 since we only put one thing in a sector for now + return sid, 0, nil } func (m *Sealing) addPiece(sectorID abi.SectorNumber, piece Piece) error { From ff8894f2ecaec0eed10899c773fb0cb9c3d4f4ec Mon Sep 17 00:00:00 2001 From: Aayush Rajasekaran Date: Tue, 23 Jun 2020 17:47:53 -0400 Subject: [PATCH 7/9] Maintain a map of UnsealedSectorInfos --- sealing.go | 55 +++++++++++++++++++++++++++++++++++++++++++++--------- 1 file changed, 46 insertions(+), 9 deletions(-) diff --git a/sealing.go b/sealing.go index 8a0483ca3..c83fae71b 100644 --- a/sealing.go +++ b/sealing.go @@ -53,7 +53,14 @@ type Sealing struct { sc SectorIDCounter verif ffiwrapper.Verifier - pcp PreCommitPolicy + unsealedInfos map[abi.SectorNumber]UnsealedSectorInfo + pcp PreCommitPolicy +} + +type UnsealedSectorInfo struct { + // stored should always equal sum of pieceSizes + stored uint64 + pieceSizes []abi.UnpaddedPieceSize } func New(api SealingAPI, events Events, maddr address.Address, ds datastore.Batching, sealer sectorstorage.SectorManager, sc SectorIDCounter, verif ffiwrapper.Verifier, pcp PreCommitPolicy) *Sealing { @@ -61,11 +68,12 @@ func New(api SealingAPI, events Events, maddr address.Address, ds datastore.Batc api: api, events: events, - maddr: maddr, - sealer: sealer, - sc: sc, - verif: verif, - pcp: pcp, + maddr: maddr, + sealer: sealer, + sc: sc, + verif: verif, + unsealedInfos: make(map[abi.SectorNumber]UnsealedSectorInfo), + pcp: pcp, } s.sectors = statemachine.New(namespace.Wrap(ds, datastore.NewKey(SectorStorePrefix)), s, SectorInfo{}) @@ -130,7 +138,18 @@ func (m *Sealing) AddPieceToAnySector(ctx context.Context, size abi.UnpaddedPiec func (m *Sealing) addPiece(sectorID abi.SectorNumber, piece Piece) error { log.Infof("Adding piece to sector %d", sectorID) - return m.sectors.Send(uint64(sectorID), SectorAddPiece{NewPiece: piece}) + err := m.sectors.Send(uint64(sectorID), SectorAddPiece{NewPiece: piece}) + if err != nil { + return err + } + + ui := m.unsealedInfos[sectorID] + m.unsealedInfos[sectorID] = UnsealedSectorInfo{ + stored: ui.stored + uint64(piece.Piece.Size.Unpadded()), + pieceSizes: append(ui.pieceSizes, piece.Piece.Size.Unpadded()), + } + + return nil } func (m *Sealing) Remove(ctx context.Context, sid abi.SectorNumber) error { @@ -139,7 +158,14 @@ func (m *Sealing) Remove(ctx context.Context, sid abi.SectorNumber) error { func (m *Sealing) StartPacking(sectorID abi.SectorNumber) error { log.Infof("Starting packing sector %d", sectorID) - return m.sectors.Send(uint64(sectorID), SectorStartPacking{}) + err := m.sectors.Send(uint64(sectorID), SectorStartPacking{}) + if err != nil { + return err + } + + delete(m.unsealedInfos, sectorID) + + return nil } // newSector creates a new sector for deal storage @@ -150,10 +176,21 @@ func (m *Sealing) newSector(sid abi.SectorNumber) error { } log.Infof("Creating sector %d", sid) - return m.sectors.Send(uint64(sid), SectorStart{ + err = m.sectors.Send(uint64(sid), SectorStart{ ID: sid, SectorType: rt, }) + + if err != nil { + return err + } + + m.unsealedInfos[sid] = UnsealedSectorInfo{ + stored: 0, + pieceSizes: nil, + } + + return nil } // newSectorCC accepts a slice of pieces with no deal (junk data) From 81a2792e37652d3d1d8d2687d4c969e01c812cf7 Mon Sep 17 00:00:00 2001 From: Aayush Rajasekaran Date: Tue, 23 Jun 2020 19:19:16 -0400 Subject: [PATCH 8/9] Move sector number calculation into newSector --- sealing.go | 32 ++++++++++++++++---------------- 1 file changed, 16 insertions(+), 16 deletions(-) diff --git a/sealing.go b/sealing.go index c83fae71b..bed821fc7 100644 --- a/sealing.go +++ b/sealing.go @@ -103,14 +103,9 @@ func (m *Sealing) AddPieceToAnySector(ctx context.Context, size abi.UnpaddedPiec return 0, 0, xerrors.Errorf("piece cannot fit into a sector") } - sid, err := m.sc.Next() + sid, err := m.newSector() // TODO: Put more than one thing in a sector if err != nil { - return 0, 0, xerrors.Errorf("getting sector number: %w", err) - } - - err = m.sealer.NewSector(context.TODO(), m.minerSector(sid)) // TODO: Put more than one thing in a sector - if err != nil { - return 0, 0, xerrors.Errorf("initializing sector: %w", err) + return 0, 0, xerrors.Errorf("creating new sector: %w", err) } ppi, err := m.sealer.AddPiece(sectorstorage.WithPriority(ctx, DealSectorPriority), m.minerSector(sid), []abi.UnpaddedPieceSize{}, size, r) @@ -118,11 +113,6 @@ func (m *Sealing) AddPieceToAnySector(ctx context.Context, size abi.UnpaddedPiec return 0, 0, xerrors.Errorf("writing piece: %w", err) } - err = m.newSector(sid) - if err != nil { - return 0, 0, xerrors.Errorf("creating new sector: %w", err) - } - err = m.addPiece(sid, Piece{ Piece: ppi, DealInfo: &d, @@ -169,10 +159,20 @@ func (m *Sealing) StartPacking(sectorID abi.SectorNumber) error { } // newSector creates a new sector for deal storage -func (m *Sealing) newSector(sid abi.SectorNumber) error { +func (m *Sealing) newSector() (abi.SectorNumber, error) { + sid, err := m.sc.Next() + if err != nil { + return 0, xerrors.Errorf("getting sector number: %w", err) + } + + err = m.sealer.NewSector(context.TODO(), m.minerSector(sid)) + if err != nil { + return 0, xerrors.Errorf("initializing sector: %w", err) + } + rt, err := ffiwrapper.SealProofTypeFromSectorSize(m.sealer.SectorSize()) if err != nil { - return xerrors.Errorf("bad sector size: %w", err) + return 0, xerrors.Errorf("bad sector size: %w", err) } log.Infof("Creating sector %d", sid) @@ -182,7 +182,7 @@ func (m *Sealing) newSector(sid abi.SectorNumber) error { }) if err != nil { - return err + return 0, xerrors.Errorf("starting the sector fsm: %w", err) } m.unsealedInfos[sid] = UnsealedSectorInfo{ @@ -190,7 +190,7 @@ func (m *Sealing) newSector(sid abi.SectorNumber) error { pieceSizes: nil, } - return nil + return sid, nil } // newSectorCC accepts a slice of pieces with no deal (junk data) From a3c686d317b363533c62ac6f3f9cbb53ca8a80d9 Mon Sep 17 00:00:00 2001 From: Aayush Rajasekaran Date: Tue, 23 Jun 2020 19:21:50 -0400 Subject: [PATCH 9/9] Try to pack deals into existing unsealed sectors before creating new ones --- sealing.go | 22 ++++++++++++++++++---- 1 file changed, 18 insertions(+), 4 deletions(-) diff --git a/sealing.go b/sealing.go index bed821fc7..9eea593a2 100644 --- a/sealing.go +++ b/sealing.go @@ -103,12 +103,13 @@ func (m *Sealing) AddPieceToAnySector(ctx context.Context, size abi.UnpaddedPiec return 0, 0, xerrors.Errorf("piece cannot fit into a sector") } - sid, err := m.newSector() // TODO: Put more than one thing in a sector + sid, err := m.getAvailableSector(size) if err != nil { return 0, 0, xerrors.Errorf("creating new sector: %w", err) } - ppi, err := m.sealer.AddPiece(sectorstorage.WithPriority(ctx, DealSectorPriority), m.minerSector(sid), []abi.UnpaddedPieceSize{}, size, r) + offset := m.unsealedInfos[sid].stored + ppi, err := m.sealer.AddPiece(sectorstorage.WithPriority(ctx, DealSectorPriority), m.minerSector(sid), m.unsealedInfos[sid].pieceSizes, size, r) if err != nil { return 0, 0, xerrors.Errorf("writing piece: %w", err) } @@ -122,8 +123,7 @@ func (m *Sealing) AddPieceToAnySector(ctx context.Context, size abi.UnpaddedPiec return 0, 0, xerrors.Errorf("adding piece to sector: %w", err) } - // offset hard-coded to 0 since we only put one thing in a sector for now - return sid, 0, nil + return sid, offset, nil } func (m *Sealing) addPiece(sectorID abi.SectorNumber, piece Piece) error { @@ -158,6 +158,20 @@ func (m *Sealing) StartPacking(sectorID abi.SectorNumber) error { return nil } +func (m *Sealing) getAvailableSector(size abi.UnpaddedPieceSize) (abi.SectorNumber, error) { + ss := m.sealer.SectorSize() + for k, v := range m.unsealedInfos { + if v.stored+uint64(size) <= uint64(ss) { + // TODO: Support multiple deal sizes in the same sector + if len(v.pieceSizes) == 0 || v.pieceSizes[0] == size { + return k, nil + } + } + } + + return m.newSector() +} + // newSector creates a new sector for deal storage func (m *Sealing) newSector() (abi.SectorNumber, error) { sid, err := m.sc.Next()