Skip to content

Commit 2e41814

Browse files
LinZexiaoLinZexiao
committed
feat: walk state for 4 finality
1 parent dc9d2be commit 2e41814

14 files changed

+1589
-332
lines changed

‎main.go

+4
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,10 @@ func main() {
3131
_ = logging.SetLogLevel("pubsub", "error")
3232
_ = logging.SetLogLevel("relay", "error")
3333
_ = logging.SetLogLevel("dht/RtRefreshManager", "error")
34+
// todo: remove it
35+
_ = logging.SetLogLevel("splitstore", "debug")
36+
_ = logging.SetLogLevel("chainsync.syncer", "debug")
37+
3438
} else {
3539
level, err := logging.LevelFromString(lvl)
3640
if err != nil {

‎pkg/config/config.go

+6-5
Original file line numberDiff line numberDiff line change
@@ -79,10 +79,11 @@ func newDefaultAPIConfig() *APIConfig {
7979
// DatastoreConfig holds all the configuration options for the datastore.
8080
// TODO: use the advanced datastore configuration from ipfs
8181
type DatastoreConfig struct {
82-
Type string `json:"type"`
83-
Path string `json:"path"`
84-
SplitstoreSize int64 `json:"splitstore_size"`
85-
SplitstoreCount int `json:"splitstore_count"`
82+
Type string `json:"type"`
83+
Path string `json:"path"`
84+
SplitstoreSize int64 `json:"splitstoreSize"`
85+
SplitstoreCount int `json:"splitstoreCount"`
86+
SplitstoreInitProtectEpoch int64 `json:"splitstoreInitProtectEpoch"`
8687
}
8788

8889
// Validators hold the list of validation functions for each configuration
@@ -98,7 +99,7 @@ func newDefaultDatastoreConfig() *DatastoreConfig {
9899
return &DatastoreConfig{
99100
Type: "badgerds",
100101
Path: "badger",
101-
SplitstoreSize: int64(3 * policy.ChainFinality),
102+
SplitstoreSize: int64(5 * policy.ChainFinality),
102103
SplitstoreCount: 3,
103104
}
104105
}

‎pkg/consensus/processor.go

+7-1
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@ import (
1111
"github.com/filecoin-project/venus/pkg/fvm"
1212
"github.com/filecoin-project/venus/pkg/vm/vmcontext"
1313
"github.com/filecoin-project/venus/venus-shared/actors/builtin/reward"
14+
"github.com/filecoin-project/venus/venus-shared/blockstore"
1415

1516
"github.com/filecoin-project/go-address"
1617
amt4 "github.com/filecoin-project/go-amt-ipld/v4"
@@ -95,6 +96,11 @@ func (p *DefaultProcessor) ApplyBlocks(ctx context.Context,
9596
)
9697

9798
makeVM := func(base cid.Cid, e abi.ChainEpoch, timestamp uint64) (vm.Interface, error) {
99+
bs := vmOpts.Bsstore
100+
if ts.Height() == 1185554 {
101+
bs = blockstore.NewLogStore("./bs.log", bs)
102+
// _, _ = bs.Has(ctx, base)
103+
}
98104
vmOpt := vm.VmOption{
99105
CircSupplyCalculator: vmOpts.CircSupplyCalculator,
100106
LookbackStateGetter: vmOpts.LookbackStateGetter,
@@ -107,7 +113,7 @@ func (p *DefaultProcessor) ApplyBlocks(ctx context.Context,
107113
Timestamp: timestamp,
108114
GasPriceSchedule: vmOpts.GasPriceSchedule,
109115
PRoot: base,
110-
Bsstore: vmOpts.Bsstore,
116+
Bsstore: bs,
111117
SysCallsImpl: vmOpts.SysCallsImpl,
112118
TipSetGetter: vmOpts.TipSetGetter,
113119
Tracing: vmOpts.Tracing,

‎pkg/repo/fsrepo.go

+3-2
Original file line numberDiff line numberDiff line change
@@ -387,8 +387,9 @@ func (r *FSRepo) openDatastore() error {
387387

388388
ssPath := filepath.Join(r.path, splitstorePrefix)
389389
opt := splitstore.Option{
390-
MaxStoreCount: r.cfg.Datastore.SplitstoreCount,
391-
StoreSize: abi.ChainEpoch(r.cfg.Datastore.SplitstoreSize),
390+
MaxLayerCount: Config.Datastore.SplitstoreCount,
391+
LayerSize: abi.ChainEpoch(Config.Datastore.SplitstoreSize),
392+
InitSyncProtect: abi.ChainEpoch(Config.Datastore.SplitstoreInitProtectEpoch),
392393
}
393394
splitstore, err := splitstore.NewSplitstore(ssPath, ds, opt)
394395
if err != nil {

‎venus-shared/blockstore/log_store.go

+102
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,102 @@
1+
package blockstore
2+
3+
import (
4+
"context"
5+
llog "log"
6+
"os"
7+
8+
blocks "github.com/ipfs/go-block-format"
9+
cid "github.com/ipfs/go-cid"
10+
)
11+
12+
type LogStore struct {
13+
logger *llog.Logger
14+
bs Blockstore
15+
}
16+
17+
// DeleteMany implements Blockstore.
18+
func (l *LogStore) DeleteMany(ctx context.Context, cids []cid.Cid) error {
19+
for _, c := range cids {
20+
l.log(c, "delete", "")
21+
}
22+
return l.bs.DeleteMany(ctx, cids)
23+
}
24+
25+
// Flush implements Blockstore.
26+
func (l *LogStore) Flush(ctx context.Context) error {
27+
l.logger.Println("flush")
28+
return l.bs.Flush(ctx)
29+
}
30+
31+
// View implements Blockstore.
32+
func (l *LogStore) View(ctx context.Context, cid cid.Cid, callback func([]byte) error) error {
33+
l.log(cid, "view", "")
34+
return l.bs.View(ctx, cid, callback)
35+
}
36+
37+
// AllKeysChan implements blockstore.Blockstore.
38+
func (l *LogStore) AllKeysChan(ctx context.Context) (<-chan cid.Cid, error) {
39+
return l.AllKeysChan(ctx)
40+
}
41+
42+
// DeleteBlock implements blockstore.Blockstore.
43+
func (l *LogStore) DeleteBlock(ctx context.Context, c cid.Cid) error {
44+
l.log(c, "delete", "")
45+
return l.bs.DeleteBlock(ctx, c)
46+
}
47+
48+
// Get implements blockstore.Blockstore.
49+
func (l *LogStore) Get(ctx context.Context, c cid.Cid) (blocks.Block, error) {
50+
l.log(c, "get", "")
51+
return l.bs.Get(ctx, c)
52+
}
53+
54+
// GetSize implements blockstore.Blockstore.
55+
func (l *LogStore) GetSize(ctx context.Context, c cid.Cid) (int, error) {
56+
l.log(c, "getsize", "")
57+
return l.bs.GetSize(ctx, c)
58+
}
59+
60+
// Has implements blockstore.Blockstore.
61+
func (l *LogStore) Has(ctx context.Context, c cid.Cid) (bool, error) {
62+
l.log(c, "has", "")
63+
return l.bs.Has(ctx, c)
64+
}
65+
66+
// HashOnRead implements blockstore.Blockstore.
67+
func (l *LogStore) HashOnRead(enabled bool) {
68+
l.bs.HashOnRead(enabled)
69+
}
70+
71+
// Put implements blockstore.Blockstore.
72+
func (l *LogStore) Put(ctx context.Context, b blocks.Block) error {
73+
l.log(b.Cid(), "put", "")
74+
return l.bs.Put(ctx, b)
75+
}
76+
77+
// DeleteMany implements blockstore.Blockstore.
78+
func (l *LogStore) PutMany(ctx context.Context, bs []blocks.Block) error {
79+
for _, b := range bs {
80+
l.log(b.Cid(), "put", "")
81+
}
82+
return l.bs.PutMany(ctx, bs)
83+
}
84+
85+
var _ Blockstore = (*LogStore)(nil)
86+
87+
func NewLogStore(path string, bs Blockstore) *LogStore {
88+
file, err := os.OpenFile(path, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0666)
89+
if err != nil {
90+
panic(err)
91+
}
92+
logger := llog.New(file, "", llog.LstdFlags)
93+
logger.Println("log store opened")
94+
return &LogStore{
95+
logger: logger,
96+
bs: bs,
97+
}
98+
}
99+
100+
func (l *LogStore) log(c cid.Cid, op, msg string) {
101+
l.logger.Printf("%s %s %s", c.String(), op, msg)
102+
}

‎venus-shared/blockstore/splitstore/compose_store.go

+36-7
Original file line numberDiff line numberDiff line change
@@ -18,23 +18,48 @@ var log = logging.New("splitstore")
1818
// Write: write to primary store only
1919
// Delete: Delete from all store
2020
type ComposeStore struct {
21-
primary blockstore.Blockstore
22-
secondary blockstore.Blockstore
21+
shouldSync bool
22+
primary blockstore.Blockstore
23+
secondary blockstore.Blockstore
2324
}
2425

2526
// NewComposeStore create a new ComposeStore with a list of blockstore
2627
// low priority come first
2728
func NewComposeStore(bs ...blockstore.Blockstore) blockstore.Blockstore {
28-
if len(bs) == 0 {
29+
switch len(bs) {
30+
case 0:
2931
return nil
32+
case 1:
33+
return bs[0]
3034
}
31-
ret := bs[0]
32-
for i := 1; i < len(bs); i++ {
35+
return Compose(bs...)
36+
}
37+
38+
func Compose(bs ...blockstore.Blockstore) *ComposeStore {
39+
switch len(bs) {
40+
case 0:
41+
return nil
42+
case 1:
43+
return &ComposeStore{
44+
shouldSync: false,
45+
primary: bs[0],
46+
secondary: bs[0],
47+
}
48+
}
49+
50+
ret := &ComposeStore{
51+
shouldSync: false,
52+
primary: bs[1],
53+
secondary: bs[0],
54+
}
55+
for i := 2; i < len(bs); i++ {
3356
ret = &ComposeStore{
34-
primary: bs[i],
35-
secondary: ret,
57+
shouldSync: i == len(bs)-1,
58+
primary: bs[i],
59+
secondary: ret,
3660
}
3761
}
62+
3863
return ret
3964
}
4065

@@ -187,6 +212,10 @@ func (cs *ComposeStore) View(ctx context.Context, c cid.Cid, cb func([]byte) err
187212

188213
// sync sync block from secondly to primary
189214
func (cs *ComposeStore) sync(ctx context.Context, c cid.Cid, b blocks.Block) {
215+
if !cs.shouldSync {
216+
return
217+
}
218+
190219
go func() {
191220
select {
192221
case <-ctx.Done():

‎venus-shared/blockstore/splitstore/compose_store_test.go

+10-14
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,16 @@ import (
1313
"github.com/stretchr/testify/require"
1414
)
1515

16+
func TestNewComposeStore(t *testing.T) {
17+
s := NewComposeStore(nil, nil)
18+
require.True(t, s.(*ComposeStore).shouldSync)
19+
20+
s = NewComposeStore(nil, nil, nil)
21+
require.True(t, s.(*ComposeStore).shouldSync)
22+
require.False(t, s.(*ComposeStore).secondary.(*ComposeStore).shouldSync)
23+
require.Nil(t, s.(*ComposeStore).secondary.(*ComposeStore).secondary)
24+
}
25+
1626
func TestComposeStoreGet(t *testing.T) {
1727
ctx := context.Background()
1828
composeStore, primaryStore, secondaryStore, tertiaryStore := getBlockstore(t)
@@ -340,20 +350,6 @@ func TestComposeStoreDelete(t *testing.T) {
340350
})
341351
}
342352

343-
func TestNewComposeStore(t *testing.T) {
344-
tempDir := t.TempDir()
345-
346-
primaryPath := filepath.Join(tempDir, "primary")
347-
optPri, err := blockstore.BadgerBlockstoreOptions(primaryPath, false)
348-
require.NoError(t, err)
349-
dsPri, err := blockstore.Open(optPri)
350-
require.NoError(t, err)
351-
352-
cs := NewComposeStore(dsPri)
353-
_, err = cs.Has(context.Background(), cid.Undef)
354-
require.NoError(t, err)
355-
}
356-
357353
func getBlockstore(t *testing.T) (compose, primary, secondary, tertiary blockstore.Blockstore) {
358354
tempDir := t.TempDir()
359355

0 commit comments

Comments
 (0)