this repo has no description
some more memory tuning
authored by
whyrusleeping
and committed by
whyrusleeping
e13676ef
668afaa2
+47
-8
+38
-3
repo/carutil/reader.go
+38
-3
repo/carutil/reader.go
···
8
8
"io"
9
9
"sync"
10
10
11
-
blocks "github.com/ipfs/go-block-format"
12
11
"github.com/ipfs/go-cid"
13
12
car "github.com/ipld/go-car"
14
13
)
···
38
37
39
38
const MaxAllowedSectionSize = 32 << 20
40
39
41
-
func (r *Reader) NextBlock(allocator *sync.Pool, allocMax uint64) (blocks.Block, error) {
40
+
func (r *Reader) NextBlock(allocator *sync.Pool, allocMax uint64) (*BasicBlock, error) {
42
41
data, err := ldRead(r.r, allocator, allocMax)
43
42
if err != nil {
44
43
return nil, err
···
49
48
return nil, err
50
49
}
51
50
52
-
return blocks.NewBlockWithCid(data[n:], c)
51
+
return NewBlockWithCid(data[n:], data, c), nil
53
52
}
54
53
55
54
func ldRead(r *bufio.Reader, alloc *sync.Pool, allocMax uint64) ([]byte, error) {
···
88
87
89
88
return buf, nil
90
89
}
90
+
91
+
type BasicBlock struct {
92
+
cid cid.Cid
93
+
data []byte
94
+
base []byte
95
+
}
96
+
97
+
func NewBlockWithCid(data, base []byte, c cid.Cid) *BasicBlock {
98
+
return &BasicBlock{data: data, cid: c, base: base}
99
+
}
100
+
101
+
// RawData returns the block raw contents as a byte slice.
102
+
func (b *BasicBlock) RawData() []byte {
103
+
return b.data
104
+
}
105
+
106
+
// Cid returns the content identifier of the block.
107
+
func (b *BasicBlock) Cid() cid.Cid {
108
+
return b.cid
109
+
}
110
+
111
+
// String provides a human-readable representation of the block CID.
112
+
func (b *BasicBlock) String() string {
113
+
return fmt.Sprintf("[Block %s]", b.Cid())
114
+
}
115
+
116
+
// Loggable returns a go-log loggable item.
117
+
func (b *BasicBlock) Loggable() map[string]interface{} {
118
+
return map[string]interface{}{
119
+
"block": b.Cid().String(),
120
+
}
121
+
}
122
+
123
+
func (b *BasicBlock) BaseBuffer() []byte {
124
+
return b.base
125
+
}
+9
-5
repo/stream.go
+9
-5
repo/stream.go
···
16
16
)
17
17
18
18
type readStreamBlockstore struct {
19
-
otherBlocks map[cid.Cid]block.Block
19
+
otherBlocks map[cid.Cid]*carutil.BasicBlock
20
20
streamComplete bool
21
21
22
22
r *carutil.Reader
···
24
24
25
25
func newStreamingBlockstore(r *carutil.Reader) *readStreamBlockstore {
26
26
return &readStreamBlockstore{
27
-
otherBlocks: make(map[cid.Cid]block.Block),
27
+
otherBlocks: make(map[cid.Cid]*carutil.BasicBlock),
28
28
r: r,
29
29
}
30
30
}
31
31
32
-
func (bs *readStreamBlockstore) readUntilBlock(ctx context.Context, cc cid.Cid) (block.Block, error) {
32
+
func (bs *readStreamBlockstore) readUntilBlock(ctx context.Context, cc cid.Cid) (*carutil.BasicBlock, error) {
33
33
for {
34
34
blk, err := bs.r.NextBlock(repoBlockBufferPool, repoBlockBufferSize)
35
35
if err != nil {
···
52
52
}
53
53
54
54
func (bs *readStreamBlockstore) Get(ctx context.Context, cc cid.Cid) (block.Block, error) {
55
+
return bs.get(ctx, cc)
56
+
}
57
+
58
+
func (bs *readStreamBlockstore) get(ctx context.Context, cc cid.Cid) (*carutil.BasicBlock, error) {
55
59
if blk, ok := bs.otherBlocks[cc]; ok {
56
60
delete(bs.otherBlocks, cc)
57
61
return blk, nil
···
70
74
}
71
75
72
76
func (bs *readStreamBlockstore) View(cc cid.Cid, cb func([]byte) error) error {
73
-
blk, err := bs.Get(context.TODO(), cc)
77
+
blk, err := bs.get(context.TODO(), cc)
74
78
if err != nil {
75
79
return err
76
80
}
···
79
83
return err
80
84
}
81
85
82
-
FreeRepoBlock(blk.RawData())
86
+
FreeRepoBlock(blk.BaseBuffer())
83
87
return nil
84
88
}
85
89