From 8d63734af1b7383b31954a98d1e4eae71b5d2b1b Mon Sep 17 00:00:00 2001 From: Brannon King Date: Wed, 21 Jul 2021 15:58:47 -0400 Subject: [PATCH 1/3] added buffer pool for pebble merge string --- claimtrie/node/noderepo/pebble.go | 64 ++++++++++++++++++++++++++++++- 1 file changed, 63 insertions(+), 1 deletion(-) diff --git a/claimtrie/node/noderepo/pebble.go b/claimtrie/node/noderepo/pebble.go index cb5d25e4cc..e5ee75edc1 100644 --- a/claimtrie/node/noderepo/pebble.go +++ b/claimtrie/node/noderepo/pebble.go @@ -2,7 +2,9 @@ package noderepo import ( "bytes" + "io" "sort" + "sync" "github.com/cockroachdb/pebble" "github.com/lbryio/lbcd/claimtrie/change" @@ -13,9 +15,69 @@ type Pebble struct { db *pebble.DB } +type pooledMerger struct { + values [][]byte + index []int + pool *sync.Pool + buffer []byte +} + +func (a *pooledMerger) Len() int { return len(a.index) } +func (a *pooledMerger) Less(i, j int) bool { return a.index[i] < a.index[j] } +func (a *pooledMerger) Swap(i, j int) { + a.index[i], a.index[j] = a.index[j], a.index[i] + a.values[i], a.values[j] = a.values[j], a.values[i] +} + +func (a *pooledMerger) MergeNewer(value []byte) error { + a.values = append(a.values, value) + a.index = append(a.index, len(a.values)) + return nil +} + +func (a *pooledMerger) MergeOlder(value []byte) error { + a.values = append(a.values, value) + a.index = append(a.index, -len(a.values)) + return nil +} + +func (a *pooledMerger) Finish(includesBase bool) ([]byte, io.Closer, error) { + sort.Sort(a) + + a.buffer = a.pool.Get().([]byte)[:0] + for i := range a.values { + a.buffer = append(a.buffer, a.values[i]...) + } + + return a.buffer, a, nil +} + +func (a *pooledMerger) Close() error { + a.pool.Put(a.buffer) + return nil +} + func NewPebble(path string) (*Pebble, error) { - db, err := pebble.Open(path, &pebble.Options{Cache: pebble.NewCache(64 << 20), BytesPerSync: 8 << 20, MaxOpenFiles: 2000}) + mp := &sync.Pool{ + New: func() interface{} { + return make([]byte, 0, 256) + }, + } + + db, err := pebble.Open(path, &pebble.Options{ + Merger: &pebble.Merger{ + Merge: func(key, value []byte) (pebble.ValueMerger, error) { + p := &pooledMerger{pool: mp} + return p, p.MergeNewer(value) + }, + Name: pebble.DefaultMerger.Name, // yes, it's a lie + }, + Cache: pebble.NewCache(64 << 20), + BytesPerSync: 8 << 20, + MaxOpenFiles: 2000, + }) + repo := &Pebble{db: db} return repo, errors.Wrapf(err, "unable to open %s", path) From 92d8b5f20ee81c97aa9f6fab434251f573819f10 Mon Sep 17 00:00:00 2001 From: Jonathan Moody <103143855+moodyjon@users.noreply.github.com> Date: Tue, 24 May 2022 09:00:09 -0400 Subject: [PATCH 2/3] Harden Marshal/Unmarshal logic for Change.SpentChildren. --- claimtrie/change/change.go | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/claimtrie/change/change.go b/claimtrie/change/change.go index aac349c651..95a0e3d5b8 100644 --- a/claimtrie/change/change.go +++ b/claimtrie/change/change.go @@ -78,9 +78,10 @@ func (c *Change) Marshal(enc *bytes.Buffer) error { binary.BigEndian.PutUint32(temp[:4], uint32(len(c.SpentChildren))) enc.Write(temp[:4]) for key := range c.SpentChildren { - binary.BigEndian.PutUint16(temp[:2], uint16(len(key))) // technically limited to 255; not sure we trust it + keySize := uint16(len(key)) + binary.BigEndian.PutUint16(temp[:2], keySize) // technically limited to 255; not sure we trust it enc.Write(temp[:2]) - enc.WriteString(key) + enc.WriteString(key[:keySize]) } } else { binary.BigEndian.PutUint32(temp[:4], 0) From 728ddc4dfa4533d5155a3c4c831d604e29d74537 Mon Sep 17 00:00:00 2001 From: Jonathan Moody <103143855+moodyjon@users.noreply.github.com> Date: Tue, 24 May 2022 09:17:45 -0400 Subject: [PATCH 3/3] Copy value received by MergeOlder/MergeNewer so caller can't trash the merge result by modifying the contents. --- claimtrie/node/noderepo/pebble.go | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) diff --git a/claimtrie/node/noderepo/pebble.go b/claimtrie/node/noderepo/pebble.go index e5ee75edc1..32ca2c0412 100644 --- a/claimtrie/node/noderepo/pebble.go +++ b/claimtrie/node/noderepo/pebble.go @@ -30,13 +30,17 @@ func (a *pooledMerger) Swap(i, j int) { } func (a *pooledMerger) MergeNewer(value []byte) error { - a.values = append(a.values, value) + vc := a.pool.Get().([]byte)[:0] + vc = append(vc, value...) + a.values = append(a.values, vc) a.index = append(a.index, len(a.values)) return nil } func (a *pooledMerger) MergeOlder(value []byte) error { - a.values = append(a.values, value) + vc := a.pool.Get().([]byte)[:0] + vc = append(vc, value...) + a.values = append(a.values, vc) a.index = append(a.index, -len(a.values)) return nil } @@ -53,6 +57,9 @@ func (a *pooledMerger) Finish(includesBase bool) ([]byte, io.Closer, error) { } func (a *pooledMerger) Close() error { + for i := range a.values { + a.pool.Put(a.values[i]) + } a.pool.Put(a.buffer) return nil }