this repo has no description
make the carstore a bit faster
+226
-41
+71
-14
carstore/bs.go
+71
-14
carstore/bs.go
···
45
45
46
46
Root string
47
47
DataStart int64
48
-
Seq int
48
+
Seq int `gorm:"index"`
49
49
Path string
50
-
User uint
50
+
User uint `gorm:"index"`
51
51
}
52
52
53
53
type blockRef struct {
54
54
gorm.Model
55
-
Cid string
55
+
Cid string `gorm:"index"`
56
56
Shard uint
57
57
Offset int64
58
58
}
···
60
60
type userView struct {
61
61
cs *CarStore
62
62
user uint
63
+
64
+
cache map[cid.Cid]blocks.Block
65
+
prefetch bool
63
66
}
64
67
65
68
var _ blockstore.Blockstore = (*userView)(nil)
···
86
89
}
87
90
88
91
func (uv *userView) Get(ctx context.Context, k cid.Cid) (blocks.Block, error) {
92
+
if uv.cache != nil {
93
+
blk, ok := uv.cache[k]
94
+
if ok {
95
+
return blk, nil
96
+
}
97
+
}
98
+
89
99
// TODO: for now, im using a join to ensure we only query blocks from the
90
100
// correct user. maybe it makes sense to put the user in the blockRef
91
101
// directly? tradeoff of time vs space
···
106
116
return nil, err
107
117
}
108
118
109
-
fi, err := os.Open(info.Path)
119
+
if uv.prefetch {
120
+
return uv.prefetchRead(ctx, k, info.Path, info.Offset)
121
+
} else {
122
+
return uv.singleRead(ctx, k, info.Path, info.Offset)
123
+
}
124
+
}
125
+
126
+
func (uv *userView) prefetchRead(ctx context.Context, k cid.Cid, path string, offset int64) (blocks.Block, error) {
127
+
fi, err := os.Open(path)
128
+
if err != nil {
129
+
return nil, err
130
+
}
131
+
132
+
cr, err := car.NewCarReader(fi)
133
+
if err != nil {
134
+
return nil, err
135
+
}
136
+
137
+
for {
138
+
blk, err := cr.Next()
139
+
if err != nil {
140
+
if err == io.EOF {
141
+
break
142
+
}
143
+
return nil, err
144
+
}
145
+
146
+
uv.cache[blk.Cid()] = blk
147
+
}
148
+
149
+
outblk, ok := uv.cache[k]
150
+
if !ok {
151
+
return nil, fmt.Errorf("requested block was not found in car slice")
152
+
}
153
+
154
+
return outblk, nil
155
+
}
156
+
157
+
func (uv *userView) singleRead(ctx context.Context, k cid.Cid, path string, offset int64) (blocks.Block, error) {
158
+
fi, err := os.Open(path)
110
159
if err != nil {
111
160
return nil, err
112
161
}
113
162
114
-
seeked, err := fi.Seek(info.Offset, io.SeekStart)
163
+
seeked, err := fi.Seek(offset, io.SeekStart)
115
164
if err != nil {
116
165
return nil, err
117
166
}
118
167
119
-
if seeked != info.Offset {
120
-
return nil, fmt.Errorf("failed to seek to offset (%d != %d)", seeked, info.Offset)
168
+
if seeked != offset {
169
+
return nil, fmt.Errorf("failed to seek to offset (%d != %d)", seeked, offset)
121
170
}
122
171
123
172
bufr := bufio.NewReader(fi)
···
172
221
// TODO: ensure that we don't write updates on top of the wrong head
173
222
// this needs to be a compare and swap type operation
174
223
var lastShard CarShard
175
-
if err := cs.meta.Model(CarShard{}).Limit(1).Order("seq desc").Find(&lastShard).Error; err != nil {
224
+
if err := cs.meta.Model(CarShard{}).Limit(1).Order("id desc").Find(&lastShard, "user = ?", user).Error; err != nil {
225
+
//if err := cs.meta.Model(CarShard{}).Where("user = ?", user).Last(&lastShard).Error; err != nil {
226
+
//if err != gorm.ErrRecordNotFound {
176
227
return nil, err
228
+
//}
177
229
}
178
230
179
231
return &DeltaSession{
180
232
fresh: blockstore.NewBlockstore(datastore.NewMapDatastore()),
181
233
blks: make(map[cid.Cid]blocks.Block),
182
234
base: &userView{
183
-
user: user,
184
-
cs: cs,
235
+
user: user,
236
+
cs: cs,
237
+
prefetch: true,
238
+
cache: make(map[cid.Cid]blocks.Block),
185
239
},
186
240
user: user,
187
241
cs: cs,
···
362
416
}
363
417
364
418
var offset int64 = hnw
419
+
brefs := make([]*blockRef, 0, len(ds.blks))
365
420
for k, blk := range ds.blks {
366
421
nw, err := LdWrite(fi, k.Bytes(), blk.RawData())
367
422
if err != nil {
368
423
return err
369
424
}
370
425
371
-
if err := ds.cs.meta.Create(&blockRef{
426
+
brefs = append(brefs, &blockRef{
372
427
Cid: k.String(),
373
428
Offset: offset,
374
429
Shard: shard.ID,
375
-
}).Error; err != nil {
376
-
return err
377
-
}
430
+
})
378
431
379
432
offset += nw
433
+
}
434
+
435
+
if err := ds.cs.meta.Create(brefs).Error; err != nil {
436
+
return err
380
437
}
381
438
382
439
return nil
+149
-24
carstore/repo_test.go
+149
-24
carstore/repo_test.go
···
11
11
"time"
12
12
13
13
"github.com/ipfs/go-cid"
14
+
flatfs "github.com/ipfs/go-ds-flatfs"
15
+
blockstore "github.com/ipfs/go-ipfs-blockstore"
14
16
"github.com/whyrusleeping/gosky/api"
15
17
"github.com/whyrusleeping/gosky/repo"
16
18
"gorm.io/driver/sqlite"
17
19
"gorm.io/gorm"
18
20
)
19
21
20
-
func TestBasicOperation(t *testing.T) {
22
+
func testCarStore() (*CarStore, func(), error) {
21
23
tempdir, err := ioutil.TempDir("", "msttest-")
22
24
if err != nil {
23
-
t.Fatal(err)
25
+
return nil, nil, err
24
26
}
25
27
26
-
defer func() {
27
-
_ = os.RemoveAll(tempdir)
28
-
}()
29
-
30
28
sharddir := filepath.Join(tempdir, "shards")
31
29
if err := os.MkdirAll(sharddir, 0775); err != nil {
32
-
t.Fatal(err)
30
+
return nil, nil, err
33
31
}
34
32
35
-
db, err := gorm.Open(sqlite.Open(filepath.Join(tempdir, "foo.db")))
33
+
dbstr := "file::memory:"
34
+
db, err := gorm.Open(sqlite.Open(dbstr))
36
35
if err != nil {
37
-
t.Fatal(err)
36
+
return nil, nil, err
38
37
}
39
38
40
39
cs, err := NewCarStore(db, sharddir)
41
40
if err != nil {
42
-
t.Fatal(err)
41
+
return nil, nil, err
43
42
}
44
43
45
-
ds, err := cs.NewDeltaSession(1, cid.Undef)
44
+
return cs, func() {
45
+
_ = os.RemoveAll(tempdir)
46
+
}, nil
47
+
}
48
+
49
+
func testFlatfsBs() (blockstore.Blockstore, func(), error) {
50
+
tempdir, err := ioutil.TempDir("", "msttest-")
46
51
if err != nil {
47
-
t.Fatal(err)
52
+
return nil, nil, err
48
53
}
49
54
55
+
ffds, err := flatfs.CreateOrOpen(tempdir, flatfs.IPFS_DEF_SHARD, true)
56
+
if err != nil {
57
+
return nil, nil, err
58
+
}
59
+
60
+
bs := blockstore.NewBlockstoreNoPrefix(ffds)
61
+
62
+
return bs, func() {
63
+
_ = os.RemoveAll(tempdir)
64
+
}, nil
65
+
}
66
+
67
+
func TestBasicOperation(t *testing.T) {
50
68
ctx := context.TODO()
51
69
52
-
nr := repo.NewRepo(ctx, ds)
53
-
54
-
if err := nr.CreateRecord(ctx, "app.bsky.feed.post", &api.PostRecord{
55
-
Text: fmt.Sprintf("hey look its a tweet %s", time.Now()),
56
-
}); err != nil {
70
+
cs, cleanup, err := testCarStore()
71
+
if err != nil {
57
72
t.Fatal(err)
58
73
}
74
+
defer cleanup()
59
75
60
-
ncid, err := nr.Commit(ctx)
76
+
ds, err := cs.NewDeltaSession(1, cid.Undef)
61
77
if err != nil {
62
78
t.Fatal(err)
63
79
}
64
80
65
-
if err := ds.CloseWithRoot(ctx, ncid); err != nil {
81
+
ncid, err := setupRepo(ctx, ds)
82
+
if err != nil {
66
83
t.Fatal(err)
67
84
}
68
85
69
-
buf := new(bytes.Buffer)
70
-
if err := cs.ReadUserCar(ctx, 1, cid.Undef, true, buf); err != nil {
86
+
if err := ds.CloseWithRoot(ctx, ncid); err != nil {
71
87
t.Fatal(err)
72
88
}
73
-
fmt.Println(buf.Len())
74
89
75
90
head := ncid
76
91
for i := 0; i < 10; i++ {
77
-
fmt.Println("head: ", head)
78
92
ds, err := cs.NewDeltaSession(1, head)
79
93
if err != nil {
80
94
t.Fatal(err)
···
103
117
head = nroot
104
118
}
105
119
106
-
buf = new(bytes.Buffer)
120
+
buf := new(bytes.Buffer)
107
121
if err := cs.ReadUserCar(ctx, 1, cid.Undef, true, buf); err != nil {
108
122
t.Fatal(err)
109
123
}
110
124
111
125
fmt.Println(buf.Len())
126
+
127
+
}
128
+
129
+
func setupRepo(ctx context.Context, bs blockstore.Blockstore) (cid.Cid, error) {
130
+
nr := repo.NewRepo(ctx, bs)
131
+
132
+
if err := nr.CreateRecord(ctx, "app.bsky.feed.post", &api.PostRecord{
133
+
Text: fmt.Sprintf("hey look its a tweet %s", time.Now()),
134
+
}); err != nil {
135
+
return cid.Undef, err
136
+
}
137
+
138
+
ncid, err := nr.Commit(ctx)
139
+
if err != nil {
140
+
return cid.Undef, err
141
+
}
142
+
143
+
return ncid, nil
144
+
}
145
+
146
+
func BenchmarkRepoWritesCarstore(b *testing.B) {
147
+
ctx := context.TODO()
148
+
149
+
cs, cleanup, err := testCarStore()
150
+
if err != nil {
151
+
b.Fatal(err)
152
+
}
153
+
defer cleanup()
154
+
155
+
ds, err := cs.NewDeltaSession(1, cid.Undef)
156
+
if err != nil {
157
+
b.Fatal(err)
158
+
}
159
+
160
+
ncid, err := setupRepo(ctx, ds)
161
+
if err != nil {
162
+
b.Fatal(err)
163
+
}
164
+
165
+
if err := ds.CloseWithRoot(ctx, ncid); err != nil {
166
+
b.Fatal(err)
167
+
}
168
+
169
+
head := ncid
170
+
b.ResetTimer()
171
+
for i := 0; i < b.N; i++ {
172
+
ds, err := cs.NewDeltaSession(1, head)
173
+
if err != nil {
174
+
b.Fatal(err)
175
+
}
176
+
177
+
rr, err := repo.OpenRepo(ctx, ds, head)
178
+
if err != nil {
179
+
b.Fatal(err)
180
+
}
181
+
182
+
if err := rr.CreateRecord(ctx, "app.bsky.feed.post", &api.PostRecord{
183
+
Text: fmt.Sprintf("hey look its a tweet %s", time.Now()),
184
+
}); err != nil {
185
+
b.Fatal(err)
186
+
}
187
+
188
+
nroot, err := rr.Commit(ctx)
189
+
if err != nil {
190
+
b.Fatal(err)
191
+
}
192
+
193
+
if err := ds.CloseWithRoot(ctx, nroot); err != nil {
194
+
b.Fatal(err)
195
+
}
196
+
197
+
head = nroot
198
+
}
199
+
}
200
+
201
+
func BenchmarkRepoWritesFlatfs(b *testing.B) {
202
+
ctx := context.TODO()
203
+
204
+
bs, cleanup, err := testFlatfsBs()
205
+
if err != nil {
206
+
b.Fatal(err)
207
+
}
208
+
defer cleanup()
209
+
210
+
ncid, err := setupRepo(ctx, bs)
211
+
if err != nil {
212
+
b.Fatal(err)
213
+
}
214
+
215
+
head := ncid
216
+
b.ResetTimer()
217
+
for i := 0; i < b.N; i++ {
218
+
219
+
rr, err := repo.OpenRepo(ctx, bs, head)
220
+
if err != nil {
221
+
b.Fatal(err)
222
+
}
223
+
224
+
if err := rr.CreateRecord(ctx, "app.bsky.feed.post", &api.PostRecord{
225
+
Text: fmt.Sprintf("hey look its a tweet %s", time.Now()),
226
+
}); err != nil {
227
+
b.Fatal(err)
228
+
}
229
+
230
+
nroot, err := rr.Commit(ctx)
231
+
if err != nil {
232
+
b.Fatal(err)
233
+
}
234
+
235
+
head = nroot
236
+
}
112
237
}
+2
go.mod
+2
go.mod
···
7
7
github.com/ipfs/go-car v0.0.4
8
8
github.com/ipfs/go-cid v0.1.0
9
9
github.com/ipfs/go-datastore v0.5.1
10
+
github.com/ipfs/go-ds-flatfs v0.5.1
10
11
github.com/ipfs/go-ipfs-blockstore v1.2.1-0.20220823165003-9904c18f1d0a
11
12
github.com/ipfs/go-ipld-cbor v0.0.6
12
13
github.com/ipfs/go-ipld-format v0.3.0
···
22
23
)
23
24
24
25
require (
26
+
github.com/alexbrainman/goissue34681 v0.0.0-20191006012335-3fc7a47baff5 // indirect
25
27
github.com/cpuguy83/go-md2man/v2 v2.0.2 // indirect
26
28
github.com/decred/dcrd/dcrec/secp256k1/v4 v4.0.1 // indirect
27
29
github.com/goccy/go-json v0.9.7 // indirect
+4
-2
go.sum
+4
-2
go.sum
···
25
25
github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
26
26
github.com/alecthomas/units v0.0.0-20190717042225-c3de453c63f4/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
27
27
github.com/alecthomas/units v0.0.0-20190924025748-f65c72e2690d/go.mod h1:rBZYJk541a8SKzHPHnH3zbiI+7dagKZ0cgpgrD7Fyho=
28
+
github.com/alexbrainman/goissue34681 v0.0.0-20191006012335-3fc7a47baff5 h1:iW0a5ljuFxkLGPNem5Ui+KBjFJzKg4Fv2fnxe4dvzpM=
29
+
github.com/alexbrainman/goissue34681 v0.0.0-20191006012335-3fc7a47baff5/go.mod h1:Y2QMoi1vgtOIfc+6DhrMOGkLoGzqSV2rKp4Sm+opsyA=
28
30
github.com/anmitsu/go-shlex v0.0.0-20161002113705-648efa622239/go.mod h1:2FmKhYUyUczH0OGQWaF5ceTx0UBShxjsH6f8oGKYe2c=
29
31
github.com/apache/thrift v0.12.0/go.mod h1:cp2SuWMxlEZw2r+iP2GNCdIi4C1qmUzdZFSVb+bacwQ=
30
32
github.com/apache/thrift v0.13.0/go.mod h1:cp2SuWMxlEZw2r+iP2GNCdIi4C1qmUzdZFSVb+bacwQ=
···
289
291
github.com/ipfs/go-ds-badger v0.0.5/go.mod h1:g5AuuCGmr7efyzQhLL8MzwqcauPojGPUaHzfGTzuE3s=
290
292
github.com/ipfs/go-ds-badger v0.2.1/go.mod h1:Tx7l3aTph3FMFrRS838dcSJh+jjA7cX9DrGVwx/NOwE=
291
293
github.com/ipfs/go-ds-badger v0.2.3/go.mod h1:pEYw0rgg3FIrywKKnL+Snr+w/LjJZVMTBRn4FS6UHUk=
294
+
github.com/ipfs/go-ds-flatfs v0.5.1 h1:ZCIO/kQOS/PSh3vcF1H6a8fkRGS7pOfwfPdx4n/KJH4=
295
+
github.com/ipfs/go-ds-flatfs v0.5.1/go.mod h1:RWTV7oZD/yZYBKdbVIFXTX2fdY2Tbvl94NsWqmoyAX4=
292
296
github.com/ipfs/go-ds-leveldb v0.0.1/go.mod h1:feO8V3kubwsEF22n0YRQCffeb79OOYIykR4L04tMOYc=
293
297
github.com/ipfs/go-ds-leveldb v0.4.1/go.mod h1:jpbku/YqBSsBc1qgME8BkWS4AxzF2cEu1Ii2r79Hh9s=
294
298
github.com/ipfs/go-ds-leveldb v0.4.2/go.mod h1:jpbku/YqBSsBc1qgME8BkWS4AxzF2cEu1Ii2r79Hh9s=
···
921
925
github.com/warpfork/go-wish v0.0.0-20200122115046-b9ea61034e4a/go.mod h1:x6AKhvSSexNrVSrViXSHUEbICjmGXhtgABaHIySUSGw=
922
926
github.com/whyrusleeping/cbor v0.0.0-20171005072247-63513f603b11 h1:5HZfQkwe0mIfyDmc1Em5GqlNRzcdtlv4HTNmdpt7XH0=
923
927
github.com/whyrusleeping/cbor-gen v0.0.0-20200123233031-1cdf64d27158/go.mod h1:Xj/M2wWU+QdTdRbu/L/1dIZY8/Wb2K9pAhtroQuxJJI=
924
-
github.com/whyrusleeping/cbor-gen v0.0.0-20221107235327-d52321526af2 h1:uKKSmHNmtxSf6+3MDfYmU7HYiV5q3DgIKjDRPkc5OIU=
925
-
github.com/whyrusleeping/cbor-gen v0.0.0-20221107235327-d52321526af2/go.mod h1:fgkXqYy7bV2cFeIEOkVTZS/WjXARfBqSH6Q2qHL33hQ=
926
928
github.com/whyrusleeping/cbor-gen v0.0.0-20221120232258-33755d0abe95 h1:f4dFX0UniZ3T3TswZsQJ82hJUII2cQOcNVZsMTxus3g=
927
929
github.com/whyrusleeping/cbor-gen v0.0.0-20221120232258-33755d0abe95/go.mod h1:fgkXqYy7bV2cFeIEOkVTZS/WjXARfBqSH6Q2qHL33hQ=
928
930
github.com/whyrusleeping/chunker v0.0.0-20181014151217-fe64bd25879f h1:jQa4QT2UP9WYv2nzyawpKMOCl+Z/jW7djv2/J50lj9E=