Mem pressure try 3

blockchain/claimtrie.go

@@ -35,7 +35,7 @@ func (b *BlockChain) ParseClaimScripts(block *btcutil.Block, bn *blockNode, view
 	ht := block.Height()
 
 	for _, tx := range block.Transactions() {
-		h := handler{ht, tx, view, map[string][]byte{}}
+		h := handler{ht, tx, view, map[change.ClaimID][]byte{}}
 		if err := h.handleTxIns(b.claimTrie); err != nil {
 			return err
 		}
@@ -67,7 +67,7 @@ type handler struct {
 	ht    int32
 	tx    *btcutil.Tx
 	view  *UtxoViewpoint
-	spent map[string][]byte
+	spent map[change.ClaimID][]byte
 }
 
 func (h *handler) handleTxIns(ct *claimtrie.ClaimTrie) error {

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)

claimtrie/change/claimid.go

@@ -39,8 +39,8 @@ func NewIDFromString(s string) (id ClaimID, err error) {
 }
 
 // Key is for in-memory maps
-func (id ClaimID) Key() string {
-	return string(id[:])
+func (id ClaimID) Key() ClaimID {
+	return id
 }
 
 // String is for anything written to a DB

claimtrie/cmd/cmd/node.go

@@ -52,10 +52,11 @@ func NewNodeDumpCommand() *cobra.Command {
 			}
 			defer repo.Close()
 
-			changes, err := repo.LoadChanges([]byte(name))
+			changes, closer, err := repo.LoadChanges([]byte(name))
 			if err != nil {
 				return errors.Wrapf(err, "load commands")
 			}
+			defer closer()
 
 			for _, chg := range changes {
 				if chg.Height > height {

claimtrie/node/manager.go

@@ -53,10 +53,11 @@ func (nm *BaseManager) NodeAt(height int32, name []byte) (*Node, error) {
 
 	n, changes, oldHeight := nm.cache.fetch(name, height)
 	if n == nil {
-		changes, err := nm.repo.LoadChanges(name)
+		changes, closer, err := nm.repo.LoadChanges(name)
 		if err != nil {
 			return nil, errors.Wrap(err, "in load changes")
 		}
+		defer closer()
 
 		if nm.tempChanges != nil { // making an assumption that we only ever have tempChanges for a single block
 			changes = append(changes, nm.tempChanges[string(name)]...)

claimtrie/node/node.go

@@ -14,12 +14,12 @@ type Node struct {
 	TakenOverAt int32     // The height at when the current BestClaim took over.
 	Claims      ClaimList // List of all Claims.
 	Supports    ClaimList // List of all Supports, including orphaned ones.
-	SupportSums map[string]int64
+	SupportSums map[change.ClaimID]int64
 }
 
 // New returns a new node.
 func New() *Node {
-	return &Node{SupportSums: map[string]int64{}}
+	return &Node{SupportSums: map[change.ClaimID]int64{}}
 }
 
 func (n *Node) HasActiveBestClaim() bool {
@@ -166,7 +166,7 @@ func (n *Node) handleExpiredAndActivated(height int32) int {
 	}
 
 	changes := 0
-	update := func(items ClaimList, sums map[string]int64) ClaimList {
+	update := func(items ClaimList, sums map[change.ClaimID]int64) ClaimList {
 		for i := 0; i < len(items); i++ {
 			c := items[i]
 			if c.Status == Accepted && c.ActiveAt <= height && c.VisibleAt <= height {
@@ -343,7 +343,7 @@ func (n *Node) SortClaimsByBid() {
 func (n *Node) Clone() *Node {
 	clone := New()
 	if n.SupportSums != nil {
-		clone.SupportSums = map[string]int64{}
+		clone.SupportSums = map[change.ClaimID]int64{}
 		for key, value := range n.SupportSums {
 			clone.SupportSums[key] = value
 		}

claimtrie/node/noderepo/noderepo_test.go

@@ -92,8 +92,9 @@ func testNodeRepo(t *testing.T, repo node.Repo, setup, cleanup func()) {
 		err := repo.AppendChanges(tt.changes)
 		r.NoError(err)
 
-		changes, err := repo.LoadChanges(testNodeName1)
+		changes, closer, err := repo.LoadChanges(testNodeName1)
 		r.NoError(err)
+		defer closer()
 		r.Equalf(tt.expected, changes[:len(tt.expected)], tt.name)
 
 		cleanup()
@@ -150,8 +151,9 @@ func testNodeRepo(t *testing.T, repo node.Repo, setup, cleanup func()) {
 			r.NoError(err)
 		}
 
-		changes, err := repo.LoadChanges(testNodeName1)
+		changes, closer, err := repo.LoadChanges(testNodeName1)
 		r.NoError(err)
+		defer closer()
 		r.Equalf(tt.expected, changes[:len(tt.expected)], tt.name)
 
 		cleanup()

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,76 @@ 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 {
+	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 {
+	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) 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 {
+	for i := range a.values {
+		a.pool.Put(a.values[i])
+	}
+	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)
@@ -43,11 +112,11 @@ func (repo *Pebble) AppendChanges(changes []change.Change) error {
 	return errors.Wrap(batch.Commit(pebble.NoSync), "in commit")
 }
 
-func (repo *Pebble) LoadChanges(name []byte) ([]change.Change, error) {
+func (repo *Pebble) LoadChanges(name []byte) ([]change.Change, func(), error) {
 
 	data, closer, err := repo.db.Get(name)
 	if err != nil && err != pebble.ErrNotFound {
-		return nil, errors.Wrapf(err, "in get %s", name) // does returning a name in an error expose too much?
+		return nil, nil, errors.Wrapf(err, "in get %s", name) // does returning a name in an error expose too much?
 	}
 	if closer != nil {
 		defer closer.Close()
@@ -56,17 +125,25 @@ func (repo *Pebble) LoadChanges(name []byte) ([]change.Change, error) {
 	return unmarshalChanges(name, data)
 }
 
-func unmarshalChanges(name, data []byte) ([]change.Change, error) {
-	// data is 84bytes+ per change
-	changes := make([]change.Change, 0, len(data)/84+1) // average is 5.1 changes
+var changesPool = sync.Pool{
+	New: func() interface{} {
+		return make([]change.Change, 0, 6)
+	},
+}
+
+func unmarshalChanges(name, data []byte) ([]change.Change, func(), error) {
+	// data is 84bytes+ per change, average is 5.1 changes
+	changes := changesPool.Get().([]change.Change)[:0]
+	closer := func() { changesPool.Put(changes) }
 
 	buffer := bytes.NewBuffer(data)
 	sortNeeded := false
 	for buffer.Len() > 0 {
 		var chg change.Change
 		err := chg.Unmarshal(buffer)
 		if err != nil {
-			return nil, errors.Wrap(err, "in decode")
+			closer()
+			return nil, nil, errors.Wrap(err, "in decode")
 		}
 		chg.Name = name
 		if len(changes) > 0 && chg.Height < changes[len(changes)-1].Height {
@@ -81,14 +158,17 @@ func unmarshalChanges(name, data []byte) ([]change.Change, error) {
 			return changes[i].Height < changes[j].Height
 		})
 	}
-	return changes, nil
+
+	return changes, closer, nil
 }
 
 func (repo *Pebble) DropChanges(name []byte, finalHeight int32) error {
-	changes, err := repo.LoadChanges(name)
+	changes, closer, err := repo.LoadChanges(name)
 	if err != nil {
 		return errors.Wrapf(err, "in load changes for %s", name)
 	}
+	defer closer()
+
 	buffer := bytes.NewBuffer(nil)
 	for i := 0; i < len(changes); i++ { // assuming changes are ordered by height
 		if changes[i].Height > finalHeight {
@@ -137,10 +217,11 @@ func (repo *Pebble) IterateChildren(name []byte, f func(changes []change.Change)
 
 	for iter.First(); iter.Valid(); iter.Next() {
 		// NOTE! iter.Key() is ephemeral!
-		changes, err := unmarshalChanges(iter.Key(), iter.Value())
+		changes, closer, err := unmarshalChanges(iter.Key(), iter.Value())
 		if err != nil {
 			return errors.Wrapf(err, "from unmarshaller at %s", iter.Key())
 		}
+		defer closer()
 		if !f(changes) {
 			break
 		}

claimtrie/node/repo.go

@@ -13,7 +13,9 @@ type Repo interface {
 
 	// LoadChanges loads changes of a node up to (includes) the specified height.
 	// If no changes found, both returned slice and error will be nil.
-	LoadChanges(name []byte) ([]change.Change, error)
+	// The returned closer func() should be called to release changes after
+	// work on them is finished.
+	LoadChanges(name []byte) (changes []change.Change, closer func(), err error)
 
 	DropChanges(name []byte, finalHeight int32) error
 

config.go

@@ -117,6 +117,7 @@ type config struct {
 	ConfigFile           string        `short:"C" long:"configfile" description:"Path to configuration file"`
 	ConnectPeers         []string      `long:"connect" description:"Connect only to the specified peers at startup"`
 	CPUProfile           string        `long:"cpuprofile" description:"Write CPU profile to the specified file"`
+	MemProfile           string        `long:"memprofile" description:"Write memory profile to the specified file"`
 	DataDir              string        `short:"b" long:"datadir" description:"Directory to store data"`
 	DbType               string        `long:"dbtype" description:"Database backend to use for the Block Chain"`
 	DebugLevel           string        `short:"d" long:"debuglevel" description:"Logging level for all subsystems {trace, debug, info, warn, error, critical} -- You may also specify <subsystem>=<level>,<subsystem2>=<level>,... to set the log level for individual subsystems -- Use show to list available subsystems"`