package store import ( "context" "encoding/binary" "errors" "sync" blocks "github.com/ipfs/go-block-format" cid "github.com/ipfs/go-cid" dstore "github.com/ipfs/go-datastore" "github.com/filecoin-project/go-state-types/abi" "github.com/filecoin-project/lotus/build" "github.com/filecoin-project/lotus/chain/types" bstore "github.com/filecoin-project/lotus/lib/blockstore" ) const CompactionThreshold = 5 * build.Finality var baseEpochKey = dstore.NewKey("baseEpoch") type SplitStore struct { baseEpoch abi.ChainEpoch curTs *types.TipSet cs *ChainStore ds dstore.Datastore hot bstore.Blockstore cold bstore.Blockstore snoop TrackingStore stateMx sync.Mutex compacting bool } type TrackingStore interface { Put(cid.Cid, abi.ChainEpoch) error PutBatch([]cid.Cid, abi.ChainEpoch) error Get(cid.Cid) (abi.ChainEpoch, error) Delete(cid.Cid) error Keys() (<-chan cid.Cid, error) } type LiveSet interface { Mark(cid.Cid) error Has(cid.Cid) (bool, error) Close() error } var _ bstore.Blockstore = (*SplitStore)(nil) // Blockstore interface func (s *SplitStore) DeleteBlock(cid cid.Cid) error { // afaict we don't seem to be using this method, so it's not implemented return errors.New("DeleteBlock not implemented on SplitStore; don't do this Luke!") //nolint } func (s *SplitStore) Has(cid cid.Cid) (bool, error) { has, err := s.hot.Has(cid) if err != nil { return false, err } if has { return true, nil } return s.cold.Has(cid) } func (s *SplitStore) Get(cid cid.Cid) (blocks.Block, error) { blk, err := s.hot.Get(cid) switch err { case nil: return blk, nil case bstore.ErrNotFound: return s.cold.Get(cid) default: return nil, err } } func (s *SplitStore) GetSize(cid cid.Cid) (int, error) { size, err := s.hot.GetSize(cid) switch err { case nil: return size, nil case bstore.ErrNotFound: return s.cold.GetSize(cid) default: return 0, err } } func (s *SplitStore) Put(blk blocks.Block) error { epoch := s.curTs.Height() err := s.snoop.Put(blk.Cid(), epoch) if err != nil { return err } return s.hot.Put(blk) } func (s *SplitStore) PutMany(blks []blocks.Block) error { err := s.hot.PutMany(blks) if err != nil { return err } epoch := s.curTs.Height() batch := make([]cid.Cid, 0, len(blks)) for _, blk := range blks { batch = append(batch, blk.Cid()) } return s.snoop.PutBatch(batch, epoch) } func (s *SplitStore) AllKeysChan(ctx context.Context) (<-chan cid.Cid, error) { ctx, cancel := context.WithCancel(ctx) chHot, err := s.hot.AllKeysChan(ctx) if err != nil { cancel() return nil, err } chCold, err := s.cold.AllKeysChan(ctx) if err != nil { cancel() return nil, err } ch := make(chan cid.Cid) go func() { defer cancel() defer close(ch) for _, in := range []<-chan cid.Cid{chHot, chCold} { for cid := range in { select { case ch <- cid: case <-ctx.Done(): return } } } }() return ch, nil } func (s *SplitStore) HashOnRead(enabled bool) { s.hot.HashOnRead(enabled) s.cold.HashOnRead(enabled) } func (s *SplitStore) View(cid cid.Cid, cb func([]byte) error) error { err := s.hot.View(cid, cb) switch err { case bstore.ErrNotFound: return s.cold.View(cid, cb) default: return err } } // State tracking func (s *SplitStore) Start(cs *ChainStore) error { s.cs = cs s.curTs = cs.GetHeaviestTipSet() // load base epoch from metadata ds // if none, then use current epoch because it's a fresh start bs, err := s.ds.Get(baseEpochKey) switch err { case nil: epoch, n := binary.Uvarint(bs) if n < 0 { panic("bogus base epoch") } s.baseEpoch = abi.ChainEpoch(epoch) case dstore.ErrNotFound: err = s.setBaseEpoch(s.curTs.Height()) if err != nil { return err } default: return err } // watch the chain cs.SubscribeHeadChanges(s.HeadChange) return nil } func (s *SplitStore) HeadChange(revert, apply []*types.TipSet) error { s.curTs = apply[len(apply)-1] epoch := s.curTs.Height() if epoch-s.baseEpoch > CompactionThreshold && !s.isCompacting() { s.setCompacting(true) go func() { defer s.setCompacting(false) s.compact() }() } return nil } func (s *SplitStore) isCompacting() bool { s.stateMx.Lock() defer s.stateMx.Unlock() return s.compacting } func (s *SplitStore) setCompacting(state bool) { s.stateMx.Lock() defer s.stateMx.Unlock() s.compacting = state } // Compaction/GC Algorithm func (s *SplitStore) compact() { // create two on disk live sets, one for marking the cold finality region // and one for marking the hot region hotSet, err := s.newLiveSet() if err != nil { // TODO do something better here panic(err) } defer hotSet.Close() //nolint:errcheck coldSet, err := s.newLiveSet() if err != nil { // TODO do something better here panic(err) } defer coldSet.Close() //nolint:errcheck // Phase 1a: mark all reachable CIDs in the hot range curTs := s.curTs epoch := curTs.Height() coldEpoch := s.baseEpoch + build.Finality err = s.cs.WalkSnapshot(context.Background(), curTs, epoch-coldEpoch+1, false, false, func(cid cid.Cid) error { return hotSet.Mark(cid) }) if err != nil { // TODO do something better here panic(err) } // Phase 1b: mark all reachable CIDs in the cold range coldTs, err := s.cs.GetTipsetByHeight(context.Background(), coldEpoch-1, curTs, true) if err != nil { // TODO do something better here panic(err) } err = s.cs.WalkSnapshot(context.Background(), coldTs, build.Finality, false, false, func(cid cid.Cid) error { return coldSet.Mark(cid) }) if err != nil { // TODO do something better here panic(err) } // Phase 2: sweep cold objects: // - If a cold object is reachable in the hot range, it stays in the hotstore. // - If a cold object is reachable in the cold range, it is moved to the coldstore. // - If a cold object is unreachable, it is deleted. ch, err := s.snoop.Keys() if err != nil { // TODO do something better here panic(err) } for cid := range ch { wrEpoch, err := s.snoop.Get(cid) if err != nil { // TODO do something better here panic(err) } // is the object stil hot? if wrEpoch >= coldEpoch { // yes, stay in the hotstore continue } // the object is cold -- check whether it is reachable in the hot range mark, err := hotSet.Has(cid) if err != nil { // TODO do something better here panic(err) } if mark { // the object is reachable in the hot range, stay in the hotstore continue } // check whether it is reachable in the cold range mark, err = coldSet.Has(cid) if err != nil { // TODO do something better here panic(err) } if mark { // the object is reachable in the cold range, move it to the cold store blk, err := s.hot.Get(cid) if err != nil { // TODO do something better here panic(err) } err = s.cold.Put(blk) if err != nil { // TODO do something better here panic(err) } } // delete the object from the hotstore err = s.hot.DeleteBlock(cid) if err != nil { // TODO do something better here panic(err) } // remove the snoop tracking err = s.snoop.Delete(cid) if err != nil { // TODO do something better here panic(err) } } err = s.setBaseEpoch(coldEpoch) if err != nil { // TODO do something better here panic(err) } } func (s *SplitStore) setBaseEpoch(epoch abi.ChainEpoch) error { s.baseEpoch = epoch // write to datastore bs := make([]byte, 16) n := binary.PutUvarint(bs, uint64(epoch)) bs = bs[:n] return s.ds.Put(baseEpochKey, bs) } func (s *SplitStore) newLiveSet() (LiveSet, error) { // TODO implementation return nil, errors.New("newLiveSet: IMPLEMENT ME!!!") //nolint }